"""What keeps one request from overwhelming the server when a source is very
large: page size limits, the cap on rows read into Python, capped counts,
keyset paging and the statement timeout."""

import uuid

import pytest
from fastapi import Depends, HTTPException
from sqlalchemy import select, text, true
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.config import settings
from app.core.database import get_db
from app.crud import record_sql
from app.main import app
from app.models.master import MasterRecord
from tests.conftest import AsyncSessionLocal


async def _master_with_records(client, headers, count: int) -> tuple[str, list[str]]:
    master = await client.post("/masters", headers=headers, json={"name": f"Large {uuid.uuid4().hex[:6]}", "field_schema": [{"name": "title", "type": "varchar"}]})
    assert master.status_code == 201, master.text
    master_id = master.json()["id"]
    ids = []
    for i in range(count):
        record = await client.post(f"/masters/{master_id}/records", headers=headers, json={"field_values": {"title": f"r{i}"}})
        assert record.status_code == 201, record.text
        ids.append(record.json()["id"])
    return master_id, ids


async def _form_with_submissions(client, headers, count: int) -> tuple[str, list[str]]:
    form = await client.post("/forms", headers=headers, json={"name": f"Large {uuid.uuid4().hex[:6]}", "field_definitions": [{"name": "subject", "type": "text", "label": "Subject"}]})
    assert form.status_code == 201, form.text
    published = await client.post(f"/forms/{form.json()['form_id']}/publish", headers=headers)
    assert published.status_code == 200, published.text
    form_id, version = published.json()["form_id"], published.json()["version"]
    ids = []
    for i in range(count):
        made = await client.post("/form-submissions", headers=headers, json={"form_id": form_id, "form_version": version, "field_values": {"subject": f"s{i}"}})
        assert made.status_code == 201, made.text
        ids.append(made.json()["id"])
    return form_id, ids


@pytest.mark.parametrize("params", [{"limit": record_sql.MAX_PAGE_SIZE + 1}, {"limit": 0}, {"skip": -1}])
async def test_a_list_request_stays_within_the_page_limits(client, auth_headers, params):
    master_id, _ = await _master_with_records(client, auth_headers, 0)
    form_id, _ = await _form_with_submissions(client, auth_headers, 0)
    assert (await client.get(f"/masters/{master_id}/records", headers=auth_headers, params=params)).status_code == 422
    assert (await client.get(f"/forms/{form_id}/submissions", headers=auth_headers, params=params)).status_code == 422


async def test_the_largest_page_is_allowed(client, auth_headers):
    master_id, _ = await _master_with_records(client, auth_headers, 1)
    listed = await client.get(f"/masters/{master_id}/records", headers=auth_headers, params={"limit": record_sql.MAX_PAGE_SIZE})
    assert listed.status_code == 200, listed.text
    assert len(listed.json()) == 1


async def test_a_dataset_query_stays_within_its_limit(client, auth_headers):
    master_id, _ = await _master_with_records(client, auth_headers, 0)
    too_many = await client.post("/datasets/query", headers=auth_headers, json={"source_type": "master", "source_id": master_id, "columns": [], "filters": [], "limit": 50_001})
    assert too_many.status_code == 422


async def test_reading_into_python_stops_at_the_cap(client, auth_headers, monkeypatch):
    master_id, _ = await _master_with_records(client, auth_headers, 3)
    stmt = select(MasterRecord.id).where(MasterRecord.master_id == uuid.UUID(master_id))
    async with AsyncSessionLocal() as db:
        monkeypatch.setattr(record_sql, "MAX_PYTHON_ROWS", 3)
        assert len(await record_sql.read_capped(db, stmt)) == 3
        monkeypatch.setattr(record_sql, "MAX_PYTHON_ROWS", 2)
        with pytest.raises(HTTPException) as refused:
            await record_sql.read_capped(db, stmt)
    assert refused.value.status_code == 422
    assert "narrow it" in refused.value.detail


async def test_a_task_data_list_stops_at_the_cap(client, auth_headers, test_tenant, monkeypatch):
    from app.crud import tasks

    master_id, _ = await _master_with_records(client, auth_headers, 3)
    monkeypatch.setattr(record_sql, "MAX_PYTHON_ROWS", 2)
    async with AsyncSessionLocal() as db:
        ok, message, _ = await tasks._exec_data_list(db, test_tenant["tenant_id"], {"master_id": master_id}, {})
    assert ok is False
    assert "set a filter" in message


async def test_counts_stop_at_the_cap_and_say_so(client, auth_headers, monkeypatch):
    master_id, _ = await _master_with_records(client, auth_headers, 3)
    form_id, _ = await _form_with_submissions(client, auth_headers, 3)

    exact = await client.get(f"/masters/{master_id}/records/count", headers=auth_headers)
    assert exact.json() == {"total": 3, "capped": False}
    assert (await client.get(f"/masters/{master_id}", headers=auth_headers)).json()["record_count_capped"] is False

    monkeypatch.setattr(record_sql, "COUNT_CAP", 2)
    assert (await client.get(f"/masters/{master_id}/records/count", headers=auth_headers)).json() == {"total": 2, "capped": True}
    assert (await client.get(f"/forms/{form_id}/submissions/count", headers=auth_headers)).json() == {"total": 2, "capped": True}

    one = (await client.get(f"/masters/{master_id}", headers=auth_headers)).json()
    assert (one["record_count"], one["record_count_capped"]) == (2, True)
    listed = next(m for m in (await client.get("/masters", headers=auth_headers)).json() if m["id"] == master_id)
    assert (listed["record_count"], listed["record_count_capped"]) == (2, True)

    model = (await client.get("/app-model", headers=auth_headers)).json()
    entity = next(e for e in model["entities"] if e["id"] == master_id)
    assert (entity["record_count"], entity["record_count_capped"]) == (2, True)


async def test_counts_leave_out_deleted_records_and_other_sources(client, auth_headers):
    master_id, ids = await _master_with_records(client, auth_headers, 3)
    await _master_with_records(client, auth_headers, 2)
    assert (await client.delete(f"/masters/{master_id}/records/{ids[0]}", headers=auth_headers)).status_code in (200, 204)
    assert (await client.get(f"/masters/{master_id}", headers=auth_headers)).json()["record_count"] == 2


async def test_keyset_paging_walks_records_like_skip_does(client, auth_headers):
    master_id, ids = await _master_with_records(client, auth_headers, 5)
    url, walked, after = f"/masters/{master_id}/records", [], None
    for _ in range(3):
        params = {"limit": 2, **({"after": after} if after else {})}
        page = (await client.get(url, headers=auth_headers, params=params)).json()
        walked += [r["id"] for r in page]
        after = page[-1]["id"]
    assert walked == ids
    assert (await client.get(url, headers=auth_headers, params={"limit": 2, "after": ids[-1]})).json() == []


async def test_keyset_paging_keeps_the_search(client, auth_headers):
    master_id, ids = await _master_with_records(client, auth_headers, 3)
    extra = await client.post(f"/masters/{master_id}/records", headers=auth_headers, json={"field_values": {"title": "r-last"}})
    page = (await client.get(f"/masters/{master_id}/records", headers=auth_headers, params={"q": "r-last", "after": ids[0]})).json()
    assert [r["id"] for r in page] == [extra.json()["id"]]


async def test_keyset_paging_walks_submissions_newest_first(client, auth_headers):
    form_id, ids = await _form_with_submissions(client, auth_headers, 5)
    url, walked, after = f"/forms/{form_id}/submissions", [], None
    for _ in range(3):
        params = {"limit": 2, **({"after": after} if after else {})}
        page = (await client.get(url, headers=auth_headers, params=params)).json()
        walked += [r["id"] for r in page]
        after = page[-1]["id"]
    assert walked == list(reversed(ids))


async def test_a_cursor_that_has_gone_falls_back_to_skip(client, auth_headers):
    master_id, ids = await _master_with_records(client, auth_headers, 3)
    page = (await client.get(f"/masters/{master_id}/records", headers=auth_headers, params={"after": str(uuid.uuid4()), "skip": 2})).json()
    assert [r["id"] for r in page] == [ids[2]]


async def test_a_cursor_from_another_source_is_not_used(client, auth_headers):
    master_id, ids = await _master_with_records(client, auth_headers, 2)
    _, other_ids = await _master_with_records(client, auth_headers, 1)
    page = (await client.get(f"/masters/{master_id}/records", headers=auth_headers, params={"after": other_ids[0]})).json()
    assert [r["id"] for r in page] == ids


async def test_the_python_path_pages_by_cursor_and_stops_at_the_cap(client, auth_headers, monkeypatch):
    """A search SQL can only narrow finishes in Python: the same paging, and the same cap."""
    master_id, ids = await _master_with_records(client, auth_headers, 4)
    monkeypatch.setattr(record_sql, "search_clause", lambda *args, **kwargs: record_sql.Clause(true(), False))
    url = f"/masters/{master_id}/records"

    page = (await client.get(url, headers=auth_headers, params={"q": "r", "after": ids[1], "limit": 1})).json()
    assert [r["id"] for r in page] == [ids[2]]
    page = (await client.get(url, headers=auth_headers, params={"q": "r", "after": str(uuid.uuid4()), "skip": 3})).json()
    assert [r["id"] for r in page] == [ids[3]]
    assert (await client.get(f"{url}/count", headers=auth_headers, params={"q": "r"})).json() == {"total": 4, "capped": False}

    monkeypatch.setattr(record_sql, "MAX_PYTHON_ROWS", 2)
    refused = await client.get(url, headers=auth_headers, params={"q": "r"})
    assert refused.status_code == 422
    assert "narrow it" in refused.text


@pytest.fixture
def slow_route():
    """A route that runs one slow statement on a request's own session."""

    async def slow(db: AsyncSession = Depends(get_db)) -> dict:
        await db.execute(text("SELECT pg_sleep(2)"))
        return {"finished": True}

    async def setting(db: AsyncSession = Depends(get_db)) -> dict:
        return {"value": (await db.execute(text("SHOW statement_timeout"))).scalar_one()}

    app.add_api_route("/__pytest/slow", slow)
    app.add_api_route("/__pytest/setting", setting)
    yield
    app.router.routes[:] = [r for r in app.router.routes if not getattr(r, "path", "").startswith("/__pytest/")]


async def test_a_slow_statement_is_stopped_with_a_clear_error(client, auth_headers, slow_route, monkeypatch):
    monkeypatch.setattr(settings, "request_statement_timeout_ms", 300)
    stopped = await client.get("/__pytest/slow", headers=auth_headers)
    assert stopped.status_code == 503, stopped.text
    assert "took too long" in stopped.json()["detail"]


async def test_the_timeout_can_be_turned_off(client, auth_headers, slow_route, monkeypatch):
    monkeypatch.setattr(settings, "request_statement_timeout_ms", 0)
    assert (await client.get("/__pytest/setting", headers=auth_headers)).json()["value"] == "0"


async def test_the_timeout_applies_to_requests_and_not_to_storage_changes(client, auth_headers, slow_route, monkeypatch):
    from app.crud import entity_storage

    monkeypatch.setattr(settings, "request_statement_timeout_ms", 45_000)
    assert (await client.get("/__pytest/setting", headers=auth_headers)).json()["value"] == "45s"

    async with AsyncSessionLocal() as db, db.begin():
        await db.execute(text("SELECT set_config('statement_timeout', '300', true)"))
        await entity_storage._no_time_limit(db)
        assert (await db.execute(text("SHOW statement_timeout"))).scalar_one() == "0"