1import * as Clock from "effect/Clock";
2import * as DateTime from "effect/DateTime";
3import * as Duration from "effect/Duration";
4import * as Effect from "effect/Effect";
5import * as Option from "effect/Option";
6import * as Pull from "effect/Pull";
7import * as Schedule from "effect/Schedule";
8import * as Schema from "effect/Schema";
9import * as SqlClient from "effect/sql/SqlClient";
10
11/**
12 * The storage and delivery rules of one environment's held webhook requests.
13 * Runs inside the environment's `HookInboxObject` Durable Object against its SQLite
14 * storage; written against `SqlClient` so tests can run it on any SQLite.
15 */
16
17/** How long a held request waits for its environment. */
18const HOOK_INBOX_TTL_MS = 24 * 60 * 60 * 1000;
19/** Requests one environment may have waiting at once. */
20const HOOK_INBOX_MAX_REQUESTS = 1_000;
21/** Body bytes one environment may have waiting at once. */
22export const HOOK_INBOX_MAX_BYTES = 50 * 1_048_576;
23/**
24 * Requests one hook may have waiting at once. The relay cannot check tokens,
25 * so this keeps junk sent to one hook id from crowding out the others.
26 */
27export const HOOK_INBOX_MAX_PER_HOOK = 100;
28/** Requests pushed per alarm run; the next run starts right away while more wait. */
29const DELIVERIES_PER_RUN = 20;
30/** Requests read at a time; a run keeps reading past busy hooks' backlogs. */
31const ROWS_READ_PER_PAGE = 50;
32/** Wait before trying a hook whose environment answered busy again. */
33const BUSY_RETRY_MS = 30_000;
34const MAX_RETRY_DELAY = Duration.minutes(10);
35/** Past this many failures the schedule is at its cap, so stepping further changes nothing. */
36const MAX_COUNTED_FAILURES = 25;
37
38/**
39 * Delays between delivery attempts while the environment is unreachable,
40 * indexed by consecutive failures since it was last reached or woke us.
41 * A wake means its tunnel just connected, but Cloudflare can take a few
42 * minutes to route the hostname to it, so the first 3 minutes retry every
43 * 10 s. After that the environment is likely gone again: 30 s, 1 min, 2 min,
44 * ... up to 10 min.
45 */
46export const retrySchedule = Schedule.spaced("10 seconds").pipe(
47 Schedule.upTo({ times: 18 }),
48 Schedule.concat(
49 Schedule.exponential("30 seconds").pipe(
50 Schedule.modifyDelay(({ duration }) =>
51 Effect.succeed(Duration.min(Duration.fromInputUnsafe(duration), MAX_RETRY_DELAY)),
52 ),
53 ),
54 ),
55);
56
57export interface HeldHook {
58 readonly id: string;
59 readonly receivedAt: string;
60 readonly method: string;
61 /** Path segments exactly as the sender sent them; the environment decodes them. */
62 readonly rawHookId: string;
63 readonly rawToken: string;
64 /** The decoded hook id, so every spelling of one hook shares its cap. */
65 readonly hookKey: string;
66 /** Without the leading `?`. */
67 readonly query: string;
68 readonly headers: Readonly<Record<string, string>>;
69 readonly body: Uint8Array;
70}
71
72/**
73 * `delivered` deletes the request. `unreachable` keeps it and backs off the
74 * whole inbox, since nothing else will get through either. `busy` keeps it
75 * and its hook's later requests for the next run, but lets other hooks'
76 * requests go ahead, so one stuck task cannot hold up the rest.
77 */
78export type DeliveryOutcome = "delivered" | "busy" | "unreachable";
79
80/** The `retrySchedule` delay after `failures` consecutive unreachable attempts. */
81export const retryDelayMs = Effect.fn("HookInboxStore.retryDelayMs")(function* (failures: number) {
82 const step = yield* Schedule.toStep(retrySchedule);
83 let delay = MAX_RETRY_DELAY;
84 for (
85 let attempt = 0;
86 attempt < Math.min(Math.max(1, failures), MAX_COUNTED_FAILURES);
87 attempt++
88 ) {
89 const next = yield* step(0, undefined).pipe(
90 Effect.map(([, duration]) => Option.some(duration)),
91 // The schedule never ends; stay at the cap if it ever does.
92 Pull.catchDone(() => Effect.succeedNone),
93 );
94 if (Option.isNone(next)) break;
95 delay = next.value;
96 }
97 return Duration.toMillis(delay);
98});
99
100const HeadersJson = Schema.fromJsonString(Schema.Record(Schema.String, Schema.String));
101const encodeHeaders = Schema.encodeSync(HeadersJson);
102const decodeHeaders = Schema.decodeUnknownOption(HeadersJson);
103
104interface HeldHookRow {
105 readonly seq: number;
106 readonly id: string;
107 readonly received_at: string;
108 readonly method: string;
109 readonly raw_hook_id: string;
110 readonly raw_token: string;
111 readonly hook_key: string;
112 readonly query: string;
113 readonly headers: string;
114 readonly body: Uint8Array;
115}
116
117interface TargetRow {
118 readonly base_url: string;
119 readonly failures: number;
120}
121
122const iso = (epochMillis: number) => DateTime.formatIso(DateTime.makeUnsafe(epochMillis));
123
124export const migrate = Effect.gen(function* () {
125 const sql = yield* SqlClient.SqlClient;
126 yield* sql`
127 CREATE TABLE IF NOT EXISTS held_hooks (
128 seq INTEGER PRIMARY KEY AUTOINCREMENT,
129 id TEXT NOT NULL UNIQUE,
130 received_at TEXT NOT NULL,
131 method TEXT NOT NULL,
132 raw_hook_id TEXT NOT NULL,
133 raw_token TEXT NOT NULL,
134 hook_key TEXT NOT NULL,
135 query TEXT NOT NULL,
136 headers TEXT NOT NULL,
137 body BLOB NOT NULL
138 )
139 `;
140 yield* sql`CREATE INDEX IF NOT EXISTS held_hooks_hook ON held_hooks (hook_key)`;
141 // Where to push, and how many attempts in a row have failed. One row.
142 yield* sql`
143 CREATE TABLE IF NOT EXISTS held_hooks_target (
144 id INTEGER PRIMARY KEY CHECK (id = 1),
145 base_url TEXT NOT NULL,
146 failures INTEGER NOT NULL DEFAULT 0
147 )
148 `;
149});
150
151const setTarget = (baseUrl: string, options: { readonly resetFailures: boolean }) =>
152 Effect.gen(function* () {
153 const sql = yield* SqlClient.SqlClient;
154 yield* options.resetFailures
155 ? sql`
156 INSERT INTO held_hooks_target (id, base_url) VALUES (1, ${baseUrl})
157 ON CONFLICT (id) DO UPDATE SET base_url = excluded.base_url, failures = 0
158 `
159 : sql`
160 INSERT INTO held_hooks_target (id, base_url) VALUES (1, ${baseUrl})
161 ON CONFLICT (id) DO UPDATE SET base_url = excluded.base_url
162 `;
163 });
164
165const readTarget = Effect.gen(function* () {
166 const sql = yield* SqlClient.SqlClient;
167 const rows = yield* sql<TargetRow>`SELECT base_url, failures FROM held_hooks_target WHERE id = 1`;
168 return rows[0] ?? null;
169});
170
171const resetFailures = Effect.gen(function* () {
172 const sql = yield* SqlClient.SqlClient;
173 yield* sql`UPDATE held_hooks_target SET failures = 0 WHERE id = 1`;
174});
175
176/**
177 * Counts one more unreachable attempt. Incremented in place, so a wake that
178 * reset the count while this run's request was in flight is not overwritten.
179 */
180const countFailure = Effect.gen(function* () {
181 const sql = yield* SqlClient.SqlClient;
182 const rows = yield* sql<{ readonly failures: number }>`
183 UPDATE held_hooks_target SET failures = min(failures + 1, ${MAX_COUNTED_FAILURES})
184 WHERE id = 1 RETURNING failures
185 `;
186 return rows[0]?.failures ?? 1;
187});
188
189const hasPending = Effect.gen(function* () {
190 const sql = yield* SqlClient.SqlClient;
191 const rows = yield* sql<{ readonly id: string }>`SELECT id FROM held_hooks LIMIT 1`;
192 return rows.length > 0;
193});
194
195/**
196 * Stores a request for later delivery to `baseUrl`. Returns the time the
197 * first attempt is due, or null when the inbox is full and nothing was stored.
198 */
199export const hold = Effect.fn("HookInboxStore.hold")(function* (hook: HeldHook, baseUrl: string) {
200 const sql = yield* SqlClient.SqlClient;
201 // One statement, so the caps hold however many requests arrive at once.
202 const inserted = yield* sql<{ readonly id: string }>`
203 INSERT INTO held_hooks
204 (id, received_at, method, raw_hook_id, raw_token, hook_key, query, headers, body)
205 SELECT ${hook.id}, ${hook.receivedAt}, ${hook.method}, ${hook.rawHookId}, ${hook.rawToken},
206 ${hook.hookKey}, ${hook.query}, ${encodeHeaders(hook.headers)}, ${hook.body}
207 WHERE (SELECT count(*) FROM held_hooks) < ${HOOK_INBOX_MAX_REQUESTS}
208 AND (SELECT count(*) FROM held_hooks WHERE hook_key = ${hook.hookKey})
209 < ${HOOK_INBOX_MAX_PER_HOOK}
210 AND (SELECT coalesce(sum(length(body)), 0) FROM held_hooks) + ${hook.body.byteLength}
211 <= ${HOOK_INBOX_MAX_BYTES}
212 ON CONFLICT (id) DO NOTHING
213 RETURNING id
214 `;
215 if (inserted.length === 0) {
216 yield* Effect.annotateCurrentSpan({ "relay.inbox.refused": yield* refusalReason(hook) });
217 return null;
218 }
219 yield* setTarget(baseUrl, { resetFailures: false });
220 const target = yield* readTarget;
221 const backlog = yield* backlogSize;
222 yield* Effect.annotateCurrentSpan({
223 "relay.inbox.held_count": backlog.count,
224 "relay.inbox.held_bytes": backlog.bytes,
225 });
226 return (yield* Clock.currentTimeMillis) + (yield* retryDelayMs(target?.failures ?? 0));
227});
228
229const backlogSize = Effect.gen(function* () {
230 const sql = yield* SqlClient.SqlClient;
231 const rows = yield* sql<{ readonly count: number; readonly bytes: number }>`
232 SELECT count(*) AS count, coalesce(sum(length(body)), 0) AS bytes FROM held_hooks
233 `;
234 return rows[0] ?? { count: 0, bytes: 0 };
235});
236
237/** Which cap refused a request, or that it was already held, for traces. */
238const refusalReason = (hook: HeldHook) =>
239 Effect.gen(function* () {
240 const sql = yield* SqlClient.SqlClient;
241 const existing = yield* sql<{ readonly id: string }>`
242 SELECT id FROM held_hooks WHERE id = ${hook.id}
243 `;
244 if (existing.length > 0) return "already_held";
245 const backlog = yield* backlogSize;
246 if (backlog.count >= HOOK_INBOX_MAX_REQUESTS) return "max_requests";
247 if (backlog.bytes + hook.body.byteLength > HOOK_INBOX_MAX_BYTES) return "max_bytes";
248 return "max_per_hook";
249 });
250
251/**
252 * The environment is reachable again at `baseUrl`. Returns whether anything
253 * is waiting, so the caller can deliver right away.
254 */
255export const wake = Effect.fn("HookInboxStore.wake")(function* (baseUrl: string) {
256 if (!(yield* hasPending)) return false;
257 yield* setTarget(baseUrl, { resetFailures: true });
258 return true;
259});
260
261export const clear = Effect.gen(function* () {
262 const sql = yield* SqlClient.SqlClient;
263 yield* sql`DELETE FROM held_hooks`;
264 yield* sql`DELETE FROM held_hooks_target`;
265});
266
267/**
268 * Pushes the oldest held requests to the environment, one at a time, in
269 * order per hook. Stops at the first sign the environment is unreachable;
270 * skips past a hook whose environment answered busy. Drops requests older
271 * than the TTL. Returns when the next run is due, or null when nothing is left.
272 */
273export const deliverDue = Effect.fn("HookInboxStore.deliverDue")(function* <R>(
274 send: (baseUrl: string, hook: HeldHook) => Effect.Effect<DeliveryOutcome, never, R>,
275) {
276 const sql = yield* SqlClient.SqlClient;
277 const startedAt = yield* Clock.currentTimeMillis;
278 const expired = yield* sql<{ readonly id: string }>`
279 DELETE FROM held_hooks WHERE received_at < ${iso(startedAt - HOOK_INBOX_TTL_MS)}
280 RETURNING id
281 `;
282 const target = yield* readTarget;
283 /** Oldest held requests after `afterSeq`, skipping hooks already found busy. */
284 const readPage = (afterSeq: number, skip: ReadonlySet<string>) =>
285 sql<HeldHookRow>`
286 SELECT seq, id, received_at, method, raw_hook_id, raw_token, hook_key, query, headers, body
287 FROM held_hooks
288 WHERE seq > ${afterSeq}
289 ${skip.size === 0 ? sql`` : sql`AND hook_key NOT IN ${sql.in([...skip])}`}
290 ORDER BY seq LIMIT ${ROWS_READ_PER_PAGE}
291 `;
292 let batch = yield* readPage(0, new Set());
293 if (batch.length === 0 || target === null) {
294 if (target === null) yield* sql`DELETE FROM held_hooks`;
295 // Empty, so the next request held starts the schedule from the top.
296 else yield* resetFailures;
297 yield* Effect.annotateCurrentSpan({ "relay.inbox.expired": expired.length });
298 return null;
299 }
300
301 let failures = target.failures;
302 let sent = 0;
303 let delivered = 0;
304 let unreadable = 0;
305 let longestWaitMs = 0;
306 const busyHooks = new Set<string>();
307 /** What this run did, for the alarm's span; never request contents. */
308 const annotateRun = (result: string) =>
309 Effect.gen(function* () {
310 const backlog = yield* backlogSize;
311 yield* Effect.annotateCurrentSpan({
312 "relay.inbox.run_result": result,
313 "relay.inbox.sent": sent,
314 "relay.inbox.delivered": delivered,
315 "relay.inbox.busy_hooks": busyHooks.size,
316 "relay.inbox.expired": expired.length,
317 "relay.inbox.unreadable": unreadable,
318 "relay.inbox.consecutive_failures": failures,
319 "relay.inbox.longest_wait_ms": longestWaitMs,
320 "relay.inbox.held_count": backlog.count,
321 "relay.inbox.held_bytes": backlog.bytes,
322 });
323 });
324 // Pages run out only once every non-busy request has been tried, so a
325 // backlog behind busy hooks never hides another hook's requests.
326 pages: while (batch.length > 0) {
327 for (const row of batch) {
328 if (sent === DELIVERIES_PER_RUN) break pages;
329 // Later requests to a busy hook wait their turn, so its order is kept.
330 if (busyHooks.has(row.hook_key)) continue;
331 const headers = decodeHeaders(row.headers);
332 if (Option.isNone(headers)) {
333 // Unreadable: it can never be delivered, and must not block the rest.
334 yield* sql`DELETE FROM held_hooks WHERE id = ${row.id}`;
335 unreadable += 1;
336 continue;
337 }
338 sent += 1;
339 const outcome = yield* send(target.base_url, {
340 id: row.id,
341 receivedAt: row.received_at,
342 method: row.method,
343 rawHookId: row.raw_hook_id,
344 rawToken: row.raw_token,
345 hookKey: row.hook_key,
346 query: row.query,
347 headers: headers.value,
348 body: row.body,
349 });
350 if (outcome === "unreachable") {
351 failures = yield* countFailure;
352 yield* annotateRun("unreachable");
353 return (yield* Clock.currentTimeMillis) + (yield* retryDelayMs(failures));
354 }
355 if (failures !== 0) {
356 failures = 0;
357 yield* resetFailures;
358 }
359 if (outcome === "busy") {
360 busyHooks.add(row.hook_key);
361 continue;
362 }
363 yield* sql`DELETE FROM held_hooks WHERE id = ${row.id}`;
364 delivered += 1;
365 const receivedAtMs = DateTime.toEpochMillis(DateTime.makeUnsafe(row.received_at));
366 longestWaitMs = Math.max(longestWaitMs, startedAt - receivedAtMs);
367 }
368 batch = yield* readPage(batch.at(-1)!.seq, busyHooks);
369 }
370 if (!(yield* hasPending)) {
371 yield* resetFailures;
372 yield* annotateRun("drained");
373 return null;
374 }
375 // Everything left this run was busy: give those tasks a moment to drain.
376 const now = yield* Clock.currentTimeMillis;
377 const idle = delivered === 0 && busyHooks.size > 0;
378 yield* annotateRun(idle ? "busy" : "more_pending");
379 return idle ? now + BUSY_RETRY_MS : now;
380});