"""
Event-loop-safety regression tests for the query service.

`app.pipeline.run_nl2sql_pipeline` is awaited from the FastAPI event loop but
runs the SQLAlchemy/MySQL round-trip through blocking synchronous code. That
work must be offloaded to a worker thread (`asyncio.to_thread`) so a slow
database can never stall the event loop and, with it, every other request in
the process.
"""
from __future__ import annotations

import asyncio
import threading
import time
from types import SimpleNamespace

from app.core.config import Settings
from app import pipeline as service_module
from app.planning.operators import Operation
from app.planning.planner import QueryPlanResult
from app.sql.builder import BuiltQuery


def _fake_plan_result() -> QueryPlanResult:
    return QueryPlanResult(
        data_query=BuiltQuery(sql="SELECT 1", params={}),
        operation=Operation.LIST,
        record_positions=[],
    )


async def test_db_execution_runs_off_the_event_loop(monkeypatch):
    """While the blocking DB work is in flight the event loop must remain
    free: the call must run in a worker thread, not on the loop thread."""

    observed_thread_ids: list[int] = []

    async def _fake_build_query_plan(**kwargs):
        return _fake_plan_result()

    def _blocking_run_built_query(data_query, record_positions=None, stage_timings=None):
        time.sleep(0.2)  # simulate a slow DB round-trip
        observed_thread_ids.append(threading.get_ident())
        return SimpleNamespace(rows=[], db_execution_time_ms=0.0)

    # `run_nl2sql_pipeline` reads the registry snapshot before planning; a dummy
    # snapshot (with an empty `tables` dict, which the domain guard reads)
    # keeps the test free of contract/DB dependencies.
    monkeypatch.setattr(
        service_module, "get_registry",
        lambda: SimpleNamespace(snapshot=SimpleNamespace(tables={})),
    )
    monkeypatch.setattr(service_module, "build_query_plan", _fake_build_query_plan)
    monkeypatch.setattr(service_module, "run_built_query", _blocking_run_built_query)

    loop_thread_id = threading.get_ident()
    task = asyncio.create_task(
        service_module.run_nl2sql_pipeline(
            question="list invoices",
            settings=Settings(),
            llm_client=None,  # never used - build_query_plan is faked
        )
    )

    # If the DB work ran on the event loop, this sleep could not return until
    # the 0.2s round-trip finished; returning while it is still in flight
    # proves the loop stays free for other requests during execution.
    await asyncio.sleep(0.05)
    assert not task.done()

    response = await task
    assert response["success"] is True
    assert observed_thread_ids[0] != loop_thread_id, "DB work must run on a worker thread, not the event loop"