left4me/l4d2host/instances.py

from pathlib import Path
import shutil
from typing import Callable

from l4d2host.process import run_command
from l4d2host.spec import load_spec
from l4d2host.systemd_user import daemon_reload, ensure_template_unit


DEFAULT_ROOT = Path("/opt/l4d2")


def initialize_instance(
    name: str,
    spec_path: Path,
    *,
    root: Path = DEFAULT_ROOT,
    on_stdout: Callable[[str], None] | None = None,
    on_stderr: Callable[[str], None] | None = None,
    passthrough: bool = False,
    should_cancel: Callable[[], bool] | None = None,
) -> None:
    spec = load_spec(spec_path)

    instance_dir = root / "instances" / name
    runtime_dir = root / "runtime" / name
    (runtime_dir / "upper").mkdir(parents=True, exist_ok=True)
    (runtime_dir / "work").mkdir(parents=True, exist_ok=True)
    (runtime_dir / "merged").mkdir(parents=True, exist_ok=True)
    instance_dir.mkdir(parents=True, exist_ok=True)

    lowerdirs = [str(root / "overlays" / overlay) for overlay in spec.overlays]
    lowerdirs.append(str(root / "installation"))

    instance_env = "\n".join(
        [
            f"L4D2_PORT={spec.port}",
            f"L4D2_ARGS={' '.join(spec.arguments)}",
            f"L4D2_LOWERDIRS={':'.join(lowerdirs)}",
        ]
    ) + "\n"
    (instance_dir / "instance.env").write_text(instance_env)

    server_cfg = "\n".join(spec.config) if spec.config else ""
    (instance_dir / "server.cfg").write_text(server_cfg)

    if root.resolve() == DEFAULT_ROOT:
        ensure_template_unit()
        daemon_reload(on_stdout=on_stdout, on_stderr=on_stderr, passthrough=passthrough, should_cancel=should_cancel)


def _load_instance_env(path: Path) -> dict[str, str]:
    result: dict[str, str] = {}
    for line in path.read_text().splitlines():
        if "=" not in line:
            continue
        key, value = line.split("=", 1)
        result[key] = value
    return result


def start_instance(
    name: str,
    *,
    root: Path = DEFAULT_ROOT,
    on_stdout: Callable[[str], None] | None = None,
    on_stderr: Callable[[str], None] | None = None,
    passthrough: bool = False,
    should_cancel: Callable[[], bool] | None = None,
) -> None:
    instance_dir = root / "instances" / name
    runtime_dir = root / "runtime" / name

    env = _load_instance_env(instance_dir / "instance.env")

    run_command(
        [
            "fuse-overlayfs",
            "-o",
            (
                f"lowerdir={env['L4D2_LOWERDIRS']},"
                f"upperdir={runtime_dir / 'upper'},"
                f"workdir={runtime_dir / 'work'}"
            ),
            str(runtime_dir / "merged"),
        ],
        on_stdout=on_stdout,
        on_stderr=on_stderr,
        passthrough=passthrough,
        should_cancel=should_cancel,
    )

    target_cfg = runtime_dir / "merged" / "left4dead2" / "cfg" / "server.cfg"
    target_cfg.parent.mkdir(parents=True, exist_ok=True)
    shutil.copy2(instance_dir / "server.cfg", target_cfg)

    run_command(
        ["systemctl", "--user", "start", f"l4d2@{name}.service"],
        on_stdout=on_stdout,
        on_stderr=on_stderr,
        passthrough=passthrough,
        should_cancel=should_cancel,
    )


def stop_instance(
    name: str,
    *,
    root: Path = DEFAULT_ROOT,
    on_stdout: Callable[[str], None] | None = None,
    on_stderr: Callable[[str], None] | None = None,
    passthrough: bool = False,
    should_cancel: Callable[[], bool] | None = None,
) -> None:
    run_command(
        ["systemctl", "--user", "stop", f"l4d2@{name}.service"],
        on_stdout=on_stdout,
        on_stderr=on_stderr,
        passthrough=passthrough,
        should_cancel=should_cancel,
    )
    run_command(
        ["fusermount3", "-u", str(root / "runtime" / name / "merged")],
        on_stdout=on_stdout,
        on_stderr=on_stderr,
        passthrough=passthrough,
        should_cancel=should_cancel,
    )


def delete_instance(
    name: str,
    *,
    root: Path = DEFAULT_ROOT,
    on_stdout: Callable[[str], None] | None = None,
    on_stderr: Callable[[str], None] | None = None,
    passthrough: bool = False,
    should_cancel: Callable[[], bool] | None = None,
) -> None:
    instance_dir = root / "instances" / name
    runtime_dir = root / "runtime" / name

    if not instance_dir.exists() and not runtime_dir.exists():
        return

    run_command(
        ["systemctl", "--user", "stop", f"l4d2@{name}.service"],
        on_stdout=on_stdout,
        on_stderr=on_stderr,
        passthrough=passthrough,
        should_cancel=should_cancel,
    )

    merged = runtime_dir / "merged"
    if merged.is_mount():
        run_command(
            ["fusermount3", "-u", str(merged)],
            on_stdout=on_stdout,
            on_stderr=on_stderr,
            passthrough=passthrough,
            should_cancel=should_cancel,
        )

    if instance_dir.exists():
        shutil.rmtree(instance_dir)
    if runtime_dir.exists():
        shutil.rmtree(runtime_dir)