Fluxer PostgreSQL→Cassandra migration tool

Type-aware migration from Fluxer's PostgreSQL fluxer_kv backend to the native Apache Cassandra schema, plus scylla-feature image builds and a full cutover runbook. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-23 11:47:06 -04:00
commit abae7f9f41
8 changed files with 565 additions and 0 deletions
@@ -0,0 +1,308 @@
+#!/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=10.1.10.36 PG_USER=fluxer PG_PASSWORD=... PG_DATABASE=fluxer \
+ *   SCYLLA_HOSTS=10.1.10.38:9042 SCYLLA_DC=atl \
+ *   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 ?? '10.1.10.X:9042').split(',');  // replace X
+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 ?? '10.1.10.36',
+    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 ?? '10.1.10.36'}:${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); });