siftlode/frontend/src/lib/diagnostics.ts

// The user's consented diagnostics level, applied to this tab (E22 S5) — and the debug session it
// may be recording (E16).
//
// `applyDiagnostics(me.diagnostics, me.debug_session)` 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 recording raises the level to its own for as long as it runs (starting it was the consent), and
// never lowers it: a user already at Teljes records at Teljes. Every event kept while it runs carries
// its id, and so does every request (`api/core`), which is how the server tags its own lines.
// 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 DebugSession,
  type DiagnosticsState,
} from "./api";
import { log, setDebugSession, 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";
}

/** The session if it is still recording: not ended, and before its cap. */
export function liveSession(
  session: DebugSession | null | undefined,
  now = Date.now(),
): DebugSession | null {
  if (!session || session.ended_at || Date.parse(session.until) <= now) return null;
  return session;
}

const DETAIL = { INFO: 0, DEBUG: 1, TRACE: 2 } as const;

let applied: string | null = null;
let lapseTimer: number | null = null;
// A recording's last seconds are sent BEFORE its cap: at the cap the server stops tagging (or, at the
// base level, taking) them, and the regular flush could be up to FLUSH_MS behind.
const CAP_FLUSH_LEAD_MS = 1_500;
let capFlushTimer: number | null = null;
let onLapsed: (() => void) | null = null;
let lastState: DiagnosticsState | null | undefined = null;
let lastSession: DebugSession | null | undefined = null;

/** Apply `state` and `session` to this tab. `lapsed` is called when a raised level or a recording
 *  runs out, or the server says it is off, so the caller can refetch `me` (and Settings and the
 *  recorder show what is in force now). */
export function applyDiagnostics(
  state: DiagnosticsState | null | undefined,
  session: DebugSession | null | undefined,
  lapsed: () => void,
): void {
  onLapsed = lapsed;
  lastState = state;
  lastSession = session;
  const own = levelInForce(state);
  const live = liveSession(session);
  const bySession = !!live && DETAIL[live.level] > DETAIL[own];
  const level = bySession ? live.level : own;
  const until = bySession ? live.until : (state?.until ?? null);
  setDebugSession(live?.id ?? null);
  const key = level === "INFO" ? "INFO" : `${level}@${until}#${live?.id ?? ""}`;
  if (key === applied) return;
  const wasRaised = applied !== null && applied !== "INFO";
  applied = key;
  if (lapseTimer !== null) window.clearTimeout(lapseTimer);
  lapseTimer = null;
  if (capFlushTimer !== null) window.clearTimeout(capFlushTimer);
  capFlushTimer = 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,
    ...(live ? { debug_session: live.id } : {}),
  });
  // Whichever runs out first — the level or the recording — changes what is in force. 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 ends = [until, live?.until].filter((t): t is string => !!t).map((t) => Date.parse(t));
  const wait = Math.min(Math.min(...ends) - Date.now(), 2 ** 31 - 1);
  lapseTimer = window.setTimeout(lapse, wait);
  if (live) {
    const lead = Date.parse(live.until) - CAP_FLUSH_LEAD_MS - Date.now();
    if (lead > 0 && lead < wait) capFlushTimer = window.setTimeout(() => void flush(), lead);
  }
}

/** Something in force ran out: apply what is left of it (the clock decides), then let `me` catch up. */
function lapse(): void {
  lapseTimer = null;
  const before = applied;
  applied = null;
  applyDiagnostics(lastState, lastSession, onLapsed ?? (() => {}));
  if (applied === "INFO" && before !== "INFO")
    log.info("diagnostics.level", "diagnostics level lapsed", { level: "INFO" });
  onLapsed?.();
}

/** The server has it off: lapse everything this tab thought was in force. */
function serverSaysOff(): void {
  lastState = null;
  lastSession = null;
  lapse();
}

// --- streaming ----------------------------------------------------------------------------------
let queue: LogEvent[] = [];
let unsubscribe: (() => void) | null = null;
let flushTimer: number | null = null;
let inFlight: Promise<void> | null = null;
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 = [];
}

function flush(): Promise<void> {
  if (inFlight) return inFlight;
  if (!queue.length || Date.now() < quietUntil) return Promise.resolve();
  inFlight = sendBatch().finally(() => {
    inFlight = null;
  });
  return inFlight;
}

/** Send everything waiting, now — the recorder awaits it before Stop: once the server has closed
 *  the session it no longer tags (or, at the base level, takes) the events of its last seconds,
 *  which are the ones closest to what went wrong. Gives up on a batch the server turned away. */
export async function flushNow(): Promise<void> {
  for (let guard = 0; guard < 20; guard++) {
    await flush();
    if (inFlight) continue;
    if (!queue.length || !unsubscribe || Date.now() < quietUntil) return;
  }
}

async function sendBatch(): Promise<void> {
  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.
      serverSaysOff();
    }
  }
}

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