1import type * as cf from "@cloudflare/workers-types";
2import * as Cause from "effect/Cause";
3import * as Clock from "effect/Clock";
4import * as Context from "effect/Context";
5import * as Duration from "effect/Duration";
6import * as Effect from "effect/Effect";
7import * as Exit from "effect/Exit";
8import * as Schema from "effect/Schema";
9import * as Scope from "effect/Scope";
10import * as Tracer from "effect/Tracer";
11import * as Logger from "effect/Logger";
12import { RuntimeContext } from "../../RuntimeContext.ts";
13import {
14 ensureAlarmTables,
15 reconcileDurableObjectAlarm,
16} from "./DurableObjectAlarmStorage.ts";
17import {
18 DurableObjectState,
19 fromDurableObjectState,
20} from "./DurableObjectState.ts";
21import { fromDurableObjectStorage } from "./DurableObjectStorage.ts";
22import { ActiveStorageTransactions } from "./DurableObjectTransactionContext.ts";
23import {
24 CallbackError,
25 type Callback,
26 type CallbackOptions,
27 type CallbackScheduleOptions,
28 type CallbackFactory,
29} from "../../Callback.ts";
30
31type InvocationServices = RuntimeContext | DurableObjectState | Scope.Scope;
32interface RegisteredCallback {
33 readonly retryDelay: number;
34 readonly handler: (
35 payload: unknown,
36 ) => Effect.Effect<unknown, unknown, InvocationServices>;
37}
38interface CallbackRegistry {
39 readonly callbacks: Map<string, RegisteredCallback>;
40 open: boolean;
41}
42type AlarmRow = {
43 callback: string;
44 id: string;
45 version: string;
46 run_at: number;
47 payload: string;
48};
49const registries = new WeakMap<cf.DurableObjectState, CallbackRegistry>();
50
51/** @internal */
52export const initializeAlarmCallbacks = (state: cf.DurableObjectState) => {
53 const registry: CallbackRegistry = { callbacks: new Map(), open: true };
54 registries.set(state, registry);
55 return () => {
56 registry.open = false;
57 };
58};
59
60/** @internal */
61export const makeDurableObjectCallbackFactory = (
62 raw: cf.DurableObjectState,
63): CallbackFactory => {
64 const state = fromDurableObjectState(raw);
65 return (name, handler, options) =>
66 makeAlarmCallback(state, name, handler, options);
67};
68
69const makeAlarmCallback = <Payload, E, R>(
70 state: DurableObjectState["Service"],
71 name: string,
72 handler: (payload: Payload) => Effect.Effect<unknown, E, R>,
73 options?: CallbackOptions,
74): Effect.Effect<
75 Callback<Payload>,
76 never,
77 RuntimeContext | Exclude<R, Scope.Scope>
78> =>
79 Effect.gen(function* () {
80 const context = (yield* Effect.context<Exclude<R, Scope.Scope>>()).pipe(
81 Context.omit(
82 ActiveStorageTransactions,
83 DurableObjectState,
84 RuntimeContext,
85 Scope.Scope,
86 Tracer.ParentSpan,
87 Tracer.Tracer,
88 Logger.CurrentLoggers,
89 ),
90 );
91 const registry = registries.get(state.raw);
92 if (!registry?.open || !name || registry.callbacks.has(name)) {
93 return yield* Effect.die(
94 new CallbackError({
95 callback: name,
96 message: !registry?.open
97 ? "Alarm callbacks must be registered during Durable Object instance initialization"
98 : "Alarm callback names must be non-empty and unique within an instance",
99 }),
100 );
101 }
102 const retryDelay = yield* Effect.sync(() =>
103 Duration.toMillis(options?.retry?.delay ?? "30 seconds"),
104 );
105 if (!Number.isFinite(retryDelay) || retryDelay <= 0) {
106 return yield* Effect.die(
107 new CallbackError({
108 callback: name,
109 message: "Alarm retry delay must be finite and positive",
110 }),
111 );
112 }
113 yield* Effect.sync(() =>
114 registry.callbacks.set(name, {
115 retryDelay,
116 handler: (payload) =>
117 Effect.gen(function* () {
118 const invocation = yield* Effect.context<InvocationServices>();
119 return yield* handler(payload as Payload).pipe(
120 Effect.provide(
121 Context.merge(invocation, context) as Context.Context<R>,
122 ),
123 );
124 }),
125 }),
126 );
127
128 const raw = state.raw.storage;
129 return {
130 schedule: Effect.fn(function* (
131 id: string,
132 schedule: CallbackScheduleOptions<Payload>,
133 ) {
134 const now = yield* Clock.currentTimeMillis;
135 const { at, payload } = yield* Effect.try({
136 try: () => {
137 if (!id) throw new Error("Alarm IDs must be non-empty");
138 if (
139 (schedule.at === undefined) ===
140 (schedule.after === undefined)
141 ) {
142 throw new Error("Specify exactly one of at or after");
143 }
144 const delay =
145 schedule.after === undefined
146 ? undefined
147 : Duration.toMillis(schedule.after);
148 const at =
149 schedule.at === undefined
150 ? now + delay!
151 : schedule.at instanceof Date
152 ? schedule.at.getTime()
153 : schedule.at;
154 if (
155 !Number.isFinite(at) ||
156 at <= 0 ||
157 (delay !== undefined && (!Number.isFinite(delay) || delay < 0))
158 ) {
159 throw new Error(
160 "Alarm time must be finite and delay must be non-negative",
161 );
162 }
163 if (!Schema.is(Schema.Json)(schedule.payload)) {
164 throw new Error("Alarm payload must be a JSON value");
165 }
166 const payload = JSON.stringify(schedule.payload);
167 if (payload === undefined)
168 throw new Error("Alarm payload must be JSON-serializable");
169 return { at, payload };
170 },
171 catch: (cause) =>
172 new CallbackError({
173 callback: name,
174 message: "Invalid alarm schedule",
175 cause,
176 }),
177 });
178 const version = yield* Effect.sync(() => crypto.randomUUID());
179 yield* state.storage
180 .transaction(
181 Effect.gen(function* () {
182 yield* ensureAlarmTables(raw);
183 yield* Effect.sync(() =>
184 raw.sql.exec(
185 `INSERT INTO alchemy_alarm_callbacks (callback, id, version, run_at, payload)
186 VALUES (?, ?, ?, ?, ?)
187 ON CONFLICT (callback, id) DO UPDATE SET
188 version = excluded.version, run_at = excluded.run_at, payload = excluded.payload`,
189 name,
190 id,
191 version,
192 at,
193 payload,
194 ),
195 );
196 yield* reconcileDurableObjectAlarm(raw);
197 }),
198 )
199 .pipe(
200 Effect.catchTag("DurableObjectStorageError", (cause) =>
201 Effect.fail(
202 new CallbackError({
203 callback: name,
204 message: "Callback storage transaction failed",
205 cause,
206 }),
207 ),
208 ),
209 );
210 }),
211 cancel: Effect.fn(function* (id: string) {
212 yield* state.storage
213 .transaction(
214 Effect.gen(function* () {
215 yield* ensureAlarmTables(raw);
216 yield* Effect.sync(() =>
217 raw.sql.exec(
218 "DELETE FROM alchemy_alarm_callbacks WHERE callback = ? AND id = ?",
219 name,
220 id,
221 ),
222 );
223 yield* reconcileDurableObjectAlarm(raw);
224 }),
225 )
226 .pipe(
227 Effect.catchTag("DurableObjectStorageError", (cause) =>
228 Effect.fail(
229 new CallbackError({
230 callback: name,
231 message: "Callback storage transaction failed",
232 cause,
233 }),
234 ),
235 ),
236 );
237 }),
238 };
239 });
240
241/** @internal */
242export const dispatchAlarmCallbacks = (
243 state: cf.DurableObjectState,
244 hasLegacyHandler: boolean,
245) =>
246 Effect.gen(function* () {
247 const registry = registries.get(state);
248 if (!registry) return;
249 const raw = state.storage;
250 if (registry.callbacks.size === 0) {
251 const hasJobs = yield* Effect.sync(
252 () =>
253 raw.sql
254 ?.exec(
255 "SELECT name FROM sqlite_master WHERE type = 'table' AND name IN ('alchemy_alarm_callbacks', 'alchemy_alarm_schema', 'alchemy_scheduled_events')",
256 )
257 .toArray().length > 0,
258 );
259 if (!hasJobs) return;
260 }
261 const storage = fromDurableObjectStorage(raw);
262 yield* ensureAlarmTables(raw);
263 if (!hasLegacyHandler) {
264 const legacy = yield* Effect.sync(() =>
265 raw.sql
266 .exec("SELECT id FROM alchemy_scheduled_events LIMIT 1")
267 .toArray(),
268 );
269 if (legacy.length > 0) {
270 return yield* Effect.fail(
271 new CallbackError({
272 callback: "scheduleEvent",
273 message:
274 "Pending legacy events require an alarm handler calling processScheduledEvents",
275 }),
276 );
277 }
278 }
279 const now = yield* Clock.currentTimeMillis;
280 const due = yield* Effect.sync(() =>
281 raw.sql
282 .exec<AlarmRow>(
283 `SELECT callback, id, version, run_at, payload FROM alchemy_alarm_callbacks
284 WHERE run_at <= ? ORDER BY run_at, callback, id LIMIT 100`,
285 now,
286 )
287 .toArray(),
288 );
289 for (const job of due) {
290 const callback = registry.callbacks.get(job.callback);
291 const claimed = yield* storage.transaction(
292 Effect.gen(function* () {
293 const now = yield* Clock.currentTimeMillis;
294 const claimed = yield* Effect.sync(
295 () =>
296 raw.sql.exec(
297 `UPDATE alchemy_alarm_callbacks SET run_at = ?
298 WHERE callback = ? AND id = ? AND version = ?`,
299 now + (callback?.retryDelay ?? 30_000),
300 job.callback,
301 job.id,
302 job.version,
303 ).rowsWritten > 0,
304 );
305 yield* reconcileDurableObjectAlarm(raw);
306 return claimed;
307 }),
308 );
309 if (!claimed) continue;
310 yield* storage.sync();
311 const result = yield* Effect.gen(function* () {
312 if (!callback) {
313 return yield* Effect.fail(
314 new CallbackError({
315 callback: job.callback,
316 message: "No handler is registered for a pending alarm",
317 }),
318 );
319 }
320 const payload = yield* Effect.try({
321 try: () => JSON.parse(job.payload),
322 catch: (cause) =>
323 new CallbackError({
324 callback: job.callback,
325 message: "Invalid persisted alarm payload",
326 cause,
327 }),
328 });
329 yield* callback.handler(payload);
330 }).pipe(Effect.scoped, Effect.exit);
331 if (Exit.isFailure(result)) {
332 if (Cause.hasInterrupts(result.cause))
333 return yield* Effect.failCause(result.cause);
334 yield* Effect.logError(
335 "Durable Object alarm callback failed",
336 result.cause,
337 );
338 continue;
339 }
340 yield* storage.transaction(
341 Effect.gen(function* () {
342 yield* Effect.sync(() =>
343 raw.sql.exec(
344 "DELETE FROM alchemy_alarm_callbacks WHERE callback = ? AND id = ? AND version = ?",
345 job.callback,
346 job.id,
347 job.version,
348 ),
349 );
350 yield* reconcileDurableObjectAlarm(raw);
351 }),
352 );
353 }
354 yield* storage.transaction(reconcileDurableObjectAlarm(raw));
355 });