Commonwealth-Online-Public/server/server_core.py

from __future__ import annotations

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

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


def get_lan_addresses() -> list[str]:
    """Return likely LAN IPv4 addresses for this machine."""
    try:
        with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as probe:
            probe.connect(("8.8.8.8", 80))
            return [probe.getsockname()[0]]
    except OSError:
        return []


class FalloutTogetherServer:
    def __init__(self, host: str = HOST, port: int = PORT) -> None:
        self.host = host
        self.port = port

        self._lock = threading.RLock()
        self._log_lock = threading.RLock()
        self._clients: dict[socket.socket, ClientSession] = {}
        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._stats: dict[str, int] = {
            "clientsConnected": 0,
            "clientsDisconnected": 0,
            "packetsReceived": 0,
            "packetsSent": 0,
            "packetsBroadcast": 0,
            "transformPacketsReceived": 0,
            "transformPacketsBroadcast": 0,
            "worldStatePacketsReceived": 0,
            "worldStatePacketsBroadcast": 0,
            "worldStateHostPacketsBroadcast": 0,
            "serverWorldStatePacketsBroadcast": 0,
            "disconnectPacketsBroadcast": 0,
        }

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

            self._prepare_server_socket()
            self._discovery = LanDiscoveryResponder(self)
            self._discovery.start()
            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._discovery = LanDiscoveryResponder(self)
            self._discovery.start()

        self._accept_loop()

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

        with self._lock:
            self._running = False
            server_socket = self._server_socket
            self._server_socket = None
            clients = list(self._clients.values())

        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)

    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,
                    "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 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)
            server_socket.bind((self.host, self.port))
            server_socket.listen()
            server_socket.settimeout(ACCEPT_TIMEOUT_SECONDS)
        except OSError:
            server_socket.close()
            raise

        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._log(f"Commonwealth Online server listening on {self.host}:{self.port}")
        self._log(f"LAN discovery listening on UDP port {DISCOVERY_PORT}")
        if self.host == "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.")
            self._log(
                "Forward this port on your router and allow it through Windows Firewall "
                "for connections from outside your network."
            )
        self._log("Waiting for newline-separated JSON transform packets...")

    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)
                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:
        client = self._assign_client(connection, address)
        self._log(f"Client connected: {client.label} (player {client.player_id})")

        with connection:
            buffer = ""

            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(),
                }
                if world_state_host_player_id is not None:
                    welcome_packet["worldStateHostPlayerId"] = world_state_host_player_id

                self._send_packet(client, welcome_packet)

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

                    buffer += chunk.decode("utf-8", errors="replace")
                    while "\n" in buffer:
                        line, buffer = buffer.split("\n", 1)
                        self._handle_line(client, line.strip())
            except ConnectionResetError:
                self._log(f"Client disconnected unexpectedly: {client.label} (player {client.player_id})")
            except OSError as error:
                if self.is_running():
                    self._log(f"Client connection error: {client.label} (player {client.player_id}): {error}")
            finally:
                self._disconnect_client(client)

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

        received_at = time.time()
        client.record_received(received_at)

        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

        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)

    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 _disconnect_client(self, client: ClientSession) -> None:
        with self._lock:
            was_world_state_host = client.player_id == self._world_state_host_player_id

        if not self._remove_client(client):
            return

        self._close_client_socket(client)
        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 _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)")

    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}"
        )

    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._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
            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 _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}"
            )
            return

        if packet_type != "transform":
            self._log(f"Packet from {client.label}: {packet}")
            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}")
            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}"
        )

    def _log(self, message: str) -> None:
        with self._log_lock:
            listeners = list(self._log_listeners)

        for listener in listeners:
            try:
                listener(message)
            except Exception:
                # Future UI listeners must not be able to break server networking.
                pass