#!/usr/bin/env python3
"""Bounded shared Notification Service deployment. Python standard library only."""

import argparse
import base64
import contextlib
import datetime
import hashlib
import json
import os
from pathlib import Path, PurePosixPath
import re
import secrets
import subprocess
import sys
import time

NAMESPACE = "shared-notification"
MANAGED = "hrynco-shared-notification"
MIGRATIONS = ["20260501214629_InitialCreate", "20260502154249_PendingChanges"]
IMAGE = re.compile(r"registry\.hrynco\.com/[a-z0-9._/-]+@sha256:[a-f0-9]{64}$")
SHA = re.compile(r"[a-f0-9]{64}$")

# Marker must be provisioned only after findmnt proves the backing filesystem UUID.
# XFS/ext4 statfs plus exact marker prevents a missing-mount fallback onto another FS.
# Marker alone is not a live query of the block-device UUID: verify findmnt at preflight.
GUARD = r'''
import ctypes, json, os
buf = ctypes.create_string_buffer(512)
libc = ctypes.CDLL(None, use_errno=True)
if libc.statfs(b'/data', ctypes.byref(buf)) != 0:
    raise RuntimeError('storage statfs failed')
magic = ctypes.cast(buf, ctypes.POINTER(ctypes.c_long))[0]
expected = {'ext4': 0xEF53, 'xfs': 0x58465342}[os.environ['EXPECTED_FS']]
if magic != expected:
    raise RuntimeError('unexpected storage filesystem')
with open('/data/.invemory-storage.json', encoding='utf-8') as stream:
    marker = json.load(stream)
if marker != {'uuid': os.environ['EXPECTED_UUID'], 'role': os.environ['EXPECTED_ROLE']}:
    raise RuntimeError('unexpected storage UUID/role marker')
print('storage guard passed')
'''

# Variables are quoted by PostgreSQL format, never interpolated into SQL by the shell.
GRANTS = r'''
\getenv runtime_password RUNTIME_PASSWORD
SELECT format('CREATE ROLE notification_runtime LOGIN PASSWORD %L NOSUPERUSER NOCREATEDB NOCREATEROLE', :'runtime_password')
WHERE NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'notification_runtime') \gexec
SELECT format('ALTER ROLE notification_runtime PASSWORD %L', :'runtime_password') \gexec
REVOKE ALL ON DATABASE notification_service FROM PUBLIC;
GRANT CONNECT ON DATABASE notification_service TO notification_runtime;
REVOKE CREATE ON SCHEMA public FROM PUBLIC;
GRANT USAGE ON SCHEMA public TO notification_runtime;
GRANT SELECT, INSERT, UPDATE, DELETE ON email_channels, email_templates, email_channel_usage TO notification_runtime;
'''

SANDBOX = r'''
BEGIN;
INSERT INTO email_channels (id, service_name, priority, provider_type, settings, warn_threshold_percent, is_active, created)
SELECT gen_random_uuid(), 'Invemory.Staging.Sandbox', 1, 1,
 '{"EmailChannelType":1,"Host":"mailpit","Port":1025,"Username":"","Password":"","UseSsl":false,"FromEmail":"no-reply@local.hrynco.test","FromName":"Invemory Sandbox"}'::jsonb, 90, true, now()
WHERE NOT EXISTS (SELECT 1 FROM email_channels WHERE service_name = 'Invemory.Staging.Sandbox');
INSERT INTO email_templates (id, service_name, key, language_code, subject, html_body, text_body, variables, created)
VALUES (gen_random_uuid(), 'Invemory.Staging.Sandbox', 'TestEmail', 'en', 'Test notification',
 '<p>Hello {{RecipientName}},</p><p>{{Message}}</p>', 'Hello {{RecipientName}}, {{Message}}',
 '[{"name":"RecipientName","required":true},{"name":"Message","required":true}]'::jsonb, now())
ON CONFLICT (service_name, key, language_code) DO NOTHING;
COMMIT;
'''

SANDBOX_CHECK = """SELECT
 (SELECT count(*) = 1 AND bool_and(provider_type = 1 AND is_active AND
 settings->>'Host' = 'mailpit' AND settings->>'Port' = '1025' AND
 settings->>'UseSsl' = 'false' AND settings->>'Username' = '' AND settings->>'Password' = '')
 FROM email_channels WHERE service_name = 'Invemory.Staging.Sandbox') AND
 EXISTS (SELECT 1 FROM email_templates WHERE service_name = 'Invemory.Staging.Sandbox'
 AND key = 'TestEmail' AND language_code = 'en'
 AND variables @> '[{"name":"RecipientName","required":true},{"name":"Message","required":true}]'::jsonb)"""


def read_json(path):
    return json.loads(Path(path).read_text(encoding="utf-8"))


def write_private(path, value):
    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
    fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
    with os.fdopen(fd, "w", encoding="utf-8") as stream:
        json.dump(value, stream, indent=2)
        stream.write("\n")


def validate(config, artifacts):
    if config.get("namespace") != NAMESPACE:
        raise ValueError("this package manages only shared-notification")
    if not re.fullmatch(r"[a-z0-9][a-z0-9.-]*", config["node"]):
        raise ValueError("invalid node name")
    storage = config["storage"]
    path = PurePosixPath(storage["basePath"])
    if not path.is_absolute() or ".." in path.parts or str(path) != storage["basePath"]:
        raise ValueError("storage must be an absolute normalized path")
    if str(path) not in ("/home/devops/shared-notification", "/mnt/shared/notification-service"):
        raise ValueError("unexpected storage path; a new location requires package review")
    if storage["filesystem"] not in ("ext4", "xfs"):
        raise ValueError("unsupported filesystem")
    if not re.fullmatch(r"[a-f0-9-]{36}", storage["uuid"]):
        raise ValueError("invalid storage UUID")
    for name in ("postgresSize", "rabbitmqSize"):
        if not re.fullmatch(r"[1-9][0-9]*Gi", storage[name]):
            raise ValueError("invalid storage capacity")
    for key in ("sandboxServiceName", "replyQueue", "rabbitVirtualHost"):
        if not re.fullmatch(r"[A-Za-z0-9._/-]{1,100}", config[key]):
            raise ValueError("invalid routing value")
    if config["sandboxServiceName"] != "Invemory.Staging.Sandbox":
        raise ValueError("initial package is restricted to the reviewed sandbox identity")
    if config["seqNamespace"] != "observability" or config["publisherNamespace"] != "invemory-staging":
        raise ValueError("unexpected dependent namespace")
    if config["seqUrl"] != "http://seq.observability.svc.cluster.local:8080":
        raise ValueError("unexpected Seq endpoint")
    for key in ("seqPodLabels", "publisherPodLabels"):
        if not isinstance(config[key], dict) or not config[key]:
            raise ValueError("exact dependent pod selector required")
    if not re.fullmatch(r"[a-f0-9]{40}", artifacts.get("sourceCommit", "")):
        raise ValueError("checked source commit required")
    if not SHA.fullmatch(artifacts.get("sourceArchiveSha256", "")):
        raise ValueError("verified source archive hash required")
    if artifacts.get("migrations") != MIGRATIONS:
        raise ValueError("migration list differs from reviewed source")
    expected = {"web", "worker", "migrator", "postgres", "rabbitmq", "mailpit", "guard"}
    if set(artifacts["images"]) != expected:
        raise ValueError("complete image set required")
    for name, artifact in artifacts["images"].items():
        if not IMAGE.fullmatch(artifact["image"]) or artifact["image"].endswith("0" * 64):
            raise ValueError("registry.hrynco.com immutable image required: " + name)
        if name in ("web", "worker", "migrator") and artifact.get("sourceArchiveSha256") != artifacts["sourceArchiveSha256"]:
            raise ValueError("application and migrator source archives differ")
    if artifacts["images"]["postgres"].get("version") != "17" or artifacts["images"]["rabbitmq"].get("version") != "4":
        raise ValueError("database/broker major upgrade requires separate reviewed procedure")


def labels(component=None):
    result = {"app.kubernetes.io/name": MANAGED, "app.kubernetes.io/managed-by": MANAGED}
    if component:
        result["app.kubernetes.io/component"] = component
    return result


def obj(kind, name, spec=None, namespace=NAMESPACE, api="v1", **extra):
    value = {"apiVersion": api, "kind": kind, "metadata": {"name": name, "labels": labels()}}
    if namespace:
        value["metadata"]["namespace"] = namespace
    if spec is not None:
        value["spec"] = spec
    value.update(extra)
    return value


def env(name, value=None, secret=None, key=None):
    if secret:
        return {"name": name, "valueFrom": {"secretKeyRef": {"name": secret, "key": key}}}
    return {"name": name, "value": str(value)}


def secure(uid=1000, gid=1000):
    return {"runAsNonRoot": True, "runAsUser": uid, "runAsGroup": gid,
            "allowPrivilegeEscalation": False, "capabilities": {"drop": ["ALL"]},
            "seccompProfile": {"type": "RuntimeDefault"}}


def manifest(config, artifacts):
    validate(config, artifacts)
    images = {key: value["image"] for key, value in artifacts["images"].items()}
    storage = config["storage"]
    output = [obj("Namespace", NAMESPACE, namespace=None)]
    output[0]["metadata"]["labels"].update({"pod-security.kubernetes.io/enforce": "restricted"})
    output.append(obj("ResourceQuota", "notification-budget", {"hard": {
        "pods": "8", "count/deployments.apps": "5", "count/jobs.batch": "3",
        "persistentvolumeclaims": "2", "requests.storage": str(int(storage["postgresSize"][:-2]) + int(storage["rabbitmqSize"][:-2])) + "Gi",
        "requests.cpu": "1600m", "requests.memory": "2Gi", "limits.cpu": "4", "limits.memory": "4Gi"}}))
    output.append(obj("LimitRange", "container-budget", {"limits": [{"type": "Container",
        "defaultRequest": {"cpu": "50m", "memory": "64Mi"}, "default": {"cpu": "250m", "memory": "256Mi"}}]}))
    for component, size, uid, gid in (("postgres", storage["postgresSize"], 999, 999), ("rabbitmq", storage["rabbitmqSize"], 100, 101)):
        pv = obj("PersistentVolume", "shared-notification-" + component, {
            "capacity": {"storage": size}, "accessModes": ["ReadWriteOnce"], "volumeMode": "Filesystem",
            "persistentVolumeReclaimPolicy": "Retain", "storageClassName": "",
            "claimRef": {"namespace": NAMESPACE, "name": component + "-data"},
            "local": {"path": storage["basePath"] + "/" + component},
            "nodeAffinity": {"required": {"nodeSelectorTerms": [{"matchExpressions": [{
                "key": "kubernetes.io/hostname", "operator": "In", "values": [config["node"]]}]}]}}}, namespace=None)
        pv["metadata"]["annotations"] = {"hrynco.com/storage-uuid": storage["uuid"], "hrynco.com/filesystem": storage["filesystem"]}
        output += [pv, obj("PersistentVolumeClaim", component + "-data", {
            "accessModes": ["ReadWriteOnce"], "volumeMode": "Filesystem", "storageClassName": "",
            "volumeName": pv["metadata"]["name"], "resources": {"requests": {"storage": size}}})]
    output.append(obj("ConfigMap", "rabbit-config", data={"rabbitmq.conf":
        "listeners.tcp.default = 5672\nmanagement.tcp.port = 15672\nmanagement.load_definitions = /etc/rabbitmq/definitions/definitions.json\n"}))
    output.append(obj("ConfigMap", "runtime-grants", data={"grants.sql": GRANTS}))
    output.append(obj("ConfigMap", "sandbox-seed", data={"sandbox.sql": SANDBOX}))
    output.append(obj("ConfigMap", "notification-config", data={
        "ASPNETCORE_ENVIRONMENT": "Production", "DOTNET_ENVIRONMENT": "Production",
        "ASPNETCORE_HTTP_PORTS": "8080", "App__RabbitMq__Host": "rabbitmq",
        "App__RabbitMq__Port": "5672", "App__RabbitMq__VirtualHost": config["rabbitVirtualHost"],
        "Serilog__WriteTo__1__Name": "Seq", "Serilog__WriteTo__1__Args__serverUrl": config["seqUrl"],
        "Serilog__Properties__Environment": "Shared", "Serilog__Properties__Service": "HrynCo.NotificationService"}))

    def workload(component, image, resources, environment=None, ports=None, volumes=None, mounts=None, replicas=1, uid=1000, gid=1000, **extra):
        container = {"name": component, "image": image, "imagePullPolicy": "IfNotPresent",
                     "resources": resources, "securityContext": secure(uid, gid)}
        if environment:
            container["env"] = environment
        if ports:
            container["ports"] = [{"name": name, "containerPort": port} for name, port in ports]
        if mounts:
            container["volumeMounts"] = mounts
        container.update(extra)
        pod = {"automountServiceAccountToken": False, "nodeSelector": {"kubernetes.io/hostname": config["node"]},
               "imagePullSecrets": [{"name": "registry-pull"}], "containers": [container],
               "terminationGracePeriodSeconds": 60}
        if volumes:
            pod["volumes"] = volumes
        if component in ("postgres", "rabbitmq"):
            pod["initContainers"] = [{"name": "storage-guard", "image": images["guard"],
                "command": ["python3", "-c", GUARD], "securityContext": secure(uid, gid),
                "resources": {"requests": {"cpu": "10m", "memory": "32Mi"}, "limits": {"cpu": "100m", "memory": "64Mi"}},
                "env": [env("EXPECTED_UUID", storage["uuid"]), env("EXPECTED_ROLE", NAMESPACE + "/" + component), env("EXPECTED_FS", storage["filesystem"])],
                "volumeMounts": [{"name": "data", "mountPath": "/data", "readOnly": True}]}]
        value = obj("Deployment", component, {"replicas": replicas, "strategy": {"type": "Recreate"},
            "selector": {"matchLabels": labels(component)}, "template": {"metadata": {"labels": labels(component),
            "annotations": {"hrynco.com/source-archive-sha256": artifacts["sourceArchiveSha256"]}}, "spec": pod}}, api="apps/v1")
        output.append(value)
        if ports:
            output.append(obj("Service", component, {"selector": labels(component), "ports": [
                {"name": name, "port": port, "targetPort": name} for name, port in ports]}))

    def res(cpu, memory, limit_cpu, limit_memory):
        return {"requests": {"cpu": cpu, "memory": memory}, "limits": {"cpu": limit_cpu, "memory": limit_memory}}

    workload("postgres", images["postgres"], res("100m", "256Mi", "500m", "512Mi"),
        [env("POSTGRES_DB", "notification_service"), env("POSTGRES_USER", "notification_owner"),
         env("POSTGRES_PASSWORD", secret="notification-private", key="postgresPassword"), env("PGDATA", "/data/pgdata")],
        [("postgres", 5432)], [{"name": "data", "persistentVolumeClaim": {"claimName": "postgres-data"}}],
        [{"name": "data", "mountPath": "/data"}], uid=999, gid=999,
        readinessProbe={"exec": {"command": ["pg_isready", "-U", "notification_owner", "-d", "notification_service"]}, "periodSeconds": 5},
        livenessProbe={"exec": {"command": ["pg_isready", "-U", "notification_owner", "-d", "notification_service"]}, "initialDelaySeconds": 30, "periodSeconds": 10})
    workload("rabbitmq", images["rabbitmq"], res("100m", "256Mi", "500m", "512Mi"),
        [env("RABBITMQ_NODENAME", "rabbit@rabbitmq"), env("RABBITMQ_DEFAULT_USER", "notification_admin"),
         env("RABBITMQ_DEFAULT_PASS", secret="notification-private", key="rabbitAdminPassword")],
        [("amqp", 5672), ("management", 15672)],
        [{"name": "data", "persistentVolumeClaim": {"claimName": "rabbitmq-data"}},
         {"name": "config", "configMap": {"name": "rabbit-config"}}, {"name": "definitions", "secret": {"secretName": "rabbit-definitions"}}],
        [{"name": "data", "mountPath": "/var/lib/rabbitmq"}, {"name": "config", "mountPath": "/etc/rabbitmq/rabbitmq.conf", "subPath": "rabbitmq.conf", "readOnly": True},
         {"name": "definitions", "mountPath": "/etc/rabbitmq/definitions", "readOnly": True}], uid=100, gid=101,
        startupProbe={"exec": {"command": ["rabbitmq-diagnostics", "-q", "ping"]}, "timeoutSeconds": 15, "periodSeconds": 20, "failureThreshold": 9},
        readinessProbe={"exec": {"command": ["rabbitmq-diagnostics", "-q", "check_running"]}, "timeoutSeconds": 15, "periodSeconds": 20},
        livenessProbe={"exec": {"command": ["rabbitmq-diagnostics", "-q", "ping"]}, "timeoutSeconds": 15, "periodSeconds": 30})
    output[-2]["spec"]["template"]["spec"]["hostname"] = "rabbitmq"
    workload("mailpit", images["mailpit"], res("25m", "32Mi", "100m", "128Mi"),
        [env("MP_SMTP_BIND_ADDR", "0.0.0.0:1025"), env("MP_UI_BIND_ADDR", "0.0.0.0:8025"), env("MP_MAX_MESSAGES", "100")],
        [("smtp", 1025), ("http", 8025)], readinessProbe={"httpGet": {"path": "/livez", "port": "http"}, "periodSeconds": 10})
    for component in ("web", "worker"):
        environment = [env("App__ConnectionString", secret="notification-private", key="runtimeConnectionString"),
                       env("Serilog__WriteTo__1__Args__apiKey", secret="notification-private", key="seqApiKey")]
        if component == "worker":
            environment += [env("App__RabbitMq__User", "notification_worker"), env("App__RabbitMq__Password", secret="notification-private", key="rabbitWorkerPassword")]
        workload(component, images[component], res("50m", "128Mi", "250m", "256Mi"), environment,
                 [("http", 8080)] if component == "web" else None, replicas=0,
                 **({"readinessProbe": {"httpGet": {"path": "/api/v1/email-channels?serviceName=" + config["sandboxServiceName"], "port": "http"}, "periodSeconds": 10}} if component == "web" else {}))
        deployment = output[-2] if component == "web" else output[-1]
        deployment["spec"]["template"]["spec"]["containers"][0]["envFrom"] = [{"configMapRef": {"name": "notification-config"}}]
    output += network_policies(config)
    return output


def peer(component=None, namespace=NAMESPACE, selector=None):
    return {"namespaceSelector": {"matchLabels": {"kubernetes.io/metadata.name": namespace}},
            "podSelector": {"matchLabels": selector if selector is not None else labels(component)}}


def network_policies(config):
    def policy(name, selector, ingress=None, egress=None, namespace=NAMESPACE):
        spec = {"podSelector": {"matchLabels": selector}, "policyTypes": []}
        if ingress is not None:
            spec.update(ingress=ingress)
            spec["policyTypes"].append("Ingress")
        if egress is not None:
            spec.update(egress=egress)
            spec["policyTypes"].append("Egress")
        return obj("NetworkPolicy", name, spec, namespace, "networking.k8s.io/v1")

    def port(number, protocol="TCP"):
        return {"protocol": protocol, "port": number}

    output = [policy("default-deny", {}, [], [])]
    output.append(policy("dns", labels(), egress=[{"to": [peer(namespace="kube-system", selector={"k8s-app": "kube-dns"})], "ports": [port(53, "UDP"), port(53)]}]))
    output.append(policy("postgres-ingress", labels("postgres"), ingress=[{"from": [peer(x) for x in ("web", "worker", "migrator", "grants", "sandbox")], "ports": [port(5432)]}]))
    output.append(policy("rabbitmq-ingress", labels("rabbitmq"), ingress=[{"from": [peer("worker"), peer(namespace=config["publisherNamespace"], selector=config["publisherPodLabels"])], "ports": [port(5672)]}]))
    output.append(policy("mailpit-ingress", labels("mailpit"), ingress=[{"from": [peer("worker"), peer("web")], "ports": [port(1025)]}]))
    for component in ("web", "worker", "migrator", "grants", "sandbox"):
        rules = [{"to": [peer("postgres")], "ports": [port(5432)]}]
        if component in ("web", "worker"):
            rules.append({"to": [peer(namespace=config["seqNamespace"], selector=config["seqPodLabels"])], "ports": [port(8080)]})
        if component == "worker":
            rules += [{"to": [peer("rabbitmq")], "ports": [port(5672)]}, {"to": [peer("mailpit")], "ports": [port(1025)]}]
        if component == "web":
            rules.append({"to": [peer("mailpit")], "ports": [port(1025)]})
        output.append(policy(component + "-egress", labels(component), egress=rules))
    # Additive allowances, separately owned; existing deny policies remain effective.
    output.append(policy("notification-seq-ingress", config["seqPodLabels"], ingress=[{
        "from": [peer("web"), peer("worker")], "ports": [port(8080)]}], namespace=config["seqNamespace"]))
    # Publisher remains LogOnly until its separate, explicitly controlled cutover.
    output.append(policy("notification-broker-egress", config["publisherPodLabels"], egress=[{
        "to": [peer("rabbitmq")], "ports": [port(5672)]}], namespace=config["publisherNamespace"]))
    return output


def rabbit_definitions(config, credentials):
    def hashed(password, salt):
        return base64.b64encode(salt + hashlib.sha256(salt + password.encode()).digest()).decode()

    users = []
    for name, key, tag in (("notification_admin", "rabbitAdminPassword", "administrator"), ("notification_worker", "rabbitWorkerPassword", ""), ("invemory_staging", "rabbitPublisherPassword", "")):
        salt = base64.b64decode(credentials[key + "Salt"])
        users.append({"name": name, "password_hash": hashed(credentials[key], salt), "hashing_algorithm": "rabbit_password_hashing_sha256", "tags": tag})
    incoming = re.escape("notification.send-email")
    reply = re.escape(config["replyQueue"])
    vhost = config["rabbitVirtualHost"]
    return {"users": users, "vhosts": [{"name": vhost}], "permissions": [
        {"user": "notification_admin", "vhost": vhost, "configure": ".*", "write": ".*", "read": ".*"},
        {"user": "notification_worker", "vhost": vhost, "configure": "^(" + incoming + "|" + reply + ")$", "write": "^(amq\\.default|" + reply + ")$", "read": "^" + incoming + "$"},
        {"user": "invemory_staging", "vhost": vhost, "configure": "^(" + incoming + "|" + reply + ")$", "write": "^(amq\\.default|" + incoming + ")$", "read": "^" + reply + "$"}],
        "queues": [{"name": name, "vhost": vhost, "durable": True, "auto_delete": False, "arguments": {}} for name in ("notification.send-email", config["replyQueue"])], "exchanges": [], "bindings": []}


def secret_manifests(config, credentials, registry):
    def secret(name, data, kind="Opaque"):
        return obj("Secret", name, type=kind, data={key: base64.b64encode(value.encode()).decode() for key, value in data.items()})

    data = {key: credentials[key] for key in ("postgresPassword", "runtimePassword", "rabbitAdminPassword", "rabbitWorkerPassword", "rabbitPublisherPassword", "seqApiKey")}
    data["ownerConnectionString"] = "Host=postgres;Database=notification_service;Username=notification_owner;Password=" + credentials["postgresPassword"]
    data["runtimeConnectionString"] = "Host=postgres;Database=notification_service;Username=notification_runtime;Password=" + credentials["runtimePassword"]
    return [secret("notification-private", data), secret("rabbit-definitions", {"definitions.json": json.dumps(rabbit_definitions(config, credentials))}),
            secret("registry-pull", {".dockerconfigjson": json.dumps(registry)}, "kubernetes.io/dockerconfigjson")]


class Cluster:
    def __init__(self, args):
        self.command = [args.kubectl, "--kubeconfig=" + args.kubeconfig, "--request-timeout=20s"]

    def run(self, *args, data=None, timeout=120, sensitive=False):
        result = subprocess.run(self.command + list(args), input=data, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=timeout)
        if result.returncode:
            # Error output can contain submitted secret resources or provider details.
            detail = "" if sensitive else result.stderr.decode(errors="replace")[:1000]
            raise RuntimeError("kubectl operation failed: " + " ".join(args[:3]) + ("\n" + detail if detail else ""))
        return result.stdout

    def get(self, kind, name=None, namespace=NAMESPACE):
        args = ["get", kind]
        if name:
            args.append(name)
        if namespace:
            args += ["-n", namespace]
        args += ["-o", "json"]
        if name:
            args.append("--ignore-not-found")
        data = self.run(*args)
        if not data.strip():
            if name:
                return None
            raise RuntimeError("kubectl returned no JSON for resource listing: " + kind)
        value = json.loads(data)
        if not name and (not isinstance(value, dict) or not isinstance(value.get("items"), list)):
            raise RuntimeError("kubectl returned an unexpected resource list: " + kind)
        return value

    def apply(self, values, dry_run=False, sensitive=False):
        args = ["apply", "-f", "-"]
        if dry_run:
            args += ["--dry-run=server"]
        self.run(*args, data=json.dumps({"apiVersion": "v1", "kind": "List", "items": values}).encode(), sensitive=sensitive)

    def scale(self, component, replicas):
        if self.get("deployment", component):
            self.run("scale", "deployment/" + component, "-n", NAMESPACE, "--replicas=" + str(replicas))
            if replicas == 0 and any(p["metadata"].get("labels", {}).get("app.kubernetes.io/component") == component for p in self.get("pods")["items"]):
                self.run("wait", "--for=delete", "pod", "-n", NAMESPACE, "-l", "app.kubernetes.io/component=" + component, "--timeout=120s", timeout=140)

    def rollout(self, component):
        self.run("rollout", "status", "deployment/" + component, "-n", NAMESPACE, "--timeout=240s", timeout=260)

    def history(self):
        table = self.run("exec", "deployment/postgres", "-n", NAMESPACE, "--", "psql", "-U", "notification_owner", "-d", "notification_service", "-At", "-c", "SELECT to_regclass('public.\"__EFMigrationsHistory\"')").decode().strip()
        if not table:
            return []
        return self.run("exec", "deployment/postgres", "-n", NAMESPACE, "--", "psql", "-U", "notification_owner", "-d", "notification_service", "-At", "-c", 'SELECT "MigrationId" FROM "__EFMigrationsHistory" ORDER BY "MigrationId"').decode().splitlines()


@contextlib.contextmanager
def lock(directory):
    directory = Path(directory)
    directory.mkdir(parents=True, exist_ok=True, mode=0o700)
    path = directory / "deployment.lock"
    try:
        fd = os.open(path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
    except FileExistsError:
        raise RuntimeError("deployment lock exists; verify no active deployment before manually removing a stale lock")
    try:
        os.write(fd, str(os.getpid()).encode())
        os.close(fd)
        yield
    finally:
        path.unlink()


def preflight(cluster, config, resources):
    node = cluster.get("node", config["node"], namespace=None)
    if not node or not any(x["type"] == "Ready" and x["status"] == "True" for x in node["status"]["conditions"]):
        raise RuntimeError("target storage node is not Ready")
    if any(x["type"] in ("DiskPressure", "MemoryPressure", "PIDPressure") and x["status"] == "True" for x in node["status"]["conditions"]):
        raise RuntimeError("target node has resource pressure")
    seq_pods = cluster.get("pods", namespace=config["seqNamespace"])["items"]
    if not any(all(p["metadata"].get("labels", {}).get(k) == v for k, v in config["seqPodLabels"].items()) for p in seq_pods):
        raise RuntimeError("Seq selector matches no existing pod; verify dependency labels")
    for item in resources:
        kind, name = item["kind"], item["metadata"]["name"]
        current = cluster.get(kind, name, item["metadata"].get("namespace"))
        if not current:
            continue
        if current["metadata"].get("labels", {}).get("app.kubernetes.io/managed-by") != MANAGED:
            raise RuntimeError("existing resource is not owned by this package: " + kind + "/" + name)
        if kind == "PersistentVolume":
            old = current["spec"]
            for key in ("local", "nodeAffinity", "capacity", "persistentVolumeReclaimPolicy"):
                if old.get(key) != item["spec"][key]:
                    raise RuntimeError("storage changes require a separate migration procedure")
            if current["metadata"].get("annotations", {}).get("hrynco.com/storage-uuid") != config["storage"]["uuid"]:
                raise RuntimeError("storage UUID change requires a separate migration procedure")
        if kind == "Deployment" and name in ("postgres", "rabbitmq"):
            if current["spec"]["template"]["spec"]["containers"][0]["image"] != item["spec"]["template"]["spec"]["containers"][0]["image"]:
                raise RuntimeError("database/broker image change requires separate backup/upgrade procedure")


def host_storage_check(config):
    """Run on the storage node, not on the developer workstation."""
    if sys.platform != "linux":
        raise RuntimeError("host storage preflight and deployment must run on the Linux storage node")
    for component, uid, gid in (("postgres", 999, 999), ("rabbitmq", 100, 101)):
        path = config["storage"]["basePath"] + "/" + component
        result = subprocess.run(["findmnt", "--json", "--target", path, "--output", "FSTYPE,UUID"], capture_output=True, check=True, timeout=20)
        filesystems = json.loads(result.stdout)["filesystems"]
        if len(filesystems) != 1 or filesystems[0]["fstype"] != config["storage"]["filesystem"] or filesystems[0]["uuid"] != config["storage"]["uuid"]:
            raise RuntimeError("host storage filesystem/UUID does not match reviewed configuration")
        info = os.stat(path)
        if info.st_uid != uid or info.st_gid != gid or info.st_mode & 0o007:
            raise RuntimeError("storage ownership or world permissions differ from provisioned contract")


def job(component, image, command=None, environment=None, volumes=None, mounts=None):
    container = {"name": component, "image": image, "securityContext": secure(),
                 "resources": {"requests": {"cpu": "100m", "memory": "128Mi"}, "limits": {"cpu": "500m", "memory": "512Mi"}}, "env": environment or []}
    if command:
        container["command"] = command
    if mounts:
        container["volumeMounts"] = mounts
    pod = {"restartPolicy": "Never", "automountServiceAccountToken": False,
           "imagePullSecrets": [{"name": "registry-pull"}], "containers": [container]}
    if volumes:
        pod["volumes"] = volumes
    return obj("Job", "notification-" + component, {"backoffLimit": 0, "activeDeadlineSeconds": 300,
        "template": {"metadata": {"labels": labels(component)}, "spec": pod}}, api="batch/v1")


def run_job(cluster, value):
    name = value["metadata"]["name"]
    cluster.run("delete", "job", name, "-n", NAMESPACE, "--ignore-not-found", "--wait=true", "--timeout=60s")
    cluster.apply([value])
    cluster.run("wait", "--for=condition=complete", "job/" + name, "-n", NAMESPACE, "--timeout=300s", timeout=320)


def backup(cluster, private_dir):
    folder = Path(private_dir) / ("backup-" + datetime.datetime.now(datetime.timezone.utc).strftime("%Y%m%dT%H%M%S%fZ"))
    folder.mkdir(mode=0o700)
    dump = cluster.run("exec", "deployment/postgres", "-n", NAMESPACE, "--", "pg_dump", "-U", "notification_owner", "-d", "notification_service", "-Fc", timeout=300, sensitive=True)
    if not dump.startswith(b"PGDMP"):
        raise RuntimeError("invalid PostgreSQL backup")
    fd = os.open(folder / "database.dump", os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
    with os.fdopen(fd, "wb") as stream:
        stream.write(dump)
    cluster.run("exec", "-i", "deployment/postgres", "-n", NAMESPACE, "--", "pg_restore", "--list", data=dump, sensitive=True)
    write_private(folder / "manifest.json", {"sha256": hashlib.sha256(dump).hexdigest(), "bytes": len(dump), "history": cluster.history(), "scope": "notification_service database only; broker PV retained, no offsite proof"})
    return folder.name


def database_ready(deployment):
    if not deployment or deployment.get("spec", {}).get("replicas") != 1:
        return False
    status = deployment.get("status", {})
    return (status.get("readyReplicas", 0) == 1 and status.get("updatedReplicas", 0) == 1 and
            status.get("observedGeneration", 0) >= deployment.get("metadata", {}).get("generation", 1))


def recovery_foundation(resources):
    # Ownership/storage/image preflight has already passed. Do not apply Web/Worker
    # targets or start their writers before history and a verified database backup.
    return [value for value in resources if not (
        value["kind"] in ("Deployment", "Service") and value["metadata"]["name"] in ("web", "worker"))]


def wait_for_worker_consumer(cluster, config, timeout=60):
    deadline = time.monotonic() + timeout
    while True:
        remaining = deadline - time.monotonic()
        if remaining <= 0:
            raise RuntimeError("expected exactly one notification.send-email consumer before timeout")
        try:
            output = cluster.run("exec", "deployment/rabbitmq", "-n", NAMESPACE, "--", "rabbitmqctl", "-q",
                                 "list_queues", "-p", config["rabbitVirtualHost"], "name", "consumers",
                                 timeout=min(20, remaining), sensitive=True).decode()
        except (RuntimeError, subprocess.TimeoutExpired):
            raise RuntimeError("could not verify notification.send-email consumer count") from None
        rows = [line.split() for line in output.splitlines() if line.split()[:1] == ["notification.send-email"]]
        if len(rows) != 1 or len(rows[0]) != 2 or rows[0][1] not in ("0", "1"):
            raise RuntimeError("expected exactly one notification.send-email consumer; unexpected broker count")
        remaining = deadline - time.monotonic()
        if remaining <= 0:
            raise RuntimeError("expected exactly one notification.send-email consumer before timeout")
        if rows[0][1] == "1":
            return
        # Process rollout can complete before the RabbitMQ subscription attaches.
        # Only zero consumers is a transient state; extra/unknown counts fail closed.
        time.sleep(min(2, remaining))


def deploy(cluster, config, artifacts, args):
    resources = manifest(config, artifacts)
    preflight(cluster, config, resources)
    host_storage_check(config)
    exists = cluster.get("namespace", NAMESPACE, namespace=None) is not None
    if not exists:
        raise RuntimeError("run bootstrap first, then inspect node capacity and server dry-run before deploy")
    if not cluster.get("secret", "notification-private") or not cluster.get("secret", "registry-pull"):
        raise RuntimeError("private credentials and registry pull Secret required")
    with lock(args.private_dir):
        # Never let an ordinary redeploy bypass migrations/schema validation.
        database = cluster.get("deployment", "postgres")
        current_worker = cluster.get("deployment", "worker")
        worker_replicas = current_worker["spec"].get("replicas", 0) if current_worker else 0
        if args.enable_worker and not args.sandbox_verified:
            raise RuntimeError("inspect sandbox channel/template and broker backlog before enabling the Worker")
        if not database_ready(database):
            if not args.migrate:
                raise RuntimeError("PostgreSQL is not ready; interrupted install/recovery requires --migrate")
            cluster.scale("worker", 0)
            cluster.scale("web", 0)
            cluster.apply(recovery_foundation(resources))
            for component in ("postgres", "rabbitmq", "mailpit"):
                cluster.rollout(component)
        # Deployment existence does not establish a running database or empty disk.
        # After recovery, query the actual retained DB rather than assume [] history.
        before = cluster.history()
        if before != MIGRATIONS and not args.migrate:
            raise RuntimeError("schema is not the reviewed target; --migrate required")
        if before != MIGRATIONS[:len(before)]:
            raise RuntimeError("database has unknown or non-prefix migration history")
        if args.migrate:
            cluster.scale("worker", 0)
            cluster.scale("web", 0)
            # Always back up the actual DB, including an empty initialized first DB.
            # A removed Deployment or interrupted first install may retain real data.
            backup(cluster, args.private_dir)
        # Preserve runtime replicas during normal reconciliation. First install stays stopped.
        for value in resources:
            if value["kind"] == "Deployment" and value["metadata"]["name"] in ("web", "worker"):
                value["spec"]["replicas"] = 0 if args.migrate else (1 if value["metadata"]["name"] == "web" else worker_replicas)
        cluster.apply(resources)
        for component in ("postgres", "rabbitmq", "mailpit"):
            cluster.rollout(component)
        if args.migrate:
            run_job(cluster, job("migrator", artifacts["images"]["migrator"]["image"], environment=[
                env("DOTNET_ENVIRONMENT", "Production"), env("App__ConnectionString", secret="notification-private", key="ownerConnectionString")]))
            if cluster.history() != MIGRATIONS:
                raise RuntimeError("migrator completed but schema history is not the reviewed target")
            run_job(cluster, job("grants", artifacts["images"]["postgres"]["image"],
                ["psql", "-h", "postgres", "-U", "notification_owner", "-d", "notification_service", "-v", "ON_ERROR_STOP=1", "-q", "-f", "/sql/grants.sql"],
                [env("PGPASSWORD", secret="notification-private", key="postgresPassword"), env("RUNTIME_PASSWORD", secret="notification-private", key="runtimePassword")],
                [{"name": "sql", "configMap": {"name": "runtime-grants"}}], [{"name": "sql", "mountPath": "/sql", "readOnly": True}]))
            run_job(cluster, job("sandbox", artifacts["images"]["postgres"]["image"],
                ["psql", "-h", "postgres", "-U", "notification_owner", "-d", "notification_service", "-v", "ON_ERROR_STOP=1", "-q", "-f", "/sql/sandbox.sql"],
                [env("PGPASSWORD", secret="notification-private", key="postgresPassword")],
                [{"name": "sql", "configMap": {"name": "sandbox-seed"}}], [{"name": "sql", "mountPath": "/sql", "readOnly": True}]))
        cluster.scale("web", 1)
        cluster.rollout("web")
        # Worker startup is separate: inspect/seed sandbox config before consuming anything.
        if args.enable_worker:
            safe = cluster.run("exec", "deployment/postgres", "-n", NAMESPACE, "--", "psql", "-U", "notification_owner", "-d", "notification_service", "-At", "-c", SANDBOX_CHECK).decode().strip()
            if safe != "t":
                raise RuntimeError("sandbox channel/template configuration does not match the safe seed")
            queues = cluster.run("exec", "deployment/rabbitmq", "-n", NAMESPACE, "--", "rabbitmqctl", "-q", "list_queues", "-p", config["rabbitVirtualHost"], "name", "messages_ready", "messages_unacknowledged").decode()
            if not any(line.split() == ["notification.send-email", "0", "0"] for line in queues.splitlines()):
                raise RuntimeError("incoming queue is not empty; inspect backlog before enabling the Worker")
            cluster.scale("worker", 1)
        # A preserved running Worker is also a required gate during ordinary deploy.
        if args.enable_worker or (not args.migrate and worker_replicas > 0):
            try:
                cluster.rollout("worker")
                wait_for_worker_consumer(cluster, config)
            except Exception:
                cluster.scale("worker", 0)
                raise
        print("deployment complete; SMTP is sandbox-only, Invemory publisher cutover is separate")


def main(argv=None):
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("command", choices=("render", "check", "bootstrap", "deploy", "backup"))
    parser.add_argument("--config", required=True)
    parser.add_argument("--artifacts", required=True)
    parser.add_argument("--kubectl", default="/snap/microk8s/current/kubectl")
    parser.add_argument("--kubeconfig", default="/home/devops/.kube/cluster-admin.conf")
    parser.add_argument("--private-dir", default="/home/devops/shared-notification-private")
    parser.add_argument("--output")
    parser.add_argument("--seq-key-file")
    parser.add_argument("--registry-auth-file")
    parser.add_argument("--migrate", action="store_true")
    parser.add_argument("--enable-worker", action="store_true")
    parser.add_argument("--sandbox-verified", action="store_true")
    args = parser.parse_args(argv)
    config, artifacts = read_json(args.config), read_json(args.artifacts)
    resources = manifest(config, artifacts)
    if args.command == "render":
        if not args.output:
            raise ValueError("render requires --output")
        Path(args.output).write_text(json.dumps({"apiVersion": "v1", "kind": "List", "items": resources}, indent=2) + "\n", encoding="utf-8")
        print("rendered reviewed resource package")
        return
    cluster = Cluster(args)
    if args.command == "bootstrap":
        if sys.platform != "linux":
            raise RuntimeError("private secret bootstrap must run on the Linux deployment host")
        if not args.seq_key_file or not args.registry_auth_file:
            raise ValueError("bootstrap requires private dedicated Seq ingest key and registry auth files")
        path = Path(args.private_dir) / "credentials.json"
        current_secret = cluster.get("secret", "notification-private")
        if path.exists():
            credentials = read_json(path)
        else:
            if current_secret:
                raise RuntimeError("private credential file missing while live Secret exists; recover original private file; do not regenerate passwords")
            credentials = {key: secrets.token_hex(32) for key in ("postgresPassword", "runtimePassword", "rabbitAdminPassword", "rabbitWorkerPassword", "rabbitPublisherPassword")}
            for key in ("rabbitAdminPassword", "rabbitWorkerPassword", "rabbitPublisherPassword"):
                credentials[key + "Salt"] = base64.b64encode(secrets.token_bytes(4)).decode()
            credentials["seqApiKey"] = Path(args.seq_key_file).read_text(encoding="utf-8").strip()
            if not credentials["seqApiKey"]:
                raise ValueError("dedicated Seq key is empty")
            write_private(path, credentials)
        if credentials["seqApiKey"] != Path(args.seq_key_file).read_text(encoding="utf-8").strip():
            raise RuntimeError("dedicated Seq key differs from preserved credentials; explicit rotation procedure required")
        registry = read_json(args.registry_auth_file)
        if set(registry.get("auths", {})) != {"registry.hrynco.com"}:
            raise ValueError("registry auth must contain registry.hrynco.com only")
        namespace = cluster.get("namespace", NAMESPACE, namespace=None)
        if namespace and namespace["metadata"].get("labels", {}).get("app.kubernetes.io/managed-by") != MANAGED:
            raise RuntimeError("refusing to adopt existing namespace")
        cluster.apply([resources[0]])
        values = secret_manifests(config, credentials, registry)
        for value in values:
            current = cluster.get("secret", value["metadata"]["name"])
            if current and current["metadata"].get("labels", {}).get("app.kubernetes.io/managed-by") != MANAGED:
                raise RuntimeError("refusing to adopt existing Secret")
            if current and value["metadata"]["name"] != "registry-pull" and current.get("data") != value["data"]:
                raise RuntimeError("preserved credentials or broker definitions differ from live Secret; explicit rotation/routing procedure required")
        cluster.apply(values, sensitive=True)
        print("private credentials preserved and Secrets reconciled; values not printed")
    elif args.command == "check":
        preflight(cluster, config, resources)
        host_storage_check(config)
        if not cluster.get("namespace", NAMESPACE, namespace=None):
            raise RuntimeError("bootstrap namespace/Secrets before server-side dry-run")
        cluster.apply(resources, dry_run=True)
        print("ownership, host storage UUID, pinned artifacts and server-side dry-run passed; measured node capacity remains mandatory")
    elif args.command == "backup":
        with lock(args.private_dir):
            if any((cluster.get("deployment", x) or {}).get("spec", {}).get("replicas", 0) for x in ("web", "worker")):
                raise RuntimeError("stop Web/Worker writers before backup")
            print("verified local backup: " + backup(cluster, args.private_dir))
    else:
        deploy(cluster, config, artifacts, args)


if __name__ == "__main__":
    try:
        main()
    except (ValueError, RuntimeError, KeyError, OSError, subprocess.TimeoutExpired, subprocess.CalledProcessError) as error:
        print("STOP: " + str(error), file=sys.stderr)
        sys.exit(1)