gateway.py

#!/usr/bin/env python3
"""
gateway.py — Lightweight HTTP gateway for the Observatory

Sits on port 7860 (the only port HF Spaces exposes) and routes:

  GET  /api/refresh        →  git pull + run SQL (supports ?target=sql|data|all)
  GET  /api/status         →  sync state + health info
  GET  /api/sources        →  return current sources.yaml as JSON
  *    /*                  →  proxy everything else to ClickHouse on 8123

Key design:
  • Port 7860 binds IMMEDIATELY — no blocking on data sync
  • Initial data sync runs in a background thread
  • ClickHouse stays online even if all git operations fail
"""

import json
import os
import subprocess
import sys
import threading
import time
import urllib.request
import urllib.error
import yaml
from http.server import ThreadingHTTPServer, BaseHTTPRequestHandler
from urllib.parse import urlparse, parse_qs
from datetime import datetime, timezone

CLICKHOUSE_URL = "http://127.0.0.1:8123"
CONFIG_PATH = "/app/sources.yaml"
REFRESH_SCRIPT = "/app/refresh_sources.sh"
HOMEPAGE_PATH = "/app/index.html"
GATEWAY_PORT = 7860

# Load homepage HTML once at import time
try:
    with open(HOMEPAGE_PATH, "rb") as _f:
        HOMEPAGE_HTML = _f.read()
except FileNotFoundError:
    HOMEPAGE_HTML = b"<h1>Market-Data Observatory</h1><p><a href='/play'>Open SQL UI</a></p>"

# ── Global sync state (thread-safe via GIL for simple reads/writes) ──────────
sync_state = {
    "status": "pending",       # pending → syncing → ready / failed
    "started_at": None,
    "completed_at": None,
    "elapsed_seconds": None,
    "error": None,
    "last_refresh": None,
}
sync_lock = threading.Lock()


def update_sync_state(**kwargs):
    with sync_lock:
        sync_state.update(kwargs)


def get_sync_state():
    with sync_lock:
        return dict(sync_state)


# ── Background initial sync ─────────────────────────────────────────────────
def initial_sync():
    """Run full refresh in background. Gateway is already serving traffic."""
    update_sync_state(
        status="syncing",
        started_at=datetime.now(timezone.utc).isoformat(),
    )

    try:
        start = time.time()
        result = subprocess.run(
            ["bash", REFRESH_SCRIPT, "full"],
            capture_output=True, text=True, timeout=600,
        )
        elapsed = round(time.time() - start, 2)

        if result.returncode == 0:
            update_sync_state(
                status="ready",
                completed_at=datetime.now(timezone.utc).isoformat(),
                elapsed_seconds=elapsed,
                last_refresh=datetime.now(timezone.utc).isoformat(),
                error=None,
            )
            print(f"[gateway] Initial sync completed in {elapsed}s")
        else:
            update_sync_state(
                status="failed",
                completed_at=datetime.now(timezone.utc).isoformat(),
                elapsed_seconds=elapsed,
                error=result.stderr or result.stdout or "Unknown error",
            )
            print(f"[gateway] Initial sync failed (exit {result.returncode})")

        if result.stdout:
            print(result.stdout)
        if result.stderr:
            print(f"[gateway] stderr: {result.stderr}", file=sys.stderr)

    except subprocess.TimeoutExpired:
        update_sync_state(
            status="failed",
            completed_at=datetime.now(timezone.utc).isoformat(),
            error="Initial sync timed out after 600s",
        )
        print("[gateway] Initial sync timed out", file=sys.stderr)
    except Exception as e:
        update_sync_state(
            status="failed",
            completed_at=datetime.now(timezone.utc).isoformat(),
            error=str(e),
        )
        print(f"[gateway] Initial sync error: {e}", file=sys.stderr)


class GatewayHandler(BaseHTTPRequestHandler):

    # ── /api/refresh ─────────────────────────────────────────────────────
    def _handle_refresh(self):
        """Run refresh_sources.sh and return JSON result.

        Supports ?target=sql|data|all (default: full)
        """
        # Parse target parameter
        parsed = urlparse(self.path)
        params = parse_qs(parsed.query)
        target = params.get("target", ["full"])[0]

        # Map target to script mode
        mode_map = {"sql": "sql", "data": "data", "all": "full", "full": "full"}
        mode = mode_map.get(target, "full")

        try:
            start = time.time()
            result = subprocess.run(
                ["bash", REFRESH_SCRIPT, mode],
                capture_output=True, text=True, timeout=300,
            )
            elapsed = round(time.time() - start, 2)

            body = {
                "action": "refresh",
                "target": target,
                "elapsed_seconds": elapsed,
                "exit_code": result.returncode,
            }

            if result.stdout:
                body["stdout"] = result.stdout

            if result.stderr:
                body["stderr"] = result.stderr

            # Update global state
            update_sync_state(
                status="ready" if result.returncode == 0 else "failed",
                last_refresh=datetime.now(timezone.utc).isoformat(),
            )

            self._json_response(200, body)

        except subprocess.TimeoutExpired:
            self._json_response(504, {"error": "refresh timed out after 300s"})
        except Exception as e:
            self._json_response(500, {"error": str(e)})

    # ── /api/status ──────────────────────────────────────────────────────
    def _handle_status(self):
        """Return sync state + ClickHouse health."""
        state = get_sync_state()

        # Check ClickHouse health
        ch_status = "unknown"
        try:
            resp = urllib.request.urlopen(f"{CLICKHOUSE_URL}/ping", timeout=2)
            ch_status = "ok" if resp.read().decode().strip() == "Ok." else "degraded"
        except Exception:
            ch_status = "unreachable"

        body = {
            "clickhouse": ch_status,
            "data_sync": state,
        }

        self._json_response(200, body)

    # ── /api/sources ─────────────────────────────────────────────────────
    def _handle_sources(self):
        """Return current sources.yaml as JSON."""
        try:
            with open(CONFIG_PATH, "r") as f:
                config = yaml.safe_load(f)
            self._json_response(200, config)
        except Exception as e:
            self._json_response(500, {"error": str(e)})

    # ── Proxy to ClickHouse ──────────────────────────────────────────────
    def _proxy_to_clickhouse(self):
        """Forward the request to ClickHouse HTTP interface."""
        content_length = int(self.headers.get("Content-Length", 0))
        body = self.rfile.read(content_length) if content_length else None

        url = f"{CLICKHOUSE_URL}{self.path}"

        req = urllib.request.Request(url, data=body, method=self.command)

        # Forward relevant headers
        for header in self.headers:
            lower = header.lower()
            if lower not in ("host", "content-length", "transfer-encoding"):
                req.add_header(header, self.headers[header])

        try:
            resp = urllib.request.urlopen(req, timeout=600)
            self.send_response(resp.status)
            for key, val in resp.headers.items():
                if key.lower() != "transfer-encoding":
                    self.send_header(key, val)
            self.send_header("Access-Control-Allow-Origin", "*")
            self.end_headers()
            while True:
                chunk = resp.read(65536)
                if not chunk:
                    break
                self.wfile.write(chunk)

        except urllib.error.HTTPError as e:
            self.send_response(e.code)
            for key, val in e.headers.items():
                if key.lower() != "transfer-encoding":
                    self.send_header(key, val)
            self.send_header("Access-Control-Allow-Origin", "*")
            self.end_headers()
            self.wfile.write(e.read())

        except Exception as e:
            self._json_response(502, {"error": f"ClickHouse unreachable: {e}"})

    # ── Homepage ──────────────────────────────────────────────────────────
    def _handle_homepage(self):
        """Serve the custom landing page."""
        self.send_response(200)
        self.send_header("Content-Type", "text/html; charset=utf-8")
        self.send_header("Content-Length", str(len(HOMEPAGE_HTML)))
        self.send_header("Cache-Control", "no-cache")
        self.end_headers()
        self.wfile.write(HOMEPAGE_HTML)

    # ── HTTP method handlers ─────────────────────────────────────────────
    def do_GET(self):
        parsed = urlparse(self.path)
        path = parsed.path
        query = parsed.query

        # Serve homepage at root ONLY if no query params (/?query=... goes to ClickHouse)
        if path == "/" and not query:
            self._handle_homepage()
        elif path == "/api/refresh":
            self._handle_refresh()
        elif path == "/api/status":
            self._handle_status()
        elif path == "/api/sources":
            self._handle_sources()
        else:
            self._proxy_to_clickhouse()

    def do_POST(self):
        self._proxy_to_clickhouse()

    def do_OPTIONS(self):
        """CORS preflight."""
        self.send_response(204)
        self.send_header("Access-Control-Allow-Origin", "*")
        self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
        self.send_header("Access-Control-Allow-Headers", "*")
        self.end_headers()

    # ── Helpers ──────────────────────────────────────────────────────────
    def _json_response(self, code, data):
        body = json.dumps(data, indent=2).encode()
        self.send_response(code)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Access-Control-Allow-Origin", "*")
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, format, *args):
        sys.stderr.write(f"[gateway] {self.address_string()} {format % args}\n")


def wait_for_clickhouse(timeout=60):
    """Block until ClickHouse is responding on 8123."""
    print(f"[gateway] Waiting for ClickHouse on {CLICKHOUSE_URL}...")
    for i in range(timeout):
        try:
            urllib.request.urlopen(f"{CLICKHOUSE_URL}/ping", timeout=2)
            print(f"[gateway] ClickHouse is ready (took {i+1}s)")
            return True
        except Exception:
            time.sleep(1)
    print(f"[gateway] WARNING: ClickHouse not ready after {timeout}s, starting anyway")
    return False


if __name__ == "__main__":
    # Wait for ClickHouse (but don't crash if it's slow)
    wait_for_clickhouse()

    # Start the HTTP server FIRST — so HF Spaces health check passes immediately
    server = ThreadingHTTPServer(("0.0.0.0", GATEWAY_PORT), GatewayHandler)
    print(f"[gateway] Listening on port {GATEWAY_PORT}")
    print(f"[gateway]   /api/refresh?target=sql|data|all  → on-demand sync")
    print(f"[gateway]   /api/status                       → sync state + health")
    print(f"[gateway]   /api/sources                      → current config as JSON")
    print(f"[gateway]   /*                                → proxy to ClickHouse")

    # Run initial data sync in background thread (non-blocking)
    print("[gateway] Starting initial data sync in background...")
    sync_thread = threading.Thread(target=initial_sync, daemon=True)
    sync_thread.start()

    # Serve forever in the foreground
    server.serve_forever()