// src/lib/gsc/sync.ts — sync a site's GSC queries into the research stage.
// Two effects: (1) overlay positioning onto existing keywords AND promote them
// into "research" (a ranking keyword is actionable), and (2) auto-create research
// keywords for unmatched queries that already rank within striking distance.
import "server-only";
import { eq, sql } from "drizzle-orm";
import * as wasm from "import-core";
import { getOrgQuotaLimits } from "@/lib/billing/quotas";
import { getUsage } from "@/lib/billing/usage";
import { chunk } from "@/lib/db/chunk";
import { db } from "@/lib/db/client";
import {
  keywords as keywordsTable,
  sites as sitesTable,
} from "@/lib/db/domain-schema";
import { getExcludedSlugs } from "@/lib/keywords/queries";
import type { GscTracking } from "@/lib/keywords/types";
import { createGscKeywordsForSite } from "./discovery";
import {
  countryForLang,
  type QueryCountryImpressions,
  resolveGscCountry,
  resolveQueryLanguages,
} from "./language";
import type { GscQueryRow } from "./types";

export type GscSyncExtras = {
  /** alpha-2 → lignes query du pays (appel filtré country). */
  byCountry?: Record<string, GscQueryRow[]>;
  /** queryLower → URL qui ranke (réduction topPageByQuery). */
  topPageByQuery?: Map<string, string>;
  /** queryLower → impressions par pays (dimension [query, country]), pour
   *  attribuer la langue et écarter les marchés hors-projet. */
  queryCountries?: Map<string, QueryCountryImpressions[]>;
};

export type GscSyncResult = {
  /** existing keywords overlaid + promoted into research */
  matched: number;
  /** new research keywords created from unmatched ranking queries */
  created: number;
  /** total GSC query rows considered */
  queries: number;
};

/** Auto-import an unmatched GSC query as a research keyword only when it ranks
 *  within striking distance (position ≤ 20 — the top10 + quick-win bands; see
 *  `gscBand`). Beyond that is long-tail noise we don't want flooding Recherche. */
const AUTO_IMPORT_MAX_POSITION = 20;

/** Rows per batched `UPDATE ... FROM (VALUES ...)` statement. Each row carries
 *  ~5 bind params, well under Postgres' ~65 535 param cap at this size. */
const BATCH_SIZE = 500;

/** GSC surfaces search-operator queries (quoted exact-match, parenthesised
 *  groups, boolean OR/AND) issued by power users, tools and bots. They're never
 *  useful keywords, so we never auto-import them. */
export function isOperatorQuery(query: string): boolean {
  return /["()]/.test(query) || /\b(?:OR|AND)\b/.test(query);
}

/** Sync GSC queries for a site into the research stage:
 *  - a query matching an existing keyword (by lower(term)) gets a positioning
 *    overlay and is promoted into the "research" stage (status idea→active,
 *    never overriding a deliberate status);
 *  - an unmatched query ranking ≤ position 20 is created as a research keyword;
 *  - an unmatched query beyond position 20 is ignored. */
export async function syncGscForSite(
  siteId: string,
  rows: GscQueryRow[],
  extras: GscSyncExtras = {},
): Promise<GscSyncResult> {
  const now = new Date().toISOString();
  const [site] = await db
    .select({
      orgId: sitesTable.orgId,
      lang: sitesTable.primaryLanguage,
      languages: sitesTable.languages,
      countries: sitesTable.targetCountries,
    })
    .from(sitesTable)
    .where(eq(sitesTable.id, siteId));
  const primary = site?.lang ?? "en";
  const countries = site?.countries ?? [];
  // The languages set up in the project — every created/overlaid keyword must
  // belong to one of these; queries attributed elsewhere are dropped.
  const configured = [...new Set([primary, ...(site?.languages ?? [])])];

  // Attribute each query to a configured language (URL locale → dominant
  // country → primary), or null when it ranks only in non-configured markets.
  const queryLang = resolveQueryLanguages({
    rows,
    configured,
    primary,
    topPageByQuery: extras.topPageByQuery,
    queryCountries: extras.queryCountries,
  });

  // Lean projection: matching only needs id/term + the fields we union/bridge.
  // The heavy serp/metrics/trend jsonb stays out of this read.
  const existing = await db
    .select({
      id: keywordsTable.id,
      term: keywordsTable.term,
      sources: keywordsTable.sources,
      status: keywordsTable.status,
      stages: keywordsTable.stages,
      gsc: keywordsTable.gsc,
    })
    .from(keywordsTable)
    .where(eq(keywordsTable.siteId, siteId));
  const byTerm = new Map(existing.map((k) => [k.term.toLowerCase(), k]));

  const excluded = await getExcludedSlugs(siteId);

  const toCreate: { row: GscQueryRow; lang: string }[] = [];
  let matched = 0;

  // Matched pass — collect the final per-keyword values (last query row wins for
  // a given id, mirroring the original last-UPDATE-wins), then flush in batches.
  // `matched` counts matched query rows (not distinct keywords) — unchanged.
  type MatchedUpdate = {
    id: string;
    gsc: GscTracking;
    sources: string[];
    status: string;
    stages: string[];
  };
  const updatesById = new Map<string, MatchedUpdate>();
  for (const row of rows) {
    if (excluded.has(wasm.slugify(row.query))) continue;
    // Drop queries attributed to a non-configured language (foreign markets):
    // no overlay, no create — this is what stops the languages from mixing.
    const lang = queryLang.get(row.query.toLowerCase());
    if (lang == null) continue;
    const found = byTerm.get(row.query.toLowerCase());
    if (!found) {
      if (
        row.position <= AUTO_IMPORT_MAX_POSITION &&
        !isOperatorQuery(row.query)
      )
        toCreate.push({ row, lang });
      continue;
    }
    const gsc: GscTracking = {
      clicks: row.clicks,
      impressions: row.impressions,
      ctr: row.ctr,
      position: row.position,
      updatedAt: now,
      // Fetch pages échoué (map absente) → préserve la page connue ; map
      // présente sans entrée → null (la query ne ranke plus sur une page).
      page: extras.topPageByQuery
        ? (extras.topPageByQuery.get(row.query.toLowerCase()) ?? null)
        : (found.gsc?.page ?? null),
    };
    const sources = found.sources.includes("gsc")
      ? found.sources
      : [...found.sources, "gsc" as const];
    // a ranking keyword is "active" — promote from "idea", but never override
    // a deliberate status (watch/active/archived).
    const status = found.status === "idea" ? "active" : found.status;
    // a ranking keyword is actionable → surface it on Recherche.
    const stages = found.stages.includes("research")
      ? found.stages
      : [...found.stages, "research" as const];
    updatesById.set(found.id, { id: found.id, gsc, sources, status, stages });
    matched++;
  }

  // Overlay par pays — matchés seulement. Le merge jsonb (`||`) doit voir la
  // valeur courante de la colonne ; on collecte donc le patch par id (last wins,
  // identique à l'ancien per-row), puis on flush en batch avec `col || patch`.
  const byCountryPatch = new Map<string, Record<string, GscTracking>>();
  const byCountry = extras.byCountry ?? {};
  for (const [country, countryRows] of Object.entries(byCountry)) {
    for (const row of countryRows) {
      // A query explicitly dropped (foreign language) gets no per-country
      // overlay either; queries absent from the main set (undefined) still pass.
      if (queryLang.get(row.query.toLowerCase()) === null) continue;
      const found = byTerm.get(row.query.toLowerCase());
      if (!found) continue;
      const tracking: GscTracking = {
        clicks: row.clicks,
        impressions: row.impressions,
        ctr: row.ctr,
        position: row.position,
        updatedAt: now,
      };
      // Merge multiple country rows for the same keyword into one patch object;
      // the FROM-VALUES flush then applies `gsc_by_country || patch` once per id.
      const patch = byCountryPatch.get(found.id) ?? {};
      patch[country] = tracking;
      byCountryPatch.set(found.id, patch);
    }
  }

  await db.transaction(async (tx) => {
    // Chunked `UPDATE ... FROM (VALUES ...)` — one round-trip per ~500 rows
    // instead of thousands of sequential UPDATEs (Vercel single-connection
    // timeout risk). Cast each VALUES column to the target column's type.
    for (const batch of chunk([...updatesById.values()], BATCH_SIZE)) {
      const values = batch.map(
        (u) =>
          sql`(${u.id}, ${JSON.stringify(u.gsc)}::jsonb, ${sql`ARRAY[${sql.join(
            u.sources.map((s) => sql`${s}`),
            sql`, `,
          )}]::text[]`}, ${u.status}, ${sql`ARRAY[${sql.join(
            u.stages.map((s) => sql`${s}`),
            sql`, `,
          )}]::text[]`})`,
      );
      await tx.execute(sql`
        UPDATE ${keywordsTable} AS k
        SET gsc = v.gsc, sources = v.sources, status = v.status, stages = v.stages
        FROM (VALUES ${sql.join(values, sql`, `)})
          AS v(id, gsc, sources, status, stages)
        WHERE k.id = v.id
      `);
    }

    // Per-country jsonb merge — batched the same way; `||` against the current
    // column value preserves untouched countries.
    const countryEntries = [...byCountryPatch.entries()];
    for (const batch of chunk(countryEntries, BATCH_SIZE)) {
      const values = batch.map(
        ([id, patch]) => sql`(${id}, ${JSON.stringify(patch)}::jsonb)`,
      );
      await tx.execute(sql`
        UPDATE ${keywordsTable} AS k
        SET gsc_by_country = k.gsc_by_country || v.patch
        FROM (VALUES ${sql.join(values, sql`, `)}) AS v(id, patch)
        WHERE k.id = v.id
      `);
    }
  });

  // Plan quota — background sync: CAP silently at the remaining keyword quota
  // (import only what fits, in the order GSC returned them); never error out.
  if (toCreate.length > 0 && site) {
    const limit = (await getOrgQuotaLimits(site.orgId)).keywords;
    if (limit !== -1) {
      const current = await getUsage(site.orgId, "keywords");
      const remaining = Math.max(0, limit - current);
      if (toCreate.length > remaining) {
        console.warn(
          `[gsc:sync] keyword quota cap: importing ${remaining}/${toCreate.length} auto-discovered queries for site ${siteId} (org ${site.orgId} at ${current}/${limit})`,
        );
        toCreate.length = remaining;
      }
    }
  }

  // Group the survivors by attributed language and create each group with its
  // own lang (the whole point — no more tagging everything as the primary). The
  // country column gets a target country that matches the lang, else the first
  // target. createGscKeywordsForSite is idempotent (onConflictDoNothing on id),
  // so a slug that collides with an existing keyword is simply skipped.
  const byLang = new Map<string, GscQueryRow[]>();
  for (const { row, lang } of toCreate) {
    const list = byLang.get(lang) ?? [];
    list.push(row);
    byLang.set(lang, list);
  }
  let created = 0;
  for (const [lang, langRows] of byLang) {
    // Fallback (pas de breakdown pour la requête) : un pays cible qui matche la
    // langue, sinon le 1er cible. Sinon on prend le pays réel dominant.
    const langFallback =
      countryForLang(lang, countries) ?? countries[0] ?? "US";
    const res = await createGscKeywordsForSite(siteId, langRows, {
      lang,
      country: (q) =>
        resolveGscCountry(
          extras.queryCountries?.get(q.toLowerCase()),
          countries,
          langFallback,
        ),
      stages: ["research"],
    });
    created += res.created;
  }
  return { matched, created, queries: rows.length };
}