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