import { test } from "node:test";
import assert from "node:assert/strict";
import {
  applySearchIn,
  newGroup,
  newLeaf,
  type GroupNode,
  type SearchIn,
} from "../searchQuery";
import {
  runQueryTree,
  type LayerCache,
  type TreeProgress,
} from "../searchEval";
import type { TranscriptDetail } from "../transcripts";
import type { SubsDetail } from "../subs";
import type { Post } from "../posts";
import {
  runLeafPipeline,
  type LayerHit,
  type LeafController,
  type LeafFetchers,
  type LeafProgress,
  type LeafRunner,
} from "./leafPipeline";
import { foldSearchIn } from "./searchIn";

// The "Search in" fold, and the rewrite run end to end through the real tree
// evaluator with a fake leaf runner: a transcripts leaf reads the kinds ticked,
// and what the visitor sees is filed under the leaf they built.

// A tiny corpus. v2 is the only video with live chat; p1 is a post.
const TEXT: Record<string, Record<string, string>> = {
  transcripts: { v1: "an alpha line", v2: "a beta line", v3: "a gamma line" },
  chat: { v2: "@u: alpha in chat" },
  posts: { p1: "a post about alpha" },
};

const noCache: LayerCache = {
  key: (a, b) => `${a}__${b}`,
  getSync: () => null,
  get: async () => null,
  put: () => {},
};

// Matches `leaf.query` against the text for the leaf's scope, one hit per
// matching slug, and reports it the way runLeafPipeline does (a chat hit
// carries its track).
const fakeRunLeaf: LeafRunner = (opts) => {
  const { leaf, scopeSlugs, emit } = opts;
  const slugs = new Set<string>();
  const hits = new Map<string, LayerHit[]>();
  for (const slug of scopeSlugs) {
    const text = TEXT[leaf.scope]?.[slug];
    if (!text || !text.includes(leaf.query)) continue;
    slugs.add(slug);
    if (leaf.contributeHits) {
      hits.set(slug, [
        {
          leafId: leaf.id,
          scope: leaf.scope,
          ...(leaf.scope === "chat" ? { track: "live_chat" } : {}),
          start: 1,
          text,
        },
      ]);
    }
  }
  let resolve!: (r: { slugs: Set<string>; hits: Map<string, LayerHit[]> }) => void;
  const done = new Promise<{ slugs: Set<string>; hits: Map<string, LayerHit[]> }>(
    (r) => (resolve = r),
  );
  queueMicrotask(() => {
    emit({
      slugs,
      hits,
      totalHits: [...hits.values()].reduce((n, l) => n + l.length, 0),
      processed: scopeSlugs.length,
      totalToProcess: scopeSlugs.length,
      capped: false,
      done: true,
    });
    resolve({ slugs, hits });
  });
  const ctrl: LeafController = { cancel() {}, setHitLimit() {}, done };
  return ctrl;
};

function run(root: GroupNode, s: SearchIn): Promise<TreeProgress> {
  const rewritten = applySearchIn(root, s);
  return new Promise((resolve) => {
    runQueryTree({
      root: rewritten.root,
      runtime: { runLeaf: fakeRunLeaf, cache: noCache },
      // The session's global scope: the videos, and the posts when Posts is on.
      globalScope: ["v1", "v2", "v3", ...(s.posts ? ["p1"] : [])],
      summaries: [],
      chatScopeSlugs: new Set(["v2"]),
      postScopeSlugs: new Set(["p1"]),
      initialHitLimit: 500,
      concurrency: 2,
      flushIntervalMs: 10,
      emit: (p) => {
        if (p.done) resolve(foldSearchIn(p, rewritten));
      },
    });
  });
}

const ALPHA = () => newGroup({ id: "root", children: [newLeaf({ id: "a", query: "alpha" })] });

test("Transcripts on: the transcript cues only (today's plain query)", async () => {
  const p = await run(ALPHA(), { transcripts: true, posts: false, chat: false });
  assert.deepEqual([...p.slugs].sort(), ["v1"]);
});

test("Transcripts + Posts (the default): videos and the post, in one result set", async () => {
  const p = await run(ALPHA(), { transcripts: true, posts: true, chat: false });
  assert.deepEqual([...p.slugs].sort(), ["p1", "v1"]);
  assert.equal(p.hits.get("p1")?.[0].scope, "posts");
  assert.equal(p.hits.get("p1")?.[0].leafId, "a", "filed under the visitor's leaf");
});

test("Live chat on: a chat hit lands in the same leaf, wearing its track", async () => {
  const p = await run(ALPHA(), { transcripts: true, posts: false, chat: true });
  assert.deepEqual([...p.slugs].sort(), ["v1", "v2"]);
  const chat = p.hits.get("v2")?.[0];
  assert.equal(chat?.leafId, "a");
  assert.equal(chat?.scope, "chat");
  assert.equal(chat?.track, "live_chat");
  // The copies' states fold into the leaf's: no copy id is left for the
  // builder to miss, and the count is the union.
  assert.deepEqual([...p.leafStates.keys()], ["a"]);
  assert.equal(p.leafStates.get("a")?.slugCount, 2);
  assert.equal(p.leafStates.get("a")?.totalHits, 2);
  assert.equal(p.leafStates.get("a")?.active, false);
});

test("Transcripts off + Live chat on: a word only in the cues returns nothing", async () => {
  const off: SearchIn = { transcripts: false, posts: false, chat: true };
  const onlyInCues = newGroup({ children: [newLeaf({ id: "b", query: "beta" })] });
  assert.equal((await run(onlyInCues, off)).slugs.size, 0);
  const p = await run(ALPHA(), off);
  assert.deepEqual([...p.slugs], ["v2"]);
});

test("a leaf asked for by name ignores the row", async () => {
  const chatByName = newGroup({
    children: [newLeaf({ id: "c", query: "alpha", scope: "chat" })],
  });
  const p = await run(chatByName, { transcripts: true, posts: true, chat: false });
  assert.deepEqual([...p.slugs], ["v2"]);
});

test("NOT reads NOT of the union, and its count is still what it matched", async () => {
  const root = newGroup({
    children: [newLeaf({ id: "a", query: "alpha", negate: true })],
  });
  const p = await run(root, { transcripts: true, posts: false, chat: true });
  // alpha is in v1's cues and v2's chat, so only v3 is left.
  assert.deepEqual([...p.slugs], ["v3"]);
  assert.equal(p.leafStates.get("a")?.slugCount, 2);
});

test("the fold before every copy has started: the largest copy, still active", () => {
  const { origin, unionOf } = applySearchIn(ALPHA(), {
    transcripts: true,
    posts: true,
    chat: true,
  });
  const state = (slugCount: number) => ({
    slugCount,
    totalHits: slugCount,
    processed: 1,
    totalToProcess: 4,
    capped: false,
    cached: false,
    active: false,
  });
  const p: TreeProgress = {
    slugs: new Set(),
    hits: new Map(),
    leafStates: new Map([
      ["a~transcripts", state(3)],
      ["a~posts", state(1)],
    ]),
    // What the evaluator reports while a copy has no result: the whole scope.
    groupStates: new Map([["a~in", { slugCount: 30_000 }]]),
    done: false,
    capped: false,
  };
  const folded = foldSearchIn(p, { origin, unionOf });
  const a = folded.leafStates.get("a");
  assert.equal(a?.slugCount, 3);
  assert.equal(a?.active, true);
  assert.equal(a?.totalHits, 4);
  assert.equal(a?.processed, 2);
  assert.equal(a?.totalToProcess, 8);
});

test("nothing rewritten: the fold hands back the same progress", () => {
  const p: TreeProgress = {
    slugs: new Set(["v1"]),
    hits: new Map(),
    leafStates: new Map(),
    groupStates: new Map(),
    done: true,
    capped: false,
  };
  assert.equal(foldSearchIn(p, { origin: new Map(), unionOf: new Map() }), p);
});

// ─── An empty scope settles (review M1) ───
// The rewrite makes copies whose scope can be empty from a plain query: a posts
// copy under a tag filter or on a channel selection with no posts, a chat copy
// where no video in scope has chat, a transcripts copy when the Type row keeps
// no video. The drivers used to start no worker for zero slugs and so never
// finalize — the query read "searching" for ever. These run the REAL drivers.

const realFetchers: LeafFetchers = {
  transcript: async (slug) =>
    ({
      slug,
      cues: [{ start: 1, end: 2, text: TEXT.transcripts[slug] ?? "" }],
    }) as unknown as TranscriptDetail,
  subs: async (slug) =>
    ({
      slug,
      tracks: { live_chat: [{ start: 1, end: 2, text: TEXT.chat[slug] ?? "" }] },
    }) as unknown as SubsDetail,
  post: async (slug) => ({ slug, text: TEXT.posts[slug] ?? "" }) as unknown as Post,
};
const realRunLeaf: LeafRunner = (opts) =>
  runLeafPipeline({ ...opts, fetchers: realFetchers });

function within<T>(p: Promise<T>, ms = 2_000): Promise<T> {
  return Promise.race([
    p,
    new Promise<T>((_, reject) =>
      setTimeout(() => reject(new Error(`not settled within ${ms} ms`)), ms),
    ),
  ]);
}

for (const scope of ["transcripts", "posts", "chat", "description"] as const) {
  test(`the real ${scope} driver settles an empty scope, and again after a raised cap`, async () => {
    const seen: LeafProgress[] = [];
    const ctrl = runLeafPipeline({
      leaf: newLeaf({ id: "x", query: "alpha", scope }),
      scopeSlugs: [],
      initialHitLimit: 10,
      concurrency: 2,
      flushIntervalMs: 5,
      emit: (p) => seen.push(p),
      fetchers: realFetchers,
    });
    const result = await within(ctrl.done);
    assert.equal(result.slugs.size, 0);
    assert.equal(seen.at(-1)?.done, true);
    // "Show more" on it resumes nothing, and must settle again.
    ctrl.setHitLimit(1_000);
    assert.equal(seen.at(-1)?.done, true);
  });
}

function runReal(
  root: GroupNode,
  s: SearchIn,
  scopes: { chat: ReadonlySet<string> | null; posts: ReadonlySet<string> | null },
  // Every folded progress, in order — what the session's readout sees.
  seen: TreeProgress[] = [],
): Promise<TreeProgress> {
  const rewritten = applySearchIn(root, s);
  return within(
    new Promise((resolve) => {
      runQueryTree({
        root: rewritten.root,
        runtime: { runLeaf: realRunLeaf, cache: noCache },
        globalScope: ["v1", "v2", "v3"],
        summaries: [],
        chatScopeSlugs: scopes.chat,
        postScopeSlugs: scopes.posts,
        initialHitLimit: 500,
        concurrency: 2,
        flushIntervalMs: 5,
        emit: (p) => {
          const folded = foldSearchIn(p, rewritten);
          seen.push(folded);
          if (p.done) resolve(folded);
        },
      });
    }),
  );
}

test("a plain query whose posts copy has nothing to read finishes (posts scope null)", async () => {
  // What the session passes under a tag filter: no posts in scope at all.
  const seen: TreeProgress[] = [];
  const p = await runReal(
    ALPHA(),
    { transcripts: true, posts: true, chat: false },
    { chat: null, posts: null },
    seen,
  );
  assert.deepEqual([...p.slugs], ["v1"]);
  assert.equal(p.leafStates.get("a")?.active, false);
  // The folded leaf's "searched N/M" only climbs (re-review R-L1): the empty
  // copy reports 0 of 0, not its parent scope as done, so the readout does
  // not start full and fall back as the transcripts copy streams.
  let lastProcessed = 0;
  let lastFraction = 0;
  for (const q of seen) {
    const a = q.leafStates.get("a");
    if (!a || a.totalToProcess === 0) continue;
    assert.ok(a.processed >= lastProcessed, `processed fell: ${lastProcessed} → ${a.processed}`);
    const f = a.processed / a.totalToProcess;
    assert.ok(f >= lastFraction, `searched fell: ${lastFraction} → ${f}`);
    lastProcessed = a.processed;
    lastFraction = f;
  }
  assert.equal(p.leafStates.get("a")?.processed, 3);
  assert.equal(p.leafStates.get("a")?.totalToProcess, 3, "the empty copy adds nothing to the total");
});

test("… and with an empty posts set (a channel selection with no posts channel)", async () => {
  const p = await runReal(ALPHA(), { transcripts: true, posts: true, chat: false }, {
    chat: null,
    posts: new Set(),
  });
  assert.deepEqual([...p.slugs], ["v1"]);
  assert.equal(p.leafStates.get("a")?.active, false);
});

test("… and with Live chat ticked where no video in scope has chat", async () => {
  const p = await runReal(ALPHA(), { transcripts: true, posts: false, chat: true }, {
    chat: new Set(),
    posts: null,
  });
  assert.deepEqual([...p.slugs], ["v1"]);
  // Posts only, with no posts: nothing, and done.
  const none = await runReal(ALPHA(), { transcripts: false, posts: true, chat: false }, {
    chat: null,
    posts: null,
  });
  assert.equal(none.slugs.size, 0);
  assert.equal(none.leafStates.get("a")?.active, false);
});