siftlode/frontend/src/lib/diagnostics.ts

// The user's consented diagnostics level, applied to this tab (E22 S5).
//
// `applyDiagnostics(me.diagnostics)` runs whenever `me` answers. It sets the client log's level, and
// while that level is raised (Részletes / Teljes) it streams the log to the server:
//   * only events kept AFTER the level was raised — the consent is for what happens from then on;
//   * in batches, every FLUSH_MS or once BATCH_MAX are waiting, and the rest with a keepalive
//     request when the tab closes;
//   * never the lines about its own requests, or each batch would carry the previous one's;
//   * and it stops at once when the level is lowered, when it lapses (a timer, not the next `me`
//     refetch), or when the server answers that it is off.
// A lost batch is dropped, not retried in a loop: this is a diagnostic stream, not a ledger, and a
// retry storm against a struggling server is the last thing a diagnosis needs. The one exception is
// being told to slow down (429): that batch goes back to the front of the queue, and nothing is sent
// for BACKOFF_MS — not even when a full batch is waiting, or every new event would send one.
import { api, DIAGNOSTICS_EVENTS_PATH, HttpError, type DiagnosticsState } from "./api";
import { log, setLevel, subscribe, type LogEvent } from "./log";

const FLUSH_MS = 5_000;
/** Per request. The server takes up to 200; staying under it leaves room for a burst. */
const BATCH_MAX = 100;
/** Waiting events kept while the server is slow or away; past it the oldest go. */
const QUEUE_MAX = 1_000;
/** Quiet time after a 429. The server counts batches per minute, per account — every tab of it. */
const BACKOFF_MS = 15_000;

/** The level actually in force: the consented one until it lapses. */
export function levelInForce(state: DiagnosticsState | null | undefined, now = Date.now()) {
  if (!state || state.level === "INFO" || !state.until) return "INFO";
  return Date.parse(state.until) > now ? state.level : "INFO";
}

let applied: string | null = null;
let lapseTimer: number | null = null;
let onLapsed: (() => void) | null = null;

/** Apply `state` to this tab. `lapsed` is called when a raised level runs out or the server says it
 *  is off, so the caller can refetch `me` (and the Settings tab can show the base level again). */
export function applyDiagnostics(
  state: DiagnosticsState | null | undefined,
  lapsed: () => void,
): void {
  onLapsed = lapsed;
  const level = levelInForce(state);
  const key = level === "INFO" ? "INFO" : `${level}@${state?.until}`;
  if (key === applied) return;
  const wasRaised = applied !== null && applied !== "INFO";
  applied = key;
  if (lapseTimer !== null) window.clearTimeout(lapseTimer);
  lapseTimer = null;
  if (level === "INFO") {
    stopStreaming();
    setLevel("INFO");
    // Only a CHANGE is worth a line; every tab starts at the base level.
    if (wasRaised) log.info("diagnostics.level", "diagnostics back at the base level", { level });
    return;
  }
  setLevel(level);
  startStreaming(); // first, so the stream opens with the line saying why it exists
  log.info("diagnostics.level", "diagnostics level raised", { level, until: state?.until });
  // setTimeout's delay is a signed 32-bit number; a longer wait would fire at once. The server
  // grants 24 h, so the clamp only matters for a clock that is badly off.
  const wait = Math.min(Date.parse(state!.until!) - Date.now(), 2 ** 31 - 1);
  lapseTimer = window.setTimeout(lapse, wait);
}

function lapse(): void {
  lapseTimer = null;
  applied = "INFO";
  stopStreaming();
  setLevel("INFO");
  log.info("diagnostics.level", "diagnostics level lapsed", { level: "INFO" });
  onLapsed?.();
}

// --- streaming ----------------------------------------------------------------------------------
let queue: LogEvent[] = [];
let unsubscribe: (() => void) | null = null;
let flushTimer: number | null = null;
let inFlight = false;
let quietUntil = 0;
// Which consent a batch was taken under. A batch still in flight when the level is lowered must not
// come back into the queue of a LATER consent: it was captured under the one just withdrawn.
let session = 0;

const isOwnRequest = (event: LogEvent) =>
  event.event === "http.request" && event.attrs?.path === DIAGNOSTICS_EVENTS_PATH;

const keepNewest = () => {
  if (queue.length > QUEUE_MAX) queue.splice(0, queue.length - QUEUE_MAX);
};

function startStreaming(): void {
  if (unsubscribe) return;
  session += 1;
  queue = [];
  quietUntil = 0;
  unsubscribe = subscribe((event) => {
    if (!event || isOwnRequest(event)) return;
    queue.push(event);
    keepNewest();
    if (queue.length >= BATCH_MAX) void flush();
  });
  flushTimer = window.setInterval(() => void flush(), FLUSH_MS);
  window.addEventListener("pagehide", flushOnUnload);
}

function stopStreaming(): void {
  unsubscribe?.();
  unsubscribe = null;
  if (flushTimer !== null) window.clearInterval(flushTimer);
  flushTimer = null;
  window.removeEventListener("pagehide", flushOnUnload);
  // What was captured under the consent but not yet sent stays in the tab's own log, not the server's.
  queue = [];
}

async function flush(): Promise<void> {
  if (inFlight || !queue.length || Date.now() < quietUntil) return;
  inFlight = true;
  const takenIn = session;
  const batch = queue.splice(0, BATCH_MAX);
  try {
    await api.sendLogEvents(batch);
  } catch (err) {
    if (err instanceof HttpError && err.status === 429) {
      quietUntil = Date.now() + BACKOFF_MS;
      if (takenIn === session && unsubscribe) {
        queue.unshift(...batch);
        keepNewest();
      }
    } else if (
      err instanceof HttpError &&
      err.status === 409 &&
      takenIn === session &&
      unsubscribe
    ) {
      // The server has it off: withdrawn in another tab, or lapsed by its clock. `unsubscribe`
      // means THIS tab does not already know — without it, the 409 answering a batch that was in
      // flight when the user picked Alap writes "lapsed" into their own log for a level they
      // revoked, and a lapse this tab has already announced gets announced a second time.
      lapse();
    }
  } finally {
    inFlight = false;
  }
}

function flushOnUnload(): void {
  while (queue.length) api.sendLogEventsOnUnload(queue.splice(0, BATCH_MAX));
}