1import * as Clock from "effect/Clock";
2import * as Effect from "effect/Effect";
3import * as Stream from "effect/Stream";
4import type { RuntimeContext } from "../../RuntimeContext.ts";
5import { DurableObjectState } from "./DurableObjectState.ts";
6import {
7 ensureAlarmTables,
8 reconcileDurableObjectAlarm,
9} from "./DurableObjectAlarmStorage.ts";
10import type { SqlStorageValue } from "./DurableObjectStorage.ts";
11
12// ---------------------------------------------------------------------------
13// Scheduled Events — SQLite-backed cron/timer for Durable Objects
14// ---------------------------------------------------------------------------
15
16const ensureTable = Effect.gen(function* () {
17 const ctx = yield* DurableObjectState;
18 yield* ensureAlarmTables(ctx.raw.storage);
19});
20
21const inTransaction = <A, R>(effect: Effect.Effect<A, never, R>) =>
22 Effect.gen(function* () {
23 const ctx = yield* DurableObjectState;
24 return yield* ctx.storage.transaction(effect).pipe(Effect.orDie);
25 });
26
27export interface ScheduledEvent {
28 id: string;
29 runAt: Date;
30 repeatMs?: number;
31 payload: unknown;
32}
33
34interface EventRow extends Record<string, SqlStorageValue> {
35 id: string;
36 run_at: number;
37 repeat_ms: number | null;
38 payload: string;
39}
40
41const toScheduledEvent = (row: EventRow): ScheduledEvent => ({
42 id: row.id,
43 runAt: new Date(row.run_at),
44 repeatMs: row.repeat_ms ?? undefined,
45 payload: JSON.parse(row.payload) as unknown,
46});
47
48/**
49 * Schedule (or reschedule) a named event on the current Durable Object.
50 *
51 * The event is persisted in a SQLite table and the DO alarm is set to
52 * the earliest pending `runAt`. If an event with the same `id` already
53 * exists it is replaced (upsert).
54 *
55 * @param id Stable identifier for the event (used for upsert / cancel).
56 * @param runAt When the event should fire.
57 * @param payload Arbitrary JSON-serialisable data delivered to the alarm handler.
58 * @param repeatMs If set, the event re-schedules itself this many ms after each fire.
59 */
60export const scheduleEvent = Effect.fn(function* (
61 id: string,
62 runAt: Date,
63 payload: unknown,
64 repeatMs?: number,
65) {
66 yield* ensureTable;
67 const ctx = yield* DurableObjectState;
68
69 yield* ctx.storage.sql.exec(
70 `INSERT OR REPLACE INTO alchemy_scheduled_events (id, run_at, repeat_ms, payload)
71 VALUES (?, ?, ?, ?)`,
72 id,
73 runAt.getTime(),
74 repeatMs ?? null,
75 JSON.stringify(payload),
76 );
77
78 yield* reconcileAlarm;
79}, inTransaction);
80
81/**
82 * Cancel a previously scheduled event by id. No-op if the event does not exist.
83 */
84export const cancelEvent = Effect.fn(function* (id: string) {
85 yield* ensureTable;
86 const ctx = yield* DurableObjectState;
87
88 yield* ctx.storage.sql.exec(
89 `DELETE FROM alchemy_scheduled_events WHERE id = ?`,
90 id,
91 );
92
93 yield* reconcileAlarm;
94}, inTransaction);
95
96/**
97 * List all currently scheduled events, ordered by `runAt` ascending.
98 */
99export const listEvents: Effect.Effect<
100 ScheduledEvent[],
101 never,
102 DurableObjectState | RuntimeContext
103> = Effect.gen(function* () {
104 yield* ensureTable;
105 const ctx = yield* DurableObjectState;
106
107 const cursor = yield* ctx.storage.sql.exec<EventRow>(
108 `SELECT id, run_at, repeat_ms, payload FROM alchemy_scheduled_events ORDER BY run_at ASC`,
109 );
110
111 return yield* cursor.pipe(Stream.map(toScheduledEvent), Stream.runCollect);
112});
113
114/**
115 * Process all events whose `runAt` <= now. Returns the fired events.
116 *
117 * - One-shot events are deleted after firing.
118 * - Repeating events have their `runAt` bumped by `repeatMs`.
119 * - The DO alarm is re-set to the next pending event (if any).
120 *
121 * Call this from your Durable Object's `alarm` handler:
122 *
123 * ```ts
124 * alarm: () => Effect.gen(function* () {
125 * const fired = yield* Cloudflare.Workers.processScheduledEvents;
126 * for (const event of fired) {
127 * // handle each event
128 * }
129 * })
130 * ```
131 */
132export const processScheduledEvents: Effect.Effect<
133 ScheduledEvent[],
134 never,
135 DurableObjectState | RuntimeContext
136> = Effect.gen(function* () {
137 yield* ensureTable;
138 const ctx = yield* DurableObjectState;
139 const now = yield* Clock.currentTimeMillis;
140
141 const cursor = yield* ctx.storage.sql.exec<EventRow>(
142 `SELECT id, run_at, repeat_ms, payload FROM alchemy_scheduled_events WHERE run_at <= ? ORDER BY run_at ASC`,
143 now,
144 );
145
146 const rows = yield* cursor.toArray();
147 const fired: ScheduledEvent[] = [];
148 for (const row of rows) {
149 yield* row.repeat_ms != null
150 ? ctx.storage.sql.exec(
151 `UPDATE alchemy_scheduled_events SET run_at = ? WHERE id = ?`,
152 now + row.repeat_ms,
153 row.id,
154 )
155 : ctx.storage.sql.exec(
156 `DELETE FROM alchemy_scheduled_events WHERE id = ?`,
157 row.id,
158 );
159 fired.push(toScheduledEvent(row));
160 }
161
162 yield* reconcileAlarm;
163 return fired;
164}).pipe(inTransaction);
165
166/**
167 * Set the DO alarm to the earliest pending event, or clear it if none remain.
168 */
169const reconcileAlarm: Effect.Effect<
170 void,
171 never,
172 DurableObjectState | RuntimeContext
173> = Effect.gen(function* () {
174 const ctx = yield* DurableObjectState;
175
176 yield* reconcileDurableObjectAlarm(ctx.raw.storage);
177});