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 channels


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("--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]:
    available_channels = {int(channel_id) for channel_id in channels.CHANNEL_MAP.keys()}

    if args.channels:
        selected = set(args.channels)
    else:
        selected = {channel_id for channel_id in available_channels if channel_id <= args.channel_limit}

    if 99 in available_channels:
        selected.add(99)

    unknown = sorted(selected - available_channels)
    if unknown:
        print(f"Unknown channels requested: {unknown}", file=sys.stderr)
        raise SystemExit(1)

    return sorted(selected)


def resolve_instances(selected_channels: list[int]) -> list[str]:
    instances: list[str] = []
    for channel_id in selected_channels:
        cores = channels.CHANNEL_MAP[int(channel_id)]
        for core_id in sorted(int(core) for core in cores.keys()):
            instances.append(f"channel{channel_id}_core{core_id}")
    return instances


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)

    selected_channels = resolve_channels(args)
    instances = resolve_instances(selected_channels)

    template_values = {
        "USER_NAME": args.user,
        "GROUP_NAME": group_name,
        "RUNTIME_ROOT": runtime_root,
        "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)

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

    enable_units = [
        "metin-server.service",
        "metin-db.service",
        "metin-db-ready.service",
        "metin-auth.service",
        *[f"metin-game@{instance}.service" for instance in instances],
    ]
    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())