Commonwealth-Online-Server/server/server_core.py

from __future__ import annotations

import json
import math
import socket
import sys
import threading
import time
from collections.abc import Callable
from typing import Any

from ban_store import BanStore
from client_session import ClientSession
from lan_discovery import DISCOVERY_PORT, LanDiscoveryResponder
from world_state_presets import (
    normalize_fw_console_arg,
    relay_weather_form_id,
)


HOST = "0.0.0.0"
PORT = 7777
ACCEPT_TIMEOUT_SECONDS = 0.5
DEFAULT_SERVER_NAME = "Commonwealth Online Server"
DEFAULT_MAX_PLAYERS = 16
DEFAULT_LOG_VERBOSITY = "info"
SESSION_ENDED_BANNED = "banned"
SESSION_ENDED_KICKED = "kicked"

_LOG_LEVELS = {
    "debug": 10,
    "info": 20,
    "warning": 30,
    "error": 40,
}


def get_lan_addresses() -> list[str]:
    """Return likely non-loopback IPv4 addresses for this machine."""
    addresses: list[str] = []

    try:
        with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as probe:
            probe.connect(("8.8.8.8", 80))
            primary = probe.getsockname()[0]
            if primary and not primary.startswith("127."):
                addresses.append(primary)
    except OSError:
        pass

    try:
        hostname = socket.gethostname()
        for info in socket.getaddrinfo(hostname, None, family=socket.AF_INET):
            candidate = info[4][0]
            if candidate and not candidate.startswith("127.") and candidate not in addresses:
                addresses.append(candidate)
    except OSError:
        pass

    return addresses


class FalloutTogetherServer:
    def __init__(
        self,
        host: str = HOST,
        port: int = PORT,
        server_name: str = DEFAULT_SERVER_NAME,
        server_description: str = "",
        max_players: int = DEFAULT_MAX_PLAYERS,
        bans_path: str | None = None,
        log_verbosity: str = DEFAULT_LOG_VERBOSITY,
    ) -> None:
        self.host = host
        self.port = port
        self.server_name = server_name or DEFAULT_SERVER_NAME
        self.server_description = server_description or ""
        self.max_players = max_players if max_players >= 1 else DEFAULT_MAX_PLAYERS
        verbosity = str(log_verbosity or DEFAULT_LOG_VERBOSITY).strip().lower()
        self.log_verbosity = verbosity if verbosity in _LOG_LEVELS else DEFAULT_LOG_VERBOSITY
        self._ban_store = BanStore(bans_path)

        self._lock = threading.RLock()
        self._log_lock = threading.RLock()
        self._clients: dict[socket.socket, ClientSession] = {}
        self._client_threads: set[threading.Thread] = set()
        self._log_listeners: list[Callable[[str], None]] = []
        self._server_socket: socket.socket | None = None
        self._accept_thread: threading.Thread | None = None
        self._discovery: LanDiscoveryResponder | None = None
        self._running = False
        self._started_at: float | None = None
        self._next_player_id = 1
        self._world_state_host_player_id: int | None = None
        self._server_world_state: dict[str, str] = {}
        self._last_npc_state: dict[str, Any] | None = None

        self._stats: dict[str, int] = {
            "clientsConnected": 0,
            "clientsDisconnected": 0,
            "packetsReceived": 0,
            "packetsSent": 0,
            "packetsBroadcast": 0,
            "transformPacketsReceived": 0,
            "transformPacketsBroadcast": 0,
            "worldStatePacketsReceived": 0,
            "worldStatePacketsBroadcast": 0,
            "npcStatePacketsReceived": 0,
            "npcStatePacketsBroadcast": 0,
            "combatHitsReceived": 0,
            "combatHitsRouted": 0,
            "worldStateHostPacketsBroadcast": 0,
            "serverWorldStatePacketsBroadcast": 0,
            "sessionEndedPacketsSent": 0,
            "bannedConnectionsRejected": 0,
            "disconnectPacketsBroadcast": 0,
        }

    def start(self) -> None:
        with self._lock:
            if self._running:
                return

            self._prepare_server_socket()
            self._start_discovery()
            self._accept_thread = threading.Thread(target=self._accept_loop, daemon=True)
            self._accept_thread.start()

    def serve_forever(self) -> None:
        with self._lock:
            if self._running:
                raise RuntimeError("Server is already running.")

            self._prepare_server_socket()
            self._start_discovery()

        self._accept_loop()

    def stop(self) -> None:
        clients: list[ClientSession]
        server_socket: socket.socket | None
        client_threads: list[threading.Thread]

        with self._lock:
            if not self._running and self._server_socket is None and self._discovery is None:
                return
            self._running = False
            server_socket = self._server_socket
            self._server_socket = None
            clients = list(self._clients.values())
            client_threads = [
                thread for thread in self._client_threads if thread is not threading.current_thread()
            ]

        discovery = self._discovery
        self._discovery = None
        if discovery is not None:
            discovery.stop()

        if server_socket is not None:
            try:
                server_socket.close()
            except OSError:
                pass

        for client in clients:
            self._close_client_socket(client)

        accept_thread = self._accept_thread
        if accept_thread is not None and accept_thread is not threading.current_thread():
            accept_thread.join(timeout=1.0)
        self._accept_thread = None

        for thread in client_threads:
            thread.join(timeout=1.0)

        with self._lock:
            self._client_threads.clear()

    def is_running(self) -> bool:
        with self._lock:
            return self._running

    def get_clients(self) -> list[dict[str, Any]]:
        with self._lock:
            clients = list(self._clients.values())

        return [client.to_snapshot() for client in clients]

    def get_stats(self) -> dict[str, Any]:
        with self._lock:
            connected_clients = len(self._clients)
            started_at = self._started_at
            stats = dict(self._stats)
            stats.update(
                {
                    "host": self.host,
                    "port": self.port,
                    "serverName": self.server_name,
                    "serverDescription": self.server_description,
                    "maxPlayers": self.max_players,
                    "lanAddresses": get_lan_addresses() if self._running and self.host == "0.0.0.0" else [],
                    "isRunning": self._running,
                    "startedAt": started_at,
                    "uptimeSeconds": time.time() - started_at if started_at is not None else 0.0,
                    "connectedClients": connected_clients,
                    "nextPlayerId": self._next_player_id,
                }
            )

        return stats

    def get_server_world_state(self) -> dict[str, str]:
        with self._lock:
            return dict(self._server_world_state)

    def set_server_time(self, hhmm: str) -> bool:
        text = str(hhmm).strip()
        if not text.isdigit() or len(text) > 4:
            self._log(f"Rejected invalid server time HHmm value: {hhmm!r}")
            return False

        normalized = text.zfill(4)
        hours = int(normalized[:2])
        minutes = int(normalized[2:])
        if hours > 23 or minutes > 59:
            self._log(f"Rejected out-of-range server time HHmm value: {hhmm!r}")
            return False

        with self._lock:
            self._server_world_state["timeHHmm"] = normalized

        self._broadcast_server_world_state()
        return True

    def set_server_weather(self, fw_console_arg: str) -> bool:
        try:
            normalized_fw_arg = normalize_fw_console_arg(fw_console_arg)
        except ValueError:
            self._log(f"Rejected invalid server weather console id: {fw_console_arg!r}")
            return False

        with self._lock:
            self._server_world_state["weatherConsoleArg"] = normalized_fw_arg
            self._server_world_state["weatherFormId"] = relay_weather_form_id(normalized_fw_arg)

        self._broadcast_server_world_state()
        return True

    def list_bans(self) -> list[dict[str, Any]]:
        return [
            {
                "ip": entry.ip,
                "reason": entry.reason,
                "bannedAt": entry.banned_at,
            }
            for entry in self._ban_store.list_bans()
        ]

    def unban_ip(self, ip: str) -> bool:
        normalized = str(ip).strip()
        removed = self._ban_store.unban_ip(normalized)
        if removed:
            self._log(f"Unbanned IP {normalized}")
        return removed

    def ban_ip(self, ip: str, reason: str = "") -> dict[str, Any]:
        normalized = str(ip).strip()
        if not normalized:
            raise ValueError("IP address cannot be empty.")

        entry = self._ban_store.ban_ip(normalized, reason=reason)
        self._log(
            f"Banned IP {entry.ip}"
            + (f" (reason: {entry.reason})" if entry.reason else "")
        )
        ended = self._end_sessions_for_ip(normalized, code=SESSION_ENDED_BANNED, reason=entry.reason)
        return {
            "ip": entry.ip,
            "reason": entry.reason,
            "bannedAt": entry.banned_at,
            "sessionsEnded": ended,
        }

    def ban_player(self, player_id: int, reason: str = "") -> dict[str, Any]:
        client = self._find_client_by_player_id(player_id)
        if client is None:
            raise KeyError(f"No connected player with id {player_id}")
        return self.ban_ip(client.address[0], reason=reason)

    def kick_player(self, player_id: int, reason: str = "") -> dict[str, Any]:
        client = self._find_client_by_player_id(player_id)
        if client is None:
            raise KeyError(f"No connected player with id {player_id}")

        ip = client.address[0]
        self._log(
            f"Kicking player {player_id} ({client.label})"
            + (f" (reason: {reason})" if reason else "")
        )
        self._end_client_session(client, code=SESSION_ENDED_KICKED, reason=reason)
        return {
            "playerId": player_id,
            "ip": ip,
            "code": SESSION_ENDED_KICKED,
            "reason": str(reason or ""),
        }

    def add_log_listener(self, callback: Callable[[str], None]) -> None:
        with self._log_lock:
            if callback not in self._log_listeners:
                self._log_listeners.append(callback)

    def remove_log_listener(self, callback: Callable[[str], None]) -> None:
        with self._log_lock:
            if callback in self._log_listeners:
                self._log_listeners.remove(callback)

    def _prepare_server_socket(self) -> None:
        server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        try:
            if hasattr(socket, "SO_EXCLUSIVEADDRUSE"):
                server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_EXCLUSIVEADDRUSE, 1)
            else:
                server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            server_socket.bind((self.host, self.port))
            server_socket.listen()
            server_socket.settimeout(ACCEPT_TIMEOUT_SECONDS)
        except OSError as error:
            server_socket.close()
            raise OSError(
                f"Could not bind the server to {self.host}:{self.port}. "
                f"The port may already be in use or unavailable. ({error})"
            ) from error

        self._server_socket = server_socket
        self._running = True
        self._started_at = time.time()
        # Each server run is a fresh session for player/host assignment.
        # Without this reset, restarts can leave stale host IDs and player IDs
        # > 1, which breaks weather authority handoff to the active host.
        self._next_player_id = 1
        self._world_state_host_player_id = None
        self._last_npc_state = None

        self._log(f"Commonwealth Online server listening on {self.host}:{self.port}")
        if self.host == "0.0.0.0":
            self._log(
                f"Bind address 0.0.0.0 means all interfaces; clients should not join 0.0.0.0."
            )
            self._log(f"Local clients can connect at 127.0.0.1:{self.port}")
            lan_addresses = get_lan_addresses()
            for address in lan_addresses:
                self._log(f"LAN clients can connect at {address}:{self.port}")
            if not lan_addresses:
                self._log("Could not detect a LAN IPv4 address for this machine.")
            if sys.platform == "win32":
                self._log(
                    "Forward this port on your router and allow it through Windows Firewall "
                    "for connections from outside your network."
                )
            else:
                self._log(
                    "Allow TCP "
                    f"{self.port} through your host firewall. Router port forwarding is only "
                    "needed for connections from outside your LAN."
                )
        else:
            self._log(f"Clients can connect at {self.host}:{self.port}")
        self._log("Waiting for newline-separated JSON transform packets...")

    def _start_discovery(self) -> None:
        discovery = LanDiscoveryResponder(self)
        try:
            discovery.start()
        except OSError as error:
            self._log(
                f"LAN discovery unavailable on UDP {DISCOVERY_PORT}: {error}. "
                "Direct TCP connections still work.",
                level="warning",
            )
            self._discovery = None
            return

        self._discovery = discovery
        self._log(f"LAN discovery listening on UDP port {DISCOVERY_PORT}")

    def _accept_loop(self) -> None:
        try:
            while self.is_running():
                with self._lock:
                    server_socket = self._server_socket

                if server_socket is None:
                    break

                try:
                    connection, address = server_socket.accept()
                except socket.timeout:
                    # Wake periodically so Ctrl+C is handled promptly in Windows terminals.
                    continue
                except OSError:
                    if self.is_running():
                        self._log("Server socket closed unexpectedly.")
                    break

                thread = threading.Thread(
                    target=self._handle_client,
                    args=(connection, address),
                    daemon=True,
                )
                with self._lock:
                    self._client_threads.add(thread)
                thread.start()
        finally:
            with self._lock:
                server_socket = self._server_socket
                self._running = False
                self._server_socket = None

            if server_socket is not None:
                try:
                    server_socket.close()
                except OSError:
                    pass

    def _assign_client(self, connection: socket.socket, address: tuple[str, int]) -> ClientSession:
        with self._lock:
            # The server owns player IDs so early clients do not need to coordinate
            # identity with each other or know anything about remote players yet.
            client = ClientSession(
                connection=connection,
                address=address,
                player_id=self._next_player_id,
                connected_at=time.time(),
            )
            self._next_player_id += 1
            self._clients[connection] = client
            self._stats["clientsConnected"] += 1
            if self._world_state_host_player_id is None:
                self._world_state_host_player_id = client.player_id

        return client

    def _remove_client(self, client: ClientSession) -> bool:
        with self._lock:
            removed = self._clients.pop(client.connection, None) is not None
            if removed:
                self._stats["clientsDisconnected"] += 1
            return removed

    def _handle_client(self, connection: socket.socket, address: tuple[str, int]) -> None:
        peer_ip = address[0]
        ban_entry = self._ban_store.get_ban(peer_ip)
        if ban_entry is not None:
            with self._lock:
                self._stats["bannedConnectionsRejected"] += 1
            self._log(f"Rejected banned IP {peer_ip}:{address[1]}")
            self._send_session_ended_raw(
                connection,
                code=SESSION_ENDED_BANNED,
                reason=ban_entry.reason,
            )
            try:
                connection.close()
            except OSError:
                pass
            return

        client = self._assign_client(connection, address)
        self._log(
            f"TCP accept: {client.label} (player {client.player_id})",
            level="debug",
        )

        try:
            with connection:
                buffer = b""

                try:
                    with self._lock:
                        world_state_host_player_id = self._world_state_host_player_id

                    welcome_packet: dict[str, Any] = {
                        "type": "welcome",
                        "playerId": client.player_id,
                        "serverTime": time.time(),
                        "serverName": self.server_name,
                    }
                    if self.server_description:
                        welcome_packet["serverDescription"] = self.server_description
                    if world_state_host_player_id is not None:
                        welcome_packet["worldStateHostPlayerId"] = world_state_host_player_id

                    self._send_packet(client, welcome_packet)
                    # Delay transform/npc snapshots until the client sends its first
                    # packet so TCP ping probes (connect+close) do not pull world state.

                    while self.is_running():
                        chunk = connection.recv(4096)
                        if not chunk:
                            break

                        buffer += chunk
                        while b"\n" in buffer:
                            line_bytes, buffer = buffer.split(b"\n", 1)
                            line = line_bytes.decode("utf-8", errors="replace").strip()
                            self._handle_line(client, line)
                except ConnectionResetError:
                    if client.packets_received > 0:
                        self._log(
                            f"Client disconnected unexpectedly: {client.label} "
                            f"(player {client.player_id})"
                        )
                    else:
                        self._log(
                            f"Short-lived connection closed: {client.label} "
                            f"(player {client.player_id})",
                            level="debug",
                        )
                except OSError as error:
                    if self.is_running():
                        self._log(
                            f"Client connection error: {client.label} "
                            f"(player {client.player_id}): {error}"
                        )
                finally:
                    self._disconnect_client(client)
        finally:
            with self._lock:
                self._client_threads.discard(threading.current_thread())

    def _handle_line(self, client: ClientSession, line: str) -> None:
        if not line:
            return

        received_at = time.time()
        packets_received = client.record_received(received_at)
        if packets_received == 1:
            self._log(f"Client connected: {client.label} (player {client.player_id})")
            self._send_existing_transforms_to_client(client)
            self._send_existing_npc_state_to_client(client)

        with self._lock:
            self._stats["packetsReceived"] += 1

        try:
            packet = json.loads(line)
        except json.JSONDecodeError as error:
            self._log(f"Invalid JSON from {client.label}: {error}: {line}")
            return

        if not isinstance(packet, dict):
            self._log(f"Invalid packet from {client.label}: expected JSON object: {packet}")
            return

        if packet.get("type") == "transform":
            packet["playerId"] = client.player_id
            packet["serverTime"] = time.time()
            client.record_transform(packet)
            with self._lock:
                self._stats["transformPacketsReceived"] += 1

        if packet.get("type") == "worldState":
            with self._lock:
                world_state_host_player_id = self._world_state_host_player_id

            if world_state_host_player_id is None or client.player_id != world_state_host_player_id:
                self._log(
                    f"Ignoring worldState from non-host {client.label} "
                    f"(player {client.player_id}, host is player {world_state_host_player_id})."
                )
                return

            packet["playerId"] = client.player_id
            packet["serverTime"] = time.time()
            with self._lock:
                self._stats["worldStatePacketsReceived"] += 1

        if packet.get("type") == "npcState":
            with self._lock:
                world_state_host_player_id = self._world_state_host_player_id

            if world_state_host_player_id is None or client.player_id != world_state_host_player_id:
                self._log(
                    f"Ignoring npcState from non-host {client.label} "
                    f"(player {client.player_id}, host is player {world_state_host_player_id})."
                )
                return

            npcs = packet.get("npcs")
            if not isinstance(npcs, list) or len(npcs) > 16 or any(not isinstance(npc, dict) for npc in npcs):
                self._log(f"Ignoring malformed npcState from host player {client.player_id}.")
                return

            packet["playerId"] = client.player_id
            packet["serverTime"] = time.time()
            packet["fullReplace"] = True
            with self._lock:
                self._last_npc_state = dict(packet)
                self._stats["npcStatePacketsReceived"] += 1

        if packet.get("type") == "combatHit":
            packet["playerId"] = client.player_id
            packet["serverTime"] = time.time()
            with self._lock:
                self._stats["combatHitsReceived"] += 1

        self._print_packet(client, packet)

        if packet.get("type") == "transform":
            self._broadcast_transform(client, packet)
        elif packet.get("type") == "worldState":
            self._broadcast_world_state(client, packet)
        elif packet.get("type") == "npcState":
            self._broadcast_npc_state(client, packet)
        elif packet.get("type") == "combatHit":
            self._route_combat_hit(client, packet)

    def _send_packet(self, client: ClientSession, packet: dict[str, Any], *, broadcast: bool = False) -> None:
        encoded = json.dumps(packet, separators=(",", ":")).encode("utf-8") + b"\n"
        client.connection.sendall(encoded)
        client.record_sent(broadcast=broadcast)

        with self._lock:
            self._stats["packetsSent"] += 1
            if broadcast:
                self._stats["packetsBroadcast"] += 1

    def _send_existing_transforms_to_client(self, new_client: ClientSession) -> None:
        with self._lock:
            snapshots = [
                dict(client.last_transform)
                for client in self._clients.values()
                if client.connection != new_client.connection and client.last_transform is not None
            ]

        if not snapshots:
            return

        successful_sends = 0
        failed = False
        for packet in snapshots:
            packet["serverTime"] = time.time()
            try:
                self._send_packet(new_client, packet, broadcast=True)
                successful_sends += 1
            except OSError:
                failed = True
                break

        with self._lock:
            self._stats["transformPacketsBroadcast"] += successful_sends

        if successful_sends:
            self._log(
                f"Sent {successful_sends} existing transform snapshot(s) to newly connected player {new_client.player_id}"
            )

        if failed:
            self._disconnect_client(new_client)

    def _send_existing_npc_state_to_client(self, new_client: ClientSession) -> None:
        with self._lock:
            snapshot = dict(self._last_npc_state) if self._last_npc_state is not None else None

        if snapshot is None or snapshot.get("playerId") == new_client.player_id:
            return

        snapshot["serverTime"] = time.time()
        try:
            self._send_packet(new_client, snapshot, broadcast=True)
            with self._lock:
                self._stats["npcStatePacketsBroadcast"] += 1
            self._log(
                f"Sent existing npcState snapshot with {len(snapshot.get('npcs', []))} "
                f"enemy/enemies to newly connected player {new_client.player_id}"
            )
        except OSError:
            self._disconnect_client(new_client)

    def _disconnect_client(self, client: ClientSession) -> None:
        with self._lock:
            was_world_state_host = client.player_id == self._world_state_host_player_id
            never_sent_packet = client.packets_received == 0

        if not self._remove_client(client):
            return

        self._close_client_socket(client)

        if never_sent_packet:
            # TCP connect probes (browser/pause ping) never send gameplay packets.
            # Avoid broadcasting a peer disconnect for those ghost sessions.
            self._log(
                f"Ignored probe/short-lived client: {client.label} (player {client.player_id})",
                level="debug",
            )
            if was_world_state_host:
                self._reassign_world_state_host()
            return

        self._log(f"Client disconnected: {client.label} (player {client.player_id})")
        self._broadcast_disconnect(client)

        if was_world_state_host:
            self._reassign_world_state_host()

    def _find_client_by_player_id(self, player_id: int) -> ClientSession | None:
        with self._lock:
            for client in self._clients.values():
                if client.player_id == player_id:
                    return client
        return None

    def _clients_for_ip(self, ip: str) -> list[ClientSession]:
        normalized = str(ip).strip()
        with self._lock:
            return [client for client in self._clients.values() if client.address[0] == normalized]

    def _build_session_ended_packet(self, code: str, reason: str = "") -> dict[str, Any]:
        packet: dict[str, Any] = {
            "type": "sessionEnded",
            "code": code,
            "serverTime": time.time(),
        }
        reason_text = str(reason or "")
        if reason_text:
            packet["reason"] = reason_text
        else:
            packet["reason"] = ""
        return packet

    def _send_session_ended_raw(
        self,
        connection: socket.socket,
        *,
        code: str,
        reason: str = "",
    ) -> None:
        packet = self._build_session_ended_packet(code, reason)
        encoded = json.dumps(packet, separators=(",", ":")).encode("utf-8") + b"\n"
        try:
            connection.sendall(encoded)
            with self._lock:
                self._stats["sessionEndedPacketsSent"] += 1
                self._stats["packetsSent"] += 1
            try:
                connection.shutdown(socket.SHUT_WR)
            except OSError:
                pass
        except OSError:
            pass

    def _end_client_session(
        self,
        client: ClientSession,
        *,
        code: str,
        reason: str = "",
    ) -> None:
        packet = self._build_session_ended_packet(code, reason)
        try:
            self._send_packet(client, packet)
            with self._lock:
                self._stats["sessionEndedPacketsSent"] += 1
            try:
                client.connection.shutdown(socket.SHUT_WR)
            except OSError:
                pass
        except OSError:
            pass
        self._disconnect_client(client)

    def _end_sessions_for_ip(self, ip: str, *, code: str, reason: str = "") -> int:
        clients = self._clients_for_ip(ip)
        for client in clients:
            self._end_client_session(client, code=code, reason=reason)
        return len(clients)

    def _close_client_socket(self, client: ClientSession) -> None:
        try:
            client.connection.close()
        except OSError:
            pass

    def _broadcast_transform(self, sender: ClientSession, packet: dict[str, Any]) -> None:
        with self._lock:
            recipients = [client for client in self._clients.values() if client.connection != sender.connection]

        failed_recipients: list[ClientSession] = []
        successful_sends = 0
        for recipient in recipients:
            try:
                # Transforms go to every other client only. The sender already knows
                # its own movement; echoing it back would create duplicate local state.
                self._send_packet(recipient, packet, broadcast=True)
                successful_sends += 1
            except OSError:
                failed_recipients.append(recipient)

        with self._lock:
            self._stats["transformPacketsBroadcast"] += successful_sends

        for recipient in failed_recipients:
            self._disconnect_client(recipient)

        self._log(
            f"Broadcast transform from player {sender.player_id} to {successful_sends} other client(s)",
            level="debug",
        )

    def _broadcast_world_state(self, sender: ClientSession, packet: dict[str, Any]) -> None:
        with self._lock:
            recipients = [client for client in self._clients.values() if client.connection != sender.connection]

        failed_recipients: list[ClientSession] = []
        successful_sends = 0
        for recipient in recipients:
            try:
                self._send_packet(recipient, packet, broadcast=True)
                successful_sends += 1
            except OSError:
                failed_recipients.append(recipient)

        with self._lock:
            self._stats["worldStatePacketsBroadcast"] += successful_sends

        for recipient in failed_recipients:
            self._disconnect_client(recipient)

        game_hour = packet.get("gameHour", "?")
        game_days_passed = packet.get("gameDaysPassed", "?")
        weather_form_id = packet.get("weatherFormId", "")
        weather_detail = f", weatherFormId={weather_form_id}" if weather_form_id else ""
        self._log(
            f"Broadcast worldState from player {sender.player_id} to {successful_sends} other client(s): "
            f"gameHour={game_hour}, gameDaysPassed={game_days_passed}{weather_detail}",
            level="debug",
        )

    def _broadcast_npc_state(self, sender: ClientSession, packet: dict[str, Any]) -> None:
        with self._lock:
            recipients = [client for client in self._clients.values() if client.connection != sender.connection]

        failed_recipients: list[ClientSession] = []
        successful_sends = 0
        for recipient in recipients:
            try:
                self._send_packet(recipient, packet, broadcast=True)
                successful_sends += 1
            except OSError:
                failed_recipients.append(recipient)

        with self._lock:
            self._stats["npcStatePacketsBroadcast"] += successful_sends

        for recipient in failed_recipients:
            self._disconnect_client(recipient)

        self._log(
            f"Broadcast npcState from player {sender.player_id} to {successful_sends} "
            f"other client(s): enemies={len(packet.get('npcs', []))}",
            level="debug",
        )

    def _broadcast_server_world_state(self) -> None:
        with self._lock:
            if not self._running:
                self._log("Cannot broadcast server world state while the server is stopped.")
                return

            recipients = list(self._clients.values())
            snapshot = dict(self._server_world_state)

        time_hhmm = snapshot.get("timeHHmm")
        weather_console_arg = snapshot.get("weatherConsoleArg")
        weather_form_id = snapshot.get("weatherFormId")
        if not time_hhmm and not weather_console_arg:
            self._log("Server world state broadcast skipped: no time or weather set.")
            return

        server_time = time.time()
        failed_recipients: list[ClientSession] = []
        successful_sends = 0

        if time_hhmm:
            time_packet: dict[str, Any] = {
                "type": "serverWorldState",
                "timeHHmm": time_hhmm,
                "serverTime": server_time,
            }
            for recipient in recipients:
                try:
                    self._send_packet(recipient, time_packet, broadcast=True)
                    successful_sends += 1
                except OSError:
                    failed_recipients.append(recipient)

            self._log(
                f"Broadcast serverWorldState time to {len(recipients)} client(s): timeHHmm={time_hhmm}"
            )

        if weather_console_arg:
            weather_packet: dict[str, Any] = {
                "type": "serverWorldState",
                "weatherConsoleArg": weather_console_arg,
                "serverTime": server_time,
            }
            if weather_form_id:
                weather_packet["weatherFormId"] = weather_form_id

            for recipient in recipients:
                try:
                    self._send_packet(recipient, weather_packet, broadcast=True)
                    successful_sends += 1
                except OSError:
                    failed_recipients.append(recipient)

            self._log(
                "Broadcast serverWorldState weather command to "
                f"{len(recipients)} client(s): fw {weather_console_arg}"
            )

        with self._lock:
            self._stats["serverWorldStatePacketsBroadcast"] += successful_sends

        for recipient in failed_recipients:
            self._disconnect_client(recipient)

    def _reassign_world_state_host(self) -> None:
        with self._lock:
            if not self._clients:
                self._world_state_host_player_id = None
                self._last_npc_state = None
                self._log("World-state host cleared; no clients remain.")
                return

            new_host_player_id = min(client.player_id for client in self._clients.values())
            self._world_state_host_player_id = new_host_player_id
            self._last_npc_state = None
            recipients = list(self._clients.values())

        packet = {
            "type": "worldStateHost",
            "worldStateHostPlayerId": new_host_player_id,
            "serverTime": time.time(),
        }

        failed_recipients: list[ClientSession] = []
        successful_sends = 0
        for recipient in recipients:
            try:
                self._send_packet(recipient, packet, broadcast=True)
                successful_sends += 1
            except OSError:
                failed_recipients.append(recipient)

        with self._lock:
            self._stats["worldStateHostPacketsBroadcast"] += successful_sends

        for recipient in failed_recipients:
            self._disconnect_client(recipient)

        self._log(
            f"Reassigned world-state host to player {new_host_player_id} "
            f"and notified {successful_sends} client(s)."
        )

    def _broadcast_disconnect(self, disconnected_client: ClientSession) -> None:
        packet = {
            "type": "disconnect",
            "playerId": disconnected_client.player_id,
            "serverTime": time.time(),
        }

        with self._lock:
            recipients = list(self._clients.values())

        failed_recipients: list[ClientSession] = []
        successful_sends = 0
        for recipient in recipients:
            try:
                self._send_packet(recipient, packet, broadcast=True)
                successful_sends += 1
            except OSError:
                failed_recipients.append(recipient)

        with self._lock:
            self._stats["disconnectPacketsBroadcast"] += successful_sends

        for recipient in failed_recipients:
            self._disconnect_client(recipient)

        self._log(
            f"Broadcast disconnect for player {disconnected_client.player_id} "
            f"to {successful_sends} other client(s)"
        )

    def _route_combat_hit(self, sender: ClientSession, packet: dict[str, Any]) -> None:
        """Route a targeted combatHit packet to the victim player only."""
        target_player_id = packet.get("targetPlayerId")
        sequence = packet.get("sequence")
        damage_value = packet.get("damage")

        # Coerce numeric JSON values: some clients/serializers emit floats for integers.
        if isinstance(target_player_id, float) and target_player_id.is_integer():
            target_player_id = int(target_player_id)
            packet["targetPlayerId"] = target_player_id
        if isinstance(sequence, float) and sequence.is_integer():
            sequence = int(sequence)
            packet["sequence"] = sequence

        if (
            not isinstance(target_player_id, int)
            or isinstance(target_player_id, bool)
            or not isinstance(sequence, int)
            or isinstance(sequence, bool)
            or not isinstance(damage_value, (int, float))
            or isinstance(damage_value, bool)
        ):
            self._log(f"Malformed combatHit packet from player {sender.player_id}: {packet}")
            return

        damage = float(damage_value)
        if (
            not 0 < target_player_id <= 0xFFFFFFFF
            or not 0 < sequence <= 0xFFFFFFFF
            or not math.isfinite(damage)
            or not 0.0 < damage <= 10000.0
        ):
            self._log(
                f"Invalid combatHit values from player {sender.player_id}: "
                f"targetPlayerId={target_player_id}, sequence={sequence}, damage={damage}"
            )
            return

        if target_player_id == sender.player_id:
            self._log(f"Ignoring self-targeted combatHit from player {sender.player_id}")
            return

        weapon_form_id = packet.get("weaponFormId")
        if weapon_form_id is not None:
            if (
                not isinstance(weapon_form_id, str)
                or not 1 <= len(weapon_form_id) <= 8
                or any(character not in "0123456789abcdefABCDEF" for character in weapon_form_id)
            ):
                self._log(
                    f"Invalid combatHit weaponFormId from player {sender.player_id}: "
                    f"{weapon_form_id!r}"
                )
                return
            packet["weaponFormId"] = weapon_form_id.upper().zfill(8)

        with self._lock:
            recipient = next(
                (client for client in self._clients.values() if client.player_id == target_player_id),
                None,
            )

        if recipient is None:
            self._log(
                f"Target player {target_player_id} not found for combatHit from player {sender.player_id}"
            )
            return

        try:
            self._send_packet(recipient, packet, broadcast=False)
            with self._lock:
                self._stats["combatHitsRouted"] += 1
            self._log(
                f"Routed combatHit from player {sender.player_id} to player {target_player_id}: "
                f"damage={damage:.1f}"
            )
        except OSError:
            self._disconnect_client(recipient)

    def _print_packet(self, client: ClientSession, packet: dict[str, Any]) -> None:
        packet_type = packet.get("type")
        if packet_type == "worldState":
            game_hour = packet.get("gameHour", "?")
            game_days_passed = packet.get("gameDaysPassed", "?")
            weather_form_id = packet.get("weatherFormId", "")
            extra_fields = []
            for field_name in ("playerId", "clientTime", "serverTime"):
                if field_name in packet:
                    extra_fields.append(f"{field_name}={packet[field_name]}")
            extra_details = ""
            if extra_fields:
                extra_details = ", " + ", ".join(extra_fields)
            weather_detail = f", weatherFormId={weather_form_id}" if weather_form_id else ""
            self._log(
                f"WorldState from {client.label}: gameHour={game_hour}, "
                f"gameDaysPassed={game_days_passed}{weather_detail}{extra_details}",
                level="debug",
            )
            return

        if packet_type != "transform":
            self._log(f"Packet from {client.label}: {packet}", level="debug")
            return

        try:
            x = float(packet["x"])
            y = float(packet["y"])
            z = float(packet["z"])
            angle_z = float(packet["angleZ"])
        except (KeyError, TypeError, ValueError):
            self._log(f"Malformed transform packet from {client.label}: {packet}", level="warning")
            return

        movement_type = packet.get("movementType", "normal")
        extra_fields = []
        for field_name in ("playerId", "cellId", "worldspaceId", "clientTime", "serverTime"):
            if field_name in packet:
                extra_fields.append(f"{field_name}={packet[field_name]}")

        extra_details = ""
        if extra_fields:
            extra_details = ", " + ", ".join(extra_fields)

        self._log(
            "Transform from "
            f"{client.label}: x={x:.2f}, y={y:.2f}, z={z:.2f}, angleZ={angle_z:.2f}, "
            f"movementType={movement_type}{extra_details}",
            level="debug",
        )

    def _should_log(self, level: str) -> bool:
        configured = _LOG_LEVELS.get(self.log_verbosity, _LOG_LEVELS[DEFAULT_LOG_VERBOSITY])
        message_level = _LOG_LEVELS.get(level, _LOG_LEVELS["info"])
        return message_level >= configured

    def _log(self, message: str, *, level: str = "info") -> None:
        if not self._should_log(level):
            return

        with self._log_lock:
            listeners = list(self._log_listeners)

        for listener in listeners:
            try:
                # Prefer level-aware callbacks; fall back for older listeners.
                try:
                    listener(message, level=level)  # type: ignore[call-arg]
                except TypeError:
                    listener(message)
            except Exception:
                # Future UI listeners must not be able to break server networking.
                pass