ratelimit: IP client = dernier hop de confiance de X-Forwarded-For (HFMD_TRUSTED_PROXY_HOPS), cache clés LRU borné + génération Redis keys:version, throttle auth par endpoint fail-closed, throttle des recherches de clés inconnues, clés expirées refusées, séries usage étendues (429/bytes/parquet), export CSV, alertes quota quotidiennes

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

Simon-Pierre Boucher committed 18 days ago (Sep 6, 2026) parent 4809bf8

4 changed files +311 −86

modified hfmarketdata/api/ratelimit/middleware.py +160 −61

		`@@ -1,29 +1,38 @@`
1	1	"""Pure ASGI rate-limit middleware (`install(app)`).
2	2
3	3	Per request under `/v1/`:
4		−1. resolve the principal — `Authorization: Bearer hfmd_live_…` or `?api_key=` (hashed lookup, 60 s
5		− in-process cache) → `key:<id>` with the key/user tier; otherwise the client IP (first hop of
6		− `X-Forwarded-For`, ngrok fronts the API) hashed with `settings.key_hash_salt` → `ip:<hash>`, tier keyless;
	4	+1. resolve the principal — `Authorization: Bearer hfmd_live_…` (or the discouraged `?api_key=`): hashed lookup
	5	+ through a bounded LRU cache (10 000 entries, 60 s TTL, negatives cached too, cross-worker invalidation via
	6	+ the Redis `keys:version` generation) → `key:<id>` with the key/user tier; otherwise the client IP hashed
	7	+ with `settings.key_hash_salt` → `ip:<hash>`, tier keyless. Cache misses are throttled per IP before SQLite
	8	+ is asked, so unknown keys cost the same as known ones;
7	9	2. expose `request.state.principal / principal_kind / tier / max_rows / user_id / api_key_id / ratelimit`;
8	10	3. keyless principals are refused (401 AUTH_REQUIRED) on routes tagged `stream`/`screener`, and on any
9	11	response whose handler set `request.state.requires_key = True`;
10		−4. `/v1/auth/*` is throttled per IP (10 / hour) regardless of key;
	12	+4. `/v1/auth/*` is throttled per IP with one bucket per endpoint (login 20/h, signup 5/h, forgot 5/h, verify
	13	+ 30/h, …); login/signup/forgot fail CLOSED when Redis is down (429, Retry-After 30);
11	14	5. reserve 1 request in the Redis sliding window before the handler (429 when exhausted, with `Retry-After`
12	15	and `error.type` ∈ requests_per_hour \| rows_per_hour \| requests_per_minute \| rows_per_minute), then charge
13	16	the rows read from `X-Row-Count` (Parquet ½, quota_exempt/304 → 0) and the extra `request_cost` after it;
14	17	6. `X-RateLimit-*` headers on every response, usage counters recorded for the dashboard.
15	18
	19	+Client IP: ngrok (and any sane reverse proxy) APPENDS the peer address to `X-Forwarded-For`, so the trustworthy
	20	+value is the N-th hop from the END (`HFMD_TRUSTED_PROXY_HOPS`, default 1), never the first one, which the client
	21	+controls. The header is ignored altogether when the direct peer is not a loopback / private address.
	22	+
16	23	Account endpoints (`/v1/limits`, `/v1/me/`, `/v1/admin/`) are not charged. `/health`, `/openapi.json`
17		−and the SPA are exempt. Redis down → fail-open (no headers), the API never goes down because of quotas.
	24	+and the SPA are exempt. Redis down → data endpoints fail open (no headers), the API never goes down because of
	25	+quotas.
18	26
19	27	Author: Simon-Pierre Boucher <contact@spboucher.ai>
20	28	"""
21	29	from __future__ import annotations
22	30
23		−import hashlib
	31	+import ipaddress
24	32	import logging
25	33	import threading
26	34	import time
	35	+from collections import OrderedDict
27	36	from dataclasses import dataclass
28	37	from datetime import datetime, timezone
29	38	from typing import Any
		`@@ -38,12 +47,14 @@ from core.errors import ApiError`
38	47
39	48	from . import redis_limiter as rl
40	49	from . import usage
41		−from .tiers import AUTH_REQUESTS_PER_WINDOW, AUTH_WINDOW_S, KEY_REQUIRED_TAGS, TIERS, Tier, row_cost, tier_for, upgrade_hint
	50	+from .tiers import (AUTH_DEFAULT_LIMIT, AUTH_FAIL_CLOSED, AUTH_FAIL_CLOSED_RETRY_S, AUTH_LIMITS, AUTH_WINDOW_S,
	51	+ KEY_LOOKUPS_PER_MINUTE_PER_IP, KEY_REQUIRED_TAGS, TIERS, Tier, row_cost, tier_for, upgrade_hint)
42	52
43	53	log = logging.getLogger("hfmarketdata.ratelimit")
44	54
45	55	KEY_PREFIX = "hfmd_live_"
46	56	KEY_CACHE_TTL_S = 60
	57	+KEY_CACHE_MAX = 10_000
47	58	EXEMPT_PREFIXES = ("/health", "/openapi.json", "/docs", "/redoc")
48	59	UNCHARGED_PREFIXES = ("/v1/limits", "/v1/me", "/v1/admin")
49	60	AUTH_PREFIX = "/v1/auth/"
		`@@ -59,7 +70,7 @@ class KeyInfo:`
59	70	key_id: int
60	71	user_id: int
61	72	tier: str
62		− key_status: str
	73	+ key_status: str # active \| revoked \| expired
63	74	user_status: str
64	75	email: str
65	76
		`@@ -77,8 +88,47 @@ class Principal:`
77	88	return self.kind == "keyless"
78	89
79	90
80		−_key_cache: dict[str, tuple[float, KeyInfo \| None]] = {}
81		−_key_cache_lock = threading.Lock()
	91	+class _LRU:
	92	+ """Tiny thread-safe LRU with per-entry expiry and a global generation number (Redis `keys:version`)."""
	93	+
	94	+ def __init__(self, maxsize: int, ttl_s: float) -> None:
	95	+ self.maxsize, self.ttl_s = maxsize, ttl_s
	96	+ self._d: OrderedDict[str, tuple[float, int \| None, Any]] = OrderedDict()
	97	+ self._lock = threading.Lock()
	98	+
	99	+ def get(self, key: str, generation: int \| None) -> tuple[bool, Any]:
	100	+ now = time.time()
	101	+ with self._lock:
	102	+ hit = self._d.get(key)
	103	+ if hit is None:
	104	+ return False, None
	105	+ exp, gen, value = hit
	106	+ if exp <= now or (generation is not None and gen is not None and gen != generation):
	107	+ self._d.pop(key, None)
	108	+ return False, None
	109	+ self._d.move_to_end(key)
	110	+ return True, value
	111	+
	112	+ def put(self, key: str, value: Any, generation: int \| None) -> None:
	113	+ with self._lock:
	114	+ self._d[key] = (time.time() + self.ttl_s, generation, value)
	115	+ self._d.move_to_end(key)
	116	+ while len(self._d) > self.maxsize:
	117	+ self._d.popitem(last=False)
	118	+
	119	+ def pop(self, key: str) -> None:
	120	+ with self._lock:
	121	+ self._d.pop(key, None)
	122	+
	123	+ def clear(self) -> None:
	124	+ with self._lock:
	125	+ self._d.clear()
	126	+
	127	+ def __len__(self) -> int:
	128	+ return len(self._d)
	129	+
	130	+
	131	+_key_cache = _LRU(KEY_CACHE_MAX, KEY_CACHE_TTL_S)
82	132
83	133
84	134	def hash_key(raw_key: str) -> str:
		`@@ -87,19 +137,21 @@ def hash_key(raw_key: str) -> str:`
87	137
88	138
89	139	def hash_ip(ip: str) -> str:
90		− return hashlib.sha256((settings.key_hash_salt + "\|ip\|" + ip).encode()).hexdigest()[:16]
	140	+ from accounts.security import hash_ip as _hip # single formula, shared with api_keys.last_used_ip
	141	+ return _hip(ip)
91	142
92	143
93	144	def invalidate_key_cache(raw_key_hash: str \| None = None) -> None:
94		− """Drop cached lookups (call after revoke / rotate / tier or status change)."""
95		− with _key_cache_lock:
96		− if raw_key_hash is None:
97		− _key_cache.clear()
98		− else:
99		− _key_cache.pop(raw_key_hash, None)
	145	+ """Drop cached lookups on this worker AND bump the Redis generation so other workers drop theirs too
	146	+ (call after revoke / rotate / tier or status change)."""
	147	+ if raw_key_hash is None:
	148	+ _key_cache.clear()
	149	+ else:
	150	+ _key_cache.pop(raw_key_hash)
	151	+ rl.bump_keys_version()
100	152
101	153
102		−def _lookup_key(key_hash: str) -> KeyInfo \| None:
	154	+def _lookup_key(key_hash: str, ip_hash: str \| None = None) -> KeyInfo \| None:
103	155	from sqlalchemy import select
104	156
105	157	from accounts.models import ApiKey, User
		`@@ -110,21 +162,32 @@ def _lookup_key(key_hash: str) -> KeyInfo \| None:`
110	162	if row is None:
111	163	return None
112	164	k, u = row
113		− if k.status == "active":
	165	+ status = k.status
	166	+ if status == "active" and k.is_expired():
	167	+ status = "expired"
	168	+ if status == "active":
114	169	k.last_used_at = datetime.now(timezone.utc).replace(tzinfo=None)
115		− return KeyInfo(k.id, u.id, k.tier_override or u.tier, k.status, u.status, u.email)
	170	+ if ip_hash:
	171	+ k.last_used_ip = ip_hash
	172	+ return KeyInfo(k.id, u.id, k.tier_override or u.tier, status, u.status, u.email)
116	173
117	174
118		−def resolve_key(raw_key: str) -> KeyInfo \| None:
	175	+def resolve_key(raw_key: str, *, ip: str \| None = None) -> KeyInfo \| None:
	176	+ """Cached lookup. On a miss the IP is throttled (KEY_LOOKUPS_PER_MINUTE_PER_IP) before SQLite is queried,
	177	+ so guessing keys costs the attacker a 429 well before it costs us a database round-trip."""
119	178	h = hash_key(raw_key)
120		− now = time.time()
121		− with _key_cache_lock:
122		− hit = _key_cache.get(h)
123		− if hit and hit[0] > now:
124		− return hit[1]
125		− info = _lookup_key(h)
126		− with _key_cache_lock:
127		− _key_cache[h] = (now + KEY_CACHE_TTL_S, info)
	179	+ gen = rl.keys_version()
	180	+ hit, info = _key_cache.get(h, gen)
	181	+ if hit:
	182	+ return info
	183	+ if ip:
	184	+ d = rl.throttle(f"keylookup:ip:{hash_ip(ip)}", limit=KEY_LOOKUPS_PER_MINUTE_PER_IP, window_s=60)
	185	+ if d is not None and not d.allowed_requests:
	186	+ retry = max(1, d.reset_requests - rl.now_ms() // 1000)
	187	+ raise ApiError(429, "RATE_LIMIT_EXCEEDED", f"Too many unknown API keys from this address. Retry in {retry} s.",
	188	+ type="requests_per_minute", headers={"Retry-After": str(retry)})
	189	+ info = _lookup_key(h, hash_ip(ip) if ip else None)
	190	+ _key_cache.put(h, info, gen)
128	191	return info
129	192
130	193
		`@@ -132,37 +195,53 @@ def extract_key(headers: Headers, query: QueryParams) -> str \| None:`
132	195	auth = headers.get("authorization", "")
133	196	if auth[:7].lower() == "bearer ":
134	197	return auth[7:].strip() or None
135		− return query.get("api_key") or None
	198	+ return query.get("api_key") or None # supported for compatibility; discouraged (ends up in logs / history)
	199	+
	200	+
	201	+def _is_trusted_peer(host: str \| None) -> bool:
	202	+ """Only loopback / private / link-local peers (our reverse proxy) may set X-Forwarded-For.
	203	+ Non-IP hosts (unix sockets, the ASGI test client) are treated as local."""
	204	+ if not host:
	205	+ return True
	206	+ try:
	207	+ ip = ipaddress.ip_address(host)
	208	+ except ValueError:
	209	+ return True
	210	+ return ip.is_loopback or ip.is_private or ip.is_link_local
136	211
137	212
138	213	def client_ip(scope: dict, headers: Headers) -> str:
139		− xff = headers.get("x-forwarded-for")
140		− if xff:
141		− first = xff.split(",")[0].strip()
142		− if first:
143		− return first
144		− real = headers.get("x-real-ip")
145		− if real:
146		− return real.strip()
147	214	client = scope.get("client")
148		− return client[0] if client else "unknown"
	215	+ peer = client[0] if client else None
	216	+ hops = settings.trusted_proxy_hops
	217	+ if hops > 0 and _is_trusted_peer(peer):
	218	+ xff = [p.strip() for p in headers.get("x-forwarded-for", "").split(",") if p.strip()]
	219	+ if xff:
	220	+ # the last `hops` entries were appended by our proxies; the one just before them is the client
	221	+ idx = len(xff) - hops
	222	+ return xff[idx] if idx >= 0 else xff[0]
	223	+ real = headers.get("x-real-ip")
	224	+ if real:
	225	+ return real.strip()
	226	+ return peer or "unknown"
149	227
150	228
151	229	def resolve_principal(scope: dict) -> Principal:
152		− """Raises ApiError(401 INVALID_API_KEY / 403 ACCOUNT_DISABLED) for a bad key; never for a missing one."""
	230	+ """Raises ApiError(401 INVALID_API_KEY / 403 ACCOUNT_DISABLED / 429) for a bad key; never for a missing one."""
153	231	headers = Headers(scope=scope)
154	232	query = QueryParams(scope.get("query_string", b""))
155	233	raw = extract_key(headers, query)
	234	+ ip = client_ip(scope, headers)
156	235	if raw:
157		− info = resolve_key(raw) if raw.startswith(KEY_PREFIX) else None
	236	+ info = resolve_key(raw, ip=ip) if raw.startswith(KEY_PREFIX) else None
158	237	if info is None or info.key_status != "active":
159		− raise ApiError(401, "INVALID_API_KEY", "The API key is unknown, malformed or revoked. Check it in your "
	238	+ what = "expired" if info is not None and info.key_status == "expired" else "unknown, malformed or revoked"
	239	+ raise ApiError(401, "INVALID_API_KEY", f"The API key is {what}. Check it in your "
160	240	f"dashboard ({settings.public_url}/dashboard) or omit it to use keyless access.")
161		− if info.user_status == "disabled": # `invited` accounts (created by an admin) can use their key right away
	241	+ if info.user_status in ("disabled", "deleted"): # `invited` accounts (created by an admin) can use their key right away
162	242	raise ApiError(403, "ACCOUNT_DISABLED", "This API key belongs to a disabled account. "
163	243	f"Contact {settings.contact_email}.")
164	244	return Principal(f"key:{info.key_id}", "key", tier_for(info.tier), info.user_id, info.key_id)
165		− ip = client_ip(scope, headers)
166	245	return Principal(f"ip:{hash_ip(ip)}", "keyless", TIERS["keyless"])
167	246
168	247
		`@@ -235,6 +314,31 @@ def _is_charged(path: str) -> bool:`
235	314	return not path.startswith(UNCHARGED_PREFIXES) and not path.startswith(AUTH_PREFIX)
236	315
237	316
	317	+def auth_endpoint(path: str) -> str:
	318	+ """`/v1/auth/login` → `login` (first segment after the prefix)."""
	319	+ return path[len(AUTH_PREFIX):].split("/")[0].split("?")[0]
	320	+
	321	+
	322	+def auth_throttle(path: str, ip: str, *, now_s: int) -> tuple[ApiError \| None, dict[str, str]]:
	323	+ """Per-endpoint, per-IP hourly throttle for /v1/auth/*. Returns (error or None, extra headers)."""
	324	+ ep = auth_endpoint(path)
	325	+ limit = AUTH_LIMITS.get(ep, AUTH_DEFAULT_LIMIT)
	326	+ d = rl.throttle(f"auth:{ep}:ip:{hash_ip(ip)}", limit=limit, window_s=AUTH_WINDOW_S, namespace="rl")
	327	+ if d is None:
	328	+ if ep in AUTH_FAIL_CLOSED:
	329	+ return ApiError(429, "RATE_LIMIT_EXCEEDED", "Authentication is temporarily unavailable (rate limiter "
	330	+ f"offline). Retry in {AUTH_FAIL_CLOSED_RETRY_S} s.", type="requests_per_hour",
	331	+ headers={"Retry-After": str(AUTH_FAIL_CLOSED_RETRY_S)}), {}
	332	+ return None, {}
	333	+ if not d.allowed_requests:
	334	+ retry = max(1, d.reset_requests - now_s)
	335	+ return ApiError(429, "RATE_LIMIT_EXCEEDED", f"Too many authentication attempts: {limit} per hour per IP "
	336	+ f"for {ep}. Retry in {retry} s.", type="requests_per_hour",
	337	+ headers={**d.headers(), "Retry-After": str(retry)},
	338	+ details={"limit": limit, "window_seconds": AUTH_WINDOW_S, "reset": d.reset_requests, "endpoint": ep}), {}
	339	+ return None, d.headers()
	340	+
	341	+
238	342	# ----------------------------------------------------------------------------------------------------
239	343	# Middleware
240	344	# ----------------------------------------------------------------------------------------------------
		`@@ -254,6 +358,9 @@ class RateLimitMiddleware:`
254	358	usage.start_background()
255	359	state: dict = scope.setdefault("state", {})
256	360	now_s = rl.now_ms() // 1000
	361	+ headers = Headers(scope=scope)
	362	+ ip = client_ip(scope, headers)
	363	+ state["client_ip_hash"] = hash_ip(ip)
257	364
258	365	try:
259	366	principal = resolve_principal(scope)
		`@@ -266,19 +373,13 @@ class RateLimitMiddleware:`
266	373	if principal.keyless and (route_tags(scope) & KEY_REQUIRED_TAGS):
267	374	return await _auth_required_error().response()(scope, receive, send)
268	375
269		− # /v1/auth/* : per-IP throttle (10 / hour), whatever the key
	376	+ # /v1/auth/* : per-IP throttle, one bucket per endpoint, whatever the key
270	377	if path.startswith(AUTH_PREFIX):
	378	+ extra: dict[str, str] = {}
271	379	if path not in AUTH_UNTHROTTLED:
272		− ip_principal = f"auth:ip:{hash_ip(client_ip(scope, Headers(scope=scope)))}"
273		− auth_tier = Tier("auth", "ip", AUTH_WINDOW_S, AUTH_REQUESTS_PER_WINDOW, 10**9, 0)
274		− d = rl.apply_tier(ip_principal, auth_tier, req_cost=1)
275		− if d is not None and not d.allowed_requests:
276		− err = _rate_limited_error(d, auth_tier, principal, now_s=now_s, scope_label="authentication endpoints (per IP)")
277		− err.message = f"Too many authentication attempts: {AUTH_REQUESTS_PER_WINDOW} per hour per IP. Retry in {err.headers['Retry-After']} s."
	380	+ err, extra = auth_throttle(path, ip, now_s=now_s)
	381	+ if err is not None:
278	382	return await err.response()(scope, receive, send)
279		− extra = d.headers() if d is not None else {}
280		− else:
281		− extra = {}
282	383	return await self._forward(scope, receive, send, principal, tier, None, charged=False, extra_headers=extra)
283	384
284	385	charged = _is_charged(path)
		`@@ -376,11 +477,10 @@ def _tier_for_principal(principal: str):`
376	477	from . import tiers as _tiers
377	478	if not principal.startswith("key:"):
378	479	return _tiers.TIERS["keyless"]
379		− now = time.time()
380		− with _key_cache_lock:
381		− hit = _key_cache.get(principal)
382		− if hit and hit[0] > now:
383		− return hit[1]
	480	+ gen = rl.keys_version()
	481	+ hit, tier = _key_cache.get(principal, gen)
	482	+ if hit:
	483	+ return tier
384	484	tier_name = "free"
385	485	try:
386	486	from sqlalchemy import select
		`@@ -394,8 +494,7 @@ def _tier_for_principal(principal: str):`
394	494	except Exception:
395	495	pass
396	496	tier = _tiers.TIERS.get(tier_name, _tiers.TIERS["free"])
397		− with _key_cache_lock:
398		− _key_cache[principal] = (now + KEY_CACHE_TTL_S, tier)
	497	+ _key_cache.put(principal, tier, gen)
399	498	return tier
400	499
401	500

modified hfmarketdata/api/ratelimit/redis_limiter.py +25 −0

		`@@ -225,6 +225,31 @@ def peek_tier(principal: str, tier: Tier, at_ms: int \| None = None) -> Decision`
225	225	return apply_tier(principal, tier, req_cost=0, rows_cost=0, at_ms=at_ms)
226	226
227	227
	228	+def throttle(principal: str, *, limit: int, window_s: int, cost: int = 1, namespace: str = "thr") -> Decision \| None:
	229	+ """Simple sliding-window counter (requests only) for auth endpoints, key lookups, per-e-mail limits…
	230	+ `None` when Redis is down — the caller decides whether to fail open or closed."""
	231	+ return apply(principal, window_s=window_s, limit_requests=limit, limit_rows=10**9, req_cost=cost, namespace=namespace)
	232	+
	233	+
	234	+KEYS_VERSION = "keys:version"
	235	+
	236	+
	237	+def keys_version() -> int \| None:
	238	+ """Global generation number of the API-key cache (bumped by `bump_keys_version`). None when Redis is down."""
	239	+ res = _call(lambda: client().get(KEYS_VERSION))
	240	+ if res is None:
	241	+ return None if not available() else 0
	242	+ try:
	243	+ return int(res)
	244	+ except (TypeError, ValueError):
	245	+ return 0
	246	+
	247	+
	248	+def bump_keys_version() -> None:
	249	+ """Invalidate every worker's key cache at once (revoke / rotate / tier or status change)."""
	250	+ _call(lambda: client().incr(KEYS_VERSION))
	251	+
	252	+
228	253	def reset_for_tests() -> None:
229	254	"""Flush every counter (fakeredis only — never call against production Redis)."""
230	255	global _down_until

modified hfmarketdata/api/ratelimit/tiers.py +10 −1

		`@@ -26,7 +26,16 @@ PARQUET_MEDIA = "application/vnd.apache.parquet"`
26	26	PARQUET_ROW_FACTOR = 0.5
27	27	KEY_REQUIRED_TAGS = frozenset({"stream", "screener"})
28	28	AUTH_WINDOW_S = 3600
29		−AUTH_REQUESTS_PER_WINDOW = 10 # /v1/auth/* per IP (signup, login, forgot, …)
	29	+# /v1/auth/* per IP and per hour, one bucket per endpoint (logout is not throttled). Login / signup / forgot
	30	+# fail CLOSED when Redis is unreachable (429, Retry-After 30) — they are the brute-force surface.
	31	+AUTH_LIMITS: dict[str, int] = {"login": 20, "signup": 5, "forgot": 5, "verify": 30, "reset": 10, "accept-invite": 10}
	32	+AUTH_DEFAULT_LIMIT = 20
	33	+AUTH_FAIL_CLOSED = frozenset({"login", "signup", "forgot"})
	34	+AUTH_FAIL_CLOSED_RETRY_S = 30
	35	+FORGOT_PER_EMAIL_PER_HOUR = 5
	36	+AUTH_REQUESTS_PER_WINDOW = AUTH_DEFAULT_LIMIT # backward-compatible alias
	37	+# unknown Bearer keys: SQLite lookups per IP per minute before the DB is even asked (cache misses only)
	38	+KEY_LOOKUPS_PER_MINUTE_PER_IP = 120
30	39
31	40
32	41	@dataclass(frozen=True)

modified hfmarketdata/api/ratelimit/usage.py +116 −24

		`@@ -1,17 +1,21 @@`
1		−"""Usage accounting: per-minute counters in Redis, folded into SQLite every minute.
	1	+"""Usage accounting: per-minute counters in Redis, folded into SQLite every minute, plus quota alerts.
2	2
3	3	Redis: HASH `usage:{principal}:{minute_epoch}` (requests, rows, rows_parquet, bytes, status_2xx, status_429),
4	4	TTL 8 days, plus a SET `usage:pending` indexing the hashes still to fold.
5	5	SQLite: `usage_minute` (7 rolling days, for the 24 h / 7 d charts) and `usage_daily` (forever).
6	6
7	7	`start_background(loop)` schedules the folder every 60 s (once per process, guarded by a Redis lock so the
8		−2 uvicorn workers never double count). `usage_series(principals, range)` serves the dashboard.
	8	+2 uvicorn workers never double count). After each fold the folder runs `quota_alerts()`: users whose keys
	9	+reached 80 % / 100 % of the rows quota of their window, or got their first 429 of the day, receive at most
	10	+ONE e-mail per UTC day (`users.quota_alerts` opt-out; dedupe key `alert:<uid>:<day>` in Redis).
	11	+`usage_series(principals, range)` serves the dashboard; `usage_csv()` the export.
9	12
10	13	Author: Simon-Pierre Boucher <contact@spboucher.ai>
11	14	"""
12	15	from __future__ import annotations
13	16
14	17	import asyncio
	18	+import io
15	19	import logging
16	20	import time
17	21	from collections import defaultdict
		`@@ -30,7 +34,11 @@ FOLD_LOCK = "usage:fold:lock"`
30	34	TTL_S = 8 * 86400
31	35	MINUTE_RETENTION_DAYS = 7
32	36	FIELDS = ("requests", "rows", "rows_parquet", "bytes", "status_2xx", "status_429")
	37	+MINUTE_FIELDS = ("requests", "rows", "rows_parquet", "bytes", "status_429")
	38	+SERIES_FIELDS = ("requests", "rows", "status_429", "bytes", "rows_parquet")
33	39	RANGES = {"24h": (timedelta(hours=24), 60), "7d": (timedelta(days=7), 3600), "30d": (timedelta(days=30), 86400)}
	40	+ALERT_WARN_RATIO = 0.8
	41	+ALERT_TTL_S = 2 * 86400
34	42
35	43	_started_loops: set[int] = set()
36	44
		`@@ -129,10 +137,11 @@ def fold(now_s: float \| None = None) -> int:`
129	137	day = m_dt.date()
130	138	um = s.get(_models().UsageMinute, (m_dt, principal))
131	139	if um is None:
132		− um = _models().UsageMinute(minute=m_dt, principal=principal, requests=0, rows=0)
	140	+ um = _models().UsageMinute(minute=m_dt, principal=principal, requests=0, rows=0,
	141	+ rows_parquet=0, bytes=0, status_429=0)
133	142	s.add(um)
134		− um.requests += counters.get("requests", 0)
135		− um.rows += counters.get("rows", 0)
	143	+ for f in MINUTE_FIELDS:
	144	+ setattr(um, f, (getattr(um, f) or 0) + counters.get(f, 0))
136	145	ud = s.get(_models().UsageDaily, (day, principal))
137	146	if ud is None:
138	147	ud = _models().UsageDaily(day=day, principal=principal)
		`@@ -173,6 +182,7 @@ async def _folder_loop(interval_s: int = 60) -> None:`
173	182	try:
174	183	await asyncio.sleep(interval_s)
175	184	await asyncio.to_thread(fold)
	185	+ await asyncio.to_thread(quota_alerts)
176	186	except asyncio.CancelledError: # pragma: no cover
177	187	return
178	188	except Exception as exc: # pragma: no cover
		`@@ -191,8 +201,13 @@ def start_background(interval_s: int = 60) -> None:`
191	201	loop.create_task(_folder_loop(interval_s), name="hfmd-usage-folder")
192	202
193	203
	204	+def _empty_bucket() -> dict[str, int]:
	205	+ return {f: 0 for f in SERIES_FIELDS}
	206	+
	207	+
194	208	def usage_series(principals: list[str], range_: str, now_s: float \| None = None) -> dict:
195		− """Series + totals for one or several principals over 24h (per minute), 7d (per hour) or 30d (per day)."""
	209	+ """Series + totals for one or several principals over 24h (per minute), 7d (per hour) or 30d (per day).
	210	+ Every point carries requests, rows, status_429, bytes and rows_parquet (folded + live minute)."""
196	211	if range_ not in RANGES:
197	212	from core.errors import ApiError
198	213	raise ApiError(400, "INVALID_PARAMETER", "range must be one of 24h, 7d, 30d")
		`@@ -201,45 +216,55 @@ def usage_series(principals: list[str], range_: str, now_s: float \| None = None)`
201	216	end = int(now) - int(now) % step + step # exclusive end of the last bucket
202	217	start_ts = end - int(span.total_seconds())
203	218	start_dt = datetime.fromtimestamp(start_ts, tz=timezone.utc).replace(tzinfo=None)
204		− buckets: dict[int, dict[str, int]] = defaultdict(lambda: {"requests": 0, "rows": 0})
	219	+ buckets: dict[int, dict[str, int]] = defaultdict(_empty_bucket)
205	220	models = _models()
206	221	pset = set(principals)
207	222	if pset:
208	223	with session() as s:
209	224	if range_ == "30d":
210		− rows = s.execute(select(models.UsageDaily.day, func.sum(models.UsageDaily.requests), func.sum(models.UsageDaily.rows))
211		− .where(models.UsageDaily.principal.in_(pset), models.UsageDaily.day >= start_dt.date())
212		− .group_by(models.UsageDaily.day)).all()
213		− for day, req, rw in rows:
	225	+ D = models.UsageDaily
	226	+ rows = s.execute(select(D.day, func.sum(D.requests), func.sum(D.rows), func.sum(D.status_429), func.sum(D.bytes),
	227	+ func.sum(D.rows_parquet))
	228	+ .where(D.principal.in_(pset), D.day >= start_dt.date()).group_by(D.day)).all()
	229	+ for day, *vals in rows:
214	230	ts = int(datetime(day.year, day.month, day.day, tzinfo=timezone.utc).timestamp())
215		− buckets[ts]["requests"] += int(req or 0)
216		− buckets[ts]["rows"] += int(rw or 0)
	231	+ for f, v in zip(SERIES_FIELDS, vals):
	232	+ buckets[ts][f] += int(v or 0)
217	233	else:
218		− rows = s.execute(select(models.UsageMinute.minute, models.UsageMinute.requests, models.UsageMinute.rows)
219		− .where(models.UsageMinute.principal.in_(pset), models.UsageMinute.minute >= start_dt)).all()
220		− for m, req, rw in rows:
	234	+ M = models.UsageMinute
	235	+ rows = s.execute(select(M.minute, M.requests, M.rows, M.status_429, M.bytes, M.rows_parquet)
	236	+ .where(M.principal.in_(pset), M.minute >= start_dt)).all()
	237	+ for m, *vals in rows:
221	238	ts = int(m.replace(tzinfo=timezone.utc).timestamp())
222	239	ts -= ts % step
223		− buckets[ts]["requests"] += int(req or 0)
224		− buckets[ts]["rows"] += int(rw or 0)
	240	+ for f, v in zip(SERIES_FIELDS, vals):
	241	+ buckets[ts][f] += int(v or 0)
225	242	for (_, minute), counters in _live_minutes(pset).items():
226	243	if minute >= start_ts:
227	244	ts = minute - minute % step
228		− buckets[ts]["requests"] += counters.get("requests", 0)
229		− buckets[ts]["rows"] += counters.get("rows", 0)
	245	+ for f in SERIES_FIELDS:
	246	+ buckets[ts][f] += counters.get(f, 0)
230	247	points = []
231	248	t = start_ts
232	249	while t < end:
233		− b = buckets.get(t, {"requests": 0, "rows": 0})
234		− points.append({"t": datetime.fromtimestamp(t, tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
235		− "requests": b["requests"], "rows": b["rows"]})
	250	+ b = buckets.get(t) or _empty_bucket()
	251	+ points.append({"t": datetime.fromtimestamp(t, tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), **b})
236	252	t += step
237		− totals = {"requests": sum(p["requests"] for p in points), "rows": sum(p["rows"] for p in points)}
	253	+ totals = {f: sum(p[f] for p in points) for f in SERIES_FIELDS}
238	254	return {"range": range_, "step_seconds": step, "from": points[0]["t"] if points else None,
239	255	"to": datetime.fromtimestamp(end, tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
240	256	"points": points, "totals": totals}
241	257
242	258
	259	+def usage_csv(series: dict) -> str:
	260	+ """CSV of a `usage_series` result: one line per point (UTC), header first."""
	261	+ buf = io.StringIO()
	262	+ buf.write("t," + ",".join(SERIES_FIELDS) + "\n")
	263	+ for p in series["points"]:
	264	+ buf.write(p["t"] + "," + ",".join(str(p[f]) for f in SERIES_FIELDS) + "\n")
	265	+ return buf.getvalue()
	266	+
	267	+
243	268	def top_principals(days: int = 7, limit: int = 20) -> list[dict]:
244	269	"""Admin: heaviest principals over the last N days (folded data only)."""
245	270	models = _models()
		`@@ -262,3 +287,70 @@ def totals_per_day(days: int = 30) -> list[dict]:`
262	287	.where(models.UsageDaily.day >= since).group_by(models.UsageDaily.day).order_by(models.UsageDaily.day)).all()
263	288	return [{"day": d.isoformat(), "requests": int(r or 0), "rows": int(w or 0), "rows_parquet": int(pq or 0),
264	289	"status_429": int(e or 0), "principals": int(n or 0)} for d, r, w, pq, e, n in rows]
	290	+
	291	+
	292	+# ------------------------------------------------------------------------------------------ quota alerts
	293	+
	294	+def _recent_principals(minutes: int = 3) -> set[str]:
	295	+ """Principals with live counters (the folder ran a moment ago, so the current minute is what is left) plus
	296	+ the ones folded over the last few minutes."""
	297	+ out = {p for (p, _m) in _live_minutes()}
	298	+ models = _models()
	299	+ since = datetime.now(timezone.utc).replace(tzinfo=None) - timedelta(minutes=minutes)
	300	+ with session() as s:
	301	+ for (p,) in s.execute(select(models.UsageMinute.principal).where(models.UsageMinute.minute >= since).distinct()).all():
	302	+ out.add(p)
	303	+ return {p for p in out if p.startswith("key:")}
	304	+
	305	+
	306	+def _alert_state(uid: int, day: str) -> bool:
	307	+ """SETNX `alert:<uid>:<day>` → True when this user has not been alerted today (Redis down → False)."""
	308	+ return bool(rl._call(lambda: rl.client().set(f"alert:{uid}:{day}", "1", nx=True, ex=ALERT_TTL_S)))
	309	+
	310	+
	311	+def quota_alerts(now_s: float \| None = None) -> int:
	312	+ """Send at most one alert per user per UTC day. Returns the number of alerts queued."""
	313	+ from .tiers import tier_for
	314	+ models = _models()
	315	+ principals = _recent_principals()
	316	+ if not principals:
	317	+ return 0
	318	+ now = datetime.fromtimestamp(time.time() if now_s is None else now_s, tz=timezone.utc)
	319	+ day = now.strftime("%Y-%m-%d")
	320	+ sent = 0
	321	+ key_ids = [int(p.split(":")[1]) for p in principals]
	322	+ with session() as s:
	323	+ rows = s.execute(select(models.ApiKey, models.User).join(models.User, models.User.id == models.ApiKey.user_id)
	324	+ .where(models.ApiKey.id.in_(key_ids))).all()
	325	+ per_user: dict[int, list] = defaultdict(list)
	326	+ for k, u in rows:
	327	+ if u.quota_alerts and u.status == "active" and u.email_verified_at is not None:
	328	+ per_user[u.id].append((k, u))
	329	+ for uid, items in per_user.items():
	330	+ u = items[0][1]
	331	+ reason = None
	332	+ for k, _u in items:
	333	+ tier = tier_for(k.tier_override or u.tier)
	334	+ d = rl.peek_tier(k.principal, tier)
	335	+ if d is not None and tier.rows > 0:
	336	+ used = tier.rows - d.remaining_rows
	337	+ ratio = used / tier.rows
	338	+ if d.remaining_rows <= 0 or ratio >= 1:
	339	+ reason = ("Rows quota reached (100 %)", f"Key “{k.name}” ({k.prefix}…) used 100 % of its "
	340	+ f"{tier.rows:,} rows per {tier.window_label}; requests get 429 until the window slides.")
	341	+ break
	342	+ if ratio >= ALERT_WARN_RATIO:
	343	+ reason = reason or ("Rows quota at 80 %", f"Key “{k.name}” ({k.prefix}…) used {ratio:.0%} of its "
	344	+ f"{tier.rows:,} rows per {tier.window_label}.")
	345	+ today_429 = s.execute(select(func.sum(models.UsageDaily.status_429)).where(
	346	+ models.UsageDaily.principal == k.principal, models.UsageDaily.day == now.date())).scalar()
	347	+ live_429 = sum(c.get("status_429", 0) for (p, _m), c in _live_minutes({k.principal}).items())
	348	+ if (int(today_429 or 0) + live_429) > 0:
	349	+ reason = reason or ("First 429 of the day", f"Key “{k.name}” ({k.prefix}…) received its first "
	350	+ "HTTP 429 (rate limit exceeded) today.")
	351	+ if reason is None or not _alert_state(uid, day):
	352	+ continue
	353	+ from accounts import mailer
	354	+ mailer.queue(s, "quota_alert", u.email, name=u.name, what=reason[0], detail=reason[1])
	355	+ sent += 1
	356	+ return sent
265	357