"""broker-control: minimal internal control API for Mosquitto.

Scope (spec section 29):
    POST /deploy   { "content": "<mosquitto.conf text>" } -> validate + atomic activate + reload
    POST /reload                                          -> signal config reload
    GET  /status                                           -> broker process state
    GET  /health/live                                      -> liveness

Security model:
    * Listens on 0.0.0.0:8090 but is ONLY reachable on the backend-net
      Docker network; the port is not published to the host.
    * Requires a bearer token (CONTROL_API_TOKEN) shared with the backend.
    * Never mounts docker.sock; signals the broker over the shared PID
      namespace (broker is PID 1 in this container's view).

Implementation notes:
    * Atomic activation writes content to a temp file in the SAME
      directory as the target, then os.replace() renames it over the old
      file. os.replace is atomic and is NOT affected by Mosquitto 2.1's
      symlink-read restrictions (which we deliberately do not relax via
      MOSQUITTO_UNSAFE_ALLOW_SYMLINKS).
    * The broker runs as uid/gid 1883 in the official image, and so does
      this sidecar (inherited from the base image), so file ownership is
      consistent.
"""

from __future__ import annotations

import json
import os
import signal
import subprocess
import sys
import tempfile
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

TOKEN = os.environ.get("CONTROL_API_TOKEN", "")
CONFIG_DIR = os.environ.get("MOSQUITTO_CONFIG_DIR", "/mosquitto/config")
ACTIVE_CONF = os.path.join(CONFIG_DIR, "mosquitto.conf")
STAGE_DIR = os.path.join(CONFIG_DIR, "staged")
LISTEN_PORT = int(os.environ.get("CONTROL_LISTEN_PORT", "8090"))

_LOCK = threading.Lock()


class _Error(Exception):
    def __init__(self, status: int, message: str) -> None:
        super().__init__(message)
        self.status = status
        self.message = message


def _validate(content: str) -> dict[str, object]:
    """Run `mosquitto -c <staged> --test-config` against staged content."""
    os.makedirs(STAGE_DIR, exist_ok=True)
    # Temp file inside STAGE_DIR so --test-config sees the same filesystem
    # (relative includes, cert paths etc. behave as in production).
    fd, path = tempfile.mkstemp(prefix="validate-", suffix=".conf", dir=STAGE_DIR)
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as f:
            f.write(content)
        proc = subprocess.run(
            ["/usr/sbin/mosquitto", "-c", path, "--test-config"],
            capture_output=True,
            text=True,
            timeout=30,
        )
        return {
            "valid": proc.returncode == 0,
            "stdout": proc.stdout,
            "stderr": proc.stderr,
        }
    except subprocess.TimeoutExpired as exc:
        return {"valid": False, "stdout": "", "stderr": f"validation timed out: {exc}"}
    finally:
        try:
            os.unlink(path)
        except OSError:
            pass


def _activate(content: str) -> None:
    """Atomically replace the active mosquitto.conf with validated content."""
    os.makedirs(os.path.dirname(ACTIVE_CONF), exist_ok=True)
    fd, tmp = tempfile.mkstemp(prefix=".mosquitto.conf.", dir=CONFIG_DIR)
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as f:
            f.write(content)
        # 0644: broker (uid 1883) must be able to read; same-uid sidecar writes it.
        os.chmod(tmp, 0o644)
        os.replace(tmp, ACTIVE_CONF)
    except BaseException:
        try:
            os.unlink(tmp)
        except OSError:
            pass
        raise


def _signal_reload() -> None:
    """Signal the broker (PID 1 in the shared PID namespace) to reload."""
    try:
        proc = subprocess.run(
            ["mosquitto_signal", "-p", "1", "config-reload"],
            capture_output=True,
            text=True,
            timeout=15,
        )
    except FileNotFoundError:
        # Fallback if the helper app is unavailable for any reason.
        os.kill(1, signal.SIGHUP)
        return
    if proc.returncode != 0:
        raise _Error(500, f"mosquitto_signal failed: {proc.stderr.strip()}")


def _restart_broker() -> None:
    """Gracefully stop the broker; Docker's restart policy revives it.

    The broker is PID 1 in the shared PID namespace. SIGTERM makes it
    shut down cleanly (persistence saved); `restart: unless-stopped`
    on the mosquitto service then starts the container again. This
    sidecar is a separate container and survives, so it can report
    status while the broker is down (spec section 29: restart broker
    where explicitly supported — no Docker socket involved).
    """
    # The HTTP response MUST be flushed before PID 1 dies, so the kill
    # happens after the handler returns — orchestrated by the route.
    os.kill(1, signal.SIGTERM)


def _broker_pid_alive() -> bool:
    try:
        # Broker is PID 1 in the shared PID namespace.
        with open("/proc/1/comm", encoding="utf-8") as f:
            return f.read().strip() == "mosquitto"
    except OSError:
        return False


def _post_broker_health() -> bool:
    """Post-signal health probe: a real dynsec roundtrip as the admin user.

    Uses mosquitto_ctrl rather than a $SYS subscription, because the 2.1
    auto-generated admin role set is not guaranteed to include $SYS read
    access — but it must always be able to administer dynamic security.

    NOTE: this sidecar shares the broker's PID namespace but NOT its
    network namespace — 127.0.0.1 here is the SIDECAR's loopback, where
    nothing listens. The broker is reached via Docker DNS ("mosquitto")
    on the backend-net network.
    """
    username = os.environ.get("MOSQUITTO_DYNSEC_USERNAME", "admin")
    token_pw = os.environ.get("MOSQUITTO_DYNSEC_PASSWORD", "")
    cmd = [
        "mosquitto_ctrl",
        "-h",
        "mosquitto",
        "-u",
        username,
        "-P",
        token_pw,
        "dynsec",
        "listClients",
    ]
    try:
        proc = subprocess.run(cmd, capture_output=True, text=True, timeout=10)
        return proc.returncode == 0
    except (subprocess.TimeoutExpired, OSError):
        return False


class Handler(BaseHTTPRequestHandler):
    server_version = "mosquitto-control/0.1"

    # -- plumbing -----------------------------------------------------------
    def _send(self, status: int, payload: dict[str, object]) -> None:
        body = json.dumps(payload).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def _authorized(self) -> bool:
        if not TOKEN:
            return False  # refuse to run unauthenticated
        header = self.headers.get("Authorization", "")
        return header == f"Bearer {TOKEN}"

    def _read_json(self) -> dict[str, object]:
        length = int(self.headers.get("Content-Length", "0"))
        if length <= 0 or length > 2_000_000:
            raise _Error(400, "missing or oversized body")
        try:
            data = json.loads(self.rfile.read(length).decode("utf-8"))
        except (UnicodeDecodeError, json.JSONDecodeError) as exc:
            raise _Error(400, f"invalid JSON body: {exc}") from exc
        if not isinstance(data, dict):
            raise _Error(400, "JSON body must be an object")
        return data

    def log_message(self, fmt: str, *args: object) -> None:
        sys.stderr.write("[control] " + (fmt % args) + "\n")

    # -- routes -------------------------------------------------------------
    def do_GET(self) -> None:
        if self.path == "/health/live":
            self._send(200, {"status": "ok"})
            return
        if not self._authorized():
            self._send(401, {"error": "unauthorized"})
            return
        if self.path == "/status":
            self._send(
                200,
                {
                    "broker_pid_alive": _broker_pid_alive(),
                    "config_file": ACTIVE_CONF,
                },
            )
            return
        self._send(404, {"error": "not found"})

    def do_POST(self) -> None:
        if not self._authorized():
            self._send(401, {"error": "unauthorized"})
            return
        try:
            if self.path == "/deploy":
                self._handle_deploy()
            elif self.path == "/restart":
                # Acknowledge BEFORE the broker dies: the response must
                # reach the client first, then the process is signaled
                # in a daemon thread so the handler completes cleanly.
                self._send(200, {"status": "restarting"})
                threading.Thread(target=_restart_broker, daemon=True).start()
            elif self.path == "/reload":
                _signal_reload()
                self._send(200, {"status": "reloading", "broker_healthy": _post_broker_health()})
            else:
                self._send(404, {"error": "not found"})
        except _Error as exc:
            self._send(exc.status, {"error": exc.message})
        except OSError as exc:
            self._send(500, {"error": f"os error: {exc}"})

    def _handle_deploy(self) -> None:
        body = self._read_json()
        content = body.get("content")
        if not isinstance(content, str) or not content.strip():
            raise _Error(400, "field 'content' (mosquitto.conf text) is required")
        with _LOCK:
            result = _validate(content)
            if not result["valid"]:
                self._send(422, {"status": "validation_failed", **result})
                return
            _activate(content)
        _signal_reload()
        self._send(
            200,
            {"status": "deployed", "broker_healthy": _post_broker_health()},
        )


def main() -> None:
    if not TOKEN:
        sys.stderr.write("[control] FATAL: CONTROL_API_TOKEN is not set\n")
        sys.exit(1)
    server = ThreadingHTTPServer(("0.0.0.0", LISTEN_PORT), Handler)
    sys.stderr.write(f"[control] listening on :{LISTEN_PORT}\n")
    server.serve_forever()


if __name__ == "__main__":
    main()