import type { Provenance, QueryRow } from "@/lib/types"; import { provenance, queryRow } from "@/features/queries/provenanceFixture"; import { RING_CAPACITY, mergeGap, pushRow, summaryOf, type LiveRow } from "./ringBuffer"; function streamed(key: number, ts: number, domain: string, sections: Parameters[0] = {}): LiveRow { return { kind: "streamed", key, event: provenance({ ...sections, request: { time: ts, domain, ...sections.request }, route: { upstream: "udp://9.9.9.9:53", ...sections.route }, }), }; } function fetchedRow(id: number, ts: number, domain: string, overrides: Partial = {}): QueryRow { return queryRow(id, { ts, domain, upstream: "udp://9.9.9.9:53", ...overrides }); } function counter(start = 100): () => number { let n = start; return () => ++n; } function domains(rows: LiveRow[]): string[] { return rows.map((row) => summaryOf(row).domain); } describe("summaryOf", () => { test("a streamed frame projects every summary field from the provenance it carries", () => { const event: Provenance = provenance({ request: { time: 1700, domain: "ads.example", client: "192.0.2.11", qtype: 28 }, policy: { action: "block", reason: "blocklist_wildcard" }, route: { kind: "blocked", upstream: "" }, response: { duration_us: 42 }, }); expect(summaryOf({ kind: "streamed", key: 1, event })).toEqual({ id: null, ts: 1700, domain: "ads.example", client_ip: "192.0.2.11", qtype: 28, blocked: true, policy_reason: "blocklist_wildcard", rcode: 0, route_kind: "blocked", response_time_us: 42, cache_hit: null, upstream: "", }); }); test("a recovered row projects its stored fields and keeps its id", () => { const row = queryRow(77, { domain: "news.example", cache_hit: true, policy_reason: "rule_allow_exact" }); expect(summaryOf({ kind: "recovered", key: 2, row })).toMatchObject({ id: 77, domain: "news.example", cache_hit: true, policy_reason: "rule_allow_exact", }); }); /** * The guard the discriminated union exists for: a field added to the wire * DTO must be either projected into the summary or consciously left to the * detail page. A silent addition fails here rather than going unrendered. */ test("every provenance field is either projected or knowingly detail-only", () => { const projected = [ "request.time", "request.domain", "request.client", "request.qtype", "policy.action", "policy.reason", "route.kind", "route.upstream", "response.duration_us", ]; const detailOnly = [ "request.qclass", "group.id", "group.name", "policy.matched", "policy.source_id", "policy.source_name", "rewrites.cname_target", "rewrites.safe_search_target", "route.forward_zone", "response.rcode", ]; const leaves = Object.entries(provenance()).flatMap(([section, fields]) => Object.keys(fields as Record).map((field) => `${section}.${field}`), ); expect(leaves.sort()).toEqual([...projected, ...detailOnly].sort()); }); }); describe("pushRow", () => { test("prepends newest-first", () => { let rows: LiveRow[] = []; rows = pushRow(rows, streamed(1, 10, "a.example")); rows = pushRow(rows, streamed(2, 11, "b.example")); expect(domains(rows)).toEqual(["b.example", "a.example"]); }); test("drops the oldest beyond capacity", () => { let rows: LiveRow[] = []; for (let i = 0; i < 5; i++) rows = pushRow(rows, streamed(i, i, `d${i}.example`), 3); expect(rows).toHaveLength(3); expect(rows.map((r) => r.key)).toEqual([4, 3, 2]); }); test("default capacity is 500", () => { let rows: LiveRow[] = []; for (let i = 0; i < RING_CAPACITY + 10; i++) rows = pushRow(rows, streamed(i, i, "x.example")); expect(rows).toHaveLength(RING_CAPACITY); }); }); describe("mergeGap", () => { test("skips rows already in the buffer and counts only new ones", () => { const buffer = [streamed(2, 100, "seen.example"), streamed(1, 99, "old.example")]; const fetched = [ fetchedRow(30, 102, "gap2.example"), fetchedRow(29, 101, "gap1.example"), fetchedRow(28, 100, "seen.example"), ]; const { rows, missed } = mergeGap(buffer, fetched, counter()); expect(missed).toBe(2); expect(domains(rows)).toEqual(["gap2.example", "gap1.example", "seen.example", "old.example"]); }); test("no additions returns the buffer unchanged with missed 0", () => { const buffer = [streamed(1, 100, "seen.example")]; const { rows, missed } = mergeGap(buffer, [fetchedRow(5, 100, "seen.example")], counter()); expect(missed).toBe(0); expect(rows).toBe(buffer); }); /** * A household repeats itself: one client, one name, three lookups inside the * same second. The stream delivered one of them before the connection broke, * so the gap fetch must recover the other two rather than let the one row in * the buffer stand for all three. */ test("repeated identical queries drop only as many rows as the buffer already holds", () => { const buffer = [streamed(1, 100, "dup.example")]; const fetched = [ fetchedRow(12, 100, "dup.example"), fetchedRow(11, 100, "dup.example"), fetchedRow(10, 100, "dup.example"), ]; const { rows, missed } = mergeGap(buffer, fetched, counter()); expect(missed).toBe(2); expect(domains(rows)).toEqual(["dup.example", "dup.example", "dup.example"]); const recoveredIds = rows.flatMap((row) => (row.kind === "recovered" ? [row.row.id] : [])); expect(new Set(recoveredIds).size).toBe(2); }); test("a gap fetch that repeats the whole buffer adds nothing", () => { const buffer = [streamed(2, 100, "dup.example"), streamed(1, 100, "dup.example")]; const fetched = [fetchedRow(12, 100, "dup.example"), fetchedRow(11, 100, "dup.example")]; const { rows, missed } = mergeGap(buffer, fetched, counter()); expect(missed).toBe(0); expect(rows).toBe(buffer); }); /** * Two queries of the same name from the same client in the same second are * still separate facts when any stored column differs — the record type or * class, the response code, the policy that decided them, how long they took, * the route taken. The gap fetch here returns the differing row *first* and * the one the buffer already holds second, so an identity blind to the column * would let the differing row consume the buffered occurrence: the buffered * query would come back duplicated and the other would vanish, at an * unchanged `missed`. Order is what exposes that — the count alone is 1 * either way. * * `blocked` and `cache_hit` have no case of their own: the server derives * them from `policy_action` and `route_kind`, so they cannot differ while * everything else holds, and the two columns they follow are covered here. */ test.each([ { column: "qtype", sections: { request: { qtype: 1 } }, held: { qtype: 1 }, differing: { qtype: 28 } }, { column: "qclass", sections: { request: { qclass: 1 } }, held: { qclass: 1 }, differing: { qclass: 3 } }, { column: "rcode", sections: { response: { rcode: 0 } }, held: { rcode: 0 }, differing: { rcode: 2 } }, { column: "response_time_us", sections: { response: { duration_us: 1234 } }, held: { response_time_us: 1234 }, differing: { response_time_us: 9999 }, }, { column: "route_kind", sections: { route: { kind: "upstream" } }, held: { route_kind: "upstream" }, differing: { route_kind: "forward_zone" }, }, { column: "policy_action", sections: { policy: { action: "allow" } }, held: { policy_action: "allow" }, differing: { policy_action: "not_evaluated" }, }, { column: "policy_reason", sections: { policy: { reason: "no_match" } }, held: { policy_reason: "no_match" }, differing: { policy_reason: "rule_allow_exact" }, }, ] satisfies readonly { column: string; sections: Parameters[0]; held: Partial; differing: Partial; }[])("rows differing only in $column survive the gap merge", ({ sections, held, differing }) => { const buffer = [streamed(1, 100, "dual.example", sections)]; const fetched = [fetchedRow(6, 100, "dual.example", differing), fetchedRow(5, 100, "dual.example", held)]; const { rows, missed } = mergeGap(buffer, fetched, counter()); expect(missed).toBe(1); expect(rows).toEqual([{ kind: "recovered", key: expect.any(Number), row: fetched[0] }, buffer[0]]); }); test("recovered rows keep their id and take a fresh key", () => { const { rows } = mergeGap([], [fetchedRow(77, 100, "gap.example")], counter(200)); const recovered = rows[0]; expect(recovered?.key).toBe(201); expect(recovered?.kind).toBe("recovered"); expect(recovered !== undefined && recovered.kind === "recovered" ? recovered.row.id : null).toBe(77); }); test("result is capped at capacity, keeping the newest", () => { const buffer = [streamed(3, 300, "live.example")]; const fetched = [fetchedRow(2, 302, "g2.example"), fetchedRow(1, 301, "g1.example")]; const { rows, missed } = mergeGap(buffer, fetched, counter(), 2); expect(missed).toBe(2); expect(domains(rows)).toEqual(["g2.example", "g1.example"]); }); test("merged rows stay sorted newest-first by ts", () => { const buffer = [streamed(4, 105, "after-reopen.example"), streamed(3, 100, "before.example")]; const fetched = [fetchedRow(9, 103, "gap.example")]; const { rows } = mergeGap(buffer, fetched, counter()); expect(rows.map((row) => summaryOf(row).ts)).toEqual([105, 103, 100]); }); });