"""Tests for the process-level period-row cache.

One dashboard page load fans out to ~10 endpoints in parallel and a
naive implementation runs ``_collect_period_rows`` ~13 times (each of
overview/movers/forecast re-fetches cur/prev/yoy independently). The
cache eliminates the dup work; these tests pin the contract:

* Two calls inside the TTL window with the same (start, end) reuse
  the warm cache — no second hit to the bucket loaders.
* Different ``bu`` filters share the underlying fetch but apply the
  filter post-cache (so a filtered call still uses the warm cache).
* Different window keys do NOT collide.
* TTL expiry invalidates correctly.
* ``invalidate_period_rows_cache()`` wipes the entire cache.
* Concurrent cold-misses on the same key compute exactly once
  (thundering-herd defence via the per-key lock).
"""
from __future__ import annotations

import asyncio
from datetime import date
from decimal import Decimal
from unittest.mock import AsyncMock, patch

import pytest

from app.domains.ai_spending import service as svc_mod
from app.domains.ai_spending.service import (
    BUCKET_FOUNDRY,
    AiSpendingService,
    BucketRow,
    _period_rows_cache,
    _period_rows_locks,
    invalidate_period_rows_cache,
)


@pytest.fixture(autouse=True)
def _clean_cache():
    """Reset cache between tests so a flaky leftover entry from one
    test doesn't bleed into the next."""
    invalidate_period_rows_cache()
    _period_rows_locks.clear()
    yield
    invalidate_period_rows_cache()
    _period_rows_locks.clear()


@pytest.fixture(autouse=True)
def _force_live_compute():
    """This file pins the L1 process cache contract, which wraps the live
    compute path. ``_load_period_rows`` now always tries the rollup first
    (closed AND open month — perf fix 2026-06), so force a rollup MISS
    (and a stub fingerprint to avoid the load_whitelist DB hit on the mock
    db) → every call falls through to ``_compute_period_rows`` (the three
    loaders). Rollup read-through itself is covered in
    ``test_period_rollup.py`` with a real DB."""
    with patch.object(
        AiSpendingService, "_rollup_fingerprint",
        new=AsyncMock(return_value="fp-test"),
    ), patch.object(
        AiSpendingService, "_read_rollup", new=AsyncMock(return_value=None),
    ):
        yield


def _row(day: date) -> BucketRow:
    return BucketRow(
        bucket=BUCKET_FOUNDRY,
        date=day,
        cost_usd=Decimal("1.00"),
        integration_id="integ-x",
    )


@pytest.mark.asyncio
async def test_two_calls_same_window_reuse_cache():
    """Second call within the TTL must NOT re-run the bucket loaders."""
    svc = AiSpendingService(db=AsyncMock())
    start, end = date(2026, 5, 1), date(2026, 6, 1)
    with patch.object(
        svc, "_bucket_seats", return_value=[],
    ) as p_seats, patch.object(
        svc, "_bucket_bailian", return_value=[],
    ) as p_bai, patch.object(
        svc, "_bucket_cost_facts", return_value=[_row(date(2026, 5, 15))],
    ) as p_cf:
        rows1 = await svc._collect_period_rows(start, end)
        rows2 = await svc._collect_period_rows(start, end)
    assert rows1 == rows2
    assert len(rows1) == 1
    # Each bucket loader called exactly once — the second call hit the
    # cache.
    assert p_seats.await_count == 1
    assert p_bai.await_count == 1
    assert p_cf.await_count == 1


@pytest.mark.asyncio
async def test_different_windows_do_not_collide():
    """Cache is keyed by (start, end). Two non-overlapping months must
    NOT share a cache slot."""
    svc = AiSpendingService(db=AsyncMock())
    with patch.object(
        svc, "_bucket_seats", return_value=[],
    ) as p_seats, patch.object(
        svc, "_bucket_bailian", return_value=[],
    ), patch.object(
        svc, "_bucket_cost_facts", return_value=[],
    ):
        await svc._collect_period_rows(date(2026, 4, 1), date(2026, 5, 1))
        await svc._collect_period_rows(date(2026, 5, 1), date(2026, 6, 1))
    assert p_seats.await_count == 2


@pytest.mark.asyncio
async def test_bu_filter_applied_after_cache_lookup():
    """A bu-filtered call should still benefit from the cache: the
    underlying loaders run once for the window, then the filter is
    applied to the warm rows."""
    svc = AiSpendingService(db=AsyncMock())
    start, end = date(2026, 5, 1), date(2026, 6, 1)
    rows_full = [_row(date(2026, 5, 15)), _row(date(2026, 5, 20))]

    async def fake_filter(rows, bu):  # noqa: ANN001
        # The filter receives the cached, unfiltered list. Pretend "Eng"
        # narrows to the first row.
        return [rows[0]] if bu == "Eng" else rows

    with patch.object(
        svc, "_bucket_seats", return_value=[],
    ) as p_seats, patch.object(
        svc, "_bucket_bailian", return_value=[],
    ), patch.object(
        svc, "_bucket_cost_facts", return_value=rows_full,
    ), patch.object(
        svc, "_filter_rows_by_bu", side_effect=fake_filter,
    ) as p_filter:
        unfiltered = await svc._collect_period_rows(start, end)
        eng_only = await svc._collect_period_rows(start, end, bu="Eng")
        other = await svc._collect_period_rows(start, end, bu="Sales")
    assert len(unfiltered) == 2
    assert len(eng_only) == 1
    assert len(other) == 2  # Sales = passthrough in our stub
    # Bucket loaders only ran once across all three calls.
    assert p_seats.await_count == 1
    # Filter only ran for the two bu-filtered calls.
    assert p_filter.await_count == 2


@pytest.mark.asyncio
async def test_invalidate_wipes_cache():
    """``invalidate_period_rows_cache`` must force the next call to
    re-run the loaders — this is what ``update_settings`` calls so an
    admin's whitelist / FX-rate edit becomes visible immediately."""
    svc = AiSpendingService(db=AsyncMock())
    start, end = date(2026, 5, 1), date(2026, 6, 1)
    with patch.object(
        svc, "_bucket_seats", return_value=[],
    ) as p_seats, patch.object(
        svc, "_bucket_bailian", return_value=[],
    ), patch.object(
        svc, "_bucket_cost_facts", return_value=[],
    ):
        await svc._collect_period_rows(start, end)
        assert _period_rows_cache  # warm
        invalidate_period_rows_cache()
        assert not _period_rows_cache  # empty after wipe
        await svc._collect_period_rows(start, end)
    assert p_seats.await_count == 2  # ran twice — cache was busted between


@pytest.mark.asyncio
async def test_ttl_expiry_re_runs_loaders(monkeypatch):
    """Once the TTL elapses the next call MUST re-run the loaders."""
    svc = AiSpendingService(db=AsyncMock())
    start, end = date(2026, 5, 1), date(2026, 6, 1)

    # Frozen monotonic clock.
    t = {"now": 1000.0}
    monkeypatch.setattr(svc_mod.time, "monotonic", lambda: t["now"])

    with patch.object(
        svc, "_bucket_seats", return_value=[],
    ) as p_seats, patch.object(
        svc, "_bucket_bailian", return_value=[],
    ), patch.object(
        svc, "_bucket_cost_facts", return_value=[],
    ):
        await svc._collect_period_rows(start, end)
        # Within TTL — still cached.
        t["now"] += svc_mod._PERIOD_ROWS_TTL_SECONDS - 1
        await svc._collect_period_rows(start, end)
        # Past TTL — cache miss.
        t["now"] += 2
        await svc._collect_period_rows(start, end)
    assert p_seats.await_count == 2


@pytest.mark.asyncio
async def test_thundering_herd_collapses_to_one_compute():
    """13 sibling endpoints can launch the same cold ``_collect_period_
    rows`` call simultaneously. The per-key lock must ensure exactly
    one of them does the work; the rest read the warm cache."""
    svc = AiSpendingService(db=AsyncMock())
    start, end = date(2026, 5, 1), date(2026, 6, 1)

    call_count = {"seats": 0}

    async def slow_seats(*_a, **_kw):
        call_count["seats"] += 1
        # Hold the lock for long enough that every parallel caller
        # would race in without the lock.
        await asyncio.sleep(0.05)
        return []

    with patch.object(
        svc, "_bucket_seats", side_effect=slow_seats,
    ), patch.object(
        svc, "_bucket_bailian", return_value=[],
    ), patch.object(
        svc, "_bucket_cost_facts", return_value=[],
    ):
        # Fan out 13 parallel callers like the dashboard does.
        await asyncio.gather(*[
            svc._collect_period_rows(start, end) for _ in range(13)
        ])
    assert call_count["seats"] == 1


@pytest.mark.asyncio
async def test_empty_window_returns_immediately_without_caching():
    """``period_end_excl <= period_start`` is a no-op; it must not
    pollute the cache."""
    svc = AiSpendingService(db=AsyncMock())
    with patch.object(svc, "_bucket_seats", return_value=[]) as p_seats:
        rows = await svc._collect_period_rows(
            date(2026, 5, 1), date(2026, 5, 1),
        )
    assert rows == []
    assert p_seats.await_count == 0
    assert not _period_rows_cache