1import * as Cloudflare from "alchemy/Cloudflare";
2import * as Test from "alchemy/Test/Bun";
3import { expect } from "bun:test";
4import * as Data from "effect/Data";
5import * as Effect from "effect/Effect";
6import { cast } from "effect/Function";
7import * as Schedule from "effect/Schedule";
8import * as Schema from "effect/Schema";
9import * as HttpBody from "effect/http/HttpBody";
10import * as HttpClient from "effect/http/HttpClient";
11import * as HttpClientResponse from "effect/http/HttpClientResponse";
12import assert from "node:assert";
13import * as fs from "node:fs";
14import * as path from "node:path";
15import Stack, { SECRETS_STORE_VALUE } from "../alchemy.run.ts";
16import type { Message } from "../src/AsyncWorker.ts";
17import { WORKFLOW_SECRET_VALUE } from "../src/NotifyWorkflow.ts";
18
19const { test, beforeAll, afterAll, deploy, destroy } = Test.make({
20 providers: Cloudflare.providers(),
21 state: Cloudflare.state(),
22 dev: true,
23});
24
25const stack = beforeAll(
26 deploy(Stack).pipe(
27 Effect.flatMap(
28 Effect.fn(function* (outputs) {
29 const {
30 asyncWorker,
31 effectWorker,
32 mediaWorker,
33 tailWorker,
34 inboxWorker,
35 liveKvNamespaceId,
36 secretsStoreId,
37 secretsSecretId,
38 } = outputs;
39 assert(typeof asyncWorker === "string");
40 assert(typeof effectWorker === "string");
41 assert(typeof mediaWorker === "string");
42 assert(typeof tailWorker === "string");
43 assert(typeof inboxWorker === "string");
44 assert(typeof liveKvNamespaceId === "string");
45 assert(typeof secretsStoreId === "string");
46 assert(typeof secretsSecretId === "string");
47 const hyperdrive = (outputs as { hyperdrive?: string }).hyperdrive;
48 yield* Effect.forEach(
49 [asyncWorker, effectWorker, mediaWorker, tailWorker, inboxWorker],
50 (url) =>
51 HttpClient.get(url).pipe(
52 Effect.flatMap(HttpClientResponse.filterStatusOk),
53 Effect.retry({
54 schedule: Schedule.max([
55 Schedule.spaced("250 millis"),
56 Schedule.recurs(40),
57 ]),
58 }),
59 ),
60 );
61 return {
62 asyncWorker,
63 effectWorker,
64 mediaWorker,
65 tailWorker,
66 inboxWorker,
67 liveKvNamespaceId,
68 secretsStoreId,
69 secretsSecretId,
70 hyperdrive,
71 };
72 }),
73 ),
74 ),
75 { timeout: 300_000 },
76);
77
78afterAll.skipIf(!!process.env.NO_DESTROY)(destroy(Stack));
79
80class NotYet extends Data.TaggedError("NotYet")<{ status?: number }> {}
81
82/**
83 * GET a route, retrying while it serves a non-200 — used for routes whose
84 * first request does heavy lazy work (launching Chrome, warming sharp).
85 */
86const getJsonReady = (
87 url: string | URL,
88 {
89 times = 10,
90 spaced = "2 seconds",
91 }: { times?: number; spaced?: `${number} ${"millis" | "seconds"}` } = {},
92) =>
93 Effect.gen(function* () {
94 const res = yield* HttpClient.get(url);
95 if (res.status !== 200) {
96 return yield* Effect.fail(new NotYet({ status: res.status }));
97 }
98 return yield* res.json;
99 }).pipe(
100 Effect.retry({
101 while: (e) => e._tag === "NotYet",
102 schedule: Schedule.max([Schedule.spaced(spaced), Schedule.recurs(times)]),
103 }),
104 );
105
106/** 8x4 solid red PNG (checked-in fixture — never generated at test time). */
107const PNG_RED_8X4 = Uint8Array.from(
108 Buffer.from(
109 "iVBORw0KGgoAAAANSUhEUgAAAAgAAAAECAIAAAA8r+mnAAAACXBIWXMAAAPoAAAD6AG1e1JrAAAAEklEQVR4nGP4z8CAFWEXJUsCAFpeH+EeQoQoAAAAAElFTkSuQmCC",
110 "base64",
111 ),
112);
113
114/**
115 * Find a persisted `.eml` under the local email storage dir whose content
116 * carries `needle`. The simulator writes relative to the process cwd, so
117 * check both the example dir and the current cwd.
118 */
119class EmlNotFound extends Data.TaggedError("EmlNotFound")<{}> {}
120const findEmlContaining = (needle: string) =>
121 Effect.sync(() => {
122 const candidates = [
123 path.resolve(import.meta.dirname, "..", ".alchemy", "local", "email"),
124 path.resolve(process.cwd(), ".alchemy", "local", "email"),
125 ];
126 for (const dir of candidates) {
127 let files: string[];
128 try {
129 files = fs.readdirSync(dir);
130 } catch {
131 continue;
132 }
133 for (const name of files) {
134 if (!name.endsWith(".eml")) continue;
135 const content = fs.readFileSync(path.join(dir, name), "utf8");
136 if (content.includes(needle)) return content;
137 }
138 }
139 return undefined;
140 }).pipe(
141 Effect.flatMap((content) =>
142 content === undefined
143 ? Effect.fail(new EmlNotFound({}))
144 : Effect.succeed(content),
145 ),
146 Effect.retry({
147 while: (e) => e._tag === "EmlNotFound",
148 schedule: Schedule.max([
149 Schedule.spaced("500 millis"),
150 Schedule.recurs(20),
151 ]),
152 }),
153 );
154
155test(
156 "deploys all workers with URLs",
157 Effect.gen(function* () {
158 const { asyncWorker, effectWorker, mediaWorker, tailWorker, inboxWorker } =
159 yield* stack;
160
161 // Local dev proxy URLs — proof no cloud deploy ran for the workers.
162 for (const url of [
163 asyncWorker,
164 effectWorker,
165 mediaWorker,
166 tailWorker,
167 inboxWorker,
168 ]) {
169 expect(url).toBeString();
170 expect(url).toMatch(/^http:\/\/localhost:\d+/);
171 }
172 }),
173);
174
175/**
176 * AsyncWorker exports a default fetch handler that calls the `Counter`
177 * Durable Object's `increment()` and returns `Hello, world! <n>`.
178 *
179 * Hitting the worker twice exercises the DO end-to-end and proves
180 * persistent state across requests — if the DO binding is missing or
181 * the class export is wrong, the first request fails outright.
182 */
183test(
184 "AsyncWorker increments the Counter Durable Object across requests",
185 Effect.gen(function* () {
186 const { asyncWorker } = yield* stack;
187
188 const first = yield* HttpClient.get(new URL("/counter", asyncWorker));
189 expect(first.status).toBe(200);
190 const firstBody = yield* first.text;
191 const firstMatch = firstBody.match(/^Hello, world! (\d+)$/);
192 expect(firstMatch).not.toBeNull();
193 const firstCount = Number(firstMatch![1]);
194
195 const second = yield* HttpClient.get(new URL("/counter", asyncWorker));
196 expect(second.status).toBe(200);
197 const secondBody = yield* second.text;
198 const secondMatch = secondBody.match(/^Hello, world! (\d+)$/);
199 expect(secondMatch).not.toBeNull();
200 const secondCount = Number(secondMatch![1]);
201
202 expect(secondCount).toBe(firstCount + 1);
203 }),
204);
205
206/**
207 * Under `dev: true` the R2 bucket is emulated by the local provider and the
208 * `r2_bucket` binding is served by the local workerd R2 simulator — no cloud
209 * bucket is ever created.
210 */
211test(
212 "AsyncWorker reads and writes the local R2 bucket",
213 Effect.gen(function* () {
214 const { asyncWorker } = yield* stack;
215 const response = yield* HttpClient.get(new URL("/r2", asyncWorker));
216 expect(response.status).toBe(200);
217 const body = (yield* response.json) as { text: string; keys: string[] };
218 expect(body.text).toBe("hello from r2");
219 expect(body.keys).toContain("hello.txt");
220 }),
221);
222
223/**
224 * Same for D1: the database is a local `dev:` row and queries run against
225 * the local workerd D1 simulator (DO SQLite).
226 */
227test(
228 "AsyncWorker queries the local D1 database",
229 Effect.gen(function* () {
230 const { asyncWorker } = yield* stack;
231 const response = yield* HttpClient.get(new URL("/d1", asyncWorker));
232 expect(response.status).toBe(200);
233 const body = (yield* response.json) as { text: string | null };
234 expect(body.text).toBe("hello from d1");
235 }),
236);
237
238test(
239 "AsyncWorker serves assets",
240 Effect.gen(function* () {
241 const { asyncWorker } = yield* stack;
242 const response = yield* HttpClient.get(new URL("/", asyncWorker));
243 expect(response.status).toBe(200);
244 const body = yield* response.text;
245 expect(body).toMatch("<h1>Hello, world!</h1>");
246 }),
247);
248
249test(
250 "AsyncWorker receives bindings, including variables and secrets",
251 Effect.gen(function* () {
252 const { asyncWorker } = yield* stack;
253 const response = yield* HttpClient.get(new URL("/env", asyncWorker));
254 expect(response.status).toBe(200);
255 const body = yield* response.json;
256 expect(body).toMatchObject({
257 MY_SECRET: "my-secret-abc123",
258 MY_VARIABLE: "my-variable-abc123",
259 COUNTER: {},
260 });
261 }),
262);
263
264/**
265 * `env: { PUBLIC_URL: Cloudflare.Worker.URL }` lowers the `self_url`
266 * sentinel to a plain-text binding holding the worker's own URL. Under the
267 * local provider (this suite runs with `dev: true`) that URL is the dev
268 * proxy's address, resolved before workerd starts. The binding value has no
269 * trailing slash; the stack's URL output does — normalize before comparing.
270 */
271test(
272 "AsyncWorker receives its own URL via env: { PUBLIC_URL: Worker.URL }",
273 Effect.gen(function* () {
274 const { asyncWorker } = yield* stack;
275 const response = yield* HttpClient.get(new URL("/env", asyncWorker));
276 expect(response.status).toBe(200);
277 const body = (yield* response.json) as { PUBLIC_URL: string };
278 expect(body.PUBLIC_URL).toBe(asyncWorker.replace(/\/$/, ""));
279 }),
280);
281
282/**
283 * The Effect-native form: `yield* Cloudflare.Worker.URL` at init attaches
284 * the `self_url` binding and returns a deferred accessor the handler reads
285 * at request time.
286 */
287test(
288 "EffectWorker reads its own URL via yield* Worker.URL",
289 Effect.gen(function* () {
290 const { effectWorker } = yield* stack;
291 const response = yield* HttpClient.get(new URL("/url", effectWorker));
292 expect(response.status).toBe(200);
293 const body = (yield* response.json) as { url: string };
294 expect(body.url).toBe(effectWorker.replace(/\/$/, ""));
295 }),
296);
297
298test(
299 "AsyncWorker sends and receives messages on the queue",
300 Effect.gen(function* () {
301 const { asyncWorker } = yield* stack;
302 const body = { text: "hello", sentAt: Date.now() };
303 yield* HttpClient.post(new URL("/queue/send", asyncWorker), {
304 body: yield* HttpBody.json(body),
305 }).pipe(Effect.flatMap(HttpClientResponse.filterStatusOk));
306 const message = yield* HttpClient.get(
307 new URL("/queue/messages", asyncWorker),
308 ).pipe(
309 Effect.flatMap(HttpClientResponse.filterStatusOk),
310 Effect.flatMap((res) => res.json),
311 Effect.map(cast<Schema.Json, Array<Message>>),
312 Effect.map((messages) =>
313 messages.find((m) => m.body.sentAt === body.sentAt),
314 ),
315 Effect.filterOrFail(
316 (message) => message !== undefined,
317 () => ({ _tag: "MessageNotFound" }) as const,
318 ),
319 Effect.retry({
320 while: (error) => error._tag === "MessageNotFound",
321 schedule: Schedule.max([
322 Schedule.spaced("250 millis"),
323 Schedule.recurs(25),
324 ]),
325 }),
326 );
327 expect(message).toMatchObject({
328 id: expect.any(String),
329 body,
330 });
331 }),
332 { timeout: 10_000 },
333);
334
335/**
336 * EffectWorker binds a KV namespace via `Cloudflare.KV.ReadWriteNamespace(KV)`
337 * and returns the result of `kv.list()` as JSON. A successful response
338 * proves the Effect-style binding wired the runtime SDK and the
339 * `WorkerEnvironment` service was provisioned for the fetch handler.
340 */
341test(
342 "EffectWorker returns a KV list result via the Effect KV binding",
343 Effect.gen(function* () {
344 const { effectWorker } = yield* stack;
345
346 const response = yield* HttpClient.get(effectWorker);
347 expect(response.status).toBe(200);
348
349 const body = (yield* response.json) as {
350 keys: Array<{ name: string }>;
351 list_complete: boolean;
352 };
353 expect(Array.isArray(body.keys)).toBe(true);
354 expect(typeof body.list_complete).toBe("boolean");
355 }),
356);
357
358test(
359 "EffectWorker sends and receives messages on the queue",
360 Effect.gen(function* () {
361 const { effectWorker } = yield* stack;
362 const body = { text: "hello", sentAt: Date.now() };
363 yield* HttpClient.post(new URL("/queue/send", effectWorker), {
364 body: yield* HttpBody.json(body),
365 }).pipe(Effect.flatMap(HttpClientResponse.filterStatusOk));
366 const message = yield* HttpClient.get(
367 new URL("/queue/messages", effectWorker),
368 ).pipe(
369 Effect.flatMap(HttpClientResponse.filterStatusOk),
370 Effect.flatMap((res) => res.json),
371 Effect.map(cast<Schema.Json, Array<Message>>),
372 Effect.map((messages) =>
373 messages.find((m) => m.body.sentAt === body.sentAt),
374 ),
375 Effect.filterOrFail(
376 (message) => message !== undefined,
377 () => ({ _tag: "MessageNotFound" }) as const,
378 ),
379 Effect.retry({
380 while: (error) => error._tag === "MessageNotFound",
381 schedule: Schedule.max([
382 Schedule.spaced("250 millis"),
383 Schedule.recurs(25),
384 ]),
385 }),
386 );
387 expect(message).toMatchObject({
388 id: expect.any(String),
389 body,
390 });
391 }),
392 { timeout: 10_000 },
393);
394
395/**
396 * Both workers import `./modules/wasm-example.wasm`, which exports a
397 * single `add(a: number, b: number): number` function. Hitting `/wasm`
398 * instantiates the module and returns `add(3, 4)` as JSON, proving that
399 * the bundler ships the wasm asset to workerd and that runtime
400 * `WebAssembly.instantiate` works for both the raw async-handler and
401 * Effect-style entrypoints.
402 */
403test(
404 "AsyncWorker /wasm instantiates the wasm module and returns add(3, 4)",
405 Effect.gen(function* () {
406 const { asyncWorker } = yield* stack;
407
408 const response = yield* HttpClient.get(new URL("/wasm", asyncWorker));
409 expect(response.status).toBe(200);
410 const body = (yield* response.json) as { result: number };
411 expect(body.result).toBe(7);
412 }),
413);
414
415test(
416 "EffectWorker /wasm instantiates the wasm module and returns add(3, 4)",
417 Effect.gen(function* () {
418 const { effectWorker } = yield* stack;
419
420 const response = yield* HttpClient.get(new URL("/wasm", effectWorker));
421 expect(response.status).toBe(200);
422 const body = (yield* response.json) as { result: number };
423 expect(body.result).toBe(7);
424 }),
425);
426
427/**
428 * Cache API: the local runtime ships an always-on `caches.default`
429 * simulator persisted under `.alchemy/local`. A per-run key keeps the
430 * first request a deterministic miss even though the storage survives
431 * across runs.
432 */
433test(
434 "AsyncWorker caches responses with the Cache API (miss then hit)",
435 Effect.gen(function* () {
436 const { asyncWorker } = yield* stack;
437 const key = crypto.randomUUID();
438
439 const first = (yield* (yield* HttpClient.get(
440 new URL(`/cache?key=${key}`, asyncWorker),
441 )).json) as { hit: boolean; body: string };
442 expect(first.hit).toBe(false);
443 expect(first.body).toBe("cached-body");
444
445 const second = (yield* (yield* HttpClient.get(
446 new URL(`/cache?key=${key}`, asyncWorker),
447 )).json) as { hit: boolean; body: string };
448 expect(second.hit).toBe(true);
449 expect(second.body).toBe("cached-body");
450 }),
451);
452
453/**
454 * Rate limit binding: THROTTLE allows 2 requests per 10s per key. The local
455 * runtime implements real throttling (not a stub), so the third call with
456 * the same (per-run) key observes `success: false`.
457 */
458test(
459 "AsyncWorker rate limit binding throttles the third call",
460 Effect.gen(function* () {
461 const { asyncWorker } = yield* stack;
462 const key = crypto.randomUUID();
463 const url = new URL(`/ratelimit?key=${key}`, asyncWorker);
464
465 const outcomes: boolean[] = [];
466 for (let i = 0; i < 3; i++) {
467 const res = yield* HttpClient.get(url);
468 expect(res.status).toBe(200);
469 const body = (yield* res.json) as { success: boolean };
470 outcomes.push(body.success);
471 }
472 expect(outcomes).toEqual([true, true, false]);
473 }),
474);
475
476/**
477 * Version metadata binding: locally stubbed with a random `id` (and empty
478 * `tag`), matching the production binding's shape.
479 */
480test(
481 "AsyncWorker exposes version metadata",
482 Effect.gen(function* () {
483 const { asyncWorker } = yield* stack;
484 const response = yield* HttpClient.get(new URL("/version", asyncWorker));
485 expect(response.status).toBe(200);
486 const body = (yield* response.json) as { id: string };
487 expect(typeof body.id).toBe("string");
488 expect(body.id.length).toBeGreaterThan(0);
489 }),
490);
491
492/**
493 * Service binding (worker → worker): AsyncWorker's `/service` route calls
494 * the Effect-native worker's `/url` route through `env.SERVICE.fetch` — no
495 * public hop. In local dev this resolves through the dev registry, which
496 * may pick a fresh peer up asynchronously — hence the bounded retry.
497 */
498test(
499 "AsyncWorker calls EffectWorker through a service binding",
500 Effect.gen(function* () {
501 const { asyncWorker, effectWorker } = yield* stack;
502 const body = (yield* getJsonReady(new URL("/service", asyncWorker))) as {
503 url: string;
504 };
505 expect(body.url).toBe(effectWorker.replace(/\/$/, ""));
506 }),
507 { timeout: 60_000 },
508);
509
510/**
511 * tailConsumers: AsyncWorker lists TailWorker as a tail consumer, so every
512 * invocation delivers a trace batch (outcome + console logs) to its `tail()`
513 * handler, which records into KV. Each poll re-invokes the producer so an
514 * early dropped batch (registry still warming) doesn't strand the test.
515 */
516test(
517 "TailWorker receives AsyncWorker's trace batches",
518 Effect.gen(function* () {
519 const { asyncWorker, tailWorker } = yield* stack;
520
521 const batches = yield* Effect.gen(function* () {
522 yield* HttpClient.get(new URL("/tail-ping", asyncWorker)).pipe(
523 Effect.flatMap((res) => res.text),
524 Effect.orDie,
525 );
526 const res = yield* HttpClient.get(new URL("/events", tailWorker));
527 if (res.status !== 200) return [] as string[];
528 const body = (yield* res.json) as { batches?: unknown };
529 return Array.isArray(body.batches) ? (body.batches as string[]) : [];
530 }).pipe(
531 Effect.repeat({
532 schedule: Schedule.spaced("1 second"),
533 until: (batches): boolean =>
534 batches.some((batch) => batch.includes("cloudflare-dev-tail-marker")),
535 times: 20,
536 }),
537 );
538
539 const batch = batches.find((b) => b.includes("cloudflare-dev-tail-marker"));
540 expect(batch).toBeDefined();
541 const items = JSON.parse(batch!) as {
542 logs?: { message?: unknown[] }[];
543 }[];
544 expect(
545 items.some((item) =>
546 item.logs?.some((log) =>
547 log.message?.includes("cloudflare-dev-tail-marker"),
548 ),
549 ),
550 ).toBe(true);
551 }),
552 { timeout: 60_000 },
553);
554
555/**
556 * Cron trigger: EffectWorker registers `cron("* * * * *", ...)` at init.
557 * Locally the fire is driven on demand via the `/cdn-cgi/handler/scheduled`
558 * trigger route (never wait for the minute boundary); the handler records
559 * `controller.scheduledTime` into a DO which `/cron/times` exposes.
560 */
561test(
562 "EffectWorker cron handler fires via the scheduled trigger route",
563 Effect.gen(function* () {
564 const { effectWorker } = yield* stack;
565
566 const scheduledTime = Date.now();
567 const trigger = yield* HttpClient.post(
568 new URL(
569 `/cdn-cgi/handler/scheduled?cron=${encodeURIComponent("* * * * *")}&time=${scheduledTime}`,
570 effectWorker,
571 ),
572 );
573 expect(trigger.status).toBe(200);
574 expect(yield* trigger.text).toBe("ok");
575
576 const times = yield* Effect.gen(function* () {
577 const res = yield* HttpClient.get(new URL("/cron/times", effectWorker));
578 if (res.status !== 200) return [] as number[];
579 const body = (yield* res.json) as { times?: unknown };
580 return Array.isArray(body.times) ? (body.times as number[]) : [];
581 }).pipe(
582 Effect.repeat({
583 schedule: Schedule.spaced("500 millis"),
584 until: (times): boolean => times.includes(scheduledTime),
585 times: 20,
586 }),
587 );
588 expect(times).toContain(scheduledTime);
589 }),
590 { timeout: 60_000 },
591);
592
593/**
594 * secret_key binding: workerd imports the checked-in HMAC material as a
595 * non-extractable CryptoKey; the route signs and verifies a message with it.
596 */
597test(
598 "EffectWorker signs and verifies with the secret_key binding",
599 Effect.gen(function* () {
600 const { effectWorker } = yield* stack;
601 const response = yield* HttpClient.get(
602 new URL("/secret-key?message=hello", effectWorker),
603 );
604 expect(response.status).toBe(200);
605 const body = (yield* response.json) as {
606 verified: boolean;
607 algorithm: string;
608 extractable: boolean;
609 signatureBase64: string;
610 };
611 expect(body.verified).toBe(true);
612 expect(body.algorithm).toBe("HMAC");
613 // Bound keys are never extractable; the local lowering matches.
614 expect(body.extractable).toBe(false);
615 expect(body.signatureBase64.length).toBeGreaterThan(0);
616 }),
617);
618
619/**
620 * Analytics Engine: `writeDataPoint` is accepted and discarded in local dev
621 * (Miniflare parity) — succeeding without a throw IS the documented local
622 * behavior.
623 */
624test(
625 "EffectWorker writes an Analytics Engine data point (local no-op)",
626 Effect.gen(function* () {
627 const { effectWorker } = yield* stack;
628 const response = yield* HttpClient.get(new URL("/analytics", effectWorker));
629 expect(response.status).toBe(200);
630 expect((yield* response.json) as { ok: boolean }).toEqual({ ok: true });
631 }),
632);
633
634/**
635 * Secrets Store: the store + secret are local `dev:` rows and the
636 * `secrets_store_secret` binding round-trips the exact seeded value.
637 */
638test(
639 "MediaWorker reads the Secrets Store secret",
640 Effect.gen(function* () {
641 const { mediaWorker, secretsStoreId, secretsSecretId } = yield* stack;
642
643 // Locally emulated — no cloud store/secret was created.
644 expect(secretsStoreId).toMatch(/^dev:/);
645 expect(secretsSecretId).toMatch(/^dev:/);
646
647 const body = (yield* (yield* HttpClient.get(
648 new URL("/secret", mediaWorker),
649 )).json) as { value: string };
650 expect(body.value).toBe(SECRETS_STORE_VALUE);
651 }),
652);
653
654/**
655 * send_email: locally the message is persisted as an `.eml` under
656 * `.alchemy/local/email` instead of being delivered; the simulator
657 * synthesizes a production-shaped Message-ID.
658 */
659test(
660 "MediaWorker sends email; the .eml lands in local storage",
661 Effect.gen(function* () {
662 const { mediaWorker } = yield* stack;
663 const marker = crypto.randomUUID();
664
665 const res = yield* HttpClient.get(
666 new URL(`/email/send?marker=${marker}`, mediaWorker),
667 );
668 expect(res.status).toBe(200);
669 const body = (yield* res.json) as { ok: boolean; messageId: string };
670 expect(body.ok).toBe(true);
671 expect(body.messageId).toMatch(/^<[A-Za-z0-9]{36}@example\.com>$/);
672
673 const eml = yield* findEmlContaining(`marker:${marker}`);
674 expect(eml).toContain("From: sender <sender@example.com>");
675 expect(eml).toContain(`Message-ID: <${marker}@example.com>`);
676 }),
677 { timeout: 60_000 },
678);
679
680/**
681 * Inbound email: POST a raw MIME message to the worker's
682 * `/cdn-cgi/handler/email` trigger route. The `email()` handler records it
683 * into KV (asserted via `/email/received`); a `reject-me` subject calls
684 * `setReject`, which the sender observes as a 400.
685 */
686test(
687 "InboxWorker email() handler accepts and rejects inbound email",
688 Effect.gen(function* () {
689 const { inboxWorker } = yield* stack;
690 const marker = crypto.randomUUID();
691 const from = "someone@example.com";
692 const to = "someone-else@example.com";
693 const triggerUrl = new URL(
694 `/cdn-cgi/handler/email?from=${encodeURIComponent(from)}&to=${encodeURIComponent(to)}`,
695 inboxWorker,
696 );
697 const mime = (subject: string) =>
698 [
699 `From: someone <${from}>`,
700 `To: someone else <${to}>`,
701 `Subject: ${subject}`,
702 `Message-ID: <incoming-${marker}@example.com>`,
703 "MIME-Version: 1.0",
704 "Content-Type: text/plain",
705 "",
706 "This is a random email body.",
707 ].join("\n");
708
709 // Accepted.
710 const okRes = yield* HttpClient.post(triggerUrl, {
711 body: HttpBody.text(mime(`accept-me:${marker}`)),
712 });
713 const okText = yield* okRes.text;
714 if (okRes.status !== 200) {
715 // The trigger route surfaces handler errors as a 500 with the stack.
716 throw new Error(`email trigger failed (${okRes.status}): ${okText}`);
717 }
718 expect(okText).toBe("Worker successfully processed email");
719
720 const received = (yield* (yield* HttpClient.get(
721 new URL("/email/received", inboxWorker),
722 )).json) as Array<{ from: string; to: string; subject: string }>;
723 const record = received.find((r) => r.subject === `accept-me:${marker}`);
724 expect(record).toMatchObject({ from, to });
725
726 // Rejected via setReject → the sender sees a 400 with the reason.
727 const rejectRes = yield* HttpClient.post(triggerUrl, {
728 body: HttpBody.text(mime(`reject-me:${marker}`)),
729 });
730 expect(rejectRes.status).toBe(400);
731 expect(yield* rejectRes.text).toBe(
732 "Worker rejected email with the following reason: I don't like this email",
733 );
734 }),
735 { timeout: 60_000 },
736);
737
738/**
739 * Browser Rendering: renders an inline HTML page through
740 * `@cloudflare/puppeteer` against the locally launched Chrome. The FIRST
741 * run on a machine downloads Chrome into the shared wrangler cache, which
742 * can take minutes — hence the generous bounded retry + timeout.
743 */
744test(
745 "MediaWorker renders a page through the browser binding",
746 Effect.gen(function* () {
747 const { mediaWorker } = yield* stack;
748 const body = (yield* getJsonReady(new URL("/browser/title", mediaWorker), {
749 times: 30,
750 spaced: "5 seconds",
751 })) as { title: string; heading: string };
752 expect(body.title).toBe("Cloudflare Dev");
753 expect(body.heading).toBe("Hello from the local browser");
754 }),
755 { timeout: 240_000 },
756);
757
758/**
759 * Images: `info()` + `transform().output()` served locally by sharp.
760 * The transform result is re-inspected through `info()` to prove real pixel
761 * work happened (8x4 contain-fit to width 4 → 4x2).
762 */
763test(
764 "MediaWorker transforms an image through the images binding",
765 Effect.gen(function* () {
766 const { mediaWorker } = yield* stack;
767
768 const info = (yield* (yield* HttpClient.post(
769 new URL("/images/info", mediaWorker),
770 { body: HttpBody.uint8Array(PNG_RED_8X4) },
771 )).json) as { format: string; width: number; height: number };
772 expect(info.format).toBe("image/png");
773 expect(info.width).toBe(8);
774 expect(info.height).toBe(4);
775
776 const transformRes = yield* HttpClient.post(
777 new URL("/images/transform?width=4", mediaWorker),
778 { body: HttpBody.uint8Array(PNG_RED_8X4) },
779 );
780 expect(transformRes.status).toBe(200);
781 const outputBytes = new Uint8Array(yield* transformRes.arrayBuffer);
782 // PNG magic bytes.
783 expect(Array.from(outputBytes.slice(0, 4))).toEqual([
784 0x89, 0x50, 0x4e, 0x47,
785 ]);
786
787 const outputInfo = (yield* (yield* HttpClient.post(
788 new URL("/images/info", mediaWorker),
789 { body: HttpBody.uint8Array(outputBytes) },
790 )).json) as { format: string; width: number; height: number };
791 expect(outputInfo.width).toBe(4);
792 expect(outputInfo.height).toBe(2);
793 }),
794 { timeout: 60_000 },
795);
796
797/**
798 * HYBRID — per-binding `Alchemy.remote()`: `IMAGES_REMOTE` proxies to
799 * the REAL Images service while the worker (and every other binding) stays
800 * local. Requires real Cloudflare credentials, which this suite already has
801 * (the state store is remote).
802 */
803test(
804 "MediaWorker images binding with Alchemy.remote() hits the real service",
805 Effect.gen(function* () {
806 const { mediaWorker } = yield* stack;
807 const info = (yield* Effect.gen(function* () {
808 const res = yield* HttpClient.post(
809 new URL("/images/info-remote", mediaWorker),
810 { body: HttpBody.uint8Array(PNG_RED_8X4) },
811 );
812 if (res.status !== 200) {
813 return yield* Effect.fail(new NotYet({ status: res.status }));
814 }
815 return yield* res.json;
816 }).pipe(
817 Effect.retry({
818 while: (e) => e._tag === "NotYet",
819 schedule: Schedule.max([
820 Schedule.spaced("2 seconds"),
821 Schedule.recurs(10),
822 ]),
823 }),
824 )) as { format: string; width: number; height: number };
825 expect(info.format).toBe("image/png");
826 expect(info.width).toBe(8);
827 expect(info.height).toBe(4);
828 }),
829 { timeout: 120_000 },
830);
831
832/**
833 * Stream: upload video bytes through the binding, read details back, then
834 * fetch the watch URL served by the local stream router middleware.
835 */
836test(
837 "MediaWorker uploads and watches a video through the stream binding",
838 Effect.gen(function* () {
839 const { mediaWorker } = yield* stack;
840 const videoBytes = new Uint8Array([0, 1, 2, 3, 4, 5, 6, 7]);
841
842 const video = (yield* (yield* HttpClient.post(
843 new URL("/stream/upload", mediaWorker),
844 { body: HttpBody.uint8Array(videoBytes) },
845 )).json) as {
846 id: string;
847 readyToStream: boolean;
848 status: { state: string };
849 size: number;
850 creator: string | null;
851 };
852 expect(video.id).toBeTruthy();
853 expect(video.readyToStream).toBe(true);
854 expect(video.status.state).toBe("ready");
855 expect(video.size).toBe(videoBytes.byteLength);
856 expect(video.creator).toBe("alchemy");
857
858 const details = (yield* (yield* HttpClient.get(
859 new URL(`/stream/details?id=${video.id}`, mediaWorker),
860 )).json) as { id: string; size: number };
861 expect(details.id).toBe(video.id);
862 expect(details.size).toBe(videoBytes.byteLength);
863
864 // The watch route is served by the stream router middleware in front of
865 // the worker.
866 const watchRes = yield* HttpClient.get(
867 new URL(`/cdn-cgi/mf/stream/${video.id}/watch`, mediaWorker),
868 );
869 expect(watchRes.status).toBe(200);
870 const watched = new Uint8Array(yield* watchRes.arrayBuffer);
871 expect(Array.from(watched)).toEqual(Array.from(videoBytes));
872
873 // Clean up so re-runs stay tidy.
874 const deleted = (yield* (yield* HttpClient.get(
875 new URL(`/stream/delete?id=${video.id}`, mediaWorker),
876 )).json) as { deleted: boolean };
877 expect(deleted.deleted).toBe(true);
878 }),
879 { timeout: 60_000 },
880);
881
882/**
883 * HYBRID — `Alchemy.remote()`: LiveKV opted out of local emulation, so its
884 * id is a REAL namespace id (no `dev:` marker) and the binding round-trip
885 * from the local worker lands in the real cloud namespace.
886 */
887test(
888 "Alchemy.remote() KV namespace is live while the worker stays local",
889 Effect.gen(function* () {
890 const { asyncWorker, liveKvNamespaceId } = yield* stack;
891
892 // A real cloud namespace id — the `dev:` marker means "emulated" and
893 // must be absent here.
894 expect(liveKvNamespaceId).not.toMatch(/^dev:/);
895
896 const key = crypto.randomUUID();
897 const value = crypto.randomUUID();
898 const put = (yield* getJsonReady(
899 new URL(`/kv-live?key=${key}&value=${value}`, asyncWorker),
900 { times: 10, spaced: "2 seconds" },
901 )) as { value: string | null };
902 expect(put.value).toBe(value);
903
904 const got = (yield* (yield* HttpClient.get(
905 new URL(`/kv-live?key=${key}`, asyncWorker),
906 )).json) as { value: string | null };
907 expect(got.value).toBe(value);
908 }),
909 { timeout: 60_000 },
910);
911
912interface WorkflowStatus {
913 status: string;
914 output?: { text?: string; secret?: string; ts?: number };
915 error?: { name: string; message: string } | null;
916}
917
918/**
919 * Start a `NotifyWorkflow` instance through `workerUrl` and poll its status
920 * until the instance reaches a terminal state. Asserts the workflow ran the
921 * KV roundtrip task (`Processed: <message>`) and resolved the plantime-bound
922 * `Alchemy.Secret` at runtime (`output.secret === WORKFLOW_SECRET_VALUE`).
923 */
924const exerciseWorkflow = (workerUrl: string, label: string) =>
925 Effect.gen(function* () {
926 const roomId = `${label}-${Math.random().toString(36).slice(2, 10)}`;
927
928 const startResponse = yield* HttpClient.post(
929 new URL(`/workflow/start/${roomId}`, workerUrl),
930 );
931 expect(startResponse.status).toBe(200);
932 const { instanceId } = (yield* startResponse.json) as {
933 instanceId: string;
934 };
935 expect(instanceId).toBeString();
936
937 const statusUrl = new URL(`/workflow/status/${instanceId}`, workerUrl);
938 const fetchStatus = HttpClient.get(statusUrl).pipe(
939 Effect.flatMap((res) => res.json),
940 Effect.map((json) => json as unknown as WorkflowStatus),
941 );
942 const status = yield* fetchStatus.pipe(
943 Effect.repeat({
944 schedule: Schedule.spaced("2 seconds"),
945 until: (s: WorkflowStatus) =>
946 s.status === "complete" || s.status === "errored",
947 times: 60,
948 }),
949 );
950
951 expect(status.error).toBeFalsy();
952 expect(status.status).toBe("complete");
953 expect(status.output?.secret).toBe(WORKFLOW_SECRET_VALUE);
954 expect(status.output?.text).toBe("Processed: hello from workflow");
955 });
956
957/**
958 * EffectWorker `/workflow/start/:roomId` kicks off `NotifyWorkflow`, which
959 * does a KV roundtrip task and resolves the `WORKFLOW_SECRET` `Alchemy.Secret`
960 * at runtime. The status route surfaces the workflow output so we can assert
961 * the workflow actually executed end-to-end (not just that it was scheduled).
962 */
963test(
964 "EffectWorker drives NotifyWorkflow to completion with secret + KV roundtrip",
965 Effect.gen(function* () {
966 const { effectWorker } = yield* stack;
967 yield* exerciseWorkflow(effectWorker, "effect");
968 }),
969 { timeout: 60_000 },
970);
971
972test(
973 "EffectWorker fetches a URL in a sandbox",
974 Effect.gen(function* () {
975 const { effectWorker } = yield* stack;
976 const response = yield* HttpClient.get(new URL("/sandbox", effectWorker));
977 expect(response.status).toBe(200);
978 const body = yield* response.text;
979 // The container echoes its `GREETING` env var, proving env vars flow
980 // through to the container (via the application config on a live deploy,
981 // and via `ctx.container.start({ env })` in local dev).
982 expect(body).toBe("Hello from Sandbox container! GREETING=hello-from-env");
983 }),
984 { timeout: 60_000 },
985);
986
987/**
988 * Hyperdrive: local dev is a passthrough to the Connection's `dev` origin,
989 * which must be a REAL reachable Postgres. Gated behind
990 * `HYPERDRIVE_DEV_URL` (see src/HyperdriveWorker.ts for a one-liner Docker
991 * setup); without it the stack omits the worker and this test skips.
992 */
993test.skipIf(!process.env.HYPERDRIVE_DEV_URL)(
994 "HyperdriveWorker queries Postgres through the hyperdrive binding",
995 Effect.gen(function* () {
996 const { hyperdrive } = yield* stack;
997 assert(typeof hyperdrive === "string");
998
999 const body = (yield* getJsonReady(new URL("/query", hyperdrive), {
1000 times: 15,
1001 spaced: "2 seconds",
1002 })) as { row: { sum: number; db: string } };
1003 expect(body.row.sum).toBe(2);
1004 expect(body.row.db.length).toBeGreaterThan(0);
1005 }),
1006 { timeout: 120_000 },
1007);