fix: complete production data migration rehearsal

frontend/scripts/migrate-supabase-production.mjs

@@ -65,6 +65,10 @@ export function readConfiguration(env) {
   if (!ownerUserId || !uuidPattern.test(ownerUserId)) {
     throw new SafeProductionMigrationError("PRODUCTION_OWNER_USER_ID must be a UUID");
   }
+  const ownerEmail = env.PRODUCTION_OWNER_EMAIL?.trim().toLowerCase();
+  if (!ownerEmail || !ownerEmail.includes("@")) {
+    throw new SafeProductionMigrationError("PRODUCTION_OWNER_EMAIL must be an email address");
+  }
   const ciphertextMode = env.PRODUCTION_CIPHERTEXT_MODE?.trim();
   if (!new Set(["preserve", "exclude"]).has(ciphertextMode)) {
     throw new SafeProductionMigrationError(
@@ -79,7 +83,7 @@ export function readConfiguration(env) {
       "preserving ciphertext requires confirmed production encryption keys",
     );
   }
-  return { sourceUrl, targetUrl, ownerUserId, ciphertextMode };
+  return { sourceUrl, targetUrl, ownerUserId, ownerEmail, ciphertextMode };
 }
 
 export function parseMode(arguments_) {
@@ -106,6 +110,7 @@ export async function readSchema(client, schema) {
         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
@@ -167,6 +172,8 @@ export async function readSchema(client, schema) {
       generated: row.generated,
       identity: row.identity,
       identityGeneration: row.identity_generation,
+      dataType: row.data_type,
+      udtName: row.udt_name,
     });
   }
   for (const row of primaryKeysResult.rows) {
@@ -188,6 +195,17 @@ 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
@@ -401,13 +419,16 @@ export async function readActiveAdminUserIds(source, sourceTables, ownerUserId)
   return new Set(result.rows.map((row) => String(row.user_id).toLowerCase()));
 }
 
-export function assertActiveAdminUsers(users, activeAdminUserIds, ownerUserId) {
+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);
@@ -506,10 +527,29 @@ function deferredForeignKeys(tableName, table, selectedTables) {
   );
 }
 
+
+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",
@@ -545,7 +585,7 @@ async function mergeSeedTable(source, target, tableName, sourceTable, targetTabl
       tableName,
       "preserve",
     );
-    const result = await target.query(sql, columns.map((column) => row[column]));
+    const result = await target.query(sql, prepareColumnValues(row, columns, targetTable));
     map.set(String(sourceRow.id), result.rows[0].id);
   }
   maps.set(`public.${tableName}`, map);
@@ -573,12 +613,12 @@ async function mergeSeedRelation(source, target, tableName, sourceTable, targetT
       tableName,
       ciphertextMode,
     );
-    await target.query(sql, columns.map((column) => row[column]));
+    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}`,
-        naturalKey.map((column) => row[column]),
+        prepareColumnValues(row, naturalKey, targetTable),
       );
       map.set(String(sourceRow.id), result.rows[0].id);
     }
@@ -590,7 +630,7 @@ async function mergeSeedRelation(source, target, tableName, sourceTable, targetT
 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 = deferredForeignKeys(tableName, targetTable, selectedTables);
+  const deferred = copiedDeferredForeignKeys(tableName, targetTable, selectedTables, columns);
   const deferredColumns = new Set(
     deferred.flatMap((foreignKey) => {
       const columnsByName = columnMap(targetTable);
@@ -617,7 +657,7 @@ async function copyTable(source, target, tableName, sourceTable, targetTable, ma
     );
     row = { ...row };
     for (const column of deferredColumns) row[column] = null;
-    await target.query(sql, columns.map((column) => row[column]));
+    await target.query(sql, prepareColumnValues(row, columns, targetTable));
   }
   for (const column of targetTable.columns.filter(
     (column) => column.identity && columns.includes(column.name),
@@ -643,7 +683,10 @@ async function restoreDeferredForeignKeys(source, target, tableName, sourceTable
     );
     await target.query(
       `update ${qualified("public", tableName)} set ${assignments.join(", ")} where ${where.join(" and ")}`,
-      [...updateColumns.map((column) => row[column]), ...sourceTable.primaryKey.map((column) => row[column])],
+      [
+        ...prepareColumnValues(row, updateColumns, targetTable),
+        ...prepareColumnValues(row, sourceTable.primaryKey, targetTable),
+      ],
     );
   }
 }
@@ -717,7 +760,7 @@ async function preflightContext(source, target, config, { requireEmpty = true }
   ]);
   const activeAdminUserIds = await readActiveAdminUserIds(source, sourcePublic, config.ownerUserId);
   const users = await readSourceUsers(source, sourceAuth, activeAdminUserIds);
-  assertActiveAdminUsers(users, activeAdminUserIds, config.ownerUserId);
+  assertActiveAdminUsers(users, activeAdminUserIds, config.ownerUserId, config.ownerEmail);
   if (requireEmpty) await assertTargetEmpty(target, targetPublic);
   await assertNoUnmappedSourceTables(source, sourcePublic, targetPublic);
   await assertActiveAdminRoles(source, sourcePublic, config.ownerUserId);
@@ -732,31 +775,31 @@ async function applyMigration(source, target, config, context) {
   await insertIdentityUsers(target, users);
 
   for (const tableName of plan.seedTables) {
-    counts[`public.${tableName}`] = await mergeSeedTable(
+    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 mergeSeedRelation(
+    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 copyTable(
+    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 restoreDeferredForeignKeys(
+    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);
@@ -817,7 +860,7 @@ async function tableManifest(source, target, tableName, sourceTable, targetTable
       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}`,
-        naturalKey.map((column) => row[column]),
+        prepareColumnValues(row, naturalKey, targetTable),
       );
       targetRows.push(...match.rows);
     }
@@ -849,39 +892,39 @@ async function queryAggregate(client, text) {
   return (await client.query(text)).rows.map((row) => normalizeValue(row));
 }
 
-async function reconciliationAggregates(client, tables) {
+async function reconciliationAggregates(client, selectedTables) {
   const result = {};
-  if (tables.has("credit_transactions")) {
+  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 (tables.has("payment_orders")) {
+  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 (tables.has("user_subscriptions")) {
+  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 (tables.has("personal_reports")) {
+  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 (tables.has("consultation_requests")) {
+  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 = [...tables.keys()].filter((table) => table.includes("rectification"));
+  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) });
@@ -908,8 +951,8 @@ async function verifyMigration(source, target, config, context) {
     );
   }
   await assertTargetAdminState(target, config.ownerUserId);
-  const sourceAggregates = await reconciliationAggregates(source, sourcePublic);
-  const targetAggregates = await reconciliationAggregates(target, targetPublic);
+  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 &&