"""Live user sessions on hosts (Linux loginctl / Windows qwinsta) via SSH or WinRM."""

from __future__ import annotations

import re
from dataclasses import dataclass

from app.models import Event, Host
from app.services.linux_admin_settings import LinuxAdminConfig
from app.services.ssh_connect import (
    HostNotLinuxError as SshHostNotLinuxError,
    HostTargetMissingError as SshHostTargetMissingError,
    SshCommandResult,
    is_linux_host,
    iter_ssh_targets,
    run_ssh_command,
)
from app.services.win_admin_settings import WinAdminConfig
from app.services.winrm_connect import (
    HostNotWindowsError,
    HostTargetMissingError,
    WinRmCmdResult,
    is_windows_host,
    run_winrm_logoff,
    run_winrm_on_host_targets,
    run_winrm_qwinsta,
)

SESSION_EVENT_TYPES_LINUX = frozenset(
    {
        "session.logind.new",
        "ssh.login.success",
        "privilege.sudo.command",
    }
)
SESSION_EVENT_TYPES_WINDOWS = frozenset({"rdp.login.success"})


@dataclass(frozen=True)
class HostSessionRow:
    session_id: str
    user: str
    tty: str | None = None
    state: str | None = None
    source_ip: str | None = None
    session_name: str | None = None


def _details_dict(event: Event) -> dict:
    raw = event.details
    return raw if isinstance(raw, dict) else {}


def event_session_id(event: Event) -> str | None:
    details = _details_dict(event)
    for key in ("session_id", "sid"):
        val = details.get(key)
        if val is not None and str(val).strip():
            return str(val).strip()
    return None


def event_supports_session_terminate(event: Event) -> bool:
    if event.type in SESSION_EVENT_TYPES_LINUX:
        return is_linux_host_event(event)
    if event.type in SESSION_EVENT_TYPES_WINDOWS:
        return is_windows_host_event(event)
    return False


def is_linux_host_event(event: Event) -> bool:
    host = event.host
    return host is not None and is_linux_host(host)


def is_windows_host_event(event: Event) -> bool:
    host = event.host
    return host is not None and is_windows_host(host)


def parse_loginctl_sessions(stdout: str) -> list[HostSessionRow]:
    rows: list[HostSessionRow] = []
    for line in stdout.splitlines():
        text = line.strip()
        if not text:
            continue
        parts = text.split()
        if len(parts) < 3:
            continue
        sid, user = parts[0], parts[2]
        tty = parts[4] if len(parts) > 4 and parts[4] != "-" else None
        state = parts[5] if len(parts) > 5 else None
        rows.append(
            HostSessionRow(
                session_id=sid,
                user=user,
                tty=tty,
                state=state,
            )
        )
    return rows


def parse_qwinsta_sessions(stdout: str, *, filter_user: str | None = None) -> list[HostSessionRow]:
    rows: list[HostSessionRow] = []
    norm_filter = (filter_user or "").strip().lower()

    def norm_user(value: str) -> str:
        return value.replace("B26\\", "").replace("b26\\", "").lower()

    for line in stdout.splitlines():
        text = line.strip()
        if not text or re.match(r"^SESSION", text, re.I) or text.startswith("---"):
            continue
        parts = text.split()
        if len(parts) < 4:
            continue
        session_name = parts[0].lstrip(">")
        user_name = parts[1]
        try:
            sid = int(parts[2])
        except ValueError:
            continue
        state = " ".join(parts[3:])
        if norm_filter and norm_filter not in norm_user(user_name):
            continue
        rows.append(
            HostSessionRow(
                session_id=str(sid),
                user=user_name,
                session_name=session_name,
                state=state,
            )
        )

    if rows or not norm_filter:
        return rows

    return parse_qwinsta_sessions(stdout, filter_user=None)


def list_linux_sessions(host: Host, cfg: LinuxAdminConfig) -> tuple[list[HostSessionRow], SshCommandResult]:
    if not is_linux_host(host):
        raise SshHostNotLinuxError("Host is not Linux")
    targets = iter_ssh_targets(host)
    cmd = "loginctl list-sessions --no-legend --no-pager 2>/dev/null || who -u"
    last: SshCommandResult | None = None
    for target in targets:
        result = run_ssh_command(
            target=target,
            user=cfg.user,
            password=cfg.password,
            remote_cmd=cmd,
            connect_timeout_sec=15,
            command_timeout_sec=45,
            need_root=True,
        )
        last = result
        if result.ok and result.stdout.strip():
            return parse_loginctl_sessions(result.stdout), result
    assert last is not None
    return [], last


def terminate_linux_session(
    host: Host,
    cfg: LinuxAdminConfig,
    session_id: str,
) -> SshCommandResult:
    sid = session_id.strip()
    if not sid:
        return SshCommandResult(ok=False, message="session_id is required", target="")
    if not is_linux_host(host):
        raise SshHostNotLinuxError("Host is not Linux")
    targets = iter_ssh_targets(host)
    remote_cmd = f"loginctl terminate-session {_shell_quote(sid)}"
    last: SshCommandResult | None = None
    for target in targets:
        result = run_ssh_command(
            target=target,
            user=cfg.user,
            password=cfg.password,
            remote_cmd=remote_cmd,
            connect_timeout_sec=15,
            command_timeout_sec=45,
            need_root=True,
        )
        last = result
        if result.ok:
            return result
    assert last is not None
    return last


def _shell_quote(value: str) -> str:
    return "'" + value.replace("'", "'\"'\"'") + "'"


def list_windows_sessions(host: Host, cfg: WinAdminConfig) -> tuple[list[HostSessionRow], WinRmCmdResult | None]:
    if not is_windows_host(host):
        raise HostNotWindowsError("Host is not Windows")

    def action(target: str) -> WinRmCmdResult:
        return run_winrm_qwinsta(target=target, user=cfg.user, password=cfg.password)

    result, _attempts = run_winrm_on_host_targets(host, user=cfg.user, password=cfg.password, action=action)
    if result is None or not result.ok:
        return [], result
    return parse_qwinsta_sessions(result.stdout), result


def terminate_windows_session(
    host: Host,
    cfg: WinAdminConfig,
    session_id: str,
) -> WinRmCmdResult | None:
    try:
        sid = int(session_id.strip())
    except ValueError:
        return WinRmCmdResult(ok=False, message="Invalid Windows session_id", target="")

    def action(target: str) -> WinRmCmdResult:
        return run_winrm_logoff(
            target=target,
            user=cfg.user,
            password=cfg.password,
            session_id=sid,
        )

    result, _attempts = run_winrm_on_host_targets(host, user=cfg.user, password=cfg.password, action=action)
    return result


def terminate_session_for_event(
    event: Event,
    *,
    linux_cfg: LinuxAdminConfig,
    win_cfg: WinAdminConfig,
    session_id: str | None = None,
) -> SshCommandResult | WinRmCmdResult:
    host = event.host
    if host is None:
        raise ValueError("Event has no host")

    sid = (session_id or event_session_id(event) or "").strip()
    if is_linux_host(host):
        if not sid:
            user = (event.actor_user or _details_dict(event).get("user") or "").strip()
            if user:
                return _terminate_linux_user_sessions(host, linux_cfg, user)
            raise ValueError("session_id is required for this event")
        return terminate_linux_session(host, linux_cfg, sid)

    if is_windows_host(host):
        if not sid:
            user = (event.actor_user or _details_dict(event).get("user") or "").strip()
            sessions, qwinsta = list_windows_sessions(host, win_cfg)
            if not qwinsta or not qwinsta.ok:
                raise ValueError(qwinsta.message if qwinsta else "qwinsta failed")
            matched = [s for s in sessions if user.lower() in s.user.lower()] if user else sessions
            if len(matched) == 1:
                sid = matched[0].session_id
            elif not matched:
                raise ValueError("No matching Windows session for user")
            else:
                raise ValueError("Multiple sessions; specify session_id")
        result = terminate_windows_session(host, win_cfg, sid)
        if result is None:
            raise ValueError("WinRM logoff failed")
        return result

    raise ValueError("Unsupported host OS for session terminate")


def _terminate_linux_user_sessions(host: Host, cfg: LinuxAdminConfig, user: str) -> SshCommandResult:
    safe_user = _shell_quote(user.strip())
    remote_cmd = f"loginctl terminate-user {safe_user}"
    targets = iter_ssh_targets(host)
    last: SshCommandResult | None = None
    for target in targets:
        result = run_ssh_command(
            target=target,
            user=cfg.user,
            password=cfg.password,
            remote_cmd=remote_cmd,
            connect_timeout_sec=15,
            command_timeout_sec=45,
            need_root=True,
        )
        last = result
        if result.ok:
            return result
    assert last is not None
    return last