from __future__ import annotations

from contextlib import contextmanager
import hashlib
import time
from collections.abc import Iterator

from sqlalchemy import create_engine, text
from sqlalchemy.engine import make_url
from sqlalchemy.exc import SQLAlchemyError


class MigrationLockTimeout(RuntimeError):
    pass


def migration_lock_key(installation_id: str, migration_track: str) -> int:
    payload = f"govoplan\0{installation_id}\0{migration_track}".encode("utf-8")
    return int.from_bytes(hashlib.sha256(payload).digest()[:8], "big", signed=True)


@contextmanager
def deployment_migration_lock(
    database_url: str,
    *,
    installation_id: str,
    migration_track: str,
    timeout_seconds: float = 900.0,
    poll_seconds: float = 2.0,
) -> Iterator[None]:
    """Serialize all release migrations before coordination tables are available."""

    if timeout_seconds < 0 or poll_seconds <= 0:
        raise ValueError("Migration lock timeout must be non-negative and polling positive")
    if make_url(database_url).get_backend_name() != "postgresql":
        yield
        return
    key = migration_lock_key(installation_id, migration_track)
    engine = create_engine(database_url)
    connection = engine.connect()
    acquired = False
    try:
        deadline = time.monotonic() + timeout_seconds
        while True:
            acquired = bool(
                connection.execute(
                    text("SELECT pg_try_advisory_lock(:key)"),
                    {"key": key},
                ).scalar_one()
            )
            if acquired:
                break
            if time.monotonic() >= deadline:
                raise MigrationLockTimeout(
                    "Timed out waiting for the deployment-wide database migration lock"
                )
            time.sleep(min(poll_seconds, max(0.05, deadline - time.monotonic())))
        yield
    finally:
        if acquired:
            try:
                connection.execute(
                    text("SELECT pg_advisory_unlock(:key)"),
                    {"key": key},
                )
            except SQLAlchemyError:
                # Closing the physical connection releases session locks.
                pass
        connection.close()
        engine.dispose()


__all__ = [
    "MigrationLockTimeout",
    "deployment_migration_lock",
    "migration_lock_key",
]
