1import * as Data from "effect/Data";
2import type * as Duration from "effect/Duration";
3import * as Effect from "effect/Effect";
4import type * as Scope from "effect/Scope";
5import { RuntimeContext } from "./RuntimeContext.ts";
6
7/** A durable callback could not be registered, scheduled, cancelled, or dispatched. */
8export class CallbackError extends Data.TaggedError("CallbackError")<{
9 readonly callback: string;
10 readonly message: string;
11 readonly cause?: unknown;
12}> {}
13
14export interface CallbackOptions {
15 /** Bounded backoff for unsuccessful deliveries and self-rescheduling without progress. */
16 readonly retry?: {
17 /** Initial delay. Defaults to 30 seconds; positive durations below one second use one second. */
18 readonly delay?: Duration.Input;
19 /** Attempts without progress before parking. Must be a positive safe integer. Defaults to eight. */
20 readonly maxAttempts?: number;
21 /** Recovery interval for parked work. Must be at least one hour and the initial delay. Defaults to the greater of those two values. */
22 readonly parkedDelay?: Duration.Input;
23 };
24}
25
26export type CallbackScheduleOptions<Payload> = {
27 /** JSON-serializable data delivered to the callback. */
28 readonly payload: Payload;
29 /**
30 * Committed source version or sequence, as a nonnegative safe integer.
31 * A strictly increasing value resets the retry budget. Replayed external
32 * schedules cannot replace newer work. Commit progress and scheduling in the
33 * same storage transaction; changing IDs, payloads, or deadlines is not progress.
34 */
35 readonly progress?: number;
36} & (
37 | {
38 /** Absolute delivery time, as a Date or positive milliseconds since the Unix epoch. */
39 readonly at: Date | number;
40 readonly after?: never;
41 }
42 | {
43 /** Minimum delay before delivery. Zero schedules the callback as soon as possible. */
44 readonly after: Duration.Input;
45 readonly at?: never;
46 }
47);
48
49/** Durable scheduling state, available without decoding the job's payload. */
50export interface CallbackStatus {
51 /** Requested delivery time in milliseconds since the Unix epoch; preserved across retries. */
52 readonly scheduledAt: number;
53 /** Earliest recovery time in milliseconds since the Unix epoch, if an attempt has started. */
54 readonly retryAt: number | undefined;
55 /** Attempts started without reported progress, capped when the job is parked. */
56 readonly attempts: number;
57 /** Whether the job is retained for recovery at the parked interval. */
58 readonly parked: boolean;
59 /** Latest committed source cursor supplied when scheduling, if any. */
60 readonly progress: number | undefined;
61}
62
63export interface Callback<Payload> {
64 /** Schedule or replace a pending job identified by this callback's name and the supplied ID. */
65 readonly schedule: (
66 id: string,
67 options: CallbackScheduleOptions<Payload>,
68 ) => Effect.Effect<void, CallbackError, RuntimeContext>;
69 /** Cancel a pending job. Cancelling an absent ID succeeds; an already-running handler is not interrupted. */
70 readonly cancel: (id: string) => Effect.Effect<void, CallbackError, RuntimeContext>;
71 /** Inspect a pending job. Returns undefined after successful completion or cancellation. */
72 readonly getStatus: (
73 id: string,
74 ) => Effect.Effect<CallbackStatus | undefined, CallbackError, RuntimeContext>;
75}
76
77/**
78 * Host-provided durable callback registration and scheduling.
79 *
80 * Implementations scope callback names and job IDs to a durable owner, persist
81 * jobs and recovery wakes, and acknowledge successful handlers without deleting
82 * same-ID replacements. Each invocation supplies a fresh Scope. Scheduling joins
83 * a storage transaction only when the host supports that transaction boundary.
84 */
85export type CallbackFactory = <Payload, E, R>(
86 name: string,
87 handler: (payload: Payload) => Effect.Effect<unknown, E, R>,
88 options?: CallbackOptions,
89) => Effect.Effect<Callback<Payload>, never, RuntimeContext | Exclude<R, Scope.Scope>>;
90
91/**
92 * Register a durable callback with the current host.
93 *
94 * Cloudflare Durable Objects supply callback registration on their per-instance
95 * RuntimeContext. Register handlers in their inner Effect. Other hosts must
96 * implement RuntimeContext.makeCallback; unsupported hosts reject registration.
97 * Delivery is at least once: external writes must be idempotent. This API does
98 * not make unrelated external operations atomic.
99 *
100 * ### Registering and Scheduling a Callback
101 * **Example:** Schedule typed work from a Durable Object instance
102 * ```typescript
103 * return Effect.gen(function* () {
104 * const onArchive = yield* Alchemy.makeCallback(
105 * "archive",
106 * Effect.fn(function* (payload: { key: string; body: string }) {
107 * yield* archive.put(payload.key, payload.body);
108 * }),
109 * );
110 * return {
111 * save: Effect.fn(function* (id: string, body: string) {
112 * yield* state.storage.transaction(
113 * Effect.gen(function* () {
114 * yield* state.storage.put(id, body);
115 * yield* onArchive.schedule(id, {
116 * after: "30 seconds",
117 * payload: { key: id, body },
118 * });
119 * }),
120 * );
121 * }),
122 * };
123 * });
124 * ```
125 *
126 * ### Cancelling or Replacing a Job
127 * **Example:** Use a stable ID within a callback
128 * ```typescript
129 * yield* onArchive.schedule("revision-42", {
130 * at: new Date("2026-10-01T09:00:00Z"),
131 * payload: { key: "42.txt", body: "hello" },
132 * });
133 * yield* onArchive.cancel("revision-42");
134 * ```
135 *
136 * ### Inspecting and Recovering Work
137 * **Example:** Resume a parked job from an incoming recovery request
138 * ```typescript
139 * const status = yield* onArchive.getStatus("revision-42");
140 * if (status?.parked) {
141 * yield* onArchive.schedule("revision-42", {
142 * after: 0,
143 * payload: { key: "42.txt", body: "hello" },
144 * });
145 * }
146 * ```
147 *
148 * Cloudflare persists an exponential retry budget before each attempt. The
149 * default initial delay is 30 seconds, with a one-second minimum. After eight
150 * attempts without progress, jobs remain stored and recover hourly. Configure
151 * `retry.delay`, `retry.maxAttempts`, and `retry.parkedDelay` when registering.
152 * Status preserves the requested `scheduledAt` separately from `retryAt`.
153 *
154 * Scheduling inside a callback inherits its budget, including when cancelling
155 * or changing IDs. Supply a strictly increasing `progress` cursor alongside
156 * committed application state to reset it. An external schedule without a
157 * cursor starts a fresh budget; stale external cursors leave existing work
158 * unchanged. Detached callback fibers cannot schedule or cancel after the
159 * invocation ends. Successful completion and cancellation remove the job.
160 *
161 * Callback names identify persisted jobs; retain handlers for old names while
162 * jobs are pending. Payloads must be JSON values compatible with pending jobs
163 * from earlier deployments; TypeScript types do not perform runtime decoding.
164 * Cloudflare commits scheduling alongside SQLite, KV, and native alarm writes
165 * inside the same Durable Object's storage transaction. Automatic recovery from
166 * instance termination relies on Cloudflare's native alarm retries. Calling
167 * `state.abort` with `{ retryAlarm: false }` removes that recovery guarantee:
168 * pending jobs remain stored, but may need an explicitly rearmed native alarm.
169 * Other implementations must document their own transaction integration.
170 */
171export const makeCallback = <Payload, E, R>(
172 name: string,
173 handler: (payload: Payload) => Effect.Effect<unknown, E, R>,
174 options?: CallbackOptions,
175): Effect.Effect<Callback<Payload>, never, RuntimeContext | Exclude<R, Scope.Scope>> =>
176 Effect.gen(function* () {
177 const context = yield* RuntimeContext;
178 if (!context.makeCallback) {
179 return yield* Effect.die(
180 new CallbackError({
181 callback: name,
182 message: `Durable callbacks are not supported by ${context.Type}`,
183 }),
184 );
185 }
186 return yield* context.makeCallback(name, handler, options);
187 });