import type { Queryable } from './schema_catalog.ts'
import type { SchemaCompatibilityIssue } from './schema_compatibility.ts'
import {
  readReplicaSchemaCatalog,
  REPLICA_FUNCTIONS,
  REPLICA_TABLES,
} from './schema_catalog.ts'
import { readReplicaSchemaCompatibilityIssues } from './schema_compatibility.ts'

interface SchemaColumn {
  table: string
  name: string
  type: string
  notNull: boolean
  default: string | null
  identity: string
  generated: string
}

interface SchemaConstraint {
  table: string
  name: string
  type: 'c' | 'p' | 'u'
  definition: string
  valid?: boolean
}

interface SchemaIndex {
  table: string
  name: string
  definition: string
  valid: boolean
  constraintOwned?: boolean
}

interface SchemaCompositeAttribute {
  position: number
  name: string
  type: string
}

interface SchemaType {
  name: string
  kind: string
  definition: unknown
}

interface TypeSyncPlan {
  statements: string[]
  reason: string | null
}

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 SchemaTable {
  name: string
}

interface SchemaCatalog {
  columns?: SchemaColumn[]
  constraints?: SchemaConstraint[]
  functions?: SchemaFunction[]
  indexes?: SchemaIndex[]
  sequences?: SchemaSequence[]
  tables?: SchemaTable[]
  types?: SchemaType[]
}

type SyncStatementKind
  = | 'table'
    | 'column'
    | 'check_constraint'
    | 'constraint'
    | 'drop_check_constraint'
    | 'function'
    | 'index'
    | 'invalid_index'
    | 'drop_index'
    | 'sequence'
    | 'type'

type SyncObjectKind = 'table' | 'column' | 'constraint' | 'function' | 'index' | 'sequence' | 'type'

export interface ReadReplicaSchemaSyncStatement {
  kind: SyncStatementKind
  table: string
  name: string
  sql: string
}

export interface ReadReplicaSchemaSyncClient extends Queryable {
  assertCanApplyReadReplicaSchemaPlan?: (
    plan: ReadReplicaSchemaSyncPlan,
  ) => void
  applyReadReplicaSchemaPlan?: (
    plan: ReadReplicaSchemaSyncPlan,
  ) => Promise<void>
}

interface SkippedChange {
  kind: SyncObjectKind
  table: string
  name: string
  reason: string
}

export interface ReadReplicaSchemaSyncPlan {
  statements: ReadReplicaSchemaSyncStatement[]
  skipped: SkippedChange[]
}

export type AdditiveSchemaSyncPlan = ReadReplicaSchemaSyncPlan

export interface ReadReplicaSchemaSyncResult {
  applied: Array<Omit<ReadReplicaSchemaSyncStatement, 'sql'>>
  skipped: SkippedChange[]
}

export type AdditiveSchemaSyncResult = ReadReplicaSchemaSyncResult

export interface ReadReplicaSchemaSyncOptions {
  statementTimeoutMs?: number
  maxDurationMs?: number
  deadline?: number
}

export type AdditiveSchemaSyncOptions = ReadReplicaSchemaSyncOptions

export interface ReadReplicaSchemaReconciliationResult
  extends ReadReplicaSchemaSyncResult {
  issues: SchemaCompatibilityIssue[]
}
const SAFE_IDENTIFIER_RE = /^[A-Z_]\w*$/i
const SAFE_SQL_FRAGMENT_RE = /^[\w .()[\],:'"{}+\-=<>!]+$/
const REPLICA_TABLE_SET = new Set<string>(REPLICA_TABLES)
const REPLICA_FUNCTION_SET = new Set<string>(REPLICA_FUNCTIONS)
// Replica tables the release job may create on the subscriber before primary
// migrations add them to the publication. Each entry pins the exact SQL and
// the exact catalog shape the migrations must produce; any other missing
// table (or a different shape) stays a blocker, since an absent established
// table means lost replica data.
interface AllowedNewReplicaTable {
  sql: string
  catalog: Required<Pick<SchemaCatalog, 'columns' | 'constraints' | 'indexes' | 'sequences'>>
}
function replicaIdentityKeyCatalog(
  table: string,
  type: 'bigint' | 'integer',
): Pick<AllowedNewReplicaTable['catalog'], 'constraints' | 'indexes' | 'sequences'> {
  return {
    constraints: [
      { definition: 'PRIMARY KEY (id)', name: `${table}_pkey`, table, type: 'p', valid: true },
    ],
    indexes: [
      { constraintOwned: true, definition: `CREATE UNIQUE INDEX ${table}_pkey ON public.${table} USING btree (id)`, name: `${table}_pkey`, table, valid: true },
    ],
    sequences: [
      { cache: '1', cycle: false, increment: '1', max: type === 'bigint' ? '9223372036854775807' : '2147483647', min: '1', name: `${table}_id_seq`, ownedColumn: 'id', ownedTable: table, start: '1', type },
    ],
  }
}

export const ALLOWED_NEW_REPLICA_TABLES: Readonly<Record<string, AllowedNewReplicaTable>> = {
  delta_files: {
    sql: [
      'CREATE TABLE public."delta_files" (',
      '  "id" bigint GENERATED BY DEFAULT AS IDENTITY NOT NULL,',
      '  "app_id" character varying NOT NULL,',
      '  "s3_path" character varying NOT NULL,',
      '  "file_name" character varying NOT NULL,',
      '  "file_hash" character varying NOT NULL,',
      '  "file_size" bigint DEFAULT 0 NOT NULL,',
      '  "path_key" bytea NOT NULL,',
      '  "entry_key" bytea NOT NULL,',
      '  "created_at" timestamp with time zone DEFAULT now() NOT NULL,',
      '  "path_org" uuid,',
      '  "path_session" bytea,',
      '  "path_sha" bytea,',
      '  "file_hash_bin" bytea,',
      '  "path_session_id" integer,',
      '  CONSTRAINT "delta_files_pkey" PRIMARY KEY ("id")',
      ')',
    ].join('\n'),
    catalog: {
      columns: [
        { default: null, generated: '', identity: 'd', name: 'id', notNull: true, table: 'delta_files', type: 'bigint' },
        { default: null, generated: '', identity: '', name: 'app_id', notNull: true, table: 'delta_files', type: 'character varying' },
        { default: null, generated: '', identity: '', name: 's3_path', notNull: true, table: 'delta_files', type: 'character varying' },
        { default: null, generated: '', identity: '', name: 'file_name', notNull: true, table: 'delta_files', type: 'character varying' },
        { default: null, generated: '', identity: '', name: 'file_hash', notNull: true, table: 'delta_files', type: 'character varying' },
        { default: '0', generated: '', identity: '', name: 'file_size', notNull: true, table: 'delta_files', type: 'bigint' },
        { default: null, generated: '', identity: '', name: 'path_key', notNull: true, table: 'delta_files', type: 'bytea' },
        { default: null, generated: '', identity: '', name: 'entry_key', notNull: true, table: 'delta_files', type: 'bytea' },
        { default: 'now()', generated: '', identity: '', name: 'created_at', notNull: true, table: 'delta_files', type: 'timestamp with time zone' },
        { default: null, generated: '', identity: '', name: 'path_org', notNull: false, table: 'delta_files', type: 'uuid' },
        { default: null, generated: '', identity: '', name: 'path_session', notNull: false, table: 'delta_files', type: 'bytea' },
        { default: null, generated: '', identity: '', name: 'path_sha', notNull: false, table: 'delta_files', type: 'bytea' },
        { default: null, generated: '', identity: '', name: 'file_hash_bin', notNull: false, table: 'delta_files', type: 'bytea' },
        { default: null, generated: '', identity: '', name: 'path_session_id', notNull: false, table: 'delta_files', type: 'integer' },
      ],
      ...replicaIdentityKeyCatalog('delta_files', 'bigint'),
    },
  },
  delta_sessions: {
    sql: [
      'CREATE TABLE public."delta_sessions" (',
      '  "id" integer GENERATED BY DEFAULT AS IDENTITY NOT NULL,',
      '  "app_id" character varying NOT NULL,',
      '  "session" bytea NOT NULL,',
      '  CONSTRAINT "delta_sessions_pkey" PRIMARY KEY ("id")',
      ')',
    ].join('\n'),
    catalog: {
      columns: [
        { default: null, generated: '', identity: 'd', name: 'id', notNull: true, table: 'delta_sessions', type: 'integer' },
        { default: null, generated: '', identity: '', name: 'app_id', notNull: true, table: 'delta_sessions', type: 'character varying' },
        { default: null, generated: '', identity: '', name: 'session', notNull: true, table: 'delta_sessions', type: 'bytea' },
      ],
      ...replicaIdentityKeyCatalog('delta_sessions', 'integer'),
    },
  },
  version_files: {
    sql: [
      'CREATE TABLE public."version_files" (',
      '  "app_version_id" bigint NOT NULL,',
      '  "file_ids" bigint[] NOT NULL,',
      '  "created_at" timestamp with time zone DEFAULT now() NOT NULL,',
      '  CONSTRAINT "version_files_pkey" PRIMARY KEY ("app_version_id")',
      ')',
    ].join('\n'),
    catalog: {
      columns: [
        { default: null, generated: '', identity: '', name: 'app_version_id', notNull: true, table: 'version_files', type: 'bigint' },
        { default: null, generated: '', identity: '', name: 'file_ids', notNull: true, table: 'version_files', type: 'bigint[]' },
        { default: 'now()', generated: '', identity: '', name: 'created_at', notNull: true, table: 'version_files', type: 'timestamp with time zone' },
      ],
      constraints: [
        { definition: 'PRIMARY KEY (app_version_id)', name: 'version_files_pkey', table: 'version_files', type: 'p', valid: true },
      ],
      indexes: [
        { constraintOwned: true, definition: 'CREATE UNIQUE INDEX version_files_pkey ON public.version_files USING btree (app_version_id)', name: 'version_files_pkey', table: 'version_files', valid: true },
      ],
      sequences: [],
    },
  },
}

const DEFAULT_SCHEMA_SYNC_STATEMENT_TIMEOUT_MS = 550_000
const DEFAULT_SCHEMA_SYNC_MAX_DURATION_MS = 585_000
const SCHEMA_SYNC_RESPONSE_BUFFER_MS = 5_000

export function planReadReplicaSchemaSync(
  expected: unknown,
  actual: unknown,
): ReadReplicaSchemaSyncPlan {
  const expectedCatalog = assertSchemaCatalog(expected)
  const actualCatalog = assertSchemaCatalog(actual)
  const actualTables = new Set(
    (actualCatalog.tables ?? []).map(table => table.name),
  )
  const actualColumnsByKey = new Map(
    (actualCatalog.columns ?? []).map(column => [columnKey(column), column]),
  )
  const actualIndexByName = new Map(
    (actualCatalog.indexes ?? []).map(index => [index.name, index]),
  )
  const expectedIndexByName = new Map(
    (expectedCatalog.indexes ?? []).map(index => [index.name, index]),
  )
  const actualConstraintsByKey = new Map(
    (actualCatalog.constraints ?? []).map(constraint => [
      constraintKey(constraint),
      constraint,
    ]),
  )
  const expectedConstraintsByKey = new Map(
    (expectedCatalog.constraints ?? []).map(constraint => [
      constraintKey(constraint),
      constraint,
    ]),
  )
  const actualFunctionsByKey = new Map(
    (actualCatalog.functions ?? []).map(fn => [functionKey(fn), fn]),
  )
  const actualTypesByName = new Map(
    (actualCatalog.types ?? []).map(type => [type.name, type]),
  )
  const skippedColumnsByTable = new Map<string, Set<string>>()
  const statements: ReadReplicaSchemaSyncStatement[] = []
  const skipped: SkippedChange[] = []
  const createdTables = new Set<string>()

  for (const table of expectedCatalog.tables ?? []) {
    const allowed = ALLOWED_NEW_REPLICA_TABLES[table.name]
    if (actualTables.has(table.name) || !allowed)
      continue
    if (!matchesAllowedNewReplicaTable(expectedCatalog, table.name, allowed)) {
      skipped.push({
        kind: 'table',
        table: table.name,
        name: table.name,
        reason: 'unexpected_new_table_definition',
      })
      continue
    }
    statements.push({ kind: 'table', table: table.name, name: table.name, sql: allowed.sql })
    createdTables.add(table.name)
  }

  for (const type of expectedCatalog.types ?? []) {
    if (typeMatches(type, actualTypesByName.get(type.name)))
      continue

    const typeSync = planTypeSync(type, actualTypesByName.get(type.name))
    if (typeSync.reason) {
      skipped.push({
        kind: 'type',
        table: 'public',
        name: type.name,
        reason: typeSync.reason,
      })
      continue
    }

    for (const sql of typeSync.statements) {
      statements.push({
        kind: 'type',
        table: 'public',
        name: type.name,
        sql,
      })
    }
  }

  for (const fn of expectedCatalog.functions ?? []) {
    if (functionMatches(fn, actualFunctionsByKey.get(functionKey(fn))))
      continue

    const sql = buildCreateOrReplaceFunctionStatement(fn)
    if (!sql) {
      skipped.push({
        kind: 'function',
        table: 'public',
        name: functionKey(fn),
        reason: 'unsupported_function_reconciliation',
      })
      continue
    }

    statements.push({
      kind: 'function',
      table: 'public',
      name: functionKey(fn),
      sql,
    })
  }

  for (const sequence of expectedCatalog.sequences ?? []) {
    // Identity sequences of a created table come with its CREATE TABLE.
    if (sequence.ownedTable && createdTables.has(sequence.ownedTable))
      continue
    const actualSequence = (actualCatalog.sequences ?? []).find(
      candidate => candidate.name === sequence.name,
    )
    if (sequenceMatches(sequence, actualCatalog.sequences ?? []))
      continue

    const sql = actualSequence
      ? buildAlterSequenceStatement(sequence, actualTables)
      : buildCreateSequenceStatement(sequence, actualTables)
    if (!sql) {
      skipped.push({
        kind: 'sequence',
        table: 'public',
        name: sequence.name,
        reason: 'unsupported_sequence_reconciliation',
      })
      continue
    }

    statements.push({
      kind: 'sequence',
      table: 'public',
      name: sequence.name,
      sql,
    })
  }

  for (const column of expectedCatalog.columns ?? []) {
    if (createdTables.has(column.table))
      continue
    const actualColumn = actualColumnsByKey.get(columnKey(column))
    if (!actualColumn) {
      const addColumnSql = buildAddColumnStatement(column, actualTables)
      if (!addColumnSql) {
        trackSkippedColumn(skippedColumnsByTable, column)
        skipped.push({
          kind: 'column',
          table: column.table,
          name: column.name,
          reason: 'unsupported_additive_column',
        })
        continue
      }

      statements.push({
        kind: 'column',
        table: column.table,
        name: column.name,
        sql: addColumnSql,
      })
      continue
    }

    if (actualColumn.notNull !== column.notNull) {
      // Tightening or loosening an existing subscriber column before the
      // publisher changes can reject replicated rows. Leave it for a proven
      // publisher-compatible transition rather than risk stopping replication.
      skipped.push({
        kind: 'column',
        table: column.table,
        name: column.name,
        reason: 'pre_primary_column_nullability',
      })
    }
  }

  for (const index of expectedCatalog.indexes ?? []) {
    if (createdTables.has(index.table))
      continue
    const actualIndex = actualIndexByName.get(index.name)
    if (indexMatches(index, actualIndex))
      continue

    if (hasUnresolvedColumnDependency(index, skippedColumnsByTable)) {
      skipped.push({
        kind: 'index',
        table: index.table,
        name: index.name,
        reason: 'unresolved_column_dependency',
      })
      continue
    }

    if (
      actualIndex?.constraintOwned !== undefined
      && actualIndex.constraintOwned
    ) {
      skipped.push({
        kind: 'index',
        table: index.table,
        name: index.name,
        reason: 'constraint_owned_index',
      })
      continue
    }

    if (actualIndex && actualIndex.table !== index.table) {
      skipped.push({
        kind: 'index',
        table: index.table,
        name: index.name,
        reason: 'index_name_conflict',
      })
      continue
    }

    if (actualIndex && actualIndex.definition !== index.definition) {
      const dropIndexSql = buildDropIndexStatement(actualIndex, actualTables)
      if (!dropIndexSql) {
        skipped.push({
          kind: 'index',
          table: index.table,
          name: index.name,
          reason:
            actualIndex.constraintOwned === undefined
              ? 'unknown_index_ownership'
              : 'unsupported_index_replacement',
        })
        continue
      }
      statements.push({
        kind: 'drop_index',
        table: actualIndex.table,
        name: actualIndex.name,
        sql: dropIndexSql,
      })
    }
    else if (actualIndex) {
      const reindexSql = buildReindexIndexStatement(index, actualTables)
      if (!reindexSql) {
        skipped.push({
          kind: 'index',
          table: index.table,
          name: index.name,
          reason:
            actualIndex.constraintOwned === undefined
              ? 'unknown_index_ownership'
              : 'unsupported_additive_index',
        })
        continue
      }

      statements.push({
        kind: 'invalid_index',
        table: index.table,
        name: index.name,
        sql: reindexSql,
      })
      continue
    }

    const addIndexSql = buildCreateIndexStatement(index, actualTables)
    if (!addIndexSql) {
      skipped.push({
        kind: 'index',
        table: index.table,
        name: index.name,
        reason: 'unsupported_additive_index',
      })
      continue
    }

    statements.push({
      kind: 'index',
      table: index.table,
      name: index.name,
      sql: addIndexSql,
    })
  }
  for (const constraint of expectedCatalog.constraints ?? []) {
    if (createdTables.has(constraint.table))
      continue
    if (constraint.type === 'c') {
      const actualConstraint = actualConstraintsByKey.get(
        constraintKey(constraint),
      )
      // Missing publisher CHECK constraints stay optional on read-only subscribers.
      if (!actualConstraint || constraintMatches(constraint, actualConstraint))
        continue

      const dropSql = buildDropCheckConstraintStatement(
        constraint,
        actualTables,
      )
      const addSql = buildAddCheckConstraintStatement(
        constraint,
        actualTables,
      )
      if (!dropSql || !addSql) {
        skipped.push({
          kind: 'constraint',
          table: constraint.table,
          name: constraint.name,
          reason: 'check_constraint_conflict',
        })
        continue
      }

      statements.push({
        kind: 'drop_check_constraint',
        table: constraint.table,
        name: constraint.name,
        sql: dropSql,
      })
      statements.push({
        kind: 'check_constraint',
        table: constraint.table,
        name: constraint.name,
        sql: addSql,
      })
      continue
    }

    const actualConstraint = actualConstraintsByKey.get(
      constraintKey(constraint),
    )
    if (constraintMatches(constraint, actualConstraint))
      continue

    if (actualConstraint) {
      skipped.push({
        kind: 'constraint',
        table: constraint.table,
        name: constraint.name,
        reason: 'constraint_conflict',
      })
      continue
    }

    const expectedIndex = expectedIndexByName.get(constraint.name)
    const actualIndex = expectedIndex
      ? actualIndexByName.get(expectedIndex.name)
      : undefined
    const sql = buildAttachConstraintStatement(
      constraint,
      expectedIndex,
      actualIndex,
      actualTables,
    )
    if (!sql) {
      skipped.push({
        kind: 'constraint',
        table: constraint.table,
        name: constraint.name,
        reason: 'unsupported_key_constraint',
      })
      continue
    }

    statements.push({
      kind: 'constraint',
      table: constraint.table,
      name: constraint.name,
      sql,
    })
  }

  for (const constraint of actualCatalog.constraints ?? []) {
    if (expectedConstraintsByKey.has(constraintKey(constraint)))
      continue

    skipped.push({
      kind: 'constraint',
      table: constraint.table,
      name: constraint.name,
      reason: 'unexpected_constraint',
    })
  }

  for (const index of actualCatalog.indexes ?? []) {
    if (expectedIndexByName.has(index.name))
      continue

    const dropIndexSql = buildDropIndexStatement(index, actualTables)
    if (!dropIndexSql) {
      skipped.push({
        kind: 'index',
        table: index.table,
        name: index.name,
        reason:
          index.constraintOwned === undefined
            ? 'unknown_index_ownership'
            : 'constraint_owned_index',
      })
      continue
    }

    statements.push({
      kind: 'drop_index',
      table: index.table,
      name: index.name,
      sql: dropIndexSql,
    })
  }

  return { statements, skipped }
}

function matchesAllowedNewReplicaTable(
  catalog: SchemaCatalog,
  table: string,
  allowed: AllowedNewReplicaTable,
): boolean {
  // Compare the fields the allowlist pins, in a stable key order.
  const same = (actual: readonly object[], expected: readonly object[]) => {
    const byText = (left: string, right: string) => left.localeCompare(right)
    const keys = expected.length ? Object.keys(expected[0]).sort(byText) : []
    const shape = (entries: readonly object[]) => entries.map(entry => JSON.stringify(entry, keys)).sort(byText)
    return JSON.stringify(shape(actual)) === JSON.stringify(shape(expected))
  }
  return same((catalog.columns ?? []).filter(entry => entry.table === table), allowed.catalog.columns)
    && same((catalog.constraints ?? []).filter(entry => entry.table === table), allowed.catalog.constraints)
    && same((catalog.indexes ?? []).filter(entry => entry.table === table), allowed.catalog.indexes)
    && same((catalog.sequences ?? []).filter(entry => entry.ownedTable === table), allowed.catalog.sequences)
}

export const planReadReplicaAdditiveSchemaSync = planReadReplicaSchemaSync

export async function applyReadReplicaSchemaSync(
  client: ReadReplicaSchemaSyncClient,
  expected: unknown,
  options: ReadReplicaSchemaSyncOptions = {},
): Promise<ReadReplicaSchemaSyncResult> {
  const deadline = options.deadline ?? (
    Date.now()
    + positiveIntegerOrDefault(
      options.maxDurationMs,
      DEFAULT_SCHEMA_SYNC_MAX_DURATION_MS,
    )
  )
  assertTimeRemaining(deadline, 'read the subscriber schema catalog')
  const actual = await readReplicaSchemaCatalog(client)
  assertTimeRemaining(deadline, 'plan subscriber schema reconciliation')
  const plan = planReadReplicaSchemaSync(expected, actual)
  const statementTimeoutMs = positiveIntegerOrDefault(
    options.statementTimeoutMs,
    DEFAULT_SCHEMA_SYNC_STATEMENT_TIMEOUT_MS,
  )
  const applyPlan = client.applyReadReplicaSchemaPlan
  const assertPlan = client.assertCanApplyReadReplicaSchemaPlan
  if (applyPlan && !assertPlan) {
    throw new Error(
      'Transactional read-replica schema execution must preflight the whole reconciliation plan before applying schema changes',
    )
  }
  assertPlan?.(plan)

  try {
    if (applyPlan) {
      if (plan.statements.length) {
        assertTimeRemaining(deadline, 'apply the transactional subscriber schema plan')
        await applyPlan(plan)
        assertTimeRemaining(deadline, 'finish the transactional subscriber schema plan')
      }
    }
    else {
      for (const statement of plan.statements) {
        await setStatementTimeoutForRemainingBudget(
          client,
          statementTimeoutMs,
          deadline,
          statement,
        )
        await client.query(statement.sql)
        assertTimeRemaining(
          deadline,
          `finish ${statement.kind} ${statement.table}.${statement.name}`,
        )
      }
    }
  }
  finally {
    await client.query('RESET statement_timeout')
  }

  assertTimeRemaining(deadline, 'read the post-DDL subscriber schema catalog')
  const finalActual = await readReplicaSchemaCatalog(client)
  assertTimeRemaining(deadline, 'validate applied subscriber schema changes')
  assertAppliedStatementsVisible(plan.statements, expected, finalActual)

  return {
    applied: plan.statements.map(({ kind, table, name }) => ({
      kind,
      table,
      name,
    })),
    skipped: plan.skipped,
  }
}
export const applyReadReplicaAdditiveSchemaSync = applyReadReplicaSchemaSync

export async function reconcileReadReplicaSchema(
  master: Queryable,
  replica: ReadReplicaSchemaSyncClient,
  options: ReadReplicaSchemaSyncOptions = {},
): Promise<ReadReplicaSchemaReconciliationResult> {
  const deadline = options.deadline ?? (
    Date.now()
    + positiveIntegerOrDefault(
      options.maxDurationMs,
      DEFAULT_SCHEMA_SYNC_MAX_DURATION_MS,
    )
  )
  assertTimeRemaining(deadline, 'read the primary schema catalog')
  const expected = await readReplicaSchemaCatalog(master)
  assertTimeRemaining(deadline, 'reconcile the subscriber schema')
  const result = await applyReadReplicaSchemaSync(replica, expected, {
    ...options,
    deadline,
  })
  assertTimeRemaining(deadline, 'read the final primary schema catalog')
  const finalExpected = await readReplicaSchemaCatalog(master)
  assertTimeRemaining(deadline, 'read the final subscriber schema catalog')
  const finalActual = await readReplicaSchemaCatalog(replica)
  assertTimeRemaining(deadline, 'compare the final selected schemas')

  return {
    ...result,
    issues: readReplicaSchemaCompatibilityIssues(finalExpected, finalActual),
  }
}

function assertAppliedStatementsVisible(
  applied: readonly ReadReplicaSchemaSyncStatement[],
  expected: unknown,
  actual: unknown,
): void {
  const remaining = new Set(
    planReadReplicaSchemaSync(expected, actual).statements.map(statementKey),
  )
  const unavailable = applied.filter(statement =>
    remaining.has(statementKey(statement)),
  )
  if (!unavailable.length)
    return

  throw new Error(
    `Read-replica schema sync did not persist ${unavailable.map(statement => `${statement.kind} ${statement.table}.${statement.name}`).join(', ')}`,
  )
}

function statementKey(
  statement: Pick<ReadReplicaSchemaSyncStatement, 'kind' | 'table' | 'name'>,
): string {
  const kind = statement.kind === 'invalid_index' ? 'index' : statement.kind
  return `${kind}:${statement.table}.${statement.name}`
}

function indexMatches(
  expected: SchemaIndex,
  actual: SchemaIndex | undefined,
): boolean {
  return (
    actual?.table === expected.table
    && actual.valid
    && actual.definition === expected.definition
  )
}

async function setStatementTimeoutForRemainingBudget(
  client: Queryable,
  maxStatementTimeoutMs: number,
  deadline: number,
  statement: ReadReplicaSchemaSyncStatement,
): Promise<void> {
  const remainingMs = deadline - Date.now() - SCHEMA_SYNC_RESPONSE_BUFFER_MS
  if (remainingMs <= 0) {
    throw new Error(
      `Read-replica schema sync exceeded max duration before ${statement.kind} ${statement.table}.${statement.name}`,
    )
  }

  await client.query(
    `SET statement_timeout = ${Math.min(maxStatementTimeoutMs, remainingMs)}`,
  )
}

function assertTimeRemaining(deadline: number, action: string): void {
  if (Date.now() >= deadline) {
    throw new Error(
      `Read-replica schema sync exceeded max duration before it could ${action}`,
    )
  }
}

function positiveIntegerOrDefault(
  value: number | undefined,
  fallback: number,
): number {
  if (typeof value !== 'number' || !Number.isFinite(value) || value <= 0)
    return fallback

  return Math.trunc(value)
}

function assertSchemaCatalog(value: unknown): SchemaCatalog {
  if (!value || typeof value !== 'object')
    throw new Error('Expected 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] && !Array.isArray(catalog[collection]))
      throw new Error(`Read-replica schema catalog ${collection} must be an array`)
  }

  for (const column of catalog.columns ?? []) assertSchemaColumn(column)
  for (const constraint of catalog.constraints ?? [])
    assertSchemaConstraint(constraint)
  for (const fn of catalog.functions ?? []) assertSchemaFunction(fn)
  for (const index of catalog.indexes ?? []) assertSchemaIndex(index)
  for (const sequence of catalog.sequences ?? []) assertSchemaSequence(sequence)
  for (const table of catalog.tables ?? []) assertSchemaTable(table)
  for (const type of catalog.types ?? []) assertSchemaType(type)
  return catalog
}

function assertSchemaColumn(value: unknown): asserts value is SchemaColumn {
  const column = assertCatalogObject(value, 'columns') as Partial<SchemaColumn>
  if (
    typeof column.table !== 'string'
    || typeof column.name !== 'string'
    || typeof column.type !== 'string'
  ) {
    throw new TypeError(
      'Read-replica schema catalog columns must include string table, name, and type',
    )
  }
  if (typeof column.notNull !== 'boolean') {
    throw new TypeError(
      'Read-replica schema catalog columns must include boolean notNull',
    )
  }
  if (column.default !== null && typeof column.default !== 'string') {
    throw new Error(
      'Read-replica schema catalog column defaults must be strings or null',
    )
  }
  if (
    typeof column.identity !== 'string'
    || typeof column.generated !== 'string'
  ) {
    throw new TypeError(
      'Read-replica schema catalog columns must include string identity and generated fields',
    )
  }
}

function assertSchemaConstraint(
  value: unknown,
): asserts value is SchemaConstraint {
  const constraint = assertCatalogObject(
    value,
    'constraints',
  ) as Partial<SchemaConstraint>
  if (
    typeof constraint.table !== 'string'
    || typeof constraint.name !== 'string'
    || typeof constraint.definition !== 'string'
    || !isConstraintType(constraint.type)
  ) {
    throw new TypeError(
      'Read-replica schema catalog constraints must include table, name, type, and definition',
    )
  }
  if (constraint.valid !== undefined && typeof constraint.valid !== 'boolean') {
    throw new TypeError(
      'Read-replica schema catalog constraint valid must be boolean when present',
    )
  }
}

function assertSchemaFunction(value: unknown): asserts value is SchemaFunction {
  const fn = assertCatalogObject(value, 'functions') as Partial<SchemaFunction>
  if (
    typeof fn.name !== 'string'
    || typeof fn.arguments !== 'string'
    || typeof fn.definition !== 'string'
  ) {
    throw new TypeError(
      'Read-replica schema catalog functions must include name, arguments, and definition',
    )
  }
}

function assertSchemaIndex(value: unknown): asserts value is SchemaIndex {
  const index = assertCatalogObject(value, 'indexes') as Partial<SchemaIndex>
  if (
    typeof index.table !== 'string'
    || typeof index.name !== 'string'
    || typeof index.definition !== 'string'
  ) {
    throw new TypeError(
      'Read-replica schema catalog indexes must include string table, name, and definition',
    )
  }
  if (typeof index.valid !== 'boolean') {
    throw new TypeError(
      'Read-replica schema catalog indexes must include boolean valid',
    )
  }
  if (
    index.constraintOwned !== undefined
    && typeof index.constraintOwned !== 'boolean'
  ) {
    throw new Error(
      'Read-replica schema catalog indexes constraintOwned must be boolean when present',
    )
  }
}

function assertSchemaSequence(value: unknown): asserts value is SchemaSequence {
  const sequence = assertCatalogObject(
    value,
    'sequences',
  ) as Partial<SchemaSequence>
  const textFields = [
    sequence.name,
    sequence.type,
    sequence.start,
    sequence.increment,
    sequence.min,
    sequence.max,
    sequence.cache,
  ]
  if (textFields.some(field => typeof field !== 'string')) {
    throw new TypeError(
      'Read-replica schema catalog sequences must include string structural fields',
    )
  }
  if (typeof sequence.cycle !== 'boolean') {
    throw new TypeError(
      'Read-replica schema catalog sequences must include boolean cycle',
    )
  }
  if (
    (sequence.ownedTable !== null && typeof sequence.ownedTable !== 'string')
    || (sequence.ownedColumn !== null && typeof sequence.ownedColumn !== 'string')
  ) {
    throw new TypeError(
      'Read-replica schema catalog sequences must include nullable ownership fields',
    )
  }
}

function assertSchemaTable(value: unknown): asserts value is SchemaTable {
  const table = assertCatalogObject(value, 'tables') as Partial<SchemaTable>
  if (typeof table.name !== 'string') {
    throw new TypeError(
      'Read-replica schema catalog tables must include string names',
    )
  }
}

function assertSchemaType(value: unknown): asserts value is SchemaType {
  const type = assertCatalogObject(value, 'types') as Partial<SchemaType>
  if (typeof type.name !== 'string' || typeof type.kind !== 'string') {
    throw new TypeError(
      'Read-replica schema catalog types must include string name and kind',
    )
  }
  if (type.kind === 'e') {
    if (
      !Array.isArray(type.definition)
      || !type.definition.every(label => typeof label === 'string')
      || new Set(type.definition).size !== type.definition.length
    ) {
      throw new TypeError(
        'Read-replica enum types must include unique string labels',
      )
    }
    return
  }
  if (type.kind !== 'c')
    return

  if (!Array.isArray(type.definition)) {
    throw new TypeError(
      'Read-replica composite types must include attribute definitions',
    )
  }

  const names = new Set<string>()
  const positions = new Set<number>()
  for (const value of type.definition) {
    assertSchemaCompositeAttribute(value)
    if (names.has(value.name) || positions.has(value.position)) {
      throw new TypeError(
        'Read-replica composite types must include unique attributes',
      )
    }
    names.add(value.name)
    positions.add(value.position)
  }
}

function assertSchemaCompositeAttribute(
  value: unknown,
): asserts value is SchemaCompositeAttribute {
  const attribute = assertCatalogObject(
    value,
    'types',
  ) as Partial<SchemaCompositeAttribute>
  if (
    typeof attribute.position !== 'number'
    || !Number.isSafeInteger(attribute.position)
    || attribute.position <= 0
    || typeof attribute.name !== 'string'
    || typeof attribute.type !== 'string'
  ) {
    throw new TypeError(
      'Read-replica composite type attributes must include position, name, and type',
    )
  }
}

function assertCatalogObject(value: unknown, collection: string): object {
  if (!value || typeof value !== 'object') {
    throw new Error(
      `Read-replica schema catalog ${collection} must contain objects`,
    )
  }

  return value
}

function typeMatches(
  expected: SchemaType,
  actual: SchemaType | undefined,
): boolean {
  if (actual?.kind !== expected.kind)
    return false

  return JSON.stringify(actual.definition) === JSON.stringify(expected.definition)
}

function planTypeSync(
  expected: SchemaType,
  actual: SchemaType | undefined,
): TypeSyncPlan {
  if (!isSafeIdentifier(expected.name))
    return { statements: [], reason: 'unsafe_type_name' }
  if (!actual) {
    const statement = buildCreateTypeStatement(expected)
    return statement
      ? { statements: [statement], reason: null }
      : { statements: [], reason: 'unsupported_type_creation' }
  }
  if (actual.kind !== expected.kind)
    return { statements: [], reason: 'type_kind_conflict' }
  if (expected.kind === 'e')
    return planEnumTypeSync(expected, actual)
  if (expected.kind === 'c')
    return planCompositeTypeSync(expected, actual)

  return { statements: [], reason: 'unsupported_type_reconciliation' }
}

function buildCreateTypeStatement(type: SchemaType): string | null {
  if (type.kind === 'e') {
    const labels = enumLabels(type)
    const values = labels && quoteEnumValues(labels)
    if (!values?.length)
      return null

    return `CREATE TYPE public.${quoteIdent(type.name)} AS ENUM (${values.join(', ')})`
  }
  if (type.kind !== 'c')
    return null

  const attributes = compositeTypeAttributes(type)
  if (!attributes?.length || !hasSequentialCompositePositions(attributes))
    return null

  const definitions = compositeAttributeDefinitions(attributes)
  if (!definitions)
    return null

  return `CREATE TYPE public.${quoteIdent(type.name)} AS (${definitions.join(', ')})`
}

function planEnumTypeSync(
  expected: SchemaType,
  actual: SchemaType,
): TypeSyncPlan {
  const expectedLabels = enumLabels(expected)
  const actualLabels = enumLabels(actual)
  if (!expectedLabels || !actualLabels) {
    return { statements: [], reason: 'unsupported_enum_reconciliation' }
  }
  if (!isEnumPrefix(actualLabels, expectedLabels)) {
    return { statements: [], reason: 'unsafe_enum_reconciliation' }
  }

  const values = quoteEnumValues(expectedLabels.slice(actualLabels.length))
  if (!values)
    return { statements: [], reason: 'unsupported_enum_reconciliation' }

  return {
    statements: values.map(
      value => `ALTER TYPE public.${quoteIdent(expected.name)} ADD VALUE IF NOT EXISTS ${value}`,
    ),
    reason: null,
  }
}

function planCompositeTypeSync(
  expected: SchemaType,
  actual: SchemaType,
): TypeSyncPlan {
  const expectedAttributes = compositeTypeAttributes(expected)
  const actualAttributes = compositeTypeAttributes(actual)
  if (!expectedAttributes || !actualAttributes) {
    return { statements: [], reason: 'unsupported_composite_reconciliation' }
  }
  if (
    !hasSequentialCompositePositions(expectedAttributes)
    || !hasSequentialCompositePositions(actualAttributes)
    || !isCompositePrefix(actualAttributes, expectedAttributes)
  ) {
    return { statements: [], reason: 'unsafe_composite_reconciliation' }
  }

  const definitions = compositeAttributeDefinitions(
    expectedAttributes.slice(actualAttributes.length),
  )
  if (!definitions) {
    return { statements: [], reason: 'unsupported_composite_reconciliation' }
  }

  return {
    statements: definitions.map(
      definition => `ALTER TYPE public.${quoteIdent(expected.name)} ADD ATTRIBUTE ${definition}`,
    ),
    reason: null,
  }
}

function enumLabels(type: SchemaType): string[] | null {
  if (
    type.kind !== 'e'
    || !Array.isArray(type.definition)
    || !type.definition.every(label => typeof label === 'string')
  ) {
    return null
  }

  return type.definition as string[]
}

function compositeTypeAttributes(
  type: SchemaType,
): SchemaCompositeAttribute[] | null {
  if (
    type.kind !== 'c'
    || !Array.isArray(type.definition)
    || !type.definition.every(isSchemaCompositeAttribute)
  ) {
    return null
  }

  return type.definition as SchemaCompositeAttribute[]
}

function isSchemaCompositeAttribute(
  value: unknown,
): value is SchemaCompositeAttribute {
  if (!value || typeof value !== 'object')
    return false

  const attribute = value as Partial<SchemaCompositeAttribute>
  return (
    typeof attribute.position === 'number'
    && Number.isSafeInteger(attribute.position)
    && attribute.position > 0
    && typeof attribute.name === 'string'
    && typeof attribute.type === 'string'
  )
}

function isEnumPrefix(actual: readonly string[], expected: readonly string[]): boolean {
  return actual.length <= expected.length
    && actual.every((label, index) => label === expected[index])
}

function isCompositePrefix(
  actual: readonly SchemaCompositeAttribute[],
  expected: readonly SchemaCompositeAttribute[],
): boolean {
  return actual.length <= expected.length
    && actual.every((attribute, index) => {
      const expectedAttribute = expected[index]
      return attribute.position === expectedAttribute.position
        && attribute.name === expectedAttribute.name
        && attribute.type === expectedAttribute.type
    })
}

function hasSequentialCompositePositions(
  attributes: readonly SchemaCompositeAttribute[],
): boolean {
  return attributes.every(
    (attribute, index) => attribute.position === index + 1,
  )
}

function compositeAttributeDefinitions(
  attributes: readonly SchemaCompositeAttribute[],
): string[] | null {
  const definitions: string[] = []
  for (const attribute of attributes) {
    if (!isSafeIdentifier(attribute.name) || !isSafeSqlFragment(attribute.type))
      return null

    definitions.push(`${quoteIdent(attribute.name)} ${attribute.type}`)
  }

  return definitions
}

function quoteEnumValues(values: readonly string[]): string[] | null {
  const quotedValues: string[] = []
  for (const value of values) {
    if (value.includes('\0'))
      return null

    quotedValues.push(
      `E'${value.replaceAll('\\', '\\\\').replaceAll('\'', '\'\'')}'`,
    )
  }

  return quotedValues
}

function buildAddColumnStatement(
  column: SchemaColumn,
  actualTables: Set<string>,
): string | null {
  if (!isSafeReplicaTable(column.table, actualTables))
    return null
  if (!isSafeIdentifier(column.name))
    return null
  if (column.identity || column.generated)
    return null
  if (!isSafeSqlFragment(column.type))
    return null
  if (
    column.default !== null
    && (!isSafeSqlFragment(column.default)
      || !isSafeConstantDefault(column.default))
  ) {
    return null
  }
  if (
    column.notNull
    && (column.default === null || isNullDefault(column.default))
  ) {
    return null
  }

  const defaultSql
    = column.default === null ? '' : ` DEFAULT ${column.default}`
  const notNullSql = column.notNull ? ' NOT NULL' : ''
  return `ALTER TABLE ${quoteQualifiedTable(column.table)} ADD COLUMN IF NOT EXISTS ${quoteIdent(column.name)} ${column.type}${defaultSql}${notNullSql}`
}

function buildCreateIndexStatement(
  index: SchemaIndex,
  actualTables: Set<string>,
): string | null {
  if (!isSafeReplicaTable(index.table, actualTables))
    return null
  if (!isSafeQuotedIdentifier(index.name))
    return null
  if (!isSafeSchemaDefinition(index.definition))
    return null

  const parsedDefinition = parseCreateIndexDefinition(index.definition)
  if (!parsedDefinition)
    return null

  const { unique, indexName, tableName, indexTail } = parsedDefinition
  if (indexName !== index.name || tableName !== index.table)
    return null

  return `CREATE ${unique}INDEX CONCURRENTLY IF NOT EXISTS ${quoteIdent(index.name)} ON ${quoteQualifiedTable(index.table)} ${indexTail}`
}

function buildDropCheckConstraintStatement(
  constraint: Pick<SchemaConstraint, 'table' | 'name'>,
  actualTables: Set<string>,
): string | null {
  if (!isSafeReplicaTable(constraint.table, actualTables))
    return null
  if (!isSafeQuotedIdentifier(constraint.name))
    return null

  return `ALTER TABLE ${quoteQualifiedTable(constraint.table)} DROP CONSTRAINT IF EXISTS ${quoteIdent(constraint.name)}`
}

function buildAddCheckConstraintStatement(
  constraint: SchemaConstraint,
  actualTables: Set<string>,
): string | null {
  if (
    constraint.type !== 'c'
    || !isSafeReplicaTable(constraint.table, actualTables)
    || !isSafeQuotedIdentifier(constraint.name)
    || !isSafeCheckConstraintDefinition(constraint.definition)
  ) {
    return null
  }

  return `ALTER TABLE ${quoteQualifiedTable(constraint.table)} ADD CONSTRAINT ${quoteIdent(constraint.name)} ${constraint.definition}`
}

function buildAttachConstraintStatement(
  constraint: SchemaConstraint,
  expectedIndex: SchemaIndex | undefined,
  actualIndex: SchemaIndex | undefined,
  actualTables: Set<string>,
): string | null {
  if (
    (constraint.type !== 'p' && constraint.type !== 'u')
    || !isSafeReplicaTable(constraint.table, actualTables)
    || !isSafeQuotedIdentifier(constraint.name)
    || expectedIndex?.name !== constraint.name
    || expectedIndex.table !== constraint.table
    || expectedIndex.constraintOwned !== true
  ) {
    return null
  }
  if (
    actualIndex
    && (
      actualIndex.table !== constraint.table
      || actualIndex.constraintOwned !== false
    )
  ) {
    return null
  }
  if (
    !actualIndex
    || !indexMatches(expectedIndex, actualIndex)
  ) {
    if (!buildCreateIndexStatement(expectedIndex, actualTables))
      return null
  }

  const kind = constraint.type === 'p' ? 'PRIMARY KEY' : 'UNIQUE'
  return `ALTER TABLE ${quoteQualifiedTable(constraint.table)} ADD CONSTRAINT ${quoteIdent(constraint.name)} ${kind} USING INDEX ${quoteIdent(expectedIndex.name)}`
}

function buildCreateOrReplaceFunctionStatement(
  fn: SchemaFunction,
): string | null {
  const expectedPrefix = `CREATE OR REPLACE FUNCTION public.${fn.name}(`
  if (
    !REPLICA_FUNCTION_SET.has(fn.name)
    || !isSafeIdentifier(fn.name)
    || fn.arguments.includes('\0')
    || !fn.definition.startsWith(expectedPrefix)
    || fn.definition.includes('\0')
  ) {
    return null
  }

  return fn.definition
}

function buildCreateSequenceStatement(
  sequence: SchemaSequence,
  actualTables: Set<string>,
): string | null {
  if (!isSafeQuotedIdentifier(sequence.name))
    return null

  const options = buildSequenceOptions(sequence, actualTables)
  if (!options)
    return null

  return `CREATE SEQUENCE IF NOT EXISTS public.${quoteIdent(sequence.name)} ${options}`
}

function buildAlterSequenceStatement(
  sequence: SchemaSequence,
  actualTables: Set<string>,
): string | null {
  if (!isSafeQuotedIdentifier(sequence.name))
    return null

  const options = buildSequenceOptions(sequence, actualTables)
  if (!options)
    return null

  return `ALTER SEQUENCE public.${quoteIdent(sequence.name)} ${options}`
}

function buildSequenceOptions(
  sequence: SchemaSequence,
  actualTables: Set<string>,
): string | null {
  if (!isSafeSequenceDataType(sequence.type))
    return null
  for (const value of [
    sequence.start,
    sequence.increment,
    sequence.min,
    sequence.max,
    sequence.cache,
  ]) {
    if (!isSafeSequenceNumber(value))
      return null
  }

  const ownedBy = buildSequenceOwnership(sequence, actualTables)
  if (!ownedBy)
    return null

  const cycle = sequence.cycle ? 'CYCLE' : 'NO CYCLE'
  return `AS ${sequence.type} START WITH ${sequence.start} INCREMENT BY ${sequence.increment} MINVALUE ${sequence.min} MAXVALUE ${sequence.max} CACHE ${sequence.cache} ${cycle} ${ownedBy}`
}

function buildSequenceOwnership(
  sequence: SchemaSequence,
  actualTables: Set<string>,
): string | null {
  if (sequence.ownedTable === null && sequence.ownedColumn === null)
    return 'OWNED BY NONE'
  if (
    typeof sequence.ownedTable !== 'string'
    || typeof sequence.ownedColumn !== 'string'
    || !isSafeReplicaTable(sequence.ownedTable, actualTables)
    || !isSafeIdentifier(sequence.ownedColumn)
  ) {
    return null
  }

  return `OWNED BY ${quoteQualifiedTable(sequence.ownedTable)}.${quoteIdent(sequence.ownedColumn)}`
}

function isSafeSequenceDataType(value: string): boolean {
  return value === 'smallint' || value === 'integer' || value === 'bigint'
}

function isSafeSequenceNumber(value: string): boolean {
  return /^-?\d+$/.test(value)
}

function buildDropIndexStatement(
  index: SchemaIndex,
  actualTables: Set<string>,
): string | null {
  if (index.constraintOwned !== false)
    return null
  if (!isSafeReplicaTable(index.table, actualTables))
    return null
  if (!isSafeQuotedIdentifier(index.name))
    return null

  return `DROP INDEX CONCURRENTLY IF EXISTS public.${quoteIdent(index.name)}`
}

function buildReindexIndexStatement(
  index: SchemaIndex,
  actualTables: Set<string>,
): string | null {
  if (
    !isSafeReplicaTable(index.table, actualTables)
    || !isSafeQuotedIdentifier(index.name)
  ) {
    return null
  }

  return `REINDEX INDEX CONCURRENTLY public.${quoteIdent(index.name)}`
}

function parseCreateIndexDefinition(
  definition: string,
): {
  unique: string
  indexName: string
  tableName: string
  indexTail: string
} | null {
  let rest = definition
  if (!rest.startsWith('CREATE '))
    return null
  rest = rest.slice('CREATE '.length)

  const unique = rest.startsWith('UNIQUE ') ? 'UNIQUE ' : ''
  if (unique)
    rest = rest.slice(unique.length)

  if (!rest.startsWith('INDEX '))
    return null
  rest = rest.slice('INDEX '.length)

  const parsedIndexName = parseSqlIdentifier(rest)
  if (!parsedIndexName)
    return null
  rest = parsedIndexName.rest.trimStart()

  const tablePrefix = 'ON public.'
  if (!rest.startsWith(tablePrefix))
    return null
  rest = rest.slice(tablePrefix.length)

  const parsedTableName = parseSqlIdentifier(rest)
  if (!parsedTableName)
    return null
  rest = parsedTableName.rest.trimStart()
  if (!rest)
    return null

  return {
    unique,
    indexName: parsedIndexName.identifier,
    tableName: parsedTableName.identifier,
    indexTail: rest,
  }
}

function parseSqlIdentifier(value: string): {
  identifier: string
  rest: string
} | null {
  if (value.startsWith('"')) {
    let identifier = ''
    for (let index = 1; index < value.length; index += 1) {
      if (value[index] !== '"') {
        identifier += value[index]
        continue
      }
      if (value[index + 1] === '"') {
        identifier += '"'
        index += 1
        continue
      }
      return { identifier, rest: value.slice(index + 1) }
    }
    return null
  }

  const match = /^([A-Z_]\w*)/i.exec(value)
  if (!match)
    return null

  return {
    identifier: match[1],
    rest: value.slice(match[1].length),
  }
}

function isSafeReplicaTable(table: string, actualTables: Set<string>): boolean {
  return (
    REPLICA_TABLE_SET.has(table)
    && actualTables.has(table)
    && isSafeIdentifier(table)
  )
}

function trackSkippedColumn(
  skippedColumnsByTable: Map<string, Set<string>>,
  column: SchemaColumn,
): void {
  const tableColumns
    = skippedColumnsByTable.get(column.table) ?? new Set<string>()
  tableColumns.add(column.name)
  skippedColumnsByTable.set(column.table, tableColumns)
}

function hasUnresolvedColumnDependency(
  index: SchemaIndex,
  skippedColumnsByTable: Map<string, Set<string>>,
): boolean {
  const skippedColumns = skippedColumnsByTable.get(index.table)
  if (!skippedColumns?.size)
    return false

  const parsedDefinition = parseCreateIndexDefinition(index.definition)
  if (!parsedDefinition)
    return false

  for (const column of skippedColumns) {
    if (indexTailReferencesColumn(parsedDefinition.indexTail, column))
      return true
  }

  return false
}

function indexTailReferencesColumn(indexTail: string, column: string): boolean {
  return (
    indexTail.includes(quoteIdent(column))
    || new RegExp(
      `(^|[^A-Za-z0-9_"])${escapeRegExp(column)}($|[^A-Za-z0-9_"])`,
    ).test(indexTail)
  )
}

function isSafeIdentifier(value: string): boolean {
  return SAFE_IDENTIFIER_RE.test(value)
}

function isSafeQuotedIdentifier(value: string): boolean {
  return value.length > 0 && !value.includes('\0')
}

function escapeRegExp(value: string): string {
  return value.replace(/[.*+?^${}()|[\]\\]/g, String.raw`\$&`)
}

function isSafeSqlFragment(value: string): boolean {
  return (
    SAFE_SQL_FRAGMENT_RE.test(value)
    && !value.includes(';')
    && !value.includes('--')
    && !value.includes('/*')
    && !value.includes('*/')
  )
}

function isSafeSchemaDefinition(value: string): boolean {
  return (
    value.length > 0
    && !value.includes('\0')
    && !value.includes(';')
    && !value.includes('--')
    && !value.includes('/*')
    && !value.includes('*/')
  )
}

function isSafeCheckConstraintDefinition(value: string): boolean {
  return (
    value.startsWith('CHECK (')
    && isSafeSchemaDefinition(value)
  )
}

function isConstraintType(value: unknown): value is SchemaConstraint['type'] {
  return value === 'c' || value === 'p' || value === 'u'
}

function constraintMatches(
  expected: SchemaConstraint,
  actual: SchemaConstraint | undefined,
): boolean {
  return (
    constraintShapeMatches(expected, actual)
    && (expected.valid === undefined || actual?.valid === undefined || expected.valid === actual.valid)
  )
}

function constraintShapeMatches(
  expected: SchemaConstraint,
  actual: SchemaConstraint | undefined,
): boolean {
  return (
    actual?.table === expected.table
    && actual.type === expected.type
    && actual.definition === expected.definition
  )
}

function functionMatches(
  expected: SchemaFunction,
  actual: SchemaFunction | undefined,
): boolean {
  return actual?.definition === expected.definition
}

function sequenceMatches(
  expected: SchemaSequence,
  actualSequences: readonly SchemaSequence[],
): boolean {
  const actual = actualSequences.find(sequence => sequence.name === expected.name)
  return (
    actual?.type === expected.type
    && actual.start === expected.start
    && actual.increment === expected.increment
    && actual.min === expected.min
    && actual.max === expected.max
    && actual.cache === expected.cache
    && actual.cycle === expected.cycle
    && actual.ownedTable === expected.ownedTable
    && actual.ownedColumn === expected.ownedColumn
  )
}

function isSafeConstantDefault(value: string): boolean {
  for (let index = 0; index < value.length; index += 1) {
    if (value[index] !== '(')
      continue

    const previous = previousNonWhitespace(value, index)
    if (previous && (isIdentifierChar(previous) || previous === '"'))
      return false
  }

  return true
}

function previousNonWhitespace(value: string, index: number): string | null {
  for (let cursor = index - 1; cursor >= 0; cursor -= 1) {
    if (!isWhitespace(value[cursor]))
      return value[cursor]
  }

  return null
}

function isWhitespace(value: string): boolean {
  return (
    value === ' '
    || value === '\n'
    || value === '\r'
    || value === '\t'
    || value === '\f'
  )
}

function isIdentifierChar(value: string): boolean {
  return (
    value === '_'
    || (value >= 'A' && value <= 'Z')
    || (value >= 'a' && value <= 'z')
    || (value >= '0' && value <= '9')
  )
}

function isNullDefault(value: string): boolean {
  return /^NULL\b/i.test(value)
}
function quoteQualifiedTable(table: string): string {
  return `public.${quoteIdent(table)}`
}

function quoteIdent(value: string): string {
  return `"${value.replaceAll('"', '""')}"`
}

function columnKey(column: Pick<SchemaColumn, 'table' | 'name'>): string {
  return `${column.table}.${column.name}`
}

function constraintKey(
  constraint: Pick<SchemaConstraint, 'table' | 'name'>,
): string {
  return `${constraint.table}.${constraint.name}`
}

function functionKey(
  fn: Pick<SchemaFunction, 'name' | 'arguments'>,
): string {
  return `${fn.name}(${fn.arguments})`
}