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

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>;
	/**
	 * Ring size. Injectable so a test can provoke an eviction with a handful of
	 * rows instead of pushing 500 frames through React state; the app never
	 * passes it, and the operator never sees it.
	 */
	capacity?: number;
}

// 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 =
	(capacity: number) =>
	(since: number): Promise<QueriesPage> =>
		api.getQueries({ since, limit: 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 capacity = opts?.capacity ?? RING_CAPACITY;
		const fetchSince = opts?.fetchSince ?? defaultFetchSince(capacity);
		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, capacity);
						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.request.time;
			bufferRef.current = pushRow(
				bufferRef.current,
				{ kind: "streamed", event: payload, key: ++keyRef.current },
				capacity,
			);
			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);
		},
	};
}