SEL-346: AMW API: per-key usage events and daily per-key summary

gateway/app/core/api_key_usage.py

@@ -0,0 +1,124 @@
+"""Metadata-only HTTP usage events for database-backed API keys."""
+
+import asyncio
+import logging
+import time
+from datetime import datetime, timedelta
+from typing import Any, cast
+from uuid import uuid4
+
+from sqlalchemy import text
+from starlette.types import ASGIApp, Message, Receive, Scope, Send
+
+from .auth import CREDENTIAL_ACCEPTANCE
+from .config import get_settings
+from .time import to_naive_utc, utc_now
+from ..db.database import get_session_factory
+from ..db.models import APIKeyUsageEventModel
+
+logger = logging.getLogger(__name__)
+_USAGE_WRITE_TIMEOUT_SECONDS = 2.0
+_MAX_PRUNE_BATCH_SIZE = 1_000
+
+
+async def prune_api_key_usage_events(
+    *, batch_size: int = _MAX_PRUNE_BATCH_SIZE, now: datetime | None = None
+) -> int:
+    """Delete at most one batch older than retention; never runs automatically."""
+    retention_days = get_settings().API_KEY_USAGE_RETENTION_DAYS
+    if not retention_days:
+        return 0
+    if not 1 <= batch_size <= _MAX_PRUNE_BATCH_SIZE:
+        raise ValueError(f"batch_size must be between 1 and {_MAX_PRUNE_BATCH_SIZE}")
+    cutoff = to_naive_utc(now) if now is not None else utc_now()
+    cutoff -= timedelta(days=retention_days)
+    async with get_session_factory()() as session:
+        result = await session.execute(
+            text(
+                "DELETE FROM api_key_usage_events WHERE id IN ("
+                "SELECT id FROM api_key_usage_events WHERE created_at < :cutoff "
+                "ORDER BY created_at, id LIMIT :batch_size)"
+            ),
+            {"cutoff": cutoff, "batch_size": batch_size},
+        )
+        await session.commit()
+        return cast(Any, result).rowcount
+
+
+async def record_api_key_usage_event(
+    *,
+    key_id: str,
+    tenant: str | None,
+    method: str,
+    endpoint: str,
+    status_code: int,
+    latency_ms: int,
+) -> None:
+    """Append one event using a separate session after the response."""
+    async with get_session_factory()() as session:
+        session.add(
+            APIKeyUsageEventModel(
+                id=str(uuid4()),
+                key_id=key_id,
+                tenant=tenant,
+                method=method,
+                endpoint=endpoint,
+                status_code=status_code,
+                latency_ms=latency_ms,
+            )
+        )
+        await session.commit()
+
+
+class APIKeyUsageMiddleware:
+    """Record the resolved key and final HTTP status once per API request."""
+
+    def __init__(self, app: ASGIApp) -> None:
+        self.app = app
+
+    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
+        if scope["type"] != "http":
+            await self.app(scope, receive, send)
+            return
+
+        # RateLimitMiddleware sets this shared holder before entering this
+        # layer; auth fills its key identity in the child task after lookup.
+        acceptance = CREDENTIAL_ACCEPTANCE.get()
+        if acceptance is None:
+            await self.app(scope, receive, send)
+            return
+
+        started = time.perf_counter()
+        status_code = 500
+
+        async def send_with_status(message: Message) -> None:
+            nonlocal status_code
+            if message["type"] == "http.response.start":
+                status_code = message["status"]
+            await send(message)
+
+        try:
+            await self.app(scope, receive, send_with_status)
+        finally:
+            # route.path is the registered template. Never substitute scope's
+            # raw path, which could contain IDs, query values, or user text.
+            endpoint = getattr(scope.get("route"), "path", None)
+            if acceptance.key_id and endpoint:
+                latency_ms = max(0, int((time.perf_counter() - started) * 1000))
+                try:
+                    await asyncio.wait_for(
+                        record_api_key_usage_event(
+                            key_id=acceptance.key_id,
+                            tenant=acceptance.tenant,
+                            method=scope.get("amw_original_method", scope["method"]),
+                            endpoint=endpoint,
+                            status_code=status_code,
+                            latency_ms=latency_ms,
+                        ),
+                        timeout=_USAGE_WRITE_TIMEOUT_SECONDS,
+                    )
+                except Exception as exc:  # noqa: BLE001 - usage cannot fail API calls
+                    logger.warning(
+                        "api_key_usage_event_write_failed error_type=%s",
+                        type(exc).__name__,
+                    )

gateway/app/core/auth.py

@@ -100,7 +100,9 @@ def require_bootstrap_admin(self) -> None:
         )
 
 
-async def _key_tenant_for_auth(key_tenant: str | None, wallet_id: str) -> str | None:
+async def _key_tenant_for_auth(
+    key_tenant: str | None, wallet_id: str, key_id: str | None = None
+) -> str | None:
     """Apply demo confinement to the key and its owning wallet."""
     from ..services.demo_tenant_boundary import (
         DEMO_KEY_ROUTE_FORBIDDEN,
@@ -112,6 +114,11 @@ async def _key_tenant_for_auth(key_tenant: str | None, wallet_id: str) -> str |
     )
 
     tenant = await resolve_effective_tenant(key_tenant, wallet_id)
+    if key_id is not None:
+        acceptance = CREDENTIAL_ACCEPTANCE.get()
+        if acceptance is not None:
+            acceptance.key_id = key_id
+            acceptance.tenant = tenant
     if tenant != DEMO_TENANT_LABEL:
         return tenant
     if not get_settings().ENABLE_DEMO_TENANT:
@@ -148,9 +155,11 @@ async def _key_tenant_for_auth(key_tenant: str | None, wallet_id: str) -> str |
 
 @dataclass
 class CredentialAcceptance:
-    """Whether ``get_auth_context`` accepted credentials during one request."""
+    """Accepted credentials and any database key resolved for the request."""
 
     accepted: bool = False
+    key_id: str | None = None
+    tenant: str | None = None
 
 
 # The converse signal, for RateLimitMiddleware's shared pre-auth bucket. A
@@ -190,6 +199,8 @@ async def get_auth_context(
     acceptance = CREDENTIAL_ACCEPTANCE.get()
     if acceptance is not None:
         acceptance.accepted = True
+        acceptance.key_id = context.key_id
+        acceptance.tenant = context.tenant
     return context
 
 
@@ -311,7 +322,9 @@ async def _resolve_auth_context(
 
     if candidate:
         if _key_is_live(candidate, utc_now()):
-            await _key_tenant_for_auth(candidate.tenant, candidate.wallet_id)
+            await _key_tenant_for_auth(
+                candidate.tenant, candidate.wallet_id, candidate.key_id
+            )
         try:
             db_key = await key_service.validate_key(stripped)
         except RuntimeError:
@@ -322,7 +335,9 @@ async def _resolve_auth_context(
     if db_key:
         from ..services.demo_tenant_boundary import parse_allowlist
 
-        tenant = await _key_tenant_for_auth(db_key.tenant, db_key.wallet_id)
+        tenant = await _key_tenant_for_auth(
+            db_key.tenant, db_key.wallet_id, db_key.key_id
+        )
         return AuthContext(
             source="db",
             raw_key=stripped,
@@ -464,7 +479,9 @@ async def _auth_from_jwt(token: str) -> AuthContext:
         and candidate.wallet_id == payload.sub
         and _key_is_live(candidate, utc_now())
     ):
-        await _key_tenant_for_auth(candidate.tenant, candidate.wallet_id)
+        await _key_tenant_for_auth(
+            candidate.tenant, candidate.wallet_id, candidate.key_id
+        )
 
     if not await key_service.consume_derived_key_use(payload.key_id, payload.sub):
         raise HTTPException(
@@ -507,7 +524,9 @@ async def _auth_from_jwt(token: str) -> AuthContext:
         )
     from ..services.demo_tenant_boundary import parse_allowlist
 
-    tenant = await _key_tenant_for_auth(key_row.tenant, key_row.wallet_id)
+    tenant = await _key_tenant_for_auth(
+        key_row.tenant, key_row.wallet_id, key_row.key_id
+    )
     return AuthContext(
         source="jwt",
         # Not a credential: a stable, non-secret handle for the API key that

gateway/app/core/config.py

@@ -5,7 +5,7 @@
 
 from decimal import Decimal
 from enum import Enum
-from pydantic import SecretStr, field_validator
+from pydantic import Field, SecretStr, field_validator
 from pydantic_settings import BaseSettings, SettingsConfigDict
 from functools import lru_cache
 
@@ -64,6 +64,9 @@ class Settings(BaseSettings):
     DB_POOL_SIZE: int = 5
     DB_MAX_OVERFLOW: int = 10
 
+    # Opt-in only; no automatic usage-event cleanup is scheduled.
+    API_KEY_USAGE_RETENTION_DAYS: int | None = Field(default=None, ge=0)
+
     # --- Authentication ---
     API_KEY_HEADER: str = "X-API-Key"
     # Comma-separated list of valid API keys (use a secrets manager in production)

gateway/app/db/api_key_usage_view.py

@@ -0,0 +1,74 @@
+"""Portable daily API-key usage view and bounded report query.
+
+Timestamps in this database are naive UTC. Percentiles use continuous linear
+interpolation, matching PostgreSQL's percentile_cont on both supported SQL
+dialects. The SQLite development schema and Alembic migration share this SQL.
+"""
+
+
+def _percentile_expression(percent: int) -> str:
+    return (
+        "SUM(CASE "
+        f"WHEN latency_rank = p{percent}_floor_rank "
+        f"THEN latency_ms * (100 - p{percent}_remainder) / 100.0 "
+        f"WHEN latency_rank = p{percent}_floor_rank + 1 "
+        f"THEN latency_ms * p{percent}_remainder / 100.0 "
+        f"ELSE 0 END) AS p{percent}_latency_ms"
+    )
+
+
+def _daily_usage_select_sql(event_filter: str = "") -> str:
+    return f"""
+WITH ranked AS (
+    SELECT
+        key_id,
+        DATE(created_at) AS day,
+        status_code,
+        latency_ms,
+        created_at,
+        ROW_NUMBER() OVER (
+            PARTITION BY key_id, DATE(created_at)
+            ORDER BY latency_ms, id
+        ) AS latency_rank,
+        COUNT(*) OVER (
+            PARTITION BY key_id, DATE(created_at)
+        ) AS call_count
+    FROM api_key_usage_events{event_filter}
+), positioned AS (
+    SELECT
+        *,
+        ((call_count - 1) * 50) / 100 + 1 AS p50_floor_rank,
+        ((call_count - 1) * 50) % 100 AS p50_remainder,
+        ((call_count - 1) * 95) / 100 + 1 AS p95_floor_rank,
+        ((call_count - 1) * 95) % 100 AS p95_remainder
+    FROM ranked
+)
+SELECT
+    key_id,
+    day,
+    COUNT(*) AS calls,
+    SUM(CASE WHEN status_code BETWEEN 400 AND 499 THEN 1 ELSE 0 END) AS calls_4xx,
+    SUM(CASE WHEN status_code BETWEEN 500 AND 599 THEN 1 ELSE 0 END) AS calls_5xx,
+    {_percentile_expression(50)},
+    {_percentile_expression(95)},
+    MIN(created_at) AS first_call_at,
+    MAX(created_at) AS last_call_at
+FROM positioned
+GROUP BY key_id, day
+"""
+
+
+CREATE_API_KEY_USAGE_DAILY_VIEW_SQL = (
+    "CREATE VIEW api_key_usage_daily AS" + _daily_usage_select_sql()
+)
+
+
+def bounded_daily_usage_sql(*, filter_key: bool) -> str:
+    """Aggregate only events in the requested UTC window, then cap result rows."""
+    event_filter = "\n    WHERE created_at >= :start_at AND created_at < :end_at"
+    if filter_key:
+        event_filter += " AND key_id = :key_id"
+    return (
+        _daily_usage_select_sql(event_filter)
+        + " ORDER BY day DESC, key_id LIMIT :limit"
+    )

gateway/app/db/database.py

@@ -36,6 +36,7 @@
 from ..core.db_urls import as_sqlalchemy_url
 from ..core.time import to_naive_utc
 from ..core.trust_mode import is_production_like_environment
+from .api_key_usage_view import CREATE_API_KEY_USAGE_DAILY_VIEW_SQL
 
 logger = logging.getLogger(__name__)
 
@@ -62,9 +63,22 @@
         "human_approvals",
         "receipt_resolutions",
         "resolution_chain_heads",
+        "api_key_usage_events",
     }
 )
 REQUIRED_TRUST_COLUMNS = {
+    "api_key_usage_events": frozenset(
+        {
+            "id",
+            "key_id",
+            "tenant",
+            "method",
+            "endpoint",
+            "status_code",
+            "latency_ms",
+            "created_at",
+        }
+    ),
     "mcp_dispatch_attempts": frozenset({"dispatch_claim_hash"}),
     "resolution_chain_heads": frozenset({"last_sequence", "last_hash"}),
     "receipt_resolutions": frozenset(
@@ -362,6 +376,10 @@ def _missing_required_tables(sync_conn) -> set[str]:  # noqa: ANN001
     return set(REQUIRED_TRUST_TABLES) - existing
 
 
+def _missing_required_views(sync_conn) -> set[str]:  # noqa: ANN001
+    return {"api_key_usage_daily"} - set(sa_inspect(sync_conn).get_view_names())
+
+
 def _missing_required_columns(sync_conn) -> set[str]:  # noqa: ANN001
     inspector = sa_inspect(sync_conn)
     existing_tables = set(inspector.get_table_names())
@@ -418,6 +436,7 @@ async def verify_required_schema(engine: AsyncEngine) -> None:
     async with engine.connect() as conn:
         missing = await conn.run_sync(_missing_required_tables)
         missing_columns = await conn.run_sync(_missing_required_columns)
+        missing_views = await conn.run_sync(_missing_required_views)
         stale = await conn.run_sync(_stale_alembic_message)
     # A stamped database can be both behind head and missing tables introduced
     # by later revisions. Report the revision mismatch first: it is the
@@ -440,6 +459,12 @@ async def verify_required_schema(engine: AsyncEngine) -> None:
             "the API (normally `alembic upgrade head`; review an unstamped "
             "legacy schema before stamping its exact existing revision)."
         )
+    if missing_views:
+        raise SchemaInitError(
+            "Required database views missing: "
+            + ", ".join(sorted(missing_views))
+            + ". Bring the schema to the packaged Alembic head before starting the API."
+        )
 
 
 async def init_db() -> None:
@@ -462,6 +487,8 @@ async def init_db() -> None:
     if allows_metadata_create_all(dialect_name=dialect, environment=environment):
         async with engine.begin() as conn:
             await conn.run_sync(SQLModel.metadata.create_all)
+            if "api_key_usage_daily" in await conn.run_sync(_missing_required_views):
+                await conn.exec_driver_sql(CREATE_API_KEY_USAGE_DAILY_VIEW_SQL)
             missing_columns = await conn.run_sync(_missing_required_columns)
         if missing_columns:
             raise SchemaInitError(