// src/lib/backlink/import.ts
import "server-only";
import { eq, sql } from "drizzle-orm";
import { chunk } from "@/lib/db/chunk";
import { db } from "@/lib/db/client";
import { linkSources } from "@/lib/db/domain-schema";
import type { CanonicalSource } from "./import-types";

/** The transaction type `db.transaction(async (tx) => …)` hands its callback. */
type Tx = Parameters<Parameters<typeof db.transaction>[0]>[0];

/** Namespace a source id per org. `link_sources.id` is the GLOBAL primary key,
 *  but the import-core crate derives ids from the URL alone (`slug_id`, dashes
 *  only — never `__`). Two orgs importing the same URL would collide on the PK
 *  and the second org's upsert would silently no-op. Rule:
 *  - raw crate id (URL slug, no org prefix) → prefix with `${orgId}__`;
 *  - id already starting with `${orgId}__` (bootstrap-provisioned rows, or a
 *    re-import of an already-namespaced id) → left as is, never double-prefixed.
 *  Matches org-bootstrap's convention (`suffixOf` = part after the first `__`). */
export function namespaceSourceId(orgId: string, id: string): string {
  return id.startsWith(`${orgId}__`) ? id : `${orgId}__${id}`;
}

/** Upsert sources for an org. New rows insert; existing (same id) update —
 *  EXCEPT `flags`, which are not in the CSV and must not be clobbered.
 *  Ids are namespaced per org (see `namespaceSourceId`) — the single choke
 *  point through which every CSV / manual import flows.
 *
 *  `tx`: an already-open transaction to run in (sec audit A2/P8 —
 *  `commitImportAction` opens ONE transaction covering
 *  `lockOrgForQuota` + `assertWithinQuota` + this insert, so the quota
 *  check and the write can't be raced apart; passing it through here avoids
 *  nesting a second transaction inside the first). Omit it (manual single
 *  add via `addSourceAction`, no quota gate) to keep the old
 *  self-contained-transaction behaviour. */
export async function upsertLinkSources(
  orgId: string,
  rows: CanonicalSource[],
  tx?: Tx,
): Promise<void> {
  if (rows.length === 0) return;
  const run = async (t: Tx | typeof db) => {
    // ~15 bind params per row → chunk under the Postgres ~65 535-param cap.
    for (const slice of chunk(rows, 500)) {
      await t
        .insert(linkSources)
        .values(
          slice.map((r) => ({
            id: namespaceSourceId(orgId, r.id),
            orgId,
            platform: r.platform,
            url: r.url,
            domainNorm: r.domainNorm,
            kind: r.kind,
            da: r.da,
            effort: r.effort,
            linkType: r.linkType,
            language: r.language,
            themes: r.themes,
            tags: r.tags,
            flags: { toxic: false, disavow: false },
            note: r.note,
          })),
        )
        .onConflictDoUpdate({
          target: linkSources.id,
          setWhere: eq(linkSources.orgId, orgId),
          set: {
            platform: sql`excluded.platform`,
            url: sql`excluded.url`,
            domainNorm: sql`excluded.domain_norm`,
            kind: sql`excluded.kind`,
            da: sql`excluded.da`,
            effort: sql`excluded.effort`,
            linkType: sql`excluded.link_type`,
            language: sql`excluded.language`,
            themes: sql`excluded.themes`,
            tags: sql`excluded.tags`,
            note: sql`excluded.note`,
            // flags intentionally omitted — preserve existing toxic/disavow.
          },
        });
    }
  };
  if (tx) {
    await run(tx);
  } else {
    await db.transaction(run);
  }
}