From f747dea9f05f9acb41c6e650a1c12848b55303c8 Mon Sep 17 00:00:00 2001 From: Dan Lynch Date: Wed, 22 Jul 2026 04:16:27 +0000 Subject: [PATCH] =?UTF-8?q?feat(pgpm-core):=20rebundlePlan=20=E2=80=94=20d?= =?UTF-8?q?ependency-aware=20chunk/merge=20primitive=20with=20byte-identic?= =?UTF-8?q?al=20gate?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pgpm/core/__tests__/rebundle/rebundle.test.ts | 139 ++++++++++++ pgpm/core/src/index.ts | 1 + pgpm/core/src/rebundle/index.ts | 2 + pgpm/core/src/rebundle/rebundle.ts | 206 ++++++++++++++++++ pgpm/core/src/rebundle/types.ts | 73 +++++++ 5 files changed, 421 insertions(+) create mode 100644 pgpm/core/__tests__/rebundle/rebundle.test.ts create mode 100644 pgpm/core/src/rebundle/index.ts create mode 100644 pgpm/core/src/rebundle/rebundle.ts create mode 100644 pgpm/core/src/rebundle/types.ts diff --git a/pgpm/core/__tests__/rebundle/rebundle.test.ts b/pgpm/core/__tests__/rebundle/rebundle.test.ts new file mode 100644 index 000000000..e39b3a1c0 --- /dev/null +++ b/pgpm/core/__tests__/rebundle/rebundle.test.ts @@ -0,0 +1,139 @@ +import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from 'fs'; +import { tmpdir } from 'os'; +import { dirname, join } from 'path'; + +import { + assembleChunkSql, + rebundlePlan, + verifyRebundleInvariant, +} from '../../src/rebundle'; +import { resolveWithPlan } from '../../src/resolution/resolve'; + +let moduleDir: string; + +const PLAN = `%syntax-version=1.0.0 +%project=my-module +%uri=my-module + +schemas/auth/schema 2024-01-01T00:00:00Z Dev # add schema +schemas/auth/tables/users [schemas/auth/schema] 2024-01-01T00:00:01Z Dev # users +schemas/auth/tables/sessions [schemas/auth/tables/users] 2024-01-01T00:00:02Z Dev # sessions +schemas/billing/schema [schemas/auth/schema] 2024-01-01T00:00:03Z Dev # billing schema +schemas/billing/tables/invoices [schemas/billing/schema schemas/auth/tables/users] 2024-01-01T00:00:04Z Dev # invoices +@v1.0.0 2024-01-01T00:00:05Z Dev # release +`; + +const CHANGES = [ + 'schemas/auth/schema', + 'schemas/auth/tables/users', + 'schemas/auth/tables/sessions', + 'schemas/billing/schema', + 'schemas/billing/tables/invoices', +]; + +function writeScript(rel: string, content: string): void { + const file = join(moduleDir, rel); + mkdirSync(dirname(file), { recursive: true }); + writeFileSync(file, content); +} + +beforeEach(() => { + moduleDir = mkdtempSync(join(tmpdir(), 'pgpm-rebundle-')); + writeFileSync(join(moduleDir, 'pgpm.plan'), PLAN); + + for (const change of CHANGES) { + writeScript(`deploy/${change}.sql`, `-- Deploy ${change}\nCREATE /* ${change} */ ;\n`); + writeScript(`revert/${change}.sql`, `-- Revert ${change}\nDROP /* ${change} */ ;\n`); + writeScript(`verify/${change}.sql`, `-- Verify ${change}\nSELECT '${change}';\n`); + } +}); + +afterEach(() => { + rmSync(moduleDir, { recursive: true, force: true }); +}); + +describe('rebundlePlan', () => { + it('groups changes into folder-derived chunks by default', () => { + const result = rebundlePlan(moduleDir); + expect(result.deployOrder).toEqual(['auth', 'billing']); + + const auth = result.chunks.find(c => c.name === 'auth')!; + const billing = result.chunks.find(c => c.name === 'billing')!; + + expect(auth.deploy).toEqual([ + 'schemas/auth/schema', + 'schemas/auth/tables/users', + 'schemas/auth/tables/sessions', + ]); + expect(billing.deploy).toEqual([ + 'schemas/billing/schema', + 'schemas/billing/tables/invoices', + ]); + }); + + it('records verify order as deploy order and revert as its reverse', () => { + const auth = rebundlePlan(moduleDir).chunks.find(c => c.name === 'auth')!; + expect(auth.verify).toEqual(auth.deploy); + expect(auth.revert).toEqual([...auth.deploy].reverse()); + }); + + it('derives chunk-level dependencies from cross-chunk change edges', () => { + const billing = rebundlePlan(moduleDir).chunks.find(c => c.name === 'billing')!; + // billing changes require auth/schema and auth/tables/users → depends on 'auth' + expect(billing.dependencies).toEqual(['auth']); + }); + + it('collapses everything into one chunk with boundary "none"', () => { + const result = rebundlePlan(moduleDir, { boundary: 'none' }); + expect(result.chunks).toHaveLength(1); + expect(result.chunks[0].deploy).toEqual(CHANGES); + }); + + it('splits a boundary group when maxChunkSize is exceeded', () => { + const result = rebundlePlan(moduleDir, { boundary: 'none', maxChunkSize: 2 }); + expect(result.chunks.map(c => c.deploy.length)).toEqual([2, 2, 1]); + // Members remain in contiguous source order across the split. + expect(result.chunks.flatMap(c => c.deploy)).toEqual(CHANGES); + }); +}); + +describe('assembleChunkSql', () => { + it('concatenates member deploy files in deploy order', () => { + const auth = rebundlePlan(moduleDir).chunks.find(c => c.name === 'auth')!; + const sql = assembleChunkSql(moduleDir, auth, 'deploy'); + expect(sql.indexOf('schemas/auth/schema')).toBeLessThan(sql.indexOf('schemas/auth/tables/users')); + expect(sql.indexOf('schemas/auth/tables/users')).toBeLessThan(sql.indexOf('schemas/auth/tables/sessions')); + }); + + it('concatenates member revert files in reverse order', () => { + const auth = rebundlePlan(moduleDir).chunks.find(c => c.name === 'auth')!; + const sql = assembleChunkSql(moduleDir, auth, 'revert'); + expect(sql.indexOf('schemas/auth/tables/sessions')).toBeLessThan(sql.indexOf('schemas/auth/schema')); + }); +}); + +describe('verifyRebundleInvariant', () => { + it('reproduces the original resolved output byte-for-byte (folder boundary)', () => { + const result = rebundlePlan(moduleDir); + const check = verifyRebundleInvariant(moduleDir, result); + expect(check).toEqual({ ok: true, mismatches: [] }); + }); + + it('is byte-identical for the single-chunk dial position', () => { + const result = rebundlePlan(moduleDir, { boundary: 'none' }); + expect(verifyRebundleInvariant(moduleDir, result).ok).toBe(true); + }); + + it('is byte-identical for the size-capped dial position', () => { + const result = rebundlePlan(moduleDir, { boundary: 'none', maxChunkSize: 2 }); + expect(verifyRebundleInvariant(moduleDir, result).ok).toBe(true); + }); + + it('assembled deploy equals resolveWithPlan deploy', () => { + const result = rebundlePlan(moduleDir); + const assembled = result.chunks + .map(c => assembleChunkSql(moduleDir, c, 'deploy')) + .join('\n'); + expect(assembled).toEqual(resolveWithPlan(moduleDir, 'deploy')); + }); +}); diff --git a/pgpm/core/src/index.ts b/pgpm/core/src/index.ts index b9ec507ce..094badea9 100644 --- a/pgpm/core/src/index.ts +++ b/pgpm/core/src/index.ts @@ -1,5 +1,6 @@ export * from './core/class/pgpm'; export * from './slice'; +export * from './rebundle'; export * from './extensions/extensions'; export * from './modules/modules'; export * from './packaging/package'; diff --git a/pgpm/core/src/rebundle/index.ts b/pgpm/core/src/rebundle/index.ts new file mode 100644 index 000000000..2e61ce403 --- /dev/null +++ b/pgpm/core/src/rebundle/index.ts @@ -0,0 +1,2 @@ +export * from './types'; +export * from './rebundle'; diff --git a/pgpm/core/src/rebundle/rebundle.ts b/pgpm/core/src/rebundle/rebundle.ts new file mode 100644 index 000000000..0711e2a41 --- /dev/null +++ b/pgpm/core/src/rebundle/rebundle.ts @@ -0,0 +1,206 @@ +import { readFileSync } from 'fs'; +import { join } from 'path'; + +import { getChanges } from '../files/plan/parser'; +import { resolveWithPlan } from '../resolution/resolve'; +import { + buildDependencyGraph, + buildPackageDependencies, + computeDeployOrder, + detectPackageCycle, + extractPackageFromPath, +} from '../slice/slice'; +import { parsePlanFile } from '../files/plan/parser'; +import { DependencyGraph, SliceWarning } from '../slice/types'; +import { Chunk, RebundleResult, RebundleStrategy } from './types'; + +/** + * Derive the boundary key for a change under the given strategy. + * Consecutive changes sharing a key may be merged into the same chunk. + */ +function boundaryKey(changeName: string, strategy: RebundleStrategy): string { + if (strategy.boundary === 'none') return ''; + const depth = strategy.depth ?? 1; + const prefix = strategy.prefixToStrip ?? 'schemas'; + return extractPackageFromPath(changeName, depth, prefix); +} + +/** + * Assign changes to chunks by walking the source plan in order and greedily + * accumulating consecutive changes that share a boundary key, sealing a chunk + * when the key changes, the size cap is reached, or a member carries a tag + * (tags mark points you can `deploy --to @tag`, so they must stay chunk seams). + * + * Because chunks are contiguous runs of the plan, member and chunk order both + * follow the source, which is what guarantees byte-identical output. + */ +function assignChunks( + orderedChanges: string[], + graph: DependencyGraph, + strategy: RebundleStrategy +): { name: string; members: string[] }[] { + const chunks: { name: string; members: string[]; key: string }[] = []; + const seen = new Map(); + const maxSize = strategy.maxChunkSize && strategy.maxChunkSize > 0 ? strategy.maxChunkSize : Infinity; + + let current: { name: string; members: string[]; key: string } | null = null; + + const seal = (): void => { + current = null; + }; + + for (const change of orderedChanges) { + const key = boundaryKey(change, strategy); + const changeHasTag = (graph.tags.get(change)?.length ?? 0) > 0; + + if ( + current && + current.key === key && + current.members.length < maxSize + ) { + current.members.push(change); + } else { + // Deterministic, collision-free chunk name: boundary key (or 'chunk'), + // suffixed with an ordinal when the same key splits across size caps. + const base = key || 'chunk'; + const ordinal = (seen.get(base) ?? 0) + 1; + seen.set(base, ordinal); + const name = ordinal === 1 ? base : `${base}~${ordinal}`; + current = { name, members: [change], key }; + chunks.push(current); + } + + // A member that is a tag anchor seals the chunk after inclusion. + if (changeHasTag) seal(); + } + + return chunks.map(({ name, members }) => ({ name, members })); +} + +/** + * Compute a rebundle plan for a pgpm module directory. + * + * Uses the existing plan parser, dependency graph, and package-dependency / + * deploy-order primitives from the slicer — chunks are treated exactly like + * packages for dependency and ordering purposes. No plan re-parsing or path + * logic is reinvented here. + */ +export function rebundlePlan(moduleDir: string, strategy: RebundleStrategy = {}): RebundleResult { + const planPath = join(moduleDir, 'pgpm.plan'); + const planResult = parsePlanFile(planPath); + if (!planResult.data) { + const msg = planResult.errors?.map(e => `Line ${e.line}: ${e.message}`).join('\n') || 'Unknown error'; + throw new Error(`Failed to parse plan file: ${msg}`); + } + + const graph = buildDependencyGraph(planResult.data); + const orderedChanges = planResult.data.changes.map(c => c.name); + + const warnings: SliceWarning[] = []; + + const rawChunks = assignChunks(orderedChanges, graph, strategy); + + // Reverse lookup change -> chunk, reusing slice's package-dep machinery. + const assignments = new Map>(); + const changeToChunk = new Map(); + for (const chunk of rawChunks) { + assignments.set(chunk.name, new Set(chunk.members)); + for (const member of chunk.members) changeToChunk.set(member, chunk.name); + } + + const chunkDeps = buildPackageDependencies(graph, assignments); + + const cycle = detectPackageCycle(chunkDeps); + if (cycle) { + warnings.push({ + type: 'cycle_detected', + message: `Chunk cycle detected: ${cycle.join(' -> ')}`, + suggestedAction: 'Widen the boundary or raise maxChunkSize so cyclic changes share a chunk', + }); + } + + const deployOrder = computeDeployOrder(chunkDeps); + + // Because chunks are contiguous runs of the source plan, the greedy creation + // order is already a valid deploy order. If the dependency-derived order + // disagrees, the boundary split a back-edge — surface it rather than silently + // reordering (that would break byte-identical output). + const creationOrder = rawChunks.map(c => c.name); + const orderMatches = + deployOrder.length === creationOrder.length && + deployOrder.every((name, i) => name === creationOrder[i]); + if (!orderMatches) { + warnings.push({ + type: 'heavy_cross_deps', + message: 'Chunk deploy order differs from source order; a boundary split a dependency edge', + suggestedAction: 'Adjust the boundary/maxChunkSize so dependent changes stay contiguous', + }); + } + + const chunkByName = new Map(rawChunks.map(c => [c.name, c.members])); + + const chunks: Chunk[] = creationOrder.map(name => { + const members = chunkByName.get(name)!; + return { + name, + deploy: members, + verify: members, + revert: [...members].reverse(), + dependencies: [...(chunkDeps.get(name) ?? new Set())].sort(), + }; + }); + + return { + chunks, + deployOrder: creationOrder, + warnings, + sourceChanges: planResult.data.changes, + }; +} + +/** + * Assemble the merged SQL for one chunk in the given direction by concatenating + * its member script files — the same file-concatenation `resolveWithPlan` uses, + * just scoped to the chunk. Members are already correctly ordered per direction. + */ +export function assembleChunkSql( + moduleDir: string, + chunk: Chunk, + scriptType: 'deploy' | 'revert' | 'verify' = 'deploy' +): string { + const members = scriptType === 'revert' ? chunk.revert : scriptType === 'verify' ? chunk.verify : chunk.deploy; + const parts: string[] = []; + for (const change of members) { + parts.push(readFileSync(join(moduleDir, scriptType, `${change}.sql`), 'utf-8')); + } + return parts.join('\n'); +} + +/** + * Byte-identical gate: assembling all chunks in deploy order must reproduce the + * module's original resolved output in every direction the engine cares about + * (deploy topological, verify in deploy order, revert reversed). This is the + * invariant that makes any granularity-dial position safe. + */ +export function verifyRebundleInvariant( + moduleDir: string, + result: RebundleResult +): { ok: boolean; mismatches: ('deploy' | 'revert' | 'verify')[] } { + const mismatches: ('deploy' | 'revert' | 'verify')[] = []; + + const assembled = (scriptType: 'deploy' | 'revert' | 'verify'): string => { + const chunks = scriptType === 'revert' ? [...result.chunks].reverse() : result.chunks; + return chunks.map(chunk => assembleChunkSql(moduleDir, chunk, scriptType)).join('\n'); + }; + + // deploy / revert have canonical resolvers; verify concatenates in deploy order. + const originalVerify = getChanges(join(moduleDir, 'pgpm.plan')) + .map(change => readFileSync(join(moduleDir, 'verify', `${change}.sql`), 'utf-8')) + .join('\n'); + + if (assembled('deploy') !== resolveWithPlan(moduleDir, 'deploy')) mismatches.push('deploy'); + if (assembled('revert') !== resolveWithPlan(moduleDir, 'revert')) mismatches.push('revert'); + if (assembled('verify') !== originalVerify) mismatches.push('verify'); + + return { ok: mismatches.length === 0, mismatches }; +} diff --git a/pgpm/core/src/rebundle/types.ts b/pgpm/core/src/rebundle/types.ts new file mode 100644 index 000000000..3ecfd93fc --- /dev/null +++ b/pgpm/core/src/rebundle/types.ts @@ -0,0 +1,73 @@ +import { Change } from '../files/types'; +import { SliceWarning } from '../slice/types'; + +/** + * Strategy for rebundling a module's changes into fewer, larger chunks. + * + * Rebundling walks the source plan in order and greedily accumulates + * consecutive changes into a chunk until a boundary is hit. Because chunks + * are contiguous runs of the source plan, the concatenated deploy/verify/ + * revert output is byte-identical to the original module — the granularity + * dial never reorders statements, it only decides where the chunk seams fall. + */ +export interface RebundleStrategy { + /** + * How chunk boundaries are chosen: + * - 'folder' (default): start a new chunk when the folder-derived key + * changes (uses the same path extraction as slice's folder strategy) + * - 'none': one chunk for the whole module, subject only to `maxChunkSize` + */ + boundary?: 'folder' | 'none'; + + /** Depth in the change path used to derive the folder key (default: 1) */ + depth?: number; + + /** Path prefix stripped before deriving the folder key (default: 'schemas') */ + prefixToStrip?: string; + + /** + * Maximum number of member changes per chunk. When reached, the current + * chunk is sealed and a new one begins even if the boundary key is unchanged. + * Undefined or 0 means unlimited. + */ + maxChunkSize?: number; +} + +/** + * A single rebundled chunk: a contiguous run of source-plan changes that will + * be merged into one migration. Member ordering is preserved in all three + * directions the pgpm engine cares about. + */ +export interface Chunk { + /** Deterministic chunk identifier */ + name: string; + + /** Member change names in deploy order (== source plan order) */ + deploy: string[]; + + /** Member change names in verify order (same as deploy) */ + verify: string[]; + + /** Member change names in revert order (reverse of deploy) */ + revert: string[]; + + /** Other chunk names this chunk depends on (always earlier in deployOrder) */ + dependencies: string[]; +} + +/** + * Result of computing a rebundle plan for a module. + */ +export interface RebundleResult { + /** Chunks in deploy order */ + chunks: Chunk[]; + + /** Chunk names in deploy order */ + deployOrder: string[]; + + /** Warnings encountered while rebundling */ + warnings: SliceWarning[]; + + /** The source changes (in plan order) that were rebundled */ + sourceChanges: Change[]; +}