From ae790dc6a2786b0813d58df01ea110e4974a6944 Mon Sep 17 00:00:00 2001 From: Justin Lin Date: Mon, 12 Oct 2026 10:26:29 +1100 Subject: [PATCH] feat: create migration script --- bun.lock | 2 + packages/database/package.json | 4 +- .../database/scripts/migrate-from-mongo.ts | 403 ++++++++++++++++++ 3 files changed, 408 insertions(+), 1 deletion(-) create mode 100644 packages/database/scripts/migrate-from-mongo.ts diff --git a/bun.lock b/bun.lock index 15d8208..72ffe0d 100644 --- a/bun.lock +++ b/bun.lock @@ -82,8 +82,10 @@ "dependencies": { "@prisma/orm-postgres": "^8.0.0-rc.17", "dotenv": "^18.0.7", + "temporal-polyfill": "^1.0.5", }, "devDependencies": { + "@odyssey/models": "workspace:*", "@prisma/cli-engine": "0.7.0", "@types/bun": "latest", "@types/node": "^26.6.5", diff --git a/packages/database/package.json b/packages/database/package.json index 00e3606..2ab68d5 100644 --- a/packages/database/package.json +++ b/packages/database/package.json @@ -3,6 +3,7 @@ "module": "src/index.ts", "type": "module", "devDependencies": { + "@odyssey/models": "workspace:*", "@prisma/cli-engine": "0.7.0", "@types/bun": "latest", "@types/node": "^26.6.5", @@ -16,6 +17,7 @@ }, "dependencies": { "@prisma/orm-postgres": "^8.0.0-rc.17", - "dotenv": "^18.0.7" + "dotenv": "^18.0.7", + "temporal-polyfill": "^1.0.5" } } diff --git a/packages/database/scripts/migrate-from-mongo.ts b/packages/database/scripts/migrate-from-mongo.ts new file mode 100644 index 0000000..600c63f --- /dev/null +++ b/packages/database/scripts/migrate-from-mongo.ts @@ -0,0 +1,403 @@ +/** + * One-shot migration of the legacy Mongo data into Postgres. + * + * MONGO_URL=... DATABASE_URL=... bun run scripts/migrate-from-mongo.ts [--dry-run] [--force] + * + * - Stop the server before running; this is a snapshot copy, not a sync. + * - Everything runs in ONE Postgres transaction, so a failure leaves Postgres untouched + * and the script can simply be re-run. + * - Refuses to run if any target table is non-empty, unless --force is given + * (--force does not delete anything; it only skips the check). + * - New ids are assigned here (sequential), so relations can be rewired from ObjectIds. + * The id sequences are advanced afterwards. + * - Rows that cannot be migrated (orphans, unknown providers) are written to + * migration-skipped.json and counted in the summary. + */ +import 'dotenv/config' +import { writeFileSync } from 'node:fs' +import { Temporal } from 'temporal-polyfill' +import { + connectDB, + User, + UserProfile, + Session, + Command, + LoginAttempt, +} from '@odyssey/models' +import { db } from '../src/prisma/db' + +const DRY_RUN = process.argv.includes('--dry-run') +const FORCE = process.argv.includes('--force') +const BATCH_SIZE = 1000 + +const LOGIN_METHODS = ['password', 'google', 'github'] as const +type LoginMethod = (typeof LOGIN_METHODS)[number] +const isLoginMethod = (v: string): v is LoginMethod => + (LOGIN_METHODS as readonly string[]).includes(v) + +type Tx = Parameters[0]>[0] + +const skipped: { + table: string + reason: string + id?: string + detail?: unknown +}[] = [] +const stats: Record< + string, + { source: number; inserted: number; skipped: number } +> = {} + +function track(table: string) { + return (stats[table] ??= { source: 0, inserted: 0, skipped: 0 }) +} + +function skip(table: string, reason: string, id?: unknown, detail?: unknown) { + track(table).skipped++ + skipped.push({ + table, + reason, + id: id === undefined ? undefined : String(id), + detail, + }) +} + +const instant = (d: Date) => Temporal.Instant.fromEpochMilliseconds(d.getTime()) +const instantOrNull = (d?: Date | null) => (d ? instant(d) : null) + +// Postgres text cannot hold NUL bytes, which some captured terminal output contains. +let nulStripped = 0 +function text(v: string): string +function text(v: string | null | undefined): string | null +function text(v: string | null | undefined) { + if (v == null) return null + if (!v.includes('\0')) return v + nulStripped++ + return v.replaceAll('\0', '') +} + +// Mongoose subdocuments carry _id; the Postgres graph column stores the bare tree. +interface GraphNode { + path: string + type: string + discovered: boolean + children?: GraphNode[] +} +function cleanGraph(node: any): GraphNode { + const out: GraphNode = { + path: node?.path ?? '/', + // legacy sessions used 'folder' for directories + type: node?.type === 'file' ? 'file' : 'directory', + discovered: node?.discovered ?? false, + } + if (node?.children?.length) out.children = node.children.map(cleanGraph) + return out +} + +// A lean Mongoose Map comes back as a plain object, a hydrated one as a Map. +const entries = (m: Map | Record | undefined) => + m instanceof Map ? [...m.entries()] : Object.entries(m ?? {}) + +async function insert(tx: Tx, table: string, plan: unknown, count: number) { + if (DRY_RUN || count === 0) { + track(table).inserted += count + return + } + const { affectedRows } = await tx.execute(plan as never) + track(table).inserted += Number(affectedRows) +} + +async function* batches(cursor: AsyncIterable, size: number) { + let buf: T[] = [] + for await (const item of cursor) { + buf.push(item) + if (buf.length >= size) { + yield buf + buf = [] + } + } + if (buf.length) yield buf +} + +async function migrate(tx: Tx) { + const t = tx.sql.public + const userIds = new Map() + const userIdsByUsername = new Map() + const sessionIds = new Map() + const seenEmails = new Set() + const seenOAuth = new Set() + let nextUserId = 1n + let nextSessionId = 1n + let nextCommandId = 1n + let nextAttemptId = 1n + + // --- User + UserLoginMethod + for await (const users of batches( + User.find().sort({ _id: 1 }).lean().cursor(), + BATCH_SIZE + )) { + const userRows: any[] = [] + const methodRows: any[] = [] + for (const u of users) { + track('User').source++ + const email = u.email.trim().toLowerCase() + if (seenEmails.has(email)) { + throw new Error( + `Duplicate email after lowercasing: ${email} (user ${u._id}). Resolve it in Mongo first.` + ) + } + seenEmails.add(email) + + const id = nextUserId++ + userIds.set(String(u._id), id) + userIdsByUsername.set(u.username, id) + userRows.push({ + id, + email, + username: u.username, + name: null, + createdAt: instant(u.createdAt), + updatedAt: instant(u.updatedAt), + isGuest: u.isGuest ?? false, + }) + + if (u.hashedPassword) { + methodRows.push({ + userId: id, + type: 'password', + hashedPassword: u.hashedPassword, + oauthId: null, + displayName: null, + createdAt: instant(u.createdAt), + updatedAt: instant(u.updatedAt), + }) + } + for (const [key, login] of entries(u.socialLogins)) { + const provider = (login.provider ?? key).toLowerCase() + if (!isLoginMethod(provider) || provider === 'password') { + skip('UserLoginMethod', 'unknown provider', u._id, { provider }) + continue + } + const dedupe = `${provider}:${login.id}` + if (seenOAuth.has(dedupe)) { + skip('UserLoginMethod', 'duplicate (type, oauthId)', u._id, { + provider, + oauthId: login.id, + }) + continue + } + seenOAuth.add(dedupe) + methodRows.push({ + userId: id, + type: provider, + hashedPassword: null, + oauthId: login.id, + displayName: login.displayName ?? null, + createdAt: instant(u.createdAt), + updatedAt: instant(u.updatedAt), + }) + } + } + await insert(tx, 'User', t.User.insert(userRows).build(), userRows.length) + track('UserLoginMethod').source += methodRows.length + await insert( + tx, + 'UserLoginMethod', + t.UserLoginMethod.insert(methodRows).build(), + methodRows.length + ) + } + + // --- UserProgress (UserProfile.progress map, flattened) + for await (const profiles of batches( + UserProfile.find().lean().cursor(), + BATCH_SIZE + )) { + const rows: any[] = [] + for (const p of profiles) { + const userId = userIds.get(String(p.user)) + const items = entries(p.progress) + track('UserProgress').source += items.length + if (userId === undefined) { + for (const [quest] of items) + skip('UserProgress', 'user not found', p.user, { quest }) + continue + } + for (const [key, prog] of items) { + rows.push({ + userId, + quest: prog.quest ?? key, + startedAt: instant(prog.startedAt), + // `completed` is no longer stored: finishedAt IS NOT NULL means completed. + finishedAt: prog.completed + ? instantOrNull(prog.finishedAt ?? (p as any).updatedAt) + : null, + }) + } + } + await insert( + tx, + 'UserProgress', + t.UserProgress.insert(rows).build(), + rows.length + ) + } + + // --- Session + for await (const sessions of batches( + Session.find().sort({ _id: 1 }).lean().cursor(), + BATCH_SIZE + )) { + const rows: any[] = [] + for (const s of sessions) { + track('Session').source++ + const userId = userIds.get(String(s.user)) + if (userId === undefined) { + skip('Session', 'user not found', s._id, { user: String(s.user) }) + continue + } + const id = nextSessionId++ + sessionIds.set(String(s._id), id) + rows.push({ + id, + userId, + quest: s.quest, + // sessions created before localization have no locale; they were zh-TW + locale: s.locale ?? 'zh-TW', + containerId: s.containerId ?? null, + containerName: s.containerName ?? null, + status: s.status, + createdAt: instant(s.createdAt), + lastActivityAt: instant(s.lastActivityAt ?? s.updatedAt), + finishedAt: instantOrNull(s.finishedAt), + completedEvents: s.completedEvents ?? [], + graph: cleanGraph(s.graph), + }) + } + await insert(tx, 'Session', t.Session.insert(rows).build(), rows.length) + } + + // --- Command (largest collection) + for await (const commands of batches( + Command.find().sort({ _id: 1 }).lean().cursor(), + BATCH_SIZE + )) { + const rows: any[] = [] + for (const c of commands) { + track('Command').source++ + const sessionId = sessionIds.get(String(c.session)) + if (sessionId === undefined) { + skip('Command', 'session not found', c._id, { + session: String(c.session), + }) + continue + } + rows.push({ + id: nextCommandId++, + sessionId, + timestamp: instant(c.createdAt), + pwd: text(c.pwd), + command: text(c.command), + output: text(c.output), + error: text(c.error), + stage: text(c.stage), + }) + } + await insert(tx, 'Command', t.Command.insert(rows).build(), rows.length) + } + + // --- LoginAttempt (userId stays null when the username no longer exists) + for await (const attempts of batches( + LoginAttempt.find().sort({ _id: 1 }).lean().cursor(), + BATCH_SIZE + )) { + const rows = attempts.map((a) => { + track('LoginAttempt').source++ + return { + id: nextAttemptId++, + username: a.username, + userId: userIdsByUsername.get(a.username) ?? null, + ip: a.ip, + loginMethod: 'password' as const, // legacy attempts were password-only + userAgent: a.userAgent, + time: instant(a.time ?? (a as any).createdAt), + success: a.success, + message: a.message ?? null, + } + }) + await insert( + tx, + 'LoginAttempt', + t.LoginAttempt.insert(rows).build(), + rows.length + ) + } + + // --- Advance the autoincrement sequences past the ids we assigned explicitly. + // Identifiers can't be bind parameters, so each statement is spelled out. + if (!DRY_RUN) { + const { raw } = db + const plans = [ + raw.sql`SELECT setval(pg_get_serial_sequence('"User"', 'id'), COALESCE((SELECT MAX(id) FROM "User"), 0) + 1, false)`, + raw.sql`SELECT setval(pg_get_serial_sequence('"Session"', 'id'), COALESCE((SELECT MAX(id) FROM "Session"), 0) + 1, false)`, + raw.sql`SELECT setval(pg_get_serial_sequence('"Command"', 'id'), COALESCE((SELECT MAX(id) FROM "Command"), 0) + 1, false)`, + raw.sql`SELECT setval(pg_get_serial_sequence('"LoginAttempt"', 'id'), COALESCE((SELECT MAX(id) FROM "LoginAttempt"), 0) + 1, false)`, + ] + for (const plan of plans) + await tx.execute(plan.affectedCount().build() as never) + } +} + +async function assertTargetEmpty() { + const { orm } = db + const found = await Promise.all([ + orm.public.User.first(), + orm.public.UserLoginMethod.first(), + orm.public.UserProgress.first(), + orm.public.Session.first(), + orm.public.Command.first(), + orm.public.LoginAttempt.first(), + ]) + if (found.some(Boolean)) { + throw new Error( + 'Target tables are not empty. Truncate them or pass --force.' + ) + } +} + +async function main() { + const mongoUrl = process.env['MONGO_URL'] + if (!mongoUrl) throw new Error('MONGO_URL is not set') + const mongo = await connectDB(mongoUrl) + console.log( + DRY_RUN ? 'DRY RUN: nothing will be written' : 'Migrating Mongo -> Postgres' + ) + + try { + if (!FORCE && !DRY_RUN) await assertTargetEmpty() + await db.transaction((tx) => migrate(tx as Tx)) + } finally { + await mongo.close() + await db.close() + } + + console.table(stats) + if (nulStripped) + console.log(`Stripped NUL bytes from ${nulStripped} command text fields`) + writeFileSync('migration-skipped.json', JSON.stringify(skipped, null, 2)) + console.log( + `Skipped rows written to migration-skipped.json (${skipped.length})` + ) + for (const [table, s] of Object.entries(stats)) { + if (!DRY_RUN && s.source !== s.inserted + s.skipped) { + throw new Error( + `Count mismatch for ${table}: ${s.source} != ${s.inserted} + ${s.skipped}` + ) + } + } +} + +main().catch((err) => { + console.error(err) + process.exit(1) +})