1import * as Cloudflare from "@/Cloudflare";
2import * as Test from "@/Test/Alchemy";
3import { expect } from "alchemy-test";
4import * as Data from "effect/Data";
5import * as Effect from "effect/Effect";
6import { MinimumLogLevel } from "effect/References";
7import * as Schedule from "effect/Schedule";
8import * as HttpClient from "effect/http/HttpClient";
9import * as HttpClientRequest from "effect/http/HttpClientRequest";
10import WriteBindingWorker from "./fixtures/write-binding.ts";
11import WriteHttpWorker from "./fixtures/write-http.ts";
12
13const { test } = Test.make({ providers: Cloudflare.providers() });
14
15const logLevel = Effect.provideService(
16 MinimumLogLevel,
17 process.env.DEBUG ? "Debug" : "Info",
18);
19
20class WorkerNotReady extends Data.TaggedError("WorkerNotReady")<{
21 status: number;
22 body: string;
23}> {}
24
25const ready = Schedule.max([Schedule.spaced("3 seconds"), Schedule.recurs(20)]);
26
27/** POST and retry until the producer route accepts the message (202). */
28const post = (base: string, path: string, body?: string) => {
29 const base$ = HttpClientRequest.post(`${base}${path}`);
30 const req =
31 body !== undefined ? HttpClientRequest.bodyText(base$, body) : base$;
32 return HttpClient.execute(req).pipe(
33 Effect.flatMap((res) =>
34 res.status === 202
35 ? Effect.succeed(res)
36 : res.text.pipe(
37 Effect.flatMap((b) =>
38 Effect.fail(new WorkerNotReady({ status: res.status, body: b })),
39 ),
40 ),
41 ),
42 Effect.retry({
43 // Ride out cold-start propagation — a fresh workers.dev URL
44 // serves 404 ("nothing here yet") or 500 (code 1104 "Script not
45 // found") for a few seconds before the script goes live. The
46 // bounded spaced schedule caps total wait so a genuine failure
47 // (worker returns its own JSON error body) still surfaces.
48 while: (e): e is WorkerNotReady => e instanceof WorkerNotReady,
49 schedule: ready,
50 }),
51 );
52};
53
54/**
55 * Cloudflare Queue is producer-only at the binding layer, so there is
56 * no Read/ReadWrite split — only a Write producer. This deploys two
57 * Workers that both bind one shared queue (native Worker binding and
58 * scoped HTTP API token), then drives every {@link WriteQueueClient}
59 * method over `fetch` — `send` and `sendBatch`, each in its JSON and
60 * `text` content-type form — and asserts the producer accepts the
61 * messages (202), proving the binding/token are wired and reach the
62 * real queue.
63 */
64test.provider(
65 "Queue write producer over binding + http",
66 (stack) =>
67 Effect.gen(function* () {
68 yield* stack.destroy();
69
70 const out = yield* stack.deploy(
71 Effect.gen(function* () {
72 const writeBinding = yield* WriteBindingWorker;
73 const writeHttp = yield* WriteHttpWorker;
74 return {
75 writeBinding: writeBinding.url,
76 writeHttp: writeHttp.url,
77 };
78 }),
79 );
80
81 const url = (u: unknown) => {
82 expect(u).toBeTypeOf("string");
83 return u as string;
84 };
85
86 // Drive the full producer surface (send + sendBatch, json + text)
87 // against one base url.
88 const exercise = (base: string, label: string) =>
89 Effect.gen(function* () {
90 expect((yield* post(base, "/send", `${label}-json`)).status).toBe(
91 202,
92 );
93 expect(
94 (yield* post(base, "/send-text", `${label}-text`)).status,
95 ).toBe(202);
96 expect((yield* post(base, "/sendBatch")).status).toBe(202);
97 expect((yield* post(base, "/sendBatch-text")).status).toBe(202);
98 });
99
100 // ── Native binding producer ──
101 yield* exercise(url(out.writeBinding), "binding");
102 // ── HTTP token producer ──
103 yield* exercise(url(out.writeHttp), "http");
104
105 yield* stack.destroy();
106 }).pipe(logLevel),
107 {
108 tags: [
109 "provider:cloudflare",
110 "provider:cloudflare:queue",
111 "provider:cloudflare:worker",
112 "live",
113 ],
114 timeout: 240_000,
115 },
116);