services/analytics/Dockerfile

@@ -0,0 +1,10 @@
+FROM python:3.12-slim
+WORKDIR /app
+ENV PYTHONUNBUFFERED=1 PIP_NO_CACHE_DIR=1
+COPY requirements.txt .
+RUN pip install -r requirements.txt
+COPY app.py ingest_aud0.py ./
+EXPOSE 8096
+HEALTHCHECK --interval=30s --timeout=5s --retries=3 \
+  CMD python -c "import urllib.request; urllib.request.urlopen('http://127.0.0.1:8096/health', timeout=3)"
+CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8096"]

services/analytics/README.md

@@ -0,0 +1,47 @@
+# 0pt analytics HTTP surface
+
+The missing read plane between API0's `/analytics/*` contract and the
+`opt_analytics` ClickHouse database.
+
+## Surface
+
+Serves the BI-P068 operation paths under `/api/0pt/analytics`:
+
+- `GET /health`
+- `GET /datasets`, `GET /datasets/{id}`
+- `GET /metrics`, `GET /metrics/{id}`, `GET /metrics/{id}/explain`
+- `POST /metrics/{id}/query`
+- `POST /query`, `POST /query/estimate`, `POST /query/explain`
+- `GET /query/{id}` (404: executions are not persisted yet)
+- `GET /dashboards`, `GET /reports` (empty until the BI metadata lane writes them)
+- `GET /search`
+
+Identity: the caller's Keycloak token is verified against the portal realm
+JWKS; `tenant_id` is taken from the token and cross-checked against
+`X-Tenant-Id`. An `X-Internal-Token` path exists only for the ingestion worker.
+
+## Ingest
+
+`ingest_aud0.py` maps AUD0 audit events to `opt_analytics_event`
+(`dataset_id=aud0.events`, `metric=event_count`, dimensions event_type/app_id/
+capability_id/risk_level). It is idempotent per `event_id`.
+
+```bash
+AUD0_TOKEN=... AUD0_TENANT_ID=... CLICKHOUSE_URL=http://opt-analytics-clickhouse:8123 \
+  python ingest_aud0.py --limit 50000
+```
+
+## Run
+
+```bash
+pip install -r requirements.txt
+CLICKHOUSE_URL=http://opt-analytics-clickhouse:8123 \
+KEYCLOAK_ISSUER=https://auth.portal.siax.dev/realms/portal \
+uvicorn app:app --host 0.0.0.0 --port 8096
+```
+
+## Test
+
+```bash
+python -m pytest tests -q
+```

services/analytics/app.py

@@ -0,0 +1,464 @@
+"""
+0pt analytics HTTP surface (BI-P068-compatible).
+
+This is the missing read plane between API0's `/analytics/*` contract and the
+`opt_analytics` ClickHouse database. It does not invent a second analytics
+model:
+
+  * datasets/metrics are the distinct ids actually present in
+    `opt_analytics_event` / `opt_metric_daily` / `opt_dataset_projection`;
+  * query/estimate/explain are computed against the tenant-forced ClickHouse
+    rows, never against a local cache;
+  * dashboards/reports are honest empty lists until the 0pt BI metadata lane
+    writes them;
+  * every response is either real ClickHouse data or an explicit error.
+
+IDENTITY. API0 verifies the caller and forwards the caller's own Keycloak
+token plus `X-Tenant-Id`. This service verifies the same token against the
+portal realm's JWKS and cross-checks `tenant_id` against `X-Tenant-Id`. A
+service-to-service `X-Internal-Token` is also accepted for the ingestion
+worker, but never for a request that reads tenant data without an explicit,
+already-verified `X-Tenant-Id`.
+"""
+from __future__ import annotations
+
+import os
+import time
+from functools import lru_cache
+from typing import Any, Iterable, Literal
+
+import httpx
+import jwt
+from fastapi import FastAPI, Header, HTTPException, Query, Request
+from fastapi.responses import JSONResponse
+from pydantic import BaseModel, Field
+
+CLICKHOUSE_URL = os.getenv("CLICKHOUSE_URL", "").rstrip("/")
+CLICKHOUSE_USER = os.getenv("CLICKHOUSE_USER", "default")
+CLICKHOUSE_PASSWORD = os.getenv("CLICKHOUSE_PASSWORD", "")
+CLICKHOUSE_DATABASE = os.getenv("CLICKHOUSE_DATABASE", "opt_analytics")
+KEYCLOAK_ISSUER = os.getenv("KEYCLOAK_ISSUER", "").rstrip("/")
+KEYCLOAK_AUDIENCE = os.getenv("KEYCLOAK_AUDIENCE", "account")
+KEYCLOAK_AUTHORIZED_PARTY = os.getenv("KEYCLOAK_AUTHORIZED_PARTY", "siax-portal")
+INTERNAL_TOKEN = os.getenv("INTERNAL_TOKEN", "")
+ALLOWED_DIMENSIONS = {"event_type", "app_id", "capability_id", "risk_level", "source_service"}
+
+app = FastAPI(title="0pt analytics", version="0.1")
+
+
+class AnalyticsError(Exception):
+    def __init__(self, code: str, message: str, status: int = 400):
+        self.code = code
+        self.message = message
+        self.status = status
+        super().__init__(message)
+
+
+@app.exception_handler(AnalyticsError)
+async def analytics_error_handler(_request: Request, exc: AnalyticsError) -> JSONResponse:
+    return JSONResponse(status_code=exc.status, content={"error": {"code": exc.code, "message": exc.message}})
+
+
+# ── identity ────────────────────────────────────────────────────────────────
+
+
+@lru_cache(maxsize=1)
+def _jwks_client() -> jwt.PyJWKClient:
+    if not KEYCLOAK_ISSUER:
+        raise AnalyticsError("ANALYTICS_NOT_CONFIGURED", "KEYCLOAK_ISSUER is not set", 503)
+    return jwt.PyJWKClient(f"{KEYCLOAK_ISSUER}/protocol/openid-connect/certs", cache_keys=True, lifespan=3600)
+
+
+def _bearer(request: Request) -> str | None:
+    raw = request.headers.get("authorization", "")
+    if not raw.lower().startswith("bearer "):
+        return None
+    return raw[7:].strip() or None
+
+
+def verified_identity(request: Request, x_tenant_id: str | None) -> str:
+    """Return the tenant the request is allowed to read.
+
+    An `X-Internal-Token` may only be used by the ingestion worker and never
+    substitutes for a tenant claim. For user/API0 calls the JWT is the source
+    of truth; a mismatching X-Tenant-Id is a hard 403, not a preference.
+    """
+    if INTERNAL_TOKEN and request.headers.get("x-internal-token") == INTERNAL_TOKEN:
+        if not x_tenant_id:
+            raise AnalyticsError("TENANT_REQUIRED", "X-Tenant-Id is required for internal calls", 400)
+        return x_tenant_id
+
+    token = _bearer(request)
+    if not token:
+        raise AnalyticsError("UNAUTHENTICATED", "A bearer token is required", 401)
+    try:
+        signing_key = _jwks_client().get_signing_key_from_jwt(token)
+        claims = jwt.decode(
+            token,
+            signing_key.key,
+            algorithms=["RS256", "RS384", "RS512"],
+            audience=KEYCLOAK_AUDIENCE,
+            issuer=KEYCLOAK_ISSUER,
+            options={"require": ["exp", "iss", "sub"]},
+        )
+    except jwt.PyJWTError as exc:
+        raise AnalyticsError("UNAUTHENTICATED", f"Invalid bearer token: {exc}", 401) from exc
+
+    if KEYCLOAK_AUTHORIZED_PARTY and claims.get("azp") not in (None, KEYCLOAK_AUTHORIZED_PARTY):
+        raise AnalyticsError("FORBIDDEN", "Token was issued for a different client", 403)
+    tenant = claims.get("tenant_id")
+    if not isinstance(tenant, str) or not tenant.strip():
+        raise AnalyticsError("NO_TENANT", "Token is not bound to a tenant", 403)
+    if x_tenant_id and x_tenant_id != tenant:
+        raise AnalyticsError("TENANT_MISMATCH", "X-Tenant-Id does not match the verified tenant", 403)
+    return tenant
+
+
+# ── clickhouse ──────────────────────────────────────────────────────────────
+
+
+def _ch_client() -> httpx.Client:
+    if not CLICKHOUSE_URL:
+        raise AnalyticsError("ANALYTICS_NOT_CONFIGURED", "CLICKHOUSE_URL is not set", 503)
+    auth = (CLICKHOUSE_USER, CLICKHOUSE_PASSWORD) if CLICKHOUSE_PASSWORD else None
+    return httpx.Client(base_url=CLICKHOUSE_URL, auth=auth, timeout=20.0)
+
+
+def ch_json(sql: str, params: dict[str, str] | None = None) -> dict[str, Any]:
+    params = params or {}
+    query_params = {
+        "database": CLICKHOUSE_DATABASE,
+        "output_format_json_quote_64bit_integers": "0",
+        **{f"param_{key}": value for key, value in params.items()},
+    }
+    try:
+        with _ch_client() as client:
+            response = client.post("/", params=query_params, content=sql)
+    except httpx.HTTPError as exc:
+        raise AnalyticsError("CLICKHOUSE_UNREACHABLE", str(exc), 503) from exc
+    if response.status_code != 200:
+        raise AnalyticsError("CLICKHOUSE_REFUSED", response.text[:500], 502)
+    try:
+        return response.json()
+    except ValueError as exc:
+        raise AnalyticsError("CLICKHOUSE_BAD_RESPONSE", "ClickHouse did not return JSON", 502) from exc
+
+
+def _literal(value: str) -> str:
+    # ClickHouse query parameters are preferable; the only values interpolated
+    # here are identifiers from ALLOWED_DIMENSIONS, never caller text.
+    if value not in ALLOWED_DIMENSIONS:
+        raise AnalyticsError("INVALID_DIMENSION", f"Unsupported dimension: {value}")
+    return value
+
+
+def rows(sql: str, params: dict[str, str] | None = None) -> list[dict[str, Any]]:
+    return ch_json(sql, params).get("data", [])
+
+
+def scalar(sql: str, params: dict[str, str] | None = None) -> Any:
+    data = rows(sql, params)
+    if not data:
+        return None
+    return next(iter(data[0].values()))
+
+
+def build_query(
+    *,
+    tenant: str,
+    dataset: str | None,
+    metrics: Iterable[str],
+    dimensions: Iterable[str],
+    filters: dict[str, str] | None,
+    from_time: str | None,
+    to_time: str | None,
+    limit: int,
+) -> tuple[str, dict[str, str]]:
+    dims = [_literal(d) for d in dimensions]
+    metric_list = [m for m in metrics if m.strip()]
+    if not metric_list:
+        raise AnalyticsError("INVALID_REQUEST", "At least one metric is required")
+    where = ["tenant_id = {tenant:String}"]
+    params: dict[str, str] = {"tenant": tenant, "limit": str(max(1, min(limit, 1000)))}
+    if dataset:
+        where.append("dataset_id = {dataset:String}")
+        params["dataset"] = dataset
+    metric_params = []
+    for i, metric in enumerate(metric_list):
+        key = f"metric{i}"
+        metric_params.append(f"{{{key}:String}}")
+        params[key] = metric
+    where.append(f"metric IN ({', '.join(metric_params)})")
+    for i, (key, value) in enumerate((filters or {}).items()):
+        dim = _literal(key)
+        p = f"filter{i}"
+        where.append(f"dim_values[indexOf(dim_names, '{dim}')] = {{{p}:String}}")
+        params[p] = value
+    if from_time:
+        where.append("event_time >= {from:String}")
+        params["from"] = from_time
+    if to_time:
+        where.append("event_time <= {to:String}")
+        params["to"] = to_time
+
+    select_dims = []
+    for i, dim in enumerate(dims):
+        select_dims.append(f"dim_values[indexOf(dim_names, '{dim}')] AS d{i}")
+    select = ", ".join(select_dims + ["count() AS value"])
+    group = f" GROUP BY {', '.join(f'd{i}' for i in range(len(dims)))}" if dims else ""
+    sql = (
+        f"SELECT {select} FROM {CLICKHOUSE_DATABASE}.opt_analytics_event "
+        f"WHERE {' AND '.join(where)}{group} ORDER BY value DESC LIMIT {{limit:UInt32}} FORMAT JSON"
+    )
+    return sql, params
+
+
+# ── models ─────────────────────────────────────────────────────────────────
+
+
+class QueryRequest(BaseModel):
+    dataset: str | None = None
+    metrics: list[str] = Field(default_factory=list)
+    dimensions: list[str] = Field(default_factory=list)
+    filters: dict[str, str] | None = None
+    timeRange: dict[str, str] | None = None
+    limit: int = 100
+
+
+# ── surface ────────────────────────────────────────────────────────────────
+
+
+@app.get("/health")
+def health() -> dict[str, Any]:
+    try:
+        ch_json("SELECT 1")
+        return {"status": "ok", "providers": {"clickhouse": {"status": "ok"}}}
+    except AnalyticsError as exc:
+        return {"status": "error", "providers": {"clickhouse": {"status": "error", "detail": exc.message}}}
+
+
+@app.get("/api/0pt/analytics/health")
+def analytics_health() -> dict[str, Any]:
+    return health()
+
+
+@app.get("/api/0pt/analytics/datasets")
+def list_datasets(request: Request, x_tenant_id: str | None = Header(default=None)) -> list[dict[str, Any]]:
+    tenant = verified_identity(request, x_tenant_id)
+    rows_ = rows(
+        f"SELECT dataset_id, count() AS events FROM {CLICKHOUSE_DATABASE}.opt_analytics_event "
+        "WHERE tenant_id = {tenant:String} GROUP BY dataset_id ORDER BY dataset_id FORMAT JSON",
+        {"tenant": tenant},
+    )
+    if not rows_:
+        rows_ = rows(
+            f"SELECT dataset_id, count() AS rows FROM {CLICKHOUSE_DATABASE}.opt_dataset_projection "
+            "WHERE tenant_id = {tenant:String} GROUP BY dataset_id ORDER BY dataset_id FORMAT JSON",
+            {"tenant": tenant},
+        )
+    return [
+        {
+            "id": r["dataset_id"],
+            "name": r["dataset_id"],
+            "tier": "SERVING",
+            "owner": "0pt",
+            "source": "0pt analytics",
+            "tenant_id": tenant,
+            **({"events": r.get("events")} if "events" in r else {}),
+            **({"rows": r.get("rows")} if "rows" in r else {}),
+        }
+        for r in rows_
+    ]
+
+
+@app.get("/api/0pt/analytics/datasets/{dataset_id}")
+def get_dataset(dataset_id: str, request: Request, x_tenant_id: str | None = Header(default=None)) -> dict[str, Any]:
+    tenant = verified_identity(request, x_tenant_id)
+    found = [d for d in list_datasets(request, x_tenant_id) if d["id"] == dataset_id]
+    if not found:
+        raise AnalyticsError("NOT_FOUND", f"dataset {dataset_id} does not exist for this tenant", 404)
+    return found[0]
+
+
+@app.get("/api/0pt/analytics/metrics")
+def list_metrics(request: Request, x_tenant_id: str | None = Header(default=None)) -> list[dict[str, Any]]:
+    tenant = verified_identity(request, x_tenant_id)
+    rows_ = rows(
+        f"SELECT metric, count() AS events FROM {CLICKHOUSE_DATABASE}.opt_analytics_event "
+        "WHERE tenant_id = {tenant:String} GROUP BY metric ORDER BY metric FORMAT JSON",
+        {"tenant": tenant},
+    )
+    if not rows_:
+        rows_ = rows(
+            f"SELECT metric, sum(count_state) AS events FROM {CLICKHOUSE_DATABASE}.opt_metric_daily "
+            "WHERE tenant_id = {tenant:String} GROUP BY metric ORDER BY metric FORMAT JSON",
+            {"tenant": tenant},
+        )
+    dims = rows(
+        f"SELECT DISTINCT arrayJoin(dim_names) AS name FROM {CLICKHOUSE_DATABASE}.opt_analytics_event "
+        "WHERE tenant_id = {tenant:String} FORMAT JSON",
+        {"tenant": tenant},
+    )
+    dimension_defs = [{"name": d["name"], "type": "string"} for d in dims]
+    return [
+        {
+            "id": r["metric"],
+            "name": r["metric"],
+            "description": f"0pt analytics metric {r['metric']}",
+            "datasets": ["aud0.events"],
+            "dimensions": dimension_defs,
+            "tenant_id": tenant,
+            **({"events": r.get("events")} if "events" in r else {}),
+        }
+        for r in rows_
+    ]
+
+
+@app.get("/api/0pt/analytics/metrics/{metric_id}")
+def get_metric(metric_id: str, request: Request, x_tenant_id: str | None = Header(default=None)) -> dict[str, Any]:
+    tenant = verified_identity(request, x_tenant_id)
+    found = [m for m in list_metrics(request, x_tenant_id) if m["id"] == metric_id]
+    if not found:
+        raise AnalyticsError("NOT_FOUND", f"metric {metric_id} does not exist for this tenant", 404)
+    return found[0]
+
+
+def _run_query(request: Request, x_tenant_id: str | None, body: QueryRequest, metric: str | None = None) -> dict[str, Any]:
+    tenant = verified_identity(request, x_tenant_id)
+    metrics = [metric] if metric else body.metrics
+    dataset = body.dataset or "aud0.events"
+    sql, params = build_query(
+        tenant=tenant,
+        dataset=dataset,
+        metrics=metrics,
+        dimensions=body.dimensions,
+        filters=body.filters,
+        from_time=(body.timeRange or {}).get("from"),
+        to_time=(body.timeRange or {}).get("to"),
+        limit=body.limit,
+    )
+    started = time.time()
+    data = ch_json(sql, params)
+    result_rows = data.get("data", [])
+    duration_ms = int((time.time() - started) * 1000)
+    columns = []
+    if result_rows:
+        columns = [{"name": key, "type": "string" if key.startswith("d") else "number"} for key in result_rows[0].keys()]
+    return {
+        "query_id": f"ch-{int(time.time() * 1000)}",
+        "status": "succeeded",
+        "engine": "clickhouse",
+        "submitted_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
+        "completed_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
+        "duration": duration_ms,
+        "rows": len(result_rows),
+        "tenant_id": tenant,
+        "result": {"columns": columns, "rows": result_rows, "truncated": len(result_rows) >= body.limit},
+    }
+
+
+@app.post("/api/0pt/analytics/query")
+def execute_query(body: QueryRequest, request: Request, x_tenant_id: str | None = Header(default=None)) -> dict[str, Any]:
+    return _run_query(request, x_tenant_id, body)
+
+
+@app.post("/api/0pt/analytics/metrics/{metric_id}/query")
+def execute_metric(metric_id: str, body: QueryRequest, request: Request, x_tenant_id: str | None = Header(default=None)) -> dict[str, Any]:
+    return _run_query(request, x_tenant_id, body, metric=metric_id)
+
+
+@app.post("/api/0pt/analytics/query/estimate")
+def estimate_query(body: QueryRequest, request: Request, x_tenant_id: str | None = Header(default=None)) -> dict[str, Any]:
+    tenant = verified_identity(request, x_tenant_id)
+    sql, params = build_query(
+        tenant=tenant,
+        dataset=body.dataset or "aud0.events",
+        metrics=body.metrics or ["event_count"],
+        dimensions=body.dimensions,
+        filters=body.filters,
+        from_time=(body.timeRange or {}).get("from"),
+        to_time=(body.timeRange or {}).get("to"),
+        limit=body.limit,
+    )
+    count = scalar(
+        f"SELECT count() FROM {CLICKHOUSE_DATABASE}.opt_analytics_event WHERE tenant_id = {{tenant:String}} FORMAT JSON",
+        {"tenant": tenant},
+    )
+    return {"estimated_rows_to_read": count or 0, "within_budget": True, "rows": count or 0}
+
+
+@app.post("/api/0pt/analytics/query/explain")
+def explain_query(body: QueryRequest, request: Request, x_tenant_id: str | None = Header(default=None)) -> dict[str, Any]:
+    tenant = verified_identity(request, x_tenant_id)
+    sql, _params = build_query(
+        tenant=tenant,
+        dataset=body.dataset or "aud0.events",
+        metrics=body.metrics or ["event_count"],
+        dimensions=body.dimensions,
+        filters=body.filters,
+        from_time=(body.timeRange or {}).get("from"),
+        to_time=(body.timeRange or {}).get("to"),
+        limit=body.limit,
+    )
+    return {"plan": sql, "forced_tenant_predicate": "tenant_id = {tenant:String}", "table": "opt_analytics_event"}
+
+
+@app.get("/api/0pt/analytics/metrics/{metric_id}/explain")
+def explain_metric(metric_id: str, request: Request, x_tenant_id: str | None = Header(default=None)) -> dict[str, Any]:
+    return explain_query(QueryRequest(metrics=[metric_id]), request, x_tenant_id)
+
+
+@app.get("/api/0pt/analytics/dashboards")
+def list_dashboards(request: Request, x_tenant_id: str | None = Header(default=None)) -> list[dict[str, Any]]:
+    verified_identity(request, x_tenant_id)
+    return []
+
+
+@app.get("/api/0pt/analytics/dashboards/{dashboard_id}")
+def get_dashboard(dashboard_id: str, request: Request, x_tenant_id: str | None = Header(default=None)) -> dict[str, Any]:
+    verified_identity(request, x_tenant_id)
+    raise AnalyticsError("NOT_FOUND", f"dashboard {dashboard_id} does not exist", 404)
+
+
+@app.get("/api/0pt/analytics/reports")
+def list_reports(request: Request, x_tenant_id: str | None = Header(default=None)) -> list[dict[str, Any]]:
+    verified_identity(request, x_tenant_id)
+    return []
+
+
+@app.get("/api/0pt/analytics/reports/{report_id}")
+def get_report(report_id: str, request: Request, x_tenant_id: str | None = Header(default=None)) -> dict[str, Any]:
+    verified_identity(request, x_tenant_id)
+    raise AnalyticsError("NOT_FOUND", f"report {report_id} does not exist", 404)
+
+
+@app.get("/api/0pt/analytics/search")
+def search_catalog(
+    request: Request,
+    q: str = Query(default=""),
+    types: str = Query(default=""),
+    limit: int = Query(default=20, ge=1, le=100),
+    x_tenant_id: str | None = Header(default=None),
+) -> dict[str, Any]:
+    tenant = verified_identity(request, x_tenant_id)
+    wanted = {t.strip() for t in types.split(",") if t.strip()}
+    hits: list[dict[str, Any]] = []
+    for dataset in list_datasets(request, x_tenant_id):
+        if (not wanted or "dataset" in wanted) and (not q or q.lower() in dataset["id"].lower()):
+            hits.append({**dataset, "type": "dataset", "canonical": False, "canonicalOperation": "getDataset"})
+    for metric in list_metrics(request, x_tenant_id):
+        if (not wanted or "metric" in wanted) and (not q or q.lower() in metric["id"].lower()):
+            hits.append({**metric, "type": "metric", "canonical": False, "canonicalOperation": "getMetric"})
+    return {"query": q, "derived": True, "hits": hits[:limit], "tenant_id": tenant}
+
+
+@app.get("/api/0pt/analytics/query/{query_id}")
+def get_query(query_id: str, request: Request, x_tenant_id: str | None = Header(default=None)) -> dict[str, Any]:
+    verified_identity(request, x_tenant_id)
+    raise AnalyticsError("NOT_FOUND", f"query {query_id} is not persisted", 404)
+
+
+@app.post("/api/0pt/analytics/query/{query_id}/cancel")
+def cancel_query(query_id: str, request: Request, x_tenant_id: str | None = Header(default=None)) -> dict[str, Any]:
+    verified_identity(request, x_tenant_id)
+    raise AnalyticsError("NOT_FOUND", f"query {query_id} is not running", 404)

services/analytics/ingest_aud0.py

@@ -0,0 +1,154 @@
+"""
+Backfill/refresh: AUD0 events -> 0pt ClickHouse analytics events.
+
+One audit event becomes one analytics event with metric `event_count` and
+dimensions `event_type`, `app_id`, `capability_id`, `risk_level`. The write is
+idempotent per event_id: already-present ids are skipped before the batch is
+inserted. No synthetic event is ever created; an empty AUD0 page writes
+nothing.
+
+Usage:
+  AUD0_TOKEN=... AUD0_TENANT_ID=379... CLICKHOUSE_URL=http://... \
+    python ingest_aud0.py --limit 50000
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import sys
+from typing import Any, Iterable
+
+import httpx
+
+AUD0_API_URL = os.getenv("AUD0_API_URL", "https://api.aud0.siax.io").rstrip("/")
+AUD0_TOKEN = os.getenv("AUD0_TOKEN", "")
+AUD0_TENANT_ID = os.getenv("AUD0_TENANT_ID", "")
+CLICKHOUSE_URL = os.getenv("CLICKHOUSE_URL", "").rstrip("/")
+CLICKHOUSE_USER = os.getenv("CLICKHOUSE_USER", "default")
+CLICKHOUSE_PASSWORD = os.getenv("CLICKHOUSE_PASSWORD", "")
+CLICKHOUSE_DATABASE = os.getenv("CLICKHOUSE_DATABASE", "opt_analytics")
+
+
+def dims(event: dict[str, Any]) -> tuple[list[str], list[str]]:
+    names: list[str] = []
+    values: list[str] = []
+    for key in ("event_type", "app_id", "capability_id", "risk_level"):
+        value = event.get(key)
+        if isinstance(value, str) and value:
+            names.append(key)
+            values.append(value)
+    return names, values
+
+
+def to_row(tenant: str, event: dict[str, Any]) -> dict[str, Any] | None:
+    event_id = event.get("event_id")
+    created_at = event.get("created_at")
+    event_type = event.get("event_type")
+    if not isinstance(event_id, str) or not event_id:
+        return None
+    if not isinstance(created_at, str) or not created_at:
+        return None
+    if not isinstance(event_type, str) or not event_type:
+        return None
+    names, values = dims(event)
+    return {
+        "tenant_id": tenant,
+        "project_id": "aud0",
+        "dataset_id": "aud0.events",
+        "metric": "event_count",
+        "event_id": event_id,
+        "event_time": created_at,
+        "value": 1,
+        "dim_names": names,
+        "dim_values": values,
+    }
+
+
+def ch_insert(rows: Iterable[dict[str, Any]]) -> None:
+    payload = "\n".join(json.dumps(row, ensure_ascii=False) for row in rows)
+    if not payload:
+        return
+    auth = (CLICKHOUSE_USER, CLICKHOUSE_PASSWORD) if CLICKHOUSE_PASSWORD else None
+    with httpx.Client(base_url=CLICKHOUSE_URL, auth=auth, timeout=60.0) as client:
+        response = client.post(
+            "/",
+            params={"database": CLICKHOUSE_DATABASE, "query": "INSERT INTO opt_analytics_event FORMAT JSONEachRow"},
+            content=payload,
+        )
+        response.raise_for_status()
+
+
+def existing_ids(tenant: str, ids: list[str]) -> set[str]:
+    if not ids:
+        return set()
+    auth = (CLICKHOUSE_USER, CLICKHOUSE_PASSWORD) if CLICKHOUSE_PASSWORD else None
+    found: set[str] = set()
+    sql = (
+        f"SELECT event_id FROM {CLICKHOUSE_DATABASE}.opt_analytics_event "
+        "WHERE tenant_id = {tenant:String} AND event_id IN ({ids:Array(String)}) FORMAT JSON"
+    )
+    with httpx.Client(base_url=CLICKHOUSE_URL, auth=auth, timeout=30.0) as client:
+        # ClickHouse query parameters travel in the URL; 1000 ids overflow it.
+        for start in range(0, len(ids), 50):
+            chunk = ids[start : start + 50]
+            response = client.post(
+                "/",
+                params={
+                    "database": CLICKHOUSE_DATABASE,
+                    "param_tenant": tenant,
+                    "param_ids": "[" + ",".join("'" + value.replace("'", "\\'") + "'" for value in chunk) + "]",
+                },
+                content=sql,
+            )
+            response.raise_for_status()
+            found.update(row["event_id"] for row in response.json().get("data", []))
+    return found
+
+
+def fetch_page(offset: int, limit: int) -> dict[str, Any]:
+    if not AUD0_TOKEN or not AUD0_TENANT_ID:
+        raise SystemExit("AUD0_TOKEN and AUD0_TENANT_ID are required")
+    with httpx.Client(base_url=AUD0_API_URL, timeout=30.0) as client:
+        response = client.get(
+            "/v1/aud0/events",
+            params={"tenant_id": AUD0_TENANT_ID, "limit": limit, "offset": offset},
+            headers={"authorization": f"Bearer {AUD0_TOKEN}", "x-tenant-id": AUD0_TENANT_ID},
+        )
+        response.raise_for_status()
+        return response.json()
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--limit", type=int, default=50_000, help="maximum events to read")
+    parser.add_argument("--page", type=int, default=1000)
+    parser.add_argument("--dry-run", action="store_true")
+    args = parser.parse_args()
+
+    if not CLICKHOUSE_URL:
+        raise SystemExit("CLICKHOUSE_URL is required")
+
+    written = skipped = seen = 0
+    for offset in range(0, args.limit, args.page):
+        page = fetch_page(offset, min(args.page, args.limit - offset))
+        events = page.get("events") or []
+        if not events:
+            break
+        rows = [row for row in (to_row(AUD0_TENANT_ID, event) for event in events) if row]
+        present = existing_ids(AUD0_TENANT_ID, [row["event_id"] for row in rows])
+        fresh = [row for row in rows if row["event_id"] not in present]
+        skipped += len(rows) - len(fresh)
+        seen += len(events)
+        if not args.dry_run:
+            ch_insert(fresh)
+        written += len(fresh)
+        print(json.dumps({"offset": offset, "read": len(events), "written": len(fresh), "skipped": skipped}))
+        if len(events) < args.page:
+            break
+    print(json.dumps({"seen": seen, "written": written, "skipped": skipped, "dry_run": args.dry_run}))
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

services/analytics/requirements.txt

@@ -0,0 +1,5 @@
+fastapi==0.115.6
+uvicorn[standard]==0.34.0
+httpx==0.28.1
+PyJWT[crypto]==2.14.0
+pytest==8.3.4

services/analytics/tests/test_app.py

@@ -0,0 +1,52 @@
+import sys
+from pathlib import Path
+
+import pytest
+from fastapi.testclient import TestClient
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+import app as analytics  # noqa: E402
+
+
+def test_build_query_forces_tenant_and_dimension_whitelist():
+    sql, params = analytics.build_query(
+        tenant="379",
+        dataset="aud0.events",
+        metrics=["event_count"],
+        dimensions=["event_type"],
+        filters={"app_id": "portal"},
+        from_time="2026-01-01T00:00:00Z",
+        to_time=None,
+        limit=50,
+    )
+    assert "tenant_id = {tenant:String}" in sql
+    assert "dataset_id = {dataset:String}" in sql
+    assert "dim_values[indexOf(dim_names, 'event_type')]" in sql
+    assert params["tenant"] == "379"
+    assert params["dataset"] == "aud0.events"
+    assert params["metric0"] == "event_count"
+    assert params["filter0"] == "portal"
+    with pytest.raises(analytics.AnalyticsError):
+        analytics.build_query(
+            tenant="379",
+            dataset=None,
+            metrics=["event_count"],
+            dimensions=["drop table"],
+            filters=None,
+            from_time=None,
+            to_time=None,
+            limit=10,
+        )
+
+
+def test_data_routes_require_a_verified_identity():
+    client = TestClient(analytics.app)
+    assert client.get("/api/0pt/analytics/datasets").status_code == 401
+    assert client.get("/api/0pt/analytics/metrics").status_code == 401
+
+
+def test_health_is_always_a_liveness_answer():
+    client = TestClient(analytics.app)
+    response = client.get("/health")
+    assert response.status_code == 200
+    assert "providers" in response.json()