1import { describe, expect, it } from "@effect/vitest";
2import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient";
3import * as Clock from "effect/Clock";
4import * as DateTime from "effect/DateTime";
5import * as Duration from "effect/Duration";
6import * as Effect from "effect/Effect";
7import * as SqlClient from "effect/sql/SqlClient";
8import * as TestClock from "effect/testing/TestClock";
9
10import * as HookInboxStore from "./HookInboxStore.ts";
11
12const BASE_URL = "https://env.example.test/";
13
14const hook = (id: string, overrides: Partial<HookInboxStore.HeldHook> = {}) =>
15 Effect.map(Clock.currentTimeMillis, (now): HookInboxStore.HeldHook => ({
16 id,
17 receivedAt: DateTime.formatIso(DateTime.makeUnsafe(now)),
18 method: "POST",
19 rawHookId: "hook-1",
20 hookKey: "hook-1",
21 rawToken: "tok%2Fen",
22 query: "a=1",
23 headers: { "content-type": "application/json", "x-sig": "s" },
24 body: new Uint8Array([0, 255, 10]),
25 ...overrides,
26 }));
27
28/** Delivers with `outcome` per request and records what the environment was sent. */
29const deliverer = (outcome: (hook: HookInboxStore.HeldHook) => HookInboxStore.DeliveryOutcome) => {
30 const sent: Array<{ readonly baseUrl: string; readonly hook: HookInboxStore.HeldHook }> = [];
31 const send = (baseUrl: string, held: HookInboxStore.HeldHook) =>
32 Effect.sync(() => {
33 sent.push({ baseUrl, hook: held });
34 return outcome(held);
35 });
36 return { sent, send };
37};
38
39const withInbox = <A, E, R>(effect: Effect.Effect<A, E, R>) =>
40 HookInboxStore.migrate.pipe(
41 Effect.andThen(effect),
42 Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" })),
43 );
44
45describe("HookInboxStore", () => {
46 it.effect("delivers held requests oldest first, byte for byte", () =>
47 withInbox(
48 Effect.gen(function* () {
49 const first = yield* hook("first");
50 expect(yield* HookInboxStore.hold(first, BASE_URL)).not.toBeNull();
51 yield* HookInboxStore.hold(yield* hook("second"), BASE_URL);
52 const { sent, send } = deliverer(() => "delivered");
53 expect(yield* HookInboxStore.deliverDue(send)).toBeNull();
54 expect(sent.map((entry) => entry.hook.id)).toEqual(["first", "second"]);
55 expect(sent[0]).toEqual({ baseUrl: BASE_URL, hook: first });
56 // Delivered requests are gone.
57 expect(yield* HookInboxStore.deliverDue(send)).toBeNull();
58 expect(sent).toHaveLength(2);
59 }),
60 ),
61 );
62
63 it.effect("keeps a request the environment did not take and backs off", () =>
64 withInbox(
65 Effect.gen(function* () {
66 yield* HookInboxStore.hold(yield* hook("first"), BASE_URL);
67 yield* HookInboxStore.hold(yield* hook("second"), BASE_URL);
68 const offline = deliverer(() => "unreachable");
69 const now = yield* Clock.currentTimeMillis;
70 // Stops at the first failure, so order is kept.
71 expect(yield* HookInboxStore.deliverDue(offline.send)).toBe(now + 10_000);
72 expect(offline.sent.map((entry) => entry.hook.id)).toEqual(["first"]);
73 expect(yield* HookInboxStore.deliverDue(offline.send)).toBe(now + 10_000);
74
75 const online = deliverer(() => "delivered");
76 expect(yield* HookInboxStore.deliverDue(online.send)).toBeNull();
77 expect(online.sent.map((entry) => entry.hook.id)).toEqual(["first", "second"]);
78 }),
79 ),
80 );
81
82 it.effect("lets other hooks through while one hook's environment is busy", () =>
83 withInbox(
84 Effect.gen(function* () {
85 const stuck = { rawHookId: "stuck", hookKey: "stuck" };
86 yield* HookInboxStore.hold(yield* hook("stuck-1", stuck), BASE_URL);
87 yield* HookInboxStore.hold(yield* hook("other-1"), BASE_URL);
88 yield* HookInboxStore.hold(yield* hook("stuck-2", stuck), BASE_URL);
89 const busy = deliverer((held) => (held.hookKey === "stuck" ? "busy" : "delivered"));
90 const now = yield* Clock.currentTimeMillis;
91 // The other hook is delivered; the busy hook keeps its order and runs again soon.
92 expect(yield* HookInboxStore.deliverDue(busy.send)).toBe(now);
93 expect(busy.sent.map((entry) => entry.hook.id)).toEqual(["stuck-1", "other-1"]);
94 // Only the busy hook is left, so the next run waits rather than spinning.
95 expect(yield* HookInboxStore.deliverDue(busy.send)).toBe(now + 30_000);
96
97 const drained = deliverer(() => "delivered");
98 expect(yield* HookInboxStore.deliverDue(drained.send)).toBeNull();
99 expect(drained.sent.map((entry) => entry.hook.id)).toEqual(["stuck-1", "stuck-2"]);
100 }),
101 ),
102 );
103
104 it.effect("reaches a hook queued behind busy hooks' full backlogs in the same run", () =>
105 withInbox(
106 Effect.gen(function* () {
107 for (const name of ["stuck-a", "stuck-b"]) {
108 const stuck = { rawHookId: name, hookKey: name };
109 for (let index = 0; index < HookInboxStore.HOOK_INBOX_MAX_PER_HOOK; index++) {
110 yield* HookInboxStore.hold(yield* hook(`${name}-${index}`, stuck), BASE_URL);
111 }
112 }
113 yield* HookInboxStore.hold(yield* hook("other-1"), BASE_URL);
114 const busy = deliverer((held) => (held.hookKey.startsWith("stuck") ? "busy" : "delivered"));
115 yield* HookInboxStore.deliverDue(busy.send);
116 expect(busy.sent.map((entry) => entry.hook.id)).toEqual([
117 "stuck-a-0",
118 "stuck-b-0",
119 "other-1",
120 ]);
121 }),
122 ),
123 );
124
125 it.effect("drops a held request it cannot read instead of stalling on it", () =>
126 withInbox(
127 Effect.gen(function* () {
128 yield* HookInboxStore.hold(yield* hook("broken"), BASE_URL);
129 yield* HookInboxStore.hold(yield* hook("fine"), BASE_URL);
130 const sql = yield* SqlClient.SqlClient;
131 yield* sql`UPDATE held_hooks SET headers = 'not json' WHERE id = 'broken'`;
132 const { sent, send } = deliverer(() => "delivered");
133 expect(yield* HookInboxStore.deliverDue(send)).toBeNull();
134 expect(sent.map((entry) => entry.hook.id)).toEqual(["fine"]);
135 }),
136 ),
137 );
138
139 it.effect("retries every 10 s for 3 minutes, then backs off to 10 minutes", () =>
140 Effect.gen(function* () {
141 const delays = yield* Effect.forEach(
142 [1, 18, 19, 20, 21, 23, 24, 40],
143 HookInboxStore.retryDelayMs,
144 );
145 expect(delays).toEqual([10_000, 10_000, 30_000, 60_000, 120_000, 480_000, 600_000, 600_000]);
146 }),
147 );
148
149 it.effect("waking resets the backoff and reports whether anything waits", () =>
150 withInbox(
151 Effect.gen(function* () {
152 expect(yield* HookInboxStore.wake(BASE_URL)).toBe(false);
153 yield* HookInboxStore.hold(yield* hook("first"), "https://old.example.test/");
154 const offline = deliverer(() => "unreachable");
155 yield* HookInboxStore.deliverDue(offline.send);
156 yield* HookInboxStore.deliverDue(offline.send);
157
158 // Past the fast phase: the environment was away a while.
159 for (let failure = 0; failure < 20; failure++) {
160 yield* HookInboxStore.deliverDue(offline.send);
161 }
162 const now = yield* Clock.currentTimeMillis;
163 expect(yield* HookInboxStore.deliverDue(offline.send)).toBeGreaterThan(now + 60_000);
164
165 // A wake means its tunnel just connected, which Cloudflare may not
166 // route to for a few minutes: back to retrying every 10 s, sent to
167 // where the environment is now.
168 expect(yield* HookInboxStore.wake(BASE_URL)).toBe(true);
169 expect(yield* HookInboxStore.deliverDue(offline.send)).toBe(now + 10_000);
170 expect(offline.sent.at(-1)?.baseUrl).toBe(BASE_URL);
171 }),
172 ),
173 );
174
175 it.effect("starts the schedule over once the inbox drains", () =>
176 withInbox(
177 Effect.gen(function* () {
178 yield* HookInboxStore.hold(yield* hook("old"), BASE_URL);
179 const offline = deliverer(() => "unreachable");
180 for (let failure = 0; failure < 30; failure++) {
181 yield* HookInboxStore.deliverDue(offline.send);
182 }
183 // Nothing got through before the request expired.
184 yield* TestClock.adjust(Duration.hours(25));
185 expect(yield* HookInboxStore.deliverDue(offline.send)).toBeNull();
186
187 const now = yield* Clock.currentTimeMillis;
188 expect(yield* HookInboxStore.hold(yield* hook("new"), BASE_URL)).toBe(now + 10_000);
189 }),
190 ),
191 );
192
193 it.effect("drops requests older than 24 hours", () =>
194 withInbox(
195 Effect.gen(function* () {
196 yield* HookInboxStore.hold(yield* hook("old"), BASE_URL);
197 yield* TestClock.adjust(Duration.hours(1));
198 yield* HookInboxStore.hold(yield* hook("new"), BASE_URL);
199 yield* TestClock.adjust(Duration.minutes(23 * 60 + 1));
200 const { sent, send } = deliverer(() => "delivered");
201 yield* HookInboxStore.deliverDue(send);
202 expect(sent.map((entry) => entry.hook.id)).toEqual(["new"]);
203 }),
204 ),
205 );
206
207 it.effect("refuses requests past the per-hook cap", () =>
208 withInbox(
209 Effect.gen(function* () {
210 for (let index = 0; index < HookInboxStore.HOOK_INBOX_MAX_PER_HOOK; index++) {
211 expect(yield* HookInboxStore.hold(yield* hook(`a-${index}`), BASE_URL)).not.toBeNull();
212 }
213 expect(yield* HookInboxStore.hold(yield* hook("one-too-many"), BASE_URL)).toBeNull();
214 // Another hook still has room.
215 const other = yield* hook("other", { rawHookId: "hook-2", hookKey: "hook-2" });
216 expect(yield* HookInboxStore.hold(other, BASE_URL)).not.toBeNull();
217 }),
218 ),
219 );
220
221 it.effect("refuses a request that would pass the byte cap", () =>
222 withInbox(
223 Effect.gen(function* () {
224 const big = new Uint8Array(HookInboxStore.HOOK_INBOX_MAX_BYTES - 10);
225 expect(
226 yield* HookInboxStore.hold(yield* hook("big", { body: big }), BASE_URL),
227 ).not.toBeNull();
228 const small = new Uint8Array(11);
229 expect(
230 yield* HookInboxStore.hold(
231 yield* hook("small", { rawHookId: "x", hookKey: "x", body: small }),
232 BASE_URL,
233 ),
234 ).toBeNull();
235 }),
236 ),
237 );
238
239 it.effect("stores a request id once", () =>
240 withInbox(
241 Effect.gen(function* () {
242 const first = yield* hook("same");
243 expect(yield* HookInboxStore.hold(first, BASE_URL)).not.toBeNull();
244 expect(yield* HookInboxStore.hold(first, BASE_URL)).toBeNull();
245 }),
246 ),
247 );
248
249 it.effect("clear drops everything held", () =>
250 withInbox(
251 Effect.gen(function* () {
252 yield* HookInboxStore.hold(yield* hook("first"), BASE_URL);
253 yield* HookInboxStore.clear;
254 const { sent, send } = deliverer(() => "delivered");
255 expect(yield* HookInboxStore.deliverDue(send)).toBeNull();
256 expect(sent).toHaveLength(0);
257 }),
258 ),
259 );
260});