/** * 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, }); });