From bb54fc7ac62283c26a6d58bed7f57eac05750fc8 Mon Sep 17 00:00:00 2001 From: ACQ Build Date: Fri, 12 Jun 2026 23:12:56 -0500 Subject: [PATCH] =?UTF-8?q?feat:=20apso=20import=20data=20=E2=80=94=20copy?= =?UTF-8?q?=20table=20data=20from=20Supabase=20into=20the=20rebuilt=20DB?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase 2 of migrating off Supabase: after 'apso import supabase' generates the schema and it's deployed, this copies the rows from the source Postgres into the target database. - src/lib/import/copy-data.ts (pure, fully unit-tested): planCopy() builds a topologically-ordered, FK-aware plan. Mapping mirrors phase 1 exactly (target table = snakeCase(entity); single-col FK -> camelCase Id via the shared deriveToName/getRelationshipIdField; all other columns identity), then verifies every mapped column against an introspection of the target and drops/reports mismatches. executeCopy() is insert-only (pre-flight aborts if any target table is non-empty), batches inserts, resets serial sequences, and runs in one transaction (rollback on error). - src/lib/import/pg-data.ts: PgSourceReader/PgTargetWriter (lazy pg, SSL on) implementing the injectable SourceReader/TargetWriter seams. - src/commands/import/data.ts: 'apso import data' with source/target conn resolution (flag / SUPABASE_DB_URL|DATABASE_URL and TARGET_DATABASE_URL| APSO_DATABASE_URL / masked prompt), --schema/--tables/--batch-size/ --dry-run/--yes; never logs or persists connection strings. Reuses phase 1's PgIntrospector for both source and target schemas. Verified end-to-end against two live PGlite databases: snake_case source -> camelCase-FK target with JSONB/null preservation, serial sequence reset (next id = 3, no collision), and FK integrity across a join. Full suite 264 tests, lint clean. Co-Authored-By: Claude Fable 5 --- src/commands/import/data.ts | 200 ++++++++++++++++ src/lib/import/copy-data.ts | 369 +++++++++++++++++++++++++++++ src/lib/import/index.ts | 20 ++ src/lib/import/pg-data.ts | 139 +++++++++++ test/lib/import/copy-data.test.ts | 381 ++++++++++++++++++++++++++++++ 5 files changed, 1109 insertions(+) create mode 100644 src/commands/import/data.ts create mode 100644 src/lib/import/copy-data.ts create mode 100644 src/lib/import/pg-data.ts create mode 100644 test/lib/import/copy-data.test.ts diff --git a/src/commands/import/data.ts b/src/commands/import/data.ts new file mode 100644 index 0000000..9c667f4 --- /dev/null +++ b/src/commands/import/data.ts @@ -0,0 +1,200 @@ +import { Flags } from "@oclif/core"; +import inquirer from "inquirer"; +import BaseCommand from "../../lib/base-command"; +import { + CopyPlan, + CopyResult, + PgIntrospector, + PgSourceReader, + PgTargetWriter, + executeCopy, + planCopy, + redactConnectionString, + targetColumnsFromSchema, +} from "../../lib/import"; +import { resolveConnectionString } from "./supabase"; + +/** Resolve the target connection string from flag/env, else undefined. */ +export function resolveTargetConnectionString( + flagValue: string | undefined, + env: NodeJS.ProcessEnv = process.env +): string | undefined { + return flagValue || env.TARGET_DATABASE_URL || env.APSO_DATABASE_URL || undefined; +} + +/** Render a copy result (or dry-run plan) as a human-readable summary. */ +export function formatCopyResult(result: CopyResult): string { + const lines: string[] = []; + const total = result.perTable.reduce((sum, t) => sum + t.rows, 0); + lines.push( + result.dryRun + ? `Plan: ${result.perTable.length} table(s), ${total} source row(s) to copy.` + : `Copied ${total} row(s) across ${result.perTable.length} table(s).` + ); + for (const t of result.perTable) { + lines.push(` ${t.table}: ${t.rows} row(s)`); + } + if (result.missingTargetTables.length > 0) { + lines.push( + ` Skipped (no matching target table): ${result.missingTargetTables.join(", ")}` + ); + } + if (result.cyclicTables.length > 0) { + lines.push( + ` FK cycle (copied best-effort, may need a second pass): ${result.cyclicTables.join(", ")}` + ); + } + if (result.skippedColumns.length > 0) { + lines.push(" Skipped columns:"); + for (const s of result.skippedColumns) { + lines.push(` - ${s.table}.${s.column} (${s.reason})`); + } + } + if (result.dryRun && result.blockedTables.length > 0) { + lines.push( + ` WARNING: target table(s) already have rows and would block a real run: ${result.blockedTables.join(", ")}` + ); + } + return lines.join("\n"); +} + +export default class ImportData extends BaseCommand { + static description = + "Copy table data from a source Supabase (Postgres) database into the target database rebuilt from the imported schema. Reads the source read-only; inserts into empty target tables."; + + static examples = [ + `$ apso import data -s "postgresql://...source..." -t "postgresql://...target..."`, + `$ apso import data --dry-run`, + `$ apso import data --tables users,orders`, + ]; + + static flags = { + source: Flags.string({ + char: "s", + description: + "Source connection string. Falls back to SUPABASE_DB_URL or DATABASE_URL, then prompts.", + }), + target: Flags.string({ + char: "t", + description: + "Target connection string. Falls back to TARGET_DATABASE_URL or APSO_DATABASE_URL, then prompts.", + }), + schema: Flags.string({ description: "Database schema", default: "public" }), + tables: Flags.string({ + description: "Comma-separated list of source tables to copy (default: all)", + }), + "batch-size": Flags.integer({ + description: "Rows per insert batch", + default: 500, + }), + "dry-run": Flags.boolean({ + description: "Plan and report row counts without writing", + default: false, + }), + yes: Flags.boolean({ + char: "y", + description: "Skip the confirmation prompt", + default: false, + }), + }; + + private async promptConnection(role: string): Promise { + const answer = await inquirer.prompt<{ value: string }>([ + { + type: "password", + name: "value", + mask: "*", + message: `${role} Postgres connection string:`, + validate: (input: string) => + input.trim().length > 0 || "A connection string is required", + }, + ]); + return answer.value.trim(); + } + + async run(): Promise { + const { flags } = await this.parse(ImportData); + + const sourceConn = + resolveConnectionString(flags.source) || + (await this.promptConnection("Source")); + const targetConn = + resolveTargetConnectionString(flags.target) || + (await this.promptConnection("Target")); + + const tableFilter = flags.tables + ? flags.tables.split(",").map((t) => t.trim()).filter(Boolean) + : undefined; + + let sourceReader: PgSourceReader | undefined; + let targetWriter: PgTargetWriter | undefined; + try { + this.log("Introspecting source and target schemas..."); + const sourceSchema = await new PgIntrospector(sourceConn).introspect( + flags.schema + ); + const targetSchema = await new PgIntrospector(targetConn).introspect( + flags.schema + ); + + const plan: CopyPlan = planCopy( + sourceSchema, + targetColumnsFromSchema(targetSchema), + tableFilter + ); + + if (plan.tables.length === 0) { + this.warn("No tables to copy (no source/target table overlap)."); + return; + } + + sourceReader = await PgSourceReader.connect(sourceConn, flags.schema); + targetWriter = await PgTargetWriter.connect(targetConn, flags.schema); + + if (flags["dry-run"]) { + const result = await executeCopy(plan, sourceReader, targetWriter, { + batchSize: flags["batch-size"], + dryRun: true, + }); + this.log(""); + this.log(formatCopyResult(result)); + this.log(""); + this.log("Dry run — no rows written."); + return; + } + + if (!flags.yes) { + const tableList = plan.tables.map((t) => t.targetTable).join(", "); + const { confirm } = await inquirer.prompt<{ confirm: boolean }>([ + { + type: "confirm", + name: "confirm", + message: `Copy data into ${plan.tables.length} target table(s) (${tableList})?`, + default: false, + }, + ]); + if (!confirm) { + this.log("Data copy cancelled."); + return; + } + } + + this.log("Copying data..."); + const result = await executeCopy(plan, sourceReader, targetWriter, { + batchSize: flags["batch-size"], + }); + this.log(""); + this.log(formatCopyResult(result)); + this.log("Done."); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + const redacted = redactConnectionString( + message.split(sourceConn).join("****").split(targetConn).join("****") + ); + this.error(redacted); + } finally { + if (sourceReader) await sourceReader.close(); + if (targetWriter) await targetWriter.close(); + } + } +} diff --git a/src/lib/import/copy-data.ts b/src/lib/import/copy-data.ts new file mode 100644 index 0000000..639f08d --- /dev/null +++ b/src/lib/import/copy-data.ts @@ -0,0 +1,369 @@ +/** + * Database Import — data copy (phase 2) + * + * Plans and executes a row copy from a source Postgres/Supabase database into a + * target database that was rebuilt from the imported schema (phase 1). The + * planning logic is pure and fully unit-testable; the actual reads/writes go + * through the injectable `SourceReader` / `TargetWriter` interfaces so the + * executor can be tested with fakes (mirroring phase 1's `Introspector`). + * + * Mapping rules mirror phase 1 exactly: + * - target table name = snakeCase(entity name) = snakeCase(source table name) + * - single-column FK = renamed to the platform's camelCase `Id` + * - every other column = copied under its original name + */ + +import { snakeCase } from "../utils/casing"; +import { getRelationshipIdField } from "../utils/relationships"; +import { deriveToName } from "./pg-to-apsorc"; +import { IntrospectedSchema, IntrospectedTable } from "./types"; + +const INTEGER_PK_UDTS = new Set(["int2", "int4", "int8"]); + +export interface ColumnMapping { + sourceColumn: string; + targetColumn: string; +} + +export interface TableCopyPlan { + sourceTable: string; + targetTable: string; + /** Kept source columns (those whose mapped target column exists), in order. */ + sourceColumns: string[]; + /** Target column names, parallel to sourceColumns. */ + targetColumns: string[]; + /** Source PK column used to order paged reads (null when there is no PK). */ + orderBy: string | null; + /** Target PK column whose sequence to reset after copy (null if not serial). */ + serialPkColumn: string | null; +} + +export interface CopyPlan { + tables: TableCopyPlan[]; + skippedColumns: Array<{ table: string; column: string; reason: string }>; + /** Source tables with no corresponding target table (skipped entirely). */ + missingTargetTables: string[]; + /** Tables involved in a FK cycle (copied best-effort, after acyclic tables). */ + cyclicTables: string[]; +} + +/** The target table name the generator produces for a given source table. */ +export function targetTableName(sourceTable: string): string { + return snakeCase(sourceTable); +} + +/** Map every source column to its target column name (FK rename or identity). */ +export function buildColumnMapping(table: IntrospectedTable): ColumnMapping[] { + const fkRefByColumn = new Map(); + for (const fk of table.foreignKeys) { + if (fk.columns.length === 1) { + fkRefByColumn.set(fk.columns[0], fk.referencedTable); + } + } + + return [...table.columns] + .sort((a, b) => a.ordinal - b.ordinal) + .map((col) => { + const refTable = fkRefByColumn.get(col.name); + if (refTable) { + const { toName } = deriveToName(col.name, refTable); + const targetColumn = getRelationshipIdField({ + name: refTable, + type: "ManyToOne", + referenceName: toName, + }); + return { sourceColumn: col.name, targetColumn }; + } + return { sourceColumn: col.name, targetColumn: col.name }; + }); +} + +/** + * Order tables so each is copied after the tables it references via FK + * (parents before children). Self-references are ignored; tables in a cycle are + * reported and appended best-effort at the end. + */ +export function topoSortTables(tables: IntrospectedTable[]): { + order: string[]; + cyclic: string[]; +} { + const names = new Set(tables.map((t) => t.name)); + const deps = new Map>(); + for (const t of tables) { + const d = new Set(); + for (const fk of t.foreignKeys) { + if (fk.referencedTable !== t.name && names.has(fk.referencedTable)) { + d.add(fk.referencedTable); + } + } + deps.set(t.name, d); + } + + const order: string[] = []; + const visited = new Set(); + let progress = true; + while (order.length < tables.length && progress) { + progress = false; + for (const t of tables) { + if (visited.has(t.name)) continue; + if ([...deps.get(t.name)!].every((x) => visited.has(x))) { + order.push(t.name); + visited.add(t.name); + progress = true; + } + } + } + + const cyclic = tables + .filter((t) => !visited.has(t.name)) + .map((t) => t.name); + for (const c of cyclic) order.push(c); // best-effort append + + return { order, cyclic }; +} + +/** + * Build a full copy plan from the source schema and the set of columns that + * actually exist in each target table. Source columns whose mapped target + * column is missing are dropped (and reported); source tables with no target + * table are skipped entirely. + */ +export function planCopy( + source: IntrospectedSchema, + targetColumnsByTable: Map>, + tableFilter?: string[] +): CopyPlan { + const filter = tableFilter && tableFilter.length > 0 ? new Set(tableFilter) : null; + const consideredTables = source.tables.filter( + (t) => !filter || filter.has(t.name) + ); + + const { order, cyclic } = topoSortTables(consideredTables); + const byName = new Map(consideredTables.map((t) => [t.name, t])); + + const tables: TableCopyPlan[] = []; + const skippedColumns: CopyPlan["skippedColumns"] = []; + const missingTargetTables: string[] = []; + + for (const sourceTable of order) { + const table = byName.get(sourceTable)!; + const targetTable = targetTableName(sourceTable); + const targetColumns = targetColumnsByTable.get(targetTable); + if (!targetColumns) { + missingTargetTables.push(sourceTable); + continue; + } + + const mapping = buildColumnMapping(table); + const keptSource: string[] = []; + const keptTarget: string[] = []; + for (const m of mapping) { + if (targetColumns.has(m.targetColumn)) { + keptSource.push(m.sourceColumn); + keptTarget.push(m.targetColumn); + } else { + skippedColumns.push({ + table: sourceTable, + column: m.sourceColumn, + reason: `target column "${m.targetColumn}" not found in ${targetTable}`, + }); + } + } + + // Determine ordering + serial PK handling from a single-column PK. + let orderBy: string | null = null; + let serialPkColumn: string | null = null; + if (table.primaryKey.length === 1) { + const pkCol = table.primaryKey[0]; + orderBy = pkCol; + const pk = table.columns.find((c) => c.name === pkCol); + const targetPk = mapping.find((m) => m.sourceColumn === pkCol)?.targetColumn; + if (pk && targetPk && INTEGER_PK_UDTS.has(pk.udtName)) { + serialPkColumn = targetPk; + } + } + + tables.push({ + sourceTable, + targetTable, + sourceColumns: keptSource, + targetColumns: keptTarget, + orderBy, + serialPkColumn, + }); + } + + return { tables, skippedColumns, missingTargetTables, cyclicTables: cyclic }; +} + +/** Coerce a value read from pg into something safe to bind on insert. */ +export function coerceValue(value: unknown): unknown { + if (value === null || value === undefined) return null; + if (value instanceof Date) return value; + if (Buffer.isBuffer(value)) return value; + if (typeof value === "object") return JSON.stringify(value); + return value; +} + +/** Build a parameterized multi-row INSERT statement for a batch. */ +export function buildInsertSql( + table: string, + columns: string[], + rows: unknown[][], + schema?: string +): { text: string; values: unknown[] } { + const tableRef = schema ? `"${schema}"."${table}"` : `"${table}"`; + const colList = columns.map((c) => `"${c}"`).join(", "); + const values: unknown[] = []; + const tuples = rows.map((row) => { + const placeholders = row.map((val) => { + values.push(coerceValue(val)); + return `$${values.length}`; + }); + return `(${placeholders.join(", ")})`; + }); + return { + text: `INSERT INTO ${tableRef} (${colList}) VALUES ${tuples.join(", ")}`, + values, + }; +} + +export interface SourceReader { + totalRows(table: string): Promise; + readBatch( + table: string, + columns: string[], + orderBy: string | null, + offset: number, + limit: number + ): Promise; +} + +export interface TargetWriter { + rowCount(table: string): Promise; + insertBatch( + table: string, + columns: string[], + rows: unknown[][] + ): Promise; + resetSequence(table: string, column: string): Promise; + begin(): Promise; + commit(): Promise; + rollback(): Promise; +} + +export interface CopyResult { + dryRun: boolean; + perTable: Array<{ table: string; rows: number }>; + skippedColumns: CopyPlan["skippedColumns"]; + missingTargetTables: string[]; + cyclicTables: string[]; + /** Target tables that already had rows (blocks a real, insert-only run). */ + blockedTables: string[]; +} + +export interface ExecuteOptions { + batchSize?: number; + dryRun?: boolean; +} + +/** + * Execute a copy plan. Insert-only: if any target table already has rows the + * run aborts before writing anything. The whole copy runs in one transaction + * on the target, so a failure rolls everything back. + */ +export async function executeCopy( + plan: CopyPlan, + source: SourceReader, + target: TargetWriter, + options: ExecuteOptions = {} +): Promise { + const batchSize = options.batchSize ?? 500; + const dryRun = options.dryRun ?? false; + + // Pre-flight: detect non-empty target tables. + const blockedTables: string[] = []; + for (const t of plan.tables) { + // eslint-disable-next-line no-await-in-loop + const count = await target.rowCount(t.targetTable); + if (count > 0) blockedTables.push(t.targetTable); + } + + const perTable: Array<{ table: string; rows: number }> = []; + + if (dryRun) { + for (const t of plan.tables) { + // eslint-disable-next-line no-await-in-loop + const rows = await source.totalRows(t.sourceTable); + perTable.push({ table: t.targetTable, rows }); + } + return { + dryRun: true, + perTable, + skippedColumns: plan.skippedColumns, + missingTargetTables: plan.missingTargetTables, + cyclicTables: plan.cyclicTables, + blockedTables, + }; + } + + if (blockedTables.length > 0) { + throw new Error( + `Aborting: target table(s) already contain rows: ${blockedTables.join( + ", " + )}. Use a fresh target or clear these tables first.` + ); + } + + await target.begin(); + try { + for (const t of plan.tables) { + let offset = 0; + let copied = 0; + // eslint-disable-next-line no-constant-condition + while (true) { + // eslint-disable-next-line no-await-in-loop + const rows = await source.readBatch( + t.sourceTable, + t.sourceColumns, + t.orderBy, + offset, + batchSize + ); + if (rows.length === 0) break; + // eslint-disable-next-line no-await-in-loop + await target.insertBatch(t.targetTable, t.targetColumns, rows); + copied += rows.length; + offset += rows.length; + if (rows.length < batchSize) break; + } + if (t.serialPkColumn && copied > 0) { + // eslint-disable-next-line no-await-in-loop + await target.resetSequence(t.targetTable, t.serialPkColumn); + } + perTable.push({ table: t.targetTable, rows: copied }); + } + await target.commit(); + } catch (error) { + await target.rollback(); + throw error; + } + + return { + dryRun: false, + perTable, + skippedColumns: plan.skippedColumns, + missingTargetTables: plan.missingTargetTables, + cyclicTables: plan.cyclicTables, + blockedTables, + }; +} + +/** Build the target column lookup used by planCopy from an introspected target. */ +export function targetColumnsFromSchema( + target: IntrospectedSchema +): Map> { + return new Map( + target.tables.map((t) => [t.name, new Set(t.columns.map((c) => c.name))]) + ); +} diff --git a/src/lib/import/index.ts b/src/lib/import/index.ts index 633e0c0..be98650 100644 --- a/src/lib/import/index.ts +++ b/src/lib/import/index.ts @@ -7,3 +7,23 @@ export type { ApsorcImportRelationshipOutput, } from "./pg-to-apsorc"; export { PgIntrospector, redactConnectionString, buildSchema, SYSTEM_SCHEMAS } from "./introspect"; +export { + planCopy, + executeCopy, + buildColumnMapping, + topoSortTables, + targetTableName, + targetColumnsFromSchema, + buildInsertSql, + coerceValue, +} from "./copy-data"; +export type { + CopyPlan, + CopyResult, + TableCopyPlan, + ColumnMapping, + SourceReader, + TargetWriter, + ExecuteOptions, +} from "./copy-data"; +export { PgSourceReader, PgTargetWriter } from "./pg-data"; diff --git a/src/lib/import/pg-data.ts b/src/lib/import/pg-data.ts new file mode 100644 index 0000000..90f606f --- /dev/null +++ b/src/lib/import/pg-data.ts @@ -0,0 +1,139 @@ +/** + * Database Import — pg-backed data access for the copy executor. + * + * Thin wrappers around `pg` implementing the SourceReader / TargetWriter + * interfaces from copy-data.ts. `pg` is imported lazily and SSL is enabled by + * default (Supabase requires it). Integration-level: covered by manual/E2E + * runs, while the copy logic itself is unit-tested with fakes. + */ + +import { SourceReader, TargetWriter } from "./copy-data"; + +interface PgClient { + connect(): Promise; + query(sql: string, params?: unknown[]): Promise<{ rows: any[]; rowCount: number }>; + end(): Promise; +} + +async function createClient(connectionString: string): Promise { + const pg = await import("pg"); + const Client = (pg as any).Client ?? (pg as any).default?.Client; + const client = new Client({ + connectionString, + ssl: { rejectUnauthorized: false }, + }); + await client.connect(); + return client as PgClient; +} + +function qualified(schema: string, table: string): string { + return `"${schema}"."${table}"`; +} + +/** Reads rows from the source database (read-only). */ +export class PgSourceReader implements SourceReader { + private readonly client: PgClient; + private readonly schema: string; + + private constructor(client: PgClient, schema: string) { + this.client = client; + this.schema = schema; + } + + static async connect( + connectionString: string, + schema: string + ): Promise { + return new PgSourceReader(await createClient(connectionString), schema); + } + + async totalRows(table: string): Promise { + const res = await this.client.query( + `SELECT count(*)::int AS n FROM ${qualified(this.schema, table)}` + ); + return res.rows[0].n; + } + + async readBatch( + table: string, + columns: string[], + orderBy: string | null, + offset: number, + limit: number + ): Promise { + const cols = columns.map((c) => `"${c}"`).join(", "); + const order = orderBy ? ` ORDER BY "${orderBy}"` : ""; + const res = await this.client.query( + `SELECT ${cols} FROM ${qualified(this.schema, table)}${order} OFFSET $1 LIMIT $2`, + [offset, limit] + ); + return res.rows.map((row) => columns.map((c) => row[c])); + } + + async close(): Promise { + await this.client.end(); + } +} + +/** Writes rows into the target database within a single transaction. */ +export class PgTargetWriter implements TargetWriter { + private readonly client: PgClient; + private readonly schema: string; + + private constructor(client: PgClient, schema: string) { + this.client = client; + this.schema = schema; + } + + static async connect( + connectionString: string, + schema: string + ): Promise { + return new PgTargetWriter(await createClient(connectionString), schema); + } + + async rowCount(table: string): Promise { + const res = await this.client.query( + `SELECT count(*)::int AS n FROM ${qualified(this.schema, table)}` + ); + return res.rows[0].n; + } + + async insertBatch( + table: string, + columns: string[], + rows: unknown[][] + ): Promise { + const { buildInsertSql } = await import("./copy-data"); + const { text, values } = buildInsertSql(table, columns, rows, this.schema); + await this.client.query(text, values); + } + + async resetSequence(table: string, column: string): Promise { + // Advance the column's owned sequence past the max value just inserted. + await this.client.query( + `SELECT setval( + pg_get_serial_sequence($1, $2), + (SELECT COALESCE(MAX("${column}"), 0) FROM ${qualified(this.schema, table)}), + true + )`, + [`${this.schema}.${table}`, column] + ); + } + + async begin(): Promise { + await this.client.query("BEGIN"); + } + + async commit(): Promise { + await this.client.query("COMMIT"); + } + + async rollback(): Promise { + await this.client.query("ROLLBACK"); + } + + async close(): Promise { + await this.client.end(); + } +} diff --git a/test/lib/import/copy-data.test.ts b/test/lib/import/copy-data.test.ts new file mode 100644 index 0000000..7f18be4 --- /dev/null +++ b/test/lib/import/copy-data.test.ts @@ -0,0 +1,381 @@ +import { expect, describe, test } from "@jest/globals"; +import { + targetTableName, + buildColumnMapping, + topoSortTables, + planCopy, + coerceValue, + buildInsertSql, + executeCopy, + targetColumnsFromSchema, + SourceReader, + TargetWriter, +} from "../../../src/lib/import/copy-data"; +import { + IntrospectedSchema, + IntrospectedTable, +} from "../../../src/lib/import/types"; + +function col(name: string, udtName = "text", ordinal = 1) { + return { + name, + udtName, + dataType: "base", + nullable: false, + default: null, + charMaxLength: null, + numericPrecision: null, + numericScale: null, + ordinal, + isEnum: false, + }; +} + +function table( + name: string, + o: Partial = {} +): IntrospectedTable { + return { + name, + columns: o.columns ?? [], + primaryKey: o.primaryKey ?? [], + foreignKeys: o.foreignKeys ?? [], + uniqueConstraints: o.uniqueConstraints ?? [], + indexes: o.indexes ?? [], + }; +} + +function schema(tables: IntrospectedTable[]): IntrospectedSchema { + return { schema: "public", tables, enums: [], skipped: { views: [], systemSchemas: [] } }; +} + +describe("targetTableName", () => { + test("snake_case source tables are unchanged", () => { + expect(targetTableName("blog_posts")).toBe("blog_posts"); + }); + test("PascalCase entity names fold to snake_case", () => { + expect(targetTableName("User")).toBe("user"); + expect(targetTableName("BlogPost")).toBe("blog_post"); + }); +}); + +describe("buildColumnMapping", () => { + test("scalar columns map to themselves; single FK renamed to camelCase Id", () => { + const t = table("posts", { + columns: [col("id", "uuid", 1), col("title", "text", 2), col("user_id", "uuid", 3)], + foreignKeys: [ + { columns: ["user_id"], referencedTable: "users", referencedColumns: ["id"], onDelete: "CASCADE" }, + ], + }); + const mapping = buildColumnMapping(t); + expect(mapping).toEqual([ + { sourceColumn: "id", targetColumn: "id" }, + { sourceColumn: "title", targetColumn: "title" }, + { sourceColumn: "user_id", targetColumn: "userId" }, + ]); + }); + + test("non-default FK column maps via derived reference name", () => { + const t = table("posts", { + columns: [col("id", "uuid", 1), col("author_id", "uuid", 2)], + foreignKeys: [ + { columns: ["author_id"], referencedTable: "users", referencedColumns: ["id"], onDelete: "NO ACTION" }, + ], + }); + expect(buildColumnMapping(t)[1]).toEqual({ + sourceColumn: "author_id", + targetColumn: "authorId", + }); + }); +}); + +describe("topoSortTables", () => { + test("parents are ordered before children", () => { + const tables = [ + table("posts", { + foreignKeys: [{ columns: ["user_id"], referencedTable: "users", referencedColumns: ["id"], onDelete: "NO ACTION" }], + }), + table("users"), + ]; + const { order, cyclic } = topoSortTables(tables); + expect(order.indexOf("users")).toBeLessThan(order.indexOf("posts")); + expect(cyclic).toEqual([]); + }); + + test("self-references do not create a cycle", () => { + const tables = [ + table("nodes", { + foreignKeys: [{ columns: ["parent_id"], referencedTable: "nodes", referencedColumns: ["id"], onDelete: "NO ACTION" }], + }), + ]; + const { order, cyclic } = topoSortTables(tables); + expect(order).toEqual(["nodes"]); + expect(cyclic).toEqual([]); + }); + + test("mutual FK cycle is reported and still ordered best-effort", () => { + const tables = [ + table("a", { foreignKeys: [{ columns: ["b_id"], referencedTable: "b", referencedColumns: ["id"], onDelete: "NO ACTION" }] }), + table("b", { foreignKeys: [{ columns: ["a_id"], referencedTable: "a", referencedColumns: ["id"], onDelete: "NO ACTION" }] }), + ]; + const { order, cyclic } = topoSortTables(tables); + expect(order).toHaveLength(2); + expect(cyclic.sort()).toEqual(["a", "b"]); + }); +}); + +describe("planCopy", () => { + const source = schema([ + table("users", { + primaryKey: ["id"], + columns: [col("id", "int4", 1), col("email", "text", 2), col("secret", "text", 3)], + }), + table("posts", { + primaryKey: ["id"], + columns: [col("id", "uuid", 1), col("user_id", "uuid", 2)], + foreignKeys: [{ columns: ["user_id"], referencedTable: "users", referencedColumns: ["id"], onDelete: "CASCADE" }], + }), + table("legacy", { primaryKey: ["id"], columns: [col("id", "uuid", 1)] }), + ]); + + const targetColumns = new Map>([ + ["users", new Set(["id", "email"])], // note: "secret" missing in target + ["posts", new Set(["id", "userId"])], // snakeCase("posts") === "posts"; FK renamed + // "legacy" intentionally absent from target + ]); + + test("orders tables, maps FK columns, and verifies against the target", () => { + const plan = planCopy(source, targetColumns); + const names = plan.tables.map((t) => t.sourceTable); + expect(names.indexOf("users")).toBeLessThan(names.indexOf("posts")); + + const users = plan.tables.find((t) => t.sourceTable === "users")!; + expect(users.targetTable).toBe("users"); + expect(users.sourceColumns).toEqual(["id", "email"]); + // serial (int4) single-id PK => sequence reset target column + expect(users.serialPkColumn).toBe("id"); + expect(users.orderBy).toBe("id"); + + const posts = plan.tables.find((t) => t.sourceTable === "posts")!; + expect(posts.targetTable).toBe("posts"); + expect(posts.sourceColumns).toEqual(["id", "user_id"]); + expect(posts.targetColumns).toEqual(["id", "userId"]); + expect(posts.serialPkColumn).toBeNull(); // uuid PK + + // "secret" dropped (missing in target); "legacy" table skipped entirely. + expect(plan.skippedColumns).toContainEqual({ + table: "users", + column: "secret", + reason: 'target column "secret" not found in users', + }); + expect(plan.missingTargetTables).toEqual(["legacy"]); + }); + + test("table filter limits the plan", () => { + const plan = planCopy(source, targetColumns, ["posts"]); + expect(plan.tables.map((t) => t.sourceTable)).toEqual(["posts"]); + }); +}); + +describe("coerceValue", () => { + test("stringifies objects/arrays, passes primitives/Date/Buffer/null", () => { + expect(coerceValue({ a: 1 })).toBe('{"a":1}'); + expect(coerceValue([1, 2])).toBe("[1,2]"); + expect(coerceValue(null)).toBeNull(); + // eslint-disable-next-line unicorn/no-useless-undefined -- exercising the undefined branch + expect(coerceValue(undefined)).toBeNull(); + expect(coerceValue(5)).toBe(5); + expect(coerceValue("x")).toBe("x"); + const d = new Date(); + expect(coerceValue(d)).toBe(d); + const b = Buffer.from("x"); + expect(coerceValue(b)).toBe(b); + }); +}); + +describe("buildInsertSql", () => { + test("builds parameterized multi-row insert with schema qualification", () => { + const { text, values } = buildInsertSql( + "post", + ["id", "userId"], + [ + ["a", "u1"], + ["b", null], + ], + "public" + ); + expect(text).toBe( + 'INSERT INTO "public"."post" ("id", "userId") VALUES ($1, $2), ($3, $4)' + ); + expect(values).toEqual(["a", "u1", "b", null]); + }); +}); + +// ---- executor with fakes ---- + +class FakeSource implements SourceReader { + private data: Record; + + constructor(data: Record) { + this.data = data; + } + + async totalRows(t: string): Promise { + return (this.data[t] ?? []).length; + } + + async readBatch( + t: string, + _columns: string[], + _orderBy: string | null, + offset: number, + limit: number + ): Promise { + return (this.data[t] ?? []).slice(offset, offset + limit); + } +} + +class FakeTarget implements TargetWriter { + inserts: Array<{ table: string; columns: string[]; rows: unknown[][] }> = []; + + sequencesReset: Array<{ table: string; column: string }> = []; + + began = false; + + committed = false; + + rolledBack = false; + + private counts: Record; + + constructor(counts: Record = {}) { + this.counts = counts; + } + + async rowCount(t: string): Promise { + return this.counts[t] ?? 0; + } + + async insertBatch(table: string, columns: string[], rows: unknown[][]): Promise { + this.inserts.push({ table, columns, rows }); + } + + async resetSequence(table: string, column: string): Promise { + this.sequencesReset.push({ table, column }); + } + + async begin(): Promise { + this.began = true; + } + + async commit(): Promise { + this.committed = true; + } + + async rollback(): Promise { + this.rolledBack = true; + } +} + +const twoTablePlan = { + tables: [ + { + sourceTable: "users", + targetTable: "users", + sourceColumns: ["id", "email"], + targetColumns: ["id", "email"], + orderBy: "id", + serialPkColumn: "id", + }, + { + sourceTable: "posts", + targetTable: "post", + sourceColumns: ["id", "user_id"], + targetColumns: ["id", "userId"], + orderBy: "id", + serialPkColumn: null, + }, + ], + skippedColumns: [], + missingTargetTables: [], + cyclicTables: [], +}; + +describe("executeCopy", () => { + test("copies in batches, resets serial sequences, commits", async () => { + const source = new FakeSource({ + users: [ + [1, "a@x.com"], + [2, "b@x.com"], + [3, "c@x.com"], + ], + posts: [["p1", 1]], + }); + const target = new FakeTarget(); + const result = await executeCopy(twoTablePlan, source, target, { + batchSize: 2, + }); + + expect(target.began).toBe(true); + expect(target.committed).toBe(true); + // users: 3 rows in batches of 2 => 2 insert calls; posts: 1 call. + expect(target.inserts.filter((i) => i.table === "users")).toHaveLength(2); + expect(target.inserts.find((i) => i.table === "post")!.columns).toEqual([ + "id", + "userId", + ]); + expect(target.sequencesReset).toEqual([{ table: "users", column: "id" }]); + expect(result.perTable).toEqual([ + { table: "users", rows: 3 }, + { table: "post", rows: 1 }, + ]); + }); + + test("aborts before writing when a target table is non-empty", async () => { + const source = new FakeSource({ users: [[1, "a"]], posts: [] }); + const target = new FakeTarget({ users: 5 }); + await expect( + executeCopy(twoTablePlan, source, target, {}) + ).rejects.toThrow(/already contain rows/); + expect(target.began).toBe(false); + expect(target.inserts).toHaveLength(0); + }); + + test("dry run reports source counts and would-block tables without writing", async () => { + const source = new FakeSource({ users: [[1, "a"], [2, "b"]], posts: [["p", 1]] }); + const target = new FakeTarget({ users: 9 }); + const result = await executeCopy(twoTablePlan, source, target, { + dryRun: true, + }); + expect(result.dryRun).toBe(true); + expect(result.perTable).toEqual([ + { table: "users", rows: 2 }, + { table: "post", rows: 1 }, + ]); + expect(result.blockedTables).toEqual(["users"]); + expect(target.began).toBe(false); + expect(target.inserts).toHaveLength(0); + }); + + test("rolls back if an insert fails mid-run", async () => { + const source = new FakeSource({ users: [[1, "a"]], posts: [["p", 1]] }); + const target = new FakeTarget(); + target.insertBatch = async () => { + throw new Error("boom"); + }; + await expect(executeCopy(twoTablePlan, source, target, {})).rejects.toThrow( + "boom" + ); + expect(target.rolledBack).toBe(true); + expect(target.committed).toBe(false); + }); +}); + +describe("targetColumnsFromSchema", () => { + test("builds a table -> column-set map", () => { + const map = targetColumnsFromSchema( + schema([table("users", { columns: [col("id"), col("email", "text", 2)] })]) + ); + expect(map.get("users")).toEqual(new Set(["id", "email"])); + }); +});