#!/usr/bin/env python3
"""Host-side management worker. No network listener or Docker socket in the API.

Only fixed operations cross the shared mailbox. Private state stays outside it.
Run with Python >= 3.10; installed by install-management.sh.
"""
import argparse
import hashlib
import json
import os
import re
import subprocess
import tempfile
import threading
import time
import uuid
from datetime import datetime, timezone
from pathlib import Path
from urllib.parse import urlsplit


class ManagementError(Exception):
    def __init__(self, code, status=400):
        super().__init__(code)
        self.code, self.status = code, status


def atomic(path, value, mode=0o600, preserve_owner=False):
    fd, name = tempfile.mkstemp(dir=path.parent, prefix=".management-")
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as out:
            os.chmod(name, mode)
            if preserve_owner and path.exists() and hasattr(os, "chown"):
                previous = path.stat()
                os.chown(name, previous.st_uid, previous.st_gid)
            out.write(value)
            out.flush()
            os.fsync(out.fileno())
        os.replace(name, path)
    finally:
        if os.path.exists(name):
            os.unlink(name)


def read_env(path):
    """Read literal dotenv values, including the quoted format we write.

    Compose performs interpolation; this reader never executes dotenv contents.
    """
    values = {}
    if not path.exists():
        return values
    for line in path.read_text(encoding="utf-8").splitlines():
        match = re.match(r"^(?:export\s+)?([A-Z][A-Z0-9_]*)\s*=(.*)$", line.strip())
        if not match:
            continue
        key, value = match.groups()
        value = value.strip()
        if value.startswith("'") and value.endswith("'"):
            value = value[1:-1].replace("\\'", "'")
        elif value.startswith('"') and value.endswith('"'):
            try:
                value = json.loads(value)
            except ValueError:
                value = value[1:-1]
        else:
            value = value.split(" #", 1)[0].strip()
        values[key] = value
    return values


def validate_url(url):
    if not isinstance(url, str) or len(url) > 2048 or re.search(r"[\s\x00-\x1f]", url):
        raise ManagementError("management.source_invalid")
    if re.fullmatch(r"git@[A-Za-z0-9.-]+:[A-Za-z0-9_./-]+", url):
        return
    parsed = urlsplit(url)
    if (parsed.scheme not in ("https", "http", "ssh") or not parsed.hostname
            or parsed.password or parsed.query or parsed.fragment
            or (parsed.username and not (parsed.scheme == "ssh" and parsed.username == "git"))
            or not parsed.path.strip("/")):
        raise ManagementError("management.source_invalid")


def version_key(tag):
    match = re.fullmatch(r"v?(\d+)\.(\d+)\.(\d+)(?:-rc\.(\d+))?", tag)
    if not match:
        return None
    major, minor, patch, rc = match.groups()
    return int(major), int(minor), int(patch), 1 if rc is None else 0, int(rc or 0)


def now():
    return datetime.now(timezone.utc).isoformat()


class Agent:
    def __init__(self, deploy, mailbox, private):
        self.deploy, self.mailbox, self.private = deploy, mailbox, private
        self.lock = threading.Lock()
        self.cache = {}
        self.job = self.load("job.json", None)
        # systemd KillMode=control-group stops descendants too. Never claim success
        # after a reboot or interrupted deployment.
        if self.job and self.job["state"] == "running":
            self.job.update(state="failed", finished_at=now(), log="Agent interrupted; inspect deployment before retrying.")
            self.save("job.json", self.job)

    def load(self, name, default):
        path = self.private / name
        return json.loads(path.read_text(encoding="utf-8")) if path.exists() else default

    def save(self, name, value):
        atomic(self.private / name, json.dumps(value, ensure_ascii=False))

    def git_environment(self):
        return {"GIT_TERMINAL_PROMPT": "0", "GIT_SSH_COMMAND": "ssh -oBatchMode=yes",
                "GIT_CONFIG_COUNT": "1", "GIT_CONFIG_KEY_0": "safe.directory", "GIT_CONFIG_VALUE_0": str(self.deploy.parent)}

    def command(self, args, timeout=30, **kwargs):
        return subprocess.run(args, cwd=self.deploy, timeout=timeout, check=True,
                              stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
                              env={**os.environ, **self.git_environment()}, **kwargs).stdout

    def sources(self):
        sources = self.load("sources.json", None)
        if sources is None:
            url = self.command(["git", "remote", "get-url", "origin"]).strip()
            validate_url(url)
            sources = [{"id": "origin", "name": "origin", "url": url}]
            self.save("sources.json", sources)
        return sources

    def refs(self, url, fresh=False):
        cached = self.cache.get(url)
        if cached and not fresh and time.monotonic() - cached[0] < 60:
            return cached[1]
        output = self.command(["git", "-c", "protocol.file.allow=never", "-c", "protocol.ext.allow=never",
                               "ls-remote", "--heads", "--tags", "--", url], timeout=20)
        refs = dict(line.split()[::-1] for line in output.splitlines() if len(line.split()) == 2)
        self.cache[url] = time.monotonic(), refs
        return refs

    def schema(self):
        return json.loads((self.deploy / "config.schema.json").read_text(encoding="utf-8"))

    def revision(self):
        return hashlib.sha256((self.deploy / "config.local").read_bytes()).hexdigest()

    def config(self):
        env = read_env(self.deploy / ".env")
        local = read_env(self.deploy / "config.local")
        entries = []
        for item in self.schema():
            key = item["key"]
            source = "env" if key in env else "local" if key in local else "default"
            effective = env.get(key, local.get(key, item["default_value"]))
            def display(value):
                return "••••••••" if item["secret"] and value else value
            entries.append({**item, "env": display(env.get(key)), "local": display(local.get(key)),
                            "effective": display(effective), "source": source})
        return {"entries": entries, "revision": self.revision(),
                "restart_required": self.load("applied.json", "") != self.revision()}

    def save_config(self, body):
        if self.job and self.job["state"] == "running":
            raise ManagementError("management.busy", 409)
        if body.get("revision") != self.revision():
            raise ManagementError("management.config_conflict", 409)
        values = body.get("values")
        if not isinstance(values, dict) or len(values) > 200:
            raise ManagementError("management.config_invalid")
        schema = {v["key"]: v for v in self.schema()}
        local = read_env(self.deploy / "config.local")
        for key, value in values.items():
            item = schema.get(key)
            if not item or item["readonly"]:
                raise ManagementError("management.config_readonly")
            if value is None:
                local.pop(key, None)
                continue
            if not isinstance(value, str) or len(value) > 8192 or any(c in value for c in "\r\n\x00"):
                raise ManagementError("management.config_invalid")
            if item["type"] == "boolean" and value not in ("true", "false", "1", "0"):
                raise ManagementError("management.config_invalid")
            if item["type"] == "integer" and not re.fullmatch(r"-?\d+", value):
                raise ManagementError("management.config_invalid")
            if key == "LOG_LEVEL" and value not in ("debug", "info", "warn", "error"):
                raise ManagementError("management.config_invalid")
            if item["type"] == "integer" and key != "TRAIN_HOUR":
                minimum = -1 if key in ("PIPU_FILE_UID", "PIPU_FILE_GID") else 0
                if int(value) < minimum:
                    raise ManagementError("management.config_invalid")
            if key == "TRAIN_HOUR" and not -1 <= int(value) <= 23:
                raise ManagementError("management.config_invalid")
            if key.endswith("_PORT") and not 1 <= int(value) <= 65535:
                raise ManagementError("management.config_invalid")
            if item["type"] == "duration" and not re.fullmatch(r"(?:\d+(?:\.\d+)?(?:ns|us|µs|ms|s|m|h))+", value):
                raise ManagementError("management.config_invalid")
            local[key] = value
        # Single quotes make dollar signs literal to Compose. Newlines are rejected.
        content = "# Managed by Pipu. Environment and .env override these values.\n"
        content += "".join(k + "='" + v.replace("'", "\\'") + "'\n" for k, v in sorted(local.items()))
        path = self.deploy / "config.local"
        previous = path.read_text(encoding="utf-8")
        atomic(path, content, preserve_owner=True)
        try:
            self.command([*self.compose(), "config", "--quiet"])
        except (subprocess.SubprocessError, OSError):
            atomic(path, previous, preserve_owner=True)
            raise ManagementError("management.config_invalid") from None
        self.audit("config.save", list(values))
        return self.config()

    def compose(self):
        return ["docker", "compose", "--env-file", "config.local", "--env-file", ".env"]

    def audit(self, action, details):
        # Deliberately no configuration values or credentials in this log.
        with (self.private / "audit.jsonl").open("a", encoding="utf-8") as out:
            out.write(json.dumps({"at": now(), "actor": getattr(self, "actor", ""), "action": action, "details": details}) + "\n")

    def start_job(self, kind, target, command, extra_env=None):
        if self.job and self.job["state"] == "running":
            raise ManagementError("management.busy", 409)
        job = {"id": str(uuid.uuid4()), "kind": kind, "state": "running", "target": target,
               "started_at": now(), "finished_at": "", "log": ""}
        self.job = job
        self.save("job.json", job)
        revision = self.revision()
        self.audit(kind + ".start", {"id": job["id"], "target": target})
        threading.Thread(target=self.run_job, args=(job, command, extra_env or {}, revision), daemon=True).start()
        return dict(job)

    def run_job(self, job, command, extra_env, revision):
        # Let the API deliver the accepted job before recreating its container.
        time.sleep(2)
        # Capture full output privately, expose only a bounded redacted tail.
        secrets = [v for k, v in {**read_env(self.deploy / "config.local"), **read_env(self.deploy / ".env")}.items()
                   if re.search(r"TOKEN|PASSWORD|KEY|SECRET|DATABASE_URL|REDIS_URL", k) and len(v) >= 4]
        def redact(text):
            for value in secrets:
                text = text.replace(value, "[redacted]")
            return re.sub(r"(https?://)[^\s/@]+:[^\s/@]+@", r"\1[redacted]@", text)[-64000:]
        logpath = self.private / (job["id"] + ".log")
        result = 1
        try:
            with logpath.open("w", encoding="utf-8") as output:
                process = subprocess.Popen(command, cwd=self.deploy, stdout=output, stderr=subprocess.STDOUT,
                                           env={**os.environ, **self.git_environment(), **extra_env}, start_new_session=True)
            deadline = time.monotonic() + 7200
            while process.poll() is None:
                if time.monotonic() > deadline:
                    import signal
                    os.killpg(process.pid, signal.SIGKILL)
                    process.wait()
                    break
                with self.lock:
                    with logpath.open("rb") as output:
                        output.seek(max(0, logpath.stat().st_size - 128000))
                        job["log"] = redact(output.read().decode("utf-8", errors="replace"))
                    self.save("job.json", job)
                time.sleep(1)
            result = process.returncode
            with logpath.open("rb") as output:
                output.seek(max(0, logpath.stat().st_size - 128000))
                job["log"] = redact(output.read().decode("utf-8", errors="replace"))
            if result == 0:
                self.save("applied.json", revision)
        except (OSError, subprocess.SubprocessError):
            result = 1
            job["log"] += "\nCould not execute deployment command. Inspect journalctl -u pipu-management."
        finally:
            with self.lock:
                job.update(state="succeeded" if result == 0 else "failed", finished_at=now())
                self.save("job.json", job)

    def handle(self, op, body):
        if op == "config.get":
            return self.config()
        if op == "config.save":
            return self.save_config(body)
        if op == "config.apply":
            return self.start_job("config", "config.local", ["bash", str(self.deploy / "apply-config.sh")])
        if op == "sources.save":
            if self.job and self.job["state"] == "running":
                raise ManagementError("management.busy", 409)
            sources = body.get("sources")
            if not isinstance(sources, list) or not 1 <= len(sources) <= 20:
                raise ManagementError("management.source_invalid")
            ids = set()
            for source in sources:
                if (not isinstance(source, dict) or not re.fullmatch(r"[A-Za-z0-9_-]{1,64}", source.get("id", ""))
                        or not isinstance(source.get("name"), str) or not 1 <= len(source["name"].strip()) <= 100
                        or source["id"] in ids):
                    raise ManagementError("management.source_invalid")
                validate_url(source.get("url"))
                ids.add(source["id"])
            self.save("sources.json", sources)
            self.audit("sources.save", list(ids))
            return {"sources": sources}
        sources = self.sources()
        selected = next((s for s in sources if s["id"] == (body.get("source_id") or sources[0]["id"])), None)
        if not selected:
            raise ManagementError("management.source_invalid")
        if op == "updates.get":
            versions, error = [], ""
            try:
                refs = self.refs(selected["url"])
                versions = sorted([r.removeprefix("refs/tags/") for r in refs if r.startswith("refs/tags/") and version_key(r[10:])], key=version_key, reverse=True)
            except (subprocess.SubprocessError, OSError):
                error = "management.source_unavailable"
            return {"sources": sources, "versions": versions, "source_error": error,
                    "current_version": read_env(self.deploy / ".env").get("PIPU_VERSION", ""), "job": self.job}
        if op == "updates.start":
            if self.job and self.job["state"] == "running":
                raise ManagementError("management.busy", 409)
            channel = body.get("channel")
            if channel not in ("latest", "version", "main"):
                raise ManagementError("management.target_invalid")
            refs = self.refs(selected["url"], fresh=True)
            if channel == "latest":
                tags = [r[10:] for r in refs if r.startswith("refs/tags/") and version_key(r[10:]) and "-rc." not in r]
                if not tags:
                    raise ManagementError("management.no_releases")
                target = max(tags, key=version_key)
            else:
                target = "main" if channel == "main" else body.get("version", "")
            if target != "main" and not version_key(target):
                raise ManagementError("management.target_invalid")
            ref = "refs/heads/main" if target == "main" else "refs/tags/" + target
            commit = refs.get(ref + "^{}", refs.get(ref))
            if not commit or not re.fullmatch(r"[a-f0-9]{40,64}", commit):
                raise ManagementError("management.target_invalid")
            return self.start_job("update", target, ["bash", str(self.deploy / "update.sh"), target],
                                  {"PIPU_UPDATE_SOURCE": selected["url"], "PIPU_UPDATE_COMMIT": commit,
                                   "PIPU_UPDATE_REF": ref})
        raise ManagementError("management.operation_invalid")

    def tick(self):
        atomic(self.mailbox / "heartbeat", str(time.time()), 0o644)
        for path in sorted(self.mailbox.glob("*.request.json")):
            if path.is_symlink() or not re.fullmatch(r"[a-f0-9-]{36}\.request\.json", path.name):
                continue
            response = self.mailbox / path.name.replace(".request.json", ".response.json")
            if response.exists():
                path.unlink(missing_ok=True)
                continue
            try:
                if path.stat().st_size > 131072:
                    raise ManagementError("management.request_invalid")
                request = json.loads(path.read_text(encoding="utf-8"))
                if time.time() - request["created_at"] > 45:
                    raise ManagementError("management.request_expired", 409)
                # Claim before executing: a worker crash must never replay a mutation.
                path.unlink(missing_ok=True)
                with self.lock:
                    self.actor = request.get("actor", "")
                    result = {"data": self.handle(request["operation"], request.get("body") or {})}
            except ManagementError as error:
                result = {"code": error.code, "status": error.status}
            except (OSError, ValueError, KeyError, TypeError, subprocess.SubprocessError):
                result = {"code": "management.agent_error", "status": 502}
            atomic(response, json.dumps(result, ensure_ascii=False), 0o644)
            path.unlink(missing_ok=True)
        for path in self.mailbox.glob("*.response.json"):
            if not path.is_symlink() and time.time() - path.stat().st_mtime > 120:
                path.unlink(missing_ok=True)


def main():
    import fcntl  # Linux host only; tests import this module on Windows too.
    parser = argparse.ArgumentParser()
    parser.add_argument("--deploy-dir", type=Path, required=True)
    parser.add_argument("--mailbox", type=Path, required=True)
    parser.add_argument("--state-dir", type=Path, required=True)
    args = parser.parse_args()
    args.state_dir.mkdir(parents=True, exist_ok=True, mode=0o700)
    with (args.state_dir / "lock").open("w") as lock:
        fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
        agent = Agent(args.deploy_dir.resolve(), args.mailbox.resolve(), args.state_dir.resolve())
        while True:
            # Refresh the installed worker only after a successful deployment has
            # persisted its final status. The running worker survives checkout/build.
            with agent.lock:
                installed = Path(__file__).resolve()
                candidate = agent.deploy / "server-agent.py"
                if (agent.job and agent.job["kind"] == "update" and agent.job["state"] == "succeeded"
                        and installed == Path("/usr/local/lib/pipu/server-agent.py") and candidate.is_file()
                        and candidate.read_bytes() != installed.read_bytes()):
                    atomic(installed, candidate.read_text(encoding="utf-8"), 0o700)
                    # The singleton lock is non-inheritable and released by exec.
                    import sys
                    os.execv(sys.executable, [sys.executable, str(installed), *sys.argv[1:]])
            agent.tick()
            time.sleep(0.2)


if __name__ == "__main__":
    main()