m2dev-server/deploy/systemd/install_systemd.py

#!/usr/bin/env python3
import argparse
import os
import shutil
import subprocess
import sys
from pathlib import Path

SCRIPT_DIR = Path(__file__).resolve().parent
REPO_ROOT = SCRIPT_DIR.parent.parent
sys.path.insert(0, str(REPO_ROOT))

import channel_inventory


TEMPLATES_DIR = SCRIPT_DIR / "templates"
BIN_DIR = SCRIPT_DIR / "bin"


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Install Metin systemd units")
    parser.add_argument("--user", required=True, help="Runtime user")
    parser.add_argument("--group", help="Runtime group, defaults to --user")
    parser.add_argument("--runtime-root", required=True, help="Absolute path to the live runtime root")
    parser.add_argument("--systemd-dir", default="/etc/systemd/system", help="systemd unit destination")
    parser.add_argument("--libexec-dir", default="/usr/local/libexec", help="Helper script destination")
    parser.add_argument("--bin-dir", default="/usr/local/bin", help="Binary/script destination")
    parser.add_argument("--sbin-dir", default="/usr/local/sbin", help="Root-only binary/script destination")
    parser.add_argument("--env-file", default="/etc/metin/metin.env", help="Optional EnvironmentFile path for runtime overrides")
    parser.add_argument("--wait-host", default="127.0.0.1", help="DB readiness host")
    parser.add_argument("--wait-port", type=int, default=9000, help="DB readiness port")
    parser.add_argument("--wait-timeout", type=int, default=30, help="DB readiness timeout in seconds")
    parser.add_argument("--restart", action="store_true", help="Restart metin-server.service after install")

    channel_group = parser.add_mutually_exclusive_group(required=True)
    channel_group.add_argument(
        "--channel-limit",
        type=int,
        help="Enable channels up to this number and keep channel 99 if present",
    )
    channel_group.add_argument(
        "--channel",
        dest="channels",
        action="append",
        type=int,
        help="Enable an explicit channel, may be passed multiple times",
    )
    return parser.parse_args()


def ensure_root() -> None:
    if os.geteuid() != 0:
        print("This installer must run as root.", file=sys.stderr)
        raise SystemExit(1)


def render_template(template_path: Path, values: dict[str, str]) -> str:
    content = template_path.read_text(encoding="utf-8")
    for key, value in values.items():
        content = content.replace(f"{{{{{key}}}}}", value)
    return content


def write_text(path: Path, content: str, mode: int) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(content, encoding="utf-8")
    os.chmod(path, mode)


def copy_file(source: Path, destination: Path, mode: int) -> None:
    destination.parent.mkdir(parents=True, exist_ok=True)
    shutil.copy2(source, destination)
    os.chmod(destination, mode)


def resolve_channels(args: argparse.Namespace) -> list[int]:
    try:
        return channel_inventory.resolve_selected_channels(
            channel_limit=args.channel_limit,
            explicit_channels=args.channels,
        )
    except ValueError as exc:
        print(str(exc), file=sys.stderr)
        raise SystemExit(1)


def resolve_instances(selected_channels: list[int]) -> list[str]:
    return channel_inventory.get_instances(selected_channels)


def run(command: list[str]) -> None:
    subprocess.run(command, check=True)


def main() -> int:
    args = parse_args()
    ensure_root()

    group_name = args.group or args.user
    runtime_root = str(Path(args.runtime_root).resolve())
    systemd_dir = Path(args.systemd_dir)
    libexec_dir = Path(args.libexec_dir)
    bin_dir = Path(args.bin_dir)
    sbin_dir = Path(args.sbin_dir)

    selected_channels = resolve_channels(args)
    instances = resolve_instances(selected_channels)
    selected_game_units = set(channel_inventory.get_game_units(selected_channels))
    all_game_units = set(channel_inventory.get_game_units())

    template_values = {
        "USER_NAME": args.user,
        "GROUP_NAME": group_name,
        "REPO_ROOT": str(REPO_ROOT),
        "RUNTIME_ROOT": runtime_root,
        "ENV_FILE": args.env_file,
        "WAIT_HOST": args.wait_host,
        "WAIT_PORT": str(args.wait_port),
        "WAIT_TIMEOUT": str(args.wait_timeout),
        "WAIT_TIMEOUT_SYSTEMD": str(args.wait_timeout + 5),
    }

    unit_names = [
        "metin-server.service",
        "metin-db.service",
        "metin-db-ready.service",
        "metin-auth.service",
        "metin-game@.service",
    ]

    for unit_name in unit_names:
        template_path = TEMPLATES_DIR / f"{unit_name}.in"
        write_text(systemd_dir / unit_name, render_template(template_path, template_values), 0o644)

    write_text(
        libexec_dir / "metin-game-instance-start",
        render_template(BIN_DIR / "metin-game-instance-start.in", template_values),
        0o755,
    )
    copy_file(BIN_DIR / "metin-wait-port", libexec_dir / "metin-wait-port", 0o755)
    write_text(
        bin_dir / "metinctl",
        render_template(BIN_DIR / "metinctl.in", template_values),
        0o755,
    )
    write_text(
        sbin_dir / "metin-collect-incident",
        render_template(BIN_DIR / "metin-collect-incident.in", template_values),
        0o700,
    )

    verify_units = [str(systemd_dir / unit_name) for unit_name in unit_names]
    run(["systemd-analyze", "verify", *verify_units])
    run(["systemctl", "daemon-reload"])

    stale_game_units = sorted(all_game_units - selected_game_units)
    if stale_game_units:
        disable_command = ["systemctl", "disable"]
        if args.restart:
            disable_command.append("--now")
        run([*disable_command, *stale_game_units])

    enable_units = [
        "metin-server.service",
        "metin-db.service",
        "metin-db-ready.service",
        "metin-auth.service",
        *sorted(selected_game_units),
    ]
    run(["systemctl", "enable", *enable_units])

    if args.restart:
        run(["systemctl", "restart", "metin-server.service"])

    print("Installed systemd units for instances:")
    for instance in instances:
        print(f" - {instance}")
    return 0


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