MikrotikManager/backend/src/db/database-source-capture-adopted.sql-helper.ts

/** Test-only adopted-writer capture regression fixture. Never imported by
 * startup, restore, dispatch or READY. Raw/unadopted SQL is NOT fenced. */
import { createHash } from "node:crypto"
import { resolve } from "node:path"
import { pathToFileURL } from "node:url"
if (process.env.MMAPP_OWN_SOURCE_CAPTURE !== "true" || !process.env.MMAPP_OWN_SOURCE_CAPTURE_ROOT || process.env.DATABASE_URL) throw Error("Owned capture runner required")
const source = (path: string) => pathToFileURL(resolve(process.env.MMAPP_OWN_SOURCE_CAPTURE_ROOT!, "backend/src", path)).href
const { PARTITION_WRITER_ADVISORY_KEY } = await import(source("services/partition-write-transaction.ts")) as typeof import("../services/partition-write-transaction.js")
const { captureMigrationData } = await import(source("db/migration-data-inventory.ts")) as typeof import("./migration-data-inventory.js")
export interface CaptureClient {
  query(sql: string, values?: unknown[]): Promise<{ rows: Record<string, unknown>[]; command?: string }>
  release(destroy?: boolean): void
}
export interface CapturePool { connect(): Promise<CaptureClient> }
export interface CapturePolicy { databaseName: string; captureRole: string; runtimeRole: string; timeoutMs: number }
export class AdoptedSourceCaptureError extends Error {
  readonly code = "ADOPTED_SOURCE_CAPTURE_REFUSED"
  constructor(message: string, readonly outcomeUnknown = false, options?: ErrorOptions) { super(message, options) }
}
const fail = (message: string): never => { throw new AdoptedSourceCaptureError(message) }
const digest = (value: unknown) => createHash("sha256").update(JSON.stringify(value)).digest("hex")
export const CAPTURE_SQL = Object.freeze({
  roleClosure: `WITH RECURSIVE roles(oid) AS (
 SELECT oid FROM pg_catalog.pg_roles WHERE rolname=$1
 UNION SELECT m.roleid FROM pg_catalog.pg_auth_members m JOIN roles r ON m.member=r.oid)
 SELECT r.oid::text AS oid,r.rolname,r.rolsuper,r.rolcreaterole,r.rolcreatedb,r.rolreplication,r.rolbypassrls,r.rolcanlogin
 FROM pg_catalog.pg_roles r JOIN roles c ON c.oid=r.oid ORDER BY r.oid LIMIT 129`,
  privilege: `SELECT
 EXISTS(SELECT 1 FROM pg_catalog.pg_database d WHERE d.datname=pg_catalog.current_database() AND (d.datdba=ANY($1::oid[]) OR EXISTS(SELECT 1 FROM unnest($1::oid[]) r WHERE pg_catalog.has_database_privilege(r,d.oid,'CREATE')))) AS database_owner_or_create,
 EXISTS(SELECT 1 FROM pg_catalog.pg_namespace n WHERE n.nspname='public' AND (n.nspowner=ANY($1::oid[]) OR EXISTS(SELECT 1 FROM unnest($1::oid[]) r WHERE pg_catalog.has_schema_privilege(r,n.oid,'CREATE')))) AS schema_owner_or_create,
 EXISTS(SELECT 1 FROM pg_catalog.pg_class c JOIN pg_catalog.pg_namespace n ON n.oid=c.relnamespace WHERE n.nspname='public' AND c.relowner=ANY($1::oid[])) AS relation_owner,
 EXISTS(SELECT 1 FROM pg_catalog.pg_proc p JOIN pg_catalog.pg_namespace n ON n.oid=p.pronamespace WHERE n.nspname='public' AND p.proowner=ANY($1::oid[])) AS routine_owner,
 EXISTS(SELECT 1 FROM pg_catalog.pg_proc p JOIN pg_catalog.pg_namespace n ON n.oid=p.pronamespace WHERE p.prosecdef AND n.nspname NOT IN ('pg_catalog','information_schema') AND EXISTS(SELECT 1 FROM unnest($1::oid[]) r WHERE pg_catalog.has_function_privilege(r,p.oid,'EXECUTE'))) AS callable_security_definer`,
  ownLock: `SELECT count(*)::int AS n FROM pg_catalog.pg_locks WHERE locktype='advisory' AND pid=pg_catalog.pg_backend_pid() AND classid=0 AND objid=$1::oid AND objsubid=1 AND mode='ExclusiveLock' AND granted`,
  context: `SELECT d.oid::text AS database_oid,d.datname,SESSION_USER::text AS session_user,CURRENT_USER::text AS current_user,
 pg_catalog.pg_backend_pid() AS backend_pid,c.system_identifier::text AS system_identifier,
 pg_catalog.pg_current_snapshot()::text AS snapshot,pg_catalog.transaction_timestamp()::text AS transaction_started,
 pg_catalog.current_setting('transaction_isolation') AS isolation,pg_catalog.current_setting('transaction_read_only') AS read_only,
 pg_catalog.pg_is_in_recovery() AS in_recovery
 FROM pg_catalog.pg_database d CROSS JOIN pg_catalog.pg_control_system() c WHERE d.datname=pg_catalog.current_database()`,
})
function policy(raw: CapturePolicy) {
  if (!raw || Object.getPrototypeOf(raw) !== Object.prototype || Object.keys(raw).sort().join(" ") !== "captureRole databaseName runtimeRole timeoutMs" ||
    Object.values(Object.getOwnPropertyDescriptors(raw)).some(d => !("value" in d) || !d.enumerable) || Reflect.ownKeys(raw).length !== 4 ||
    ![raw.databaseName, raw.captureRole, raw.runtimeRole].every(v => typeof v === "string" && /^[a-z][a-z0-9_]{0,62}$/.test(v)) ||
    raw.captureRole === raw.runtimeRole || !Number.isSafeInteger(raw.timeoutMs) || raw.timeoutMs < 100 || raw.timeoutMs > 30000) fail("Exact bounded trusted capture policy required")
  return Object.freeze(structuredClone(raw))
}
/** Owns the actual exclusive lock on the SAME checked-out backend used for
 * capture. It waits for existing adopted transactions and blocks later adopted
 * transactions/DDL before their first relation access. No lock callback seam. */
export async function openAdoptedSourceCapture(pool: CapturePool, raw: CapturePolicy, options: { signal?: AbortSignal } = {}) {
  const cfg = policy(raw)
  if (options.signal?.aborted) fail("Capture aborted before checkout")
  const client = await pool.connect()
  const deadlineAt = Date.now() + cfg.timeoutMs
  let closed = false, busy = false, used = false, transaction = false, commitSent = false
  const release = (discard: boolean) => {
    if (!closed) { closed = true; clearTimeout(timer); options.signal?.removeEventListener("abort", onAbort); client.release(discard) }
  }
  const destroy = () => release(true)
  const onAbort = () => destroy()
  const timer = setTimeout(destroy, cfg.timeoutMs); timer.unref()
  options.signal?.addEventListener("abort", onAbort, { once: true })
  const fence = () => {
    if (closed || options.signal?.aborted || Date.now() >= deadlineAt) { destroy(); fail("Capture closed, aborted or expired") }
  }
  const query: CaptureClient["query"] = async (sql, values) => {
    fence(); const result = await client.query(sql, values); fence(); return result
  }
  async function rolePolicy() {
    const roles = (await query(CAPTURE_SQL.roleClosure, [cfg.runtimeRole])).rows
    if (!roles.length || roles.length > 128 || !roles.some(r => r.rolname === cfg.runtimeRole && r.rolcanlogin === true) || roles.some(r =>
      typeof r.oid !== "string" || !/^[1-9][0-9]*$/.test(r.oid) || typeof r.rolname !== "string" || r.rolname.startsWith("pg_") ||
      [r.rolsuper,r.rolcreaterole,r.rolcreatedb,r.rolreplication,r.rolbypassrls].some(v => v !== false))) fail("Runtime role/membership has privileged bypass")
    const rows = (await query(CAPTURE_SQL.privilege, [roles.map(r => r.oid)])).rows
    if (rows.length !== 1 || Object.values(rows[0]).some(v => v !== false)) fail("Runtime role can own/create application schema or execute security-definer bypass")
    return digest({ roles, privileges: rows[0] })
  }
  async function ownedLock() { const rows = (await query(CAPTURE_SQL.ownLock, [PARTITION_WRITER_ADVISORY_KEY])).rows; if (rows.length !== 1 || rows[0].n !== 1) fail("Same-backend exclusive adopted-writer lease missing") }
  try {
    await query("SET lock_timeout='" + cfg.timeoutMs + "ms'")
    await query("SET statement_timeout='" + cfg.timeoutMs + "ms'")
    const identity = (await query("SELECT pg_catalog.current_database() AS database_name,CURRENT_USER::text AS current_user,SESSION_USER::text AS session_user")).rows
    if (identity.length !== 1 || identity[0].database_name !== cfg.databaseName || identity[0].current_user !== cfg.captureRole || identity[0].session_user !== cfg.captureRole) fail("Capture connection is not the explicitly provisioned source/operator")
    // Authenticate privileges before waiting: a superuser/owner application
    // role must not be mislabeled as an enforceable limited writer.
    const rolePolicySha256 = await rolePolicy()
    // SESSION lease in autocommit first: its waiting SELECT must finish BEFORE
    // BEGIN RR. Taking an xact lock inside RR could pin an old snapshot while
    // waiting and miss the writer's COMMIT even after the lock is granted.
    await query("SELECT pg_catalog.pg_advisory_lock($1::bigint)", [PARTITION_WRITER_ADVISORY_KEY]); await ownedLock()
    return Object.freeze({
      async capture() {
        if (closed || busy || used) fail("Capture handle closed/concurrent/reused")
        used = true; busy = true
        try {
          await ownedLock()
          if (await rolePolicy() !== rolePolicySha256) fail("Role policy changed while acquiring writer drain")
          await query("BEGIN ISOLATION LEVEL REPEATABLE READ READ ONLY"); transaction = true
          for (const statement of ["SET LOCAL search_path=pg_catalog", "SET LOCAL TimeZone='UTC'", "SET LOCAL DateStyle='ISO, YMD'", "SET LOCAL standard_conforming_strings=on", "SET LOCAL statement_timeout='5s'",
            "SET LOCAL row_security=off", "SET LOCAL extra_float_digits=3", "SET LOCAL bytea_output=hex", "SET LOCAL IntervalStyle=postgres", "SET LOCAL client_encoding='UTF8'", "SET LOCAL lc_monetary='C'", "SET LOCAL lc_numeric='C'"]) await query(statement)
          const first = (await query(CAPTURE_SQL.context)).rows
          if (first.length !== 1 || first[0].datname !== cfg.databaseName || first[0].current_user !== cfg.captureRole || first[0].session_user !== cfg.captureRole || first[0].isolation !== "repeatable read" || first[0].read_only !== "on" || first[0].in_recovery !== false) fail("Owned source snapshot context unavailable")
          const captured = await captureMigrationData({ query }, { scope: { applicationNamespaces: ["public"], excludedControlNamespaces: ["mmapp_control"] }, deadlineAt, maxRows: 100000, maxBytes: 16777216 })
          const last = (await query(CAPTURE_SQL.context)).rows
          if (JSON.stringify(first) !== JSON.stringify(last)) fail("Source/backend/snapshot changed across schema/data capture")
          await ownedLock(); commitSent = true
          const committed = await query("COMMIT")
          if (committed.command !== "COMMIT") fail("Owned capture COMMIT acknowledgement unavailable")
          transaction = false
          // Role catalogs are re-read OUTSIDE the old RR snapshot so membership
          // changes during capture are not hidden by snapshot consistency.
          if (await rolePolicy() !== rolePolicySha256) fail("Role policy changed during capture")
          await ownedLock()
          return Object.freeze({ format: "adopted-source-capture-v1" as const, source: Object.freeze(structuredClone(first[0])),
            rolePolicySha256, schemaSha256: captured.schemaSha256, dataSha256: captured.sha256, rowCount: captured.rowCount, sequenceCount: captured.sequences.length,
            adoptedWriterDrainAuthenticated: true as const, ownedRepeatableReadCaptureAuthenticated: true as const,
            allWriterDrainAuthenticated: false as const, sourceGenerationAuthenticated: false as const, readyAuthorized: false as const })
        } catch (cause) {
          destroy()
          throw new AdoptedSourceCaptureError("Owned adopted-writer source capture refused; no replay", transaction || commitSent, { cause })
        } finally { busy = false }
      },
      async close() {
        if (closed) return
        if (busy) fail("Capture still active")
        try {
          const released = (await query("SELECT pg_catalog.pg_advisory_unlock($1::bigint) AS unlocked", [PARTITION_WRITER_ADVISORY_KEY])).rows
          if (released.length !== 1 || released[0].unlocked !== true) fail("Adopted-writer lease release unknown")
          await query("RESET lock_timeout"); await query("RESET statement_timeout")
          release(false)
        } catch (cause) { destroy(); throw new AdoptedSourceCaptureError("Capture owner release unconfirmed", true, { cause }) }
      },
    })
  } catch (cause) { destroy(); throw cause }
}