Add dev-prod harmonize tool for bidirectional catalog and image sync.
Schema migrates dev to prod only; catalog rows merge by updated_at and images by mtime with union merge and conflict reporting. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Danila KhodjaefandCursor committed
1 parent
81ebad2120
commit
f247b418d8
21 files changed
+1336
-4
No files matched your search
@@ -0,0 +1,304 @@
|
||||
/**
|
||||
* 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',
|
||||
];
|
||||
|
||||
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}`,
|
||||
};
|
||||
|
||||
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,
|
||||
});
|
||||
});
|
||||
Reference in new issue
Block a user