// src/lib/pages/crawl-runner.ts — cœur orchestrable du job `site-crawl`.
// Découverte sitemap → pool de fetch 6 → parse métas → upsert site_pages →
// mapping keywords↔pages (gsc > content > text). TOUT en DI (db/fetch/parse)
// pour être testé par test:all sur le vrai Postgres (pattern seranking/sync).
// Le fichier service (services/analytics/src/site-crawl.ts) n'est qu'un
// adaptateur mince qui câble les vraies dépendances.

import { createHash } from "node:crypto";
import { and, eq, inArray, sql } from "drizzle-orm";
import type { PostgresJsDatabase } from "drizzle-orm/postgres-js";
import {
  contentItems,
  contentKeywords,
  keywords,
  sitePages,
  sites,
} from "@/lib/db/domain-schema";
import type { FetchHtml } from "@/lib/keywords/onpage-fetch";
import { computeIssues } from "./issues";
import { normalizePageUrl, type PageRef, textualMatch } from "./mapping";
import { discoverSitemapUrls, type FetchText } from "./sitemap";
import type { CrawledPage, PageMeta } from "./types";

/** Concurrence du pool de fetch (spec : 6). */
const POOL = 6;
/** Lignes par lot d'upsert `site_pages`. */
const UPSERT_BATCH = 200;
/** Lignes par transaction d'UPDATE mapping. */
const WRITE_CHUNK = 500;
/** Fréquence de mise à jour de la progression (une page sur N). */
const PROGRESS_EVERY = 5;
/** Émission forcée si aucun update depuis ce délai (ms) — évite l'aveugle sur
 *  un pool lent (peu de pages mais fetch longs). Date.now() OK ici : code
 *  service, pas workflow déterministe. */
const PROGRESS_THROTTLE_MS = 2_000;

/** DB Drizzle sur le schéma domaine (exactement le `db` du service analytics,
 *  `drizzle(client, { schema: domainSchema })`). Le `db` de l'app (schéma plus
 *  large) porte un générique invariant non assignable tel quel — le test le
 *  cast. Le runner n'utilise que select/insert/update/transaction (pas `.query`). */
export type CrawlDb = PostgresJsDatabase<
  typeof import("@/lib/db/domain-schema")
>;

export type CrawlPhase = "discover" | "crawl" | "map" | "done";

/** État de progression exposé dans `analytics_jobs.totals` (jsonb réutilisé). */
export type CrawlTotals = {
  phase: CrawlPhase;
  done: number;
  total: number;
  pages: number;
  errors: number;
  mapped: number;
  unmapped: number;
  truncated?: boolean;
  message?: string;
};

export type CrawlRunnerDeps = {
  db: CrawlDb;
  /** Fetch texte pour la découverte sitemap (content-type xml/text assoupli). */
  fetchText: FetchText;
  /** Fetch HTML d'une page (null = échec → status 0). */
  fetchHtml: FetchHtml;
  /** Extraction des métas (wasm `parse_page_meta` en prod, fake en test). */
  parseMeta: (html: string) => PageMeta;
  /** Callback de progression (updateJob périodique côté service). */
  onProgress?: (totals: CrawlTotals) => Promise<void>;
};

/** id stable d'une page : `${siteId}-${sha256(urlNormalisée).slice(0,16)}`. */
function pageId(siteId: string, normalizedUrl: string): string {
  const hash = createHash("sha256")
    .update(normalizedUrl)
    .digest("hex")
    .slice(0, 16);
  return `${siteId}-${hash}`;
}

/**
 * Exécute `fn` sur `items` avec au plus `limit` en vol ; l'ordre est conservé.
 * Copie locale de `mapPool` de keywords/serp-actions (ce fichier est
 * `"use server"`, non importable ici) — même sémantique.
 */
async function mapPool<T, R>(
  items: T[],
  limit: number,
  fn: (item: T, index: number) => Promise<R>,
): Promise<R[]> {
  const out = new Array<R>(items.length);
  let next = 0;
  async function worker(): Promise<void> {
    while (next < items.length) {
      const i = next++;
      out[i] = await fn(items[i], i);
    }
  }
  await Promise.all(
    Array.from({ length: Math.min(limit, items.length) }, worker),
  );
  return out;
}

/** Page « vide » quand le fetch échoue : status 0, toutes les métas nulles. */
function failedPage(url: string): CrawledPage {
  return {
    url,
    status: 0,
    title: null,
    metaDescription: null,
    h1: null,
    h1Count: 0,
    canonical: null,
    robots: null,
    lang: null,
    wordCount: 0,
  };
}

/**
 * Crawle un site (via son sitemap) et met à jour l'inventaire + le mapping.
 * Ne touche JAMAIS au statut du job : c'est l'adaptateur service qui
 * complete/fail. Une page morte = ligne `status` 0, jamais un throw.
 * L'absence d'URL de site est la seule condition d'échec dur (throw).
 */
export async function runSiteCrawl(
  siteId: string,
  deps: CrawlRunnerDeps,
): Promise<CrawlTotals> {
  const { db, fetchText, fetchHtml, parseMeta, onProgress } = deps;
  const progress = async (t: CrawlTotals): Promise<void> => {
    if (onProgress) await onProgress(t);
  };

  const [site] = await db
    .select({ url: sites.url })
    .from(sites)
    .where(eq(sites.id, siteId));
  if (!site) throw new Error(`site introuvable: ${siteId}`);
  if (!site.url) {
    throw new Error(
      "site.url absente — renseigne l'URL du site avant de lancer un crawl",
    );
  }

  // --- Phase 1 : découverte -------------------------------------------------
  await progress({
    phase: "discover",
    done: 0,
    total: 0,
    pages: 0,
    errors: 0,
    mapped: 0,
    unmapped: 0,
  });
  const { urls, truncated } = await discoverSitemapUrls(site.url, fetchText);

  if (urls.length === 0) {
    const totals: CrawlTotals = {
      phase: "done",
      done: 0,
      total: 0,
      pages: 0,
      errors: 0,
      mapped: 0,
      unmapped: 0,
      message: "Aucun sitemap trouvé",
    };
    await progress(totals);
    return totals;
  }

  // --- Phase 2 : crawl (pool 6) --------------------------------------------
  const total = urls.length;
  let done = 0;
  const crawled = new Array<CrawledPage>(urls.length);
  // Update « done:0 » émis AVANT le pool : la barre s'affiche à total connu dès
  // la 1re page, même sur un petit site (< PROGRESS_EVERY pages).
  await progress({
    phase: "crawl",
    done: 0,
    total,
    pages: 0,
    errors: 0,
    mapped: 0,
    unmapped: 0,
    truncated,
  });
  let lastEmit = Date.now();

  await mapPool(urls, POOL, async (url, i) => {
    const html = await fetchHtml(url);
    if (html === null) {
      crawled[i] = failedPage(url);
    } else {
      try {
        // parseMeta = wasm `parse_page_meta` : un panic (throw) rejetterait
        // tout le Promise.all → job failed. On confine l'échec à la page :
        // une page fautive = status 0, jamais un job échoué (invariant).
        crawled[i] = { url, status: 200, ...parseMeta(html) };
      } catch {
        crawled[i] = failedPage(url);
      }
    }
    done++;
    // Émission sur cadence (1 page sur PROGRESS_EVERY) OU sur throttle temporel
    // (>2 s sans update) — la barre bouge même quand peu de pages fetchent lentement.
    if (
      done % PROGRESS_EVERY === 0 ||
      Date.now() - lastEmit >= PROGRESS_THROTTLE_MS
    ) {
      lastEmit = Date.now();
      const errorsSoFar = crawled.filter((p) => p && p.status === 0).length;
      await progress({
        phase: "crawl",
        done,
        total,
        pages: done,
        errors: errorsSoFar,
        mapped: 0,
        unmapped: 0,
        truncated,
      });
    }
  });

  // --- Upsert site_pages par lots de 200 (idempotent) ----------------------
  const crawledAt = new Date();
  const rowById = new Map<
    string,
    typeof sitePages.$inferInsert & { title: string | null; h1: string | null }
  >();
  for (const p of crawled) {
    const id = pageId(siteId, normalizePageUrl(p.url));
    // Dédup intra-crawl : deux URLs normalisées-égales → une seule ligne (le
    // dernier gagne), sinon collision de PK à l'insert.
    rowById.set(id, {
      id,
      siteId,
      url: p.url,
      status: p.status,
      title: p.title,
      metaDescription: p.metaDescription,
      h1: p.h1,
      h1Count: p.h1Count,
      canonical: p.canonical,
      robots: p.robots,
      lang: p.lang,
      wordCount: p.wordCount,
      issues: computeIssues(p),
      crawledAt,
    });
  }
  const rows = [...rowById.values()];

  for (let i = 0; i < rows.length; i += UPSERT_BATCH) {
    await db
      .insert(sitePages)
      .values(rows.slice(i, i + UPSERT_BATCH))
      .onConflictDoUpdate({
        target: sitePages.id,
        set: {
          url: sql`excluded.url`,
          status: sql`excluded.status`,
          title: sql`excluded.title`,
          metaDescription: sql`excluded.meta_description`,
          h1: sql`excluded.h1`,
          h1Count: sql`excluded.h1_count`,
          canonical: sql`excluded.canonical`,
          robots: sql`excluded.robots`,
          lang: sql`excluded.lang`,
          wordCount: sql`excluded.word_count`,
          issues: sql`excluded.issues`,
          crawledAt: sql`excluded.crawled_at`,
        },
      });
  }

  const errors = rows.filter((r) => r.status === 0).length;

  // --- Phase 3 : mapping keywords ← pages -----------------------------------
  await progress({
    phase: "map",
    done: total,
    total,
    pages: rows.length,
    errors,
    mapped: 0,
    unmapped: 0,
    truncated,
  });

  // Pages fraîches (celles qu'on vient d'upserter) — source de vérité du mapping.
  const pageByNormUrl = new Map<string, string>();
  const pageRefs: PageRef[] = [];
  for (const r of rows) {
    pageByNormUrl.set(normalizePageUrl(r.url), r.id);
    pageRefs.push({
      id: r.id,
      url: r.url,
      title: r.title,
      h1: r.h1,
    });
  }

  const kws = await db
    .select({
      id: keywords.id,
      term: keywords.term,
      gsc: keywords.gsc,
      targetPageSource: keywords.targetPageSource,
    })
    .from(keywords)
    .where(eq(keywords.siteId, siteId));

  // URL de contenu publiée par keyword (content_keywords → content_items.url).
  const contentRows = await db
    .select({
      keywordId: contentKeywords.keywordId,
      url: contentItems.url,
    })
    .from(contentKeywords)
    .innerJoin(contentItems, eq(contentKeywords.contentItemId, contentItems.id))
    .where(eq(contentKeywords.siteId, siteId));
  const contentUrlByKw = new Map<string, string>();
  for (const c of contentRows) {
    if (c.url && !contentUrlByKw.has(c.keywordId)) {
      contentUrlByKw.set(c.keywordId, c.url);
    }
  }

  type MapUpdate = {
    keywordId: string;
    targetPageId: string;
    source: "gsc" | "content" | "text";
    score: number | null;
  };
  const updates: MapUpdate[] = [];
  const clears: string[] = [];

  for (const kw of kws) {
    // Une assignation `manual` n'est JAMAIS écrasée par l'auto-mapping.
    if (kw.targetPageSource === "manual") continue;

    let match: MapUpdate | null = null;

    // 1. gsc : gsc.page normalisée == url d'une page crawlée.
    const gscPage = kw.gsc?.page;
    if (gscPage) {
      const pid = pageByNormUrl.get(normalizePageUrl(gscPage));
      if (pid) {
        match = {
          keywordId: kw.id,
          targetPageId: pid,
          source: "gsc",
          score: null,
        };
      }
    }

    // 2. content : URL d'un content_item lié, matchée à une page crawlée.
    if (!match) {
      const cUrl = contentUrlByKw.get(kw.id);
      if (cUrl) {
        const pid = pageByNormUrl.get(normalizePageUrl(cUrl));
        if (pid) {
          match = {
            keywordId: kw.id,
            targetPageId: pid,
            source: "content",
            score: null,
          };
        }
      }
    }

    // 3. text : scoring slug/H1/title (seuil 50).
    if (!match) {
      const tm = textualMatch(kw.term, pageRefs);
      if (tm) {
        match = {
          keywordId: kw.id,
          targetPageId: tm.pageId,
          source: "text",
          score: tm.score,
        };
      }
    }

    if (match) updates.push(match);
    else clears.push(kw.id); // gap de couverture (ré-éligible au prochain crawl)
  }

  // Efface les assignations auto orphelines (non-manual sans match) — idempotent.
  for (let i = 0; i < clears.length; i += WRITE_CHUNK) {
    await db
      .update(keywords)
      .set({
        targetPageId: null,
        targetPageSource: null,
        targetPageScore: null,
      })
      // Garde-fou anti-course : un `manual` posé PENDANT le crawl (non-manual
      // au SELECT) ne doit jamais être effacé par ce clear.
      .where(
        and(
          inArray(keywords.id, clears.slice(i, i + WRITE_CHUNK)),
          sql`${keywords.targetPageSource} is distinct from 'manual'`,
        ),
      );
  }

  // Applique les matches (page/source/score diffèrent par kw) par lots.
  for (let i = 0; i < updates.length; i += WRITE_CHUNK) {
    const batch = updates.slice(i, i + WRITE_CHUNK);
    await db.transaction(async (tx) => {
      for (const u of batch) {
        await tx
          .update(keywords)
          .set({
            targetPageId: u.targetPageId,
            targetPageSource: u.source,
            targetPageScore: u.score,
          })
          // Garde-fou anti-course : ne jamais écraser un manual concurrent.
          .where(
            and(
              eq(keywords.id, u.keywordId),
              sql`${keywords.targetPageSource} is distinct from 'manual'`,
            ),
          );
      }
    });
  }

  const totals: CrawlTotals = {
    phase: "done",
    done: total,
    total,
    pages: rows.length,
    errors,
    mapped: updates.length,
    unmapped: clears.length,
    truncated,
  };
  await progress(totals);
  return totals;
}