import * as workflows from "@distilled.cloud/cloudflare/workflows"; import type { ConfigError } from "effect/Config"; import * as Context from "effect/Context"; import * as Effect from "effect/Effect"; import type { Scope } from "effect/Scope"; import type { Input } from "../../Input.ts"; import type { PlatformServices } from "../../Platform.ts"; import * as Provider from "../../Provider.ts"; import { Resource } from "../../Resource.ts"; import type { RuntimeContext } from "../../RuntimeContext.ts"; import { CloudflareEnvironment } from "../CloudflareEnvironment.ts"; import { Worker, type WorkerServices } from "../Workers/Worker.ts"; type TypeId = "Cloudflare.Workflow"; declare const TypeId: "Cloudflare.Workflow"; declare const WorkflowEvent_base: Context.ServiceClass; /** * Service that carries the current workflow event payload. * `yield* WorkflowEvent` inside a workflow body to access it. */ export declare class WorkflowEvent extends WorkflowEvent_base { } export interface WorkflowCronSchedule { cron: string; scheduledTime: number; } export type WorkflowBackoff = "constant" | "linear" | "exponential"; export interface WorkflowStepConfig { retries?: { limit: number; delay: string | number; backoff?: WorkflowBackoff; }; timeout?: string | number; } export interface WorkflowStepContextData { step: { name: string; count: number; }; attempt: number; config: WorkflowStepConfig; } declare const WorkflowStepContext_base: Context.ServiceClass; /** * Runtime information for the current `task` attempt. */ export declare class WorkflowStepContext extends WorkflowStepContext_base { } export interface WorkflowRollbackContext { error: Error; output: Output | undefined; } export interface WorkflowRollbackOptions { rollback: (context: WorkflowRollbackContext) => Effect.Effect; rollbackConfig?: WorkflowStepConfig; } /** * Optional configuration for a `task` step: retry policy, timeout, and a * rollback handler with its own retry config. */ export interface WorkflowTaskConfig extends WorkflowStepConfig { rollback?: (context: WorkflowRollbackContext) => Effect.Effect; rollbackConfig?: WorkflowStepConfig; } /** * Internal step descriptor passed from `task` to the bridge. Bundles the step * name and Effect together with the `WorkflowTaskConfig` fields. */ export interface WorkflowTaskOptions extends WorkflowTaskConfig { name: string; effect: Effect.Effect; } export interface WorkflowWaitForEventOptions { type: string; timeout?: string | number; } /** * The event delivered to a `waitForEvent` step. Mirrors the native * `WorkflowStepEvent` shape from `cloudflare:workers` 1:1. */ export interface WorkflowStepEvent { payload: Payload; timestamp: Date; type: string; } type ExcludeWorkflowStepContext = R extends { readonly key: "Cloudflare.WorkflowStepContext"; } ? never : R; declare const WorkflowStep_base: Context.ServiceClass(options: WorkflowTaskOptions): Effect.Effect; sleep(name: string, duration: string | number): Effect.Effect; sleepUntil(name: string, timestamp: Date | number): Effect.Effect; waitForEvent(name: string, options: WorkflowWaitForEventOptions): Effect.Effect>; }>; /** * Internal service that wraps the Cloudflare `WorkflowStep` object. * Not accessed directly by users -- use `task`, `sleep`, `sleepUntil`, and * `waitForEvent` instead. */ export declare class WorkflowStep extends WorkflowStep_base { } /** * Execute a named, durable workflow step. The effect is run inside the * Cloudflare step transaction so its result is automatically persisted * and replayed on retries. * * Any services the inner effect requires (e.g. `WorkerEnvironment` from a * binding like `kv.put` / `kv.get`) are threaded through automatically by * capturing the surrounding workflow body's context and providing it to * the inner effect before it runs inside `step.do`. * * The step name comes first, followed by the Effect. Retry config, timeout, * and a rollback handler can be passed in the optional third `options` arg. */ export declare function task(name: string, effect: Effect.Effect, options?: WorkflowTaskConfig): Effect.Effect>; /** * Pause the workflow for the given duration. */ export declare const sleep: (name: string, duration: string | number) => Effect.Effect; /** * Pause the workflow until the given timestamp. */ export declare const sleepUntil: (name: string, timestamp: Date | number) => Effect.Effect; /** * Pause the workflow until an external event is delivered with * `WorkflowInstance.sendEvent`. Resolves with the full * {@link WorkflowStepEvent} (`{ payload, timestamp, type }`), exactly like * the native `step.waitForEvent`. */ export declare const waitForEvent: (name: string, options: WorkflowWaitForEventOptions) => Effect.Effect, never, WorkflowStep>; /** * The services available inside a workflow run body. * * `WorkerEnvironment` is provided to the body at runtime by the workflow * export wrapper (see `make(env)` below), so users can access env bindings * from inside workflow steps via `yield* WorkerEnvironment` — the type must * reflect that or `yield* WorkerEnvironment` fails to type-check inside a * body even though it succeeds at runtime. * * A fresh `Scope` is provided per run-invocation by `WorkflowBridge.run` and * threaded into every `task` via the surrounding body context, so `@binding` * helpers that acquire per-run resources against the ambient scope (e.g. * `Drizzle.Postgres`) resolve them inside workflow steps just as they do in * a Worker `fetch`/`queue` handler. */ export type WorkflowRunServices = WorkflowEvent | WorkflowStep | WorkerServices | Scope; export type WorkflowServices = WorkflowRunServices | PlatformServices | RuntimeContext; /** * Metadata stored in the worker export map to distinguish workflow exports * from durable object exports at bundle-generation time. */ export interface WorkflowExport { readonly kind: "workflow"; readonly make: (env: unknown) => Effect.Effect>; } /** * A workflow implementation is a function from a typed `Input` payload to * an Effect that produces the workflow's `Result`. The Effect requires * `WorkflowRunServices` (event + step + env) to execute. */ export type WorkflowImpl = (input: Input) => Effect.Effect; export declare const isWorkflowExport: (value: unknown) => value is WorkflowExport; /** * Limits applied to the workflow on create or update. */ export interface WorkflowLimits { /** * Maximum number of steps a single workflow instance may execute. */ steps?: number; } /** * 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; /** * Limits applied to the workflow. Only applies when the workflow is hosted by * the Worker that declares the binding; ignored when `scriptName` is set. */ limits?: WorkflowLimits; } /** * Props for the Effect-native form of {@link Workflow} * (`Workflow(name, props, impl)`). Used when the workflow's implementation is * defined inline by the hosting Worker. */ export interface WorkflowProps { /** * Limits applied to the workflow. */ limits?: WorkflowLimits; } /** * 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 */ limits?: WorkflowLimits; /** @internal phantom */ Params?: Params; } /** * Type guard for the reference (async) form of a Workflow. */ export declare const isWorkflowLike: (value: unknown) => value is WorkflowLike; /** * Type guard for workflow binding metadata in the Worker binding contract. */ export declare const isWorkflowBinding: (binding: { type: string; }) => binding is { type: "workflow"; name: string; workflowName: string; className: string; scriptName?: string; }; /** * Handle returned to the caller at deploy/bind time. Allows starting * workflow instances and checking their status from the Api layer. */ export interface WorkflowHandle { Type: TypeId; name: string; /** * Start a workflow instance. Pass payload through `params`; omit `id` to let * Cloudflare generate an instance ID. */ create(options?: WorkflowInstanceCreateOptions): Effect.Effect>; createBatch(batch: WorkflowInstanceCreateOptions[]): Effect.Effect[]>; get(instanceId: string): Effect.Effect>; } /** Options for starting a workflow instance. */ export interface WorkflowInstanceCreateOptions { id?: string; params?: Input; retention?: WorkflowInstanceRetention; } export interface WorkflowInstanceRetention { successRetention?: string | number; errorRetention?: string | number; } /** Handle for a single Cloudflare workflow instance. */ export interface WorkflowInstance { id: string; status(): Effect.Effect>; pause(): Effect.Effect; resume(): Effect.Effect; restart(options?: WorkflowInstanceRestartOptions): Effect.Effect; terminate(): Effect.Effect; sendEvent(event: WorkflowInstanceEvent): Effect.Effect; } export interface WorkflowInstanceRestartOptions { from?: { name: string; count?: number; type?: "do" | "sleep" | "waitForEvent"; }; } export interface WorkflowInstanceEvent { type: string; payload?: Payload; } export interface WorkflowInstanceStatus { status: "queued" | "running" | "paused" | "errored" | "terminated" | "complete" | "waiting" | "waitingForPause" | "unknown" | (string & {}); output?: Result; error?: { name: string; message: string; } | null; rollback?: { outcome: "complete" | "failed"; error: { name: string; message: string; } | null; } | null; } export interface WorkflowClass extends Effect.Effect { <_Self>(): { (name: string, impl: Effect.Effect, ConfigError, InitReq>): Effect.Effect, never, Worker | Exclude> & { new (_: never): WorkflowImpl; }; (name: string, props: WorkflowProps, impl: Effect.Effect, ConfigError, InitReq>): Effect.Effect, never, Worker | Exclude> & { new (_: never): WorkflowImpl; }; }; (name: string, props?: WorkflowRefProps): WorkflowLike; (name: string, impl: Effect.Effect, ConfigError, InitReq>): Effect.Effect, never, Worker | Exclude>; (name: string, props: WorkflowProps, impl: Effect.Effect, ConfigError, InitReq>): Effect.Effect, never, Worker | Exclude>; } declare const WorkflowScope_base: Context.ServiceClass>; export declare class WorkflowScope extends WorkflowScope_base { } /** * A Cloudflare Workflow that orchestrates durable, multi-step tasks with * automatic retries and at-least-once delivery. * * A Workflow follows the same two-phase pattern as Workers and Durable * Objects. The outer `Effect.gen` resolves shared dependencies. The inner * `Effect.fn` is the workflow body — a function from a typed `input` * payload to an Effect that runs steps using `task`, `sleep`, and * `sleepUntil`. `task` takes the step name and Effect, plus an optional * config object for retries, timeout, and a rollback handler. * * ```typescript * Effect.gen(function* () { * // Phase 1: resolve dependencies * const notifier = yield* NotificationService; * * return Effect.fn(function* (input: { orderId: string }) { * // Phase 2: workflow body (durable steps) * const result = yield* Cloudflare.Workflows.task("process", doWork(input.orderId)); * yield* Cloudflare.Workflows.sleep("cooldown", "10 seconds"); * return result; * }); * }) * ``` * * * ### Defining a Workflow * **Example:** Minimal workflow * ```typescript * export default class MyWorkflow extends Cloudflare.Workflow()( * "MyWorkflow", * Effect.gen(function* () { * return Effect.fn(function* (input: { name: string }) { * return { received: input.name }; * }); * }), * ) {} * ``` * * **Example:** Setting a step limit * ```typescript * export default class MyWorkflow extends Cloudflare.Workflow()( * "MyWorkflow", * { limits: { steps: 25000 } }, * Effect.gen(function* () { * return Effect.fn(function* (input: { name: string }) { * return { received: input.name }; * }); * }), * ) {} * ``` * * ### Step Primitives * **Example:** Running a named task * ```typescript * const result = yield* Cloudflare.Workflows.task( * "process-order", * Effect.succeed({ orderId: "abc", total: 42 }), * ); * ``` * * **Example:** Configuring retries and reading step context * ```typescript * const result = yield* Cloudflare.Workflows.task( * "call-api", * Effect.gen(function* () { * const context = yield* Cloudflare.Workflows.WorkflowStepContext; * return { attempt: context.attempt }; * }), * { retries: { limit: 3, delay: "5 seconds", backoff: "linear" } }, * ); * ``` * * **Example:** Registering rollback * ```typescript * yield* Cloudflare.Workflows.task("reserve-inventory", reserveInventory, { * rollback: ({ output }) => * output ? releaseInventory(output.reservationId) : Effect.void, * rollbackConfig: { retries: { limit: 3, delay: "10 seconds" } }, * }); * ``` * * **Example:** Sleeping between steps * ```typescript * yield* Cloudflare.Workflows.sleep("cooldown", "30 seconds"); * ``` * * **Example:** Waiting for an external event * ```typescript * const event = yield* Cloudflare.Workflows.waitForEvent<{ approved: boolean }>( * "approval", * { type: "approval", timeout: "1 day" }, * ); * // Same shape as the native step.waitForEvent result: * event.payload.approved; * ``` * * **Example:** Accessing env bindings inside a task * Bind a resource (e.g. `Namespace`, `Bucket`) in the workflow's * outer init phase to get a typed Effect-native client, then use it * directly inside `task`. `task` threads the binding's service * requirement (`WorkerEnvironment`) through automatically so the inner * Effect needs no extra plumbing. * * ```typescript * Effect.gen(function* () { * const kv = yield* Cloudflare.KV.ReadWriteNamespace(KV); * * return Effect.fn(function* (input: { roomId: string; message: string }) { * const { roomId, message } = input; * * const stored = yield* Cloudflare.Workflows.task( * "kv-roundtrip", * Effect.gen(function* () { * const key = `workflow:${roomId}`; * yield* kv.put(key, message); * return yield* kv.get(key); * }).pipe(Effect.orDie), * ); * * return stored; * }); * }); * ``` * * ### Starting and Monitoring Instances * `create` mirrors Cloudflare's native Workflow API: pass workflow input in * `params`, pass `id` only when you need a deterministic instance ID, and omit * `id` to let Cloudflare generate one. * * **Example:** Creating an instance from a Worker * ```typescript * const workflow = yield* MyWorkflow; * const instance = yield* workflow.create({ params: { orderId: "abc" } }); * ``` * * **Example:** Creating an instance with id and retention * ```typescript * const instance = yield* workflow.create({ * id: "order-abc", * params: { orderId: "abc" }, * retention: { successRetention: "1 day", errorRetention: "7 days" }, * }); * ``` * * **Example:** Creating a batch * ```typescript * const instances = yield* workflow.createBatch([ * { id: "order-a", params: { orderId: "a" } }, * { id: "order-b", params: { orderId: "b" } }, * ]); * ``` * * **Example:** Checking instance status * ```typescript * const workflow = yield* MyWorkflow; * const handle = yield* workflow.get(instanceId); * const status = yield* handle.status(); * ``` * * **Example:** Sending events and restarting instances * ```typescript * const instance = yield* workflow.get(instanceId); * yield* instance.sendEvent({ type: "approval", payload: { approved: true } }); * yield* instance.restart({ from: { name: "approval", type: "waitForEvent" } }); * ``` * * ### Triggering from a Worker * Wire the workflow into HTTP routes so callers can fire instances * and poll for completion. * * **Example:** Workflow start + status routes * ```typescript * // src/worker.ts * const notifier = yield* MyWorkflow; * * return { * fetch: Effect.gen(function* () { * const request = yield* HttpServerRequest; * * if (request.url.startsWith("/workflow/start/")) { * const id = request.url.split("/").pop()!; * const instance = yield* notifier.create({ params: { orderId: id } }); * return HttpServerResponse.json({ instanceId: instance.id }); * } * * if (request.url.startsWith("/workflow/status/")) { * const id = request.url.split("/").pop()!; * const instance = yield* notifier.get(id); * return HttpServerResponse.json(yield* instance.status()); * } * * return HttpServerResponse.text("Not Found", { status: 404 }); * }), * }; * ``` * * ### 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 }); * }, * }; * ``` * * ### 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, * }), * }, * }); * ``` * * ### Testing Workflows * Workflows run asynchronously, so tests start an instance and poll until it * reaches a terminal status. Keep polling bounded with `Effect.repeat`. * * **Example:** Polling for workflow completion * ```typescript * test( * "workflow completes", * Effect.gen(function* () { * const { url } = yield* stack; * * const start = yield* HttpClient.post(`${url}/workflow/start/x`); * const { instanceId } = (yield* start.json) as { instanceId: string }; * * const status = yield* HttpClient.get( * `${url}/workflow/status/${instanceId}`, * ).pipe( * Effect.flatMap((res) => res.json), * Effect.map((json) => json as { status: string }), * Effect.repeat({ * schedule: Schedule.spaced("2 seconds"), * until: (status) => * status.status === "complete" || status.status === "errored", * times: 30, * }), * ); * * expect(status.status).toBe("complete"); * }), * { timeout: 120_000 }, * ); * ``` * * @resource * @product Workflows * @category Workers & Compute */ export declare const Workflow: WorkflowClass; export interface WorkflowResourceProps { /** * Account-global Workflow name. */ workflowName: string; className: string; scriptName: string; limits?: WorkflowLimits; } export interface WorkflowResourceAttrs { workflowId: string; workflowName: string; className: string; scriptName: string; accountId: string; } declare const WorkflowResourceTypeId = "Cloudflare.Workflow"; export interface WorkflowResource extends Resource { } export declare const WorkflowResource: import("../../Resource.ts").ResourceClass; export declare const ProviderLive: () => import("effect/Layer").Layer, never, CloudflareEnvironment | workflows.CloudflareOpContext>; /** * Local (dev) provider — the workflow is purely virtual: a `dev:` id keyed * into the local workerd workflow engine. The host worker's `workflow` * binding is lowered onto the local runtime's Workflow Engine DO by * `LocalWorkerProvider` (`Workflows.local(...)`), so no runtime layer is * needed here; instance state persists under the worker's local storage. */ export declare const ProviderLocal: () => import("effect/Layer").Layer, never, CloudflareEnvironment>; export declare const WorkflowProvider: () => import("effect/Layer").Layer, never, import("../../AlchemyContext.ts").AlchemyContext | CloudflareEnvironment | workflows.CloudflareOpContext>; export {}; //# sourceMappingURL=Workflow.d.ts.map