check_schema_drift.py

"""Ask a hosted database whether it still is what this repository says it is.

`tests/test_migrations.py` proves the revisions build the right schema, and
`alembic check` inside it proves the models and the migrations agree. Both
questions are asked of a throwaway Postgres that the suite created seconds
earlier. Neither is a claim about Supabase.

The gap that leaves is the one this closes. A table altered in the Supabase
dashboard - a column widened to unblock something, row-level security clicked
off while chasing a 403 - produces no diff, no pull request and no failing
test. The next migration then runs against a schema nobody described, which is
the point at which it is discovered.

**Three states, and only two of them are faults.** The distinction is the
whole design, because a check that goes red for a normal condition is one
people learn to scroll past - the same reasoning
`.github/workflows/check-upstream-freshness.yml` records for declining to fail
on stale upstream data.

    BEHIND    The database is at a revision this repository knows, with more
              revisions after it. That is every moment between a migration
              merging and somebody applying it, which for production is a
              deliberate wait (RELEASING.md 8c: expand and contract across two
              releases, never both at once). Reported, never failed.

    UNKNOWN   The database is at a revision this repository has never heard
              of. Something applied a migration from somewhere else - another
              branch, a second ledger, a hand-run `alembic stamp`. Fails.

    DRIFTED   The database is at head, and `alembic check` still finds a
              difference. Everything the repository knows about has been
              applied and the schema is *still* not what the models describe,
              which leaves editing-by-hand as the explanation. Fails.

An empty database is not a fault either: it is indistinguishable from one that
has not been stood up yet, which is exactly where UA and production both are
as this is written (LAUNCH_CHECKLIST.md 5).

**Read-only.** It reads `alembic_version` and reflects the schema; autogenerate
comparison writes nothing. It is not stdlib-only, unlike
`check_supabase_config.py` - it needs Alembic's own revision graph to tell
BEHIND from UNKNOWN, and reimplementing that against the filenames would be a
second, worse copy of it.

**What it deliberately does not check: row-level security.** `supabase_keepalive.py`
already asks that question twice a day, over PostgREST with the anon key -
which is the front door an attacker would use, and a stronger claim than
anything this script could make holding the owner's credentials, since RLS
does not apply to a table's owner. One home per item (CONTRIBUTING.md).
"""

from __future__ import annotations

import argparse
import sys
from pathlib import Path

import sqlalchemy
from alembic.config import Config
from alembic.runtime.migration import MigrationContext
from alembic.script import ScriptDirectory
from alembic.util.exc import AutogenerateDiffsDetected
from sqlalchemy.engine.url import make_url

from alembic import command

BACKEND_DIR = Path(__file__).resolve().parent

# Supavisor's transaction mode. The one port a migration must never use: it
# hands each transaction whatever backend is free, so DDL and the advisory
# lock land on different sessions. Session mode is 5432 on the same host and
# is the right one - see unsuitable_reason().
TRANSACTION_POOLER_PORT = 6543

AT_HEAD = "AT_HEAD"
BEHIND = "BEHIND"
UNKNOWN = "UNKNOWN"
EMPTY = "EMPTY"

# Which states mean somebody should do something about the database rather
# than about this check. Kept as data so the test suite asserts the policy
# rather than restating it.
FAILING_STATES = frozenset({UNKNOWN})


def load_settings():
    """`app.config.settings`, imported late and without a traceback.

    Two reasons this is not a module-level import.

    **A missing setting must not print the connection string.**
    `app/config.py` builds `Settings()` at import time, and several of its
    fields have no default on purpose - so an environment carrying only
    DATABASE_URL raises `ValidationError`, and pydantic's rendering of that
    includes `input_value={'database_url': 'postgre...'}`. GitHub masks
    registered secrets in logs by matching the value, and a truncated middle
    matches nothing, so the fragments print. That happened, on the merge
    commit of #411. Only `loc` is read below; the input never is.

    **A migration job is not the app.** It needs a database URL, and the
    Supabase fields it will never touch are required all the same. The
    workflows pass them rather than this file defaulting them, because a
    default here would also apply to the app, where failing loudly is the
    documented intent.
    """
    from pydantic import ValidationError

    try:
        from app.config import settings
    except ValidationError as error:
        missing = ", ".join(str(item["loc"][0]).upper() for item in error.errors())
        raise SettingsMissing(missing) from None
    return settings


class SettingsMissing(Exception):
    """Named rather than a bare exit, so main() reports it in one place and
    the tests can assert on which settings were missing without parsing a
    message."""


def alembic_config() -> Config:
    """Alembic's own config, resolved absolutely.

    `alembic/env.py` overrides `sqlalchemy.url` from `app.config.settings`, so
    this script cannot disagree with the app about which database it means -
    and pointing it somewhere else is a matter of setting DATABASE_URL, the
    same lever a deploy pulls.
    """
    config = Config(str(BACKEND_DIR / "alembic.ini"))
    config.set_main_option("script_location", str(BACKEND_DIR / "alembic"))
    return config


def unsuitable_reason(url: str) -> str | None:
    """Why this connection string cannot apply a migration, or None.

    Both failures below are quiet and arrive far from their cause, which is
    the only reason a string is inspected here rather than simply used.

    **No `+psycopg`.** Supabase's dashboard hands out `postgresql://...`, and
    `app/config.py` uses DATABASE_URL exactly as given - deliberately, since
    silently rewriting a credential is worse than refusing one. SQLAlchemy
    resolves a bare `postgresql://` to psycopg2, which this backend does not
    install (requirements.in pins `psycopg[binary]`, v3), so the failure is an
    import error about a driver nobody chose.

    **Port 6543.** Supavisor's transaction mode, and the pooled string
    `LAUNCH_CHECKLIST.md` 6.2 correctly tells you to give the running *app*.
    For a migration it is wrong for the reason `app/db/session.py` documents
    at length: each transaction gets whatever backend is free, so `CREATE
    TABLE`, `ALTER TABLE` and Alembic's advisory lock do not share a session.
    Session mode on 5432 is the same host with the property migrations need.

    The direct endpoint (`db.<ref>.supabase.co`) is *not* rejected, and that
    is a judgement rather than an oversight: it is the best target for a
    migration and the only wrong thing about it is reachability. Supabase
    serves it over IPv6 unless the project buys the IPv4 add-on, and GitHub's
    hosted runners are IPv4-only - so it works from a maintainer's laptop and
    times out from CI. A warning belongs there; a refusal does not.
    """
    parsed = make_url(url)
    if parsed.get_driver_name() != "psycopg":
        return (
            f"DATABASE_URL names the '{parsed.drivername}' driver. This backend installs psycopg v3 only, so the URL "
            f"has to say so: postgresql+psycopg://... Supabase's dashboard gives you postgresql://..., and the "
            f"'+psycopg' is the edit to make when you paste it."
        )
    if parsed.port == TRANSACTION_POOLER_PORT:
        return (
            f"DATABASE_URL points at port {TRANSACTION_POOLER_PORT}, Supavisor's transaction mode. A migration needs "
            f"one session that stays put; that pooler gives each transaction a different backend. Use session mode - "
            f"the same host on port 5432 - or the direct endpoint if this network has IPv6."
        )
    return None


def warn_if_hard_to_reach(url: str) -> str | None:
    """The direct endpoint is right, and unreachable from a hosted runner."""
    host = make_url(url).host or ""
    if host.startswith("db.") and host.endswith(".supabase.co"):
        return (
            "DATABASE_URL is Supabase's direct endpoint, which is the best target for a migration but is served over "
            "IPv6 unless the project has the IPv4 add-on. GitHub's hosted runners are IPv4-only, so if this run times "
            "out connecting, that is why - switch to the session pooler (port 5432 on aws-<region>.pooler.supabase.com)."
        )
    return None


def classify(db_revision: str | None, head: str, known: set[str], pending: list[str]) -> tuple[str, str]:
    """Where the database sits relative to this checkout, and why in one line.

    Pure, so the states that matter can be tested without standing up a
    database in each of them - which for UNKNOWN would mean manufacturing a
    revision the repository does not contain.
    """
    if db_revision is None:
        return EMPTY, "No alembic_version row - this database has never had a migration applied."
    if db_revision not in known:
        return UNKNOWN, (
            f"The database is at {db_revision}, which is not a revision in this repository. Something applied a "
            f"migration from somewhere else, so what is in this database is not described by anything here."
        )
    if db_revision == head:
        return AT_HEAD, f"At head ({head})."
    return BEHIND, f"At {db_revision}, {len(pending)} revision(s) behind head ({head}): {', '.join(pending)}."


def read_current_revision(url: str) -> str | None:
    engine = sqlalchemy.create_engine(url)
    try:
        with engine.connect() as connection:
            heads = MigrationContext.configure(connection).get_current_heads()
    finally:
        engine.dispose()
    # A branched history would return several. This project has a single
    # linear chain, and a second head is itself worth failing on rather than
    # silently picking one - classify() sees a revision it cannot match.
    if not heads:
        return None
    return heads[0] if len(heads) == 1 else "+".join(sorted(heads))


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    parser.add_argument("--label", default="database", help="What to call this database in the output, e.g. production.")
    parser.add_argument(
        "--url-only",
        action="store_true",
        help="Check the connection string is one a migration can use, then stop. Touches no network.",
    )
    args = parser.parse_args(argv)

    # Before anything else, including reading the URL: app.config is what
    # holds it, and constructing it is the step that can fail.
    try:
        settings = load_settings()
    except SettingsMissing as missing:
        print(
            f"::error::{args.label}: the environment is missing {missing}, which app/config.py requires before any "
            f"setting can be read - including DATABASE_URL. A migration job does not use them, but Settings() will "
            f"not build without them, and neither will alembic/env.py. The workflows pass them; see migrate.yml."
        )
        return 2

    # Before anything reaches the network, because both of these fail in ways
    # that do not name themselves - and migrate.yml runs this first, so a
    # mistyped secret is caught before `alembic upgrade head` half-applies
    # anything.
    problem = unsuitable_reason(settings.database_url)
    if problem:
        print(f"::error::{problem}")
        return 2
    warning = warn_if_hard_to_reach(settings.database_url)
    if warning:
        print(f"::warning::{warning}")
    if args.url_only:
        print(f"{args.label}: the connection string is usable for a migration.")
        return 0

    config = alembic_config()
    script = ScriptDirectory.from_config(config)
    head = script.get_current_head()
    known = {revision.revision for revision in script.walk_revisions()}

    print(f"Checking {args.label} against {len(known)} revision(s) in this checkout.")

    try:
        db_revision = read_current_revision(settings.database_url)
    except sqlalchemy.exc.SQLAlchemyError as error:
        # Being unable to reach the database is a broken check, not a verdict
        # about the schema. Say which, so a network blip is never read as
        # drift.
        print(f"::error::Could not read {args.label}'s current revision: {error.__class__.__name__}: {error}")
        return 2

    pending = []
    if db_revision in known and db_revision != head:
        pending = [
            revision.revision for revision in script.iterate_revisions(head, db_revision) if revision.revision != db_revision
        ]
        pending.reverse()

    state, detail = classify(db_revision, head, known, pending)
    print(f"{state}: {detail}")

    if state == AT_HEAD:
        try:
            command.check(config)
        except AutogenerateDiffsDetected as diffs:
            print(
                f"::error::{args.label} is at head and still differs from the models. Every migration this repository "
                f"holds has been applied, so this schema was changed by something other than a migration - the "
                f"Supabase dashboard, most likely. Write a revision that captures it, or put the schema back.\n{diffs}"
            )
            return 1
        print("No difference between the models and the live schema.")
        return 0

    if state in FAILING_STATES:
        return 1

    # BEHIND and EMPTY both reach here. Neither is a fault; both are worth
    # saying out loud, because "nothing to do" and "somebody still has to
    # apply this" look identical in a green run otherwise.
    print(f"::notice::{args.label}: {detail}")
    return 0


if __name__ == "__main__":
    sys.exit(main())