plugin-helper/src/plugin_helper/installer.py

from __future__ import annotations

import shutil
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
from zipfile import ZipFile

from .fsutil import ensure_inside, ensure_relative, sha256_bytes, sha256_file
from .state import backups_dir, installed_state_path, load_installed_state, save_installed_state


def _timestamp() -> str:
    return datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")


def _backup_existing(instance_path: Path, backup_root: Path, rel_target: str) -> str | None:
    target = ensure_inside(instance_path, instance_path / ensure_relative(rel_target))
    if not target.exists():
        return None
    backup_path = backup_root / rel_target
    backup_path.parent.mkdir(parents=True, exist_ok=True)
    shutil.copy2(target, backup_path)
    return str(backup_path)


def apply_plan(plan: dict[str, Any], state_root: Path) -> dict[str, Any]:
    instance = plan["instance"]
    install_id = plan.get("installId")
    instance_path = Path(plan["instancePath"])
    if not instance_path.is_dir():
        raise FileNotFoundError(f"instance path does not exist: {instance_path}")

    backup_root = backups_dir(state_root, instance, install_id=install_id) / f"apply-{_timestamp()}"
    installed_state = load_installed_state(state_root, instance, install_id=install_id)
    installed_state.setdefault("beatSaberVersion", plan.get("beatSaberVersion"))
    installed_state.setdefault("plugins", {})

    applied: list[dict[str, Any]] = []
    for change in plan.get("changes", []):
        source = Path(change["source"])
        if sha256_file(source) != change["sourceSha256"]:
            raise ValueError(f"source hash changed: {source}")
        rel_target = ensure_relative(change["target"]).as_posix()
        target = ensure_inside(instance_path, instance_path / rel_target)
        backup = _backup_existing(instance_path, backup_root, rel_target)
        target.parent.mkdir(parents=True, exist_ok=True)

        if change["action"] == "copy":
            shutil.copy2(source, target)
        elif change["action"] == "extract":
            with ZipFile(source) as archive:
                data = archive.read(change["archiveMember"])
            if sha256_bytes(data) != change["sha256"]:
                raise ValueError(f"archive member hash changed: {change['archiveMember']}")
            target.write_bytes(data)
        else:
            raise ValueError(f"unsupported action: {change['action']}")

        actual_sha = sha256_file(target)
        if actual_sha != change["sha256"]:
            raise ValueError(f"installed file hash mismatch: {rel_target}")

        plugin_state = installed_state["plugins"].setdefault(
            change["plugin"],
            {
                "installedAt": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"),
                "files": [],
            },
        )
        plugin_state["files"] = [
            item for item in plugin_state.get("files", []) if item.get("path") != rel_target
        ]
        plugin_state["files"].append(
            {
                "path": rel_target,
                "sha256": actual_sha,
                "size": target.stat().st_size,
            }
        )
        installed_state.setdefault("disabledPlugins", {}).pop(change["plugin"], None)
        applied.append({"path": rel_target, "plugin": change["plugin"], "backup": backup})

    save_installed_state(state_root, instance, installed_state, install_id=install_id)
    return {"applied": applied, "statePath": str(installed_state_path(state_root, instance, install_id=install_id))}


def uninstall_plugin(
    instance: str,
    instance_path: Path,
    state_root: Path,
    plugin_id: str,
    force: bool = False,
    *,
    install_id: str | None = None,
) -> dict[str, Any]:
    installed_state = load_installed_state(state_root, instance, install_id=install_id)
    plugin_state = installed_state.get("plugins", {}).get(plugin_id)
    if not plugin_state:
        raise KeyError(f"plugin is not recorded in managed install state: {plugin_id}")

    removed: list[str] = []
    skipped: list[dict[str, str]] = []
    for item in plugin_state.get("files", []):
        rel_path = ensure_relative(item["path"]).as_posix()
        target = ensure_inside(instance_path, instance_path / rel_path)
        if not target.exists():
            removed.append(rel_path)
            continue
        current_sha = sha256_file(target)
        if current_sha != item.get("sha256") and not force:
            skipped.append({"path": rel_path, "reason": "hash mismatch"})
            continue
        target.unlink()
        removed.append(rel_path)

    if skipped and not force:
        return {"removed": removed, "skipped": skipped, "stateUpdated": False}

    installed_state.get("plugins", {}).pop(plugin_id, None)
    save_installed_state(state_root, instance, installed_state, install_id=install_id)
    return {"removed": removed, "skipped": skipped, "stateUpdated": True}


def disable_plugin(
    instance: str,
    instance_path: Path,
    state_root: Path,
    plugin_id: str,
    force: bool = False,
    *,
    install_id: str | None = None,
) -> dict[str, Any]:
    installed_state = load_installed_state(state_root, instance, install_id=install_id)
    plugin_state = installed_state.get("plugins", {}).get(plugin_id)
    if not plugin_state:
        if plugin_id in installed_state.get("disabledPlugins", {}):
            raise KeyError(f"plugin is already disabled: {plugin_id}")
        raise KeyError(f"plugin is not recorded in managed install state: {plugin_id}")

    removed: list[str] = []
    skipped: list[dict[str, str]] = []
    for item in plugin_state.get("files", []):
        rel_path = ensure_relative(item["path"]).as_posix()
        target = ensure_inside(instance_path, instance_path / rel_path)
        if not target.exists():
            removed.append(rel_path)
            continue
        current_sha = sha256_file(target)
        if current_sha != item.get("sha256") and not force:
            skipped.append({"path": rel_path, "reason": "hash mismatch"})
            continue
        target.unlink()
        removed.append(rel_path)

    if skipped and not force:
        return {"removed": removed, "skipped": skipped, "stateUpdated": False}

    disabled_state = dict(plugin_state)
    disabled_state["disabledAt"] = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
    installed_state.setdefault("disabledPlugins", {})[plugin_id] = disabled_state
    installed_state.get("plugins", {}).pop(plugin_id, None)
    save_installed_state(state_root, instance, installed_state, install_id=install_id)
    return {"removed": removed, "skipped": skipped, "stateUpdated": True}