/**
 * Realtime integration (TASK-0092 slice 4): the flush-rebase exercised
 * over the PRODUCTION persistence path — the room's `flush()` persists
 * through a real `HttpDocStore` (a `node:http` fake standing in for the
 * API's `/internal/*` surface, not an in-memory seam). This spans the
 * seams no unit test covers: room `flush()` → `HttpDocStore.persist` →
 * the HTTP flush's `pages.version` CAS → the conflict response
 * (`externalBlocks`) → the room's slice-1 rebase (absorb the REST write
 * for blocks it didn't touch) → the retry persist at the fresh version.
 *
 * Acceptance criterion 2 (flush-rebase): "a concurrent REST write … while
 * the room is dirty → the flush's CAS fails → the rebase path runs → the
 * final state has BOTH the REST write and the CRDT edits (no clobber), and
 * the next flush is a no-op." Same contract, but the concurrent writer now
 * arrives through the real flush-CAS over HTTP (the slice-3 REST batch and
 * the flush serialize on the one `pages.version` column).
 *
 * Docker-free: the API is faked (no Postgres); the store, server, and
 * client are the real production classes. The Postgres-backed
 * `PrismaDocStore` variant of this scenario is the Docker-gated DoD item
 * (Testcontainers), recorded in the task file.
 */
import { createServer, type IncomingMessage, type Server, type ServerResponse } from "node:http";
import { AddressInfo } from "node:net";
import { afterAll, beforeAll, describe, expect, it } from "vitest";
import * as Y from "yjs";
import { HttpDocStore } from "@factory/realtime/src/http-store";
import {
  buildRealtimeFixture,
  connectClient,
  type RealtimeFixture,
  type RtClient,
} from "./rt-helpers";
import {
  yDocToBlockList,
  FRAGMENT_NAME,
  type BlockLike,
} from "@factory/collaboration/codec";

const TOKEN = "svc-token";

const PARA = (id: string, text: string, position: string): BlockLike => ({
  id,
  type: "paragraph",
  content: { text },
  parentId: null,
  position,
});

interface WireBlock {
  id: string;
  parentId: string | null;
  type: string;
  position: string;
  content: Record<string, unknown>;
  properties?: Record<string, unknown>;
}

interface FakeApi {
  state: {
    pageId: string;
    blocks: WireBlock[];
    version: number;
    casBase: number;
  };
  /** Simulate a REST batch (slice 3): the write lands + the CAS advanced
   *  the version (a concurrent non-Yjs writer). */
  restWrite: (blocks: WireBlock[]) => void;
  server: Server;
  url: string;
}

/** The API's `/internal/blocks/*` surface, faked over node:http (the same
 *  wire contract the API implements: cursor read + flush CAS). */
async function startFakeApi(pageId: string): Promise<FakeApi> {
  const state = {
    pageId,
    blocks: [] as WireBlock[],
    version: 1,
    casBase: 1,
  };
  const server = createServer((req: IncomingMessage, res: ServerResponse) => {
    const send = (status: number, body: unknown) => {
      res.writeHead(status, { "content-type": "application/json" });
      res.end(JSON.stringify(body));
    };
    if (req.headers.authorization !== `Bearer ${TOKEN}`) {
      return send(401, { error: { code: "UNAUTHORIZED", message: "bad token" } });
    }
    if (req.method === "GET" && typeof req.url === "string") {
      const u = new URL(req.url, "http://x");
      if (u.pathname === `/internal/blocks/${pageId}`) {
        const pageSize = Number(u.searchParams.get("pageSize") ?? "500");
        const cursor = u.searchParams.get("cursor");
        let rows = state.blocks;
        if (cursor !== null) rows = rows.filter((b) => b.position > (cursor as string));
        const slice = rows.slice(0, pageSize);
        const hasMore = rows.length > pageSize;
        const last = slice[slice.length - 1];
        return send(200, {
          pageId,
          version: state.version,
          blocks: slice,
          nextCursor: hasMore && last ? last.position : null,
          hasMore,
        });
      }
      return send(404, { error: { code: "NOT_FOUND", message: "no page" } });
    }
    if (req.method === "POST" && req.url === "/internal/blocks/flush") {
      let raw = "";
      req.on("data", (c) => (raw += c));
      req.on("end", () => {
        const body = JSON.parse(raw) as { pageId: string; baseVersion: number; blocks: WireBlock[] };
        if (body.pageId !== state.pageId) return send(404, { error: { code: "NOT_FOUND", message: "no page" } });
        if (body.baseVersion !== state.casBase) {
          return send(200, {
            version: state.version,
            conflict: { externalBlocks: state.blocks },
          });
        }
        state.blocks = body.blocks;
        state.version = state.casBase + 1;
        state.casBase = state.version;
        return send(200, { version: state.version });
      });
      return;
    }
    send(404, { error: { code: "NOT_FOUND", message: "no route" } });
  });
  await new Promise<void>((r) => server.listen(0, "127.0.0.1", () => r()));
  const addr = server.address() as AddressInfo;
  return {
    state,
    restWrite: (blocks) => {
      // A REST batch (slice 3): the write lands + the version advances
      // (the flush's CAS now sees a stale base → the conflict path).
      state.blocks = blocks;
      state.version += 1;
      state.casBase = state.version;
    },
    server,
    url: `http://127.0.0.1:${addr.port}`,
  };
}

function applyAll(doc: Y.Doc, c: RtClient): void {
  for (const f of c.textFrames()) {
    if (f.type === "sync.state" && typeof f.chunk === "string" && f.chunk.length > 0) {
      Y.applyUpdate(doc, Buffer.from(f.chunk, "base64"), "it");
    }
  }
  for (const b of c.binaryFrames()) {
    if (b.kind === "update") Y.applyUpdate(doc, b.bytes, "it");
  }
}

async function joinAndSync(fx: RealtimeFixture, room: string): Promise<{ client: RtClient; doc: Y.Doc }> {
  const client = await connectClient(fx, { room });
  await client.nextFrame("hello");
  client.sendText({ type: "sync.request" });
  await client.nextFrameWhere((f) => f.type === "sync.state" && f.done === true, 8_000);
  const doc = new Y.Doc();
  applyAll(doc, client);
  return { client, doc };
}

function textSlotOf(doc: Y.Doc, blockId: string): Y.XmlText {
  const fragment = doc.getXmlFragment(FRAGMENT_NAME);
  const walk = (els: Y.XmlElement[]): Y.XmlElement | null => {
    for (const el of els) {
      if (el.getAttribute("id") === blockId) return el;
      const nested = el.toArray().filter((c): c is Y.XmlElement => c instanceof Y.XmlElement);
      const found = walk(nested);
      if (found !== null) return found;
    }
    return null;
  };
  const el = walk(fragment.toArray() as Y.XmlElement[]);
  if (el === null) throw new Error(`block "${blockId}" not found in the doc`);
  const text = el.toArray().find((c): c is Y.XmlText => c instanceof Y.XmlText && c.getAttribute("name") === "text");
  if (text === undefined) throw new Error(`block "${blockId}" has no text slot`);
  return text;
}

function blockTextUpdate(doc: Y.Doc, blockId: string, append: string): Uint8Array {
  const before = Y.encodeStateAsUpdate(doc);
  const slot = textSlotOf(doc, blockId);
  slot.insert(slot.length, append);
  return Y.encodeStateAsUpdate(doc, before);
}

async function editBlock(fx: RealtimeFixture, client: RtClient, doc: Y.Doc, blockId: string, append: string): Promise<void> {
  void fx;
  client.sendBinary(1, blockTextUpdate(doc, blockId, append));
  await client.nextFrame("ack");
}

describe("flush-rebase over the HTTP store (TASK-0092 slice 4, acceptance criterion 2)", () => {
  const pageId = "p-http";
  let fx: RealtimeFixture;
  let api: FakeApi;
  let store: HttpDocStore;

  beforeAll(async () => {
    api = await startFakeApi(pageId);
    store = new HttpDocStore({ apiUrl: api.url, serviceToken: TOKEN });
    // Seed through the (faked) API, NOT the in-memory store — the server
    // persists to the HttpDocStore, so the fake IS the database.
    api.state.blocks = [
      { id: "a", parentId: null, type: "paragraph", position: "1", content: { text: "base a" } },
      { id: "b", parentId: null, type: "paragraph", position: "2", content: { text: "base b" } },
    ];
    fx = await buildRealtimeFixture({
      pages: new Set([pageId]),
      workspaces: { [pageId]: "w1" },
      levels: { [`page:${pageId}`]: "edit" },
      store,
    });
  });
  afterAll(async () => {
    await fx.stop();
    await new Promise<void>((r) => api.server.close(() => r()));
  });

  it("hydration over HTTP: the joiner's doc == the API's blocks (the production hydrate path)", async () => {
    const { client, doc } = await joinAndSync(fx, `page:${pageId}`);
    try {
      const blocks = yDocToBlockList(doc);
      expect(blocks.map((b) => b.id).sort()).toEqual(["a", "b"]);
      const a = blocks.find((b) => b.id === "a")!;
      expect(a.content).toEqual({ text: "base a" });
      // The room's CAS base == the API's version (1).
      expect(api.state.version).toBe(1);
    } finally {
      client.close();
      doc.destroy();
    }
  });

  it("a concurrent REST write + a room edit on DIFFERENT blocks → both survive (no clobber), next flush is a no-op", async () => {
    const { client, doc } = await joinAndSync(fx, `page:${pageId}`);
    try {
      // The room's client edits block "a" (the CRDT path).
      await editBlock(fx, client, doc, "a", "+room edit");
      // A concurrent REST write (bypassing the room) edits block "b" +
      // advances pages.version — exactly the race the flush's HTTP CAS
      // detects (the slice-3 REST batch and the flush on one column).
      api.restWrite([
        { id: "a", parentId: null, type: "paragraph", position: "1", content: { text: "base a" } },
        { id: "b", parentId: null, type: "paragraph", position: "2", content: { text: "base b+rest edit" } },
      ]);
      const versionAfterRest = api.state.version;

      // Wait for the flush (1 s debounce → the HTTP CAS fails → rebase →
      // the retry persist at the fresh version).
      const persisted = await client.nextFrame("doc.persisted", 15_000);
      expect(String(persisted.pageId)).toBe(pageId);

      // The FINAL store state (the fake API) has BOTH edits (the
      // acceptance criterion) — the rebase absorbed the REST write for
      // block "b" (the room didn't touch it) and kept the room's "a".
      const a = api.state.blocks.find((b) => b.id === "a")!;
      const b = api.state.blocks.find((b) => b.id === "b")!;
      expect(a.content).toEqual({ text: "base a+room edit" });
      expect(b.content).toEqual({ text: "base b+rest edit" });

      // The retry persist won at the external version + 1 (the rebase
      // re-based onto the REST write, then flushed over it).
      expect(Number(persisted.version)).toBe(versionAfterRest + 1);

      // The flush emitted `page.updated` with the additive blockDelta
      // (both real changes vs the base → 2 updates).
      const evts = fx.events.pageUpdated(pageId);
      expect(evts.length).toBeGreaterThanOrEqual(1);
      const delta = evts[evts.length - 1]!.payload.blockDelta;
      expect(delta).toMatchObject({ inserted: 0, updated: 2, moved: 0, deleted: 0 });

      // The room is now CLEAN + in sync — the NEXT flush is a no-op (the
      // idempotency assert): a clean persist emits a ZERO blockDelta + the
      // store content is unchanged (the HTTP CAS now matches, so it writes
      // the identical set at version + 1, no external conflict).
      const room = fx.server.manager.get(`page:${pageId}`)!;
      expect(room.isDirty).toBe(false);
      await room.flush();
      const noop = fx.events.pageUpdated(pageId)[fx.events.pageUpdated(pageId).length - 1]!;
      expect(noop.payload.blockDelta).toEqual({ inserted: 0, updated: 0, moved: 0, deleted: 0 });
      const unchanged = api.state.blocks;
      expect(unchanged.find((b) => b.id === "a")!.content).toEqual({ text: "base a+room edit" });
      expect(unchanged.find((b) => b.id === "b")!.content).toEqual({ text: "base b+rest edit" });
    } finally {
      client.close();
      doc.destroy();
    }
  });
});