"""
Encrypted database management using SQLCipher.
Handles per-user encrypted databases with browser-friendly authentication.
"""

import os
import threading
import time
from pathlib import Path
from typing import Any, Dict, Optional

from loguru import logger
from sqlalchemy import create_engine, event, text
from sqlalchemy.engine import Engine
from sqlalchemy.orm import Session, sessionmaker
from sqlalchemy.pool import QueuePool, StaticPool

from ..config.paths import get_data_directory, get_user_database_filename
from ..settings.env_registry import get_env_setting

# ``redact_secrets`` is imported lazily inside each except handler that
# uses it. A top-level import would trip a circular-import chain:
#
#   security/__init__.py
#     → security/file_integrity/integrity_manager.py
#       → database/session_context.py
#         → database/encrypted_db.py  (mid-load, ``db_manager`` not yet
#                                       defined → ImportError that
#                                       integrity_manager catches and
#                                       sets ``_has_session_context`` to
#                                       False for the rest of the run)
#
# This file sits on the import path that ``security/__init__.py``
# re-enters via the file_integrity submodule. Other files in
# PRs #4168/#4175/#4181 can import ``redact_secrets`` at the top
# because they are not on that path.
from .sqlcipher_compat import get_sqlcipher_module
from .pool_config import POOL_PRE_PING, POOL_RECYCLE_SECONDS
from .sqlcipher_utils import (
    set_sqlcipher_key,
    set_sqlcipher_rekey,
    apply_cipher_defaults_before_key,
    apply_sqlcipher_pragmas,
    apply_performance_pragmas,
    verify_sqlcipher_connection,
    create_database_salt,
    has_per_database_salt,
    get_key_from_password,
    get_sqlcipher_version,
    create_sqlcipher_connection,
)


class DatabaseInitializationError(Exception):
    """Raised when a per-user database opens but its schema can't be initialised.

    Distinct from credential / decryption failures (which return ``None``
    from :py:meth:`DatabaseManager.open_user_database`) so callers — chiefly
    the login route — can avoid penalising the user's lockout counter and
    surface a different error message. The credentials are valid; the
    database state isn't.
    """


def _best_effort_chmod(path, mode: int, *, warn: bool = False) -> None:
    """Tighten permissions on ``path`` to ``mode``, never raising.

    Permission hardening must NOT be able to break database creation. On
    filesystems that don't support POSIX chmod — some Docker bind mounts and
    network/FUSE volumes, notably Docker Desktop on macOS/Windows — os.chmod
    can raise OSError even for the file's owner. In that case we leave the
    path at its default (umask) mode rather than failing registration/login.

    ``warn=True`` surfaces the failure at warning level: a silent downgrade
    leaves the path group/world-readable, which matters for DB files holding
    user data (the unencrypted fallback is plaintext). The owner-only
    directory chmod is defense-in-depth, so it stays at debug.
    """
    try:
        os.chmod(str(path), mode)
    except OSError:
        log = logger.warning if warn else logger.debug
        log(
            f"Could not set permissions {oct(mode)} on {path}; "
            "left at filesystem default",
            exc_info=True,
        )


class DatabaseManager:
    """Manages encrypted SQLCipher databases for each user."""

    def __init__(self):
        self.connections: Dict[str, Engine] = {}
        self._connections_lock = threading.RLock()
        # Per-user locks serializing the cold-open (engine build + migration)
        # so two concurrent first-opens of one user never run alembic against
        # the same database file at once. Created lazily under
        # _connections_lock; see open_user_database / _get_init_lock.
        self._init_locks: Dict[str, threading.Lock] = {}
        self.data_dir = get_data_directory() / "encrypted_databases"
        self.data_dir.mkdir(parents=True, exist_ok=True)
        # Restrict the per-user DB directory to the owner (0o700). Belt-and-
        # suspenders with the per-file 0o600 chmod: it also covers SQLite's
        # WAL/SHM sidecars and any temp files (which the unencrypted fallback
        # writes in PLAINTEXT and which SQLite may recreate at umask perms on
        # each connect), so no sibling-readable user data is exposed to other
        # local accounts regardless of per-file modes.
        _best_effort_chmod(self.data_dir, 0o700)

        # Check SQLCipher availability
        self.has_encryption = self._check_encryption_available()

        # ----------------------------------------------------------------
        # Pool class selection — see ADR-0004
        #
        # We use QueuePool (pool_size=20, max_overflow=40,
        # pool_timeout=10) for production and StaticPool for tests.
        #
        # Why pool_size=20:
        #
        # 1. SQLCipher + WAL mode can leak file handles when connections
        #    close out of open-order. Fewer pooled connections = fewer
        #    opportunities for out-of-order closes during pool_recycle.
        #    See: https://github.com/sqlcipher/android-database-sqlcipher/issues/6
        #    See: https://github.com/dotnet/efcore/issues/35010
        #
        # 2. SQLite serializes all writes through a single file lock.
        #    Multiple pooled connections don't improve throughput — they
        #    just hold FDs (up to 3 per connection in WAL mode).
        #
        # 3. The cleanup scheduler periodically calls engine.dispose()
        #    to release all pooled connections, preventing long-lived
        #    handles from accumulating over days of idle operation.
        #
        # Why pool_size=20 and not 1: inject_current_user() creates a
        # QueuePool session on every request via g.db_session. With the
        # UI polling /api/research/<id>/status every 1-2s plus other
        # API calls and before_request middleware, pool_size=1
        # (max_overflow=2, so 3 total) is easily exhausted — causing
        # 30-second timeouts and PendingRollbackError cascades.
        # pool_size=20 + max_overflow=40 (60 total) provides ample
        # headroom for concurrent requests and multiple browser tabs.
        #
        # Why not NullPool: SQLCipher's PRAGMA key adds ~0.2ms per
        # connection open. With 20-30 queries per page load, NullPool
        # adds a noticeable 4-6ms overhead vs QueuePool's ~1.5ms.
        # ----------------------------------------------------------------
        self._use_static_pool = bool(os.environ.get("TESTING"))
        self._pool_class = StaticPool if self._use_static_pool else QueuePool

    def _get_pool_kwargs(self) -> Dict[str, Any]:
        """Get pool configuration kwargs based on pool type.

        StaticPool doesn't support pool_size or max_overflow.
        QueuePool uses moderate sizing to handle concurrent web requests
        while limiting FD usage. See ADR-0004 for rationale.
        """
        if self._use_static_pool:
            return {}
        return {
            "pool_size": 20,
            "max_overflow": 40,
            "pool_timeout": 10,
            "pool_pre_ping": POOL_PRE_PING,
            "pool_recycle": POOL_RECYCLE_SECONDS,
        }

    def _is_valid_encryption_key(self, password: str) -> bool:
        """
        Check if the provided password is valid (not None, empty, or whitespace-only).

        Args:
            password: The password to check

        Returns:
            True if the password is valid, False otherwise
        """
        return password is not None and password.strip() != ""

    def is_user_connected(self, username: str) -> bool:
        """Check if a user has an active database connection.

        Thread-safe accessor for external callers.

        Args:
            username: The username to check

        Returns:
            True if the user has an active connection
        """
        with self._connections_lock:
            return username in self.connections

    def _check_encryption_available(self) -> bool:
        """Check if SQLCipher is available for encryption."""
        try:
            import os as os_module
            import tempfile

            # Test if SQLCipher actually works, not just if it imports
            with tempfile.NamedTemporaryFile(delete=False) as tmp:
                tmp_path = tmp.name

            try:
                # Try to create a test encrypted database
                sqlcipher_module = get_sqlcipher_module()
                sqlcipher = sqlcipher_module.dbapi2

                conn = sqlcipher.connect(tmp_path)
                try:
                    cursor = conn.cursor()
                    # Use creation_mode=True since we're creating a new test database
                    apply_cipher_defaults_before_key(cursor)
                    # Use centralized key setting
                    set_sqlcipher_key(cursor, "testpass")
                    # Apply post-key pragmas (kdf_iter for new DB)
                    apply_sqlcipher_pragmas(cursor, creation_mode=True)
                    apply_performance_pragmas(cursor)

                    # Check SQLCipher version
                    version = get_sqlcipher_version(cursor)
                    if version:
                        major = (
                            version.split(".")[0]
                            if "." in version
                            else version[0]
                        )
                        if major.isdigit() and int(major) < 4:
                            logger.warning(
                                f"SQLCipher version {version} detected. "
                                "Version 4.x+ is recommended for proper PRAGMA ordering."
                            )

                    cursor.close()
                    # Now use the connection for table operations
                    conn.execute("CREATE TABLE test (id INTEGER PRIMARY KEY)")
                    conn.execute("INSERT INTO test VALUES (1)")
                    result = conn.execute("SELECT * FROM test").fetchone()

                    if result != (1,):
                        raise RuntimeError("SQLCipher encryption test failed")
                    logger.info(
                        "SQLCipher available and working - databases will be encrypted"
                    )
                    return True
                finally:
                    from ..utilities.resource_utils import safe_close

                    safe_close(conn, "SQLCipher test connection")
            except Exception:
                logger.warning("SQLCipher module found but not working")
                raise ImportError("SQLCipher not functional")
            finally:
                # Clean up test file
                try:
                    os_module.unlink(tmp_path)
                except OSError as e:
                    logger.debug(
                        f"Failed to clean up temp file {tmp_path}: {e}"
                    )

        except ImportError:
            # Check if user has explicitly allowed unencrypted databases.
            # Registry handles deprecated LDR_ALLOW_UNENCRYPTED fallback automatically.
            allow_unencrypted = get_env_setting(
                "bootstrap.allow_unencrypted", False
            )

            if not allow_unencrypted:
                # No ``password`` is in scope here (this runs at
                # bootstrap, before any user authenticates), and the
                # caught ``ImportError`` is purely about the SQLCipher
                # package being missing. Still drop the traceback to
                # keep this file's logging discipline uniform with the
                # rest of the #4182 sweep: no exception traceback is
                # written to the production log from this module. The
                # install-hint message is the user-facing diagnostic;
                # the traceback adds no value beyond it. ``warning``
                # rather than ``error``/``exception`` so the custom
                # "logger.exception in except blocks" lint passes
                # without re-introducing a traceback render.
                logger.warning(
                    "SECURITY ERROR: SQLCipher is not installed!\n"
                    "Your databases will NOT be encrypted.\n"
                    "To fix this:\n"
                    "1. Install SQLCipher: sudo apt install sqlcipher libsqlcipher-dev\n"
                    "2. Reinstall project: pdm install\n"
                    "Or use Docker with SQLCipher pre-installed.\n\n"
                    "To explicitly allow unencrypted databases (NOT RECOMMENDED):\n"
                    "export LDR_BOOTSTRAP_ALLOW_UNENCRYPTED=true"
                )
                raise RuntimeError(
                    "SQLCipher not available. Set LDR_BOOTSTRAP_ALLOW_UNENCRYPTED=true to proceed without encryption (NOT RECOMMENDED)"
                )
            logger.warning(
                "WARNING: Running with UNENCRYPTED databases!\n"
                "This means:\n"
                "- Passwords don't protect data access\n"
                "- API keys are stored in plain text\n"
                "- Anyone with file access can read all data\n"
                "Install SQLCipher for secure operation!"
            )
            return False

    def _get_user_db_path(self, username: str) -> Path:
        """Get the path for a user's encrypted database."""
        return self.data_dir / get_user_database_filename(username)

    def _apply_pragmas(self, connection, connection_record):
        """Apply pragmas for optimal performance."""
        # Check if this is SQLCipher or regular SQLite
        is_encrypted = self.has_encryption

        # Use centralized performance pragma application

        apply_performance_pragmas(connection)

        # SQLCipher-specific pragmas
        if is_encrypted:
            from .sqlcipher_utils import get_sqlcipher_settings

            settings = get_sqlcipher_settings()
            pragmas = [
                f"PRAGMA kdf_iter = {settings['kdf_iterations']}",
                f"PRAGMA cipher_page_size = {settings['page_size']}",
            ]
            for pragma in pragmas:
                try:
                    connection.execute(pragma)
                except Exception as e:
                    logger.debug(f"Could not apply pragma '{pragma}': {e}")
        else:
            # Regular SQLite pragma
            try:
                connection.execute(
                    "PRAGMA mmap_size = 268435456"
                )  # 256MB memory mapping
            except Exception as e:
                logger.debug(f"Could not apply mmap_size pragma: {e}")

    @staticmethod
    def _make_sqlcipher_connection(
        db_path: Path,
        password: str,
        isolation_level: Optional[str] = "IMMEDIATE",
        check_same_thread: bool = False,
    ) -> Any:
        """Create a properly initialized SQLCipher connection.

        Follows the canonical SQLCipher initialization order: set key,
        apply cipher pragmas, verify, then apply performance pragmas.
        Cipher pragmas (page size, HMAC algorithm, KDF iterations) must
        be configured before the first query (verification) because that
        query triggers page decryption with the active cipher settings.

        Args:
            db_path: Path to the database file
            password: The database encryption passphrase
            isolation_level: SQLite isolation level (``""`` for deferred
                transactions, ``None`` for autocommit)
            check_same_thread: SQLite check_same_thread flag

        Returns:
            A raw ``sqlcipher3`` connection ready for use.

        Raises:
            ValueError: If the database key cannot be verified.
        """
        sqlcipher3 = get_sqlcipher_module()
        conn = sqlcipher3.connect(
            str(db_path),
            isolation_level=isolation_level,
            check_same_thread=check_same_thread,
        )
        cursor = conn.cursor()

        try:
            set_sqlcipher_key(cursor, password, db_path=db_path)
            apply_sqlcipher_pragmas(cursor, creation_mode=False)

            if not verify_sqlcipher_connection(cursor):
                raise ValueError("Failed to verify database key")  # noqa: TRY301 — cleanup in except before re-raise

            apply_performance_pragmas(cursor)
        except Exception:
            try:
                cursor.close()
            except Exception:  # noqa: BLE001
                logger.warning("Failed to close cursor during cleanup")
            from ..utilities.resource_utils import safe_close

            safe_close(conn, "encrypted DB connection")
            raise

        cursor.close()
        return conn

    def create_user_database(self, username: str, password: str) -> Engine:
        """Create a new encrypted database for a user."""

        # Validate the encryption key
        if not self._is_valid_encryption_key(password):
            logger.error(
                f"Invalid encryption key for user {username}: password is None or empty"
            )
            raise ValueError(
                "Invalid encryption key: password cannot be None or empty"
            )

        db_path = self._get_user_db_path(username)

        if db_path.exists():
            raise ValueError(f"Database already exists for user {username}")

        # Create connection string - use regular SQLite when SQLCipher not available
        if self.has_encryption:
            # Create directory if it doesn't exist
            db_path.parent.mkdir(parents=True, exist_ok=True)

            # Create per-database salt for new databases (v2 security improvement)
            create_database_salt(db_path)
            logger.info(f"Created per-database salt for {username}")

            # Pre-derive key before closures to avoid capturing plaintext password
            hex_key = get_key_from_password(password, db_path=db_path).hex()

            # Create database structure using raw SQLCipher outside SQLAlchemy
            try:
                conn = create_sqlcipher_connection(
                    db_path,
                    password=password,
                    creation_mode=True,
                    connect_kwargs={
                        # DEFERRED (empty string) so pure-SELECT transactions
                        # acquire only SQLite's SHARED lock, letting WAL-mode
                        # concurrent readers proceed while a writer is active.
                        # IMMEDIATE was previously set "defensively" and made
                        # every transaction (even reads) take a RESERVED lock,
                        # which was the single biggest contention source on
                        # the login-hang path. Race-prone check-then-insert
                        # call sites were made race-free at the application
                        # layer in the preceding prerequisite PR.
                        "isolation_level": "",
                        "check_same_thread": False,
                    },
                )
                _best_effort_chmod(db_path, 0o600, warn=True)
                try:
                    # Get the CREATE TABLE statements from SQLAlchemy models
                    from sqlalchemy.dialects import sqlite
                    from sqlalchemy.schema import CreateIndex, CreateTable

                    from .models import Base

                    # Indexes must be emitted explicitly — SQLAlchemy compiles
                    # `index=True`/`unique=True` and `Index(...)` to separate
                    # CREATE [UNIQUE] INDEX statements, not inline.
                    sqlite_dialect = sqlite.dialect()
                    for table in Base.metadata.sorted_tables:
                        if table.name == "users":
                            continue
                        create_sql = str(
                            CreateTable(table, if_not_exists=True).compile(
                                dialect=sqlite_dialect
                            )
                        )
                        logger.debug(f"Creating table {table.name}")
                        conn.execute(create_sql)
                        for index in table.indexes:
                            index_sql = str(
                                CreateIndex(index, if_not_exists=True).compile(
                                    dialect=sqlite_dialect
                                )
                            )
                            conn.execute(index_sql)

                    conn.commit()
                finally:
                    from ..utilities.resource_utils import safe_close

                    safe_close(conn, "user DB setup connection")

                logger.info(
                    f"Database structure created successfully for {username}"
                )

            except Exception as e:
                # ``password`` is in lexical scope (function parameter) and
                # is passed into ``create_sqlcipher_connection`` /
                # ``set_sqlcipher_key`` above. A traceback rendered with
                # loguru ``diagnose=True`` would dump frame locals — which
                # includes the plaintext SQLCipher master password. Use
                # ``logger.warning`` to drop the traceback chain and
                # redact the password from str(e) for defense in depth.
                # SQLCipher master passwords are unrecoverable (see
                # TRUST.md §5) so the impact of a leak is permanent.
                from ..security.log_sanitizer import redact_secrets

                safe_msg = redact_secrets(str(e), password)
                logger.warning(f"Error creating database structure: {safe_msg}")
                # Cleanup partial DB file on failure
                if db_path.exists():
                    db_path.unlink(missing_ok=True)
                raise

            # Small delay to ensure file is fully written
            import time

            time.sleep(0.1)

            # Now create SQLAlchemy engine using custom connection creator
            def create_engine_connection():
                """Create a properly initialized SQLCipher connection."""
                return create_sqlcipher_connection(
                    db_path,
                    hex_key=hex_key,
                    creation_mode=False,
                    connect_kwargs={
                        # DEFERRED (empty string) so pure-SELECT transactions
                        # acquire only SQLite's SHARED lock, letting WAL-mode
                        # concurrent readers proceed while a writer is active.
                        # IMMEDIATE was previously set "defensively" and made
                        # every transaction (even reads) take a RESERVED lock,
                        # which was the single biggest contention source on
                        # the login-hang path. Race-prone check-then-insert
                        # call sites were made race-free at the application
                        # layer in the preceding prerequisite PR.
                        "isolation_level": "",
                        "check_same_thread": False,
                    },
                )

            # Create engine with custom creator function and optimized cache
            engine = create_engine(
                "sqlite://",
                creator=create_engine_connection,
                poolclass=self._pool_class,
                echo=False,
                query_cache_size=1000,
                **self._get_pool_kwargs(),
            )
        else:
            logger.warning(
                f"SQLCipher not available - creating UNENCRYPTED database for user {username}"
            )
            # Fall back to regular SQLite with query cache
            engine = create_engine(
                f"sqlite:///{db_path}",
                connect_args={"check_same_thread": False, "timeout": 30},
                poolclass=self._pool_class,
                echo=False,
                query_cache_size=1000,
                **self._get_pool_kwargs(),
            )

            # For unencrypted databases, just apply pragmas
            event.listen(engine, "connect", self._apply_pragmas)

        # Tables have already been created using raw SQLCipher above
        # No need to create them again with SQLAlchemy

        # Initialize database tables using centralized initialization
        from .initialize import initialize_database

        # Mirror of the fail-loud change #3635 made to open_user_database.
        # Previously this swallowed the exception with "tables exist but
        # schema version not stamped — migrations will be retried on next
        # process restart". That left a half-broken DB on disk: tables
        # present, no alembic_version row. The next login then re-ran
        # alembic, hit the same error, and (post-#3635) 503'd — so the
        # user could register but never log in again. Better to fail
        # registration loudly with the partial DB removed, so the real
        # cause (e.g. world-writable migrations dir) gets fixed instead
        # of producing a permanently-locked-out account.
        try:
            Session = sessionmaker(bind=engine)
            with Session() as session:
                initialize_database(engine, session)
        except Exception as e:
            # ``password`` is in scope and was passed into the engine
            # creator closure above. Drop the traceback to avoid leaking
            # frame locals under ``diagnose=True`` and redact the
            # password from str(e) defensively.
            from ..security.log_sanitizer import redact_secrets

            safe_msg = redact_secrets(str(e), password)
            logger.warning(
                f"Database migration failed for {username} during creation"
                f" — removing partial DB: {safe_msg}"
            )
            engine.dispose()
            if db_path.exists():
                db_path.unlink(missing_ok=True)
            raise

        # Restrict DB file to owner-only (0o600). The encrypted branch
        # chmod's right after create_sqlcipher_connection, but the
        # unencrypted fallback creates the file lazily on first connect
        # (during initialize_database above), so it is only guaranteed to
        # exist here. This file holds PLAINTEXT user data — leaving it at
        # umask-default perms (commonly 0o644) would expose it to other
        # local accounts.
        if not self.has_encryption and db_path.exists():
            _best_effort_chmod(db_path, 0o600, warn=True)

        # Store connection AFTER migrations complete
        with self._connections_lock:
            self.connections[username] = engine

        logger.info(f"Created encrypted database for user {username}")
        return engine

    def _get_init_lock(self, username: str) -> threading.Lock:
        """Return the per-user cold-open lock, creating it on first use.

        Guards the engine-build + migration in ``open_user_database`` so two
        concurrent first-opens of the same user serialize instead of running
        migrations against one database file simultaneously. Per-user (not the
        global ``_connections_lock``) so different users still open in
        parallel.
        """
        with self._connections_lock:
            lock = self._init_locks.get(username)
            if lock is None:
                lock = threading.Lock()
                self._init_locks[username] = lock
            return lock

    def open_user_database(
        self, username: str, password: str
    ) -> Optional[Engine]:
        """Open an existing encrypted database for a user."""
        open_start = time.perf_counter()

        # Validate the encryption key
        if not self._is_valid_encryption_key(password):
            logger.error(
                f"Invalid encryption key when opening database for user {username}: password is None or empty"
            )
            # TODO: Fix the root cause - research threads are not getting the correct password
            logger.error(
                "TODO: This usually means the research thread is not receiving the user's "
                "password for database encryption. Need to ensure password is passed from "
                "the main thread to research threads."
            )
            raise ValueError(
                "Invalid encryption key: password cannot be None or empty"
            )

        # Check if already open
        with self._connections_lock:
            if username in self.connections:
                return self.connections[username]

        # Serialize the cold-open (engine build + migration) per user so two
        # concurrent first-opens never run alembic against the same database
        # file at once -- alembic's module-level proxy and the version-row
        # UPDATE are not safe under concurrent migration of one DB.
        init_lock = self._get_init_lock(username)
        with init_lock:
            # Re-check: another thread may have completed the cold-open while
            # we were waiting for the per-user lock.
            with self._connections_lock:
                if username in self.connections:
                    return self.connections[username]
            return self._open_user_database_cold(username, password, open_start)

    def _open_user_database_cold(
        self, username: str, password: str, open_start: float
    ) -> Optional[Engine]:
        """Build the engine and run migrations for a not-yet-cached user DB.

        Must be called while holding this user's init lock (acquired in
        ``open_user_database``) so concurrent first-opens of the same user do
        not migrate one database file simultaneously.
        """
        db_path = self._get_user_db_path(username)

        # Prevent timing attacks: always derive key before checking file existence
        # This ensures both existing and non-existent users take the same amount of time,
        # preventing username enumeration via timing analysis.
        # Pre-derive key before closures to avoid capturing plaintext password
        hex_key = get_key_from_password(password, db_path=db_path).hex()

        if not db_path.exists():
            logger.error(f"No database found for user {username}")
            return None

        # Warn if this is a legacy database without per-database salt
        if self.has_encryption and not has_per_database_salt(db_path):
            logger.warning(
                f"Database for user '{username}' uses the legacy shared salt "
                f"(deprecated). For improved security, consider creating a new "
                f"account to get a per-database salt. Legacy databases remain "
                f"fully functional but are less resistant to multi-target attacks."
            )

        # Create connection string - use regular SQLite when SQLCipher not available
        if self.has_encryption:

            def create_open_connection():
                """Create a properly initialized SQLCipher connection."""
                return create_sqlcipher_connection(
                    db_path,
                    hex_key=hex_key,
                    creation_mode=False,
                    connect_kwargs={
                        # DEFERRED (empty string) so pure-SELECT transactions
                        # acquire only SQLite's SHARED lock, letting WAL-mode
                        # concurrent readers proceed while a writer is active.
                        # IMMEDIATE was previously set "defensively" and made
                        # every transaction (even reads) take a RESERVED lock,
                        # which was the single biggest contention source on
                        # the login-hang path. Race-prone check-then-insert
                        # call sites were made race-free at the application
                        # layer in the preceding prerequisite PR.
                        "isolation_level": "",
                        "check_same_thread": False,
                    },
                )

            # Create engine with custom creator function and optimized cache
            engine = create_engine(
                "sqlite://",
                creator=create_open_connection,
                poolclass=self._pool_class,
                echo=False,
                query_cache_size=1000,
                **self._get_pool_kwargs(),
            )
        else:
            logger.warning(
                f"SQLCipher not available - opening UNENCRYPTED database for user {username}"
            )
            # Fall back to regular SQLite (no password protection!)
            engine = create_engine(
                f"sqlite:///{db_path}",
                connect_args={"check_same_thread": False, "timeout": 30},
                poolclass=self._pool_class,
                echo=False,
                query_cache_size=1000,
                **self._get_pool_kwargs(),
            )

            # For unencrypted databases, just apply pragmas
            event.listen(engine, "connect", self._apply_pragmas)

        try:
            # Test connection by running a simple query
            with engine.connect() as conn:
                conn.execute(text("SELECT 1"))

            # Run database initialization (creates missing tables and runs migrations)
            from .initialize import initialize_database

            # Create backup before migration to protect against schema change failures
            from .alembic_runner import needs_migration

            if needs_migration(engine):
                try:
                    from .backup.backup_service import BackupService

                    result = BackupService(
                        username=username, password=password
                    ).create_backup(force=True)
                    if result.success:
                        logger.info(
                            f"Pre-migration backup created: {result.backup_path}"
                        )
                    else:
                        logger.error(
                            f"Pre-migration backup failed: {result.error}"
                        )
                except Exception as e:
                    # ``password`` is in scope and was passed into
                    # ``BackupService`` above. Drop traceback + redact
                    # so the upstream error (which can carry the
                    # password through ``set_sqlcipher_key`` frames) is
                    # not written to log sinks.
                    from ..security.log_sanitizer import redact_secrets

                    safe_msg = redact_secrets(str(e), password)
                    logger.warning(
                        f"Pre-migration backup failed — proceeding with migration: {safe_msg}"
                    )

            # Init failures need to be distinguishable from credential
            # failures at the call site: the credentials worked (we
            # decrypted and ran SELECT 1), but the schema couldn't be
            # brought up. Re-raise as a typed error so the login route
            # can skip the lockout counter and surface a server-error
            # message instead of "Invalid username or password".
            try:
                initialize_database(engine)
            except Exception as init_err:
                elapsed_ms = (time.perf_counter() - open_start) * 1000
                # ``password`` is in scope (function parameter). The
                # initializer ran on an engine whose creator captured
                # ``hex_key`` derived from the password — a SQLAlchemy
                # ``OperationalError`` traceback could surface frames
                # holding the plaintext password under ``diagnose=True``.
                from ..security.log_sanitizer import redact_secrets

                safe_msg = redact_secrets(str(init_err), password)
                logger.warning(
                    f"Database migration failed for {username} "
                    f"after {elapsed_ms:.0f}ms — refusing login: {safe_msg}"
                )
                engine.dispose()
                # Sanitised message + ``from None``: ``init_err`` (its
                # str() and its chained traceback frames) can carry the
                # plaintext password, and the caller that catches this
                # typed error logs it with ``logger.exception``
                # (thread_local_session) — which would render the chain
                # and defeat the redaction applied just above. Break the
                # chain per ADR-0003; the redacted detail is already
                # logged at this site.
                raise DatabaseInitializationError(
                    f"Database initialisation failed for {username}: {safe_msg}"
                ) from None

            # Store connection AFTER migrations complete
            with self._connections_lock:
                self.connections[username] = engine

            elapsed_ms = (time.perf_counter() - open_start) * 1000
            if elapsed_ms > 100:
                logger.info(
                    f"Opened encrypted database for user {username} "
                    f"(cold-open wall clock: {elapsed_ms:.0f}ms)"
                )
            else:
                logger.info(
                    f"Opened encrypted database for user {username} "
                    f"({elapsed_ms:.0f}ms)"
                )
            return engine

        except DatabaseInitializationError:
            # Already logged + engine disposed at the raise site. Re-raise
            # past the catch-all below so callers see the typed error.
            raise
        except Exception as e:
            elapsed_ms = (time.perf_counter() - open_start) * 1000
            # Catches connection errors during ``engine.connect()`` —
            # the connection creator passes the password into
            # ``set_sqlcipher_key`` so a failure there can carry the
            # password in the traceback's frame locals. Drop the
            # traceback and redact str(e).
            from ..security.log_sanitizer import redact_secrets

            safe_msg = redact_secrets(str(e), password)
            logger.warning(
                f"Failed to open database for user {username} "
                f"after {elapsed_ms:.0f}ms: {safe_msg}"
            )
            engine.dispose()
            return None

    def get_session(self, username: str) -> Optional[Session]:
        """Create a new session for a user's database."""
        with self._connections_lock:
            if username not in self.connections:
                # Use debug level for this common scenario to reduce log noise
                logger.debug(f"No open database for user {username}")
                return None
            engine = self.connections[username]
            # Create session inside lock to prevent race with close_user_database()
            SessionLocal = sessionmaker(bind=engine)
            return SessionLocal()

    def get_connected_usernames(self) -> set:
        """Return a snapshot of usernames with open connections."""
        with self._connections_lock:
            return set(self.connections.keys())

    def _checkpoint_wal(self, engine, context: str = ""):
        """Checkpoint WAL before disposing engine to flush pending writes."""
        try:
            with engine.connect() as conn:
                # TRUNCATE returns (busy, log_pages, checkpointed_pages).
                # busy=1 means a reader/writer held a WAL lock and the WAL
                # was NOT truncated — surface that so we can tell from logs
                # whether the helper actually shrank the file.
                result = conn.execute(
                    text("PRAGMA wal_checkpoint(TRUNCATE)")
                ).fetchone()
                if result and result[0] == 1:
                    logger.debug(
                        f"WAL checkpoint busy {context} — WAL not truncated"
                    )
        except Exception:
            logger.debug(f"WAL checkpoint failed {context}", exc_info=True)

    def close_user_database(self, username: str):
        """Close a user's database connection."""
        with self._connections_lock:
            if username in self.connections:
                try:
                    self._checkpoint_wal(
                        self.connections[username], f"for {username}"
                    )
                    self.connections[username].dispose()
                except Exception:
                    logger.warning(
                        f"Failed to dispose engine for {username}",
                    )
                del self.connections[username]
                # Deliberately do NOT pop _init_locks[username] here. A
                # concurrent open_user_database may already hold a reference to
                # this user's lock (fetched via _get_init_lock) and be about to
                # enter its cold-open; dropping the entry would let a later open
                # create a *second* lock for the same user, so two cold-opens
                # could migrate one DB file at once -- the very race this lock
                # exists to prevent. The dict is bounded by the number of
                # distinct usernames (one small Lock each), so retaining it is
                # cheap; close_all_databases clears it wholesale at shutdown.
                logger.info(f"Closed database for user {username}")

    def close_all_databases(self):
        """Close all open user database connections and release file locks."""
        with self._connections_lock:
            for username, engine in list(self.connections.items()):
                try:
                    self._checkpoint_wal(engine, f"for {username}")
                    engine.dispose()
                except Exception:
                    logger.debug(f"Error disposing engine for {username}")
            self.connections.clear()
            self._init_locks.clear()

    def check_database_integrity(self, username: str) -> bool:
        """Check integrity of a user's encrypted database."""
        with self._connections_lock:
            if username not in self.connections:
                return False
            engine = self.connections[username]

        try:
            with engine.connect() as conn:
                # Quick integrity check
                result = conn.execute(text("PRAGMA quick_check"))
                if result.fetchone()[0] != "ok":
                    return False

                # SQLCipher integrity check
                result = conn.execute(text("PRAGMA cipher_integrity_check"))
                # If this returns any rows, there are HMAC failures
                failures = list(result)
                if failures:
                    logger.error(
                        f"Integrity check failed for {username}: {len(failures)} HMAC failures"
                    )
                    return False

                return True

        except Exception as e:
            # No ``password`` parameter on this method and the engine
            # was created via a hex-key closure (the plaintext password
            # is not retained on the engine itself), so the traceback
            # frames in this method do not hold a credential. Still
            # drop the traceback to stay uniform with the rest of the
            # #4182 sweep and to guard against a caller frame that
            # happens to hold one (e.g., a route handler that
            # retrieved the password from the session store).
            from ..security.log_sanitizer import redact_secrets

            safe_msg = redact_secrets(str(e), None)
            logger.warning(
                f"Integrity check error for user: {username}: {safe_msg}"
            )
            return False

    def change_password(
        self, username: str, old_password: str, new_password: str
    ) -> bool:
        """Change the encryption password for a user's database.

        This rekeys the SQLCipher database — no separate auth-DB
        password-hash update is needed because passwords are never
        stored.  Login verification is done by attempting decryption.
        """
        if not self.has_encryption:
            logger.warning(
                "Cannot change password - SQLCipher not available (databases are unencrypted)"
            )
            return False

        db_path = self._get_user_db_path(username)

        if not db_path.exists():
            return False

        try:
            # Close existing connection if any
            self.close_user_database(username)

            # Open with old password
            engine = self.open_user_database(username, old_password)
            if not engine:
                return False

            # Rekey the database (only works with SQLCipher)
            with engine.connect() as conn:
                # Use centralized rekey function
                set_sqlcipher_rekey(conn, new_password, db_path=db_path)

            logger.info(f"Password changed for user {username}")
            return True

        except Exception as e:
            # Both ``old_password`` and ``new_password`` are in lexical
            # scope. ``open_user_database`` / ``set_sqlcipher_rekey``
            # carry these into nested frames — a traceback rendered
            # with ``diagnose=True`` would leak them. Redact both and
            # drop the traceback chain.
            from ..security.log_sanitizer import redact_secrets

            safe_msg = redact_secrets(str(e), old_password, new_password)
            logger.warning(
                f"Failed to change password for user: {username}: {safe_msg}"
            )
            return False
        finally:
            # Close the connection
            self.close_user_database(username)

    def user_exists(self, username: str) -> bool:
        """Check if a user exists in the auth database."""
        from .auth_db import auth_db_session
        from .models.auth import User

        with auth_db_session() as session:
            user = session.query(User).filter_by(username=username).first()
            return user is not None

    def get_memory_usage(self) -> Dict[str, Any]:
        """Get memory usage statistics."""
        with self._connections_lock:
            num_connections = len(self.connections)
        return {
            "active_connections": num_connections,
            "active_sessions": 0,  # Sessions are created on-demand, not tracked
            "estimated_memory_mb": num_connections
            * 3.5,  # ~3.5MB per connection
        }

    def create_thread_safe_session_for_metrics(
        self, username: str, password: str
    ):
        """
        Create a new database session safe for use in background threads.

        Previously this method created a dedicated NullPool engine per
        (username, thread_id) pair, which leaked file descriptors under
        load (SQLCipher + WAL holds 3 FDs per active connection and
        orphaned engines accumulated when @thread_cleanup did not fire).

        It now routes through the shared per-user QueuePool engine at
        ``self.connections[username]``. That engine is already created
        with ``check_same_thread=False`` (so background threads are
        safe), is bounded by ``pool_size + max_overflow``, and is
        subject to the periodic ``dispose()`` workaround in
        ``connection_cleanup.py`` that mitigates the SQLCipher+WAL
        out-of-order-close FD leak.

        Args:
            username: The username
            password: The user's password (encryption key), used only
                to open the user database on cache miss.

        Returns:
            A SQLAlchemy Session bound to the per-user QueuePool engine.
        """
        db_path = self._get_user_db_path(username)

        if not db_path.exists():
            raise ValueError(f"No database found for user {username}")

        with self._connections_lock:
            engine = self.connections.get(username)

        if engine is None:
            # Cache miss — open the user database. This is idempotent:
            # after the first call it just returns the cached engine.
            engine = self.open_user_database(username, password)
            if engine is None:
                raise ValueError(f"Failed to open database for user {username}")

        # Use SQLAlchemy's default expire_on_commit=True.
        Session = sessionmaker(bind=engine)
        return Session()


# Global instance
db_manager = DatabaseManager()
