server.py

"""
Ion Storm main server.

Features:
1) Keep TCP control port 5000 open for desktop client compatibility.
2) Support WebSocket control on port 5000 for web client room creation.
3) Allocate independent room ports for either TCP rooms or WebSocket rooms.
4) Reclaim idle room ports that have no connected clients.
"""

from __future__ import annotations

import base64
import hashlib
import json
import random
import secrets
import socket
import struct
import threading
import time
from dataclasses import dataclass
from typing import Any, Dict, Optional, Tuple

from ChemGame import GameEngine, HostServer, _recv_json_nonblocking, _send_json


MAIN_PORT = 5000
ROOM_PORT_MIN = 10000
ROOM_PORT_MAX = 59999
MAX_ALLOC_ATTEMPTS = 400
ROOM_EMPTY_RECLAIM_SECONDS = 15.0
WATCHDOG_INTERVAL_SECONDS = 2.0

HEARTBEAT_INTERVAL_SECONDS = 1.0
HEARTBEAT_LAG_SECONDS = 3.0
HEARTBEAT_TIMEOUT_SECONDS = 8.0
HEARTBEAT_WATCHDOG_SECONDS = 1.0

_WS_GUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"


def _set_listener_socket_opts(sock: socket.socket) -> None:
    """
    Configure listener sockets to avoid accidental port sharing.

    On Windows, prefer SO_EXCLUSIVEADDRUSE so another process cannot bind the
    same port with SO_REUSEADDR and steal part of incoming traffic.
    """
    if hasattr(socket, "SO_EXCLUSIVEADDRUSE"):
        try:
            sock.setsockopt(socket.SOL_SOCKET, socket.SO_EXCLUSIVEADDRUSE, 1)
            return
        except OSError:
            pass
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)


def _ws_send_frame(sock: socket.socket, opcode: int, payload: bytes = b"") -> None:
    fin_opcode = 0x80 | (opcode & 0x0F)
    n = len(payload)
    if n < 126:
        header = bytes([fin_opcode, n])
    elif n <= 0xFFFF:
        header = bytes([fin_opcode, 126]) + struct.pack("!H", n)
    else:
        header = bytes([fin_opcode, 127]) + struct.pack("!Q", n)
    sock.sendall(header + payload)


def _ws_send_text(sock: socket.socket, text: str) -> None:
    _ws_send_frame(sock, 0x1, text.encode("utf-8"))


def _ws_send_json(sock: socket.socket, payload: Dict[str, Any]) -> None:
    _ws_send_text(sock, json.dumps(payload, ensure_ascii=False))


def _ws_try_parse_frame(buffer: bytes) -> Tuple[Optional[Tuple[int, bytes]], bytes]:
    if len(buffer) < 2:
        return None, buffer

    b1 = buffer[0]
    b2 = buffer[1]
    opcode = b1 & 0x0F
    masked = (b2 & 0x80) != 0
    payload_len = b2 & 0x7F
    idx = 2

    if payload_len == 126:
        if len(buffer) < idx + 2:
            return None, buffer
        payload_len = struct.unpack("!H", buffer[idx : idx + 2])[0]
        idx += 2
    elif payload_len == 127:
        if len(buffer) < idx + 8:
            return None, buffer
        payload_len = struct.unpack("!Q", buffer[idx : idx + 8])[0]
        idx += 8

    mask_key = b""
    if masked:
        if len(buffer) < idx + 4:
            return None, buffer
        mask_key = buffer[idx : idx + 4]
        idx += 4

    if len(buffer) < idx + payload_len:
        return None, buffer

    payload = buffer[idx : idx + payload_len]
    rest = buffer[idx + payload_len :]

    if masked:
        payload = bytes(payload[i] ^ mask_key[i % 4] for i in range(payload_len))

    return (opcode, payload), rest


def _ws_recv_json_nonblocking(
    sock: socket.socket, buffer: bytes
) -> Tuple[Optional[Dict[str, Any]], bytes, bool]:
    while True:
        parsed, buffer = _ws_try_parse_frame(buffer)
        if parsed is None:
            try:
                chunk = sock.recv(4096)
            except (socket.timeout, BlockingIOError, InterruptedError):
                return None, buffer, False
            if not chunk:
                return None, buffer, True
            buffer += chunk
            continue

        opcode, payload = parsed
        if opcode == 0x8:  # close
            return None, buffer, True
        if opcode == 0x9:  # ping
            try:
                _ws_send_frame(sock, 0xA, payload)
            except Exception:
                return None, buffer, True
            continue
        if opcode == 0xA:  # pong
            continue
        if opcode != 0x1:  # text only
            continue

        line = payload.decode("utf-8", errors="replace").strip()
        if not line:
            return None, buffer, False
        try:
            msg = json.loads(line)
        except Exception:
            msg = {"type": "_invalid"}
        return msg, buffer, False


def _recv_http_headers(
    conn: socket.socket,
    initial_bytes: bytes = b"",
    max_bytes: int = 32768,
) -> Tuple[Optional[bytes], bytes]:
    buffer = bytearray(initial_bytes)
    while b"\r\n\r\n" not in buffer:
        if len(buffer) > max_bytes:
            return None, b""
        try:
            chunk = conn.recv(4096)
        except (socket.timeout, BlockingIOError, InterruptedError):
            continue
        if not chunk:
            return None, b""
        buffer.extend(chunk)
    head, tail = bytes(buffer).split(b"\r\n\r\n", 1)
    return head + b"\r\n\r\n", tail


def _ws_accept(conn: socket.socket, initial_bytes: bytes = b"") -> Tuple[bool, str, bytes]:
    header_bytes, tail = _recv_http_headers(conn, initial_bytes=initial_bytes)
    if not header_bytes:
        return False, "", b""

    text = header_bytes.decode("iso-8859-1", errors="replace")
    lines = text.split("\r\n")
    if not lines:
        return False, "", b""

    req_line = lines[0].strip()
    parts = req_line.split(" ")
    if len(parts) < 2 or parts[0].upper() != "GET":
        return False, "", b""
    path = parts[1]

    headers: Dict[str, str] = {}
    for line in lines[1:]:
        if not line or ":" not in line:
            continue
        k, v = line.split(":", 1)
        headers[k.strip().lower()] = v.strip()

    key = headers.get("sec-websocket-key", "").strip()
    upgrade = headers.get("upgrade", "").strip().lower()
    connection = headers.get("connection", "").strip().lower()
    if not key or upgrade != "websocket" or "upgrade" not in connection:
        return False, path, b""

    accept = base64.b64encode(
        hashlib.sha1((key + _WS_GUID).encode("utf-8")).digest()
    ).decode("ascii")

    response = (
        "HTTP/1.1 101 Switching Protocols\r\n"
        "Upgrade: websocket\r\n"
        "Connection: Upgrade\r\n"
        f"Sec-WebSocket-Accept: {accept}\r\n"
        "\r\n"
    )
    conn.sendall(response.encode("ascii"))
    return True, path, tail


@dataclass
class RoomRecord:
    server: Any
    protocol: str
    created_at: float
    empty_since: Optional[float] = None


class WebRoomServer:
    def __init__(
        self,
        port: int,
        room_size: int = 2,
        initial_hand_size: Optional[int] = None,
    ) -> None:
        self.port = int(port)
        self.room_size = max(2, min(8, int(room_size)))
        self.initial_hand_size = initial_hand_size

        self.listener: Optional[socket.socket] = None
        self.engine: Optional[GameEngine] = None
        self.lock = threading.Lock()
        self.running = False
        self.error: Optional[str] = None

        self.lobby_players: Dict[int, str] = {}
        self.clients: Dict[int, Dict[str, Any]] = {}
        self.reconnect_keys: Dict[int, str] = {}

    def _log(self, message: str) -> None:
        stamp = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
        print(f"[web-room:{self.port} {stamp}] {message}", flush=True)

    def start(self) -> None:
        self.listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        _set_listener_socket_opts(self.listener)
        self.listener.settimeout(0.5)
        self.listener.bind(("0.0.0.0", self.port))
        self.listener.listen(max(2, self.room_size * 2))
        self.running = True
        self._log(
            f"Listening on 0.0.0.0:{self.port}, room_size={self.room_size}, "
            f"initial_hand_size={self.initial_hand_size}"
        )
        threading.Thread(target=self._accept_loop, daemon=True).start()
        threading.Thread(target=self._heartbeat_watchdog_loop, daemon=True).start()

    def client_count(self) -> int:
        with self.lock:
            return len(self.clients)

    def _find_open_slot_locked(self) -> Optional[int]:
        for idx in range(self.room_size):
            if idx not in self.lobby_players:
                return idx
        return None

    @staticmethod
    def _new_reconnect_key() -> str:
        return secrets.token_urlsafe(12)

    def _ensure_reconnect_key_locked(self, slot: int) -> str:
        key = str(self.reconnect_keys.get(slot, "")).strip()
        if not key:
            key = self._new_reconnect_key()
            self.reconnect_keys[slot] = key
        return key

    def _resolve_reconnect_slot_locked(
        self,
        remote_name: str,
        resume_index: Optional[int],
        resume_key: str,
    ) -> Tuple[Optional[int], Optional[str]]:
        offline_slots: list[int] = []
        for idx in range(self.room_size):
            if idx not in self.lobby_players:
                continue
            if idx in self.clients:
                continue
            offline_slots.append(idx)

        if not offline_slots:
            return None, "当前没有可重连的离线座位。"

        if resume_key:
            matched = [
                idx
                for idx in offline_slots
                if str(self.reconnect_keys.get(idx, "")).strip() == resume_key
            ]
            if len(matched) == 1:
                return matched[0], None
            all_matched = [
                idx
                for idx in range(self.room_size)
                if str(self.reconnect_keys.get(idx, "")).strip() == resume_key
            ]
            if all_matched:
                return None, "该座位已在线，无法重复登录。"
            return None, "重连凭据无效，请使用原会话信息重连。"

        if resume_index is not None:
            idx = int(resume_index)
            if idx < 0 or idx >= self.room_size:
                return None, "重连座位号无效。"
            if idx not in self.lobby_players:
                return None, "该座位不存在。"
            if idx in self.clients:
                return None, "该座位已在线。"
            seat_name = str(self.lobby_players.get(idx, "")).strip()
            if seat_name and remote_name and seat_name != remote_name:
                return None, f"重连昵称不匹配，{idx + 1}号位应为“{seat_name}”。"
            return idx, None

        if remote_name:
            matched_by_name = [
                idx
                for idx in offline_slots
                if str(self.lobby_players.get(idx, "")).strip() == remote_name
            ]
            if len(matched_by_name) == 1:
                return matched_by_name[0], None
            if len(matched_by_name) > 1:
                return None, "存在重名离线玩家，请使用原会话重连。"

        return None, "对局已开始，仅支持已掉线玩家重连。"

    def _status_text_from_net_state(self, net_state: str) -> str:
        if net_state == "lagging":
            return "网卡"
        return "在线"

    def _slot_status_text(
        self,
        idx: int,
        game_started: bool,
        client_states: Dict[int, str],
    ) -> str:
        if idx in client_states:
            return self._status_text_from_net_state(client_states[idx])
        if game_started:
            return "离线"
        return ""

    def _decorate_state_with_player_status(self, state: Dict[str, Any], game_started: bool) -> None:
        players = state.get("players")
        if not isinstance(players, list):
            return
        with self.lock:
            client_states = {
                idx: str(info.get("net_state", "online"))
                for idx, info in self.clients.items()
            }
        for idx, entry in enumerate(players):
            if isinstance(entry, dict):
                entry["status"] = self._slot_status_text(
                    idx=idx,
                    game_started=game_started,
                    client_states=client_states,
                )

    def _heartbeat_watchdog_loop(self) -> None:
        while self.running:
            time.sleep(HEARTBEAT_WATCHDOG_SECONDS)
            drop_list: list[int] = []
            status_changed = False
            game_started = False
            now = time.time()
            with self.lock:
                game_started = self.engine is not None
                for idx, info in list(self.clients.items()):
                    last_seen = float(info.get("last_seen", now))
                    elapsed = now - last_seen
                    if elapsed >= HEARTBEAT_TIMEOUT_SECONDS:
                        drop_list.append(idx)
                        continue
                    expected = "lagging" if elapsed >= HEARTBEAT_LAG_SECONDS else "online"
                    if str(info.get("net_state", "online")) != expected:
                        info["net_state"] = expected
                        status_changed = True
            for idx in drop_list:
                self._drop_client(idx, " 心跳超时下线。")
            if status_changed:
                if game_started:
                    self._broadcast_state()
                else:
                    self._broadcast_lobby_state()

    def _build_lobby_state(self, viewer_idx: int, mode: str) -> Dict[str, Any]:
        with self.lock:
            lobby_players = dict(self.lobby_players)
            client_states = {
                idx: str(info.get("net_state", "online"))
                for idx, info in self.clients.items()
            }
            error_text = self.error

        players_data: list[Dict[str, Any]] = []
        for idx in range(self.room_size):
            name = str(lobby_players.get(idx, "")) if idx in lobby_players else ""
            entry: Dict[str, Any] = {
                "name": name,
                "hand_size": "",
                "status": self._slot_status_text(
                    idx=idx,
                    game_started=False,
                    client_states=client_states,
                ),
            }
            if idx == viewer_idx and idx in lobby_players:
                entry["hand"] = {}
            players_data.append(entry)

        joined = len(lobby_players)
        left = self.room_size - joined
        if left > 0:
            wait_msg = f"房间等待中：已加入 {joined}/{self.room_size}，还需 {left} 人。"
        else:
            wait_msg = "房间已满，正在开始对局..."

        logs: list[str] = []
        if error_text:
            logs.append(error_text)

        return {
            "mode": mode,
            "game_started": False,
            "message": wait_msg,
            "you_index": viewer_idx,
            "current_player": None,
            "actions_left": 0,
            "direction": 1,
            "turn_number": 0,
            "winner": None,
            "deck_size": 0,
            "discard_size": 0,
            "can_act": False,
            "room_size": self.room_size,
            "joined_count": joined,
            "players": players_data,
            "board": {},
            "log": logs,
        }

    def _broadcast_lobby_state(self) -> None:
        with self.lock:
            clients_snapshot = {
                idx: info["sock"]
                for idx, info in self.clients.items()
            }

        for idx, sock in clients_snapshot.items():
            try:
                state = self._build_lobby_state(viewer_idx=idx, mode="client")
                _ws_send_json(sock, {"type": "state", "state": state})
            except Exception:
                pass

    def _try_start_game(self) -> None:
        start_game = False
        with self.lock:
            if self.engine is None and len(self.lobby_players) == self.room_size:
                names = [self.lobby_players[i] for i in range(self.room_size)]
                self.engine = GameEngine(names, initial_hand_size=self.initial_hand_size)
                start_game = True
        if start_game:
            self._broadcast_state()
    def _accept_loop(self) -> None:
        if self.listener is None:
            return
        while self.running:
            try:
                conn, _ = self.listener.accept()
            except socket.timeout:
                continue
            except Exception as exc:
                self.error = f"监听异常: {exc}"
                self._log(self.error)
                return

            try:
                self._log("Incoming TCP connection accepted.")
                conn.settimeout(0.4)
                ok, _path, buffer = _ws_accept(conn)
                if not ok:
                    self._log("WebSocket handshake rejected.")
                    conn.close()
                    continue
                self._log("WebSocket handshake success, waiting for join.")

                join_msg: Optional[Dict[str, Any]] = None
                deadline = time.time() + 8.0
                while self.running and time.time() < deadline:
                    msg, buffer, closed = _ws_recv_json_nonblocking(conn, buffer)
                    if closed:
                        conn.close()
                        join_msg = None
                        break
                    if msg is None:
                        continue
                    join_msg = msg
                    break

                if join_msg is None:
                    try:
                        _ws_send_json(conn, {"type": "error", "message": "Join timeout."})
                    except Exception:
                        pass
                    self._log("Join timeout, closing connection.")
                    conn.close()
                    continue

                if join_msg.get("type") != "join":
                    _ws_send_json(conn, {"type": "error", "message": "首条消息必须是 join。"})
                    self._log(f"First message is not join: {join_msg}")
                    conn.close()
                    continue

                remote_name = str(join_msg.get("name", "玩家")).strip() or "玩家"
                raw_resume_index = join_msg.get("resume_index")
                resume_index: Optional[int]
                if raw_resume_index is None:
                    resume_index = None
                else:
                    try:
                        resume_index = int(raw_resume_index)
                    except Exception:
                        resume_index = None
                resume_key = str(join_msg.get("resume_key", "")).strip()

                reject_msg: Optional[str] = None
                slot: Optional[int] = None
                reconnect_key = ""
                game_started = False
                is_rejoin = False

                with self.lock:
                    game_started = self.engine is not None
                    if game_started:
                        slot, reject_msg = self._resolve_reconnect_slot_locked(
                            remote_name=remote_name,
                            resume_index=resume_index,
                            resume_key=resume_key,
                        )
                        if slot is not None:
                            seat_name = str(self.lobby_players.get(slot, remote_name)).strip() or remote_name
                            now = time.time()
                            reconnect_key = self._ensure_reconnect_key_locked(slot)
                            self.clients[slot] = {
                                "sock": conn,
                                "buffer": buffer,
                                "name": seat_name,
                                "last_seen": now,
                                "net_state": "online",
                            }
                            is_rejoin = True
                            if self.engine is not None:
                                self.engine.skip_turns.pop(slot, None)
                                self.engine._log(f"{seat_name} 已重连并回到 {slot + 1} 号位。")
                            self.error = f"{seat_name} 已重连。"
                    elif len(self.lobby_players) >= self.room_size:
                        reject_msg = "房间已满。"
                    else:
                        slot = self._find_open_slot_locked()
                        if slot is None:
                            reject_msg = "房间暂无可用座位。"
                        else:
                            now = time.time()
                            reconnect_key = self._ensure_reconnect_key_locked(slot)
                            self.lobby_players[slot] = remote_name
                            self.clients[slot] = {
                                "sock": conn,
                                "buffer": buffer,
                                "name": remote_name,
                                "last_seen": now,
                                "net_state": "online",
                            }

                if reject_msg is not None or slot is None:
                    _ws_send_json(conn, {"type": "error", "message": reject_msg or "加入房间失败。"})
                    self._log(f"Join rejected: {reject_msg or 'unknown'}")
                    conn.close()
                    continue

                _ws_send_json(
                    conn,
                    {
                        "type": "welcome",
                        "player_index": slot,
                        "room_size": self.room_size,
                        "reconnect_key": reconnect_key,
                        "rejoined": is_rejoin,
                    },
                )
                self._log(
                    f"Join accepted: slot={slot}, name={self.lobby_players.get(slot, remote_name)}, "
                    f"rejoined={is_rejoin}"
                )
                threading.Thread(target=self._client_loop, args=(slot,), daemon=True).start()
                if game_started:
                    self._broadcast_state()
                else:
                    self._broadcast_lobby_state()
                    self._try_start_game()
            except Exception as exc:
                self.error = f"接入异常: {exc}"
                self._log(self.error)
                try:
                    conn.close()
                except Exception:
                    pass

    def _drop_client(self, player_idx: int, disconnected_text: str) -> None:
        client: Optional[Dict[str, Any]] = None
        game_started = False
        name = f"玩家{player_idx + 1}"
        advance_turn = False

        with self.lock:
            client = self.clients.pop(player_idx, None)
            game_started = self.engine is not None
            if client is not None:
                name = str(client.get("name", name))

            if not game_started and player_idx in self.lobby_players:
                self.lobby_players.pop(player_idx, None)
                self.reconnect_keys.pop(player_idx, None)

            if game_started and self.engine is not None:
                self.engine.skip_turns[player_idx] = 10**9
                self.engine._log(f"{name} 掉线，后续回合将自动跳过。")
                self.engine.on_player_disconnected(player_idx)
                if (
                    self.engine.current_player == player_idx
                    and self.engine._pending_waiting_player() is None
                ):
                    advance_turn = True

            self.error = f"{name}{disconnected_text}"

            if advance_turn and self.engine is not None and self.engine.winner is None:
                self.engine._end_turn_internal()

        if client is not None:
            sock = client.get("sock")
            if isinstance(sock, socket.socket):
                try:
                    sock.shutdown(socket.SHUT_RDWR)
                except Exception:
                    pass
                try:
                    sock.close()
                except Exception:
                    pass

        if game_started:
            self._broadcast_state()
        else:
            self._broadcast_lobby_state()

    def _client_loop(self, player_idx: int) -> None:
        while self.running:
            with self.lock:
                client = self.clients.get(player_idx)
                if client is None:
                    return
                conn = client["sock"]
                buffer = client["buffer"]

            try:
                msg, buffer, closed = _ws_recv_json_nonblocking(conn, buffer)
                status_recovered = False
                game_started = False
                with self.lock:
                    if player_idx in self.clients:
                        info = self.clients[player_idx]
                        info["buffer"] = buffer
                        if msg is not None:
                            info["last_seen"] = time.time()
                            if str(info.get("net_state", "online")) != "online":
                                info["net_state"] = "online"
                                status_recovered = True
                        game_started = self.engine is not None

                if closed:
                    self._drop_client(player_idx, " 已断开连接。")
                    return
                if msg is None:
                    continue

                if status_recovered:
                    if game_started:
                        self._broadcast_state()
                    else:
                        self._broadcast_lobby_state()

                mtype = msg.get("type")
                if mtype == "h":
                    continue
                if mtype == "bye":
                    self._drop_client(player_idx, " 已结束会话。")
                    return
                if mtype == "action":
                    self.apply_action(player_idx, msg.get("data", {}))
                elif mtype == "_invalid":
                    _ws_send_json(conn, {"type": "error", "message": "JSON 格式无效。"})
            except Exception:
                self._drop_client(player_idx, " 网络异常断开。")
                return

    def _broadcast_state(self) -> None:
        with self.lock:
            engine = self.engine
            if engine is None:
                return
            clients_snapshot = {
                idx: info["sock"]
                for idx, info in self.clients.items()
            }

        for idx, sock in clients_snapshot.items():
            try:
                client_state = engine.serialize_state(
                    viewer_idx=idx,
                    reveal_all_hands=False,
                    mode="client",
                )
                client_state["room_size"] = self.room_size
                self._decorate_state_with_player_status(client_state, game_started=True)
                _ws_send_json(sock, {"type": "state", "state": client_state})
            except Exception:
                pass

    def apply_action(self, player_idx: int, action: Dict[str, Any]) -> Tuple[bool, str]:
        with self.lock:
            if self.engine is None:
                return False, "房间人数未满，无法出牌。"
            ok, msg = self.engine.apply_action(player_idx, action)
            err_sock: Optional[socket.socket] = None
            if not ok:
                client = self.clients.get(player_idx)
                if client is not None:
                    err_sock = client.get("sock")

        if ok:
            self._broadcast_state()
        elif err_sock is not None:
            try:
                _ws_send_json(err_sock, {"type": "error", "message": msg})
            except Exception:
                pass
        return ok, msg

    def stop(self) -> None:
        self.running = False
        self._log("Stopping room server.")
        with self.lock:
            clients = list(self.clients.values())
            self.clients.clear()

        for client in clients:
            sock = client.get("sock")
            if isinstance(sock, socket.socket):
                try:
                    sock.shutdown(socket.SHUT_RDWR)
                except Exception:
                    pass
                try:
                    sock.close()
                except Exception:
                    pass

        if self.listener is not None:
            try:
                self.listener.close()
            except Exception:
                pass


class MainPortServer:
    def __init__(self) -> None:
        self.listener: Optional[socket.socket] = None
        self.running = False
        self.lock = threading.Lock()
        self.rooms: Dict[int, RoomRecord] = {}

    @staticmethod
    def _log(message: str) -> None:
        stamp = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
        print(f"[server {stamp}] {message}", flush=True)

    def start(self) -> None:
        self.listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        _set_listener_socket_opts(self.listener)
        self.listener.settimeout(0.5)
        self.listener.bind(("0.0.0.0", MAIN_PORT))
        self.listener.listen(128)
        self.running = True
        threading.Thread(target=self._accept_loop, daemon=True).start()
        threading.Thread(target=self._watchdog_loop, daemon=True).start()
        self._log(f"Main port listening on 0.0.0.0:{MAIN_PORT}")
        self._log(
            f"Room port range: {ROOM_PORT_MIN}-{ROOM_PORT_MAX} "
            f"(excluding {MAIN_PORT}), reclaim idle after {int(ROOM_EMPTY_RECLAIM_SECONDS)}s"
        )
    def stop(self) -> None:
        self.running = False
        if self.listener is not None:
            try:
                self.listener.close()
            except Exception:
                pass
        with self.lock:
            records = list(self.rooms.values())
            self.rooms.clear()
        for rec in records:
            try:
                rec.server.stop()
            except Exception:
                pass
        self._log("Stopped all room servers.")

    @staticmethod
    def _can_bind_port(port: int) -> bool:
        probe = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        _set_listener_socket_opts(probe)
        probe.settimeout(0.3)
        try:
            probe.bind(("0.0.0.0", port))
            return True
        except OSError:
            return False
        finally:
            try:
                probe.close()
            except Exception:
                pass

    @staticmethod
    def _room_client_count(room: Any) -> int:
        if isinstance(room, HostServer):
            with room.lock:
                return len(room.clients)
        if isinstance(room, WebRoomServer):
            return room.client_count()
        fn = getattr(room, "client_count", None)
        if callable(fn):
            try:
                return int(fn())
            except Exception:
                return 0
        return 0

    def _cleanup_stopped_rooms_locked(self) -> None:
        dead_ports = [
            port
            for port, rec in self.rooms.items()
            if not bool(getattr(rec.server, "running", False))
        ]
        for port in dead_ports:
            self.rooms.pop(port, None)
            self._log(f"Removed stopped room record: {port}")

    def _watchdog_loop(self) -> None:
        while self.running:
            time.sleep(WATCHDOG_INTERVAL_SECONDS)
            now = time.time()
            to_reclaim: list[tuple[int, Any, str]] = []

            with self.lock:
                self._cleanup_stopped_rooms_locked()
                for port, rec in list(self.rooms.items()):
                    room = rec.server
                    if not bool(getattr(room, "running", False)):
                        self.rooms.pop(port, None)
                        self._log(f"Removed non-running room: {port}")
                        continue

                    client_count = self._room_client_count(room)
                    if client_count > 0:
                        rec.empty_since = None
                        continue

                    if rec.empty_since is None:
                        rec.empty_since = now
                        continue

                    if now - rec.empty_since >= ROOM_EMPTY_RECLAIM_SECONDS:
                        self.rooms.pop(port, None)
                        to_reclaim.append((port, room, rec.protocol))

            for port, room, proto in to_reclaim:
                try:
                    room.stop()
                except Exception:
                    pass
                self._log(
                    f"Reclaimed {proto} room port {port} "
                    f"(no clients for {int(ROOM_EMPTY_RECLAIM_SECONDS)}s)."
                )

    def _allocate_room_port(self, room_size: int, initial_hand_size: int, protocol: str) -> int:
        p = str(protocol or "tcp").strip().lower()
        if p not in {"tcp", "web"}:
            raise RuntimeError(f"Unsupported protocol: {p}")

        for _ in range(MAX_ALLOC_ATTEMPTS):
            port = random.randint(ROOM_PORT_MIN, ROOM_PORT_MAX)
            if port == MAIN_PORT:
                continue
            with self.lock:
                self._cleanup_stopped_rooms_locked()
                if port in self.rooms:
                    continue
            if not self._can_bind_port(port):
                continue

            if p == "web":
                room: Any = WebRoomServer(
                    port=port,
                    room_size=room_size,
                    initial_hand_size=initial_hand_size,
                )
            else:
                room = HostServer(
                    host_name="",
                    port=port,
                    room_size=room_size,
                    initial_hand_size=initial_hand_size,
                    include_host_player=False,
                )
            try:
                room.start()
            except Exception:
                continue

            with self.lock:
                if port in self.rooms:
                    try:
                        room.stop()
                    except Exception:
                        pass
                    continue
                self.rooms[port] = RoomRecord(
                    server=room,
                    protocol=p,
                    created_at=time.time(),
                    empty_since=time.time(),
                )
            self._log(
                f"Created {p} room: port={port}, room_size={room_size}, "
                f"initial_hand_size={initial_hand_size}"
            )
            return port

        raise RuntimeError("No available room port could be allocated.")

    @staticmethod
    def _peer_text(conn: socket.socket) -> str:
        try:
            host, port = conn.getpeername()
            return f"{host}:{port}"
        except Exception:
            return "unknown"

    def _accept_loop(self) -> None:
        if self.listener is None:
            return
        while self.running:
            try:
                conn, _addr = self.listener.accept()
            except socket.timeout:
                continue
            except Exception as exc:
                if self.running:
                    self._log(f"Main accept loop error: {exc}")
                    time.sleep(0.05)
                continue
            threading.Thread(target=self._handle_conn, args=(conn,), daemon=True).start()

    def _handle_conn(self, conn: socket.socket) -> None:
        peer = self._peer_text(conn)
        conn.settimeout(0.4)
        self._log(f"Control request connected: {peer}")
        try:
            initial = b""
            deadline = time.time() + 3.0
            while len(initial) < 3 and time.time() < deadline:
                try:
                    chunk = conn.recv(3 - len(initial))
                except socket.timeout:
                    continue
                if not chunk:
                    return
                initial += chunk
            if not initial:
                self._log(f"Control request empty preface from {peer}")
                return

            probe = initial.lstrip()
            probe_up = probe[:8].upper()
            is_ws = False
            if probe_up.startswith(b"GET"):
                is_ws = True
            elif probe_up.startswith(b"{") or probe_up.startswith(b"["):
                is_ws = False
            elif b"GET" in probe_up:
                # Tolerate uncommon splitting/leading bytes; browser WS control is always GET.
                is_ws = True

            self._log(
                f"Control preface from {peer}: {initial!r} "
                f"-> {'WS' if is_ws else 'TCP'}"
            )
            if is_ws:
                self._handle_ws_control_conn(conn, initial, peer)
            else:
                self._handle_tcp_control_conn(conn, initial, peer)
        finally:
            try:
                conn.close()
            except Exception:
                pass

    def _parse_create_room_payload(self, req: Dict[str, Any]) -> Tuple[int, int, str]:
        try:
            room_size = int(req.get("room_size", 3))
        except Exception:
            raise ValueError("Invalid room_size.")
        if room_size < 2 or room_size > 8:
            raise ValueError("room_size must be in [2, 8].")

        try:
            initial_hand_size = int(req.get("initial_hand_size", 7))
        except Exception:
            raise ValueError("Invalid initial_hand_size.")
        if initial_hand_size < 2 or initial_hand_size > 10:
            raise ValueError("initial_hand_size must be in [2, 10].")

        protocol = str(req.get("protocol", "tcp")).strip().lower() or "tcp"
        if protocol not in {"tcp", "web"}:
            raise ValueError("protocol must be 'tcp' or 'web'.")

        return room_size, initial_hand_size, protocol

    def _handle_tcp_control_conn(self, conn: socket.socket, initial: bytes, peer: str) -> None:
        buffer = initial
        req = None
        deadline = time.time() + 8.0
        while time.time() < deadline:
            msg, buffer, closed = _recv_json_nonblocking(conn, buffer)
            if closed:
                self._log(f"TCP control closed before payload: {peer}")
                return
            if msg is None:
                time.sleep(0.02)
                continue
            req = msg
            break

        if req is None:
            _send_json(conn, {"type": "error", "message": "Request timeout."})
            self._log(f"TCP control request timeout: {peer}")
            return

        if str(req.get("type", "")).strip() != "create_room":
            _send_json(conn, {"type": "error", "message": "Unsupported request type."})
            self._log(f"Unsupported TCP control request from {peer}: {req}")
            return

        try:
            room_size, initial_hand_size, protocol = self._parse_create_room_payload(req)
        except ValueError as exc:
            _send_json(conn, {"type": "error", "message": str(exc)})
            self._log(f"Invalid TCP create_room from {peer}: {exc}")
            return

        try:
            port = self._allocate_room_port(room_size, initial_hand_size, protocol)
        except Exception as exc:
            _send_json(conn, {"type": "error", "message": f"Create room failed: {exc}"})
            self._log(f"Create room failed for {peer}: {exc}")
            return

        _send_json(
            conn,
            {
                "type": "room_created",
                "port": port,
                "main_port": MAIN_PORT,
                "protocol": protocol,
                "room_size": room_size,
                "initial_hand_size": initial_hand_size,
            },
        )
        self._log(f"Assigned {protocol} room port {port} to {peer} (TCP control)")

    def _handle_ws_control_conn(self, conn: socket.socket, initial: bytes, peer: str) -> None:
        ok, _path, buffer = _ws_accept(conn, initial)
        if not ok:
            self._log(f"WS control handshake failed: {peer}")
            return

        req = None
        deadline = time.time() + 8.0
        while time.time() < deadline:
            msg, buffer, closed = _ws_recv_json_nonblocking(conn, buffer)
            if closed:
                self._log(f"WS control closed before payload: {peer}")
                return
            if msg is None:
                time.sleep(0.02)
                continue
            req = msg
            break

        if req is None:
            _ws_send_json(conn, {"type": "error", "message": "Request timeout."})
            self._log(f"WS control request timeout: {peer}")
            return

        if str(req.get("type", "")).strip() != "create_room":
            _ws_send_json(conn, {"type": "error", "message": "Unsupported request type."})
            self._log(f"Unsupported WS control request from {peer}: {req}")
            return

        # Web control defaults to web rooms if protocol not explicitly set.
        if "protocol" not in req:
            req = dict(req)
            req["protocol"] = "web"

        try:
            room_size, initial_hand_size, protocol = self._parse_create_room_payload(req)
        except ValueError as exc:
            _ws_send_json(conn, {"type": "error", "message": str(exc)})
            self._log(f"Invalid WS create_room from {peer}: {exc}")
            return

        try:
            port = self._allocate_room_port(room_size, initial_hand_size, protocol)
        except Exception as exc:
            _ws_send_json(conn, {"type": "error", "message": f"Create room failed: {exc}"})
            self._log(f"Create room failed for {peer}: {exc}")
            return

        _ws_send_json(
            conn,
            {
                "type": "room_created",
                "port": port,
                "main_port": MAIN_PORT,
                "protocol": protocol,
                "room_size": room_size,
                "initial_hand_size": initial_hand_size,
            },
        )
        self._log(f"Assigned {protocol} room port {port} to {peer} (WS control)")


def main() -> None:
    server = MainPortServer()
    server.start()
    try:
        while True:
            time.sleep(1.0)
    except KeyboardInterrupt:
        pass
    finally:
        server.stop()
        print("[server] Stopped.", flush=True)


if __name__ == "__main__":
    main()