beaver_project/deploy-control/server.py

#!/usr/bin/env python3
from __future__ import annotations

import json
import os
import re
import subprocess
import time
from http import HTTPStatus
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from typing import Any
from urllib import error as urllib_error
from urllib import request as urllib_request


BASE_DIR = Path(__file__).resolve().parent
APP_INSTANCE_DIR = Path(os.environ.get("APP_INSTANCE_DIR", BASE_DIR.parent / "app-instance")).resolve()
ROUTER_PROXY_DIR = Path(os.environ.get("ROUTER_PROXY_DIR", BASE_DIR.parent / "router-proxy")).resolve()
CREATE_INSTANCE_SCRIPT = Path(
    os.environ.get("CREATE_INSTANCE_SCRIPT", APP_INSTANCE_DIR / "create-instance.sh")
).resolve()
REMOVE_INSTANCE_SCRIPT = Path(
    os.environ.get("REMOVE_INSTANCE_SCRIPT", APP_INSTANCE_DIR / "remove-instance.sh")
).resolve()
REGISTRY_TOOL = Path(
    os.environ.get("REGISTRY_TOOL", APP_INSTANCE_DIR / "instance-registry.py")
).resolve()
REGISTRY_PATH = Path(
    os.environ.get("REGISTRY_PATH", APP_INSTANCE_DIR / "runtime" / "registry" / "instances.json")
).resolve()
PROXY_RELOAD_SCRIPT = Path(
    os.environ.get("PROXY_RELOAD_SCRIPT", ROUTER_PROXY_DIR / "reload-proxy.sh")
).resolve()

API_TOKEN = os.environ.get("DEPLOY_CONTROL_API_TOKEN", "").strip()
INSTANCE_IMAGE = os.environ.get("APP_INSTANCE_IMAGE", "nano/app-instance:latest").strip()
INSTANCE_NETWORK_NAME = os.environ.get("APP_INSTANCE_NETWORK_NAME", "nano-instance-edge").strip()
DEFAULT_PROVIDER = os.environ.get("APP_INSTANCE_PROVIDER", "openai").strip()
DEFAULT_MODEL = os.environ.get("APP_INSTANCE_MODEL", "openai/gpt-5").strip()
DEFAULT_API_KEY = os.environ.get("APP_INSTANCE_API_KEY", "").strip()
DEFAULT_API_BASE = os.environ.get("APP_INSTANCE_API_BASE", "").strip()
DEFAULT_AUTHZ_BASE_URL = os.environ.get("DEFAULT_AUTHZ_BASE_URL", "").strip()
DEFAULT_AUTHZ_OUTLOOK_MCP_URL = os.environ.get("DEFAULT_AUTHZ_OUTLOOK_MCP_URL", "").strip()
DEFAULT_OUTLOOK_MCP_SERVER_ID = os.environ.get("DEFAULT_OUTLOOK_MCP_SERVER_ID", "outlook_mcp").strip() or "outlook_mcp"
PUBLIC_SCHEME = os.environ.get("DEPLOY_PUBLIC_SCHEME", "http").strip() or "http"
PUBLIC_BASE_DOMAIN = os.environ.get("DEPLOY_PUBLIC_BASE_DOMAIN", "127.0.0.1.nip.io").strip()
PUBLIC_HOST_TEMPLATE = os.environ.get("DEPLOY_PUBLIC_HOST_TEMPLATE", "{slug}.{base_domain}").strip()
PUBLIC_PORT = int(os.environ.get("DEPLOY_PUBLIC_PORT", "8088").strip() or "8088")
AUTO_START_PROXY = os.environ.get("DEPLOY_AUTO_START_PROXY", "1").strip() not in {"0", "false", "False"}
HEALTH_TIMEOUT_SECONDS = float(os.environ.get("DEPLOY_HEALTH_TIMEOUT_SECONDS", "60").strip() or "60")
HEALTH_INTERVAL_SECONDS = float(os.environ.get("DEPLOY_HEALTH_INTERVAL_SECONDS", "1").strip() or "1")
INSTANCE_INTERNAL_PORT = int(os.environ.get("APP_INSTANCE_INTERNAL_PORT", "8080").strip() or "8080")
SERVER_HOST = os.environ.get("DEPLOY_CONTROL_HOST", "0.0.0.0").strip() or "0.0.0.0"
SERVER_PORT = int(os.environ.get("DEPLOY_CONTROL_PORT", "8090").strip() or "8090")


class ApiError(Exception):
    def __init__(self, status_code: int, detail: str):
        super().__init__(detail)
        self.status_code = status_code
        self.detail = detail


def slugify(value: str) -> str:
    slug = re.sub(r"[^a-z0-9._-]+", "-", value.strip().lower()).strip("-")
    if not slug:
        raise ApiError(HTTPStatus.BAD_REQUEST, "instance id produced an empty slug")
    return slug


def run_command(args: list[str], *, cwd: Path | None = None, extra_env: dict[str, str] | None = None) -> str:
    env = os.environ.copy()
    if extra_env:
        env.update(extra_env)
    completed = subprocess.run(
        args,
        cwd=str(cwd) if cwd else None,
        env=env,
        text=True,
        capture_output=True,
        check=False,
    )
    if completed.returncode != 0:
        detail = completed.stderr.strip() or completed.stdout.strip() or "command failed"
        raise ApiError(HTTPStatus.BAD_GATEWAY, detail)
    return completed.stdout.strip()


def load_registry() -> dict[str, Any]:
    if not REGISTRY_PATH.exists():
        return {"instances": []}
    try:
        data = json.loads(REGISTRY_PATH.read_text(encoding="utf-8"))
    except json.JSONDecodeError:
        return {"instances": []}
    if not isinstance(data, dict):
        return {"instances": []}
    if not isinstance(data.get("instances"), list):
        data["instances"] = []
    return data


def get_registry_record(*, instance_id: str | None = None, username: str | None = None) -> dict[str, Any] | None:
    args = [str(REGISTRY_TOOL), "--registry", str(REGISTRY_PATH), "get"]
    if instance_id:
        args.extend(["--instance-id", instance_id])
    if username:
        args.extend(["--username", username])
    completed = subprocess.run(args, text=True, capture_output=True, check=False)
    if completed.returncode != 0:
        return None
    try:
        data = json.loads(completed.stdout)
    except json.JSONDecodeError:
        return None
    return data if isinstance(data, dict) else None


def ensure_network() -> None:
    result = subprocess.run(
        ["docker", "network", "inspect", INSTANCE_NETWORK_NAME],
        text=True,
        capture_output=True,
        check=False,
    )
    if result.returncode == 0:
        return
    run_command(["docker", "network", "create", INSTANCE_NETWORK_NAME])


def ensure_proxy() -> None:
    if AUTO_START_PROXY:
        run_command([str(PROXY_RELOAD_SCRIPT), "--start-if-missing"])
        return
    run_command([str(PROXY_RELOAD_SCRIPT)])


def build_public_host(*, slug: str, instance_id: str, username: str) -> str:
    try:
        host = PUBLIC_HOST_TEMPLATE.format(
            slug=slug,
            instance_id=instance_id,
            username=slugify(username),
            base_domain=PUBLIC_BASE_DOMAIN,
        ).strip()
    except KeyError as exc:
        raise ApiError(HTTPStatus.INTERNAL_SERVER_ERROR, f"invalid host template key: {exc}") from exc
    if not host:
        raise ApiError(HTTPStatus.INTERNAL_SERVER_ERROR, "public host template produced an empty host")
    return host


def build_public_url(host: str) -> str:
    default_port = 80 if PUBLIC_SCHEME == "http" else 443
    netloc = host if PUBLIC_PORT == default_port else f"{host}:{PUBLIC_PORT}"
    return f"{PUBLIC_SCHEME}://{netloc}"


def build_internal_api_base_url(record: dict[str, Any]) -> str:
    container_name = str(record.get("container_name", "") or "").strip()
    if container_name:
        return f"http://{container_name}:{INSTANCE_INTERNAL_PORT}"
    fallback = str(record.get("api_base_url", "") or record.get("public_url", "") or "").strip()
    return fallback


def wait_for_backend(record: dict[str, Any]) -> None:
    host_port = int(record.get("host_port", 0) or 0)
    container_name = str(record.get("container_name", "") or "").strip()
    targets: list[str] = []
    if container_name:
        targets.append(f"http://{container_name}:{INSTANCE_INTERNAL_PORT}/api/ping")
    if host_port > 0:
        targets.append(f"http://127.0.0.1:{host_port}/api/ping")
    if not targets:
        raise ApiError(HTTPStatus.BAD_GATEWAY, "instance health target missing from registry")

    deadline = time.time() + HEALTH_TIMEOUT_SECONDS
    last_error = "backend not ready"
    while time.time() < deadline:
        for target in targets:
            try:
                with urllib_request.urlopen(target, timeout=5) as response:
                    payload = json.loads(response.read().decode("utf-8"))
                if payload.get("message") == "pong":
                    return
                last_error = f"unexpected ping response from {target}"
            except (urllib_error.URLError, TimeoutError, json.JSONDecodeError) as exc:
                last_error = f"{target}: {exc}"
        time.sleep(HEALTH_INTERVAL_SECONDS)
    raise ApiError(HTTPStatus.BAD_GATEWAY, f"instance health check failed: {last_error}")


def create_or_get_instance(payload: dict[str, Any]) -> dict[str, Any]:
    username = str(payload.get("username", "") or "").strip()
    password = str(payload.get("password", "") or "")
    email = str(payload.get("email", "") or "").strip()
    if not username:
        raise ApiError(HTTPStatus.BAD_REQUEST, "username is required")
    if not password:
        raise ApiError(HTTPStatus.BAD_REQUEST, "password is required")

    instance_id = str(payload.get("instance_id", "") or username).strip()
    slug = slugify(instance_id)
    existing = get_registry_record(username=username) or get_registry_record(instance_id=instance_id)
    created = False

    if existing is None:
        ensure_network()
        public_host = build_public_host(slug=slug, instance_id=instance_id, username=username)
        public_url = build_public_url(public_host)
        provider = str(payload.get("provider", "") or DEFAULT_PROVIDER).strip() or DEFAULT_PROVIDER
        model = str(payload.get("model", "") or DEFAULT_MODEL).strip() or DEFAULT_MODEL
        api_key = str(payload.get("api_key", "") or DEFAULT_API_KEY).strip()
        api_base = str(payload.get("api_base", "") or DEFAULT_API_BASE).strip()
        authz_base_url = str(payload.get("authz_base_url", "") or DEFAULT_AUTHZ_BASE_URL).strip()
        authz_outlook_mcp_url = str(
            payload.get("authz_outlook_mcp_url", "") or DEFAULT_AUTHZ_OUTLOOK_MCP_URL
        ).strip()
        backend_name = str(payload.get("backend_name", "") or username).strip() or username
        image_name = str(payload.get("image_name", "") or INSTANCE_IMAGE).strip() or INSTANCE_IMAGE

        if not api_key:
            raise ApiError(HTTPStatus.BAD_REQUEST, "api key is required for new instances")

        command = [
            str(CREATE_INSTANCE_SCRIPT),
            "--image",
            image_name,
            "--instance-id",
            instance_id,
            "--auth-username",
            username,
            "--auth-password",
            password,
            "--username",
            username,
            "--email",
            email,
            "--provider",
            provider,
            "--model",
            model,
            "--api-key",
            api_key,
            "--backend-name",
            backend_name,
            "--public-url",
            public_url,
            "--instance-host",
            public_host,
            "--network",
            INSTANCE_NETWORK_NAME,
        ]
        if api_base:
            command.extend(["--api-base", api_base])
        if authz_base_url:
            command.extend(["--authz-base-url", authz_base_url])
        if authz_outlook_mcp_url:
            command.extend(["--authz-outlook-mcp-url", authz_outlook_mcp_url])
            command.extend(["--outlook-mcp-server-id", DEFAULT_OUTLOOK_MCP_SERVER_ID])
        if payload.get("replace") is True:
            command.append("--replace")

        run_command(command, cwd=APP_INSTANCE_DIR)
        existing = get_registry_record(instance_id=instance_id)
        created = True

    if existing is None:
        raise ApiError(HTTPStatus.BAD_GATEWAY, "instance was created but registry record is missing")

    wait_for_backend(existing)
    ensure_proxy()

    return {
        "created": created,
        "instance": existing,
        "public_url": str(existing.get("public_url", "") or ""),
        "frontend_base_url": str(existing.get("frontend_base_url", "") or existing.get("public_url", "") or ""),
        "api_base_url": build_internal_api_base_url(existing),
    }


def resolve_instance(payload: dict[str, Any]) -> dict[str, Any]:
    username = str(payload.get("username", "") or "").strip()
    if not username:
        raise ApiError(HTTPStatus.BAD_REQUEST, "username is required")
    record = get_registry_record(username=username)
    if record is None:
        raise ApiError(HTTPStatus.NOT_FOUND, "instance not found")
    return {
        "instance": record,
        "public_url": str(record.get("public_url", "") or ""),
        "frontend_base_url": str(record.get("frontend_base_url", "") or record.get("public_url", "") or ""),
        "api_base_url": build_internal_api_base_url(record),
    }


def remove_instance(instance_id: str, purge_data: bool) -> dict[str, Any]:
    if not instance_id.strip():
        raise ApiError(HTTPStatus.BAD_REQUEST, "instance id is required")
    command = [str(REMOVE_INSTANCE_SCRIPT), "--instance-id", instance_id]
    if purge_data:
        command.append("--purge-data")
    output = run_command(command, cwd=APP_INSTANCE_DIR)
    ensure_proxy()
    result: dict[str, str] = {}
    for line in output.splitlines():
        if "=" not in line:
            continue
        key, value = line.split("=", 1)
        result[key] = value
    return result


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

    def _json_response(self, status_code: int, payload: dict[str, Any]) -> None:
        body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
        self.send_response(status_code)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def _read_json_body(self) -> dict[str, Any]:
        raw_length = self.headers.get("Content-Length", "0").strip() or "0"
        try:
            length = int(raw_length)
        except ValueError as exc:
            raise ApiError(HTTPStatus.BAD_REQUEST, "invalid content length") from exc
        raw = self.rfile.read(length) if length > 0 else b"{}"
        try:
            payload = json.loads(raw.decode("utf-8"))
        except json.JSONDecodeError as exc:
            raise ApiError(HTTPStatus.BAD_REQUEST, "invalid JSON body") from exc
        if not isinstance(payload, dict):
            raise ApiError(HTTPStatus.BAD_REQUEST, "JSON body must be an object")
        return payload

    def _require_auth(self) -> None:
        if not API_TOKEN:
            return
        authorization = self.headers.get("Authorization", "").strip()
        expected = f"Bearer {API_TOKEN}"
        if authorization != expected:
            raise ApiError(HTTPStatus.UNAUTHORIZED, "unauthorized")

    def log_message(self, format: str, *args: Any) -> None:
        return

    def do_GET(self) -> None:  # noqa: N802
        try:
            if self.path == "/healthz":
                self._json_response(HTTPStatus.OK, {"ok": True, "instances": len(load_registry().get("instances", []))})
                return
            raise ApiError(HTTPStatus.NOT_FOUND, "not found")
        except ApiError as exc:
            self._json_response(exc.status_code, {"detail": exc.detail})

    def do_POST(self) -> None:  # noqa: N802
        try:
            self._require_auth()
            if self.path == "/api/instances/register":
                payload = self._read_json_body()
                self._json_response(HTTPStatus.OK, create_or_get_instance(payload))
                return
            if self.path == "/api/instances/resolve":
                payload = self._read_json_body()
                self._json_response(HTTPStatus.OK, resolve_instance(payload))
                return
            raise ApiError(HTTPStatus.NOT_FOUND, "not found")
        except ApiError as exc:
            self._json_response(exc.status_code, {"detail": exc.detail})

    def do_DELETE(self) -> None:  # noqa: N802
        try:
            self._require_auth()
            if not self.path.startswith("/api/instances/"):
                raise ApiError(HTTPStatus.NOT_FOUND, "not found")
            instance_id = self.path.rsplit("/", 1)[-1]
            purge_data = self.headers.get("X-Purge-Data", "").strip() == "1"
            self._json_response(HTTPStatus.OK, remove_instance(instance_id, purge_data))
        except ApiError as exc:
            self._json_response(exc.status_code, {"detail": exc.detail})


def main() -> int:
    server = ThreadingHTTPServer((SERVER_HOST, SERVER_PORT), Handler)
    print(f"deploy-control listening on {SERVER_HOST}:{SERVER_PORT}")
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        pass
    finally:
        server.server_close()
    return 0


if __name__ == "__main__":
    raise SystemExit(main())