Merge pull request #1569 from constructive-io/feat/pgpm-transform

feat(pgpm): `pgpm transform` — re-dial an existing module/workspace; wire `--partition` into `pgpm export`

Author: pyramation

pgpm/cli/__tests__/transform-e2e.test.ts

@@ -0,0 +1,313 @@
+/**
+ * e2e for `pgpm transform` against live Postgres.
+ *
+ * Fixture module: two schemas, tables + FK, function, trigger, policy, grant,
+ * index. Transforms to each granularity, deploys each output, asserts catalog
+ * equivalence with the original, exercises --check/--dry-run/overwrite guard,
+ * verifies, reverts, partitions into pkg-app + pkg-security, and round-trips
+ * consolidated -> atomic -> consolidated.
+ *
+ * PREREQUISITES: a running PostgreSQL instance via standard PG* env vars.
+ */
+import { diffCatalogSnapshots, snapshotCatalog } from '@pgpmjs/export';
+import * as fs from 'fs';
+import * as path from 'path';
+import { teardownPgPools } from 'pg-cache';
+
+import { CLIDeployTestFixture } from '../test-utils';
+
+jest.setTimeout(180000);
+
+afterAll(async () => {
+  await teardownPgPools();
+});
+
+const MODULE_NAME = 'transform-e2e';
+const WS = 'transform-ws';
+
+interface ChangeSpec {
+  name: string;
+  deps?: string[];
+  deploy: string;
+}
+
+/** Two schemas, tables + FK, function, trigger, policy, grant, index. */
+const CHANGES: ChangeSpec[] = [
+  {
+    name: 'schemas/tfx_app/schema',
+    deploy: 'CREATE SCHEMA tfx_app;'
+  },
+  {
+    name: 'schemas/tfx_sec/schema',
+    deploy: 'CREATE SCHEMA tfx_sec;'
+  },
+  {
+    name: 'schemas/tfx_app/tables/users/table',
+    deps: ['schemas/tfx_app/schema'],
+    deploy: 'CREATE TABLE tfx_app.users (id serial PRIMARY KEY, email text NOT NULL);'
+  },
+  {
+    name: 'schemas/tfx_sec/tables/audit/table',
+    deps: ['schemas/tfx_sec/schema', 'schemas/tfx_app/tables/users/table'],
+    deploy: [
+      'CREATE TABLE tfx_sec.audit (',
+      '  id serial PRIMARY KEY,',
+      '  user_id int NOT NULL REFERENCES tfx_app.users (id),',
+      '  note text',
+      ');'
+    ].join('\n')
+  },
+  {
+    name: 'schemas/tfx_app/procedures/touch',
+    deps: ['schemas/tfx_app/schema'],
+    deploy:
+      'CREATE FUNCTION tfx_app.touch() RETURNS trigger AS $$ BEGIN RETURN NEW; END $$ LANGUAGE plpgsql;'
+  },
+  {
+    name: 'schemas/tfx_app/tables/users/triggers/users_touch',
+    deps: ['schemas/tfx_app/tables/users/table', 'schemas/tfx_app/procedures/touch'],
+    deploy:
+      'CREATE TRIGGER users_touch BEFORE UPDATE ON tfx_app.users FOR EACH ROW EXECUTE FUNCTION tfx_app.touch();'
+  },
+  {
+    name: 'schemas/tfx_app/tables/users/policies/users_self',
+    deps: ['schemas/tfx_app/tables/users/table'],
+    deploy: [
+      'ALTER TABLE tfx_app.users ENABLE ROW LEVEL SECURITY;',
+      'CREATE POLICY users_self ON tfx_app.users FOR SELECT USING (true);'
+    ].join('\n')
+  },
+  {
+    name: 'schemas/tfx_app/tables/users/grants/select',
+    deps: ['schemas/tfx_app/tables/users/table'],
+    deploy: 'GRANT SELECT ON tfx_app.users TO PUBLIC;'
+  },
+  {
+    name: 'schemas/tfx_app/tables/users/indexes/users_email_idx',
+    deps: ['schemas/tfx_app/tables/users/table'],
+    deploy: 'CREATE INDEX users_email_idx ON tfx_app.users (email);'
+  }
+];
+
+const writeModule = (moduleDir: string) => {
+  const workspaceDir = path.dirname(moduleDir);
+  fs.mkdirSync(workspaceDir, { recursive: true });
+  fs.writeFileSync(path.join(workspaceDir, 'pgpm.json'), '{\n    "packages": [\n        "*"\n    ]\n}');
+  fs.mkdirSync(moduleDir, { recursive: true });
+  const planLines = CHANGES.map((c, i) => {
+    const deps = c.deps && c.deps.length > 0 ? ` [${c.deps.join(' ')}]` : '';
+    return `${c.name}${deps} 2024-01-0${(i % 9) + 1}T00:00:00Z test <test@example.com> # add ${c.name}`;
+  });
+  fs.writeFileSync(
+    path.join(moduleDir, 'pgpm.plan'),
+    `%syntax-version=1.0.0\n%project=${MODULE_NAME}\n%uri=https://github.com/test/${MODULE_NAME}\n\n${planLines.join('\n')}\n`
+  );
+  fs.writeFileSync(
+    path.join(moduleDir, `${MODULE_NAME}.control`),
+    `comment = 'Transform e2e module'\ndefault_version = '0.0.1'\nrelocatable = false\nsuperuser = false\n`
+  );
+  for (const c of CHANGES) {
+    const filePath = path.join(moduleDir, 'deploy', `${c.name}.sql`);
+    fs.mkdirSync(path.dirname(filePath), { recursive: true });
+    fs.writeFileSync(filePath, `-- Deploy ${c.name} to pg\n\nBEGIN;\n\n${c.deploy}\n\nCOMMIT;\n`);
+    const revertPath = path.join(moduleDir, 'revert', `${c.name}.sql`);
+    fs.mkdirSync(path.dirname(revertPath), { recursive: true });
+    fs.writeFileSync(revertPath, `-- Revert ${c.name} from pg\n\nBEGIN;\n\n-- noop\n\nCOMMIT;\n`);
+    const verifyPath = path.join(moduleDir, 'verify', `${c.name}.sql`);
+    fs.mkdirSync(path.dirname(verifyPath), { recursive: true });
+    fs.writeFileSync(verifyPath, `-- Verify ${c.name} on pg\n\nBEGIN;\n\nROLLBACK;\n`);
+  }
+};
+
+const PARTITION_CONFIG = {
+  defaultPackage: 'pkg-app',
+  splitRiders: ['grant'],
+  rules: [
+    {
+      package: 'pkg-security',
+      select: [{ kind: 'policy' }, { statementKind: 'grant' }]
+    }
+  ]
+};
+
+describe('pgpm transform e2e', () => {
+  let fixture: CLIDeployTestFixture;
+  let originalDb: any;
+  let wsDir: string;
+  let moduleDir: string;
+
+  beforeAll(async () => {
+    fixture = new CLIDeployTestFixture();
+    wsDir = path.join(fixture.tempFixtureDir, WS);
+    moduleDir = path.join(wsDir, MODULE_NAME);
+    writeModule(moduleDir);
+
+    originalDb = await fixture.setupTestDatabase();
+    await fixture.runTerminalCommands(
+      `
+      cd ${WS}/${MODULE_NAME}
+      pgpm deploy --database ${originalDb.name} --package ${MODULE_NAME} --yes
+      `,
+      { database: originalDb.name }
+    );
+  });
+
+  afterAll(async () => {
+    await fixture.cleanup();
+  });
+
+  it('--dry-run prints the plan without writing anything', async () => {
+    await fixture.runTerminalCommands(
+      `
+      cd ${WS}/${MODULE_NAME}
+      pgpm transform --granularity object --dry-run
+      `,
+      {}
+    );
+    expect(fs.existsSync(path.join(wsDir, `${MODULE_NAME}-object`))).toBe(false);
+  });
+
+  it.each(['atomic', 'object', 'consolidated'] as const)(
+    'transform --granularity %s deploys with an equivalent catalog, verifies, and reverts clean',
+    async granularity => {
+      await fixture.runTerminalCommands(
+        `
+        cd ${WS}/${MODULE_NAME}
+        pgpm transform --granularity ${granularity}
+        `,
+        {}
+      );
+
+      const outDir = path.join(wsDir, `${MODULE_NAME}-${granularity}`);
+      expect(fs.existsSync(path.join(outDir, 'pgpm.plan'))).toBe(true);
+      expect(fs.existsSync(path.join(outDir, `${MODULE_NAME}-${granularity}.control`))).toBe(true);
+
+      const testDb = await fixture.setupTestDatabase();
+      await fixture.runTerminalCommands(
+        `
+        cd ${WS}/${MODULE_NAME}-${granularity}
+        pgpm deploy --database ${testDb.name} --package ${MODULE_NAME}-${granularity} --yes
+        `,
+        { database: testDb.name }
+      );
+
+      const snapOriginal = await snapshotCatalog(originalDb);
+      const snapTransformed = await snapshotCatalog(testDb);
+      expect(diffCatalogSnapshots(snapOriginal, snapTransformed)).toEqual([]);
+
+      await fixture.runTerminalCommands(
+        `
+        cd ${WS}/${MODULE_NAME}-${granularity}
+        pgpm verify --database ${testDb.name} --package ${MODULE_NAME}-${granularity} --yes
+        pgpm revert --database ${testDb.name} --package ${MODULE_NAME}-${granularity} --yes
+        `,
+        { database: testDb.name }
+      );
+
+      expect(await testDb.exists('schema', 'tfx_app')).toBe(false);
+      expect(await testDb.exists('schema', 'tfx_sec')).toBe(false);
+      const remaining = await testDb.query(
+        `SELECT COUNT(*)::int AS count FROM pgpm_migrate.changes WHERE package = $1`,
+        [`${MODULE_NAME}-${granularity}`]
+      );
+      expect(remaining.rows[0].count).toBe(0);
+    }
+  );
+
+  it('rewrites an existing output when --write is passed', async () => {
+    // The refusal path (no --write) exits the process, so it is covered by the
+    // checkOverwrite unit tests; here we prove --write re-generates in place.
+    const outDir = path.join(wsDir, `${MODULE_NAME}-object`);
+    expect(fs.existsSync(outDir)).toBe(true);
+
+    await fixture.runTerminalCommands(
+      `
+      cd ${WS}/${MODULE_NAME}
+      pgpm transform --granularity object --write
+      `,
+      {}
+    );
+    expect(fs.existsSync(path.join(outDir, 'pgpm.plan'))).toBe(true);
+  });
+
+  it('--check proves the transform is structurally lossless', async () => {
+    await fixture.runTerminalCommands(
+      `
+      cd ${WS}/${MODULE_NAME}
+      pgpm transform --granularity consolidated --write --check
+      `,
+      {}
+    );
+  });
+
+  it('partitions into pkg-app + pkg-security with catalog equivalence', async () => {
+    fs.writeFileSync(path.join(wsDir, 'partition.json'), JSON.stringify(PARTITION_CONFIG, null, 2));
+
+    await fixture.runTerminalCommands(
+      `
+      cd ${WS}/${MODULE_NAME}
+      pgpm transform --granularity object --partition ../partition.json
+      `,
+      {}
+    );
+
+    const appDir = path.join(wsDir, 'pkg-app');
+    const secDir = path.join(wsDir, 'pkg-security');
+    expect(fs.existsSync(path.join(appDir, 'pgpm.plan'))).toBe(true);
+    expect(fs.existsSync(path.join(secDir, 'pgpm.plan'))).toBe(true);
+
+    // security package requires the app package (policies/grants target app tables)
+    const secControl = fs.readFileSync(path.join(secDir, 'pkg-security.control'), 'utf-8');
+    expect(secControl).toMatch(/requires = '.*pkg-app.*'/);
+
+    const testDb = await fixture.setupTestDatabase();
+    await fixture.runTerminalCommands(
+      `
+      cd ${WS}/pkg-app
+      pgpm deploy --database ${testDb.name} --package pkg-app --yes
+      cd ../pkg-security
+      pgpm deploy --database ${testDb.name} --package pkg-security --yes
+      `,
+      { database: testDb.name }
+    );
+
+    const snapOriginal = await snapshotCatalog(originalDb);
+    const snapPartitioned = await snapshotCatalog(testDb);
+    expect(diffCatalogSnapshots(snapOriginal, snapPartitioned)).toEqual([]);
+  });
+
+  it('round-trips consolidated -> atomic -> consolidated to the same object set', async () => {
+    // consolidated already exists at <module>-consolidated (from --check test)
+    const c1 = path.join(wsDir, `${MODULE_NAME}-consolidated`);
+    expect(fs.existsSync(path.join(c1, 'pgpm.plan'))).toBe(true);
+
+    await fixture.runTerminalCommands(
+      `
+      cd ${WS}/${MODULE_NAME}-consolidated
+      pgpm transform --granularity atomic --out ../roundtrip-atomic --write
+      cd ../roundtrip-atomic/${MODULE_NAME}-consolidated-atomic
+      pgpm transform --granularity consolidated --out ../../roundtrip-consolidated --write
+      `,
+      {}
+    );
+
+    const c2 = path.join(
+      wsDir,
+      'roundtrip-consolidated',
+      `${MODULE_NAME}-consolidated-atomic-consolidated`
+    );
+    expect(fs.existsSync(path.join(c2, 'pgpm.plan'))).toBe(true);
+
+    const changeSet = (planPath: string): string[] =>
+      fs
+        .readFileSync(planPath, 'utf-8')
+        .split('\n')
+        .filter(line => line && !line.startsWith('%'))
+        .map(line => line.split(' ')[0])
+        .sort();
+
+    expect(changeSet(path.join(c2, 'pgpm.plan'))).toEqual(
+      changeSet(path.join(c1, 'pgpm.plan'))
+    );
+  });
+});

pgpm/cli/__tests__/transform-unit.test.ts

@@ -0,0 +1,67 @@
+import { mkdirSync, mkdtempSync, rmSync } from 'fs';
+import { tmpdir } from 'os';
+import { join } from 'path';
+
+import { checkOverwrite, orderPackages, resolveOutBase } from '../src/commands/transform';
+
+describe('resolveOutBase', () => {
+  it('defaults to the source module parent (packages land as siblings)', () => {
+    expect(resolveOutBase('/ws/my-mod')).toBe('/ws');
+  });
+
+  it('resolves --out when given', () => {
+    expect(resolveOutBase('/ws/my-mod', '/tmp/out')).toBe('/tmp/out');
+  });
+});
+
+describe('checkOverwrite', () => {
+  let dir: string;
+
+  beforeEach(() => {
+    dir = mkdtempSync(join(tmpdir(), 'pgpm-transform-guard-'));
+  });
+
+  afterEach(() => {
+    rmSync(dir, { recursive: true, force: true });
+  });
+
+  it('refuses to overwrite the source module without --write', () => {
+    const mod = join(dir, 'my-mod');
+    mkdirSync(mod);
+    expect(checkOverwrite(mod, mod, false)).toMatch(/source module/);
+    expect(checkOverwrite(mod, mod, true)).toBeNull();
+  });
+
+  it('refuses to clobber an existing output directory without --write', () => {
+    const mod = join(dir, 'my-mod');
+    const out = join(dir, 'my-mod-object');
+    mkdirSync(mod);
+    mkdirSync(out);
+    expect(checkOverwrite(out, mod, false)).toMatch(/already exists/);
+    expect(checkOverwrite(out, mod, true)).toBeNull();
+  });
+
+  it('allows a fresh output directory', () => {
+    const mod = join(dir, 'my-mod');
+    mkdirSync(mod);
+    expect(checkOverwrite(join(dir, 'fresh'), mod, false)).toBeNull();
+  });
+});
+
+describe('orderPackages', () => {
+  it('orders prerequisites before dependents', () => {
+    const ordered = orderPackages([
+      { name: 'pkg-security', requires: ['pkg-app'], rows: [] },
+      { name: 'pkg-app', requires: [], rows: [] }
+    ]);
+    expect(ordered.map(p => p.name)).toEqual(['pkg-app', 'pkg-security']);
+  });
+
+  it('keeps independent packages in input order', () => {
+    const ordered = orderPackages([
+      { name: 'a', requires: [], rows: [] },
+      { name: 'b', requires: [], rows: [] }
+    ]);
+    expect(ordered.map(p => p.name)).toEqual(['a', 'b']);
+  });
+});

pgpm/cli/src/commands.ts

@@ -29,6 +29,7 @@ import slice from './commands/slice';
 import syncVersions from './commands/sync-versions';
 import tag from './commands/tag';
 import testPackages from './commands/test-packages';
+import transform from './commands/transform';
 import tune from './commands/tune';
 import updateCmd from './commands/update';
 import upgrade from './commands/upgrade';
@@ -112,6 +113,7 @@ export const createPgpmCommandMap = (skipPgTeardown: boolean = false): Record<st
     slice,
     'sync-versions': syncVersions,
     'test-packages': pgt(testPackages),
+    transform: pgt(transform),
     tune: pgt(tune),
     upgrade: pgt(upgrade),
     up: pgt(upgrade),