1import { DurableObject } from "cloudflare:workers";
2import type {
3 Column,
4 Delivery,
5 LegacyRow,
6 SchemaObject,
7 SchemaVersion,
8 Snapshot,
9} from "./types.ts";
10
11// Freeze the pre-callback scheduler; importing Alchemy here would weaken the upgrade test.
12export class UpgradeObject extends DurableObject<unknown> {
13 constructor(ctx: DurableObjectState, env: unknown) {
14 super(ctx, env);
15 ctx.storage.sql.exec(`
16 CREATE TABLE IF NOT EXISTS alchemy_scheduled_events (
17 id TEXT PRIMARY KEY,
18 run_at INTEGER NOT NULL,
19 repeat_ms INTEGER,
20 payload TEXT NOT NULL
21 );
22 CREATE INDEX IF NOT EXISTS idx_alchemy_scheduled_events_run_at
23 ON alchemy_scheduled_events (run_at);
24 `);
25 ctx.blockConcurrencyWhile(async () => {
26 const constructors =
27 (await ctx.storage.get<string[]>("constructors")) ?? [];
28 await ctx.storage.put("constructors", [...constructors, "v1"]);
29 });
30 }
31
32 private reconcileAlarm() {
33 const next = this.ctx.storage.sql
34 .exec<{ run_at: number | null }>(
35 "SELECT MIN(run_at) AS run_at FROM alchemy_scheduled_events",
36 )
37 .one().run_at;
38 return next === null
39 ? this.ctx.storage.deleteAlarm()
40 : this.ctx.storage.setAlarm(next);
41 }
42
43 async seed(proof = true) {
44 if (await this.ctx.storage.get("marker")) {
45 throw new Error("Upgrade fixture must start with an unseeded object");
46 }
47 await this.ctx.storage.put("marker", "written-by-v1");
48 const now = Date.now();
49 for (const [id, runAt, repeatMs] of [
50 ["v1-proof", now + 500, null],
51 ["legacy-one", now + 300_000, null],
52 ["legacy-repeat", now + 301_000, 2_000],
53 ["legacy-cancel", now + 302_000, null],
54 ] as const) {
55 if (id === "v1-proof" && !proof) continue;
56 this.ctx.storage.sql.exec(
57 `INSERT INTO alchemy_scheduled_events (id, run_at, repeat_ms, payload)
58 VALUES (?, ?, ?, ?)`,
59 id,
60 runAt,
61 repeatMs,
62 JSON.stringify({ callback: "archive", value: id }),
63 );
64 }
65 await this.reconcileAlarm();
66 return this.snapshot();
67 }
68
69 async alarm() {
70 const now = Date.now();
71 const due = this.ctx.storage.sql
72 .exec<LegacyRow>(
73 "SELECT id, run_at, repeat_ms, payload FROM alchemy_scheduled_events WHERE run_at <= ? ORDER BY run_at ASC",
74 now,
75 )
76 .toArray();
77 const deliveries =
78 (await this.ctx.storage.get<Delivery[]>("deliveries")) ?? [];
79 for (const event of due) {
80 if (event.repeat_ms !== null) {
81 this.ctx.storage.sql.exec(
82 "UPDATE alchemy_scheduled_events SET run_at = ? WHERE id = ?",
83 now + event.repeat_ms,
84 event.id,
85 );
86 } else {
87 this.ctx.storage.sql.exec(
88 "DELETE FROM alchemy_scheduled_events WHERE id = ?",
89 event.id,
90 );
91 }
92 deliveries.push({
93 version: "v1",
94 channel: "legacy",
95 id: event.id,
96 payload: JSON.parse(event.payload),
97 });
98 }
99 await this.ctx.storage.put("deliveries", deliveries);
100 await this.reconcileAlarm();
101 }
102
103 async snapshot(): Promise<Snapshot> {
104 const schema = this.ctx.storage.sql
105 .exec<SchemaObject>(
106 "SELECT name, type, sql FROM sqlite_master WHERE name LIKE 'alchemy_%' OR name LIKE 'idx_alchemy_%' ORDER BY name",
107 )
108 .toArray();
109 const schemaRows = schema.some((row) => row.name === "alchemy_alarm_schema")
110 ? this.ctx.storage.sql
111 .exec<SchemaVersion>(
112 "SELECT id, version FROM alchemy_alarm_schema ORDER BY id",
113 )
114 .toArray()
115 : [];
116 return {
117 schema,
118 schemaRows,
119 schemaVersion: schemaRows[0]?.version ?? null,
120 version: "v1",
121 id: this.ctx.id.toString(),
122 marker: (await this.ctx.storage.get<string>("marker")) ?? null,
123 constructors:
124 (await this.ctx.storage.get<string[]>("constructors")) ?? [],
125 legacyRows: this.ctx.storage.sql
126 .exec<LegacyRow>(
127 "SELECT id, run_at, repeat_ms, payload FROM alchemy_scheduled_events ORDER BY id",
128 )
129 .toArray(),
130 legacyColumns: this.ctx.storage.sql
131 .exec<Column>(
132 "SELECT name, type, \"notnull\", pk FROM pragma_table_info('alchemy_scheduled_events') ORDER BY cid",
133 )
134 .toArray(),
135 callbacks: [],
136 tables: this.ctx.storage.sql
137 .exec<{ name: string }>(
138 "SELECT name FROM sqlite_master WHERE type = 'table' AND name LIKE 'alchemy_%' ORDER BY name",
139 )
140 .toArray()
141 .map((row) => row.name),
142 alarm: await this.ctx.storage.getAlarm(),
143 deliveries: (await this.ctx.storage.get<Delivery[]>("deliveries")) ?? [],
144 };
145 }
146}
147
148export default {
149 async fetch(
150 request: Request,
151 env: { UpgradeObject: DurableObjectNamespace<UpgradeObject> },
152 ) {
153 if (
154 request.headers.get("x-alarm-worker-version") !== null &&
155 request.headers.get("x-alarm-worker-version") !== "v1"
156 ) {
157 return new Response("Alarm worker version mismatch", {
158 status: 409,
159 headers: { "x-alarm-worker-version": "v1" },
160 });
161 }
162 const url = new URL(request.url);
163 const name = url.searchParams.get("name") ?? "persisted-object";
164 const object = env.UpgradeObject.getByName(name);
165 const path = url.pathname;
166 if (path === "/seed" && request.method === "POST") {
167 return Response.json(await object.seed(name === "persisted-object"));
168 }
169 if (path === "/snapshot") {
170 return Response.json(await object.snapshot());
171 }
172 return new Response("Not Found", { status: 404 });
173 },
174};