feat: support postgres autonomy runtime

2026-08-17 13:43:50 +01:00
parent 4aa2367971
commit f0b598a3b8
11 changed files with 301 additions and 24 deletions
@@ -0,0 +1,120 @@
+#!/usr/bin/env node
+/*
+ * Re-sync one logical SQLite database into an existing PostgreSQL schema.
+ * Unlike the initial migration, this updates conflicting primary-key rows so
+ * mutable runtime state such as job status, replay cursors, and broker snapshots
+ * can be cut over safely.
+ */
+const Database = require('better-sqlite3');
+const { Pool } = require('pg');
+
+const sqlitePath = process.env.SQLITE_PATH || process.env.SQLITE_INTELLIGENCE_PATH || '/data/intelligence.sqlite';
+const schema = process.env.POSTGRES_SCHEMA || 'intelligence';
+const connectionString = process.env.DATABASE_URL;
+if (!connectionString) throw new Error('DATABASE_URL is required');
+
+const batchSize = Math.max(1, Number(process.env.POSTGRES_SYNC_BATCH_SIZE) || 500);
+const sqlite = new Database(sqlitePath, { readonly: true });
+const pool = new Pool({ connectionString, max: 2 });
+
+function quote(name) { return `"${String(name).replaceAll('"', '""')}"`; }
+function pgType(sqliteType) {
+  const type = String(sqliteType || '').toUpperCase();
+  if (type.includes('INT')) return 'BIGINT';
+  if (type.includes('REAL') || type.includes('FLOA') || type.includes('DOUB')) return 'DOUBLE PRECISION';
+  if (type.includes('BLOB')) return 'BYTEA';
+  return 'TEXT';
+}
+function normalizeValue(value) {
+  if (typeof value === 'string' && value.includes('\u0000')) return value.replace(/\u0000/g, '');
+  return value ?? null;
+}
+
+function sourceTables() {
+  return sqlite.prepare("SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%' ORDER BY name").all().map((row) => row.name);
+}
+
+function columnsFor(table) {
+  return sqlite.prepare(`PRAGMA table_info(${quote(table)})`).all();
+}
+
+async function ensureTable(client, table, columns) {
+  const primary = columns.filter((column) => column.pk).sort((a, b) => a.pk - b.pk).map((column) => quote(column.name));
+  const definitions = columns.map((column) => `${quote(column.name)} ${pgType(column.type)}${column.notnull ? ' NOT NULL' : ''}`);
+  if (primary.length) definitions.push(`PRIMARY KEY (${primary.join(', ')})`);
+  await client.query(`CREATE TABLE IF NOT EXISTS ${quote(schema)}.${quote(table)} (${definitions.join(', ')})`);
+}
+
+async function repairIdentity(client, table, columns) {
+  const id = columns.find((column) => column.pk === 1 && column.name === 'id' && /INT/i.test(column.type));
+  if (!id) return;
+  const seq = `${schema}_${table}_id_seq`;
+  await client.query(`CREATE SEQUENCE IF NOT EXISTS ${quote(schema)}.${quote(seq)}`);
+  await client.query(`ALTER TABLE ${quote(schema)}.${quote(table)} ALTER COLUMN id SET DEFAULT nextval('${quote(schema)}.${quote(seq)}')`);
+  await client.query(`ALTER SEQUENCE ${quote(schema)}.${quote(seq)} OWNED BY ${quote(schema)}.${quote(table)}.id`);
+  await client.query(`SELECT setval('${quote(schema)}.${quote(seq)}', COALESCE((SELECT MAX(id) FROM ${quote(schema)}.${quote(table)}), 0) + 1, false)`);
+}
+
+async function syncTable(client, table) {
+  const columns = columnsFor(table);
+  if (!columns.length) return;
+  await ensureTable(client, table, columns);
+  const names = columns.map((column) => column.name);
+  const pk = columns.filter((column) => column.pk).sort((a, b) => a.pk - b.pk).map((column) => column.name);
+  const sourceRows = sqlite.prepare(`SELECT COUNT(*) AS count FROM ${quote(table)}`).get().count;
+  if (!sourceRows) {
+    await repairIdentity(client, table, columns);
+    console.log(`[sync] ${schema}.${table}: empty`);
+    return;
+  }
+  if (!pk.length) {
+    const targetRows = await client.query(`SELECT COUNT(*)::bigint AS count FROM ${quote(schema)}.${quote(table)}`);
+    if (Number(targetRows.rows[0].count) === Number(sourceRows)) {
+      console.log(`[sync] ${schema}.${table}: no primary key, count matches (${sourceRows})`);
+      return;
+    }
+    throw new Error(`${schema}.${table} has no primary key and count differs; refusing ambiguous sync`);
+  }
+  const placeholders = names.map((_, index) => `$${index + 1}`).join(', ');
+  const updates = names.filter((name) => !pk.includes(name)).map((name) => `${quote(name)}=EXCLUDED.${quote(name)}`).join(', ');
+  const conflict = pk.map(quote).join(', ');
+  const sql = `INSERT INTO ${quote(schema)}.${quote(table)} (${names.map(quote).join(', ')}) VALUES (${placeholders}) ON CONFLICT (${conflict}) ${updates ? `DO UPDATE SET ${updates}` : 'DO NOTHING'}`;
+  const key = columns.find((column) => column.pk === 1 && /INT/i.test(column.type));
+  let cursor = 0;
+  let copied = 0;
+  console.log(`[sync] ${schema}.${table}: ${sourceRows} source rows`);
+  while (true) {
+    const rows = key
+      ? sqlite.prepare(`SELECT ${names.map(quote).join(', ')} FROM ${quote(table)} WHERE ${quote(key.name)} > ? ORDER BY ${quote(key.name)} LIMIT ?`).all(cursor, batchSize)
+      : sqlite.prepare(`SELECT ${names.map(quote).join(', ')} FROM ${quote(table)} LIMIT ? OFFSET ?`).all(batchSize, copied);
+    if (!rows.length) break;
+    await client.query('BEGIN');
+    try {
+      for (const row of rows) await client.query(sql, names.map((name) => normalizeValue(row[name])));
+      await client.query('COMMIT');
+    } catch (error) {
+      await client.query('ROLLBACK');
+      throw error;
+    }
+    copied += rows.length;
+    if (key) cursor = Number(rows[rows.length - 1][key.name]);
+    process.stdout.write(`\r[sync] ${schema}.${table}: ${copied}/${sourceRows}`);
+  }
+  await repairIdentity(client, table, columns);
+  console.log(`\r[sync] ${schema}.${table}: synced ${sourceRows}`);
+}
+
+(async () => {
+  const client = await pool.connect();
+  try {
+    await client.query(`CREATE SCHEMA IF NOT EXISTS ${quote(schema)}`);
+    for (const table of sourceTables()) await syncTable(client, table);
+  } finally {
+    client.release();
+    await pool.end();
+    sqlite.close();
+  }
+})().catch((error) => {
+  console.error('[sync] fatal:', error);
+  process.exit(1);
+});