rename web/ to admin/, along with the web-named build and cli identifiers
This commit is contained in:
@@ -0,0 +1,86 @@
|
||||
import { act, fireEvent, render, screen } from "@testing-library/react";
|
||||
import type { LiveQueryEvent } from "@/lib/types";
|
||||
import { FakeEventSource } from "./fakeEventSource";
|
||||
import LiveLogPage from "./LiveLogPage";
|
||||
|
||||
function frame(ts: number, domain: string, overrides: Partial<LiveQueryEvent> = {}): { data: string } {
|
||||
const payload: LiveQueryEvent = {
|
||||
ts,
|
||||
domain,
|
||||
client_ip: "192.0.2.10",
|
||||
qtype: 1,
|
||||
blocked: false,
|
||||
block_reason: "",
|
||||
response_time_us: 500,
|
||||
cache_hit: true,
|
||||
upstream: "",
|
||||
...overrides,
|
||||
};
|
||||
return { data: JSON.stringify(payload) };
|
||||
}
|
||||
|
||||
function renderPage() {
|
||||
const sources: FakeEventSource[] = [];
|
||||
const createEventSource = (url: string) => {
|
||||
const es = new FakeEventSource(url);
|
||||
sources.push(es);
|
||||
return es;
|
||||
};
|
||||
render(<LiveLogPage createEventSource={createEventSource} />);
|
||||
return sources;
|
||||
}
|
||||
|
||||
test("streams rows, flags blocked ones, and freezes the display", () => {
|
||||
const sources = renderPage();
|
||||
expect(screen.getByText("Connecting…")).toBeTruthy();
|
||||
|
||||
act(() => sources[0]!.emit("open"));
|
||||
expect(screen.getByRole("status", { name: "Live" })).toBeTruthy();
|
||||
expect(screen.getByText("Waiting for queries…")).toBeTruthy();
|
||||
|
||||
act(() => {
|
||||
sources[0]!.emit("query", frame(1000, "ok.example"));
|
||||
sources[0]!.emit(
|
||||
"query",
|
||||
frame(1001, "ads.example", { blocked: true, block_reason: "blocklist:stevenblack", qtype: 28 }),
|
||||
);
|
||||
});
|
||||
|
||||
expect(screen.getByText("ok.example")).toBeTruthy();
|
||||
expect(screen.getByText("Blocked")).toBeTruthy();
|
||||
expect(screen.getByText("blocklist:stevenblack")).toBeTruthy();
|
||||
expect(screen.getByText("AAAA")).toBeTruthy();
|
||||
// StyleX compiles to opaque class names, so the check is structural: a blocked
|
||||
// row carries every class a plain row does, plus the ones the flag adds.
|
||||
const blockedRow = screen.getByText("ads.example").closest("tr");
|
||||
const plainRow = screen.getByText("ok.example").closest("tr");
|
||||
const blockedClasses = new Set(blockedRow?.className.split(" "));
|
||||
const plainClasses = plainRow?.className.split(" ") ?? [];
|
||||
expect(plainClasses.every((name) => blockedClasses.has(name))).toBe(true);
|
||||
expect(blockedClasses.size).toBeGreaterThan(plainClasses.length);
|
||||
|
||||
const freeze = screen.getByRole("button", { name: "Freeze" });
|
||||
fireEvent.click(freeze);
|
||||
expect(freeze.getAttribute("aria-pressed")).toBe("true");
|
||||
|
||||
act(() => sources[0]!.emit("query", frame(1002, "later.example")));
|
||||
expect(screen.queryByText("later.example")).toBeNull();
|
||||
expect(screen.getByText(/3 in buffer/)).toBeTruthy();
|
||||
|
||||
fireEvent.click(screen.getByRole("button", { name: "Resume" }));
|
||||
expect(screen.getByText("later.example")).toBeTruthy();
|
||||
});
|
||||
|
||||
test("repeated connection failures show the viewer-cap state with a retry button", () => {
|
||||
const sources = renderPage();
|
||||
act(() => {
|
||||
sources[0]!.emit("error");
|
||||
sources[0]!.emit("error");
|
||||
sources[0]!.emit("error");
|
||||
});
|
||||
expect(screen.getByRole("alert").textContent).toContain("too many live viewers");
|
||||
|
||||
fireEvent.click(screen.getByRole("button", { name: "Retry" }));
|
||||
expect(sources).toHaveLength(2);
|
||||
expect(screen.getByText("Connecting…")).toBeTruthy();
|
||||
});
|
||||
@@ -0,0 +1,257 @@
|
||||
import * as stylex from "@stylexjs/stylex";
|
||||
import { QueryCells, QueryTableHead } from "@/features/queries/QueryLogPage";
|
||||
import { RING_CAPACITY } from "./ringBuffer";
|
||||
import { useLiveQueries, type EventSourceFactory, type StreamStatus } from "./useLiveQueries";
|
||||
import { styles as shared } from "@/ui/styles";
|
||||
import { colors } from "@/ui/tokens.stylex";
|
||||
|
||||
const DARK = "@media (prefers-color-scheme: dark)";
|
||||
|
||||
const styles = stylex.create({
|
||||
toolbar: {
|
||||
display: "flex",
|
||||
flexWrap: "wrap",
|
||||
alignItems: "center",
|
||||
gap: "0.75rem",
|
||||
},
|
||||
heading: {
|
||||
fontSize: "1.5rem",
|
||||
lineHeight: "2rem",
|
||||
fontWeight: 600,
|
||||
},
|
||||
toolbarButton: {
|
||||
fontWeight: 500,
|
||||
},
|
||||
pill: {
|
||||
borderRadius: "9999px",
|
||||
paddingInline: "0.625rem",
|
||||
paddingBlock: "0.125rem",
|
||||
fontSize: "0.75rem",
|
||||
lineHeight: "1rem",
|
||||
fontWeight: 500,
|
||||
},
|
||||
/** Four stream states need four tints; only two of them map onto a token role. */
|
||||
pillConnecting: {
|
||||
backgroundColor: { default: "oklch(96.7% 0.001 286.375)", [DARK]: "oklch(27.4% 0.006 286.033)" },
|
||||
color: { default: "oklch(37% 0.013 285.805)", [DARK]: "oklch(87.1% 0.006 286.286)" },
|
||||
},
|
||||
pillOpen: {
|
||||
backgroundColor: { default: "oklch(96.2% 0.044 156.743)", [DARK]: "oklch(39.3% 0.095 152.535)" },
|
||||
color: { default: "oklch(44.8% 0.119 151.328)", [DARK]: "oklch(92.5% 0.084 155.995)" },
|
||||
},
|
||||
pillRetrying: {
|
||||
backgroundColor: { default: "oklch(96.2% 0.059 95.617)", [DARK]: "oklch(41.4% 0.112 45.904)" },
|
||||
color: { default: "oklch(47.3% 0.137 46.201)", [DARK]: "oklch(92.4% 0.12 95.746)" },
|
||||
},
|
||||
pillCapped: {
|
||||
backgroundColor: { default: "oklch(93.6% 0.032 17.717)", [DARK]: "oklch(39.6% 0.141 25.723)" },
|
||||
color: { default: "oklch(44.4% 0.177 26.899)", [DARK]: "oklch(88.5% 0.062 18.334)" },
|
||||
},
|
||||
note: {
|
||||
marginTop: "0.5rem",
|
||||
fontSize: "0.875rem",
|
||||
lineHeight: "1.25rem",
|
||||
color: colors.textMuted,
|
||||
},
|
||||
/** Informational, neither a warning nor a failure, so the blue ramp stands alone. */
|
||||
resumed: {
|
||||
marginTop: "0.75rem",
|
||||
display: "flex",
|
||||
alignItems: "center",
|
||||
gap: "0.75rem",
|
||||
borderRadius: "0.25rem",
|
||||
borderWidth: 1,
|
||||
borderStyle: "solid",
|
||||
borderColor: { default: "oklch(80.9% 0.105 251.813)", [DARK]: "oklch(37.9% 0.146 265.522)" },
|
||||
backgroundColor: { default: "oklch(97% 0.014 254.604)", [DARK]: "oklch(28.2% 0.091 267.935)" },
|
||||
color: { default: "oklch(42.4% 0.199 265.638)", [DARK]: "oklch(88.2% 0.059 254.128)" },
|
||||
paddingInline: "0.75rem",
|
||||
paddingBlock: "0.5rem",
|
||||
fontSize: "0.875rem",
|
||||
lineHeight: "1.25rem",
|
||||
},
|
||||
dismiss: {
|
||||
borderStyle: "none",
|
||||
backgroundColor: "transparent",
|
||||
padding: 0,
|
||||
color: "inherit",
|
||||
fontSize: "inherit",
|
||||
fontWeight: 500,
|
||||
textDecorationLine: "underline",
|
||||
},
|
||||
failureNote: {
|
||||
marginTop: "0.75rem",
|
||||
fontSize: "0.875rem",
|
||||
lineHeight: "1.25rem",
|
||||
color: colors.dangerText,
|
||||
},
|
||||
cappedBox: {
|
||||
marginTop: "1rem",
|
||||
borderRadius: "0.25rem",
|
||||
borderWidth: 1,
|
||||
borderStyle: "solid",
|
||||
borderColor: colors.dangerBorder,
|
||||
backgroundColor: colors.dangerSurface,
|
||||
padding: "1rem",
|
||||
},
|
||||
cappedHeading: {
|
||||
fontWeight: 600,
|
||||
color: colors.dangerText,
|
||||
},
|
||||
cappedDetail: {
|
||||
marginTop: "0.25rem",
|
||||
fontSize: "0.875rem",
|
||||
lineHeight: "1.25rem",
|
||||
color: colors.dangerText,
|
||||
},
|
||||
empty: {
|
||||
marginTop: "1.5rem",
|
||||
color: colors.textMuted,
|
||||
},
|
||||
tableWrap: {
|
||||
marginTop: "1rem",
|
||||
overflowX: "auto",
|
||||
borderRadius: "0.25rem",
|
||||
borderWidth: 1,
|
||||
borderStyle: "solid",
|
||||
borderColor: colors.border,
|
||||
},
|
||||
table: {
|
||||
width: "100%",
|
||||
fontSize: "0.875rem",
|
||||
lineHeight: "1.25rem",
|
||||
},
|
||||
/** `divide-y`: a hairline between rows, so the first row carries none. */
|
||||
row: {
|
||||
borderTopWidth: { default: 1, ":first-child": 0 },
|
||||
borderTopStyle: "solid",
|
||||
borderTopColor: colors.border,
|
||||
},
|
||||
rowBlocked: {
|
||||
backgroundColor: {
|
||||
default: "oklch(97.1% 0.013 17.38)",
|
||||
[DARK]: "oklch(25.8% 0.092 26.042 / 0.4)",
|
||||
},
|
||||
},
|
||||
footnote: {
|
||||
marginTop: "0.75rem",
|
||||
fontSize: "0.875rem",
|
||||
lineHeight: "1.25rem",
|
||||
color: colors.textMuted,
|
||||
},
|
||||
});
|
||||
|
||||
const PILL_LABELS: Record<StreamStatus, string> = {
|
||||
connecting: "Connecting…",
|
||||
open: "Live",
|
||||
retrying: "Reconnecting…",
|
||||
capped: "Disconnected",
|
||||
};
|
||||
|
||||
function pillStyle(status: StreamStatus) {
|
||||
if (status === "open") return styles.pillOpen;
|
||||
if (status === "retrying") return styles.pillRetrying;
|
||||
if (status === "capped") return styles.pillCapped;
|
||||
return styles.pillConnecting;
|
||||
}
|
||||
|
||||
function StatusPill({ status }: { status: StreamStatus }) {
|
||||
const label = PILL_LABELS[status];
|
||||
return (
|
||||
<span role="status" aria-label={label} {...stylex.props(styles.pill, pillStyle(status))}>
|
||||
{label}
|
||||
</span>
|
||||
);
|
||||
}
|
||||
|
||||
/** Freeze is display-only: the stream stays open and the 500-row ring buffer keeps
|
||||
* filling; Resume shows the current buffer (anything pushed out meanwhile is gone). */
|
||||
export default function LiveLogPage({ createEventSource }: { createEventSource?: EventSourceFactory } = {}) {
|
||||
const live = useLiveQueries({ createEventSource });
|
||||
|
||||
return (
|
||||
<section>
|
||||
<div {...stylex.props(styles.toolbar)}>
|
||||
<h1 {...stylex.props(styles.heading)}>Live</h1>
|
||||
<StatusPill status={live.status} />
|
||||
<button
|
||||
type="button"
|
||||
onClick={live.toggleFreeze}
|
||||
aria-pressed={live.frozen}
|
||||
{...stylex.props(shared.button, styles.toolbarButton, shared.focusRing)}
|
||||
>
|
||||
{live.frozen ? "Resume" : "Freeze"}
|
||||
</button>
|
||||
</div>
|
||||
|
||||
{live.frozen && (
|
||||
<p {...stylex.props(styles.note)} role="status">
|
||||
Display frozen — new queries keep buffering ({live.liveCount} in buffer, newest {RING_CAPACITY}{" "}
|
||||
kept).
|
||||
</p>
|
||||
)}
|
||||
|
||||
{live.missed !== null && (
|
||||
<div role="status" {...stylex.props(styles.resumed)}>
|
||||
<span>
|
||||
Stream resumed —{" "}
|
||||
{live.missed === 0 ? "no queries missed" : `${live.missed} missed queries recovered`}.
|
||||
</span>
|
||||
<button
|
||||
type="button"
|
||||
onClick={live.dismissMissed}
|
||||
{...stylex.props(styles.dismiss, shared.focusRing)}
|
||||
>
|
||||
Dismiss
|
||||
</button>
|
||||
</div>
|
||||
)}
|
||||
|
||||
{live.resyncFailed && (
|
||||
<p role="alert" {...stylex.props(styles.failureNote)}>
|
||||
Stream resumed, but re-syncing the gap failed — some queries may be missing here.
|
||||
</p>
|
||||
)}
|
||||
|
||||
{live.status === "capped" && (
|
||||
<div role="alert" {...stylex.props(styles.cappedBox)}>
|
||||
<h2 {...stylex.props(styles.cappedHeading)}>Live stream unavailable</h2>
|
||||
<p {...stylex.props(styles.cappedDetail)}>
|
||||
The connection failed repeatedly — possibly too many live viewers (the server caps streams per
|
||||
address), or the server is unreachable.
|
||||
</p>
|
||||
<button type="button" onClick={live.retry} {...stylex.props(shared.retryButton, shared.focusRing)}>
|
||||
Retry
|
||||
</button>
|
||||
</div>
|
||||
)}
|
||||
|
||||
{live.rows.length === 0 ? (
|
||||
live.status !== "capped" && (
|
||||
<p {...stylex.props(styles.empty)}>
|
||||
{live.status === "open" ? "Waiting for queries…" : "No queries received yet."}
|
||||
</p>
|
||||
)
|
||||
) : (
|
||||
<>
|
||||
<div {...stylex.props(styles.tableWrap)}>
|
||||
<table {...stylex.props(styles.table)}>
|
||||
<QueryTableHead />
|
||||
<tbody>
|
||||
{live.rows.map((row) => (
|
||||
<tr key={row.key} {...stylex.props(styles.row, row.blocked && styles.rowBlocked)}>
|
||||
<QueryCells row={row} />
|
||||
</tr>
|
||||
))}
|
||||
</tbody>
|
||||
</table>
|
||||
</div>
|
||||
<p {...stylex.props(styles.footnote)}>
|
||||
Showing {live.rows.length} {live.rows.length === 1 ? "query" : "queries"} (newest first, last{" "}
|
||||
{RING_CAPACITY} kept).
|
||||
</p>
|
||||
</>
|
||||
)}
|
||||
</section>
|
||||
);
|
||||
}
|
||||
@@ -0,0 +1,41 @@
|
||||
import { EVENT_SOURCE_CLOSED, type EventSourceLike } from "./useLiveQueries";
|
||||
|
||||
const CONNECTING = 0;
|
||||
const OPEN = 1;
|
||||
|
||||
/** Test double for the injected EventSource constructor. */
|
||||
export class FakeEventSource implements EventSourceLike {
|
||||
readonly url: string;
|
||||
closed = false;
|
||||
readyState: number = CONNECTING;
|
||||
private listeners = new Map<string, Array<(event: { data?: unknown }) => void>>();
|
||||
|
||||
constructor(url: string) {
|
||||
this.url = url;
|
||||
}
|
||||
|
||||
addEventListener(type: string, listener: (event: { data?: unknown }) => void): void {
|
||||
const existing = this.listeners.get(type) ?? [];
|
||||
existing.push(listener);
|
||||
this.listeners.set(type, existing);
|
||||
}
|
||||
|
||||
close(): void {
|
||||
this.closed = true;
|
||||
this.readyState = EVENT_SOURCE_CLOSED;
|
||||
}
|
||||
|
||||
emit(type: string, event: { data?: unknown } = {}): void {
|
||||
if (type === "open") this.readyState = OPEN;
|
||||
for (const listener of this.listeners.get(type) ?? []) listener(event);
|
||||
}
|
||||
|
||||
/**
|
||||
* A non-200 response: the browser closes the source, then dispatches one
|
||||
* error event and never retries.
|
||||
*/
|
||||
failFatal(): void {
|
||||
this.readyState = EVENT_SOURCE_CLOSED;
|
||||
this.emit("error");
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,101 @@
|
||||
import type { LiveQueryEvent, QueryRow } from "@/lib/types";
|
||||
import { RING_CAPACITY, mergeGap, pushRow, type LiveRow } from "./ringBuffer";
|
||||
|
||||
function event(ts: number, domain: string, overrides: Partial<LiveQueryEvent> = {}): LiveQueryEvent {
|
||||
return {
|
||||
ts,
|
||||
domain,
|
||||
client_ip: "192.0.2.10",
|
||||
qtype: 1,
|
||||
blocked: false,
|
||||
block_reason: "",
|
||||
response_time_us: 500,
|
||||
cache_hit: false,
|
||||
upstream: "udp://9.9.9.9:53",
|
||||
...overrides,
|
||||
};
|
||||
}
|
||||
|
||||
function liveRow(key: number, ts: number, domain: string, overrides: Partial<LiveQueryEvent> = {}): LiveRow {
|
||||
return { ...event(ts, domain, overrides), key };
|
||||
}
|
||||
|
||||
function fetchedRow(id: number, ts: number, domain: string, overrides: Partial<LiveQueryEvent> = {}): QueryRow {
|
||||
return { id, ...event(ts, domain, overrides) };
|
||||
}
|
||||
|
||||
function counter(start = 100): () => number {
|
||||
let n = start;
|
||||
return () => ++n;
|
||||
}
|
||||
|
||||
describe("pushRow", () => {
|
||||
test("prepends newest-first", () => {
|
||||
let rows: LiveRow[] = [];
|
||||
rows = pushRow(rows, liveRow(1, 10, "a.example"));
|
||||
rows = pushRow(rows, liveRow(2, 11, "b.example"));
|
||||
expect(rows.map((r) => r.domain)).toEqual(["b.example", "a.example"]);
|
||||
});
|
||||
|
||||
test("drops the oldest beyond capacity", () => {
|
||||
let rows: LiveRow[] = [];
|
||||
for (let i = 0; i < 5; i++) rows = pushRow(rows, liveRow(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, liveRow(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 = [liveRow(2, 100, "seen.example"), liveRow(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(rows.map((r) => r.domain)).toEqual(["gap2.example", "gap1.example", "seen.example", "old.example"]);
|
||||
});
|
||||
|
||||
test("no additions returns the buffer unchanged with missed 0", () => {
|
||||
const buffer = [liveRow(1, 100, "seen.example")];
|
||||
const { rows, missed } = mergeGap(buffer, [fetchedRow(5, 100, "seen.example")], counter());
|
||||
expect(missed).toBe(0);
|
||||
expect(rows).toBe(buffer);
|
||||
});
|
||||
|
||||
test("rows differing only in qtype are not deduplicated", () => {
|
||||
const buffer = [liveRow(1, 100, "dual.example", { qtype: 1 })];
|
||||
const fetched = [fetchedRow(5, 100, "dual.example", { qtype: 28 })];
|
||||
const { missed } = mergeGap(buffer, fetched, counter());
|
||||
expect(missed).toBe(1);
|
||||
});
|
||||
|
||||
test("assigns fresh keys from the counter and drops the id", () => {
|
||||
const { rows } = mergeGap([], [fetchedRow(77, 100, "gap.example")], counter(200));
|
||||
expect(rows[0]?.key).toBe(201);
|
||||
expect("id" in (rows[0] ?? {})).toBe(false);
|
||||
});
|
||||
|
||||
test("result is capped at capacity, keeping the newest", () => {
|
||||
const buffer = [liveRow(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(rows.map((r) => r.domain)).toEqual(["g2.example", "g1.example"]);
|
||||
});
|
||||
|
||||
test("merged rows stay sorted newest-first by ts", () => {
|
||||
const buffer = [liveRow(4, 105, "after-reopen.example"), liveRow(3, 100, "before.example")];
|
||||
const fetched = [fetchedRow(9, 103, "gap.example")];
|
||||
const { rows } = mergeGap(buffer, fetched, counter());
|
||||
expect(rows.map((r) => r.ts)).toEqual([105, 103, 100]);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,53 @@
|
||||
import type { LiveQueryEvent, QueryRow } from "@/lib/types";
|
||||
|
||||
/** A live stream row; `key` is a client-side monotonic counter (SSE frames carry no id). */
|
||||
export interface LiveRow extends LiveQueryEvent {
|
||||
key: number;
|
||||
}
|
||||
|
||||
export const RING_CAPACITY = 500;
|
||||
|
||||
/** Prepend `row` (rows are newest-first) and drop the oldest beyond `capacity`. */
|
||||
export function pushRow(rows: LiveRow[], row: LiveRow, capacity: number = RING_CAPACITY): LiveRow[] {
|
||||
const next = [row, ...rows];
|
||||
return next.length > capacity ? next.slice(0, capacity) : next;
|
||||
}
|
||||
|
||||
// `since` on GET /api/queries is inclusive, so the re-sync fetch returns the
|
||||
// last-seen row(s) again; live rows have no id, so identity is this tuple.
|
||||
function signature(row: LiveQueryEvent): string {
|
||||
return `${row.ts}|${row.domain}|${row.client_ip}|${row.qtype ?? -1}|${row.blocked}|${row.upstream}`;
|
||||
}
|
||||
|
||||
/**
|
||||
* Merge rows fetched for a reconnect gap (newest-first, from GET /api/queries)
|
||||
* into the buffer. Rows already present are skipped; `missed` counts what was
|
||||
* actually added. The result stays newest-first (stable sort by ts) and capped.
|
||||
*/
|
||||
export function mergeGap(
|
||||
rows: LiveRow[],
|
||||
fetched: QueryRow[],
|
||||
nextKey: () => number,
|
||||
capacity: number = RING_CAPACITY,
|
||||
): { rows: LiveRow[]; missed: number } {
|
||||
const seen = new Set(rows.map(signature));
|
||||
const added: LiveRow[] = [];
|
||||
for (const row of fetched) {
|
||||
const event: LiveQueryEvent = {
|
||||
ts: row.ts,
|
||||
domain: row.domain,
|
||||
client_ip: row.client_ip,
|
||||
qtype: row.qtype,
|
||||
blocked: row.blocked,
|
||||
block_reason: row.block_reason,
|
||||
response_time_us: row.response_time_us,
|
||||
cache_hit: row.cache_hit,
|
||||
upstream: row.upstream,
|
||||
};
|
||||
if (seen.has(signature(event))) continue;
|
||||
added.push({ ...event, key: nextKey() });
|
||||
}
|
||||
if (added.length === 0) return { rows, missed: 0 };
|
||||
const merged = [...added, ...rows].sort((a, b) => b.ts - a.ts).slice(0, capacity);
|
||||
return { rows: merged, missed: added.length };
|
||||
}
|
||||
@@ -0,0 +1,254 @@
|
||||
import { act, renderHook, waitFor } from "@testing-library/react";
|
||||
import { ApiError } from "@/lib/api";
|
||||
import type { LiveQueryEvent, QueriesPage, QueryRow } from "@/lib/types";
|
||||
import { FakeEventSource } from "./fakeEventSource";
|
||||
import { CAP_ERROR_THRESHOLD, useLiveQueries } from "./useLiveQueries";
|
||||
|
||||
afterEach(() => vi.unstubAllGlobals());
|
||||
|
||||
function stubLocationAssign() {
|
||||
const assign = vi.fn();
|
||||
vi.stubGlobal("location", { pathname: "/live", search: "", assign });
|
||||
return assign;
|
||||
}
|
||||
|
||||
function frame(ts: number, domain: string, overrides: Partial<LiveQueryEvent> = {}): { data: string } {
|
||||
const payload: LiveQueryEvent = {
|
||||
ts,
|
||||
domain,
|
||||
client_ip: "192.0.2.10",
|
||||
qtype: 1,
|
||||
blocked: false,
|
||||
block_reason: "",
|
||||
response_time_us: 500,
|
||||
cache_hit: false,
|
||||
upstream: "udp://9.9.9.9:53",
|
||||
...overrides,
|
||||
};
|
||||
return { data: JSON.stringify(payload) };
|
||||
}
|
||||
|
||||
function fetchedRow(id: number, ts: number, domain: string): QueryRow {
|
||||
return {
|
||||
id,
|
||||
ts,
|
||||
domain,
|
||||
client_ip: "192.0.2.10",
|
||||
qtype: 1,
|
||||
blocked: false,
|
||||
block_reason: "",
|
||||
response_time_us: 500,
|
||||
cache_hit: false,
|
||||
upstream: "udp://9.9.9.9:53",
|
||||
};
|
||||
}
|
||||
|
||||
function setup(fetchSince?: (since: number) => Promise<QueriesPage>, probeSession?: () => Promise<unknown>) {
|
||||
const sources: FakeEventSource[] = [];
|
||||
const createEventSource = (url: string) => {
|
||||
const es = new FakeEventSource(url);
|
||||
sources.push(es);
|
||||
return es;
|
||||
};
|
||||
const probe = probeSession ?? (() => Promise.resolve());
|
||||
const hook = renderHook(() => useLiveQueries({ createEventSource, fetchSince, probeSession: probe }));
|
||||
return { sources, hook };
|
||||
}
|
||||
|
||||
test("open then frames: rows newest-first with increasing keys", () => {
|
||||
const { sources, hook } = setup();
|
||||
expect(sources).toHaveLength(1);
|
||||
expect(hook.result.current.status).toBe("connecting");
|
||||
|
||||
act(() => sources[0]!.emit("open"));
|
||||
expect(hook.result.current.status).toBe("open");
|
||||
|
||||
act(() => {
|
||||
sources[0]!.emit("query", frame(1000, "a.example"));
|
||||
sources[0]!.emit("query", frame(1001, "b.example"));
|
||||
});
|
||||
const rows = hook.result.current.rows;
|
||||
expect(rows.map((r) => r.domain)).toEqual(["b.example", "a.example"]);
|
||||
expect(rows[0]!.key).toBeGreaterThan(rows[1]!.key);
|
||||
});
|
||||
|
||||
test("malformed and non-string frames are ignored", () => {
|
||||
const { sources, hook } = setup();
|
||||
act(() => {
|
||||
sources[0]!.emit("open");
|
||||
sources[0]!.emit("query", { data: "{not json" });
|
||||
sources[0]!.emit("query", {});
|
||||
});
|
||||
expect(hook.result.current.rows).toHaveLength(0);
|
||||
});
|
||||
|
||||
test("error then reopen re-syncs the gap since the last seen ts", async () => {
|
||||
const fetchSince = vi.fn((since: number): Promise<QueriesPage> => {
|
||||
return Promise.resolve({
|
||||
queries: [fetchedRow(9, 1002, "gap.example"), fetchedRow(8, since, "a.example")],
|
||||
next_before: null,
|
||||
});
|
||||
});
|
||||
const { sources, hook } = setup(fetchSince);
|
||||
|
||||
act(() => sources[0]!.emit("open"));
|
||||
expect(fetchSince).not.toHaveBeenCalled();
|
||||
|
||||
act(() => sources[0]!.emit("query", frame(1000, "a.example")));
|
||||
act(() => sources[0]!.emit("error"));
|
||||
expect(hook.result.current.status).toBe("retrying");
|
||||
|
||||
act(() => sources[0]!.emit("open"));
|
||||
expect(hook.result.current.status).toBe("open");
|
||||
expect(fetchSince).toHaveBeenCalledWith(1000);
|
||||
|
||||
await waitFor(() => expect(hook.result.current.missed).toBe(1));
|
||||
expect(hook.result.current.rows.map((r) => r.domain)).toEqual(["gap.example", "a.example"]);
|
||||
|
||||
act(() => hook.result.current.dismissMissed());
|
||||
expect(hook.result.current.missed).toBeNull();
|
||||
});
|
||||
|
||||
test("failed re-sync sets resyncFailed", async () => {
|
||||
const fetchSince = vi.fn((): Promise<QueriesPage> => Promise.reject(new Error("boom")));
|
||||
const { sources, hook } = setup(fetchSince);
|
||||
act(() => sources[0]!.emit("open"));
|
||||
act(() => sources[0]!.emit("query", frame(1000, "a.example")));
|
||||
act(() => sources[0]!.emit("error"));
|
||||
act(() => sources[0]!.emit("open"));
|
||||
await waitFor(() => expect(hook.result.current.resyncFailed).toBe(true));
|
||||
});
|
||||
|
||||
test("a 401 gap re-sync redirects to login instead of setting resyncFailed", async () => {
|
||||
const assign = stubLocationAssign();
|
||||
const fetchSince = vi.fn((): Promise<QueriesPage> => Promise.reject(new ApiError(401, "unauthorized")));
|
||||
const { sources, hook } = setup(fetchSince);
|
||||
act(() => sources[0]!.emit("open"));
|
||||
act(() => sources[0]!.emit("query", frame(1000, "a.example")));
|
||||
act(() => sources[0]!.emit("error"));
|
||||
act(() => sources[0]!.emit("open"));
|
||||
await waitFor(() => expect(assign).toHaveBeenCalledWith("/login?redirect=%2Flive"));
|
||||
expect(hook.result.current.resyncFailed).toBe(false);
|
||||
});
|
||||
|
||||
test("cap trip with a valid session probes once and stays capped", async () => {
|
||||
const assign = stubLocationAssign();
|
||||
const probeSession = vi.fn((): Promise<unknown> => Promise.resolve({}));
|
||||
const { sources, hook } = setup(undefined, probeSession);
|
||||
act(() => {
|
||||
for (let i = 0; i < CAP_ERROR_THRESHOLD; i++) sources[0]!.emit("error");
|
||||
});
|
||||
expect(hook.result.current.status).toBe("capped");
|
||||
expect(probeSession).toHaveBeenCalledTimes(1);
|
||||
await act(async () => {});
|
||||
expect(assign).not.toHaveBeenCalled();
|
||||
expect(hook.result.current.status).toBe("capped");
|
||||
});
|
||||
|
||||
test("cap trip with an expired session redirects to login", async () => {
|
||||
const assign = stubLocationAssign();
|
||||
const probeSession = vi.fn((): Promise<unknown> => Promise.reject(new ApiError(401, "unauthorized")));
|
||||
const { sources } = setup(undefined, probeSession);
|
||||
act(() => {
|
||||
for (let i = 0; i < CAP_ERROR_THRESHOLD; i++) sources[0]!.emit("error");
|
||||
});
|
||||
await waitFor(() => expect(assign).toHaveBeenCalledWith("/login?redirect=%2Flive"));
|
||||
expect(probeSession).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
test("repeated errors without open hit the cap state; retry reconnects", () => {
|
||||
const { sources, hook } = setup();
|
||||
act(() => {
|
||||
for (let i = 0; i < CAP_ERROR_THRESHOLD; i++) sources[0]!.emit("error");
|
||||
});
|
||||
expect(hook.result.current.status).toBe("capped");
|
||||
expect(sources[0]!.closed).toBe(true);
|
||||
|
||||
act(() => hook.result.current.retry());
|
||||
expect(sources).toHaveLength(2);
|
||||
expect(hook.result.current.status).toBe("connecting");
|
||||
|
||||
act(() => sources[1]!.emit("open"));
|
||||
expect(hook.result.current.status).toBe("open");
|
||||
});
|
||||
|
||||
test("a fatal rejection caps on the first error event and probes the session", async () => {
|
||||
const assign = stubLocationAssign();
|
||||
const probeSession = vi.fn((): Promise<unknown> => Promise.resolve({}));
|
||||
const { sources, hook } = setup(undefined, probeSession);
|
||||
|
||||
act(() => sources[0]!.failFatal());
|
||||
|
||||
expect(hook.result.current.status).toBe("capped");
|
||||
expect(probeSession).toHaveBeenCalledTimes(1);
|
||||
expect(sources[0]!.closed).toBe(true);
|
||||
await act(async () => {});
|
||||
expect(assign).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
test("a fatal rejection with an expired session redirects to login", async () => {
|
||||
const assign = stubLocationAssign();
|
||||
const probeSession = vi.fn((): Promise<unknown> => Promise.reject(new ApiError(401, "unauthorized")));
|
||||
const { sources } = setup(undefined, probeSession);
|
||||
|
||||
act(() => sources[0]!.failFatal());
|
||||
|
||||
await waitFor(() => expect(assign).toHaveBeenCalledWith("/login?redirect=%2Flive"));
|
||||
expect(probeSession).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
test("a transient error leaves the source open and still takes three to cap", () => {
|
||||
const probeSession = vi.fn((): Promise<unknown> => Promise.resolve({}));
|
||||
const { sources, hook } = setup(undefined, probeSession);
|
||||
|
||||
for (let i = 0; i < CAP_ERROR_THRESHOLD - 1; i++) {
|
||||
act(() => sources[0]!.emit("error"));
|
||||
expect(hook.result.current.status).toBe("retrying");
|
||||
expect(sources[0]!.closed).toBe(false);
|
||||
expect(probeSession).not.toHaveBeenCalled();
|
||||
}
|
||||
|
||||
act(() => sources[0]!.emit("error"));
|
||||
expect(hook.result.current.status).toBe("capped");
|
||||
expect(probeSession).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
test("a successful open resets the consecutive error count", () => {
|
||||
const { sources, hook } = setup();
|
||||
act(() => sources[0]!.emit("error"));
|
||||
act(() => sources[0]!.emit("error"));
|
||||
act(() => sources[0]!.emit("open"));
|
||||
act(() => sources[0]!.emit("error"));
|
||||
expect(hook.result.current.status).toBe("retrying");
|
||||
expect(sources[0]!.closed).toBe(false);
|
||||
});
|
||||
|
||||
test("freeze keeps the display fixed while the buffer keeps filling", () => {
|
||||
const { sources, hook } = setup();
|
||||
act(() => sources[0]!.emit("open"));
|
||||
act(() => sources[0]!.emit("query", frame(1000, "a.example")));
|
||||
|
||||
act(() => hook.result.current.toggleFreeze());
|
||||
expect(hook.result.current.frozen).toBe(true);
|
||||
|
||||
act(() => {
|
||||
sources[0]!.emit("query", frame(1001, "b.example"));
|
||||
sources[0]!.emit("query", frame(1002, "c.example"));
|
||||
});
|
||||
expect(hook.result.current.rows.map((r) => r.domain)).toEqual(["a.example"]);
|
||||
expect(hook.result.current.liveCount).toBe(3);
|
||||
|
||||
act(() => hook.result.current.toggleFreeze());
|
||||
expect(hook.result.current.frozen).toBe(false);
|
||||
expect(hook.result.current.rows.map((r) => r.domain)).toEqual(["c.example", "b.example", "a.example"]);
|
||||
});
|
||||
|
||||
test("stale sources are ignored after retry and closed on unmount", () => {
|
||||
const { sources, hook } = setup();
|
||||
act(() => hook.result.current.retry());
|
||||
act(() => sources[0]!.emit("query", frame(1000, "stale.example")));
|
||||
expect(hook.result.current.rows).toHaveLength(0);
|
||||
|
||||
hook.unmount();
|
||||
expect(sources[1]!.closed).toBe(true);
|
||||
});
|
||||
@@ -0,0 +1,187 @@
|
||||
import { useCallback, useEffect, useRef, useState } from "react";
|
||||
import * as api from "@/lib/api";
|
||||
import { handleUnauthorized } from "@/lib/queryClient";
|
||||
import type { LiveQueryEvent, QueriesPage } from "@/lib/types";
|
||||
import { RING_CAPACITY, mergeGap, pushRow, type LiveRow } from "./ringBuffer";
|
||||
|
||||
export type StreamStatus = "connecting" | "open" | "retrying" | "capped";
|
||||
|
||||
/** Minimal EventSource surface so tests can inject a fake. */
|
||||
export interface EventSourceLike {
|
||||
addEventListener(type: string, listener: (event: { data?: unknown }) => void): void;
|
||||
close(): void;
|
||||
readyState: number;
|
||||
}
|
||||
|
||||
/** `EventSource.CLOSED`: the browser gave up and will not retry. */
|
||||
export const EVENT_SOURCE_CLOSED = 2;
|
||||
|
||||
export type EventSourceFactory = (url: string) => EventSourceLike;
|
||||
|
||||
export interface LiveQueriesOptions {
|
||||
url?: string;
|
||||
createEventSource?: EventSourceFactory;
|
||||
fetchSince?: (since: number) => Promise<QueriesPage>;
|
||||
/** Cheap session-gated GET fired once on entering capped, to distinguish an expired session from a real cap. */
|
||||
probeSession?: () => Promise<unknown>;
|
||||
}
|
||||
|
||||
// A transient drop is invisible to EventSource beyond a bare `error` event;
|
||||
// this many consecutive errors without an intervening `open` (the browser
|
||||
// retries every 3s per the server's `retry: 3000`) stops the stream and
|
||||
// surfaces a manual-retry state. A non-200 response instead fails the source
|
||||
// permanently after one error event, and is handled by readyState below.
|
||||
export const CAP_ERROR_THRESHOLD = 3;
|
||||
|
||||
const defaultEventSource: EventSourceFactory = (url) => new EventSource(url);
|
||||
const defaultFetchSince = (since: number): Promise<QueriesPage> => api.getQueries({ since, limit: RING_CAPACITY });
|
||||
const defaultProbeSession = (): Promise<unknown> => api.getPause();
|
||||
|
||||
function isUnauthorized(error: unknown): boolean {
|
||||
return error instanceof api.ApiError && error.status === 401;
|
||||
}
|
||||
|
||||
export interface LiveQueries {
|
||||
/** Newest-first; the freeze-time snapshot while frozen. */
|
||||
rows: LiveRow[];
|
||||
/** Size of the live buffer, which keeps filling while frozen. */
|
||||
liveCount: number;
|
||||
status: StreamStatus;
|
||||
/** Rows recovered by the reconnect re-sync; null until a re-sync happens or after dismissal. */
|
||||
missed: number | null;
|
||||
resyncFailed: boolean;
|
||||
frozen: boolean;
|
||||
toggleFreeze: () => void;
|
||||
retry: () => void;
|
||||
dismissMissed: () => void;
|
||||
}
|
||||
|
||||
export function useLiveQueries(options?: LiveQueriesOptions): LiveQueries {
|
||||
const [rows, setRows] = useState<LiveRow[]>([]);
|
||||
const [status, setStatus] = useState<StreamStatus>("connecting");
|
||||
const [missed, setMissed] = useState<number | null>(null);
|
||||
const [resyncFailed, setResyncFailed] = useState(false);
|
||||
const [frozen, setFrozen] = useState(false);
|
||||
const [frozenRows, setFrozenRows] = useState<LiveRow[]>([]);
|
||||
|
||||
const bufferRef = useRef<LiveRow[]>([]);
|
||||
const keyRef = useRef(0);
|
||||
const lastSeenTsRef = useRef<number | null>(null);
|
||||
const everOpenRef = useRef(false);
|
||||
const errorsRef = useRef(0);
|
||||
const esRef = useRef<EventSourceLike | null>(null);
|
||||
|
||||
const optionsRef = useRef(options);
|
||||
optionsRef.current = options;
|
||||
|
||||
const connect = useCallback(() => {
|
||||
esRef.current?.close();
|
||||
errorsRef.current = 0;
|
||||
setStatus("connecting");
|
||||
const opts = optionsRef.current;
|
||||
const fetchSince = opts?.fetchSince ?? defaultFetchSince;
|
||||
const probeSession = opts?.probeSession ?? defaultProbeSession;
|
||||
const es = (opts?.createEventSource ?? defaultEventSource)(opts?.url ?? api.liveQueriesUrl);
|
||||
esRef.current = es;
|
||||
|
||||
es.addEventListener("open", () => {
|
||||
if (esRef.current !== es) return;
|
||||
errorsRef.current = 0;
|
||||
setStatus("open");
|
||||
const since = lastSeenTsRef.current;
|
||||
if (everOpenRef.current && since !== null) {
|
||||
setResyncFailed(false);
|
||||
fetchSince(since).then(
|
||||
(page) => {
|
||||
if (esRef.current !== es) return;
|
||||
const merged = mergeGap(bufferRef.current, page.queries, () => ++keyRef.current);
|
||||
bufferRef.current = merged.rows;
|
||||
setRows(merged.rows);
|
||||
setMissed(merged.missed);
|
||||
},
|
||||
(error: unknown) => {
|
||||
if (esRef.current !== es) return;
|
||||
if (isUnauthorized(error)) {
|
||||
handleUnauthorized(error);
|
||||
return;
|
||||
}
|
||||
setResyncFailed(true);
|
||||
},
|
||||
);
|
||||
}
|
||||
everOpenRef.current = true;
|
||||
});
|
||||
|
||||
es.addEventListener("query", (event) => {
|
||||
if (esRef.current !== es) return;
|
||||
if (typeof event.data !== "string") return;
|
||||
let payload: LiveQueryEvent;
|
||||
try {
|
||||
payload = JSON.parse(event.data) as LiveQueryEvent;
|
||||
} catch {
|
||||
return;
|
||||
}
|
||||
lastSeenTsRef.current = payload.ts;
|
||||
bufferRef.current = pushRow(bufferRef.current, { ...payload, key: ++keyRef.current });
|
||||
setRows(bufferRef.current);
|
||||
});
|
||||
|
||||
const giveUp = () => {
|
||||
es.close();
|
||||
setStatus("capped");
|
||||
// EventSource cannot surface a 401; an expired session looks
|
||||
// identical to the cap. Probe once on entering capped so the
|
||||
// user lands on login instead of a misleading capped message.
|
||||
probeSession().catch(handleUnauthorized);
|
||||
};
|
||||
|
||||
es.addEventListener("error", () => {
|
||||
if (esRef.current !== es) return;
|
||||
// A 429 or 401 closes the source outright — no retry follows, so
|
||||
// the consecutive-error counter would never reach its threshold.
|
||||
if (es.readyState === EVENT_SOURCE_CLOSED) {
|
||||
errorsRef.current = CAP_ERROR_THRESHOLD;
|
||||
giveUp();
|
||||
return;
|
||||
}
|
||||
errorsRef.current += 1;
|
||||
if (errorsRef.current >= CAP_ERROR_THRESHOLD) {
|
||||
giveUp();
|
||||
} else {
|
||||
setStatus("retrying");
|
||||
}
|
||||
});
|
||||
}, []);
|
||||
|
||||
useEffect(() => {
|
||||
connect();
|
||||
return () => {
|
||||
esRef.current?.close();
|
||||
esRef.current = null;
|
||||
};
|
||||
}, [connect]);
|
||||
|
||||
const toggleFreeze = () => {
|
||||
if (frozen) {
|
||||
setFrozen(false);
|
||||
} else {
|
||||
setFrozen(true);
|
||||
setFrozenRows(bufferRef.current);
|
||||
}
|
||||
};
|
||||
|
||||
return {
|
||||
rows: frozen ? frozenRows : rows,
|
||||
liveCount: rows.length,
|
||||
status,
|
||||
missed,
|
||||
resyncFailed,
|
||||
frozen,
|
||||
toggleFreeze,
|
||||
retry: connect,
|
||||
dismissMissed: () => {
|
||||
setMissed(null);
|
||||
setResyncFailed(false);
|
||||
},
|
||||
};
|
||||
}
|
||||
Reference in New Issue
Block a user