"""
Automatic cleanup of idle database connections.

Periodically closes database connections for users who have no active sessions
and no active research, preventing resource leaks when users close their browser
without logging out.

Also periodically disposes all QueuePool engines to release accumulated WAL/SHM
file handles. See ADR-0004 for why this is necessary with SQLCipher + WAL mode.
"""

import os
import time
from pathlib import Path

from apscheduler.schedulers.background import BackgroundScheduler
from loguru import logger

from ...database.session_passwords import session_password_store
from ...database.thread_local_session import cleanup_dead_threads
from ...web.routes.globals import get_usernames_with_active_research

# ---------------------------------------------------------------------------
# File Descriptor Monitoring
# ---------------------------------------------------------------------------
# WHY: After days of idle operation in Docker, the app crashed with
#   OSError: [Errno 24] Too many open files
# This monitoring logs the FD count every 5 minutes so we can correlate
# FD growth with specific events and find leaks.
#
# WHAT IT LOGS:
#   - open_fds: total open file descriptors for the process
#   - pool_engines: number of per-user QueuePool engines
#   - pool_checked_out: connections currently checked out from QueuePool
#   - protected_users: users with active sessions
#
# HOW TO USE: grep "Resource monitor" in container logs. If open_fds
# grows steadily over hours, something is leaking.
# ---------------------------------------------------------------------------

# Dispose all pool engines every 30 minutes to release WAL/SHM handles.
# SQLCipher + WAL mode leaks handles when connections close out of order
# (which QueuePool's pool_recycle causes). Periodic dispose() closes ALL
# pooled connections at once, resetting the handle state cleanly.
# The next DB operation transparently reopens a fresh connection.
_DISPOSE_INTERVAL_SECONDS = 1800
_last_dispose_time = 0.0


def _pop_per_user_locks(username: str) -> None:
    """Pop ``username`` from the four module-level per-user lock dicts.

    The library-init, backup, queue-processor, and library-RAG modules
    each maintain a ``dict[..., threading.Lock]`` keyed by username (or
    by ``(username, ...)``) for serialising per-user critical sections.
    None of them had a removal hook, so without this the dicts
    accumulated one entry per username across the process lifetime —
    bounded by total users (~296 bytes/entry × 4 = ~1.2 KB/user/dict)
    but real on long-lived self-hosted instances with churn. Called
    from the user-close paths so the next login starts with fresh
    locks (the locks hold no state worth preserving across
    login/logout).

    Lazy-imported here to keep this module's import graph shallow:
    ``connection_cleanup`` runs at startup and shouldn't pull in the
    queue / backup / library-init / library-RAG modules eagerly.
    """
    try:
        from ...database.library_init import pop_user_init_lock

        pop_user_init_lock(username)
    except Exception:
        # Surface at WARNING to match the sibling scheduler-unregister
        # error handler in this same module (line ~111). A failure
        # here means the lock-dict entry will accumulate on every
        # subsequent close cycle for this user; we want it visible.
        logger.warning(f"Failed to pop _user_init_locks for {username}")

    try:
        from ...database.backup.backup_service import pop_user_lock

        pop_user_lock(username)
    except Exception:
        logger.warning(f"Failed to pop _user_locks for {username}")

    try:
        from ...web.queue.processor_v2 import queue_processor

        queue_processor.pop_user_critical_lock(username)
    except Exception:
        logger.warning(f"Failed to pop _user_critical_locks for {username}")

    try:
        from ...research_library.services.library_rag_service import (
            pop_faiss_locks_for_user,
        )

        pop_faiss_locks_for_user(username)
    except Exception:
        logger.warning(f"Failed to pop _faiss_write_locks for {username}")


def _count_open_fds() -> int:
    """Count open file descriptors for the current process."""
    proc_fd = Path("/proc/self/fd")
    if proc_fd.is_dir():
        try:
            return len(list(proc_fd.iterdir()))
        except OSError:
            pass
    import resource

    soft_limit = resource.getrlimit(resource.RLIMIT_NOFILE)[0]
    count = 0
    for fd in range(soft_limit):
        try:
            os.fstat(fd)
            count += 1
        except OSError:
            pass
    return count


def cleanup_idle_connections(session_manager, db_manager):
    """Close db connections for users with no active sessions and no active research."""
    # 1. Purge expired sessions first
    session_manager.cleanup_expired_sessions()

    # 2. Get protected usernames (active sessions OR active research)
    active_usernames = session_manager.get_active_usernames()
    researching_usernames = get_usernames_with_active_research()
    protected = active_usernames | researching_usernames

    # 3. Get usernames with open connections
    connected_usernames = db_manager.get_connected_usernames()

    # 4. Find idle candidates
    candidates = connected_usernames - protected

    # 5. Double-check before closing (narrows race window)
    closed = 0
    for username in candidates:
        if session_manager.has_active_sessions_for(username):
            logger.debug(
                f"Skipped {username} (active session appeared since snapshot)"
            )
            continue  # User logged in since snapshot
        if username in get_usernames_with_active_research():
            logger.debug(
                f"Skipped {username} (active research appeared since snapshot)"
            )
            continue  # Research started since snapshot
        # Unregister news scheduler jobs (matches logout pattern in routes.py)
        try:
            from ...scheduler.background import (
                get_background_job_scheduler,
            )

            sched = get_background_job_scheduler()
            if sched.is_running:
                sched.unregister_user(username)
        except Exception:
            logger.warning(
                f"Failed to unregister scheduler for {username}",
            )
        try:
            db_manager.close_user_database(username)
            session_password_store.clear_all_for_user(username)
            closed += 1
            logger.debug(f"Closed idle connection for {username}")
        except Exception:
            logger.warning(f"Connection cleanup failed for {username}")
        # Pop lock-dict entries regardless of whether close succeeded.
        # The lock-dict cleanup is independent of DB-engine teardown;
        # putting it inside the try above would skip pop on the very
        # failure path it most matters for (close raises -> next login
        # rebuilds the engine but the stale lock entries linger).
        _pop_per_user_locks(username)

    if closed:
        logger.info(f"Connection cleanup: closed {closed} idle connection(s)")
    logger.debug(
        f"Connection cleanup: evaluated {len(candidates)} candidate(s), "
        f"closed {closed}, protected {len(protected)} active user(s)"
    )

    # Sweep dead-thread sessions and credentials — safety net when neither
    # HTTP requests nor the queue processor are triggering sweeps.
    cleanup_dead_threads()

    # --- Periodic pool dispose to release WAL/SHM handles ---
    # SQLCipher + WAL mode accumulates file handles when QueuePool recycles
    # connections out of open-order (ADR-0004). Periodically calling
    # dispose() on all engines closes ALL pooled connections, releasing any
    # leaked handles. The pool is transparently recreated on the next DB
    # operation.
    #
    # Safe to run against engines with checked-out connections: SA 2.0
    # `QueuePool.dispose` only drains idle queue entries and
    # `Engine.dispose` calls `pool.recreate()`; a thread holding a
    # checked-out connection keeps using it until return. SA docs are
    # explicit — "Connections that are still checked out will not be
    # closed". The post-login bulk write (_perform_post_login_tasks in
    # web/auth/routes.py) is additionally protected by being a single
    # atomic transaction, so any interruption (dispose, crash, OOM)
    # rolls back cleanly without leaving partial state.
    #
    # Do not add a `checkedout() > 0` skip guard here without first
    # reproducing a real torn-write against the actual SA source path:
    # see PR #3487 discussion — the speculative skip introduces an
    # unbounded "skip forever" risk on busy engines in exchange for
    # preventing a failure mode that SA 2.0 does not produce.
    global _last_dispose_time
    now = time.monotonic()
    if now - _last_dispose_time >= _DISPOSE_INTERVAL_SECONDS:
        _last_dispose_time = now
        disposed = 0
        with db_manager._connections_lock:
            for username, engine in list(db_manager.connections.items()):
                try:
                    db_manager._checkpoint_wal(engine, f"for {username}")
                    engine.dispose()
                    disposed += 1
                except Exception as exc:
                    # Surface the failure. Pre-fix this was logger.debug,
                    # which hid the symptom — if WAL checkpoint or pool
                    # dispose repeatedly fails (disk pressure, lock
                    # starvation, etc.) the WAL file silently grows on
                    # disk and pooled connections leak. The 30-min
                    # periodic-dispose workaround for ADR-0004's WAL/SHM
                    # handle leak depends on this loop succeeding.
                    #
                    # Only the exception's TYPE NAME is logged, matching
                    # the codebase's `_report_silent_exception` pattern
                    # (utilities/log_utils.py:146-194). The exception
                    # value itself can carry sensitive locals (DB paths,
                    # query fragments, etc.) and our sensitive-logging
                    # hook flags any `f"...{exc}"` interpolation.
                    exc_type = type(exc).__name__
                    logger.warning(
                        f"Error disposing engine for {username}: {exc_type}"
                    )
        if disposed:
            logger.info(
                f"Pool dispose: reset {disposed} engine(s) to release "
                f"WAL/SHM handles"
            )

    # --- FD monitoring ---
    try:
        fd_count = _count_open_fds()
        pool_engine_count = len(db_manager.connections)
        pool_checked_out = 0
        with db_manager._connections_lock:
            for engine in db_manager.connections.values():
                try:
                    pool_checked_out += engine.pool.checkedout()
                except Exception:  # noqa: silent-exception
                    pass
        logger.debug(
            f"Resource monitor: open_fds={fd_count}, "
            f"pool_engines={pool_engine_count}, "
            f"pool_checked_out={pool_checked_out}, "
            f"protected_users={len(protected)}"
        )
        if fd_count > 800:
            logger.warning(
                f"High FD count ({fd_count}) — approaching system limit. "
                f"Check for resource leaks."
            )
    except Exception:
        logger.debug("FD monitoring failed")  # noqa: silent-exception


def start_connection_cleanup_scheduler(
    session_manager, db_manager, interval_seconds=300
):
    """Start APScheduler job for periodic connection cleanup.

    Args:
        session_manager: The SessionManager singleton.
        db_manager: The DatabaseManager singleton.
        interval_seconds: How often to run cleanup (default: 5 minutes).

    Returns:
        The BackgroundScheduler instance (for shutdown registration).
    """
    scheduler = BackgroundScheduler()
    scheduler.add_job(
        cleanup_idle_connections,
        "interval",
        seconds=interval_seconds,
        args=[session_manager, db_manager],
        id="cleanup_idle_connections",
        jitter=30,
    )
    scheduler.start()
    logger.info(
        f"Connection cleanup scheduler started "
        f"(interval={interval_seconds}s, jitter=30s)"
    )
    return scheduler
