/**
 * Backfills the denormalised plan_id / material_* columns on `recommendation`
 * from the plan_recommendations and recommendation_materials join tables in a
 * SINGLE keyset-paginated pass, then (optionally) builds the supporting indexes
 * and validates the foreign keys — all ONLINE and resumable.
 *
 * Why this lives outside drizzle:
 *   drizzle-kit migrate wraps every pending migration in ONE transaction, so it
 *   cannot COMMIT between batches, cannot run CREATE INDEX CONCURRENTLY, and
 *   holds the ADD COLUMN's AccessExclusiveLock for the whole run. This script
 *   instead commits each batch, keeping locks tiny, WAL/bloat bounded, EBS IO
 *   burst balance healthy, and progress visible + resumable.
 *   See docs/adr/0001-data-backfills-outside-drizzle.md
 *
 * Run AFTER `npm run migration:migrate` has applied 0222/0224 (the column adds):
 *   npm run backfill:recommendation-denormalization
 *
 * ── Making it fast (see docs/runbooks/recommendation-denormalization-backfill.md)
 * The dominant cost is index write-amplification: `recommendation` is packed
 * (fillfactor 100), so each UPDATE can't do a HOT update and must insert into
 * every index on the table for every one of ~21m rows. To go roughly an order
 * of magnitude faster in a low-usage window:
 *   1. DROP the 3 secondary indexes  (src/app/db/sql/drop-recommendation-backfill-indexes.sql)
 *   2. run this script                (BACKFILL_FINALIZE=false for a partial/interrupted window)
 *   3. REBUILD them after             (src/app/db/sql/rebuild-recommendation-backfill-indexes.sql)
 * The backfill is resumable: stop any time, rebuild the indexes, and the app is
 * back to today's behaviour (the half-filled columns are NULL-tolerant). Re-run
 * to continue — the IS NULL guards never rewrite a row that's already done.
 *
 * This pass sets plan_id AND the four material_* columns in ONE write per row
 * (the previous version did two full passes, rewriting every row twice).
 *
 * Safe to re-run: every step is idempotent (IS NULL guards, IF NOT EXISTS,
 * VALIDATE on an already-valid constraint is a no-op).
 *
 * Tunables via env:
 *   BACKFILL_BATCH_SIZE            rows scanned per committed batch (default 100000)
 *   BACKFILL_SLEEP_MS             pause between batches, eases IO pressure (default 0)
 *   BACKFILL_SYNC_COMMIT          'on' keeps synchronous_commit; default 'off' (faster
 *                                 batch commits — safe here because the run is resumable)
 *   BACKFILL_MAINTENANCE_WORK_MEM session maintenance_work_mem for index builds (default '1GB')
 *   BACKFILL_WORK_MEM             session work_mem for the batch joins (default '128MB')
 *   BACKFILL_INDEX_CONCURRENTLY   'true' builds the new indexes CONCURRENTLY (default plain
 *                                 CREATE INDEX — faster, but briefly blocks writes)
 *   BACKFILL_FINALIZE             'false' skips the index build + FK validate; use this for
 *                                 partial/interrupted windows, finalize on the last one
 */
import dotenv from "dotenv";
import { Pool, type PoolClient } from "pg";

dotenv.config({ path: ".env.local" });

const BATCH_SIZE = Number(process.env.BACKFILL_BATCH_SIZE ?? 100_000);
const SLEEP_MS = Number(process.env.BACKFILL_SLEEP_MS ?? 0);
const SYNC_COMMIT = process.env.BACKFILL_SYNC_COMMIT === "on" ? "on" : "off";
const MAINTENANCE_WORK_MEM = process.env.BACKFILL_MAINTENANCE_WORK_MEM ?? "1GB";
const WORK_MEM = process.env.BACKFILL_WORK_MEM ?? "128MB";
const INDEX_CONCURRENTLY = process.env.BACKFILL_INDEX_CONCURRENTLY === "true";
const FINALIZE = process.env.BACKFILL_FINALIZE !== "false";

// Range + phase controls, for running several workers over disjoint id ranges in
// parallel (near-linear speed-up — the ranges never contend for the same rows).
const START_ID = process.env.BACKFILL_START_ID ?? "0";
const END_ID = process.env.BACKFILL_END_ID ?? null; // inclusive upper bound; null = unbounded
const SKIP_CARDINALITY = process.env.BACKFILL_SKIP_CARDINALITY === "true";
const ONLY_FINALIZE = process.env.BACKFILL_ONLY_FINALIZE === "true";

const pool = new Pool({
  host: process.env.DB_HOST,
  port: Number(process.env.DB_PORT),
  user: process.env.DB_USERNAME,
  password: process.env.DB_PASSWORD,
  database: process.env.DB_NAME,
});

const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));

/** Abort before touching data if the 1:1 assumption is violated. */
async function assertSingleCardinality(
  client: PoolClient,
  joinTable: string,
): Promise<void> {
  const { rows } = await client.query(
    `SELECT count(*)::int AS violations
       FROM (
         SELECT recommendation_id
           FROM ${joinTable}
          GROUP BY recommendation_id
         HAVING count(*) > 1
       ) dupes`,
  );
  const violations = rows[0].violations as number;
  if (violations > 0) {
    throw new Error(
      `Aborting: ${violations} recommendation(s) map to multiple ${joinTable} ` +
        `rows. The backfill cannot pick deterministically. Resolve cardinality first.`,
    );
  }
}

/**
 * Keyset-paginate the whole `recommendation` table by id and, for each batch,
 * set plan_id and the four material_* columns from the join tables in a single
 * UPDATE. Each batch is its own autocommitted statement (no surrounding BEGIN),
 * so locks are held only for the rows in that batch.
 *
 * We scan by id (not `WHERE col IS NULL LIMIT n`) so rows with no match — or a
 * legitimately NULL material_id — don't get re-selected forever; they're left as
 * they are and we move past them.
 *
 * The guard only touches a row when there is real work to do (a NULL column that
 * has a matching source row), and COALESCE never overwrites an already-set value.
 * So a row is written at most once, and a re-run after interruption skips every
 * row already done — the pass is idempotent and resumable.
 *
 * The single-cardinality assertion above guarantees the LEFT JOINs cannot fan a
 * row out into multiple update targets.
 */
const fmtDuration = (secs: number): string => {
  const s = Math.max(0, Math.round(secs));
  const h = Math.floor(s / 3600);
  const m = Math.floor((s % 3600) / 60);
  return h > 0 ? `${h}h${m}m` : m > 0 ? `${m}m${s % 60}s` : `${s}s`;
};

async function backfillDenormalized(client: PoolClient): Promise<void> {
  let lastId = START_ID;
  let scannedTotal = 0;
  let updatedTotal = 0;
  const startedAt = Date.now();

  // Upper bound of the id keyset, for percent-complete + ETA. When END_ID is set
  // (a parallel worker), use it; otherwise ask the table (fast, PK).
  let maxId: number;
  if (END_ID != null) {
    maxId = Number(END_ID);
  } else {
    const { rows: maxRows } = await client.query(
      `SELECT max(id) AS max_id FROM recommendation`,
    );
    maxId = Number(maxRows[0].max_id ?? 0);
  }
  console.log(
    `Backfilling id ${Number(START_ID).toLocaleString()} → ${maxId.toLocaleString()}...`,
  );

  for (;;) {
    const { rows } = await client.query(
      `WITH batch AS (
         SELECT id FROM recommendation
          WHERE id > $1
            AND ($3::bigint IS NULL OR id <= $3::bigint)
          ORDER BY id
          LIMIT $2
       ),
       upd AS (
         UPDATE recommendation r
            SET plan_id                = COALESCE(r.plan_id, pr.plan_id),
                material_id            = COALESCE(r.material_id, rm.material_id),
                material_quantity      = COALESCE(r.material_quantity, rm.quantity),
                material_quantity_unit = COALESCE(r.material_quantity_unit, rm.quantity_unit),
                material_depth         = COALESCE(r.material_depth, rm.depth)
           FROM batch b
           LEFT JOIN plan_recommendations pr     ON pr.recommendation_id = b.id
           LEFT JOIN recommendation_materials rm ON rm.recommendation_id = b.id
          WHERE r.id = b.id
            AND ( (r.plan_id     IS NULL AND pr.plan_id     IS NOT NULL)
               OR (r.material_id IS NULL AND rm.material_id IS NOT NULL) )
         RETURNING r.id
       )
       SELECT (SELECT max(id) FROM batch)  AS max_id,
              (SELECT count(*) FROM batch) AS scanned,
              (SELECT count(*) FROM upd)   AS updated`,
      [lastId, BATCH_SIZE, END_ID],
    );
    const { max_id, scanned, updated } = rows[0] as {
      max_id: string | null;
      scanned: number;
      updated: number;
    };

    if (!scanned || max_id === null) break;

    scannedTotal += Number(scanned);
    updatedTotal += Number(updated);
    lastId = max_id;

    const elapsed = (Date.now() - startedAt) / 1000;
    const rate = Math.round(scannedTotal / elapsed);
    const from = Number(START_ID);
    const span = maxId - from;
    const pct = span > 0 ? Math.min(100, ((Number(lastId) - from) / span) * 100) : 0;
    const eta = pct > 0 ? fmtDuration((elapsed / pct) * (100 - pct)) : "?";
    console.log(
      `${pct.toFixed(1)}% (id ${lastId}) — scanned ${scannedTotal.toLocaleString()}, ` +
        `updated ${updatedTotal.toLocaleString()} — ${rate.toLocaleString()} rows/s, ETA ${eta}`,
    );

    if (SLEEP_MS > 0) await sleep(SLEEP_MS);
  }

  console.log(
    `Backfill pass done: scanned ${scannedTotal.toLocaleString()}, ` +
      `updated ${updatedTotal.toLocaleString()} in ` +
      `${Math.round((Date.now() - startedAt) / 1000)}s`,
  );
}

async function main(): Promise<void> {
  const client = await pool.connect();
  try {
    console.log(
      `Config: batch=${BATCH_SIZE.toLocaleString()} rows, sleep=${SLEEP_MS}ms, ` +
        `synchronous_commit=${SYNC_COMMIT}, work_mem=${WORK_MEM}, ` +
        `maintenance_work_mem=${MAINTENANCE_WORK_MEM}, finalize=${FINALIZE}, ` +
        `range=(${Number(START_ID).toLocaleString()}, ${END_ID ?? "∞"}]` +
        `${ONLY_FINALIZE ? ", ONLY_FINALIZE" : ""}\n`,
    );

    // Session tuning. synchronous_commit=off is safe here: the backfill is
    // idempotent and resumable, so a crash that loses the last few batch commits
    // just means the next run continues from there. work_mem feeds the batch
    // joins; maintenance_work_mem feeds the CREATE INDEX in the finalize step.
    await client.query(`SET synchronous_commit = ${SYNC_COMMIT}`);
    await client.query(`SET work_mem = '${WORK_MEM}'`);
    await client.query(`SET maintenance_work_mem = '${MAINTENANCE_WORK_MEM}'`);

    if (ONLY_FINALIZE) {
      // Parallel workflow: the data was written by range workers; this run just
      // builds the indexes + validates the FKs. No cardinality check, no pass.
      console.log("ONLY_FINALIZE: skipping cardinality check + backfill pass.");
    } else {
      // 1. Guard the 1:1 assumption before mutating anything. Skippable on
      //    secondary parallel workers (one worker having checked is enough).
      if (SKIP_CARDINALITY) {
        console.log("Skipping cardinality check (BACKFILL_SKIP_CARDINALITY=true).");
      } else {
        console.log("Checking cardinality...");
        await assertSingleCardinality(client, "plan_recommendations");
        await assertSingleCardinality(client, "recommendation_materials");
      }

      // 2. Single combined pass: plan_id + material_* in one write per row.
      console.log("Backfilling plan_id + material_* (single pass)...");
      await backfillDenormalized(client);
    }

    if (FINALIZE || ONLY_FINALIZE) {
      // 3. Build the two new indexes. Plain CREATE INDEX is faster in a quiet
      //    window; BACKFILL_INDEX_CONCURRENTLY=true avoids the brief write lock.
      //    Autocommitted (no surrounding BEGIN), which CONCURRENTLY requires.
      const cc = INDEX_CONCURRENTLY ? "CONCURRENTLY " : "";
      console.log(
        `Creating indexes (${INDEX_CONCURRENTLY ? "concurrently" : "plain"})...`,
      );
      await client.query(
        `CREATE INDEX ${cc}IF NOT EXISTS idx_recommendation_plan_id
           ON recommendation USING btree (plan_id)`,
      );
      await client.query(
        `CREATE INDEX ${cc}IF NOT EXISTS idx_recommendation_material_id
           ON recommendation USING btree (material_id)`,
      );

      // 4. Validate the FKs online (ShareUpdateExclusiveLock — allows reads/writes).
      console.log("Validating foreign keys...");
      await client.query(
        `ALTER TABLE recommendation
           VALIDATE CONSTRAINT recommendation_plan_id_plan_id_fk`,
      );
      await client.query(
        `ALTER TABLE recommendation
           VALIDATE CONSTRAINT recommendation_material_id_material_id_fk`,
      );
    } else {
      console.log(
        "Skipping index build + FK validate (BACKFILL_FINALIZE=false). " +
          "Run once more with finalize enabled in your final window.",
      );
    }

    // 5. Report any rows still unlinked. material_id NULL is expected (not every
    //    recommendation has a material); plan_id NULL should trend toward ~0.
    const { rows } = await client.query(
      `SELECT count(*) FILTER (WHERE plan_id IS NULL)     AS plan_null,
              count(*) FILTER (WHERE material_id IS NULL) AS material_null
         FROM recommendation`,
    );
    console.log(
      `\nRemaining NULLs — plan_id: ${Number(rows[0].plan_null).toLocaleString()}, ` +
        `material_id: ${Number(rows[0].material_null).toLocaleString()}`,
    );
    console.log(
      FINALIZE ? "Backfill complete." : "Backfill pass complete (not finalized).",
    );
  } finally {
    client.release();
    await pool.end();
  }
}

main().catch((err) => {
  console.error("Backfill failed:", err);
  process.exit(1);
});
