"""Fields marked "Indexed" get a partial expression index (app/crud/
field_indexes.py) that queries built by app/crud/record_sql.py actually use."""

import uuid
from decimal import Decimal

import pytest_asyncio
from sqlalchemy import select

from app.core.database import engine
from tests.conftest import AsyncSessionLocal
from app.crud import field_indexes, record_sql
from app.models.master import MasterRecord


async def _indexes(source_id: uuid.UUID, kind: str = "master") -> dict[str, bool]:
    async with AsyncSessionLocal() as db:
        return await field_indexes.existing_indexes(db, kind, source_id)


async def _drop_all(kind: str, source_id: uuid.UUID) -> None:
    await field_indexes.wait_for_pending()
    async with AsyncSessionLocal() as db:
        async with db.begin():  # the job is queued in this transaction
            await field_indexes.sync_field_indexes(db, kind, source_id, [])
    await field_indexes.wait_for_pending()


@pytest_asyncio.fixture
async def cleanup():
    """Sources whose field indexes to drop when the test ends, pass or fail —
    the tenant teardown deletes rows, not indexes."""
    sources: list[tuple[str, uuid.UUID]] = []
    yield sources
    for kind, source_id in sources:
        await _drop_all(kind, source_id)


async def _plan(stmt) -> str:
    compiled = stmt.compile(dialect=engine.dialect)
    params = tuple(compiled.params[name] for name in compiled.positiontup)
    async with engine.connect() as conn:
        await conn.exec_driver_sql("SET enable_seqscan = off")
        rows = (await conn.exec_driver_sql("EXPLAIN " + str(compiled), params)).all()
        await conn.exec_driver_sql("RESET enable_seqscan")
    return "\n".join(r[0] for r in rows)


async def test_indexed_fields_get_used_and_dropped(client, auth_headers, test_tenant, cleanup):
    resp = await client.post(
        "/masters",
        headers=auth_headers,
        json={
            "name": f"Indexed {uuid.uuid4().hex[:6]}",
            "field_schema": [
                {"name": "amount", "type": "number", "indexed": True},
                {"name": "code", "type": "text", "indexed": True},
                {"name": "note", "type": "text"},
            ],
        },
    )
    assert resp.status_code == 201, resp.text
    master_id = uuid.UUID(resp.json()["id"])
    cleanup.append(("master", master_id))
    assert resp.json()["field_schema"][0]["indexed"] is True
    await field_indexes.wait_for_pending()
    built = await _indexes(master_id)
    assert len(built) == 2 and all(built.values())

    async with AsyncSessionLocal() as db:
        async with db.begin():
            db.add_all(
                MasterRecord(tenant_id=test_tenant["tenant_id"], master_id=master_id, field_values={"amount": i, "code": f"C{i:04d}"})
                for i in range(300)
            )
    async with engine.begin() as conn:
        await conn.exec_driver_sql("ANALYZE master_records")

    tenant_id = test_tenant["tenant_id"]
    src = record_sql.RecordSource("master", master_id, resp.json()["field_schema"])
    fv = src.fv
    statements = [
        select(MasterRecord.id).where(*src.base_where(tenant_id), record_sql.numeric_key(fv, "amount") > Decimal(290)),
        select(MasterRecord.id).where(*src.base_where(tenant_id), record_sql.text_key(fv, "code") == "C0042"),
        # A whole translated filter, not just the bare expression.
        select(MasterRecord.id).where(
            *src.base_where(tenant_id),
            record_sql.filter_clause(src, {"field": "amount", "operator": "between", "value": [10, 12]}, tenant_id).sql,
        ),
    ]
    for stmt in statements:
        plan = await _plan(stmt)
        assert any(name in plan for name in built), plan

    # Results through the API are unchanged by the index.
    resp = await client.post(
        "/datasets/query",
        headers=auth_headers,
        json={
            "source_type": "master",
            "source_id": str(master_id),
            "filters": [{"field": "amount", "operator": "gte", "value": 297}],
            "sort_field": "amount",
            "sort_dir": "desc",
        },
    )
    assert [r["amount"] for r in resp.json()["rows"]] == [299, 298, 297]

    # Un-indexing a field drops its index; deleting the datasource drops the rest.
    schema = [
        {"name": "amount", "type": "number", "indexed": False},
        {"name": "code", "type": "text", "indexed": True},
        {"name": "note", "type": "text"},
    ]
    resp = await client.patch(f"/masters/{master_id}", headers=auth_headers, json={"field_schema": schema})
    assert resp.status_code == 200, resp.text
    await field_indexes.wait_for_pending()
    assert len(await _indexes(master_id)) == 1

    resp = await client.delete(f"/masters/{master_id}", headers=auth_headers)
    assert resp.status_code == 204, resp.text
    await field_indexes.wait_for_pending()
    assert await _indexes(master_id) == {}


async def test_form_fields_can_be_indexed_while_live(client, auth_headers, cleanup):
    resp = await client.post(
        "/forms",
        headers=auth_headers,
        json={"name": "Indexed form", "field_definitions": [{"name": "ref", "type": "text", "label": "Ref"}, {"name": "qty", "type": "number", "label": "Qty"}]},
    )
    form_id = uuid.UUID(resp.json()["form_id"])
    cleanup.append(("form", form_id))
    assert (await client.post(f"/forms/{form_id}/publish", headers=auth_headers)).status_code == 200

    resp = await client.put(f"/forms/{form_id}/indexed-fields", headers=auth_headers, json={"field_names": ["ref", "qty"]})
    assert resp.status_code == 200, resp.text
    assert resp.json()["status"] == "published"
    assert [f["indexed"] for f in resp.json()["field_definitions"]] == [True, True]
    await field_indexes.wait_for_pending()
    assert len(await _indexes(form_id, "form")) == 2

    resp = await client.put(f"/forms/{form_id}/indexed-fields", headers=auth_headers, json={"field_names": ["nope"]})
    assert resp.status_code == 422

    resp = await client.put(f"/forms/{form_id}/indexed-fields", headers=auth_headers, json={"field_names": []})
    assert resp.status_code == 200
    await field_indexes.wait_for_pending()
    assert await _indexes(form_id, "form") == {}


async def test_invalid_leftover_index_is_rebuilt(client, auth_headers, test_tenant, cleanup):
    """A CONCURRENTLY build that fails leaves an INVALID index behind; the
    next sync drops and rebuilds it."""
    resp = await client.post("/masters", headers=auth_headers, json={"name": f"Leftover {uuid.uuid4().hex[:6]}", "field_schema": [{"name": "code", "type": "text"}]})
    master_id = uuid.UUID(resp.json()["id"])
    cleanup.append(("master", master_id))
    async with AsyncSessionLocal() as db:
        async with db.begin():
            db.add_all(MasterRecord(tenant_id=test_tenant["tenant_id"], master_id=master_id, field_values={"code": "same"}) for _ in range(2))

    fields = [{"name": "code", "type": "text", "indexed": True}]
    async with engine.connect() as conn:
        partition = await field_indexes.record_partition(conn, "master", master_id)
    ((name, _),) = field_indexes.wanted_indexes("master", master_id, fields, partition).items()
    async with engine.connect() as conn:
        conn = await conn.execution_options(isolation_level="AUTOCOMMIT")
        try:  # a unique index over duplicate values fails part-way
            await conn.exec_driver_sql(f"CREATE UNIQUE INDEX CONCURRENTLY {name} ON {partition} ((field_values ->> 'code')) WHERE master_id = '{master_id}'::uuid")
        except Exception:  # noqa: BLE001 - the failure is the point
            pass
    assert await _indexes(master_id) == {name: False}

    async with AsyncSessionLocal() as db:
        async with db.begin():  # the job is queued in this transaction
            await field_indexes.sync_field_indexes(db, "master", master_id, fields)
    await field_indexes.wait_for_pending()
    assert await _indexes(master_id) == {name: True}