import path from "node:path";
import { existsSync } from "node:fs";
import { readdir, stat } from "node:fs/promises";
import { dataFile, stateFile } from "./paths";
import { readJson } from "./state";
import { primeWords, sourceVideo } from "./clips";
import { archiveMomentUrl } from "./archive";
import type { AsrWord } from "./types";
import {
  CONTEXT_WORDS,
  inChunkOverlap,
  isFillerQuery,
  normTerms,
  previewWindow,
  hitId,
  type PhraseHit,
  type PhraseQuery,
  type PhraseResult,
} from "./phrase-types";

// The shapes and constants live in lib/phrase-types.ts, which has no server
// imports, so PhraseConsole.tsx can take them as VALUES without dragging
// node:fs into the browser bundle. Re-exported so the server has one import.
export * from "./phrase-types";

// ---------------------------------------------------------------------------
// THE INDEX: a space-delimited blob per video, not a parallel string[].
//
// `" the thing is "` makes strict adjacency a native indexOf, which is the
// entire argument. Measured against this corpus (300 files, 623,079 tokens):
//
//   naive scan        527ms cold load, 50-60ms per query, 77 MB heap
//   this              ~712ms to build,  6ms for `the`,   ~105 MB heap
//
// and it costs LESS memory than an array of normalised strings would, because
// one long string has one header instead of 623,079 of them.
//
// Per video: the blob, an Int32Array of each term's char offset, and an
// Int32Array mapping term index -> word index. The second is built
// unconditionally -- it is 2.5 MB across the whole corpus, and one code path
// that always works beats a null fast-path that is right 99.997% of the time.
// Only the 18 tokens with an internal space make it differ from the identity,
// and it is consulted ONLY on a hit, so they cost nothing in the hot path.
// ---------------------------------------------------------------------------

type VideoIndex = {
  video: string;
  blob: string;
  /** Char offset of each term within `blob`. Ascending by construction. */
  off: Int32Array;
  /** Term index -> word index. Not the identity: `L.A.` is one word, two terms. */
  termWord: Int32Array;
  /**
   * The SAME array lib/clips.ts holds, not a copy -- primeWords() put it there
   * as this index was built, so a re-transcribed episode cannot leave
   * /browse/find and a clip card disagreeing about what was said.
   */
  words: AsrWord[];
  mtimeMs: number;
  size: number;
};

type Corpus = {
  videos: VideoIndex[];
  byVideo: Map<string, VideoIndex>;
  words: number;
  builtAt: number;
};

let corpus: Corpus | null = null;
let building: Promise<Corpus> | null = null;

const asrDir = () => dataFile("asr");

function buildVideoIndex(
  video: string,
  words: AsrWord[],
  mtimeMs: number,
  size: number,
): VideoIndex {
  const parts: string[] = [];
  const off: number[] = [];
  const termWord: number[] = [];
  // The blob opens with a space, so the first term is at offset 1 and a needle
  // that also opens with a space can match it.
  let pos = 1;
  for (let wi = 0; wi < words.length; wi += 1) {
    for (const t of normTerms(words[wi].w)) {
      off.push(pos);
      termWord.push(wi);
      parts.push(t);
      pos += t.length + 1;
    }
  }
  return {
    video,
    blob: ` ${parts.join(" ")} `,
    off: Int32Array.from(off),
    termWord: Int32Array.from(termWord),
    words,
    mtimeMs,
    size,
  };
}

// INVALIDATION BY (mtimeMs, size), PER FILE.
//
// A readdir plus 300 stats measures 3-4ms -- 0.06% of even a 6ms query once the
// stats are parallel -- so exact invalidation is free. A TTL would be worse in
// both directions: it would cost a 712ms rebuild every minute in a tool nobody
// is querying, and it would still serve a stale answer for up to that minute
// after a re-transcription. Only changed files are re-read.
async function build(): Promise<Corpus> {
  const dir = asrDir();
  let files: string[] = [];
  try {
    files = (await readdir(dir)).filter((f) => f.endsWith(".json"));
  } catch {
    files = [];
  }
  files.sort();

  const prev = corpus?.byVideo;
  const built = await Promise.all(
    files.map(async (f): Promise<VideoIndex | null> => {
      const video = f.slice(0, -".json".length);
      const full = path.join(dir, f);
      let st;
      try {
        st = await stat(full);
      } catch {
        return null;
      }
      const old = prev?.get(video);
      if (old && old.mtimeMs === st.mtimeMs && old.size === st.size) return old;
      const raw = await readJson<{ words: AsrWord[] }>(full, { words: [] });
      // primeWords sorts and publishes into lib/clips.ts's cache, so this parse
      // serves the console AND every clip card. One read, two callers.
      const words = primeWords(video, raw.words ?? []);
      return buildVideoIndex(video, words, st.mtimeMs, st.size);
    }),
  );

  const videos = built.filter((v): v is VideoIndex => v !== null);
  corpus = {
    videos,
    byVideo: new Map(videos.map((v) => [v.video, v])),
    words: videos.reduce((n, v) => n + v.words.length, 0),
    builtAt: Date.now(),
  };
  return corpus;
}

function ensureCorpus(): Promise<Corpus> {
  if (!building) {
    building = build().finally(() => {
      building = null;
    });
  }
  return building;
}

/** What is in memory right now. Never builds -- this is for a status line. */
export function indexState(): { videos: number; words: number; builtAt: number | null } {
  return corpus
    ? { videos: corpus.videos.length, words: corpus.words, builtAt: corpus.builtAt }
    : { videos: 0, words: 0, builtAt: null };
}

/**
 * The episode ids, from a bare readdir. NEVER builds the index.
 *
 * This is what a page with no query is allowed to do -- the same restraint
 * listSongs() shows by never probing. Building 105 MB of index to draw an empty
 * search box would be the whole cost of the feature paid for nothing.
 */
export async function episodeIds(): Promise<string[]> {
  try {
    return (await readdir(asrDir()))
      .filter((f) => f.endsWith(".json"))
      .map((f) => f.slice(0, -".json".length))
      .sort();
  } catch {
    return [];
  }
}

// ---------------------------------------------------------------------------
// Flagged sources.
//
// suspect-sources.json is keyed `<video>@<start>`, and reducing it to a set of
// VIDEOS is exactly what app/api/clip/[key]/evidence already did inline. One
// reduction, one place, so the two surfaces cannot disagree about which
// episodes carry a second speaker.
//
// The label is "flagged source", not "flagged moment". The judgement behind the
// data is per-episode in effect -- a clip somewhere in it was called "not Jer"
// -- and claiming per-moment precision for it would be worse than no flag.
// ---------------------------------------------------------------------------

export async function suspectVideos(): Promise<Set<string>> {
  const list = await readJson<unknown>(stateFile("suspect-sources.json"), []);
  const out = new Set<string>();
  for (const s of Array.isArray(list) ? list : []) {
    const k = typeof s === "string" ? s : (s as { k?: string })?.k;
    if (!k) continue;
    const at = k.lastIndexOf("@");
    out.add(at > 0 ? k.slice(0, at) : k);
  }
  return out;
}

// ---------------------------------------------------------------------------
// The scan.
//
// A RawHit is deliberately tiny: `the` is 23,081 matches, and 23,081 of these
// is a few hundred KB and about 3ms. Only the PAGE SLICE is ever expanded into
// a PhraseHit -- that is where the title join, the existsSync, the archive URL
// and the context line happen, and doing any of them 23,081 times to show 50
// rows would be the whole cost of the query.
// ---------------------------------------------------------------------------

type RawHit = {
  vi: number;
  i: number;
  n: number;
  start: number;
  cmin: number | null;
  dupes: number;
  stutter: boolean;
};

function scan(idx: VideoIndex, vi: number, needle: string, lastTerm: number, out: RawHit[]): void {
  const { blob, off, termWord, words } = idx;
  // A monotonic cursor rather than a binary search: matches come out of
  // indexOf in ascending order and `off` is ascending by construction, so the
  // whole scan walks the offsets once.
  let cur = 0;
  for (let k = blob.indexOf(needle); k >= 0; k = blob.indexOf(needle, k + 1)) {
    const at = k + 1;
    while (cur < off.length && off[cur] < at) cur += 1;
    if (cur >= off.length || off[cur] !== at) continue;
    const t1 = cur + lastTerm;
    if (t1 >= termWord.length) break;
    const w0 = termWord[cur];
    const w1 = termWord[t1];
    const first = words[w0];
    if (!first) continue;
    let cmin: number | null = null;
    for (let x = w0; x <= w1; x += 1) {
      const c = words[x]?.conf;
      if (typeof c === "number") cmin = cmin === null ? c : Math.min(cmin, c);
    }
    out.push({ vi, i: w0, n: w1 - w0 + 1, start: first.start, cmin, dupes: 0, stutter: false });
  }
}

// CHUNK TWINS, and the discriminator is exact rather than a guess.
//
// Fold a hit into its predecessor when it is the same video, the same run,
// within 0.6s, AND the later start falls in a seam band. Outside the band the
// second hit SURVIVES and is flagged `stutter`: 1,224 of the 3,472 adjacent
// same-word pairs are genuine repeats ("Very, very"), and a supercut wants
// those. `collapsed` is reported, never silently applied.
function dedupe(hits: RawHit[], fold: boolean): { kept: RawHit[]; collapsed: number } {
  const kept: RawHit[] = [];
  let collapsed = 0;
  let prev: RawHit | null = null;
  for (const h of hits) {
    const near = prev !== null && prev.vi === h.vi && h.start - prev.start < 0.6;
    if (near && inChunkOverlap(h.start)) {
      if (fold) {
        (prev as RawHit).dupes += 1;
        collapsed += 1;
        continue;
      }
    } else if (near) {
      h.stutter = true;
    }
    kept.push(h);
    prev = h;
  }
  return { kept, collapsed };
}

const hasVideoCache = new Map<string, boolean>();
function hasVideo(video: string): boolean {
  const hit = hasVideoCache.get(video);
  if (hit !== undefined) return hit;
  const ok = existsSync(sourceVideo(video));
  hasVideoCache.set(video, ok);
  return ok;
}

function expand(
  h: RawHit,
  idx: VideoIndex,
  ctx: { titles: Record<string, string>; dates: Record<string, string>; suspect: Set<string> },
): PhraseHit {
  const { words, video } = idx;
  const last = words[h.i + h.n - 1] ?? words[h.i];
  const start = words[h.i].start;
  const end = last.end;

  let sum = 0;
  let seen = 0;
  let cmin: number | null = null;
  for (let x = h.i; x < h.i + h.n; x += 1) {
    const c = words[x]?.conf;
    if (typeof c === "number") {
      sum += c;
      seen += 1;
      cmin = cmin === null ? c : Math.min(cmin, c);
    }
  }

  const text = (ws: AsrWord[]) => ws.map((w) => w.w).join(" ");
  const { from, to } = previewWindow({ start, end });

  return {
    id: hitId(video, h.i),
    video,
    // titles.json covers 298 of 300. The bare id is a worse label but a true
    // one, and an episode with no title must still be findable.
    title: ctx.titles[video] ?? video,
    date: ctx.dates[video] ?? "",
    i: h.i,
    n: h.n,
    start,
    end,
    text: text(words.slice(h.i, h.i + h.n)),
    before: text(words.slice(Math.max(0, h.i - CONTEXT_WORDS), h.i)),
    after: text(words.slice(h.i + h.n, h.i + h.n + CONTEXT_WORDS)),
    confMin: cmin === null ? null : +cmin.toFixed(4),
    confMean: seen ? +(sum / seen).toFixed(4) : null,
    from,
    to,
    suspect: ctx.suspect.has(video),
    dupes: h.dupes,
    stutter: h.stutter,
    hasVideo: hasVideo(video),
    archive: archiveMomentUrl(video, start),
  };
}

const MB = 1 << 20;

const empty = (query: PhraseQuery, notes: string[] = []): PhraseResult => ({
  query,
  hits: [],
  total: 0,
  collapsed: 0,
  suppressed: 0,
  videos: 0,
  ms: 0,
  corpus: { ...indexState(), heapMb: Math.round(process.memoryUsage().heapUsed / MB) },
  notes,
});

/** Is this a real episode? A bare readdir, so it costs nothing to ask. */
export async function isEpisode(video: string): Promise<boolean> {
  return (await episodeIds()).includes(video);
}

export async function searchPhrases(query: PhraseQuery): Promise<PhraseResult> {
  const t0 = Date.now();
  if (query.terms.length === 0) return empty(query);

  const [idx, titles, dates, suspect] = await Promise.all([
    ensureCorpus(),
    readJson<Record<string, string>>(stateFile("titles.json"), {}),
    readJson<Record<string, string>>(stateFile("dates.json"), {}),
    suspectVideos(),
  ]);

  const needle = ` ${query.terms.join(" ")} `;
  const lastTerm = query.terms.length - 1;
  const scope = query.video
    ? idx.videos.filter((v) => v.video === query.video)
    : idx.videos;

  const raw: RawHit[] = [];
  let suppressed = 0;
  for (let vi = 0; vi < scope.length; vi += 1) {
    const v = scope[vi];
    // Filters apply AS the video is scanned, so `total` is the total of what is
    // shown and `suppressed` says what went. A count that included rows the
    // filter removed would make every filter look broken.
    if (query.clean && suspect.has(v.video)) {
      const dropped: RawHit[] = [];
      scan(v, vi, needle, lastTerm, dropped);
      suppressed += dropped.length;
      continue;
    }
    const mine: RawHit[] = [];
    scan(v, vi, needle, lastTerm, mine);
    for (const h of mine) {
      // An unknown confidence cannot be asserted to clear a floor, so a floor
      // drops it. It is never read as 1.0.
      if (query.minConf !== null && (h.cmin === null || h.cmin < query.minConf)) {
        suppressed += 1;
        continue;
      }
      raw.push(h);
    }
  }

  const { kept, collapsed } = dedupe(raw, query.dupes);

  if (query.order === "conf") {
    kept.sort((a, b) => {
      const ac = a.cmin === null ? 2 : a.cmin;
      const bc = b.cmin === null ? 2 : b.cmin;
      return ac - bc || a.vi - b.vi || a.start - b.start;
    });
  }

  const from = (query.page - 1) * query.per;
  const slice = kept.slice(from, from + query.per);
  const ctx = { titles, dates, suspect };
  const hits = slice.map((h) => expand(h, scope[h.vi], ctx));

  const notes: string[] = [];
  if (isFillerQuery(query.terms)) {
    notes.push(
      "The ASR barely transcribes fillers: `um` appears 2 times in 623,079 tokens, " +
        "`uh` 5, `huh` 9. That is exactly why the um palette was mined ACOUSTICALLY " +
        "into cand2/ rather than read out of the transcript — so a near-empty result " +
        "here is the transcript being honest, not the search being broken.",
    );
  }
  if (collapsed) {
    notes.push(
      `${collapsed} chunk ${collapsed === 1 ? "twin" : "twins"} folded — repeats inside a ` +
        "240s/3s transcription seam. Genuine stutters are kept and marked.",
    );
  }
  if (suppressed) notes.push(`${suppressed} hidden by the filters.`);

  return {
    query,
    hits,
    total: kept.length,
    collapsed,
    suppressed,
    videos: new Set(kept.map((h) => h.vi)).size,
    ms: Date.now() - t0,
    corpus: {
      videos: idx.videos.length,
      words: idx.words,
      builtAt: idx.builtAt,
      heapMb: Math.round(process.memoryUsage().heapUsed / MB),
    },
    notes,
  };
}