feat(pgpm): object-level apply routing (route a table and a function to different schemas)

__fixtures__/apply/routing/packages/catalog-module/catalog-module.control

@@ -0,0 +1,7 @@
+# catalog-module extension
+comment = 'catalog-module extension'
+default_version = '0.0.1'
+module_pathname = '$libdir/catalog-module'
+requires = 'plpgsql'
+relocatable = false
+superuser = false

__fixtures__/apply/routing/packages/catalog-module/deploy/schemas/catalog/procedures/product_count.sql

@@ -0,0 +1,13 @@
+-- Deploy schemas/catalog/procedures/product_count to pg
+
+-- requires: schemas/catalog/schema
+-- requires: schemas/reporting/schema
+-- requires: schemas/catalog/tables/products/table
+
+BEGIN;
+
+CREATE FUNCTION catalog.product_count() RETURNS bigint AS $$
+  SELECT count(*) FROM catalog.products;
+$$ LANGUAGE sql STABLE;
+
+COMMIT;

__fixtures__/apply/routing/packages/catalog-module/deploy/schemas/catalog/schema.sql

@@ -0,0 +1,7 @@
+-- Deploy schemas/catalog/schema to pg
+
+BEGIN;
+
+CREATE SCHEMA catalog;
+
+COMMIT;

__fixtures__/apply/routing/packages/catalog-module/deploy/schemas/catalog/tables/products/table.sql

@@ -0,0 +1,12 @@
+-- Deploy schemas/catalog/tables/products/table to pg
+
+-- requires: schemas/catalog/schema
+
+BEGIN;
+
+CREATE TABLE catalog.products (
+  id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
+  name text NOT NULL
+);
+
+COMMIT;

__fixtures__/apply/routing/packages/catalog-module/deploy/schemas/reporting/schema.sql

@@ -0,0 +1,7 @@
+-- Deploy schemas/reporting/schema to pg
+
+BEGIN;
+
+CREATE SCHEMA reporting;
+
+COMMIT;

__fixtures__/apply/routing/packages/catalog-module/pgpm.plan

@@ -0,0 +1,8 @@
+%syntax-version=1.0.0
+%project=catalog-module
+%uri=catalog-module
+
+schemas/catalog/schema 2024-01-01T00:00:00Z Dev <dev@example.com> # add catalog schema
+schemas/reporting/schema 2024-01-01T00:00:01Z Dev <dev@example.com> # add reporting schema
+schemas/catalog/tables/products/table [schemas/catalog/schema] 2024-01-01T00:00:02Z Dev <dev@example.com> # add products table
+schemas/catalog/procedures/product_count [schemas/catalog/schema schemas/reporting/schema schemas/catalog/tables/products/table] 2024-01-01T00:00:03Z Dev <dev@example.com> # add product_count

__fixtures__/apply/routing/packages/catalog-module/revert/schemas/catalog/procedures/product_count.sql

@@ -0,0 +1,7 @@
+-- Revert schemas/catalog/procedures/product_count from pg
+
+BEGIN;
+
+DROP FUNCTION catalog.product_count;
+
+COMMIT;

__fixtures__/apply/routing/packages/catalog-module/revert/schemas/catalog/schema.sql

@@ -0,0 +1,7 @@
+-- Revert schemas/catalog/schema from pg
+
+BEGIN;
+
+DROP SCHEMA catalog;
+
+COMMIT;

__fixtures__/apply/routing/packages/catalog-module/revert/schemas/catalog/tables/products/table.sql

@@ -0,0 +1,7 @@
+-- Revert schemas/catalog/tables/products/table from pg
+
+BEGIN;
+
+DROP TABLE catalog.products;
+
+COMMIT;

__fixtures__/apply/routing/packages/catalog-module/revert/schemas/reporting/schema.sql

@@ -0,0 +1,7 @@
+-- Revert schemas/reporting/schema from pg
+
+BEGIN;
+
+DROP SCHEMA reporting;
+
+COMMIT;

__fixtures__/apply/routing/packages/catalog-module/verify/schemas/catalog/procedures/product_count.sql

@@ -0,0 +1,7 @@
+-- Verify schemas/catalog/procedures/product_count on pg
+
+BEGIN;
+
+SELECT catalog.product_count();
+
+ROLLBACK;

__fixtures__/apply/routing/packages/catalog-module/verify/schemas/catalog/schema.sql

@@ -0,0 +1,7 @@
+-- Verify schemas/catalog/schema on pg
+
+BEGIN;
+
+SELECT pg_catalog.has_schema_privilege('catalog', 'usage');
+
+ROLLBACK;

__fixtures__/apply/routing/packages/catalog-module/verify/schemas/catalog/tables/products/table.sql

@@ -0,0 +1,7 @@
+-- Verify schemas/catalog/tables/products/table on pg
+
+BEGIN;
+
+SELECT id, name FROM catalog.products WHERE FALSE;
+
+ROLLBACK;

__fixtures__/apply/routing/packages/catalog-module/verify/schemas/reporting/schema.sql

@@ -0,0 +1,7 @@
+-- Verify schemas/reporting/schema on pg
+
+BEGIN;
+
+SELECT pg_catalog.has_schema_privilege('reporting', 'usage');
+
+ROLLBACK;

__fixtures__/apply/routing/packages/shop-a/pgpm.apply.json

@@ -0,0 +1,15 @@
+{
+  "source": "catalog-module",
+  "schemas": {
+    "catalog": "shop_a",
+    "reporting": "analytics_a"
+  },
+  "route": [
+    {
+      "fromSchema": "catalog",
+      "kind": "function",
+      "name": "product_count",
+      "toSchema": "analytics_a"
+    }
+  ]
+}

__fixtures__/apply/routing/packages/shop-b/pgpm.apply.json

@@ -0,0 +1,15 @@
+{
+  "source": "catalog-module",
+  "schemas": {
+    "catalog": "shop_b",
+    "reporting": "analytics_b"
+  },
+  "route": [
+    {
+      "fromSchema": "catalog",
+      "kind": "function",
+      "name": "product_count",
+      "toSchema": "analytics_b"
+    }
+  ]
+}

__fixtures__/apply/routing/pgpm.json

@@ -0,0 +1,5 @@
+{
+    "packages": [
+        "packages/*"
+    ]
+}

pgpm/core/__tests__/apply/apply-routing.test.ts

@@ -0,0 +1,183 @@
+import { rmSync } from 'fs';
+
+import {
+  clearApplyMaterializationCache,
+  materializeApplyModule,
+  parseApplySpec,
+  readApplySpec
+} from '../../src/apply';
+import { PgpmPackage } from '../../src/core/class/pgpm';
+import { CoreDeployTestFixture } from '../../test-utils/CoreDeployTestFixture';
+import { TestDatabase } from '../../test-utils/TestDatabase';
+import { TestFixture } from '../../test-utils/TestFixture';
+
+const at = '/ws/packages/shop-a/pgpm.apply.json';
+
+describe('apply spec parsing — object routes', () => {
+  it('accepts object routes alongside a schema default', () => {
+    const spec = parseApplySpec(
+      JSON.stringify({
+        source: 'catalog-module',
+        schemas: { catalog: 'shop_a', reporting: 'analytics_a' },
+        route: [
+          { fromSchema: 'catalog', kind: 'function', name: 'product_count', toSchema: 'analytics_a' }
+        ]
+      }),
+      at
+    );
+    expect(spec.schemas).toEqual({ catalog: 'shop_a', reporting: 'analytics_a' });
+    expect(spec.route).toEqual([
+      { fromSchema: 'catalog', kind: 'function', name: 'product_count', toSchema: 'analytics_a' }
+    ]);
+  });
+
+  it('accepts a route-only spec (no schema default)', () => {
+    const spec = parseApplySpec(
+      JSON.stringify({
+        source: 'catalog-module',
+        route: [{ fromSchema: 'catalog', kind: 'table', name: 'products', toSchema: 'shop_a' }]
+      }),
+      at
+    );
+    expect(spec.schemas).toBeUndefined();
+    expect(spec.route).toHaveLength(1);
+  });
+
+  it.each([
+    [{ source: 'x' }, /at least one of "schemas".*or "route"/],
+    [{ source: 'x', route: [] }, /"route" must be a non-empty array/],
+    [{ source: 'x', route: [{ fromSchema: 'a', kind: 'widget', name: 'n', toSchema: 'b' }] }, /route" entry/],
+    [{ source: 'x', route: [{ fromSchema: 'a', name: 'n', toSchema: 'b' }] }, /route" entry/],
+    [{ source: 'x', route: [{ fromSchema: '', kind: 'table', name: 'n', toSchema: 'b' }] }, /route" entry/]
+  ])('rejects invalid route specs %#', (spec, err) => {
+    expect(() => parseApplySpec(JSON.stringify(spec), at)).toThrow(err);
+  });
+});
+
+describe('materializeApplyModule — object routing', () => {
+  let fixture: TestFixture;
+
+  beforeAll(() => {
+    fixture = new TestFixture('apply', 'routing');
+  });
+
+  afterAll(() => fixture.cleanup());
+
+  it('fans a function and its table into different schemas, rewriting the cross-ref', async () => {
+    const sourceDir = fixture.fixturePath('packages', 'catalog-module');
+    const spec = readApplySpec(fixture.fixturePath('packages', 'shop-a'));
+    const { bundle, outDir } = await materializeApplyModule({ sourceDir, spec });
+    try {
+      // the table follows the schema-level default; the function is routed out
+      expect(bundle.manifest.deployOrder).toEqual([
+        'schemas/shop_a/schema',
+        'schemas/analytics_a/schema',
+        'schemas/shop_a/tables/products/table',
+        'schemas/analytics_a/procedures/product_count'
+      ]);
+
+      const proc = bundle.changes.find(
+        c => c.name === 'schemas/analytics_a/procedures/product_count'
+      )!;
+      // definition lands in the routed schema, its body reference follows the
+      // table into the schema-level target
+      expect(proc.deploy!.sql).toContain('analytics_a.product_count');
+      expect(proc.deploy!.sql).toContain('shop_a.products');
+      expect(proc.deploy!.sql).not.toMatch(/\bcatalog\./);
+
+      // the routed function still depends on the (renamed) table + schemas
+      expect(proc.dependencies).toEqual(
+        expect.arrayContaining([
+          'schemas/shop_a/schema',
+          'schemas/analytics_a/schema',
+          'schemas/shop_a/tables/products/table'
+        ])
+      );
+
+      // target schemas are created idempotently (destinations may pre-exist)
+      const schemaChange = bundle.changes.find(c => c.name === 'schemas/analytics_a/schema')!;
+      expect(schemaChange.deploy!.sql).toMatch(/CREATE SCHEMA IF NOT EXISTS analytics_a/i);
+    } finally {
+      rmSync(outDir, { recursive: true, force: true });
+    }
+  });
+});
+
+describe('apply object routing deployment (e2e)', () => {
+  let fixture: CoreDeployTestFixture;
+  let db: TestDatabase;
+
+  const functionExists = async (schema: string, name: string): Promise<boolean> => {
+    const res = await db.query(
+      `SELECT 1 FROM pg_proc p JOIN pg_namespace n ON n.oid = p.pronamespace
+       WHERE n.nspname = $1 AND p.proname = $2`,
+      [schema, name]
+    );
+    return res.rows.length > 0;
+  };
+
+  beforeAll(() => {
+    fixture = new CoreDeployTestFixture('apply', 'routing');
+  });
+
+  afterAll(async () => {
+    await fixture.cleanup();
+  });
+
+  beforeEach(async () => {
+    clearApplyMaterializationCache();
+    db = await fixture.setupTestDatabase();
+  });
+
+  test('routes a table and a function into different schemas across two instances', async () => {
+    // shop-a: both target schemas are created fresh by the apply
+    await fixture.deployModule('shop-a', db.name, ['apply', 'routing']);
+
+    // shop-b: the destination schema already exists — apply must not fail on it
+    await db.query('CREATE SCHEMA shop_b');
+    await fixture.deployModule('shop-b', db.name, ['apply', 'routing']);
+
+    // the source module itself is never deployed
+    expect(await db.exists('schema', 'catalog')).toBe(false);
+
+    // table follows the schema-level default; function is routed elsewhere
+    expect(await db.exists('table', 'shop_a.products')).toBe(true);
+    expect(await db.exists('table', 'shop_b.products')).toBe(true);
+    expect(await functionExists('analytics_a', 'product_count')).toBe(true);
+    expect(await functionExists('analytics_b', 'product_count')).toBe(true);
+    // the function did NOT land in the table's schema
+    expect(await functionExists('shop_a', 'product_count')).toBe(false);
+
+    // the table is transpiled twice — counts are independent per instance
+    await db.query(`INSERT INTO shop_a.products (name) VALUES ('a'), ('b')`);
+    await db.query(`INSERT INTO shop_b.products (name) VALUES ('c')`);
+    // the routed function reads across schemas into its instance's table
+    const a = await db.query('SELECT analytics_a.product_count() AS n');
+    const b = await db.query('SELECT analytics_b.product_count() AS n');
+    expect(Number(a.rows[0].n)).toBe(2);
+    expect(Number(b.rows[0].n)).toBe(1);
+
+    // registry attribution lands on the instances, not the source
+    const packages = new Set((await db.getDeployedChanges()).map((c: any) => c.package));
+    expect(packages.has('shop-a')).toBe(true);
+    expect(packages.has('shop-b')).toBe(true);
+    expect(packages.has('catalog-module')).toBe(false);
+  });
+
+  test('verify and revert work against the routed, re-derived instance', async () => {
+    await db.query('CREATE SCHEMA shop_b');
+    await fixture.deployModule('shop-a', db.name, ['apply', 'routing']);
+    await fixture.deployModule('shop-b', db.name, ['apply', 'routing']);
+
+    await fixture.verifyModule('shop-a', db.name, ['apply', 'routing']);
+    await fixture.verifyModule('shop-b', db.name, ['apply', 'routing']);
+
+    // revert the last-deployed instance; the other (deployed earlier) is
+    // untouched — each instance's routed objects revert independently
+    await fixture.revertModule('shop-b', db.name, ['apply', 'routing']);
+    expect(await functionExists('analytics_b', 'product_count')).toBe(false);
+    expect(await db.exists('table', 'shop_b.products')).toBe(false);
+    expect(await functionExists('analytics_a', 'product_count')).toBe(true);
+    expect(await db.exists('table', 'shop_a.products')).toBe(true);
+  });
+});