// src/lib/backlink/import-actions.ts
"use server";
import { and, eq, inArray } from "drizzle-orm";
import * as wasm from "import-core";
import { revalidatePath } from "next/cache";
import { getActiveOrgId } from "@/lib/active-org";
import {
  assertWithinQuotaTx,
  getOrgQuotaLimits,
  lockOrgForQuota,
  QuotaError,
  type QuotaExceeded,
  quotaPayload,
} from "@/lib/billing/quotas";
import { db } from "@/lib/db/client";
import { linkSources } from "@/lib/db/domain-schema";
import { namespaceSourceId, upsertLinkSources } from "./import";
import type {
  CanonicalSource,
  PreviewResult,
  PreviewRow,
} from "./import-types";

const RESERVOIR_PATH = "/backlink/reservoir";

interface RowResult {
  ok: boolean;
  error?: string;
  source?: CanonicalSource;
}

function okSources(rows: RowResult[]): CanonicalSource[] {
  return rows
    .filter((r) => r.ok && r.source)
    .map((r) => r.source as CanonicalSource);
}

/** Parse + classify new/update against the org's existing ids. No write. */
export async function previewImportAction(
  csvText: string,
): Promise<PreviewResult> {
  const rows = wasm.parse_csv(csvText) as RowResult[];
  const orgId = await getActiveOrgId();
  // Stored ids are org-namespaced (see upsertLinkSources) — classify
  // new/update against the namespaced form, not the raw crate slug.
  const ids = okSources(rows).map((s) => namespaceSourceId(orgId, s.id));
  const existing = ids.length
    ? new Set(
        (
          await db
            .select({ id: linkSources.id })
            .from(linkSources)
            .where(
              and(inArray(linkSources.id, ids), eq(linkSources.orgId, orgId)),
            )
        ).map((r) => r.id),
      )
    : new Set<string>();

  const out: PreviewRow[] = rows.map((r) => {
    if (!r.ok || !r.source) return { status: "error", error: r.error };
    const s = r.source;
    return {
      status: existing.has(namespaceSourceId(orgId, s.id)) ? "update" : "new",
      id: s.id,
      platform: s.platform,
      url: s.url,
      da: s.da,
    };
  });
  return {
    rows: out,
    counts: {
      created: out.filter((r) => r.status === "new").length,
      updated: out.filter((r) => r.status === "update").length,
      errors: out.filter((r) => r.status === "error").length,
    },
  };
}

/** Re-parse server-side (never trust the client) + upsert the ok rows. */
export async function commitImportAction(csvText: string): Promise<{
  created: number;
  updated: number;
  errors: number;
  /** Set (with zero counts) when the import would exceed the backlink quota. */
  quota?: QuotaExceeded;
}> {
  const rows = wasm.parse_csv(csvText) as RowResult[];
  const sources = okSources(rows);
  const orgId = await getActiveOrgId();
  // Stored ids are org-namespaced (see upsertLinkSources).
  const ids = sources.map((s) => namespaceSourceId(orgId, s.id));
  const existing = ids.length
    ? new Set(
        (
          await db
            .select({ id: linkSources.id })
            .from(linkSources)
            .where(
              and(inArray(linkSources.id, ids), eq(linkSources.orgId, orgId)),
            )
        ).map((r) => r.id),
      )
    : new Set<string>();
  const updated = sources.filter((s) =>
    existing.has(namespaceSourceId(orgId, s.id)),
  ).length;
  // Plan quota (backlinkRows). This action inserts `link_sources`, not
  // `backlinks` rows directly — the fan-out to per-site backlink rows happens
  // later (site creation copies the whole catalog; prospect assignment adds
  // source×site pairs). Increment = the NEW sources in this import: the
  // conservative floor of the rows the catalog growth will materialize
  // (≥1 backlink row per source once any site targets it).
  const created = sources.length - updated;
  // Sec audit A2/P8 — the quota check and the upsert now share ONE
  // transaction, gated by lockOrgForQuota, so a concurrent import for the
  // same org can't read the same pre-write count and both pass. Limit
  // resolved BEFORE the transaction — see assertWithinQuotaTx's doc.
  const limits = created > 0 ? await getOrgQuotaLimits(orgId) : null;
  try {
    await db.transaction(async (tx) => {
      if (created > 0 && limits) {
        await lockOrgForQuota(tx, orgId);
        await assertWithinQuotaTx(
          tx,
          orgId,
          "backlinkRows",
          limits.backlinkRows,
          created,
        );
      }
      await upsertLinkSources(orgId, sources, tx);
    });
  } catch (e) {
    if (e instanceof QuotaError) {
      return { created: 0, updated: 0, errors: 0, quota: quotaPayload(e) };
    }
    throw e;
  }
  revalidatePath(RESERVOIR_PATH);
  return {
    created,
    updated,
    errors: rows.length - sources.length,
  };
}

/** Manual single add. */
export async function addSourceAction(
  input: Record<string, string>,
): Promise<{ ok: true } | { ok: false; error: string }> {
  const res = wasm.normalize_row(input) as RowResult;
  if (!res.ok || !res.source) {
    return { ok: false, error: res.error ?? "Source invalide" };
  }
  const orgId = await getActiveOrgId();
  await upsertLinkSources(orgId, [res.source]);
  revalidatePath(RESERVOIR_PATH);
  return { ok: true };
}