Art-gallery/scripts/harmonize-db.js

/**
 * Bidirectional catalog DB harmonize (dev <-> prod) by updated_at.
 *
 * Usage:
 *   npm run harmonize:db
 *   npm run harmonize:db -- --dry-run
 *   npm run harmonize:db -- --prefer=dev
 */
const fs = require('fs');
const path = require('path');
const crypto = require('crypto');
const pg = require('pg');
const {
  loadDevPgConfig,
  loadProdPgConfig,
  confirmProdAction,
  rootDir,
} = require('./db-env');
const { printCliResult } = require('./lib/cli-result');

const { Client } = pg;

const CATALOG_TABLES = [
  'historical_eras',
  'art_movements',
  'artists',
  'artist_periods',
  'paintings',
  'painting_influences',
  'painting_influence_sources',
  'painting_annotations',
  'entity_translations',
];

const NATURAL_KEY_FN = {
  historical_eras: (r) => String(r.name || '').trim().toLowerCase(),
  art_movements: (r) => String(r.name || '').trim().toLowerCase(),
  artists: (r) => String(r.name || '').trim().toLowerCase(),
  artist_periods: (r) => `${r.artist_id}:${String(r.name || '').trim().toLowerCase()}`,
  paintings: (r) => `${r.artist_id}:${String(r.title || '').trim().toLowerCase()}`,
  painting_influences: (r) => `${r.painting_id}:${r.influenced_by_painting_id}`,
  painting_influence_sources: (r) => `${r.painting_id}:${r.source_type}:${r.source_painting_id || 0}:${r.source_artist_id || 0}:${r.source_movement_id || 0}`,
  painting_annotations: (r) => `${r.painting_id}:${String(r.label || '').trim().toLowerCase()}:${r.sort_order}`,
  entity_translations: (r) => `${r.entity_type}:${r.entity_id}:${r.locale}:${r.field_name}`,
};

function parseArgs(argv) {
  const tablesArg = argv.find((a) => a.startsWith('--tables='));
  const preferArg = argv.find((a) => a.startsWith('--prefer='));
  return {
    dryRun: argv.includes('--dry-run'),
    verbose: argv.includes('--verbose'),
    tables: tablesArg ? tablesArg.slice('--tables='.length).split(',').map((t) => t.trim()).filter(Boolean) : null,
    prefer: preferArg ? preferArg.slice('--prefer='.length).trim().toLowerCase() : null,
  };
}

function stableRowHash(row, columns) {
  const payload = {};
  for (const col of columns) {
    const val = row[col];
    if (val instanceof Date) payload[col] = val.toISOString();
    else if (val != null && typeof val === 'object') payload[col] = JSON.stringify(val);
    else payload[col] = val;
  }
  return crypto.createHash('sha256').update(JSON.stringify(payload)).digest('hex');
}

function compareUpdatedAt(a, b) {
  const ta = a?.updated_at ? new Date(a.updated_at).getTime() : 0;
  const tb = b?.updated_at ? new Date(b.updated_at).getTime() : 0;
  if (ta === tb) return 0;
  return ta > tb ? 1 : -1;
}

async function getTableColumns(client, tableName) {
  const { rows } = await client.query(
    `SELECT column_name
     FROM information_schema.columns
     WHERE table_schema = 'public' AND table_name = $1
     ORDER BY ordinal_position`,
    [tableName],
  );
  return rows.map((r) => r.column_name);
}

async function fetchTableRows(client, tableName) {
  const { rows } = await client.query(`SELECT * FROM ${tableName} ORDER BY id`);
  const map = new Map();
  for (const row of rows) map.set(row.id, row);
  return map;
}

function buildInsertSql(tableName, columns) {
  const cols = columns.map((c) => `"${c}"`).join(', ');
  const placeholders = columns.map((_, i) => `$${i + 1}`).join(', ');
  return `INSERT INTO ${tableName} (${cols}) OVERRIDING SYSTEM VALUE VALUES (${placeholders})`;
}

function buildUpsertSql(tableName, columns) {
  const insertCols = columns.map((c) => `"${c}"`).join(', ');
  const placeholders = columns.map((_, i) => `$${i + 1}`).join(', ');
  const updates = columns
    .filter((c) => c !== 'id')
    .map((c) => `"${c}" = EXCLUDED."${c}"`)
    .join(', ');
  return `INSERT INTO ${tableName} (${insertCols}) OVERRIDING SYSTEM VALUE VALUES (${placeholders})
    ON CONFLICT (id) DO UPDATE SET ${updates}
    WHERE ${tableName}.updated_at < EXCLUDED.updated_at`;
}

function rowValues(row, columns) {
  return columns.map((col) => row[col]);
}

async function insertRow(client, tableName, columns, row, dryRun) {
  if (dryRun) return;
  const sql = buildInsertSql(tableName, columns);
  await client.query(sql, rowValues(row, columns));
}

async function upsertRow(client, tableName, columns, row, dryRun) {
  if (dryRun) return;
  const sql = buildUpsertSql(tableName, columns);
  await client.query(sql, rowValues(row, columns));
}

async function fixSequence(client, tableName) {
  await client.query(
    `SELECT setval(pg_get_serial_sequence($1, 'id'), COALESCE((SELECT MAX(id) FROM ${tableName}), 1), true)`,
    [tableName],
  );
}

async function harmonizeTable(tableName, devClient, prodClient, options, stats) {
  const naturalKeyFn = NATURAL_KEY_FN[tableName];
  const columns = await getTableColumns(devClient, tableName);
  if (!columns.includes('updated_at')) {
    throw new Error(`Table ${tableName} missing updated_at — run migrate-sync-timestamps first`);
  }

  const devRows = await fetchTableRows(devClient, tableName);
  const prodRows = await fetchTableRows(prodClient, tableName);
  const allIds = new Set([...devRows.keys(), ...prodRows.keys()]);

  for (const id of allIds) {
    const devRow = devRows.get(id);
    const prodRow = prodRows.get(id);

    if (devRow && !prodRow) {
      stats.devToProd += 1;
      stats.actions.push({ table: tableName, id, action: 'dev→prod insert' });
      await insertRow(prodClient, tableName, columns, devRow, options.dryRun);
      continue;
    }

    if (prodRow && !devRow) {
      stats.prodToDev += 1;
      stats.actions.push({ table: tableName, id, action: 'prod→dev insert' });
      await insertRow(devClient, tableName, columns, prodRow, options.dryRun);
      continue;
    }

    const devKey = naturalKeyFn(devRow);
    const prodKey = naturalKeyFn(prodRow);
    if (devKey !== prodKey) {
      stats.conflicts += 1;
      stats.conflictDetails.push({
        table: tableName,
        id,
        reason: 'id_collision',
        devKey,
        prodKey,
      });
      continue;
    }

    const devHash = stableRowHash(devRow, columns);
    const prodHash = stableRowHash(prodRow, columns);
    if (devHash === prodHash) {
      stats.skipped += 1;
      continue;
    }

    const cmp = compareUpdatedAt(devRow, prodRow);
    if (cmp > 0) {
      stats.devToProd += 1;
      stats.actions.push({ table: tableName, id, action: 'dev→prod update' });
      await upsertRow(prodClient, tableName, columns, devRow, options.dryRun);
    } else if (cmp < 0) {
      stats.prodToDev += 1;
      stats.actions.push({ table: tableName, id, action: 'prod→dev update' });
      await upsertRow(devClient, tableName, columns, prodRow, options.dryRun);
    } else if (options.prefer === 'dev') {
      stats.devToProd += 1;
      stats.actions.push({ table: tableName, id, action: 'dev→prod update (prefer tie)' });
      await upsertRow(prodClient, tableName, columns, devRow, options.dryRun);
    } else if (options.prefer === 'prod') {
      stats.prodToDev += 1;
      stats.actions.push({ table: tableName, id, action: 'prod→dev update (prefer tie)' });
      await upsertRow(devClient, tableName, columns, prodRow, options.dryRun);
    } else {
      stats.conflicts += 1;
      stats.conflictDetails.push({
        table: tableName,
        id,
        reason: 'equal_updated_at',
        devUpdatedAt: devRow.updated_at,
        prodUpdatedAt: prodRow.updated_at,
      });
    }
  }

  if (!options.dryRun) {
    await fixSequence(devClient, tableName);
    await fixSequence(prodClient, tableName);
  }
}

function timestampSlug(date = new Date()) {
  const pad = (n) => String(n).padStart(2, '0');
  return [
    date.getFullYear(),
    pad(date.getMonth() + 1),
    pad(date.getDate()),
    '_',
    pad(date.getHours()),
    pad(date.getMinutes()),
    pad(date.getSeconds()),
  ].join('');
}

async function main() {
  const options = parseArgs(process.argv.slice(2));
  const tables = options.tables || CATALOG_TABLES;

  for (const table of tables) {
    if (!CATALOG_TABLES.includes(table)) {
      throw new Error(`Unknown catalog table: ${table}`);
    }
  }

  if (!options.dryRun && process.env.CONFIRM_PROD !== '1') {
    await confirmProdAction(
      'Harmonize will merge catalog rows between gallery_dev and gallery_prod (prod may be modified).',
    );
  }

  const devClient = new Client(loadDevPgConfig());
  const prodClient = new Client(loadProdPgConfig());
  await devClient.connect();
  await prodClient.connect();

  const stats = {
    devToProd: 0,
    prodToDev: 0,
    skipped: 0,
    conflicts: 0,
    actions: [],
    conflictDetails: [],
  };

  try {
    for (const table of tables) {
      if (options.verbose) console.log(`Harmonizing ${table}…`);
      await harmonizeTable(table, devClient, prodClient, options, stats);
    }
  } finally {
    await devClient.end();
    await prodClient.end();
  }

  const reportDir = path.join(rootDir, 'db', 'SyncReports');
  fs.mkdirSync(reportDir, { recursive: true });
  const reportPath = path.join(reportDir, `harmonize_db_${timestampSlug()}.json`);
  fs.writeFileSync(reportPath, JSON.stringify({
    generatedAt: new Date().toISOString(),
    dryRun: options.dryRun,
    prefer: options.prefer,
    tables,
    stats: {
      devToProd: stats.devToProd,
      prodToDev: stats.prodToDev,
      skipped: stats.skipped,
      conflicts: stats.conflicts,
    },
    conflicts: stats.conflictDetails,
    actions: options.verbose ? stats.actions : stats.actions.slice(0, 200),
  }, null, 2));

  const mode = options.dryRun ? ' (dry-run)' : '';
  printCliResult({
    script: 'harmonize-db',
    ok: true,
    summary: `DB harmonize complete${mode}: dev→prod ${stats.devToProd}, prod→dev ${stats.prodToDev}, skipped ${stats.skipped}, conflicts ${stats.conflicts}.`,
    details: [`Report: ${reportPath}`],
  });
}

main().catch((err) => {
  printCliResult({
    script: 'harmonize-db',
    ok: false,
    summary: err.message,
  });
});