Jyotisha/frontend/scripts/migrate-supabase-production.mjs

import { createHash } from "node:crypto";
import { dirname, resolve } from "node:path";
import { fileURLToPath, pathToFileURL } from "node:url";
import { Pool, types as pgTypes } from "pg";

import { runMigrations } from "./db-migrate.mjs";
import { normalizeSupabaseUsers } from "./import-supabase-auth-users.mjs";

export class SafeProductionMigrationError extends Error {}

const PRODUCTION_OWNER_USER_ID = "b8907d0c-6ed0-4270-b866-7e83bb4a1b26";
const PRODUCTION_OWNER_EMAIL = "luna@copse.life";
const IDENTITY_STATE_TABLES = [
  "users",
  "accounts",
  "sessions",
  "verifications",
  "otp_rate_limits",
  "two_factors",
];

// Preserve PostgreSQL microseconds. JavaScript Date truncates timestamps to milliseconds.
pgTypes.setTypeParser(1114, (value) => value);
pgTypes.setTypeParser(1184, (value) => value);

const ALLOWED_TARGET_ROWS = new Set([
  "public.admin_permissions",
  "public.admin_role_permissions",
  "public.admin_roles",
  "public.billing_products",
  "public.feature_flags",
  "public.notification_templates",
  "public.product_entitlements",
]);

const SEED_TABLES = new Map([
  ["admin_permissions", ["permission_key"]],
  ["admin_roles", ["code"]],
  ["billing_products", ["code", "version"]],
  ["feature_flags", ["flag_key", "version"]],
  ["notification_templates", ["template_key", "channel", "version"]],
]);

const SEED_RELATIONS = new Map([
  ["admin_role_permissions", ["role_id", "permission_id"]],
  ["product_entitlements", ["product_id", "feature_key"]],
]);

const CIPHERTEXT_COLUMNS = new Map([
  ["epay_settings", new Set(["encrypted_key"])],
  ["model_providers", new Set(["encrypted_api_key"])],
]);

const uuidPattern = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i;

function quoted(identifier) {
  return `"${String(identifier).replaceAll('"', '""')}"`;
}

function qualified(schema, table) {
  return `${quoted(schema)}.${quoted(table)}`;
}

function requiredUrl(env, name) {
  const value = env[name]?.trim();
  if (!value) throw new SafeProductionMigrationError(`${name} is required`);
  if (!/^postgres(?:ql)?:\/\//.test(value)) {
    throw new SafeProductionMigrationError(`${name} must be a PostgreSQL URL`);
  }
  return value;
}

export function readConfiguration(env) {
  const sourceUrl = requiredUrl(env, "SUPABASE_SOURCE_DATABASE_URL");
  const targetUrl = requiredUrl(env, "PRODUCTION_TARGET_DATABASE_URL");
  if (sourceUrl === targetUrl) {
    throw new SafeProductionMigrationError("source and target databases must differ");
  }
  const ownerUserId = env.PRODUCTION_OWNER_USER_ID?.trim().toLowerCase();
  if (!ownerUserId || !uuidPattern.test(ownerUserId)) {
    throw new SafeProductionMigrationError("PRODUCTION_OWNER_USER_ID must be a UUID");
  }
  if (ownerUserId !== PRODUCTION_OWNER_USER_ID) {
    throw new SafeProductionMigrationError("PRODUCTION_OWNER_USER_ID does not match the approved production Owner");
  }
  const ownerEmail = env.PRODUCTION_OWNER_EMAIL?.trim().toLowerCase();
  if (!ownerEmail || !ownerEmail.includes("@")) {
    throw new SafeProductionMigrationError("PRODUCTION_OWNER_EMAIL must be an email address");
  }
  if (ownerEmail !== PRODUCTION_OWNER_EMAIL) {
    throw new SafeProductionMigrationError("PRODUCTION_OWNER_EMAIL does not match the approved production Owner");
  }
  const ciphertextMode = env.PRODUCTION_CIPHERTEXT_MODE?.trim();
  if (!new Set(["preserve", "exclude"]).has(ciphertextMode)) {
    throw new SafeProductionMigrationError(
      "PRODUCTION_CIPHERTEXT_MODE must be preserve or exclude",
    );
  }
  if (
    ciphertextMode === "preserve" &&
    env.PRODUCTION_CIPHERTEXT_KEYS_CONFIRMED !== "true"
  ) {
    throw new SafeProductionMigrationError(
      "preserving ciphertext requires confirmed production encryption keys",
    );
  }
  return { sourceUrl, targetUrl, ownerUserId, ownerEmail, ciphertextMode };
}

export function parseMode(arguments_) {
  const modes = arguments_.filter((argument) =>
    ["--preflight", "--apply", "--verify"].includes(argument),
  );
  if (modes.length !== 1 || modes.length !== arguments_.length) {
    throw new SafeProductionMigrationError(
      "choose exactly one of --preflight, --apply, or --verify",
    );
  }
  return modes[0].slice(2);
}

function asArray(value) {
  if (Array.isArray(value)) return value;
  if (typeof value !== "string") return [];
  return value.replace(/^\{/, "").replace(/\}$/, "").split(",").filter(Boolean);
}

export async function readSchema(client, schema) {
  const columnsResult = await client.query(
      `
        select c.table_name, c.column_name, c.is_nullable = 'YES' as nullable,
               c.column_default, c.is_generated <> 'NEVER' as generated,
               c.is_identity = 'YES' as identity, c.identity_generation,
               c.data_type, c.udt_name,
               c.ordinal_position
        from information_schema.columns c
        join information_schema.tables t
          on t.table_schema = c.table_schema and t.table_name = c.table_name
        where c.table_schema = $1 and t.table_type = 'BASE TABLE'
        order by c.table_name, c.ordinal_position
      `,
      [schema],
    );
  const primaryKeysResult = await client.query(
      `
        select kcu.table_name,
               array_agg(kcu.column_name order by kcu.ordinal_position) as columns
        from information_schema.table_constraints tc
        join information_schema.key_column_usage kcu
          on kcu.constraint_schema = tc.constraint_schema
         and kcu.constraint_name = tc.constraint_name
         and kcu.table_name = tc.table_name
        where tc.table_schema = $1 and tc.constraint_type = 'PRIMARY KEY'
        group by kcu.table_name
      `,
      [schema],
    );
  const foreignKeysResult = await client.query(
      `
        select n.nspname as schema_name, r.relname as table_name,
               rn.nspname as ref_schema, rr.relname as ref_table,
               array(
                 select a.attname
                 from unnest(c.conkey) with ordinality as key(attnum, ord)
                 join pg_attribute a on a.attrelid = c.conrelid and a.attnum = key.attnum
                 order by key.ord
               ) as columns,
               array(
                 select a.attname
                 from unnest(c.confkey) with ordinality as key(attnum, ord)
                 join pg_attribute a on a.attrelid = c.confrelid and a.attnum = key.attnum
                 order by key.ord
               ) as ref_columns
        from pg_constraint c
        join pg_class r on r.oid = c.conrelid
        join pg_namespace n on n.oid = r.relnamespace
        join pg_class rr on rr.oid = c.confrelid
        join pg_namespace rn on rn.oid = rr.relnamespace
        where c.contype = 'f' and n.nspname = $1
      `,
      [schema],
    );

  const tables = new Map();
  for (const row of columnsResult.rows) {
    if (!tables.has(row.table_name)) {
      tables.set(row.table_name, { columns: [], primaryKey: [], foreignKeys: [] });
    }
    tables.get(row.table_name).columns.push({
      name: row.column_name,
      nullable: row.nullable,
      defaultValue: row.column_default,
      generated: row.generated,
      identity: row.identity,
      identityGeneration: row.identity_generation,
      dataType: row.data_type,
      udtName: row.udt_name,
    });
  }
  for (const row of primaryKeysResult.rows) {
    if (tables.has(row.table_name)) tables.get(row.table_name).primaryKey = asArray(row.columns);
  }
  for (const row of foreignKeysResult.rows) {
    if (!tables.has(row.table_name)) continue;
    tables.get(row.table_name).foreignKeys.push({
      columns: asArray(row.columns),
      refSchema: row.ref_schema,
      refTable: row.ref_table,
      refColumns: asArray(row.ref_columns),
    });
  }
  return tables;
}

function columnMap(table) {
  return new Map(table.columns.map((column) => [column.name, column]));
}

export function prepareColumnValues(row, columns, table) {
  const columnsByName = columnMap(table);
  return columns.map((name) => {
    const value = row[name];
    if (value === null || value === undefined) return value;
    return ["json", "jsonb"].includes(columnsByName.get(name)?.dataType)
      ? JSON.stringify(value)
      : value;
  });
}

function commonColumns(sourceTable, targetTable) {
  const sourceColumns = new Set(sourceTable.columns.map((column) => column.name));
  return targetTable.columns
    .filter((column) => !column.generated && sourceColumns.has(column.name))
    .map((column) => column.name);
}

function assertCompatibleTable(tableName, sourceTable, targetTable) {
  if (targetTable.primaryKey.length === 0) {
    throw new SafeProductionMigrationError(`target table has no primary key: ${tableName}`);
  }
  const sourceColumns = new Set(sourceTable.columns.map((column) => column.name));
  const targetColumns = new Set(targetTable.columns.map((column) => column.name));
  if (sourceTable.columns.some((column) => !targetColumns.has(column.name))) {
    throw new SafeProductionMigrationError(`target schema is missing a source column: ${tableName}`);
  }
  for (const column of targetTable.columns) {
    if (
      !column.generated &&
      !column.nullable &&
      column.defaultValue === null &&
      !sourceColumns.has(column.name)
    ) {
      throw new SafeProductionMigrationError(`source schema is missing a required target column: ${tableName}`);
    }
  }
  for (const primaryKeyColumn of targetTable.primaryKey) {
    if (!sourceColumns.has(primaryKeyColumn)) {
      throw new SafeProductionMigrationError(`source schema is missing a target primary key: ${tableName}`);
    }
  }
}

function strictDependencies(tableName, table, selectedTables) {
  const columns = columnMap(table);
  return table.foreignKeys
    .filter(
      (foreignKey) =>
        foreignKey.refSchema === "public" &&
        foreignKey.refTable !== tableName &&
        selectedTables.has(foreignKey.refTable) &&
        foreignKey.columns.every((name) => columns.get(name)?.nullable === false),
    )
    .map((foreignKey) => foreignKey.refTable);
}

export function transferPlan(sourceTables, targetTables) {
  const selected = new Set(
    [...sourceTables.keys()].filter((table) => targetTables.has(table)),
  );
  for (const table of selected) {
    const targetTable = targetTables.get(table);
    assertCompatibleTable(table, sourceTables.get(table), targetTable);
    const columns = columnMap(targetTable);
    if (targetTable.foreignKeys.some(
      (foreignKey) => foreignKey.refSchema === "public" &&
        foreignKey.refTable === table &&
        foreignKey.columns.every((name) => columns.get(name)?.nullable === false),
    )) {
      throw new SafeProductionMigrationError("non-nullable self reference blocks migration");
    }
  }

  const general = new Set(
    [...selected].filter(
      (table) => !SEED_TABLES.has(table) && !SEED_RELATIONS.has(table),
    ),
  );
  const remaining = new Set(general);
  const ordered = [];
  while (remaining.size > 0) {
    const ready = [...remaining]
      .filter((table) =>
        strictDependencies(table, targetTables.get(table), general).every(
          (dependency) => !remaining.has(dependency),
        ),
      )
      .sort();
    if (ready.length === 0) {
      throw new SafeProductionMigrationError("non-nullable foreign-key cycle blocks migration");
    }
    for (const table of ready) {
      remaining.delete(table);
      ordered.push(table);
    }
  }
  return {
    selected,
    ordered,
    seedTables: [...SEED_TABLES.keys()].filter((table) => selected.has(table)),
    seedRelations: [...SEED_RELATIONS.keys()].filter((table) => selected.has(table)),
  };
}

function normalizeValue(value) {
  if (value instanceof Date) return value.toISOString();
  if (Buffer.isBuffer(value)) return value.toString("base64");
  if (Array.isArray(value)) return value.map(normalizeValue);
  if (value && typeof value === "object") {
    return Object.fromEntries(
      Object.keys(value).sort().map((key) => [key, normalizeValue(value[key])]),
    );
  }
  return value;
}

function canonicalRow(row, columns) {
  return Object.fromEntries(columns.map((column) => [column, normalizeValue(row[column])]));
}

export function rowsSha256(rows, columns) {
  const hash = createHash("sha256");
  const values = rows.map((row) => JSON.stringify(canonicalRow(row, columns))).sort();
  for (const value of values) hash.update(value).update("\n");
  return hash.digest("hex");
}

async function readRows(client, schema, table, columns, orderColumns = []) {
  if (columns.length === 0) return [];
  const order = orderColumns.length
    ? ` order by ${orderColumns.map(quoted).join(", ")}`
    : "";
  return (
    await client.query(
      `select ${columns.map(quoted).join(", ")} from ${qualified(schema, table)}${order}`,
    )
  ).rows;
}

async function countRows(client, schema, table) {
  const result = await client.query(`select count(*)::bigint as count from ${qualified(schema, table)}`);
  return Number(result.rows[0].count);
}

function bannedState(value, now = new Date()) {
  if (value === null || value === undefined || value === "") {
    return { banned: false, banExpires: null };
  }
  if (String(value).toLowerCase() === "infinity") {
    return { banned: true, banExpires: null };
  }
  const date = value instanceof Date ? value : new Date(value);
  if (!Number.isFinite(date.getTime())) {
    throw new SafeProductionMigrationError("source contains an invalid banned_until value");
  }
  return date > now
    ? { banned: true, banExpires: value }
    : { banned: false, banExpires: null };
}

export function normalizeAuthUsers(rows, now = new Date(), activeAdminUserIds = new Set()) {
  let portableUsers;
  try {
    portableUsers = normalizeSupabaseUsers(rows);
  } catch {
    throw new SafeProductionMigrationError("source contains invalid or duplicate auth identities");
  }
  return portableUsers.map((user, index) => {
    const { banned, banExpires } = bannedState(rows[index].banned_until, now);
    return {
      id: user.id,
      name: user.name,
      email: user.email,
      email_verified: user.emailVerified,
      email_verified_at: user.emailVerifiedAt === null ? null : rows[index].email_confirmed_at,
      image: user.image,
      role: activeAdminUserIds.has(user.id) ? "admin" : "user",
      banned,
      ban_reason: banned ? "migrated blocked-user state" : null,
      ban_expires: banExpires,
      created_at: rows[index].created_at,
      updated_at: rows[index].updated_at,
      two_factor_enabled: false,
    };
  });
}

export function normalizeAuthUser(row, now = new Date()) {
  return normalizeAuthUsers([row], now)[0];
}

async function readSourceUsers(source, sourceAuthSchema, activeAdminUserIds) {
  const table = sourceAuthSchema.get("users");
  if (!table) throw new SafeProductionMigrationError("source auth.users is missing");
  const available = new Set(table.columns.map((column) => column.name));
  const required = [
    "id",
    "email",
    "raw_user_meta_data",
    "email_confirmed_at",
    "created_at",
    "updated_at",
  ];
  if (required.some((column) => !available.has(column))) {
    throw new SafeProductionMigrationError("source auth.users is missing portable identity columns");
  }
  const columns = [...required, ...(available.has("banned_until") ? ["banned_until"] : [])];
  return normalizeAuthUsers(
    await readRows(source, "auth", "users", columns, ["id"]),
    new Date(),
    activeAdminUserIds,
  );
}

export async function readActiveAdminUserIds(source, sourceTables, ownerUserId) {
  // Legacy production used a single ADMIN_EMAILS allowlist and has no admin tables.
  if (!sourceTables.has("admin_users")) return new Set([ownerUserId]);
  const result = await source.query(
    "select user_id from public.admin_users where revoked_at is null order by user_id",
  );
  return new Set(result.rows.map((row) => String(row.user_id).toLowerCase()));
}

export function assertActiveAdminUsers(users, activeAdminUserIds, ownerUserId, ownerEmail) {
  if (!activeAdminUserIds.has(ownerUserId)) {
    throw new SafeProductionMigrationError("the designated Owner is not an active source administrator");
  }
  const usersById = new Map(users.map((user) => [user.id, user]));
  const owner = usersById.get(ownerUserId);
  if (!owner) throw new SafeProductionMigrationError("the designated Owner is absent from source auth users");
  if (owner.email !== ownerEmail) {
    throw new SafeProductionMigrationError("the designated Owner UUID does not match PRODUCTION_OWNER_EMAIL");
  }
  if (owner.banned) throw new SafeProductionMigrationError("the designated Owner is blocked");
  for (const userId of activeAdminUserIds) {
    const user = usersById.get(userId);
    if (!user) throw new SafeProductionMigrationError("an active source administrator is absent from auth users");
    if (user.banned) throw new SafeProductionMigrationError("an active source administrator is blocked");
  }
}

export async function assertTargetEmpty(target, targetTables) {
  for (const table of IDENTITY_STATE_TABLES) {
    if (await countRows(target, "identity", table)) {
      throw new SafeProductionMigrationError("target identity database is not empty");
    }
  }
  if (await countRows(target, "auth", "users")) {
    throw new SafeProductionMigrationError("target auth compatibility table is not empty");
  }
  for (const table of targetTables.keys()) {
    if (ALLOWED_TARGET_ROWS.has(`public.${table}`)) continue;
    if (await countRows(target, "public", table)) {
      throw new SafeProductionMigrationError("target business database is not empty");
    }
  }
}

async function assertNoUnmappedSourceTables(source, sourceTables, targetTables) {
  for (const table of sourceTables.keys()) {
    if (targetTables.has(table)) continue;
    if (await countRows(source, "public", table)) {
      throw new SafeProductionMigrationError("source contains an unsupported non-empty public table");
    }
  }
}

export async function assertActiveAdminRoles(source, sourceTables, ownerUserId) {
  if (!sourceTables.has("admin_users")) return;
  if (!sourceTables.has("admin_user_roles") || !sourceTables.has("admin_roles")) {
    const result = await source.query(
      `select count(*)::bigint as count from public.admin_users where revoked_at is null and user_id <> $1`,
      [ownerUserId],
    );
    if (Number(result.rows[0].count) > 0) {
      throw new SafeProductionMigrationError("an active source administrator has no canonical target role");
    }
    return;
  }
  const result = await source.query(
    `
      select count(*)::bigint as count
      from public.admin_users au
      where au.revoked_at is null and au.user_id <> $1
        and not exists (
          select 1
          from public.admin_user_roles aur
          join public.admin_roles ar on ar.id = aur.role_id
          where aur.admin_user_id = au.user_id
            and ar.code in ('owner','model_admin','billing_admin','operations','support','auditor')
        )
    `,
    [ownerUserId],
  );
  if (Number(result.rows[0].count) > 0) {
    throw new SafeProductionMigrationError("an active source administrator has no canonical target role");
  }
  const otherOwners = await source.query(
    `
      select count(*)::bigint as count
      from public.admin_users au
      join public.admin_user_roles aur on aur.admin_user_id = au.user_id
      join public.admin_roles ar on ar.id = aur.role_id
      where au.revoked_at is null and ar.code = 'owner' and au.user_id <> $1
    `,
    [ownerUserId],
  );
  if (Number(otherOwners.rows[0].count) > 0) {
    throw new SafeProductionMigrationError("source contains more than one active Owner");
  }
}

function remapForeignKeys(row, table, maps) {
  const result = { ...row };
  for (const foreignKey of table.foreignKeys) {
    const map = maps.get(`${foreignKey.refSchema}.${foreignKey.refTable}`);
    if (!map || foreignKey.columns.length !== 1 || foreignKey.refColumns[0] !== "id") continue;
    const column = foreignKey.columns[0];
    if (result[column] === null || result[column] === undefined) continue;
    const mapped = map.get(String(result[column]));
    if (!mapped) throw new SafeProductionMigrationError("a configuration foreign key could not be mapped");
    result[column] = mapped;
  }
  return result;
}

function applyCiphertextPolicy(row, table, ciphertextMode) {
  if (ciphertextMode !== "exclude") return row;
  const columns = CIPHERTEXT_COLUMNS.get(table);
  if (!columns) return row;
  return Object.fromEntries(
    Object.entries(row).map(([column, value]) => [column, columns.has(column) ? null : value]),
  );
}

function deferredForeignKeys(tableName, table, selectedTables) {
  const columns = columnMap(table);
  return table.foreignKeys.filter(
    (foreignKey) =>
      foreignKey.refSchema === "public" &&
      selectedTables.has(foreignKey.refTable) &&
      (foreignKey.refTable === tableName ||
        foreignKey.columns.some((name) => columns.get(name)?.nullable === true)),
  );
}


export function copiedDeferredForeignKeys(tableName, table, selectedTables, copiedColumns) {
  return deferredForeignKeys(tableName, table, selectedTables).filter(
    (foreignKey) => foreignKey.columns.every((column) => copiedColumns.includes(column)),
  );
}

function parameterList(length) {
  return Array.from({ length }, (_, index) => `$${index + 1}`).join(", ");
}

async function migrateTable(tableName, operation) {
  try {
    return await operation();
  } catch (error) {
    if (error instanceof SafeProductionMigrationError) throw error;
    const diagnostic = [error?.code, error?.constraint, error?.column].filter(Boolean).join(":");
    throw new SafeProductionMigrationError(
      `failed to migrate public.${tableName}${diagnostic ? ` (${diagnostic})` : ""}`,
    );
  }
}

async function insertIdentityUsers(target, users) {
  const columns = [
    "id", "name", "email", "email_verified", "email_verified_at", "image", "role",
    "banned", "ban_reason", "ban_expires", "created_at", "updated_at", "two_factor_enabled",
  ];
  const sql = `insert into identity.users (${columns.map(quoted).join(", ")}) values (${parameterList(columns.length)})`;
  for (const user of users) {
    await target.query(sql, columns.map((column) => user[column]));
  }
}

async function mergeSeedTable(source, target, tableName, sourceTable, targetTable, maps) {
  const naturalKey = SEED_TABLES.get(tableName);
  const columns = commonColumns(sourceTable, targetTable).filter((column) => column !== "id");
  if (naturalKey.some((column) => !columns.includes(column))) {
    throw new SafeProductionMigrationError("a seed table is missing its natural key");
  }
  const rows = await readRows(source, "public", tableName, ["id", ...columns], sourceTable.primaryKey);
  const map = new Map();
  const updateColumns = columns.filter((column) => !naturalKey.includes(column));
  const assignments = updateColumns.length
    ? updateColumns.map((column) => `${quoted(column)} = excluded.${quoted(column)}`).join(", ")
    : `${quoted(naturalKey[0])} = excluded.${quoted(naturalKey[0])}`;
  const sql = `
    insert into ${qualified("public", tableName)} (${columns.map(quoted).join(", ")})
    values (${parameterList(columns.length)})
    on conflict (${naturalKey.map(quoted).join(", ")}) do update set ${assignments}
    returning id
  `;
  for (const sourceRow of rows) {
    const row = applyCiphertextPolicy(
      remapForeignKeys(sourceRow, targetTable, maps),
      tableName,
      "preserve",
    );
    const result = await target.query(sql, prepareColumnValues(row, columns, targetTable));
    map.set(String(sourceRow.id), result.rows[0].id);
  }
  maps.set(`public.${tableName}`, map);
  return rows.length;
}

async function mergeSeedRelation(source, target, tableName, sourceTable, targetTable, maps, ciphertextMode) {
  const naturalKey = SEED_RELATIONS.get(tableName);
  const columns = commonColumns(sourceTable, targetTable).filter((column) => column !== "id");
  const rows = await readRows(source, "public", tableName, commonColumns(sourceTable, targetTable), sourceTable.primaryKey);
  const updateColumns = columns.filter((column) => !naturalKey.includes(column));
  const conflict = updateColumns.length
    ? `do update set ${updateColumns.map((column) => `${quoted(column)} = excluded.${quoted(column)}`).join(", ")}`
    : "do nothing";
  const sql = `
    insert into ${qualified("public", tableName)} (${columns.map(quoted).join(", ")})
    values (${parameterList(columns.length)})
    on conflict (${naturalKey.map(quoted).join(", ")}) ${conflict}
  `;
  let map;
  if (commonColumns(sourceTable, targetTable).includes("id")) map = new Map();
  for (const sourceRow of rows) {
    const row = applyCiphertextPolicy(
      remapForeignKeys(sourceRow, targetTable, maps),
      tableName,
      ciphertextMode,
    );
    await target.query(sql, prepareColumnValues(row, columns, targetTable));
    if (map) {
      const where = naturalKey.map((column, index) => `${quoted(column)} = $${index + 1}`).join(" and ");
      const result = await target.query(
        `select id from ${qualified("public", tableName)} where ${where}`,
        prepareColumnValues(row, naturalKey, targetTable),
      );
      map.set(String(sourceRow.id), result.rows[0].id);
    }
  }
  if (map) maps.set(`public.${tableName}`, map);
  return rows.length;
}

async function copyTable(source, target, tableName, sourceTable, targetTable, maps, selectedTables, ciphertextMode) {
  const columns = commonColumns(sourceTable, targetTable);
  const rows = await readRows(source, "public", tableName, columns, sourceTable.primaryKey);
  const deferred = copiedDeferredForeignKeys(tableName, targetTable, selectedTables, columns);
  const deferredColumns = new Set(
    deferred.flatMap((foreignKey) => {
      const columnsByName = columnMap(targetTable);
      return foreignKey.columns.filter((column) => columnsByName.get(column)?.nullable);
    }),
  );
  const updateColumns = columns.filter((column) => !targetTable.primaryKey.includes(column));
  const conflict = updateColumns.length
    ? `do update set ${updateColumns.map((column) => `${quoted(column)} = excluded.${quoted(column)}`).join(", ")}`
    : "do nothing";
  const overriding = columns.some((column) => columnMap(targetTable).get(column)?.identity)
    ? " overriding system value"
    : "";
  const sql = `
    insert into ${qualified("public", tableName)} (${columns.map(quoted).join(", ")})${overriding}
    values (${parameterList(columns.length)})
    on conflict (${targetTable.primaryKey.map(quoted).join(", ")}) ${conflict}
  `;
  for (const sourceRow of rows) {
    let row = applyCiphertextPolicy(
      remapForeignKeys(sourceRow, targetTable, maps),
      tableName,
      ciphertextMode,
    );
    row = { ...row };
    for (const column of deferredColumns) row[column] = null;
    await target.query(sql, prepareColumnValues(row, columns, targetTable));
  }
  for (const column of targetTable.columns.filter(
    (column) => column.identity && columns.includes(column.name),
  )) {
    await target.query(
      `select setval(pg_get_serial_sequence($1, $2), coalesce(max(${quoted(column.name)}), 1), max(${quoted(column.name)}) is not null) from ${qualified("public", tableName)}`,
      [`public.${tableName}`, column.name],
    );
  }
  return { count: rows.length, deferred };
}

async function restoreDeferredForeignKeys(source, target, tableName, sourceTable, targetTable, foreignKeys, maps) {
  if (foreignKeys.length === 0) return;
  const updateColumns = [...new Set(foreignKeys.flatMap((foreignKey) => foreignKey.columns))];
  const columns = [...new Set([...sourceTable.primaryKey, ...updateColumns])];
  const rows = await readRows(source, "public", tableName, columns, sourceTable.primaryKey);
  for (const sourceRow of rows) {
    const row = remapForeignKeys(sourceRow, targetTable, maps);
    const assignments = updateColumns.map((column, index) => `${quoted(column)} = $${index + 1}`);
    const where = sourceTable.primaryKey.map(
      (column, index) => `${quoted(column)} = $${updateColumns.length + index + 1}`,
    );
    await target.query(
      `update ${qualified("public", tableName)} set ${assignments.join(", ")} where ${where.join(" and ")}`,
      [
        ...prepareColumnValues(row, updateColumns, targetTable),
        ...prepareColumnValues(row, sourceTable.primaryKey, targetTable),
      ],
    );
  }
}

async function forceOwner(target, ownerUserId) {
  const role = await target.query("select id from public.admin_roles where code = 'owner'");
  if (role.rows.length !== 1) throw new SafeProductionMigrationError("target Owner role is missing");
  await target.query(
    `
      insert into public.admin_users (user_id, created_by, revoked_at, revoked_by)
      values ($1, $1, null, null)
      on conflict (user_id) do update set
        revoked_at = null,
        revoked_by = null,
        updated_at = case
          when public.admin_users.revoked_at is not null or public.admin_users.revoked_by is not null
            then now()
          else public.admin_users.updated_at
        end
    `,
    [ownerUserId],
  );
  await target.query(
    `
      insert into public.admin_user_roles (admin_user_id, role_id, assigned_by)
      values ($1, $2, $1)
      on conflict (admin_user_id, role_id) do nothing
    `,
    [ownerUserId, role.rows[0].id],
  );
}

export async function assertTargetAdminState(target, ownerUserId) {
  const result = await target.query(
    `
      select
        count(*) filter (where au.user_id = $1 and au.revoked_at is null and ar.code = 'owner')::int as owner_count,
        count(*) filter (where au.revoked_at is null and ar.code = 'owner')::int as total_owner_count,
        (
          select count(*)::int from public.admin_users active
          where active.revoked_at is null and not exists (
            select 1 from public.admin_user_roles roles where roles.admin_user_id = active.user_id
          )
        ) as admins_without_roles,
        (
          select count(*)::int from public.admin_users active
          where active.revoked_at is null and not exists (
            select 1 from identity.users users
            where users.id = active.user_id and users.role = 'admin' and users.banned = false
          )
        ) as unusable_identity_admins
      from public.admin_users au
      join public.admin_user_roles aur on aur.admin_user_id = au.user_id
      join public.admin_roles ar on ar.id = aur.role_id
    `,
    [ownerUserId],
  );
  if (
    result.rows[0].owner_count !== 1 ||
    result.rows[0].total_owner_count !== 1 ||
    result.rows[0].admins_without_roles !== 0 ||
    result.rows[0].unusable_identity_admins !== 0
  ) {
    throw new SafeProductionMigrationError("target administrator reconciliation failed");
  }
}

async function preflightContext(source, target, config, { requireEmpty = true } = {}) {
  const [sourcePublic, sourceAuth, targetPublic] = await Promise.all([
    readSchema(source, "public"),
    readSchema(source, "auth"),
    readSchema(target, "public"),
  ]);
  const activeAdminUserIds = await readActiveAdminUserIds(source, sourcePublic, config.ownerUserId);
  const users = await readSourceUsers(source, sourceAuth, activeAdminUserIds);
  assertActiveAdminUsers(users, activeAdminUserIds, config.ownerUserId, config.ownerEmail);
  if (requireEmpty) await assertTargetEmpty(target, targetPublic);
  await assertNoUnmappedSourceTables(source, sourcePublic, targetPublic);
  await assertActiveAdminRoles(source, sourcePublic, config.ownerUserId);
  const plan = transferPlan(sourcePublic, targetPublic);
  return { sourcePublic, targetPublic, users, plan };
}

async function applyMigration(source, target, config, context) {
  const { sourcePublic, targetPublic, users, plan } = context;
  const counts = { identity_users: users.length };
  const maps = new Map();
  await insertIdentityUsers(target, users);

  for (const tableName of plan.seedTables) {
    counts[`public.${tableName}`] = await migrateTable(tableName, () => mergeSeedTable(
      source, target, tableName, sourcePublic.get(tableName), targetPublic.get(tableName), maps,
    ));
  }
  for (const tableName of plan.seedRelations) {
    counts[`public.${tableName}`] = await migrateTable(tableName, () => mergeSeedRelation(
      source, target, tableName, sourcePublic.get(tableName), targetPublic.get(tableName), maps,
      config.ciphertextMode,
    ));
  }

  const deferredByTable = new Map();
  for (const tableName of plan.ordered) {
    const result = await migrateTable(tableName, () => copyTable(
      source, target, tableName, sourcePublic.get(tableName), targetPublic.get(tableName), maps,
      plan.selected, config.ciphertextMode,
    ));
    counts[`public.${tableName}`] = result.count;
    deferredByTable.set(tableName, result.deferred);
  }
  for (const tableName of plan.ordered) {
    await migrateTable(tableName, () => restoreDeferredForeignKeys(
      source, target, tableName, sourcePublic.get(tableName), targetPublic.get(tableName),
      deferredByTable.get(tableName), maps,
    ));
  }
  await forceOwner(target, config.ownerUserId);
  await assertTargetAdminState(target, config.ownerUserId);
  return counts;
}

async function migrationFilesAreCurrent(targetUrl) {
  const scriptDirectory = dirname(fileURLToPath(import.meta.url));
  const output = [];
  const status = await runMigrations({
    connectionString: targetUrl,
    migrationsDirectories: [
      resolve(scriptDirectory, "../db/migrations"),
      resolve(scriptDirectory, "../supabase/migrations"),
    ],
    logger: { log: (value) => output.push(String(value)) },
    check: true,
  });
  if (status !== 0) throw new SafeProductionMigrationError("target has pending schema migrations");
}

async function buildMapsForVerification(source, target, sourcePublic, plan) {
  const maps = new Map();
  for (const [tableName, naturalKey] of SEED_TABLES) {
    if (!plan.selected.has(tableName)) continue;
    const sourceTable = sourcePublic.get(tableName);
    const sourceRows = await readRows(
      source, "public", tableName,
      ["id", ...naturalKey], sourceTable.primaryKey,
    );
    const map = new Map();
    for (const row of sourceRows) {
      const where = naturalKey.map((column, index) => `${quoted(column)} = $${index + 1}`).join(" and ");
      const match = await target.query(
        `select id from ${qualified("public", tableName)} where ${where}`,
        naturalKey.map((column) => row[column]),
      );
      if (match.rows.length !== 1) throw new SafeProductionMigrationError("seed reconciliation failed");
      map.set(String(row.id), match.rows[0].id);
    }
    maps.set(`public.${tableName}`, map);
  }
  return maps;
}

async function tableManifest(source, target, tableName, sourceTable, targetTable, maps, ciphertextMode, seedSubset) {
  let columns = commonColumns(sourceTable, targetTable);
  if (seedSubset) columns = columns.filter((column) => column !== "id");
  const sourceRows = await readRows(source, "public", tableName, columns, sourceTable.primaryKey);
  const expected = sourceRows.map((row) =>
    applyCiphertextPolicy(remapForeignKeys(row, targetTable, maps), tableName, ciphertextMode),
  );
  let targetRows;
  if (seedSubset) {
    const naturalKey = SEED_TABLES.get(tableName) ?? SEED_RELATIONS.get(tableName);
    targetRows = [];
    for (const row of expected) {
      const where = naturalKey.map((column, index) => `${quoted(column)} = $${index + 1}`).join(" and ");
      const match = await target.query(
        `select ${columns.map(quoted).join(", ")} from ${qualified("public", tableName)} where ${where}`,
        prepareColumnValues(row, naturalKey, targetTable),
      );
      targetRows.push(...match.rows);
    }
  } else {
    targetRows = await readRows(target, "public", tableName, columns, targetTable.primaryKey);
  }
  const sourceHash = rowsSha256(expected, columns);
  const targetHash = rowsSha256(targetRows, columns);
  const keyColumns = seedSubset
    ? (SEED_TABLES.get(tableName) ?? SEED_RELATIONS.get(tableName))
    : sourceTable.primaryKey;
  const sourceKeyHash = rowsSha256(expected, keyColumns);
  const targetKeyHash = rowsSha256(targetRows, keyColumns);
  return {
    source_count: expected.length,
    target_count: targetRows.length,
    primary_key_sha256: sourceKeyHash,
    target_primary_key_sha256: targetKeyHash,
    normalized_sha256: sourceHash,
    target_normalized_sha256: targetHash,
    ok:
      expected.length === targetRows.length &&
      sourceKeyHash === targetKeyHash &&
      sourceHash === targetHash,
  };
}

async function queryAggregate(client, text) {
  return (await client.query(text)).rows.map((row) => normalizeValue(row));
}

async function reconciliationAggregates(client, selectedTables) {
  const result = {};
  if (selectedTables.has("credit_transactions")) {
    result.credits = await queryAggregate(
      client,
      `select transaction_type as state, count(*)::bigint as count, coalesce(sum(amount),0)::text as amount from public.credit_transactions group by transaction_type order by transaction_type`,
    );
  }
  if (selectedTables.has("payment_orders")) {
    result.orders = await queryAggregate(
      client,
      `select status, count(*)::bigint as count, coalesce(sum(money_cents),0)::text as money_cents, coalesce(sum(refund_amount_cents),0)::text as refund_cents from public.payment_orders group by status order by status`,
    );
  }
  if (selectedTables.has("user_subscriptions")) {
    result.subscriptions = await queryAggregate(
      client,
      `select status, count(*)::bigint as count from public.user_subscriptions group by status order by status`,
    );
  }
  if (selectedTables.has("personal_reports")) {
    result.personal_reports = await queryAggregate(
      client,
      `select status, count(*)::bigint as count from public.personal_reports group by status order by status`,
    );
  }
  if (selectedTables.has("consultation_requests")) {
    result.consultations = await queryAggregate(
      client,
      `select status, count(*)::bigint as count from public.consultation_requests group by status order by status`,
    );
  }
  const rectificationTables = [...selectedTables].filter((table) => table.includes("rectification"));
  result.rectification = [];
  for (const table of rectificationTables.sort()) {
    result.rectification.push({ table, count: await countRows(client, "public", table) });
  }
  return result;
}

async function verifyMigration(source, target, config, context) {
  const { sourcePublic, targetPublic, users, plan } = context;
  const maps = await buildMapsForVerification(source, target, sourcePublic, plan);
  const identityColumns = Object.keys(users[0] ?? normalizeAuthUser({
    id: "00000000-0000-4000-8000-000000000000",
    email: "empty@example.invalid",
    raw_user_meta_data: {}, email_confirmed_at: null, created_at: null, updated_at: null,
  }));
  const targetUsers = await readRows(target, "identity", "users", identityColumns, ["id"]);
  const identityHash = rowsSha256(users, identityColumns);
  const targetIdentityHash = rowsSha256(targetUsers, identityColumns);
  const tables = {};
  for (const tableName of [...plan.selected].sort()) {
    tables[`public.${tableName}`] = await tableManifest(
      source, target, tableName, sourcePublic.get(tableName), targetPublic.get(tableName), maps,
      config.ciphertextMode, SEED_TABLES.has(tableName) || SEED_RELATIONS.has(tableName),
    );
  }
  await assertTargetAdminState(target, config.ownerUserId);
  const sourceAggregates = await reconciliationAggregates(source, plan.selected);
  const targetAggregates = await reconciliationAggregates(target, plan.selected);
  const aggregatesOk = JSON.stringify(sourceAggregates) === JSON.stringify(targetAggregates);
  const ok =
    users.length === targetUsers.length &&
    identityHash === targetIdentityHash &&
    Object.values(tables).every((table) => table.ok) &&
    aggregatesOk;
  return {
    mode: "verify",
    ok,
    identity: {
      source_count: users.length,
      target_count: targetUsers.length,
      normalized_sha256: identityHash,
      target_normalized_sha256: targetIdentityHash,
    },
    tables,
    aggregates: { source: sourceAggregates, target: targetAggregates, ok: aggregatesOk },
    ciphertext_mode: config.ciphertextMode,
  };
}

export async function run(mode, env, dependencies = {}) {
  const config = readConfiguration(env);
  const PoolClass = dependencies.Pool ?? Pool;
  await (dependencies.checkMigrations ?? migrationFilesAreCurrent)(config.targetUrl);
  const sourcePool = new PoolClass({
    connectionString: config.sourceUrl,
    application_name: "jyotisha-production-migration-source",
    max: 1,
  });
  const targetPool = new PoolClass({
    connectionString: config.targetUrl,
    application_name: "jyotisha-production-migration-target",
    max: 1,
  });
  let source;
  let target;
  let sourceTransaction = false;
  let targetTransaction = false;
  try {
    source = await sourcePool.connect();
    target = await targetPool.connect();
    await source.query("begin isolation level repeatable read read only");
    sourceTransaction = true;
    if (mode === "verify") {
      await target.query("begin isolation level repeatable read read only");
    } else {
      await target.query("begin");
      if (mode === "apply") {
        await target.query("select pg_advisory_xact_lock(hashtext('jyotisha_production_data_migration'))");
      }
    }
    targetTransaction = true;
    const context = await preflightContext(source, target, config, {
      requireEmpty: mode !== "verify",
    });

    if (mode === "preflight") {
      await target.query("rollback");
      targetTransaction = false;
      await source.query("commit");
      sourceTransaction = false;
      return {
        mode,
        ok: true,
        source_users: context.users.length,
        source_public_tables: context.plan.selected.size,
        target_business_empty: true,
        owner_ready: true,
        ciphertext_mode: config.ciphertextMode,
      };
    }
    if (mode === "apply") {
      const counts = await applyMigration(source, target, config, context);
      await target.query("commit");
      targetTransaction = false;
      await source.query("commit");
      sourceTransaction = false;
      return { mode, ok: true, imported: counts, ciphertext_mode: config.ciphertextMode };
    }
    const manifest = await verifyMigration(source, target, config, context);
    await target.query("commit");
    targetTransaction = false;
    await source.query("commit");
    sourceTransaction = false;
    return manifest;
  } catch (error) {
    if (targetTransaction && target) await target.query("rollback").catch(() => {});
    if (sourceTransaction && source) await source.query("rollback").catch(() => {});
    throw error;
  } finally {
    source?.release();
    target?.release();
    await sourcePool.end().catch(() => {});
    await targetPool.end().catch(() => {});
  }
}

function safeMessage(error) {
  return error instanceof SafeProductionMigrationError
    ? error.message
    : "production data migration failed";
}

const invokedPath = process.argv[1] ? pathToFileURL(resolve(process.argv[1])).href : "";
if (import.meta.url === invokedPath) {
  let mode;
  try {
    mode = parseMode(process.argv.slice(2));
    const result = await run(mode, process.env);
    process.stdout.write(`${JSON.stringify(result)}\n`);
    if (!result.ok) process.exitCode = 2;
  } catch (error) {
    process.stderr.write(`${safeMessage(error)}\n`);
    process.exitCode = 1;
  }
}