textmachine/frontend/src/showcase/useRunStream.ts

// Live progress, wired into the query cache. One place, because the stream and the reads answer
// the same questions and two copies of the answer would drift on the first busy run.
//
// The stream WRITES into the cache rather than into a state of its own: server data has exactly one
// home (FRONTEND_PLAN.md rule 5), and a component that read "progress from the stream, everything
// else from the query" would be the second home.

import { useQueryClient } from '@tanstack/react-query';
import { useEffect, useState } from 'react';

import { bookScope, keys, subscribeToRun } from '../api';
import type { BookDetail, ChapterList, ConnectionState, Frame, Id } from '../api';

export function useRunStream(bookId: Id | undefined, runId: Id | undefined): ConnectionState {
  const client = useQueryClient();
  const [state, setState] = useState<ConnectionState>('connecting');

  // ⚠ The run id travels INSIDE the book card, and a resync empties that card for the length of one
  // read — so for that moment the caller has no id to give, and the subscription would be torn down
  // and opened again over a gap in the data that merely DESCRIBES it. Measured, not reasoned:
  // with an answer that takes 120 ms the stream opened twice (the acceptance of S3.7 asked for
  // exactly this measurement, and the earlier claim that it does not happen was taken against an
  // instant mock).
  //
  // Held per BOOK: another book is another stream, and there the id must not survive. Legal by the
  // contract as well — `run` is "the current or last run", so once a book has one it never goes
  // back to none.
  // State and not a ref, although a ref is what this looks like: a ref may not be read while
  // rendering (`react-hooks/refs`), and the value is needed to decide what to render.
  const [held, setHeld] = useState<{ bookId?: Id; runId?: Id }>({});
  if (held.bookId !== bookId) {
    setHeld({ bookId, runId });
    // The state belongs to the stream it was measured on. Left alone, the first frame painted for
    // another book carried the previous book's verdict — a status strip saying "live progress is
    // unavailable" about a connection that had not been opened yet.
    setState('connecting');
  } else if (runId !== undefined && runId !== held.runId) setHeld({ bookId, runId });
  const streamId = runId ?? held.runId;

  useEffect(() => {
    if (bookId === undefined || streamId === undefined) return;
    const runId = streamId;

    // Keys whose re-read is already on its way. Per subscription rather than per module: it dies
    // with the stream it belongs to, and nothing leaks into the next book.
    const asked = new Set<string>();

    return subscribeToRun({
      bookId,
      runId,
      onState: setState,
      onFrame: (frame, revision) => {
        apply(frame, revision, bookId, client, asked);
      },
      onResync: () => {
        // The marks go with the snapshots. They say "a re-read of this key is already on its way",
        // and after a full replacement that sentence is about an epoch that no longer exists — a
        // frame arriving in the window right after a resync would be eaten by a mark left from
        // before it (acceptance of S3.7).
        asked.clear();
        // Replaying history is forbidden by the contract, so the only correct answer to "I cannot
        // resume you" is to read the snapshots again.
        //
        // RESET, not invalidate: a resync follows a FULL replacement, where the revision is not
        // promised to keep growing. Invalidating kept the old snapshots, and with them the marks
        // every later read is compared against — the new epoch's lower numbers were then dropped
        // as stale, and the screen stayed on data that no longer exists. The stream side already
        // forgets its own mark; this is the same move on the cache side. Scoped to the book: the
        // library, the usage and a second book were not replaced.
        void client.resetQueries(bookScope(bookId));
      },
    });
  }, [bookId, streamId, client]);

  // Without a run there is no channel, and the answer says so rather than keeping the last word of
  // a stream nobody holds any more — the status strip would go on claiming a live connection
  // (named by the adversarial review). Derived, not stored: state written from an effect is a
  // cascade of renders, and there is nothing here to store that the arguments do not already say.
  return bookId === undefined || streamId === undefined ? 'connecting' : state;
}

type Client = ReturnType<typeof useQueryClient>;

function apply(
  frame: Frame,
  revision: number,
  bookId: Id,
  client: Client,
  asked: Set<string>,
): void {
  switch (frame.event) {
    case 'progress':
      patch<BookDetail>(client, keys.book(bookId), revision, asked, (detail) => ({
        ...detail,
        book: { ...detail.book, progress: frame.data.progress },
      }));
      return;

    case 'status':
      patch<BookDetail>(client, keys.book(bookId), revision, asked, (detail) => ({
        ...detail,
        book: { ...detail.book, status: frame.data.status },
        run: detail.run
          ? { ...detail.run, status: frame.data.status, paused_reason: frame.data.paused_reason }
          : detail.run,
      }));
      // A stage boundary or a stop is exactly where the translated text is written (contract,
      // `Unit`), and an open chapter is on a keep-alive tab that never unmounts — nothing else
      // would ever ask for it again. Whole-book scope on purpose: while `units_done` stays a
      // single counter (K-10) a chapter frame may not arrive once in a whole draft wave.
      rereadAll(client, keys.bookUnits(bookId), asked);
      // The chapter LIST goes stale at the same boundary and for the same reason (the acceptance of
      // S3.7 turned Ф-49's own premise back on it): every row of it carries `units_done` and
      // `note_count`, a stage boundary moves them all, and the per-chapter frame that would patch
      // them may not arrive once in a whole wave while `units_done` stays a single counter (K-10).
      reread(client, keys.chapters(bookId), asked);
      // And the LIBRARY, which carries the same status in its own list: the tree's badge is the
      // main carrier of the state (a word, not a dot), and it went on saying "queued" while the run
      // it names was already halted — seen on the frame of the ceiling stop. Re-read and never
      // patched: the library is a scope of its own with a revision of its own, and stamping it with
      // the book's number would be comparing two counters that are not comparable (contract,
      // `Revision`).
      reread(client, keys.library(), asked);
      return;

    case 'chapter':
      patch<ChapterList>(client, keys.chapters(bookId), revision, asked, (list) => ({
        ...list,
        chapters: list.chapters.map((chapter) =>
          chapter.id === frame.data.chapter_id
            ? { ...chapter, units_done: frame.data.units_done, note_count: frame.data.note_count }
            : chapter,
        ),
      }));
      // The frame carries counters, not text: "the live signal arrives as an event; the text
      // arrives with a read after the boundary". Scoped to the chapter the frame names — re-reading
      // every open tab on every frame would cost a read per tab per burst.
      reread(client, keys.units(bookId, frame.data.chapter_id), asked);
      return;

    // A note is one-shot and a bank change replaces rows wholesale: neither is a delta this client
    // can apply, so both are answered by re-reading rather than by patching.
    case 'note':
      reread(client, keys.notes(bookId), asked);
      return;

    case 'bank':
      reread(client, keys.bank(bookId), asked);
      return;

    // The ceiling halt is a status change the client must not guess at: it re-reads the book and
    // shows whatever the platform now says, including the machine reason for the pause.
    case 'ceiling':
      reread(client, keys.book(bookId), asked);
      return;

    case 'hello':
    case 'resync_required':
      return;
  }
}

/**
 * Asks a key to answer again. EVERY frame goes through here, whether it patches something or only
 * says "this changed" — because both vanish in the same window.
 *
 * ⚠ Reset when the entry is EMPTY, invalidate when it holds something, and the difference is not a
 * nicety: a query with no data ignores an invalidation. The library restarts an in-flight fetch
 * only when data is already there (`Query.fetch`: `state.data !== undefined && cancelRefetch`);
 * otherwise it hands back the promise of the very read that is stale — the read prepared BEFORE the
 * frame, which is what the frame's own mark has already been counted against.
 *
 * The mark keeps that to ONE restart while the entry stays empty. Measured, not assumed: a burst of
 * five frames on an empty entry cost six reads without it and two with it, and the server MAY
 * coalesce, so bursts are the normal shape of a run. What the restarted read misses of a burst the
 * next frame patches, because by then the entry is no longer empty.
 *
 * ⚠ The mark lives exactly as long as the read it stands for, and not a moment longer. Set and left
 * behind — which is what it was — it says "on its way" about a read that ENDED: a read that failed
 * leaves the entry empty, so every later frame for that key found the mark and returned, and the
 * key was silenced for the rest of the run (acceptance of S3.7). A reset that starts no read at all
 * — a key nobody is watching — resolves at once and clears it just the same.
 */
function reread(client: Client, key: readonly unknown[], asked: Set<string>): void {
  const mark = key.join('/');
  if (client.getQueryData(key) !== undefined) {
    asked.delete(mark);
    void client.invalidateQueries({ queryKey: key });
    return;
  }
  if (asked.has(mark)) return;
  asked.add(mark);
  void client.resetQueries({ queryKey: key }).finally(() => asked.delete(mark));
}

/**
 * The same, for a PREFIX that stands for several entries — every chapter of a book. Asked of the
 * cache one exact key at a time rather than of the prefix as a whole: a prefix has no entry of its
 * own, so it would look empty and reset every open chapter at once, blanking readers that were
 * showing text a moment ago.
 */
function rereadAll(client: Client, prefix: readonly unknown[], asked: Set<string>): void {
  for (const query of client.getQueryCache().findAll({ queryKey: prefix })) {
    reread(client, query.queryKey, asked);
  }
}

/**
 * Applies one frame to one cache entry, and moves that entry's revision with it: the revision is
 * what a late read is compared against, and without it the next refetch on focus overwrites the
 * live numbers with older ones.
 *
 * A frame that finds NO entry has nothing to patch, so the only trace it can leave is a re-read.
 */
function patch<T extends { revision: number }>(
  client: Client,
  key: readonly unknown[],
  revision: number,
  asked: Set<string>,
  change: (held: T) => T,
): void {
  const held = client.getQueryData<T>(key);
  if (held === undefined) {
    reread(client, key, asked);
    return;
  }
  asked.delete(key.join('/'));
  client.setQueryData<T>(key, { ...change(held), revision });
}