from __future__ import annotations

import hashlib
import os
import shlex
import threading
from functools import lru_cache
from pathlib import Path
from typing import Any, Callable

from common.runtime_paths import resource_path


XRAY_ASSET_FILENAME = "xray-linux-amd64"
XRAY_ASSET_SHA256 = "4bedbb1fa52a5647fe6a2587c0ee0a2834298f699e8ae6e9ee66ba985c602824"
XRAY_REMOTE_PATH = "/usr/local/bin/xray"


def _emit(emit: Callable[[str], None] | None, message: str) -> None:
    if emit is not None:
        emit(message)


def _sha256(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as source:
        for chunk in iter(lambda: source.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


@lru_cache(maxsize=1)
def bundled_xray_path() -> Path:
    path = Path(resource_path(XRAY_ASSET_FILENAME))
    if not path.is_file():
        raise FileNotFoundError(f"软件内置Xray不存在: {path}")
    actual_sha256 = _sha256(path)
    if actual_sha256 != XRAY_ASSET_SHA256:
        raise RuntimeError(
            "软件内置Xray校验失败: "
            f"期望 {XRAY_ASSET_SHA256}，实际 {actual_sha256}"
        )
    return path


def _remote_exec(ssh: Any, command: str, *, timeout: int = 30) -> tuple[int, str, str]:
    _stdin, stdout, stderr = ssh.exec_command(command, timeout=timeout)
    output = stdout.read().decode("utf-8", errors="replace")
    error = stderr.read().decode("utf-8", errors="replace")
    return stdout.channel.recv_exit_status(), output, error


def _remote_xray_is_valid(ssh: Any) -> bool:
    remote = shlex.quote(XRAY_REMOTE_PATH)
    try:
        status, _output, _error = _remote_exec(
            ssh,
            f"test -x {remote} && {remote} version >/dev/null 2>&1",
            timeout=20,
        )
    except Exception:
        return False
    return status == 0


def ensure_remote_xray(
    ssh: Any,
    emit: Callable[[str], None] | None = None,
) -> bool:
    """Ensure a usable Xray exists remotely without downloading it from the server."""
    local_path = bundled_xray_path()
    if _remote_xray_is_valid(ssh):
        return False

    suffix = f"{os.getpid()}-{threading.get_ident()}"
    remote_temp = f"/usr/local/bin/.xray-{XRAY_ASSET_SHA256[:12]}-{suffix}.part"
    quoted_temp = shlex.quote(remote_temp)
    quoted_target = shlex.quote(XRAY_REMOTE_PATH)

    _emit(emit, "正在上传软件内置Xray...")
    status, _output, error = _remote_exec(
        ssh,
        "mkdir -p /usr/local/bin",
        timeout=20,
    )
    if status != 0:
        raise RuntimeError(f"创建Xray目录失败: {error.strip() or status}")

    try:
        sftp = ssh.open_sftp()
        try:
            channel = getattr(sftp, "get_channel", lambda: None)()
            if channel is not None:
                channel.settimeout(180)
            sftp.put(str(local_path), remote_temp)
        finally:
            sftp.close()

        verify_command = (
            "set -e; "
            f"printf '%s  %s\\n' {shlex.quote(XRAY_ASSET_SHA256)} {quoted_temp} "
            "| sha256sum -c - >/dev/null; "
            f"chmod 0755 {quoted_temp}; "
            f"mv -f {quoted_temp} {quoted_target}; "
            f"{quoted_target} version >/dev/null 2>&1"
        )
        status, _output, error = _remote_exec(ssh, verify_command, timeout=60)
        if status != 0:
            raise RuntimeError(f"远端Xray校验或安装失败: {error.strip() or status}")
    except Exception:
        try:
            _remote_exec(ssh, f"rm -f {quoted_temp}", timeout=10)
        except Exception:
            pass
        raise

    _emit(emit, "软件内置Xray已上传并校验。")
    return True
