import { promises as fs } from "node:fs";
import { join, basename } from "node:path";
import { sql } from "@/db";
import { STORAGE_DIR } from "@/lib/env";
import { pdfInfo, pdfIntegrity, pdfText, pdfTextPages, pdfThumbnail, sha256File, storeFile, storagePath, ensureStorage } from "@/lib/pdf/process";
import { proposeMetadata, titleSimilarity } from "@/lib/pdf/extract";
import { chunkPages } from "@/lib/pdf/chunk";
import { lexvec } from "@/lib/ai/lexvec";
import { embed, embeddingsEnabled, EMBEDDING_MODEL } from "@/lib/ai/embeddings";
import { enqueue } from "./queue";
import { runBackup } from "./backup";
import { rebuildRelations } from "./relations";

const vec = (v: number[]) => `[${v.map((x) => x.toFixed(6)).join(",")}]`;

export async function processPdf(p: { fileId: string; paperId?: string; versionId?: string }) {
  const [f] = await sql`SELECT * FROM files WHERE id = ${p.fileId}`; if (!f) throw new Error("file not found");
  const path = storagePath(f.storage_key);
  const integ = await pdfIntegrity(path);
  const info = integ.info ?? (integ.ok ? await pdfInfo(path) : null);
  await sql`UPDATE files SET integrity_ok = ${integ.ok}, pages = ${info?.pages ?? null}, pdf_meta = ${info ? sql.json(info.raw as never) : null} WHERE id = ${p.fileId}`;
  if (!integ.ok) { await sql`UPDATE files SET text_status = 'failed' WHERE id = ${p.fileId}`; throw new Error(`integrity: ${integ.reason}`); }
  // Preview (first page) — content-addressed by sha.
  const previewBase = join(/*turbopackIgnore: true*/ STORAGE_DIR, "previews", f.sha256);
  try { await fs.access(previewBase + ".png"); } catch { await pdfThumbnail(path, previewBase, 60); }
  if (p.versionId) await sql`UPDATE paper_versions SET pages = coalesce(pages, ${info?.pages ?? null}), sha256 = ${f.sha256}, bytes = ${Number(f.bytes)} WHERE id = ${p.versionId}`;
  // Full text → chunks (only for the current version of a paper).
  if (p.paperId) {
    const pages = await pdfTextPages(path);
    const chunks = chunkPages(pages);
    await sql.begin(async (tx) => {
      await tx`DELETE FROM paper_chunks WHERE paper_id = ${p.paperId!} AND (version_id = ${p.versionId ?? null} OR version_id IS NULL)`;
      for (let i = 0; i < chunks.length; i += 200) {
        const batch = chunks.slice(i, i + 200);
        await tx`INSERT INTO paper_chunks ${tx(batch.map((c) => ({ paper_id: p.paperId!, version_id: p.versionId ?? null, chunk_index: c.index, page_from: c.pageFrom, page_to: c.pageTo, text: c.text })))}`;
      }
    });
    const [paper] = await sql`SELECT title, subtitle, abstract, keywords FROM papers WHERE id = ${p.paperId}`;
    const base = `${paper.title} ${paper.subtitle ?? ""} ${(paper.keywords ?? []).join(" ")} ${(paper.abstract as string[]).join(" ")}`;
    const lv = lexvec(base + " " + chunks.map((c) => c.text).join(" ").slice(0, 60000));
    await sql`UPDATE papers SET lexvec = ${vec(lv)}::vector WHERE id = ${p.paperId}`;
    await sql`UPDATE files SET text_status = 'done' WHERE id = ${p.fileId}`;
    if (embeddingsEnabled()) await enqueue("embed_paper", { paperId: p.paperId }, { dedupe: true });
    await enqueue("relations", {}, { dedupe: true, runAfter: new Date(Date.now() + 30_000) });
  } else await sql`UPDATE files SET text_status = 'skipped' WHERE id = ${p.fileId}`;
}

/** Paper-level lexvec from metadata only (no manuscript yet). */
export async function lexvecFromMetadata(paperId: string) {
  const [paper] = await sql`SELECT title, subtitle, abstract, keywords, research_question, contributions, methodology, data_sections FROM papers WHERE id = ${paperId}`;
  if (!paper) return;
  const text = [paper.title, paper.subtitle, (paper.keywords ?? []).join(" "), ...(paper.abstract as string[]), paper.research_question, JSON.stringify(paper.contributions), JSON.stringify(paper.methodology), JSON.stringify(paper.data_sections)].filter(Boolean).join(" ");
  await sql`UPDATE papers SET lexvec = ${vec(lexvec(text))}::vector WHERE id = ${paperId}`;
}

export async function embedPaper(p: { paperId: string }) {
  const [paper] = await sql`SELECT id, title, subtitle, abstract, keywords FROM papers WHERE id = ${p.paperId}`; if (!paper) return;
  const head = `${paper.title}${paper.subtitle ? ": " + paper.subtitle : ""}\n\n${(paper.abstract as string[]).join("\n\n")}\n\nKeywords: ${(paper.keywords ?? []).join(", ")}`;
  const e = await embed([head]); if (!e) return;
  await sql`UPDATE papers SET embedding = ${vec(e[0])}::vector, embedding_model = ${EMBEDDING_MODEL} WHERE id = ${p.paperId}`;
  const chunks = await sql`SELECT id, text FROM paper_chunks WHERE paper_id = ${p.paperId} AND embedding IS NULL ORDER BY chunk_index`;
  for (let i = 0; i < chunks.length; i += 64) {
    const batch = chunks.slice(i, i + 64); const vs = await embed(batch.map((c) => c.text)); if (!vs) return;
    await sql.begin(async (tx) => { for (let k = 0; k < batch.length; k++) await tx`UPDATE paper_chunks SET embedding = ${vec(vs[k])}::vector WHERE id = ${batch[k].id}`; });
  }
  await enqueue("relations", {}, { dedupe: true, runAfter: new Date(Date.now() + 30_000) });
}

export async function dailyStats() {
  await sql`INSERT INTO paper_stats_daily (paper_id, day, views, downloads)
    SELECT paper_id, day, count(*) FILTER (WHERE kind = 'view'), count(*) FILTER (WHERE kind = 'download')
    FROM events WHERE NOT is_bot AND day >= current_date - 2 GROUP BY paper_id, day
    ON CONFLICT (paper_id, day) DO UPDATE SET views = EXCLUDED.views, downloads = EXCLUDED.downloads`;
  await sql`UPDATE papers p SET view_count = s.v, download_count = s.d FROM (SELECT paper_id, sum(views) v, sum(downloads) d FROM paper_stats_daily GROUP BY paper_id) s WHERE s.paper_id = p.id AND (p.view_count <> s.v OR p.download_count <> s.d)`;
  await sql`DELETE FROM events WHERE day < current_date - 400`;
  await sql`DELETE FROM search_log WHERE at < now() - interval '180 days'`;
  await sql`DELETE FROM jobs WHERE status IN ('done', 'failed') AND finished_at < now() - interval '30 days'`;
}

export async function ingestScan(p: { dirs?: string[] } = {}) {
  await ensureStorage();
  const [extra] = await sql`SELECT value FROM settings WHERE key = 'ingest.extra_dirs'`;
  const dirs = [join(STORAGE_DIR, "inbox"), ...(p.dirs ?? []), ...((extra?.value as string[] | undefined) ?? [])];
  let found = 0, queued = 0;
  for (const dir of dirs) {
    let entries: string[] = [];
    try { entries = (await fs.readdir(dir)).filter((n) => n.toLowerCase().endsWith(".pdf")); } catch { continue; }
    for (const name of entries) {
      found++;
      const src = join(dir, name); const sha = await sha256File(src);
      const [known] = await sql`SELECT 1 FROM ingest_candidates WHERE sha256 = ${sha} LIMIT 1`;
      if (known) continue;
      const [existing] = await sql`SELECT f.id FROM files f WHERE f.sha256 = ${sha} LIMIT 1`;
      let fileId: string;
      if (existing) fileId = existing.id;
      else {
        const st = await fs.stat(src); const key = await storeFile(src, sha);
        const [f] = await sql`INSERT INTO files (kind, storage_key, original_name, mime, bytes, sha256) VALUES ('manuscript', ${key}, ${basename(src)}, 'application/pdf', ${st.size}, ${sha}) RETURNING id`;
        fileId = f.id;
      }
      const [c] = await sql`INSERT INTO ingest_candidates (source_path, file_id, sha256, status) VALUES (${src}, ${fileId}, ${sha}, 'queued') ON CONFLICT DO NOTHING RETURNING id`;
      if (c) { queued++; await enqueue("ingest_candidate", { candidateId: c.id }); }
    }
  }
  await sql`INSERT INTO settings (key, value) VALUES ('ingest.last_scan', ${sql.json({ at: new Date().toISOString(), dirs, found, queued } as never)}) ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value, updated_at = now()`;
  return { found, queued };
}

export async function ingestCandidate(p: { candidateId: string }) {
  const [c] = await sql`SELECT c.*, f.storage_key FROM ingest_candidates c JOIN files f ON f.id = c.file_id WHERE c.id = ${p.candidateId}`; if (!c) return;
  await sql`UPDATE ingest_candidates SET status = 'processing' WHERE id = ${c.id}`;
  const path = storagePath(c.storage_key);
  const flags: { code: string; message: string }[] = [];
  const integ = await pdfIntegrity(path);
  if (!integ.ok) flags.push({ code: "integrity", message: integ.reason ?? "PDF failed integrity check" });
  const info = integ.info ?? null;
  await sql`UPDATE files SET integrity_ok = ${integ.ok}, pages = ${info?.pages ?? null}, pdf_meta = ${info ? sql.json(info.raw as never) : null} WHERE id = ${c.file_id}`;
  let proposal: Record<string, unknown> = {};
  if (integ.ok) {
    const first = await pdfText(path, { first: 1, last: 3 });
    const prop = proposeMetadata(first, info?.title, info?.author);
    proposal = { ...prop, pages: info?.pages ?? null, pdfMeta: { title: info?.title, author: info?.author, subject: info?.subject, keywords: info?.keywords, creationDate: info?.creationDate, modDate: info?.modDate, producer: info?.producer }, sourceFile: basename(c.source_path), sha256: c.sha256 };
    if (!prop.title) flags.push({ code: "title", message: "No title could be proposed" });
    if (!prop.authors.length) flags.push({ code: "authors", message: "No author line detected" });
    if (!prop.abstract) flags.push({ code: "abstract", message: "No abstract detected" });
    if (!prop.wp) flags.push({ code: "identifier", message: "No working paper number detected on the title page" });
    if (!prop.dateText) flags.push({ code: "date", message: "No release date detected; do not use the file timestamp" });
    if (first.trim().length < 200) flags.push({ code: "scanned", message: "Little or no extractable text (scanned PDF?)" });
    try { await pdfThumbnail(path, join(STORAGE_DIR, "previews", c.sha256), 60); } catch { flags.push({ code: "preview", message: "Preview generation failed" }); }
    // Duplicate detection: checksum, identifier, title similarity, author overlap.
    const dups: { paperId: string; identifier: string; title: string; reasons: string[]; score: number }[] = [];
    const papers = await sql`SELECT p.id, p.identifier, p.title, p.number, s.code, (SELECT array_agg(a.display_name) FROM paper_authors pa JOIN authors a ON a.id = pa.author_id WHERE pa.paper_id = p.id) AS authors,
      EXISTS (SELECT 1 FROM paper_versions v WHERE v.paper_id = p.id AND v.sha256 = ${c.sha256}) AS same_file FROM papers p JOIN series s ON s.id = p.series_id`;
    for (const pr of papers) {
      const reasons: string[] = []; let score = 0;
      if (pr.same_file) { reasons.push("identical checksum"); score += 1; }
      if (prop.wp && pr.code === prop.wp.seriesCode && pr.number === prop.wp.number) { reasons.push(`same identifier ${pr.identifier}`); score += 0.6; }
      const ts = prop.title ? titleSimilarity(prop.title, pr.title) : 0; if (ts >= 0.6) { reasons.push(`title similarity ${(ts * 100).toFixed(0)}%`); score += ts * 0.5; }
      const ov = prop.authors.filter((a) => (pr.authors ?? []).some((b: string) => titleSimilarity(a, b) >= 0.5)).length; if (ov && ts >= 0.3) { reasons.push(`${ov} shared author(s)`); score += 0.1 * ov; }
      if (reasons.length) dups.push({ paperId: pr.id, identifier: pr.identifier, title: pr.title, reasons, score: Math.min(1, score) });
    }
    dups.sort((a, b) => b.score - a.score);
    if (dups.length) flags.push({ code: "duplicate", message: `Possible duplicate of ${dups[0].identifier} (${dups[0].reasons.join(", ")})` });
    await sql`UPDATE ingest_candidates SET proposal = ${sql.json(proposal as never)}, flags = ${sql.json(flags as never)}, duplicates = ${sql.json(dups.slice(0, 5) as never)}, status = 'ready' WHERE id = ${c.id}`;
  } else await sql`UPDATE ingest_candidates SET flags = ${sql.json(flags as never)}, status = 'ready' WHERE id = ${c.id}`;
}

export const handlers: Record<string, (payload: Record<string, unknown>) => Promise<unknown>> = {
  process_pdf: (p) => processPdf(p as { fileId: string; paperId?: string; versionId?: string }),
  embed_paper: (p) => embedPaper(p as { paperId: string }),
  relations: () => rebuildRelations(),
  daily_stats: () => dailyStats(),
  backup: () => runBackup(),
  ingest_scan: (p) => ingestScan(p as { dirs?: string[] }),
  ingest_candidate: (p) => ingestCandidate(p as { candidateId: string }),
};