nxdns/admin/src/features/activity/useLiveQueries.test.tsx

import { act, renderHook, waitFor } from "@testing-library/react";
import { ApiError } from "@/lib/api";
import type { QueriesPage, QueryRow } from "@/lib/types";
import { provenance, queryRow } from "@/features/provenance/provenanceFixture";
import { summaryOf, type LiveRow } from "./ringBuffer";
import { FakeEventSource } from "./fakeEventSource";
import { CAP_ERROR_THRESHOLD, useLiveQueries } from "./useLiveQueries";

afterEach(() => vi.unstubAllGlobals());

function stubLocationAssign() {
	const assign = vi.fn();
	vi.stubGlobal("location", { pathname: "/activity", search: "?mode=live", assign });
	return assign;
}

function frame(ts: number, domain: string): { data: string } {
	const payload = provenance({
		request: { time: ts, domain },
		route: { upstream: "udp://9.9.9.9:53" },
	});
	return { data: JSON.stringify(payload) };
}

function fetchedRow(id: number, ts: number, domain: string): QueryRow {
	return queryRow(id, { ts, domain, upstream: "udp://9.9.9.9:53" });
}

function domains(rows: LiveRow[]): string[] {
	return rows.map((row) => summaryOf(row).domain);
}

const FULL_COVERAGE = { complete: true, available_since: 0 };

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(domains(rows)).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,
			coverage: FULL_COVERAGE,
		});
	});
	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(domains(hook.result.current.rows)).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=%2Factivity%3Fmode%3Dlive"));
	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=%2Factivity%3Fmode%3Dlive"));
	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=%2Factivity%3Fmode%3Dlive"));
	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(domains(hook.result.current.rows)).toEqual(["a.example"]);
	expect(hook.result.current.liveCount).toBe(3);

	act(() => hook.result.current.toggleFreeze());
	expect(hook.result.current.frozen).toBe(false);
	expect(domains(hook.result.current.rows)).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);
});