// The stream against the query cache: a mock network answers the reads, a stand-in transport
// delivers the frames. Every case here is a defect that is invisible from either side alone — the
// frame handler looks correct next to the reads, and the reads look correct next to the frames.
//
// The world is local rather than taken from `src/mock/`: these tests turn on WHEN a read is made
// and at which revision it answers, and a shared fixture would decide both.

import { QueryClient, QueryClientProvider, useQuery } from '@tanstack/react-query';
import { render, waitFor } from '@testing-library/react';
import { HttpResponse, delay, http } from 'msw';
import { setupServer } from 'msw/node';
import { afterAll, afterEach, beforeAll, beforeEach, expect, test } from 'vitest';

import {
  bankQuery,
  bookQuery,
  chaptersQuery,
  keys,
  libraryQuery,
  unitsQuery,
  usageQuery,
} from '../api';
import type { BookDetail, ChapterList, UnitList } from '../api';
import type { components } from '../api/schema';
import { useRunStream } from './useRunStream';

type Schemas = components['schemas'];

const bookId = 'bk_1';
const runId = 'run_1';

/** The book's revision, moved by hand: a frame and a read are ordered against each other by it. */
let revision = 1841;
const asked: string[] = [];
const askedFor = (path: string) => asked.filter((seen) => seen === path).length;

const book: Schemas['Book'] = {
  id: bookId,
  title: 'Gu Zhenren',
  source_lang: 'zh',
  target_lang: 'ru',
  chapter_count: 2,
  added_at: '2026-08-01T00:00:00Z',
  status: 'translating',
  progress: { draft: { done: 3, total: 10 }, edit: { done: 0, total: 10 } },
  note_count: 0,
};

const chapter = (id: string): Schemas['Chapter'] => ({
  id,
  number: 1,
  heading: id,
  units_total: 2,
  units_done: 0,
  note_count: 0,
});

const run: Schemas['Run'] = {
  id: runId,
  revision: 1841,
  status: 'translating',
  verify_bank: true,
  ceiling_chapters: 60,
  paused_reason: null,
  started_at: '2026-08-01T00:00:00Z',
};

const unit: Schemas['Unit'] = {
  id: 'u_1',
  source: '第一节',
  target: 'First',
  state: 'translated',
};

const server = setupServer(
  http.get('*/v0/books', () =>
    HttpResponse.json<Schemas['Library']>({ revision, next_cursor: null, books: [book] }),
  ),
  http.get('*/v0/books/:bookId', () =>
    HttpResponse.json<Schemas['BookDetail']>({ revision, book, run: { ...run, revision } }),
  ),
  http.get('*/v0/books/:bookId/chapters', () =>
    HttpResponse.json<Schemas['ChapterList']>({
      revision,
      next_cursor: null,
      chapters: [chapter('ch_1'), chapter('ch_2')],
    }),
  ),
  http.get('*/v0/books/:bookId/chapters/:chapterId/units', () =>
    HttpResponse.json<Schemas['UnitList']>({ revision, next_cursor: null, units: [unit] }),
  ),
  http.get('*/v0/books/:bookId/bank', () =>
    HttpResponse.json<Schemas['Bank']>({
      revision,
      next_cursor: null,
      total: 1,
      signed: 0,
      terms: [
        {
          id: 't_1',
          src: '方源',
          dst: 'Fang Yuan',
          kind: 'name',
          status: 'approved',
          origin: 'seed',
          sense: '',
          since_chapter: 0,
          until_chapter: 0,
        },
      ],
    }),
  ),
  http.get('*/v0/usage', () =>
    HttpResponse.json<Schemas['Usage']>({
      state: 'ok',
      remaining_percent: 62,
      paused_reason: null,
    }),
  ),
);

beforeAll(() => {
  server.listen({ onUnhandledRequest: 'error' });
  server.events.on('request:start', ({ request }) => asked.push(new URL(request.url).pathname));
  Reflect.set(globalThis, 'EventSource', FakeEventSource);
});
afterAll(() => {
  server.close();
  Reflect.deleteProperty(globalThis, 'EventSource');
});
beforeEach(() => {
  revision = 1841;
  asked.length = 0;
  FakeEventSource.opened = 0;
});
afterEach(() => {
  // Unmounting closes the subscription, and closing it forgets the stream's high-water mark —
  // a mark left behind would make the next test's frames look stale.
  for (const close of mounted.splice(0)) close();
  server.resetHandlers();
});

/** The transport, replaced: the test DOM has no `EventSource` at all (measured — no global). */
class FakeEventSource {
  static latest: FakeEventSource | null = null;
  static opened = 0;

  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)[]>();

  constructor() {
    FakeEventSource.latest = this;
    FakeEventSource.opened += 1;
  }

  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: number): void {
    const message = {
      data: JSON.stringify(data),
      lastEventId: String(id),
    } as MessageEvent<string>;
    for (const listener of this.listeners.get(name) ?? []) listener(message);
  }
}

/** Emits a frame at a revision the server has already moved to — the order a real run has. */
function emit(name: string, data: unknown, at = revision): void {
  revision = at;
  (FakeEventSource.latest as FakeEventSource).emit(name, data, at);
}

const hello = () => {
  emit('hello', { contract: '0.2.0', run_id: runId, revision }, revision);
};

/**
 * The screen as the stream sees it: a live subscription plus the reads that are open at the same
 * time. Two chapters, because "re-read the chapter the frame names" and "re-read everything" are
 * indistinguishable with one.
 */
function Screen({ chapters: open }: { chapters: string[] }) {
  useRunStream(bookId, runId);
  useQuery(bookQuery(bookId));
  useQuery(libraryQuery());
  useQuery(chaptersQuery(bookId));
  useQuery(bankQuery(bookId));
  useQuery(usageQuery());
  return (
    <>
      {open.map((chapterId) => (
        <Chapter key={chapterId} chapterId={chapterId} />
      ))}
    </>
  );
}

function Chapter({ chapterId }: { chapterId: string }) {
  const units = useQuery(unitsQuery(bookId, chapterId));
  return <p>{units.data?.units.length ?? 0}</p>;
}

/**
 * The same screen, wired the way the real one is: the run id arrives INSIDE the book detail. That
 * wiring is what makes emptying the detail a question about the STREAM and not only about the data,
 * and it is the only shape in which the restart below can happen at all.
 */
function WiredScreen() {
  const detail = useQuery(bookQuery(bookId));
  useRunStream(bookId, detail.data?.run?.id);
  useQuery(chaptersQuery(bookId));
  return null;
}

const mounted: (() => void)[] = [];

function mount(open: string[] = [], screen = <Screen chapters={open} />) {
  // Retries off: a failing read would otherwise be answered three times and the counts below would
  // measure the retry policy instead of the invalidation.
  const client = new QueryClient({ defaultOptions: { queries: { retry: false } } });
  const view = render(<QueryClientProvider client={client}>{screen}</QueryClientProvider>);
  mounted.push(() => {
    view.unmount();
    client.clear();
  });
  return client;
}

const settled = (client: QueryClient) =>
  waitFor(() => {
    expect(client.isFetching()).toBe(0);
  });

const unitsOf = (id: string) => `/v0/books/${bookId}/chapters/${id}/units`;

// Ф-49. The open chapter sits on a keep-alive tab: it never unmounts, so none of the three triggers
// of `staleTime` (focus, reconnect, mount) ever fires and nothing else asks for the text again.
test('a chapter frame re-reads THAT chapter and leaves the others alone', async () => {
  const client = mount(['ch_1', 'ch_2']);
  await settled(client);
  expect(askedFor(unitsOf('ch_1'))).toBe(1);

  hello();
  emit('chapter', { chapter_id: 'ch_1', units_done: 2, note_count: 1 }, 1850);

  await waitFor(() => {
    expect(askedFor(unitsOf('ch_1'))).toBe(2);
  });
  await settled(client);
  expect(askedFor(unitsOf('ch_2'))).toBe(1);
  // The counters of the frame land on the chapter list at the frame's revision, as before.
  const list = client.getQueryData<ChapterList>(keys.chapters(bookId));
  expect(list).toMatchObject({ revision: 1850 });
  expect(list?.chapters[0]).toMatchObject({ units_done: 2, note_count: 1 });
});

// Ф-49, second half. While `units_done` stays a single counter (contract, K-10) a whole draft wave
// can pass without one chapter frame, and the text is written at the stage boundary — which arrives
// as a status frame and as nothing else.
test('a status frame re-reads the text of every open chapter', async () => {
  const client = mount(['ch_1', 'ch_2']);
  await settled(client);

  hello();
  emit('status', { status: 'awaiting_bank', paused_reason: null }, 1850);

  await waitFor(() => {
    expect(askedFor(unitsOf('ch_1'))).toBe(2);
    expect(askedFor(unitsOf('ch_2'))).toBe(2);
  });
});

// S4. The same status stands in TWO answers — the book's card and the row of the library — and the
// tree's badge is the one the eye reads first. Patching only the card left the badge saying "queued"
// under a run that had already halted (seen on the frame of a ceiling stop).
// Acceptance of S3.7, finding 8: the counters of every chapter move at a stage boundary too, and
// the frame that would patch a single row may not arrive at all while `units_done` is one counter.
test('a status frame re-reads the chapter list, whose counters move at the same boundary', async () => {
  const client = mount();
  await settled(client);
  expect(askedFor(`/v0/books/${bookId}/chapters`)).toBe(1);

  hello();
  emit('status', { status: 'awaiting_bank', paused_reason: null }, 1850);

  await waitFor(() => {
    expect(askedFor(`/v0/books/${bookId}/chapters`)).toBe(2);
  });
});

test('a status frame re-reads the library, where the same status is a row', async () => {
  const client = mount();
  await settled(client);
  expect(askedFor('/v0/books')).toBe(1);

  hello();
  emit('status', { status: 'paused', paused_reason: 'credit_exhausted' }, 1850);

  await waitFor(() => {
    expect(askedFor('/v0/books')).toBe(2);
  });
});

// Ф-50. The window is every screen start and every resync: the frame arrives while the first read
// is still in flight, so there is nothing in the cache to patch — and the stream has already moved
// its high-water mark, so the answer prepared BEFORE the frame is taken as fresh and the frame is
// gone without a trace.
test('a frame that finds no snapshot yet leaves a trace instead of vanishing', async () => {
  const client = mount();

  hello();
  emit('status', { status: 'paused', paused_reason: 'credit_exhausted' }, 1850);
  emit('chapter', { chapter_id: 'ch_1', units_done: 2, note_count: 1 }, 1851);

  await settled(client);
  // Two reads of each, not one: the first answer was prepared before the frame and cannot carry
  // it, so the frame's own re-read is what puts the new state on the screen.
  expect(askedFor(`/v0/books/${bookId}`)).toBe(2);
  expect(askedFor(`/v0/books/${bookId}/chapters`)).toBe(2);
});

// The same window, the frames that only SAY "this changed": a note, a bank rebuild, a ceiling halt.
// They were left on `invalidateQueries` by the first pass of this fix and vanished exactly as the
// patched ones did — found by the adversarial review, not by me.
test('a frame that only says "this changed" leaves the same trace', async () => {
  const client = mount();

  hello();
  emit('bank', { total: 60, signed: 12, pending_decisions: 48 }, 1850);
  emit('ceiling', { halted: true }, 1851);

  await settled(client);
  expect(askedFor(`/v0/books/${bookId}/bank`)).toBe(2);
  expect(askedFor(`/v0/books/${bookId}`)).toBe(2);
});

// The cost of that trace, measured rather than assumed: the server MAY coalesce, so frames arrive
// in bursts, and a burst landing on the same empty entry must not restart its read once per frame.
test('a burst of frames on an empty cache costs one re-read, not one per frame', async () => {
  const client = mount();

  hello();
  for (let step = 0; step < 5; step += 1) {
    emit('progress', { progress: book.progress }, 1850 + step);
  }

  await settled(client);
  expect(askedFor(`/v0/books/${bookId}`)).toBe(2);
});

// Acceptance of S3.7, finding 2: the mark says "a re-read of this key is on its way", and it was
// never taken off. A read that FAILS leaves the entry empty, so from then on every frame for that
// key found the mark and returned — the key was silenced for the rest of the run, on a screen whose
// reads had recovered long ago.
test('a key whose re-read failed is not silenced for the rest of the run', async () => {
  let refuse = true;
  server.use(
    http.get('*/v0/books/:bookId/bank', () => {
      if (refuse) return HttpResponse.json({ title: 'no' }, { status: 503 });
      return HttpResponse.json<Schemas['Bank']>({
        revision,
        next_cursor: null,
        total: 1,
        signed: 1,
        terms: [],
      });
    }),
  );
  const client = mount();
  await settled(client);
  const before = askedFor(`/v0/books/${bookId}/bank`);

  hello();
  emit('bank', { total: 60, signed: 12, pending_decisions: 48 }, 1850);
  await settled(client);
  refuse = false;

  // The second frame has to reach the network. On the pre-fix code the mark from the first one was
  // still there, and this read never happened.
  emit('bank', { total: 60, signed: 13, pending_decisions: 47 }, 1851);
  await waitFor(() => {
    expect(askedFor(`/v0/books/${bookId}/bank`)).toBeGreaterThan(before + 1);
  });
});

// Acceptance of S3.7, finding 3: the marks belong to an epoch. A resync throws the snapshots away,
// and a mark left from before it eats the first frame of the new epoch — exactly in the window
// where the client has nothing else to go on.
test('a resync clears the marks with the snapshots', async () => {
  // The read is SLOW on purpose: the window this guards is the one where the resync's own re-read
  // has not landed yet, so the cache is empty and the mark is the only thing deciding. With a fast
  // read the data arrives first and the mark is bypassed — the test would pass on the defect.
  server.use(
    http.get('*/v0/books/:bookId/bank', async () => {
      await delay(120);
      return HttpResponse.json<Schemas['Bank']>({
        revision,
        next_cursor: null,
        total: 1,
        signed: 1,
        terms: [],
      });
    }),
  );
  const client = mount();

  hello();
  // NOT settled first, and that is the point: the first read is still in flight, so this frame
  // finds no snapshot to patch, leaves a mark and restarts the read.
  emit('bank', { total: 60, signed: 12, pending_decisions: 48 }, 1850);
  emit('resync_required', { reason: 'the book was rebuilt' }, 12);
  const before = askedFor(`/v0/books/${bookId}/bank`);

  // Inside the window: the re-read is still in flight, the cache is empty, and this frame must
  // still reach the network. On the pre-fix code the mark of the dead epoch ate it.
  emit('bank', { total: 60, signed: 13, pending_decisions: 47 }, 13);
  await waitFor(() => {
    expect(askedFor(`/v0/books/${bookId}/bank`)).toBeGreaterThan(before);
  });
  await settled(client);
});

// Ф-51. A resync follows a FULL replacement, where the revision is not promised to keep growing —
// so the snapshot's revision has to go with the snapshot.
test('a resync drops the book from the cache, marks and all', async () => {
  const client = mount(['ch_1']);
  await settled(client);
  client.setQueryData<BookDetail>(keys.book(bookId), (held) =>
    held === undefined ? held : { ...held, revision: 9000 },
  );

  hello();
  // The new epoch numbers lower than what was held. A keyless invalidation kept the 9000, and the
  // read guard then dropped every answer of the new epoch as stale.
  emit('resync_required', { reason: 'book rebuilt' }, 12);

  await settled(client);
  expect(client.getQueryData<BookDetail>(keys.book(bookId))?.revision).toBe(12);
  expect(client.getQueryData<UnitList>(keys.units(bookId, 'ch_1'))?.units).toHaveLength(1);
});

// The cost the resync must NOT pay: the run id travels inside the book detail, so emptying that
// detail took the subscription down with it and opened a second connection. Found by the
// adversarial review of this very fix, not by the fix's own tests.
//
// ⚠ The read is SLOW here, and that is the acceptance of S3.7 talking (finding 3): with an instant
// answer the observer may never render between the reset and the new data, so the test would be
// green on a stream that does restart. The window is opened on purpose and measured inside it.
test('a resync does not restart the live stream, even while the re-read is in flight', async () => {
  server.use(
    http.get('*/v0/books/:bookId', async () => {
      await delay(120);
      return HttpResponse.json<Schemas['BookDetail']>({
        revision,
        book,
        run: { ...run, revision },
      });
    }),
  );
  const client = mount([], <WiredScreen />);
  await settled(client);
  expect(FakeEventSource.opened).toBe(1);

  hello();
  emit('resync_required', { reason: 'the book was rebuilt' }, 12);

  // Inside the window: the card is empty and the answer has not come back yet.
  await waitFor(() => {
    expect(client.getQueryData<BookDetail>(keys.book(bookId))).toBeUndefined();
  });
  expect(FakeEventSource.opened).toBe(1);

  await settled(client);
  await waitFor(() => {
    expect(client.getQueryData<BookDetail>(keys.book(bookId))?.revision).toBe(12);
  });
  expect(FakeEventSource.opened).toBe(1);
});

test('a resync of one book does not touch the library, the usage or another book', async () => {
  const client = mount();
  await settled(client);
  client.setQueryData<Schemas['Usage']>(keys.usage(), {
    state: 'low',
    remaining_percent: 3,
    paused_reason: null,
  });
  const other: ChapterList = { revision: 7, next_cursor: null, chapters: [] };
  client.setQueryData<ChapterList>(keys.chapters('bk_other'), other);

  hello();
  emit('resync_required', { reason: 'bank rebuilt' }, 12);

  await settled(client);
  expect(client.getQueryData<Schemas['Usage']>(keys.usage())).toMatchObject({ state: 'low' });
  expect(client.getQueryData<ChapterList>(keys.chapters('bk_other'))).toBe(other);
});