1import { makeCallback } from "alchemy/Callback";
2import type { RuntimeContext } from "alchemy/RuntimeContext";
3import {
4 CurrentActivity,
5 decodeExit,
6 decodeResult,
7 encodeExit,
8 encodeResult,
9 makeCurrentActivity,
10 payloadCodec,
11} from "alchemy/WorkflowEncoding";
12import { StorageName } from "alchemy/Workflows";
13import * as Cause from "effect/Cause";
14import * as Clock from "effect/Clock";
15import * as Context from "effect/Context";
16import * as Duration from "effect/Duration";
17import * as Effect from "effect/Effect";
18import * as Exit from "effect/Exit";
19import * as Fiber from "effect/Fiber";
20import * as Layer from "effect/Layer";
21import * as Option from "effect/Option";
22import * as Schedule from "effect/Schedule";
23import * as Schema from "effect/Schema";
24import type * as Scope from "effect/Scope";
25import { Workflow, WorkflowEngine } from "effect/workflow";
26import {
27 DurableObject as CloudflareDurableObject,
28 type DurableObjectJurisdiction,
29} from "../Workers/DurableObject.ts";
30import { DurableObjectState } from "../Workers/DurableObjectState.ts";
31import type { Worker } from "../Workers/Worker.ts";
32
33/**
34 * Runs Effect workflows on Durable Objects. Each workflow gets its own Durable
35 * Object class on the host Worker, and each execution is one instance of it,
36 * named by its execution id, with its journal in that instance's SQLite.
37 *
38 * A run replays the workflow from the top: finished activities return their
39 * stored results, and the first unfinished one runs and is stored. A run that
40 * waits on a `DurableClock` or `DurableDeferred` that isn't ready suspends,
41 * and the clock firing or the deferred being completed runs it again. Every
42 * run happens in one durable callback per execution, so runs never overlap,
43 * and a run that fails or is cut short is retried.
44 *
45 * ### Running workflows on Durable Objects
46 * **Example:** Provide the engine to a Worker
47 * ```typescript
48 * import * as Cloudflare from "@alchemy.run/cloudflare";
49 * import * as Workflows from "alchemy/Workflows";
50 * import * as Effect from "effect/Effect";
51 * import * as Layer from "effect/Layer";
52 * import * as HttpServerResponse from "effect/http/HttpServerResponse";
53 * import * as Schema from "effect/Schema";
54 * import { Activity, Workflow } from "effect/workflow";
55 *
56 * const SendWelcome = Workflow.make("SendWelcome", {
57 * payload: { userId: Schema.String },
58 * idempotencyKey: ({ userId }) => userId,
59 * });
60 *
61 * const SendWelcomeLive = Workflows.layer(
62 * SendWelcome,
63 * Effect.fn(function* ({ userId }) {
64 * yield* Activity.make({ name: "send", execute: Effect.log(`welcome, ${userId}`) });
65 * }),
66 * );
67 *
68 * export default class AppWorker extends Cloudflare.Worker<AppWorker>()(
69 * "AppWorker",
70 * { main: import.meta.url },
71 * Effect.gen(function* () {
72 * const welcome = yield* Workflows.RunWorkflow(SendWelcome);
73 * return {
74 * fetch: Effect.gen(function* () {
75 * const run = yield* welcome.start({ userId: "u-1" });
76 * return HttpServerResponse.text(run.executionId);
77 * }),
78 * };
79 * }).pipe(
80 * Effect.provide(
81 * SendWelcomeLive.pipe(
82 * Layer.provideMerge(Cloudflare.Workflows.layerDurableObjectEngine()),
83 * ),
84 * ),
85 * ),
86 * ) {}
87 * ```
88 *
89 * ### Running executions in a jurisdiction
90 * **Example:** Every execution in the EU
91 * ```typescript
92 * import * as Cloudflare from "@alchemy.run/cloudflare";
93 * import * as Layer from "effect/Layer";
94 * import { SendWelcomeLive } from "./SendWelcome.ts";
95 *
96 * const WorkflowsLive = SendWelcomeLive.pipe(
97 * Layer.provideMerge(Cloudflare.Workflows.layerDurableObjectEngine({ jurisdiction: "eu" })),
98 * );
99 * ```
100 *
101 * @layer
102 * @provides WorkflowEngine
103 * @product Durable Objects
104 */
105export function layerDurableObjectEngine({
106 jurisdiction,
107}: DurableObjectEngineOptions = {}): Layer.Layer<WorkflowEngine.WorkflowEngine, never, Worker> {
108 // The same options return the same Layer, so providing it in several places
109 // builds one engine.
110 let engine = engines.get(jurisdiction);
111 if (engine === undefined) engines.set(jurisdiction, (engine = makeEngine(jurisdiction)));
112 return engine;
113}
114
115/** Options for {@link layerDurableObjectEngine}. */
116export interface DurableObjectEngineOptions {
117 /**
118 * Create and store every workflow execution object only inside this
119 * jurisdiction, e.g. `"eu"`. An execution in a jurisdiction is a different
120 * object from the one with the same name outside it, so setting this does
121 * not move executions that already exist.
122 *
123 * Enforced when deployed, not emulated in `alchemy dev` (local workerd does
124 * not implement jurisdictions).
125 *
126 * @default undefined (no jurisdiction restriction)
127 */
128 readonly jurisdiction?: DurableObjectJurisdiction;
129}
130
131const engines = new Map<
132 DurableObjectJurisdiction | undefined,
133 Layer.Layer<WorkflowEngine.WorkflowEngine, never, Worker>
134>();
135
136const makeEngine = (
137 jurisdiction: DurableObjectJurisdiction | undefined,
138): Layer.Layer<WorkflowEngine.WorkflowEngine, never, Worker> =>
139 Layer.effect(
140 WorkflowEngine.WorkflowEngine,
141 Effect.gen(function* () {
142 // Workflows register while their Layers build, inside the Worker's init
143 // phase; declaring a Durable Object class needs that context.
144 const init = yield* Effect.context<Worker>();
145 const namespaces = new Map<string, Namespace>();
146
147 const namespaceOf = (workflowName: string) =>
148 Effect.suspend(() => {
149 const namespace = namespaces.get(workflowName);
150 return namespace
151 ? Effect.succeed(namespace)
152 : Effect.die(new Error(`Workflow ${workflowName} is not registered`));
153 });
154 const stubOf = (workflowName: string, executionId: string) =>
155 Effect.map(namespaceOf(workflowName), (namespace) => namespace.getByName(executionId));
156
157 const engine: WorkflowEngine.WorkflowEngine["Service"] = WorkflowEngine.makeUnsafe({
158 register: (workflow, execute) =>
159 Effect.gen(function* () {
160 if (namespaces.has(workflow._tag)) return;
161 const namespace = yield* declare(
162 `${workflowClassName(workflow)}Workflow`,
163 executionObject({ workflow: workflow as AnyWorkflow, execute, engine: () => engine }),
164 ).pipe(Effect.provide(init));
165 namespaces.set(
166 workflow._tag,
167 jurisdiction === undefined ? namespace : namespace.jurisdiction(jurisdiction),
168 );
169 }),
170
171 execute: (workflow, options) =>
172 Effect.gen(function* () {
173 const payload = yield* Schema.encodeEffect(payloadCodec(workflow as AnyWorkflow))(
174 options.payload,
175 ).pipe(Effect.orDie);
176 const parent: ParentRef | null = options.parent
177 ? {
178 workflowName: options.parent.workflow._tag,
179 executionId: options.parent.executionId,
180 }
181 : null;
182 const stub = yield* stubOf(workflow._tag, options.executionId);
183 const result = yield* call<string | null>(
184 stub.start(options.executionId, JSON.stringify(payload), JSON.stringify(parent)),
185 );
186 if (options.discard) return undefined;
187 return result === null
188 ? Workflow.Suspended.make({})
189 : yield* decodeResult(workflow as AnyWorkflow, result);
190 }) as Effect.Effect<any>,
191
192 poll: (workflow, executionId) =>
193 Effect.gen(function* () {
194 const stub = yield* stubOf(workflow._tag, executionId);
195 const polled = yield* call<{
196 readonly started: boolean;
197 readonly result: string | null;
198 }>(stub.poll());
199 if (!polled.started) return Option.none();
200 if (polled.result === null) return Option.some(Workflow.Suspended.make({}));
201 return Option.some(yield* decodeResult(workflow as AnyWorkflow, polled.result));
202 }),
203
204 interrupt: (workflow, executionId) =>
205 Effect.flatMap(stubOf(workflow._tag, executionId), (stub) => call(stub.interrupt())),
206
207 interruptUnsafe: (workflow, executionId) =>
208 Effect.flatMap(stubOf(workflow._tag, executionId), (stub) =>
209 call(stub.interruptUnsafe()),
210 ),
211
212 resume: (workflow, executionId) =>
213 Effect.flatMap(stubOf(workflow._tag, executionId), (stub) => call(stub.resume())),
214
215 // An activity's result is stored the first time it completes and
216 // replayed on every run after that.
217 activityExecute: (activity, attempt) =>
218 Effect.gen(function* () {
219 const execution = yield* currentExecution("activity execution");
220 const instance = yield* WorkflowEngine.WorkflowInstance;
221 const key = `${activity.name}/${attempt}`;
222 const stored = yield* execution.query<{ exit: string }>(
223 "SELECT exit FROM workflow_activity WHERE key = ?",
224 key,
225 );
226 if (stored.length > 0) {
227 return new Workflow.Complete({ exit: yield* decodeExit(stored[0]!.exit) });
228 }
229 const activityInstance = WorkflowEngine.WorkflowInstance.initial(
230 instance.workflow,
231 instance.executionId,
232 );
233 activityInstance.interrupted = instance.interrupted;
234 const result = yield* activity.executeEncoded.pipe(
235 Workflow.intoResult,
236 Effect.provide(
237 Layer.mergeAll(
238 Layer.succeed(WorkflowEngine.WorkflowInstance, activityInstance),
239 Layer.succeed(
240 CurrentActivity,
241 makeCurrentActivity(
242 instance.workflow._tag,
243 instance.executionId,
244 activity.name,
245 ),
246 ),
247 ),
248 ),
249 );
250 if (result._tag === "Complete") {
251 yield* execution.query(
252 "INSERT OR REPLACE INTO workflow_activity (key, exit) VALUES (?, ?)",
253 key,
254 yield* encodeExit(result.exit),
255 );
256 }
257 return result;
258 }),
259
260 deferredResult: (deferred) =>
261 Effect.gen(function* () {
262 const execution = yield* currentExecution("DurableDeferred reads");
263 const stored = yield* execution.query<{ exit: string }>(
264 "SELECT exit FROM workflow_deferred WHERE name = ?",
265 deferred.name,
266 );
267 if (stored.length === 0) return Option.none();
268 return Option.some(yield* decodeExit(stored[0]!.exit));
269 }),
270
271 // Completing a deferred of the running execution writes locally;
272 // anything else goes to the execution's Durable Object.
273 deferredDone: (options) =>
274 Effect.gen(function* () {
275 const exit = yield* encodeExit(options.exit);
276 const current = yield* Effect.serviceOption(CurrentExecution);
277 if (
278 Option.isSome(current) &&
279 current.value.workflowName === options.workflowName &&
280 current.value.executionId === options.executionId
281 ) {
282 return yield* current.value.complete(options.deferredName, exit);
283 }
284 const stub = yield* stubOf(options.workflowName, options.executionId);
285 yield* call(stub.deferredDone(options.deferredName, exit));
286 }),
287
288 // A clock keeps the due time it was first given, however many times
289 // the run replays past `DurableClock.sleep`.
290 scheduleClock: (_workflow, options) =>
291 Effect.gen(function* () {
292 const execution = yield* currentExecution("DurableClock scheduling");
293 const now = yield* Clock.currentTimeMillis;
294 yield* execution.query(
295 "INSERT OR IGNORE INTO workflow_clock (name, deferred, due) VALUES (?, ?, ?)",
296 options.clock.name,
297 options.clock.deferred.name,
298 now + Duration.toMillis(options.clock.duration),
299 );
300 const [clock] = yield* execution.query<{ due: number }>(
301 "SELECT due FROM workflow_clock WHERE name = ?",
302 options.clock.name,
303 );
304 yield* execution.scheduleClock(
305 options.clock.name,
306 options.clock.deferred.name,
307 clock!.due,
308 );
309 }),
310 });
311
312 return engine;
313 }),
314 );
315
316// ---------------------------------------------------------------- the execution object
317
318type AnyWorkflow = Workflow.AnyWithProps;
319
320type Execute = (
321 payload: object,
322 executionId: string,
323) => Effect.Effect<
324 unknown,
325 unknown,
326 WorkflowEngine.WorkflowInstance | WorkflowEngine.WorkflowEngine
327>;
328
329interface Registration {
330 readonly workflow: AnyWorkflow;
331 readonly execute: Execute;
332 readonly engine: () => WorkflowEngine.WorkflowEngine["Service"];
333}
334
335interface ParentRef {
336 readonly workflowName: string;
337 readonly executionId: string;
338}
339
340/**
341 * The execution a run belongs to. The engine's in-run operations (activities,
342 * deferreds, clocks) read and write its journal through this.
343 */
344class CurrentExecution extends Context.Service<
345 CurrentExecution,
346 {
347 readonly workflowName: string;
348 readonly executionId: string;
349 readonly query: <T extends Record<string, string | number | null>>(
350 sql: string,
351 ...bindings: ReadonlyArray<unknown>
352 ) => Effect.Effect<ReadonlyArray<T>>;
353 readonly complete: (deferredName: string, exit: string) => Effect.Effect<void>;
354 readonly scheduleClock: (
355 name: string,
356 deferredName: string,
357 due: number,
358 ) => Effect.Effect<void>;
359 }
360>()("alchemy/Cloudflare/Workflows/DurableObjectEngine/CurrentExecution") {}
361
362const currentExecution = (operation: string) =>
363 Effect.flatMap(Effect.serviceOption(CurrentExecution), (current) =>
364 Option.isSome(current)
365 ? Effect.succeed(current.value)
366 : Effect.die(
367 new Error(`DurableObjectEngine: ${operation} only runs inside a workflow execution`),
368 ),
369 );
370
371const SCHEMA = `
372CREATE TABLE IF NOT EXISTS workflow_execution (
373 id INTEGER PRIMARY KEY CHECK (id = 1),
374 execution_id TEXT NOT NULL,
375 payload TEXT NOT NULL,
376 parent TEXT,
377 interrupted INTEGER NOT NULL DEFAULT 0,
378 result TEXT
379);
380CREATE TABLE IF NOT EXISTS workflow_activity (key TEXT PRIMARY KEY, exit TEXT NOT NULL);
381CREATE TABLE IF NOT EXISTS workflow_deferred (name TEXT PRIMARY KEY, exit TEXT NOT NULL);
382CREATE TABLE IF NOT EXISTS workflow_clock (
383 name TEXT PRIMARY KEY,
384 deferred TEXT NOT NULL,
385 due INTEGER NOT NULL
386);
387CREATE TABLE IF NOT EXISTS workflow_progress (
388 id INTEGER PRIMARY KEY CHECK (id = 1),
389 seq INTEGER NOT NULL
390);
391`;
392
393interface ExecutionRow {
394 readonly execution_id: string;
395 readonly payload: string;
396 readonly parent: string | null;
397 readonly interrupted: number;
398 readonly result: string | null;
399 readonly [column: string]: string | number | null;
400}
401
402/** The single job id every wake-up schedules, so pending wake-ups collapse into one run. */
403const RUN = "run";
404
405/** One workflow execution: its journal, and the runs that replay it. */
406const executionObject = (registration: Registration) =>
407 Effect.gen(function* () {
408 const state = yield* DurableObjectState;
409 return Effect.gen(function* () {
410 const context = yield* Effect.context<RuntimeContext>();
411 const query = <T extends Record<string, string | number | null>>(
412 sql: string,
413 ...bindings: ReadonlyArray<unknown>
414 ): Effect.Effect<ReadonlyArray<T>> =>
415 state.storage.sql.exec<T>(sql, ...bindings).pipe(
416 Effect.flatMap((cursor) => cursor.toArray()),
417 Effect.provide(context),
418 );
419 yield* query(SCHEMA);
420
421 const readExecution = Effect.map(
422 query<ExecutionRow>("SELECT * FROM workflow_execution"),
423 (rows) => rows[0],
424 );
425
426 // A wake-up scheduled from inside a callback inherits that callback's
427 // retry budget unless it reports progress. Each new journal entry that
428 // schedules a wake-up advances this cursor in the same transaction, so a
429 // wake-up after a successful step runs on time instead of waiting out a
430 // backoff.
431 const advance = Effect.map(
432 query<{ seq: number }>(
433 "INSERT INTO workflow_progress (id, seq) VALUES (1, 0) ON CONFLICT (id) DO UPDATE SET seq = seq + 1 RETURNING seq",
434 ),
435 (rows) => rows[0]!.seq,
436 );
437 const transaction = <A>(effect: Effect.Effect<A, never, RuntimeContext>) =>
438 state.storage.transaction(effect).pipe(Effect.orDie, Effect.provide(context));
439
440 // Every wake-up (a start, a completed deferred, a fired clock, an
441 // interrupt) records what changed and schedules this one job. Jobs run
442 // one at a time and are retried until a run succeeds.
443 const runs = yield* makeCallback(RUN, () => run, { retry: { delay: "5 seconds" } });
444 const wake = (progress?: number) =>
445 runs
446 .schedule(RUN, { after: 0, payload: {}, progress })
447 .pipe(Effect.orDie, Effect.provide(context));
448
449 const complete = (deferredName: string, exit: string) =>
450 transaction(
451 Effect.gen(function* () {
452 const inserted = yield* query(
453 "INSERT OR IGNORE INTO workflow_deferred (name, exit) VALUES (?, ?) RETURNING name",
454 deferredName,
455 exit,
456 );
457 if (inserted.length > 0) return yield* wake(yield* advance);
458 // Completed already, and that completion woke the run. Replacing a
459 // pending wake-up here would hand it this caller's retry budget.
460 const pending = yield* runs.getStatus(RUN).pipe(Effect.orDie);
461 if (pending === undefined) yield* wake();
462 }),
463 );
464
465 const clocks = yield* makeCallback(
466 "clock",
467 (job: { readonly deferred: string }) =>
468 Effect.gen(function* () {
469 yield* complete(job.deferred, yield* encodeExit(Exit.void));
470 }),
471 { retry: { delay: "5 seconds" } },
472 );
473 // Replays reach the same sleep again: a clock that's done or already
474 // pending is left alone.
475 const scheduleClock = (name: string, deferredName: string, due: number) =>
476 transaction(
477 Effect.gen(function* () {
478 const done = yield* query(
479 "SELECT 1 FROM workflow_deferred WHERE name = ?",
480 deferredName,
481 );
482 if (done.length > 0) return;
483 if ((yield* clocks.getStatus(name).pipe(Effect.orDie)) !== undefined) return;
484 yield* clocks
485 .schedule(name, {
486 at: due,
487 payload: { deferred: deferredName },
488 progress: yield* advance,
489 })
490 .pipe(Effect.orDie);
491 }),
492 );
493
494 const run: Effect.Effect<void> = Effect.gen(function* () {
495 const row = yield* readExecution;
496 if (row === undefined || row.result !== null) return;
497 const { workflow, execute } = registration;
498 const payload = yield* Schema.decodeEffect(payloadCodec(workflow))(
499 JSON.parse(row.payload),
500 ).pipe(Effect.orDie);
501 const instance = WorkflowEngine.WorkflowInstance.initial(workflow, row.execution_id);
502 const result = yield* execute(payload as object, row.execution_id).pipe(
503 // An interrupted execution replays to where it waits, then is
504 // interrupted there, which fails it and runs its compensations.
505 Effect.onExit(() => {
506 if (row.interrupted !== 1) return Effect.void;
507 instance.interrupted = true;
508 instance.suspended = false;
509 return Effect.withFiber((fiber) => fiber.pipe(Fiber.interrupt, Effect.interruptible));
510 }),
511 Workflow.intoResult,
512 Effect.provide(
513 Layer.mergeAll(
514 Layer.succeed(WorkflowEngine.WorkflowInstance, instance),
515 Layer.succeed(WorkflowEngine.WorkflowEngine, registration.engine()),
516 Layer.succeed(CurrentExecution, {
517 workflowName: workflow._tag,
518 executionId: row.execution_id,
519 query,
520 complete,
521 scheduleClock,
522 }),
523 ),
524 ),
525 );
526 // Suspended: a clock or a completed deferred wakes it again.
527 if (result._tag === "Suspended") return;
528 yield* query(
529 "UPDATE workflow_execution SET result = ?",
530 yield* encodeResult(workflow, result.exit),
531 );
532 if (row.parent !== null) yield* resumeParent(registration, JSON.parse(row.parent));
533 });
534
535 const result = Effect.map(readExecution, (row) => row?.result ?? null);
536
537 return {
538 /** Create the execution if it's new, and return its result once it has one. */
539 start: (executionId: string, payload: string, parent: string) =>
540 reportDefects(
541 Effect.gen(function* () {
542 const existing = yield* readExecution;
543 if (existing === undefined) {
544 yield* query(
545 "INSERT INTO workflow_execution (id, execution_id, payload, parent) VALUES (1, ?, ?, ?)",
546 executionId,
547 payload,
548 parent === "null" ? null : parent,
549 );
550 yield* wake();
551 }
552 return yield* result;
553 }),
554 ),
555 poll: () =>
556 reportDefects(
557 Effect.map(readExecution, (row) => ({
558 started: row !== undefined,
559 result: row?.result ?? null,
560 })),
561 ),
562 deferredDone: (deferredName: string, exit: string) =>
563 reportDefects(complete(deferredName, exit)),
564 resume: () => reportDefects(wake()),
565 // The next run replays with the instance marked interrupted, which
566 // fails the workflow and runs its compensations.
567 interrupt: () =>
568 reportDefects(
569 query("UPDATE workflow_execution SET interrupted = 1").pipe(Effect.andThen(wake())),
570 ),
571 // Stop without running compensations.
572 interruptUnsafe: () =>
573 reportDefects(
574 Effect.gen(function* () {
575 yield* runs.cancel(RUN).pipe(Effect.orDie, Effect.provide(context));
576 yield* query(
577 "UPDATE workflow_execution SET interrupted = 1, result = ? WHERE result IS NULL",
578 yield* encodeResult(registration.workflow, Exit.interrupt()),
579 );
580 }),
581 ),
582 };
583 });
584 });
585
586/** A child execution finished: its parent replays and reads the child's result. */
587const resumeParent = (registration: Registration, parent: ParentRef) =>
588 Effect.gen(function* () {
589 const engine = registration.engine();
590 yield* engine.resume({ _tag: parent.workflowName } as Workflow.Any, parent.executionId);
591 });
592
593// ---------------------------------------------------------------- Durable Object calls
594
595const DefectTag = "alchemy/Cloudflare/Workflows/DurableObjectEngine/Defect";
596
597/**
598 * Workers RPC replaces thrown errors with an opaque "internal error". Return
599 * failures as data instead, and rethrow them as defects on the calling side.
600 */
601const reportDefects = <A, E, R>(effect: Effect.Effect<A, E, R>) =>
602 effect.pipe(
603 Effect.catchCause((cause) =>
604 Effect.succeed({ [DefectTag]: Cause.pretty(cause) } as unknown as A),
605 ),
606 );
607
608/**
609 * Call an execution's Durable Object. Every method is idempotent, so a call
610 * that fails in transit (an instance resetting, or still running the previous
611 * script during a deploy) is retried for about a minute.
612 */
613const call = <A = void>(effect: Effect.Effect<unknown, unknown>): Effect.Effect<A> =>
614 effect.pipe(
615 Effect.catchDefect((defect) => Effect.fail(defect)),
616 Effect.retry({
617 schedule: Schedule.min([Schedule.exponential("100 millis"), Schedule.spaced("2 seconds")]),
618 times: 35,
619 }),
620 Effect.orDie,
621 Effect.flatMap((value) =>
622 value !== null && typeof value === "object" && DefectTag in value
623 ? Effect.die(new Error(String((value as Record<string, unknown>)[DefectTag])))
624 : Effect.succeed(value as A),
625 ),
626 );
627
628interface Stub {
629 readonly [method: string]: (...args: ReadonlyArray<unknown>) => Effect.Effect<unknown, unknown>;
630}
631
632interface Namespace {
633 readonly getByName: (name: string) => Stub;
634 readonly jurisdiction: (jurisdiction: DurableObjectJurisdiction) => Namespace;
635}
636
637type ObjectImpl = Effect.Effect<
638 Effect.Effect<Record<string, unknown>, never, DurableObjectState | RuntimeContext | Scope.Scope>,
639 never,
640 DurableObjectState
641>;
642
643/**
644 * Declare a Durable Object class on the host Worker. The inline form of
645 * `Cloudflare.DurableObject` is typed for single-stage implementations, so it
646 * is re-typed here for the two-stage form it runs.
647 */
648/**
649 * The class name a workflow's runs are stored under: its `StorageName`, which
650 * may be dotted like a Domain model's `Billing.OpenAccount`, or else its name.
651 */
652const workflowClassName = (workflow: Workflow.Any) => {
653 const storageName = Context.getOrUndefined(workflow.annotations, StorageName);
654 return storageName === undefined
655 ? workflow._tag.replace(/[^A-Za-z0-9_]/g, "_")
656 : storageName.replace(/[^A-Za-z0-9_.]/g, "_");
657};
658
659const declare = CloudflareDurableObject as unknown as (
660 name: string,
661 impl: ObjectImpl,
662) => Effect.Effect<Namespace, never, Worker>;