1import type * as cf from "@cloudflare/workers-types";
2import * as Clock from "effect/Clock";
3import * as Data from "effect/Data";
4import * as Effect from "effect/Effect";
5import { ActiveStorageTransactions } from "./DurableObjectTransactionContext.ts";
6
7const SCHEMA_VERSION = 1;
8
9export class UnsupportedAlarmSchemaVersion extends Data.TaggedError(
10 "UnsupportedAlarmSchemaVersion",
11)<{
12 readonly version: number;
13 readonly supportedVersion: number;
14}> {}
15
16export const ensureAlarmTables = (storage: cf.DurableObjectStorage) =>
17 Effect.gen(function* () {
18 const transaction = (yield* ActiveStorageTransactions).get(storage);
19 if (transaction?.alarmTablesEnsured) return;
20 yield* initializeAlarmTables(storage);
21 if (transaction !== undefined) transaction.alarmTablesEnsured = true;
22 });
23
24const initializeAlarmTables = (storage: cf.DurableObjectStorage) =>
25 Effect.sync(() => {
26 const hasVersion =
27 storage.sql
28 .exec<{ name: string }>(
29 "SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'alchemy_alarm_schema'",
30 )
31 .toArray().length > 0;
32 const version = hasVersion
33 ? storage.sql
34 .exec<{ version: number }>(
35 "SELECT version FROM alchemy_alarm_schema WHERE id = 1",
36 )
37 .one().version
38 : 0;
39 if (version === SCHEMA_VERSION) return;
40 if (version !== 0) {
41 throw new UnsupportedAlarmSchemaVersion({
42 version,
43 supportedVersion: SCHEMA_VERSION,
44 });
45 }
46
47 storage.transactionSync(() => {
48 // Version 0 is the original, unversioned scheduleEvent schema.
49 storage.sql.exec(`
50 CREATE TABLE IF NOT EXISTS alchemy_scheduled_events (
51 id TEXT PRIMARY KEY,
52 run_at INTEGER NOT NULL,
53 repeat_ms INTEGER,
54 payload TEXT NOT NULL
55 );
56 CREATE INDEX IF NOT EXISTS idx_alchemy_scheduled_events_run_at
57 ON alchemy_scheduled_events (run_at);
58 CREATE TABLE alchemy_alarm_callbacks (
59 callback TEXT NOT NULL,
60 id TEXT NOT NULL,
61 version TEXT NOT NULL,
62 run_at INTEGER NOT NULL,
63 payload TEXT NOT NULL,
64 PRIMARY KEY (callback, id)
65 );
66 CREATE INDEX idx_alchemy_alarm_callbacks_run_at
67 ON alchemy_alarm_callbacks (run_at, callback, id);
68 CREATE TABLE IF NOT EXISTS alchemy_alarm_schema (
69 id INTEGER PRIMARY KEY CHECK (id = 1),
70 version INTEGER NOT NULL
71 );
72 INSERT INTO alchemy_alarm_schema (id, version) VALUES (1, 1)
73 ON CONFLICT (id) DO UPDATE SET version = excluded.version;
74 `);
75 });
76 });
77
78// Both schedulers share the Durable Object's single native alarm.
79export const reconcileDurableObjectAlarm = (storage: cf.DurableObjectStorage) =>
80 Effect.gen(function* () {
81 const transaction = (yield* ActiveStorageTransactions).get(storage);
82 if (transaction !== undefined) {
83 transaction.alarmDirty = true;
84 } else {
85 yield* reconcileAlarm(storage);
86 }
87 });
88
89// Flush before commit or an explicit alarm operation, never after rollback.
90export const flushDurableObjectAlarm = (storage: cf.DurableObjectStorage) =>
91 Effect.gen(function* () {
92 const transaction = (yield* ActiveStorageTransactions).get(storage);
93 if (transaction?.alarmDirty && !transaction.rolledBack) {
94 yield* reconcileAlarm(storage);
95 transaction.alarmDirty = false;
96 }
97 });
98
99const reconcileAlarm = (storage: cf.DurableObjectStorage) =>
100 Effect.gen(function* () {
101 const next = yield* Effect.sync(
102 () =>
103 storage.sql
104 .exec<{ run_at: number | null }>(`
105 SELECT MIN(run_at) AS run_at FROM (
106 SELECT MIN(run_at) AS run_at FROM alchemy_scheduled_events
107 UNION ALL
108 SELECT MIN(run_at) AS run_at FROM alchemy_alarm_callbacks
109 )
110 `)
111 .one().run_at,
112 );
113 const now = yield* Clock.currentTimeMillis;
114 // A continuation must use a new timestamp, not the alarm currently being acknowledged.
115 yield* Effect.promise(() =>
116 next === null
117 ? storage.deleteAlarm()
118 : storage.setAlarm(Math.max(next, now + 1)),
119 );
120 });