"""
Queue processor v2 - uses encrypted user databases instead of service.db
Supports both direct execution and queue modes.
"""

import threading
import time
import uuid
from typing import Any, Dict, Optional

from loguru import logger
from sqlalchemy.orm import Session

from ...constants import ResearchStatus
from ...database.encrypted_db import db_manager
from ...database.models import (
    QueuedResearch,
    ResearchHistory,
    UserActiveResearch,
)
from ...database.queue_service import UserQueueService
from ...database.session_context import get_user_db_session
from ...database.session_passwords import session_password_store
from ...exceptions import DuplicateResearchError, SystemAtCapacityError
from ...security.log_sanitizer import redact_secrets
from ...notifications.queue_helpers import (
    send_research_completed_notification_from_session,
    send_research_failed_notification_from_session,
)
from ..services.research_service import (
    run_research_process,
    start_research_process,
)

# Retry configuration constants for notification database queries
MAX_RESEARCH_LOOKUP_RETRIES = 3
INITIAL_RESEARCH_LOOKUP_DELAY = 0.5  # seconds
RETRY_BACKOFF_MULTIPLIER = 2

# Give up on a queued research after this many consecutive spawn failures.
# Each failure leaves is_processing=False so the next loop tick retries.
SPAWN_RETRY_LIMIT = 3


class QueueProcessorV2:
    """
    Processes queued researches using encrypted user databases.
    This replaces the service.db approach.
    """

    def __init__(self, check_interval=10):
        """
        Initialize the queue processor.

        Args:
            check_interval: How often to check for work (seconds)
        """
        self.check_interval = check_interval
        self.running = False
        self.thread = None
        self._loop_iteration = 0
        # Wakes the loop out of its inter-iteration wait so stop() returns
        # in milliseconds instead of blocking for up to check_interval.
        # The test suite stops the processor after every app test
        # (tests/conftest.py reset_singletons), so a blocking stop() adds
        # ~check_interval seconds of teardown to each of those tests.
        self._stop_event = threading.Event()

        # Per-user settings will be retrieved from each user's database
        # when processing their queue using SettingsManager
        logger.info(
            "Queue processor v2 initialized - will use per-user settings from SettingsManager"
        )

        # Track which users we should check
        self._users_to_check: set[tuple[str, str]] = set()
        self._users_lock = threading.Lock()

        # Track pending operations from background threads
        self.pending_operations = {}
        self._pending_operations_lock = threading.Lock()

        # Per-user serialisation for the "count active → start direct"
        # critical section. Without this the count-then-insert races with
        # itself for the same user (e.g. two concurrent research submissions
        # from two browser tabs), and IMMEDIATE isolation was what used to
        # paper over it at the DB layer.
        self._user_critical_locks: Dict[str, threading.Lock] = {}
        self._user_critical_locks_lock = threading.Lock()

        # Count consecutive spawn failures per research_id. Entries are
        # popped on success or after hitting SPAWN_RETRY_LIMIT (then the
        # research is marked FAILED). In-memory is sufficient: a restart
        # resets the counter and the research gets a fresh N retries,
        # which is the desired behavior if the underlying system issue
        # (thread pool, memory) cleared.
        # Access is guarded by _spawn_retry_counts_lock because the
        # increment path is a read-modify-write and the loop and direct
        # request paths can interleave.
        self._spawn_retry_counts: dict[str, int] = {}
        self._spawn_retry_counts_lock = threading.Lock()

    def _get_user_critical_lock(self, username: str) -> threading.Lock:
        """Get (or lazily create) the per-user lock used to serialise the
        count-active-and-start-direct critical section for a given user.
        """
        with self._user_critical_locks_lock:
            lock = self._user_critical_locks.get(username)
            if lock is None:
                lock = threading.Lock()
                self._user_critical_locks[username] = lock
            return lock

    def pop_user_critical_lock(self, username: str) -> None:
        """Remove the per-user critical-section lock for ``username``.

        Called from the user-close path so this instance dict doesn't
        accumulate one entry per username across the process lifetime.
        The next direct-research submission for that user lazily
        re-creates the lock if needed — the lock has no state that
        needs to persist across login/logout.
        """
        with self._user_critical_locks_lock:
            self._user_critical_locks.pop(username, None)

    def _bump_spawn_retry_count(self, research_id: str) -> int:
        """Atomically increment and return the spawn-retry counter for
        ``research_id``. Extracted so tests can exercise the real
        locked increment path instead of duplicating the lock in the
        test worker (which would be a tautology).
        """
        with self._spawn_retry_counts_lock:
            attempts = self._spawn_retry_counts.get(research_id, 0) + 1
            self._spawn_retry_counts[research_id] = attempts
            return attempts

    @staticmethod
    def _commit_with_safe_rollback(db_session: Session, context: str) -> bool:
        """Commit ``db_session`` with best-effort rollback on failure.

        Returns ``True`` on success, ``False`` if the commit raised.
        The failure path logs via ``logger.exception`` and attempts a
        rollback, itself guarded so a subsequent rollback failure is
        logged at debug level rather than propagated.

        Extracted because the ``try: commit / except: log + try:
        rollback`` idiom repeats at ≥5 sites in this module; inlining
        hides the defensive structure behind nested ``try`` blocks and
        makes each callsite longer than the work it describes.
        """
        try:
            db_session.commit()
            return True
        except Exception:
            logger.exception(f"Commit failed: {context}")
            try:
                db_session.rollback()
            except Exception:
                logger.debug(
                    f"Rollback after commit failure ({context})",
                    exc_info=True,
                )
            return False

    def _delete_queue_row_safely(
        self, db_session: Session, username: str, research_id: str
    ) -> None:
        """Best-effort delete of the ``QueuedResearch`` row for
        ``(username, research_id)``.

        Rolls back any pending state first (the session may be in
        ``PendingRollbackError`` from a failed commit inside
        ``_start_research``), re-queries the row fresh, deletes it if
        present, and commits via ``_commit_with_safe_rollback``.

        Use this for ``DuplicateResearchError`` cleanup where the goal
        is "drop the queue row regardless of session state." Do NOT
        use it for paths that need the delete to be atomic with other
        writes (e.g. the terminal FAILED path bundles
        ``ResearchHistory.status = FAILED`` with the queue-row delete
        in a single commit — that stays inline).
        """
        try:
            db_session.rollback()
        except Exception:
            logger.debug(
                f"Rollback before queue-row delete for {research_id}",
                exc_info=True,
            )
        try:
            fresh_queued = (
                db_session.query(QueuedResearch)
                .filter_by(username=username, research_id=research_id)
                .first()
            )
            if fresh_queued:
                db_session.delete(fresh_queued)
            self._commit_with_safe_rollback(
                db_session,
                f"queue-row delete for research {research_id}",
            )
        except Exception:
            logger.exception(
                f"Failed to query/delete queue row for {research_id}"
            )
            try:
                db_session.rollback()
            except Exception:
                logger.debug(
                    f"Rollback after queue-row delete failure for {research_id}",
                    exc_info=True,
                )

    def start(self):
        """Start the queue processor thread."""
        if self.running:
            logger.warning("Queue processor already running")
            return

        self.running = True
        # Re-arm the wait for restart: stop() leaves the event set, and
        # create_app() restarts this singleton after the test-suite
        # teardown stops it.
        self._stop_event.clear()
        self.thread = threading.Thread(
            target=self._process_queue_loop, daemon=True
        )
        self.thread.start()
        logger.info("Queue processor v2 started")

    def stop(self):
        """Stop the queue processor thread."""
        self.running = False
        self._stop_event.set()
        if self.thread:
            self.thread.join(timeout=10)
        logger.info("Queue processor v2 stopped")

    def notify_user_activity(self, username: str, session_id: str):
        """
        Notify that a user has activity and their queue should be checked.

        Args:
            username: The username
            session_id: The Flask session ID (for password access)
        """
        with self._users_lock:
            self._users_to_check.add((username, session_id))
            logger.debug(f"User {username} added to queue check list")

    def notify_research_queued(self, username: str, research_id: str, **kwargs):
        """
        Notify that a research was queued.
        In direct mode, this immediately starts the research if slots are available.
        In queue mode, it adds to the queue.

        Args:
            username: The username
            research_id: The research ID
            **kwargs: Additional parameters for direct execution (query, mode, etc.)
        """
        # Pre-declared so the except handlers below can pass it to
        # redact_secrets even on paths where it is never assigned.
        password = None
        # Check user's queue_mode setting when we have database access
        if kwargs:
            session_id = kwargs.get("session_id")
            if session_id:
                # Check if we can start it directly
                password = session_password_store.get_session_password(
                    username, session_id
                )
                if password:
                    try:
                        # Open database and check settings + active count
                        engine = db_manager.open_user_database(
                            username, password
                        )
                        if engine:
                            with get_user_db_session(username) as db_session:
                                # Get user's settings using SettingsManager
                                from ...settings.manager import SettingsManager

                                settings_manager = SettingsManager(db_session)

                                # Get user's queue_mode setting (env > DB > default)
                                queue_mode = settings_manager.get_setting(
                                    "app.queue_mode", "direct"
                                )

                                # Get user's max concurrent setting (env > DB > default)
                                max_concurrent = settings_manager.get_setting(
                                    "app.max_concurrent_researches", 3
                                )

                                logger.debug(
                                    f"User {username} settings: queue_mode={queue_mode}, "
                                    f"max_concurrent={max_concurrent}"
                                )

                                # Only try direct execution if user has queue_mode="direct"
                                if queue_mode == "direct":
                                    # Serialise the count→check→start critical
                                    # section at the application layer. Two
                                    # concurrent submissions for the same user
                                    # must not both observe the same active
                                    # count and both start — that would exceed
                                    # max_concurrent. A per-user Python lock
                                    # gives us that atomicity independent of
                                    # the DB isolation level.
                                    with self._get_user_critical_lock(username):
                                        active_count = (
                                            db_session.query(UserActiveResearch)
                                            .filter_by(
                                                username=username,
                                                status=ResearchStatus.IN_PROGRESS,
                                            )
                                            .count()
                                        )

                                        if active_count < max_concurrent:
                                            # We have slots - start directly!
                                            logger.info(
                                                f"Direct mode: Starting research {research_id} immediately "
                                                f"(active: {active_count}/{max_concurrent})"
                                            )

                                            # Start the research directly
                                            self._start_research_directly(
                                                username,
                                                research_id,
                                                password,
                                                **kwargs,
                                            )
                                            return
                                    logger.info(
                                        f"Direct mode: Max concurrent reached ({active_count}/"
                                        f"{max_concurrent}), queueing {research_id}"
                                    )
                                else:
                                    logger.info(
                                        f"User {username} has queue_mode={queue_mode}, "
                                        f"queueing research {research_id}"
                                    )
                    except Exception as e:
                        # ``password`` is in scope — drop the traceback
                        # chain and redact str(e) so the SQLCipher master
                        # password can't leak via diagnose=True frame
                        # locals (see the generic handler in
                        # _start_research_directly for the full rationale).
                        safe_msg = redact_secrets(str(e), password)
                        logger.warning(
                            f"Error in direct execution for {username}: {safe_msg}"
                        )

        # Fall back to queue mode (or if direct mode failed)
        try:
            with get_user_db_session(username) as session:
                queue_service = UserQueueService(session)
                queue_service.add_task_metadata(
                    task_id=research_id,
                    task_type="research",
                    priority=0,
                )
                logger.info(
                    f"Research {research_id} queued for user {username}"
                )
        except Exception as e:
            # ``password`` may be bound above — same redaction rationale.
            safe_msg = redact_secrets(str(e), password)
            logger.warning(
                f"Failed to update queue status for {username}: {safe_msg}"
            )

    def _start_research_directly(
        self, username: str, research_id: str, password: str, **kwargs
    ):
        """
        Start a research directly without queueing.

        Args:
            username: The username
            research_id: The research ID
            password: The user's password
            **kwargs: Research parameters (query, mode, settings, etc.)
        """
        query = kwargs.get("query")
        mode = kwargs.get("mode")
        settings_snapshot = kwargs.get("settings_snapshot", {})

        # Create active research record
        try:
            with get_user_db_session(username) as db_session:
                active_record = UserActiveResearch(
                    username=username,
                    research_id=research_id,
                    status=ResearchStatus.IN_PROGRESS,
                    thread_id="pending",
                    settings_snapshot=settings_snapshot,
                )
                db_session.add(active_record)
                db_session.commit()

                # Update task status if it exists
                queue_service = UserQueueService(db_session)
                queue_service.update_task_status(research_id, "processing")
        except Exception as e:
            # ``password`` is a parameter of this method — drop the
            # traceback chain and redact str(e) (full rationale at the
            # generic handler below).
            safe_msg = redact_secrets(str(e), password)
            logger.warning(
                f"Failed to create active research record for {research_id}: {safe_msg}"
            )
            return

        # Extract parameters from kwargs
        model_provider = kwargs.get("model_provider")
        model = kwargs.get("model")
        custom_endpoint = kwargs.get("custom_endpoint")
        search_engine = kwargs.get("search_engine")

        # Start the research process
        try:
            research_thread = start_research_process(
                research_id,
                query,
                mode,
                run_research_process,
                username=username,
                user_password=password,
                model_provider=model_provider,
                model=model,
                custom_endpoint=custom_endpoint,
                search_engine=search_engine,
                max_results=kwargs.get("max_results"),
                time_period=kwargs.get("time_period"),
                iterations=kwargs.get("iterations"),
                questions_per_iteration=kwargs.get("questions_per_iteration"),
                strategy=kwargs.get("strategy", "source-based"),
                settings_snapshot=settings_snapshot,
            )

            # Update thread ID
            try:
                with get_user_db_session(username) as db_session:
                    active_record = (
                        db_session.query(UserActiveResearch)
                        .filter_by(username=username, research_id=research_id)
                        .first()
                    )
                    if active_record:
                        active_record.thread_id = str(research_thread.ident)
                        db_session.commit()
            except Exception as e:
                # ``password`` is in scope — same redaction rationale.
                safe_msg = redact_secrets(str(e), password)
                logger.warning(
                    f"Failed to update thread ID for {research_id}: {safe_msg}"
                )

            logger.info(
                f"Direct execution: Started research {research_id} for user {username} "
                f"in thread {research_thread.ident}"
            )

        except DuplicateResearchError:
            # A live thread already owns this research_id. Do NOT delete
            # the UserActiveResearch row or mark ResearchHistory FAILED —
            # that state belongs to the live thread, and mutating it
            # would terminate a running research from the user's
            # perspective while it keeps executing. Same contract as the
            # queue processor's dedicated dup branch (#3506).
            logger.warning(
                f"Duplicate live thread detected for {research_id} "
                "in direct mode; leaving state intact"
            )
            return
        except SystemAtCapacityError:
            # System at concurrent-research capacity in the direct-execution
            # path. Roll back the IN_PROGRESS active row and mark history
            # back to QUEUED so the queue processor can pick it up later.
            logger.info(
                f"Direct execution hit capacity for {research_id}; re-queueing"
            )
            try:
                with get_user_db_session(username) as db_session:
                    active_record = (
                        db_session.query(UserActiveResearch)
                        .filter_by(username=username, research_id=research_id)
                        .first()
                    )
                    if active_record:
                        db_session.delete(active_record)
                    research_row = (
                        db_session.query(ResearchHistory)
                        .filter_by(id=research_id)
                        .first()
                    )
                    if research_row:
                        research_row.status = ResearchStatus.QUEUED
                    # Bump queued_tasks so _process_user_queue's
                    # `queued_tasks == 0` gate doesn't treat the queue as
                    # empty and strand the QueuedResearch row the submit
                    # path already created. The direct path returns before
                    # the normal add_task_metadata call, so this is the
                    # single, non-double-counting increment;
                    # _start_queued_researches later dispatches the row and
                    # update_task_status() transitions this TaskMetadata
                    # queued->processing (balancing the counter).
                    UserQueueService(db_session).add_task_metadata(
                        task_id=research_id,
                        task_type="research",
                        priority=0,
                    )
                    db_session.commit()
            except Exception as e:
                # ``password`` is in scope — same redaction rationale.
                safe_msg = redact_secrets(str(e), password)
                logger.warning(
                    f"Cleanup after capacity reject failed for "
                    f"{research_id}; the stale UserActiveResearch row is "
                    f"recovered by reclaim_stale_user_active_research: {safe_msg}"
                )
            return
        except Exception as e:
            # ``password`` is in lexical scope (function parameter,
            # passed through to ``start_research_process``). A
            # SQLAlchemy / requests exception from anywhere in
            # ``start_research_process`` could carry frame locals that
            # include the SQLCipher master password (which is
            # unrecoverable — see TRUST.md §5). Drop the traceback chain
            # and redact str(e) defensively.
            safe_msg = redact_secrets(str(e), password)
            logger.warning(
                f"Failed to start research {research_id} directly: {safe_msg}"
            )
            # Clean up the active record AND mark the research terminal
            # FAILED so the user-visible state matches reality (no running
            # thread, not IN_PROGRESS). Same contract as the queue
            # processor's terminal-failure branch (#3481).
            try:
                with get_user_db_session(username) as db_session:
                    active_record = (
                        db_session.query(UserActiveResearch)
                        .filter_by(username=username, research_id=research_id)
                        .first()
                    )
                    if active_record:
                        db_session.delete(active_record)
                    research_row = (
                        db_session.query(ResearchHistory)
                        .filter_by(id=research_id)
                        .first()
                    )
                    if research_row:
                        research_row.status = ResearchStatus.FAILED
                    db_session.commit()
            except Exception as e2:
                # ``password`` is in scope — same redaction rationale.
                safe_msg = redact_secrets(str(e2), password)
                logger.warning(
                    f"Failed to clean up active research record for {research_id}: {safe_msg}"
                )

    def notify_research_completed(
        self, username: str, research_id: str, user_password: str | None = None
    ):
        """
        Notify that a research completed.
        Updates the user's queue status in their database.

        Args:
            username: The username
            research_id: The research ID
            user_password: User password for database access. Required for queue
                          updates and database lookups during notification sending.
                          Optional only because some callers may not have it
                          available, in which case only basic updates occur.
        """
        try:
            # get_user_db_session is already imported at module level (line 19)
            # It accepts optional password parameter and returns a context manager
            with get_user_db_session(username, user_password) as session:
                queue_service = UserQueueService(session)
                queue_service.update_task_status(
                    research_id, ResearchStatus.COMPLETED
                )
                logger.info(
                    f"Research {research_id} completed for user {username}"
                )

                # Send notification using helper from notification module
                send_research_completed_notification_from_session(
                    username=username,
                    research_id=research_id,
                    db_session=session,
                )

        except Exception as e:
            # ``user_password`` is a parameter of this method — drop the
            # traceback chain and redact str(e) so the SQLCipher master
            # password can't leak via diagnose=True frame locals.
            safe_msg = redact_secrets(str(e), user_password)
            logger.warning(
                f"Failed to update completion status for {username}: {safe_msg}"
            )

        # Auto-convert research to document in History collection.
        # Documents only — FAISS indexing is triggered separately by the user
        # via "Index All" on the History page.
        from ...research_library.search.services.research_history_indexer import (
            auto_convert_research,
        )

        auto_convert_research(username, research_id, db_password=user_password)

    def notify_research_failed(
        self,
        username: str,
        research_id: str,
        error_message: str | None = None,
        user_password: str | None = None,
    ):
        """
        Notify that a research failed.
        Updates the user's queue status in their database and sends notification.

        Args:
            username: The username
            research_id: The research ID
            error_message: Optional error message
            user_password: User password for database access. Required for queue
                          updates and database lookups during notification sending.
                          Optional only because some callers may not have it
                          available, in which case only basic updates occur.
        """
        try:
            # get_user_db_session is already imported at module level (line 19)
            # It accepts optional password parameter and returns a context manager
            with get_user_db_session(username, user_password) as session:
                queue_service = UserQueueService(session)
                queue_service.update_task_status(
                    research_id,
                    ResearchStatus.FAILED,
                    error_message=error_message,
                )
                logger.info(
                    f"Research {research_id} failed for user {username}: "
                    f"{error_message}"
                )

                # Send notification using helper from notification module
                send_research_failed_notification_from_session(
                    username=username,
                    research_id=research_id,
                    error_message=error_message or "Unknown error",
                    db_session=session,
                )

        except Exception as e:
            # ``user_password`` is a parameter of this method — same
            # redaction rationale as notify_research_completed.
            safe_msg = redact_secrets(str(e), user_password)
            logger.warning(
                f"Failed to update failure status for {username}: {safe_msg}"
            )

    def _process_queue_loop(self):
        """Main loop that processes the queue."""
        while self.running:
            try:
                # Get list of users to check (don't clear immediately)
                with self._users_lock:
                    users_to_check = list(self._users_to_check)

                # Process each user's queue
                users_to_remove = []
                for user_session in users_to_check:
                    try:
                        username, session_id = user_session
                        # _process_user_queue returns True if queue is empty
                        queue_empty = self._process_user_queue(
                            username, session_id
                        )
                        if queue_empty:
                            users_to_remove.append(user_session)
                    except Exception:
                        logger.exception(
                            f"Error processing queue for {user_session}"
                        )
                        # Don't remove on error - the _process_user_queue method
                        # determines whether to keep checking based on error type

                # Only remove users whose queues are now empty
                with self._users_lock:
                    for user_session in users_to_remove:
                        self._users_to_check.discard(user_session)

            except Exception:
                logger.exception("Error in queue processor loop")
            finally:
                # Clean up thread-local database session after each iteration.
                # The loop opens a new session each iteration via get_user_db_session();
                # closing it returns the connection to the shared QueuePool promptly.
                try:
                    from ...database.thread_local_session import (
                        cleanup_current_thread,
                        cleanup_dead_threads,
                    )

                    cleanup_current_thread()
                except Exception:
                    logger.debug(
                        "thread-local cleanup on shutdown", exc_info=True
                    )

                # Periodic dead-thread credential sweep (every ~60s).
                # One of three sweep trigger points (app_factory
                # teardown, connection_cleanup scheduler, and here).
                self._loop_iteration += 1
                if self._loop_iteration % 6 == 0:  # Every ~60s (10s × 6)
                    try:
                        cleanup_dead_threads()
                    except Exception:
                        logger.debug(
                            "periodic dead-thread sweep", exc_info=True
                        )

            # Event.wait, not time.sleep: stop() must be able to interrupt
            # this pause, otherwise shutdown blocks for up to
            # check_interval seconds.
            self._stop_event.wait(self.check_interval)

    def _process_user_queue(self, username: str, session_id: str) -> bool:
        """
        Process the queue for a specific user.

        Args:
            username: The username
            session_id: The Flask session ID

        Returns:
            True if the queue is empty, False if there are still items
        """
        # Get the user's password from session store
        password = session_password_store.get_session_password(
            username, session_id
        )
        if not password:
            logger.debug(
                f"No password available for user {username}, skipping queue check"
            )
            return True  # Remove from checking - session expired

        # Open the user's encrypted database
        try:
            # First ensure the database is open
            engine = db_manager.open_user_database(username, password)
            if not engine:
                logger.error(f"Failed to open database for user {username}")
                return False  # Keep checking - could be temporary DB issue

            # Get a session and process the queue
            with get_user_db_session(username, password) as db_session:
                queue_service = UserQueueService(db_session)

                # Get user's settings using SettingsManager
                from ...settings.manager import SettingsManager

                settings_manager = SettingsManager(db_session)

                # Get user's max concurrent setting (env > DB > default)
                max_concurrent = settings_manager.get_setting(
                    "app.max_concurrent_researches", 3
                )

                # Get queue status
                queue_status = queue_service.get_queue_status() or {
                    "active_tasks": 0,
                    "queued_tasks": 0,
                }

                # Calculate available slots
                available_slots = max_concurrent - queue_status["active_tasks"]

                if available_slots <= 0:
                    # No slots available, but queue might not be empty
                    return False  # Keep checking

                if queue_status["queued_tasks"] == 0:
                    # Queue is empty
                    return True  # Remove from checking

                logger.info(
                    f"Processing queue for {username}: "
                    f"{queue_status['active_tasks']} active, "
                    f"{queue_status['queued_tasks']} queued, "
                    f"{available_slots} slots available"
                )

                # Process queued researches
                self._start_queued_researches(
                    db_session,
                    queue_service,
                    username,
                    password,
                    available_slots,
                )

                # Check if there are still items in queue
                updated_status = queue_service.get_queue_status() or {
                    "queued_tasks": 0
                }
                return bool(updated_status["queued_tasks"] == 0)

        except Exception as e:
            # ``password`` (from the session store above) is in scope —
            # drop the traceback chain and redact str(e).
            safe_msg = redact_secrets(str(e), password)
            logger.warning(
                f"Error processing queue for user {username}: {safe_msg}"
            )
            return False  # Keep checking - errors might be temporary

    def _reclaim_stranded_queue_rows(
        self, db_session: Session, username: str
    ) -> int:
        """Reclaim queue rows stranded by a crash or restart.

        A row is stranded when ``is_processing=True`` but no live thread
        exists in ``_active_research`` for its ``research_id``. This can
        happen after a crash/restart between the pre-spawn IN_PROGRESS
        commit and the queue-row deletion in ``_start_queued_researches``
        — the row is invisible to the normal ``is_processing=False``
        query and would never be retried.

        Reverts ``QueuedResearch.is_processing`` to False and — if
        ``ResearchHistory.status`` is still IN_PROGRESS with no live
        thread — reverts that to QUEUED so the next tick can freshly
        spawn. Returns the number of rows reclaimed.
        """
        from ..routes.globals import is_research_active

        stranded = (
            db_session.query(QueuedResearch)
            .filter_by(username=username, is_processing=True)
            .all()
        )
        reclaimed = 0
        for row in stranded:
            if is_research_active(row.research_id):
                # A legitimate in-flight claim; don't touch.
                continue
            row.is_processing = False
            research = (
                db_session.query(ResearchHistory)
                .filter_by(id=row.research_id)
                .first()
            )
            status_changed = (
                research is not None
                and research.status == ResearchStatus.IN_PROGRESS
            )
            if status_changed:
                research.status = ResearchStatus.QUEUED
            reclaimed += 1
            logger.warning(
                f"Reclaimed stranded queue row for research "
                f"{row.research_id} (user {username}): no live thread, "
                "resetting is_processing=False"
                + (" and status=QUEUED" if status_changed else "")
            )
        if reclaimed:
            if not self._commit_with_safe_rollback(
                db_session,
                f"reclaim of stranded rows for user {username}",
            ):
                return 0
        return reclaimed

    def _start_queued_researches(
        self,
        db_session: Session,
        queue_service: UserQueueService,
        username: str,
        password: str,
        available_slots: int,
    ):
        """Start queued researches up to available slots."""
        # Before picking work, reclaim any rows stranded by a prior
        # crash — otherwise they are invisible to the is_processing=False
        # filter below and would never retry.
        self._reclaim_stranded_queue_rows(db_session, username)

        # Get queued researches
        queued = (
            db_session.query(QueuedResearch)
            .filter_by(username=username, is_processing=False)
            .order_by(QueuedResearch.position)
            .limit(available_slots)
            .all()
        )

        for queued_research in queued:
            research_id = queued_research.research_id
            try:
                # Atomically claim this item by flipping is_processing from
                # False to True in a single UPDATE. If another worker has
                # already claimed it since our SELECT above, the UPDATE will
                # match zero rows and we skip. Under non-IMMEDIATE isolation
                # the previous SELECT+assign pattern would race and two
                # workers could both process the same queued item.
                claimed = (
                    db_session.query(QueuedResearch)
                    .filter_by(
                        id=queued_research.id,
                        is_processing=False,
                    )
                    .update(
                        {QueuedResearch.is_processing: True},
                        synchronize_session=False,
                    )
                )
                db_session.commit()
                if not claimed:
                    logger.debug(
                        f"Queued research {research_id} "
                        f"already claimed by another worker; skipping"
                    )
                    continue
                # Refresh local object state now that we hold the claim
                db_session.refresh(queued_research)

                # Update task status
                queue_service.update_task_status(research_id, "processing")

                # Start the research
                self._start_research(
                    db_session,
                    username,
                    password,
                    queued_research,
                )

                # Success — clear any prior spawn-failure count and
                # remove the queue row.
                with self._spawn_retry_counts_lock:
                    self._spawn_retry_counts.pop(research_id, None)
                db_session.delete(queued_research)
                db_session.commit()

                logger.info(
                    f"Started queued research {research_id} for user {username}"
                )

            except DuplicateResearchError:
                # Raised by _start_research when a prior attempt's thread
                # is still live, OR when the ResearchHistory row is in a
                # non-QUEUED state (IN_PROGRESS from a prior attempt's
                # successful pre-spawn commit; terminal COMPLETED /
                # FAILED / SUSPENDED from a thread that already finished
                # and cleaned up). In every case the correct behavior is
                # the same: clear the stale queue row and the retry
                # counter, and do NOT fall through to the FAILED/notify
                # path — that would terminate-status a live thread or
                # emit a false failure for a completed one.
                logger.warning(
                    f"Research {research_id} is already started "
                    "(live thread or non-QUEUED status); clearing stale "
                    "queue row"
                )
                with self._spawn_retry_counts_lock:
                    self._spawn_retry_counts.pop(research_id, None)
                self._delete_queue_row_safely(db_session, username, research_id)
                continue

            except SystemAtCapacityError:
                # System hit the global concurrent-research capacity while
                # dispatching this queued item. _start_research already
                # reset the ResearchHistory row back to QUEUED before
                # re-raising. This is a transient condition, NOT a spawn
                # failure, so it must NOT count toward SPAWN_RETRY_LIMIT —
                # otherwise a busy system would wrongly mark a perfectly
                # valid queued research FAILED after a few ticks. Just
                # release the processing claim so the next tick retries.
                # Mirrors the dedicated handler in _start_research_directly.
                logger.info(
                    f"System at capacity dispatching queued research "
                    f"{research_id}; leaving queued for next tick"
                )
                # Revert the queued->processing claim from
                # update_task_status("processing") above. The research stays
                # queued for the next tick, so its slot must return to
                # queued_tasks rather than leaking into active_tasks on
                # every capacity-rejected retry.
                queue_service.update_task_status(research_id, "queued")
                fresh_queued = (
                    db_session.query(QueuedResearch)
                    .filter_by(username=username, research_id=research_id)
                    .first()
                )
                if fresh_queued:
                    fresh_queued.is_processing = False
                    self._commit_with_safe_rollback(
                        db_session,
                        "is_processing reset after capacity reject for "
                        f"research {research_id}",
                    )
                continue

            except Exception as e:
                # ``password`` is a parameter of this method — drop the
                # traceback chain and redact str(e).
                safe_msg = redact_secrets(str(e), password)
                logger.warning(
                    f"Error starting queued research {research_id}: {safe_msg}"
                )
                # Session may be in PendingRollbackError state after a
                # failed commit inside _start_research.
                try:
                    db_session.rollback()
                except Exception as rb_err:
                    # No exc_info: ``password`` is in this frame and a
                    # rendered traceback could expose it via
                    # diagnose=True frame locals.
                    logger.debug(
                        "Rollback after start failure: "
                        f"{redact_secrets(str(rb_err), password)}"
                    )

                attempts = self._bump_spawn_retry_count(research_id)

                # Re-query in case rollback expired the ORM object.
                fresh_queued = (
                    db_session.query(QueuedResearch)
                    .filter_by(username=username, research_id=research_id)
                    .first()
                )

                if attempts < SPAWN_RETRY_LIMIT:
                    # Transient failure — allow the next loop tick to
                    # retry. _start_research rolls back its own
                    # IN_PROGRESS write on spawn failure, so the only
                    # fix-up needed here is resetting is_processing.
                    logger.warning(
                        f"Spawn failed for research {research_id} "
                        f"(attempt {attempts}/{SPAWN_RETRY_LIMIT}), "
                        "leaving queued for retry"
                    )
                    if fresh_queued:
                        fresh_queued.is_processing = False
                        self._commit_with_safe_rollback(
                            db_session,
                            f"is_processing reset for research {research_id}",
                        )
                    continue

                # Exhausted retries — mark terminal FAILED, delete the
                # queue row to stop re-dispatch, and notify the user.
                # The spawn failure was already logged (redacted) at the
                # top of this except block; no need to repeat it here.
                logger.warning(
                    f"Spawn failed for research {research_id} "
                    f"after {attempts} attempts; marking FAILED"
                )
                with self._spawn_retry_counts_lock:
                    self._spawn_retry_counts.pop(research_id, None)
                try:
                    research = (
                        db_session.query(ResearchHistory)
                        .filter_by(id=research_id)
                        .first()
                    )
                    if research:
                        research.status = ResearchStatus.FAILED
                    if fresh_queued:
                        db_session.delete(fresh_queued)
                    db_session.commit()
                except Exception as e2:
                    # ``password`` is in scope — same redaction rationale.
                    safe_msg = redact_secrets(str(e2), password)
                    logger.warning(
                        "Failed to persist terminal FAILED state for "
                        f"research {research_id}: {safe_msg}"
                    )
                    try:
                        db_session.rollback()
                    except Exception as rb_err:
                        # No exc_info: same frame-locals rationale as the
                        # rollback handler above.
                        logger.debug(
                            "Rollback after terminal update failure: "
                            f"{redact_secrets(str(rb_err), password)}"
                        )

                # notify_research_failed opens its own session and
                # sends the user notification. Called exactly once
                # per research_id because the counter is popped above.
                self.notify_research_failed(
                    username=username,
                    research_id=research_id,
                    error_message=(
                        f"Failed to start research after {attempts} attempts"
                    ),
                    user_password=password,
                )

    def _start_research(
        self,
        db_session: Session,
        username: str,
        password: str,
        queued_research,
    ):
        """Start a queued research.

        Commits ``ResearchHistory.status = IN_PROGRESS`` BEFORE spawning
        the thread. If we did this after, a fast-completing thread
        (which opens its own DB session) could write ``COMPLETED`` and
        then our post-spawn commit would overwrite that with
        ``IN_PROGRESS``, stranding the research as stuck IN_PROGRESS
        after it had already finished.

        If ``start_research_process`` raises, reset status back to
        ``QUEUED`` and re-raise so the caller's 3-strike retry logic
        handles it. ``DuplicateResearchError`` is re-raised as-is
        because a thread is already running for this research; mutating
        status further would be wrong.
        """
        research_id = queued_research.research_id
        research = (
            db_session.query(ResearchHistory).filter_by(id=research_id).first()
        )

        if not research:
            raise ValueError(f"Research {research_id} not found")

        # Guard against re-entering _start_research on a retry when a
        # prior attempt's post-spawn UserActiveResearch commit failed:
        #   - IN_PROGRESS means the prior thread is (or was) running.
        #   - COMPLETED/FAILED means the prior thread already finished
        #     and cleaned itself up out of _active_research, so a bare
        #     retry would both overwrite the terminal status with
        #     IN_PROGRESS and then spawn a *second* thread (because
        #     check_and_start_research sees no live entry), re-running
        #     the whole research.
        # In all three cases the correct behavior is the same: raise
        # DuplicateResearchError so the caller's existing dup branch
        # deletes the queue row without mutating status or notifying.
        if research.status != ResearchStatus.QUEUED:
            raise DuplicateResearchError(
                f"Research {research_id} is already started "
                f"(status={research.status})"
            )

        # Claim IN_PROGRESS before spawn to close the
        # thread-completes-before-parent-commits race.
        research.status = ResearchStatus.IN_PROGRESS
        db_session.commit()

        # Extract settings
        settings_snapshot = queued_research.settings_snapshot or {}

        # Handle new vs legacy structure
        if (
            isinstance(settings_snapshot, dict)
            and "submission" in settings_snapshot
        ):
            submission_params = settings_snapshot.get("submission", {})
            complete_settings = settings_snapshot.get("settings_snapshot", {})
        else:
            submission_params = settings_snapshot
            # A legacy-flat queued row (enqueued by an older version, pre-
            # "submission" wrapper) carries no settings_snapshot. Seed the run's
            # primary engine from the submitted search_engine so the worker's
            # egress build (resolve_run_primary_engine) doesn't fail closed on
            # an empty snapshot and refuse the run.
            _legacy_engine = submission_params.get("search_engine")
            complete_settings = (
                {"search.tool": _legacy_engine} if _legacy_engine else {}
            )

        try:
            research_thread = start_research_process(
                research_id,
                queued_research.query,
                queued_research.mode,
                run_research_process,
                username=username,
                user_password=password,  # Pass password for metrics
                model_provider=submission_params.get("model_provider"),
                model=submission_params.get("model"),
                custom_endpoint=submission_params.get("custom_endpoint"),
                search_engine=submission_params.get("search_engine"),
                max_results=submission_params.get("max_results"),
                time_period=submission_params.get("time_period"),
                iterations=submission_params.get("iterations"),
                questions_per_iteration=submission_params.get(
                    "questions_per_iteration"
                ),
                strategy=submission_params.get("strategy", "source-based"),
                settings_snapshot=complete_settings,
            )
        except DuplicateResearchError:
            # A live thread already exists for this research_id (e.g.
            # previous attempt's post-spawn commit failed). Do NOT
            # reset status — that would contradict the running thread.
            raise
        except SystemAtCapacityError:
            # System at concurrent-research capacity. No thread was
            # spawned. Reset to QUEUED so the next dispatch tick can try
            # again — this is not a permanent spawn failure and should
            # NOT count toward SPAWN_RETRY_LIMIT.
            logger.info(
                f"System at capacity when dispatching {research_id}; "
                "re-queueing for next tick"
            )
            research.status = ResearchStatus.QUEUED
            self._commit_with_safe_rollback(
                db_session,
                f"status reset to QUEUED after capacity reject for research {research_id}",
            )
            raise
        except Exception:
            # Genuine spawn failure: no thread exists. Roll back the
            # IN_PROGRESS claim so the retry sees a clean QUEUED row.
            research.status = ResearchStatus.QUEUED
            self._commit_with_safe_rollback(
                db_session,
                f"status reset to QUEUED after spawn failure for research {research_id}",
            )
            raise

        # Thread is running. Record the active-research row. If this
        # commit fails the live thread is unrecorded but still running.
        # Raise DuplicateResearchError instead of letting a generic
        # exception propagate, so the caller's dup branch cleans up the
        # queue row without bumping the retry counter — if we let this
        # count as a spawn failure, three consecutive post-spawn commit
        # failures (or one at LIMIT-1) would push the counter to
        # SPAWN_RETRY_LIMIT and mark a LIVE thread as terminal FAILED.
        active_record = UserActiveResearch(
            username=username,
            research_id=research_id,
            status=ResearchStatus.IN_PROGRESS,
            thread_id=str(research_thread.ident),
            settings_snapshot=queued_research.settings_snapshot,
        )
        db_session.add(active_record)
        if not self._commit_with_safe_rollback(
            db_session,
            f"UserActiveResearch persist after spawn for research {research_id}",
        ):
            # Thread is live; the commit failing leaves the UAR row
            # unrecorded but the thread running. Raise
            # DuplicateResearchError so the caller's dup branch deletes
            # the queue row without bumping the retry counter — if we
            # let a plain exception count as a spawn failure, a commit
            # failure at SPAWN_RETRY_LIMIT - 1 would mark a LIVE thread
            # as terminal FAILED.
            raise DuplicateResearchError(
                f"Research {research_id} thread is live; "
                "UserActiveResearch commit failed"
            )

    def process_user_request(self, username: str, session_id: str) -> int:
        """
        Process queue for a user during their request.
        This is called from request context to check and start queued items.

        Returns:
            Number of researches started
        """
        # Pre-declared so the except handler can pass it to redact_secrets
        # even if the exception is raised before the retrieve below.
        password = None
        try:
            # Add user to check list
            self.notify_user_activity(username, session_id)

            # Force immediate check (don't wait for loop)
            password = session_password_store.get_session_password(
                username, session_id
            )
            if password:
                # Open database and check queue
                engine = db_manager.open_user_database(username, password)
                if engine:
                    with get_user_db_session(username) as db_session:
                        queue_service = UserQueueService(db_session)
                        status = queue_service.get_queue_status()

                        if status and status["queued_tasks"] > 0:
                            logger.info(
                                f"User {username} has {status['queued_tasks']} "
                                f"queued tasks, triggering immediate processing"
                            )
                            # Process will happen in background thread
                            return int(status["queued_tasks"])

            return 0

        except Exception as e:
            # ``password`` may be bound above — drop the traceback chain
            # and redact str(e).
            safe_msg = redact_secrets(str(e), password)
            logger.warning(
                f"Error in process_user_request for {username}: {safe_msg}"
            )
            return 0

    def queue_progress_update(
        self, username: str, research_id: str, progress: float
    ):
        """
        Queue a progress update that needs database access.
        For compatibility with old processor during migration.

        Args:
            username: The username
            research_id: The research ID
            progress: The progress value (0-100)
        """
        # In processor_v2, we can update directly if we have database access
        # or queue it for later processing
        operation_id = str(uuid.uuid4())
        with self._pending_operations_lock:
            self.pending_operations[operation_id] = {
                "username": username,
                "operation_type": "progress_update",
                "research_id": research_id,
                "progress": progress,
                "timestamp": time.time(),
            }
        logger.debug(
            f"Queued progress update for research {research_id}: {progress}%"
        )

    def queue_error_update(
        self,
        username: str,
        research_id: str,
        status: str,
        error_message: str,
        metadata: Dict[str, Any],
        completed_at: str,
        report_path: Optional[str] = None,
    ):
        """
        Queue an error status update that needs database access.
        For compatibility with old processor during migration.

        Args:
            username: The username
            research_id: The research ID
            status: The status to set (failed, suspended, etc.)
            error_message: The error message
            metadata: Research metadata
            completed_at: Completion timestamp
            report_path: Optional path to error report
        """
        operation_id = str(uuid.uuid4())
        with self._pending_operations_lock:
            self.pending_operations[operation_id] = {
                "username": username,
                "operation_type": "error_update",
                "research_id": research_id,
                "status": status,
                "error_message": error_message,
                "metadata": metadata,
                "completed_at": completed_at,
                "report_path": report_path,
                "timestamp": time.time(),
            }
        logger.info(
            f"Queued error update for research {research_id} with status {status}"
        )

    def process_pending_operations_for_user(
        self, username: str, db_session: Session
    ) -> int:
        """
        Process pending operations for a user when we have database access.
        Called from request context where encrypted database is accessible.
        For compatibility with old processor during migration.

        Args:
            username: Username to process operations for
            db_session: Active database session for the user

        Returns:
            Number of operations processed
        """
        # Find pending operations for this user (with lock)
        operations_to_process = []
        with self._pending_operations_lock:
            for op_id, op_data in list(self.pending_operations.items()):
                if op_data["username"] == username:
                    operations_to_process.append((op_id, op_data))
                    # Remove immediately to prevent duplicate processing
                    del self.pending_operations[op_id]

        if not operations_to_process:
            return 0

        processed_count = 0

        # Process operations outside the lock (to avoid holding lock during DB operations)
        for op_id, op_data in operations_to_process:
            try:
                operation_type = op_data.get("operation_type")

                if operation_type == "progress_update":
                    # Update progress in database
                    from ...database.models import ResearchHistory

                    research = (
                        db_session.query(ResearchHistory)
                        .filter_by(id=op_data["research_id"])
                        .first()
                    )
                    if research:
                        # Update the progress column directly
                        research.progress = op_data["progress"]
                        db_session.commit()
                        processed_count += 1

                elif operation_type == "error_update":
                    # Update error status in database
                    from ...database.models import ResearchHistory

                    research = (
                        db_session.query(ResearchHistory)
                        .filter_by(id=op_data["research_id"])
                        .first()
                    )
                    if research:
                        research.status = op_data["status"]
                        research.error_message = op_data["error_message"]
                        research.research_meta = op_data["metadata"]
                        research.completed_at = op_data["completed_at"]
                        if op_data.get("report_path"):
                            research.report_path = op_data["report_path"]
                        db_session.commit()
                        processed_count += 1

            except Exception:
                logger.exception(f"Error processing operation {op_id}")
                # Rollback to clear the failed transaction state
                try:
                    db_session.rollback()
                except Exception:
                    logger.warning(
                        f"Failed to rollback after error in operation {op_id}"
                    )

        return processed_count


# Global queue processor instance
queue_processor = QueueProcessorV2()
