update

api/app/auth.py

@@ -0,0 +1,50 @@
+import sqlite3
+import os
+from pathlib import Path
+from fastapi import HTTPException, Header
+
+DB_PATH = os.getenv("DB_PATH", "/data/api_keys.db")
+SEED_API_KEY = os.getenv("SEED_API_KEY")
+SEED_GAME = os.getenv("SEED_GAME")
+
+def get_db():
+    # Ensure the directory exists
+    Path(DB_PATH).parent.mkdir(parents=True, exist_ok=True)
+    conn = sqlite3.connect(DB_PATH)
+    conn.row_factory = sqlite3.Row
+    return conn
+
+def init_db():
+    conn = get_db()
+    conn.execute("""
+        CREATE TABLE IF NOT EXISTS api_keys (
+            key TEXT PRIMARY KEY,
+            game TEXT NOT NULL,
+            active BOOLEAN DEFAULT 1,
+            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+        )
+    """)
+    # Insert seed key if provided via env
+    if SEED_API_KEY and SEED_GAME:
+        try:
+            conn.execute(
+                "INSERT OR IGNORE INTO api_keys(key, game) VALUES (?, ?)",
+                (SEED_API_KEY, SEED_GAME)
+            )
+            conn.commit()
+        except Exception:
+            pass
+    conn.close()
+
+async def verify_api_key(x_api_key: str = Header(None)) -> str:
+    if not x_api_key:
+        raise HTTPException(status_code=401, detail="X-API-Key header missing")
+    conn = get_db()
+    row = conn.execute(
+        "SELECT game FROM api_keys WHERE key = ? AND active = 1",
+        (x_api_key,)
+    ).fetchone()
+    conn.close()
+    if not row:
+        raise HTTPException(status_code=403, detail="Invalid or inactive API key")
+    return row["game"]

api/app/enrichment.py

@@ -0,0 +1,17 @@
+def classify_price(robux: int) -> str:
+    if robux <= 99:
+        return "low"
+    elif robux <= 499:
+        return "medium"
+    else:
+        return "high"
+
+def enrich_event(event, game: str):
+    """Adds game tag and derived fields."""
+    event.data = event.data or {}
+    # Inject game name into event data as a tag
+    event.data["game"] = game
+    # Derive price group for robux purchases
+    if event.type == "robux_purchase" and "robux" in event.data:
+        event.data["priceGroup"] = classify_price(event.data["robux"])
+    return event

api/app/influx.py

@@ -0,0 +1,46 @@
+import os
+import logging
+from influxdb_client import InfluxDBClient, Point
+from influxdb_client.client.write_api import SYNCHRONOUS
+
+logger = logging.getLogger("influx")
+
+INFLUX_URL = os.getenv("INFLUX_URL", "")
+INFLUX_TOKEN = os.getenv("INFLUX_TOKEN", "")
+INFLUX_ORG = os.getenv("INFLUX_ORG", "")
+INFLUX_BUCKET = os.getenv("INFLUX_BUCKET", "")
+
+client = None
+write_api = None
+
+if INFLUX_URL and INFLUX_TOKEN:
+    try:
+        client = InfluxDBClient(url=INFLUX_URL, token=INFLUX_TOKEN, org=INFLUX_ORG)
+        write_api = client.write_api(write_options=SYNCHRONOUS)
+        logger.info("InfluxDB client initialised")
+    except Exception as e:
+        logger.error(f"InfluxDB client creation failed: {e}")
+        client = None
+else:
+    logger.warning("InfluxDB environment variables missing – writing disabled")
+
+def write_event(event) -> bool:
+    if write_api is None:
+        logger.debug("InfluxDB not available, skipping write")
+        return False
+    try:
+        p = Point(event.type).time(event.time * 1_000_000_000)
+        p.tag("game", event.data.get("game", "unknown"))
+        p.tag("serverId", event.serverId)
+        # Optional tags
+        if "oreType" in event.data:
+            p.tag("oreType", event.data["oreType"])
+        # Write all numeric/string fields
+        for k, v in event.data.items():
+            if isinstance(v, (int, float, str, bool)):
+                p.field(k, v)
+        write_api.write(bucket=INFLUX_BUCKET, record=p)
+        return True
+    except Exception as e:
+        logger.error(f"Write failed: {e}")
+        return False

api/app/main.py

@@ -1,7 +1,26 @@
-from fastapi import FastAPI
+from fastapi import FastAPI, Depends, HTTPException
+from .auth import init_db, verify_api_key
+from .models import Event, BatchEvents
+from .enrichment import enrich_event
+from .influx import write_event
+import logging
+
+logging.basicConfig(level=logging.INFO)
+init_db()
 
 app = FastAPI(title="Signal - Roblox Telemetry")
 
 @app.get("/health")
 async def health():
-    return {"status": "ok"}
+    return {"status": "ok"}
+
+@app.post("/api/log")
+async def ingest_event(payload: Event | BatchEvents, game: str = Depends(verify_api_key)):
+    if isinstance(payload, BatchEvents):
+        for event in payload.events:
+            enriched = enrich_event(event, game)
+            write_event(enriched)
+    else:
+        enriched = enrich_event(payload, game)
+        write_event(enriched)
+    return {"success": True}

api/app/models.py

@@ -0,0 +1,11 @@
+from pydantic import BaseModel
+from typing import Any, List, Optional
+
+class Event(BaseModel):
+    type: str
+    time: int            # unix timestamp
+    serverId: str
+    data: Optional[dict] = {}
+
+class BatchEvents(BaseModel):
+    events: List[Event]