1// Alchemy modifications are licensed under Apache-2.0.
2// This file includes third-party code; see /THIRD_PARTY_LICENSES.md.
3// Alchemy modifications: uses Effect Schema instead of Zod, preserving strict option validation.
4import { RpcTarget } from "cloudflare:workers";
5import { ms } from "itty-time";
6import * as Schema from "effect/Schema";
7import * as Result from "effect/Result";
8import type { ResolvedStepConfig } from "./context.ts";
9
10const WorkflowSubscriptionEventCommonSchema = Schema.Struct({
11 instanceId: Schema.String,
12 eventId: Schema.Number.check(Schema.isFinite()),
13 timestamp: Schema.Number.check(Schema.isFinite()),
14});
15const StepDurationSchema = Schema.declare<ResolvedStepConfig["timeout"]>(
16 (value): value is ResolvedStepConfig["timeout"] =>
17 typeof value === "number" ||
18 (typeof value === "string" && !Number.isNaN(ms(value))),
19);
20const ResolvedStepDelaySchema = Schema.Union([
21 StepDurationSchema,
22 Schema.Literal("[dynamic]"),
23]);
24const ResolvedStepConfigSchema = Schema.Struct({
25 retries: Schema.Struct({
26 limit: Schema.declare<number>(
27 (value): value is number =>
28 typeof value === "number" &&
29 (Number.isFinite(value) || value === Infinity),
30 ),
31 delay: ResolvedStepDelaySchema,
32 backoff: Schema.optional(
33 Schema.Literals(["constant", "linear", "exponential"]),
34 ),
35 }),
36 timeout: StepDurationSchema,
37 sensitive: Schema.optional(Schema.Literal("output")),
38});
39
40/**
41 * Parses a resolved step config read from persisted local Workflow state.
42 *
43 * @param value Persisted value to parse.
44 * @returns The resolved config, or `undefined` when the value is invalid.
45 */
46export function parseResolvedStepConfig(value: unknown) {
47 const result = Schema.decodeUnknownResult(ResolvedStepConfigSchema)(value);
48 return Result.isSuccess(result) ? result.success : undefined;
49}
50const WorkflowSubscriptionEventSchema = Schema.Union([
51 Schema.Struct({
52 ...WorkflowSubscriptionEventCommonSchema.fields,
53 type: Schema.Literal("workflow_queued"),
54 }),
55 Schema.Struct({
56 ...WorkflowSubscriptionEventCommonSchema.fields,
57 type: Schema.Literal("workflow_started"),
58 params: Schema.optional(Schema.Unknown),
59 }),
60 Schema.Struct({
61 ...WorkflowSubscriptionEventCommonSchema.fields,
62 type: Schema.Literal("workflow_running"),
63 }),
64 Schema.Struct({
65 ...WorkflowSubscriptionEventCommonSchema.fields,
66 type: Schema.Literal("workflow_paused"),
67 }),
68 Schema.Struct({
69 ...WorkflowSubscriptionEventCommonSchema.fields,
70 type: Schema.Literal("workflow_waiting_for_pause"),
71 }),
72 Schema.Struct({
73 ...WorkflowSubscriptionEventCommonSchema.fields,
74 type: Schema.Literal("workflow_waiting"),
75 }),
76 Schema.Struct({
77 ...WorkflowSubscriptionEventCommonSchema.fields,
78 type: Schema.Literal("workflow_completed"),
79 output: Schema.optional(Schema.Unknown),
80 }),
81 Schema.Struct({
82 ...WorkflowSubscriptionEventCommonSchema.fields,
83 type: Schema.Literal("workflow_errored"),
84 error: Schema.Struct({ name: Schema.String, message: Schema.String }),
85 }),
86 Schema.Struct({
87 ...WorkflowSubscriptionEventCommonSchema.fields,
88 type: Schema.Literal("workflow_terminated"),
89 }),
90 Schema.Struct({
91 ...WorkflowSubscriptionEventCommonSchema.fields,
92 type: Schema.Literal("step_started"),
93 stepName: Schema.String,
94 config: Schema.optional(ResolvedStepConfigSchema),
95 }),
96 Schema.Struct({
97 ...WorkflowSubscriptionEventCommonSchema.fields,
98 type: Schema.Literal("step_completed"),
99 stepName: Schema.String,
100 output: Schema.optional(Schema.Unknown),
101 }),
102 Schema.Struct({
103 ...WorkflowSubscriptionEventCommonSchema.fields,
104 type: Schema.Literal("step_errored"),
105 stepName: Schema.String,
106 }),
107 Schema.Struct({
108 ...WorkflowSubscriptionEventCommonSchema.fields,
109 type: Schema.Literal("attempt_started"),
110 stepName: Schema.String,
111 attempt: Schema.Number.check(Schema.isFinite()),
112 }),
113 Schema.Struct({
114 ...WorkflowSubscriptionEventCommonSchema.fields,
115 type: Schema.Literal("attempt_completed"),
116 stepName: Schema.String,
117 attempt: Schema.Number.check(Schema.isFinite()),
118 }),
119 Schema.Struct({
120 ...WorkflowSubscriptionEventCommonSchema.fields,
121 type: Schema.Literal("attempt_errored"),
122 stepName: Schema.String,
123 attempt: Schema.Number.check(Schema.isFinite()),
124 retryDelayMs: Schema.optional(Schema.Number.check(Schema.isFinite())),
125 error: Schema.Struct({ name: Schema.String, message: Schema.String }),
126 }),
127 Schema.Struct({
128 ...WorkflowSubscriptionEventCommonSchema.fields,
129 type: Schema.Literal("sleep_started"),
130 stepName: Schema.String,
131 durationMs: Schema.Number.check(Schema.isFinite()),
132 }),
133 Schema.Struct({
134 ...WorkflowSubscriptionEventCommonSchema.fields,
135 type: Schema.Literal("sleep_completed"),
136 stepName: Schema.String,
137 }),
138 Schema.Struct({
139 ...WorkflowSubscriptionEventCommonSchema.fields,
140 type: Schema.Literal("wait_started"),
141 stepName: Schema.String,
142 eventType: Schema.String,
143 }),
144 Schema.Struct({
145 ...WorkflowSubscriptionEventCommonSchema.fields,
146 type: Schema.Literal("wait_completed"),
147 stepName: Schema.String,
148 }),
149 Schema.Struct({
150 ...WorkflowSubscriptionEventCommonSchema.fields,
151 type: Schema.Literal("wait_timed_out"),
152 stepName: Schema.String,
153 }),
154 Schema.Struct({
155 ...WorkflowSubscriptionEventCommonSchema.fields,
156 type: Schema.Literal("rollback_started"),
157 }),
158 Schema.Struct({
159 ...WorkflowSubscriptionEventCommonSchema.fields,
160 type: Schema.Literal("rollback_step_started"),
161 stepName: Schema.String,
162 config: Schema.optional(ResolvedStepConfigSchema),
163 }),
164 Schema.Struct({
165 ...WorkflowSubscriptionEventCommonSchema.fields,
166 type: Schema.Literal("rollback_step_completed"),
167 stepName: Schema.String,
168 }),
169 Schema.Struct({
170 ...WorkflowSubscriptionEventCommonSchema.fields,
171 type: Schema.Literal("rollback_step_errored"),
172 stepName: Schema.String,
173 error: Schema.Struct({ name: Schema.String, message: Schema.String }),
174 }),
175 Schema.Struct({
176 ...WorkflowSubscriptionEventCommonSchema.fields,
177 type: Schema.Literal("rollback_attempt_started"),
178 stepName: Schema.String,
179 attempt: Schema.Number.check(Schema.isFinite()),
180 }),
181 Schema.Struct({
182 ...WorkflowSubscriptionEventCommonSchema.fields,
183 type: Schema.Literal("rollback_attempt_completed"),
184 stepName: Schema.String,
185 attempt: Schema.Number.check(Schema.isFinite()),
186 }),
187 Schema.Struct({
188 ...WorkflowSubscriptionEventCommonSchema.fields,
189 type: Schema.Literal("rollback_attempt_errored"),
190 stepName: Schema.String,
191 attempt: Schema.Number.check(Schema.isFinite()),
192 retryDelayMs: Schema.optional(Schema.Number.check(Schema.isFinite())),
193 error: Schema.Struct({ name: Schema.String, message: Schema.String }),
194 }),
195 Schema.Struct({
196 ...WorkflowSubscriptionEventCommonSchema.fields,
197 type: Schema.Literal("rollback_completed"),
198 }),
199 Schema.Struct({
200 ...WorkflowSubscriptionEventCommonSchema.fields,
201 type: Schema.Literal("rollback_errored"),
202 }),
203]);
204
205export type WorkflowSubscriptionEvent =
206 typeof WorkflowSubscriptionEventSchema.Type;
207
208const workflowSubscriptionEventTypeNames = new Set(
209 WorkflowSubscriptionEventSchema.members.map(
210 (option) => option.fields.type.literal,
211 ),
212);
213const WorkflowSubscriptionEventTypeSchema = Schema.declare<
214 WorkflowSubscriptionEvent["type"]
215>(
216 (value): value is WorkflowSubscriptionEvent["type"] =>
217 typeof value === "string" &&
218 workflowSubscriptionEventTypeNames.has(
219 value as WorkflowSubscriptionEvent["type"],
220 ),
221);
222
223export type WorkflowSubscriptionOptions = {
224 cursor?: number;
225 filter?: Array<WorkflowSubscriptionEvent["type"]>;
226};
227
228const WORKFLOW_SUBSCRIPTION_OPTIONS_SCHEMA = Schema.Struct({
229 cursor: Schema.optional(
230 Schema.Number.check(
231 Schema.isFinite(),
232 Schema.isInt(),
233 Schema.isGreaterThanOrEqualTo(0),
234 ),
235 ),
236 filter: Schema.optional(
237 Schema.mutable(Schema.Array(WorkflowSubscriptionEventTypeSchema)),
238 ),
239});
240
241export function parseWorkflowSubscriptionOptions(options: unknown) {
242 if (options === undefined) {
243 return {} satisfies WorkflowSubscriptionOptions;
244 }
245
246 const parsed = Schema.decodeUnknownResult(
247 WORKFLOW_SUBSCRIPTION_OPTIONS_SCHEMA,
248 )(options, { onExcessProperty: "error" });
249 if (Result.isFailure(parsed)) {
250 throw new Error("Invalid Workflow subscription options");
251 }
252
253 return parsed.success satisfies WorkflowSubscriptionOptions;
254}
255
256export type WorkflowSubscriptionState = {
257 instanceId: string;
258 params: unknown;
259 lastEventId: number;
260 filter: ReadonlySet<WorkflowSubscriptionEvent["type"]> | undefined;
261 waiter: { resolve: () => void } | undefined;
262 closed: boolean;
263};
264
265export interface WorkflowSubscription extends Disposable {
266 next(): Promise<IteratorResult<WorkflowSubscriptionEvent, undefined>>;
267}
268
269export function isTerminalEvent(event: WorkflowSubscriptionEvent): boolean {
270 return (
271 event.type === "workflow_completed" ||
272 event.type === "workflow_errored" ||
273 event.type === "workflow_terminated"
274 );
275}
276
277export class WorkflowSubscriptionTarget
278 extends RpcTarget
279 implements WorkflowSubscription
280{
281 readonly #nextEvent: () => Promise<
282 IteratorResult<WorkflowSubscriptionEvent, undefined>
283 >;
284 readonly #onClose: () => void;
285 #nextRequest = Promise.resolve<unknown>(undefined);
286 #closed = false;
287
288 constructor(
289 nextEvent: () => Promise<
290 IteratorResult<WorkflowSubscriptionEvent, undefined>
291 >,
292 onClose: () => void,
293 ) {
294 super();
295 this.#nextEvent = nextEvent;
296 this.#onClose = onClose;
297 }
298
299 async next(): Promise<IteratorResult<WorkflowSubscriptionEvent, undefined>> {
300 if (this.#closed) {
301 return { done: true, value: undefined };
302 }
303
304 const request = this.#nextRequest.then(async () => {
305 if (this.#closed) {
306 return { done: true, value: undefined } as const;
307 }
308
309 try {
310 const result = await this.#nextEvent();
311 if (this.#closed) {
312 return { done: true, value: undefined } as const;
313 }
314 if (result.done || isTerminalEvent(result.value)) {
315 this.#finish();
316 }
317 return result;
318 } catch (error) {
319 this.#finish();
320 throw error;
321 }
322 });
323 this.#nextRequest = request;
324 return request;
325 }
326
327 [Symbol.dispose](): void {
328 this.#finish();
329 }
330
331 #finish(): void {
332 if (this.#closed) {
333 return;
334 }
335 this.#closed = true;
336 this.#onClose();
337 }
338}