"""Dashboard summaries of very large sources: stored and refreshed in the
background (app/crud/summary_cache.py), with the same answers as a live read."""

import uuid
from decimal import Decimal

import pytest
from sqlalchemy import text

from app.core import jobs
from app.core.config import settings
from app.crud import record_sql, summary_cache
from tests.conftest import AsyncSessionLocal


async def _master(client, headers, rows) -> str:
    made = await client.post(
        "/masters", headers=headers,
        json={"name": f"Large {uuid.uuid4().hex[:6]}", "field_schema": [{"name": "status", "type": "varchar"}, {"name": "amount", "type": "numeric"}]},
    )
    assert made.status_code == 201, made.text
    master_id = made.json()["id"]
    for status, amount in rows:
        added = await client.post(f"/masters/{master_id}/records", headers=headers, json={"field_values": {"status": status, "amount": amount}})
        assert added.status_code == 201, added.text
    return master_id


ROWS = [("active", 10.5), ("active", 2), ("retired", 7.25)]


async def _group(client, headers, master_id, **extra):
    done = await client.post(
        "/datasets/query", headers=headers,
        json={"source_type": "master", "source_id": master_id, "columns": [], "filters": [], "group_by": "status", "aggregate_field": "amount", "aggregate_fn": "sum", **extra},
    )
    assert done.status_code == 200, done.text
    return done.json()["rows"]


async def _total(client, headers, master_id, **extra):
    done = await client.post("/datasets/query", headers=headers, json={"source_type": "master", "source_id": master_id, "columns": [], "filters": [], "limit": 1, **extra})
    assert done.status_code == 200, done.text
    return done.json()["total"]


async def _stored(tenant_id) -> int:
    async with AsyncSessionLocal() as db:
        return (await db.execute(text("SELECT count(*) FROM summary_cache WHERE tenant_id = :t"), {"t": tenant_id})).scalar_one()


@pytest.fixture
def large(monkeypatch):
    """Every source with more than one record counts as very large."""
    monkeypatch.setattr(record_sql, "COUNT_CAP", 1)


async def test_a_smaller_source_always_reads_live(client, auth_headers, test_tenant):
    master_id = await _master(client, auth_headers, ROWS)
    assert await _total(client, auth_headers, master_id) == 3
    await _group(client, auth_headers, master_id)
    assert await _stored(test_tenant["tenant_id"]) == 0


async def test_stored_answers_match_a_live_read(client, auth_headers, test_tenant, monkeypatch):
    master_id = await _master(client, auth_headers, ROWS)
    live_group, live_total = await _group(client, auth_headers, master_id), await _total(client, auth_headers, master_id)

    monkeypatch.setattr(record_sql, "COUNT_CAP", 1)
    assert await _group(client, auth_headers, master_id) == live_group  # worked out, then stored
    assert await _total(client, auth_headers, master_id) == live_total
    assert await _stored(test_tenant["tenant_id"]) == 2
    assert await _group(client, auth_headers, master_id) == live_group  # read back
    assert await _total(client, auth_headers, master_id) == live_total


async def test_a_series_chart_matches_a_live_read(client, auth_headers, test_tenant, monkeypatch):
    master_id = await _master(client, auth_headers, ROWS)
    body = {"source_type": "master", "source_id": master_id, "filters": [], "x_field": "status", "series": [{"aggregate": "count"}, {"field": "amount", "aggregate": "sum"}]}
    live = await client.post("/datasets/query-series", headers=auth_headers, json=body)
    assert live.status_code == 200, live.text

    monkeypatch.setattr(record_sql, "COUNT_CAP", 1)
    assert (await client.post("/datasets/query-series", headers=auth_headers, json=body)).json() == live.json()
    assert await _stored(test_tenant["tenant_id"]) == 1
    assert (await client.post("/datasets/query-series", headers=auth_headers, json=body)).json() == live.json()


async def test_a_fresh_result_is_served_until_it_goes_stale(client, auth_headers, test_tenant, large, monkeypatch):
    master_id = await _master(client, auth_headers, ROWS)
    assert await _total(client, auth_headers, master_id) == 3
    await client.post(f"/masters/{master_id}/records", headers=auth_headers, json={"field_values": {"status": "new", "amount": 1}})
    assert await _total(client, auth_headers, master_id) == 3  # within the stored result's life

    monkeypatch.setattr(settings, "summary_cache_seconds", 1)
    async with AsyncSessionLocal() as db, db.begin():
        await db.execute(text("UPDATE summary_cache SET computed_at = now() - interval '1 hour' WHERE tenant_id = :t"), {"t": test_tenant["tenant_id"]})
    assert await _total(client, auth_headers, master_id) == 3  # stale: still answers at once, and queues a refresh
    await jobs.run_due([summary_cache.JOB_KIND])
    monkeypatch.setattr(settings, "summary_cache_seconds", 900)
    assert await _total(client, auth_headers, master_id) == 4


async def test_too_slow_the_first_time_it_is_prepared_in_the_background(client, auth_headers, test_tenant, large, monkeypatch):
    master_id = await _master(client, auth_headers, ROWS)

    async def too_slow(db, sql):
        raise summary_cache._TooSlow

    monkeypatch.setattr(summary_cache, "_attempt", too_slow)
    refused = await client.post("/datasets/query", headers=auth_headers, json={"source_type": "master", "source_id": master_id, "columns": [], "filters": [], "limit": 1})
    assert refused.status_code == 503
    assert "being prepared" in refused.json()["detail"]

    await jobs.run_due([summary_cache.JOB_KIND])
    assert await _total(client, auth_headers, master_id) == 3


async def test_a_real_timeout_leaves_the_request_usable(client, auth_headers, test_tenant, large, monkeypatch):
    """Postgres really cancels the statement, in the savepoint, under the request's own session."""
    master_id = await _master(client, auth_headers, ROWS)
    real = summary_cache._run_raw

    async def slow(db, sql):
        await db.execute(text("SELECT pg_sleep(1)"))
        return await real(db, sql)

    monkeypatch.setattr(summary_cache, "_run_raw", slow)
    monkeypatch.setattr(summary_cache, "IN_REQUEST_MS", 100)
    refused = await client.post("/datasets/query", headers=auth_headers, json={"source_type": "master", "source_id": master_id, "columns": [], "filters": [], "limit": 1})
    assert refused.status_code == 503, refused.text
    assert "being prepared" in refused.json()["detail"]
    monkeypatch.setattr(summary_cache, "_run_raw", real)
    await jobs.run_due([summary_cache.JOB_KIND])
    assert await _total(client, auth_headers, master_id) == 3


async def test_different_filters_keep_their_own_results(client, auth_headers, large):
    master_id = await _master(client, auth_headers, ROWS)
    only_active = [{"field": "status", "operator": "equals", "value": "active"}]
    assert await _total(client, auth_headers, master_id) == 3
    assert await _total(client, auth_headers, master_id, filters=only_active) == 2
    assert await _total(client, auth_headers, master_id) == 3


async def test_results_stay_within_their_tenant(client, auth_headers, test_tenant, large):
    master_id = await _master(client, auth_headers, ROWS)
    await _total(client, auth_headers, master_id)
    from sqlalchemy.ext.asyncio import AsyncSession

    from app.core.database import engine

    async with AsyncSession(engine) as db, db.begin():
        await db.execute(text("SELECT set_config('app.current_tenant_id', :v, true)"), {"v": str(uuid.uuid4())})
        assert (await db.execute(text("SELECT count(*) FROM summary_cache"))).scalar_one() == 0


async def test_it_can_be_turned_off(client, auth_headers, test_tenant, large, monkeypatch):
    monkeypatch.setattr(settings, "summary_cache_seconds", 0)
    master_id = await _master(client, auth_headers, ROWS)
    assert await _total(client, auth_headers, master_id) == 3
    assert await _stored(test_tenant["tenant_id"]) == 0


def test_stored_values_keep_their_types():
    values = [Decimal("10.50"), None, True, 3, "x", uuid.uuid4(), {"a": 1}, ["a"], {"$": "other", "v": 1}]
    assert [summary_cache._decode(summary_cache._encode(v)) for v in values] == values