// src/lib/seranking/sync.ts — orchestrateur du « Sync SE-Ranking » : résout la
// sélection, groupe par base régionale, appelle l'API bulk (via le limiteur),
// normalise en Rust (wasm) et merge ENRICHMENT-SAFE : un enrichissement ne
// réécrit jamais l'identité d'un mot-clé (term/lang/country/status/theme
// intacts) et ne crée jamais de ligne — UPDATE only.
import "server-only";
import { inArray, sql } from "drizzle-orm";
import * as wasm from "import-core";
import {
  assertWithinQuota,
  assertWithinQuotaTx,
  getOrgQuotaLimits,
  lockOrgForQuota,
} from "@/lib/billing/quotas";
import { chunk } from "@/lib/db/chunk";
import { db } from "@/lib/db/client";
import { keywords } from "@/lib/db/domain-schema";
import { setEnrichStatus } from "@/lib/keywords/enrich-status";
import {
  type KeywordSelection,
  selectionWhere,
} from "@/lib/keywords/list-query";
import { consumeSync, type SerankingAccess } from "./balance";
import { bulkKeywordMetrics, limiterForKey, SerankingError } from "./client";
import { serankingSource } from "./country-db";

/** Cap aligné sur MAX_FILTER_EXPORT (actions.ts) — un sync ne traite jamais
 *  plus que ce que les exports/bulk actions acceptent. Exporté : réutilisé par
 *  `estimateSerankingSyncAction` (actions.ts) pour capper la même sélection. */
export const MAX_SYNC = 20_000;
/** Limite API : 5000 kw par requête bulk. */
const API_CHUNK = 5_000;
/** Lignes UPDATE par transaction d'écriture. */
const WRITE_CHUNK = 500;

export interface SerankingSyncResult {
  /** Mots-clés ayant reçu des métriques. */
  enriched: number;
  /** Mots-clés répondus `is_data_found=false` (aucune donnée SE-Ranking). */
  noData: number;
  /** Pays cibles sans base régionale SE-Ranking (groupes sautés). */
  skippedCountries: string[];
  /** Requêtes API bulk émises (×100 = crédits consommés). */
  requests: number;
}

export interface SelectedKw {
  id: string;
  term: string;
  country: string;
  sources: string[];
}

/** Sortie normalisée du wasm `parse_seranking_api` (camelCase via serde). */
export type SerankingApiRow = {
  term: string;
  found: boolean;
  metrics: {
    volume: number | null;
    kd: number | null;
    cpc: number | null;
    growthPct: number | null;
    competitiveDensity: number | null;
    numberOfResults: number | null;
  } | null;
  intent: string[];
  trend: { month: string; volume: number }[];
};

/** Clé composée base×terme de `termOf` (le même terme peut exister dans
 *  plusieurs bases régionales). */
function termKey(dbCode: string, lowerTerm: string): string {
  return `${dbCode}:${lowerTerm}`;
}

/** Pose un état sur le rail d'enrichissement (provider `seranking`) en
 *  best-effort : un échec d'écriture de statut ne doit JAMAIS faire échouer le
 *  sync (le spinner par-ligne est cosmétique ; l'enrichissement, lui, est déjà
 *  commité). No-op sur tableau vide. */
async function markEnrich(
  ids: string[],
  status: "syncing" | "done" | "error",
  error?: string,
): Promise<void> {
  if (ids.length === 0) return;
  try {
    await setEnrichStatus(ids, "seranking", {
      status,
      at: new Date().toISOString(),
      ...(error ? { error } : {}),
    });
  } catch (e) {
    console.warn(`[seranking] enrich-status ${status} write failed`, e);
  }
}

/** Groupe la sélection par base SE-Ranking, dédupe par terme lowercasé (un
 *  terme envoyé une fois peut porter plusieurs ids), collecte les pays sans
 *  base. Pur — testé sans DB. */
export function groupByDb(rows: SelectedKw[]): {
  groups: Map<string, Map<string, string[]>>;
  termOf: Map<string, string>;
  skippedCountries: string[];
} {
  const groups = new Map<string, Map<string, string[]>>();
  const termOf = new Map<string, string>();
  const skipped = new Set<string>();
  for (const r of rows) {
    const dbCode = serankingSource(r.country);
    if (!dbCode) {
      skipped.add(r.country.toUpperCase());
      continue;
    }
    const lower = r.term.trim().toLowerCase();
    if (!lower) continue;
    let group = groups.get(dbCode);
    if (!group) {
      group = new Map();
      groups.set(dbCode, group);
    }
    const ids = group.get(lower);
    if (ids) ids.push(r.id);
    else {
      group.set(lower, [r.id]);
      termOf.set(termKey(dbCode, lower), r.term.trim());
    }
  }
  return { groups, termOf, skippedCountries: [...skipped].sort() };
}

/** Estimation pure (dialog de confirmation, `estimateSerankingSyncAction`) :
 *  réutilise `groupByDb` — `requests` = Σ ceil(termes uniques du groupe /
 *  API_CHUNK) (même découpage que la boucle d'envoi réelle), `keywords` = nb
 *  de lignes couvertes par ces groupes (les pays sautés en sont exclus, comme
 *  `skippedCountries`), `countries` = nb de bases régionales couvertes (≈ nb
 *  de pays cibles couverts — une base SE-Ranking par pays, cf. country-db.ts).
 *  Ne fait AUCUN appel réseau. */
export function estimateSync(rows: SelectedKw[]): {
  keywords: number;
  requests: number;
  countries: number;
  skippedCountries: string[];
} {
  const { groups, skippedCountries } = groupByDb(rows);
  let keywordsCount = 0;
  let requests = 0;
  for (const terms of groups.values()) {
    for (const ids of terms.values()) keywordsCount += ids.length;
    requests += Math.ceil(terms.size / API_CHUNK);
  }
  return {
    keywords: keywordsCount,
    requests,
    countries: groups.size,
    skippedCountries,
  };
}

/** Applique un lot de résultats API aux mots-clés (par ids), enrichment-safe,
 *  par transactions verrouillées quota (WRITE_CHUNK lignes chacune). Retourne
 *  le nb de kw enrichis. */
async function applyRows(
  orgId: string,
  updates: Array<{ ids: string[]; row: SerankingApiRow }>,
  syncedAt: string,
): Promise<number> {
  if (updates.length === 0) return 0;
  // Limite résolue AVANT la transaction — getOrgQuotaLimits passe par le pool
  // `db` et deadlockerait appelé depuis un tx ouvert (voir doc quotas.ts).
  const limits = await getOrgQuotaLimits(orgId);
  let enriched = 0;
  for (const slice of chunk(updates, WRITE_CHUNK)) {
    await db.transaction(async (tx) => {
      await lockOrgForQuota(tx, orgId);
      // Delta réel : ids qui n'ont pas encore la source (quota kw×source —
      // ajouter "seranking" à un kw existant coûte 1 unité, re-sync = 0).
      const allIds = slice.flatMap((u) => u.ids);
      const existing = await tx
        .select({ id: keywords.id, sources: keywords.sources })
        .from(keywords)
        .where(inArray(keywords.id, allIds));
      const delta = existing.filter(
        (k) => !k.sources.includes("seranking"),
      ).length;
      if (delta > 0) {
        await assertWithinQuotaTx(
          tx,
          orgId,
          "keywords",
          limits.keywords,
          delta,
        );
      }
      for (const { ids, row } of slice) {
        if (!row.metrics) continue;
        const metricsJson = JSON.stringify({
          seranking: { ...row.metrics, updatedAt: syncedAt },
        });
        const trendJson = JSON.stringify(row.trend);
        const trendBySourceJson = JSON.stringify({ seranking: row.trend });
        const intentArr = row.intent;
        const res = await tx
          .update(keywords)
          .set({
            metrics: sql`${keywords.metrics} || ${metricsJson}::jsonb`,
            sources: sql`(select coalesce(array_agg(distinct s), '{}') from unnest(${keywords.sources} || '{seranking}'::text[]) s)`,
            intent: sql`(select coalesce(array_agg(distinct i), '{}') from unnest(${keywords.intent} || ${sql.param(intentArr)}::text[]) i)`,
            // primarySource bascule vers seranking UNIQUEMENT si la source
            // primaire actuelle ne porte aucun volume (cas GSC/manuel) —
            // même condition que hasNoMetric côté UI.
            primarySource: sql`CASE WHEN (${keywords.metrics} -> ${keywords.primarySource} ->> 'volume') IS NULL THEN 'seranking' ELSE ${keywords.primarySource} END`,
            // Spread conditionnel — ne PAS passer de clés `undefined` à .set().
            ...(row.trend.length
              ? {
                  trendBySource: sql`${keywords.trendBySource} || ${trendBySourceJson}::jsonb`,
                  // trend global (fallback sparkline) : rempli seulement s'il est vide.
                  trend: sql`CASE WHEN jsonb_array_length(${keywords.trend}) = 0 THEN ${trendJson}::jsonb ELSE ${keywords.trend} END`,
                }
              : {}),
          })
          .where(inArray(keywords.id, ids))
          .returning({ id: keywords.id });
        enriched += res.length;
      }
    });
  }
  return enriched;
}

/** Sync SE-Ranking d'une sélection du réservoir. Le pré-check quota tourne
 *  AVANT le premier appel API (ne pas dépenser de crédits pour échouer au
 *  moment d'écrire) ; la vérification autoritaire reste dans la transaction
 *  d'écriture (lockOrgForQuota + assertWithinQuotaTx). `access` (résolu en
 *  amont par l'appelant via `getSerankingAccess`) pilote le métering : mode
 *  metered → 1 `consume` avant CHAQUE requête bulk (stop propre
 *  `SerankingError("no-syncs")` à l'épuisement, writes déjà commités
 *  conservés — jamais de refund) ; mode byok → la requête part avec la clé +
 *  le limiteur dédiés à l'org ; mode unlimited → comportement inchangé (clé
 *  plateforme, aucune consommation). */
export async function syncSerankingForSelection(
  siteId: string,
  orgId: string,
  sel: KeywordSelection,
  access: SerankingAccess,
  deps?: {
    fetchBulk?: typeof bulkKeywordMetrics;
    parse?: (json: string) => SerankingApiRow[];
    consume?: (orgId: string) => Promise<boolean>;
  },
): Promise<SerankingSyncResult> {
  const fetchBulk = deps?.fetchBulk ?? bulkKeywordMetrics;
  const parse =
    deps?.parse ??
    ((json: string) => wasm.parse_seranking_api(json) as SerankingApiRow[]);
  const consume = deps?.consume ?? consumeSync;

  const rows: SelectedKw[] = await db
    .select({
      id: keywords.id,
      term: keywords.term,
      country: keywords.country,
      sources: keywords.sources,
    })
    .from(keywords)
    .where(selectionWhere(siteId, sel))
    .limit(MAX_SYNC);

  const { groups, termOf, skippedCountries } = groupByDb(rows);

  // Pré-check quota (advisory, hors transaction — voir doc assertWithinQuota).
  const projectedDelta = rows.filter(
    (r) => !r.sources.includes("seranking"),
  ).length;
  if (projectedDelta > 0) {
    await assertWithinQuota(orgId, "keywords", projectedDelta);
  }

  // Timestamp ISO COMPLET : une date seule ("YYYY-MM-DD") est parsée minuit
  // UTC par le front → « Synchronisé il y a 9 h » juste après un sync (UTC+7).
  const syncedAt = new Date().toISOString();
  let enriched = 0;
  let noData = 0;
  let requests = 0;

  for (const [dbCode, terms] of groups) {
    const lowerTerms = [...terms.keys()];
    for (const slice of chunk(lowerTerms, API_CHUNK)) {
      // Rail par-ligne : tous les ids du chunk passent `syncing` avant l'appel
      // API (le spinner du réservoir tourne pendant la requête bulk). Rail
      // provider-agnostique : ne touche que la clé `seranking`.
      const sliceIds = slice.flatMap((t) => terms.get(t) ?? []);
      await markEnrich(sliceIds, "syncing");
      // Consommation AVANT l'appel API (metered uniquement) — décrément
      // atomique sans verrou (voir balance.ts) ; false = solde épuisé, on
      // s'arrête net, le récap partiel (enriched/noData/requests) reste basé
      // sur les chunks déjà commités.
      if (access.mode === "metered") {
        const ok = await consume(orgId);
        if (!ok) {
          // Solde épuisé : ne pas laisser ce chunk `syncing` pour toujours
          // (spinner infini côté UI) — on le repasse `error` avant de sortir.
          await markEnrich(sliceIds, "error", "quota");
          throw new SerankingError(
            "no-syncs",
            `${enriched} enrichi(s) avant épuisement du solde`,
          );
        }
      }
      // À partir d'ici les ids du chunk sont `syncing`. TOUTE erreur (fetch,
      // parse, écriture) les laisserait `syncing` pour toujours → spinner infini
      // côté UI. try/catch : on repasse le chunk en `error:internal` best-effort
      // (markEnrich avale déjà ses propres échecs d'écriture de statut) AVANT de
      // propager. Même philosophie que le fix Task 8. Le stop metered (solde
      // épuisé) est traité PLUS HAUT, hors de ce try, pour conserver son
      // `error:quota` sans être écrasé par `internal`.
      try {
        const payload = slice.map((t) => termOf.get(termKey(dbCode, t)) ?? t);
        const raw = await fetchBulk({
          source: dbCode,
          keywords: payload,
          ...(access.mode === "byok"
            ? { apiKey: access.apiKey, limiter: limiterForKey(orgId) }
            : {}),
        });
        requests++;
        const parsed = parse(raw);
        const updates: Array<{ ids: string[]; row: SerankingApiRow }> = [];
        for (const row of parsed) {
          const ids = terms.get(row.term.trim().toLowerCase());
          if (!ids) continue; // écho inconnu — ignoré
          if (!row.found || !row.metrics) {
            noData += ids.length;
            continue;
          }
          updates.push({ ids, row });
        }
        enriched += await applyRows(orgId, updates, syncedAt);

        // Rail par-ligne, transition finale : les ids enrichis passent `done`,
        // le reste du chunk (no-data / écho absent de la réponse) `error:
        // not_found` — jamais laissés `syncing`.
        const doneIds = updates.flatMap((u) => u.ids);
        const doneSet = new Set(doneIds);
        await markEnrich(doneIds, "done");
        await markEnrich(
          sliceIds.filter((id) => !doneSet.has(id)),
          "error",
          "not_found",
        );
      } catch (e) {
        await markEnrich(sliceIds, "error", "internal");
        throw e;
      }
    }
  }

  // Observabilité ops : crédits consommés (100/requête bulk).
  console.info(
    `[seranking] org=${orgId} site=${siteId} requests=${requests} credits=${requests * 100} enriched=${enriched} noData=${noData} skipped=${skippedCountries.join(",") || "-"}`,
  );

  return { enriched, noData, skippedCountries, requests };
}