server.py

"""
Mauritius Bus Route Planner — tiny stdlib HTTP server.

Serves the static frontend and a small JSON API backed by busapi.py
(journey planning via mauritius-buses.com) and geocode.py (stop -> coordinates
via OpenStreetMap / Nominatim).

    python3 server.py            # http://localhost:8000
    PORT=9000 python3 server.py

Endpoints:
    GET /api/stops                       -> [{id, name}, ...]
    GET /api/plan?day=&from=&to=         -> itinerary with coords attached
"""

import hashlib
import json
import math
import os
import sys
import threading
import time
import traceback
import warnings
import gzip as _gzip
from email.utils import formatdate
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import urlparse, parse_qs

# Text assets worth gzipping on the wire (the bundled network.js/json are ~2.3MB
# uncompressed but ~5x smaller gzipped — a big win for first load and the Android
# data refresh). Binary types (png/ico) are already compressed; skip them.
GZIP_EXTS = {".html", ".js", ".css", ".json", ".svg"}

# Silence the harmless LibreSSL/urllib3 notice on stock macOS Python.
warnings.filterwarnings("ignore", message=".*OpenSSL.*")

import busapi
import events as events_mod
import geocode
import nearby
import roads

try:
    import psutil  # process + system CPU/RAM metrics for /metrics

    _PROC = psutil.Process()
    _PROC.cpu_percent(interval=None)  # prime the per-process CPU delta
    psutil.cpu_percent(interval=None)  # prime the system CPU delta
except Exception:  # noqa: BLE001 — metrics are best-effort; never block startup
    psutil = None
    _PROC = None

ROOT = os.path.dirname(os.path.abspath(__file__))
PUBLIC = os.path.join(ROOT, "public")
PORT = int(os.environ.get("PORT", "8000"))

# /api/health reports "degraded" (HTTP 503) below this many cached events, so a
# monitor can catch a silently-broken scraper (e.g. a source site redesign).
EVENTS_MIN_HEALTHY = int(os.environ.get("EVENTS_MIN_HEALTHY", "5"))

# Client error beacon: clients POST a tiny JSON blob ({msg, where, platform}) to
# /api/clienterror when a JS error / unhandled rejection fires, so real-world
# breakage is visible in the Space logs. Privacy-first: NOTHING is stored or
# forwarded — we only log to stderr, never persist, and carry no identifiers.
# A coarse global rate limit stops a looping client from flooding the log.
_CLIENT_ERR_MAX_BYTES = 4096
_CLIENT_ERR_WINDOW = 60.0  # seconds
_CLIENT_ERR_MAX_PER_WINDOW = 30
_client_err_hits = []  # recent timestamps (trimmed each call)

# Per-IP rate limit on /api/*. A token bucket smooths bursts while stopping any
# single client from flooding the upstream proxies (Overpass, Nominatim, the
# event scrape). Tuned so normal app use never trips it: a bucket of RATE_BURST
# tokens, refilled RATE_RPS/sec. Tunable via env on the Space.
RATE_RPS = float(os.environ.get("API_RATE_RPS", "5"))  # sustained req/s per IP
RATE_BURST = float(os.environ.get("API_RATE_BURST", "40"))  # burst bucket capacity


class _RateLimiter:
    """Token bucket keyed by client IP. Thread-safe (ThreadingHTTPServer), and
    self-pruning so a flood of unique IPs can't grow the map without bound."""

    def __init__(self, rps, burst):
        self.rps = rps
        self.burst = burst
        self._buckets = {}  # ip -> [tokens, last_monotonic]
        self._lock = threading.Lock()

    def allow(self, ip):
        now = time.monotonic()
        with self._lock:
            tokens, last = self._buckets.get(ip, (self.burst, now))
            tokens = min(self.burst, tokens + (now - last) * self.rps)
            ok = tokens >= 1.0
            self._buckets[ip] = (tokens - 1.0 if ok else tokens, now)
            if len(self._buckets) > 4096:  # prune idle/refilled buckets
                self._buckets = {
                    k: v
                    for k, v in self._buckets.items()
                    if v[0] < self.burst or (now - v[1]) < 120
                }
            return ok


RATE_LIMITER = _RateLimiter(RATE_RPS, RATE_BURST)

# External services we proxy. Map each hostname to a stable, low-cardinality
# label so the dashboard can chart per-upstream error rate and latency — the
# usual root cause of a user-facing failure, otherwise hidden in route latency.
_UPSTREAM_HOSTS = {
    "overpass-api.de": "overpass",
    "nominatim.openstreetmap.org": "nominatim",
    "router.project-osrm.org": "osrm",
    "routing.openstreetmap.de": "osrm",
    "partyapp.mu": "events_partyapp",
    "www.partyapp.mu": "events_partyapp",
    "otayo.com": "events_otayo",
    "www.otayo.com": "events_otayo",
}


def _upstream_label(url):
    try:
        return _UPSTREAM_HOSTS.get(urlparse(url).hostname or "", "other")
    except Exception:  # noqa: BLE001
        return "other"


def _instrument_requests():
    """Wrap requests.get/post once to time external calls and record per-upstream
    success/latency in METRICS — without touching the worker modules (they call
    requests.get/post by attribute, so this is picked up transparently). A
    timeout/connection error counts as an error and is re-raised unchanged."""
    try:
        import requests as _rq
    except Exception:  # noqa: BLE001
        return
    for name in ("get", "post"):
        orig = getattr(_rq, name, None)
        if orig is None or getattr(orig, "_mobaz_wrapped", False):
            continue

        def make(orig):
            def wrapper(url, *a, **kw):
                dep = _upstream_label(url if isinstance(url, str) else "")
                t0 = time.perf_counter()
                ok = False
                try:
                    resp = orig(url, *a, **kw)
                    ok = getattr(resp, "ok", True)
                    return resp
                finally:
                    METRICS.observe_upstream(dep, time.perf_counter() - t0, ok)

            wrapper._mobaz_wrapped = True
            return wrapper

        setattr(_rq, name, make(orig))


# ---------------------------------------------------------------------------
# Metrics — a tiny, dependency-free Prometheus exposition. Prometheus (running
# in the same container) scrapes /metrics; Grafana then renders it. We keep the
# label cardinality bounded by mapping each request to a fixed "route" label
# (never the raw path/query) so the series count stays small and stable.
# ---------------------------------------------------------------------------
_LATENCY_BUCKETS = (0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10)
_KNOWN_ROUTES = {
    "/api/plan",
    "/api/stops",
    "/api/nearby",
    "/api/geocode",
    "/api/walk",
    "/api/events",
    "/api/health",
    "/api/clienterror",
    "/metrics",
    "/status",
}


def _route_label(path):
    """Collapse a request path to a low-cardinality route label."""
    if path in _KNOWN_ROUTES:
        return path
    if path.startswith("/api/"):
        return "/api/other"
    if path.startswith("/data/"):
        return "/data"
    if path in ("/", "", "/index.html"):
        return "/"
    return "/static"


class _Metrics:
    def __init__(self):
        self._lock = threading.Lock()
        self.start = time.time()
        self.requests = {}  # (route, method, code) -> count
        self.bucket_counts = [0] * len(_LATENCY_BUCKETS)
        self.duration_sum = 0.0
        self.duration_count = 0
        self.client_errors = 0
        self.beacons_dropped = 0  # rate-limited client beacons
        self.inflight = 0  # requests currently being served
        self.clients = []  # sliding window of (ts, client_ip)
        self.plan_results = {}  # outcome -> count (success/no_route/...)
        self.upstream = {}  # dep -> {ok, err, sum} for external calls

    def observe(self, route, method, code, dur):
        with self._lock:
            key = (route, method, int(code))
            self.requests[key] = self.requests.get(key, 0) + 1
            self.duration_sum += dur
            self.duration_count += 1
            for i, le in enumerate(_LATENCY_BUCKETS):
                if dur <= le:
                    self.bucket_counts[i] += 1

    def observe_plan(self, outcome):
        """Count journey-plan outcomes so 'no route found' is visible (a data
        gap), distinct from a successful plan or a server error."""
        with self._lock:
            self.plan_results[outcome] = self.plan_results.get(outcome, 0) + 1

    def observe_upstream(self, dep, dur, ok):
        """Record one external dependency call (Overpass/Nominatim/OSRM/scrape):
        count, error count, and total time so the dashboard can show per-upstream
        error rate and average latency — the usual root cause of user-facing
        failures, otherwise buried inside route latency."""
        with self._lock:
            s = self.upstream.get(dep)
            if s is None:
                s = self.upstream[dep] = {"ok": 0, "err": 0, "sum": 0.0}
            s["ok" if ok else "err"] += 1
            s["sum"] += dur

    def inc_client_error(self, dropped=False):
        with self._lock:
            if dropped:
                self.beacons_dropped += 1
            else:
                self.client_errors += 1

    def inflight_inc(self):
        with self._lock:
            self.inflight += 1

    def inflight_dec(self):
        with self._lock:
            self.inflight = max(0, self.inflight - 1)

    def record_client(self, ip):
        now = time.time()
        with self._lock:
            self.clients.append((now, ip))
            if len(self.clients) > 5000:  # hard cap so a flood can't grow it unbounded
                self.clients = self.clients[-5000:]

    def active_clients(self, window=300):
        """Unique client IPs seen in the last `window` seconds (a stateless API
        has no real 'connections', so this is the meaningful 'active clients')."""
        cutoff = time.time() - window
        with self._lock:
            self.clients = [(t, ip) for (t, ip) in self.clients if t >= cutoff]
            return len({ip for _, ip in self.clients})

    def _system_metrics(self):
        """Process + host CPU/RAM via psutil (best-effort; empty if unavailable)."""
        out = {}
        if psutil is None or _PROC is None:
            return out
        try:
            out["mobaz_process_cpu_percent"] = float(_PROC.cpu_percent(interval=None))
            out["mobaz_process_resident_memory_bytes"] = int(_PROC.memory_info().rss)
            out["mobaz_process_threads"] = int(_PROC.num_threads())
            try:
                out["mobaz_process_open_fds"] = int(_PROC.num_fds())
            except Exception:  # noqa: BLE001 — not on every platform
                pass
            out["mobaz_system_cpu_percent"] = float(psutil.cpu_percent(interval=None))
            vm = psutil.virtual_memory()
            out["mobaz_system_memory_used_percent"] = float(vm.percent)
            out["mobaz_system_memory_total_bytes"] = int(vm.total)
            try:
                out["mobaz_system_load1"] = float(os.getloadavg()[0])
            except Exception:  # noqa: BLE001 — no getloadavg on some platforms
                pass
        except Exception:  # noqa: BLE001
            pass
        return out

    def _gauges(self):
        """Best-effort point-in-time gauges (never raise into /metrics)."""
        out = {}
        try:
            out["mobaz_stops"] = len(busapi.get_stops())
        except Exception:  # noqa: BLE001
            out["mobaz_stops"] = 0
        try:
            _ev = events_mod.health()
            out["mobaz_events_cached"] = _ev.get("count", 0)
            # Seconds since the last successful scrape (-1 = never): a freshness
            # signal that catches a silently-broken scraper even while it serves
            # the last good list.
            _age = _ev.get("age_seconds")
            out["mobaz_events_age_seconds"] = int(_age) if _age is not None else -1
        except Exception:  # noqa: BLE001
            out["mobaz_events_cached"] = 0
        try:
            out["mobaz_geocode_cache_entries"] = len(geocode._load_coords_cache())
        except Exception:  # noqa: BLE001
            out["mobaz_geocode_cache_entries"] = 0
        try:
            with nearby._cache_lock:
                out["mobaz_nearby_cache_entries"] = len(nearby._cache)
        except Exception:  # noqa: BLE001
            out["mobaz_nearby_cache_entries"] = 0
        return out

    def render(self):
        """Prometheus text exposition (version 0.0.4)."""
        with self._lock:
            requests = dict(self.requests)
            buckets = list(self.bucket_counts)
            dur_sum, dur_count = self.duration_sum, self.duration_count
            client_errors, dropped = self.client_errors, self.beacons_dropped
            inflight = self.inflight
            uptime = time.time() - self.start
            plan_results = dict(self.plan_results)
            upstream = {k: dict(v) for k, v in self.upstream.items()}
        L = []
        L.append("# HELP mobaz_up Whether the MoBaz backend is serving (always 1).")
        L.append("# TYPE mobaz_up gauge")
        L.append("mobaz_up 1")
        L.append("# HELP mobaz_uptime_seconds Seconds since the backend started.")
        L.append("# TYPE mobaz_uptime_seconds gauge")
        L.append("mobaz_uptime_seconds %.0f" % uptime)
        L.append(
            "# HELP mobaz_http_requests_total HTTP requests by route, method and code."
        )
        L.append("# TYPE mobaz_http_requests_total counter")
        for (route, method, code), n in sorted(requests.items()):
            L.append(
                'mobaz_http_requests_total{route="%s",method="%s",code="%d"} %d'
                % (route, method, code, n)
            )
        L.append(
            "# HELP mobaz_http_request_duration_seconds Request latency histogram."
        )
        L.append("# TYPE mobaz_http_request_duration_seconds histogram")
        cumulative = 0
        for i, le in enumerate(_LATENCY_BUCKETS):
            cumulative = buckets[i]
            L.append(
                'mobaz_http_request_duration_seconds_bucket{le="%s"} %d'
                % (le, cumulative)
            )
        L.append('mobaz_http_request_duration_seconds_bucket{le="+Inf"} %d' % dur_count)
        L.append("mobaz_http_request_duration_seconds_sum %.6f" % dur_sum)
        L.append("mobaz_http_request_duration_seconds_count %d" % dur_count)
        L.append(
            "# HELP mobaz_client_errors_total Client-side JS errors reported via the beacon."
        )
        L.append("# TYPE mobaz_client_errors_total counter")
        L.append("mobaz_client_errors_total %d" % client_errors)
        L.append(
            "# HELP mobaz_client_error_beacons_dropped_total Beacons dropped by the rate limit."
        )
        L.append("# TYPE mobaz_client_error_beacons_dropped_total counter")
        L.append("mobaz_client_error_beacons_dropped_total %d" % dropped)
        L.append("# HELP mobaz_inflight_requests Requests currently being served.")
        L.append("# TYPE mobaz_inflight_requests gauge")
        L.append("mobaz_inflight_requests %d" % inflight)
        L.append(
            "# HELP mobaz_active_clients Unique client IPs seen in the last 5 minutes."
        )
        L.append("# TYPE mobaz_active_clients gauge")
        L.append("mobaz_active_clients %d" % self.active_clients())
        # Journey-plan outcomes (success / no_route / error) — surfaces data gaps.
        L.append(
            "# HELP mobaz_plan_results_total Journey-plan outcomes by result type."
        )
        L.append("# TYPE mobaz_plan_results_total counter")
        for outcome, n in sorted(plan_results.items()):
            L.append('mobaz_plan_results_total{result="%s"} %d' % (outcome, n))
        # External dependency calls (Overpass/Nominatim/OSRM/scrape): count + time.
        L.append(
            "# HELP mobaz_upstream_requests_total External calls by dependency and outcome."
        )
        L.append("# TYPE mobaz_upstream_requests_total counter")
        for dep, s in sorted(upstream.items()):
            L.append(
                'mobaz_upstream_requests_total{dep="%s",outcome="ok"} %d'
                % (dep, s["ok"])
            )
            L.append(
                'mobaz_upstream_requests_total{dep="%s",outcome="error"} %d'
                % (dep, s["err"])
            )
        L.append(
            "# HELP mobaz_upstream_request_duration_seconds_sum Total time in external calls."
        )
        L.append("# TYPE mobaz_upstream_request_duration_seconds_sum counter")
        for dep, s in sorted(upstream.items()):
            L.append(
                'mobaz_upstream_request_duration_seconds_sum{dep="%s"} %.6f'
                % (dep, s["sum"])
            )
        # Data/cache gauges + process & host CPU/RAM (psutil).
        allg = {}
        allg.update(self._gauges())
        allg.update(self._system_metrics())
        for name, val in allg.items():
            L.append("# TYPE %s gauge" % name)
            L.append(("%s %d" if isinstance(val, int) else "%s %.4f") % (name, val))
        return "\n".join(L) + "\n"


METRICS = _Metrics()
_instrument_requests()  # time external Overpass/Nominatim/OSRM/scrape calls


def _status_html():
    """A dependency-free server-rendered metrics summary. Grafana (at /) is the
    rich view; this stays up even if Grafana is down and needs no datasource."""
    with METRICS._lock:
        reqs = dict(METRICS.requests)
        uptime = time.time() - METRICS.start
        dur_sum, dur_count = METRICS.duration_sum, METRICS.duration_count
        client_errors = METRICS.client_errors
    gauges = METRICS._gauges()
    total = sum(reqs.values())
    avg_ms = (dur_sum / dur_count * 1000) if dur_count else 0.0
    active = METRICS.active_clients()
    inflight = METRICS.inflight
    # Memory only (reading RSS is side-effect-free; CPU% would reset the delta
    # that the Prometheus scrape relies on, so leave CPU to Grafana).
    rss_mb = 0
    if _PROC is not None:
        try:
            rss_mb = _PROC.memory_info().rss / (1024 * 1024)
        except Exception:  # noqa: BLE001
            rss_mb = 0
    # Requests grouped by route (sum across method/code).
    by_route = {}
    for (route, _m, _c), n in reqs.items():
        by_route[route] = by_route.get(route, 0) + n
    rows = (
        "".join(
            "<tr><td>%s</td><td>%d</td></tr>" % (r, n)
            for r, n in sorted(by_route.items(), key=lambda kv: -kv[1])
        )
        or "<tr><td colspan=2>no requests yet</td></tr>"
    )
    cards = "".join(
        "<div class=card><div class=k>%s</div><div class=v>%s</div></div>" % (k, v)
        for k, v in [
            ("uptime", "%dh %dm" % (uptime // 3600, (uptime % 3600) // 60)),
            ("requests", total),
            ("avg latency", "%.1f ms" % avg_ms),
            ("active clients (5m)", active),
            ("in-flight", inflight),
            ("memory", "%.0f MB" % rss_mb),
            ("stops", gauges.get("mobaz_stops", 0)),
            ("events cached", gauges.get("mobaz_events_cached", 0)),
            ("geocode cache", gauges.get("mobaz_geocode_cache_entries", 0)),
            ("nearby cache", gauges.get("mobaz_nearby_cache_entries", 0)),
            ("client errors", client_errors),
        ]
    )
    return (
        "<!doctype html><meta charset=utf-8><title>MoBaz · status</title>"
        "<meta name=viewport content='width=device-width,initial-scale=1'>"
        "<style>body{font:15px/1.5 system-ui,sans-serif;margin:0;background:#0b1020;color:#e7ecf5;padding:24px}"
        "h1{font-size:20px;margin:0 0 4px}.sub{color:#9fb0cc;margin:0 0 20px}"
        ".grid{display:grid;grid-template-columns:repeat(auto-fill,minmax(150px,1fr));gap:12px;margin-bottom:24px}"
        ".card{background:#161d33;border:1px solid #243056;border-radius:12px;padding:14px}"
        ".k{color:#9fb0cc;font-size:12px;text-transform:uppercase;letter-spacing:.04em}"
        ".v{font-size:22px;font-weight:700;margin-top:4px}"
        "table{border-collapse:collapse;width:100%;max-width:480px;background:#161d33;border-radius:12px;overflow:hidden}"
        "td{padding:8px 14px;border-top:1px solid #243056}tr:first-child td{border-top:0}"
        "a{color:#7aa2ff}.links{margin-top:20px}</style>"
        "<h1>MoBaz · backend status</h1>"
        "<p class=sub>Live counters from the running Space. Full dashboards in "
        "<a href='/'>Grafana</a>.</p>"
        "<div class=grid>" + cards + "</div>"
        "<h2 style='font-size:15px'>Requests by route</h2>"
        "<table>" + rows + "</table>"
        "<p class=links><a href='/metrics'>raw /metrics</a> · "
        "<a href='/api/health'>/api/health</a> · <a href='/app'>open the app</a></p>"
    )


CONTENT_TYPES = {
    ".html": "text/html; charset=utf-8",
    ".js": "application/javascript; charset=utf-8",
    ".css": "text/css; charset=utf-8",
    ".json": "application/json; charset=utf-8",
    ".svg": "image/svg+xml",
    ".png": "image/png",
    ".ico": "image/x-icon",
}


# --------------------------------------------------------------------------- #
# Geocoding sanity: drop mis-resolved stops.
#
# Many Mauritian places share a name (several "Belle Vue", "St. Martin", ...),
# so a stop can resolve to a same-named village on the far side of the island.
# Such a point forces a large *detour* relative to going straight from its
# previous to its next neighbour, whereas a correctly placed stop sits roughly
# on the line between them. We null out the worst detour-causing points until
# the path is consistent. This leaves genuinely distant endpoints alone.
# --------------------------------------------------------------------------- #
DETOUR_KM = 9.0  # interior point forcing more extra distance than this -> drop
ENDPOINT_KM = 13.0  # terminal stop jumping more than this to its neighbour -> drop


def _km(a, b):
    return math.hypot((a[0] - b[0]) * 111.0, (a[1] - b[1]) * 105.0)


def _despike(stops):
    """Remove (null) coordinates that are clearly mis-geocoded for this leg."""
    # Iteratively drop the single worst off-corridor node.
    while True:
        idx = [i for i, s in enumerate(stops) if s.get("coord")]
        if len(idx) < 3:
            break
        # Group consecutive resolved stops at (almost) the same spot into one
        # node, so a block of identical bad points is handled together.
        nodes = []  # [[positions_in_idx], coord]
        for k, i in enumerate(idx):
            c = stops[i]["coord"]
            if nodes and _km(nodes[-1][1], c) < 0.4:
                nodes[-1][0].append(k)
            else:
                nodes.append([[k], c])
        if len(nodes) < 3:
            break
        worst = None
        for n in range(1, len(nodes) - 1):
            p, c, q = nodes[n - 1][1], nodes[n][1], nodes[n + 1][1]
            detour = _km(p, c) + _km(c, q) - _km(p, q)
            if detour > DETOUR_KM and (worst is None or detour > worst[0]):
                worst = (detour, n)
        if not worst:
            break
        for k in nodes[worst[1]][0]:
            stops[idx[k]]["coord"] = None

    # One-sided endpoint check: a terminal stop jumping implausibly far to its
    # nearest surviving neighbour is almost certainly wrong.
    idx = [i for i, s in enumerate(stops) if s.get("coord")]
    if (
        len(idx) >= 2
        and _km(stops[idx[0]]["coord"], stops[idx[1]]["coord"]) > ENDPOINT_KM
    ):
        stops[idx[0]]["coord"] = None
    idx = [i for i, s in enumerate(stops) if s.get("coord")]
    if (
        len(idx) >= 2
        and _km(stops[idx[-1]]["coord"], stops[idx[-2]]["coord"]) > ENDPOINT_KM
    ):
        stops[idx[-1]]["coord"] = None


def _attach_coords(result):
    """Resolve every stop name in the itinerary to [lat, lon] in place."""
    names = []
    for opt in result.get("options", []):
        for leg in opt["legs"]:
            if leg["type"] == "bus":
                names += [s["name"] for s in leg["stops"]]
            elif leg.get("to"):
                names.append(leg["to"])
    coords = geocode.resolve_many(names)
    for opt in result.get("options", []):
        for leg in opt["legs"]:
            if leg["type"] == "bus":
                for s in leg["stops"]:
                    s["coord"] = coords.get(s["name"])
                _despike(leg["stops"])
                pts = [s["coord"] for s in leg["stops"] if s["coord"]]
                leg["fromCoord"] = pts[0] if pts else None
                leg["toCoord"] = pts[-1] if pts else None
                # Snap to roads so the leg follows streets, not straight hops.
                leg["geometry"] = roads.snap_route(pts)
            else:
                leg["toCoord"] = coords.get(leg.get("to"))
    return result


class Handler(BaseHTTPRequestHandler):
    server_version = "MauritiusBusPlanner/1.0"

    def log_message(self, fmt, *args):
        sys.stderr.write("%s - %s\n" % (self.address_string(), fmt % args))

    # --- metrics instrumentation -------------------------------------------
    # send_response is the single choke point every reply flows through (JSON,
    # static file, 204, error), so we record the request count + latency here
    # exactly once, using the route/start stamped by the dispatcher.
    def send_response(self, code, message=None):
        super().send_response(code, message)
        if not getattr(self, "_recorded", False):
            self._recorded = True
            t0 = getattr(self, "_t0", None)
            if t0 is not None:
                route = getattr(self, "_route", _route_label(urlparse(self.path).path))
                METRICS.observe(
                    route, self.command or "GET", code, time.perf_counter() - t0
                )
                METRICS.inflight_dec()

    def _begin(self, path):
        """Stamp route/timer, count the request as in-flight, and note the client
        IP (behind nginx the real IP is in X-Forwarded-For)."""
        self._t0 = time.perf_counter()
        self._route = _route_label(path)
        METRICS.inflight_inc()
        xff = self.headers.get("X-Forwarded-For")
        ip = (
            xff.split(",")[0].strip()
            if xff
            else (self.client_address[0] if self.client_address else "?")
        )
        self._client_ip = ip
        METRICS.record_client(ip)

    def _cors(self):
        # The packaged app calls this API from a file:// origin, so allow any.
        self.send_header("Access-Control-Allow-Origin", "*")
        self.send_header("Access-Control-Allow-Methods", "GET, OPTIONS")

    def do_OPTIONS(self):
        self.send_response(204)
        self._cors()
        self.send_header("Content-Length", "0")
        self.end_headers()

    def _send_json(self, obj, status=200):
        body = json.dumps(obj, ensure_ascii=False).encode("utf-8")
        gzipped = False
        if self._accepts_gzip() and len(body) > 1024:
            body = _gzip.compress(body, 6)
            gzipped = True
        self.send_response(status)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        if gzipped:
            self.send_header("Content-Encoding", "gzip")
            self.send_header("Vary", "Accept-Encoding")
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Cache-Control", "no-store")
        self._cors()
        self.end_headers()
        self.wfile.write(body)

    def _etag_matches(self, etag):
        """True if the client's If-None-Match lists `etag` (so we can 304)."""
        inm = self.headers.get("If-None-Match")
        return bool(inm) and etag in [t.strip() for t in inm.split(",")]

    def _send_304(self, etag):
        self.send_response(304)
        self.send_header("ETag", etag)
        self.send_header("Cache-Control", "no-cache")
        self._cors()
        self.end_headers()

    def _send_rate_limited(self):
        body = b'{"error":"rate limited, slow down"}'
        self.send_response(429)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Retry-After", "2")
        self._cors()
        self.end_headers()
        self.wfile.write(body)

    def _send_json_cacheable(self, obj):
        """Like _send_json but with an ETag so an unchanged body costs the client
        only a 304 (no re-download). Used for /api/stops, which the apps fetch
        repeatedly but which rarely changes. `no-cache` = always revalidate."""
        body = json.dumps(obj, ensure_ascii=False).encode("utf-8")
        etag = '"%s"' % hashlib.md5(body).hexdigest()[:16]
        if self._etag_matches(etag):
            return self._send_304(etag)
        gzipped = False
        if self._accepts_gzip() and len(body) > 1024:
            body = _gzip.compress(body, 6)
            gzipped = True
        self.send_response(200)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        if gzipped:
            self.send_header("Content-Encoding", "gzip")
            self.send_header("Vary", "Accept-Encoding")
        self.send_header("Content-Length", str(len(body)))
        self.send_header("ETag", etag)
        self.send_header("Cache-Control", "no-cache")
        self._cors()
        self.end_headers()
        self.wfile.write(body)

    def _send_text(self, text, ctype="text/plain; charset=utf-8", status=200):
        body = text.encode("utf-8")
        gzipped = False
        if self._accepts_gzip() and len(body) > 1024:
            body = _gzip.compress(body, 6)
            gzipped = True
        self.send_response(status)
        self.send_header("Content-Type", ctype)
        if gzipped:
            self.send_header("Content-Encoding", "gzip")
            self.send_header("Vary", "Accept-Encoding")
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Cache-Control", "no-store")
        self._cors()
        self.end_headers()
        self.wfile.write(body)

    def _accepts_gzip(self):
        return "gzip" in (self.headers.get("Accept-Encoding") or "").lower()

    def _send_file(self, path):
        ext = os.path.splitext(path)[1].lower()
        # ETag from mtime+size lets the client (and the Android data refresh)
        # revalidate cheaply: an unchanged file returns 304 with no body. This
        # is the big win for /data/network.json (~2.3 MB) on every app launch.
        st = os.stat(path)
        etag = '"%x-%x"' % (int(st.st_mtime), st.st_size)
        if self._etag_matches(etag):
            return self._send_304(etag)
        with open(path, "rb") as fh:
            body = fh.read()
        gzipped = False
        if ext in GZIP_EXTS and self._accepts_gzip() and len(body) > 1024:
            body = _gzip.compress(body, 6)
            gzipped = True
        self.send_response(200)
        self.send_header(
            "Content-Type", CONTENT_TYPES.get(ext, "application/octet-stream")
        )
        if gzipped:
            self.send_header("Content-Encoding", "gzip")
            self.send_header("Vary", "Accept-Encoding")
        self.send_header("Content-Length", str(len(body)))
        self.send_header("ETag", etag)
        self.send_header("Last-Modified", formatdate(st.st_mtime, usegmt=True))
        self.send_header("Cache-Control", "no-cache")
        self._cors()
        self.end_headers()
        self.wfile.write(body)

    def do_GET(self):
        parsed = urlparse(self.path)
        path = parsed.path
        self._begin(path)
        # Per-IP rate limit on the API (health is exempt — it's the monitor's
        # probe and must always answer). Returns 429 + Retry-After; the 429 is
        # captured in mobaz_http_requests_total{code="429"} via send_response.
        if path.startswith("/api/") and path != "/api/health":
            if not RATE_LIMITER.allow(getattr(self, "_client_ip", "?")):
                return self._send_rate_limited()
        try:
            if path == "/metrics":
                return self._send_text(METRICS.render(), "text/plain; version=0.0.4")

            if path == "/status":
                return self._send_text(_status_html(), "text/html; charset=utf-8")

            if path == "/api/stops":
                return self._send_json_cacheable(busapi.get_stops())

            if path == "/api/plan":
                q = parse_qs(parsed.query)
                day = int((q.get("day") or ["0"])[0])
                frm = (q.get("from") or [""])[0]
                to = (q.get("to") or [""])[0]
                if not frm or not to:
                    return self._send_json({"error": "from and to are required"}, 400)
                if frm == to:
                    METRICS.observe_plan("same_stop")
                    return self._send_json(
                        {"error": "Origin and destination are the same."}, 400
                    )
                result = busapi.plan(day, frm, to)
                _attach_coords(result)
                # A plan that finds no itinerary is a data gap, not an error —
                # count it separately so the dashboard surfaces routable demand.
                METRICS.observe_plan("success" if result.get("options") else "no_route")
                return self._send_json(result)

            if path == "/api/walk":
                # Pedestrian route from the user's location to their boarding
                # stop. ?from=lat,lon&to=lat,lon -> {geometry, distance, duration}.
                q = parse_qs(parsed.query)
                try:
                    fa = [float(x) for x in (q.get("from") or [""])[0].split(",")]
                    ta = [float(x) for x in (q.get("to") or [""])[0].split(",")]
                except ValueError:
                    fa = ta = []
                if len(fa) != 2 or len(ta) != 2:
                    return self._send_json(
                        {"error": "from and to must be 'lat,lon'"}, 400
                    )
                route = roads.walk_route(fa, ta)
                if not route:
                    return self._send_json({"error": "no walking route"}, 502)
                return self._send_json(route)

            if path == "/api/geocode":
                # Free-text destination address -> OSM-corrected candidates.
                q = (parse_qs(parsed.query).get("q") or [""])[0]
                if not q.strip():
                    return self._send_json({"results": []})
                return self._send_json({"results": geocode.geocode_search(q)})

            if path == "/api/nearby":
                # Nearby activities/attractions around ?lat=&lon= (proxies OSM
                # Overpass — which has no CORS — and attaches photos). Returns
                # [{name, coord, cat, emoji, dist, photo}].
                q = parse_qs(parsed.query)
                try:
                    lat = float((q.get("lat") or [""])[0])
                    lon = float((q.get("lon") or [""])[0])
                except ValueError:
                    return self._send_json({"error": "lat and lon are required"}, 400)
                radius = int((q.get("radius") or ["7000"])[0])
                cat = (q.get("cat") or [""])[0]
                if cat and cat in nearby.CATEGORIES:
                    results = nearby.nearby_category(lat, lon, radius, cat)
                else:  # "top"/unknown → rich attractions feed
                    results = nearby.nearby_activities(lat, lon, radius)
                return self._send_json({"results": results})

            if path == "/api/events":
                # Upcoming Mauritius events scraped from partyapp.mu (public site).
                return self._send_json({"events": events_mod.get_events()})

            if path == "/api/health":
                # Machine-readable status for monitoring. Returns 200 when the
                # core data is present, 503 when degraded (no stops, or the
                # event scrape has produced nothing) so an external check can
                # alert. Never triggers a blocking scrape.
                ev = events_mod.health()
                try:
                    stops = len(busapi.get_stops())
                except Exception:  # noqa: BLE001
                    stops = 0
                try:
                    geo_cache = len(geocode._load_coords_cache())
                except Exception:  # noqa: BLE001
                    geo_cache = 0
                degraded = stops == 0 or ev["count"] < EVENTS_MIN_HEALTHY
                body = {
                    "status": "degraded" if degraded else "ok",
                    "stops": stops,
                    "events": ev,
                    "geocode_cache": geo_cache,
                }
                return self._send_json(body, 503 if degraded else 200)

            # Static files.
            rel = path.lstrip("/") or "index.html"
            full = os.path.normpath(os.path.join(PUBLIC, rel))
            if not full.startswith(PUBLIC):
                return self._send_json({"error": "forbidden"}, 403)
            if os.path.isdir(full):
                full = os.path.join(full, "index.html")
            if os.path.isfile(full):
                return self._send_file(full)
            return self._send_json({"error": "not found"}, 404)

        except Exception as exc:  # noqa: BLE001
            # Log with the request path + a timestamp so a 500 in the wild is
            # traceable in the Space logs (which is the only visibility we have).
            ts = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
            print(f"[{ts}] ERROR {self.path}: {type(exc).__name__}: {exc}", flush=True)
            traceback.print_exc()
            return self._send_json({"error": f"{type(exc).__name__}: {exc}"}, 500)

    def do_POST(self):
        parsed = urlparse(self.path)
        self._begin(parsed.path)
        if parsed.path != "/api/clienterror":
            return self._send_json({"error": "not found"}, 404)
        # Read at most a few KB; oversized or malformed bodies are dropped quietly.
        try:
            n = min(int(self.headers.get("Content-Length") or 0), _CLIENT_ERR_MAX_BYTES)
            raw = self.rfile.read(n) if n > 0 else b""
            payload = json.loads(raw.decode("utf-8")) if raw else {}
        except Exception:  # noqa: BLE001
            payload = {}
        # Coarse global rate limit so a looping client can't flood the logs.
        now = time.time()
        _client_err_hits[:] = [
            t for t in _client_err_hits if now - t < _CLIENT_ERR_WINDOW
        ]
        if len(_client_err_hits) < _CLIENT_ERR_MAX_PER_WINDOW:
            _client_err_hits.append(now)
            msg = str(payload.get("msg", ""))[:300]
            where = str(payload.get("where", ""))[:120]
            platform = str(payload.get("platform", ""))[:60]
            ts = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
            # stderr only — never persisted, no client identity captured.
            print(f"[{ts}] CLIENT-ERR [{platform}] {where}: {msg}", flush=True)
            METRICS.inc_client_error()
        else:
            METRICS.inc_client_error(dropped=True)
        # Always 204: the beacon is fire-and-forget and must never surface to users.
        self.send_response(204)
        self._cors()
        self.send_header("Content-Length", "0")
        self.end_headers()


def main():
    # Warm the caches so the first request is fast and failures surface early.
    print("Loading bus stops...", flush=True)
    stops = busapi.get_stops()
    print(f"  {len(stops)} stops cached.", flush=True)
    print("Building OpenStreetMap geocoding index...", flush=True)
    try:
        idx = geocode._build_osm_index()
        print(f"  {len(idx)} named features indexed.", flush=True)
    except Exception as exc:  # noqa: BLE001
        print(
            f"  WARN: OSM index unavailable ({exc}); will rely on Nominatim.",
            flush=True,
        )

    # Warm the events cache in the background so the Events tab is instant on the
    # first request without delaying server start-up.
    print("Pre-warming events (background)...", flush=True)
    events_mod.prewarm()

    try:
        httpd = ThreadingHTTPServer(("0.0.0.0", PORT), Handler)
    except OSError as exc:
        if exc.errno in (48, 98):  # EADDRINUSE (macOS / Linux)
            print(
                f"\nPort {PORT} is already in use — another copy may still be running.",
                file=sys.stderr,
            )
            print(f"  Free it:   lsof -ti tcp:{PORT} | xargs kill", file=sys.stderr)
            print(
                f"  Or pick another port:   PORT={PORT + 1} ./run.sh", file=sys.stderr
            )
            sys.exit(1)
        raise

    print(
        f"\nMauritius Bus Route Planner running at http://localhost:{PORT}", flush=True
    )
    print("Press Ctrl+C to stop.", flush=True)
    try:
        httpd.serve_forever()
    except KeyboardInterrupt:
        print("\nShutting down.")
        httpd.server_close()


if __name__ == "__main__":
    main()