"""Rate limiting (token bucket, tiers, headers), API keys and streamed exports."""
from __future__ import annotations

import csv
import hashlib
import io
import json

import pytest
import test_api_support as support

from companyatlas.api.ratelimit import RateLimiter, flush_usage, resolve_api_key
from companyatlas.db import fetch_one, transaction

client = support.client
fixture_data = support.fixture_data
ADMIN = support.ADMIN
RAW_API_KEY = support.RAW_API_KEY

pytestmark = pytest.mark.asyncio(loop_scope="session")
V = "/api/v1"


async def test_token_bucket_semantics() -> None:
    rl = RateLimiter(limits={"anonymous": 3, "paid": 6, "internal": None})
    t = 1000.0
    results = [rl.take("ip:1", "anonymous", now=t) for _ in range(4)]
    assert [r[0] for r in results] == [True, True, True, False]
    assert results[0][1] == 3 and results[0][2] == 2 and results[3][2] == 0 and results[3][3] > 0
    allowed, *_ = rl.take("ip:1", "anonymous", now=t + 20)       # 3/min → one token back after 20 s
    assert allowed is True
    assert rl.take("ip:2", "anonymous", now=t)[0] is True         # independent bucket
    assert rl.take("key:x", "internal", now=t) == (True, 0, 0, 0.0)
    assert rl.take("key:y", "unknown-tier", now=t)[1] == 3         # unknown tier falls back to anonymous


async def test_rate_limit_headers_and_tiers(client):  # type: ignore[no-untyped-def]
    from companyatlas.api import ratelimit

    r = await client.get(f"{V}/methodology")
    assert r.headers["x-ratelimit-tier"] == "internal"                # loopback without X-Forwarded-For = our own SSR
    public = {"X-Forwarded-For": "203.0.113.7"}
    r = await client.get(f"{V}/methodology", headers=public)
    anon = ratelimit.limiter.limits["anonymous"]
    assert r.headers["x-ratelimit-tier"] == "anonymous" and r.headers["x-ratelimit-limit"] == str(anon) and int(r.headers["x-ratelimit-remaining"]) < anon
    assert ratelimit.TIER_LIMITS_PER_MIN == {"anonymous": 120, "authenticated": 600, "paid": 3000, "internal": None}
    r = await client.get(f"{V}/methodology", headers={"X-CA-API-Key": RAW_API_KEY})
    assert r.headers["x-ratelimit-tier"] == "paid" and r.headers["x-ratelimit-limit"] == "3000"
    r = await client.get(f"{V}/methodology", headers={"X-CA-API-Key": "ca_paid_not_a_real_key_000000", **public})
    assert r.headers["x-ratelimit-tier"] == "anonymous"
    r = await client.get(f"{V}/methodology", headers=ADMIN)
    assert r.headers["x-ratelimit-tier"] == "admin" and "x-ratelimit-remaining" not in r.headers
    r = await client.get("/health")
    assert "x-ratelimit-tier" not in r.headers                       # bypassed path
    info = await resolve_api_key(RAW_API_KEY)
    assert info and info["tier"] == "paid"
    await flush_usage()
    async with transaction() as conn:
        row = await fetch_one(conn, "select request_count, last_used_at from api_keys where key_hash = :h", h=hashlib.sha256(RAW_API_KEY.encode()).hexdigest())
    assert row and row["request_count"] >= 1 and row["last_used_at"] is not None


async def test_429_from_exhausted_bucket(client, monkeypatch):  # type: ignore[no-untyped-def]
    from companyatlas.api import ratelimit

    tiny = RateLimiter(limits={"anonymous": 2, "authenticated": 600, "paid": 3000, "internal": None})
    monkeypatch.setattr(ratelimit, "limiter", tiny)
    headers = {"X-Forwarded-For": "203.0.113.9, 10.0.0.1"}
    codes = [(await client.get(f"{V}/methodology", headers=headers)).status_code for _ in range(3)]
    assert codes == [200, 200, 429]
    r = await client.get(f"{V}/methodology", headers=headers)
    assert r.status_code == 429 and int(r.headers["retry-after"]) >= 1 and r.headers["x-ratelimit-remaining"] == "0" and r.json() == {"detail": "rate limit exceeded"}
    assert (await client.get(f"{V}/methodology", headers={"X-Forwarded-For": "203.0.113.10"})).status_code == 200


async def test_export_events_formats(client, fixture_data):  # type: ignore[no-untyped-def]
    r = await client.get(f"{V}/export/events.csv", params={"country": "ZZ"})
    assert r.status_code == 200 and r.headers["content-type"].startswith("text/csv") and "attachment" in r.headers["content-disposition"]
    rows = list(csv.reader(io.StringIO(r.text)))
    assert rows[0][:4] == ["id", "detected_at", "company_slug", "company_name"] and len(rows) == 5
    r = await client.get(f"{V}/export/events.ndjson", params={"country": "ZZ", "event_type": "PRICING"})
    lines = [json.loads(x) for x in r.text.splitlines() if x]
    assert r.headers["content-type"].startswith("application/x-ndjson") and len(lines) == 1 and lines[0]["id"] == fixture_data["ev_pricing"]
    r = await client.get(f"{V}/export/events.json", params={"company": fixture_data["alpha_slug"], "limit": 2})
    data = r.json()
    assert isinstance(data, list) and len(data) == 2 and data[0]["company"]["slug"] == fixture_data["alpha_slug"]
    assert (await client.get(f"{V}/export/events.xml")).status_code == 404
    assert (await client.get(f"{V}/export/events.json", params={"company": "nope"})).status_code == 404


async def test_export_companies_and_jobs(client, fixture_data):  # type: ignore[no-untyped-def]
    r = await client.get(f"{V}/export/companies.csv", params={"country": "ZZ"})
    rows = list(csv.reader(io.StringIO(r.text)))
    assert rows[0][0] == "id" and "activity_score" in rows[0] and len(rows) == 3
    r = await client.get(f"{V}/export/companies.json", params={"industry": fixture_data["industry"]})
    assert {c["slug"] for c in r.json()} == {fixture_data["alpha_slug"], fixture_data["beta_slug"]}
    r = await client.get(f"{V}/export/jobs.ndjson", params={"company": fixture_data["alpha_slug"], "status": "all"})
    lines = [json.loads(x) for x in r.text.splitlines() if x]
    assert len(lines) == 3 and all(j["company_slug"] == fixture_data["alpha_slug"] for j in lines)
    r = await client.get(f"{V}/export/jobs.csv", params={"company": fixture_data["alpha_slug"], "ai": 1})
    rows = list(csv.reader(io.StringIO(r.text)))
    assert len(rows) == 2 and rows[1][2] == "Senior ML Engineer"