#!/usr/bin/env python3
"""Explicit, offline-capable pilot migration of an installer stack; Python 3.10+."""
import argparse
import copy
import hashlib
import importlib.util
import json
import os
from pathlib import Path
import re
import secrets
import shutil
import sys
import time
import urllib.parse

HERE = Path(__file__).resolve().parent
spec = importlib.util.spec_from_file_location("fedeo_update", HERE / "selfhost-update.py")
u = importlib.util.module_from_spec(spec)
spec.loader.exec_module(u)
Error = u.UpdateError
GIB = 1024 ** 3


class TransientStorageError(Error):
    pass


def require(condition, message):
    if not condition:
        raise Error(message)


def sha_file(path):
    h = hashlib.sha256()
    with Path(path).open("rb") as stream:
        for chunk in iter(lambda: stream.read(1024 * 1024), b""):
            h.update(chunk)
    return h.hexdigest()


def literal(value):
    """Resolved Compose JSON still undergoes interpolation when used as a file."""
    if isinstance(value, str):
        return value.replace("$", "$$")
    if isinstance(value, list):
        return [literal(v) for v in value]
    if isinstance(value, dict):
        return {k: literal(v) for k, v in value.items()}
    return value


def resolved(value):
    """Compose config serializes literal dollar signs escaped for another Compose read."""
    if isinstance(value, str):
        return value.replace("$$", "$")
    if isinstance(value, list):
        return [resolved(v) for v in value]
    if isinstance(value, dict):
        return {k: resolved(v) for k, v in value.items()}
    return value


def validate_minio_listing(root, bucket, inventory, *, exact=False):
    """MinIO can hide a/b from LIST when a exists, while GET a/b still works.

    Only inspect paths in the known single-drive layout; never decode or modify
    xl.meta. A cold equality check guards the completeness of the S3 inventory.
    """
    source = root / "minio"
    format_info = u.read_json(source / ".minio.sys/format.json", {})
    require(format_info.get("format") == "xl-single" and format_info.get("xl", {}).get("version") == "3",
            "MinIO-Datenformat ist für den zusätzlichen Vollständigkeitsabgleich nicht freigegeben.")
    bucket_root = source / bucket
    physical = {str(p.parent.relative_to(bucket_root)) for p in bucket_root.rglob("xl.meta") if p.is_file()}
    for key in physical:
        parts = key.split("/")
        require(all("/".join(parts[:n]) not in physical for n in range(1, len(parts))),
                "MinIO enthält einen Datei-/Verzeichniskonflikt; die S3-Auflistung kann Objekte verbergen.")
    if exact:
        require(physical == {o["Key"] for o in inventory["objects"]},
                "Kalte MinIO-Metadaten und S3-Inventar weichen ab; keine vollständige Übernahme nachweisbar.")


def validate_config(config, root):
    services = config.get("services", {})
    required = {"db", "minio", "createbuckets", "backend", "frontend", "traefik", "node-exporter"}
    require(required <= services.keys() and not (services.keys() - required - u.LEGACY_MATRIX_SERVICES),
            "Nur der Installer-Stack wird unterstützt; zusätzliche/fehlende Dienste zuerst klären.")
    for service in services.values():
        require(not any(service.get(k) for k in ("build", "profiles", "secrets", "configs", "volumes_from")),
                "Builds, Profile und externe Konfigurationen benötigen einen eigenen Migrationsplan.")
    store = services["minio"]
    require(not store.get("ports") and set(store.get("networks", {})) == {"internal"},
            "MinIO muss ausschließlich im internen Docker-Netz erreichbar sein.")
    require(not services["db"].get("ports"), "Veröffentlichte Datenbankports benötigen einen gesonderten Schreiber-Nachweis.")
    require(set(store.get("environment", {})) == {"MINIO_ROOT_USER", "MINIO_ROOT_PASSWORD"},
            "Zusätzliche MinIO-Konfiguration benötigt eine gesonderte Prüfung.")
    for name, target, directory in (("minio", "/data", "minio"), ("db", "/var/lib/postgresql/data", "postgres")):
        mounts = services[name].get("volumes", [])
        require(len(mounts) == 1 and mounts[0].get("type") == "bind" and mounts[0].get("target") == target
                and mounts[0].get("source") == str(root / directory), "Standard-Datenpfade fehlen: " + name)
        require((root / directory).is_dir() and (root / directory).resolve() == root / directory,
                "Datenpfad fehlt oder ist ein Link: " + directory)
    env = services["backend"].get("environment", {})
    require(env.get("FEDEO_FILE_BACKEND", "s3") == "s3" and not env.get("INFISICAL_CLIENT_ID"),
            "Externe Speicher/Secret-Provider benötigen einen eigenen Adapter.")
    require(env.get("S3_ENDPOINT", "").rstrip("/") == "http://minio:9000", "Abweichender S3-Endpunkt.")
    require(all(env.get(k) for k in ("S3_BUCKET", "S3_REGION", "S3_ACCESS_KEY", "S3_SECRET_KEY")), "S3-Konfiguration unvollständig.")
    require(re.fullmatch(r"[a-z0-9][a-z0-9.-]{1,61}[a-z0-9]", env["S3_BUCKET"]), "Ungültiger Bucketname.")
    require(env["S3_ACCESS_KEY"] == store["environment"]["MINIO_ROOT_USER"] and
            env["S3_SECRET_KEY"] == store["environment"]["MINIO_ROOT_PASSWORD"],
            "Getrennte MinIO-Benutzer vorhanden; deren Rechte zunächst vollständig inventarisieren.")
    db = urllib.parse.urlsplit(env.get("DATABASE_URL", ""))
    db_env = services["db"].get("environment", {})
    require(db.hostname == "db" and db.port in (None, 5432) and
            urllib.parse.unquote(db.path.lstrip("/")) == db_env.get("POSTGRES_DB") and
            urllib.parse.unquote(db.username or "") == db_env.get("POSTGRES_USER") and
            urllib.parse.unquote(db.password or "") == db_env.get("POSTGRES_PASSWORD"),
            "DATABASE_URL stimmt nicht mit der lokalen Sicherung überein.")
    return env


def index_containers(config, containers):
    """Keep previously retired Matrix containers without reactivating their services."""
    indexed, retained = {}, {}
    for container in containers:
        labels = container["Config"].get("Labels", {})
        name = labels.get("com.docker.compose.service")
        require(name and name not in indexed and labels.get("com.docker.compose.oneoff", "False").lower() != "true",
                "Unbekannte oder doppelte Projektcontainer vorhanden.")
        if name not in config["services"]:
            require(name in u.LEGACY_MATRIX_SERVICES, "Unbekannter Projektcontainer: " + name)
            state, host = container["State"], container["HostConfig"]
            require(not any(state.get(k) for k in ("Running", "Restarting", "Paused")) and
                    host.get("RestartPolicy", {}).get("Name") == "no" and not host.get("AutoRemove"),
                    "Alter Matrix-Container ist nicht dauerhaft stillgelegt: " + name)
            volumes = []
            for mount in container.get("Mounts", []):
                if mount["Type"] == "bind":
                    volumes.append({"type": "bind", "source": mount["Source"], "target": mount["Destination"]})
                else:
                    # These image-declared volumes belong to already retired services.
                    # Preserve their containers/volumes; never remove or restart them.
                    require(mount["Type"] == "volume" and (name, mount["Destination"]) in {
                        ("matrix-redis", "/data"), ("matrix-turn", "/var/lib/coturn")},
                        "Abweichender Datenmount eines alten Matrix-Containers: " + name)
            retained[name] = {"volumes": volumes}
        indexed[name] = container
    return indexed, retained


class Migration:
    def __init__(self, root, tools_image=None):
        os.umask(0o077)
        self.engine = u.Updater(root)
        self.root = self.engine.root
        self.home = self.engine.home / "storage-migration"
        require(not self.home.is_symlink(), "Migrationsverzeichnis darf kein Link sein.")
        self.home.mkdir(mode=0o700, exist_ok=True)
        self.runner = u.Runner(self.home / "commands.log")
        self.state = u.read_json(self.home / "state.json", {})
        self.tools_image = tools_image or self.state.get("toolsImage")
        require(not self.state or tools_image in (None, self.state.get("toolsImage")), "Image eines bestehenden Auftrags darf nicht ersetzt werden.")
        self.weed = u.read_json(HERE / "storage/seaweedfs.json")
        self.helper_name = "fedeo-storage-helper-" + hashlib.sha256(str(self.root).encode()).hexdigest()[:12]
        self.restore_name = self.helper_name + "-restore"

    def docker(self, *args, **kwargs):
        return self.runner.run(["docker", *args], **kwargs)

    def compose(self, file, *args, **kwargs):
        return self.docker("compose", "--project-directory", str(self.root), "--env-file", str(self.root / ".env"),
                           "-p", self.state["project"], "-f", str(file), *args, **kwargs)

    def save(self, phase=None, **fields):
        self.state.update(fields, updatedAt=u.now())
        if phase:
            self.state["phase"] = phase
        u.write_json(self.home / "state.json", self.state)

    def guard(self):
        # Compatible with the already installed 3.3 updater: it refuses recoveryRequired.
        # Kept after cutover until a provider-aware application updater is installed.
        u.write_json(self.engine.home / "status.json", {
            "status": "failed", "phase": "storage_migration", "job": str(self.home),
            "startedAt": self.state["startedAt"], "updatedAt": u.now(), "recoveryRequired": True,
            "maintenance": not self.state.get("writesReleased", False), "mutationStarted": True,
            "error": "Speichermigration: selfhost-storage-migrate.py status prüfen. Herkömmliche Updates sind gesperrt."})

    def fingerprint(self):
        return {name: sha_file(self.root / name) for name in ("docker-compose.yml", ".env")}

    def inspect(self, container):
        return json.loads(self.docker("inspect", container))[0]

    def wait_ready(self, container, timeout=180):
        for _ in range(timeout):
            state = self.inspect(container)["State"]
            require(state.get("Running"), "Ein benötigter Container wurde beendet.")
            if state.get("Health", {}).get("Status") in (None, "healthy"):
                return
            time.sleep(1)
        raise Error("Container wurde innerhalb der Wartezeit nicht bereit.")

    def cleanup_helper(self):
        ids = self.docker("ps", "-aq", "--filter", "name=^/" + self.helper_name + "$").split()
        for container in ids:
            require(self.inspect(container)["Config"].get("Labels", {}).get("io.fedeo.storage-job") == str(self.home),
                    "Fremder Container belegt den Hilfsnamen.")
            self.docker("rm", "-f", container)

    def helper(self, *args, **kwargs):
        deadline = time.monotonic() + 180
        while True:
            try:
                return self.helper_once(*args, **kwargs)
            except TransientStorageError:
                if time.monotonic() >= deadline:
                    raise
                time.sleep(2)

    def helper_once(self, command, *, source=False, admin=False, endpoint="http://objectstore:9000", directory="copy", backend_sdk=False):
        self.cleanup_helper()
        config = u.read_json(self.home / "credentials.json")
        job = {"bucket": config["bucket"], "directory": "/work", "result": "/work/result.json"}
        if command == "contract":
            job["foreignBucket"] = "fedeo-denied-" + secrets.token_hex(8)
        if source:
            job["source"] = config["source"]
        job["target"] = {**config["admin" if admin else "target"], "endpoint": endpoint}
        u.write_json(self.home / "job.json", job)
        work = self.home / directory
        work.mkdir(mode=0o700, exist_ok=True)
        (work / "error.json").unlink(missing_ok=True)
        runtime = [self.tools_image, command]
        if backend_sdk:
            runtime = ["--entrypoint", "node", "-e", "FEDEO_STORAGE_SDK_ROOT=" + self.state["backendWorkingDir"],
                       "-v", str(HERE / "storage") + ":/fedeo-check:ro", self.state["containers"]["backend"]["image"],
                       "/fedeo-check/cli.mjs", command]
        try:
            self.docker("run", "--rm", "--pull", "never", "--name", self.helper_name,
                        "--label", "io.fedeo.storage-job=" + str(self.home), "--network", self.state["network"],
                        "--user", str(os.getuid()) + ":" + str(os.getgid()),
                        "--read-only", "--cap-drop=ALL", "--security-opt", "no-new-privileges", "--tmpfs", "/tmp",
                        "-v", str(self.home / "job.json") + ":/config/job.json:ro", "-v", str(work) + ":/work",
                        *runtime, timeout=24 * 3600, output=self.home / "progress.jsonl")
            return u.read_json(work / "result.json")
        except Error:
            detail = u.read_json(work / "error.json", {})
            if detail.get("status") in (500, 502, 503, 504) or detail.get("code") in ("ECONNREFUSED", "ECONNRESET", "ETIMEDOUT", "EHOSTUNREACH"):
                raise TransientStorageError("S3 noch nicht bereit; Wiederholung blieb erfolglos.") from None
            raise
        finally:
            self.cleanup_helper()

    def plan(self):
        require(not self.state, "Ein Plan existiert bereits; status prüfen oder vorhandenen Auftrag fortsetzen.")
        require(not self.engine.status().get("recoveryRequired"), "Zuerst den offenen Update-Auftrag klären.")
        require(self.tools_image, "--tools-image mit lokalem Hilfsimage angeben.")
        require(all((self.root / n).is_file() and not (self.root / n).is_symlink() for n in (".env", "docker-compose.yml")),
                "Reguläre .env und docker-compose.yml erforderlich.")
        require(not any((self.root / n).exists() for n in ("docker-compose.override.yml", "docker-compose.override.yaml", "compose.yml", "compose.yaml")),
                "Compose-Overrides zuerst in einem gesonderten Plan berücksichtigen.")
        endpoint = self.docker("context", "inspect", "--format", "{{.Endpoints.docker.Host}}").strip()
        require(os.environ.get("DOCKER_HOST", endpoint).startswith("unix://"),
                "Nur ein lokaler Docker-Daemon wird unterstützt.")
        info = json.loads(self.docker("info", "--format", "{{json .}}"))
        require(info["OSType"] == "linux" and info["Architecture"] in ("amd64", "x86_64"), "Pilot benötigt Linux amd64.")
        config = resolved(json.loads(self.docker("compose", "--project-directory", str(self.root), "--env-file", str(self.root / ".env"),
                                       "-f", str(self.root / "docker-compose.yml"), "config", "--format", "json")))
        env = validate_config(config, self.root)
        project = config["name"]
        ids = self.docker("ps", "-aq", "--filter", "label=com.docker.compose.project=" + project).split()
        containers = [self.inspect(i) for i in ids]
        by_service, retained = index_containers(config, containers)
        # Validate retained bind mounts with the same strict path rules as configured
        # Matrix services. Do not put retired services back into either Compose file.
        self.engine.current = copy.deepcopy(config)
        self.engine.current["services"].update(retained)
        self.engine.retired_services = sorted(set(self.engine.current["services"]) & u.LEGACY_MATRIX_SERVICES)
        self.engine.matrix_backup_paths = []
        self.engine.validate_legacy_matrix()
        require(all(by_service.get(n, {}).get("State", {}).get("Running") for n in ("db", "minio", "backend", "frontend", "traefik")),
                "Die bestehende Instanz muss vor der Planung laufen.")
        for name in ("minio", "backend", "db"):
            actual = dict(v.split("=", 1) for v in by_service[name]["Config"]["Env"] if "=" in v)
            require(all(actual.get(k) == str(v) for k, v in config["services"][name].get("environment", {}).items()),
                    "Laufender Container und Compose-Umgebung weichen ab: " + name)
        network = config["networks"]["internal"]["name"]
        peers = json.loads(self.docker("network", "inspect", network))[0].get("Containers", {})
        require(set(peers) <= {c["Id"] for c in containers}, "Weitere Container im internen Netzwerk; externe Schreiber zuerst klären.")
        for name, target in (("minio", "/data"), ("db", "/var/lib/postgresql/data")):
            mounts = [m for m in by_service[name]["Mounts"] if m["Destination"] == target]
            require(len(mounts) == 1 and mounts[0]["Source"] == str(self.root / ("minio" if name == "minio" else "postgres")),
                    "Laufender Datenmount weicht ab: " + name)
        actual_store = by_service["minio"]
        require(not actual_store["HostConfig"].get("PortBindings") and set(actual_store["NetworkSettings"]["Networks"]) == {network},
                "Laufendes MinIO hat zusätzliche Netzwerkzugänge.")
        require(len(actual_store["Mounts"]) == 1, "Laufendes MinIO hat zusätzliche Mounts.")
        require(not any(p.is_symlink() for p in (self.root / "minio").rglob("*")), "Links in MinIO-Daten benötigen eigene Backup-Prüfung.")
        tools = json.loads(self.docker("image", "inspect", self.tools_image))[0]
        weed = json.loads(self.docker("image", "inspect", self.weed["image"]))[0]
        self.tools_image = tools["Id"]
        # Containerd may discard an untagged image index after a mutable build tag
        # moves. Keep an independent immutable job reference throughout the pilot.
        tools_tag = "fedeo-storage-tools:job-" + secrets.token_hex(12)
        self.docker("image", "tag", self.tools_image, tools_tag)
        self.state = {"schemaVersion": 1, "phase": "planning", "startedAt": u.now(), "project": project,
                      "network": network, "toolsImage": self.tools_image, "weedImage": weed["Id"],
                      "toolsRetainTag": tools_tag,
                      "version": env.get("FEDEO_VERSION"), "fingerprint": self.fingerprint(), "writesReleased": False,
                      "backendWorkingDir": by_service["backend"]["Config"].get("WorkingDir") or "/",
                      "matrixPaths": self.engine.matrix_backup_paths, "retainedServices": sorted(retained),
                      "containers": {name: {"id": c["Id"], "image": c["Image"],
                          "running": c["State"]["Running"], "restart": c["HostConfig"]["RestartPolicy"],
                          "health": c["State"].get("Health", {}).get("Status"),
                          "networks": c["NetworkSettings"]["Networks"]} for name, c in by_service.items()}}
        # Planning errors must not leave a usable partial state file.
        u.write_json(self.home / "original.json", config)
        for name in ("docker-compose.yml", ".env"):
            u.atomic(self.home / ("original" + name.replace("/", "-")), (self.root / name).read_bytes())
        u.write_json(self.home / "previous-update.json", u.read_json(self.engine.home / "status.json", {"status": "idle"}))
        admin = {"accessKeyId": "fedeo-migration-" + secrets.token_hex(8), "secretAccessKey": secrets.token_urlsafe(36)}
        app = {"accessKeyId": env["S3_ACCESS_KEY"], "secretAccessKey": env["S3_SECRET_KEY"]}
        connection = {"region": env["S3_REGION"], "credentials": app}
        u.write_json(self.home / "credentials.json", {"bucket": env["S3_BUCKET"],
                     "source": {**connection, "endpoint": env["S3_ENDPOINT"]}, "target": connection,
                     "admin": {**connection, "credentials": admin}})
        inventory = self.helper("inventory", source=True, directory="preflight")
        validate_minio_listing(self.root, env["S3_BUCKET"], inventory)
        physical = int(self.runner.run(["du", "-sb", str(self.root / "minio")]).split()[0])
        db_size = int(self.docker("exec", by_service["db"]["Id"], "sh", "-ec",
            'psql -U "$POSTGRES_USER" -d "$POSTGRES_DB" -Atc "SELECT pg_database_size(current_database())"').strip())
        image_ids = sorted({c["Image"] for c in containers} | {tools["Id"], weed["Id"]})
        image_bytes = sum(i["Size"] for i in json.loads(self.docker("image", "inspect", *image_ids)))
        matrix_bytes = sum(int(self.runner.run(["du", "-sb", str(self.root / p)]).split()[0]) for p in self.state["matrixPaths"])
        # Source backup + target + restored target + offline images + DB + spool + generous metadata reserve.
        needed = physical + inventory["bytes"] * 3 + inventory["count"] * 256 * 1024 + db_size * 2 + image_bytes * 2 + matrix_bytes + 5 * GIB
        require(shutil.disk_usage(self.root).free >= needed, f"Zu wenig freier Platz; mindestens {needed} Bytes erforderlich.")
        self.state.update(imageIds=image_ids, requiredFreeBytes=needed, freeBytes=shutil.disk_usage(self.root).free,
                          count=inventory["count"], bytes=inventory["bytes"])
        require(not (self.root / "storage").exists(), "storage/ existiert bereits; kein leeres Migrationsziel.")
        s3config = {"identities": [
            {"name": "migration-admin", "credentials": [{"accessKey": admin["accessKeyId"], "secretKey": admin["secretAccessKey"]}],
             "actions": ["Admin", "Read", "Write", "List", "Tagging"]},
            {"name": "fedeo", "credentials": [{"accessKey": app["accessKeyId"], "secretKey": app["secretAccessKey"]}],
             "actions": [action + ":" + env["S3_BUCKET"] for action in ("Read", "Write", "List")]}
        ]}
        u.write_json(self.home / "s3.json", s3config)
        self.make_compose(config)
        self.save("planned")

    def make_compose(self, original):
        target = copy.deepcopy(original)
        store = copy.deepcopy(self.weed)
        store["image"] = self.state["weedImage"]
        store.update(container_name="fedeo-objectstore", restart="no", networks={"internal": {}},
                     user=str(os.getuid()) + ":" + str(os.getgid()),
                     environment={"FEDEO_S3_CONFIG_SHA256": sha_file(self.home / "s3.json")},
                     volumes=[{"type": "bind", "source": str(self.root / "storage/seaweedfs"), "target": "/data"},
                              {"type": "bind", "source": str(self.root / "storage/config"), "target": "/etc/fedeo", "read_only": True}])
        stage = {"name": self.state["project"], "services": {"objectstore": store},
                 "networks": {"internal": {"external": True, "name": self.state["network"]}}}
        u.write_json(self.home / "stage.json", literal(stage))
        del target["services"]["minio"]
        del target["services"]["createbuckets"]
        target["services"]["objectstore"] = copy.deepcopy(store)
        target["services"]["objectstore"]["networks"]["internal"] = {"aliases": ["minio"]}
        target["services"]["objectstore"]["restart"] = "unless-stopped"
        dependencies = target["services"]["backend"].setdefault("depends_on", {})
        dependencies.pop("minio", None)
        dependencies.pop("createbuckets", None)
        dependencies["objectstore"] = {"condition": "service_healthy"}
        # Keep exactly the running application images; never couple this to an application/DB update.
        for name, c in self.state["containers"].items():
            if name in target["services"]:
                target["services"][name]["image"] = c["image"]
        target["services"]["backend"]["environment"]["FEDEO_RUN_MIGRATIONS"] = "false"
        # SDK 3.962 streaming trailers leave aws-chunked in SeaweedFS object metadata.
        # SigV4 remains active; the migration tool supplies an explicit SHA-256 checksum.
        target["services"]["backend"]["environment"]["AWS_REQUEST_CHECKSUM_CALCULATION"] = "WHEN_REQUIRED"
        u.write_json(self.home / "target.json", literal(target))
        # Fail before maintenance if Compose cannot parse the generated configurations.
        self.compose(self.home / "stage.json", "config", "--quiet")
        self.compose(self.home / "target.json", "config", "--quiet")

    def verify_original(self):
        require(self.fingerprint() == self.state["fingerprint"], "Compose/.env wurde nach der Planung verändert.")
        for name, c in self.state["containers"].items():
            actual = self.inspect(c["id"])
            require(actual["Image"] == c["image"], "Containeridentität verändert: " + name)
            if name in self.state.get("retainedServices", []):
                index_containers({"services": {}}, [actual])

    def freeze(self):
        self.save("freezing")
        self.guard()
        for name, c in self.state["containers"].items():
            if name != "db":
                self.docker("update", "--restart=no", c["id"])
        ids = [c["id"] for name, c in self.state["containers"].items() if name != "db"]
        if ids:
            self.docker("stop", "-t", "60", *ids)
        self.assert_frozen()

    def assert_frozen(self):
        require(self.inspect(self.state["containers"]["minio"]["id"])["HostConfig"]["RestartPolicy"]["Name"] == "no",
                "MinIO könnte automatisch wieder starten.")
        for name, c in self.state["containers"].items():
            if name in ("db", "minio"):
                continue
            actual = self.inspect(c["id"])
            require(not actual["State"]["Running"] and actual["HostConfig"]["RestartPolicy"]["Name"] == "no",
                    "Schreibpause wurde aufgehoben: " + name)
        known = {c["id"] for c in self.state["containers"].values()}
        peers = json.loads(self.docker("network", "inspect", self.state["network"]))[0].get("Containers", {})
        for cid in set(peers) - known:
            c = self.inspect(cid)
            labels = c["Config"].get("Labels", {})
            target = (labels.get("com.docker.compose.project") == self.state["project"] and
                      labels.get("com.docker.compose.service") == "objectstore" and c["Image"] == self.state["weedImage"])
            own_helper = labels.get("io.fedeo.storage-job") == str(self.home)
            require(target or own_helper, "Ein zusätzlicher Container hat sich mit dem Speichernetz verbunden.")

    def check_backup(self):
        backup = self.home / "backup"
        complete = u.read_json(backup / "complete.json", {})
        require(complete.get("sha256"), "Abgeschlossene Sicherung fehlt.")
        for name, expected in complete["sha256"].items():
            require(Path(name).name == name and sha_file(backup / name) == expected,
                    "Backup-Prüfsumme stimmt nicht: " + name)

    def backup(self):
        self.save("backing_up")
        backup = self.home / "backup"
        backup.mkdir(mode=0o700, exist_ok=True)
        require(not self.inspect(self.state["containers"]["minio"]["id"])["State"]["Running"], "MinIO für kalte Sicherung stoppen.")
        db = self.state["containers"]["db"]["id"]
        self.docker("exec", db, "sh", "-ec", 'exec pg_dump -U "$POSTGRES_USER" -d "$POSTGRES_DB" -Fc', output=backup / "database.dump")
        self.docker("exec", "-i", db, "pg_restore", "--list", input_file=backup / "database.dump", output=backup / "database.list")
        self.docker("exec", db, "sh", "-ec", 'exec pg_dumpall -U "$POSTGRES_USER" --globals-only', output=backup / "globals.sql")
        paths = ["minio", *self.state["matrixPaths"], "storage/config", ".env", "docker-compose.yml"]
        if (self.root / "traefik/letsencrypt").is_dir():
            paths.append("traefik/letsencrypt")
        self.runner.run(["tar", "-cpf", str(backup / "files.tar"), "-C", str(self.root), *paths], timeout=7200)
        self.docker("image", "save", "-o", str(backup / "images.tar"), *self.state["imageIds"], timeout=7200)
        u.write_json(backup / "complete.json", {"at": u.now(), "sha256": {p.name: sha_file(p) for p in backup.iterdir() if p.is_file() and p.name != "complete.json"}})
        self.save("backed_up")

    def migrate(self, writers_exclusive):
        require(writers_exclusive, "--writers-exclusive bestätigt: keine externen S3-/DB-Schreiber und kein paralleles manuelles Compose.")
        require(self.state.get("phase") in ("planned", "freezing", "backing_up", "backed_up", "copying", "verifying", "verified"),
                "Migration kann in dieser Phase nicht gestartet werden.")
        self.cleanup_helper()
        self.verify_original()
        require(shutil.disk_usage(self.root).free >= (self.state["requiredFreeBytes"] if self.state["phase"] == "planned" else GIB),
                "Nicht genug freier Speicher für Migration/Fortsetzung.")
        if self.state["phase"] in ("planned", "freezing", "backing_up"):
            (self.root / "storage/config").mkdir(parents=True, mode=0o700, exist_ok=True)
            (self.root / "storage/seaweedfs/master").mkdir(parents=True, mode=0o700, exist_ok=True)
            u.atomic(self.root / "storage/config/s3.json", (self.home / "s3.json").read_bytes())
            u.atomic(self.root / "storage/config/start-seaweedfs.sh", (HERE / "storage/start-seaweedfs.sh").read_bytes())
            self.freeze()
            self.docker("start", self.state["containers"]["minio"]["id"])
            self.wait_ready(self.state["containers"]["minio"]["id"])
            current = self.helper("inventory", source=True, directory="frozen-inventory")
            extra = max(0, current["bytes"] - self.state["bytes"]) * 4 + max(0, current["count"] - self.state["count"]) * 256 * 1024
            require(shutil.disk_usage(self.root).free >= self.state["requiredFreeBytes"] + extra,
                    "Bestand seit Planung gewachsen; Speicherreserve reicht nicht mehr.")
            self.save(count=current["count"], bytes=current["bytes"])
            self.docker("stop", "-t", "60", self.state["containers"]["minio"]["id"])
            validate_minio_listing(self.root, u.read_json(self.home / "credentials.json")["bucket"], current, exact=True)
            self.backup()
        self.assert_frozen()
        self.guard()
        self.check_backup()
        self.docker("start", self.state["containers"]["minio"]["id"])
        self.wait_ready(self.state["containers"]["minio"]["id"])
        self.compose(self.home / "stage.json", "up", "-d", "--pull", "never", "--wait", "--wait-timeout", "180")
        self.helper("init", admin=True, directory="init")
        self.helper("contract", directory="contract")
        self.helper("contract", directory="backend-contract", backend_sdk=True)
        self.save("copying")
        self.helper("copy", source=True)
        self.assert_frozen()
        self.save("verifying")
        self.helper("verify", source=True)
        self.assert_frozen()
        self.save("verified")

    def restore_test(self):
        # Start a separate container on newly extracted directories. Never count a
        # successful archive listing or restart of the original directory as restore.
        self.save("restore_testing")
        restore = self.home / "restore-test"
        if restore.exists():
            # A previous interrupted restore is retained; use a fresh directory.
            restore = self.home / ("restore-test-" + secrets.token_hex(4))
        restore.mkdir(mode=0o700)
        self.runner.run(["tar", "-xpf", str(self.home / "backup/seaweedfs.tar"), "-C", str(restore)], timeout=7200)
        known = self.docker("ps", "-aq", "--filter", "name=^/" + self.restore_name + "$").split()
        for cid in known:
            require(self.inspect(cid)["Config"].get("Labels", {}).get("io.fedeo.storage-job") == str(self.home), "Fremder Restore-Container.")
            self.docker("rm", "-f", cid)
        target = u.read_json(self.home / "stage.json")["services"]["objectstore"]
        try:
            self.docker("run", "-d", "--pull", "never", "--name", self.restore_name,
                        "--label", "io.fedeo.storage-job=" + str(self.home), "--network", self.state["network"],
                        "--user", str(os.getuid()) + ":" + str(os.getgid()),
                        "-v", str(restore / "storage/seaweedfs") + ":/data",
                        "-v", str(restore / "storage/config") + ":/etc/fedeo:ro",
                        "-e", "FEDEO_S3_CONFIG_SHA256=" + sha_file(self.home / "s3.json"),
                        "--entrypoint", target["entrypoint"][0], self.state["weedImage"],
                        *target["entrypoint"][1:], *target["command"])
            self.helper("restore-verify", endpoint="http://" + self.restore_name + ":9000")
            self.helper("contract", endpoint="http://" + self.restore_name + ":9000", directory="restore-contract")
            self.save("cutover", restoreVerifiedAt=u.now(), seaweedBackupSha256=sha_file(self.home / "backup/seaweedfs.tar"))
        finally:
            self.docker("rm", "-f", self.restore_name)

    def activate(self):
        require(self.state.get("phase") in ("verified", "cutover", "restore_testing"), "Erst Kopie und Prüfung abschließen.")
        require(not self.state.get("writesReleased"), "Schreibfreigabe ist bereits dauerhaft erfolgt.")
        self.cleanup_helper()
        self.assert_frozen()
        # A killed contract test can leave only its journalled probe keys behind.
        # Bring the target up without the legacy alias, remove those exact keys,
        # then take the cold backup. Never delete unknown target objects.
        self.compose(self.home / "stage.json", "up", "-d", "--pull", "never", "--wait", "--wait-timeout", "180")
        for directory in ("contract", "backend-contract", "acceptance", "backend-acceptance"):
            if (self.home / directory / "probe.json").is_file():
                self.helper("cleanup-probe", directory=directory)
        if self.state["phase"] == "verified":
            self.verify_original()
            self.helper("verify", source=True)
            self.save("cutover")
        self.guard()
        self.check_backup()
        old = self.state["containers"]["minio"]["id"]
        self.docker("stop", "-t", "60", old)
        validate_minio_listing(self.root, u.read_json(self.home / "credentials.json")["bucket"],
                               u.read_json(self.home / "copy/inventory.json"), exact=True)
        if self.state["network"] in self.inspect(old)["NetworkSettings"]["Networks"]:
            self.docker("network", "disconnect", self.state["network"], old)
        self.compose(self.home / "stage.json", "stop", "-t", "60", "objectstore")
        # A cold backup is required before serving any business write on the new provider.
        self.runner.run(["tar", "-cpf", str(self.home / "backup/seaweedfs.tar"), "-C", str(self.root), "storage"], timeout=7200)
        self.restore_test()
        u.atomic(self.root / "docker-compose.yml", (self.home / "target.json").read_bytes())
        self.compose(self.root / "docker-compose.yml", "up", "-d", "--no-deps", "--pull", "never", "--wait", "--wait-timeout", "180", "objectstore")
        self.helper("restore-verify", endpoint="http://minio:9000")
        self.helper("contract", endpoint="http://minio:9000", directory="acceptance")
        self.helper("contract", endpoint="http://minio:9000", directory="backend-acceptance", backend_sdk=True)
        # Persist before starting any backend/background process; no automatic rollback beyond here.
        self.save("released", writesReleased=True, releasedAt=u.now(), provider="seaweedfs")
        self.guard()
        self.start_application()

    def start_application(self):
        require(self.state.get("writesReleased"), "Schreibfreigabe fehlt.")
        require(sha_file(self.root / "docker-compose.yml") == sha_file(self.home / "target.json"), "Aktive Compose-Datei verändert.")
        names = [n for n, c in self.state["containers"].items() if c["running"] and n not in ("minio", "createbuckets", "traefik")]
        # Preserve Matrix without letting an already failing optional healthcheck
        # cause a new FEDEO outage after cutover. Core services must always pass.
        degraded = [n for n in names if n in u.LEGACY_MATRIX_SERVICES
                    and self.state["containers"][n].get("health") == "unhealthy"]
        self.compose(self.root / "docker-compose.yml", "up", "-d", "--pull", "never", *names)
        self.compose(self.root / "docker-compose.yml", "up", "-d", "--no-deps", "--pull", "never",
                     "--wait", "--wait-timeout", "240", *[n for n in names if n not in degraded])
        for name in degraded:
            cid = self.compose(self.root / "docker-compose.yml", "ps", "--all", "--quiet", name).strip()
            require(cid and "\n" not in cid and self.inspect(cid)["State"]["Running"],
                    "Alter Matrix-Dienst wurde nicht wieder gestartet: " + name)
        self.compose(self.root / "docker-compose.yml", "up", "-d", "--no-deps", "--pull", "never", "traefik")
        self.save("complete", preexistingUnhealthyServices=degraded)

    def rollback(self):
        require(self.state and not self.state.get("writesReleased"), "Nach Schreibfreigabe ist eine geprüfte Rückmigration erforderlich.")
        require(sha_file(self.root / ".env") == self.state["fingerprint"][".env"], ".env verändert; Rückkehr manuell prüfen.")
        require(sha_file(self.root / "docker-compose.yml") in {self.state["fingerprint"]["docker-compose.yml"], sha_file(self.home / "target.json")},
                "Compose-Datei außerhalb der Migration verändert; Rückkehr manuell prüfen.")
        self.cleanup_helper()
        for cid in self.docker("ps", "-aq", "--filter", "name=^/" + self.restore_name + "$").split():
            require(self.inspect(cid)["Config"].get("Labels", {}).get("io.fedeo.storage-job") == str(self.home), "Fremder Restore-Container.")
            self.docker("rm", "-f", cid)
        require(self.state["phase"] != "planning", "Unvollständiger Plan.")
        self.save("rolling_back")
        self.guard()
        # May not exist when the failure occurred before the first target start.
        ids = self.docker("ps", "-aq", "--filter", "label=com.docker.compose.project=" + self.state["project"],
                          "--filter", "label=com.docker.compose.service=objectstore").split()
        for cid in ids:
            self.docker("update", "--restart=no", cid)
            self.docker("stop", "-t", "60", cid)
            if self.state["network"] in self.inspect(cid)["NetworkSettings"]["Networks"]:
                self.docker("network", "disconnect", self.state["network"], cid)
        u.atomic(self.root / "docker-compose.yml", (self.home / "originaldocker-compose.yml").read_bytes())
        for name, c in self.state["containers"].items():
            actual = self.inspect(c["id"])
            if name == "minio" and self.state["network"] not in actual["NetworkSettings"]["Networks"]:
                aliases = c["networks"][self.state["network"]].get("Aliases") or ["minio"]
                args = [arg for alias in aliases for arg in ("--alias", alias)]
                self.docker("network", "connect", *args, self.state["network"], c["id"])
            policy = c["restart"]["Name"]
            if policy == "on-failure" and c["restart"].get("MaximumRetryCount"):
                policy += ":" + str(c["restart"]["MaximumRetryCount"])
            self.docker("update", "--restart=" + policy, c["id"])
        for name in ["db", "minio", *[n for n in self.state["containers"] if n not in ("db", "minio", "traefik")], "traefik"]:
            c = self.state["containers"].get(name)
            if c and c["running"]:
                self.docker("start", c["id"])
                self.wait_ready(c["id"])
        self.save("rolled_back")
        u.write_json(self.engine.home / "status.json", u.read_json(self.home / "previous-update.json"))

    def public_status(self):
        return {k: self.state[k] for k in ("phase", "version", "count", "bytes", "requiredFreeBytes", "freeBytes",
                                          "writesReleased", "releasedAt", "updatedAt") if k in self.state}


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("command", choices=["plan", "migrate", "activate", "resume-start", "rollback", "status"])
    parser.add_argument("--directory", default="/opt/fedeo")
    parser.add_argument("--tools-image")
    parser.add_argument("--writers-exclusive", action="store_true")
    args = parser.parse_args()
    migration = Migration(args.directory, args.tools_image)
    try:
        if args.command == "status":
            print(json.dumps(migration.public_status(), indent=2)); return
        with migration.engine.lock():
            if args.command == "plan": migration.plan()
            elif args.command == "migrate": migration.migrate(args.writers_exclusive)
            elif args.command == "activate": migration.activate()
            elif args.command == "resume-start": migration.start_application()
            elif args.command == "rollback": migration.rollback()
            print(json.dumps(migration.public_status(), indent=2))
    except (Error, OSError, ValueError, KeyboardInterrupt) as error:
        print("Migration angehalten: " + str(error) + "\nGeschütztes Protokoll: " + str(migration.home / "commands.log"), file=sys.stderr)
        return 1


if __name__ == "__main__":
    sys.exit(main())