#!/usr/bin/env python3
"""Build ERMD 2.0 per-file manifests and catalog data.

ERMD 1.x treated one ZIP as one release. ERMD 2.0 serves each pack as an
isolated folder under ``packs/<pack-id>/files`` and writes a manifest containing
SHA-256 and byte size for every managed file. The client can therefore download
only missing/changed files and prune only files a prior manifest owned.

This script is intentionally server-side only. It never executes imported batch
files, shortcuts, EXEs, or PowerShell. Windows-only preparation (for example
Gideon UI generation) happens on the publisher PC before the prepared pack is
uploaded to this database.
"""
from __future__ import annotations

import argparse
import fnmatch
import hashlib
import json
import os
import posixpath
import re
import shutil
import sys
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Iterable
from urllib.parse import quote

SCHEMA_VERSION = 2
DEFAULT_DB_ROOT = "/mnt/user/appdata/webserver-php8.4/html/downloads/ERMD2.0DB"
DEFAULT_PUBLIC_BASE_URL = "https://files.vmvproductions.net/downloads/ERMD2.0DB/"

DEFAULT_IGNORE_BASENAMES = {
    "desktop.ini",
    "thumbs.db",
    ".ds_store",
}
DEFAULT_IGNORE_SUFFIXES = {".log", ".tmp", ".part", ".erdmd-part"}


def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat(timespec="seconds")


def atomic_write_json(path: Path, payload: Any) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    temporary = path.with_name(path.name + ".tmp")
    temporary.write_text(json.dumps(payload, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
    os.replace(temporary, path)


def load_json(path: Path, fallback: Any) -> Any:
    try:
        value = json.loads(path.read_text(encoding="utf-8"))
        return value
    except Exception:
        return fallback


def sha256_file(path: Path, chunk_size: int = 1024 * 1024) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        while True:
            chunk = handle.read(chunk_size)
            if not chunk:
                break
            digest.update(chunk)
    return digest.hexdigest().lower()


def sha256_bytes(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest().lower()


# ---------------------------------------------------------------------------
# Hard-delete compatibility blocklist
# ---------------------------------------------------------------------------
# No blocked payload is quarantined or copied. A filename hit seeds the
# SHA-256 registry, so a later renamed PE copy is deleted by hash.
BLOCKLIST_SCHEMA = 1
DEFAULT_BLOCKED_ARTIFACTS: tuple[dict[str, Any], ...] = (
    {
        "id": "center-camera",
        "reason": "Known incompatible legacy Center Camera DLL",
        "filenames": ["center_camera.dll"],
        "sha256": [],
        "action": "delete",
    },
)


def _normalise_block_hash(value: Any) -> str:
    text = str(value or "").strip().casefold()
    return text if len(text) == 64 and all(char in "0123456789abcdef" for char in text) else ""


def _normalise_block_rule(raw: Any) -> dict[str, Any] | None:
    if not isinstance(raw, dict):
        return None
    rule_id = str(raw.get("id") or "").strip().casefold()
    if not rule_id:
        return None
    names: list[str] = []
    seen_names: set[str] = set()
    for value in raw.get("filenames") if isinstance(raw.get("filenames"), list) else []:
        name = Path(str(value or "")).name.casefold().strip()
        if name and name not in seen_names:
            seen_names.add(name)
            names.append(name)
    hashes: list[str] = []
    seen_hashes: set[str] = set()
    for value in raw.get("sha256") if isinstance(raw.get("sha256"), list) else []:
        digest = _normalise_block_hash(value)
        if digest and digest not in seen_hashes:
            seen_hashes.add(digest)
            hashes.append(digest)
    return {
        "id": rule_id,
        "reason": str(raw.get("reason") or "Blocked incompatible artifact").strip(),
        "filenames": names,
        "sha256": hashes,
        "action": "delete",
    }


def blocklist_path(db_root: Path) -> Path:
    return db_root / "staging" / "blocked_artifacts.json"


def load_blocklist(db_root: Path) -> dict[str, Any]:
    path = blocklist_path(db_root)
    raw = load_json(path, {})
    rules: dict[str, dict[str, Any]] = {}
    values = raw.get("blocked_artifacts") if isinstance(raw, dict) else []
    if isinstance(values, list):
        for item in values:
            rule = _normalise_block_rule(item)
            if rule:
                rules[rule["id"]] = rule
    for default_raw in DEFAULT_BLOCKED_ARTIFACTS:
        default = _normalise_block_rule(default_raw)
        assert default is not None
        current = rules.get(default["id"])
        if current is None:
            rules[default["id"]] = default
            continue
        current["filenames"] = list(dict.fromkeys([*current["filenames"], *default["filenames"]]))
    payload = {
        "schema_version": BLOCKLIST_SCHEMA,
        "blocked_artifacts": sorted(rules.values(), key=lambda item: item["id"]),
    }
    if not path.exists() or raw != payload:
        atomic_write_json(path, payload)
    return payload


def _is_pe_file(path: Path) -> bool:
    try:
        with path.open("rb") as handle:
            return handle.read(2) == b"MZ"
    except OSError:
        return False


def _remove_empty_parents(path: Path, root: Path) -> None:
    current = path.parent
    while current != root and current.exists():
        try:
            current.rmdir()
        except OSError:
            return
        current = current.parent


def scrub_blocked_artifacts(db_root: Path, root: Path) -> list[dict[str, str]]:
    """Permanently delete blocked files below a runtime or pack files root."""
    if not root.is_dir():
        return []
    registry = load_blocklist(db_root)
    rules = [_normalise_block_rule(item) for item in registry.get("blocked_artifacts") or []]
    rules = [item for item in rules if item]
    registry_rules_by_id = {
        str(item.get("id") or "").casefold(): item
        for item in registry.get("blocked_artifacts") or []
        if isinstance(item, dict)
    }
    filename_index: dict[str, list[dict[str, Any]]] = {}
    hash_index: dict[str, list[dict[str, Any]]] = {}
    for rule in rules:
        for filename in rule["filenames"]:
            filename_index.setdefault(filename, []).append(rule)
        for digest in rule["sha256"]:
            hash_index.setdefault(digest, []).append(rule)

    registry_changed = False
    removed: list[dict[str, str]] = []
    for candidate in sorted((item for item in root.rglob("*") if item.is_file() and not item.is_symlink()), key=lambda item: item.as_posix().casefold()):
        name_matches = filename_index.get(candidate.name.casefold(), [])
        should_hash = bool(name_matches) or _is_pe_file(candidate)
        digest = sha256_file(candidate) if should_hash else ""
        hash_matches = hash_index.get(digest, []) if digest else []
        matches: list[dict[str, Any]] = []
        seen: set[str] = set()
        for rule in [*name_matches, *hash_matches]:
            if rule["id"] not in seen:
                seen.add(rule["id"])
                matches.append(rule)
        if not matches:
            continue
        if name_matches and digest:
            for rule in name_matches:
                hashes = rule.setdefault("sha256", [])
                persisted = registry_rules_by_id.get(rule["id"])
                persisted_hashes = persisted.setdefault("sha256", []) if isinstance(persisted, dict) else hashes
                if digest not in hashes:
                    hashes.append(digest)
                    hash_index.setdefault(digest, []).append(rule)
                if digest not in persisted_hashes:
                    persisted_hashes.append(digest)
                    registry_changed = True
        relative = candidate.relative_to(root).as_posix()
        candidate.unlink()
        _remove_empty_parents(candidate, root)
        rule_ids = ", ".join(rule["id"] for rule in matches)
        removed.append({"path": relative, "sha256": digest, "match": "filename" if name_matches else "sha256", "rules": rule_ids})
    if registry_changed:
        atomic_write_json(blocklist_path(db_root), registry)
    return removed


def canonical_hash(value: Any) -> str:
    encoded = json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")
    return sha256_bytes(encoded)


def normalise_relative(value: str) -> str:
    result = str(value or "").replace("\\", "/").strip().strip("/")
    if not result or result.startswith("/") or ".." in Path(result).parts:
        raise ValueError(f"Unsafe relative path: {value!r}")
    return result


def leaf_folder_name(value: str) -> str:
    result = str(value or "").strip()
    if not result or Path(result).name != result or result in {".", ".."}:
        raise ValueError(f"Install folder must be one safe folder name: {value!r}")
    return result


def public_url(base: str, relative: str) -> str:
    cleaned = normalise_relative(relative)
    return base.rstrip("/") + "/" + quote(cleaned, safe="/")


class HashCache:
    def __init__(self, path: Path):
        self.path = path
        raw = load_json(path, {})
        self.data: dict[str, Any] = raw if isinstance(raw, dict) else {}
        self.changed = False

    def digest(self, root: Path, path: Path) -> str:
        relative = path.resolve().relative_to(root.resolve()).as_posix()
        stat = path.stat()
        signature = {"size": int(stat.st_size), "mtime_ns": int(stat.st_mtime_ns)}
        item = self.data.get(relative)
        if isinstance(item, dict) and item.get("signature") == signature and isinstance(item.get("sha256"), str):
            return str(item["sha256"]).lower()
        value = sha256_file(path)
        self.data[relative] = {"signature": signature, "sha256": value}
        self.changed = True
        return value

    def save(self) -> None:
        if self.changed:
            atomic_write_json(self.path, self.data)


def _is_ignored(relative: str, extra_patterns: list[str]) -> bool:
    rel = relative.replace("\\", "/").strip("/")
    parts = [part.casefold() for part in Path(rel).parts]
    base = Path(rel).name.casefold()
    if base in DEFAULT_IGNORE_BASENAMES:
        return True
    if base.endswith("_log.txt") or any(base.endswith(suffix) for suffix in DEFAULT_IGNORE_SUFFIXES):
        return True
    if "logs" in parts or ".ermd" in parts:
        return True
    for pattern in extra_patterns:
        if fnmatch.fnmatchcase(rel.casefold(), str(pattern).replace("\\", "/").casefold()):
            return True
    return False


def iter_payload_files(files_root: Path, extra_patterns: list[str], excluded_paths: set[str]) -> Iterable[Path]:
    if not files_root.is_dir():
        return
    for current, dirnames, filenames in os.walk(files_root, followlinks=False):
        current_path = Path(current)
        kept_dirs = []
        for name in dirnames:
            child = current_path / name
            if child.is_symlink():
                raise RuntimeError(f"Symlinked directory is not permitted in pack payload: {child}")
            kept_dirs.append(name)
        dirnames[:] = kept_dirs
        for name in filenames:
            path = current_path / name
            if path.is_symlink():
                raise RuntimeError(f"Symlinked file is not permitted in pack payload: {path}")
            if not path.is_file():
                continue
            relative = path.relative_to(files_root).as_posix()
            if relative in excluded_paths or _is_ignored(relative, extra_patterns):
                continue
            yield path


def find_assignment_list(text: str, assignment: str) -> str | None:
    match = re.search(rf"(?m)^[ \t]*{re.escape(assignment)}[ \t]*=[ \t]*\[", text)
    if not match:
        return None
    start = text.find("[", match.start(), match.end())
    depth = 0
    quote_char = ""
    escaped = False
    for index in range(start, len(text)):
        char = text[index]
        if quote_char:
            if escaped:
                escaped = False
            elif char == "\\":
                escaped = True
            elif char == quote_char:
                quote_char = ""
            continue
        if char in {"'", '"'}:
            quote_char = char
        elif char == "[":
            depth += 1
        elif char == "]":
            depth -= 1
            if depth == 0:
                return text[start + 1 : index]
    return None


def parse_toml_string_array(text: str, assignment: str) -> list[str]:
    body = find_assignment_list(text, assignment)
    if body is None:
        return []
    # Mod Engine's documented examples use quoted strings. Preserve only those
    # because malformed/unquoted entries should not become silently executable.
    out: list[str] = []
    for match in re.finditer(r'(["\'])(.*?)(?<!\\)\1', body, re.DOTALL):
        value = match.group(2).replace("\\\"", '"').replace("\\\\", "\\")
        try:
            out.append(normalise_relative(value))
        except ValueError:
            continue
    return out


def parse_modengine_mod_paths(text: str) -> list[dict[str, str]]:
    body = find_assignment_list(text, "mods")
    if body is None:
        return []
    result: list[dict[str, str]] = []
    for obj in re.findall(r"\{([^{}]*)\}", body, flags=re.DOTALL):
        name_match = re.search(r'name\s*=\s*"([^"]*)"', obj)
        path_match = re.search(r'path\s*=\s*"([^"]*)"', obj)
        if not path_match:
            continue
        try:
            path = normalise_relative(path_match.group(1))
        except ValueError:
            continue
        name = name_match.group(1).strip() if name_match else Path(path).name
        result.append({"name": name or Path(path).name or "mod", "path": path})
    return result


def dedupe_paths(values: Iterable[str]) -> list[str]:
    seen: set[str] = set()
    output: list[str] = []
    for value in values:
        try:
            path = normalise_relative(value)
        except ValueError:
            continue
        key = path.casefold()
        if key not in seen:
            seen.add(key)
            output.append(path)
    return output


def dedupe_mod_paths(values: Iterable[dict[str, Any]]) -> list[dict[str, str]]:
    seen: set[str] = set()
    output: list[dict[str, str]] = []
    for raw in values:
        if not isinstance(raw, dict):
            continue
        try:
            path = normalise_relative(str(raw.get("path") or ""))
        except ValueError:
            continue
        key = path.casefold()
        if key in seen:
            continue
        seen.add(key)
        name = str(raw.get("name") or Path(path).name or "mod").strip()
        output.append({"name": name, "path": path})
    return output


def default_meta(pack_id: str) -> dict[str, Any]:
    return {
        "name": pack_id.replace("-", " ").title(),
        "version": "0.0.0",
        "description": "",
        "image": "",
        "nexus_url": "",
        "source": {},
        "tags": [],
        "install_folder": "",
        "runtime": {"engine": ""},
        "modengine2": {},
        # ERMD packs always carry a pack-specific Seamless Co-op dependency.
        "seamlesscoop": {"enabled": True},
        "ignore": [],
    }


def _config_parent_relative(value: str) -> str:
    cleaned = normalise_relative(value)
    parent = Path(cleaned).parent.as_posix()
    return "" if parent in {"", "."} else parent


def enforce_universal_seamless_meta(meta: dict[str, Any]) -> None:
    """Derive the only valid Seamless runtime target for a pack layout.

    Standard ERMD ME2 packs use config_eldenring.toml at pack root, while
    prepared packs retain their author-owned config hierarchy. In both cases
    Seamless lives beside the active config so its DLL path is always simply
    ``SeamlessCoop/ersc.dll`` from that config.
    """
    runtime = meta.get("runtime") if isinstance(meta.get("runtime"), dict) else {}
    me2 = meta.get("modengine2") if isinstance(meta.get("modengine2"), dict) else {}
    engine = str(runtime.get("engine") or "").casefold()
    config_path = str(me2.get("config_path") or "").replace("\\", "/").strip("/")
    launch_path = str(runtime.get("launch_path") or "").replace("\\", "/").strip("/")
    seamless = meta.setdefault("seamlesscoop", {})
    if not isinstance(seamless, dict):
        seamless = {}
        meta["seamlesscoop"] = seamless
    # ME3 profiles own native DLL locations.  Publisher Stage 8 resolves the
    # authored ersc.dll native reference and stores it as target_subdir.  Never
    # replace it with a sibling folder beside the profile: Convergence and ERTE
    # intentionally use different profile-relative layouts.
    existing_target = str(seamless.get("target_subdir") or "").replace("\\", "/").strip("/")
    try:
        if engine == "modengine3" and existing_target:
            target = normalise_relative(existing_target)
        else:
            anchor = _config_parent_relative(config_path) if (engine == "modengine2" and config_path) else _config_parent_relative(launch_path) if launch_path else ""
            target = "/".join(item for item in [anchor, "SeamlessCoop"] if item) or "SeamlessCoop"
    except ValueError:
        target = "SeamlessCoop"
    seamless.update({
        "enabled": True,
        "runtime_id": "seamlesscoop",
        "target_subdir": target,
        "settings_path": "ersc_settings.ini",
        "template_path": "templates/ersc_settings.ini",
        "cooppassword": str(seamless.get("cooppassword") or ""),
        "save_file_extension": str(seamless.get("save_file_extension") or ""),
    })


def _path_relative_to_config(pack_path: str, config_path: str) -> str:
    target = normalise_relative(pack_path)
    base = _config_parent_relative(config_path) if config_path else ""
    return posixpath.relpath(target, start=base or ".").lstrip("./") if base else target


def load_meta(pack_dir: Path) -> dict[str, Any]:
    value = load_json(pack_dir / "pack.meta.json", {})
    if not isinstance(value, dict):
        value = {}
    merged = default_meta(pack_dir.name)
    for key, item in value.items():
        merged[key] = item
    for key in ("source", "runtime", "modengine2", "seamlesscoop"):
        if not isinstance(merged.get(key), dict):
            merged[key] = {}
    if not isinstance(merged.get("tags"), list):
        merged["tags"] = []
    if not isinstance(merged.get("ignore"), list):
        merged["ignore"] = []
    enforce_universal_seamless_meta(merged)
    return merged


def ensure_template(
    *,
    pack_dir: Path,
    files_root: Path,
    template_relative: str,
    source_relative: str,
    refresh_templates: bool,
    fallback_source: Path | None = None,
) -> Path | None:
    """Return a preserved per-pack config template.

    Most finished packs carry their own source config under ``files/``.  Raw
    imports that opt into a shared runtime may not, so ``fallback_source`` lets
    the runtime's pristine config seed the per-pack template without turning that
    generated file into a globally enforced runtime payload.
    """
    template = pack_dir / Path(normalise_relative(template_relative))
    source = files_root / Path(normalise_relative(source_relative))
    if refresh_templates or not template.exists():
        chosen = source if source.exists() and source.is_file() else fallback_source
        if chosen and chosen.exists() and chosen.is_file():
            template.parent.mkdir(parents=True, exist_ok=True)
            shutil.copy2(chosen, template)
    return template if template.exists() and template.is_file() else None



def relative_url_for_pack(pack_id: str, path: str) -> str:
    return f"packs/{normalise_relative(pack_id)}/{normalise_relative(path)}"


def build_generated_files(
    *,
    db_root: Path,
    base_url: str,
    pack_id: str,
    pack_dir: Path,
    files_root: Path,
    meta: dict[str, Any],
    cache: HashCache,
    refresh_templates: bool,
) -> tuple[list[dict[str, Any]], set[str]]:
    generated: list[dict[str, Any]] = []
    excluded: set[str] = set()
    me2 = meta.get("modengine2") if isinstance(meta.get("modengine2"), dict) else {}
    runtime = meta.get("runtime") if isinstance(meta.get("runtime"), dict) else {}
    engine = str(runtime.get("engine") or "").casefold()
    config_relative = str(me2.get("config_path") or "config_eldenring.toml")
    config_source = files_root / Path(config_relative)
    if engine == "modengine2" or config_source.exists():
        template_relative = str(me2.get("template_path") or "templates/config_eldenring.toml")
        template = ensure_template(
            pack_dir=pack_dir,
            files_root=files_root,
            template_relative=template_relative,
            source_relative=config_relative,
            refresh_templates=refresh_templates,
        )
        if not template:
            raise RuntimeError(f"{pack_id}: Mod Engine 2 config template is missing ({template_relative}).")
        template_text = template.read_text(encoding="utf-8", errors="strict")
        explicit_dlls = [str(item) for item in me2.get("external_dlls", []) if isinstance(item, (str, Path))]
        template_dlls = parse_toml_string_array(template_text, "external_dlls")
        auto_root = str(me2.get("auto_dll_root") or "DLL Files")
        auto_dlls: list[str] = []
        root = files_root / auto_root
        if root.is_dir():
            for dll in sorted(root.rglob("*.dll"), key=lambda item: item.as_posix().casefold()):
                if dll.is_file() and not dll.is_symlink():
                    pack_relative = dll.relative_to(files_root).as_posix()
                    auto_dlls.append(_path_relative_to_config(pack_relative, config_relative))
        # Seamless is a universal ERMD runtime. Its folder is derived beside the
        # active config, making the generated ME2 reference stable for raw and
        # prepared pack layouts alike.
        seamless = meta.get("seamlesscoop") if isinstance(meta.get("seamlesscoop"), dict) else {}
        seamless_target = normalise_relative(str(seamless.get("target_subdir") or "SeamlessCoop"))
        seamless_dll = _path_relative_to_config(f"{seamless_target}/ersc.dll", config_relative)
        template_mods = parse_modengine_mod_paths(template_text)
        explicit_mods = me2.get("asset_paths") if isinstance(me2.get("asset_paths"), list) else []
        generated.append({
            "kind": "modengine2_config",
            "path": normalise_relative(config_relative),
            "template_url": public_url(base_url, relative_url_for_pack(pack_id, template_relative)),
            "template_sha256": cache.digest(db_root, template),
            "external_dlls": dedupe_paths(template_dlls + explicit_dlls + auto_dlls + [seamless_dll]),
            "mods": dedupe_mod_paths(template_mods + explicit_mods),
        })
        excluded.add(normalise_relative(config_relative))

    # Seamless Co-op settings are deliberately *not* generated as a normal pack
    # file.  The shared runtime owns ersc.dll/locales while each pack supplies
    # its own password/save suffix.  build_pack_manifest attaches those values
    # to the seamless runtime dependency, and the reader writes the INI only
    # after it has materialised the runtime target folder.
    seamless = meta.get("seamlesscoop") if isinstance(meta.get("seamlesscoop"), dict) else {}
    if bool(seamless.get("enabled")):
        target_subdir = str(seamless.get("target_subdir") or "SeamlessCoop").strip().strip("/")
        settings_name = normalise_relative(str(seamless.get("settings_path") or "ersc_settings.ini"))
        settings_relative = normalise_relative(f"{target_subdir}/{settings_name}" if target_subdir else settings_name)
        # A pack may already ship a settings file.  It is always replaced by the
        # runtime-generated copy, never hash-enforced as source payload.
        excluded.add(settings_relative)
    return generated, excluded


def build_runtime_manifest(
    db_root: Path,
    base_url: str,
    runtime_dir: Path,
    cache: HashCache,
    runtime_id_override: str | None = None,
) -> dict[str, Any] | None:
    files_root = runtime_dir / "files"
    if not files_root.is_dir():
        return None
    runtime_id = normalise_relative(runtime_id_override or runtime_dir.name)
    files: list[dict[str, Any]] = []
    for path in iter_payload_files(files_root, [], set()):
        relative = path.relative_to(files_root).as_posix()
        # Pack-specific configuration is always generated from a pack template,
        # never treated as globally hash-enforced runtime payload. Seamless uses
        # ersc_settings.ini; Mod Engine 2 uses config_eldenring.toml.
        lower_relative = relative.casefold()
        if lower_relative.endswith("ersc_settings.ini"):
            continue
        if runtime_id.casefold() == "modengine2" and lower_relative == "config_eldenring.toml":
            continue
        stat = path.stat()
        files.append({"path": relative, "size": int(stat.st_size), "sha256": cache.digest(db_root, path)})
    runtime_base_url = public_url(base_url, f"runtimes/{runtime_id}/files")
    settings_template: dict[str, Any] | None = None
    # ersc_settings.ini must remain outside `files` because it is personalised
    # per pack, but it must be advertised as a template so readers can generate
    # it after downloading the Seamless runtime.
    if runtime_id.casefold() == "seamlesscoop":
        template = files_root / "ersc_settings.ini"
        if template.is_file():
            settings_template = {
                "template_path": "ersc_settings.ini",
                "template_sha256": cache.digest(db_root, template),
            }
    release_material = {"runtime_id": runtime_id, "files": files, "settings_template": settings_template}
    manifest = {
        "schema_version": SCHEMA_VERSION,
        "runtime_id": runtime_id,
        "generated_at": utc_now(),
        "base_url": runtime_base_url,
        "files": files,
        "release": {"id": canonical_hash(release_material), "file_count": len(files), "payload_bytes": sum(item["size"] for item in files)},
    }
    if settings_template:
        manifest["settings_template"] = settings_template
    atomic_write_json(runtime_dir / "manifest.json", manifest)
    return manifest


def runtime_dependency(
    *,
    db_root: Path,
    base_url: str,
    pack_id: str,
    runtime_id: str,
    target_subdir: str,
    runtime_manifest_overrides: dict[str, dict[str, Any]] | None = None,
) -> tuple[dict[str, Any], set[str]]:
    """Return a dependency plus exact pack-relative paths owned by runtime.

    Runtime-owned paths are excluded from the pack payload so a raw pack never
    competes with its own shared runtime during integrity checks.
    """
    runtime_id = normalise_relative(runtime_id)
    target_subdir = str(target_subdir or "").replace("\\", "/").strip().strip("/")
    if target_subdir:
        target_subdir = normalise_relative(target_subdir)
    runtime_manifest_path = db_root / "runtimes" / runtime_id / "manifest.json"
    runtime_manifest = (runtime_manifest_overrides or {}).get(runtime_id)
    if not isinstance(runtime_manifest, dict):
        runtime_manifest = load_json(runtime_manifest_path, None)
    runtime_release = runtime_manifest.get("release") if isinstance(runtime_manifest, dict) else {}
    runtime_release_id = str(runtime_release.get("id") or "").strip() if isinstance(runtime_release, dict) else ""
    if not runtime_release_id or not isinstance(runtime_manifest, dict) or not isinstance(runtime_manifest.get("files"), list):
        raise RuntimeError(
            f"{pack_id}: required runtime '{runtime_id}' has no valid manifest/release. "
            "Publish or rebuild the runtime before publishing this pack."
        )
    owned: set[str] = set()
    for entry in runtime_manifest["files"]:
        if not isinstance(entry, dict):
            continue
        try:
            rel = normalise_relative(str(entry.get("path") or ""))
        except ValueError:
            continue
        owned.add(f"{target_subdir}/{rel}".strip("/"))
    return ({
        "id": runtime_id,
        "manifest_url": public_url(base_url, f"runtimes/{runtime_id}/manifest.json"),
        "target_subdir": target_subdir,
        "release_id": runtime_release_id,
    }, owned)


def build_pack_manifest(
    *,
    db_root: Path,
    base_url: str,
    pack_dir: Path,
    cache: HashCache,
    refresh_templates: bool,
    pack_id_override: str | None = None,
    runtime_manifest_overrides: dict[str, dict[str, Any]] | None = None,
) -> tuple[dict[str, Any], dict[str, Any]]:
    pack_id = normalise_relative(pack_id_override or pack_dir.name)
    files_root = pack_dir / "files"
    if not files_root.is_dir():
        raise RuntimeError(f"{pack_id}: missing files/ directory")
    meta = load_meta(pack_dir)
    generated, excluded = build_generated_files(
        db_root=db_root,
        base_url=base_url,
        pack_id=pack_id,
        pack_dir=pack_dir,
        files_root=files_root,
        meta=meta,
        cache=cache,
        refresh_templates=refresh_templates,
    )
    runtime_deps: list[dict[str, Any]] = []
    runtime_owned: set[str] = set()
    runtime = meta.get("runtime") if isinstance(meta.get("runtime"), dict) else {}
    engine = str(runtime.get("engine") or "").casefold()
    if engine == "modengine2" and bool(runtime.get("use_shared_runtime")):
        dep, owned = runtime_dependency(
            db_root=db_root, base_url=base_url, pack_id=pack_id,
            runtime_id=str(runtime.get("runtime_id") or "modengine2"), target_subdir="",
            runtime_manifest_overrides=runtime_manifest_overrides,
        )
        runtime_deps.append(dep)
        runtime_owned.update(owned)

    seamless = meta.get("seamlesscoop") if isinstance(meta.get("seamlesscoop"), dict) else {}
    if bool(seamless.get("enabled")):
        seamless_runtime_id = normalise_relative(str(seamless.get("runtime_id") or "seamlesscoop"))
        settings_name = normalise_relative(str(seamless.get("settings_path") or "ersc_settings.ini"))
        seamless_manifest = (runtime_manifest_overrides or {}).get(seamless_runtime_id)
        if not isinstance(seamless_manifest, dict):
            seamless_manifest = load_json(db_root / "runtimes" / seamless_runtime_id / "manifest.json", None)
        template = seamless_manifest.get("settings_template") if isinstance(seamless_manifest, dict) else None
        if not isinstance(template, dict) or not str(template.get("template_path") or "").strip():
            raise RuntimeError(
                f"{pack_id}: Seamless Co-op runtime is missing its ersc_settings.ini template. "
                f"Place the pristine file at runtimes/{seamless_runtime_id}/files/ersc_settings.ini and rebuild."
            )
        dep, owned = runtime_dependency(
            db_root=db_root, base_url=base_url, pack_id=pack_id,
            runtime_id=seamless_runtime_id,
            target_subdir=str(seamless.get("target_subdir") or "SeamlessCoop"),
            runtime_manifest_overrides=runtime_manifest_overrides,
        )
        dep["settings"] = {
            "path": settings_name,
            "cooppassword": str(seamless.get("cooppassword") or ""),
            "save_file_extension": str(seamless.get("save_file_extension") or ""),
        }
        if not any(str(existing.get("id") or "") == dep["id"] and str(existing.get("target_subdir") or "") == dep["target_subdir"] for existing in runtime_deps):
            runtime_deps.append(dep)
            runtime_owned.update(owned)

    extra_ignore = [str(item) for item in meta.get("ignore", []) if isinstance(item, str)]
    files: list[dict[str, Any]] = []
    for path in sorted(iter_payload_files(files_root, extra_ignore, excluded | runtime_owned), key=lambda item: item.as_posix().casefold()):
        relative = path.relative_to(files_root).as_posix()
        stat = path.stat()
        files.append({"path": relative, "size": int(stat.st_size), "sha256": cache.digest(db_root, path)})

    install_folder = str(meta.get("install_folder") or meta.get("name") or pack_id).strip()
    install_folder = leaf_folder_name(install_folder)
    release_material = {
        "pack_id": pack_id,
        "version": str(meta.get("version") or ""),
        "files": files,
        "generated_files": generated,
        "runtime_dependencies": runtime_deps,
        "runtime": runtime,
    }
    release_id = canonical_hash(release_material)
    manifest = {
        "schema_version": SCHEMA_VERSION,
        "pack_id": pack_id,
        "install_folder": install_folder,
        "generated_at": utc_now(),
        "base_url": public_url(base_url, f"packs/{pack_id}/files"),
        "files": files,
        "generated_files": generated,
        "runtime_dependencies": runtime_deps,
        "runtime": runtime,
        "release": {
            "id": release_id,
            "version": str(meta.get("version") or ""),
            "file_count": len(files),
            "payload_bytes": sum(item["size"] for item in files),
        },
    }
    atomic_write_json(pack_dir / "manifest.json", manifest)

    image = str(meta.get("image") or "").strip()
    if image and not re.match(r"^https?://", image, flags=re.IGNORECASE):
        image = public_url(base_url, image)
    source = meta.get("source") if isinstance(meta.get("source"), dict) else {}
    runtime_badges = {
        "engine": str(runtime.get("engine") or ""),
        "seamlesscoop": bool(seamless.get("enabled")),
        "dll_module_count": len(next((item.get("external_dlls", []) for item in generated if item.get("kind") == "modengine2_config"), [])),
        "launch_path": str(runtime.get("launch_path") or ""),
        # Profile-only ME3 packs (for example ERTE) need a player-local ME3
        # installation. Self-contained packs (for example Convergence) retain
        # their bundled runtime and launcher in the payload.
        "me3_mode": str(runtime.get("me3_mode") or ""),
        "profile_path": str(runtime.get("profile_path") or ""),
        "requires_external_me3": bool(runtime.get("requires_external_me3")),
    }
    catalog_entry = {
        "id": pack_id,
        "name": str(meta.get("name") or pack_id),
        "version": str(meta.get("version") or ""),
        "description": str(meta.get("description") or ""),
        "image": image,
        "nexus_url": str(meta.get("nexus_url") or source.get("nexus_url") or ""),
        "source": source,
        "tags": [str(item) for item in meta.get("tags", []) if str(item).strip()],
        "manifest_url": public_url(base_url, f"packs/{pack_id}/manifest.json"),
        "release_id": release_id,
        "payload_bytes": manifest["release"]["payload_bytes"],
        "file_count": manifest["release"]["file_count"],
        "release": manifest["release"],
        "runtime": runtime_badges,
    }
    return manifest, catalog_entry


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--db-root", default=os.environ.get("ERMD2_DB_ROOT", DEFAULT_DB_ROOT))
    parser.add_argument("--public-base-url", default=os.environ.get("ERMD2_PUBLIC_BASE_URL", DEFAULT_PUBLIC_BASE_URL))
    parser.add_argument("--pack", action="append", default=[], help="Build only a named pack ID. Catalog is still regenerated from all valid manifests.")
    parser.add_argument("--refresh-templates", action="store_true", help="Refresh generated config templates from files/ before building.")
    parser.add_argument("--stage-pack-dir", default="", help="Build a manifest inside a hidden incoming pack directory without publishing it.")
    parser.add_argument("--stage-pack-id", default="", help="Final pack ID to use with --stage-pack-dir.")
    parser.add_argument("--stage-runtime-dir", action="append", default=[], help="Stage runtime in ID=/absolute/path form for an atomic watcher publish.")
    parser.add_argument("--catalog-only", action="store_true", help="Regenerate catalog from existing manifests without rebuilding manifests.")
    parser.add_argument(
        "--trusted-catalog-only",
        action="store_true",
        help="Watcher-only catalog refresh after staged validation; skip the redundant whole-database blocklist sweep.",
    )
    parser.add_argument(
        "--skip-runtime-rebuild",
        action="store_true",
        help="For a pack-only watcher rebuild, reuse existing runtime manifests instead of scanning/rebuilding every shared runtime.",
    )
    args = parser.parse_args()
    if args.skip_runtime_rebuild and not args.pack:
        parser.error("--skip-runtime-rebuild requires at least one --pack.")
    if args.trusted_catalog_only and not args.catalog_only:
        parser.error("--trusted-catalog-only requires --catalog-only.")

    db_root = Path(args.db_root).resolve()
    packs_root = db_root / "packs"
    runtimes_root = db_root / "runtimes"
    generated_root = db_root / "generated"
    catalog_path = db_root / "catalog" / "packs.json"
    if not packs_root.is_dir():
        print(f"ERROR: missing packs directory: {packs_root}", file=sys.stderr)
        return 2

    selected = {normalise_relative(item) for item in args.pack}
    staged_preflight = bool(args.stage_pack_dir)

    # Final server-side safety net. This catches direct edits and manual uploads
    # that bypass Publisher. Blocked binaries are deleted before any manifest is
    # generated, so they cannot be published or hashed as pack content.
    #
    # A hidden staged import is validated separately below.  Its pack payload is
    # explicitly scrubbed there, while the watcher has already scrubbed every
    # staged runtime payload before invoking this process.  Re-scanning every
    # already-published pack/runtime during that preflight is redundant and was
    # a major source of avoidable publish time.
    if not args.trusted_catalog_only:
        load_blocklist(db_root)
    blocklist_removed: list[tuple[str, dict[str, str]]] = []
    if not staged_preflight and not args.trusted_catalog_only:
        if runtimes_root.is_dir() and not args.skip_runtime_rebuild:
            for runtime_dir in runtimes_root.iterdir():
                if runtime_dir.is_dir() and not runtime_dir.name.startswith("."):
                    for event in scrub_blocked_artifacts(db_root, runtime_dir / "files"):
                        blocklist_removed.append((f"runtime {runtime_dir.name}", event))
        for pack_dir in packs_root.iterdir():
            if pack_dir.is_dir() and not pack_dir.name.startswith("."):
                if selected and pack_dir.name not in selected:
                    continue
                for event in scrub_blocked_artifacts(db_root, pack_dir / "files"):
                    blocklist_removed.append((f"pack {pack_dir.name}", event))
    for owner, event in blocklist_removed:
        print(f"Blocked and deleted: {owner}/{event['path']} ({event['match']}: {event['rules']})")

    cache = HashCache(generated_root / "hash_cache.json")

    # Watcher preflight mode: build manifests in hidden incoming trees before
    # their public paths are atomically swapped into place.
    staged_runtime_manifests: dict[str, dict[str, Any]] = {}
    for raw in args.stage_runtime_dir:
        try:
            runtime_id, raw_path = str(raw).split("=", 1)
            runtime_id = normalise_relative(runtime_id)
            runtime_dir = Path(raw_path).resolve()
            manifest = build_runtime_manifest(db_root, args.public_base_url, runtime_dir, cache, runtime_id_override=runtime_id)
            if not isinstance(manifest, dict):
                raise RuntimeError("missing files/ directory")
            manifest["runtime_id"] = runtime_id
            manifest["base_url"] = public_url(args.public_base_url, f"runtimes/{runtime_id}/files")
            material = {"runtime_id": runtime_id, "files": manifest["files"], "settings_template": manifest.get("settings_template")}
            manifest["release"]["id"] = canonical_hash(material)
            atomic_write_json(runtime_dir / "manifest.json", manifest)
            staged_runtime_manifests[runtime_id] = manifest
            print(f"Staged runtime built: {runtime_id} ({len(manifest['files'])} files)")
        except Exception as exc:
            print(f"ERROR: staged runtime {raw}: {exc}", file=sys.stderr)
            return 1

    if args.stage_pack_dir:
        try:
            stage_dir = Path(args.stage_pack_dir).resolve()
            stage_id = normalise_relative(args.stage_pack_id or stage_dir.name)
            scrub_blocked_artifacts(db_root, stage_dir / "files")
            manifest, _entry = build_pack_manifest(
                db_root=db_root, base_url=args.public_base_url, pack_dir=stage_dir,
                cache=cache, refresh_templates=bool(args.refresh_templates),
                pack_id_override=stage_id, runtime_manifest_overrides=staged_runtime_manifests,
            )
            cache.save()
            print(f"Staged pack built: {stage_id} ({len(manifest['files'])} files, {manifest['release']['payload_bytes']:,} bytes)")
            return 0
        except Exception as exc:
            print(f"ERROR: staged pack: {exc}", file=sys.stderr)
            return 1

    entries: list[dict[str, Any]] = []
    failures: list[str] = []

    if runtimes_root.is_dir() and not args.catalog_only and not args.skip_runtime_rebuild:
        for runtime_dir in sorted((path for path in runtimes_root.iterdir() if path.is_dir() and not path.name.startswith(".")), key=lambda item: item.name.casefold()):
            try:
                runtime = build_runtime_manifest(db_root, args.public_base_url, runtime_dir, cache)
                if runtime:
                    print(f"Runtime built: {runtime_dir.name} ({len(runtime['files'])} files)")
            except Exception as exc:
                failures.append(f"runtime {runtime_dir.name}: {exc}")

    for pack_dir in sorted((path for path in packs_root.iterdir() if path.is_dir() and not path.name.startswith(".")), key=lambda item: item.name.casefold()):
        if selected and pack_dir.name not in selected:
            # Preserve its previously generated catalog entry by reading the
            # existing manifest/meta instead of treating it as missing.
            manifest = load_json(pack_dir / "manifest.json", None)
            meta = load_meta(pack_dir)
            if isinstance(manifest, dict) and isinstance(manifest.get("release"), dict):
                image = str(meta.get("image") or "")
                if image and not re.match(r"^https?://", image, flags=re.IGNORECASE):
                    image = public_url(args.public_base_url, image)
                source = meta.get("source") if isinstance(meta.get("source"), dict) else {}
                entries.append({
                    "id": pack_dir.name,
                    "name": str(meta.get("name") or pack_dir.name),
                    "version": str(meta.get("version") or manifest["release"].get("version") or ""),
                    "description": str(meta.get("description") or ""),
                    "image": image,
                    "nexus_url": str(meta.get("nexus_url") or source.get("nexus_url") or ""),
                    "source": source,
                    "tags": [str(item) for item in meta.get("tags", []) if str(item).strip()],
                    "manifest_url": public_url(args.public_base_url, f"packs/{pack_dir.name}/manifest.json"),
                    "release_id": str(manifest["release"].get("id") or ""),
                    "payload_bytes": int(manifest["release"].get("payload_bytes") or 0),
                    "file_count": int(manifest["release"].get("file_count") or 0),
                    "release": manifest["release"],
                    "runtime": manifest.get("runtime") or {},
                })
            continue
        try:
            if args.catalog_only:
                manifest = load_json(pack_dir / "manifest.json", None)
                meta = load_meta(pack_dir)
                if not isinstance(manifest, dict) or not isinstance(manifest.get("release"), dict):
                    raise RuntimeError("missing valid manifest.json")
                image = str(meta.get("image") or "")
                if image and not re.match(r"^https?://", image, flags=re.IGNORECASE):
                    image = public_url(args.public_base_url, image)
                source = meta.get("source") if isinstance(meta.get("source"), dict) else {}
                entry = {
                    "id": pack_dir.name, "name": str(meta.get("name") or pack_dir.name),
                    "version": str(meta.get("version") or manifest["release"].get("version") or ""),
                    "description": str(meta.get("description") or ""), "image": image,
                    "nexus_url": str(meta.get("nexus_url") or source.get("nexus_url") or ""),
                    "source": source, "tags": [str(item) for item in meta.get("tags", []) if str(item).strip()],
                    "manifest_url": public_url(args.public_base_url, f"packs/{pack_dir.name}/manifest.json"),
                    "release_id": str(manifest["release"].get("id") or ""),
                    "payload_bytes": int(manifest["release"].get("payload_bytes") or 0),
                    "file_count": int(manifest["release"].get("file_count") or 0),
                    "release": manifest["release"], "runtime": manifest.get("runtime") or {},
                }
                print(f"Catalog entry retained: {pack_dir.name}")
            else:
                manifest, entry = build_pack_manifest(
                    db_root=db_root, base_url=args.public_base_url, pack_dir=pack_dir,
                    cache=cache, refresh_templates=bool(args.refresh_templates),
                )
                print(f"Pack built: {pack_dir.name} ({len(manifest['files'])} files, {manifest['release']['payload_bytes']:,} bytes)")
            entries.append(entry)
        except Exception as exc:
            failures.append(f"pack {pack_dir.name}: {exc}")

    entries.sort(key=lambda item: str(item.get("name") or "").casefold())
    catalog = {
        "schema_version": SCHEMA_VERSION,
        "generated_at": utc_now(),
        "packs": entries,
    }
    atomic_write_json(catalog_path, catalog)
    cache.save()
    print(f"Catalog written: {catalog_path} ({len(entries)} packs)")
    if failures:
        print("Build warnings/errors:", file=sys.stderr)
        for failure in failures:
            print(f" - {failure}", file=sys.stderr)
        return 1
    return 0


if __name__ == "__main__":
    raise SystemExit(main())