diff --git a/packages/alchemy/src/Cloudflare/Workers/InferEnv.ts b/packages/alchemy/src/Cloudflare/Workers/InferEnv.ts index db0c0db69f..e44244485c 100644 --- a/packages/alchemy/src/Cloudflare/Workers/InferEnv.ts +++ b/packages/alchemy/src/Cloudflare/Workers/InferEnv.ts @@ -4,6 +4,7 @@ import type * as Effect from "effect/Effect"; import type { Redacted } from "effect/Redacted"; import type * as Stream from "effect/Stream"; import type { Rpc } from "../../Rpc.ts"; +import type { WorkflowLike } from "../Workflows/Workflow.ts"; // NOTE: import the service modules directly rather than `import * as Cloudflare // from "../index.ts"`. Importing the whole Cloudflare barrel here creates a // circular re-export when the barrel does `export * from "./Workers/index.ts"` @@ -79,15 +80,17 @@ export type GetBindingType = ? WorkerVersionMetadata : T extends WorkerLoaderResource ? WorkerLoader - : T extends DurableObjectLike - ? DurableObjectNamespace< - Exclude - > - : T extends Redacted - ? // redacteds are always stored as secret_text, so are always string - // we JSON.stringify when not a Redacted - string - : T; + : T extends WorkflowLike + ? Workflow + : T extends DurableObjectLike + ? DurableObjectNamespace< + Exclude + > + : T extends Redacted + ? // redacteds are always stored as secret_text, so are always string + // we JSON.stringify when not a Redacted + string + : T; /** * Cloudflare service-binding wire shape for an Effect-native Worker. diff --git a/packages/alchemy/src/Cloudflare/Workers/WorkerAsyncBindings.ts b/packages/alchemy/src/Cloudflare/Workers/WorkerAsyncBindings.ts index d31bcb5732..f872ff8b58 100644 --- a/packages/alchemy/src/Cloudflare/Workers/WorkerAsyncBindings.ts +++ b/packages/alchemy/src/Cloudflare/Workers/WorkerAsyncBindings.ts @@ -21,6 +21,7 @@ import { isQueue } from "../Queues/Queue.ts"; import { isBucket } from "../R2/Bucket.ts"; import { isSecret } from "../SecretsStore/Secret.ts"; import { isIndex } from "../Vectorize/VectorizeIndex.ts"; +import { isWorkflowLike, WorkflowResource } from "../Workflows/Workflow.ts"; import { isAssets } from "./Assets.ts"; import { isBrowser } from "./Browser.ts"; import { isDurableObjectLike } from "./DurableObject.ts"; @@ -64,6 +65,20 @@ export const bindWorkerAsyncBindings = Effect.fn(function* ( ? getHyperdriveDevOrigin(binding) : undefined, }); + + // A locally-hosted Workflow (no `scriptName`) must be registered with + // Cloudflare via `putWorkflow` once the host Worker exists. Cross-script + // references (with `scriptName`) are reference-only — the host owns the + // workflow resource. `scriptName: resource.workerName` makes the + // WorkflowResource depend on the Worker so it reconciles afterwards. + if (isWorkflowLike(binding) && !binding.scriptName) { + const workflowName = binding.workflowName ?? binding.name; + yield* WorkflowResource(workflowName, { + workflowName, + className: binding.className ?? binding.name, + scriptName: resource.workerName, + }); + } } else { return yield* Effect.die(`Unknown binding type: ${bindingName}`); } @@ -155,6 +170,14 @@ const toBinding = ( className: binding.className ?? binding.name, scriptName: binding.scriptName, }; + } else if (isWorkflowLike(binding)) { + return { + type: "workflow", + name: bindingName, + workflowName: binding.workflowName ?? binding.name, + className: binding.className ?? binding.name, + scriptName: binding.scriptName, + }; } else if (isDatabase(binding)) { return { type: "d1", diff --git a/packages/alchemy/src/Cloudflare/Workers/WorkerBinding.ts b/packages/alchemy/src/Cloudflare/Workers/WorkerBinding.ts index a289258500..3fff77dda3 100644 --- a/packages/alchemy/src/Cloudflare/Workers/WorkerBinding.ts +++ b/packages/alchemy/src/Cloudflare/Workers/WorkerBinding.ts @@ -20,6 +20,7 @@ import type { Queue } from "../Queues/Queue.ts"; import type { Bucket } from "../R2/Bucket.ts"; import type { Secret } from "../SecretsStore/Secret.ts"; import type { Index as VectorizeIndex } from "../Vectorize/VectorizeIndex.ts"; +import type { WorkflowLike } from "../Workflows/Workflow.ts"; import type { Assets } from "./Assets.ts"; import type { BrowserBinding } from "./BrowserBinding.ts"; import type { DurableObjectLike } from "./DurableObject.ts"; @@ -66,7 +67,8 @@ export type WorkerBindingResource = | Worker | WorkerLoader | VersionMetadataBinding - | DurableObjectLike; + | DurableObjectLike + | WorkflowLike; export type WorkerBindings = { [bindingName in string]: WorkerBindingResource; diff --git a/packages/alchemy/src/Cloudflare/Workflows/Workflow.ts b/packages/alchemy/src/Cloudflare/Workflows/Workflow.ts index 7096b871e7..c1cb35717b 100644 --- a/packages/alchemy/src/Cloudflare/Workflows/Workflow.ts +++ b/packages/alchemy/src/Cloudflare/Workflows/Workflow.ts @@ -5,6 +5,7 @@ import * as Effect from "effect/Effect"; import * as Stream from "effect/Stream"; import { AlchemyContext } from "../../AlchemyContext.ts"; import { ExecutionContext } from "../../ExecutionContext.ts"; +import type { Input } from "../../Input.ts"; import { ALCHEMY_PHASE } from "../../Phase.ts"; import type { PlatformServices } from "../../Platform.ts"; import * as Provider from "../../Provider.ts"; @@ -149,6 +150,52 @@ export const isWorkflowExport = (value: unknown): value is WorkflowExport => "kind" in value && (value as any).kind === "workflow"; +/** + * Props for the reference (async) form of {@link Workflow}. Used when binding + * a Workflow class to a plain async Worker (one without an Effect runtime) via + * the Worker's `env`. Mirrors `DurableObjectProps`. + */ +export interface WorkflowRefProps { + /** + * Name of the exported `WorkflowEntrypoint` class. + * + * @default name + */ + className?: string; + /** + * Worker script that hosts the Workflow class. Omit this when the workflow + * is hosted by the Worker that declares the binding. + */ + scriptName?: Input; +} + +/** + * A lightweight reference to a Workflow, produced by the props-only form of + * {@link Workflow} (`Workflow(name, { className })`). Carries just enough + * metadata to emit the `workflow` binding for an async Worker and to drive + * the `putWorkflow` lifecycle. Mirrors `DurableObjectLike`. + */ +export interface WorkflowLike { + kind: TypeId; + name: string; + /** @internal phantom */ + workflowName?: string; + /** @internal phantom */ + className?: string; + /** @internal phantom */ + scriptName?: Input; + /** @internal phantom */ + Params?: Params; +} + +/** + * Type guard for the reference (async) form of a Workflow. + */ +export const isWorkflowLike = (value: unknown): value is WorkflowLike => + typeof value === "object" && + value !== null && + (value as { kind?: unknown }).kind === TypeId; + /** * Type guard for workflow binding metadata in the Worker binding contract. */ @@ -204,6 +251,10 @@ export interface WorkflowClass extends Effect.Effect< new (_: never): WorkflowImpl; }; }; + ( + name: string, + props?: WorkflowRefProps, + ): WorkflowLike; ( name: string, impl: Effect.Effect, ConfigError, InitReq>, @@ -346,6 +397,70 @@ export class WorkflowScope extends Context.Service< * }; * ``` * + * @section Binding in an Async Worker + * When using an Async Worker (plain `async fetch` handler, no Effect + * runtime), declare Workflows in the `env` prop of the Worker resource. + * Pass a `Workflow` reference with a `className` matching the exported + * `WorkflowEntrypoint` subclass in your worker source file. If `className` + * is omitted, it defaults to the binding name. Use `Cloudflare.InferEnv` + * to get a fully typed `env` object that includes the workflow binding. + * + * @example Declaring a Workflow binding in the stack + * ```typescript + * // alchemy.run.ts + * export type WorkerEnv = Cloudflare.InferEnv; + * + * export const Worker = Cloudflare.Worker("Worker", { + * main: "./src/worker.ts", + * env: { + * MY_WORKFLOW: Cloudflare.Workflow<{ value: string }>("MyWorkflow", { + * className: "MyWorkflow", + * }), + * }, + * }); + * ``` + * + * @example Using the Workflow from a plain async handler + * ```typescript + * // src/worker.ts + * import { WorkflowEntrypoint, type WorkflowEvent, type WorkflowStep } from "cloudflare:workers"; + * import type { WorkerEnv } from "../alchemy.run.ts"; + * + * export class MyWorkflow extends WorkflowEntrypoint { + * async run(event: Readonly>, step: WorkflowStep) { + * return await step.do("greet", async () => `Hello, ${event.payload.value}!`); + * } + * } + * + * export default { + * async fetch(request: Request, env: WorkerEnv) { + * const instance = await env.MY_WORKFLOW.create({ params: { value: "world" } }); + * return Response.json({ instanceId: instance.id }); + * }, + * }; + * ``` + * + * @section Cross-Script Binding in an Async Worker + * Async Workers can also bind to a Workflow hosted by another Worker + * script. The host Worker declares and exports the `WorkflowEntrypoint` + * class. The consumer Worker declares a `Workflow` with `scriptName` set + * to the host Worker's script name. Cross-script references are bindings + * only — Alchemy does not drive `putWorkflow` for the foreign class, so + * deploy the host first. + * + * @example Consumer Worker binds to the host script + * ```typescript + * const consumer = yield* Cloudflare.Worker("Consumer", { + * main: "./src/consumer.ts", + * env: { + * MY_WORKFLOW: Cloudflare.Workflow("MyWorkflow", { + * className: "MyWorkflow", + * scriptName: host.workerName, + * }), + * }, + * }); + * ``` + * * @section Testing Workflows * Workflows run asynchronously, so tests start an instance and * poll until it reaches a terminal status. A simple recipe with @@ -381,91 +496,105 @@ export class WorkflowScope extends Context.Service< * ``` */ export const Workflow: WorkflowClass = taggedFunction(WorkflowScope, (( - ...args: [] | [name: string, impl: Effect.Effect>] -) => - args.length === 0 - ? Workflow - : effectClass( - Effect.gen(function* () { - const [name, impl] = args; - const worker = yield* Worker; - - // Add the workflow binding to the Worker metadata - yield* worker.bind`${name}`({ - bindings: [ - { - type: "workflow", - name, - workflowName: name, - className: name, - }, - ], - }); + ...args: + | [] + | [name: string, impl: Effect.Effect>] + | [name: string, props?: WorkflowRefProps] +) => { + if (args.length === 0) { + return Workflow; + } + const [name, second] = args; + if (!Effect.isEffect(second)) { + // Props-only (async) reference form: returns a plain `WorkflowLike` that an + // async Worker binds via `env`. `WorkerAsyncBindings` emits the `workflow` + // binding and drives `putWorkflow` for locally-hosted workflows. + const props = second as WorkflowRefProps | undefined; + return { + kind: TypeId, + name, + workflowName: name, + className: props?.className ?? name, + scriptName: props?.scriptName, + } satisfies WorkflowLike; + } + const impl = second; + return effectClass( + Effect.gen(function* () { + const worker = yield* Worker; - // Create the Workflow API resource (putWorkflow / deleteWorkflow) - yield* WorkflowResource(name, { + // Add the workflow binding to the Worker metadata + yield* worker.bind`${name}`({ + bindings: [ + { + type: "workflow", + name, workflowName: name, className: name, - scriptName: worker.workerName, - }); - - const services = - yield* Effect.context>(); - - const binding = yield* Effect.all([ - WorkerEnvironment, - ALCHEMY_PHASE, - ]).pipe( - Effect.flatMap(([env, phase]) => { - if (env === undefined || phase === "plan") { - return Effect.succeed(undefined as any); - } - const wf = env[name]; - if (!wf) { - return Effect.die( - new Error(`Workflow '${name}' not found in env`), - ); - } - return Effect.succeed(wf); - }), - ); + }, + ], + }); - const self: WorkflowHandle = { - Type: TypeId, - name, - create: (input: unknown) => - Effect.tryPromise(() => binding.create({ params: input })).pipe( - Effect.map(wrapInstance), - Effect.orDie, - ), - get: (instanceId: string) => - Effect.tryPromise(() => binding.get(instanceId)).pipe( - Effect.map(wrapInstance), - Effect.orDie, - ), - }; + // Create the Workflow API resource (putWorkflow / deleteWorkflow) + yield* WorkflowResource(name, { + workflowName: name, + className: name, + scriptName: worker.workerName, + }); - const fn = yield* impl.pipe( - Effect.provideService(WorkflowScope, self as any), - ); + const services = yield* Effect.context>(); - yield* worker.export(name, { - kind: "workflow", - make: (env: unknown) => - Effect.succeed(((input: unknown) => - fn(input).pipe( - Effect.provideService( - WorkerEnvironment, - env as Record, - ), - )) as WorkflowImpl).pipe( - Effect.provideContext(services), + const binding = yield* Effect.all([ + WorkerEnvironment, + ALCHEMY_PHASE, + ]).pipe( + Effect.flatMap(([env, phase]) => { + if (env === undefined || phase === "plan") { + return Effect.succeed(undefined as any); + } + const wf = env[name]; + if (!wf) { + return Effect.die(new Error(`Workflow '${name}' not found in env`)); + } + return Effect.succeed(wf); + }), + ); + + const self: WorkflowHandle = { + Type: TypeId, + name, + create: (input: unknown) => + Effect.tryPromise(() => binding.create({ params: input })).pipe( + Effect.map(wrapInstance), + Effect.orDie, + ), + get: (instanceId: string) => + Effect.tryPromise(() => binding.get(instanceId)).pipe( + Effect.map(wrapInstance), + Effect.orDie, + ), + }; + + const fn = yield* impl.pipe( + Effect.provideService(WorkflowScope, self as any), + ); + + yield* worker.export(name, { + kind: "workflow", + make: (env: unknown) => + Effect.succeed(((input: unknown) => + fn(input).pipe( + Effect.provideService( + WorkerEnvironment, + env as Record, ), - } satisfies WorkflowExport); + )) as WorkflowImpl).pipe(Effect.provideContext(services)), + } satisfies WorkflowExport); - return self; - }), - )) as any); + return self; + }), + ); +}) as any); // --------------------------------------------------------------------------- // WorkflowResource -- manages the Cloudflare Workflows API lifecycle diff --git a/packages/alchemy/src/Platform.ts b/packages/alchemy/src/Platform.ts index 48616e2676..9551489fc3 100644 --- a/packages/alchemy/src/Platform.ts +++ b/packages/alchemy/src/Platform.ts @@ -242,7 +242,12 @@ export const Platform = < type: R["Type"], hooks: { createRuntimeContext: (id: string) => BaseRuntimeContext; - onCreate?: (resource: R, props: any) => Effect.Effect; + // `onCreate` runs inside the resource-construction context, which already + // carries the Stack's providers — so the hook may yield child resources + // (e.g. an async Worker registering a `WorkflowResource` for a bound + // Workflow). Allow an ambient requirement (`any`) rather than forcing + // `never`; it is discharged by the surrounding provider context. + onCreate?: (resource: R, props: any) => Effect.Effect; }, methods?: { [key: string]: any }, ): any => { diff --git a/packages/alchemy/test/Cloudflare/Workers/WorkflowAsync.test.ts b/packages/alchemy/test/Cloudflare/Workers/WorkflowAsync.test.ts new file mode 100644 index 0000000000..3c04f5911f --- /dev/null +++ b/packages/alchemy/test/Cloudflare/Workers/WorkflowAsync.test.ts @@ -0,0 +1,211 @@ +import * as Cloudflare from "@/Cloudflare"; +import * as Test from "@/Test/Vitest"; +import { expect } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import { MinimumLogLevel } from "effect/References"; +import * as Schedule from "effect/Schedule"; +import * as HttpClient from "effect/unstable/http/HttpClient"; +import Stack from "./fixtures/workflow-async/stack.ts"; + +const { test, beforeAll, afterAll, deploy, destroy } = Test.make({ + providers: Cloudflare.providers(), +}); + +const logLevel = Effect.provideService( + MinimumLogLevel, + process.env.DEBUG ? "Debug" : "Info", +); + +const stack = beforeAll( + deploy(Stack).pipe( + // Let the freshly-deployed worker (and its Workflow binding) settle before + // the first run so a step doesn't error mid-propagation. + Effect.tap(Effect.sleep("5 seconds")), + ), +); +afterAll.skipIf(!!process.env.NO_DESTROY)(destroy(Stack)); + +interface WorkflowStatus { + status: string; + output?: { greeting: string }; + error?: { message?: string } | null; +} + +// Start a fresh workflow instance and poll until it reaches a terminal state. +// A transient `errored` during edge/binding propagation fails this effect so +// the caller can retry with a brand-new instance. +const runWorkflowToCompletion = (url: string) => + Effect.gen(function* () { + const client = yield* HttpClient.HttpClient; + + // Cloudflare's edge takes a few seconds to start serving a fresh + // workers.dev URL, so retry until it returns 200 (a fresh URL also + // returns 404 transiently, which is not an HTTP error so Effect.retry + // does not catch it unless we explicitly fail on non-200). + const startRes = yield* client.post(`${url}/workflow/start/world`).pipe( + Effect.flatMap((res) => + res.status === 200 + ? Effect.succeed(res) + : Effect.fail(new Error(`Worker not ready: ${res.status}`)), + ), + Effect.retry({ + schedule: Schedule.exponential("500 millis"), + times: 15, + }), + ); + const { instanceId } = (yield* startRes.json) as { instanceId: string }; + expect(instanceId).toBeTypeOf("string"); + + const lastStatus = yield* client + .get(`${url}/workflow/status/${instanceId}`) + .pipe( + Effect.flatMap((res) => res.json), + Effect.map((json) => json as unknown as WorkflowStatus), + Effect.repeat({ + schedule: Schedule.spaced("2 seconds"), + until: (s) => s.status === "complete" || s.status === "errored", + times: 12, + }), + ); + + // Surface a non-complete terminal state as a failure so the outer retry + // can take another swing (a fresh worker occasionally errors a step while + // its bindings are still propagating). + if (lastStatus.status !== "complete") { + return yield* Effect.fail( + new Error( + `workflow ${lastStatus.status}: ${JSON.stringify(lastStatus.error)}`, + ), + ); + } + return lastStatus; + }); + +test( + "async worker can run a class-based workflow bound via env", + Effect.gen(function* () { + const out = yield* stack; + const url = out.url; + expect(url).toBeTypeOf("string"); + + const lastStatus = yield* runWorkflowToCompletion(url).pipe( + Effect.retry({ schedule: Schedule.spaced("3 seconds"), times: 2 }), + ); + + expect(lastStatus.status).toBe("complete"); + expect(lastStatus.error).toBeFalsy(); + expect(lastStatus.output?.greeting).toBe("Hello, world!"); + }).pipe(logLevel), + { timeout: 30_000 }, +); + +// --------------------------------------------------------------------------- +// Cross-script binding: a consumer Worker binds a Workflow hosted by another +// Worker script via `scriptName`. The host owns the workflow (props-only form +// with no `scriptName` → drives `putWorkflow`); the consumer's binding is a +// reference only. Inline `script` keeps both workers in this one file. +// --------------------------------------------------------------------------- + +// Host hosts the WorkflowEntrypoint class AND drives a workflow instance. +const hostWorkflowScript = `import { WorkflowEntrypoint } from "cloudflare:workers"; +export class MyWorkflow extends WorkflowEntrypoint { + async run(event, step) { + const greeting = await step.do("greet", async () => \`Hello, \${event.payload.value}!\`); + return await step.do("finalize", async () => ({ greeting })); + } +} +export default { + async fetch(request, env) { + const url = new URL(request.url); + if (url.pathname.startsWith("/workflow/start/")) { + const value = url.pathname.split("/workflow/start/")[1] ?? "world"; + const instance = await env.MY_WORKFLOW.create({ params: { value } }); + return Response.json({ instanceId: instance.id }); + } + if (url.pathname.startsWith("/workflow/status/")) { + const id = url.pathname.split("/workflow/status/")[1] ?? ""; + const instance = await env.MY_WORKFLOW.get(id); + return Response.json(await instance.status()); + } + return new Response("ok"); + }, +}; +`; + +// Consumer has no class — it only references the host's workflow. +const consumerWorkflowScript = `export default { + async fetch(request, env) { + const url = new URL(request.url); + if (url.pathname.startsWith("/workflow/start/")) { + const value = url.pathname.split("/workflow/start/")[1] ?? "world"; + const instance = await env.MY_WORKFLOW.create({ params: { value } }); + return Response.json({ instanceId: instance.id }); + } + if (url.pathname.startsWith("/workflow/status/")) { + const id = url.pathname.split("/workflow/status/")[1] ?? ""; + const instance = await env.MY_WORKFLOW.get(id); + return Response.json(await instance.status()); + } + return new Response("ok"); + }, +}; +`; + +test.provider( + "async worker workflow binding accepts scriptName (cross-script)", + (scratch) => + Effect.gen(function* () { + // Deploy the host first so its workflow exists (putWorkflow) before the + // consumer references it by scriptName. + yield* scratch.deploy( + Effect.gen(function* () { + return { + host: yield* Cloudflare.Worker("host-workflow-worker", { + script: hostWorkflowScript, + env: { + MY_WORKFLOW: Cloudflare.Workflow("MyWorkflow"), + }, + }), + }; + }), + ); + + const deployed = yield* scratch.deploy( + Effect.gen(function* () { + const host = yield* Cloudflare.Worker("host-workflow-worker", { + script: hostWorkflowScript, + env: { + MY_WORKFLOW: Cloudflare.Workflow("MyWorkflow"), + }, + }); + const consumer = yield* Cloudflare.Worker( + "consumer-workflow-worker", + { + script: consumerWorkflowScript, + env: { + MY_WORKFLOW: Cloudflare.Workflow("MyWorkflow", { + scriptName: host.workerName, + }), + }, + }, + ); + return { consumer, host }; + }), + ); + + // Start + complete a workflow instance through the CONSUMER's binding, + // exercising both `create` and `get` across the cross-script reference. + const lastStatus = yield* runWorkflowToCompletion( + deployed.consumer.url!, + ).pipe( + Effect.retry({ schedule: Schedule.spaced("3 seconds"), times: 2 }), + ); + + expect(lastStatus.status).toBe("complete"); + expect(lastStatus.error).toBeFalsy(); + expect(lastStatus.output?.greeting).toBe("Hello, world!"); + + yield* scratch.destroy(); + }).pipe(logLevel), + { timeout: 120_000 }, +); diff --git a/packages/alchemy/test/Cloudflare/Workers/fixtures/workflow-async/stack.ts b/packages/alchemy/test/Cloudflare/Workers/fixtures/workflow-async/stack.ts new file mode 100644 index 0000000000..23f67631e7 --- /dev/null +++ b/packages/alchemy/test/Cloudflare/Workers/fixtures/workflow-async/stack.ts @@ -0,0 +1,32 @@ +import * as Cloudflare from "@/Cloudflare"; +import * as Alchemy from "@/index"; +import * as Effect from "effect/Effect"; +import * as path from "pathe"; + +export type AsyncWorkflowEnv = Cloudflare.InferEnv; + +// Async (non-Effect) Worker that hosts a `WorkflowEntrypoint` class and binds +// it through `env` using the props-only `Cloudflare.Workflow` reference form. +export const AsyncWorkflowWorker = Cloudflare.Worker("AsyncWorkflowWorker", { + main: path.resolve(import.meta.dirname, "worker.ts"), + url: true, + env: { + MY_WORKFLOW: Cloudflare.Workflow<{ value: string }>("MyWorkflow", { + className: "MyWorkflow", + }), + }, +}); + +export default Alchemy.Stack( + "AsyncWorkflowBindingStack", + { + providers: Cloudflare.providers(), + state: Cloudflare.state(), + }, + Effect.gen(function* () { + const worker = yield* AsyncWorkflowWorker; + return { + url: worker.url.as(), + }; + }), +); diff --git a/packages/alchemy/test/Cloudflare/Workers/fixtures/workflow-async/worker.ts b/packages/alchemy/test/Cloudflare/Workers/fixtures/workflow-async/worker.ts new file mode 100644 index 0000000000..2deb14b794 --- /dev/null +++ b/packages/alchemy/test/Cloudflare/Workers/fixtures/workflow-async/worker.ts @@ -0,0 +1,46 @@ +import { + WorkflowEntrypoint, + type WorkflowEvent, + type WorkflowStep, +} from "cloudflare:workers"; +import type { AsyncWorkflowEnv } from "./stack.ts"; + +interface Params { + value: string; +} + +// Plain (non-Effect) Workflow class hosted by an async Worker. Bound to the +// Worker via `env.MY_WORKFLOW` using the props-only `Cloudflare.Workflow` +// reference form — the analogue of binding a class-based Durable Object. +export class MyWorkflow extends WorkflowEntrypoint { + async run(event: Readonly>, step: WorkflowStep) { + const greeting = await step.do( + "greet", + async () => `Hello, ${event.payload.value}!`, + ); + + await step.sleep("cooldown", "1 second"); + + return await step.do("finalize", async () => ({ greeting })); + } +} + +export default { + async fetch(request: Request, env: AsyncWorkflowEnv): Promise { + const url = new URL(request.url); + + if (url.pathname.startsWith("/workflow/start/")) { + const value = url.pathname.split("/workflow/start/")[1] ?? "world"; + const instance = await env.MY_WORKFLOW.create({ params: { value } }); + return Response.json({ instanceId: instance.id }); + } + + if (url.pathname.startsWith("/workflow/status/")) { + const id = url.pathname.split("/workflow/status/")[1] ?? ""; + const instance = await env.MY_WORKFLOW.get(id); + return Response.json(await instance.status()); + } + + return new Response("ok"); + }, +};