/**
 * AI Software Factory — OpenSearch content-index reindex/rollover (TASK-0099 §5).
 *
 * Rolls the `factory-content` alias to a NEW index (`factory-content-<to>`,
 * default: the next version after the current alias target) with a fresh
 * mapping (knn_vector `vector` — faiss/l2 on OpenSearch 2.15, see below —
 * dimension/m/efConstruction tunable),
 * backfills it from the `search_documents` Postgres mirror in 500-doc
 * `_bulk` batches, asserts the count equality, then ATOMICALLY swaps the
 * alias (`POST /_aliases`) and optionally deletes the old index.
 *
 * Why: `dense_vector` dimensions are IMMUTABLE per index — a model change
 * (1536 → 3072, …) cannot re-map in place; it must roll a new index forward
 * through the alias. This script is the runbook for that (and for the v1 →
 * v2 hybrid-foundation mapping change, TASK-0099).
 *
 * Self-contained by design (recorded decision): the root has no
 * `@factory/*` hoist and the repo allows no new dependencies, so the
 * mapping is re-declared below (mirror of packages/search mapping.ts —
 * keep the two in sync) and OpenSearch is driven with raw `fetch`.
 * Postgres uses the repo's existing `@prisma/client` (no new dep).
 *
 *   pnpm search:reindex --i-know-this-swaps-the-alias [--to v2]
 *       [--dimension 1536] [--dry-run] [--strict-equal] [--cleanup-v1]
 *
 * Flags:
 *   --i-know-this-swaps-the-alias  required (unless --dry-run): this command
 *                                  re-points the LIVE search alias.
 *   --to <ver>        target version suffix (default: current+1, or v1 fresh).
 *   --dimension <n>   dense_vector dimension (default 1536).
 *   --dry-run         read-only: print the plan, create/write nothing.
 *   --strict-equal    fail (exit 1) if target count != mirror count
 *                     (default: warn only).
 *   --cleanup-v1      delete the OLD index after a successful swap.
 *
 * Env: SEARCH_URL (http://localhost:9200), SEARCH_USERNAME (admin),
 *      SEARCH_PASSWORD (changeme), DATABASE_URL
 *      (postgresql://factory:factory@localhost:5432/factory),
 *      SEARCH_KNN_M (16), SEARCH_KNN_EF_CONSTRUCTION (100).
 *
 * Backfill source: the `search_documents` MIRROR (whole docs only — chunk
 * docs have no mirror rows). Phase-12 loose end: backfill from the SOURCE
 * tables (url/grants/createdByName) instead of the mirror; until then a
 * fresh v2 index contains whole docs only, and chunk docs reappear as the
 * worker re-indexes pages/records (or on the next content event).
 */
import { PrismaClient } from "@prisma/client";

// ---------------------------------------------------------------------------
// Flags + env
// ---------------------------------------------------------------------------

const ALIAS = "factory-content";
const BULK_BATCH = 500; // _bulk line batch (mirrors the worker's BATCH)

interface Flags {
  guard: boolean;
  to: string | null;
  dimension: number;
  dryRun: boolean;
  strictEqual: boolean;
  cleanupV1: boolean;
}

function fail(msg: string): never {
  console.error(`✖ ${msg}`);
  process.exit(1);
}

function parseFlags(argv: string[]): Flags {
  const f: Flags = {
    guard: false,
    to: null,
    dimension: 1536,
    dryRun: false,
    strictEqual: false,
    cleanupV1: false,
  };
  for (let i = 0; i < argv.length; i++) {
    const a = argv[i]!;
    switch (a) {
      case "--i-know-this-swaps-the-alias":
        f.guard = true;
        break;
      case "--to": {
        const v = argv[++i];
        if (!v || !/^v[0-9]+$/.test(v))
          fail(`--to expects a version like "v2", got: ${v ?? "<none>"}`);
        f.to = v;
        break;
      }
      case "--dimension": {
        const v = argv[++i];
        const n = v ? Number.parseInt(v, 10) : NaN;
        if (!v || !Number.isInteger(n) || n <= 0)
          fail(`--dimension expects a positive integer, got: ${v ?? "<none>"}`);
        f.dimension = n;
        break;
      }
      case "--dry-run":
        f.dryRun = true;
        break;
      case "--strict-equal":
        f.strictEqual = true;
        break;
      case "--cleanup-v1":
        f.cleanupV1 = true;
        break;
      default:
        fail(`unknown flag: ${a} (see header of scripts/opensearch-reindex.ts)`);
    }
  }
  return f;
}

// ---------------------------------------------------------------------------
// OpenSearch (raw fetch — no SDK dependency)
// ---------------------------------------------------------------------------

interface Os {
  base: string;
  user: string;
  pass: string;
}

async function os(osCfg: Os, path: string, init?: RequestInit): Promise<unknown> {
  const headers: Record<string, string> = {
    "content-type": "application/json",
    authorization: "Basic " + Buffer.from(`${osCfg.user}:${osCfg.pass}`).toString("base64"),
  };
  const res = await fetch(osCfg.base.replace(/\/+$/, "") + path, { ...init, headers });
  const text = await res.text();
  if (!res.ok) {
    throw new Error(
      `OpenSearch ${init?.method ?? "GET"} ${path} → HTTP ${res.status}: ${text.slice(0, 500)}`,
    );
  }
  return text ? (JSON.parse(text) as unknown) : null;
}

async function osCurrentAliasTarget(osCfg: Os): Promise<string | null> {
  const res = await os(osCfg, `/_alias/${ALIAS}`);
  const idxs = Object.keys((res as Record<string, unknown>) ?? {});
  return idxs.length > 0 ? idxs[0]! : null;
}

interface IndexInfo {
  dimension: number | null;
  count: number | null;
}

async function osIndexInfo(osCfg: Os, index: string): Promise<IndexInfo> {
  const mapping = (await os(osCfg, `/${index}/_mapping`)) as Record<
    string,
    { mappings?: Record<string, unknown> }
  >;
  const m = mapping[index]?.mappings as
    { properties?: { vector?: { dimension?: number } } } | undefined;
  const dimension = m?.properties?.vector?.dimension ?? null;
  const countRes = (await os(osCfg, `/${index}/_count`)) as { count: number };
  return { dimension, count: countRes.count ?? 0 };
}

// ---------------------------------------------------------------------------
// Mapping (MIRROR of packages/search/src/mapping.ts contentMapping — keep in
// sync; re-declared here because the script must stay dependency-free).
// ---------------------------------------------------------------------------

function contentMapping(
  dimension: number,
  m: number,
  efConstruction: number,
): Record<string, unknown> {
  return {
    settings: {
      index: {
        refresh_interval: "1s", // lag SLA: keep 1 s (search §1.4)
        number_of_replicas: 1,
        knn: true,
        "knn.algo_param.ef_search": 100,
      },
    },
    mappings: {
      properties: {
        docId: { type: "keyword" },
        // MIRRORS packages/search contentMapping: plain keyword — the
        // OpenSearch in docker/CI (2.15) rejects a mapping-level `enum` on
        // keyword mappers; the allowed set is enforced at the application
        // layer (the client only indexes known SearchDocType values).
        type: { type: "keyword" },
        id: { type: "keyword" },
        workspaceId: { type: "keyword" },
        teamId: { type: "keyword" },
        title: {
          type: "text",
          fields: { keyword: { type: "keyword", ignore_above: 512 } },
        },
        body: { type: "text" },
        chunkId: { type: "keyword" },
        chunkIndex: { type: "integer" },
        parentDocId: { type: "keyword" },
        vectorModel: { type: "keyword" },
        vector: {
          // MIRRORS the package (packages/search/src/mapping.ts — keep in
          // sync): OpenSearch 2.15 has the 1.x `knn_vector` + `method` HNSW
          // form — NO `dense_vector` handler. TASK-0125 (verified live):
          // engine = faiss (nmslib hard-rejects the kNN pre-filter that the
          // hybrid leg needs) and space_type = l2 (FAISS 2.15 rejects
          // cosinesimil/inner_product; l2 ranks identically to cosine for
          // L2-normalized real embeddings).
          type: "knn_vector",
          dimension,
          method: {
            name: "hnsw",
            space_type: "l2",
            engine: "faiss",
            parameters: { m, ef_construction: efConstruction },
          },
        },
        grants: {
          properties: {
            public: { type: "boolean" },
            workspace: { type: "boolean" },
            user: { type: "boolean" },
            teams: { type: "keyword" },
            users: { type: "keyword" },
          },
        },
        archived: { type: "boolean" },
        pageId: { type: "keyword" },
        blockId: { type: "keyword" },
        databaseId: { type: "keyword" },
        recordId: { type: "keyword" },
        createdBy: { type: "keyword" },
        createdByName: { type: "keyword" },
        icon: { type: "keyword", index: false },
        url: { type: "keyword", index: false },
        size: { type: "long" },
        mimeType: { type: "keyword" },
        createdAt: { type: "date" },
        updatedAt: { type: "date" },
      },
    },
  };
}

// ---------------------------------------------------------------------------
// Mirror → index doc (whole docs only; chunk docs have no mirror rows)
// ---------------------------------------------------------------------------

interface MirrorRow {
  id: string;
  workspaceId: string;
  type: string;
  refId: string;
  title: string;
  body: string | null;
  permissions: unknown;
  metadata: unknown;
  indexedAt: Date;
}

/** The index `_id` / `docId`: `<type>:<mirror refId>` (user refId is already
 *  workspace-scoped: `user:<ws>:<uid>`). */
function docIdFor(row: MirrorRow): string {
  return `${row.type}:${row.refId}`;
}

/** Rebuild the wire-form index doc from a mirror row (metadata = the
 *  deep-link facet snapshot the worker stored in upsertMirror). */
function mirrorToDoc(row: MirrorRow): Record<string, unknown> {
  const meta = (row.metadata ?? {}) as Record<string, unknown>;
  const str = (k: string): string | undefined =>
    typeof meta[k] === "string" ? (meta[k] as string) : undefined;
  const doc: Record<string, unknown> = {
    docId: docIdFor(row),
    type: row.type,
    // the doc's `id` is the plain row id (user docs: the bare uid)
    id: row.type === "user" ? (row.refId.split(":").pop() ?? row.refId) : row.refId,
    workspaceId: row.workspaceId,
    title: row.title,
    body: row.body ?? "",
    grants: (row.permissions ?? {}) as Record<string, unknown>,
    archived: typeof meta.archived === "boolean" ? meta.archived : false,
    url: str("url") ?? "",
    createdAt: str("createdAt") ?? row.indexedAt.toISOString(),
    updatedAt: str("updatedAt") ?? row.indexedAt.toISOString(),
  };
  // facet fields — only present when the mirror recorded them
  for (const k of [
    "teamId",
    "pageId",
    "blockId",
    "databaseId",
    "recordId",
    "createdBy",
    "icon",
    "size",
    "mimeType",
  ] as const) {
    const v = meta[k];
    if (v !== null && v !== undefined) doc[k] = v;
  }
  return doc;
}

// ---------------------------------------------------------------------------
// Main
// ---------------------------------------------------------------------------

async function main(): Promise<void> {
  const flags = parseFlags(process.argv.slice(2));
  if (!flags.guard && !flags.dryRun) {
    fail(
      "refusing to reindex without the guard flag — this command re-points the LIVE " +
        `"${ALIAS}" alias.\n  Re-run with --i-know-this-swaps-the-alias (or --dry-run to inspect only).`,
    );
  }

  const osCfg: Os = {
    base: process.env.SEARCH_URL ?? "http://localhost:9200",
    user: process.env.SEARCH_USERNAME ?? "admin",
    pass: process.env.SEARCH_PASSWORD ?? "changeme",
  };
  const dimension = flags.dimension;
  const knnM = Number.parseInt(process.env.SEARCH_KNN_M ?? "16", 10);
  const efConstruction = Number.parseInt(process.env.SEARCH_KNN_EF_CONSTRUCTION ?? "100", 10);

  console.log(`▸ reindex plan: alias="${ALIAS}" target-dim=${dimension} dry-run=${flags.dryRun}`);

  // 1. Resolve the current alias target + live state.
  const current = await osCurrentAliasTarget(osCfg).catch((e: Error) => {
    if (/404/.test(e.message)) return null;
    throw e;
  });
  console.log(`  current alias target: ${current ?? "(none — fresh cluster)"}`);

  // 2. Target version: explicit --to, else next after current (v1 → v2),
  //    else v1 for a fresh cluster.
  let target: string;
  if (flags.to) {
    target = `${ALIAS}-${flags.to}`;
  } else if (current) {
    const m = /-v([0-9]+)$/.exec(current);
    if (!m) fail(`cannot derive the next version from "${current}" — pass --to explicitly`);
    target = `${ALIAS}-v${Number.parseInt(m[1]!, 10) + 1}`;
  } else {
    target = `${ALIAS}-v1`;
  }
  console.log(`  target index: ${target}`);

  // 3. Read the mirror count (the backfill source of truth).
  const prisma = new PrismaClient({
    datasources: {
      db: {
        url: process.env.DATABASE_URL ?? "postgresql://factory:factory@localhost:5432/factory",
      },
    },
  });
  let mirrorCount = 0;
  try {
    mirrorCount = await prisma.searchDocument.count();
  } finally {
    if (flags.dryRun) await prisma.$disconnect();
  }
  console.log(`  mirror rows (search_documents, whole docs): ${mirrorCount}`);

  // 4. Existing target index? (idempotent re-run: same --to re-bakes + re-swaps)
  let targetInfo: IndexInfo | null = null;
  try {
    targetInfo = await osIndexInfo(osCfg, target);
  } catch (e) {
    if (!/404/.test((e as Error).message)) throw e;
  }
  if (targetInfo) {
    console.log(`  target already exists: dim=${targetInfo.dimension} docs=${targetInfo.count}`);
    if (targetInfo.dimension !== null && targetInfo.dimension !== dimension) {
      fail(
        `target "${target}" exists with dimension ${targetInfo.dimension} ≠ requested ${dimension} — ` +
          `dense_vector dimensions are immutable. Pick another --to (or delete the index).`,
      );
    }
  }

  if (flags.dryRun) {
    console.log(
      `\n[dry-run] plan only — would:\n` +
        `  1. ${targetInfo ? "reuse" : "create"} ${target} (dim=${dimension}, m=${knnM}, ef_construction=${efConstruction})\n` +
        `  2. bulk-index ${mirrorCount} mirror docs in ${BULK_BATCH}-doc batches\n` +
        `  3. refresh + count check (strict=${flags.strictEqual})\n` +
        `  4. atomically move "${ALIAS}": ${current ?? "(none)"} → ${target}\n` +
        `  ${flags.cleanupV1 && current ? `5. delete old index ${current}\n` : ""}` +
        `No writes were made.`,
    );
    return;
  }

  // 5. Create the target index (idempotent — only when missing).
  if (!targetInfo) {
    await os(osCfg, `/${target}`, {
      method: "PUT",
      body: JSON.stringify(contentMapping(dimension, knnM, efConstruction)),
    });
    console.log(
      `  created ${target} (dim=${dimension}, m=${knnM}, ef_construction=${efConstruction})`,
    );
  }

  // 6. Backfill from the mirror (cursor-paged reads, 500-doc _bulk writes).
  let written = 0;
  let cursor: string | null = null;
  const PAGE = 1000;
  for (;;) {
    // exactOptionalPropertyTypes: `where?: X` does not accept `X | undefined`,
    // so the cursor-paged read is a plain conditional (no `| undefined`).
    const rows: MirrorRow[] = cursor
      ? ((await prisma.searchDocument.findMany({
          where: { id: { gt: cursor } },
          orderBy: { id: "asc" },
          take: PAGE,
        })) as MirrorRow[])
      : ((await prisma.searchDocument.findMany({
          orderBy: { id: "asc" },
          take: PAGE,
        })) as MirrorRow[]);
    if (rows.length === 0) break;
    const line: string[] = [];
    for (const row of rows) {
      line.push(`{"index":{"_index":"${target}","_id":"${docIdFor(row)}"}}`);
      line.push(JSON.stringify(mirrorToDoc(row)));
    }
    const docCount = line.length / 2;
    for (let i = 0; i < line.length; i += BULK_BATCH * 2) {
      const batchDocs = Math.min(BULK_BATCH, docCount - i / 2);
      const body = line.slice(i, i + BULK_BATCH * 2).join("\n") + "\n";
      const res = (await os(osCfg, `/${target}/_bulk`, { method: "POST", body })) as {
        errors?: boolean;
        items?: { index?: { error?: { type?: string; reason?: string }; status?: number } }[];
      };
      if (res.errors) {
        const bad = res.items?.find((it) => it.index?.error) ?? null;
        throw new Error(`_bulk errors on ${target}: ${JSON.stringify(bad?.index?.error ?? {})}`);
      }
      written += batchDocs;
    }
    cursor = rows[rows.length - 1]!.id;
    if (rows.length < PAGE) break;
  }
  await prisma.$disconnect();
  console.log(`  bulk-indexed ${written} docs (mirror rows read: ${written})`);

  // 7. Refresh + count check (idempotent docs: the target holds exactly the
  //    mirror's whole docs — chunk docs have no mirror rows, see header).
  await os(osCfg, `/${target}/_refresh`, { method: "POST" });
  const after = await osIndexInfo(osCfg, target);
  console.log(`  target count after backfill: ${after.count} (mirror: ${mirrorCount})`);
  const equal = after.count === mirrorCount;
  if (!equal) {
    const msg = `count mismatch: target ${after.count} ≠ mirror ${mirrorCount}`;
    if (flags.strictEqual) fail(msg);
    console.warn(`  ⚠ ${msg} (continue — pass --strict-equal to hard-fail)`);
  } else {
    console.log(`  ✓ count equality holds (${mirrorCount})`);
  }

  // 8. Atomic alias swap (one /_aliases call — never remove-then-add).
  const actions: Record<string, unknown>[] = [{ add: { index: target, alias: ALIAS } }];
  if (current) actions.unshift({ remove: { index: current, alias: ALIAS } });
  await os(osCfg, "/_aliases", { method: "POST", body: JSON.stringify({ actions }) });
  console.log(`  ✓ alias "${ALIAS}" now → ${target}${current ? ` (was ${current})` : ""}`);

  // 9. Optional cleanup of the old index.
  if (flags.cleanupV1 && current) {
    await os(osCfg, `/${current}`, { method: "DELETE" });
    console.log(`  ✓ deleted old index ${current}`);
  } else if (current) {
    console.log(`  (kept ${current} — pass --cleanup-v1 to delete it after verifying)`);
  }

  console.log("✔ reindex complete");
}

main().catch((e: Error) => fail(e.message));