1import * as Cloudflare from "@/Cloudflare";
2import { CloudflareEnvironment } from "@/Cloudflare/CloudflareEnvironment";
3import * as Test from "@/Test/Alchemy";
4import * as Output from "@/Output";
5import * as ai from "@distilled.cloud/cloudflare/ai";
6import * as queues from "@distilled.cloud/cloudflare/queues";
7import { expect } from "alchemy-test";
8import * as Effect from "effect/Effect";
9import * as Schema from "effect/Schema";
10import * as Schedule from "effect/Schedule";
11import * as HttpClient from "effect/http/HttpClient";
12
13const { test } = Test.make({ providers: Cloudflare.providers() });
14
15test.provider(
16 "a model handle validates, replaces and preserves catalog models on destroy",
17 (stack) =>
18 Effect.gen(function* () {
19 yield* stack.destroy();
20 const { accountId } = yield* yield* CloudflareEnvironment;
21 const first = yield* stack.deploy(
22 Cloudflare.AI.Model("Model", { modelName: "@cf/baai/bge-m3" }),
23 );
24 expect(first).toEqual({ accountId, modelName: "@cf/baai/bge-m3" });
25 const updated = yield* stack.deploy(
26 Cloudflare.AI.Model("Model", {
27 modelName: "@cf/baai/bge-base-en-v1.5",
28 }),
29 );
30 expect(updated.modelName).toBe("@cf/baai/bge-base-en-v1.5");
31 yield* ai.getModelSchema({ accountId, model: first.modelName });
32 yield* stack.destroy();
33 yield* ai.getModelSchema({ accountId, model: updated.modelName });
34 }).pipe(Effect.ensuring(stack.destroy().pipe(Effect.orDie))),
35 {
36 tags: ["provider:cloudflare", "provider:cloudflare:ai", "live"],
37 timeout: 120_000,
38 exclusive: true,
39 },
40);
41
42test.provider(
43 "an unresolved model identity replaces its downstream subscription",
44 (stack) =>
45 Effect.gen(function* () {
46 yield* stack.destroy();
47 const program = (suffix: string) =>
48 Effect.gen(function* () {
49 const identity = yield* Cloudflare.R2.Bucket("Identity", {
50 name: `alchemy-model-subscription-identity-${suffix}`,
51 });
52 const model = yield* Cloudflare.AI.Model("Model", {
53 modelName: identity.bucketName.pipe(
54 Output.map((name) =>
55 name.endsWith("-a")
56 ? "@cf/baai/bge-m3"
57 : "@cf/baai/bge-base-en-v1.5",
58 ),
59 ),
60 });
61 const queue = yield* Cloudflare.Queues.Queue("Queue");
62 const subscription = yield* Cloudflare.Queues.Subscription("Events", {
63 source: model,
64 events: ["batch.queued"],
65 queueId: queue.queueId,
66 });
67 return { model, subscription };
68 });
69 const initial = yield* stack.deploy(program("a"));
70 const replaced = yield* stack.deploy(program("b"));
71 expect(replaced.model.modelName).toBe("@cf/baai/bge-base-en-v1.5");
72 expect(replaced.subscription.subscriptionId).not.toBe(
73 initial.subscription.subscriptionId,
74 );
75 const observed = yield* queues.getSubscription({
76 accountId: replaced.subscription.accountId,
77 subscriptionId: replaced.subscription.subscriptionId,
78 });
79 expect(observed.source).toEqual(
80 expect.objectContaining({
81 type: "workersAi.model",
82 modelName: replaced.model.modelName,
83 }),
84 );
85 yield* stack.destroy();
86 }).pipe(Effect.ensuring(stack.destroy().pipe(Effect.orDie))),
87 {
88 tags: [
89 "provider:cloudflare",
90 "provider:cloudflare:ai",
91 "provider:cloudflare:queue",
92 "provider:cloudflare:r2",
93 "live",
94 ],
95 timeout: 120_000,
96 exclusive: true,
97 },
98);
99
100test.provider(
101 "a persisted model reference receives a real asynchronous batch event",
102 (stack) =>
103 Effect.gen(function* () {
104 yield* stack.destroy();
105 const { accountId } = yield* yield* CloudflareEnvironment;
106 const model = Cloudflare.AI.Model("Model", {
107 modelName: "@cf/baai/bge-m3",
108 });
109 yield* stack.deploy(model);
110 const deployed = yield* stack.deploy(
111 Effect.gen(function* () {
112 yield* model;
113 const queue = yield* Cloudflare.Queues.Queue("BatchEvents");
114 const worker = yield* Cloudflare.Worker("BatchWorker", {
115 main: `${import.meta.dirname}/fixtures/model-batch.ts`,
116 env: { AI: Cloudflare.Workers.AI() },
117 });
118 const subscription = yield* Cloudflare.Queues.Subscription(
119 "ModelEvents",
120 {
121 source: yield* Cloudflare.AI.Model.ref("Model"),
122 events: ["batch.queued", "batch.succeeded", "batch.failed"],
123 queueId: queue.queueId,
124 },
125 );
126 return { queue, worker, subscription };
127 }),
128 );
129 yield* queues.createConsumer({
130 accountId,
131 queueId: deployed.queue.queueId,
132 type: "http_pull",
133 });
134 const client = HttpClient.filterStatusOk(yield* HttpClient.HttpClient);
135 yield* client.get(deployed.worker.url!).pipe(
136 Effect.flatMap((response) => response.text),
137 Effect.retry({ schedule: Schedule.spaced("2 seconds"), times: 8 }),
138 );
139 const batch = yield* client.post(deployed.worker.url!).pipe(
140 Effect.flatMap((response) => response.json),
141 Effect.flatMap(
142 Schema.decodeUnknownEffect(
143 Schema.Struct({ request_id: Schema.String }),
144 ),
145 ),
146 );
147 const bodies: string[] = [];
148 const delivered = () =>
149 bodies.some(
150 (body) =>
151 body.includes("cf.workersAi.model.batch.") &&
152 body.includes(batch.request_id) &&
153 body.includes(deployed.subscription.subscriptionId) &&
154 body.includes(accountId),
155 );
156 yield* Effect.gen(function* () {
157 const pulled = yield* queues
158 .pullMessage({
159 accountId,
160 queueId: deployed.queue.queueId,
161 batchSize: 100,
162 visibilityTimeoutMs: 1000,
163 })
164 .pipe(
165 Effect.retry({
166 while: (error) => error._tag === "QueueHttpPullNotEnabled",
167 schedule: Schedule.spaced("2 seconds"),
168 times: 8,
169 }),
170 );
171 for (const message of pulled.messages ?? [])
172 if (message.body) bodies.push(message.body);
173 const acks = (pulled.messages ?? []).flatMap((message) =>
174 message.leaseId ? [{ leaseId: message.leaseId }] : [],
175 );
176 if (acks.length)
177 yield* queues.ackMessage({
178 accountId,
179 queueId: deployed.queue.queueId,
180 acks,
181 });
182 }).pipe(
183 Effect.repeat({
184 schedule: Schedule.spaced("5 seconds"),
185 times: 10,
186 until: delivered,
187 }),
188 Effect.timeout("60 seconds"),
189 );
190 expect(delivered()).toBe(true);
191 yield* stack.deploy(model);
192 yield* ai.getModelSchema({ accountId, model: "@cf/baai/bge-m3" });
193 yield* stack.destroy();
194 }).pipe(Effect.ensuring(stack.destroy().pipe(Effect.orDie))),
195 {
196 tags: [
197 "provider:cloudflare",
198 "provider:cloudflare:ai",
199 "provider:cloudflare:queue",
200 "provider:cloudflare:worker",
201 "live",
202 ],
203 timeout: 120_000,
204 exclusive: true,
205 },
206);