1import * as Cloudflare from "@alchemy.run/cloudflare";
2import { CloudflareEnvironment } from "@alchemy.run/cloudflare/CloudflareEnvironment";
3import { expect } from "@alchemy.run/test";
4import * as Test from "@alchemy.run/test/Alchemy";
5import * as workers from "@distilled.cloud/cloudflare/workers";
6import * as workflows from "@distilled.cloud/cloudflare/workflows";
7import * as Provider from "alchemy/Provider";
8import * as Effect from "effect/Effect";
9import * as HttpClient from "effect/http/HttpClient";
10import { MinimumLogLevel } from "effect/References";
11import * as Schedule from "effect/Schedule";
12import * as Stream from "effect/Stream";
13import LimitsWorkflowWorker from "./fixtures/workflow-limits/limits-worker.ts";
14import { STEP_LIMIT } from "./fixtures/workflow-limits/limits-workflow.ts";
15import ScheduledWorkflowWorker from "./fixtures/workflow-schedules/scheduled-worker.ts";
16import { YEARLY_CRON } from "./fixtures/workflow-schedules/scheduled-workflow.ts";
17import Stack from "./fixtures/workflow/stack.ts";
18import WorkflowTestWorker from "./fixtures/workflow/workflow-worker.ts";
19
20const { test, beforeAll, afterAll, deploy, destroy } = Test.make({
21 providers: Cloudflare.providers(),
22});
23
24const logLevel = Effect.provideService(MinimumLogLevel, process.env.DEBUG ? "Debug" : "Info");
25
26const stack = beforeAll(
27 deploy(Stack).pipe(
28 // Let the freshly-deployed worker (and its Workflow binding) settle before
29 // the first run so a step doesn't error mid-propagation.
30 Effect.tap(Effect.sleep("5 seconds")),
31 ),
32);
33afterAll.skipIf(!!process.env.NO_DESTROY)(destroy(Stack));
34
35interface WorkflowStatus {
36 status: string;
37 output?: {
38 greeting: string;
39 envBindingCount: number;
40 workflowName: string;
41 stepAttempt: number;
42 instanceId: string;
43 };
44 error?: { message?: string } | null;
45 rollback?: {
46 outcome: "complete" | "failed";
47 error: { message?: string } | null;
48 } | null;
49}
50
51const isTerminal = (status: WorkflowStatus) =>
52 status.status === "complete" || status.status === "errored" || status.status === "terminated";
53
54const waitForStatus = (
55 client: HttpClient.HttpClient,
56 url: string,
57 id: string,
58 until: (status: WorkflowStatus) => boolean = isTerminal,
59 times = 30,
60) =>
61 client.get(`${url}/workflow/status/${id}`).pipe(
62 // The status endpoint transiently returns a 500 (HTML error page, not
63 // JSON) while the freshly-deployed worker's Workflow binding is still
64 // propagating. Only decode JSON on a 200; treat any other status as a
65 // non-terminal "pending" so the poll keeps swinging instead of dying
66 // on a JSON decode error.
67 Effect.flatMap((res) =>
68 res.status === 200
69 ? res.json.pipe(Effect.map((json) => json as unknown as WorkflowStatus))
70 : Effect.succeed({ status: "pending" } as WorkflowStatus),
71 ),
72 Effect.repeat({
73 schedule: Schedule.spaced("2 seconds"),
74 until,
75 times,
76 }),
77 );
78
79// Start a fresh workflow instance and poll until it reaches a terminal state.
80// A transient `errored` during edge/binding propagation fails this effect so
81// the caller can retry with a brand-new instance.
82const runWorkflowToCompletion = (url: string) =>
83 Effect.gen(function* () {
84 const client = yield* HttpClient.HttpClient;
85
86 // Cloudflare's edge takes a few seconds to start serving a fresh
87 // workers.dev URL, so retry until it returns 200 (a fresh URL also
88 // returns 404 transiently, which is not an HTTP error so Effect.retry
89 // does not catch it unless we explicitly fail on non-200).
90 const startRes = yield* client.post(`${url}/workflow/start/world`).pipe(
91 Effect.flatMap((res) =>
92 res.status === 200
93 ? Effect.succeed(res)
94 : Effect.fail(new Error(`Worker not ready: ${res.status}`)),
95 ),
96 Effect.retry({
97 // Cap the exponential at 3s — uncapped, 15 retries grow past 30s of
98 // sleep after only six attempts and blow the test timeout.
99 schedule: Schedule.min([Schedule.exponential("500 millis"), Schedule.spaced("3 seconds")]),
100 times: 15,
101 }),
102 );
103 const { instanceId } = (yield* startRes.json) as { instanceId: string };
104 expect(instanceId).toBeTypeOf("string");
105
106 const lastStatus = yield* waitForStatus(client, url, instanceId);
107
108 // Surface a non-complete terminal state as a failure so the outer retry
109 // can take another swing (a fresh worker occasionally errors a step while
110 // its bindings are still propagating).
111 if (lastStatus.status !== "complete") {
112 return yield* Effect.fail(
113 new Error(`workflow ${lastStatus.status}: ${JSON.stringify(lastStatus.error)}`),
114 );
115 }
116 return lastStatus;
117 });
118
119test(
120 "deployed worker can run a workflow to completion",
121 Effect.gen(function* () {
122 const out = yield* stack;
123 const url = out.url;
124 expect(url).toBeTypeOf("string");
125
126 const lastStatus = yield* runWorkflowToCompletion(url).pipe(
127 Effect.retry({ schedule: Schedule.spaced("3 seconds"), times: 2 }),
128 );
129
130 expect(lastStatus.status).toBe("complete");
131 expect(lastStatus.error).toBeFalsy();
132 expect(lastStatus.output?.greeting).toBe("Hello, world!");
133 // `event.workflowName` carries the account-global *physical* name, which
134 // is derived from the host Worker's unique name (stack-id-stage prefix).
135 expect(lastStatus.output?.workflowName).toMatch(/^workflowbindingstack-/);
136 expect(lastStatus.output?.stepAttempt).toBe(1);
137 expect(lastStatus.rollback).toBeNull();
138 // The body yields `WorkerEnvironment` — if the regression from PR #71 ever
139 // returns, the body dies on the first yield and `output` is undefined.
140 expect(lastStatus.output?.envBindingCount).toBeGreaterThan(0);
141 }).pipe(logLevel),
142 {
143 tags: [
144 "provider:cloudflare",
145 "provider:cloudflare:worker",
146 "provider:cloudflare:workflow",
147 "live",
148 ],
149 timeout: 180_000,
150 },
151);
152
153test(
154 "workflow can wait for and receive external events",
155 Effect.gen(function* () {
156 const { url } = yield* stack;
157 const client = yield* HttpClient.HttpClient;
158
159 const { instanceId, lastStatus } = yield* Effect.gen(function* () {
160 const { instanceId } = yield* client.post(`${url}/workflow/wait/world`).pipe(
161 Effect.flatMap((res) =>
162 res.status === 200
163 ? res.json.pipe(
164 Effect.flatMap((body) => {
165 const instanceId = (body as { instanceId?: unknown }).instanceId;
166 return typeof instanceId === "string"
167 ? Effect.succeed({ instanceId })
168 : Effect.fail(new Error("Worker returned no workflow id"));
169 }),
170 )
171 : Effect.fail(new Error(`Worker not ready: ${res.status}`)),
172 ),
173 Effect.retry({
174 schedule: Schedule.exponential("500 millis"),
175 times: 15,
176 }),
177 );
178
179 // Cloudflare reports an instance parked in `waitForEvent` as `running`
180 // (`waiting` is reserved for sleeps), so the wait step itself is not
181 // observable through the coarse instance status. Wait for the instance
182 // to start, then deliver the event.
183 const runningStatus = yield* waitForStatus(
184 client,
185 url,
186 instanceId,
187 (status) => status.status === "running" || isTerminal(status),
188 );
189 if (runningStatus.status !== "running") {
190 return yield* Effect.fail(
191 new Error(`workflow ${runningStatus.status}: ${JSON.stringify(runningStatus.error)}`),
192 );
193 }
194
195 // An event sent in the gap before the `waitForEvent` step registers can
196 // be missed, so re-send until the workflow acknowledges it by reaching a
197 // terminal status.
198 const lastStatus = yield* Effect.gen(function* () {
199 const sendRes = yield* client.post(`${url}/workflow/send/${instanceId}/external-ok`);
200 if (sendRes.status !== 200) {
201 return yield* Effect.fail(new Error(`sendEvent failed: ${sendRes.status}`));
202 }
203 const status = yield* waitForStatus(client, url, instanceId, isTerminal, 5);
204 return isTerminal(status)
205 ? status
206 : yield* Effect.fail(new Error(`workflow still ${status.status}`));
207 }).pipe(Effect.retry({ times: 3 }));
208 return { instanceId, lastStatus };
209 }).pipe(
210 // A just-deployed workflow can error its first instance while the
211 // binding propagates. Retry the scenario with a fresh instance.
212 Effect.retry({ schedule: Schedule.spaced("3 seconds"), times: 2 }),
213 );
214 expect(lastStatus.status).toBe("complete");
215 expect(lastStatus.error).toBeFalsy();
216 expect(lastStatus.output?.greeting).toBe("external-ok");
217 expect(lastStatus.output?.instanceId).toBe(instanceId);
218 }).pipe(logLevel),
219 {
220 tags: [
221 "provider:cloudflare",
222 "provider:cloudflare:worker",
223 "provider:cloudflare:workflow",
224 "live",
225 ],
226 timeout: 180_000,
227 },
228);
229
230// Canonical `list()` test (account collection): deploy the worker+workflow
231// fixture (which creates a `WorkflowResource` named "TestWorkflow"), then
232// enumerate every workflow in the account via the typed provider and assert the
233// deployed one is present. Bracket with destroy so the test is isolated.
234test.provider(
235 "list enumerates the deployed workflow",
236 (stack) =>
237 Effect.gen(function* () {
238 yield* stack.destroy();
239
240 yield* stack.deploy(
241 Effect.gen(function* () {
242 return yield* WorkflowTestWorker;
243 }),
244 );
245
246 const provider = yield* Provider.findProvider(Cloudflare.Workflows.WorkflowResource);
247 const all = yield* provider.list();
248
249 // Physical names are derived from the host Worker name (which carries
250 // the stack-id-stage prefix), not the exported class name.
251 expect(all.some((w) => w.workflowName.startsWith("workflowbindingstack-"))).toBe(true);
252
253 yield* stack.destroy();
254 }).pipe(logLevel),
255 {
256 tags: [
257 "provider:cloudflare",
258 "provider:cloudflare:worker",
259 "provider:cloudflare:workflow",
260 "live",
261 ],
262 timeout: 120_000,
263 },
264);
265
266// ---------------------------------------------------------------------------
267// Per-workflow limits: deploy a workflow declared with a step limit through the
268// Effect-native form, then read it back out-of-band from the versions API (the
269// only read that surfaces `limits`) to confirm it was applied.
270// ---------------------------------------------------------------------------
271
272// Physical workflow names are derived from the host Worker name and class, so
273// read the name off the deployed binding rather than assuming the class name.
274const readWorkflowName = (scriptName: string) =>
275 Effect.gen(function* () {
276 const { accountId } = yield* yield* CloudflareEnvironment;
277 const settings = yield* workers.getScriptScriptAndVersionSetting({
278 accountId,
279 scriptName,
280 });
281 const binding = (settings.bindings ?? []).find(
282 (b): b is Extract<typeof b, { type: "workflow" }> => b.type === "workflow",
283 );
284 return binding === undefined
285 ? yield* Effect.fail(new Error(`no workflow binding on '${scriptName}'`))
286 : binding.workflowName;
287 });
288
289// Read the applied step limit out-of-band via the versions API, retrying until
290// it propagates (bounded, so a missing limit fails fast).
291const waitForAppliedStepLimit = (workflowName: string, expected: number) =>
292 Effect.gen(function* () {
293 const { accountId } = yield* yield* CloudflareEnvironment;
294 const versions = yield* workflows.listVersions
295 .items({ accountId, workflowName })
296 .pipe(Stream.runCollect);
297 return Array.from(versions)
298 .map((v) => v.limits?.steps ?? undefined)
299 .find((steps) => steps !== undefined);
300 }).pipe(
301 Effect.flatMap((steps) =>
302 steps === expected
303 ? Effect.succeed(steps)
304 : Effect.fail(new Error(`steps limit not applied yet: ${steps}`)),
305 ),
306 Effect.retry({ schedule: Schedule.spaced("2 seconds"), times: 15 }),
307 );
308
309test.provider(
310 "effect-native workflow applies a per-workflow step limit",
311 (scratch) =>
312 Effect.gen(function* () {
313 yield* scratch.destroy();
314
315 const deployed = yield* scratch.deploy(
316 Effect.gen(function* () {
317 return { worker: yield* LimitsWorkflowWorker };
318 }),
319 );
320
321 const workflowName = yield* readWorkflowName(deployed.worker.workerName);
322 const applied = yield* waitForAppliedStepLimit(workflowName, STEP_LIMIT);
323 expect(applied).toBe(STEP_LIMIT);
324
325 yield* scratch.destroy();
326 }).pipe(logLevel),
327 {
328 tags: [
329 "provider:cloudflare",
330 "provider:cloudflare:worker",
331 "provider:cloudflare:workflow",
332 "live",
333 ],
334 timeout: 120_000,
335 },
336);
337
338// ---------------------------------------------------------------------------
339// #1473 regression: native Workflow schedules on the Effect-native form,
340// driven by the file-based `fixtures/workflow-schedules` worker. Deploy a
341// workflow declared with `schedules`, then read it back out-of-band from
342// getWorkflow (the read that surfaces `schedules`) to confirm it was
343// applied. The cron is yearly so the test does not wait for a fire.
344// ---------------------------------------------------------------------------
345
346const waitForAppliedSchedules = (workflowName: string, expected: string[]) =>
347 Effect.gen(function* () {
348 const { accountId } = yield* yield* CloudflareEnvironment;
349 const workflow = yield* workflows.getWorkflow({
350 accountId,
351 workflowName,
352 });
353 return (workflow.schedules ?? []).map((s) => s.cron);
354 }).pipe(
355 Effect.flatMap((crons) =>
356 crons.length === expected.length && crons.every((cron, index) => cron === expected[index])
357 ? Effect.succeed(crons)
358 : Effect.fail(new Error(`schedules not applied yet: ${JSON.stringify(crons)}`)),
359 ),
360 Effect.retry({ schedule: Schedule.spaced("2 seconds"), times: 15 }),
361 );
362
363test.provider(
364 "effect-native workflow applies native cron schedules",
365 (scratch) =>
366 Effect.gen(function* () {
367 yield* scratch.destroy();
368
369 const deployed = yield* scratch.deploy(
370 Effect.gen(function* () {
371 return { worker: yield* ScheduledWorkflowWorker };
372 }),
373 );
374
375 const workflowName = yield* readWorkflowName(deployed.worker.workerName);
376 const applied = yield* waitForAppliedSchedules(workflowName, [YEARLY_CRON]);
377 expect(applied).toEqual([YEARLY_CRON]);
378
379 yield* scratch.destroy();
380 }).pipe(logLevel),
381 {
382 tags: [
383 "provider:cloudflare",
384 "provider:cloudflare:worker",
385 "provider:cloudflare:workflow",
386 "live",
387 ],
388 timeout: 120_000,
389 },
390);