#!/usr/bin/env node
/**
 * Fluxer: PostgreSQL KV table → ScyllaDB/Cassandra migration
 *
 * Prerequisites:
 *   1. ScyllaDB is running and the 'fluxer' keyspace exists
 *   2. fluxer_api has started once against ScyllaDB so all tables are created
 *   3. PostgreSQL is still up and serving the original data
 *
 * Usage (prefer an --env-file with docker; values are taken literally):
 *   PG_HOST=CHANGEME PG_USER=fluxer PG_PASSWORD=... PG_DATABASE=fluxer \
 *   SCYLLA_HOSTS=CHANGEME:9042 SCYLLA_DC=CHANGEME \
 *   CASSANDRA_USERNAME=fluxer CASSANDRA_PASSWORD=... \
 *   npm run dry-run      # verify without writing
 *   npm run migrate      # real migration
 */

import pg from 'pg';
import cassandra from 'cassandra-driver';

const { Client: PgClient } = pg;
const { types: cassTypes, auth: cassAuth } = cassandra;

// ── Configuration ──────────────────────────────────────────────────────────
// PostgreSQL connection comes from PG_HOST/PG_PORT/PG_USER/PG_PASSWORD/PG_DATABASE (see main()).
const SCYLLA_HOSTS = (process.env.SCYLLA_HOSTS ?? 'CHANGEME:9042').split(',');
const SCYLLA_DC    = process.env.SCYLLA_DC    ?? 'datacenter1';
const KEYSPACE     = process.env.CASSANDRA_KEYSPACE ?? 'fluxer';
const KV_TABLE     = process.env.KV_TABLE     ?? 'fluxer_kv';
const BATCH_SIZE   = parseInt(process.env.BATCH_SIZE  ?? '200');
const CONCURRENCY  = parseInt(process.env.CONCURRENCY ?? '16');

// Fluxer snowflake epoch and message bucket window (10 days in ms)
const BUCKET_DURATION_MS = 864_000_000n;

function messageBucket(messageIdStr) {
  return (BigInt(messageIdStr) >> 22n) / BUCKET_DURATION_MS;
}

// ── Column-type introspection ───────────────────────────────────────────────
// The generic transform can't know a destination column's real CQL type (e.g.
// varint vs bigint, or map<> vs a plain array), so we read the live schema from
// system_schema.columns and coerce each value to match its target column.
let colTypes = {}; // { tableName: { columnName: cqlType } }

// Fixed node (6 bytes) + clock (2 bytes) so timeuuids reconstructed from a stored
// timestamp are DETERMINISTIC — re-running upserts the same row instead of creating
// a duplicate. Postgres' KV row_key already guarantees a unique (pk, event_id)
// timestamp per partition, so ticks=0 won't collide within a partition.
const TUUID_NODE  = Buffer.from('fluxer'); // exactly 6 bytes
const TUUID_CLOCK = Buffer.from('kv');     // exactly 2 bytes

async function loadColumnTypes(cassClient, keyspace) {
  const res = await cassClient.execute(
    `SELECT table_name, column_name, type FROM system_schema.columns WHERE keyspace_name = ?`,
    [keyspace], { prepare: true }
  );
  const map = {};
  for (const r of res.rows) (map[r.table_name] ??= {})[r.column_name] = r.type;
  return map;
}

// Coerce a transformed value to satisfy its destination CQL column type.
function coerceForColumn(value, cqlType) {
  if (value == null || !cqlType) return value;
  // varint: the driver rejects Long (that's bigint); it accepts a string.
  if (cqlType === 'varint') return value?.toString?.() ?? String(value);
  // map<...>: source may give an array of [k,v] pairs or a plain object → Map.
  if (cqlType.startsWith('map<')) {
    if (value instanceof Map) return value;
    if (Array.isArray(value)) return new Map(value);
    if (typeof value === 'object') return new Map(Object.entries(value));
    return value;
  }
  // timeuuid: Postgres stored these as a plain timestamp (the KV model has no
  // timeuuid). Reconstruct a time-ordered UUID from the date so clustering order
  // is preserved; a string value is already a uuid and passes through.
  if (cqlType === 'timeuuid') {
    return value instanceof Date
      ? cassTypes.TimeUuid.fromDate(value, 0, TUUID_NODE, TUUID_CLOCK)
      : value;
  }
  // uuid: expect a real uuid string already.
  if (cqlType === 'uuid') return value;
  return value;
}

// ── Type transformation ─────────────────────────────────────────────────────
// Fluxer's PostgreSQL KV layer encodes non-JSON-native types with a
// __fluxer_type tag. Map them to the types cassandra-driver expects.
function transformValue(v) {
  if (v === null || v === undefined) return v;
  if (Array.isArray(v)) return v.map(transformValue);
  if (typeof v === 'object') {
    if ('__fluxer_type' in v) {
      switch (v.__fluxer_type) {
        case 'bigint':
          // Snowflake IDs are 64-bit — use Long to avoid precision loss
          return cassTypes.Long.fromString(String(v.value));
        case 'date':
          return new Date(v.value);
        case 'buffer':
          return Buffer.from(v.value, 'base64');
        case 'set':
          return Array.isArray(v.value) ? v.value.map(transformValue) : [];
        case 'map': {
          const out = {};
          for (const [k, val] of Object.entries(v.value ?? {})) {
            out[String(transformValue(k))] = transformValue(val);
          }
          return out;
        }
        default:
          return v.value ?? null;
      }
    }
    // Plain nested object — recurse
    const out = {};
    for (const [k, val] of Object.entries(v)) {
      out[k] = transformValue(val);
    }
    return out;
  }
  return v;
}

// ── Concurrency helper ──────────────────────────────────────────────────────
async function runConcurrently(items, limit, fn) {
  for (let i = 0; i < items.length; i += limit) {
    const results = await Promise.allSettled(items.slice(i, i + limit).map(fn));
    const failed = results.filter(r => r.status === 'rejected');
    if (failed.length > 0) throw failed[0].reason;
  }
}

// ── Per-table migration ─────────────────────────────────────────────────────
async function migrateTable(pgClient, cassClient, tableName, dryRun) {
  const { rows: [{ count }] } = await pgClient.query(
    `SELECT COUNT(*) FROM ${KV_TABLE} WHERE table_name = $1`,
    [tableName]
  );
  const total = parseInt(count);

  if (total === 0) {
    console.log(`[${tableName}] empty, skipping`);
    return { migrated: 0, errors: 0 };
  }

  const seenBuckets = new Set();
  let migrated = 0;
  let errors   = 0;
  process.stdout.write(`[${tableName}] 0/${total}`);

  for (let offset = 0; offset < total; offset += BATCH_SIZE) {
    const { rows } = await pgClient.query(
      `SELECT row_data, expires_at
         FROM ${KV_TABLE}
        WHERE table_name = $1
        ORDER BY row_key
        LIMIT $2 OFFSET $3`,
      [tableName, BATCH_SIZE, offset]
    );

    await runConcurrently(rows, CONCURRENCY, async (row) => {
      try {
        const data = transformValue(row.row_data);

        // Messages: derive the bucket from the snowflake message_id and also
        // populate channel_message_buckets so the Rust shard can list buckets.
        if (tableName === 'messages' && data.message_id != null) {
          const bucket = messageBucket(String(data.message_id));
          data.bucket = cassTypes.Long.fromString(String(bucket));

          const key = `${data.channel_id}:${bucket}`;
          if (!seenBuckets.has(key)) {
            seenBuckets.add(key);
            if (!dryRun) {
              await cassClient.execute(
                `INSERT INTO ${KEYSPACE}.channel_message_buckets (channel_id, bucket) VALUES (?, ?)`,
                [data.channel_id, data.bucket],
                { prepare: true }
              ).catch(() => {}); // idempotent
            }
          }
        }

        const cols   = Object.keys(data);
        const types  = colTypes[tableName] ?? {};
        const vals   = cols.map(c => coerceForColumn(data[c], types[c]));
        const phs    = cols.map(() => '?').join(', ');
        const ttlSec = row.expires_at
          ? Math.max(1, Math.floor((new Date(row.expires_at).getTime() - Date.now()) / 1000))
          : null;

        let query = `INSERT INTO ${KEYSPACE}.${tableName} (${cols.join(', ')}) VALUES (${phs})`;
        if (ttlSec) query += ` USING TTL ${ttlSec}`;

        if (!dryRun) {
          await cassClient.execute(query, vals, { prepare: true });
        }
        migrated++;
      } catch (err) {
        errors++;
        // Log first few errors per table to surface schema mismatches
        if (errors <= 3) console.error(`\n  [${tableName}] row error: ${err.message}`);
      }
    });

    process.stdout.write(`\r[${tableName}] ${migrated}/${total} (${Math.round(migrated / total * 100)}%)`);
  }

  const errStr = errors > 0 ? ` — ${errors} errors` : '';
  console.log(` ✓${errStr}`);
  return { migrated, errors };
}

// Migrate in dependency order; messages last (needs bucket table populated)
const PRIORITY_ORDER = [
  'users', 'guilds', 'roles', 'channels', 'guild_members',
  'auth_sessions', 'read_states', 'guild_bans', 'emojis', 'stickers',
  'attachments', 'webhooks', 'invites',
];

// ── Main ────────────────────────────────────────────────────────────────────
async function main() {
  const dryRun = process.argv.includes('--dry-run');
  if (dryRun) console.log('=== DRY RUN — nothing will be written to ScyllaDB ===\n');

  // PostgreSQL — component-based config so passwords with URL-special chars
  // (@, #, %, etc.) don't need percent-encoding; node-postgres handles them raw.
  const pgClient = new PgClient({
    host:     process.env.PG_HOST ?? 'CHANGEME',
    port:     parseInt(process.env.PG_PORT ?? '5432'),
    user:     process.env.PG_USER ?? 'fluxer',
    password: process.env.PG_PASSWORD ?? '',
    database: process.env.PG_DATABASE ?? 'fluxer',
  });
  await pgClient.connect();
  console.log(`PostgreSQL  ✓  ${process.env.PG_USER ?? 'fluxer'}@${process.env.PG_HOST ?? 'CHANGEME'}:${process.env.PG_PORT ?? '5432'}/${process.env.PG_DATABASE ?? 'fluxer'}`);

  // ScyllaDB
  const authProvider = process.env.CASSANDRA_USERNAME
    ? new cassAuth.PlainTextAuthProvider(
        process.env.CASSANDRA_USERNAME,
        process.env.CASSANDRA_PASSWORD ?? ''
      )
    : undefined;

  const cassClient = new cassandra.Client({
    contactPoints: SCYLLA_HOSTS,
    localDataCenter: SCYLLA_DC,
    keyspace: KEYSPACE,
    authProvider,
    socketOptions: { connectTimeout: 30_000 },
    queryOptions: { consistency: cassTypes.consistencies.localQuorum },
  });
  await cassClient.connect();
  colTypes = await loadColumnTypes(cassClient, KEYSPACE);
  console.log(`ScyllaDB    ✓  ${SCYLLA_HOSTS.join(', ')} / keyspace=${KEYSPACE} / ${Object.keys(colTypes).length} tables introspected\n`);

  // Discover tables
  const { rows: tableRows } = await pgClient.query(
    `SELECT table_name, COUNT(*) AS cnt
       FROM ${KV_TABLE}
      GROUP BY table_name
      ORDER BY table_name`
  );

  const allTables = tableRows.map(r => r.table_name);
  const ordered = [
    ...PRIORITY_ORDER.filter(t => allTables.includes(t)),
    ...allTables.filter(t => !PRIORITY_ORDER.includes(t) && t !== 'messages'),
    ...(allTables.includes('messages') ? ['messages'] : []),
  ];

  const width = Math.max(...tableRows.map(r => r.table_name.length));
  console.log('Tables to migrate:');
  for (const r of tableRows) {
    console.log(`  ${r.table_name.padEnd(width)}  ${r.cnt} rows`);
  }
  console.log();

  let totalMigrated = 0;
  let totalErrors   = 0;

  for (const tableName of ordered) {
    try {
      const { migrated, errors } = await migrateTable(pgClient, cassClient, tableName, dryRun);
      totalMigrated += migrated;
      totalErrors   += errors;
    } catch (err) {
      console.error(`\n[${tableName}] FAILED: ${err.message}`);
      console.error('  Ensure fluxer_api has run once with FLUXER_DATABASE_BACKEND=cassandra');
      console.error('  to auto-create the Cassandra schema before running this script.\n');
      totalErrors++;
    }
  }

  await pgClient.end();
  await cassClient.shutdown();

  console.log('\n' + '─'.repeat(55));
  console.log(`Total migrated: ${totalMigrated} rows | Errors: ${totalErrors}`);
  if (dryRun) console.log('(dry run — no data was written)');
  if (totalErrors > 0) process.exit(1);
}

main().catch(err => { console.error(err); process.exit(1); });