jjbrunton/Sauci
1
0 Fork 0
mirror of https://github.com/jjbrunton/Sauci.git synced 2026-11-01 18:19:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(api): harden production data import

Browse Source 
Preserve live Dokploy state while importing legacy Supabase data. Add schema parity, safe relationship reconciliation, storage quarantine, and production-shaped integration coverage.
This commit is contained in:
jjbrunton committed 2026-08-29 00:04:11 +01:00
1 parent 78900a24fb
commit aedba2a480
8 files changed
+795 -85
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

apps/api/src/migration/database.ts
+108 -19
View File
@@ -1,14 +1,31 @@
import type { Pool, PoolClient } from 'pg';
import type { DatabaseMigrationOptions, TableResult } from './types.js';
interface TableSpec { name: string; key: string[]; replace?: boolean }
interface TableSpec {
name: string;
key: string[];
replace?: boolean;
/** Target-column to legacy source-column rename. */
columnMap?: Record<string, string>;
/** Target-column SQL expressions for legacy rows that need a derived value. */
sourceExpressions?: Record<string, string>;
/** Columns populated after all rows exist, for self-referential foreign keys. */
deferredColumns?: string[];
/** Keep a newer standalone-backend row when the same key exists in both systems. */
preserveNewer?: boolean;
/** Singleton runtime configuration is owned by the live target during an initial merge. */
preserveExistingOnInitial?: boolean;
}
export const TABLES: TableSpec[] = [
{ name: 'couples', key: ['id'] }, { name: 'profiles', key: ['id'] },
{ name: 'couples', key: ['id'], sourceExpressions: { invite_code: 'upper("couples"."invite_code")' } },
{ name: 'profiles', key: ['id'], preserveNewer: true },
{ name: 'admin_users', key: ['id'] }, { name: 'master_keys', key: ['id'] },
{ name: 'categories', key: ['id'] }, { name: 'question_packs', key: ['id'] },
{ name: 'topics', key: ['id'] }, { name: 'pack_topics', key: ['pack_id', 'topic_id'] },
{ name: 'questions', key: ['id'] }, { name: 'app_config', key: ['id'], replace: true },
{ name: 'ai_config', key: ['id'], replace: true },
{ name: 'questions', key: ['id'], deferredColumns: ['inverse_of'] },
{ name: 'content_reviews', key: ['id'] },
{ name: 'app_config', key: ['id'], replace: true, preserveExistingOnInitial: true },
{ name: 'ai_config', key: ['id'], replace: true, preserveExistingOnInitial: true },
{ name: 'dare_packs', key: ['id'] }, { name: 'dares', key: ['id'] },
{ name: 'sent_dares', key: ['id'] }, { name: 'dare_messages', key: ['id'] },
{ name: 'couple_packs', key: ['couple_id', 'pack_id'] }, { name: 'responses', key: ['id'] },
@@ -18,8 +35,21 @@ export const TABLES: TableSpec[] = [
{ name: 'notification_preferences', key: ['user_id'] }, { name: 'feedback', key: ['id'] },
{ name: 'subscriptions', key: ['id'] }, { name: 'revenuecat_webhook_events', key: ['id'] },
{ name: 'redemption_codes', key: ['id'] }, { name: 'code_redemptions', key: ['id'] },
{ name: 'feature_interests', key: ['user_id', 'feature'] }, { name: 'live_draw_sessions', key: ['couple_id'] },
{ name: 'catchup_reminder_tracking', key: ['user_id'] }, { name: 'audit_logs', key: ['id'] },
{ name: 'feature_interests', key: ['user_id', 'feature'], columnMap: { feature: 'feature_name' } },
{
name: 'live_draw_sessions', key: ['couple_id'],
sourceExpressions: {
updated_by: '(select p.id from profiles p where p.couple_id="live_draw_sessions"."couple_id" order by p.created_at,p.id limit 1)',
},
},
{ name: 'catchup_reminder_tracking', key: ['user_id'] },
{
name: 'audit_logs', key: ['id'],
sourceExpressions: {
actor_user_id: '"audit_logs"."admin_user_id"',
admin_user_id: '(select au.id from admin_users au where au.user_id="audit_logs"."admin_user_id" limit 1)',
},
},
];
const quote = (value: string) => `"${value.replaceAll('"', '""')}"`;
@@ -27,18 +57,42 @@ async function columns(pool: Pool | PoolClient, table: string): Promise<string[]
const result = await pool.query<{ column_name: string }>(`select column_name from information_schema.columns where table_schema=current_schema() and table_name=$1 order by ordinal_position`, [table]);
return result.rows.map((row) => row.column_name);
}
async function jsonColumns(pool: Pool | PoolClient, table: string): Promise<Set<string>> {
const result = await pool.query<{ column_name: string }>(`select column_name from information_schema.columns where table_schema=current_schema() and table_name=$1 and data_type in ('json','jsonb')`, [table]);
return new Set(result.rows.map((row) => row.column_name));
}
async function count(pool: Pool | PoolClient, table: string): Promise<number> {
return Number((await pool.query<{ count: string }>(`select count(*) count from ${quote(table)}`)).rows[0]?.count ?? 0);
}
async function upsertRows(client: PoolClient, spec: TableSpec, names: string[], rows: Record<string, unknown>[]): Promise<number> {
if (spec.replace) await client.query(`delete from ${quote(spec.name)}`);
async function upsertRows(client: PoolClient, spec: TableSpec, names: string[], rows: Record<string, unknown>[], jsonNames: Set<string>, finalSync: boolean): Promise<number> {
if (spec.preserveExistingOnInitial && !finalSync && await count(client, spec.name)) return 0;
if (spec.replace && finalSync) await client.query(`delete from ${quote(spec.name)}`);
if (!rows.length) return 0;
const update = names.filter((name) => !spec.key.includes(name));
const deferred = new Set(spec.deferredColumns ?? []);
const immediateNames = names.filter((name) => !deferred.has(name));
const update = immediateNames.filter((name) => !spec.key.includes(name));
const conflict = spec.key.map(quote).join(',');
const action = update.length ? `do update set ${update.map((name) => `${quote(name)}=excluded.${quote(name)}`).join(',')}` : 'do nothing';
const freshness = spec.preserveNewer && names.includes('updated_at')
? ` where ${quote(spec.name)}.${quote('updated_at')} is null or (excluded.${quote('updated_at')} is not null and excluded.${quote('updated_at')}>${quote(spec.name)}.${quote('updated_at')})`
: '';
const action = update.length ? `do update set ${update.map((name) => `${quote(name)}=excluded.${quote(name)}`).join(',')}${freshness}` : 'do nothing';
for (const row of rows) {
await client.query(`insert into ${quote(spec.name)} (${names.map(quote).join(',')}) values (${names.map((_, index) => `$${index + 1}`).join(',')}) on conflict (${conflict}) ${action}`, names.map((name) => row[name]));
await client.query(
`insert into ${quote(spec.name)} (${immediateNames.map(quote).join(',')}) values (${immediateNames.map((_, index) => `$${index + 1}`).join(',')}) on conflict (${conflict}) ${action}`,
immediateNames.map((name) => jsonNames.has(name) && row[name] !== null ? JSON.stringify(row[name]) : row[name]),
);
}
if (deferred.size) {
const deferredNames = names.filter((name) => deferred.has(name));
for (const row of rows) {
const values = deferredNames.map((name) => jsonNames.has(name) && row[name] !== null ? JSON.stringify(row[name]) : row[name]);
const whereValues = spec.key.map((name) => row[name]);
await client.query(
`update ${quote(spec.name)} set ${deferredNames.map((name, index) => `${quote(name)}=$${index + 1}`).join(',')} where ${spec.key.map((name, index) => `${quote(name)}=$${deferredNames.length + index + 1}`).join(' and ')}`,
[...values, ...whereValues],
);
}
}
return rows.length;
}
@@ -59,16 +113,24 @@ async function pruneRows(client: PoolClient, spec: TableSpec, sourceRows: Record
export async function migrateDatabase(source: Pool, target: Pool, options: DatabaseMigrationOptions): Promise<TableResult[]> {
const results: TableResult[] = [];
for (const spec of TABLES) {
const [sourceColumns, targetColumns] = await Promise.all([columns(source, spec.name), columns(target, spec.name)]);
const [sourceColumns, targetColumns, targetJsonColumns] = await Promise.all([columns(source, spec.name), columns(target, spec.name), jsonColumns(target, spec.name)]);
if (!sourceColumns.length || !targetColumns.length) {
results.push({ table: spec.name, sourceRows: 0, targetRows: targetColumns.length ? await count(target, spec.name) : 0, importedRows: 0, prunedRows: 0, status: 'skipped', note: !sourceColumns.length ? 'source table absent' : 'target table absent' });
continue;
}
const common = targetColumns.filter((name) => sourceColumns.includes(name));
const sourceOnly = sourceColumns.filter((name) => !targetColumns.includes(name));
const sourceName = (targetName: string) => spec.columnMap?.[targetName] ?? targetName;
const common = targetColumns.filter((name) => sourceColumns.includes(sourceName(name)) || Boolean(spec.sourceExpressions?.[name]));
const consumedSourceColumns = new Set(common.map(sourceName).filter((name) => sourceColumns.includes(name)));
const sourceOnly = sourceColumns.filter((name) => !consumedSourceColumns.has(name));
if (!spec.key.every((key) => common.includes(key))) throw new Error(`${spec.name} is missing required migration key columns`);
const sourceRows = (await source.query<Record<string, unknown>>(`select ${common.map(quote).join(',')} from ${quote(spec.name)}`)).rows;
let importedRows = 0; let prunedRows = 0;
const selections = common.map((targetName) => {
const expression = spec.sourceExpressions?.[targetName] ?? quote(sourceName(targetName));
return sourceName(targetName) === targetName && !spec.sourceExpressions?.[targetName]
? expression
: `${expression} as ${quote(targetName)}`;
});
const sourceRows = (await source.query<Record<string, unknown>>(`select ${selections.join(',')} from ${quote(spec.name)}`)).rows;
let importedRows = 0; let prunedRows = 0; let reconciledProfileCouples = 0;
const skip = !options.finalSync && options.checkpoint.completedTables.includes(spec.name);
if (!options.dryRun && !skip) {
const client = await target.connect();
@@ -77,7 +139,19 @@ export async function migrateDatabase(source: Pool, target: Pool, options: Datab
// Historical imports must not enqueue user notifications, classification,
// Discord events, or other operational side effects.
await client.query("set local sauci.suppress_operations = 'on'");
importedRows = await upsertRows(client, spec, common, sourceRows);
importedRows = await upsertRows(client, spec, common, sourceRows, targetJsonColumns, options.finalSync);
if (spec.name === 'profiles' && !options.finalSync) {
for (const row of sourceRows) {
if (!row.couple_id) continue;
const reconciled = await client.query(
`update profiles p set couple_id=$2
where p.id=$1 and p.couple_id is distinct from $2
and (select count(*) from profiles member where member.couple_id=p.couple_id)<=1`,
[row.id, row.couple_id],
);
reconciledProfileCouples += reconciled.rowCount ?? 0;
}
}
if (options.finalSync && options.prune) prunedRows = await pruneRows(client, spec, sourceRows);
await client.query('commit');
} catch (error) { await client.query('rollback'); throw error; } finally { client.release(); }
@@ -85,8 +159,23 @@ export async function migrateDatabase(source: Pool, target: Pool, options: Datab
await options.saveCheckpoint?.(options.checkpoint);
}
const targetRows = await count(target, spec.name);
const notes = [skip ? 'resumed from checkpoint' : '', sourceOnly.length ? `legacy-only columns intentionally omitted: ${sourceOnly.join(', ')}` : ''].filter(Boolean);
results.push({ table: spec.name, sourceRows: sourceRows.length, targetRows, importedRows, prunedRows, status: sourceRows.length === targetRows ? 'equal' : 'different', ...(notes.length ? { note: notes.join('; ') } : {}) });
const targetKeys = targetColumns.length
? (await target.query<Record<string, unknown>>(`select ${spec.key.map(quote).join(',')} from ${quote(spec.name)}`)).rows
: [];
const keyValue = (row: Record<string, unknown>) => JSON.stringify(spec.key.map((key) => row[key]));
const targetKeySet = new Set(targetKeys.map(keyValue));
const coveredSourceRows = sourceRows.filter((row) => targetKeySet.has(keyValue(row))).length;
const preservedSingleton = Boolean(spec.preserveExistingOnInitial && !options.finalSync && targetRows && !coveredSourceRows);
const parity = options.finalSync ? sourceRows.length === targetRows : preservedSingleton || coveredSourceRows === sourceRows.length;
const preservedTargetRows = Math.max(0, targetRows - coveredSourceRows);
const notes = [
skip ? 'resumed from checkpoint' : '',
preservedSingleton ? 'kept existing live target singleton during initial merge' : '',
reconciledProfileCouples ? `restored ${reconciledProfileCouples} established couple membership(s) over one-person target couples` : '',
!options.finalSync && preservedTargetRows ? `preserved ${preservedTargetRows} target-only row(s)` : '',
sourceOnly.length ? `legacy-only columns intentionally omitted: ${sourceOnly.join(', ')}` : '',
].filter(Boolean);
results.push({ table: spec.name, sourceRows: sourceRows.length, targetRows, importedRows, prunedRows, status: parity ? 'equal' : 'different', ...(notes.length ? { note: notes.join('; ') } : {}) });
}
return results;
}