feat(security): ops alerts, shared backoff, atomic quota and daily stats for CrowdSec

Add an alerting/stats layer over the existing CrowdSec integration:

- New crowdsec-alerts.ts: cooldown-gated ops alerts (Redis NX lock, TTL from
  HEALTH_ALERT_COOLDOWN_MIN) fanning out through the app's sendAlert service.
  Raised for daily quota exhaustion, block bursts (5-min window past
  CROWDSEC_ALERT_BLOCK_BURST), and signal-push failures.
- New crowdsec-stats.ts: daily counters (lookups/blocks/reports/report_fail)
  in Redis with a 14-day reader for the admin panel.
- Shared 403/429 backoff: the pause marker now lives in Redis
  (crowdsec:backoff-until) so every instance honours it, not just the process
  that hit the limit.
- Atomic quota reservation: INCR-before-call with self-rollback on overshoot,
  so concurrent instances can never slip calls past the daily ceiling.
- Admin anti-DDoS page gains a last-14-days activity table next to the quota bar.

src/lib/crowdsec-api.ts

@@ -1,7 +1,9 @@
 import "server-only";
 
 import { env } from "@/env";
+import { raiseCrowdsecAlert } from "@/lib/crowdsec-alerts";
 import { reportCrowdsecSignal } from "@/lib/crowdsec-report";
+import { bumpCrowdsecStat } from "@/lib/crowdsec-stats";
 import { logger } from "@/lib/logger";
 import { redis } from "@/lib/redis";
 import { UNKNOWN_CLIENT_IP } from "./client-ip";
@@ -113,6 +115,11 @@ const LAST_VERIFY_KEY = "crowdsec:last-verify";
 const BLOCK_META_PREFIX = "antiddos:block:meta:";
 const QUOTA_PREFIX = "crowdsec:usage:";
 const QUOTA_KEY_TTL_SECONDS = 48 * 3_600;
+/** Shared 403/429 pause marker, so every instance respects the backoff. */
+const BACKOFF_KEY = "crowdsec:backoff-until";
+/** Short-window block burst counter: crowdsec:burst:{unix-5min-bucket}. */
+const BURST_PREFIX = "crowdsec:burst:";
+const BURST_WINDOW_SECONDS = 300;
 /** In-process verdict cache cap so a flood of distinct IPs cannot grow it forever. */
 const MEMORY_VERDICT_CACHE_MAX = 2_000;
 /** Warn at this fraction of the daily quota, once per day. */
@@ -313,6 +320,47 @@ async function acquireLookupLock(ip: string): Promise<boolean> {
 let backoffUntil = 0;
 let quotaWarnedDate: string | null = null;
 
+/**
+ * Next moment (epoch ms) the CTI API may be called again — the max of the
+ * in-process view and the shared Redis marker so every instance respects a
+ * backoff discovered by any of them. Redis is only read when the local view is
+ * not already active, keeping the hot path cheap.
+ */
+async function getBackoffUntil(): Promise<number> {
+	if (Date.now() < backoffUntil) return backoffUntil;
+	if (redis) {
+		try {
+			const raw = await redis.get(BACKOFF_KEY);
+			const shared = Number(raw ?? 0);
+			if (Number.isFinite(shared) && shared > backoffUntil) {
+				backoffUntil = shared;
+			}
+		} catch {
+			// Redis hiccup — local view is enough
+		}
+	}
+	return backoffUntil;
+}
+
+async function setBackoff(ms: number): Promise<void> {
+	const until = Date.now() + ms;
+	backoffUntil = until;
+	if (redis) {
+		try {
+			// EX rounds up so the marker outlives the wait it encodes, plus a
+			// second of slack for the read path.
+			await redis.set(
+				BACKOFF_KEY,
+				String(until),
+				"EX",
+				Math.ceil(ms / 1000) + 1,
+			);
+		} catch {
+			// local view still protects this instance
+		}
+	}
+}
+
 function quotaDate(): string {
 	return new Date().toISOString().slice(0, 10);
 }
@@ -350,33 +398,87 @@ export async function getCrowdsecQuotaUsage(): Promise<CrowdsecQuotaUsage> {
 	return { date, used, quota, exhausted: quota > 0 && used >= quota };
 }
 
+/**
+ * Reserve one API call against today's quota. Atomic: the counter is INCR'd
+ * BEFORE the call and compared to the ceiling, so concurrent instances can
+ * never slip calls past the budget; a reserve that overshoots rolls itself
+ * back. Returns false once the budget is spent (and raises an ops alert).
+ */
 async function reserveQuota(): Promise<boolean> {
 	const quota = dailyQuota();
 	if (quota <= 0) return true;
+	if (!redis) return true; // no shared counter → unlimited best-effort
 	const date = quotaDate();
-	const usage = await getCrowdsecQuotaUsage();
-	if (usage.used >= quota) {
-		// Stop consulting the API for the rest of the day: a flood of distinct
-		// bucket-tripping IPs would otherwise burn every remaining call and
-		// then sit in a 429 storm anyway.
-		return false;
-	}
-	if (usage.used >= quota * QUOTA_WARN_RATIO && quotaWarnedDate !== date) {
-		quotaWarnedDate = date;
-		logger.warn("[crowdsec-api] CTI daily quota nearing its limit", {
-			used: usage.used,
-			quota,
-		});
-	}
-	if (redis) {
-		try {
-			await redis.incr(quotaKey(date));
-			await redis.expire(quotaKey(date), QUOTA_KEY_TTL_SECONDS);
-		} catch {
-			// best effort — an uncounted call is better than a failed lookup
+	const key = quotaKey(date);
+	try {
+		const used = await redis.incr(key);
+		await redis.expire(key, QUOTA_KEY_TTL_SECONDS);
+		if (used > quota) {
+			// Concurrent reserves nudged us past the ceiling — give the slot
+			// back and refuse: the budget would be spent the very next call
+			// anyway, so stopping here is both safe and quota-exact.
+			await redis.decr(key);
+			logger.warn(
+				"[crowdsec-api] CTI daily quota exhausted — pausing lookups until tomorrow",
+				{ quota },
+			);
+			void raiseCrowdsecAlert("quota", {
+				type: "ddos",
+				severity: "warning",
+				message: `CrowdSec reputation quota exhausted for today (${used} of ${quota} enrichment calls) — lookups are paused until tomorrow.`,
+				context: { used, quota, date },
+			});
+			return false;
 		}
+		if (used >= quota * QUOTA_WARN_RATIO && quotaWarnedDate !== date) {
+			quotaWarnedDate = date;
+			logger.warn("[crowdsec-api] CTI daily quota nearing its limit", {
+				used,
+				quota,
+			});
+		}
+		return true;
+	} catch {
+		// Redis hiccup at a moment we could not count — allow the call rather
+		// than break the gate; the verdict cache still limits frequency.
+		return true;
+	}
+}
+
+/** Block burst threshold from env, defensively coerced (falls back to 10). */
+function dailyBlockBurstThreshold(): number {
+	const raw = Number(env.CROWDSEC_ALERT_BLOCK_BURST ?? 10);
+	return Number.isFinite(raw) && raw > 0 ? Math.floor(raw) : 10;
+}
+
+/**
+ * A burst of new community-reputation blocks is usually an automated attack
+ * wave. Count blocks into a rolling 5-minute bucket and alert once per
+ * cooldown window when they cross CROWDSEC_ALERT_BLOCK_BURST. Fire-and-forget.
+ */
+async function trackBlockBurst(): Promise<void> {
+	if (!redis) return;
+	const bucket = Math.floor(Date.now() / 1000 / BURST_WINDOW_SECONDS);
+	const key = `${BURST_PREFIX}${bucket}`;
+	const threshold = dailyBlockBurstThreshold();
+	try {
+		const count = await redis.incr(key);
+		await redis.expire(key, BURST_WINDOW_SECONDS * 2);
+		if (count >= threshold) {
+			void raiseCrowdsecAlert("block-burst", {
+				type: "ddos",
+				severity: "warning",
+				message: `CrowdSec community reputation blocked ${count} IPs in the last ${BURST_WINDOW_SECONDS / 60} minutes — likely an automated attack wave.`,
+				context: {
+					blocks: count,
+					windowSeconds: BURST_WINDOW_SECONDS,
+					threshold,
+				},
+			});
+		}
+	} catch {
+		// alert is best-effort — never break the block path
 	}
-	return true;
 }
 
 /**
@@ -390,7 +492,7 @@ export async function lookupCrowdsecVerdict(
 ): Promise<CrowdsecVerdict | null> {
 	if (!crowdsecEnabled()) return null;
 	if (!ip || ip === UNKNOWN_CLIENT_IP) return null;
-	if (Date.now() < backoffUntil) return null;
+	if (Date.now() < (await getBackoffUntil())) return null;
 
 	const cached = await readVerdictCache(ip);
 	if (cached) return cached;
@@ -417,6 +519,9 @@ export async function lookupCrowdsecVerdict(
 		}
 
 		const response = await crowdsecRequest(`/smoke/${encodeURIComponent(ip)}`);
+		// The enrichment call happened — count it for the daily histogram,
+		// regardless of whether the verdict was positive, negative, or n/a.
+		void bumpCrowdsecStat("lookups");
 
 		if (response.status === 404) {
 			// Unknown to the community — cache the negative result so a clean
@@ -426,13 +531,13 @@ export async function lookupCrowdsecVerdict(
 			return verdict;
 		}
 		if (response.status === 403) {
-			backoffUntil = Date.now() + AUTH_BACKOFF_MS;
+			await setBackoff(AUTH_BACKOFF_MS);
 			throw new CrowdsecApiError(
 				`CrowdSec API key rejected (HTTP 403): ${await errorDetail(response)}`,
 			);
 		}
 		if (response.status === 429) {
-			backoffUntil = Date.now() + RATE_LIMIT_BACKOFF_MS;
+			await setBackoff(RATE_LIMIT_BACKOFF_MS);
 			logger.warn("[crowdsec-api] CTI API rate limit hit — backing off", {
 				ip,
 				backoffMs: RATE_LIMIT_BACKOFF_MS,
@@ -477,7 +582,7 @@ export async function maybeAutoBlockCrowdsec(input: {
 	// The gate only ever reads its block key through shared Redis — without it
 	// there is nowhere durable to record the block.
 	if (!redis) return;
-	if (Date.now() < backoffUntil) return;
+	if (Date.now() < (await getBackoffUntil())) return;
 
 	try {
 		const verdict = await lookupCrowdsecVerdict(ip);
@@ -528,6 +633,9 @@ export async function maybeAutoBlockCrowdsec(input: {
 		// Opt-in community signal push (CAPI), fire-and-forget: never awaited,
 		// never throws, and internally deduped per IP.
 		void reportCrowdsecSignal({ ip, category, ttlSeconds, verdict, meta });
+		// Daily histogram + burst detection (cooldown-gated ops alert).
+		void bumpCrowdsecStat("blocks");
+		void trackBlockBurst();
 	} catch (error) {
 		logger.error("[crowdsec-api] Automatic IP block failed", {
 			ip,