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
11const HOOK_INBOX_TTL_MS = 24 * 60 * 60 * 1000;
12
13const HOOK_INBOX_MAX_REQUESTS = 1_000;
14
15export const HOOK_INBOX_MAX_BYTES = 50 * 1_048_576;
16// The relay cannot check tokens; isolate each hook's junk with its own cap.
17export const HOOK_INBOX_MAX_PER_HOOK = 100;
18
19const DELIVERIES_PER_RUN = 20;
20
21const ROWS_READ_PER_PAGE = 50;
22
23const BUSY_RETRY_MS = 30_000;
24const MAX_RETRY_DELAY = Duration.minutes(10);
25
26const MAX_COUNTED_FAILURES = 25;
27
28/** Cloudflare can take minutes to route a new tunnel; retry every 10 s for three minutes before backing off. */
29const retrySchedule = Schedule.spaced("10 seconds").pipe(
30 Schedule.upTo({ times: 18 }),
31 Schedule.concat(
32 Schedule.exponential("30 seconds").pipe(
33 Schedule.modifyDelay(({ duration }) =>
34 Effect.succeed(Duration.min(Duration.fromInputUnsafe(duration), MAX_RETRY_DELAY)),
35 ),
36 ),
37 ),
38);
39
40export interface HeldHook {
41 readonly id: string;
42 readonly receivedAt: string;
43 readonly method: string;
44 /** Path segments exactly as the sender sent them; the environment decodes them. */
45 readonly rawHookId: string;
46 readonly rawToken: string;
47 /** The decoded hook id, so every spelling of one hook shares its cap. */
48 readonly hookKey: string;
49
50 readonly query: string;
51 readonly headers: Readonly<Record<string, string>>;
52 readonly body: Uint8Array;
53}
54
55export type DeliveryOutcome = "delivered" | "busy" | "unreachable";
56
57const retryDelayMs = Effect.fn("HookInboxStore.retryDelayMs")(function* (failures: number) {
58 const step = yield* Schedule.toStep(retrySchedule);
59 let delay = MAX_RETRY_DELAY;
60 for (
61 let attempt = 0;
62 attempt < Math.min(Math.max(1, failures), MAX_COUNTED_FAILURES);
63 attempt++
64 ) {
65 const next = yield* step(0, undefined).pipe(
66 Effect.map(([, duration]) => Option.some(duration)),
67
68 Pull.catchDone(() => Effect.succeedNone),
69 );
70 if (Option.isNone(next)) break;
71 delay = next.value;
72 }
73 return Duration.toMillis(delay);
74});
75
76const HeadersJson = Schema.fromJsonString(Schema.Record(Schema.String, Schema.String));
77const encodeHeaders = Schema.encodeSync(HeadersJson);
78const decodeHeaders = Schema.decodeUnknownOption(HeadersJson);
79
80interface HeldHookRow {
81 readonly seq: number;
82 readonly id: string;
83 readonly received_at: string;
84 readonly method: string;
85 readonly raw_hook_id: string;
86 readonly raw_token: string;
87 readonly hook_key: string;
88 readonly query: string;
89 readonly headers: string;
90 readonly body: Uint8Array;
91}
92
93interface TargetRow {
94 readonly base_url: string;
95 readonly failures: number;
96}
97
98const iso = (epochMillis: number) => DateTime.formatIso(DateTime.makeUnsafe(epochMillis));
99
100export const migrate = Effect.gen(function* () {
101 const sql = yield* SqlClient.SqlClient;
102 yield* sql`
103 CREATE TABLE IF NOT EXISTS held_hooks (
104 seq INTEGER PRIMARY KEY AUTOINCREMENT,
105 id TEXT NOT NULL UNIQUE,
106 received_at TEXT NOT NULL,
107 method TEXT NOT NULL,
108 raw_hook_id TEXT NOT NULL,
109 raw_token TEXT NOT NULL,
110 hook_key TEXT NOT NULL,
111 query TEXT NOT NULL,
112 headers TEXT NOT NULL,
113 body BLOB NOT NULL
114 )
115 `;
116 yield* sql`CREATE INDEX IF NOT EXISTS held_hooks_hook ON held_hooks (hook_key)`;
117
118 yield* sql`
119 CREATE TABLE IF NOT EXISTS held_hooks_target (
120 id INTEGER PRIMARY KEY CHECK (id = 1),
121 base_url TEXT NOT NULL,
122 failures INTEGER NOT NULL DEFAULT 0
123 )
124 `;
125});
126
127const setTarget = Effect.fn("HookInboxStore.setTarget")(function* (
128 baseUrl: string,
129 options: { readonly resetFailures: boolean },
130) {
131 const sql = yield* SqlClient.SqlClient;
132 yield* options.resetFailures
133 ? sql`
134 INSERT INTO held_hooks_target (id, base_url) VALUES (1, ${baseUrl})
135 ON CONFLICT (id) DO UPDATE SET base_url = excluded.base_url, failures = 0
136 `
137 : sql`
138 INSERT INTO held_hooks_target (id, base_url) VALUES (1, ${baseUrl})
139 ON CONFLICT (id) DO UPDATE SET base_url = excluded.base_url
140 `;
141});
142
143const readTarget = Effect.gen(function* () {
144 const sql = yield* SqlClient.SqlClient;
145 const rows = yield* sql<TargetRow>`SELECT base_url, failures FROM held_hooks_target WHERE id = 1`;
146 return rows[0] ?? null;
147});
148
149const resetFailures = Effect.gen(function* () {
150 const sql = yield* SqlClient.SqlClient;
151 yield* sql`UPDATE held_hooks_target SET failures = 0 WHERE id = 1`;
152});
153
154// Increment in place to preserve wake resets made during an in-flight request.
155const countFailure = Effect.gen(function* () {
156 const sql = yield* SqlClient.SqlClient;
157 const rows = yield* sql<{ readonly failures: number }>`
158 UPDATE held_hooks_target SET failures = min(failures + 1, ${MAX_COUNTED_FAILURES})
159 WHERE id = 1 RETURNING failures
160 `;
161 return rows[0]?.failures ?? 1;
162});
163
164const hasPending = Effect.gen(function* () {
165 const sql = yield* SqlClient.SqlClient;
166 const rows = yield* sql<{ readonly id: string }>`SELECT id FROM held_hooks LIMIT 1`;
167 return rows.length > 0;
168});
169
170/** First attempt's due time, or null if the full inbox stored nothing. */
171export const hold = Effect.fn("HookInboxStore.hold")(function* (hook: HeldHook, baseUrl: string) {
172 const sql = yield* SqlClient.SqlClient;
173 // One statement enforces caps across concurrent requests.
174 const inserted = yield* sql<{ readonly id: string }>`
175 INSERT INTO held_hooks
176 (id, received_at, method, raw_hook_id, raw_token, hook_key, query, headers, body)
177 SELECT ${hook.id}, ${hook.receivedAt}, ${hook.method}, ${hook.rawHookId}, ${hook.rawToken},
178 ${hook.hookKey}, ${hook.query}, ${encodeHeaders(hook.headers)}, ${hook.body}
179 WHERE (SELECT count(*) FROM held_hooks) < ${HOOK_INBOX_MAX_REQUESTS}
180 AND (SELECT count(*) FROM held_hooks WHERE hook_key = ${hook.hookKey})
181 < ${HOOK_INBOX_MAX_PER_HOOK}
182 AND (SELECT coalesce(sum(length(body)), 0) FROM held_hooks) + ${hook.body.byteLength}
183 <= ${HOOK_INBOX_MAX_BYTES}
184 ON CONFLICT (id) DO NOTHING
185 RETURNING id
186 `;
187 if (inserted.length === 0) {
188 yield* Effect.annotateCurrentSpan({ "relay.inbox.refused": yield* refusalReason(hook) });
189 return null;
190 }
191 yield* setTarget(baseUrl, { resetFailures: false });
192 const target = yield* readTarget;
193 const backlog = yield* backlogSize;
194 yield* Effect.annotateCurrentSpan({
195 "relay.inbox.held_count": backlog.count,
196 "relay.inbox.held_bytes": backlog.bytes,
197 });
198 return (yield* Clock.currentTimeMillis) + (yield* retryDelayMs(target?.failures ?? 0));
199});
200
201const backlogSize = Effect.gen(function* () {
202 const sql = yield* SqlClient.SqlClient;
203 const rows = yield* sql<{ readonly count: number; readonly bytes: number }>`
204 SELECT count(*) AS count, coalesce(sum(length(body)), 0) AS bytes FROM held_hooks
205 `;
206 return rows[0] ?? { count: 0, bytes: 0 };
207});
208
209const refusalReason = Effect.fn("HookInboxStore.refusalReason")(function* (hook: HeldHook) {
210 const sql = yield* SqlClient.SqlClient;
211 const existing = yield* sql<{ readonly id: string }>`
212 SELECT id FROM held_hooks WHERE id = ${hook.id}
213 `;
214 if (existing.length > 0) return "already_held";
215 const backlog = yield* backlogSize;
216 if (backlog.count >= HOOK_INBOX_MAX_REQUESTS) return "max_requests";
217 if (backlog.bytes + hook.body.byteLength > HOOK_INBOX_MAX_BYTES) return "max_bytes";
218 return "max_per_hook";
219});
220
221export const wake = Effect.fn("HookInboxStore.wake")(function* (baseUrl: string) {
222 if (!(yield* hasPending)) return false;
223 yield* setTarget(baseUrl, { resetFailures: true });
224 return true;
225});
226
227export const clear = Effect.gen(function* () {
228 const sql = yield* SqlClient.SqlClient;
229 yield* sql`DELETE FROM held_hooks`;
230 yield* sql`DELETE FROM held_hooks_target`;
231});
232
233export const deliverDue = Effect.fn("HookInboxStore.deliverDue")(function* <R>(
234 send: (baseUrl: string, hook: HeldHook) => Effect.Effect<DeliveryOutcome, never, R>,
235) {
236 const sql = yield* SqlClient.SqlClient;
237 const startedAt = yield* Clock.currentTimeMillis;
238 const expired = yield* sql<{ readonly id: string }>`
239 DELETE FROM held_hooks WHERE received_at < ${iso(startedAt - HOOK_INBOX_TTL_MS)}
240 RETURNING id
241 `;
242 const target = yield* readTarget;
243
244 const readPage = (afterSeq: number, skip: ReadonlySet<string>) =>
245 sql<HeldHookRow>`
246 SELECT seq, id, received_at, method, raw_hook_id, raw_token, hook_key, query, headers, body
247 FROM held_hooks
248 WHERE seq > ${afterSeq}
249 ${skip.size === 0 ? sql`` : sql`AND hook_key NOT IN ${sql.in([...skip])}`}
250 ORDER BY seq LIMIT ${ROWS_READ_PER_PAGE}
251 `;
252 let batch = yield* readPage(0, new Set());
253 if (batch.length === 0 || target === null) {
254 if (target === null) yield* sql`DELETE FROM held_hooks`;
255 else yield* resetFailures;
256 yield* Effect.annotateCurrentSpan({ "relay.inbox.expired": expired.length });
257 return null;
258 }
259
260 let failures = target.failures;
261 let sent = 0;
262 let delivered = 0;
263 let unreadable = 0;
264 let longestWaitMs = 0;
265 const busyHooks = new Set<string>();
266
267 // Delivery results and counts belong to the enclosing deliverDue span.
268 const annotateRun = Effect.fnUntraced(function* (result: string) {
269 const backlog = yield* backlogSize;
270 yield* Effect.annotateCurrentSpan({
271 "relay.inbox.run_result": result,
272 "relay.inbox.sent": sent,
273 "relay.inbox.delivered": delivered,
274 "relay.inbox.busy_hooks": busyHooks.size,
275 "relay.inbox.expired": expired.length,
276 "relay.inbox.unreadable": unreadable,
277 "relay.inbox.consecutive_failures": failures,
278 "relay.inbox.longest_wait_ms": longestWaitMs,
279 "relay.inbox.held_count": backlog.count,
280 "relay.inbox.held_bytes": backlog.bytes,
281 });
282 });
283 const stillHeld = (seq: number) =>
284 sql<{ readonly seq: number }>`SELECT seq FROM held_hooks WHERE seq = ${seq}`.pipe(
285 Effect.map((rows) => rows.length > 0),
286 );
287 // Exhaust non-busy requests before stopping so busy hooks cannot hide others' backlog.
288 pages: while (batch.length > 0) {
289 for (const row of batch) {
290 if (sent === DELIVERIES_PER_RUN) break pages;
291
292 if (busyHooks.has(row.hook_key)) continue;
293 // Clear can discard this prefetched page while an upstream send is in flight.
294 if (!(yield* stillHeld(row.seq))) {
295 yield* annotateRun("cleared");
296 return null;
297 }
298 const headers = decodeHeaders(row.headers);
299 if (Option.isNone(headers)) {
300 yield* sql`DELETE FROM held_hooks WHERE id = ${row.id}`;
301 unreadable += 1;
302 continue;
303 }
304 sent += 1;
305 const outcome = yield* send(target.base_url, {
306 id: row.id,
307 receivedAt: row.received_at,
308 method: row.method,
309 rawHookId: row.raw_hook_id,
310 rawToken: row.raw_token,
311 hookKey: row.hook_key,
312 query: row.query,
313 headers: headers.value,
314 body: row.body,
315 });
316 // A cleared request must neither advance this run nor schedule a retry.
317 if (!(yield* stillHeld(row.seq))) {
318 yield* annotateRun("cleared");
319 return null;
320 }
321 if (outcome === "unreachable") {
322 failures = yield* countFailure;
323 yield* annotateRun("unreachable");
324 return (yield* Clock.currentTimeMillis) + (yield* retryDelayMs(failures));
325 }
326 if (failures !== 0) {
327 failures = 0;
328 yield* resetFailures;
329 }
330 if (outcome === "busy") {
331 busyHooks.add(row.hook_key);
332 continue;
333 }
334 yield* sql`DELETE FROM held_hooks WHERE id = ${row.id}`;
335 delivered += 1;
336 const receivedAtMs = DateTime.toEpochMillis(DateTime.makeUnsafe(row.received_at));
337 longestWaitMs = Math.max(longestWaitMs, startedAt - receivedAtMs);
338 }
339 batch = yield* readPage(batch.at(-1)!.seq, busyHooks);
340 }
341 if (!(yield* hasPending)) {
342 yield* resetFailures;
343 yield* annotateRun("drained");
344 return null;
345 }
346
347 const now = yield* Clock.currentTimeMillis;
348 const idle = delivered === 0 && busyHooks.size > 0;
349 yield* annotateRun(idle ? "busy" : "more_pending");
350 return idle ? now + BUSY_RETRY_MS : now;
351});