textmachine/frontend/src/api/stream.ts

// Live progress. `EventSource` and nothing else: the transport is ratified (STACK_DECISIONS §5),
// and the gate forbids it anywhere but here.
//
// Two things the browser already does for us, and doing them again would be a second carrier of
// the same rule: it re-sends `Last-Event-ID` on reconnect, and it reconnects on its own. What is
// ours is the version handshake, the revision guard and the resync answer.

import { acceptFrame, forget } from './revision';
import { narrow } from './contract';
import { bookStatus, pausedReason } from './vocabulary';
import type { BookStatus, Note, PausedReason } from './contract';
import type { components } from './schema';

type Schemas = components['schemas'];

/** Major version this build speaks. A major bump means refuse and tell the user, per the contract. */
export const supportedMajor = 0;

export type ConnectionState = 'connecting' | 'open' | 'lost' | 'refused';

// The vocabularies of a frame are narrowed exactly like those of a read. The stream is the SECOND
// door into the application, and a door that skipped the narrowing would put a raw wire value into
// the same cache the reads fill — the crash of BACKLOG Ф-22, arriving by the other route.
export type Frame =
  | { event: 'hello'; data: Schemas['EventHello'] }
  | { event: 'status'; data: { status: BookStatus | null; paused_reason: PausedReason | null } }
  | { event: 'progress'; data: Schemas['EventProgress'] }
  | { event: 'chapter'; data: Schemas['EventChapter'] }
  | { event: 'note'; data: { note: Note } }
  | { event: 'bank'; data: Schemas['EventBank'] }
  | { event: 'ceiling'; data: Schemas['EventCeiling'] }
  | { event: 'resync_required'; data: Schemas['EventResyncRequired'] };

/** Frame names of version 0.x. A name outside this list is ignored, not an error: minor bumps add. */
const frameNames = [
  'hello',
  'status',
  'progress',
  'chapter',
  'note',
  'bank',
  'ceiling',
  'resync_required',
] as const;

/**
 * Turns one wire frame into a typed one. Separate from the transport on purpose: `EventSource`
 * does not exist under the test DOM (measured — no global at all), so the dispatch has to be
 * reachable without a browser.
 */
export function readFrame(name: string, data: string): Frame | null {
  if (!(frameNames as readonly string[]).includes(name)) return null;

  let payload: unknown;
  try {
    payload = JSON.parse(data);
  } catch {
    // A frame we cannot parse is dropped rather than thrown: one bad frame must not take the
    // stream down and with it the live progress of an hours-long run.
    return null;
  }

  if (name === 'status') {
    const wire = payload as Schemas['EventStatus'];
    return {
      event: 'status',
      data: {
        status: bookStatus.read(wire.status),
        paused_reason: wire.paused_reason === null ? null : pausedReason.read(wire.paused_reason),
      },
    };
  }
  if (name === 'note') {
    const wire = payload as Schemas['EventNote'];
    return { event: 'note', data: { note: narrow.note(wire.note) } };
  }
  return { event: name, data: payload } as Frame;
}

export function majorOf(version: string): number {
  return Number.parseInt(version.split('.')[0] ?? '', 10);
}

export interface Subscription {
  runId: string;
  /** Revision scope. The frame `id` is the BOOK's revision, so the guard is keyed by book. */
  bookId: string;
  /** The frame, and the book revision it carries — the number a late read is compared against. */
  onFrame: (frame: Frame, revision: number) => void;
  onState: (state: ConnectionState) => void;
  /** The stream cannot be resumed: re-read the snapshots. Replaying history is forbidden. */
  onResync: (reason: string) => void;
}

export function subscribeToRun(subscription: Subscription): () => void {
  const scope = `stream:${subscription.bookId}`;
  const source = new EventSource(`/v0/runs/${encodeURIComponent(subscription.runId)}/events`);
  let closed = false;

  const close = () => {
    if (closed) return;
    closed = true;
    source.close();
    forget(scope);
  };

  subscription.onState('connecting');

  const handle = (message: MessageEvent<string>, name: string) => {
    // A closed subscription must stay closed even if a frame is already in flight: the effect that
    // owns it has been torn down, and delivering into it would write to a screen that is gone.
    if (closed) return;

    const frame = readFrame(name, message.data);
    if (!frame) return;

    // The handshake is the stream's own business and does not reach the subscriber.
    if (frame.event === 'hello') {
      if (majorOf(frame.data.contract) !== supportedMajor) {
        subscription.onState('refused');
        close();
        return;
      }
      subscription.onState('open');
      return;
    }

    // The frame id is the book revision. A frame older than what has been applied is dropped here,
    // once, rather than in every handler downstream. A resync is exempt: it is the answer to "I
    // cannot resume you", so filtering it by the mark it exists to reset would strand the client.
    const revision = Number(message.lastEventId);
    if (frame.event === 'resync_required') {
      subscription.onResync(frame.data.reason);
      forget(scope);
      return;
    }
    if (message.lastEventId !== '' && !acceptFrame(scope, revision)) return;

    subscription.onFrame(frame, Number.isFinite(revision) ? revision : 0);
  };

  for (const name of frameNames) {
    source.addEventListener(name, (message: MessageEvent<string>) => {
      handle(message, name);
    });
  }

  source.onerror = () => {
    // CONNECTING means the browser is retrying by itself and will re-send `Last-Event-ID`; CLOSED
    // means it gave up. The screen says "live progress is unavailable" either way, and the reads
    // keep working — that separation is the whole point of pushing progress rather than polling.
    subscription.onState(source.readyState === source.CLOSED ? 'refused' : 'lost');
  };

  return close;
}