"""Versioned deployment archives and conservative managed-file ownership (stdlib only)."""
import hashlib
import json
import os
from pathlib import Path, PurePosixPath
import re
import stat
import zipfile

MANIFEST = "bundle-manifest.json"
STATE = ".bundle-state"
FORMAT = 1
MAX_FILE = 64 * 1024 * 1024
MAX_TOTAL = 256 * 1024 * 1024
APP_SERVICES = ("web", "worker-default", "worker-high", "worker-log", "scheduler")
INFRA_SERVICES = ("postgres", "redis", "minio", "minio-init", "alpha-processing")
OPTIONAL_INFRA_SERVICES = ("litellm",)
DATA_COMMANDS = {"seed_pi_continuum_menu"}


class BundleError(RuntimeError):
    pass


def digest(data):
    return hashlib.sha256(data).hexdigest()


def canonical(value):
    return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode()


def read_json(path):
    return json.loads(Path(path).read_bytes(), object_pairs_hook=unique_object)


def unique_object(pairs):
    result = {}
    for key, value in pairs:
        if key in result:
            raise BundleError("duplicate JSON key: " + key)
        result[key] = value
    return result


def write_json(path, value):
    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
    temporary = path.with_name(path.name + ".tmp")
    with open(temporary, "wb") as stream:
        os.chmod(temporary, 0o600)
        stream.write(canonical(value) + b"\n")
        stream.flush()
        os.fsync(stream.fileno())
    os.replace(temporary, path)
    fd = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY)
    try:
        os.fsync(fd)
    finally:
        os.close(fd)


def managed(name):
    p = PurePosixPath(name)
    if (not name or p.is_absolute() or str(p) != name or ".." in p.parts
            or "\\" in name or any(ord(c) < 32 for c in name)):
        return False
    if name in {"compose.yml", "README.md", ".env.example", "release-policy.json"}:
        return True
    if p.parts[0] == "bin" and len(p.parts) > 1 and "__pycache__" not in p.parts:
        return not name.endswith((".pyc", ".pyo"))
    if p.parts[:2] in (("conf", "nginx"), ("conf", "postgres"), ("conf", "litellm")):
        return "certs" not in p.parts and len(p.parts) > 2
    return False


def checked_path(root, name):
    if not managed(name):
        raise BundleError("not a managed path: " + name)
    p = Path(root)
    for part in PurePosixPath(name).parts:
        p = p / part
        if p.is_symlink():
            raise BundleError("symlink in installation path: " + name)
    return p


def file_record(path):
    s = path.lstat()
    if not stat.S_ISREG(s.st_mode):
        raise BundleError("managed path is not a regular file: " + str(path))
    return {"sha256": digest(path.read_bytes()), "mode": stat.S_IMODE(s.st_mode), "size": s.st_size}


def validate_manifest(m):
    try:
        _validate_manifest(m)
    except (AttributeError, KeyError, TypeError) as exc:
        raise BundleError("malformed release manifest") from exc


def _validate_manifest(m):
    if m.get("format") != FORMAT or m.get("product") != "pi-continuum":
        raise BundleError("unsupported bundle product/format")
    if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]{0,127}", m.get("release", "")):
        raise BundleError("invalid release identifier")
    if not re.fullmatch(r"[0-9a-f]{40}", m.get("commit", "")):
        raise BundleError("manifest needs a full commit identifier")
    if not isinstance(m.get("ancestors"), list) or any(not re.fullmatch(r"[0-9a-f]{40}", x) for x in m["ancestors"]):
        raise BundleError("invalid compatible ancestor commits")
    if m.get("contract") != "pi-continuum-update-v1":
        raise BundleError("unsupported update contract")
    files = m.get("files")
    if not isinstance(files, dict) or not {"compose.yml", ".env.example"} <= files.keys():
        raise BundleError("manifest lacks full deployment file inventory")
    for name, entry in files.items():
        if not managed(name) or entry.get("mode") not in (0o644, 0o755):
            raise BundleError("invalid managed file path/permission: " + name)
        if not re.fullmatch(r"[0-9a-f]{64}", entry.get("sha256", "")):
            raise BundleError("invalid file digest: " + name)
        if not isinstance(entry.get("size"), int) or not 0 <= entry["size"] <= MAX_FILE:
            raise BundleError("invalid file size: " + name)
    app = m.get("application", {})
    if app.get("action") not in ("replace", "keep"):
        raise BundleError("release must declare application action")
    if not re.fullmatch(r"sha256:[0-9a-f]{64}", app.get("id", "")):
        raise BundleError("release must pin the app image ID")
    if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._/:@-]*", app.get("ref", "")):
        raise BundleError("invalid app image reference")
    if app["action"] == "keep" and not app.get("compatible_ids"):
        raise BundleError("file-only release must declare compatible app image IDs")
    if not isinstance(app.get("compatible_ids", []), list) or any(
            not re.fullmatch(r"sha256:[0-9a-f]{64}", value) for value in app.get("compatible_ids", [])):
        raise BundleError("invalid compatible app image IDs")
    required_infra = set(INFRA_SERVICES) | {"nginx"}
    if not required_infra <= set(m.get("infrastructure", {})) <= required_infra | set(OPTIONAL_INFRA_SERVICES):
        raise BundleError("manifest must declare all infrastructure image compatibility requirements")
    db = m.get("database", {})
    if not db.get("migrations") or not db.get("tables"):
        raise BundleError("release must describe the required DB schema")
    if not isinstance(db["migrations"], list) or any(
            not isinstance(pair, list) or len(pair) != 2 or any(not isinstance(v, str) or not v for v in pair)
            for pair in db["migrations"]):
        raise BundleError("invalid migration history contract")
    if not isinstance(db["tables"], dict) or any(not isinstance(cols, dict) or not cols for cols in db["tables"].values()):
        raise BundleError("invalid table contract")
    if not isinstance(m.get("data_updates"), list):
        raise BundleError("release must explicitly list data updates")
    ids = set()
    for task in m.get("data_updates", []):
        if task.get("command") not in DATA_COMMANDS or task.get("id") in ids or not task.get("id"):
            raise BundleError("unsupported or duplicate data update")
        if not isinstance(task.get("backward_compatible"), bool):
            raise BundleError("data update must declare its rollback boundary")
        ids.add(task["id"])
    if not isinstance(m.get("required_env"), list) or any(not re.fullmatch(r"[A-Z][A-Z0-9_]*", x) for x in m["required_env"]):
        raise BundleError("invalid required environment names")


def extract_bundle(archive, destination):
    """Never extractall(): reject aliases, links, duplicates, bombs and unlisted entries."""
    destination = Path(destination)
    with zipfile.ZipFile(archive) as z:
        infos = z.infolist()
        names = [i.filename for i in infos]
        if len(names) != len(set(names)) or MANIFEST not in names or len(names) > 10000:
            raise BundleError("duplicate archive entries or missing manifest")
        if sum(i.file_size for i in infos) > MAX_TOTAL or any(i.file_size > MAX_FILE for i in infos):
            raise BundleError("archive exceeds deployment size limits")
        for info in infos:
            mode = info.external_attr >> 16
            if info.flag_bits & 1 or not stat.S_ISREG(mode) or stat.S_IMODE(mode) not in (0o644, 0o755):
                raise BundleError("invalid ZIP file type/permission: " + info.filename)
            if info.filename != MANIFEST and not managed(info.filename):
                raise BundleError("protected or unsafe ZIP path: " + info.filename)
        m = json.loads(z.read(MANIFEST), object_pairs_hook=unique_object)
        validate_manifest(m)
        if set(names) != set(m["files"]) | {MANIFEST}:
            raise BundleError("ZIP entries differ from manifest")
        for name, entry in m["files"].items():
            info = z.getinfo(name)
            data = z.read(name)
            if (digest(data) != entry["sha256"] or len(data) != entry["size"]
                    or stat.S_IMODE(info.external_attr >> 16) != entry["mode"]):
                raise BundleError("ZIP integrity/permission mismatch: " + name)
            dest = checked_path(destination, name)
            dest.parent.mkdir(parents=True, exist_ok=True)
            dest.write_bytes(data)
            dest.chmod(entry["mode"])
        (destination / MANIFEST).write_bytes(canonical(m) + b"\n")
    return m


def plan_files(root, previous, target):
    before, after = previous["files"], target["files"]
    conflicts, add, change, delete = [], [], [], []
    for name, record in before.items():
        path = checked_path(root, name)
        if not path.exists() or file_record(path) != record:
            conflicts.append(name)
    for name, record in after.items():
        path = checked_path(root, name)
        if name not in before:
            if path.exists():
                conflicts.append(name)  # even identical unknown files need explicit ownership
            else:
                add.append(name)
        elif record != before[name]:
            change.append(name)
    delete = sorted(set(before) - set(after))
    if conflicts:
        raise BundleError("local managed-file conflict; no changes applied: " + ", ".join(sorted(set(conflicts))))
    return {"add": sorted(add), "change": sorted(change), "delete": delete}


def check_database(expected, actual):
    def expand(items, seen=()):
        result = set()
        for item in items:
            key = "/".join(item)
            if key in seen:
                raise BundleError("cyclic migration replacement contract")
            replacement = expected.get("replacements", {}).get(key)
            result.update(expand(replacement, seen + (key,)) if replacement else [tuple(item)])
        return result
    if expand(expected["migrations"]) != expand(actual.get("migrations") or []):
        raise BundleError("DB migration history incompatible; operator schema procedure required")
    for table, columns in expected["tables"].items():
        if actual.get("tables", {}).get(table) != columns:
            raise BundleError("DB table schema incompatible or unknown: " + table)


def make_zip(source, output, metadata):
    m = dict(metadata)
    m.update(format=FORMAT, product="pi-continuum", contract="pi-continuum-update-v1")
    m["files"] = {}
    for path in sorted(Path(source).rglob("*")):
        name = path.relative_to(source).as_posix()
        if path.is_symlink() and (managed(name) or name in ("bin", "conf", "conf/nginx", "conf/postgres", "conf/litellm")):
            raise BundleError("symlink in bundle source: " + name)
        if managed(name) and not path.is_dir():
            m["files"][name] = file_record(path)
    validate_manifest(m)
    with zipfile.ZipFile(output, "w", compression=zipfile.ZIP_DEFLATED) as z:
        for name, rec in m["files"].items():
            info = zipfile.ZipInfo(name)
            info.create_system = 3
            info.external_attr = (stat.S_IFREG | rec["mode"]) << 16
            info.compress_type = zipfile.ZIP_DEFLATED
            z.writestr(info, (Path(source) / name).read_bytes())
        info = zipfile.ZipInfo(MANIFEST)
        info.create_system = 3
        info.external_attr = (stat.S_IFREG | 0o644) << 16
        z.writestr(info, canonical(m) + b"\n")
    return m