Fix federation flows and add versioned updater

docker/host-updater.py

@@ -0,0 +1,197 @@
+#!/usr/bin/env python3
+
+import json
+import os
+import signal
+import socketserver
+import subprocess
+import threading
+from datetime import datetime, timezone
+from http import HTTPStatus
+from http.server import BaseHTTPRequestHandler
+
+SOCKET_PATH = os.environ.get("HOST_UPDATER_SOCKET", "/var/run/synapsis-updater/updater.sock")
+TOKEN = os.environ.get("HOST_UPDATER_TOKEN", "")
+STATUS_FILE = os.environ.get("HOST_UPDATER_STATUS_FILE", "/opt/synapsis/updater-status.json")
+LOG_FILE = os.environ.get("HOST_UPDATER_LOG_FILE", "/opt/synapsis/updater.log")
+UPDATE_SCRIPT = os.environ.get("HOST_UPDATER_SCRIPT", "/opt/synapsis/update-local.sh")
+INSTALL_DIR = os.environ.get("INSTALL_DIR", "/opt/synapsis")
+
+status_lock = threading.Lock()
+current_process = None
+
+
+def now_iso():
+    return datetime.now(timezone.utc).isoformat()
+
+
+def ensure_parent_dirs():
+    os.makedirs(os.path.dirname(SOCKET_PATH), exist_ok=True)
+    os.makedirs(os.path.dirname(STATUS_FILE), exist_ok=True)
+    os.makedirs(os.path.dirname(LOG_FILE), exist_ok=True)
+
+
+def load_status():
+    if not os.path.exists(STATUS_FILE):
+        return {
+            "available": True,
+            "status": "idle",
+            "message": "Ready to update.",
+            "lastStartedAt": None,
+            "lastFinishedAt": None,
+            "lastExitCode": None,
+            "lastError": None,
+            "pid": None,
+        }
+
+    with open(STATUS_FILE, "r", encoding="utf-8") as handle:
+        return json.load(handle)
+
+
+def save_status(status):
+    with open(STATUS_FILE, "w", encoding="utf-8") as handle:
+        json.dump(status, handle, indent=2)
+
+
+def update_status(**changes):
+    with status_lock:
+        status = load_status()
+        status.update(changes)
+        save_status(status)
+        return status
+
+
+def is_authorized(headers):
+    expected = f"Bearer {TOKEN}"
+    return bool(TOKEN) and headers.get("Authorization", "") == expected
+
+
+def watch_process(process):
+    exit_code = process.wait()
+    update_status(
+        status="success" if exit_code == 0 else "error",
+        message="Synapsis update completed." if exit_code == 0 else "Synapsis update failed.",
+        lastFinishedAt=now_iso(),
+        lastExitCode=exit_code,
+        lastError=None if exit_code == 0 else f"Updater exited with code {exit_code}",
+        pid=None,
+    )
+
+
+class ThreadedUnixServer(socketserver.ThreadingMixIn, socketserver.UnixStreamServer):
+    daemon_threads = True
+
+
+class Handler(BaseHTTPRequestHandler):
+    server_version = "SynapsisHostUpdater/1.0"
+
+    def log_message(self, format, *args):
+        return
+
+    def send_json(self, status_code, payload):
+        body = json.dumps(payload).encode("utf-8")
+        self.send_response(status_code)
+        self.send_header("Content-Type", "application/json")
+        self.send_header("Content-Length", str(len(body)))
+        self.end_headers()
+        self.wfile.write(body)
+
+    def do_GET(self):
+        if self.path == "/health":
+            self.send_json(HTTPStatus.OK, {"ok": True})
+            return
+
+        if self.path == "/status":
+            if not is_authorized(self.headers):
+                self.send_json(HTTPStatus.UNAUTHORIZED, {"error": "Unauthorized"})
+                return
+            self.send_json(HTTPStatus.OK, load_status())
+            return
+
+        self.send_json(HTTPStatus.NOT_FOUND, {"error": "Not found"})
+
+    def do_POST(self):
+        global current_process
+
+        if self.path != "/update":
+            self.send_json(HTTPStatus.NOT_FOUND, {"error": "Not found"})
+            return
+
+        if not is_authorized(self.headers):
+            self.send_json(HTTPStatus.UNAUTHORIZED, {"error": "Unauthorized"})
+            return
+
+        with status_lock:
+            if current_process and current_process.poll() is None:
+                self.send_json(
+                    HTTPStatus.CONFLICT,
+                    {
+                        "ok": False,
+                        "status": "updating",
+                        "message": "An update is already running.",
+                    },
+                )
+                return
+
+            log_handle = open(LOG_FILE, "a", encoding="utf-8")
+            current_process = subprocess.Popen(
+                [UPDATE_SCRIPT],
+                cwd=INSTALL_DIR,
+                stdout=log_handle,
+                stderr=subprocess.STDOUT,
+                start_new_session=True,
+                env={
+                    **os.environ,
+                    "INSTALL_DIR": INSTALL_DIR,
+                },
+            )
+
+            update_status(
+                status="updating",
+                message="Synapsis update started.",
+                lastStartedAt=now_iso(),
+                lastFinishedAt=None,
+                lastExitCode=None,
+                lastError=None,
+                pid=current_process.pid,
+            )
+
+            thread = threading.Thread(target=watch_process, args=(current_process,), daemon=True)
+            thread.start()
+
+        self.send_json(
+            HTTPStatus.ACCEPTED,
+            {
+                "ok": True,
+                "status": "updating",
+                "message": "Synapsis update started.",
+            },
+        )
+
+
+def cleanup_socket(*_args):
+    try:
+        if os.path.exists(SOCKET_PATH):
+            os.remove(SOCKET_PATH)
+    finally:
+        raise SystemExit(0)
+
+
+def main():
+    ensure_parent_dirs()
+
+    if os.path.exists(SOCKET_PATH):
+        os.remove(SOCKET_PATH)
+
+    signal.signal(signal.SIGTERM, cleanup_socket)
+    signal.signal(signal.SIGINT, cleanup_socket)
+
+    update_status(available=True)
+
+    with ThreadedUnixServer(SOCKET_PATH, Handler) as server:
+        os.chmod(SOCKET_PATH, 0o666)
+        server.serve_forever()
+
+
+if __name__ == "__main__":
+    main()