"""Usage accounting: Redis minute counters → SQLite fold → series for the dashboard."""
from __future__ import annotations

from datetime import datetime, timezone


def test_record_fold_and_series(app):
    from accounts.models import UsageDaily, UsageMinute
    from core.db import session
    from ratelimit import redis_limiter as rl
    from ratelimit import usage

    rl.reset_for_tests()
    now = 1_800_000_000          # epoch s, minute-aligned
    p = "key:4242"
    usage.record(p, requests=1, rows=1000, status=200, ts_s=now - 180)
    usage.record(p, requests=2, rows=500, rows_parquet=500, bytes_=1234, status=200, ts_s=now - 180)
    usage.record(p, requests=0, status=429, ts_s=now - 120)
    usage.record(p, requests=1, rows=10, status=200, ts_s=now + 5)       # current minute: must stay live
    assert rl.client().scard(usage.PENDING_SET) == 3

    folded = usage.fold(now_s=now + 10)
    assert folded == 2
    assert rl.client().scard(usage.PENDING_SET) == 1          # the live minute remains
    with session() as s:
        m = s.get(UsageMinute, (datetime.fromtimestamp(now - 180, tz=timezone.utc).replace(tzinfo=None), p))
        assert m.requests == 3 and m.rows == 1500
        d = s.get(UsageDaily, (datetime.fromtimestamp(now - 180, tz=timezone.utc).date(), p))
        assert (d.requests, d.rows, d.rows_parquet, d.bytes, d.status_2xx, d.status_429) == (3, 1500, 500, 1234, 2, 1)
    # folding again is a no-op (idempotent)
    assert usage.fold(now_s=now + 10) == 0

    series = usage.usage_series([p], "24h", now_s=now + 10)
    assert series["step_seconds"] == 60 and len(series["points"]) == 1440
    assert series["totals"] == {"requests": 4, "rows": 1510, "status_429": 1, "bytes": 1234, "rows_parquet": 500}   # folded + live minute
    assert m.status_429 == 0 and m.rows_parquet == 500 and m.bytes == 1234   # usage_minute carries the new columns (the 429 is in the next minute)
    by_t = {pt["t"]: pt for pt in series["points"]}
    assert by_t[datetime.fromtimestamp(now - 180, tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")]["rows"] == 1500
    week = usage.usage_series([p], "7d", now_s=now + 10)
    assert week["step_seconds"] == 3600 and week["totals"]["requests"] == 4
    month = usage.usage_series([p], "30d", now_s=now + 10)
    assert month["step_seconds"] == 86400 and month["totals"]["rows"] == 1510
    assert usage.usage_series([], "24h", now_s=now)["totals"] == {"requests": 0, "rows": 0, "status_429": 0, "bytes": 0, "rows_parquet": 0}
    csv = usage.usage_csv(series).splitlines()
    assert csv[0] == "t,requests,rows,status_429,bytes,rows_parquet" and len(csv) == 1441
    assert any(line.endswith(",3,1500,0,1234,500") for line in csv[1:])
    assert any(t["principal"] == p and t["requests"] >= 3 for t in usage.top_principals(days=100_000))


def test_middleware_records_usage(client, make_user):
    from ratelimit import usage
    uid, key, _, _ = make_user()
    client.get("/v1/_test/frame?n=250", headers={"Authorization": f"Bearer {key}"})
    client.get("/v1/_test/frame?n=250&format=parquet", headers={"Authorization": f"Bearer {key}"})
    from accounts import service
    from core.db import session
    with session() as s:
        from accounts.models import User
        principals = [k.principal for k in service.list_keys(s, s.get(User, uid))]
    series = usage.usage_series(principals, "24h")
    assert (series["totals"]["requests"], series["totals"]["rows"], series["totals"]["rows_parquet"]) == (2, 375, 125)
    assert series["totals"]["bytes"] > 0 and series["totals"]["status_429"] == 0


def test_invalid_range_is_uniform_error(client, make_user):
    _, key, _, _ = make_user()
    r = client.get("/v1/me/usage?range=1y", headers={"Authorization": f"Bearer {key}"})
    assert r.status_code == 422 and r.json()["error"]["code"] == "VALIDATION_ERROR"


def test_quota_alerts_once_per_day_and_opt_out(app, make_user, outbox):
    """B4: 80 % / 100 % of the rows window or a 429 → one mail per user per day; opt-out honoured."""
    from accounts.models import User
    from core.db import session
    from ratelimit import redis_limiter as rl
    from ratelimit import usage
    from ratelimit.tiers import TIERS
    uid, key, email, _ = make_user()
    with session() as s:
        from accounts import service
        principal = service.active_keys(s, s.get(User, uid))[0].principal
    rl.reset_for_tests()
    free = TIERS["free"]
    rl.apply_tier(principal, free, req_cost=1, rows_cost=int(free.rows * 0.85), force=True)   # 85 % of the window
    usage.record(principal, requests=1, rows=10, status=200)                                     # "recent" principal
    assert usage.quota_alerts() == 1
    alert = [m for m in outbox if m.kind == "quota_alert"][-1]
    assert alert.to == email and "80 %" in alert.ctx["what"]
    assert usage.quota_alerts() == 0                                                              # deduped for today
    with session() as s:
        s.get(User, uid).quota_alerts = 0
    rl._call(lambda: rl.client().delete(*[k for k in rl.client().keys("alert:*")]) if rl.client().keys("alert:*") else None)
    assert usage.quota_alerts() == 0                                                              # opt-out
    with session() as s:
        s.get(User, uid).quota_alerts = 1
    usage.record(principal, requests=0, status=429)
    assert usage.quota_alerts() == 1
    alert = [m for m in outbox if m.kind == "quota_alert"][-1]
    assert "100 %" in alert.ctx["what"] or "429" in alert.ctx["what"] or "80 %" in alert.ctx["what"]