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 /** Recovery delay after an unsuccessful or interrupted delivery. Defaults to 30 seconds. */
16 readonly retry?: {
17 /** Must be a finite, positive duration. */
18 readonly delay: Duration.Input;
19 };
20}
21
22export type CallbackScheduleOptions<Payload> = {
23 /** JSON-serializable data delivered to the callback. */
24 readonly payload: Payload;
25} & (
26 | {
27 /** Absolute delivery time, as a Date or positive milliseconds since the Unix epoch. */
28 readonly at: Date | number;
29 readonly after?: never;
30 }
31 | {
32 /** Minimum delay before delivery. Zero schedules the callback as soon as possible. */
33 readonly after: Duration.Input;
34 readonly at?: never;
35 }
36);
37
38export interface Callback<Payload> {
39 /** Schedule or replace a pending job identified by this callback's name and the supplied ID. */
40 readonly schedule: (
41 id: string,
42 options: CallbackScheduleOptions<Payload>,
43 ) => Effect.Effect<void, CallbackError, RuntimeContext>;
44 /** Cancel a pending job. Cancelling an absent ID succeeds; an already-running handler is not interrupted. */
45 readonly cancel: (
46 id: string,
47 ) => Effect.Effect<void, CallbackError, RuntimeContext>;
48}
49
50/**
51 * Host-provided durable callback registration and scheduling.
52 *
53 * Implementations scope callback names and job IDs to a durable owner, persist
54 * jobs and recovery wakes, and acknowledge successful handlers without deleting
55 * same-ID replacements. Each invocation supplies a fresh Scope. Scheduling joins
56 * a storage transaction only when the host supports that transaction boundary.
57 */
58export type CallbackFactory = <Payload, E, R>(
59 name: string,
60 handler: (payload: Payload) => Effect.Effect<unknown, E, R>,
61 options?: CallbackOptions,
62) => Effect.Effect<
63 Callback<Payload>,
64 never,
65 RuntimeContext | Exclude<R, Scope.Scope>
66>;
67
68/**
69 * Register a durable callback with the current host.
70 *
71 * Cloudflare Durable Objects supply callback registration on their per-instance
72 * RuntimeContext. Register handlers in their inner Effect. Other hosts must
73 * implement RuntimeContext.makeCallback; unsupported hosts reject registration.
74 * Delivery is at least once: external writes must be idempotent. This API does
75 * not make unrelated external operations atomic.
76 *
77 * ### Registering and Scheduling a Callback
78 * **Example:** Schedule typed work from a Durable Object instance
79 * ```typescript
80 * return Effect.gen(function* () {
81 * const onArchive = yield* Alchemy.makeCallback(
82 * "archive",
83 * Effect.fn(function* (payload: { key: string; body: string }) {
84 * yield* archive.put(payload.key, payload.body);
85 * }),
86 * );
87 * return {
88 * save: Effect.fn(function* (id: string, body: string) {
89 * yield* state.storage.transaction(
90 * Effect.gen(function* () {
91 * yield* state.storage.put(id, body);
92 * yield* onArchive.schedule(id, {
93 * after: "30 seconds",
94 * payload: { key: id, body },
95 * });
96 * }),
97 * );
98 * }),
99 * };
100 * });
101 * ```
102 *
103 * ### Cancelling or Replacing a Job
104 * **Example:** Use a stable ID within a callback
105 * ```typescript
106 * yield* onArchive.schedule("revision-42", {
107 * at: new Date("2026-10-01T09:00:00Z"),
108 * payload: { key: "42.txt", body: "hello" },
109 * });
110 * yield* onArchive.cancel("revision-42");
111 * ```
112 *
113 * Callback names identify persisted jobs; retain handlers for old names while
114 * jobs are pending. Payloads must be JSON values compatible with pending jobs
115 * from earlier deployments; TypeScript types do not perform runtime decoding.
116 * Cloudflare commits scheduling alongside SQLite, KV, and native alarm writes
117 * inside the same Durable Object's storage transaction. Automatic recovery from
118 * instance termination relies on Cloudflare's native alarm retries. Calling
119 * `state.abort` with `{ retryAlarm: false }` removes that recovery guarantee:
120 * pending jobs remain stored, but may need an explicitly rearmed native alarm.
121 * Other implementations must document their own transaction integration.
122 */
123export const makeCallback = <Payload, E, R>(
124 name: string,
125 handler: (payload: Payload) => Effect.Effect<unknown, E, R>,
126 options?: CallbackOptions,
127): Effect.Effect<
128 Callback<Payload>,
129 never,
130 RuntimeContext | Exclude<R, Scope.Scope>
131> =>
132 Effect.gen(function* () {
133 const context = yield* RuntimeContext;
134 if (!context.makeCallback) {
135 return yield* Effect.die(
136 new CallbackError({
137 callback: name,
138 message: `Durable callbacks are not supported by ${context.Type}`,
139 }),
140 );
141 }
142 return yield* context.makeCallback(name, handler, options);
143 });