"""No-seq-scan gate on the hot indexed query paths at ~100x scale (M8;
04-architecture "Performance baseline"; T-RPT-006's CI half).

The guard: for the queries that SHOULD ride an index on time_entries —
the default date-ordered list page, a selective date range, and a single
row by id — the plan must use an index on time_entries, never a Seq
Scan. This catches an accidentally-dropped index or a query rewrite that
defeats one, which a small-fixture EXPLAIN cannot (Postgres correctly
seq-scans a tiny table).

Deliberately NOT asserted: the unbounded default totals/summary counts
(date=all) DO scan — aggregating the whole backlog is O(N) by nature and
the correct plan. The full-scale latency numbers for those live in
docs/PERF.md; here we only pin that the INDEXED paths stay indexed.
"""

import datetime as dt

from sqlalchemy import text

from app.models import TimeEntry
from app.services.authz.scope import scope
from app.services.reporting.entries import PAGE_SIZE, _row_select, fetch_entry_row
from app.services.reporting.filters import (
    EntryFilters,
    EntryJoins,
    apply_filters,
)


def _explain(session, engine, stmt) -> str:
    compiled = stmt.compile(engine, compile_kwargs={"literal_binds": True})
    rows = session.execute(text("EXPLAIN " + str(compiled))).fetchall()
    return "\n".join(r[0] for r in rows)


def _list_stmt(actor, f: EntryFilters, *, ordered: bool = True):
    joins = EntryJoins()
    stmt = apply_filters(scope(actor, "entries:read", _row_select(joins)), joins, f)
    if ordered:
        stmt = stmt.order_by(TimeEntry.start_at.desc())
    return stmt.limit(PAGE_SIZE)


def _assert_indexed(plan: str) -> None:
    assert "Seq Scan on time_entries" not in plan, (
        "hot path regressed to a Seq Scan on time_entries:\n" + plan
    )
    assert "ix_time_entries" in plan, (
        "expected an index on time_entries in the plan:\n" + plan
    )


def test_default_ordered_list_page_uses_index(perf_engine, perf_session, perf_rian):
    """The default /entries page — date=all, ordered by start_at desc,
    LIMIT 100 — rides ix_time_entries_tenant_start for BOTH the tenant
    filter and the ordering, avoiding a full sort of the backlog."""
    stmt = _list_stmt(perf_rian, EntryFilters())
    plan = _explain(perf_session, perf_engine, stmt)
    _assert_indexed(plan)


def test_selective_date_range_uses_index(perf_engine, perf_session, perf_rian):
    """A one-week window is selective at scale -> index/bitmap scan on
    ix_time_entries_tenant_start, never a Seq Scan."""
    f = EntryFilters(start=dt.datetime(2026, 6, 10), end=dt.datetime(2026, 6, 17))
    plan = _explain(perf_session, perf_engine, _list_stmt(perf_rian, f))
    _assert_indexed(plan)


def test_month_window_totals_uses_index(perf_engine, perf_session, perf_rian):
    """A month-bounded totals aggregate (the interactive scoped path) is
    index-served — the win over the unbounded default totals."""
    from app.services.reporting.entries import entry_totals

    joins = EntryJoins()
    f = EntryFilters(start=dt.datetime(2026, 6, 1), end=dt.datetime(2026, 7, 1))
    # rebuild the totals aggregate select the way entry_totals does, then
    # EXPLAIN it (entry_totals runs it; here we inspect the plan).
    from sqlalchemy import func

    from app.models import TimeEntry as TE

    stmt = apply_filters(
        scope(
            perf_rian,
            "entries:read",
            joins.select_from(func.count().label("n"), func.sum(TE.duration_seconds)),
        ),
        joins,
        f,
    )
    plan = _explain(perf_session, perf_engine, stmt)
    _assert_indexed(plan)
    # sanity: the endpoint actually returns for the same filter
    totals = entry_totals(perf_session, perf_rian, f)
    assert totals["row_count"] >= 0


def test_single_entry_by_id_uses_pk(perf_engine, perf_session, perf_rian):
    """fetch_entry_row (the mutation refetch path) resolves one row by id
    — a primary-key lookup, never a Seq Scan."""
    entry_id = perf_session.execute(
        text("select id from time_entries limit 1")
    ).scalar()
    joins = EntryJoins()
    stmt = scope(perf_rian, "entries:read", _row_select(joins)).where(
        TimeEntry.id == entry_id
    )
    plan = _explain(perf_session, perf_engine, stmt)
    assert "Seq Scan on time_entries" not in plan, plan
    # and it really fetches the row
    row = fetch_entry_row(perf_session, perf_rian, entry_id)
    assert row is not None