1/**
2 * Internal scaffolding for AWS Lambda Durable Functions — NOT exported from
3 * the Lambda service `index.ts`.
4 *
5 * A function created with `DurableConfig` receives every invocation as a
6 * dedicated durable-execution envelope (`DurableExecutionInvocationInput`):
7 * the customer payload rides inside the checkpoint log's EXECUTION operation,
8 * and the handler must speak the checkpoint/replay protocol
9 * (`CheckpointDurableExecution` / `GetDurableExecutionState`) and return a
10 * `SUCCEEDED | FAILED | PENDING` envelope.
11 *
12 * Alchemy owns the Lambda entrypoint, so this protocol is routed exactly like
13 * an event source: {@link isDurableExecutionEvent} is the shape predicate
14 * (the `isSQSEvent` analogue) and {@link makeDurableListener} produces the
15 * `host.listen` listener that drives the orchestrator. The replay engine
16 * itself is AWS's open-source `@aws/durable-execution-sdk-js`
17 * (`withDurableExecution`), wrapped behind the Effect-native `DurableStep`
18 * service — the same posture as the Cloudflare `WorkflowBridge` wrapping the
19 * native `step` object.
20 *
21 * The SDK module is loaded with a dynamic `import()` so it never loads at
22 * plan/deploy time. The `DurableFunction` wrapper merges it into
23 * `build.install` automatically, which both vendors it into the artifact's
24 * `node_modules` and externalizes it from the bundle (install roots are
25 * always external) — plain Functions never resolve it.
26 */
27import type * as lambda from "aws-lambda";
28import * as Cause from "effect/Cause";
29import * as Effect from "effect/Effect";
30import * as Exit from "effect/Exit";
31import * as Layer from "effect/Layer";
32import * as Scope from "effect/Scope";
33import { isScopeEjected } from "../../Http.ts";
34import type * as Serverless from "../../Serverless/index.ts";
35import { toSeconds } from "../../Util/Duration.ts";
36import {
37 DurableExecutionContext,
38 DurableStep,
39 type DurableCallbackOptions,
40 type DurableRetryPolicy,
41 type DurableStepOptions,
42} from "./Durable.ts";
43import { HandlerContext } from "./Function.ts";
44
45/** Module id of AWS's Durable Execution SDK (an optional peer dependency). */
46export const DURABLE_SDK_MODULE = "@aws/durable-execution-sdk-js";
47
48// ---------------------------------------------------------------------------
49// The durable invocation envelope
50// ---------------------------------------------------------------------------
51
52/**
53 * Raw invocation payload delivered to a `DurableConfig`-enabled function.
54 * Mirrors the SDK's `DurableExecutionInvocationInput`.
55 */
56export interface DurableExecutionInvocationEvent {
57 DurableExecutionArn: string;
58 CheckpointToken: string;
59 UpdatedOperationIds?: string[];
60 InitialExecutionState: {
61 Operations?: {
62 Type?: string;
63 ExecutionDetails?: { InputPayload?: string };
64 }[];
65 NextMarker?: string;
66 };
67}
68
69/**
70 * Shape predicate for durable-execution invocations — the durable analogue of
71 * `isSQSEvent`. A durable function's invocations always arrive in this
72 * envelope (the durable execution wraps even the first invocation).
73 */
74export const isDurableExecutionEvent = (
75 event: any,
76): event is DurableExecutionInvocationEvent =>
77 typeof event === "object" &&
78 event !== null &&
79 typeof event.DurableExecutionArn === "string" &&
80 typeof event.CheckpointToken === "string" &&
81 typeof event.InitialExecutionState === "object" &&
82 event.InitialExecutionState !== null;
83
84// ---------------------------------------------------------------------------
85// The alchemy payload envelope (multi-workflow routing)
86// ---------------------------------------------------------------------------
87
88/**
89 * Lambda has a single exported handler (unlike workerd's named class
90 * exports), so when several DurableFunctions share one host the start payload
91 * carries a discriminator. Alchemy owns both ends — the `start` client wraps
92 * the params, the listener unwraps them — so user code never sees this.
93 */
94export interface DurableEnvelope {
95 $alchemy: { workflow: string };
96 params: unknown;
97}
98
99export const encodeDurableEnvelope = (
100 workflow: string,
101 params: unknown,
102): string =>
103 JSON.stringify({ $alchemy: { workflow }, params } satisfies DurableEnvelope);
104
105const asDurableEnvelope = (payload: unknown): DurableEnvelope | undefined =>
106 typeof payload === "object" &&
107 payload !== null &&
108 "$alchemy" in payload &&
109 typeof (payload as DurableEnvelope).$alchemy === "object" &&
110 (payload as DurableEnvelope).$alchemy !== null &&
111 typeof (payload as DurableEnvelope).$alchemy.workflow === "string"
112 ? (payload as DurableEnvelope)
113 : undefined;
114
115/**
116 * Extract the customer input payload from the checkpoint log's EXECUTION
117 * operation (first page — the EXECUTION operation is always the log's first
118 * entry, so pagination never hides it).
119 */
120export const readDurableInputPayload = (
121 event: DurableExecutionInvocationEvent,
122): unknown => {
123 const operations = event.InitialExecutionState?.Operations ?? [];
124 const execution = operations.find((op) => op?.Type === "EXECUTION");
125 const raw = execution?.ExecutionDetails?.InputPayload;
126 if (raw === undefined) return undefined;
127 try {
128 return JSON.parse(raw);
129 } catch {
130 return raw;
131 }
132};
133
134// ---------------------------------------------------------------------------
135// Native SDK surface (structural types only — the SDK's own types never leak
136// into alchemy's public d.ts, keeping the dependency optional)
137// ---------------------------------------------------------------------------
138
139interface NativeRetryDecision {
140 shouldRetry: boolean;
141 delay?: { seconds: number };
142}
143
144interface NativeStepConfig {
145 retryStrategy?: (error: Error, attemptCount: number) => NativeRetryDecision;
146 semantics?: string;
147}
148
149interface NativeDurableContext {
150 lambdaContext: lambda.Context;
151 executionContext: { readonly durableExecutionArn: string };
152 step<T>(
153 name: string,
154 fn: (stepContext: unknown) => Promise<T>,
155 config?: NativeStepConfig,
156 ): Promise<T>;
157 wait(name: string, duration: { seconds: number }): Promise<void>;
158 waitForCallback<T>(
159 name: string,
160 submitter: (callbackId: string, context: unknown) => Promise<void>,
161 config?: {
162 timeout?: { seconds: number };
163 heartbeatTimeout?: { seconds: number };
164 },
165 ): Promise<T>;
166}
167
168interface NativeDurableSdk {
169 withDurableExecution: (
170 handler: (event: any, context: NativeDurableContext) => Promise<any>,
171 ) => (event: any, context: lambda.Context) => Promise<any>;
172}
173
174/**
175 * Load the Durable Execution SDK. The module id is an install root of every
176 * DurableFunction bundle (the wrapper merges it into `build.install`), so
177 * resolution happens at runtime inside the sandbox — from the artifact's
178 * vendored `node_modules`, otherwise from the managed runtime's bundled
179 * copy.
180 */
181const loadDurableSdk: Effect.Effect<NativeDurableSdk> = Effect.promise(
182 () => import(DURABLE_SDK_MODULE) as Promise<NativeDurableSdk>,
183).pipe(
184 Effect.catchDefect((defect) =>
185 Effect.die(
186 new Error(
187 `Failed to load "${DURABLE_SDK_MODULE}" — install it in the project ` +
188 `that defines the DurableFunction (npm i ${DURABLE_SDK_MODULE}) so ` +
189 `the bundler can vendor it into the artifact.`,
190 { cause: defect },
191 ),
192 ),
193 ),
194);
195
196// ---------------------------------------------------------------------------
197// DurableStep over the native DurableContext
198// ---------------------------------------------------------------------------
199
200const toNativeRetryStrategy = (
201 retry: DurableRetryPolicy,
202): ((error: Error, attemptCount: number) => NativeRetryDecision) => {
203 const base = toSeconds(retry.delay) ?? 1;
204 const cap = toSeconds(retry.maxDelay);
205 const backoff = retry.backoff ?? "exponential";
206 return (_error, attemptCount) => {
207 if (attemptCount > retry.limit) {
208 return { shouldRetry: false };
209 }
210 const raw =
211 backoff === "constant"
212 ? base
213 : backoff === "linear"
214 ? base * attemptCount
215 : base * 2 ** Math.max(0, attemptCount - 1);
216 const seconds = Math.max(1, Math.ceil(cap ? Math.min(raw, cap) : raw));
217 return { shouldRetry: true, delay: { seconds } };
218 };
219};
220
221const toNativeStepConfig = (
222 options: DurableStepOptions<any>,
223): NativeStepConfig | undefined => {
224 if (!options.retry && !options.semantics) return undefined;
225 return {
226 ...(options.retry
227 ? { retryStrategy: toNativeRetryStrategy(options.retry) }
228 : {}),
229 ...(options.semantics
230 ? {
231 semantics:
232 options.semantics === "at-most-once"
233 ? "AT_MOST_ONCE_PER_RETRY"
234 : "AT_LEAST_ONCE_PER_RETRY",
235 }
236 : {}),
237 };
238};
239
240const wrapDurableContext = (
241 dctx: NativeDurableContext,
242): DurableStep["Service"] => ({
243 step: <T>(options: DurableStepOptions<T>) =>
244 Effect.tryPromise(() =>
245 dctx.step<T>(
246 options.name,
247 () => Effect.runPromise(options.effect),
248 toNativeStepConfig(options),
249 ),
250 ).pipe(Effect.orDie),
251 wait: (name, duration) =>
252 Effect.tryPromise(() =>
253 dctx.wait(name, { seconds: Math.max(1, toSeconds(duration) ?? 1) }),
254 ).pipe(Effect.orDie),
255 waitForCallback: <T>(options: DurableCallbackOptions) =>
256 Effect.tryPromise(() =>
257 dctx.waitForCallback<T>(
258 options.name,
259 (callbackId) => Effect.runPromise(options.submitter(callbackId)),
260 {
261 ...(options.timeout !== undefined
262 ? { timeout: { seconds: toSeconds(options.timeout)! } }
263 : {}),
264 ...(options.heartbeatTimeout !== undefined
265 ? {
266 heartbeatTimeout: {
267 seconds: toSeconds(options.heartbeatTimeout)!,
268 },
269 }
270 : {}),
271 },
272 ),
273 ).pipe(Effect.orDie),
274});
275
276// ---------------------------------------------------------------------------
277// The listener
278// ---------------------------------------------------------------------------
279
280/**
281 * Build the `host.listen` listener that drives one named durable orchestrator.
282 *
283 * The listener recognizes the durable envelope, peeks the alchemy payload
284 * envelope to decline invocations addressed to a different DurableFunction on
285 * the same host, and otherwise hands the raw event to the SDK's
286 * `withDurableExecution` wrapper. The wrapper loads/replays the checkpoint
287 * log and calls back into the Effect body with a fresh per-invocation `Scope`
288 * and the `DurableStep`/`DurableExecutionContext` services.
289 */
290type WrappedDurableHandler = (
291 event: any,
292 context: lambda.Context,
293) => Promise<any>;
294
295export const makeDurableListener = (options: {
296 name: string;
297 run: (input: unknown) => Effect.Effect<unknown>;
298}): Effect.Effect<Serverless.FunctionListener> =>
299 // `Effect.sync`, NOT `Effect.gen` yielding `loadDurableSdk`: the listener is
300 // CONSTRUCTED when the host resolves `runtimeContext.exports`, which happens
301 // at PLAN/DEPLOY time (`Platform.ts` does `yield* runtimeContext.exports`).
302 // Importing `@aws/durable-execution-sdk-js` there both contradicts the "never
303 // at plan time" contract and stalls the deploy (the SDK's module init
304 // touches the AWS client). Defer the load to the first real invocation and
305 // memoize it.
306 Effect.sync(() => {
307 // Bind the SDK-wrapped handler once, on first invocation, and reuse it.
308 let wrapped: WrappedDurableHandler | undefined;
309 const ensureWrapped: Effect.Effect<WrappedDurableHandler> = Effect.suspend(
310 () =>
311 wrapped !== undefined
312 ? Effect.succeed(wrapped)
313 : loadDurableSdk.pipe(
314 Effect.map((sdk) => {
315 wrapped = sdk.withDurableExecution(async (event, dctx) => {
316 // The SDK extracts the customer payload from the EXECUTION
317 // operation and hands it to us on every (re-)invocation.
318 const envelope = asDurableEnvelope(event);
319 const params =
320 envelope !== undefined ? envelope.params : event;
321
322 // Fresh request scope per durable invocation, matching the
323 // Lambda dispatcher / Worker / Workflow bridges.
324 const scope = Scope.makeUnsafe();
325 const exit = await Effect.runPromiseExit(
326 options.run(params).pipe(
327 Effect.provide(
328 Layer.mergeAll(
329 Layer.succeed(DurableStep, wrapDurableContext(dctx)),
330 Layer.succeed(DurableExecutionContext, {
331 executionArn:
332 dctx.executionContext.durableExecutionArn,
333 }),
334 Layer.succeed(HandlerContext, dctx.lambdaContext),
335 Layer.succeed(Scope.Scope, scope),
336 ),
337 ),
338 ),
339 );
340 if (!isScopeEjected(scope)) {
341 await Scope.close(scope, exit).pipe(
342 Effect.ignoreCause({
343 log: "Warn",
344 message: "Durable invocation scope close failed",
345 }),
346 Effect.runPromise,
347 );
348 }
349 if (Exit.isSuccess(exit)) {
350 return exit.value;
351 }
352 throw Cause.squash(exit.cause);
353 });
354 return wrapped;
355 }),
356 ),
357 );
358
359 // `HandlerContext` is a runtime-only requirement satisfied unconditionally
360 // by the Lambda dispatcher, which provides it (and `Scope`) to every
361 // listener's returned effect. Cast the listener down to the `Req = never`
362 // `FunctionListener` so the requirement never leaks into the host's init
363 // effect — the exact contract `serve`/`makeFunctionHttpHandler` rely on.
364 return ((event: any) => {
365 if (!isDurableExecutionEvent(event)) return;
366 const payload = readDurableInputPayload(event);
367 const envelope = asDurableEnvelope(payload);
368 if (
369 envelope !== undefined &&
370 envelope.$alchemy.workflow !== options.name
371 ) {
372 // Addressed to another DurableFunction on this host — decline so its
373 // listener picks it up.
374 return;
375 }
376 return Effect.gen(function* () {
377 const context = yield* HandlerContext;
378 // Lazily load + memoize the SDK on the first real invocation.
379 const run = yield* ensureWrapped;
380 // The SDK owns the checkpoint protocol from here: it replays the log,
381 // runs new work, and returns the SUCCEEDED/FAILED/PENDING envelope
382 // Lambda interprets. A rejection here is a protocol-level failure —
383 // let it surface as an invocation error.
384 return yield* Effect.promise(() => run(event, context));
385 });
386 }) as unknown as Serverless.FunctionListener;
387 });