1import * as Cloudflare from "@/Cloudflare";
2import { CloudflareEnvironment } from "@/Cloudflare/CloudflareEnvironment";
3import * as Test from "@/Test/Alchemy";
4import * as images from "@distilled.cloud/cloudflare/images";
5import * as queues from "@distilled.cloud/cloudflare/queues";
6import { expect } from "alchemy-test";
7import * as Effect from "effect/Effect";
8import * as Schedule from "effect/Schedule";
9import * as Schema from "effect/Schema";
10import {
11 makeSubscriptionCleanup,
12 matchesSubscriptionEvent,
13 SubscriptionEvent,
14} from "../Queue/SubscriptionSources.ts";
15
16const { test } = Test.make({ providers: Cloudflare.providers() });
17const canUpload = process.env.CLOUDFLARE_TEST_IMAGES_UPLOAD === "1";
18
19test.provider.skipIf(canUpload)(
20 "image uploads expose the typed entitlement rejection",
21 (stack) =>
22 Effect.gen(function* () {
23 yield* stack.destroy();
24 const { accountId } = yield* yield* CloudflareEnvironment;
25 const cleanup = makeSubscriptionCleanup();
26 const error = yield* Effect.acquireRelease(
27 images.createV1({
28 accountId,
29 id: "alchemy-subscription-entitlement-probe",
30 url: "https://developers.cloudflare.com/og-docs.png",
31 }),
32 (image) =>
33 image.id
34 ? images.deleteV1({ accountId, imageId: image.id }).pipe(cleanup)
35 : Effect.void,
36 ).pipe(Effect.scoped, Effect.flip);
37 expect(error).toMatchObject({
38 _tag: "ImagesAccessNotEnabled",
39 code: 5403,
40 });
41 yield* stack.destroy();
42 }).pipe(
43 Effect.ensuring(
44 stack
45 .destroy()
46 .pipe(
47 Effect.timeout("15 seconds"),
48 Effect.orDie,
49 Effect.interruptible,
50 ),
51 ),
52 ),
53 { tags: ["provider:cloudflare", "provider:cloudflare:images", "live"] },
54);
55
56// Uploads require Images entitlement; the testing account returns ImagesAccessNotEnabled (5403).
57test.provider.skipIf(!canUpload)(
58 "an Images variant reference receives an account-wide upload event",
59 (stack) =>
60 Effect.gen(function* () {
61 yield* stack.destroy();
62 const { accountId } = yield* yield* CloudflareEnvironment;
63 const cleanup = makeSubscriptionCleanup();
64 const variant = Cloudflare.Images.Variant("Variant", {
65 name: "alchemySubscriptionUpload",
66 fit: "contain",
67 width: 100,
68 height: 100,
69 });
70 yield* stack.deploy(variant);
71 const deployed = yield* stack.deploy(
72 Effect.gen(function* () {
73 const source = yield* variant;
74 const queue = yield* Cloudflare.Queues.Queue("ImageEvents");
75 const subscription = yield* Cloudflare.Queues.Subscription(
76 "Uploads",
77 {
78 source: yield* Cloudflare.Images.Variant.ref("Variant"),
79 events: ["image.uploaded"],
80 queueId: queue.queueId,
81 },
82 );
83 return { source, queue, subscription };
84 }),
85 );
86 yield* queues.createConsumer({
87 accountId,
88 queueId: deployed.queue.queueId,
89 type: "http_pull",
90 });
91 yield* Effect.gen(function* () {
92 const upload = (suffix: string) =>
93 Effect.acquireRelease(
94 images.createV1({
95 accountId,
96 id: `${deployed.subscription.subscriptionId}-${suffix}`,
97 url: "https://developers.cloudflare.com/og-docs.png",
98 }),
99 (image) =>
100 image.id
101 ? images.deleteV1({ accountId, imageId: image.id }).pipe(
102 Effect.catchTag("ImageNotFound", () => Effect.void),
103 cleanup,
104 )
105 : Effect.void,
106 ).pipe(
107 Effect.flatMap((image) =>
108 image.id
109 ? Effect.succeed(image.id)
110 : Effect.fail(new Error("Image upload returned no identity")),
111 ),
112 );
113 const events: SubscriptionEvent[] = [];
114 const delivered = (imageId: string) =>
115 events.some((event) =>
116 matchesSubscriptionEvent(event, {
117 source: "images",
118 type: "image.uploaded",
119 accountId,
120 subscriptionId: deployed.subscription.subscriptionId,
121 identity: imageId,
122 }),
123 );
124 const pull = Effect.gen(function* () {
125 const pulled = yield* queues
126 .pullMessage({
127 accountId,
128 queueId: deployed.queue.queueId,
129 batchSize: 100,
130 visibilityTimeoutMs: 30_000,
131 })
132 .pipe(
133 Effect.retry({
134 while: (error) => error._tag === "QueueHttpPullNotEnabled",
135 schedule: Schedule.spaced("2 seconds"),
136 times: 8,
137 }),
138 );
139 const decoded = yield* Effect.forEach(
140 pulled.messages ?? [],
141 ({ body }) => Schema.decodeUnknownEffect(SubscriptionEvent)(body),
142 );
143 events.push(...decoded);
144 if (decoded.length)
145 yield* Effect.logInfo("Image subscription receipts", decoded);
146 const acks = (pulled.messages ?? []).flatMap(({ leaseId }) =>
147 leaseId ? [{ leaseId }] : [],
148 );
149 if (acks.length) {
150 const result = yield* queues.ackMessage({
151 accountId,
152 queueId: deployed.queue.queueId,
153 acks,
154 });
155 expect(result.ackCount).toBe(acks.length);
156 expect(Object.keys(result.warnings ?? {})).toHaveLength(0);
157 }
158 });
159 yield* pull;
160 const probes: string[] = [];
161 const ready = () => probes.some(delivered);
162 yield* Effect.gen(function* () {
163 probes.push(yield* upload(`probe-${probes.length}`));
164 yield* pull;
165 }).pipe(
166 Effect.repeat({
167 schedule: Schedule.spaced("3 seconds"),
168 times: 8,
169 until: ready,
170 }),
171 );
172 expect(ready()).toBe(true);
173 // A readiness probe must not satisfy the final upload assertion.
174 const imageId = yield* upload("final");
175 expect(probes).not.toContain(imageId);
176 yield* pull.pipe(
177 Effect.repeat({
178 schedule: Schedule.spaced("3 seconds"),
179 times: 8,
180 until: () => delivered(imageId),
181 }),
182 );
183 expect(delivered(imageId)).toBe(true);
184 }).pipe(Effect.timeout("60 seconds"), Effect.scoped);
185 yield* stack.deploy(variant);
186 yield* images.getV1Variant({
187 accountId,
188 variantId: deployed.source.variantName,
189 });
190 yield* stack.destroy();
191 }).pipe(
192 Effect.scoped,
193 Effect.ensuring(
194 stack
195 .destroy()
196 .pipe(
197 Effect.timeout("15 seconds"),
198 Effect.orDie,
199 Effect.interruptible,
200 ),
201 ),
202 ),
203 {
204 tags: [
205 "provider:cloudflare",
206 "provider:cloudflare:images",
207 "provider:cloudflare:queue",
208 "live",
209 ],
210 timeout: 120_000,
211 exclusive: true,
212 },
213);