import { stableJson } from './schema_json.ts' interface SchemaColumn { table: string name: string position: number type: string notNull: boolean default: string | null identity: string generated: string } interface SchemaConstraint { table: string name: string type: string definition: string valid?: boolean } interface SchemaIndex { table: string name: string definition: string valid: boolean constraintOwned?: boolean } interface SchemaTable { name: string kind: string reloptions: unknown } interface SchemaType { name: string kind: string definition: unknown } interface SchemaSequence { name: string type: string start: string increment: string min: string max: string cache: string cycle: boolean ownedTable: string | null ownedColumn: string | null } interface SchemaFunction { name: string arguments: string definition: string } interface SchemaCatalog { version?: number columns?: SchemaColumn[] constraints?: SchemaConstraint[] indexes?: SchemaIndex[] tables?: SchemaTable[] types?: SchemaType[] sequences?: SchemaSequence[] functions?: SchemaFunction[] } export interface SchemaCompatibilityIssue { kind: 'catalog_version' | 'column' | 'constraint' | 'function' | 'index' | 'sequence' | 'table' | 'type' object: string reason: string } // Strict comparison is retained for snapshot tooling. The release gate below is // directional because PostgreSQL logical subscribers may deliberately retain // columns and supporting objects that the publisher has already dropped. export function readReplicaSchemaCompatibilityIssues(expected: unknown, actual: unknown): SchemaCompatibilityIssue[] { return readReplicaSchemaCompatibilityIssuesInternal(expected, actual, false) } // A logical subscriber must accept every publisher column, but it can safely // retain nullable or default-backed legacy columns until the publisher removes // them. This is the compatibility relation used before primary migrations. export function readReplicaSubscriberCompatibilityIssues(expected: unknown, actual: unknown): SchemaCompatibilityIssue[] { return readReplicaSchemaCompatibilityIssuesInternal(expected, actual, true) } function readReplicaSchemaCompatibilityIssuesInternal( expected: unknown, actual: unknown, allowSafeSubscriberOnlyObjects: boolean, ): SchemaCompatibilityIssue[] { const expectedCatalog = assertSchemaCatalog(expected, 'expected') const actualCatalog = assertSchemaCatalog(actual, 'actual') const issues: SchemaCompatibilityIssue[] = [] if (expectedCatalog.version !== actualCatalog.version) { issues.push({ kind: 'catalog_version', object: 'catalog', reason: `expected version ${expectedCatalog.version ?? 'missing'}, found ${actualCatalog.version ?? 'missing'}`, }) } compareTables(expectedCatalog, actualCatalog, issues) compareColumns(expectedCatalog, actualCatalog, issues, allowSafeSubscriberOnlyObjects) compareConstraints(expectedCatalog, actualCatalog, issues, allowSafeSubscriberOnlyObjects) compareIndexes(expectedCatalog, actualCatalog, issues, allowSafeSubscriberOnlyObjects) compareTypes(expectedCatalog, actualCatalog, issues, allowSafeSubscriberOnlyObjects) compareSequences(expectedCatalog, actualCatalog, issues, allowSafeSubscriberOnlyObjects) compareFunctions(expectedCatalog, actualCatalog, issues, allowSafeSubscriberOnlyObjects) return issues } function compareTables( expected: SchemaCatalog, actual: SchemaCatalog, issues: SchemaCompatibilityIssue[], ): void { const expectedTables = new Map((expected.tables ?? []).map(table => [table.name, table])) const actualTables = new Map((actual.tables ?? []).map(table => [table.name, table])) for (const expectedTable of expected.tables ?? []) { const actualTable = actualTables.get(expectedTable.name) if (!actualTable) { issues.push({ kind: 'table', object: expectedTable.name, reason: 'missing required table' }) continue } if (actualTable.kind !== expectedTable.kind) issues.push({ kind: 'table', object: expectedTable.name, reason: `expected kind ${expectedTable.kind}, found ${actualTable.kind}` }) } for (const actualTable of actual.tables ?? []) { if (!expectedTables.has(actualTable.name)) issues.push({ kind: 'table', object: actualTable.name, reason: 'unexpected table' }) } } function compareColumns( expected: SchemaCatalog, actual: SchemaCatalog, issues: SchemaCompatibilityIssue[], allowSafeSubscriberOnlyObjects: boolean, ): void { const expectedColumns = new Map((expected.columns ?? []).map(column => [columnKey(column), column])) const actualColumns = new Map((actual.columns ?? []).map(column => [columnKey(column), column])) compareExpectedColumns(expected.columns ?? [], actualColumns, issues) compareSubscriberOnlyColumns( actual.columns ?? [], expectedColumns, issues, allowSafeSubscriberOnlyObjects, ) } function compareExpectedColumns( expectedColumns: SchemaColumn[], actualColumns: Map, issues: SchemaCompatibilityIssue[], ): void { for (const expectedColumn of expectedColumns) { const key = columnKey(expectedColumn) const actualColumn = actualColumns.get(key) if (!actualColumn) { issues.push({ kind: 'column', object: key, reason: 'missing required column' }) continue } compareMatchingColumns(expectedColumn, actualColumn, issues) } } function compareMatchingColumns( expectedColumn: SchemaColumn, actualColumn: SchemaColumn, issues: SchemaCompatibilityIssue[], ): void { const key = columnKey(expectedColumn) if (actualColumn.type !== expectedColumn.type) { issues.push({ kind: 'column', object: key, reason: `expected type ${expectedColumn.type}, found ${actualColumn.type}`, }) } if (actualColumn.notNull !== expectedColumn.notNull) { issues.push({ kind: 'column', object: key, reason: expectedColumn.notNull ? 'publisher is NOT NULL while subscriber accepts NULL' : 'subscriber is NOT NULL while publisher accepts NULL', }) } if (actualColumn.identity !== expectedColumn.identity) { issues.push({ kind: 'column', object: key, reason: `expected identity kind ${displayValue(expectedColumn.identity)}, found ${displayValue(actualColumn.identity)}`, }) } if (actualColumn.generated !== expectedColumn.generated) { issues.push({ kind: 'column', object: key, reason: `expected generated kind ${displayValue(expectedColumn.generated)}, found ${displayValue(actualColumn.generated)}`, }) } } function compareSubscriberOnlyColumns( actualColumns: SchemaColumn[], expectedColumns: Map, issues: SchemaCompatibilityIssue[], allowSafeSubscriberOnlyObjects: boolean, ): void { for (const actualColumn of actualColumns) { const key = columnKey(actualColumn) if (expectedColumns.has(key)) continue if (allowSafeSubscriberOnlyObjects && subscriberOnlyColumnCanAcceptPublisherRows(actualColumn)) continue issues.push({ kind: 'column', object: key, reason: allowSafeSubscriberOnlyObjects ? 'subscriber-only column can reject replicated rows' : 'unexpected column', }) } } function compareConstraints( expected: SchemaCatalog, actual: SchemaCatalog, issues: SchemaCompatibilityIssue[], allowSafeSubscriberOnlyObjects: boolean, ): void { const expectedConstraints = new Map((expected.constraints ?? []).map(constraint => [constraintKey(constraint), constraint])) const actualConstraints = new Map((actual.constraints ?? []).map(constraint => [constraintKey(constraint), constraint])) for (const expectedConstraint of expected.constraints ?? []) { const key = constraintKey(expectedConstraint) const actualConstraint = actualConstraints.get(key) if (expectedConstraint.type === 'c') { // Publisher CHECK constraints are optional on read-only logical subscribers. // Missing CHECK constraints are ignored; stale subscriber CHECK constraints must not block deploy. if ( actualConstraint && actualConstraint.type !== 'c' ) { issues.push({ kind: 'constraint', object: key, reason: 'subscriber CHECK constraint differs' }) } else if ( actualConstraint && !allowSafeSubscriberOnlyObjects && actualConstraint.definition !== expectedConstraint.definition ) { issues.push({ kind: 'constraint', object: key, reason: 'subscriber CHECK constraint differs' }) } continue } if (!actualConstraint) { issues.push({ kind: 'constraint', object: key, reason: 'missing required constraint' }) continue } if (actualConstraint.type !== expectedConstraint.type) { issues.push({ kind: 'constraint', object: key, reason: `expected type ${expectedConstraint.type}, found ${actualConstraint.type}`, }) } if (actualConstraint.definition !== expectedConstraint.definition) issues.push({ kind: 'constraint', object: key, reason: 'constraint definition differs' }) if (expectedConstraint.valid !== undefined || actualConstraint.valid !== undefined) { if (actualConstraint.valid !== expectedConstraint.valid) { issues.push({ kind: 'constraint', object: key, reason: `expected valid ${displayValue(expectedConstraint.valid)}, found ${displayValue(actualConstraint.valid)}`, }) } } } for (const actualConstraint of actual.constraints ?? []) { const key = constraintKey(actualConstraint) if (!expectedConstraints.has(key)) issues.push({ kind: 'constraint', object: key, reason: 'unexpected constraint' }) } } function compareIndexes( expected: SchemaCatalog, actual: SchemaCatalog, issues: SchemaCompatibilityIssue[], allowSafeSubscriberOnlyObjects: boolean, ): void { const expectedIndexes = new Map((expected.indexes ?? []).map(index => [index.name, index])) const actualIndexes = new Map((actual.indexes ?? []).map(index => [index.name, index])) compareExpectedIndexes(expected.indexes ?? [], actualIndexes, issues) compareSubscriberOnlyIndexes( actual.indexes ?? [], expectedIndexes, issues, allowSafeSubscriberOnlyObjects, ) } function compareExpectedIndexes( expectedIndexes: SchemaIndex[], actualIndexes: Map, issues: SchemaCompatibilityIssue[], ): void { for (const expectedIndex of expectedIndexes) { const actualIndex = actualIndexes.get(expectedIndex.name) if (!actualIndex) { issues.push({ kind: 'index', object: expectedIndex.name, reason: 'missing required index' }) continue } compareMatchingIndexes(expectedIndex, actualIndex, issues) } } function compareMatchingIndexes( expectedIndex: SchemaIndex, actualIndex: SchemaIndex, issues: SchemaCompatibilityIssue[], ): void { if (actualIndex.table !== expectedIndex.table) { issues.push({ kind: 'index', object: expectedIndex.name, reason: `expected table ${expectedIndex.table}, found ${actualIndex.table}`, }) } if (actualIndex.definition !== expectedIndex.definition) issues.push({ kind: 'index', object: expectedIndex.name, reason: 'index definition differs' }) if (actualIndex.valid !== expectedIndex.valid) { issues.push({ kind: 'index', object: expectedIndex.name, reason: expectedIndex.valid && !actualIndex.valid ? 'index is invalid' : `expected valid ${displayValue(expectedIndex.valid)}, found ${displayValue(actualIndex.valid)}`, }) } if (indexConstraintOwnershipDiffers(expectedIndex, actualIndex)) { issues.push({ kind: 'index', object: expectedIndex.name, reason: `expected constraintOwned ${displayValue(expectedIndex.constraintOwned)}, found ${displayValue(actualIndex.constraintOwned)}`, }) } } function indexConstraintOwnershipDiffers( expectedIndex: SchemaIndex, actualIndex: SchemaIndex, ): boolean { return expectedIndex.constraintOwned !== undefined && actualIndex.constraintOwned !== undefined && actualIndex.constraintOwned !== expectedIndex.constraintOwned } function compareSubscriberOnlyIndexes( actualIndexes: SchemaIndex[], expectedIndexes: Map, issues: SchemaCompatibilityIssue[], allowSafeSubscriberOnlyObjects: boolean, ): void { for (const actualIndex of actualIndexes) { if (expectedIndexes.has(actualIndex.name)) continue if (allowSafeSubscriberOnlyObjects && subscriberOnlyIndexCannotRejectPublisherRows(actualIndex)) continue issues.push({ kind: 'index', object: actualIndex.name, reason: allowSafeSubscriberOnlyObjects ? 'subscriber-only unique index can reject replicated rows' : 'unexpected index adds storage and write cost', }) } } function compareTypes( expected: SchemaCatalog, actual: SchemaCatalog, issues: SchemaCompatibilityIssue[], allowSafeSubscriberOnlyObjects: boolean, ): void { const expectedTypes = new Map((expected.types ?? []).map(type => [type.name, type])) const actualTypes = new Map((actual.types ?? []).map(type => [type.name, type])) for (const expectedType of expected.types ?? []) { const actualType = actualTypes.get(expectedType.name) if (!actualType) { issues.push({ kind: 'type', object: expectedType.name, reason: 'missing required type' }) continue } if (actualType.kind !== expectedType.kind) { issues.push({ kind: 'type', object: expectedType.name, reason: `expected kind ${expectedType.kind}, found ${actualType.kind}`, }) continue } if (!sameValue(actualType.definition, expectedType.definition)) issues.push({ kind: 'type', object: expectedType.name, reason: 'type definition differs' }) } if (allowSafeSubscriberOnlyObjects) return for (const actualType of actual.types ?? []) { if (!expectedTypes.has(actualType.name)) issues.push({ kind: 'type', object: actualType.name, reason: 'unexpected type' }) } } function compareSequences( expected: SchemaCatalog, actual: SchemaCatalog, issues: SchemaCompatibilityIssue[], allowSafeSubscriberOnlyObjects: boolean, ): void { const expectedSequences = new Map((expected.sequences ?? []).map(sequence => [sequence.name, sequence])) const actualSequences = new Map((actual.sequences ?? []).map(sequence => [sequence.name, sequence])) const fields = [ ['type', 'type'], ['start', 'start'], ['increment', 'increment'], ['min', 'min'], ['max', 'max'], ['cache', 'cache'], ['cycle', 'cycle'], ['ownedTable', 'owned table'], ['ownedColumn', 'owned column'], ] as const for (const expectedSequence of expected.sequences ?? []) { const actualSequence = actualSequences.get(expectedSequence.name) if (!actualSequence) { issues.push({ kind: 'sequence', object: expectedSequence.name, reason: 'missing required sequence' }) continue } for (const [field, label] of fields) { if (actualSequence[field] !== expectedSequence[field]) { issues.push({ kind: 'sequence', object: expectedSequence.name, reason: `expected ${label} ${displayValue(expectedSequence[field])}, found ${displayValue(actualSequence[field])}`, }) } } } if (allowSafeSubscriberOnlyObjects) return for (const actualSequence of actual.sequences ?? []) { if (!expectedSequences.has(actualSequence.name)) issues.push({ kind: 'sequence', object: actualSequence.name, reason: 'unexpected sequence' }) } } function compareFunctions( expected: SchemaCatalog, actual: SchemaCatalog, issues: SchemaCompatibilityIssue[], allowSafeSubscriberOnlyObjects: boolean, ): void { const expectedFunctions = new Map((expected.functions ?? []).map(fn => [functionKey(fn), fn])) const actualFunctions = new Map((actual.functions ?? []).map(fn => [functionKey(fn), fn])) for (const expectedFunction of expected.functions ?? []) { const key = functionKey(expectedFunction) const actualFunction = actualFunctions.get(key) if (!actualFunction) { issues.push({ kind: 'function', object: key, reason: 'missing required function overload' }) continue } if (actualFunction.definition !== expectedFunction.definition) issues.push({ kind: 'function', object: key, reason: 'function definition differs' }) } if (allowSafeSubscriberOnlyObjects) return for (const actualFunction of actual.functions ?? []) { const key = functionKey(actualFunction) if (!expectedFunctions.has(key)) issues.push({ kind: 'function', object: key, reason: 'unexpected function overload' }) } } function subscriberOnlyColumnCanAcceptPublisherRows(column: SchemaColumn): boolean { return !column.notNull || hasNonNullDefault(column.default) } function hasNonNullDefault(defaultValue: string | null): boolean { if (defaultValue === null) return false const normalized = defaultValue.trim().toUpperCase() return normalized !== 'NULL' && !/^NULL::[A-Z0-9_." ]+$/u.test(normalized) } function subscriberOnlyIndexCannotRejectPublisherRows(index: SchemaIndex): boolean { return index.constraintOwned !== true && !/^CREATE\s+UNIQUE\s+INDEX\b/iu.test(index.definition) } function assertSchemaCatalog(value: unknown, label: string): SchemaCatalog { if (!value || typeof value !== 'object') throw new Error(`Expected ${label} read-replica schema catalog JSON object`) const catalog = value as SchemaCatalog for (const collection of ['columns', 'constraints', 'functions', 'indexes', 'sequences', 'tables', 'types'] as const) { if (catalog[collection] !== undefined && !Array.isArray(catalog[collection])) throw new Error(`Read-replica schema catalog ${collection} must be an array`) } return catalog } function sameValue(left: unknown, right: unknown): boolean { return stableJson(left) === stableJson(right) } function columnKey(column: Pick): string { return `${column.table}.${column.name}` } function constraintKey(constraint: Pick): string { return `${constraint.table}.${constraint.name}` } function functionKey(fn: Pick): string { return `${fn.name}(${fn.arguments})` } function displayValue(value: unknown): string { if (value === undefined) return 'missing' if (value === null) return 'null' if (value === '') return 'none' if (typeof value === 'string') return value return stableJson(value) }