1import * as Cloudflare from "@/Cloudflare";
2import { CloudflareEnvironment } from "@/Cloudflare/CloudflareEnvironment";
3import * as Test from "@/Test/Alchemy";
4import * as workers from "@distilled.cloud/cloudflare/workers";
5import { expect } from "alchemy-test";
6import * as Data from "effect/Data";
7import * as Effect from "effect/Effect";
8import { MinimumLogLevel } from "effect/References";
9import * as Schedule from "effect/Schedule";
10import * as HttpClient from "effect/http/HttpClient";
11import Stack from "./fixtures/stack.ts";
12
13const { test, beforeAll, afterAll, deploy, destroy } = Test.make({
14 providers: Cloudflare.providers(),
15});
16
17const logLevel = Effect.provideService(
18 MinimumLogLevel,
19 process.env.DEBUG ? "Debug" : "Info",
20);
21
22const forbiddenBlips = {
23 while: (e: { _tag: string }) => e._tag === "Forbidden",
24 schedule: Schedule.exponential("500 millis"),
25 times: 8,
26} as const;
27
28class WorkerNotReady extends Data.TaggedError("WorkerNotReady")<{
29 status: number;
30 body: string;
31}> {}
32
33const asyncWorkerScript = `export default {
34 async fetch() {
35 return new Response("ok");
36 },
37};
38`;
39
40// A freshly-deployed workers.dev URL takes a few seconds to start serving,
41// so the first request is retried until the worker answers.
42const sendEvent = (url: string, nonce: string) =>
43 HttpClient.get(`${url}/send?nonce=${nonce}`).pipe(
44 Effect.flatMap((res) =>
45 res.status === 200
46 ? res.json
47 : res.text.pipe(
48 Effect.flatMap((body) =>
49 Effect.fail(new WorkerNotReady({ status: res.status, body })),
50 ),
51 ),
52 ),
53 Effect.retry({
54 while: (e): e is WorkerNotReady => e instanceof WorkerNotReady,
55 schedule: Schedule.max([
56 Schedule.min([
57 Schedule.exponential("500 millis"),
58 Schedule.spaced("5 seconds"),
59 ]),
60 Schedule.recurs(30),
61 ]),
62 }),
63 );
64
65const stack = beforeAll(deploy(Stack));
66afterAll.skipIf(!!process.env.NO_DESTROY)(destroy(Stack));
67
68test(
69 "async worker sends events through the env Pipelines stream binding",
70 Effect.gen(function* () {
71 const { asyncWorkerUrl } = yield* stack;
72
73 const body = yield* sendEvent(asyncWorkerUrl, "async");
74
75 // `send` only exists on a real `pipelines` binding — a `json` binding
76 // would deliver the stream's attributes as a plain object.
77 expect(body).toMatchObject({ mode: "async", sent: true, kind: "function" });
78 }),
79 {
80 tags: [
81 "provider:cloudflare",
82 "provider:cloudflare:pipelines",
83 "provider:cloudflare:worker",
84 "live",
85 ],
86 timeout: 240_000,
87 },
88);
89
90test(
91 "effect worker sends events through the env Pipelines stream binding",
92 Effect.gen(function* () {
93 const { effectWorkerUrl } = yield* stack;
94
95 const body = yield* sendEvent(effectWorkerUrl, "effect");
96
97 expect(body).toMatchObject({
98 mode: "effect",
99 sent: true,
100 kind: "function",
101 });
102 }),
103 {
104 tags: [
105 "provider:cloudflare",
106 "provider:cloudflare:pipelines",
107 "provider:cloudflare:worker",
108 "live",
109 ],
110 timeout: 240_000,
111 },
112);
113
114test.provider(
115 "the deployed worker carries a pipelines binding holding the stream id",
116 (stack) =>
117 Effect.gen(function* () {
118 const { accountId } = yield* yield* CloudflareEnvironment;
119
120 yield* stack.destroy();
121
122 const { stream, worker } = yield* stack.deploy(
123 Effect.gen(function* () {
124 const stream = yield* Cloudflare.Pipelines.Stream("InlineStream", {});
125 const worker = yield* Cloudflare.Worker("pipelines-binding-worker", {
126 script: asyncWorkerScript,
127 env: {
128 EVENTS: stream,
129 },
130 });
131 return { stream, worker };
132 }),
133 );
134
135 const settings = yield* workers
136 .getScriptScriptAndVersionSetting({
137 accountId,
138 scriptName: worker.workerName,
139 })
140 .pipe(Effect.retry(forbiddenBlips));
141
142 const binding = (settings.bindings ?? []).find(
143 (b) => b.name === "EVENTS",
144 );
145
146 expect(binding).toMatchObject({
147 type: "pipelines",
148 name: "EVENTS",
149 pipeline: stream.streamId,
150 });
151
152 yield* stack.destroy();
153 }).pipe(logLevel),
154 {
155 tags: [
156 "provider:cloudflare",
157 "provider:cloudflare:pipelines",
158 "provider:cloudflare:worker",
159 "live",
160 ],
161 timeout: 300_000,
162 },
163);