Downloader analytics version 1

server/installer/data-downloader/backend/app.py

@@ -11,9 +11,11 @@
 
 from fastapi import BackgroundTasks, FastAPI, HTTPException
 from fastapi.middleware.cors import CORSMiddleware
+from fastapi.middleware.gzip import GZipMiddleware
 from fastapi.responses import HTMLResponse
 from pydantic import BaseModel
 
+from backend import series_queries
 from backend.config import get_settings
 from backend.dbc_utils import group_sensors_by_message, load_dbc_db, refresh_dbc
 from backend.services import DataDownloaderService
@@ -43,6 +45,14 @@ class CanFramesBatchPayload(BaseModel):
     frames: List[CanFramePayload] = []
 
 
+class SeriesPayload(BaseModel):
+    season: str
+    signals: List[str]
+    start: datetime
+    end: datetime
+    target_points: int = series_queries.DEFAULT_TARGET_POINTS
+
+
 settings = get_settings()
 service = DataDownloaderService(settings)
 logger = logging.getLogger(__name__)
@@ -55,6 +65,7 @@ class CanFramesBatchPayload(BaseModel):
     allow_methods=["*"],
     allow_headers=["*"],
 )
+app.add_middleware(GZipMiddleware, minimum_size=1000)
 
 
 @app.get("/api/health")
@@ -193,6 +204,44 @@ def query_signal(payload: DataQueryPayload, season: str | None = None) -> dict:
         raise HTTPException(status_code=503, detail=f"Database query failed: {exc}")
 
 
+@app.post("/api/series")
+def query_series(payload: SeriesPayload) -> dict:
+    try:
+        series_queries.validate_request(
+            payload.signals, payload.start, payload.end, payload.target_points
+        )
+        # Metadata seasons use uppercase names; request season is the lowercase table.
+        season_by_table = {item["table"]: item for item in service.get_seasons()}
+        season_meta = season_by_table.get(payload.season)
+        if season_meta is None:
+            raise ValueError(f"unknown season table: {payload.season}")
+        known_signals = set(
+            service.get_sensors(season=season_meta["name"]).get("sensors", [])
+        )
+        unknown = sorted(set(payload.signals) - known_signals)
+        if unknown:
+            raise ValueError(f"unknown signal(s): {', '.join(unknown)}")
+        result = series_queries.execute_series_query(
+            postgres_dsn=settings.postgres_dsn,
+            table=payload.season,
+            signals=payload.signals,
+            start=payload.start,
+            end=payload.end,
+            target_points=payload.target_points,
+        )
+        return {
+            "season": payload.season,
+            "start": series_queries.normalize_utc(payload.start).isoformat(),
+            "end": series_queries.normalize_utc(payload.end).isoformat(),
+            "series": result,
+        }
+    except ValueError as exc:
+        raise HTTPException(status_code=400, detail=str(exc))
+    except Exception as exc:
+        logger.exception("Series query failed for %s", payload.signals)
+        raise HTTPException(status_code=503, detail=f"Database query failed: {exc}")
+
+
 # ── CAN frames batch ingest (for flight-recorder sync) ─────────────────────────
 
 # Cache of known signal columns per table, thread-safe

server/installer/data-downloader/backend/series_queries.py

@@ -0,0 +1,180 @@
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+import re
+from typing import Literal
+
+import psycopg2
+
+
+RAW_THRESHOLD = 100_000
+DEFAULT_TARGET_POINTS = 4_000
+MAX_TARGET_POINTS = 20_000
+MAX_SIGNALS = 12
+MAX_WINDOW_DAYS = 7
+MAX_TOTAL_POINTS = 1_500_000
+STATEMENT_TIMEOUT_MS = 15_000
+
+_IDENTIFIER = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
+
+
+def normalize_utc(dt: datetime) -> datetime:
+    if dt.tzinfo is None:
+        dt = dt.replace(tzinfo=timezone.utc)
+    return dt.astimezone(timezone.utc)
+
+
+def validate_request(
+    signals: list[str],
+    start: datetime,
+    end: datetime,
+    target_points: int,
+) -> None:
+    start_utc = normalize_utc(start)
+    end_utc = normalize_utc(end)
+    if not signals or len(signals) > MAX_SIGNALS:
+        raise ValueError(f"select between 1 and {MAX_SIGNALS} signals")
+    if len(set(signals)) != len(signals):
+        raise ValueError("signal names must be unique")
+    if any(not _IDENTIFIER.fullmatch(signal) for signal in signals):
+        raise ValueError("invalid signal name")
+    if start_utc >= end_utc:
+        raise ValueError("start must be before end")
+    if end_utc - start_utc > timedelta(days=MAX_WINDOW_DAYS):
+        raise ValueError("window cannot exceed seven days")
+    if not 1 <= target_points <= MAX_TARGET_POINTS:
+        raise ValueError(f"target_points must be between 1 and {MAX_TARGET_POINTS}")
+
+
+def bucket_interval(start: datetime, end: datetime, target_points: int) -> str:
+    total_ms = int((normalize_utc(end) - normalize_utc(start)).total_seconds() * 1000)
+    return f"{max(1, (total_ms + target_points - 1) // target_points)} milliseconds"
+
+
+def choose_mode(estimate: int) -> Literal["raw", "envelope"]:
+    return "raw" if estimate <= RAW_THRESHOLD else "envelope"
+
+
+def _quoted(signal: str) -> str:
+    if not _IDENTIFIER.fullmatch(signal):
+        raise ValueError("invalid signal name")
+    return f'"{signal}"'
+
+
+def _table(table: str) -> str:
+    if not re.fullmatch(r"[a-z][a-z0-9_]*", table):
+        raise ValueError("invalid season table")
+    return table
+
+
+def build_estimate_sql(table: str, signal: str) -> str:
+    column = _quoted(signal)
+    return (
+        "SELECT COUNT(*) FROM ("
+        f"SELECT 1 FROM {_table(table)} "
+        "WHERE time >= %(start)s AND time <= %(end)s "
+        f"AND {column} IS NOT NULL LIMIT {RAW_THRESHOLD + 1}"
+        ") AS bounded_count"
+    )
+
+
+def build_raw_sql(table: str, signal: str) -> str:
+    column = _quoted(signal)
+    return (
+        f"SELECT time, {column} FROM {_table(table)} "
+        "WHERE time >= %(start)s AND time <= %(end)s "
+        f"AND {column} IS NOT NULL ORDER BY time"
+    )
+
+
+def build_envelope_sql(table: str, signal: str) -> str:
+    column = _quoted(signal)
+    return (
+        "SELECT time_bucket(%(bucket)s::interval, time) AS bucket, "
+        f"MIN({column}), MAX({column}), AVG({column}) "
+        f"FROM {_table(table)} "
+        "WHERE time >= %(start)s AND time <= %(end)s "
+        f"AND {column} IS NOT NULL "
+        "GROUP BY bucket ORDER BY bucket"
+    )
+
+
+def response_estimate(
+    mode: Literal["raw", "envelope"],
+    raw_estimate: int,
+    target_points: int,
+) -> int:
+    return raw_estimate if mode == "raw" else target_points
+
+
+def _epoch_ms(value: datetime) -> int:
+    return int(normalize_utc(value).timestamp() * 1000)
+
+
+def execute_series_query(
+    *,
+    postgres_dsn: str,
+    table: str,
+    signals: list[str],
+    start: datetime,
+    end: datetime,
+    target_points: int,
+) -> dict[str, dict]:
+    start_utc = normalize_utc(start)
+    end_utc = normalize_utc(end)
+    validate_request(signals, start_utc, end_utc, target_points)
+    params = {"start": start_utc, "end": end_utc}
+    result: dict[str, dict] = {}
+
+    with psycopg2.connect(postgres_dsn) as conn:
+        with conn.cursor() as cur:
+            # One repeatable-read snapshot for estimate + fetch; cap query runtime.
+            cur.execute(
+                "SET TRANSACTION ISOLATION LEVEL REPEATABLE READ READ ONLY"
+            )
+            cur.execute(f"SET LOCAL statement_timeout = {STATEMENT_TIMEOUT_MS}")
+            estimates: dict[str, int] = {}
+            modes: dict[str, Literal["raw", "envelope"]] = {}
+            for signal in signals:
+                cur.execute(build_estimate_sql(table, signal), params)
+                estimates[signal] = int(cur.fetchone()[0])
+                modes[signal] = choose_mode(estimates[signal])
+
+            projected = sum(
+                response_estimate(modes[s], estimates[s], target_points) for s in signals
+            )
+            if projected > MAX_TOTAL_POINTS:
+                raise ValueError(
+                    f"estimated response exceeds {MAX_TOTAL_POINTS:,} points"
+                )
+
+            bucket = bucket_interval(start_utc, end_utc, target_points)
+            resolution_ms = int(bucket.split()[0])
+            for signal in signals:
+                mode = modes[signal]
+                if mode == "raw":
+                    cur.execute(build_raw_sql(table, signal), params)
+                    rows = cur.fetchall()
+                    result[signal] = {
+                        "mode": "raw",
+                        "resolution_ms": None,
+                        "point_count": len(rows),
+                        "t": [_epoch_ms(row[0]) for row in rows],
+                        "v": [float(row[1]) for row in rows],
+                    }
+                else:
+                    cur.execute(
+                        build_envelope_sql(table, signal),
+                        {**params, "bucket": bucket},
+                    )
+                    rows = cur.fetchall()
+                    result[signal] = {
+                        "mode": "envelope",
+                        "resolution_ms": resolution_ms,
+                        "point_count": len(rows),
+                        "t": [_epoch_ms(row[0]) for row in rows],
+                        "min": [float(row[1]) for row in rows],
+                        "max": [float(row[2]) for row in rows],
+                        "avg": [float(row[3]) for row in rows],
+                    }
+    return result

server/installer/data-downloader/backend/tests/test_compose_api_env_allowlist.py

@@ -0,0 +1,131 @@
+"""Assert data-downloader-api uses an explicit env allowlist, not broad env_file."""
+from __future__ import annotations
+
+import re
+from pathlib import Path
+
+
+# Env names config.py reads; compose must set each explicitly.
+REQUIRED_API_ENV = frozenset(
+    {
+        "DATA_DIR",
+        "POSTGRES_DSN",
+        "DEFAULT_SEASON_TABLE",
+        "SEASONS",
+        "SCANNER_BIN",
+        "SCANNER_INCLUDE_COUNTS",
+        "SCANNER_INITIAL_CHUNK_DAYS",
+        "SENSOR_WINDOW_DAYS",
+        "SENSOR_LOOKBACK_DAYS",
+        "SCAN_INTERVAL_SECONDS",
+        "SCAN_DAILY_TIME",
+        "ALLOWED_ORIGINS",
+        "GITHUB_DBC_TOKEN",
+        "GITHUB_DBC_REPO",
+        "GITHUB_DBC_BRANCH",
+        "GITHUB_DBC_PATH",
+        "DBC_FILE_PATH",
+    }
+)
+
+# Stack secrets that must never land in the API container.
+FORBIDDEN_API_ENV = (
+    "GRAFANA_API_TOKEN",
+    "SLACK_BOT_TOKEN",
+    "ANTHROPIC_API_KEY",
+)
+
+_COMPOSE_DEFAULT_RE = re.compile(r"\$\{[A-Z0-9_]+:-(.*)\}$")
+
+
+def _installer_compose_path() -> Path:
+    # backend/tests -> backend -> data-downloader -> installer
+    return Path(__file__).resolve().parents[3] / "docker-compose.yml"
+
+
+def _service_block(compose_text: str, service: str) -> str:
+    """Return the indented body of a top-level compose service (stdlib only)."""
+    marker = f"  {service}:"
+    lines = compose_text.splitlines()
+    start = None
+    for i, line in enumerate(lines):
+        if line == marker:
+            start = i + 1
+            break
+    if start is None:
+        raise AssertionError(f"service {service!r} not found in compose file")
+
+    body: list[str] = []
+    for line in lines[start:]:
+        # Next top-level key or sibling service (exactly two spaces + name:)
+        if line.startswith("  ") and not line.startswith("    "):
+            break
+        if line and not line.startswith(" ") and not line.startswith("\t"):
+            break
+        body.append(line)
+    return "\n".join(body)
+
+
+def _environment_mapping(service_body: str) -> dict[str, str]:
+    """Parse key/value pairs under the service's environment: mapping."""
+    lines = service_body.splitlines()
+    env_start = None
+    for i, line in enumerate(lines):
+        if line.strip() == "environment:" or line.rstrip() == "    environment:":
+            env_start = i + 1
+            break
+    if env_start is None:
+        return {}
+
+    mapping: dict[str, str] = {}
+    for line in lines[env_start:]:
+        stripped = line.strip()
+        if not stripped or stripped.startswith("#"):
+            continue
+        # Left environment block when indentation returns to service-property level
+        if line.startswith("    ") and not line.startswith("      "):
+            break
+        if ":" not in stripped:
+            continue
+        key, raw_value = stripped.split(":", 1)
+        key = key.strip()
+        if not key:
+            continue
+        mapping[key] = raw_value.strip().strip('"').strip("'")
+    return mapping
+
+
+def _compose_fallback(value: str) -> str | None:
+    """Return the :-default from ${VAR:-default}, or None if not that form."""
+    match = _COMPOSE_DEFAULT_RE.fullmatch(value.strip())
+    return match.group(1) if match else None
+
+
+def test_data_downloader_api_env_is_explicit_allowlist():
+    compose_path = _installer_compose_path()
+    assert compose_path.is_file(), f"missing compose file: {compose_path}"
+    text = compose_path.read_text(encoding="utf-8")
+    block = _service_block(text, "data-downloader-api")
+
+    assert "env_file:" not in block, (
+        "data-downloader-api must not use env_file; inject only an explicit allowlist"
+    )
+
+    for forbidden in FORBIDDEN_API_ENV:
+        assert forbidden not in block, (
+            f"data-downloader-api must not reference unrelated secret {forbidden}"
+        )
+
+    env_map = _environment_mapping(block)
+    env_keys = set(env_map)
+    assert env_keys == set(REQUIRED_API_ENV), (
+        "data-downloader-api environment keys must exactly match config.py allowlist; "
+        f"extra={sorted(env_keys - REQUIRED_API_ENV)} "
+        f"missing={sorted(REQUIRED_API_ENV - env_keys)}"
+    )
+
+    lookback_fallback = _compose_fallback(env_map["SENSOR_LOOKBACK_DAYS"])
+    assert lookback_fallback == "30", (
+        "SENSOR_LOOKBACK_DAYS compose fallback must match config.py default of 30 "
+        f"(got {lookback_fallback!r}); production .env may still set 365 explicitly"
+    )