apps/api/app/core/bootstrap.py

"""Schema bootstrap.

The project deliberately ships no migration tool: the schema is created from the
ORM metadata and then hardened with raw DDL (the append-only trigger, functional
indexes). That keeps the free-tier deployment story to a single command while
still putting the integrity rules *in the database*.

`init_schema` is idempotent and safe to call on every boot. Three properties
matter once the database is a *hosted* one rather than a container on localhost:

* **It must be cheap when there is nothing to do.** `create_all` plus the DDL
  loop is ~40 statements, and every statement is a network round trip. Against a
  database one ocean away that is tens of seconds of start-up on every boot, and
  a platform start-up probe will kill the container long before it finishes. So
  a single-round-trip check runs first and skips the whole thing when the schema
  is already current.

* **It must not open a hole while it runs.** The hardening DDL recreates the
  append-only trigger with `DROP` then `CREATE`. Between those two statements the
  audit log is writable. Skipping the DDL entirely when the guard is already
  installed means the normal boot never opens that window at all.

* **Concurrent boots must not race.** Two replicas starting together could
  interleave their `DROP`/`CREATE`. The apply path takes a transaction-scoped
  advisory lock, so they serialise and the loser sees a current schema.
"""

from __future__ import annotations

import asyncio

from sqlalchemy import CheckConstraint, UniqueConstraint, bindparam, text
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.ext.asyncio import AsyncConnection, AsyncEngine

from app.core.logging import get_logger
from app.models.tables import DDL_STATEMENTS, Base

logger = get_logger(__name__)

# Objects created by the raw DDL rather than by the ORM metadata.
_GUARD_TRIGGER = "trg_audit_log_no_update"
_GUARD_FUNCTION = "ledgerlens_audit_log_is_append_only"
_DDL_INDEXES = frozenset({"ix_extractions_vendor_lower"})

# Arbitrary but fixed: the advisory-lock key that serialises schema application
# across processes. Any constant works as long as every replica uses the same one.
_SCHEMA_LOCK_KEY = 0x1EDBE12E

_EXPECTED_TABLES = frozenset(Base.metadata.tables)
_EXPECTED_INDEXES = (
    frozenset(
        index.name
        for table in Base.metadata.tables.values()
        for index in table.indexes
        if index.name is not None
    )
    | _DDL_INDEXES
)

# Named CHECK and UNIQUE constraints. `table.indexes` above does not include
# these, so without them "the schema is current" was a claim about tables and
# indexes only — while the integrity rules this project actually relies on are
# constraints: the enum CHECKs, and `UNIQUE(document_id, fingerprint)` that stops
# re-screening duplicating a finding. A database carrying every table and index
# but none of those would have passed the probe, and in production
# (`DB_MANAGE_SCHEMA=false`) nothing would ever have healed it.
#
# `UNIQUE(file_hash)` — the idempotency key — is already covered: it is declared
# `unique=True, index=True`, so SQLAlchemy emits it as a named unique *index* and
# it appears above. The one gap left is `extractions.document_id`, whose
# column-level `unique=True` produces a constraint the metadata leaves unnamed;
# Postgres names it `extractions_document_id_key`, but that name is the server's
# invention rather than something declared here, so it is not asserted.
_EXPECTED_CONSTRAINTS = frozenset(
    constraint.name
    for table in Base.metadata.tables.values()
    for constraint in table.constraints
    if isinstance(constraint, CheckConstraint | UniqueConstraint) and constraint.name is not None
)

# One statement, one round trip: everything needed to decide whether the boot can
# skip the DDL. Counting rather than listing keeps the result tiny.
_SCHEMA_PROBE = (
    text(
        """
        SELECT
          (SELECT count(*) FROM pg_class c
             JOIN pg_namespace n ON n.oid = c.relnamespace
            WHERE n.nspname = current_schema()
              AND c.relkind = 'r'
              AND c.relname IN :tables)                       AS tables_present,
          (SELECT count(*) FROM pg_class c
             JOIN pg_namespace n ON n.oid = c.relnamespace
            WHERE n.nspname = current_schema()
              AND c.relkind = 'i'
              AND c.relname IN :indexes)                      AS indexes_present,
          (SELECT count(*) FROM pg_constraint c
             JOIN pg_namespace n ON n.oid = c.connamespace
            WHERE n.nspname = current_schema()
              AND c.conname IN :constraints)                  AS constraints_present,
          (SELECT count(*) FROM pg_trigger
            WHERE tgname = :guard_trigger AND NOT tgisinternal) AS guard_trigger,
          (SELECT count(*) FROM pg_proc p
             JOIN pg_namespace n ON n.oid = p.pronamespace
            WHERE n.nspname = current_schema()
              AND p.proname = :guard_function)                AS guard_function
        """
    )
    .bindparams(
        bindparam("tables", expanding=True),
        bindparam("indexes", expanding=True),
        bindparam("constraints", expanding=True),
    )
    .columns()
)


async def _schema_is_current(connection: AsyncConnection) -> bool:
    """True when every table, index and integrity guard is already installed.

    Deliberately conservative: anything missing returns False and the caller runs
    the full idempotent apply. Adding a table or index therefore heals itself on
    the next boot without needing a migration.
    """
    row = (
        await connection.execute(
            _SCHEMA_PROBE,
            {
                "tables": list(_EXPECTED_TABLES),
                "indexes": list(_EXPECTED_INDEXES),
                "constraints": list(_EXPECTED_CONSTRAINTS),
                "guard_trigger": _GUARD_TRIGGER,
                "guard_function": _GUARD_FUNCTION,
            },
        )
    ).one()
    return bool(
        row.tables_present == len(_EXPECTED_TABLES)
        and row.indexes_present == len(_EXPECTED_INDEXES)
        and row.constraints_present == len(_EXPECTED_CONSTRAINTS)
        and row.guard_trigger == 1
        and row.guard_function == 1
    )


async def _apply_schema(connection: AsyncConnection) -> None:
    """Create tables and apply the hardening DDL. Caller holds the advisory lock."""
    await connection.run_sync(Base.metadata.create_all)
    for statement in DDL_STATEMENTS:
        await connection.execute(text(statement))


class SchemaNotManagedError(RuntimeError):
    """The schema is incomplete and this process is not permitted to build it."""


async def init_schema(engine: AsyncEngine, *, manage: bool = True) -> None:
    """Create tables and apply hardening DDL. Idempotent and safe to run on every boot.

    With ``manage=False`` the schema is *verified* and never modified. That is what
    lets production run as a role holding DML rights and nothing else — no
    ownership, and therefore no ability to `ALTER TABLE audit_log DISABLE TRIGGER`,
    which is the one way to get around the append-only guarantee (AUDIT.md §4c,
    Residual 1).

    It refuses to start rather than continuing against a half-built schema. A
    service that boots and then fails on the first insert with a permission error
    is strictly harder to diagnose than one that says, at boot, exactly which
    command to run and as whom.
    """
    async with engine.connect() as connection:
        if await _schema_is_current(connection):
            logger.info("schema_ready", extra={"tables": len(_EXPECTED_TABLES), "applied": False})
            return

    if not manage:
        logger.error("schema_incomplete_and_unmanaged")
        msg = (
            "the schema is missing or incomplete and DB_MANAGE_SCHEMA is false, so this "
            "process will not create it. Run the migration as the owning role first:\n"
            "  DATABASE_URL='<owner url>' python scripts/grant_app_role.py --migrate\n"
            "This is deliberate: the application runs without ownership so it cannot "
            "disable the audit log's append-only trigger."
        )
        raise SchemaNotManagedError(msg)

    async with engine.begin() as connection:
        # Serialise concurrent boots; released automatically when the transaction ends.
        await connection.execute(
            text("SELECT pg_advisory_xact_lock(:key)"), {"key": _SCHEMA_LOCK_KEY}
        )
        # Re-check under the lock: a replica that raced us may have just finished.
        if await _schema_is_current(connection):
            logger.info("schema_ready", extra={"tables": len(_EXPECTED_TABLES), "applied": False})
            return
        await _apply_schema(connection)

    logger.info("schema_ready", extra={"tables": len(_EXPECTED_TABLES), "applied": True})


async def wait_for_database(
    engine: AsyncEngine,
    *,
    max_attempts: int = 5,
    initial_delay_s: float = 1.0,
) -> None:
    """Block until the database answers, with bounded exponential backoff.

    A serverless Postgres (Neon, Aurora Serverless) suspends its compute when
    idle and takes several seconds to wake. Without this, the first connection of
    a cold boot can exceed the driver timeout and the process dies with an opaque
    error even though the database is perfectly healthy a second later.

    Raises the last error once the attempts are exhausted, so a genuinely
    unreachable database still fails loudly rather than hanging for ever.
    """
    delay = initial_delay_s
    for attempt in range(1, max_attempts + 1):
        try:
            async with engine.connect() as connection:
                await connection.execute(text("SELECT 1"))
            if attempt > 1:
                logger.info("database_ready", extra={"attempts": attempt})
            return
        except (SQLAlchemyError, OSError) as exc:  # TimeoutError subclasses OSError
            if attempt == max_attempts:
                logger.error(
                    "database_unreachable",
                    extra={"attempts": attempt, "error": type(exc).__name__},
                )
                raise
            logger.warning(
                "database_not_ready_retrying",
                extra={"attempt": attempt, "delay_s": delay, "error": type(exc).__name__},
            )
            await asyncio.sleep(delay)
            delay *= 2


async def drop_schema(engine: AsyncEngine) -> None:
    """Drop every LedgerLens table. Used only by the integration test fixtures."""
    async with engine.begin() as connection:
        # The append-only trigger blocks DELETE but not DROP TABLE; remove it first
        # so a test teardown cannot be tripped up by its own guard rail.
        await connection.execute(text(f"DROP TRIGGER IF EXISTS {_GUARD_TRIGGER} ON audit_log;"))
        await connection.run_sync(Base.metadata.drop_all)
    logger.info("schema_dropped")