cyclone/src/lib/tail-stream.ts

// ---------------------------------------------------------------------------
// Live-tail NDJSON stream parser (sub-project 5, Phase 4 Task 16).
//
// Opens `GET /api/{resource}/stream` with `Accept: application/x-ndjson`,
// pipes the response body through a `TextDecoderStream` + a manual
// newline-splitter, and yields one typed `TailEvent` per line. The
// higher-level `useTailStream` hook (Phase 5) is what subscribes to this
// generator and dispatches into `tail-store`; this module is intentionally
// pure parsing — no React, no Zustand, no fetch options other than the
// ones documented in the spec.
// ---------------------------------------------------------------------------

/**
 * One event yielded by `streamTail`. The data shapes mirror the backend's
 * `GET /api/{resource}/stream` NDJSON contract (see `backend/api.py`).
 *
 * `item.data` is intentionally `unknown` — the consumer casts to the
 * page-specific shape (Claim / Remittance / Activity). The other variants
 * carry the data shape the spec mandates, so consumers get a friendly
 * type for the non-item events.
 */
export type TailEvent =
  | { type: "item"; data: unknown }
  | { type: "snapshot_end"; data: { count: number } }
  | { type: "heartbeat"; data: { ts: string } }
  | { type: "item_dropped"; data: { id: string } }
  | { type: "error"; data: { message: string } };

export type TailResource =
  | "claims"
  | "remittances"
  | "activity"
  | "acks"
  | "ta1_acks"
  | "claim-acks";

export interface StreamTailOptions {
  /** Forwarded to `fetch`; aborting the controller cleanly exits the generator. */
  signal?: AbortSignal;
  /** Override the base URL (defaults to "" — i.e. same-origin). Used in tests. */
  baseUrl?: string;
}

const KNOWN_TYPES: ReadonlySet<TailEvent["type"]> = new Set([
  "item",
  "snapshot_end",
  "heartbeat",
  "item_dropped",
  "error",
]);

/**
 * Open a live-tail NDJSON stream and yield one typed event per line.
 *
 * The generator is single-use: call `gen.return()` (or break out of a
 * `for await` loop) to close the underlying fetch. When `opts.signal`
 * aborts, the generator exits cleanly without throwing.
 *
 * @example
 *   for await (const ev of streamTail("claims")) {
 *     if (ev.type === "item") console.log(ev.data);
 *   }
 */
export async function* streamTail(
  resource: TailResource,
  opts?: StreamTailOptions,
): AsyncIterableIterator<TailEvent> {
  const base = opts?.baseUrl ?? "";
  const url = `${base}/api/${resource}/stream`;

  let res: Response;
  try {
    res = await fetch(url, {
      headers: { Accept: "application/x-ndjson" },
      signal: opts?.signal,
    });
  } catch (err) {
    // Aborting the signal makes fetch reject with a DOMException; the spec
    // says "if signal aborts, exit cleanly (don't throw)", so we just
    // return — the consumer sees a completed iterator.
    if (opts?.signal?.aborted) return;
    throw err;
  }

  if (!res.ok) {
    throw new Error(`stream open failed: ${res.status}`);
  }
  if (!res.body) {
    throw new Error("stream has no body");
  }

  // TextDecoderStream handles the chunked UTF-8 boundary problem for us;
  // after this, we have a stream of decoded strings. We then split each
  // chunk on \n and JSON.parse each non-empty line.
  const stringStream = res.body.pipeThrough(new TextDecoderStream());
  const reader = stringStream.getReader();

  let buffer = "";
  try {
    // eslint-disable-next-line no-constant-condition
    while (true) {
      // Check the abort signal between reads so a quick abort doesn't
      // require us to wait for a network read to complete.
      if (opts?.signal?.aborted) return;

      let chunk: string | undefined;
      let done: boolean;
      try {
        const result = await reader.read();
        chunk = result.value;
        done = result.done;
      } catch (err) {
        if (opts?.signal?.aborted) return;
        throw err;
      }

      if (done) break;
      if (chunk) buffer += chunk;

      let nl = buffer.indexOf("\n");
      while (nl !== -1) {
        const line = buffer.slice(0, nl).replace(/\r$/, "");
        buffer = buffer.slice(nl + 1);
        if (line.length > 0) {
          let parsed: unknown;
          try {
            parsed = JSON.parse(line);
          } catch (err) {
            // Malformed lines are the backend's bug, not ours. Log and
            // continue so a single bad frame doesn't kill the stream.
            console.warn(
              "tail-stream: malformed NDJSON line, skipping",
              { line, err },
            );
            nl = buffer.indexOf("\n");
            continue;
          }

          if (
            typeof parsed === "object" &&
            parsed !== null &&
            "type" in parsed &&
            typeof (parsed as { type: unknown }).type === "string" &&
            KNOWN_TYPES.has((parsed as { type: string }).type as TailEvent["type"])
          ) {
            yield parsed as TailEvent;
          } else {
            // Unknown / wrong-shape line — log and skip. A future
            // backend event type shouldn't crash the page.
            console.warn(
              "tail-stream: unknown event shape, skipping",
              { line },
            );
          }
        }
        nl = buffer.indexOf("\n");
      }
    }

    // Flush any trailing partial line (no terminating newline).
    const tail = buffer.replace(/\r$/, "");
    if (tail.length > 0) {
      try {
        const parsed = JSON.parse(tail) as unknown;
        if (
          typeof parsed === "object" &&
          parsed !== null &&
          "type" in parsed &&
          typeof (parsed as { type: unknown }).type === "string" &&
          KNOWN_TYPES.has((parsed as { type: string }).type as TailEvent["type"])
        ) {
          yield parsed as TailEvent;
        } else {
          console.warn(
            "tail-stream: unknown trailing event shape, skipping",
            { line: tail },
          );
        }
      } catch (err) {
        console.warn(
          "tail-stream: malformed trailing NDJSON line, skipping",
          { line: tail, err },
        );
      }
    }
  } finally {
    // Release the reader lock so the underlying connection can close
    // when the test / consumer drops the iterator.
    try {
      reader.releaseLock();
    } catch {
      // already released — ignore
    }
  }
}