"""Regression tests for benchmark-result persistence dedup.

Background
----------
``BenchmarkResult`` has a ``UniqueConstraint(benchmark_run_id, query_hash)``
(``uix_run_query``). Two code paths persist results from the shared
``run_data["results"]`` list: ``sync_pending_results`` (request thread, via the
``/api/results`` route) and ``_sync_results_to_database`` (worker thread, on
completion). The old check-then-insert was neither atomic across threads nor
deduped within a batch, so a duplicate ``(benchmark_run_id, query_hash)`` —
from a concurrent sync OR a dataset that repeats a question — raised::

    sqlcipher3.dbapi2.IntegrityError: UNIQUE constraint failed:
        benchmark_results.benchmark_run_id, benchmark_results.query_hash

and because it failed at ``commit()`` the whole pending batch rolled back, so
no results were stored and the UI showed "Found 0 results".

These tests exercise ``_persist_unsaved_results`` against a real SQLite DB so
the unique constraint is actually enforced.
"""

import threading
from unittest.mock import MagicMock, Mock, patch

import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from local_deep_research.benchmarks.web_api.benchmark_service import (
    BenchmarkService,
)
from local_deep_research.database.models.base import Base
from local_deep_research.database.models.benchmark import (
    BenchmarkResult,
    BenchmarkRun,
    BenchmarkStatus,
)

# Engines created by the helpers below, disposed after each test so the
# file/memory connections don't leak fds — this suite runs under `-n auto`
# where fd pressure has crashed xdist workers before (see fd_canary / #3816).
_ENGINES = []


@pytest.fixture(autouse=True)
def _dispose_engines():
    yield
    for engine in _ENGINES:
        engine.dispose()
    _ENGINES.clear()


def _make_session():
    engine = create_engine("sqlite:///:memory:")
    _ENGINES.append(engine)
    Base.metadata.create_all(engine)
    session = sessionmaker(bind=engine)()
    session.add(
        BenchmarkRun(
            id=1,
            config_hash="cfg",
            query_hash_list=[],
            search_config={},
            evaluation_config={},
            datasets_config={},
            status=BenchmarkStatus.IN_PROGRESS,
        )
    )
    session.commit()
    return session


def _result(example_id, query_hash):
    return {
        "example_id": example_id,
        "query_hash": query_hash,
        "dataset_type": "simpleqa",
        "question": "Q?",
        "correct_answer": "A",
        "task_index": 0,
    }


def _count(session):
    return session.query(BenchmarkResult).filter_by(benchmark_run_id=1).count()


def test_duplicate_query_hash_within_batch_inserts_once():
    """A dataset that repeats a question (same query_hash) must not trip the
    unique constraint — the duplicate is collapsed to a single row."""
    svc = BenchmarkService(socket_service=MagicMock())
    session = _make_session()
    run_data = {
        "results": [
            _result("ex1", "dup"),
            _result("ex2", "dup"),  # same query_hash as ex1
            _result("ex3", "unique"),
        ]
    }

    staged = svc._persist_unsaved_results(session, 1, run_data)
    session.commit()  # would raise IntegrityError before the fix

    # ex1 and ex3 are staged; ex2 (duplicate query_hash) is collapsed.
    assert staged == [0, 2]
    assert _count(session) == 2


def test_repeated_sync_does_not_double_insert():
    """A second sync of the same results (e.g. request thread after the worker
    already saved) must be a no-op even if the in-memory saved_indices was
    lost — the DB is the source of truth."""
    svc = BenchmarkService(socket_service=MagicMock())
    session = _make_session()
    run_data = {"results": [_result("ex1", "h1")]}

    assert svc._persist_unsaved_results(session, 1, run_data) == [0]
    session.commit()

    # Simulate a concurrent/restarted sync that doesn't know ex1 was saved
    # (saved_indices empty) — dedup must still hold via the DB query_hashes.
    run_data["saved_indices"] = set()
    assert svc._persist_unsaved_results(session, 1, run_data) == []
    session.commit()  # must not raise

    assert _count(session) == 1


def test_staged_not_marked_saved_until_commit():
    """Regression for the silent-drop bug: a commit failure must NOT leave
    results flagged saved. The helper must not mark ``saved_indices`` itself,
    so the next sync retries the rolled-back rows."""
    svc = BenchmarkService(socket_service=MagicMock())
    session = _make_session()
    run_data = {"results": [_result("ex1", "h1")]}

    staged = svc._persist_unsaved_results(session, 1, run_data)
    assert staged == [0]
    # Helper must not have touched saved_indices (caller owns that, post-commit).
    assert "saved_indices" not in run_data

    # Simulate a failed commit: roll back instead of committing.
    session.rollback()

    # Nothing was persisted and nothing was marked saved → the row is retried.
    assert _count(session) == 0
    assert svc._persist_unsaved_results(session, 1, run_data) == [0]
    session.commit()
    assert _count(session) == 1


# ---------------------------------------------------------------------------
# Concurrency: the lock that serializes the two sync paths
# ---------------------------------------------------------------------------


def _file_db(tmp_path):
    """File-backed SQLite engine (shared across threads/connections) seeded
    with BenchmarkRun id=1. In-memory ``:memory:`` is per-connection, so it
    can't be shared between threads — a file DB can."""
    engine = create_engine(
        f"sqlite:///{tmp_path / 'bench.db'}",
        connect_args={"timeout": 30},  # wait out the writer lock, don't error
    )
    _ENGINES.append(engine)
    Base.metadata.create_all(engine)
    SessionLocal = sessionmaker(bind=engine)
    s = SessionLocal()
    s.add(
        BenchmarkRun(
            id=1,
            config_hash="cfg",
            query_hash_list=[],
            search_config={},
            evaluation_config={},
            datasets_config={},
            status=BenchmarkStatus.IN_PROGRESS,
        )
    )
    s.commit()
    s.close()
    return SessionLocal


def _row_count(SessionLocal):
    s = SessionLocal()
    try:
        return s.query(BenchmarkResult).filter_by(benchmark_run_id=1).count()
    finally:
        s.close()


def test_lock_serializes_concurrent_same_hash_inserts(tmp_path):
    """WITH the real ``_results_sync_lock``, two threads inserting the same
    query_hash serialize: the second acquires the lock only after the first
    commits, sees the committed row, and skips. No IntegrityError, one row."""
    SessionLocal = _file_db(tmp_path)
    svc = BenchmarkService(socket_service=MagicMock())
    run_data = {"results": [_result("a", "H")]}

    errors = []

    def worker():
        sess = SessionLocal()
        try:
            with svc._results_sync_lock:
                svc._persist_unsaved_results(sess, 1, run_data)
                sess.commit()
        except Exception as e:  # noqa: BLE001 - test records any failure
            errors.append(type(e).__name__)
        finally:
            sess.close()

    threads = [threading.Thread(target=worker) for _ in range(2)]
    for t in threads:
        t.start()
    for t in threads:
        t.join(timeout=30)
        assert not t.is_alive(), "worker thread hung"

    assert errors == [], errors
    assert _row_count(SessionLocal) == 1


def test_without_lock_interleaved_commits_collide(tmp_path):
    """Negative control proving the lock is load-bearing. WITHOUT
    serialization, forcing both threads to stage (each reads an empty
    ``seen_hashes``) before either commits makes the second commit collide on
    ``uix_run_query`` — exactly the failure the lock in the two sync methods
    prevents. If this ever stops raising, the dedup is no longer relying on
    serialization and the lock could be silently removed."""
    SessionLocal = _file_db(tmp_path)
    svc = BenchmarkService(socket_service=MagicMock())
    run_data = {"results": [_result("a", "H")]}

    # timeout so a worker that dies before the barrier fails loudly instead
    # of hanging CI forever.
    after_stage = threading.Barrier(2, timeout=10)
    errors = []

    def worker():
        sess = SessionLocal()
        try:
            svc._persist_unsaved_results(sess, 1, run_data)
            after_stage.wait()  # both stage before either commits → forced race
            sess.commit()
        except Exception as e:  # noqa: BLE001
            errors.append(type(e).__name__)
        finally:
            sess.close()

    threads = [threading.Thread(target=worker) for _ in range(2)]
    for t in threads:
        t.start()
    for t in threads:
        t.join(timeout=30)
        assert not t.is_alive(), "worker thread hung"

    # Without serialization the loser's commit fails — UNIQUE collision on a
    # SQLite that waits out the busy lock (the common case, timeout=30), or
    # "database is locked" on one that returns BUSY immediately. Either proves
    # the unsynchronized inserts can't both land; the lock is what prevents it.
    assert any(("Integrity" in e or "Operational" in e) for e in errors), errors
    # The winner's single row survived; the loser rolled back.
    assert _row_count(SessionLocal) == 1


def _mock_session_cm(mock_session):
    """A get_user_db_session replacement yielding ``mock_session``."""

    def factory(*_a, **_k):
        cm = MagicMock()
        cm.__enter__ = Mock(return_value=mock_session)
        cm.__exit__ = Mock(return_value=False)
        return cm

    return factory


def test_real_sync_methods_acquire_the_lock():
    """Deterministic guard that the PRODUCTION methods take
    ``_results_sync_lock`` around their DB work.

    The negative-control test proves the race needs that serialization; this
    proves ``sync_pending_results`` and ``_sync_results_to_database`` actually
    enter the lock, so deleting either ``with self._results_sync_lock`` turns
    this red. (A MagicMock session can't reproduce the race, hence a structural
    assertion rather than a threaded one.)"""
    svc = BenchmarkService(socket_service=MagicMock())

    entries = []
    real_lock = svc._results_sync_lock

    class TrackingLock:
        def __enter__(self):
            entries.append(True)
            return real_lock.__enter__()

        def __exit__(self, *a):
            return real_lock.__exit__(*a)

    svc._results_sync_lock = TrackingLock()

    mock_session = MagicMock()
    mock_session.query.return_value.filter.return_value.all.return_value = []

    target = "local_deep_research.database.session_context.get_user_db_session"

    # sync_pending_results (request-thread path)
    svc.active_runs[1] = {
        "data": {"username": "u", "user_password": None},
        "results": [_result("a", "h1")],
    }
    with patch(target, _mock_session_cm(mock_session)):
        svc.sync_pending_results(1, "u")
    assert len(entries) == 1, "sync_pending_results did not enter the lock"

    # _sync_results_to_database (worker-thread path)
    mock_session.query.return_value.filter.return_value.first.return_value = (
        MagicMock(status=BenchmarkStatus.COMPLETED)
    )
    svc.active_runs[2] = {
        "data": {"username": "u", "user_password": None},
        "results": [_result("b", "h2")],
        "thread_complete": True,
        "completion_info": {
            "status": BenchmarkStatus.COMPLETED,
            "completed_examples": 1,
            "failed_examples": 0,
        },
    }
    with patch(target, _mock_session_cm(mock_session)):
        svc._sync_results_to_database(2)
    assert len(entries) == 2, "_sync_results_to_database did not enter the lock"


def test_real_sync_methods_commit_inside_the_lock():
    """Stronger than entry-counting: assert the lock is HELD at the instant
    ``session.commit()`` runs in both production methods. Catches a refactor
    that enters the lock but commits OUTSIDE the critical section (which
    reopens the cross-thread race) — not just outright lock removal."""
    svc = BenchmarkService(socket_service=MagicMock())

    state = {"held": False}
    real_lock = svc._results_sync_lock

    class TrackingLock:
        def __enter__(self):
            state["held"] = True
            return real_lock.__enter__()

        def __exit__(self, *a):
            state["held"] = False
            return real_lock.__exit__(*a)

    svc._results_sync_lock = TrackingLock()

    commit_held = []
    mock_session = MagicMock()
    mock_session.query.return_value.filter.return_value.all.return_value = []
    mock_session.query.return_value.filter.return_value.first.return_value = (
        MagicMock(status=BenchmarkStatus.COMPLETED)
    )
    mock_session.commit.side_effect = lambda: commit_held.append(state["held"])

    target = "local_deep_research.database.session_context.get_user_db_session"

    svc.active_runs[1] = {
        "data": {"username": "u", "user_password": None},
        "results": [_result("a", "h1")],
    }
    with patch(target, _mock_session_cm(mock_session)):
        svc.sync_pending_results(1, "u")

    svc.active_runs[2] = {
        "data": {"username": "u", "user_password": None},
        "results": [_result("b", "h2")],
        "thread_complete": True,
        "completion_info": {
            "status": BenchmarkStatus.COMPLETED,
            "completed_examples": 1,
            "failed_examples": 0,
        },
    }
    with patch(target, _mock_session_cm(mock_session)):
        svc._sync_results_to_database(2)

    # Both methods committed, and the lock was held at each commit.
    assert commit_held == [True, True], commit_held


def test_callers_do_not_mark_saved_on_commit_failure():
    """Caller-level rollback-safety: if ``commit()`` raises, ``saved_indices``
    must stay untouched so the rows retry next sync. The helper test guards the
    helper (it never marks); this guards the CALLER ordering — mark strictly
    after a successful commit. A reorder to mark-before-commit turns this red."""
    svc = BenchmarkService(socket_service=MagicMock())

    mock_session = MagicMock()
    mock_session.query.return_value.filter.return_value.all.return_value = []
    mock_session.query.return_value.filter.return_value.first.return_value = (
        MagicMock(status=BenchmarkStatus.COMPLETED)
    )
    mock_session.commit.side_effect = RuntimeError("commit boom")

    target = "local_deep_research.database.session_context.get_user_db_session"

    # sync_pending_results path
    run1 = {
        "data": {"username": "u", "user_password": None},
        "results": [_result("a", "h1")],
    }
    svc.active_runs[1] = run1
    with patch(target, _mock_session_cm(mock_session)):
        svc.sync_pending_results(1, "u")
    assert "saved_indices" not in run1, run1.get("saved_indices")

    # _sync_results_to_database path
    run2 = {
        "data": {"username": "u", "user_password": None},
        "results": [_result("b", "h2")],
        "thread_complete": True,
        "completion_info": {
            "status": BenchmarkStatus.COMPLETED,
            "completed_examples": 1,
            "failed_examples": 0,
        },
    }
    svc.active_runs[2] = run2
    with patch(target, _mock_session_cm(mock_session)):
        svc._sync_results_to_database(2)
    assert "saved_indices" not in run2, run2.get("saved_indices")