textmachine/frontend/src/api/stream.test.ts

// The stream, driven through a stand-in transport. Measured: the test DOM has no `EventSource`
// global at all, and Chromium does have one that MSW's worker intercepts — so the browser covers
// the transport (screenshot cycle) and this covers the rules layered on top of it, which is where
// the mistakes live.

import { afterAll, afterEach, beforeAll, expect, test, vi } from 'vitest';

import { forgetAll } from './revision';
import { subscribeToRun } from './stream';
import type { ConnectionState, Frame } from './stream';

class FakeEventSource {
  static latest: FakeEventSource | null = null;

  readonly CONNECTING = 0;
  readonly OPEN = 1;
  readonly CLOSED = 2;
  readyState = 0;
  onerror: (() => void) | null = null;

  private readonly listeners = new Map<string, ((message: MessageEvent<string>) => void)[]>();

  // Plain field and assignment: a parameter property is not erasable syntax, and the project
  // compiles with `erasableSyntaxOnly`.
  readonly url: string;

  constructor(url: string) {
    this.url = url;
    FakeEventSource.latest = this;
  }

  addEventListener(name: string, listener: (message: MessageEvent<string>) => void): void {
    this.listeners.set(name, [...(this.listeners.get(name) ?? []), listener]);
  }

  close(): void {
    this.readyState = this.CLOSED;
  }

  emit(name: string, data: unknown, id: string): void {
    const message = { data: JSON.stringify(data), lastEventId: id } as MessageEvent<string>;
    for (const listener of this.listeners.get(name) ?? []) listener(message);
  }
}

const original = Reflect.get(globalThis, 'EventSource') as unknown;
beforeAll(() => {
  Reflect.set(globalThis, 'EventSource', FakeEventSource);
});
afterAll(() => {
  Reflect.set(globalThis, 'EventSource', original);
});
afterEach(() => {
  forgetAll();
});

function listen() {
  const frames: Frame[] = [];
  const states: ConnectionState[] = [];
  const resyncs: string[] = [];
  const close = subscribeToRun({
    runId: 'run_41',
    bookId: 'bk_1',
    onFrame: (frame) => frames.push(frame),
    onState: (state) => states.push(state),
    onResync: (reason) => resyncs.push(reason),
  });
  const source = FakeEventSource.latest as FakeEventSource;
  return { frames, states, resyncs, close, source };
}

const progress = (done: number) => ({
  progress: { draft: { done, total: 100 }, edit: { done: 0, total: 100 } },
});

test('the handshake opens the stream and the frames arrive typed', () => {
  const { frames, states, source } = listen();
  source.emit('hello', { contract: '0.2.0', run_id: 'run_41', revision: 10 }, '10');
  source.emit('progress', progress(7), '11');

  expect(states).toEqual(['connecting', 'open']);
  expect(frames).toHaveLength(1);
  expect(frames[0]).toMatchObject({ event: 'progress' });
});

test('a major version this build does not speak closes the stream instead of guessing', () => {
  const { frames, states, source } = listen();
  source.emit('hello', { contract: '1.0.0', run_id: 'run_41', revision: 10 }, '10');
  source.emit('progress', progress(7), '11');

  expect(states).toContain('refused');
  expect(source.readyState).toBe(source.CLOSED);
  expect(frames).toHaveLength(0);
});

test('a frame older than one already applied is dropped, and an equal one is not', () => {
  const { frames, source } = listen();
  source.emit('hello', { contract: '0.2.0', run_id: 'run_41', revision: 10 }, '10');
  source.emit('progress', progress(20), '20');
  // A sibling of the frame just applied: one transaction is one revision but several frames, so
  // rejecting equality would throw away the siblings — the same reason catch-up reads `>=`.
  source.emit('chapter', { chapter_id: 'ch_1', units_done: 1, note_count: 0 }, '20');
  // Genuinely stale.
  source.emit('progress', progress(5), '12');

  expect(frames.map((frame) => frame.event)).toEqual(['progress', 'chapter']);
});

test('resync is reported and does not reach the frame handler', () => {
  const { frames, resyncs, source } = listen();
  source.emit('hello', { contract: '0.2.0', run_id: 'run_41', revision: 10 }, '10');
  source.emit('resync_required', { reason: 'the book was rebuilt' }, '30');

  expect(resyncs).toEqual(['the book was rebuilt']);
  expect(frames).toHaveLength(0);
});

test('after a resync the guard starts over, so the fresh numbering is not rejected', () => {
  const { frames, source } = listen();
  source.emit('hello', { contract: '0.2.0', run_id: 'run_41', revision: 900 }, '900');
  source.emit('progress', progress(50), '900');
  source.emit('resync_required', { reason: 'the bank was rebuilt' }, '901');
  // A full replacement may restart the numbering; a guard that kept the old high-water mark would
  // silently swallow every frame of the new epoch.
  source.emit('progress', progress(1), '1');

  expect(frames).toHaveLength(2);
});

// The defect this locks: the stream writes into the SAME cache the reads fill, so a frame that
// skipped the narrowing would put a raw wire value where every screen expects a narrowed one — the
// Ф-22 crash arriving by the other door. Found by re-reading, not by the typechecker: the wire type
// and the narrowed type are both assignable to a string field.
test('a frame passes the same vocabulary seam a read does', () => {
  const { frames, source } = listen();
  source.emit('hello', { contract: '0.2.0', run_id: 'run_41', revision: 1 }, '1');
  source.emit('status', { status: 'paused', paused_reason: 'credit_exhausted' }, '2');
  source.emit(
    'status',
    { status: 'from_the_future', paused_reason: 'reason_from_the_future' },
    '3',
  );
  source.emit('note', { note: { severity: 'screaming', message: 'something' } }, '4');

  expect(frames[0]).toMatchObject({
    event: 'status',
    data: { status: 'paused', paused_reason: 'credit_exhausted' },
  });
  // Unknown becomes null here, once, rather than reaching a component that indexes a map by it.
  expect(frames[1]).toMatchObject({ event: 'status', data: { status: null, paused_reason: null } });
  expect(frames[2]).toMatchObject({ event: 'note', data: { note: { severity: null } } });
});

test('a dropped connection is reported without taking the reads down with it', () => {
  const { states, source } = listen();
  source.emit('hello', { contract: '0.2.0', run_id: 'run_41', revision: 10 }, '10');

  source.readyState = source.CONNECTING;
  source.onerror?.();
  expect(states.at(-1)).toBe('lost');

  source.readyState = source.CLOSED;
  source.onerror?.();
  expect(states.at(-1)).toBe('refused');
});

test('unsubscribing closes the transport once', () => {
  const { close, source } = listen();
  const closed = vi.spyOn(source, 'close');
  close();
  close();
  expect(closed).toHaveBeenCalledTimes(1);
});