#!/usr/bin/env python3 """Manage local L0 independently of App installation and L1 processes.""" import argparse import json import os from pathlib import Path import secrets import subprocess import time from local_deployment import bind, cli, compose, image_file, phase, read_env, read_json, run, wait_http, worker_image_identity, write, write_env, write_json def l0_compose(root, settings, serving): state = root / "l0" base = {"image": settings["runtime_image"], "user": f"{os.getuid()}:{os.getgid()}", "restart": "unless-stopped"} environment = { "NEXUSCRM_RUNTIME_API_ADMIN_ADDR": ":9091", "NEXUSCRM_RUNTIME_API_GRPC_ADDR": ":9090", "NEXUSCRM_RUNTIME_API_TLS_CERTIFICATE_FILE": "/run/tls/server.crt", "NEXUSCRM_RUNTIME_API_TLS_PRIVATE_KEY_FILE": "/run/tls/server.key", } volumes = [bind(state / "secrets/tls" / name, "/run/tls/" + name) for name in ("server.crt", "server.key")] runtime = {**base, "environment": environment, "volumes": volumes, "ports": [f"127.0.0.1:{settings['grpc_port']}:9090", f"127.0.0.1:{settings['admin_port']}:9091"]} services = {"runtime-api": runtime} if not serving: runtime["command"] = ["bootstrap"] else: runtime["env_file"] = [str(state / "config/runtime.env")] for key, source, target in ( ("NEXUSCRM_RUNTIME_API_SERVING_CONFIG_FILE", "config/serving.json", "/etc/nexuscrm/serving.json"), ("NEXUSCRM_RUNTIME_API_SIGNING_KEY_FILE", "secrets/signing-keys.json", "/run/secrets/signing-keys.json"), ("NEXUSCRM_RUNTIME_WORKLOAD_IDENTITY_TOKEN_FILE", "secrets/workload-identities.json", "/run/secrets/workload-identities.json"), ): environment[key] = target volumes.append(bind(state / source, target)) for name in ("catalog", "files", "uploads"): volumes.append(bind(state / "data" / name, "/var/lib/nexuscrm/" + name, name == "catalog")) for path in sorted((state / "config/workers").glob("*.json")): stem = path.stem worker_env = {"NEXUSCRM_DB_MAX_OPEN_CONNS": "4", "NEXUSCRM_DB_MAX_IDLE_CONNS": "2", "NEXUSCRM_RUNTIME_WORKER_ADMIN_ADDR": ":9191", "NEXUSCRM_RUNTIME_WORKER_CONFIG_FILE": "/etc/nexuscrm/worker.json", "NEXUSCRM_RUNTIME_WORKLOAD_IDENTITY_TOKEN_FILE": "/run/secrets/workload.token"} mounts = [bind(path, "/etc/nexuscrm/worker.json"), bind(state / f"secrets/workers/{stem}.token", "/run/secrets/workload.token"), bind(state / "data/catalog", "/var/lib/nexuscrm/catalog")] for name in ("files", "uploads"): mounts.append(bind(state / "data" / name, "/var/lib/nexuscrm/" + name, False)) if stem == "record-event-dispatch": worker_env["NEXUSCRM_RUNTIME_WORKER_TIMED_FLOW_OWNER_CONFIG_FILE"] = "/etc/nexuscrm/scheduled-flow.json" mounts.append(bind(state / "config/workers/scheduled-flow.json", "/etc/nexuscrm/scheduled-flow.json")) services["worker-" + stem] = {**base, "image": settings["worker_image"], "env_file": [str(state / "config/runtime.env")], "environment": worker_env, "volumes": mounts, "healthcheck": {"test": ["CMD-SHELL", "wget -qO- http://127.0.0.1:9191/readyz >/dev/null"], "interval": "3s", "timeout": "2s", "retries": 30}} return {"name": settings["project"] + "-l0", "services": services, "networks": {"default": {"name": settings["project"] + "-backend", "external": True}}} def current_serving_mode(root): """The selected process composition owns mode; prepared metadata does not.""" command = read_json(root / "l0/compose.json")["services"]["runtime-api"].get("command") if command == ["bootstrap"]: return False if command is None or command == []: return True raise ValueError("unknown local Runtime command") def wait_process_health(identifiers, expected_services, *, inspect=None, timeout=180, clock=time.monotonic, pause=time.sleep): if inspect is None: inspect = lambda: json.loads(run(("docker", "inspect", *identifiers), capture=True, timeout=10)) health_services = {name for name, config in expected_services.items() if "healthcheck" in config} deadline = clock() + timeout pending = {} while clock() < deadline: observed = inspect() services = {item["Config"]["Labels"]["com.docker.compose.service"]: item["State"] for item in observed} if len(observed) != len(expected_services) or set(services) != set(expected_services): raise RuntimeError("L0 process inventory differs from its selected composition") pending = {name: state.get("Health", {}).get("Status", state.get("Status", "missing")) for name, state in services.items() if not state.get("Running") or (name in health_services and state.get("Health", {}).get("Status") != "healthy")} if not pending: return # Runtime owns readiness, including the scheduled-work cutover window. # A transient unhealthy/restarting observation is not a terminal verdict. pause(2) raise RuntimeError("L0 health deadline exceeded: " + json.dumps(pending, sort_keys=True)) def start_processes(root, settings, serving): compose(root, "l0", "up", "-d") identifiers = run(("docker", "compose", "-f", root / "l0/compose.json", "ps", "--all", "--quiet"), capture=True).split() if not identifiers: raise RuntimeError("L0 composition created no processes") expected = l0_compose(root, settings, serving)["services"] wait_process_health(identifiers, expected) wait_http(f"http://127.0.0.1:{settings['admin_port']}/" + ("readyz" if serving else "livez")) def initialize(args, root): if (root / "l0/settings.json").exists(): raise RuntimeError("L0 is already initialized; use start or activate") settings = {"runtime_image": args.runtime_image, "worker_image": args.worker_image, "project": args.project, "grpc_port": args.grpc_port, "admin_port": args.admin_port} if not args.runtime_image or not args.worker_image: raise ValueError("init requires --runtime-image and --worker-image") write_json(root / "database/compose.json", {"name": args.project + "-db", "services": {"tidb": { "image": "pingcap/tidb:v8.5.0", "command": ["--store=unistore", "--path=/data"], "ports": [f"127.0.0.1:{args.database_port}:4000"], "volumes": ["data:/data"]}}, "volumes": {"data": {}}, "networks": {"default": {"name": args.project + "-backend"}}}) compose(root, "database", "up", "-d") host_env = {"TIDB_HOST": "127.0.0.1", "TIDB_PORT": str(args.database_port), "TIDB_USER": "root", "TIDB_PASSWORD": "", "TIDB_DATABASE": "nexuscrm_local", "TIDB_TLS_DISABLE": "1", "JWT_SECRET": secrets.token_hex(32), "NEXUS_ADMIN_PASSWORD": secrets.token_urlsafe(18), "PUBLIC_BASE_URL": args.public_origin, "NEXUS_ENV": "development", "NEXUS_LOGIN_METHODS": "password", "NEXUS_PASSWORD_GRANT_ENABLED": "true"} write_env(root / "database.env", host_env) container_env = {**host_env, "TIDB_HOST": "tidb", "TIDB_PORT": "4000"} write_env(root / "l0/config/database.env", container_env) write(root / "admin-password", host_env["NEXUS_ADMIN_PASSWORD"] + "\n") # An open TCP port precedes TiDB's SQL readiness during startup. for attempt in range(60): ready = subprocess.run(("mysql", "--protocol=tcp", "-h", "127.0.0.1", "-P", str(args.database_port), "-u", "root", "--connect-timeout=2", "-e", "SELECT 1"), capture_output=True, timeout=5) if ready.returncode == 0: break time.sleep(1) else: raise RuntimeError("local TiDB did not start") run(("mysql", "--protocol=tcp", "-h", "127.0.0.1", "-P", args.database_port, "-u", "root", "-e", "CREATE DATABASE IF NOT EXISTS nexuscrm_local")) with phase(root, "l0-init"): run(("docker", "run", "--rm", "--network", args.project + "-backend", "--env-file", root / "l0/config/database.env", "--entrypoint", "/usr/local/bin/l0-controlplane", args.runtime_image, "init-l0"), timeout=300) tls = root / "l0/secrets/tls" tls.mkdir(parents=True, exist_ok=True) run((os.environ.get("OPENSSL", "openssl"), "req", "-x509", "-newkey", "rsa:2048", "-nodes", "-days", "365", "-subj", "/CN=localhost", "-addext", "subjectAltName=DNS:localhost,DNS:host.docker.internal,IP:127.0.0.1", "-keyout", tls / "server.key", "-out", tls / "server.crt"), capture=True) (tls / "server.key").chmod(0o600) write(root / "l0/config/runtime-server-name", "localhost\n") image_file(args.runtime_image, "/etc/nexuscrm/host-compatibility.json", root / "l0/config/host-compatibility.json") write_json(root / "l0/settings.json", settings) write_json(root / "l0/compose.json", l0_compose(root, settings, False)) compose(root, "l0", "up", "-d") wait_http(f"http://127.0.0.1:{args.admin_port}/livez") def upgrade(args, root): from standalone_l0_release import catalog_activator, upgrade_worker_bindings if not args.runtime_image or not args.worker_image: raise ValueError("upgrade requires --runtime-image and --worker-image") settings = read_json(root / "l0/settings.json") selected = (args.runtime_image, args.worker_image) images = [json.loads(run(("docker", "image", "inspect", image), capture=True))[0] for image in selected] immutable = [ref if "@sha256:" in ref else info["Id"] for ref, info in zip(selected, images)] worker_digest, worker_source = worker_image_identity(args.worker_image, images[1]) state = root / "l0" serving = current_serving_mode(root) installed = (state / "config/serving.json").exists() and (state / "config/workers").is_dir() catalog = args.catalog_image if installed and not catalog: catalog = (state / "config/catalog-controlplane-image").read_text().strip() with phase(root, "l0-upgrade"): compose(root, "l0", "stop") if installed: activate = catalog_activator(image=catalog, database_env=state / "config/database.env", evidence=state / "data/l0-worker-upgrade", network=settings["project"] + "-backend", execute=lambda command: run(command, timeout=300)) upgrade_worker_bindings(state_root=state, target_digest=worker_digest, target_source=worker_source, envv=read_env(root / "database.env"), activate=activate) settings.update(runtime_image=immutable[0], worker_image=immutable[1]) image_file(settings["runtime_image"], "/etc/nexuscrm/host-compatibility.json", state / "config/host-compatibility.json") write_json(state / "settings.json", settings) write_json(state / "compose.json", l0_compose(root, settings, serving)) start_processes(root, settings, serving) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("command", choices=("init", "activate", "start", "stop", "status", "down", "upgrade")) parser.add_argument("--root", type=Path, required=True) parser.add_argument("--runtime-image") parser.add_argument("--worker-image") parser.add_argument("--catalog-image", help="existing control-plane image for L0 upgrade") parser.add_argument("--project", default="nexuscrm-local") parser.add_argument("--database-port", type=int, default=4410) parser.add_argument("--grpc-port", type=int, default=3312) parser.add_argument("--admin-port", type=int, default=3313) parser.add_argument("--public-origin", default="http://127.0.0.1:3300") args = parser.parse_args() if args.command not in ("init", "upgrade") and (args.runtime_image or args.worker_image or args.catalog_image): parser.error("image selection is only available during init or upgrade") root = args.root.resolve() root.mkdir(parents=True, exist_ok=True) if args.command == "init": initialize(args, root) return if args.command == "upgrade": upgrade(args, root) return if args.command in ("activate", "start"): settings = read_json(root / "l0/settings.json") serving = args.command == "activate" or current_serving_mode(root) write_json(root / "l0/compose.json", l0_compose(root, settings, serving)) start_processes(root, settings, serving) else: compose(root, "l0", {"stop": "stop", "status": "ps", "down": "down"}[args.command]) if __name__ == "__main__": cli(main)