"""
dxcluster.py — DX Cluster client (Telnet).

Opens a TCP connection to a DX Cluster server, sends the login and
password (if required), then parses incoming lines in the DX de format:
  DX de <spotter>:  <freq>  <call>       <comment>  <utc>

Each logged-in user can have their own connection with their own settings
(address, port, login, password) — managed per-user in ClusterManager.

Spots are broadcast via a callback (WS to the specific user) in this format:
  {"type": "dx_spot", "freq_hz": int, "call": str, "spotter": str,
   "comment": str, "utc": str, "ts": epoch_seconds, "band": str, "mode": str}

NOTE: the "message" field of a dx_status broadcast IS displayed directly
in the UI for the error/disconnected states (public/js/dxcluster.js's
updateStatusBadge, not routed through I18n) - keep any text assigned to
it in Polish; only comments/docstrings/print() in this file are in scope
for the backend English translation pass.
"""

import asyncio
import ipaddress
import re
import socket
import time
from typing import Callable, Optional

try:
    from config import DATA as _DATA_DIR
except Exception:
    from pathlib import Path as _Path
    _DATA_DIR = _Path(__file__).parent

# Dziennik polaczenia DX Clustera - zgloszone na zywo: "hamctrl nie ma
# zaimplementowanych logow", wiec przy "ciagle sie rozlancza i laczy" nie
# bylo jak sprawdzic co faktycznie sie dzieje (produkcyjny EXE dziala bez
# okna konsoli - print() ginie bez sladu). Osobny, maly plik tekstowy w
# tym samym katalogu co config.DATA (czyli %APPDATA%\HAMCTRL na
# Windowsie), zeby bylo co wkleic/przeczytac po fakcie zamiast zgadywac.
_LOG_FILE = _DATA_DIR / "dxcluster.log"
_LOG_MAX_BYTES = 1_000_000  # ~1MB - obcinamy zamiast rosnac w nieskonczonosc


def _log_line(msg: str):
    try:
        ts = time.strftime('%Y-%m-%d %H:%M:%S')
        line = f"[{ts}] {msg}\n"
        if _LOG_FILE.exists() and _LOG_FILE.stat().st_size > _LOG_MAX_BYTES:
            # Prosta rotacja: zostaw tylko ostatnia polowe pliku
            try:
                data = _LOG_FILE.read_text(encoding='utf-8', errors='ignore')
                _LOG_FILE.write_text(data[len(data) // 2:], encoding='utf-8')
            except Exception:
                pass
        with open(_LOG_FILE, 'a', encoding='utf-8') as f:
            f.write(line)
    except Exception:
        pass  # logowanie nigdy nie moze wywrocic polaczenia


async def _resolve_and_check_safe(host: str, port: int) -> tuple[bool, str]:
    """
    SSRF guard (added 2026-09-02, found via a live permission audit): ANY
    logged-in user - operator or viewer, no admin gate on this endpoint -
    can set their OWN DX cluster host/port via /api/dxcluster/config, and
    /api/dxcluster/connect then makes THE SERVER open a raw TCP connection
    to it and send a "login" line. Without this check that's a working
    SSRF primitive: point it at 127.0.0.1:<any local port>, an internal
    LAN address, or another machine's admin interface, and the server
    will connect and speak to it on the user's behalf. Resolves the
    HOSTNAME to its actual IP (not just string-matching "127.0.0.1") so a
    DNS name that merely points at an internal address is caught too, and
    checks every address a name resolves to, not just the first.
    """
    if not host:
        return False, "Brak adresu hosta"
    try:
        loop = asyncio.get_event_loop()
        infos = await asyncio.wait_for(
            loop.getaddrinfo(host, port, type=socket.SOCK_STREAM), timeout=5.0)
    except Exception as e:
        return False, f"Nie mozna rozwiazac adresu {host!r}: {e}"
    for info in infos:
        ip_str = info[4][0]
        try:
            ip = ipaddress.ip_address(ip_str)
        except ValueError:
            continue
        if (ip.is_private or ip.is_loopback or ip.is_link_local
                or ip.is_reserved or ip.is_multicast or ip.is_unspecified):
            return False, (f"Adres {host!r} ({ip_str}) wskazuje na siec "
                            f"prywatna/lokalna - zablokowane")
    return True, ""


# Band prefixes from the IARU table (to place a spot on the waterfall).
# 160m/60m/6m DELIBERATELY NARROWER (real PL/EU allocation) - the same
# values as webapp.py::_BAND_RANGES (the table also used by the TX block).
# This used to be a separate, independently-maintained copy that drifted
# out of sync - the SAME bug already fixed once in webapp.py, just in a
# different file.
_BAND_RANGES = [
    ('160m', 1810000,   2000000),
    ('80m',  3500000,   3800000),
    ('60m',  5351500,   5366500),
    ('40m',  7000000,   7200000),
    ('30m',  10100000,  10150000),
    ('20m',  14000000,  14350000),
    ('17m',  18068000,  18168000),
    ('15m',  21000000,  21450000),
    ('12m',  24890000,  24990000),
    ('10m',  28000000,  29700000),
    ('6m',   50000000,  52000000),
    ('4m',   70000000,  70500000),
    ('2m',   144000000, 146000000),
    ('70cm', 430000000, 440000000),
]


def _get_band(freq_hz: int) -> str:
    for name, lo, hi in _BAND_RANGES:
        if lo <= freq_hz <= hi:
            return name
    return '?'


def _guess_mode(freq_hz: int, comment: str) -> str:
    """
    Spot mode heuristic: comment -> exact digital mode frequencies ->
    IARU R1 bandplan.

    Order matters:
      1. The comment (as SEPARATE WORDS \\b...\\b, so 'OK1CW' doesn't give CW)
      2. EXACT FT8/FT4/MSK144 frequencies with +-2 kHz tolerance.
         This must come BEFORE the bandplan, since e.g. 2m FT8 (144.174)
         falls within the 144.150-144.400 phone segment and the bandplan
         would incorrectly return SSB.
      3. Bandplan (CW/DIGI/SSB/FM segments) - HF, VHF, UHF, microwave.
    """
    c = (comment or '').upper()

    # ── 1. Mode stated directly in the comment ──────────────────────────────
    for pat, mode in (
        (r'\bFT8\b',                'FT8'),
        (r'\bFT4\b',                'FT4'),
        (r'\bMSK144\b',             'MSK144'),
        (r'\bJS8\b',                'DIGI'),
        (r'\bQ65\b',                'DIGI'),
        (r'\bJT(?:65|9|6M)\b',      'DIGI'),
        (r'\bRTTY\b',               'RTTY'),
        (r'\bPSK(?:31|63)?\b',      'PSK'),
        (r'\bWSPR\b',               'DIGI'),
        (r'\bSSB\b',                'SSB'),
        (r'\b(?:LSB|USB)\b',        'SSB'),
        (r'\bCW\b',                 'CW'),
        (r'\bFM\b',                 'FM'),
        (r'\bAM\b',                 'AM'),
    ):
        if re.search(pat, c):
            return mode

    khz = freq_hz / 1000.0

    # ── 2. Exact digital mode frequencies (+-2 kHz tolerance) ────────────────
    # An FT8 signal has ~50 Hz carriers within a 3 kHz band, spots vary by a
    # few hundred Hz, hence the tolerance. MUST come before the bandplan
    # (see the docstring).
    FT8_KHZ = [
        1840, 3573, 5357, 7074, 10136, 14074, 18100, 21074, 24915, 28074,
        50313, 50323,            # 6m (dwie czestotliwosci)
        70100, 70154,            # 4m
        144174,                  # 2m
        222065,                  # 1.25m
        432174,                  # 70cm
        1296174,                 # 23cm
        2320174,                 # 13cm
        3400174, 5760174, 10368174, 24048174,   # 9cm/6cm/3cm/1.2cm
    ]
    FT4_KHZ = [
        3575, 7047.5, 10140, 14080, 18104, 21140, 24919, 28180,
        50318,                   # 6m
        144170,                  # 2m
    ]
    MSK144_KHZ = [50260, 70230, 144360, 432360]   # meteor scatter

    def _near(target_list, tol=2.0):
        return any(abs(khz - t) <= tol for t in target_list)

    if _near(FT8_KHZ):    return 'FT8'
    if _near(FT4_KHZ):    return 'FT4'
    if _near(MSK144_KHZ): return 'MSK144'

    # WSPR (narrow, 1 kHz tolerance)
    WSPR_KHZ = [1836.6, 3568.6, 7038.6, 10138.7, 14095.6, 18104.6,
                21094.6, 24924.6, 28124.6, 50293, 144489]
    if _near(WSPR_KHZ, tol=1.0): return 'DIGI'

    # ── 3. Satellites — BEFORE the bandplan, since FM segments would mask them ──
    # 2m: 145.800-146.000, 70cm: 435.000-438.000 (IARU satellite segments)
    if 145800 <= khz <= 146000 or 435000 <= khz <= 438000:
        return 'SAT'

    # ── 4. IARU Region 1 bandplan (kHz) ──────────────────────────────────────
    BANDPLAN = [
        # ── HF ──
        (1810, 1838, 'CW'), (1838, 1843, 'DIGI'), (1843, 2000, 'SSB'),
        (3500, 3570, 'CW'), (3570, 3600, 'DIGI'), (3600, 3800, 'SSB'),
        (5250, 5450, 'SSB'),                       # 60m (channelized, USB)
        (7000, 7040, 'CW'), (7040, 7060, 'DIGI'), (7060, 7200, 'SSB'),
        (10100, 10130, 'CW'), (10130, 10150, 'DIGI'),   # 30m - no phone
        (14000, 14070, 'CW'), (14070, 14099, 'DIGI'), (14101, 14350, 'SSB'),
        (18068, 18095, 'CW'), (18095, 18109, 'DIGI'), (18111, 18168, 'SSB'),
        (21000, 21070, 'CW'), (21070, 21150, 'DIGI'), (21151, 21450, 'SSB'),
        (24890, 24915, 'CW'), (24915, 24929, 'DIGI'), (24931, 24990, 'SSB'),
        (28000, 28070, 'CW'), (28070, 28190, 'DIGI'), (28225, 29000, 'SSB'),
        (29000, 29700, 'FM'),
        # ── 6m ──
        (50000, 50100, 'CW'), (50100, 50300, 'SSB'),
        (50300, 50400, 'DIGI'), (50400, 52000, 'SSB'),
        # ── 4m (70 MHz) — SSB from 70.200, narrower DIGI ──
        (69900, 70100, 'CW'), (70100, 70200, 'DIGI'),
        (70200, 70300, 'SSB'), (70300, 70500, 'FM'),
        # ── 2m ──
        (144000, 144150, 'CW'), (144150, 144400, 'SSB'),
        (144400, 144500, 'DIGI'), (144500, 145800, 'FM'),
        (146000, 148000, 'FM'),                    # region 2
        # ── 1.25m (222 MHz, region 2) ──
        (222000, 222150, 'CW'), (222150, 222250, 'SSB'), (222250, 225000, 'FM'),
        # ── 70cm ──
        (430000, 432000, 'FM'), (432000, 432100, 'CW'),
        (432100, 432400, 'SSB'), (432400, 432500, 'DIGI'),
        (432500, 435000, 'FM'), (438000, 440000, 'FM'),
        # ── 23cm ──
        (1296000, 1296150, 'CW'), (1296150, 1296400, 'SSB'),
        (1296400, 1296600, 'DIGI'), (1296600, 1300000, 'FM'),
        # ── 13cm ──
        (2320000, 2320150, 'CW'), (2320150, 2320400, 'SSB'), (2320400, 2450000, 'FM'),
        # ── 9cm ──
        (3400000, 3400150, 'CW'), (3400150, 3400400, 'SSB'),
        # ── 6cm ──
        (5760000, 5760150, 'CW'), (5760150, 5760400, 'SSB'),
        # ── 3cm ──
        (10368000, 10368150, 'CW'), (10368150, 10368400, 'SSB'),
        # ── 1.2cm ──
        (24048000, 24048150, 'CW'), (24048150, 24048400, 'SSB'),
    ]
    for lo, hi, mode in BANDPLAN:
        if lo <= khz < hi:
            return mode

    return '?'


# Regex for the "DX de <spotter>:  <freq_khz>  <call>  <comment>  <utc>" line
# Example: "DX de SP3ABC-#:  14074.0  DX1CALL      FT8 -12 dB               1234Z"
_DX_RE = re.compile(
    r'^DX\s+de\s+([\w\-#/]+):?\s+([\d.]+)\s+([\w/\-]+)\s+(.*?)(\d{4}Z)?\s*$',
    re.IGNORECASE
)

# FIX (2026-10-21, live report: spot comments showed "1755Z JO82" /
# "1755Z00" glued on, or just the time when there was no comment): _DX_RE
# above only recognises the time when it's the LAST thing on the line.
# Many clusters append more after it (the spotter's locator "1755Z JO82",
# stray characters, BEL \x07 control chars), so the time silently ended up
# inside the comment. This finds the time token at the tail regardless of
# what follows it and cuts it (plus everything after) off the comment.
_DX_TAIL_TIME_RE = re.compile(r'(?<!\d)(\d{4})Z\S*(?:\s+\S+)?\s*$', re.IGNORECASE)
_CTRL_CHARS_RE = re.compile(r'[\x00-\x1f\x7f]')


class DXClusterClient:
    """A single connection to a DX Cluster server for one user."""

    IDLE_PROBE_S = 300.0   # cisza po ktorej wysylamy sonde (pusta linie)
    PROBE_WAIT_S = 120.0   # ile czekamy na odpowiedz na sonde

    def __init__(self, host: str, port: int, login: str, password: str = "",
                 on_spot: Optional[Callable[[dict], None]] = None,
                 on_status: Optional[Callable[[str, str], None]] = None):
        self.host = host
        self.port = port
        self.login = login
        self.password = password
        self.on_spot = on_spot
        self.on_status = on_status  # (status, message) - "connecting" / "connected" / "disconnected" / "error"

        self._reader: Optional[asyncio.StreamReader] = None
        self._writer: Optional[asyncio.StreamWriter] = None
        self._task: Optional[asyncio.Task] = None
        self._connected = False
        self._should_run = False
        self._reconnect_delay = 5.0
        self._max_reconnect_delay = 60.0

    def is_connected(self) -> bool:
        return self._connected

    async def _status(self, status: str, msg: str = ""):
        _log_line(f"{self.login}@{self.host}:{self.port} -> {status}" + (f" ({msg})" if msg else ""))
        if self.on_status:
            try:
                res = self.on_status(status, msg)
                if asyncio.iscoroutine(res): await res
            except Exception as e:
                print(f"[dx] on_status callback error: {e}")

    async def connect(self):
        """Start the connection with background auto-reconnect."""
        if self._task and not self._task.done():
            return
        self._should_run = True
        self._task = asyncio.create_task(self._run_loop())

    async def disconnect(self):
        """Stop the connection."""
        self._should_run = False
        if self._writer:
            try:
                self._writer.close()
                await self._writer.wait_closed()
            except Exception:
                pass
        self._writer = None
        self._reader = None
        self._connected = False
        if self._task:
            self._task.cancel()
            try:
                await self._task
            except (asyncio.CancelledError, Exception):
                pass
            self._task = None
        await self._status("disconnected", "")

    async def _run_loop(self):
        """Connection loop with auto-reconnect on disconnect."""
        delay = self._reconnect_delay
        while self._should_run:
            try:
                await self._status("connecting", f"{self.host}:{self.port}")
                self._reader, self._writer = await asyncio.wait_for(
                    asyncio.open_connection(self.host, self.port),
                    timeout=10.0
                )
                self._connected = True
                delay = self._reconnect_delay  # reset the delay after success
                await self._status("connected", "")

                try:
                    sock = self._writer.get_extra_info('socket')
                    if sock is not None:
                        sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
                except Exception:
                    pass

                # Login sequence
                await self._do_login()

                # Receive lines
                # Wykrywanie martwego polaczenia (zgloszone na zywo: admin w
                # nocy "wylaczyl sie" i wrocil dopiero po recznym
                # rozlacz/polacz). readline() bez limitu czasu wisial w
                # nieskonczonosc na polotwartym TCP (serwer/router zerwal
                # polaczenie bez FIN) - _connected zostawal True, wiec ani
                # petla reconnect, ani auto_connect przy logowaniu nic nie
                # robily. Teraz: po IDLE_PROBE_S ciszy wysylamy pusta linie
                # (DXSpider odpowiada promptem), a jesli przez PROBE_WAIT_S
                # dalej nic nie przyjdzie - zrywamy i laczymy od nowa.
                probe_sent = False
                while self._should_run:
                    try:
                        line = await asyncio.wait_for(
                            self._reader.readline(),
                            timeout=self.PROBE_WAIT_S if probe_sent else self.IDLE_PROBE_S)
                    except asyncio.TimeoutError:
                        if probe_sent:
                            _log_line(f"{self.login}@{self.host}:{self.port} brak odpowiedzi na sonde "
                                      f"- polaczenie martwe, reconnect")
                            raise ConnectionError("brak danych z serwera - polaczenie martwe")
                        _log_line(f"{self.login}@{self.host}:{self.port} cisza {self.IDLE_PROBE_S:.0f}s - wysylam sonde")
                        self._writer.write(b"\r\n")
                        await self._writer.drain()
                        probe_sent = True
                        continue
                    probe_sent = False
                    if not line:
                        break  # server closed
                    try:
                        text = line.decode('utf-8', errors='ignore').strip()
                    except Exception:
                        continue
                    if text:
                        await self._process_line(text)

            except asyncio.CancelledError:
                raise
            except asyncio.TimeoutError:
                # NOTE: kept in Polish - this message IS shown directly in
                # the UI status badge (public/js/dxcluster.js's
                # updateStatusBadge), not routed through I18n.
                await self._status("error", "Timeout polaczenia")
            except Exception as e:
                await self._status("error", str(e))

            # Cleanup and wait before reconnecting
            self._connected = False
            if self._writer:
                try: self._writer.close()
                except: pass
                self._writer = None

            if not self._should_run:
                break

            # NOTE: kept in Polish - see the note above on the "error" status message.
            await self._status("disconnected", f"reconnect za {delay:.0f}s")
            try:
                await asyncio.sleep(delay)
            except asyncio.CancelledError:
                break
            delay = min(delay * 1.5, self._max_reconnect_delay)

    async def _do_login(self):
        """Send the login and password. The server may prompt with 'login:',
        'Please enter your call:', 'password:' etc. We use a simple strategy:
        send the call, wait 500ms, then send the password if there is one."""
        if not self._writer:
            return
        # Basic strategy — wait for the prompt, then send the login
        await asyncio.sleep(0.5)
        try:
            self._writer.write((self.login + "\r\n").encode('ascii'))
            await self._writer.drain()
            _log_line(f"{self.login}@{self.host}:{self.port} sent login")
        except Exception as e:
            print(f"[dx] error sending login: {e}")
            _log_line(f"{self.login}@{self.host}:{self.port} ERROR sending login: {e}")
            return
        # If there's a password, wait briefly and send it
        if self.password:
            await asyncio.sleep(1.0)
            try:
                self._writer.write((self.password + "\r\n").encode('ascii'))
                await self._writer.drain()
                _log_line(f"{self.login}@{self.host}:{self.port} sent password ({len(self.password)} chars)")
            except Exception as e:
                print(f"[dx] error sending password: {e}")
                _log_line(f"{self.login}@{self.host}:{self.port} ERROR sending password: {e}")

    async def _process_line(self, text: str):
        """Parse a line and (if it's a spot) call the callback."""
        text = _CTRL_CHARS_RE.sub('', text).rstrip()
        m = _DX_RE.match(text)
        if not m:
            # Non-spot lines (banner, prompts, sysop announcements, kick
            # messages like "Reconnected as X ... this instance is
            # disconnected") - exactly the lines that explain an
            # unexpected disconnect, so these go to the log (spot lines
            # themselves don't - too frequent, and not diagnostically
            # useful for connection-stability issues).
            _log_line(f"{self.login}@{self.host}:{self.port} << {text}")
            return

        spotter    = m.group(1)
        freq_str   = m.group(2)
        call       = m.group(3)
        comment    = (m.group(4) or '').strip()
        utc        = m.group(5) or ''
        if not utc:
            _t = _DX_TAIL_TIME_RE.search(comment)
            if _t:
                utc = _t.group(1) + 'Z'
                comment = comment[:_t.start()].strip()

        # Freq is usually in kHz with a fraction (14074.0)
        try:
            freq_hz = int(round(float(freq_str) * 1000))
        except ValueError:
            return

        spot = {
            "type":    "dx_spot",
            "freq_hz": freq_hz,
            "call":    call.upper(),
            "spotter": spotter.upper(),
            "comment": comment[:60],  # capped at 60 characters
            "utc":     utc,
            "ts":      time.time(),
            "band":    _get_band(freq_hz),
            "mode":    _guess_mode(freq_hz, comment),
        }

        if self.on_spot:
            try:
                res = self.on_spot(spot)
                if asyncio.iscoroutine(res): await res
            except Exception as e:
                print(f"[dx] on_spot callback error: {e}")

    async def send_command(self, cmd: str):
        """Send a command to the server (e.g. 'sh/dx 20m FT8', 'set/qth', 'q')."""
        if not self._writer or not self._connected:
            return False
        try:
            self._writer.write((cmd + "\r\n").encode('ascii'))
            await self._writer.drain()
            return True
        except Exception as e:
            print(f"[dx] send_command error: {e}")
            return False


class ClusterManager:
    """Manages DX Cluster connections for all users.
    Each user has at most 1 connection. The broadcast callback delivers
    messages to the right user's WebSocket."""

    def __init__(self, on_broadcast: Callable[[str, dict], None]):
        """on_broadcast(user_id, message) — sends over WS to a specific user."""
        self.on_broadcast = on_broadcast
        self._clients: dict[str, DXClusterClient] = {}
        # Cache of the last N spots per user - returned when a tab is reopened.
        # 300 (bylo 100): na zapchanym pasmie zalew spotow cyfrowych (FT8)
        # wypychal z tego bufora spoty CW/SSB w czasie znacznie krotszym niz
        # ich docelowy czas zycia na fladze wodospadu (SPOT_FLAG_TTL_S w
        # waterfall.js, 3 min) - zgloszone na zywo jako "flagi z dx clustra
        # pala sie zbyt krotko".
        self._spot_history: dict[str, list[dict]] = {}
        self._max_history = 300

    def get_client(self, user_id: str) -> Optional[DXClusterClient]:
        return self._clients.get(user_id)

    def connected_user_ids(self) -> list[str]:
        """user_id's with a live (connect() called) client - used by
        webapp.py's idle watchdog to find candidates for auto-disconnect."""
        return list(self._clients.keys())

    def get_history(self, user_id: str) -> list[dict]:
        return list(self._spot_history.get(user_id, []))

    def get_merged_history(self) -> list[dict]:
        """Union of every connected login's spot history - the admin's
        always-on one AND any other user's own, if they also connect one.
        Matches the live broadcast (webapp.py::_dx_broadcast now sends every
        dx_spot to all online users, not just the one whose connection
        received it) so a fresh page load/reconnect sees the same picture
        the live feed would build up. Raw/undeduped on purpose - the client
        (dxcluster.js::_addSpot) replays this through the same de-dup path
        used for live spots, so there's only one place that knows about the
        call+frequency tolerance."""
        merged: list[dict] = []
        for hist in self._spot_history.values():
            merged.extend(hist)
        merged.sort(key=lambda s: s.get("ts", 0))
        return merged[-self._max_history:]

    async def connect_user(self, user_id: str, host: str, port: int,
                            login: str, password: str = ""):
        """Create/reset the connection for a user."""
        # Close the previous connection if one exists
        old = self._clients.pop(user_id, None)
        if old:
            await old.disconnect()

        async def _on_spot(spot):
            # Cache in the history
            hist = self._spot_history.setdefault(user_id, [])
            hist.append(spot)
            if len(hist) > self._max_history:
                del hist[:len(hist) - self._max_history]
            # Broadcast to the user
            res = self.on_broadcast(user_id, spot)
            if asyncio.iscoroutine(res): await res

        async def _on_status(status, msg):
            res = self.on_broadcast(user_id, {
                "type": "dx_status", "status": status, "message": msg,
            })
            if asyncio.iscoroutine(res): await res

        # SSRF guard - see _resolve_and_check_safe's docstring. Checked
        # here (not just in webapp.py's /api/dxcluster/config save) so
        # EVERY path that actually opens the connection is covered,
        # including the auto_connect-on-login path in ws_handler.
        _safe, _why = await _resolve_and_check_safe(host, port)
        if not _safe:
            await _on_status("error", _why)
            return None

        client = DXClusterClient(host, port, login, password,
                                  on_spot=_on_spot, on_status=_on_status)
        self._clients[user_id] = client
        await client.connect()
        return client

    async def disconnect_user(self, user_id: str):
        client = self._clients.pop(user_id, None)
        if client:
            await client.disconnect()

    async def send_command(self, user_id: str, cmd: str) -> bool:
        client = self._clients.get(user_id)
        if not client: return False
        return await client.send_command(cmd)

    async def shutdown_all(self):
        for user_id in list(self._clients.keys()):
            await self.disconnect_user(user_id)