#!/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);
});