import { type DuckDBConnection, DuckDBInstance } from "@duckdb/node-api";
import * as wasm from "import-core";
import type { KeywordSource, SearchIntent } from "@/lib/keywords/types";
import { resolveReadable } from "./storage";
import type { CanonicalKeyword, CuratedResult } from "./types";

/** Canonical slug for a keyword term — delegates to the Rust/WASM `slugify`
 *  (crates/import-core/src/normalize.rs) also used by Server Actions, GSC and
 *  briefs-par-terme. `keywords.id = ${siteId}-${slug}` is the merge key of
 *  every import; a divergent implementation here would silently duplicate
 *  rows. Parity with the old TS regex implementation is locked by
 *  test/slugify.test.ts — see that file for the corpus. */
export function slugifyTerm(s: string): string {
  return wasm.slugify(s);
}

const INTENTS: Record<string, SearchIntent> = {
  informational: "informational",
  commercial: "commercial",
  transactional: "transactional",
  navigational: "navigational",
};

const splitIntents = (s: string): SearchIntent[] =>
  s
    .split(/[,;|]/)
    .map((x) => INTENTS[x.trim().toLowerCase()])
    .filter(Boolean) as SearchIntent[];

/** SE-Ranking encodes intent as single letters (e.g. "L, C"); map to canonical
 *  names, dropping unknowns. Mirrors the WASM `map_intent_code`. */
const INTENT_CODES: Record<string, SearchIntent> = {
  i: "informational",
  c: "commercial",
  t: "transactional",
  n: "navigational",
  l: "local",
};

const splitIntentCodes = (s: string): SearchIntent[] =>
  s
    .split(/[,;|]/)
    .map((x) => INTENT_CODES[x.trim().toLowerCase()])
    .filter(Boolean) as SearchIntent[];

const splitSerp = (s: string): string[] =>
  s
    .split(/[,;|]/)
    .map((x) => x.trim())
    .filter(Boolean);

const num = (v: unknown): number => {
  const n = Number(String(v ?? "").replace(/[^0-9.-]/g, ""));
  return Number.isFinite(n) ? n : 0;
};

const numOrNull = (v: unknown): number | null => {
  const s = String(v ?? "").trim();
  if (!s) return null;
  const n = Number(s.replace(/[^0-9.-]/g, ""));
  return Number.isFinite(n) ? n : null;
};

const COUNTRY: Record<string, string> = {
  france: "FR",
  "united states": "US",
  "united kingdom": "GB",
  canada: "CA",
  australia: "AU",
};

const countryFromLocation = (loc: string): string =>
  COUNTRY[loc.trim().toLowerCase()] ?? "US";

/** KWFinder/Mangools renamed its headline volume column across export versions:
 *  newer exports use "Last 12 months", older ones "Last Known Values". Accept
 *  both (detection + projection). Order = preference if a file ever has both. */
const MANGOOLS_VOLUME_COLS = [
  "Avg. Search Volume (Last Known Values)",
  "Avg. Search Volume (Last 12 months)",
] as const;
/** Stable name the projected volume column is aliased to, so the row mapping
 *  reads one key regardless of which export variant the file uses. */
const MANGOOLS_VOLUME_ALIAS = MANGOOLS_VOLUME_COLS[0];

/** Run `fn` with a fresh DuckDB instance + connection, then close BOTH in a
 *  `finally` so native resources don't leak on the long-running worker (a
 *  per-call instance/connection was never released before). For non-`local`
 *  storage, install httpfs + an R2 secret so `read_csv('r2://…')` can
 *  authenticate. Creds come from env (trusted, not user input). */
async function withConnection<T>(
  fn: (c: DuckDBConnection) => Promise<T>,
): Promise<T> {
  const inst = await DuckDBInstance.create(":memory:");
  let c: DuckDBConnection | undefined;
  try {
    c = await inst.connect();
    // VPS hygiene (ELEVA-78): DuckDB defaults (memory_limit = 80% RAM, threads =
    // all cores) assume a dedicated box and starve the co-located Postgres on the
    // shared VPS. Bound resources per connection. `preserve_insertion_order=false`
    // lets DuckDB stream/reorder freely — fine here, output is deduped by slug.
    await c.runAndReadAll("SET memory_limit = '1500MB'");
    await c.runAndReadAll("SET threads = 2");
    await c.runAndReadAll("SET preserve_insertion_order = false");
    if (process.env.STORAGE_DRIVER !== "local") {
      const q = (s: string | undefined) => `'${(s ?? "").replace(/'/g, "''")}'`;
      await c.runAndReadAll("INSTALL httpfs");
      await c.runAndReadAll("LOAD httpfs");
      await c.runAndReadAll(
        `CREATE OR REPLACE SECRET r2 (TYPE R2, KEY_ID ${q(process.env.R2_ACCESS_KEY_ID)}, SECRET ${q(process.env.R2_SECRET_ACCESS_KEY)}, ACCOUNT_ID ${q(process.env.R2_ACCOUNT_ID)})`,
      );
    }
    return await fn(c);
  } finally {
    c?.closeSync();
    inst.closeSync();
  }
}

async function rowsOf(
  c: DuckDBConnection,
  loc: string,
  cols: string,
): Promise<Record<string, string>[]> {
  const reader = await c.runAndReadAll(
    `select ${cols} from read_csv(?, header=true, all_varchar=true)`,
    [resolveReadable(loc)],
  );
  return reader
    .getRowObjects()
    .map((row) =>
      Object.fromEntries(
        Object.entries(row).map(([k, v]) => [k, String(v ?? "")]),
      ),
    ) as Record<string, string>[];
}

/** Header row only — used both to sniff the source and to pick the right
 *  variant column name for projection. */
async function columnNames(
  c: DuckDBConnection,
  loc: string,
): Promise<string[]> {
  const reader = await c.runAndReadAll(
    "select * from read_csv(?, header=true, all_varchar=true) limit 0",
    [resolveReadable(loc)],
  );
  return reader.columnNames();
}

/** Sniff the source from the CSV header. Accepts an optional connection so a
 *  caller already inside a `withConnection` (e.g. `processCsv`) reuses it
 *  instead of spinning up a second DuckDB instance; standalone callers omit it
 *  and get their own short-lived instance. */
export async function detectSource(
  loc: string,
  conn?: DuckDBConnection,
): Promise<KeywordSource | "unknown"> {
  if (!conn) return withConnection((c) => detectSource(loc, c));
  const cols = await columnNames(conn, loc);
  if (MANGOOLS_VOLUME_COLS.some((c) => cols.includes(c))) return "mangools";
  if (
    cols.includes("Competitive Density") &&
    cols.includes("Number of Results")
  )
    return "semrush";
  // Ahrefs Keywords Explorer / SERP overview export (must be CSV UTF-8).
  if (cols.includes("Difficulty") && cols.includes("Traffic potential"))
    return "ahrefs";
  // SE-Ranking "Research → bulk" export (BOM-prefixed; DuckDB strips the BOM).
  if (cols.includes("Search vol.") && cols.includes("Search intent"))
    return "seranking";
  return "unknown";
}

export async function processCsv(
  loc: string,
  source: KeywordSource,
  lang: string,
  /** Default country for sources that don't carry a per-row country (semrush).
   *  Derived from the site's target country by the worker (was hardcoded "US"). */
  country = "US",
): Promise<CuratedResult> {
  // One DuckDB instance + connection for the whole job (one-time httpfs/R2 setup
  // runs once), threaded through the header sniff and the row read. Previously
  // each helper opened its own instance → 2–3 per job.
  return withConnection(async (c) => {
    // Mangools/KWFinder: the volume column name varies by export version, so pick
    // whichever variant the file actually has and alias it to a stable name.
    const mangoolsHeader =
      source === "mangools" ? await columnNames(c, loc) : [];
    const mangoolsVolumeCol =
      MANGOOLS_VOLUME_COLS.find((col) => mangoolsHeader.includes(col)) ??
      MANGOOLS_VOLUME_ALIAS;

    const cols =
      source === "semrush"
        ? `"Keyword","Volume","Keyword Difficulty","CPC (USD)","Competitive Density","Number of Results","Intent","SERP Features"`
        : source === "ahrefs"
          ? `"Keyword","Country","Difficulty","Volume","CPC","Intents"`
          : source === "seranking"
            ? `"Keyword","Difficulty","Search vol.","Search intent","SERP features","CPC","Competition"`
            : `"Keyword","${mangoolsVolumeCol}" AS "${MANGOOLS_VOLUME_ALIAS}","Keyword Difficulty","CPC/USD","Keyword interest growth (%)","Search Intent","Serp Features","Location"`;

    const recs = await rowsOf(c, loc, cols);
    return curate(recs, source, lang, country);
  });
}

/** Pure row-mapping/dedup step, split out of `processCsv` so the DuckDB
 *  connection lifecycle (the `withConnection` body) stays a thin shell. */
function curate(
  recs: Record<string, string>[],
  source: KeywordSource,
  lang: string,
  country: string,
): CuratedResult {
  const updatedAt = new Date().toISOString();
  const bySlug = new Map<string, CanonicalKeyword>();

  for (const r of recs) {
    const term = (r.Keyword ?? "").trim();
    if (!term) continue;
    const slug = slugifyTerm(term);
    if (!slug) continue;

    const k: CanonicalKeyword =
      source === "semrush"
        ? {
            slug,
            term,
            lang,
            country,
            intent: splitIntents(r.Intent ?? ""),
            serpFeatures: splitSerp(r["SERP Features"] ?? ""),
            primarySource: "semrush",
            metrics: {
              volume: num(r.Volume),
              kd: num(r["Keyword Difficulty"]),
              cpc: num(r["CPC (USD)"]),
              growthPct: null,
              competitiveDensity: numOrNull(r["Competitive Density"]),
              numberOfResults: numOrNull(r["Number of Results"]),
              updatedAt,
            },
          }
        : source === "ahrefs"
          ? {
              slug,
              term,
              lang,
              country: (r.Country ?? "").trim().toUpperCase() || "US",
              intent: splitIntents(r.Intents ?? ""),
              serpFeatures: [],
              primarySource: "ahrefs",
              metrics: {
                volume: num(r.Volume),
                kd: num(r.Difficulty),
                cpc: num(r.CPC),
                growthPct: null,
                competitiveDensity: null,
                numberOfResults: null,
                updatedAt,
              },
            }
          : source === "seranking"
            ? {
                slug,
                term,
                lang,
                // No location column → use the site's market country (param).
                country,
                intent: splitIntentCodes(r["Search intent"] ?? ""),
                serpFeatures: splitSerp(r["SERP features"] ?? ""),
                primarySource: "seranking",
                metrics: {
                  volume: num(r["Search vol."]),
                  kd: num(r.Difficulty),
                  cpc: num(r.CPC),
                  growthPct: null,
                  // "Competition" (0..1 ad competition) → competitive-density slot.
                  competitiveDensity: numOrNull(r.Competition),
                  numberOfResults: null,
                  updatedAt,
                },
              }
            : {
                slug,
                term,
                lang,
                country: countryFromLocation(r.Location ?? ""),
                intent: splitIntents(r["Search Intent"] ?? ""),
                serpFeatures: splitSerp(r["Serp Features"] ?? ""),
                primarySource: "mangools",
                metrics: {
                  volume: num(r["Avg. Search Volume (Last Known Values)"]),
                  kd: num(r["Keyword Difficulty"]),
                  cpc: num(r["CPC/USD"]),
                  growthPct: numOrNull(r["Keyword interest growth (%)"]),
                  competitiveDensity: null,
                  numberOfResults: null,
                  updatedAt,
                },
              };

    const prev = bySlug.get(slug);
    if (!prev || k.metrics.volume > prev.metrics.volume) {
      bySlug.set(slug, k);
    }
  }

  return { rows: [...bySlug.values()], totals: { read: recs.length } };
}