"""ASGI middleware contract: headers on every response, 429 envelope, keyless vs key, costs, key gating, fail-open."""
from __future__ import annotations

HDRS = ["X-RateLimit-Limit-Requests", "X-RateLimit-Remaining-Requests", "X-RateLimit-Limit-Rows",
        "X-RateLimit-Remaining-Rows", "X-RateLimit-Reset"]


def ip(n: int) -> dict:
    """The proxy (ngrok) APPENDS the real peer: the last hop is the client, the first one is whatever it claimed."""
    return {"X-Forwarded-For": f"172.16.0.1, 10.0.0.{n}"}


def test_headers_on_success_and_error_responses(client):
    r = client.get("/v1/status", headers=ip(1))
    assert r.status_code == 200
    for h in HDRS:
        assert h in r.headers, h
    assert r.headers["X-RateLimit-Limit-Requests"] == "30" and r.headers["X-RateLimit-Remaining-Requests"] == "29"
    assert r.headers["X-RateLimit-Limit-Rows"] == "100000"
    r = client.get("/v1/bars/stock/NOPE?timeframe=1day", headers=ip(1))
    assert r.status_code == 404 and r.headers["X-RateLimit-Remaining-Requests"] == "28"
    assert "X-RateLimit-Reset" in r.headers
    # exempt paths carry no counters and are not charged
    r = client.get("/health", headers=ip(1))
    assert r.status_code == 200 and "X-RateLimit-Limit-Requests" not in r.headers
    assert client.get("/v1/status", headers=ip(1)).headers["X-RateLimit-Remaining-Requests"] == "27"


def test_rows_are_charged_from_x_row_count(client):
    r = client.get("/v1/bars/stock/AAPL?timeframe=1day&limit=100", headers=ip(2))
    assert r.headers["X-Row-Count"] == "100"
    assert r.headers["X-RateLimit-Remaining-Rows"] == str(100_000 - 100)
    r = client.get("/v1/_test/frame?n=1000", headers=ip(2))
    assert r.headers["X-RateLimit-Remaining-Rows"] == str(100_000 - 1100)


def test_parquet_costs_half_and_csv_full(client):
    r = client.get("/v1/_test/frame?n=1001&format=parquet", headers=ip(3))
    assert r.status_code == 200 and r.headers["content-type"].startswith("application/vnd.apache.parquet")
    assert r.headers["X-RateLimit-Remaining-Rows"] == str(100_000 - 501)
    r = client.get("/v1/_test/frame?n=1001&format=csv", headers=ip(3))
    assert r.headers["X-RateLimit-Remaining-Rows"] == str(100_000 - 501 - 1001)


def test_quota_exempt_and_request_cost(client):
    r = client.get("/v1/_test/exempt?n=5000", headers=ip(4))
    assert r.headers["X-Row-Count"] == "5000" and r.headers["X-RateLimit-Remaining-Rows"] == "100000"
    assert r.headers["X-RateLimit-Remaining-Requests"] == "29"
    r = client.get("/v1/_test/expensive", headers=ip(4))
    assert r.headers["X-RateLimit-Remaining-Requests"] == "27"   # cost 2


def test_keyless_429_envelope_retry_after_and_upgrade_hint(client):
    for _ in range(30):
        assert client.get("/v1/status", headers=ip(5)).status_code == 200
    r = client.get("/v1/status", headers=ip(5))
    assert r.status_code == 429
    body = r.json()
    assert body["error"]["code"] == "RATE_LIMIT_EXCEEDED" and body["error"]["type"] == "requests_per_hour"
    assert "free account" in body["error"]["message"] and body["detail"]
    assert int(r.headers["Retry-After"]) >= 1
    assert r.headers["X-RateLimit-Remaining-Requests"] == "0"
    assert body["error"]["docs"].endswith("#rate_limit_exceeded")
    # another IP is unaffected
    assert client.get("/v1/status", headers=ip(6)).status_code == 200


def test_keyless_rows_per_hour_429(client):
    for _ in range(20):
        assert client.get("/v1/_test/frame?n=5000", headers=ip(7)).status_code == 200
    r = client.get("/v1/status", headers=ip(7))
    assert r.status_code == 429 and r.json()["error"]["type"] == "rows_per_hour"
    assert r.headers["X-RateLimit-Remaining-Rows"] == "0"


def test_key_gets_free_tier_limits_and_query_param_works(client, make_user):
    _, key, _, _ = make_user()
    r = client.get("/v1/status", headers={**ip(8), "Authorization": f"Bearer {key}"})
    assert r.status_code == 200
    assert r.headers["X-RateLimit-Limit-Requests"] == "120" and r.headers["X-RateLimit-Limit-Rows"] == "1000000"
    assert r.headers["X-RateLimit-Remaining-Requests"] == "119"
    r = client.get(f"/v1/status?api_key={key}", headers=ip(8))
    assert r.headers["X-RateLimit-Remaining-Requests"] == "118"
    # keyless traffic from the same IP is a different principal
    assert client.get("/v1/status", headers=ip(8)).headers["X-RateLimit-Limit-Requests"] == "30"


def test_high_usage_tier(client, make_user):
    _, key, _, _ = make_user(tier="high_usage")
    r = client.get("/v1/status", headers={"Authorization": f"Bearer {key}"})
    assert r.headers["X-RateLimit-Limit-Requests"] == "600" and r.headers["X-RateLimit-Limit-Rows"] == "10000000"


def test_free_429_type_is_per_minute(client, make_user, monkeypatch):
    _, key, _, _ = make_user()
    from ratelimit import redis_limiter as rl
    t0 = 1_800_000_000_000
    monkeypatch.setattr(rl, "now_ms", lambda: t0)
    h = {"Authorization": f"Bearer {key}"}
    for _ in range(120):
        assert client.get("/v1/status", headers=h).status_code == 200
    r = client.get("/v1/status", headers=h)
    assert r.status_code == 429 and r.json()["error"]["type"] == "requests_per_minute"
    assert r.headers["Retry-After"] == "60"
    assert "high-usage" in r.json()["error"]["message"]
    monkeypatch.setattr(rl, "now_ms", lambda: t0 + 61_000)      # window rolled over
    assert client.get("/v1/status", headers=h).status_code == 200


def test_invalid_key_is_401_not_keyless(client):
    r = client.get("/v1/status", headers={"Authorization": "Bearer hfmd_live_doesnotexist000000000000000000"})
    assert r.status_code == 401 and r.json()["error"]["code"] == "INVALID_API_KEY"
    r = client.get("/v1/status?api_key=garbage")
    assert r.status_code == 401


def test_keyless_cannot_use_key_required_endpoints(client, make_user):
    r = client.get("/v1/_test/tagged", headers=ip(9))          # route tagged `stream`
    assert r.status_code == 401 and r.json()["error"]["code"] == "AUTH_REQUIRED"
    assert "free account" in r.json()["error"]["message"]
    r = client.get("/v1/_test/needkey", headers=ip(9))         # handler sets request.state.requires_key
    assert r.status_code == 401 and r.json()["error"]["code"] == "AUTH_REQUIRED"
    assert "secret" not in r.text
    _, key, _, _ = make_user()
    assert client.get("/v1/_test/tagged", headers={"Authorization": f"Bearer {key}"}).status_code == 200
    assert client.get("/v1/_test/needkey", headers={"Authorization": f"Bearer {key}"}).json()["data"]["secret"] is True


def test_rows_cap_per_request(client, make_user):
    r = client.get("/v1/_test/frame?n=10&limit=5001", headers=ip(10))
    assert r.status_code == 400
    body = r.json()
    assert body["error"]["code"] == "ROW_LIMIT_EXCEEDED" and body["error"]["details"]["max_rows"] == 5000
    assert client.get("/v1/_test/frame?n=10&limit=5000", headers=ip(10)).status_code == 200
    _, key, _, _ = make_user()
    assert client.get("/v1/_test/frame?n=10&limit=5001", headers={"Authorization": f"Bearer {key}"}).status_code == 200
    r = client.get("/v1/_test/frame?n=10&limit=50001", headers={"Authorization": f"Bearer {key}"})
    assert r.status_code == 400 and r.json()["error"]["details"]["max_rows"] == 50000


def test_auth_endpoints_throttled_per_ip(client):
    h = {**ip(11), "Content-Type": "application/json"}
    for _ in range(20):
        r = client.post("/v1/auth/login", json={"email": "nobody@example.com", "password": "x"}, headers=h)
        assert r.status_code == 401
    r = client.post("/v1/auth/login", json={"email": "nobody@example.com", "password": "x"}, headers=h)
    assert r.status_code == 429
    assert r.json()["error"]["type"] == "requests_per_hour" and "Retry-After" in r.headers
    assert "authentication attempts" in r.json()["error"]["message"]
    # the data quota of that IP is untouched
    assert client.get("/v1/status", headers=ip(11)).headers["X-RateLimit-Remaining-Requests"] == "29"


def test_unhandled_exception_gets_envelope_and_headers(client):
    r = client.get("/v1/_test/boom", headers=ip(12))
    assert r.status_code == 500 and r.json()["error"]["code"] == "INTERNAL_ERROR"
    assert "X-RateLimit-Remaining-Requests" in r.headers


def test_fail_open_when_redis_down(client, monkeypatch):
    import redis as redis_lib

    from ratelimit import redis_limiter as rl

    class Broken:
        def __getattr__(self, name):
            def _raise(*a, **k):
                raise redis_lib.exceptions.ConnectionError("down")
            return _raise

    monkeypatch.setattr(rl, "_client", Broken())
    monkeypatch.setattr(rl, "_sha", None)
    monkeypatch.setattr(rl, "_down_until", 0.0)
    r = client.get("/v1/status", headers=ip(13))
    assert r.status_code == 200 and "X-RateLimit-Limit-Requests" not in r.headers
    r = client.get("/v1/limits", headers=ip(13))
    assert r.status_code == 200 and r.json()["data"]["principal"]["redis"] is False


def test_limits_endpoint_public_and_uncharged(client, make_user):
    r = client.get("/v1/limits", headers=ip(14))
    assert r.status_code == 200
    d = r.json()["data"]
    assert set(d["tiers"]) == {"keyless", "free", "high_usage"}
    assert d["tiers"]["free"]["requests"] == 120 and d["tiers"]["keyless"]["max_rows_per_request"] == 5000
    assert d["principal"]["kind"] == "keyless" and d["principal"]["requests"]["remaining"] == 30
    assert client.get("/v1/limits", headers=ip(14)).json()["data"]["principal"]["requests"]["remaining"] == 30
    _, key, _, _ = make_user()
    client.get("/v1/status", headers={"Authorization": f"Bearer {key}"})
    d = client.get("/v1/limits", headers={"Authorization": f"Bearer {key}"}).json()["data"]
    assert d["principal"]["tier"] == "free" and d["principal"]["requests"]["remaining"] == 119


def test_openapi_lists_new_routes_with_errors(client):
    spec = client.get("/openapi.json").json()
    op = spec["paths"]["/v1/auth/signup"]["post"]
    assert op["summary"] and "202" in op["responses"] and "400" in op["responses"] and "WEAK_PASSWORD" in op["responses"]["400"]["description"]
    assert "429" in op["responses"]
    op = spec["paths"]["/v1/admin/users"]["post"]
    assert "409" in op["responses"] and "EMAIL_TAKEN" in op["responses"]["409"]["description"]
    assert "X-RateLimit-Limit-Requests" in spec["paths"]["/v1/limits"]["get"]["responses"]["200"]["headers"]
    assert "EMAIL_TAKEN" in spec["components"]["schemas"]["Error"]["properties"]["error"]["properties"]["code"]["enum"]
    assert spec["paths"]["/v1/me/keys"]["post"]["responses"]["201"]["content"]["application/json"]["example"]["data"]["key"].startswith("hfmd_live_")


def test_browser_session_gets_account_tier_on_data_endpoints(web, make_user, signin):
    """A signed-in browser (cookie, no Bearer) is the `user:<id>` principal with the account's tier — the charts
    page and the playground get 120 req/min without pasting a key. Anonymous cookies stay keyless."""
    uid, _raw, email, pw = make_user(tier="free", with_key=False)
    r = web.get("/v1/status", headers=ip(41))
    assert r.headers["X-RateLimit-Limit-Requests"] == "30"                      # keyless before sign-in
    signin(web, email, pw)
    r = web.get("/v1/bars/stock/AAPL?timeframe=1day&limit=5", headers=ip(41))
    assert r.status_code == 200
    assert r.headers["X-RateLimit-Limit-Requests"] == "120" and r.headers["X-RateLimit-Limit-Rows"] == "1000000"
    lim = web.get("/v1/limits", headers=ip(41)).json()["data"]["principal"]
    assert lim["principal"] == f"user:{uid}" and lim["kind"] == "session" and lim["tier"] == "free"
    # the session's usage is part of the account's dashboard series
    me = web.get("/v1/me/usage?range=24h", headers={"X-Requested-With": "hfmd"}).json()["data"]
    assert f"user:{uid}" in me["principals"]
    # a Bearer key still wins over the cookie
    uid2, raw2, _, _ = make_user(tier="high_usage")
    r = web.get("/v1/status", headers={**ip(41), "Authorization": f"Bearer {raw2}"})
    assert r.headers["X-RateLimit-Limit-Requests"] == "600"
    # sign out everywhere → the OLD cookie (another device) is stale → keyless again; the current browser is refreshed
    old_cookie = web.cookies.get("hfmd_session")
    r = web.post("/v1/me/sessions/revoke-all", json={}, headers={"Content-Type": "application/json", "X-Requested-With": "hfmd"})
    assert r.status_code in (200, 204), r.text
    from fastapi.testclient import TestClient
    with TestClient(web.app) as other:
        r = other.get("/v1/status", headers={**ip(41), "Cookie": f"hfmd_session={old_cookie}"})
        assert r.headers["X-RateLimit-Limit-Requests"] == "30"
        assert other.get("/v1/me", headers={"Cookie": f"hfmd_session={old_cookie}"}).status_code == 401
    assert web.get("/v1/status", headers=ip(41)).headers["X-RateLimit-Limit-Requests"] == "120"