1"""How far apart are this database's schema and this checkout's code?
 2
 3Two directions, and they mean very different things:
 4
 5- **behind** — migrations on disk the database hasn't applied. Self-correcting;
 6  `plain postgres sync` applies them on the next `plain dev`.
 7- **ahead** — migrations recorded in the database with no file on disk. Not
 8  self-correcting, and the usual cause is switching branches: you applied a
 9  migration on one branch, switched away, and the table is still there.
10"""
11
12from __future__ import annotations
13
14from collections.abc import Generator
15from contextlib import contextmanager
16from typing import Any
17
18
19@contextmanager
20def _connected(url: str) -> Generator[Any]:
21    """Open `url` and install it as the active connection.
22
23    The migration recorder reads applied rows through `Migration.query`, which
24    uses the *global* connection rather than one passed in — so it has to be
25    installed, the same way `use_test_database` does it.
26    """
27    from plain.postgres.connection import DatabaseConnection
28    from plain.postgres.database_url import parse_database_url
29    from plain.postgres.db import _db_conn
30    from plain.postgres.sources import DirectSource
31
32    conn = DatabaseConnection(DirectSource(parse_database_url(url)))
33    token = _db_conn.set(conn)
34    try:
35        yield conn
36    finally:
37        _db_conn.reset(token)
38        conn.close()
39
40
41def pending_migration_count(url: str) -> int:
42    """Migrations on disk that this database hasn't applied yet."""
43    from plain.postgres.migrations.executor import MigrationExecutor
44
45    with _connected(url) as conn:
46        executor = MigrationExecutor(conn)
47        return len(executor.migration_plan(executor.loader.graph.leaf_nodes()))
48
49
50def migrations_not_on_disk(url: str) -> list[tuple[str, str]]:
51    """Migrations this database has applied that no longer exist as files.
52
53    This is the "your database is ahead of your code" signal. The same
54    comparison backs the `postgres.prunable_migrations` preflight check, but
55    the remedy differs by context: prune is right when a migration was deleted
56    for good, and wrong when you simply switched branches and will switch back.
57    """
58    from plain.postgres.migrations.loader import MigrationLoader
59    from plain.postgres.migrations.recorder import MigrationRecorder
60
61    with _connected(url) as conn:
62        loader = MigrationLoader(conn, ignore_no_migrations=True)
63        applied = MigrationRecorder(conn).applied_migrations()
64        on_disk = loader.disk_migrations or {}
65        return sorted(m for m in applied if m not in on_disk)