// @ts-check
/**
* MariaDB → PostgreSQL Data Migration Script
*
* Reads all data from MariaDB (source) and inserts into PostgreSQL (target).
* Handles type conversions: binary(16)→uuid, longtext→jsonb, etc.
*
* Usage:
* node scripts/migrate-pg.js [--dry-run] [--tables ObjectBase,Member,Links]
*
* Loads MariaDB credentials from Vault (same path as the backend).
* PostgreSQL credentials from env or defaults (local test container).
*
* ENV:
* PGHOST, PGUSER, PGPASSWORD, PGDATABASE (PG target, optional — defaults to localhost:5433)
*/
import { createPool as mariadbPool } from 'mariadb'
import pg from 'pg'
import { loadSecretsFromVault } from '@commtool/vault-secrets'
const BATCH_SIZE = 500
// ---------------------------------------------------------------------------
// UUID helpers
// ---------------------------------------------------------------------------
/**
* MariaDB binary(16) mixed-endian buffer → standard uuid string.
* @param {Buffer} buf
* @returns {string}
*/
const bufferToUUID = (buf) => {
if (!buf || buf.length !== 16) return ''
const d = [...buf]
return (
d[4].toString(16).padStart(2, '0') + d[5].toString(16).padStart(2, '0') +
d[6].toString(16).padStart(2, '0') + d[7].toString(16).padStart(2, '0') + '-' +
d[2].toString(16).padStart(2, '0') + d[3].toString(16).padStart(2, '0') + '-' +
d[0].toString(16).padStart(2, '0') + d[1].toString(16).padStart(2, '0') + '-' +
d[8].toString(16).padStart(2, '0') + d[9].toString(16).padStart(2, '0') + '-' +
d[10].toString(16).padStart(2, '0') + d[11].toString(16).padStart(2, '0') +
d[12].toString(16).padStart(2, '0') + d[13].toString(16).padStart(2, '0') +
d[14].toString(16).padStart(2, '0') + d[15].toString(16).padStart(2, '0')
)
}
// ---------------------------------------------------------------------------
// Type conversion map
// ---------------------------------------------------------------------------
/**
* Known binary(16) columns that need uuid conversion.
*/
const UUID_COLUMNS = new Set([
'UID', 'UIDBelongsTo', 'UIDTarget', 'UIDuser', 'UIDRoot', 'UIDObjectID',
'UIDoldTarget', 'UIDnewTarget', 'UIDOrganization', 'UIDFiscalYear',
'UIDCostCenter', 'UIDDefaultAccount', 'UIDTransaction', 'UIDAccount',
])
/**
* Known longtext JSON columns.
*/
const JSON_COLUMNS = new Set(['Data'])
/**
* Known binary/bytea columns.
*/
const BYTEA_COLUMNS = new Set(['IBANHash', 'ContentHash'])
/**
* Known timestamp columns that need timestamptz handling.
*/
const TIMESTAMP_COLUMNS = new Set(['ValidFrom', 'ValidUntil', 'CreatedAt', 'Timestamp', 'appliedAt'])
// ---------------------------------------------------------------------------
// Main migration
// ---------------------------------------------------------------------------
const migrate = async () => {
const args = process.argv.slice(2)
const dryRun = args.includes('--dry-run')
const tableFilter = args.find(a => a.startsWith('--tables='))
const onlyTables = tableFilter ? tableFilter.split('=')[1].split(',') : null
// Load MariaDB credentials from Vault (same path as the backend)
console.log('Loading secrets from Vault...')
await loadSecretsFromVault()
// Tables in dependency order
const ALL_TABLES = [
{ name: 'ObjectBase', key: 'UID' },
{ name: 'Member', key: 'UID', childOf: 'ObjectBase' },
{ name: 'Links', key: 'UID', childOf: 'ObjectBase' },
{ name: 'Visible', key: 'UID', childOf: 'ObjectBase' },
{ name: 'Transactions', key: 'UID' },
{ name: 'TransactionLines', key: 'UID' },
{ name: 'AIEmbeddings', key: 'UID' },
{ name: 'TreeQueue', key: null },
{ name: 'eventLog', key: null },
{ name: 'dbVersion', key: null },
]
const tables = onlyTables
? ALL_TABLES.filter(t => onlyTables.includes(t.name))
: ALL_TABLES
// Connect to MariaDB (credentials from Vault → process.env)
const DB_HOST = process.env.DB_HOST || 'mariadb'
const DB_USER = process.env.DB_USER || 'root'
const DB_PASS = process.env.DB_PASS || ''
const DB_DATABASE = process.env.DB_DATABASE || 'member'
console.log(`Connecting to MariaDB at ${DB_HOST}...`)
const mdb = await mariadbPool({
host: DB_HOST,
user: DB_USER,
password: DB_PASS,
database: DB_DATABASE,
connectionLimit: 5,
supportBigNumbers: true,
bigNumberStrings: true,
})
// Connect to PostgreSQL (local test container by default)
const PG_HOST = process.env.PGHOST || 'localhost'
const PG_USER = process.env.PGUSER || 'members'
const PG_PASS = process.env.PGPASSWORD || 'secret'
const PG_DATABASE = process.env.PGDATABASE || 'members'
console.log(`Connecting to PostgreSQL at ${PG_HOST}...`)
const pgPool = new pg.Pool({
host: PG_HOST,
port: parseInt(process.env.PGPORT || '5433', 10),
user: PG_USER,
password: PG_PASS,
database: PG_DATABASE,
max: 5,
})
// Test connections
const mConn = await mdb.getConnection()
await mConn.query('SELECT 1')
mConn.release()
const pgClient = await pgPool.connect()
await pgClient.query('SELECT 1')
pgClient.release()
console.log('Both connections OK')
// -----------------------------------------------------------------------
// Migrate each table
// -----------------------------------------------------------------------
for (const table of tables) {
console.log(`\n--- Migrating ${table.name} ---`)
// Check if PG table exists
const { rowCount: pgExists } = await pgClient.query(
`SELECT 1 FROM information_schema.tables WHERE table_name = $1 AND table_schema = 'public'`,
[table.name.toLowerCase()]
)
if (pgExists === 0) {
console.log(` SKIP: ${table.name} does not exist in PostgreSQL`)
continue
}
// Read all rows from MariaDB
const rows = await mdb.query(`SELECT * FROM \`${table.name}\``)
console.log(` Source rows: ${rows.length}`)
if (rows.length === 0) continue
// Get PG column info
const { rows: pgCols } = await pgClient.query(
`SELECT column_name, data_type, udt_name FROM information_schema.columns WHERE table_name = $1 AND table_schema = 'public' ORDER BY ordinal_position`,
[table.name.toLowerCase()]
)
const pgColMap = new Map(pgCols.map(c => [c.column_name, c]))
// Build column list for PG (only columns that exist in PG)
const mariadbCols = Object.keys(rows[0])
const validCols = mariadbCols.filter(col => pgColMap.has(col.toLowerCase()))
// Handle ObjectBase: add OrgUID if it doesn't exist in source
if (table.name === 'ObjectBase' && !validCols.includes('OrgUID')) {
// ... we'll set a default
}
// Prepare PG insert SQL with ON CONFLICT DO NOTHING for idempotency
const pgColsList = validCols.filter(c => c.toLowerCase() !== 'tuid' && c.toLowerCase() !== 'tuidbelongsto' && c.toLowerCase() !== 'tuidtarget' && c.toLowerCase() !== 'tuiduser')
const placeholders = pgColsList.map((_, i) => `$${i + 1}`)
const insertSQL = `INSERT INTO ${table.name} (${pgColsList.map(c => `"${c}"`).join(', ')}) VALUES (${placeholders.join(', ')}) ON CONFLICT DO NOTHING`
// Process in batches
let inserted = 0
let errors = 0
for (let i = 0; i < rows.length; i += BATCH_SIZE) {
const batch = rows.slice(i, i + BATCH_SIZE)
for (const row of batch) {
const pgRow = pgColsList.map(col => {
const val = row[col]
// Handle null
if (val === null || val === undefined) return null
// UUID columns: binary(16) buffer → uuid string
if (UUID_COLUMNS.has(col) && Buffer.isBuffer(val)) {
return bufferToUUID(val)
}
// JSON columns: string → JSON.parse'd then JSON.stringify'd for jsonb
if (JSON_COLUMNS.has(col)) {
if (typeof val === 'string') {
try { return JSON.parse(val) }
catch { return val }
}
return val
}
// BYTEA columns: Buffer → Buffer (pg accepts Buffer for bytea)
if (BYTEA_COLUMNS.has(col) && Buffer.isBuffer(val)) {
return val
}
// Timestamps: Date or string → ISO string
if (TIMESTAMP_COLUMNS.has(col) && val instanceof Date) {
return val.toISOString()
}
// Geometry: WKB buffer → hex for ST_GeomFromWKB
if (col === 'Geo' && Buffer.isBuffer(val)) {
return { wkb: val.toString('hex') }
}
return val
})
if (dryRun) {
inserted++
continue
}
try {
await pgClient.query(insertSQL, pgRow)
inserted++
} catch (err) {
errors++
if (errors <= 5) {
console.error(` ERROR on ${table.name} row ${i + inserted + errors}: ${err.message}`)
}
}
}
const total = Math.min(i + BATCH_SIZE, rows.length)
process.stdout.write(` ${total}/${rows.length} rows processed (${inserted} inserted, ${errors} errors)\r`)
}
console.log(`\n DONE: ${inserted} inserted, ${errors} errors`)
}
// -----------------------------------------------------------------------
// Handle history tables
// -----------------------------------------------------------------------
if (!onlyTables || onlyTables.includes('history')) {
const histTables = ['ObjectBase_history', 'Links_history', 'Transactions_history', 'TransactionLines_history']
for (const table of histTables) {
// Read from MariaDB system versioning
// MariaDB stores history in the same table (system-versioned),
// so we need to read FROM table FOR SYSTEM_TIME ALL
const baseName = table.replace('_history', '')
const { rowCount: pgExists } = await pgClient.query(
`SELECT 1 FROM information_schema.tables WHERE table_name = $1 AND table_schema = 'public'`,
[table.toLowerCase()]
)
if (pgExists === 0) {
console.log(`\n--- ${table}: not found in PG, skipping ---`)
continue
}
// Read history rows from MariaDB partition tables
// MariaDB stores history in partition pminus* tables
// We read the current version first
let allRows = []
try {
const mdbConn = await mdb.getConnection()
// Read using FOR SYSTEM_TIME ALL to get all historical versions
const mdbTable = baseName
allRows = await mdbConn.query(`SELECT * FROM \`${mdbTable}\` FOR SYSTEM_TIME ALL`)
mdbConn.release()
} catch (e) {
console.log(` NOTE: ${baseName} FOR SYSTEM_TIME ALL not supported or empty, trying alternate read`)
// Fallback: read only current rows
allRows = []
}
// Separate current vs history
// Actually, since MariaDB stores everything in one table for system-versioned,
// FOR SYSTEM_TIME ALL returns ALL versions INCLUDING current.
// We need to insert all but the current "latest" version into _history.
// The current version in PG is what's in the main table.
// So insert everything into _history, then remove duplicates (the main table rows).
console.log(`\n--- ${table}: ${allRows.length} total versions found ---`)
if (allRows.length === 0) continue
let histInserted = 0
let histErrors = 0
for (let i = 0; i < allRows.length; i += BATCH_SIZE) {
const batch = allRows.slice(i, i + BATCH_SIZE)
for (const row of batch) {
const pgRow = validCols.map(col => {
const val = row[col]
if (val === null || val === undefined) return null
if (UUID_COLUMNS.has(col) && Buffer.isBuffer(val)) return bufferToUUID(val)
if (JSON_COLUMNS.has(col)) {
if (typeof val === 'string') {
try { return JSON.parse(val) }
catch { return val }
}
return val
}
if (BYTEA_COLUMNS.has(col) && Buffer.isBuffer(val)) return val
if (TIMESTAMP_COLUMNS.has(col) && val instanceof Date) return val.toISOString()
return val
})
if (dryRun) { histInserted++; continue }
try {
await pgClient.query(
`INSERT INTO ${table} (${pgColsList.map(c => `"${c}"`).join(', ')}) VALUES (${placeholders.join(', ')}) ON CONFLICT DO NOTHING`,
pgRow
)
histInserted++
} catch (err) {
histErrors++
}
}
process.stdout.write(` ${Math.min(i + BATCH_SIZE, allRows.length)}/${allRows.length} history rows\r`)
}
console.log(`\n HISTORY DONE: ${histInserted} inserted, ${histErrors} errors`)
}
}
// Cleanup
await mdb.end()
await pgPool.end()
console.log('\n=== Migration complete ===')
if (dryRun) console.log('(dry run - no data was written)')
}
migrate().catch(err => {
console.error('\nMigration failed:', err)
process.exit(1)
})