"""
Programmatic Alembic migration runner for per-user encrypted databases.

This module provides functions to run Alembic migrations against SQLCipher
encrypted databases without using the Alembic CLI. Each user database
tracks its own migration version via the alembic_version table.
"""

import os
import time
from pathlib import Path
from typing import Optional

from alembic import command
from alembic.config import Config
from alembic.runtime.migration import MigrationContext
from alembic.script import ScriptDirectory
from loguru import logger
from sqlalchemy import Connection, Engine, inspect
from sqlalchemy.exc import IntegrityError, OperationalError


def get_migrations_dir() -> Path:
    """
    Get the path to the migrations directory with security validation.

    Validates that the migrations directory is within the expected package
    boundary to prevent symlink attacks that could redirect migration loading
    to arbitrary locations.

    Returns:
        Path to the migrations directory

    Raises:
        ValueError: If the migrations path is outside expected boundaries
    """
    migrations_dir = Path(__file__).parent / "migrations"
    real_path = migrations_dir.resolve()
    expected_parent = Path(__file__).parent.resolve()

    # Security: Ensure migrations directory is within expected package boundary
    # This prevents symlink attacks that could load arbitrary Python code
    if not real_path.is_relative_to(expected_parent):
        raise ValueError(
            "Invalid migrations path (possible symlink attack): "
            "migrations dir resolves outside expected package boundary"
        )

    return migrations_dir


def _validate_migrations_permissions(migrations_dir: Path) -> None:
    """
    Validate migration files are not world-writable.

    World-writable migration files could be replaced with malicious code
    that would execute during database migrations with the application's
    privileges.

    Args:
        migrations_dir: Path to the migrations directory

    Raises:
        ValueError: If any migration file is world-writable

    Note:
        This check is skipped on Windows where file permissions work differently.
    """
    if os.name == "nt":  # Skip permission checks on Windows
        return

    versions_dir = migrations_dir / "versions"
    if not versions_dir.exists():
        return

    # Check the versions directory itself
    st = versions_dir.stat()
    if st.st_mode & 0o002:
        raise ValueError(
            f"Migrations directory has insecure permissions (world-writable): "
            f"{versions_dir}. Fix with: chmod o-w {versions_dir}"
        )

    for migration_file in versions_dir.glob("*.py"):
        st = migration_file.stat()
        if st.st_mode & 0o002:  # World-writable bit
            raise ValueError(
                f"Migration file has insecure permissions (world-writable): "
                f"{migration_file.name}. "
                f"Fix with: chmod o-w {migration_file}"
            )


def get_alembic_config(engine: Engine) -> Config:
    """
    Create an Alembic Config object for programmatic usage.

    Args:
        engine: SQLAlchemy engine to run migrations against

    Returns:
        Configured Alembic Config object
    """
    migrations_dir = get_migrations_dir()

    # Create config object without ini file
    config = Config()

    # Set script location
    config.set_main_option("script_location", str(migrations_dir))

    # Set SQLAlchemy URL (not actually used since we pass connection directly)
    # But Alembic requires it to be set
    config.set_main_option("sqlalchemy.url", "sqlite:///:memory:")

    return config


def get_current_revision(engine: Engine) -> Optional[str]:
    """
    Get the current migration revision for a database.

    Args:
        engine: SQLAlchemy engine

    Returns:
        Current revision string or None if no migrations have run
    """
    with engine.connect() as conn:
        context = MigrationContext.configure(conn)
        return context.get_current_revision()


def get_head_revision() -> Optional[str]:
    """
    Get the latest migration revision.

    Returns:
        Head revision string, or None if no migrations exist
    """
    migrations_dir = get_migrations_dir()
    config = Config()
    config.set_main_option("script_location", str(migrations_dir))

    script = ScriptDirectory.from_config(config)
    return script.get_current_head()


def needs_migration(engine: Engine) -> bool:
    """
    Check if a database needs migrations.

    Args:
        engine: SQLAlchemy engine

    Returns:
        True if migrations are pending
    """
    head = get_head_revision()

    if head is None:
        # No migrations exist yet
        return False

    current = get_current_revision(engine)

    if current is None:
        # Check if this is a fresh database or existing without migrations
        inspector = inspect(engine)
        tables = inspector.get_table_names()

        if not tables:
            # Fresh database, needs initial migration
            return True
        if "alembic_version" not in tables:
            # Existing database without Alembic - needs stamping then check
            return True

    return current != head


def stamp_database(engine: Engine, revision: str = "head") -> None:
    """
    Stamp a database with a revision without running migrations.
    Used for baselining existing databases.

    Concurrency: If two callers race to stamp a fresh database, one will hit
    "table alembic_version already exists" (OperationalError) or a duplicate
    PK on version_num (IntegrityError). Both outcomes are benign — the DB
    ends up stamped — so we swallow them after verifying the table+row are
    in place. A genuine failure (no row appeared) is re-raised.

    Args:
        engine: SQLAlchemy engine
        revision: Revision to stamp (default "head")
    """
    config = get_alembic_config(engine)

    try:
        with engine.begin() as conn:
            config.attributes["connection"] = conn
            command.stamp(config, revision)
    except (IntegrityError, OperationalError) as exc:
        # Only swallow errors that look like a benign concurrent-stamp
        # race on the alembic_version table itself. A genuine failure
        # (disk full, SQLITE_BUSY on an unrelated table, corruption,
        # etc.) must propagate so callers see the real error.
        msg = str(exc).lower()
        looks_like_race = (
            "alembic_version" in msg  # IntegrityError or table-exists race
            or "already exists" in msg  # CREATE TABLE race
        )
        if not looks_like_race or get_current_revision(engine) is None:
            raise
        # Race-loss path: another caller stamped first. Don't claim we
        # stamped it ourselves — log at debug only.
        logger.debug(
            f"stamp_database({revision}) lost race to concurrent caller "
            f"({type(exc).__name__}); database is stamped, continuing"
        )
        return

    logger.info(f"Stamped database at revision: {revision}")


def _drop_orphan_alembic_temp_tables(conn: Connection) -> None:
    """Drop leftover ``_alembic_tmp_<table>`` tables from prior failed
    batch_alter_table runs (issue #3817).

    ``op.batch_alter_table`` rebuilds a table by creating
    ``_alembic_tmp_<table>``, copying data, dropping the original, and
    renaming. On a clean run alembic drops the temp table automatically.
    If a previous attempt failed in a way that bypassed transaction
    rollback (e.g., an older migration runner that auto-committed each
    migration, or a process killed mid-DDL on a non-transactional
    sqlite build), the temp table persists. The next attempt then fails
    at ``op.batch_alter_table`` with ``table _alembic_tmp_* already exists``.

    This runs in autocommit mode at the SQLite level — each ``DROP TABLE``
    briefly takes the file write lock and releases it. If a concurrent
    migration is mid-batch_alter_table on the same table, our DROP blocks
    on the SQLite write lock (busy_timeout=10000); by the time we
    acquire it, the concurrent migration's rename has consumed the temp
    table and our DROP IF EXISTS is a no-op. The race is benign.
    """
    inspector = inspect(conn)
    temp_tables = [
        name
        for name in inspector.get_table_names()
        if name.startswith("_alembic_tmp_")
    ]
    if not temp_tables:
        return
    logger.warning(
        f"Found {len(temp_tables)} orphan alembic temp table(s) from a "
        f"prior failed migration: {sorted(temp_tables)}. Dropping before retry."
    )
    for name in temp_tables:
        # Identifier is constrained to the ``_alembic_tmp_`` prefix + a
        # parent table name from ``inspector.get_table_names()``; both
        # come from the database's own catalog and cannot contain
        # injection vectors.
        # bearer:disable python_lang_sql_injection
        conn.exec_driver_sql(f'DROP TABLE IF EXISTS "{name}"')  # noqa: S608


def _disable_fk_for_migration(conn: Connection) -> None:
    """Disable FK enforcement on the migration connection BEFORE any
    transaction opens (issue #3990).

    ``apply_performance_pragmas`` set ``PRAGMA foreign_keys = ON`` at
    connect. SQLite then *silently ignores* further toggles of
    ``foreign_keys`` once any transaction (explicit or driver-implicit)
    is active. The sqlite3/sqlcipher3 driver auto-begins on the first
    DML; PRAGMA itself isn't DML, so issuing the PRAGMA before any DML
    is the only window where it actually takes effect.

    With multi-migration upgrades (revision 0001 → 0009), the first
    migration to issue DML auto-begins the driver transaction and
    freezes FK in the connect-time ON state for the rest of the upgrade.
    That defeats migration 0007's defensive PRAGMA OFF and makes its
    orphan-scrub DELETE fail with ``foreign key mismatch`` on tables
    whose FK target lacks a UNIQUE backing — exactly the broken-schema
    state migration 0007 is meant to repair.

    The caller is responsible for re-enabling FK after the migration
    transaction commits, BEFORE returning the connection to the pool —
    see ``run_migrations``.
    """
    conn.exec_driver_sql("PRAGMA foreign_keys = OFF")
    # ``exec_driver_sql`` triggers SQLAlchemy autobegin even though no
    # sqlite-level transaction was opened (PRAGMA isn't DML, so the
    # driver doesn't auto-begin). Roll back the no-op SQLAlchemy
    # transaction so the caller's ``conn.begin()`` is allowed to start
    # a fresh one. ``PRAGMA foreign_keys`` is connection-level state and
    # survives ROLLBACK at the SQLite level — see
    # https://www.sqlite.org/pragma.html#pragma_foreign_keys.
    conn.rollback()


def run_migrations(engine: Engine, target: str = "head") -> None:
    """
    Run pending migrations on a database.

    The initial migration is idempotent (only creates tables that don't exist),
    so this function runs migrations rather than just stamping existing
    databases. This ensures any missing tables are created.

    When ``target == "head"`` and the database is already at head, the call
    short-circuits without opening a write transaction — calling
    ``command.upgrade()`` unconditionally would still open a write transaction
    via ``engine.begin()`` (taking a RESERVED lock on the SQLite file) just to
    discover there's nothing to apply, serialising concurrent readers behind
    a no-op on every cold engine reopen.

    Security validations performed before running migrations:
    - Migration directory path is within expected package boundary
    - Migration files are not world-writable

    Pre-upgrade hygiene (run outside the migration transaction):
    - Drop orphan ``_alembic_tmp_*`` tables from prior failed
      ``batch_alter_table`` runs (issue #3817).
    - Disable ``PRAGMA foreign_keys`` so 0007's orphan scrub can run
      (issue #3990). Re-enabled after a successful upgrade, before the
      connection returns to the pool.

    On failure inside the migration transaction, the inner
    ``conn.begin()`` rolls back automatically — the database stays at
    its previous revision. The original exception is re-raised so
    callers can decide how to handle it.

    Args:
        engine: SQLAlchemy engine to migrate
        target: Target revision (default "head" for latest)

    Raises:
        Exception: If migration fails (database is safely rolled back)
    """
    migration_start = time.perf_counter()

    # Security: Validate migrations directory and file permissions
    migrations_dir = get_migrations_dir()
    _validate_migrations_permissions(migrations_dir)

    head = get_head_revision()

    if head is None:
        # No migrations exist yet - nothing to do
        logger.debug("No migrations found, skipping")
        return

    current = get_current_revision(engine)

    # BUG-3747: Pre-Alembic baseline detection.
    #
    # A database that has schema tables but no alembic_version row was
    # created before commit 4fde036df (v1.4.0, 2026-03-21) via
    # Base.metadata.create_all(). Without stamping, command.upgrade() runs
    # 0001 (no-op for existing tables) followed by 0002+ against a legacy
    # column shape. Migration 0007's index backfill silently fails on
    # missing columns (e.g. settings.category), leaving the DB in a
    # corrupted state. Stamping at "0001" bypasses the broken path.
    if current is None:
        inspector = inspect(engine)
        existing_tables = set(inspector.get_table_names())

        # Defensive guard: refuse what looks like an auth database. The
        # auth DB has its own initialization path (`init_auth_database()`
        # in `auth_db.py`) and contains ONLY the `users` table. Pre-
        # Alembic user DBs ALSO contain `users` (created by the old
        # `Base.metadata.create_all()` path before migration 0001 added
        # the explicit skip), so we cannot just check "users present".
        # Instead we check the auth-DB *shape*: only `users`, optionally
        # alongside `alembic_version`. A real user DB always has 50+
        # other tables. If the auth engine is ever accidentally routed
        # through this function, this guard will refuse loudly rather
        # than silently pollute the auth DB with user-DB tables.
        non_metadata_tables = existing_tables - {"alembic_version"}
        if non_metadata_tables == {"users"}:
            raise RuntimeError(
                "Refusing to run migrations on what looks like an auth "
                f"database (only 'users' table present; tables: "
                f"{sorted(existing_tables)}). Auth DB is initialized via "
                "init_auth_database()."
            )

        # User-DB sentinels: both tables date to project inception
        # (2025-06-29) and have never been renamed. We require BOTH —
        # any single one could be present on a partial-init test DB
        # (e.g. one that ran `Setting.__table__.create()` directly)
        # where we'd want 0001's `create_all()` to add the missing
        # tables, not be skipped by stamping. A real pre-Alembic
        # production DB has 60+ tables and definitely has both sentinels.
        PRE_ALEMBIC_SENTINELS = {"settings", "research_history"}
        if PRE_ALEMBIC_SENTINELS.issubset(existing_tables):
            logger.warning(
                "BUG-3747: pre-Alembic database detected "
                f"({len(existing_tables)} tables, no alembic_version). "
                "Stamping at revision 0001 before applying migrations."
            )
            stamp_database(engine, "0001")
            current = get_current_revision(engine)
            logger.info(
                f"BUG-3747: pre-Alembic DB stamped at {current}; "
                "proceeding with upgrade to head"
            )

    # Short-circuit when the database is already at head. Calling
    # command.upgrade() unconditionally opens a write transaction via
    # engine.begin() even when there is nothing to apply — SQLite takes
    # a RESERVED lock on the file as soon as the first DML lands inside
    # that transaction, serialising concurrent readers behind a no-op on
    # every cold engine reopen. The fresh-DB path (current is None) still
    # runs the upgrade so tables and the alembic_version row get created.
    if current is not None and current == head and target == "head":
        logger.info(f"Database already at revision {head}; skipping upgrade")
        return

    if current is None:
        logger.warning(
            "Database has no migration history — applying migrations "
            f"(target={target})"
        )
    elif current != head and target == "head":
        logger.warning(
            f"Database schema outdated (revision {current}, "
            f"head is {head}) — applying migrations"
        )

    config = get_alembic_config(engine)

    try:
        with engine.connect() as conn:
            _drop_orphan_alembic_temp_tables(conn)
            _disable_fk_for_migration(conn)
            with conn.begin():
                config.attributes["connection"] = conn
                command.upgrade(config, target)
            # Re-enable FK on this connection BEFORE it returns to the
            # pool so subsequent checkouts see the production-default ON
            # state. The migration transaction has just committed, so we
            # are back outside any active transaction — PRAGMA toggles
            # work again. We can't rely on ``engine.dispose()`` to force
            # a fresh connection because engines built with ``creator=``
            # have ``url.database is None``, which fails the dispose
            # guard below and leaves FK=OFF leaking into the pool.
            conn.exec_driver_sql("PRAGMA foreign_keys = ON")
            conn.rollback()
    except Exception:
        logger.exception(
            "Database migration failed — database remains at previous "
            "revision (auto-rollback by transaction manager)"
        )
        raise

    # Belt-and-suspenders: dispose pooled connections after a successful
    # upgrade. With FK explicitly re-enabled above this is no longer
    # load-bearing for FK state, but it forces the next checkout through
    # ``apply_performance_pragmas`` which also resets temp_store, cache_size,
    # journal_mode, etc. — protecting against any future migration that
    # touches connection-level PRAGMAs not handled by the FK fix-up.
    # Skip for ``:memory:`` engines — those use a single shared connection
    # and disposing it would destroy the just-migrated database.
    db_name = engine.url.database
    if db_name and db_name != ":memory:":
        engine.dispose()

    new_revision = get_current_revision(engine)
    elapsed_ms = (time.perf_counter() - migration_start) * 1000
    if current != new_revision:
        logger.warning(
            f"Database migrated: {current} -> {new_revision} "
            f"({elapsed_ms:.0f}ms)"
        )
    elif elapsed_ms > 100:
        logger.info(
            f"Database already at revision {new_revision} "
            f"(no-op upgrade took {elapsed_ms:.0f}ms)"
        )
    else:
        logger.info(
            f"Database already at revision {new_revision} ({elapsed_ms:.0f}ms)"
        )
