path: root/serve.py

#!/usr/bin/env python3
"""World map server: the viewer + a JSON API over a finished map build. Binds 127.0.0.1 only.

  python serve.py --world WORLD_DIR [--world OTHER_DIR …] [--res N] [--port 8765] [--lore NOTES_DIR]

Several --world: one server, each world at /w/<id>/ (the viewer's world menu switches).
"""
from __future__ import annotations

import argparse
import errno
import gzip
import hashlib
import os
import secrets
import shutil
import tempfile
import signal
import threading
import types
from datetime import date
import json
import mimetypes
import re
import sys
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import parse_qs, unquote, urlparse

os.environ.setdefault("OPENBLAS_THREAD_TIMEOUT", "4")   # idle BLAS threads: sleep soon, don't spin (same results)
import numpy as np  # noqa: E402

HERE = Path(__file__).resolve().parent
VIEWER = HERE / "viewer"
sys.path.insert(0, str(HERE))

import worldgen_path  # noqa: E402,F401  (mapgen on sys.path)
import servecache
from tilecap import TileCap  # noqa: E402
import tiles  # noqa: E402
from mapgen import render as RN  # noqa: E402

LOAD_KEYS = ["g_ids", "z_surface_m", "ocean", "landform", "lithology", "T_mean", "T_range", "T_jun", "T_dec",
             "P_ann", "P_jun", "P_dec", "holdridge", "seasonality", "ground", "ice", "discharge_km3_yr", "lake",
             "endorheic", "dist_ocean_km", "plate", "age_class", "po2_bar", "gravity_g", "g_xyz",
             "recv", "river", "z_filled_m", "seabed_type", "seabed_mineral", "bottom_temp_c", "sediment_m",
             "vent_potential", "pressure_bar", "o2_fraction", "fire_reactivity", "plant_height_x", "deposits",
             "deposit_main", "lake_level_m", "sst", "current_speed", "productivity"]
OVERLAYS = [("coast", "Coastline", "coast.geojson"), ("lakes", "Lakes", "lakes.geojson"),
            ("rivers", "Rivers", "rivers.geojson"), ("plates", "Plate boundaries", "plate_boundaries.geojson")]
MAX_BODY = 64 * 1024
PROFILE_POINTS = (2, 100)
PROFILE_N = (16, 1024)
FIELDS = [  # key, label, unit, group, kind ("num" | "bool" | "cat:<legend key>" | "bits:<legend key>" | "derived")
    ("z_surface_m", "Elevation", "m", "Terrain", "num"),
    ("ocean", "Sea", "", "Terrain", "bool"),
    ("landform", "Landform", "", "Terrain", "cat:landform"),
    ("lithology", "Rock", "", "Terrain", "cat:lithology"),
    ("T_mean", "Mean temperature", "°C", "Climate", "num"),
    ("T_summer", "Summer temperature", "°C", "Climate", "derived"),
    ("T_winter", "Winter temperature", "°C", "Climate", "derived"),
    ("T_range", "Seasonal range", "°C", "Climate", "num"),
    ("P_ann", "Rainfall, year", "mm/yr", "Climate", "num"),
    ("P_jun", "Rainfall, June season", "mm/yr", "Climate", "num"),
    ("P_dec", "Rainfall, December season", "mm/yr", "Climate", "num"),
    ("holdridge", "Life zone (Holdridge)", "", "Environment", "cat:holdridge"),
    ("seasonality", "Seasonality", "", "Environment", "cat:seasonality"),
    ("ground", "Ground", "", "Environment", "cat:ground"),
    ("ice", "Ice", "", "Environment", "cat:ice"),
    ("deposits", "Mineral deposits", "", "Environment", "bits:deposits"),
    ("discharge_km3_yr", "River flow", "km³/yr", "Water", "num"),
    ("lake", "Lake", "", "Water", "bool"),
    ("endorheic", "Endorheic basin", "", "Water", "bool"),
    ("dist_ocean_km", "Distance to coast", "km", "Water", "num"),
    ("plate", "Plate", "", "Tectonics", "cat:plates"),
    ("age_class", "Crust", "", "Tectonics", "cat:age_class"),
    ("seabed_type", "Sea-floor type", "", "Sea floor", "cat:seabed_type"),
    ("seabed_mineral", "Sea-floor minerals", "", "Sea floor", "cat:seabed_mineral"),
    ("bottom_temp_c", "Bottom water temperature", "°C", "Sea floor", "num"),
    ("sediment_m", "Sediment", "m", "Sea floor", "num"),
    ("vent_potential", "Vent potential", "0–1", "Sea floor", "num"),
    ("vent", "Vent field strength", "0–1", "Sea floor", "num"),
    ("canyon_m", "Canyon depth", "m", "Sea floor", "num"),
    ("sst", "Sea-surface temperature", "°C", "Sea", "num"),
    ("current_speed", "Current speed", "m/s", "Sea", "num"),
    ("productivity", "Sea productivity", "0–1", "Sea", "num"),
    ("po2_bar", "O₂ partial pressure", "bar", "World", "num"),
    ("gravity_g", "Gravity", "g", "World", "num"),
    ("o2_fraction", "O₂ fraction", "", "World", "num"),
    ("pressure_bar", "Air pressure", "bar", "World", "num"),
    ("fire_reactivity", "Fire reactivity", "×", "World", "num"),
    ("plant_height_x", "Plant height", "×", "World", "num"),
]


class MissingBuild(SystemExit):
    pass


def world_cache_key(out: Path) -> str:
    """Names the world's serve cache: the arrays it holds (LOAD_KEYS, float32) and this build of cells.npz."""
    st = (Path(out) / "cells.npz").stat()
    h = hashlib.sha1(f"{','.join(LOAD_KEYS)}:f32:{st.st_size}:{st.st_mtime_ns}".encode())
    return h.hexdigest()[:10]


def load_worlds(root: Path, res: int, lore_dir: Path | None = None, log=print, only=None) -> list:
    """(era, World) of every world the viewer can show: the base build first, then each built era (only: just that
    one; none when it is not built)."""
    import refine
    from mapgen import config as C
    if not (Path(root) / "out" / f"r{res}" / "cells.npz").exists():
        return [("base", World(root, res, lore_dir))]            # raises MissingBuild (names what is missing)
    eras = (C.load(root)[1].get("eras") or {})
    out = []
    for name, d in refine.world_dirs(root, res, log):
        if only is not None and name != only:
            continue
        label = (eras.get(name) or {}).get("label", name if name != "base" else "Base")
        out.append((name, World(root, res, lore_dir, out=d, era={"name": name, "label": label})))
    return out


class World:
    def __init__(self, data_root: Path, res: int, lore_dir: Path | None = None, use_cache: bool = True,
                 out: Path | None = None, era: dict | None = None):
        out = Path(out) if out is not None else data_root / "out" / f"r{res}"   # an era: out/r<res>/eras/<era>
        need = [out / "cells.npz", out / "cells_meta.json", out / "viewer" / "layers.json"]
        missing = [str(p) for p in need if not p.exists()]
        if missing:
            raise MissingBuild(f"missing build outputs {missing}: run `mapgen.py build` "
                               f"({'--final' if res >= 5 else f'--res {res}'}) first")
        self.res, self.out = res, out
        self.era = era or {"name": "base", "label": "Base"}
        self.cells_meta = json.loads((out / "cells_meta.json").read_text())
        if not {"name", "planet"} <= self.cells_meta.keys() and (data_root / "config" / "world.toml").exists():
            import tomllib                                                 # older builds: name and planet from the config
            cfg = tomllib.loads((data_root / "config" / "world.toml").read_text())
            if cfg.get("render", {}).get("name"):
                self.cells_meta.setdefault("name", cfg["render"]["name"])
            self.cells_meta.setdefault("planet", cfg.get("planet", {}))
        style = self.cells_meta.get("style")
        if style is not None and style not in RN.STYLES:      # a build from before a palette was renamed: every
            raise MissingBuild(f"{out} was built with render style {style!r}, which this worldgen no longer has "   # relief tile
                               f"(styles: {sorted(RN.STYLES)}): rebuild it (`mapgen.py build --res {res}`)")  # would fail
        self.layers = json.loads((out / "viewer" / "layers.json").read_text())
        self.legends = self.cells_meta["legends"]
        self.units = dict(self.cells_meta.get("units", {}))
        spill = (Path(tempfile.mkdtemp(prefix=".spill-", dir=out))
                 if use_cache and servecache.low_memory() else None)   # fields on disk while the cache is written

        def build():
            keep = servecache.spiller(spill)
            with np.load(out / "cells.npz") as z:
                return {k: keep(k, z[k].astype(np.float32) if z[k].dtype == np.float64 else z[k])
                        for k in LOAD_KEYS if k in z.files}, {}
        self.cache = servecache.cache_dir(out, world_cache_key(out)) if use_cache else None
        try:
            self.arrays = servecache.load_or_build(self.cache, build)[0] if use_cache else build()[0]
        finally:
            if spill is not None:
                shutil.rmtree(spill, ignore_errors=True)
        self.ids = self.arrays["g_ids"]
        self.lore_ids = sorted({m.group(1) for p in (Path(lore_dir).glob("**/*.md") if lore_dir else ())
                                for m in [re.search(r"^id:\s*(\S+)", p.read_text(), re.M)] if m})

    def index_of(self, lat: float, lon: float) -> int:
        import h3.api.basic_int as h3
        cell = np.uint64(h3.latlng_to_cell(lat, lon, self.res))
        i = int(np.searchsorted(self.ids, cell))
        if i >= len(self.ids) or self.ids[i] != cell:          # never hand back a neighbour's data
            raise LookupError(f"no cell at {lat:.4f}, {lon:.4f}")
        return i

    def meta(self) -> dict:
        planet = {"radius_km": self.cells_meta["radius_km"], "day_hours": 24.0, "tilt_deg": 23.44, "year_days": 365,
                  **self.cells_meta.get("planet", {})}
        return {"res": self.res, "radius_km": self.cells_meta["radius_km"], "name": self.cells_meta.get("name", "World"),
                "planet": planet, "units": {**self.units, "day_hours": planet["day_hours"]},
                "layers": [{**L, "file": f"/data/viewer/{L['file']}"} for L in self.layers],
                "overlays": [{"id": i, "name": n, "file": f"/data/geo/{f}"} for i, n, f in OVERLAYS
                             if (self.out / "geo" / f).exists()],
                "lore_ids": self.lore_ids}

    def _alt(self, key, v):
        u = self.units
        if not {"span_m", "league_km", "moment_s"} <= u.keys():       # no in-world units in the config: SI only
            return None
        if key == "z_surface_m":
            return f"{v / u['span_m']:,.0f} spans"
        if key == "dist_ocean_km":
            return f"{v / u['league_km']:,.1f} leagues"
        if key == "gravity_g":
            return f"{v * 9.80665 / u['span_m'] * u['moment_s'] ** 2:,.0f} spans/moment²"
        return None

    def describe(self, lat: float, lon: float) -> dict:
        import h3.api.basic_int as h3
        rs = getattr(self, "regions", None)
        wt = float(rs.weight(np.array([latlon_xyz(lat, lon)]))[0]) if rs is not None and not rs.empty else 0.0
        j = rs.index_of(lat, lon) if wt > 0.5 else None          # where a refined area shows, its fine cell answers
        a, i = (rs.arrays, j) if j is not None else (self.arrays, self.index_of(lat, lon))
        north = lat >= 0
        derived = {"T_summer": a["T_jun"][i] if north else a["T_dec"][i],
                   "T_winter": a["T_dec"][i] if north else a["T_jun"][i]}
        fields = []
        sea = bool(a["ocean"][i])
        for key, label, unit, group, kind in FIELDS:
            if (kind != "derived" and key not in a) or (group in ("Sea floor", "Sea") and not sea):
                continue
            raw = derived[key] if kind == "derived" else a[key][i]
            f = {"key": key, "label": label, "group": group, "unit": unit}
            if kind == "bool":
                f["value"] = bool(raw)
                f["text"] = "yes" if raw else "no"
            elif kind.startswith("bits:"):                  # legend index i + 1 = bit i
                names = self.legends.get(kind[5:], [])
                f["value"] = int(raw)
                f["text"] = ", ".join(names[b + 1] for b in range(len(names) - 1) if int(raw) >> b & 1) or "none"
            elif kind.startswith("cat:"):
                names = self.legends.get(kind[4:], [])
                f["value"] = int(raw)
                f["text"] = names[int(raw)] if 0 <= int(raw) < len(names) else str(int(raw))
            else:
                f["value"] = round(float(raw), 4)
                alt = self._alt(key, float(raw))
                if alt:
                    f["alt"] = alt
            fields.append(f)
        cell = int(a["g_ids"][i])
        return {"cell": h3.int_to_str(cell), "lat": lat, "lon": lon, "center": list(h3.cell_to_latlng(cell)),
                "inner_km": self._inner_km(cell, rs),   # clients may reuse this answer anywhere this close to the centre
                "refined": j is not None, "fields": fields, "era": self.era}

    def _inner_km(self, cell, rs=None) -> float:
        """0.8 × the distance from the cell centre to its nearest edge (great circles through the boundary); 0 where
        a refined area's edge band may lie within that (the answer changes there)."""
        import h3.api.basic_int as h3
        c = np.array(latlon_xyz(*h3.cell_to_latlng(cell)))
        b = [np.array(latlon_xyz(*v)) for v in h3.cell_to_boundary(cell)]
        near = min(np.arcsin(abs(c @ (n / np.linalg.norm(n)))) for n in (np.cross(b[k], b[(k + 1) % len(b)]) for k in range(len(b))))
        r = float(0.8 * near * self.cells_meta["radius_km"])
        if rs is not None and not rs.empty and rs.edge_km(np.array([c]))[0] < r + 2 * tiles.refine.BLEND_KM + rs.spacing_km:
            return 0.0
        return r


LORE_ID = re.compile(r"^[a-z0-9]+(?:-[a-z0-9]+)*$")


class PinStore:
    def __init__(self, path: Path, res: int):
        """res: H3 resolution of each pin's `cell` — always the final map's, whatever the server runs."""
        self.path, self.res, self.lock = path, res, threading.Lock()

    def _load(self) -> list:
        return json.loads(self.path.read_text())["pins"] if self.path.exists() else []

    def _save(self, pins: list) -> None:
        self.path.parent.mkdir(parents=True, exist_ok=True)
        fd, tmp = tempfile.mkstemp(dir=self.path.parent, prefix=".pins-", suffix=".json")
        try:
            os.fchmod(fd, 0o644)                                    # mkstemp makes 0600; pins.json is a normal repo file
            with os.fdopen(fd, "w", encoding="utf-8") as f:
                json.dump({"pins": pins}, f, indent=1, ensure_ascii=False)
                f.write("\n")
                f.flush()
                os.fsync(f.fileno())
            os.replace(tmp, self.path)
        except BaseException:
            Path(tmp).unlink(missing_ok=True)
            raise

    def validate(self, data) -> dict:
        if not isinstance(data, dict):
            raise ValueError("body must be a JSON object")
        name = data.get("name", "")
        if not isinstance(name, str) or not 1 <= len(name.strip()) <= 120:
            raise ValueError("name must be 1–120 characters")
        name = name.strip()
        try:
            lat, lon = float(data["lat"]), float(data["lon"])
        except (KeyError, TypeError, ValueError):
            raise ValueError("lat and lon must be numbers")
        if not (-90.0 <= lat <= 90.0) or not np.isfinite(lon):
            raise ValueError("lat must be in [-90, 90]")
        lore = data.get("lore") or None
        if lore is not None and not (isinstance(lore, str) and LORE_ID.fullmatch(lore)):
            raise ValueError("lore must be a kebab-case lore id")
        note = data.get("note") or ""
        if not isinstance(note, str) or len(note) > 2000:
            raise ValueError("note must be ≤ 2000 characters")
        epoch = data.get("epoch") or None                    # the epoch a place belongs to (a note id), for epoch layers
        if epoch is not None and not (isinstance(epoch, str) and LORE_ID.fullmatch(epoch)):
            raise ValueError("epoch must be a kebab-case lore id")
        lon = lon if -180.0 <= lon < 180.0 else (lon + 180.0) % 360.0 - 180.0   # wrap only when needed: keep typed values exact
        return {"name": name, "lat": lat, "lon": lon, "lore": lore, "note": note, "epoch": epoch}

    def _cell(self, lat, lon) -> str:
        import h3.api.basic_int as h3
        return h3.int_to_str(h3.latlng_to_cell(lat, lon, self.res))

    def list(self) -> list:
        with self.lock:
            return self._load()

    def create(self, data) -> dict:
        pin = self.validate(data)
        pin.update(id=f"pin-{secrets.token_hex(4)}", cell=self._cell(pin["lat"], pin["lon"]),
                   created=date.today().isoformat())
        with self.lock:
            pins = self._load()
            pins.append(pin)
            self._save(pins)
        return pin

    def update(self, pid: str, data) -> dict | None:
        if not isinstance(data, dict):
            raise ValueError("body must be a JSON object")
        with self.lock:
            pins = self._load()
            for k, p in enumerate(pins):
                if p["id"] == pid:
                    merged = {**p, **self.validate({**p, **data})}
                    merged["cell"] = self._cell(merged["lat"], merged["lon"])
                    pins[k] = merged
                    self._save(pins)
                    return merged
        return None

    def delete(self, pid: str) -> bool:
        with self.lock:
            pins = self._load()
            keep = [p for p in pins if p["id"] != pid]
            if len(keep) == len(pins):
                return False
            self._save(keep)
            return True


class RegionsFileError(ValueError):
    """places/regions.json exists but cannot be read (e.g. a hand edit went wrong): reported, never overwritten."""


class RegionStore:
    """places/regions.json: outlines to refine ."""
    MAX_KM2 = 2.0e6                    # above this the viewer asks first (slow, memory-hungry builds)
    HARD_KM2 = 10 * MAX_KM2            # refused: a build this big could run the server out of memory

    def __init__(self, path: Path):
        self.path, self.lock = path, threading.Lock()

    def _load(self) -> list:
        if not self.path.exists():
            return []
        try:
            regions = json.loads(self.path.read_text())["regions"]
            if not isinstance(regions, list) or not all(isinstance(r, dict) and "outline" in r for r in regions):
                raise TypeError("regions must be a list of {id, name, outline}")
            return regions
        except (ValueError, KeyError, TypeError) as e:
            raise RegionsFileError(f"{self.path.name} cannot be read ({e}); fix or remove it") from None

    def _save(self, regions: list) -> None:
        self.path.parent.mkdir(parents=True, exist_ok=True)
        fd, tmp = tempfile.mkstemp(dir=self.path.parent, prefix=".regions-", suffix=".json")
        try:
            os.fchmod(fd, 0o644)
            with os.fdopen(fd, "w", encoding="utf-8") as f:
                json.dump({"regions": regions}, f, indent=1, ensure_ascii=False)
                f.write("\n")
            os.replace(tmp, self.path)
        except BaseException:
            Path(tmp).unlink(missing_ok=True)
            raise

    @staticmethod
    def _crosses(pts) -> bool:
        def ccw(a, b, c):
            return (c[1] - a[1]) * (b[0] - a[0]) - (b[1] - a[1]) * (c[0] - a[0])
        n = len(pts)
        seg = [(pts[k], pts[(k + 1) % n]) for k in range(n)]
        for i in range(n):
            for j in range(i + 1, n):
                if abs(i - j) in (1, n - 1):
                    continue                                       # neighbours share a point
                (a, b), (c, d) = seg[i], seg[j]
                if ccw(a, b, c) * ccw(a, b, d) < 0 and ccw(c, d, a) * ccw(c, d, b) < 0:
                    return True
        return False

    def validate(self, data) -> dict:
        if not isinstance(data, dict):
            raise ValueError("body must be a JSON object")
        name = data.get("name", "")
        if not isinstance(name, str) or not 1 <= len(name.strip()) <= 80:
            raise ValueError("name must be 1–80 characters")
        out = data.get("outline")
        if not isinstance(out, list) or len(out) < 3 or len(out) > 500:
            raise ValueError("outline needs 3–500 [lat, lon] points")
        try:
            pts = [(float(p[0]), float(p[1])) for p in out]
        except (TypeError, ValueError, IndexError, KeyError):
            raise ValueError("outline points must be [lat, lon] numbers")
        if not all(np.isfinite(la) and np.isfinite(lo) and -90 <= la <= 90 for la, lo in pts):
            raise ValueError("lat must be in [-90, 90] and lon finite")
        lon0 = pts[0][1]
        pts = [(la, lon0 + ((lo - lon0 + 180) % 360) - 180) for la, lo in pts]   # unwrap across ±180°
        if self._crosses([(lo, la) for la, lo in pts]):
            raise ValueError("the outline crosses itself")
        return {"name": name.strip(), "outline": [[round(la, 6), round(((lo + 180) % 360) - 180, 6)] for la, lo in pts]}

    @staticmethod
    def area_km2(outline, radius_km: float) -> float:
        import h3.api.basic_int as h3
        cells = h3.polygon_to_cells(h3.LatLngPoly([tuple(p) for p in outline]), 5)
        return sum(h3.cell_area(c, unit="rads^2") for c in cells) * radius_km ** 2

    def list(self) -> list:
        with self.lock:
            return self._load()

    def create(self, data, radius_km: float) -> dict:
        r = self.validate(data)
        km2 = self.area_km2(r["outline"], radius_km)
        if km2 > self.HARD_KM2:
            raise ValueError(f"region too large: ≈ {km2:,.0f} km² (limit {self.HARD_KM2:,.0f}); draw it in parts")
        r.update(id=f"region-{secrets.token_hex(4)}", created=date.today().isoformat(), area_km2=round(km2))
        with self.lock:
            regions = self._load()
            regions.append(r)
            self._save(regions)
        return r

    def delete(self, rid: str) -> bool:
        with self.lock:
            regions = self._load()
            keep = [r for r in regions if r["id"] != rid]
            if len(keep) == len(regions):
                return False
            self._save(keep)
            return True


PRERENDER_Z = 11                   # region tiles are rendered and saved ahead down to z11 (≈ 20 km tiles) …
PRERENDER_MAX = 30000              # … as deep as fits this many tile files (a continent stops a few zooms earlier)


class RefineJobs:
    """One background worker refining regions (coalescing): saves never wait for a build."""
    def __init__(self, server, root: Path, res: int):
        self.server, self.root, self.res = server, root, res
        self.lock = threading.Lock()
        self.pending = False
        self.running = False
        self.gen = 0                                            # pre-render generation: a newer one stops older ones
        self.st = {"state": "idle", "stage": "", "fraction": 0.0, "error": None, "finished": None, "tiles": None}

    def status(self) -> dict:
        with self.lock:
            return dict(self.st)

    def request(self) -> None:
        with self.lock:
            self.pending = True
            if self.st["state"] != "building":
                self.st.update(state="queued", stage="", fraction=0.0)
            if self.running:
                return
            self.running = True
        threading.Thread(target=self._run, daemon=True).start()

    def _progress(self, stage: str, f: float) -> None:
        with self.lock:
            self.st.update(state="building", stage=stage, fraction=round(float(f), 3))

    def _run(self) -> None:
        try:
            self._loop()
        finally:
            with self.lock:                                     # whatever happened, later saves start a new worker
                self.running = False

    def _loop(self) -> None:
        import refine
        while True:
            with self.lock:
                if not self.pending:
                    return
                self.pending = False
                self.st.update(state="building", stage="starting", fraction=0.0, error=None)
            try:
                refine.build_areas(self.root, self.res, self.server.regions.path, log=lambda s: None,
                                   regions_root=self.server.regions_dir, progress=self._progress)
                tiles.reload_all(list(self.server.sources.values()))
                for name, s in self.server.sources.items():
                    self.server.worlds[name].regions = s.regions
                with self.lock:
                    done = not self.pending                     # a save came in meanwhile: not ready yet
                    self.st.update(state="ready" if done else "queued", stage="", fraction=1.0 if done else 0.0,
                                   finished=date.today().isoformat(), tiles=None)
                if done:
                    self.prerender()
            except (Exception, SystemExit) as e:                # report (MissingBuild is a SystemExit); the map
                with self.lock:                                 # keeps what it had
                    self.st.update(state="failed", error=f"{type(e).__name__}: {e}")

    def prerender(self) -> None:
        """Render and save the refined areas' tiles in the background (after a build, or at start); a newer build or
        request stops it. Progress in status()["tiles"]."""
        tiles_ = self.server.tiles
        with self.lock:
            self.gen += 1
            gen = self.gen

        def progress(done, total):
            with self.lock:
                if self.gen == gen:
                    self.st["tiles"] = {"done": done, "total": total}

        def stop():
            if self.server.socket.fileno() == -1:              # the server was closed
                return True
            with self.lock:
                return self.gen != gen or self.pending or self.st["state"] in ("queued", "building")

        def run():
            pool = tiles_.pool
            try:
                tiles_.prerender(self.server.prerender_z, progress=progress, stop=stop, max_tiles=PRERENDER_MAX,
                                 threads=max(1, pool.alive - 2) if pool else 1)
            except Exception as e:                                  # noqa: BLE001 — the map still works without
                with self.lock:
                    self.st["tiles_error"] = f"{type(e).__name__}: {e}"
        threading.Thread(target=run, daemon=True).start()


def refined_flags(server, regions: list) -> list:
    """The regions, each with "refined": the map shows the area the last build made of its (unchanged) outline."""
    import refine
    rec = (refine.read_record(server.regions_dir, server.tiles.fingerprint) or {}).get("regions", {})
    names = set(server.tiles.regions.area_names)
    return [{**r, "refined": bool((e := rec.get(refine.region_key(r))) and e["outline"] == refine.outline_hash(r)
                                  and e["area"] in names)} for r in regions]


def regions_missing(server) -> bool:
    """Whether the outlines changed since the last successful build, or any world's (era's) areas are missing (e.g.
    after a world or era rebuild). Reads the builds' records: no area work (a continent's takes ≈ 20 s)."""
    regions = server.regions.list()
    return any(_regions_missing(server, src.fingerprint, regions) for src in server.sources.values())


def _regions_missing(server, fingerprint: str, regions: list) -> bool:
    import refine
    have = {p.name for p in refine.built_areas(server.regions_dir, fingerprint)}
    rec = refine.read_record(server.regions_dir, fingerprint)
    if rec is None:
        return bool(regions) or bool(have)
    m = rec.get("regions", {})
    if {refine.region_key(r) for r in regions} != set(m) or any(
            m[refine.region_key(r)]["outline"] != refine.outline_hash(r) for r in regions):
        return True
    return {e["area"] for e in m.values() if e["area"]} != have


def h3_res(server) -> int:
    import h3.api.basic_int as h3
    return h3.get_resolution(int(server.world.arrays["g_ids"][0]))


NAME_RE = re.compile(r"^[a-z0-9][a-z0-9._-]{0,63}$")
EXPORT_MAX_SAMPLES = 8193          # 8193² float32 ≈ 270 MB per layer: the largest export the viewer may ask for


class ExportJobs:
    """Terrain exports asked for in the viewer, one at a time, on the server's own tile source."""
    def __init__(self, server):
        self.server = server
        self.lock = threading.Lock()
        self.st = {"state": "idle", "name": None, "fraction": 0.0, "error": None}
        d = Path(server.exports_dir)                            # work folders of exports cut off by a stop
        for junk in d.glob(".*-*") if d.exists() else []:
            if junk.is_dir():
                shutil.rmtree(junk, ignore_errors=True)

    def status(self) -> dict:
        with self.lock:
            return dict(self.st)

    @staticmethod
    def validate(data, radius_km: float = 12742.0) -> dict:
        import export
        if not isinstance(data, dict):
            raise ValueError("body must be a JSON object")
        try:
            lat, lon = float(data["lat"]), float(data["lon"])
            size_km, res_m = float(data["size_km"]), float(data["res_m"])
        except (KeyError, TypeError, ValueError):
            raise ValueError("lat, lon, size_km and res_m must be numbers")
        if not (-90 <= lat <= 90 and np.isfinite(lon)):
            raise ValueError("lat must be in [-90, 90] and lon finite")
        if not (0.5 <= size_km <= 200):
            raise ValueError("size_km must be 0.5–200")
        if not (1 <= res_m <= 5000):
            raise ValueError("res_m must be 1–5000")
        n = export.samples_for(size_km, res_m)
        if n > EXPORT_MAX_SAMPLES:
            raise ValueError(f"{n} × {n} samples is too many (≤ {EXPORT_MAX_SAMPLES}): a coarser resolution or a smaller area")
        lon = (lon + 180.0) % 360.0 - 180.0
        P = export.plan(lat, lon, size_km, res_m, radius_km)      # ValueError near a pole
        if P["tiles"] > export.MAX_TILES:                         # tiles narrow toward the poles: memory
            raise ValueError(f"this square needs {P['tiles']:,} map tiles (≤ {export.MAX_TILES:,}; they narrow toward "
                             f"the poles): a coarser resolution or a smaller square")
        name = data.get("name") or export.default_name(lat, lon, size_km, res_m)
        if not isinstance(name, str) or not NAME_RE.fullmatch(name) or ".." in name:
            raise ValueError("name: 1–64 of a–z, 0–9, . _ - (starting with a letter or digit)")
        era = data.get("era")
        if era is not None and not isinstance(era, str):
            raise ValueError("era must be a string")
        return {"lat": lat, "lon": lon, "size_km": size_km, "res_m": res_m, "name": name, "samples": n, "era": era}

    def start(self, p: dict) -> None:
        if (Path(self.server.exports_dir) / p["name"]).exists():
            raise FileExistsError(f"an export named {p['name']} exists")
        with self.lock:
            if self.st["state"] == "running":
                raise RuntimeError("an export is running")
            self.st = {"state": "running", "name": p["name"], "fraction": 0.0, "error": None}
        threading.Thread(target=self._run, args=(p,), daemon=True).start()

    def _run(self, p: dict) -> None:
        import export

        def progress(k, n):
            with self.lock:
                self.st["fraction"] = round(0.95 * k / n, 3)
        try:
            export.export(self.server.data_root, self.server.res, p["lat"], p["lon"], p["size_km"], p["res_m"],
                          out_dir=Path(self.server.exports_dir), name=p["name"], pins_path=self.server.pins_path,
                          src=self.server.sources[p.get("era") or self.server.default_era], progress=progress)
            with self.lock:
                self.st.update(state="done", fraction=1.0)
        except (Exception, SystemExit) as e:                    # noqa: BLE001 — reported in the viewer
            with self.lock:
                self.st.update(state="failed", error=f"{type(e).__name__}: {e}")

    def list(self) -> list:
        out = []
        d = Path(self.server.exports_dir)
        for m in sorted(d.glob("*/meta.json")) if d.exists() else []:
            if m.parent.name.startswith("."):                   # an export still being written
                continue
            try:
                meta = json.loads(m.read_text())
            except (OSError, ValueError):
                continue
            out.append({k: meta.get(k) for k in ("name", "center", "size_m", "res_m", "samples", "created")}
                       | {"path": str(m.parent)})
        return out


def vents_of(src) -> dict:
    """Every vent of the source's refined areas (vents.npz per area; none for older results), one per cell."""
    import refine
    import seafloor
    seen, out = set(), []
    for p in refine.built_areas(src.regions_dir, src.fingerprint):
        f = p / "vents.npz"
        if not f.exists():
            continue
        with np.load(f) as v:
            for c, la, lo, t, tc, fl, mi in zip(v["cell"].tolist(), v["lat"].tolist(), v["lon"].tolist(),
                                                v["type"].tolist(), v["temp_c"].tolist(), v["flow"].tolist(),
                                                v["mineral"].tolist()):
                if c in seen:
                    continue
                seen.add(c)
                out.append({"lat": round(la, 5), "lon": round(lo, 5), "type": int(t), "temp_c": round(tc, 1),
                            "flow": int(fl), "mineral": int(mi)})
    return {"vents": out, "types": seafloor.VENT_TYPES, "minerals": seafloor.VENT_MINERALS}


def plateau_geojson(root: Path, radius_km: float) -> dict:
    """The sunken plateaus' outlines (their noise-warped edge, no margin) as GeoJSON lines."""
    import refine
    from mapgen import config as C
    cfg, tect = C.load(root)
    seed = int(cfg["build"]["seed"])
    feats = []
    for p in tect.get("plateau", []):
        ring = refine.plateau_outline(p, seed, radius_km, margin_km=0.0)
        feats.append({"type": "Feature", "properties": {"name": p["name"], "islands": bool(p.get("islands", False))},
                      "geometry": {"type": "LineString", "coordinates": [[lo, la] for la, lo in ring + ring[:1]]}})
    return {"type": "FeatureCollection", "features": feats}


def latlon_xyz(lat, lon):
    la, lo = np.radians(lat), np.radians(lon)
    return np.cos(la) * np.cos(lo), np.cos(la) * np.sin(lo), np.sin(la)


def _profile(world, path, n: int, surface=None) -> dict:
    """Elevation along great-circle legs, n samples evenly spaced by distance."""
    if not PROFILE_POINTS[0] <= len(path) <= PROFILE_POINTS[1] or any(len(p) != 2 for p in path):
        raise ValueError("path needs 2–100 lat,lon points")
    if not PROFILE_N[0] <= n <= PROFILE_N[1]:
        raise ValueError("n must be 16–1024")
    lat = np.radians([p[0] for p in path])
    lon = np.radians([p[1] for p in path])
    if not np.all(np.isfinite(lat)) or np.any(np.abs(lat) > np.pi / 2):
        raise ValueError("lat must be in [-90, 90]")
    if not np.all(np.isfinite(lon)):
        raise ValueError("lon must be a finite number")
    P = np.stack([np.cos(lat) * np.cos(lon), np.cos(lat) * np.sin(lon), np.sin(lat)], axis=1)
    ang = np.arccos(np.clip(np.sum(P[:-1] * P[1:], axis=1), -1.0, 1.0))
    if np.any(ang > np.pi - 1e-6):
        raise ValueError("a leg joins antipodal points: its great circle is undefined")
    cum = np.concatenate([[0.0], np.cumsum(ang)])
    ds = np.linspace(0.0, cum[-1], n)
    k = np.clip(np.searchsorted(cum, ds, side="right") - 1, 0, len(ang) - 1)
    a = ang[k]
    t = np.where(a > 0, (ds - cum[k]) / np.where(a > 0, a, 1.0), 0.0)
    s = np.where(a > 1e-12, np.sin(a), 1.0)
    w0 = np.where(a > 1e-12, np.sin((1 - t) * a) / s, 1 - t)
    w1 = np.where(a > 1e-12, np.sin(t * a) / s, t)
    Q = w0[:, None] * P[k] + w1[:, None] * P[k + 1]
    Q /= np.linalg.norm(Q, axis=1, keepdims=True)
    qlat = np.degrees(np.arcsin(np.clip(Q[:, 2], -1, 1)))
    qlon = np.degrees(np.arctan2(Q[:, 1], Q[:, 0]))
    R = float(world.cells_meta["radius_km"])
    d = ds * R
    step = float(d[1] - d[0]) if n > 1 else 0.0
    idx = np.array([world.index_of(float(a_), float(b_)) for a_, b_ in zip(qlat, qlon)])
    zc = world.arrays["z_surface_m"][idx].astype(np.float64)
    if surface is None:
        z, water = zc, world.arrays["ocean"][idx].astype(bool) | world.arrays["lake"][idx].astype(bool)
    else:                                          # the procedural surface: same heights and coasts as the tiles
        z, water = surface(qlat, qlon, max(step / 2.0, 0.005))
        z, water = np.asarray(z, dtype=np.float64), np.asarray(water, dtype=bool)
    dz = np.diff(z)
    stats = {"min_m": round(float(z.min()), 1), "max_m": round(float(z.max()), 1),
             "ascent_m": round(float(dz[dz > 0].sum()), 1), "descent_m": round(float(-dz[dz < 0].sum()), 1),
             "water_km": round(step * float(np.sum(water[:-1] & water[1:]) + 0.5 * np.sum(water[:-1] ^ water[1:])), 3),
             "length_km": round(float(d[-1]), 3)}
    samples = [{"d_km": round(float(d_), 3), "lat": round(float(a_), 6), "lon": round(float(b_), 6),
                "z_m": round(float(z_), 1), "z_cell_m": round(float(c_), 1), "water": bool(w_)}
               for d_, a_, b_, z_, c_, w_ in zip(d, qlat, qlon, z, zc, water)]
    return {"samples": samples, "stats": stats}


def _safe(base: Path, rel: str) -> Path | None:
    try:
        p = (base / unquote(rel)).resolve()
    except (ValueError, OSError):            # e.g. an embedded NUL (%00)
        return None
    return p if p.is_file() and p.is_relative_to(base.resolve()) else None


GZIP_TYPES = {"application/octet-stream", "application/json", "application/geo+json", "text/javascript", "text/html",
              "text/css", "image/svg+xml", "text/plain"}   # mesh tiles shrink to ~60%; JPEG/PNG are compressed already
GZIP_MIN = 1024


def _gzip_ok(ctype: str, body: bytes, accept: str) -> bool:
    """Send body gzip-encoded? (browsers decompress it themselves; public links pay for every byte)"""
    if ctype.split(";")[0] not in GZIP_TYPES or len(body) < GZIP_MIN:
        return False
    for part in accept.split(","):
        name, _, q = part.strip().partition(";")
        if name.strip().lower() in ("gzip", "*"):
            return q.replace(" ", "") not in ("q=0", "q=0.0", "q=0.00", "q=0.000")
    return False


BUSY_RETRY_S = 2                   # a public server's answer when every render slot is taken
PRIVATE_API = ("/api/pins", "/api/regions", "/api/exports")   # not served by a --public server


class Handler(BaseHTTPRequestHandler):
    server_version = "worldmap-viewer/1"
    _prefix = ""

    def log_message(self, fmt, *args):
        pass

    def setup(self):
        self._front = self.server
        super().setup()

    def _route(self) -> bool:
        """Hub (several worlds): /w/<id>/… goes to that world, prefix stripped; / redirects to the first world;
        other paths go to the first world. False: answered here."""
        self.server, self._prefix = self._front, ""
        hub = getattr(self._front, "hub", None)
        url = urlparse(self.path)
        if url.path == "/api/worlds":
            return self._host_ok() and self._route_worlds(hub or {}, getattr(self._front, "default_world", None))
        if hub is None:
            return True
        parts = url.path.split("/", 3)                           # ['', 'w', '<id>', 'rest']
        if url.path == "/" or (len(parts) >= 3 and parts[1] == "w" and parts[2] in hub and len(parts) == 3):
            wid = parts[2] if url.path != "/" else self._front.default_world
            self.send_response(302)
            self.send_header("Location", f"/w/{wid}/")
            self.send_header("Content-Length", "0")
            self.end_headers()
            return False
        if len(parts) >= 3 and parts[1] == "w":
            if parts[2] not in hub:
                self._json(404, {"error": f"unknown world {parts[2]!r} (worlds: {list(hub)})"})
                return False
            self.server, self._prefix = hub[parts[2]], f"/w/{parts[2]}"
            self.path = "/" + (parts[3] if len(parts) == 4 else "") + (f"?{url.query}" if url.query else "")
            if urlparse(self.path).path == "/api/worlds":
                return self._host_ok() and self._route_worlds(hub, parts[2])
        else:
            self.server = hub[self._front.default_world]
        return True

    def _route_worlds(self, hub, current) -> bool:
        ws = [{"id": k, "name": s.world.cells_meta.get("name", k), "url": f"/w/{k}/"} for k, s in hub.items()]
        self._json(200, {"worlds": ws, "current": current})
        return False

    def _px(self, v):
        """Absolute URLs in a JSON answer, under this world's prefix (hub)."""
        if isinstance(v, dict):
            return {k: self._px(x) for k, x in v.items()}
        if isinstance(v, list):
            return [self._px(x) for x in v]
        if isinstance(v, str) and v.startswith(("/api/", "/data/", "/tiles/")):
            return self._prefix + v
        return v

    def _send(self, status: int, body: bytes, ctype: str, cache: str = "no-cache", extra: dict | None = None) -> None:
        gz = _gzip_ok(ctype, body, (getattr(self, "headers", None) or {}).get("Accept-Encoding", ""))
        if gz:
            body = gzip.compress(body, compresslevel=1, mtime=0)   # level 1: ~same size as 6 at a fifth of the time
        try:
            self.send_response(status)
            self.send_header("Content-Type", ctype)
            if ctype.split(";")[0] in GZIP_TYPES:
                self.send_header("Vary", "Accept-Encoding")
            if gz:
                self.send_header("Content-Encoding", "gzip")
            self.send_header("Content-Length", str(len(body)))
            self.send_header("Cache-Control", cache)
            for k, v in (extra or {}).items():
                self.send_header(k, v)
            self.end_headers()
            self.wfile.write(body)
        except (BrokenPipeError, ConnectionResetError):
            self.close_connection = True                 # the client hung up (closed tab, timed-out request): no one to answer

    def _json(self, status: int, obj) -> None:
        self._send(status, json.dumps(obj).encode(), "application/json")

    def _file(self, path: Path | None) -> None:
        if path is None:
            return self._json(404, {"error": "not found"})
        ctype = {".js": "text/javascript", ".mjs": "text/javascript", ".geojson": "application/geo+json",
                 ".json": "application/json"}.get(path.suffix) or mimetypes.guess_type(path.name)[0] \
            or "application/octet-stream"
        self._send(200, path.read_bytes(), ctype)

    def _host_ok(self) -> bool:
        """Refuse foreign Host headers (DNS rebinding): only this machine's own page may talk to us."""
        port = self._front.server_address[1]
        host = self.headers.get("Host", "")
        allowed = getattr(self._front, "allowed_hosts", ())       # --allow-host: the public name a proxy/tunnel passes on
        if host not in (f"127.0.0.1:{port}", f"localhost:{port}") and host.lower() not in allowed:
            self._json(403, {"error": "forbidden host"})
            return False
        if self.headers.get("Sec-Fetch-Site") == "cross-site" and urlparse(self.path).path.startswith(("/api/", "/tiles/")):
            self._json(403, {"error": "cross-site request refused"})   # other sites must not drive the API or fill the tile cache
            return False
        return True

    def do_GET(self):
        if not self._route() or not self._host_ok():
            return None
        url = urlparse(self.path)
        q = parse_qs(url.query)
        if url.path == "/api/meta":
            era = self._era(q)
            if era is None:
                return None
            w, src = self.server.worlds[era], self.server.sources[era]
            meta = w.meta()
            for L in meta["layers"]:
                L["file"] += f"?era={era}"
            for o in meta["overlays"]:
                o["file"] += f"?era={era}"
            if self.server.plateaus["features"]:
                meta["overlays"].append({"id": "plateaus", "name": "Sunken plateaus", "file": "/api/plateaus.geojson"})
            if self._public():
                meta = {**meta, "lore_ids": []}
            return self._json(200, self._px({**meta, "read_only": self._public(), "tiles": src.meta(), "era": era,
                                             "default_era": self.server.default_era,
                                             "eras": [{"name": n, "label": x.era["label"]} for n, x in self.server.worlds.items()]}))
        if url.path.startswith(PRIVATE_API) and self._public():
            return self._json(404, {"error": "not on the public map"})
        if url.path.startswith("/api/"):
            return self.api_get(url)
        if url.path.startswith("/tiles/"):
            return self._tile(url.path)
        if url.path == "/":
            return self._file(_safe(self.server.viewer_dir, "index.html"))
        for prefix, sub in (("/data/viewer/", "viewer"), ("/data/geo/", "geo")):
            if url.path.startswith(prefix):
                era = self._era(q)
                if era is None:
                    return None
                return self._file(_safe(self.server.worlds[era].out / sub, url.path[len(prefix):]))
        return self._file(_safe(self.server.viewer_dir, url.path.lstrip("/")))

    def _era(self, q) -> str | None:
        """The era a request names (default: the server's), or None after answering 400."""
        name = q.get("era", [self.server.default_era])[0] or self.server.default_era
        if name not in self.server.worlds:
            self._json(400, {"error": f"unknown era {name!r} (eras: {list(self.server.worlds)})"})
            return None
        return name

    def api_get(self, url):
        if url.path == "/api/cell":
            q = parse_qs(url.query)
            try:
                lat, lon = float(q["lat"][0]), float(q["lon"][0])
            except (KeyError, ValueError, IndexError):
                return self._json(400, {"error": "lat and lon must be numbers"})
            if not -90.0 <= lat <= 90.0:
                return self._json(400, {"error": "lat must be in [-90, 90]"})
            if not np.isfinite(lon):
                return self._json(400, {"error": "lon must be a finite number"})
            lon = (lon + 180.0) % 360.0 - 180.0
            era = self._era(q)
            if era is None:
                return None
            w = self.server.worlds[era]
            try:
                d = w.describe(lat, lon)
            except LookupError as e:
                return self._json(404, {"error": str(e)})
            z = float(self.server.sources[era].z_at(lat, lon)[0])
            d["fields"].insert(1, {"key": "z_detail_m", "label": "Local elevation (procedural)", "group": "Terrain",
                                   "unit": "m", "value": round(z, 1), "alt": w._alt("z_surface_m", z)})
            return self._json(200, d)
        if url.path == "/api/vents":
            era = self._era(parse_qs(url.query))
            return None if era is None else self._json(200, vents_of(self.server.sources[era]))
        if url.path == "/api/plateaus.geojson":
            return self._send(200, json.dumps(self.server.plateaus).encode(), "application/geo+json")
        if url.path == "/api/locate":
            return self._locate(parse_qs(url.query))
        if url.path == "/api/profile":
            return self._profile(parse_qs(url.query))
        if url.path.startswith("/api/pins"):
            return self._pins("GET")
        if url.path.startswith("/api/regions"):
            return self._regions("GET")
        if url.path == "/api/exports":
            return self._exports("GET")
        return self._json(404, {"error": "unknown endpoint"})

    def _tile(self, path: str):
        m = re.fullmatch(r"/tiles/(v\d+(?:\.\d+)?-[0-9a-f]+)/([a-z0-9_]+)/(\d+)/(\d+)/(\d+)\.(jpg|png|bin)", path)
        src = next((s for s in self.server.sources.values() if s.tag == m.group(1)), None) if m else None
        if not m or src is None or m.group(6) != tiles.EXT.get(m.group(2), "jpg"):
            return self._json(404, {"error": "no such tile"})
        try:
            data = src.tile(m.group(2), int(m.group(3)), int(m.group(4)), int(m.group(5)),
                            gate=getattr(self._front, "render_gate", None))
        except KeyError:
            return self._json(404, {"error": "no such tile"})
        except tiles.Busy:
            return self._send(503, b'{"error": "busy rendering, try again shortly"}', "application/json",
                              extra={"Retry-After": str(BUSY_RETRY_S)})
        ctype = {"jpg": "image/jpeg", "png": "image/png", "bin": "application/octet-stream"}[m.group(6)]
        # tile URLs carry their version tag: a public map lets browsers and caches keep them for good
        return self._send(200, data, ctype, "public, max-age=31536000, immutable" if self._public() else "max-age=86400")

    def _locate(self, q):
        import h3.api.basic_int as h3
        s = q.get("cell", [""])[0].strip().lower()
        try:
            c = h3.str_to_int(s)
            ok = bool(h3.is_valid_cell(c))
        except Exception:                    # h3 raises its own error types for junk ids
            ok = False
        if not ok:
            return self._json(400, {"error": "not an H3 cell id"})
        lat, lon = h3.cell_to_latlng(c)
        return self._json(200, {"cell": s, "lat": lat, "lon": lon, "res": h3.get_resolution(c)})

    def _profile(self, q):
        try:
            path = [tuple(float(v) for v in p.split(",")) for p in q.get("path", [""])[0].split(";") if p]
            n = int(q.get("n", ["256"])[0])
        except ValueError:
            return self._json(400, {"error": "path must be lat,lon;lat,lon;… and n an integer"})
        era = self._era(q)
        if era is None:
            return None
        try:
            return self._json(200, _profile(self.server.worlds[era], path, n, self.server.sources[era].surface_at))
        except ValueError as e:
            return self._json(400, {"error": str(e)})
        except LookupError as e:
            return self._json(404, {"error": str(e)})

    def _body(self):
        n = int(self.headers.get("Content-Length") or 0)
        if n < 0:
            raise ValueError("bad Content-Length")
        if n > MAX_BODY:
            raise ValueError("request body too large")
        try:
            return json.loads(self.rfile.read(n) or b"null")
        except json.JSONDecodeError:
            raise ValueError("body is not valid JSON")

    def _pins(self, method: str):
        parts = urlparse(self.path).path.split("/")          # ['', 'api', 'pins', '<id>'?]
        if parts[1:3] != ["api", "pins"] or len(parts) > 4:
            return self._json(404, {"error": "unknown endpoint"})
        pid = parts[3] if len(parts) == 4 else None
        if method in ("POST", "PUT") and self.headers.get_content_type() != "application/json":
            return self._json(415, {"error": "Content-Type must be application/json"})   # blocks cross-site form posts
        store = self.server.pins
        try:
            if method == "GET" and pid is None:
                return self._json(200, {"pins": store.list()})
            if method == "POST" and pid is None:
                return self._json(201, store.create(self._body()))
            if method == "PUT" and pid:
                pin = store.update(pid, self._body())
                return self._json(200, pin) if pin else self._json(404, {"error": "no such pin"})
            if method == "DELETE" and pid:
                if store.delete(pid):
                    self.send_response(204)
                    self.end_headers()
                    return None
                return self._json(404, {"error": "no such pin"})
        except ValueError as e:
            return self._json(400, {"error": str(e)})
        except OSError as e:
            return self._json(500, {"error": f"could not save pins: {e}"})
        return self._json(405, {"error": "method not allowed"})

    def _regions(self, method: str):
        parts = urlparse(self.path).path.split("/")          # ['', 'api', 'regions', '<id>'?]
        if parts[1:3] != ["api", "regions"] or len(parts) > 4:
            return self._json(404, {"error": "unknown endpoint"})
        rid = parts[3] if len(parts) == 4 else None
        if method == "POST" and self.headers.get_content_type() != "application/json":
            return self._json(415, {"error": "Content-Type must be application/json"})   # blocks cross-site form posts
        store, jobs = self.server.regions, getattr(self.server, "jobs", None)
        if method != "GET" and getattr(self.server, "regions_read_only", False):
            return self._json(403, {"error": "regions are read-only on this server"})
        try:
            if method == "GET" and rid is None:
                era = self._era(parse_qs(urlparse(self.path).query))
                if era is None:
                    return None
                status = jobs.status() if jobs else {"state": "idle"}
                try:
                    regions = refined_flags(self.server, store.list())
                except RegionsFileError as e:
                    regions, status = [], {**status, "state": "failed", "error": str(e)}
                return self._json(200, {"regions": regions, "status": status, "tiles": self.server.sources[era].meta()})
            if method == "POST" and rid is None:
                r = store.create(self._body(), float(self.server.world.cells_meta["radius_km"]))
                if jobs:
                    jobs.request()
                return self._json(201, r)
            if method == "DELETE" and rid:
                if store.delete(rid):
                    if jobs:
                        jobs.request()
                    self.send_response(204)
                    self.end_headers()
                    return None
                return self._json(404, {"error": "no such region"})
        except ValueError as e:
            return self._json(400, {"error": str(e)})
        except OSError as e:
            return self._json(500, {"error": f"could not save regions: {e}"})
        return self._json(405, {"error": "method not allowed"})

    def _exports(self, method: str):
        if urlparse(self.path).path != "/api/exports":
            return self._json(404, {"error": "unknown endpoint"})
        jobs = self.server.exports
        if method == "GET":
            return self._json(200, {"exports": jobs.list(), "status": jobs.status()})
        if method != "POST":
            return self._json(405, {"error": "method not allowed"})
        if self.headers.get_content_type() != "application/json":
            return self._json(415, {"error": "Content-Type must be application/json"})
        try:
            p = jobs.validate(self._body(), float(self.server.world.cells_meta["radius_km"]))
            if p["era"] is not None and p["era"] not in self.server.worlds:
                raise ValueError(f"unknown era {p['era']!r} (eras: {list(self.server.worlds)})")
            jobs.start(p)
        except ValueError as e:
            return self._json(400, {"error": str(e)})
        except (FileExistsError, RuntimeError) as e:
            return self._json(409, {"error": str(e)})
        return self._json(202, p)

    def _public(self) -> bool:
        """--public: a read-only map for anyone — nothing can be written, and pins, regions, exports and the lore
        ids (the author's notes and the names of lore entries) are not served."""
        return bool(getattr(self._front, "public", False))

    def _write(self, method: str):
        if not self._host_ok():
            return None
        if self._public():
            return self._json(403, {"error": "this map is read-only"})
        path = urlparse(self.path).path
        if path.startswith("/api/exports"):
            return self._exports(method)
        return self._regions(method) if path.startswith("/api/regions") else self._pins(method)

    def do_POST(self):
        return self._write("POST") if self._route() else None

    def do_PUT(self):
        return self._write("PUT") if self._route() else None

    def do_DELETE(self):
        return self._write("DELETE") if self._route() else None


def make_server(data_root: Path, res: int, port: int = 8765, viewer_dir: Path = VIEWER, lore_dir: Path | None = None,
                pins_path: Path | None = None, regions_path: Path | None = None,
                regions_dir: Path | None = None, workers: int = 0,
                exports_dir: Path | None = None, regions_read_only: bool = False,
                bind: str = "127.0.0.1") -> ThreadingHTTPServer:
    """workers > 0: that many forked render processes (before any thread starts: call from the main thread)."""
    server = ThreadingHTTPServer((bind, port), Handler)   # bind first: a busy port fails before the slow load
    try:
        if setup_world(server, data_root, res, viewer_dir, lore_dir, pins_path, regions_path, regions_dir, workers,
                       exports_dir, regions_read_only):
            server.jobs.request()                            # outlines without results for this world: build them
    except BaseException:   # any failed setup frees the port
        server.server_close()
        raise
    return server


def make_hub(entries: list, port: int = 8765, viewer_dir: Path = VIEWER, lore_dir: Path | None = None,
             workers: int = 0, extra_workers: int = 2, bind: str = "127.0.0.1") -> ThreadingHTTPServer:
    """One server for several worlds: entries = [(id, data_root, res)], each at /w/<id>/; the first is the default.
    The first world gets `workers` render processes, the others `extra_workers` (all forked before any thread)."""
    server = ThreadingHTTPServer((bind, port), Handler)
    try:
        server.hub, missing = {}, []
        for i, (wid, root, res) in enumerate(entries):
            st = types.SimpleNamespace(socket=server.socket, server_address=server.server_address)
            if setup_world(st, root, res, viewer_dir, lore_dir, workers=workers if i == 0 else min(workers, extra_workers)):
                missing.append(st)
            server.hub[wid] = st
        server.default_world = entries[0][0]
        for st in missing:
            st.jobs.request()
    except BaseException:
        server.server_close()
        raise
    return server


def world_states(server) -> list:
    return list(server.hub.values()) if getattr(server, "hub", None) else [server]


def setup_world(server, data_root: Path, res: int, viewer_dir: Path = VIEWER, lore_dir: Path | None = None,
                pins_path: Path | None = None, regions_path: Path | None = None,
                regions_dir: Path | None = None, workers: int = 0,
                exports_dir: Path | None = None, regions_read_only: bool = False) -> bool:
    """Load a world onto `server` (a bound server or a hub's per-world namespace). True: refined regions are missing
    (the caller starts server.jobs.request() once every render process is forked)."""
    from mapgen import config as C
    worlds = load_worlds(data_root, res, lore_dir, log=lambda m: None)
    server.worlds = dict(worlds)
    default = ((C.load(data_root)[1].get("eras") or {}).get("default"))
    server.default_era = default if default in server.worlds else worlds[0][0]
    world = server.worlds[server.default_era]
    server.world, server.viewer_dir = world, viewer_dir
    server.pins_path = pins_path or (data_root / "places" / "pins.json")
    cfg = C.load(data_root)[0]
    server.pins = PinStore(server.pins_path, int(cfg["build"]["res_final"]))
    server.regions = RegionStore(regions_path or (data_root / "places" / "regions.json"))
    server.regions_dir = regions_dir or (data_root / "out" / f"r{res}" / "regions")
    server.regions_read_only = regions_read_only
    seed = int(cfg["build"]["seed"])
    base_name, base_world = worlds[0]
    base_src = tiles.TileSource(base_world, seed, regions_dir=server.regions_dir, name=base_name)
    server.sources = {base_name: base_src}
    for name, w in worlds[1:]:
        server.sources[name] = tiles.TileSource(w, seed, regions_dir=server.regions_dir, name=name,
                                                share=base_src)
    tiles.join(list(server.sources.values()))
    server.tiles = server.sources[server.default_era]
    server.plateaus = plateau_geojson(data_root, float(world.cells_meta["radius_km"]))
    server.tiles.start_workers(workers)
    server.prerender_z = PRERENDER_Z if workers > 0 else 0   # without render processes: no rendering ahead
    server.jobs = RefineJobs(server, data_root, res)
    server.data_root, server.res = data_root, res
    server.exports_dir = exports_dir or (data_root / "exports")
    server.exports = ExportJobs(server)
    server.surface = server.tiles.surface_at
    for name, w in server.worlds.items():
        w.regions = server.sources[name].regions
    try:
        return regions_missing(server)
    except RegionsFileError:
        return False                                     # reported by GET /api/regions; the map still works


def pick_res(root: Path, cfg: dict) -> int:
    """The final map if its viewer textures exist, else the dev map."""
    final = int(cfg["build"]["res_final"])
    return final if (root / "out" / f"r{final}" / "viewer" / "layers.json").exists() else int(cfg["build"]["res_dev"])


def world_id(root: Path, cfg: dict, taken) -> str:
    base = re.sub(r"[^a-z0-9]+", "-", (cfg.get("render", {}).get("name") or root.name).lower()).strip("-") or "world"
    wid, n = base, 2
    while wid in taken:
        wid, n = f"{base}-{n}", n + 1
    return wid


ENV_PREFIX = "WORLDMAP_"   # every option can come from the environment (a container): --tile-cache-gb ← WORLDMAP_TILE_CACHE_GB


def env_list(v: str) -> list:
    return [x for x in re.split(r"[,\s]+", v) if x]


def env_defaults(ap, environ) -> None:
    """Options from WORLDMAP_<OPTION> variables as defaults (a flag given on the command line still wins).
    Switches take 1/true/yes/on or 0/false/no/off; repeatable options (--world, --allow-host) a comma or space
    separated list, applied after parsing."""
    vals = {}
    for act in ap._actions:
        opt = next((o for o in act.option_strings if o.startswith("--")), None)
        if opt is None or act.dest == "help" or isinstance(act, argparse._AppendAction):
            continue
        name = ENV_PREFIX + opt[2:].upper().replace("-", "_")
        if name not in environ:
            continue
        v = environ[name].strip()
        if isinstance(act, argparse._StoreTrueAction):
            if v.lower() not in ("1", "true", "yes", "on", "0", "false", "no", "off", ""):
                raise ValueError(f"{name}={v!r}: use 1/0, true/false, yes/no or on/off")
            vals[act.dest] = v.lower() in ("1", "true", "yes", "on")
        else:
            try:
                vals[act.dest] = act.type(v) if act.type else v
            except ValueError:
                raise ValueError(f"{name}={v!r} is not a valid {opt} value") from None
    ap.set_defaults(**vals)


def _stop(signum, frame):
    raise KeyboardInterrupt


def default_workers() -> int:
    return max(0, min(6, (os.cpu_count() or 2) - 2))


def main(argv=None, root: Path | None = None) -> int:
    from mapgen import config as C
    ap = argparse.ArgumentParser(prog="serve.py")
    ap.add_argument("--res", type=int)
    ap.add_argument("--port", type=int, default=8765)
    ap.add_argument("--world", action="append", help="a world's folder (its config/, out/, places/); default: the current "
                    "folder. Repeat for several worlds on one server (the first opens by default)")
    ap.add_argument("--lore", help="a folder of notes (*.md with an `id:` line) that pins may link to")
    ap.add_argument("--low-memory", action="store_true",
                    help="build caches with fields on disk: slower first start, lower peak; same results "
                         "(or set WORLDGEN_LOW_MEMORY=1)")
    ap.add_argument("--public", action="store_true",
                    help="read-only map for anyone: no writes; pins, regions, exports and lore ids not served")
    ap.add_argument("--allow-host", action="append", default=[], metavar="NAME",
                    help="also accept requests for this Host (the public name a reverse proxy or tunnel passes on, "
                         "e.g. maps.example.org; repeat for more)")
    ap.add_argument("--bind", default="127.0.0.1", metavar="ADDR",
                    help="address to listen on (default 127.0.0.1: this machine only). Another address (e.g. 0.0.0.0 "
                         "in a container behind a proxy) needs --public")
    ap.add_argument("--tile-cache-gb", type=float, metavar="GB",
                    help="cap on the saved tiles (all worlds): past it the least recently used are deleted. "
                         "Default: no cap")
    ap.add_argument("--tile-keep-z", type=int, default=-1, metavar="Z",
                    help="with --tile-cache-gb: relief and mesh tiles down to zoom Z are never deleted and not "
                         "counted (the prerendered map a deploy ships)")
    ap.add_argument("--max-renders", type=int, metavar="N",
                    help="at most N tiles rendering or waiting for a render at once (all worlds); more → 503 + "
                         "Retry-After (the page retries). Default: no limit; --public: 4 × workers (at least 4)")
    ap.add_argument("--workers", type=int,
                    help="render processes for the first world (default: CPUs − 2, at most 6; 0: render in the server); "
                         "on a shared box leave cores for the other services")
    try:
        env_defaults(ap, os.environ)
    except ValueError as e:
        ap.error(str(e))
    a = ap.parse_args(argv)
    if not a.world and os.environ.get(ENV_PREFIX + "WORLD"):
        a.world = env_list(os.environ[ENV_PREFIX + "WORLD"])
    if not a.allow_host and os.environ.get(ENV_PREFIX + "ALLOW_HOST"):
        a.allow_host = env_list(os.environ[ENV_PREFIX + "ALLOW_HOST"])
    if a.bind not in ("127.0.0.1", "localhost") and not a.public:
        ap.error("--bind beyond this machine needs --public: a map reachable from outside must be read-only")
    if a.allow_host and not a.public:
        ap.error("--allow-host needs --public: a map reachable from outside must be read-only")
    if a.tile_cache_gb is not None and a.tile_cache_gb <= 0:
        ap.error("--tile-cache-gb must be > 0")
    if a.max_renders is not None and a.max_renders < 1:
        ap.error("--max-renders must be ≥ 1")
    if a.workers is not None and a.workers < 0:
        ap.error("--workers must be ≥ 0")
    if a.low_memory:
        os.environ["WORLDGEN_LOW_MEMORY"] = "1"
    roots = [Path(w).resolve() for w in a.world] if a.world else [root or Path.cwd()]
    lore = Path(a.lore).resolve() if a.lore else None
    workers = default_workers() if a.workers is None else a.workers
    entries = []
    for r in roots:
        cfg = C.load(r)[0]
        entries.append((world_id(r, cfg, {e[0] for e in entries}), r, a.res if a.res is not None else pick_res(r, cfg)))
    try:
        if len(entries) == 1:
            server = make_server(entries[0][1], entries[0][2], a.port, lore_dir=lore, workers=workers, bind=a.bind)
        else:
            server = make_hub(entries, a.port, lore_dir=lore, workers=workers, bind=a.bind)
    except MissingBuild as e:
        print(f"error: {e.code} (or pick another with --res N)", file=sys.stderr)
        return 2
    except OSError as e:
        if e.errno != errno.EADDRINUSE:
            raise
        print(f"error: port {a.port} is in use — is the map server already running? Try --port {a.port + 1}", file=sys.stderr)
        return 2
    server.public = a.public
    limit = a.max_renders if a.max_renders is not None else (max(4, 4 * workers) if a.public else None)
    server.render_gate = threading.BoundedSemaphore(limit) if limit else None
    server.allowed_hosts = {h.strip().lower() for h in a.allow_host if h.strip()}
    states = world_states(server)
    if a.tile_cache_gb is not None:
        cap = TileCap(int(a.tile_cache_gb * 1e9), a.tile_keep_z)
        for st in states:
            for src in st.sources.values():
                src.cap = cap
                cap.add_root(src.cache_root)
        cap.start()                                          # counts what is saved (and trims a lowered cap)
    for st in states:
        for w in st.worlds.values():
            servecache.prune(w.out, [w.cache] if w.cache else [])
        for src in st.sources.values():
            src.prune_region_caches()
            src.carry_stale()                                # regions rebuilt while stopped: keep what still holds
            threading.Thread(target=src.warm, daemon=True).start()   # KD-tree + elevation raster for deep zoom
        st.jobs.prerender()                                  # refined areas' tiles not yet saved (after a restart)
    for (wid, _, res), st in zip(entries, states):
        url = f"http://127.0.0.1:{a.port}" + (f"/w/{wid}/" if len(entries) > 1 else "")
        print(f"{st.world.cells_meta.get('name', 'World')} map: {url}  (H3 res {res})")
    print("Ctrl-C to stop")
    old_term = signal.signal(signal.SIGTERM, _stop)          # systemd/podman stop: as PID 1 in a container the
    try:                                                     # default action would be ignored
        server.serve_forever()
    except KeyboardInterrupt:
        pass
    finally:
        signal.signal(signal.SIGTERM, old_term)
        server.server_close()                                  # stops pre-rendering
        for st in states:
            if st.tiles.pool:
                st.tiles.pool.close()
    return 0


if __name__ == "__main__":
    sys.exit(main())