1import * as Cloudflare from "@/Cloudflare/index.ts";
2import * as Alchemy from "@/index.ts";
3import * as Cause from "effect/Cause";
4import * as Effect from "effect/Effect";
5import * as Exit from "effect/Exit";
6import { HttpServerRequest } from "effect/http/HttpServerRequest";
7import * as HttpServerResponse from "effect/http/HttpServerResponse";
8import type {
9 CallbackRow,
10 Column,
11 Delivery,
12 LegacyRow,
13 MigrationProbe,
14 SchemaObject,
15 SchemaVersion,
16 Snapshot,
17} from "./types.ts";
18
19export class UpgradeObject extends Cloudflare.DurableObject<UpgradeObject>()(
20 "UpgradeObject",
21 Effect.gen(function* () {
22 const state = yield* Cloudflare.DurableObjectState;
23 const storage = state.storage;
24
25 return Effect.gen(function* () {
26 const constructors = (yield* storage.get<string[]>("constructors")) ?? [];
27 yield* storage.put("constructors", [...constructors, "v2"]);
28
29 const record = Effect.fn(function* (delivery: Delivery) {
30 const deliveries = (yield* storage.get<Delivery[]>("deliveries")) ?? [];
31 yield* storage.put("deliveries", [...deliveries, delivery]);
32 });
33 const onArchive = yield* Alchemy.makeCallback(
34 "archive",
35 Effect.fn(function* (payload: { value: string }) {
36 yield* record({
37 version: "v2",
38 channel: "callback",
39 id: payload.value,
40 payload,
41 });
42 }),
43 );
44
45 const snapshot = Effect.fn(
46 function* (migrate: boolean) {
47 if (migrate) yield* Cloudflare.listEvents;
48 const schema = yield* (yield* storage.sql.exec<SchemaObject>(
49 "SELECT name, type, sql FROM sqlite_master WHERE name LIKE 'alchemy_%' OR name LIKE 'idx_alchemy_%' ORDER BY name",
50 )).toArray();
51 const tables = schema
52 .filter((row) => row.type === "table")
53 .map((row) => row.name);
54 const schemaRows = tables.includes("alchemy_alarm_schema")
55 ? yield* (yield* storage.sql.exec<SchemaVersion>(
56 "SELECT id, version FROM alchemy_alarm_schema ORDER BY id",
57 )).toArray()
58 : [];
59 const result: Snapshot = {
60 schema,
61 schemaRows,
62 schemaVersion: schemaRows[0]?.version ?? null,
63 version: "v2",
64 id: yield* Effect.sync(() => state.id.toString()),
65 marker: (yield* storage.get<string>("marker")) ?? null,
66 constructors: (yield* storage.get<string[]>("constructors")) ?? [],
67 legacyRows: yield* (yield* storage.sql.exec<LegacyRow>(
68 "SELECT id, run_at, repeat_ms, payload FROM alchemy_scheduled_events ORDER BY id",
69 )).toArray(),
70 legacyColumns: yield* (yield* storage.sql.exec<Column>(
71 "SELECT name, type, \"notnull\", pk FROM pragma_table_info('alchemy_scheduled_events') ORDER BY cid",
72 )).toArray(),
73 callbacks: tables.includes("alchemy_alarm_callbacks")
74 ? yield* (yield* storage.sql.exec<CallbackRow>(
75 "SELECT callback, id, version, run_at, payload FROM alchemy_alarm_callbacks ORDER BY callback, id",
76 )).toArray()
77 : [],
78 tables,
79 alarm: yield* storage.getAlarm(),
80 deliveries: (yield* storage.get<Delivery[]>("deliveries")) ?? [],
81 };
82 return result;
83 },
84 Effect.provideService(Cloudflare.DurableObjectState, state),
85 );
86
87 return {
88 // V1 Workers can still reach the upgraded object during edge rollout.
89 snapshot: () => snapshot(true),
90 reconstruct: () =>
91 state.abort("alarm upgrade reconstruction", { retryAlarm: false }),
92 probe: Effect.fn(
93 function* (kind: "future" | "rollback") {
94 if (kind === "future") {
95 yield* Cloudflare.listEvents;
96 yield* onArchive.schedule("future-preserved", {
97 after: "5 minutes",
98 payload: { value: "future-preserved" },
99 });
100 yield* storage.sql.exec(
101 "UPDATE alchemy_alarm_schema SET version = 2 WHERE id = 1",
102 );
103 } else {
104 // Fail after the migration has created its first index and callback table.
105 yield* storage.sql.exec(
106 "DROP INDEX idx_alchemy_scheduled_events_run_at",
107 );
108 yield* storage.sql.exec(
109 "CREATE TABLE idx_alchemy_alarm_callbacks_run_at (value TEXT)",
110 );
111 }
112 const before = yield* snapshot(false);
113 const exit = yield* Effect.exit(Cloudflare.listEvents);
114 let failure: MigrationProbe["failure"] = null;
115 if (Exit.isFailure(exit)) {
116 const error = Cause.squash(exit.cause);
117 failure =
118 error instanceof Cloudflare.UnsupportedAlarmSchemaVersion
119 ? {
120 tag: error._tag,
121 message: Cause.pretty(exit.cause),
122 version: error.version,
123 supportedVersion: error.supportedVersion,
124 }
125 : {
126 tag: error instanceof Error ? error.name : "Defect",
127 message: Cause.pretty(exit.cause),
128 version: null,
129 supportedVersion: null,
130 };
131 }
132 const after = yield* snapshot(false);
133 let retryBefore: Snapshot | null = null;
134 let recovered: Snapshot | null = null;
135 if (kind === "rollback") {
136 yield* storage.sql.exec(
137 "DROP TABLE idx_alchemy_alarm_callbacks_run_at",
138 );
139 yield* storage.sql.exec(`
140 CREATE TABLE alchemy_alarm_schema (
141 id INTEGER PRIMARY KEY CHECK (id = 1),
142 version INTEGER NOT NULL
143 );
144 INSERT INTO alchemy_alarm_schema (id, version) VALUES (1, 0);
145 `);
146 retryBefore = yield* snapshot(false);
147 recovered = yield* snapshot(true);
148 }
149 const result: MigrationProbe = {
150 before,
151 after,
152 failure,
153 retryBefore,
154 recovered,
155 };
156 return result;
157 },
158 Effect.provideService(Cloudflare.DurableObjectState, state),
159 ),
160 alarm: Effect.fn(function* () {
161 const events = yield* Cloudflare.processScheduledEvents.pipe(
162 Effect.provideService(Cloudflare.DurableObjectState, state),
163 );
164 for (const event of events) {
165 yield* record({
166 version: "v2",
167 channel: "legacy",
168 id: event.id,
169 payload: event.payload,
170 });
171 }
172 }),
173 execute: Effect.fn(
174 function* (action: string) {
175 if (action === "callback-first") {
176 yield* onArchive.schedule("pending", {
177 after: "1 minute",
178 payload: { value: "canceled-callback" },
179 });
180 } else if (action === "cancel-legacy") {
181 yield* Cloudflare.cancelEvent("legacy-cancel");
182 } else if (action === "cancel-callback") {
183 yield* onArchive.cancel("pending");
184 } else if (action === "release") {
185 // Keep the migrated payloads and intervals; only bring their due times forward.
186 const at = yield* Effect.sync(() => new Date(Date.now() + 1_000));
187 yield* storage.sql.exec(
188 "UPDATE alchemy_scheduled_events SET run_at = ?",
189 at.getTime(),
190 );
191 yield* onArchive.schedule("delivered", {
192 at,
193 payload: { value: "new-callback" },
194 });
195 } else if (action === "only-callback") {
196 yield* onArchive.schedule("survivor", {
197 after: "2 seconds",
198 payload: { value: "callback-survivor" },
199 });
200 for (const event of yield* Cloudflare.listEvents) {
201 yield* Cloudflare.cancelEvent(event.id);
202 }
203 } else if (action === "only-legacy") {
204 const at = yield* Effect.sync(() => new Date(Date.now() + 2_000));
205 yield* Cloudflare.scheduleEvent("legacy-survivor", at, {
206 value: "legacy-survivor",
207 });
208 yield* onArchive.schedule("canceled", {
209 after: "1 second",
210 payload: { value: "must-not-fire" },
211 });
212 yield* onArchive.cancel("canceled");
213 } else if (action !== "snapshot") {
214 return yield* Effect.die(new Error(`Unknown action: ${action}`));
215 }
216 return yield* snapshot(true);
217 },
218 Effect.provideService(Cloudflare.DurableObjectState, state),
219 ),
220 };
221 });
222 }),
223) {}
224
225export default class AlarmUpgradeWorker extends Cloudflare.Worker<AlarmUpgradeWorker>()(
226 "AlarmUpgradeWorker",
227 { main: import.meta.url },
228 Effect.gen(function* () {
229 const objects = yield* UpgradeObject;
230 return {
231 fetch: Effect.gen(function* () {
232 const request = yield* HttpServerRequest;
233 const expectedVersion = request.headers["x-alarm-worker-version"];
234 if (expectedVersion !== undefined && expectedVersion !== "v2") {
235 return HttpServerResponse.text("Alarm worker version mismatch", {
236 status: 409,
237 headers: { "x-alarm-worker-version": "v2" },
238 });
239 }
240 const url = new URL(request.url, "http://localhost");
241 const action = url.pathname.slice(1);
242 const object = objects.getByName(
243 url.searchParams.get("name") ?? "persisted-object",
244 );
245 if (request.method !== "POST" && action !== "snapshot") {
246 return HttpServerResponse.text("Method Not Allowed", { status: 405 });
247 }
248 if (action === "reconstruct") {
249 const aborted = yield* object.reconstruct().pipe(
250 Effect.catchCause((cause) => {
251 // The method declares no application errors, but aborting the
252 // object rejects its RPC transport with RpcCallError.
253 const error = Cause.squash(cause);
254 if (
255 error instanceof Cloudflare.RpcCallError &&
256 error.cause instanceof Error &&
257 error.cause.message.includes("alarm upgrade reconstruction")
258 )
259 return Effect.succeed(true);
260 return Effect.failCause(cause);
261 }),
262 );
263 return yield* HttpServerResponse.json({ aborted });
264 }
265 if (action === "migration-future" || action === "migration-rollback") {
266 const probe = yield* object
267 .probe(action === "migration-future" ? "future" : "rollback")
268 .pipe(Effect.orDie);
269 return yield* HttpServerResponse.json(probe);
270 }
271 const snapshot = yield* object.execute(action).pipe(Effect.orDie);
272 return yield* HttpServerResponse.json(snapshot);
273 }).pipe(
274 Effect.catchCause((cause) =>
275 Effect.succeed(
276 HttpServerResponse.text(Cause.pretty(cause), { status: 500 }),
277 ),
278 ),
279 ),
280 };
281 }),
282) {}