Open History to see the last commit
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 Deferred from "effect/Deferred";
6import * as Effect from "effect/Effect";
7import * as Fiber from "effect/Fiber";
8import * as SqlClient from "effect/sql/SqlClient";
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
28const deliverer = (outcome: (hook: HookInboxStore.HeldHook) => HookInboxStore.DeliveryOutcome) => {
29 const sent: Array<{ readonly baseUrl: string; readonly hook: HookInboxStore.HeldHook }> = [];
30 const send = (baseUrl: string, held: HookInboxStore.HeldHook) =>
31 Effect.sync(() => {
32 sent.push({ baseUrl, hook: held });
33 return outcome(held);
34 });
35 return { sent, send };
36};
37
38const withInbox = <A, E, R>(effect: Effect.Effect<A, E, R>) =>
39 HookInboxStore.migrate.pipe(
40 Effect.andThen(effect),
41 Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" })),
42 );
43
44describe("HookInboxStore", () => {
45 it.effect.each(["delivered", "busy", "unreachable"] as const)(
46 "stops a prefetched batch cleared during a %s send",
47 (outcome) =>
48 withInbox(
49 Effect.gen(function* () {
50 for (const id of ["first", "second", "third"]) {
51 yield* HookInboxStore.hold(yield* hook(id), BASE_URL);
52 }
53 const started = yield* Deferred.make<void>();
54 const released = yield* Deferred.make<void>();
55 const sent: Array<string> = [];
56 const run = yield* HookInboxStore.deliverDue((_baseUrl, held) =>
57 Effect.gen(function* () {
58 sent.push(held.id);
59 if (held.id === "first") {
60 yield* Deferred.succeed(started, undefined);
61 yield* Deferred.await(released);
62 }
63 return outcome;
64 }),
65 ).pipe(Effect.forkChild);
66 yield* Deferred.await(started);
67 yield* HookInboxStore.clear;
68 yield* Deferred.succeed(released, undefined);
69 const nextAt = yield* Fiber.join(run);
70 expect(sent).toEqual(["first"]);
71 expect(nextAt).toBeNull();
72 const sql = yield* SqlClient.SqlClient;
73 expect(yield* sql`SELECT id FROM held_hooks`).toEqual([]);
74 expect(yield* sql`SELECT id FROM held_hooks_target`).toEqual([]);
75 }),
76 ),
77 );
78
79 it.effect("delivers held requests oldest first, byte for byte", () =>
80 withInbox(
81 Effect.gen(function* () {
82 const first = yield* hook("first");
83 expect(yield* HookInboxStore.hold(first, BASE_URL)).not.toBeNull();
84 yield* HookInboxStore.hold(yield* hook("second"), BASE_URL);
85 const { sent, send } = deliverer(() => "delivered");
86 expect(yield* HookInboxStore.deliverDue(send)).toBeNull();
87 expect(sent.map((entry) => entry.hook.id)).toEqual(["first", "second"]);
88 expect(sent[0]).toEqual({ baseUrl: BASE_URL, hook: first });
89 expect(yield* HookInboxStore.deliverDue(send)).toBeNull();
90 expect(sent).toHaveLength(2);
91 }),
92 ),
93 );
94
95 it.effect("keeps a request the environment did not take and backs off", () =>
96 withInbox(
97 Effect.gen(function* () {
98 yield* HookInboxStore.hold(yield* hook("first"), BASE_URL);
99 yield* HookInboxStore.hold(yield* hook("second"), BASE_URL);
100 const offline = deliverer(() => "unreachable");
101 const now = yield* Clock.currentTimeMillis;
102 expect(yield* HookInboxStore.deliverDue(offline.send)).toBe(now + 10_000);
103 expect(offline.sent.map((entry) => entry.hook.id)).toEqual(["first"]);
104 expect(yield* HookInboxStore.deliverDue(offline.send)).toBe(now + 10_000);
105
106 const online = deliverer(() => "delivered");
107 expect(yield* HookInboxStore.deliverDue(online.send)).toBeNull();
108 expect(online.sent.map((entry) => entry.hook.id)).toEqual(["first", "second"]);
109 }),
110 ),
111 );
112
113 it.effect("lets other hooks through while one hook's environment is busy", () =>
114 withInbox(
115 Effect.gen(function* () {
116 const stuck = { rawHookId: "stuck", hookKey: "stuck" };
117 yield* HookInboxStore.hold(yield* hook("stuck-1", stuck), BASE_URL);
118 yield* HookInboxStore.hold(yield* hook("other-1"), BASE_URL);
119 yield* HookInboxStore.hold(yield* hook("stuck-2", stuck), BASE_URL);
120 const busy = deliverer((held) => (held.hookKey === "stuck" ? "busy" : "delivered"));
121 const now = yield* Clock.currentTimeMillis;
122 expect(yield* HookInboxStore.deliverDue(busy.send)).toBe(now);
123 expect(busy.sent.map((entry) => entry.hook.id)).toEqual(["stuck-1", "other-1"]);
124 expect(yield* HookInboxStore.deliverDue(busy.send)).toBe(now + 30_000);
125
126 const drained = deliverer(() => "delivered");
127 expect(yield* HookInboxStore.deliverDue(drained.send)).toBeNull();
128 expect(drained.sent.map((entry) => entry.hook.id)).toEqual(["stuck-1", "stuck-2"]);
129 }),
130 ),
131 );
132
133 it.effect("reaches a hook queued behind busy hooks' full backlogs in the same run", () =>
134 withInbox(
135 Effect.gen(function* () {
136 for (const name of ["stuck-a", "stuck-b"]) {
137 const stuck = { rawHookId: name, hookKey: name };
138 for (let index = 0; index < HookInboxStore.HOOK_INBOX_MAX_PER_HOOK; index++) {
139 yield* HookInboxStore.hold(yield* hook(`${name}-${index}`, stuck), BASE_URL);
140 }
141 }
142 yield* HookInboxStore.hold(yield* hook("other-1"), BASE_URL);
143 const busy = deliverer((held) => (held.hookKey.startsWith("stuck") ? "busy" : "delivered"));
144 yield* HookInboxStore.deliverDue(busy.send);
145 expect(busy.sent.map((entry) => entry.hook.id)).toEqual([
146 "stuck-a-0",
147 "stuck-b-0",
148 "other-1",
149 ]);
150 }),
151 ),
152 );
153
154 it.effect("drops a held request it cannot read instead of stalling on it", () =>
155 withInbox(
156 Effect.gen(function* () {
157 yield* HookInboxStore.hold(yield* hook("broken"), BASE_URL);
158 yield* HookInboxStore.hold(yield* hook("fine"), BASE_URL);
159 const sql = yield* SqlClient.SqlClient;
160 yield* sql`UPDATE held_hooks SET headers = 'not json' WHERE id = 'broken'`;
161 const { sent, send } = deliverer(() => "delivered");
162 expect(yield* HookInboxStore.deliverDue(send)).toBeNull();
163 expect(sent.map((entry) => entry.hook.id)).toEqual(["fine"]);
164 }),
165 ),
166 );
167
168 it.effect("refuses requests past the per-hook cap", () =>
169 withInbox(
170 Effect.gen(function* () {
171 for (let index = 0; index < HookInboxStore.HOOK_INBOX_MAX_PER_HOOK; index++) {
172 expect(yield* HookInboxStore.hold(yield* hook(`a-${index}`), BASE_URL)).not.toBeNull();
173 }
174 expect(yield* HookInboxStore.hold(yield* hook("one-too-many"), BASE_URL)).toBeNull();
175 const other = yield* hook("other", { rawHookId: "hook-2", hookKey: "hook-2" });
176 expect(yield* HookInboxStore.hold(other, BASE_URL)).not.toBeNull();
177 }),
178 ),
179 );
180
181 it.effect("refuses a request that would pass the byte cap", () =>
182 withInbox(
183 Effect.gen(function* () {
184 const big = new Uint8Array(HookInboxStore.HOOK_INBOX_MAX_BYTES - 10);
185 expect(
186 yield* HookInboxStore.hold(yield* hook("big", { body: big }), BASE_URL),
187 ).not.toBeNull();
188 const small = new Uint8Array(11);
189 expect(
190 yield* HookInboxStore.hold(
191 yield* hook("small", { rawHookId: "x", hookKey: "x", body: small }),
192 BASE_URL,
193 ),
194 ).toBeNull();
195 }),
196 ),
197 );
198
199 it.effect("stores a request id once", () =>
200 withInbox(
201 Effect.gen(function* () {
202 const first = yield* hook("same");
203 expect(yield* HookInboxStore.hold(first, BASE_URL)).not.toBeNull();
204 expect(yield* HookInboxStore.hold(first, BASE_URL)).toBeNull();
205 }),
206 ),
207 );
208});