import { sql } from "drizzle-orm";
import { chunk } from "@/lib/db/chunk";
import { keywords } from "@/lib/db/domain-schema";
import type { KeywordStage } from "@/lib/keywords/types";
import { db } from "./db";
import type { CanonicalKeyword } from "./types";

export async function upsertCanonical(
  siteId: string,
  rows: CanonicalKeyword[],
  /** Stages to tag every imported keyword with (e.g. ["research"] when the
   *  import targets Recherche). Unioned with any existing stages on conflict. */
  stages: KeywordStage[] = [],
): Promise<{ upserted: number; created: number; matched: number }> {
  if (rows.length === 0) return { upserted: 0, created: 0, matched: 0 };
  const values = rows.map((k) => ({
    id: `${siteId}-${k.slug}`,
    siteId,
    term: k.term,
    lang: k.lang,
    country: k.country,
    theme: null,
    status: "idea" as const,
    intent: k.intent,
    primarySource: k.primarySource,
    sources: [k.primarySource],
    metrics: { [k.primarySource]: k.metrics },
    trend: [],
    stages,
    serpFeatures: k.serpFeatures,
  }));
  // `xmax = 0` is true for a freshly inserted row, false for one updated by the
  // ON CONFLICT path — the standard Postgres trick to tell created from matched.
  // Requires a plain (non-partitioned) `keywords` table (0057) — unsupported on
  // the hash-partitioned parent ("cannot retrieve a system column in this
  // context").
  let created = 0;
  await db.transaction(async (tx) => {
    // Target case is 10k–100k rows; ~13 bind params per row → chunk under the
    // Postgres ~65 535-param cap, atomically (one transaction).
    for (const slice of chunk(values, 1000)) {
      const out = await tx
        .insert(keywords)
        .values(slice)
        .onConflictDoUpdate({
          target: [keywords.id],
          set: {
            term: sql`excluded.term`,
            lang: sql`excluded.lang`,
            country: sql`excluded.country`,
            intent: sql`excluded.intent`,
            primarySource: sql`excluded.primary_source`,
            sources: sql`(select array_agg(distinct s) from unnest(${keywords.sources} || excluded.sources) as s)`,
            // Union: existing keywords GAIN the imported stage without losing the
            // stages they already have (coalesce keeps '{}' when both empty).
            stages: sql`(select coalesce(array_agg(distinct s), '{}') from unnest(${keywords.stages} || excluded.stages) as s)`,
            metrics: sql`${keywords.metrics} || excluded.metrics`,
            serpFeatures: sql`excluded.serp_features`,
          },
        })
        .returning({ created: sql<boolean>`(xmax = 0)` });
      for (const r of out) if (r.created) created++;
    }
  });
  const upserted = values.length;
  return { upserted, created, matched: upserted - created };
}