1import * as Cloudflare from "@/Cloudflare/index.ts";
2import * as Alchemy from "@/index.ts";
3import * as State from "@/State/State";
4import * as Test from "@/Test/Alchemy";
5import * as workflows from "@distilled.cloud/cloudflare/workflows";
6import { expect } from "alchemy-test";
7import * as Data from "effect/Data";
8import * as Effect from "effect/Effect";
9import * as Path from "effect/Path";
10import { MinimumLogLevel } from "effect/References";
11import * as Schedule from "effect/Schedule";
12import * as HttpClient from "effect/http/HttpClient";
13import ExplicitNameWorkflowWorker, {
14 EXPLICIT_WORKFLOW_NAME,
15} from "./fixtures/explicit-name-worker.ts";
16import WorkflowLocalWorker from "./fixtures/workflow-worker.ts";
17
18// `dev: true` runs local providers behind the RPC sidecar proxy by default,
19// matching the process topology of the real `alchemy dev` command (see
20// MakeOptions.sidecar in Test/Core.ts).
21const { test } = Test.make({
22 providers: Cloudflare.providers(),
23 dev: true,
24});
25
26const logLevel = Effect.provideService(
27 MinimumLogLevel,
28 process.env.DEBUG ? "Debug" : "Info",
29);
30
31class WorkerNotReady extends Data.TaggedError("WorkerNotReady")<{
32 status: number;
33}> {}
34
35class WorkflowLinkNotReady extends Data.TaggedError("WorkflowLinkNotReady")<{
36 instanceId: string;
37}> {}
38
39interface WorkflowStatus {
40 status: string;
41 output?: {
42 greeting: string;
43 retryAttempt: number;
44 retryConfig: Cloudflare.Workflows.WorkflowStepConfig;
45 timeoutConfig: Cloudflare.Workflows.WorkflowStepConfig;
46 defaultsOk: boolean;
47 workflowName: string;
48 instanceId: string;
49 };
50 error?: { name?: string; message?: string } | null;
51 rollback?: {
52 outcome: "complete" | "failed";
53 error: { message?: string } | null;
54 } | null;
55}
56
57const isTerminal = (status: WorkflowStatus): boolean =>
58 status.status === "complete" ||
59 status.status === "errored" ||
60 status.status === "terminated";
61
62/**
63 * Start a workflow instance over HTTP, retrying while the freshly-served
64 * worker is still coming up (local workerd boots fast; a fresh workers.dev
65 * URL takes a few seconds to start serving 200s).
66 */
67const startInstance = (url: string, path = "/workflow/start/world") =>
68 Effect.gen(function* () {
69 const client = yield* HttpClient.HttpClient;
70 const res = yield* client.post(`${url}${path}`).pipe(
71 Effect.flatMap((res) =>
72 res.status === 200
73 ? Effect.succeed(res)
74 : Effect.fail(new WorkerNotReady({ status: res.status })),
75 ),
76 Effect.retry({
77 while: (e) => e._tag === "WorkerNotReady" && e.status === 404,
78 schedule: Schedule.spaced("2 seconds"),
79 times: 10,
80 }),
81 );
82 const { instanceId } = (yield* res.json) as { instanceId: string };
83 expect(instanceId).toBeTypeOf("string");
84 return instanceId;
85 });
86
87/** Poll one instance without retrying failed workflow executions. */
88const waitForTerminal = (url: string, instanceId: string) =>
89 Effect.gen(function* () {
90 const client = yield* HttpClient.HttpClient;
91 return yield* client.get(`${url}/workflow/status/${instanceId}`).pipe(
92 Effect.flatMap((res) =>
93 res.status === 200
94 ? res.json.pipe(
95 Effect.map((json) => json as unknown as WorkflowStatus),
96 )
97 : res.text.pipe(
98 Effect.flatMap((body) =>
99 Effect.fail(
100 new Error(`Workflow status ${res.status}: ${body}`),
101 ),
102 ),
103 ),
104 ),
105 Effect.repeat({
106 schedule: Schedule.spaced("3 seconds"),
107 until: isTerminal,
108 times: 10,
109 }),
110 );
111 });
112
113const runInstance = (url: string, path: string, live = false) =>
114 Effect.gen(function* () {
115 const instanceId = yield* startInstance(url, path);
116 const status = yield* waitForTerminal(url, instanceId);
117 // A newly deployed live workflow can precede its worker/engine link.
118 if (
119 live &&
120 status.status === "errored" &&
121 status.error?.message === "Worker not found."
122 ) {
123 return yield* Effect.fail(new WorkflowLinkNotReady({ instanceId }));
124 }
125 return { instanceId, status };
126 }).pipe(
127 Effect.retry({
128 while: (error) => error instanceof WorkflowLinkNotReady,
129 schedule: Schedule.spaced("3 seconds"),
130 times: 2,
131 }),
132 );
133
134const probeWorkflow = Effect.fn(function* (url: string) {
135 const ready = yield* Effect.gen(function* () {
136 const id = yield* startInstance(url, "/workflow/probe");
137 const status = yield* waitForTerminal(url, id);
138 yield* Effect.logInfo(
139 `Workflow readiness probe ${id}: ${JSON.stringify(status)}`,
140 );
141 if (
142 status.status === "errored" &&
143 (status.error?.message === "Worker not found." ||
144 (status.error?.name === "TypeError" &&
145 (status.error.message ===
146 'The RPC receiver does not implement the method "run".' ||
147 status.error.message ===
148 "The entrypoint name LocalTestWorkflow was not found in this worker. Ensure the worker exports an entrypoint with that name.")))
149 )
150 return false;
151 expect(status).toMatchObject({
152 status: "complete",
153 output: { ready: true },
154 });
155 return true;
156 }).pipe(
157 Effect.repeat({
158 schedule: Schedule.spaced("1 second"),
159 times: 8,
160 until: (ready) => ready,
161 }),
162 Effect.timeout("45 seconds"),
163 );
164 expect(ready, "Workflow entrypoint did not propagate").toBe(true);
165});
166
167const assertRollback = (url: string, live = false) =>
168 Effect.gen(function* () {
169 const client = yield* HttpClient.HttpClient;
170 const { instanceId, status } = yield* runInstance(
171 url,
172 "/workflow/rollback",
173 live,
174 );
175 expect(status).toMatchObject({ status: "errored" });
176 expect(status.error?.message).toContain("rollback requested");
177 // Local bindings omit rollback metadata; the persisted records verify execution.
178 if (live) {
179 expect(status.rollback).toEqual({ outcome: "complete", error: null });
180 }
181
182 const response = yield* client.get(
183 `${url}/workflow/rollback-result/${instanceId}`,
184 );
185 expect(response.status).toBe(200);
186 expect(yield* response.json).toEqual(
187 ["undefined", "both", "timeout-only", "retries-only"].map((step) => ({
188 output: { value: "reserved", step },
189 error: "rollback requested",
190 })),
191 );
192 });
193
194const assertFailureScenarios = (url: string, live = false) =>
195 Effect.gen(function* () {
196 const client = yield* HttpClient.HttpClient;
197 for (const scenario of [
198 "timeout-zero",
199 "rollback-timeout-zero",
200 "retry-exhaustion",
201 "rollback-retry-exhaustion",
202 ]) {
203 const { instanceId, status } = yield* runInstance(
204 url,
205 `/workflow/scenario/${scenario}`,
206 live,
207 );
208 const zeroTimeout = scenario.endsWith("timeout-zero");
209 const response = yield* client.get(
210 `${url}/workflow/record/${instanceId}/${zeroTimeout ? "protected" : "attempts"}`,
211 );
212 expect(response.status).toBe(200);
213 const record = yield* response.json;
214 yield* Effect.log("Workflow failure scenario", {
215 scenario,
216 status,
217 record,
218 });
219 expect(status).toMatchObject({ status: "errored" });
220 if (zeroTimeout) {
221 if (live && scenario === "rollback-timeout-zero") {
222 // Cloudflare validates rollback config on execution, without exposing its error.
223 expect(status.error?.message).toBe("rollback requested");
224 expect(status.rollback).toEqual({ outcome: "failed", error: null });
225 } else {
226 expect(status.error?.message).toContain("invalid format");
227 }
228 expect(record).toBeNull();
229 } else {
230 expect(status.error?.message).toContain(
231 scenario === "retry-exhaustion"
232 ? "retry budget exhausted"
233 : "rollback requested",
234 );
235 expect(record).toEqual({ attempt: 2 });
236 if (live && scenario === "rollback-retry-exhaustion") {
237 expect(status.rollback?.outcome).toBe("failed");
238 expect(status.rollback?.error?.message).toContain(
239 "rollback budget exhausted",
240 );
241 }
242 }
243 }
244 });
245
246/**
247 * Read the persisted state row of the nested `Cloudflare.Workflow` resource
248 * (the WorkflowResource the `Workflow` effect-class registers under the host
249 * worker) from the scratch stack's private state store.
250 */
251const readWorkflowRow = (stack: Test.ScratchStack) =>
252 Effect.gen(function* () {
253 const state = yield* yield* State.State;
254 const fqns = yield* state.list({ stack: stack.name, stage: stack.stage });
255 for (const fqn of fqns) {
256 const row = yield* state.get({
257 stack: stack.name,
258 stage: stack.stage,
259 fqn,
260 });
261 if (
262 row &&
263 !State.isActionState(row) &&
264 row.resourceType === "Cloudflare.Workflow"
265 ) {
266 return row;
267 }
268 }
269 return undefined;
270 }).pipe(Effect.provide(stack.state));
271
272const asyncWorkflowMain = Effect.gen(function* () {
273 const path = yield* Path.Path;
274 return path.resolve(import.meta.dirname, "fixtures/async-workflow-worker.ts");
275});
276
277/**
278 * Under `alchemy dev` the Workflow resource is emulated by the local provider
279 * (a `dev:` id, no cloud API calls) and the host worker's `workflow` binding
280 * is lowered onto the local workerd workflow engine. This exercises the full
281 * local roundtrip: create an instance through the binding and poll it to
282 * completion against the local simulator.
283 */
284test.provider(
285 "workflow runs to completion against the local simulator",
286 (stack) =>
287 Effect.gen(function* () {
288 yield* stack.destroy();
289
290 const deployed = yield* stack.deploy(
291 Effect.gen(function* () {
292 const worker = yield* WorkflowLocalWorker;
293 return { worker };
294 }),
295 );
296
297 // The worker serves from the local dev proxy — proof the stack ran in
298 // local mode.
299 expect(deployed.worker.url).toMatch(/^http:\/\/localhost:\d+$/);
300
301 // The local provider fabricated a `dev:` workflow id (proof no cloud
302 // call ran) and the row is stamped with the local provider mode.
303 const row = yield* readWorkflowRow(stack);
304 expect(row).toBeDefined();
305 expect(row!.attr?.workflowId).toMatch(/^dev:/);
306 expect(row!.providerMode).toBe("local");
307
308 // Drive the workflow through the binding against local workerd.
309 const url = deployed.worker.url!;
310 const instanceId = yield* startInstance(url);
311 const status = yield* waitForTerminal(url, instanceId);
312
313 expect(status).toMatchObject({ status: "complete" });
314 expect(status.output?.greeting).toBe("Hello, world!");
315 expect(status.output?.instanceId).toBe(instanceId);
316
317 expect(status.error).toBeFalsy();
318 expect(status.output?.retryAttempt).toBe(2);
319 expect(status.output?.retryConfig.retries).toEqual({
320 limit: 2,
321 delay: "1 second",
322 backoff: "constant",
323 });
324 expect(status.output?.retryConfig.timeout).toBeDefined();
325 expect(status.output?.timeoutConfig.timeout).toBe("30 seconds");
326 expect(status.output?.timeoutConfig.retries).toBeDefined();
327 expect(status.output?.defaultsOk).toBe(true);
328 yield* Effect.log("Workflow resolved configuration", status.output);
329 yield* assertRollback(url);
330 yield* assertFailureScenarios(url);
331
332 const client = yield* HttpClient.HttpClient;
333 const events = yield* client
334 .get(`${url}/workflow/events/${instanceId}`)
335 .pipe(Effect.flatMap((response) => response.json));
336 expect(events).toEqual([
337 expect.objectContaining({ type: "workflow_queued" }),
338 ]);
339 const deleted = yield* client
340 .post(`${url}/workflow/delete/${instanceId}`)
341 .pipe(Effect.flatMap((response) => response.json));
342 expect(deleted).toEqual({
343 deleted: [],
344 errors: [expect.objectContaining({ id: instanceId })],
345 });
346 const batchId = yield* startInstance(url);
347 const batch = yield* client
348 .post(`${url}/workflow/delete-batch/${batchId}`)
349 .pipe(Effect.flatMap((response) => response.json));
350 expect(batch).toEqual({
351 deleted: [{ id: batchId }],
352 errors: [expect.objectContaining({ id: "missing-instance" })],
353 });
354
355 yield* stack.destroy();
356 }).pipe(logLevel),
357 {
358 tags: [
359 "provider:cloudflare",
360 "provider:cloudflare:r2",
361 "provider:cloudflare:worker",
362 "provider:cloudflare:workflow",
363 "local",
364 ],
365 timeout: 120_000,
366 },
367);
368
369// Exercise physical names through real workerd bindings, not just metadata.
370test.provider(
371 "async Worker binding preserves an explicit physical Workflow name",
372 (stack) =>
373 Effect.gen(function* () {
374 yield* stack.destroy();
375
376 const workflowName = "existing-workflow-physical-name";
377 const main = yield* asyncWorkflowMain;
378 const deployWith = (name?: string) =>
379 stack.deploy(
380 Cloudflare.Worker("ExplicitWorkflowWorker", {
381 main,
382 env: {
383 WORKFLOW_NAME: name ?? workflowName,
384 EXISTING_WORKFLOW: Cloudflare.Workflow("ExistingWorkflow", {
385 workflowName: name,
386 }),
387 },
388 }),
389 );
390 const created = yield* deployWith(workflowName);
391 const workflowRow = yield* readWorkflowRow(stack);
392 expect(workflowRow?.providerMode).toBe("local");
393 expect(workflowRow?.attr?.workflowId).toMatch(/^dev:/);
394 expect(workflowRow?.attr?.workflowName).toBe(workflowName);
395 const instance = yield* startInstance(created.url!);
396 expect(
397 (yield* waitForTerminal(created.url!, instance)).output?.workflowName,
398 ).toBe(workflowName);
399
400 yield* deployWith();
401 const preserved = yield* readWorkflowRow(stack);
402 expect(preserved?.attr?.workflowId).toBe(workflowRow?.attr?.workflowId);
403 expect(preserved?.attr?.workflowName).toBe(workflowName);
404
405 const renamed = yield* deployWith(`${workflowName}-renamed`);
406 const replaced = yield* readWorkflowRow(stack);
407 expect(replaced?.attr?.workflowId).not.toBe(
408 workflowRow?.attr?.workflowId,
409 );
410 expect(replaced?.attr?.workflowName).toBe(`${workflowName}-renamed`);
411 const client = yield* HttpClient.HttpClient;
412 const ready = yield* client.get(`${renamed.url!}/workflow/name`).pipe(
413 Effect.flatMap((response) => response.text),
414 Effect.repeat({
415 schedule: Schedule.spaced("1 second"),
416 until: (name) => name === `${workflowName}-renamed`,
417 times: 8,
418 }),
419 );
420 expect(ready).toBe(`${workflowName}-renamed`);
421 const renamedInstance = yield* startInstance(renamed.url!);
422 expect(
423 (yield* waitForTerminal(renamed.url!, renamedInstance)).output
424 ?.workflowName,
425 ).toBe(`${workflowName}-renamed`);
426
427 yield* stack.destroy();
428 }).pipe(logLevel),
429 {
430 tags: [
431 "provider:cloudflare",
432 "provider:cloudflare:worker",
433 "provider:cloudflare:workflow",
434 "local",
435 ],
436 timeout: 120_000,
437 },
438);
439
440test.provider(
441 "cross-script binding runs the host's explicit Workflow name",
442 (stack) =>
443 Effect.gen(function* () {
444 yield* stack.destroy();
445
446 const workflowName = "existing-cross-script-workflow";
447 const main = yield* asyncWorkflowMain;
448 const host = Cloudflare.Worker("ExplicitWorkflowHost", {
449 main,
450 env: {
451 EXISTING_WORKFLOW: Cloudflare.Workflow("ExistingWorkflow", {
452 workflowName,
453 }),
454 },
455 });
456 yield* stack.deploy(host);
457 const deployed = yield* stack.deploy(
458 Effect.gen(function* () {
459 const worker = yield* host;
460 return yield* Cloudflare.Worker("ExplicitWorkflowConsumer", {
461 main,
462 env: {
463 EXISTING_WORKFLOW: Cloudflare.Workflow("ExistingWorkflow", {
464 scriptName: worker.workerName,
465 workflowName,
466 }),
467 },
468 });
469 }),
470 );
471 const instance = yield* startInstance(deployed.url!);
472 const status = yield* waitForTerminal(deployed.url!, instance);
473 expect(status.status).toBe("complete");
474 expect(status.output?.workflowName).toBe(workflowName);
475
476 yield* stack.destroy();
477 }).pipe(logLevel),
478 {
479 tags: [
480 "provider:cloudflare",
481 "provider:cloudflare:worker",
482 "provider:cloudflare:workflow",
483 "local",
484 ],
485 timeout: 120_000,
486 },
487);
488
489test.provider(
490 "Effect-native Worker binding preserves an explicit physical Workflow name",
491 (stack) =>
492 Effect.gen(function* () {
493 yield* stack.destroy();
494 const deployed = yield* stack.deploy(ExplicitNameWorkflowWorker);
495
496 const workflowRow = yield* readWorkflowRow(stack);
497 expect(workflowRow?.providerMode).toBe("local");
498 expect(workflowRow?.attr?.workflowName).toBe(EXPLICIT_WORKFLOW_NAME);
499 const instance = yield* startInstance(deployed.url!);
500 const status = yield* waitForTerminal(deployed.url!, instance);
501 expect(status.status).toBe("complete");
502 expect(status.output?.workflowName).toBe(EXPLICIT_WORKFLOW_NAME);
503
504 yield* stack.destroy();
505 }).pipe(logLevel),
506 {
507 tags: [
508 "provider:cloudflare",
509 "provider:cloudflare:worker",
510 "provider:cloudflare:workflow",
511 "local",
512 ],
513 timeout: 120_000,
514 },
515);
516
517test.provider(
518 "Alchemy.remote() preserves an explicit Workflow name and schedules",
519 (stack) =>
520 Effect.gen(function* () {
521 yield* stack.destroy();
522 const deployed = yield* stack.deploy(
523 ExplicitNameWorkflowWorker.pipe(Alchemy.remote()),
524 );
525 expect(deployed.url).not.toMatch(/^http:\/\/localhost/);
526
527 const row = yield* readWorkflowRow(stack);
528 expect(row?.providerMode).toBe("live");
529 expect(row?.attr?.workflowId).not.toMatch(/^dev:/);
530 const workflowName = row!.attr!.workflowName;
531 const accountId = row!.attr!.accountId;
532 const observed = yield* workflows.getWorkflow({
533 accountId,
534 workflowName,
535 });
536 expect(observed.id).toBe(row!.attr!.workflowId);
537 expect(observed.name).toBe(EXPLICIT_WORKFLOW_NAME);
538 expect(observed.schedules?.map((schedule) => schedule.cron)).toEqual([
539 "0 0 1 1 *",
540 ]);
541
542 const { status } = yield* runInstance(
543 deployed.url!,
544 "/workflow/start/world",
545 true,
546 );
547 expect(status).toMatchObject({ status: "complete" });
548 expect(status.error).toBeFalsy();
549 expect(status.output?.greeting).toBe("Hello, world!");
550 expect(status.output?.workflowName).toBe(EXPLICIT_WORKFLOW_NAME);
551
552 yield* stack.destroy();
553 const gone = yield* workflows
554 .getWorkflow({ accountId, workflowName })
555 .pipe(
556 Effect.as(false),
557 Effect.catchTag("WorkflowNotFound", () => Effect.succeed(true)),
558 );
559 expect(gone).toBe(true);
560 }).pipe(logLevel),
561 {
562 tags: [
563 "provider:cloudflare",
564 "provider:cloudflare:worker",
565 "provider:cloudflare:workflow",
566 "live",
567 ],
568 timeout: 120_000,
569 },
570);
571
572/**
573 * `Alchemy.remote()` opts the whole worker + hosted workflow OUT of local
574 * emulation: even under `alchemy dev` the worker deploys to real Cloudflare
575 * and `putWorkflow` registers a real account-level Workflow. The workflow
576 * class is hosted BY the worker script, so both must run live together —
577 * a live workflow cannot reference a script that only exists in local
578 * workerd. After destroy, an out-of-band `getWorkflow` proves the cloud
579 * workflow is gone (pins the stamped-mode delete path).
580 */
581test.provider(
582 "Alchemy.remote() worker + workflow run live in dev and delete on destroy",
583 (stack) =>
584 Effect.gen(function* () {
585 yield* stack.destroy();
586
587 const deployed = yield* stack.deploy(
588 Effect.gen(function* () {
589 const worker = yield* WorkflowLocalWorker;
590 return { worker };
591 }).pipe(Alchemy.remote()),
592 );
593
594 // Live deploy: a real workers.dev URL, not the local dev proxy.
595 expect(deployed.worker.url).not.toMatch(/^http:\/\/localhost/);
596
597 // The workflow has a real Cloudflare id and a `live` mode stamp.
598 const row = yield* readWorkflowRow(stack);
599 expect(row).toBeDefined();
600 expect(row!.attr?.workflowId).not.toMatch(/^dev:/);
601 expect(row!.providerMode).toBe("live");
602 const workflowName = row!.attr!.workflowName;
603 const workflowAccountId = row!.attr!.accountId;
604
605 // Out-of-band: the workflow exists on real Cloudflare.
606 const live = yield* workflows.getWorkflow({
607 accountId: workflowAccountId,
608 workflowName,
609 });
610 expect(live.id).toBe(row!.attr!.workflowId);
611
612 const url = deployed.worker.url!;
613 yield* probeWorkflow(url);
614 const { status } = yield* runInstance(url, "/workflow/start/world", true);
615 expect(status).toMatchObject({ status: "complete" });
616 expect(status.error).toBeFalsy();
617 expect(status.output?.greeting).toBe("Hello, world!");
618 expect(status.output?.retryAttempt).toBe(2);
619 expect(status.output?.retryConfig.retries).toEqual({
620 limit: 2,
621 delay: "1 second",
622 backoff: "constant",
623 });
624 expect(status.output?.retryConfig.timeout).toBeDefined();
625 expect(status.output?.timeoutConfig.timeout).toBe("30 seconds");
626 expect(status.output?.timeoutConfig.retries).toBeDefined();
627 expect(status.output?.defaultsOk).toBe(true);
628 yield* Effect.log("Workflow resolved configuration", status.output);
629 yield* assertRollback(url, true);
630 yield* assertFailureScenarios(url, true);
631
632 yield* stack.destroy();
633
634 // The live-stamped row was deleted through the live provider even in a
635 // dev run — the cloud workflow is gone.
636 const gone = yield* workflows
637 .getWorkflow({ accountId: workflowAccountId, workflowName })
638 .pipe(
639 Effect.as(false),
640 Effect.catchTag("WorkflowNotFound", () => Effect.succeed(true)),
641 );
642 expect(gone).toBe(true);
643 }).pipe(logLevel),
644 {
645 tags: [
646 "provider:cloudflare",
647 "provider:cloudflare:r2",
648 "provider:cloudflare:worker",
649 "provider:cloudflare:workflow",
650 "live",
651 ],
652 timeout: 120_000,
653 },
654);