nxdns/admin/src/features/activity/ringBuffer.ts

import type { LiveQueryEvent, QueryRow } from "@/lib/types";
import { summarizeEvent, summarizeRow, type QuerySummary } from "@/features/provenance/querySummary";

/**
 * A row in the live buffer. `key` is a client-side monotonic counter, because
 * neither arm has a stable identity of its own on arrival.
 *
 * The two arms are genuinely different facts, not two encodings of one. A
 * streamed frame carries the full provenance of a query the server has not
 * written yet; a row recovered by the reconnect gap-fetch is the stored summary
 * of a query that *was* written, and cannot fabricate the provenance it never
 * received. Only the recovered arm has a row id to link to.
 */
export type LiveRow = { key: number } & (
	{ kind: "streamed"; event: LiveQueryEvent } | { kind: "recovered"; row: QueryRow }
);

/** The arm that carries its own provenance, and so its own detail surface. */
export type StreamedRow = Extract<LiveRow, { kind: "streamed" }>;

/** The flat cells both arms render, and the shared identity for gap dedupe. */
export function summaryOf(row: LiveRow): QuerySummary {
	return row.kind === "streamed" ? summarizeEvent(row.event) : summarizeRow(row.row);
}

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;
}

/**
 * What the gap merge compares two queries by: the whole stored row bar its id.
 *
 * `since` on GET /api/queries is inclusive, so the re-sync fetch returns the
 * last-seen row(s) again and the merge has to recognise them. The id cannot
 * serve as the identity — a streamed frame precedes its own insert and has none
 * — so the comparison is by value, and every stored column has to take part.
 * Two queries alike in name, client and second but differing in class, rcode,
 * the policy that decided them or the route taken are separate facts; if they
 * hashed alike, the fetched row that does *not* match the buffered one would
 * consume its occurrence, duplicating one query and losing the other.
 *
 * `QuerySummary` is the wrong basis for that: it is what the table renders, and
 * it drops qclass and policy_action. `Omit<QueryRow, "id">`
 * instead makes the compiler demand a derivation for every stored column, so a
 * column added to the row cannot quietly fall out of the identity.
 */
type GapIdentity = Omit<QueryRow, "id">;

function identityOfRow(row: QueryRow): GapIdentity {
	return {
		ts: row.ts,
		domain: row.domain,
		client_ip: row.client_ip,
		qtype: row.qtype,
		qclass: row.qclass,
		rcode: row.rcode,
		blocked: row.blocked,
		response_time_us: row.response_time_us,
		cache_hit: row.cache_hit,
		upstream: row.upstream,
		policy_action: row.policy_action,
		policy_reason: row.policy_reason,
		route_kind: row.route_kind,
	};
}

/**
 * The same identity out of a live frame, which carries every stored column in
 * its provenance. The columns the server derives rather than sends — `blocked`
 * and `cache_hit` — come through `summarizeEvent` so that derivation keeps
 * living in exactly one place.
 */
function identityOfEvent(event: LiveQueryEvent): GapIdentity {
	const summary = summarizeEvent(event);
	return {
		ts: summary.ts,
		domain: summary.domain,
		client_ip: summary.client_ip,
		qtype: summary.qtype,
		qclass: event.request.qclass,
		rcode: event.response.rcode,
		blocked: summary.blocked,
		response_time_us: summary.response_time_us,
		cache_hit: summary.cache_hit,
		upstream: summary.upstream,
		policy_action: event.policy.action,
		policy_reason: summary.policy_reason,
		route_kind: event.route.kind,
	};
}

function identityOf(row: LiveRow): GapIdentity {
	return row.kind === "streamed" ? identityOfEvent(row.event) : identityOfRow(row.row);
}

/** Sorted keys so the hash cannot depend on the order the two arms happen to build their literals in. */
function signature(identity: GapIdentity): string {
	return JSON.stringify(identity, Object.keys(identity).sort());
}

/**
 * How many times each signature is already in the buffer. A signature is not
 * unique: one client asking for one name twice within the same second is an
 * ordinary household pattern, and the two queries are separate facts. Counting
 * the occurrences lets the merge drop exactly as many fetched rows as the
 * buffer already holds, instead of letting one buffered row hide all of them.
 */
function occurrences(rows: LiveRow[]): Map<string, number> {
	const counts = new Map<string, number>();
	for (const row of rows) {
		const key = signature(identityOf(row));
		counts.set(key, (counts.get(key) ?? 0) + 1);
	}
	return counts;
}

/**
 * Merge rows fetched for a reconnect gap (newest-first, from GET /api/queries)
 * into the buffer. Each fetched row consumes one buffered occurrence of its
 * signature and is skipped; the rest are genuinely missed and `missed` counts
 * them. 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 buffered = occurrences(rows);
	const added: LiveRow[] = [];
	for (const row of fetched) {
		const key = signature(identityOfRow(row));
		const count = buffered.get(key) ?? 0;
		if (count > 0) {
			buffered.set(key, count - 1);
			continue;
		}
		added.push({ kind: "recovered", row, key: nextKey() });
	}
	if (added.length === 0) return { rows, missed: 0 };
	const merged = [...added, ...rows].sort((a, b) => summaryOf(b).ts - summaryOf(a).ts).slice(0, capacity);
	return { rows: merged, missed: added.length };
}