import { test } from "node:test";
import assert from "node:assert/strict";
import { mkdtemp, mkdir, writeFile, readdir, rm } from "node:fs/promises";
import os from "node:os";
import http from "node:http";
import type { AddressInfo } from "node:net";
import path from "node:path";
import { fileURLToPath } from "node:url";
import { Client } from "@modelcontextprotocol/client";
import {
  StdioClientTransport,
  getDefaultEnvironment,
} from "@modelcontextprotocol/client/stdio";
import { pageFileName } from "yt-dlp-transcript-common/lib/manifest";

// ─── Real-process protocol coverage ───
//
// Every other server test links a client and server over InMemoryTransport,
// which only ever exercises the 2025 ("legacy") era — the era decision lives in
// the serving entry (serveStdio), which an in-memory pair never runs. So a
// 2026-only regression would be invisible to the ~100 tests next door.
//
// This file is the guard: it spawns the real `src/index.ts` over stdio and
// drives it twice — once negotiating the modern (2026-07-28) era, once on the
// plain legacy handshake — asserting both still serve the same tools. Keep it
// unconditional and in the default `test` script.

const HERE = path.dirname(fileURLToPath(import.meta.url));
const ENTRY = path.join(HERE, "index.ts");
const TSX = path.join(HERE, "..", "node_modules", ".bin", "tsx");

// The exact advertised tool list, in order. `tools/list` is cached (see
// CACHE_HINTS) and its order is part of what clients see, so pin it: a rename
// or an accidental reshuffle has to be a deliberate edit here.
const EXPECTED_TOOLS = [
  "list_channels",
  "list_tags",
  "list_reports",
  "get_report",
  "search_transcripts",
  "enumerate_matches",
  "get_transcript",
  "get_post",
  "get_thread",
  "get_transcripts",
  "get_video_metadata",
  "fetch_clip",
  "get_job",
  "enqueue",
  "channel_coverage",
  "notes",
  "list_sources",
  "resolve_source",
  "open_link",
  "sweep_plan",
  "ask_plan",
];

// A minimal composed public dir: one channel, two videos, real shard layout.
async function writeFixture(opts: { tags?: boolean } = {}): Promise<string> {
  const dir = await mkdtemp(path.join(os.tmpdir(), "mcp-protocol-"));
  await writeFile(
    path.join(dir, "corpus.json"),
    JSON.stringify({
      channels: [{ slug: "chan-a", name: "Channel A", videoCount: 2 }],
    }),
  );
  const chDir = path.join(dir, "transcripts", "chan-a");
  await mkdir(chDir, { recursive: true });
  await writeFile(
    path.join(chDir, "manifest.json"),
    JSON.stringify({
      version: 1,
      channelSlug: "chan-a",
      pageCount: 1,
      maxPageBytes: 1000,
      generatedAt: "2026-01-01",
      slugToPage: { a1: 0, a2: 0 },
    }),
  );
  // Curated tags, opt-in per fixture. The DEFAULT is a site with no
  // /tags.json, because that is what every site built before corpus spec 4
  // looks like and it is the path most likely to be got wrong.
  if (opts.tags) {
    await writeFile(
      path.join(dir, "tags.json"),
      JSON.stringify({
        version: 1,
        tags: [
          {
            id: "eva-collab",
            label: "Collab",
            group: "eva",
            groupLabel: "Eva",
            order: 1,
            count: 1,
            channels: { "chan-a": 1 },
          },
          {
            id: "loose-tag",
            label: "Loose",
            order: 2,
            count: 1,
            channels: { "chan-a": 1 },
          },
        ],
      }),
    );
  }
  const rec = (
    id: string,
    title: string,
    text: string,
    curatedTags?: string[],
  ) => ({
    slug: `chan-a/${id}`,
    id,
    channelSlug: "chan-a",
    title,
    uploadDate: "20240101",
    duration: 300,
    channel: "Channel A",
    description: "",
    tags: [],
    isLivestream: false,
    ageRestricted: false,
    platform: "youtube",
    webpageUrl: `https://example.test/${id}`,
    // Omitted when empty, exactly as the index build writes it — so the
    // no-tags fixture's records are byte-identical to a pre-spec-4 site's.
    ...(curatedTags && curatedTags.length > 0 ? { curatedTags } : {}),
    cues: [{ start: 12, end: 15, text }],
  });
  // The records agree with the /tags.json above: one video per tag, which is
  // what its counts claim.
  await writeFile(
    path.join(chDir, pageFileName(0)),
    JSON.stringify([
      rec("a1", "Coffee one", "i love coffee", opts.tags ? ["eva-collab"] : undefined),
      rec("a2", "Tea two", "i love tea", opts.tags ? ["loose-tag"] : undefined),
    ]),
  );
  return dir;
}

type Session = {
  client: Client;
  stateDir: string;
  close: () => Promise<void>;
};

// Spawn the real server over stdio against `corpusDir`, connecting with the
// given negotiation mode. Each session gets its own state dir so the assertion
// that nothing is persisted is meaningful (and so a stale state file on the
// developer's machine can never leak into the test).
async function connect(
  corpusDir: string,
  versionNegotiation?: { mode: "auto" | "legacy" },
  extraEnv: Record<string, string> = {},
): Promise<Session> {
  const stateDir = await mkdtemp(path.join(os.tmpdir(), "mcp-state-"));
  const transport = new StdioClientTransport({
    command: TSX,
    args: [ENTRY, "--local", corpusDir],
    // The old controller keyed a state file off this; nothing reads it now,
    // and the assertion below is that the dir stays empty regardless.
    env: { ...getDefaultEnvironment(), TRANSCRIPT_MCP_STATE_DIR: stateDir, ...extraEnv },
    stderr: "pipe",
  });
  const client = new Client(
    { name: "protocol-test", version: "0" },
    { capabilities: {}, ...(versionNegotiation ? { versionNegotiation } : {}) },
  );
  await client.connect(transport);
  return {
    client,
    stateDir,
    close: async () => {
      await client.close();
      await rm(stateDir, { recursive: true, force: true });
    },
  };
}

function firstText(res: unknown): string {
  const content = (res as { content: { type: string; text: string }[] }).content;
  return content.map((c) => c.text).join("\n");
}

test("stdio: the modern (2026-07-28) era negotiates and serves the tools", async (t) => {
  const dir = await writeFixture();
  t.after(() => rm(dir, { recursive: true, force: true }));
  const s = await connect(dir, { mode: "auto" });
  t.after(() => s.close());

  assert.equal(
    s.client.getProtocolEra(),
    "modern",
    "server/discover should select the modern era over stdio",
  );

  const tools = await s.client.listTools();
  assert.deepEqual(tools.tools.map((x) => x.name), EXPECTED_TOOLS);

  const res = await s.client.callTool({
    name: "search_transcripts",
    arguments: { query: "coffee" },
  });
  const out = firstText(res);
  assert.match(out, /Coffee one/);
  assert.match(out, /total 1 match/);
});

test("stdio: the legacy (2025) handshake still serves the same tools", async (t) => {
  const dir = await writeFixture();
  t.after(() => rm(dir, { recursive: true, force: true }));
  // No versionNegotiation at all — the SDK's default posture, and what a
  // client that has never heard of 2026-07-28 sends.
  const s = await connect(dir);
  t.after(() => s.close());

  assert.equal(s.client.getProtocolEra(), "legacy");

  const tools = await s.client.listTools();
  assert.deepEqual(tools.tools.map((x) => x.name), EXPECTED_TOOLS);

  const res = await s.client.callTool({
    name: "search_transcripts",
    arguments: { query: "tea" },
  });
  assert.match(firstText(res), /Tea two/);
});

test("stdio: list_tags reports the vocabulary a site publishes", async (t) => {
  const dir = await writeFixture({ tags: true });
  t.after(() => rm(dir, { recursive: true, force: true }));
  const s = await connect(dir, { mode: "auto" });
  t.after(() => s.close());

  const out = firstText(await s.client.callTool({ name: "list_tags", arguments: {} }));
  assert.match(out, /2 curated tag\(s\)/);
  // Grouped, with the label, this site's count and the per-channel breakdown.
  assert.match(out, /### Eva/);
  assert.match(out, /- eva-collab · Collab · 1 video\(s\) — chan-a 1/);
  // An ungrouped tag still appears, under a bucket that says so.
  assert.match(out, /### \(ungrouped\)/);
  assert.match(out, /- loose-tag · Loose · 1 video\(s\) — chan-a 1/);
  // And it names the filter it feeds, so the id never has to be guessed.
  assert.match(out, /tags:\["eva-collab"\]/);
});

test("stdio: a site with no /tags.json says so instead of returning nothing", async (t) => {
  const dir = await writeFixture();
  t.after(() => rm(dir, { recursive: true, force: true }));
  const s = await connect(dir, { mode: "auto" });
  t.after(() => s.close());

  const out = firstText(await s.client.callTool({ name: "list_tags", arguments: {} }));
  assert.match(out, /publishes no curated tags/);
  assert.match(out, /corpus spec 4/);

  // …and a search that filters by a tag anyway is warned, not quietly empty.
  // This is the pre-spec-4 path: correct (no record carries a tag) but
  // indistinguishable from "searched and found nothing" without the warning.
  const search = firstText(
    await s.client.callTool({
      name: "search_transcripts",
      arguments: { query: "coffee", tags: ["eva-collab"] },
    }),
  );
  assert.match(search, /No matches for "coffee"/);
  assert.match(search, /posts: skipped — a tag filter was given/);
  assert.match(search, /publishes no \/tags\.json/);
  assert.match(search, /not evidence of absence/);
});

test("stdio: a tag the site does not publish is named, not silently empty", async (t) => {
  const dir = await writeFixture({ tags: true });
  t.after(() => rm(dir, { recursive: true, force: true }));
  const s = await connect(dir, { mode: "auto" });
  t.after(() => s.close());

  const out = firstText(
    await s.client.callTool({
      name: "search_transcripts",
      arguments: { query: "coffee", tags: ["eva-collab", "not-a-real-tag"] },
    }),
  );
  assert.match(out, /tag\(s\) this source does not publish: not-a-real-tag/);
  // The filter it DID understand is echoed too.
  assert.match(out, /tags: eva-collab OR not-a-real-tag/);
});

test("stdio: enumerate_matches filters by tag and names the filter", async (t) => {
  const dir = await writeFixture({ tags: true });
  t.after(() => rm(dir, { recursive: true, force: true }));
  const s = await connect(dir, { mode: "auto" });
  t.after(() => s.close());

  // Both records match "i love"; only a1 carries eva-collab.
  const all = firstText(
    await s.client.callTool({
      name: "enumerate_matches",
      arguments: { query: "i love" },
    }),
  );
  assert.match(all, /2 match\(es\)/);

  const tagged = firstText(
    await s.client.callTool({
      name: "enumerate_matches",
      arguments: { query: "i love", tags: ["eva-collab"] },
    }),
  );
  assert.match(tagged, /1 match\(es\)/);
  assert.match(tagged, /- a1 \| video/);
  assert.doesNotMatch(tagged, /- a2 \| video/);
  // The worklist's footer names the filter that produced it, so a count can
  // never be quoted without the scope that made it.
  assert.match(tagged, /filters — tags: eva-collab/);
  // …and the complete-set line is still the honest one.
  assert.match(tagged, /complete set: yes/);
  // The posts corpus is not part of that count and the footer says why: a post
  // carries no curated tags, so a tag filter drops the corpus whole. Without
  // the sentence "no post matches" and "posts were never searched" read the
  // same on the wire.
  assert.match(
    tagged,
    /posts: skipped — a tag filter was given and posts carry no curated tags \(the export UI does the same\)/,
  );
});

test("stdio: prompts are served on both eras", async (t) => {
  const dir = await writeFixture();
  t.after(() => rm(dir, { recursive: true, force: true }));
  const modern = await connect(dir, { mode: "auto" });
  t.after(() => modern.close());
  const legacy = await connect(dir, { mode: "legacy" });
  t.after(() => legacy.close());

  for (const s of [modern, legacy]) {
    const prompts = await s.client.listPrompts();
    assert.deepEqual(prompts.prompts.map((p) => p.name), ["sweep"]);
  }
});

test("stdio: a read-only session writes no state file", async (t) => {
  const dir = await writeFixture();
  t.after(() => rm(dir, { recursive: true, force: true }));
  const s = await connect(dir, { mode: "auto" });
  t.after(() => s.close());

  await s.client.callTool({ name: "list_channels", arguments: {} });
  await s.client.callTool({
    name: "search_transcripts",
    arguments: { query: "coffee" },
  });

  const left = await readdir(s.stateDir).catch(() => [] as string[]);
  assert.deepEqual(left, [], "the server must not persist anything");
});

// fetch_clip over the real process, on the modern era: the env it was
// registered with reaches the editor as the bearer token, and a client that
// asks for progress gets one notification per poll — the keep-alive for a
// client whose request timeout resets on progress. The editor is a stub on an
// ephemeral port; a resume by `job` needs no corpus read.
test("stdio (modern): fetch_clip asks the editor with its registered token and reports progress", async (t) => {
  const seen: string[] = [];
  let polls = 0;
  const editor = http.createServer((req, res) => {
    seen.push(`${req.method} ${req.url} ${req.headers.authorization}`);
    polls++;
    const body =
      polls < 3
        ? { status: "running", jobId: "j1" }
        : { status: "done", jobId: "j1", file: "/corpus/clips/7.00-23.00.mp4", from: 7, to: 23, bytes: 5 };
    res.writeHead(200, { "content-type": "application/json" });
    res.end(JSON.stringify(body));
  });
  await new Promise<void>((r) => editor.listen(0, "127.0.0.1", r));
  t.after(() => new Promise<void>((r) => editor.close(() => r())));
  const { port } = editor.address() as AddressInfo;

  const dir = await writeFixture();
  t.after(() => rm(dir, { recursive: true, force: true }));
  const s = await connect(dir, { mode: "auto" }, {
    ARCHILYZER_EDITOR_URL: `http://127.0.0.1:${port}`,
    WORKER_TOKEN: "tok-proto",
  });
  t.after(() => s.close());
  assert.equal(s.client.getProtocolEra(), "modern");

  const progress: number[] = [];
  const res = await s.client.callTool(
    { name: "fetch_clip", arguments: { job: "j1" } },
    { onprogress: (p) => progress.push(p.progress), resetTimeoutOnProgress: true },
  );
  assert.match(firstText(res), /^Fetched 7\.00–23\.00 \(job j1, \d+s waited\)\./);
  assert.deepEqual(seen, [
    "GET /api/media/fetch-window/j1 Bearer tok-proto",
    "GET /api/media/fetch-window/j1 Bearer tok-proto",
    "GET /api/media/fetch-window/j1 Bearer tok-proto",
  ]);
  assert.deepEqual(progress, [1, 2]);
});