Trixy7/trixy_core/sync/connection.py

# -*- coding: utf-8 -*-
"""
SyncConnection — Command-Only TCP-Verbindung zum Server.

Leichtgewichtige Verbindung fuer Datensynchronisation.
Nur Command-Socket (Port 2101), keine Audio-Ports.
Handshake analog zu ClientConnection.
"""

from __future__ import annotations

import asyncio
import struct
from dataclasses import dataclass
from typing import TYPE_CHECKING, Callable, Awaitable

from trixy_core.network.protocol import (
    TrixyProtocol,
    ProtocolFlags,
    ProtocolMessage,
    MAGIC,
    MAGIC_LENGTH,
    COMMAND_HELLO,
    COMMAND_PING,
    COMMAND_PONG,
)
from trixy_core.network.cmd import (
    SatelliteConnect,
    SatelliteAccepted,
    SatelliteDenied,
)
from trixy_core.utils.debug import pinfo, pdebug, perror, pwarn
from trixy_core.utils.version import VERSION_STRING

if TYPE_CHECKING:
    pass


@dataclass
class SyncConnectionConfig:
    """Konfiguration fuer die Sync-Verbindung."""
    host: str = ""
    port: int = 2101
    probe_interval_seconds: float = 30.0
    sync_interval_seconds: float = 300.0
    connect_timeout_seconds: float = 5.0


class SyncConnection:
    """
    Command-Only TCP-Verbindung zum Server fuer Sync.

    Fuehrt den Standard-Handshake durch (TRXYHELO → SatelliteConnect → SatelliteAccepted),
    registriert den Standalone als Satellite.
    Kein Audio — nur Command-Socket.
    """

    def __init__(
        self,
        config: SyncConnectionConfig,
        room: str,
        alias: str,
        mac_address: str,
    ) -> None:
        """
        Initialisiert die SyncConnection.

        Args:
            config: Verbindungskonfiguration
            room: Raum-Kennung
            alias: Anzeigename
            mac_address: MAC-Adresse des Standalone
        """
        self._config = config
        self._room = room
        self._alias = alias
        self._mac_address = mac_address

        # Protokoll
        self._protocol = TrixyProtocol()

        # Socket
        self._reader: asyncio.StreamReader | None = None
        self._writer: asyncio.StreamWriter | None = None

        # Zustand
        self._connected = False
        self._satellite_id: str = ""
        self._running = False

        # Tasks
        self._receive_task: asyncio.Task | None = None
        self._keepalive_task: asyncio.Task | None = None

        # Callbacks
        self._on_connected: Callable[[], Awaitable[None]] | None = None
        self._on_disconnected: Callable[[str], Awaitable[None]] | None = None
        self._on_message: Callable[[ProtocolMessage], Awaitable[None]] | None = None

    @property
    def is_connected(self) -> bool:
        """Ist die Verbindung aktiv?"""
        return self._connected

    @property
    def satellite_id(self) -> str:
        """Vom Server zugewiesene Satellite-ID."""
        return self._satellite_id

    def on_connected(self, callback: Callable[[], Awaitable[None]]) -> None:
        """Registriert Callback fuer Verbindungsaufbau."""
        self._on_connected = callback

    def on_disconnected(self, callback: Callable[[str], Awaitable[None]]) -> None:
        """Registriert Callback fuer Verbindungsabbruch."""
        self._on_disconnected = callback

    def on_message(self, callback: Callable[[ProtocolMessage], Awaitable[None]]) -> None:
        """Registriert Callback fuer eingehende Nachrichten."""
        self._on_message = callback

    async def connect(self) -> bool:
        """
        Verbindet sich mit dem Server (Command-Socket + Handshake).

        Returns:
            True bei erfolgreicher Verbindung
        """
        try:
            pdebug(f"SyncConnection: Verbinde zu {self._config.host}:{self._config.port}...")

            # Command-Socket verbinden
            try:
                self._reader, self._writer = await asyncio.wait_for(
                    asyncio.open_connection(self._config.host, self._config.port),
                    timeout=self._config.connect_timeout_seconds,
                )
            except asyncio.TimeoutError:
                pdebug(f"SyncConnection: Timeout zu {self._config.host}:{self._config.port}")
                return False
            except OSError as e:
                pdebug(f"SyncConnection: Verbindungsfehler: {e}")
                return False

            # Handshake durchfuehren
            if not await self._perform_handshake():
                await self._close_socket()
                return False

            # Empfangs-Tasks starten
            self._running = True
            self._receive_task = asyncio.create_task(self._receive_loop())
            self._keepalive_task = asyncio.create_task(self._keepalive_loop())

            self._connected = True
            pinfo(f"SyncConnection: Verbunden mit Server {self._config.host}:{self._config.port}")

            # Callback
            if self._on_connected:
                try:
                    await self._on_connected()
                except Exception as e:
                    perror(f"SyncConnection: on_connected Fehler: {e}")

            return True

        except Exception as e:
            perror(f"SyncConnection: Verbindungsfehler: {e}")
            await self._close_socket()
            return False

    async def disconnect(self, reason: str = "Sync-Disconnect") -> None:
        """
        Trennt die Verbindung zum Server.

        Args:
            reason: Grund fuer die Trennung
        """
        if not self._connected:
            return

        pdebug(f"SyncConnection: Trenne Verbindung: {reason}")
        self._running = False

        # Tasks beenden
        for task in [self._receive_task, self._keepalive_task]:
            if task and not task.done():
                task.cancel()
                try:
                    await task
                except asyncio.CancelledError:
                    pass

        await self._close_socket()
        self._connected = False
        self._satellite_id = ""

        # Callback
        if self._on_disconnected:
            try:
                await self._on_disconnected(reason)
            except Exception as e:
                perror(f"SyncConnection: on_disconnected Fehler: {e}")

    async def send_command(self, cmd: object) -> bool:
        """
        Sendet eine Nachricht zum Server.

        Args:
            cmd: Die zu sendende CommandMessage

        Returns:
            True bei Erfolg
        """
        if not self._connected or not self._writer:
            return False

        return await self._send_message(cmd)

    # =========================================================================
    # Handshake
    # =========================================================================

    async def _perform_handshake(self) -> bool:
        """Fuehrt den Handshake mit dem Server durch."""
        pdebug("SyncConnection: Starte Handshake...")

        # HELLO senden
        try:
            self._writer.write(COMMAND_HELLO)
            await self._writer.drain()
        except Exception as e:
            perror(f"SyncConnection: HELLO senden fehlgeschlagen: {e}")
            return False

        # SatelliteConnect senden
        connect_msg = SatelliteConnect(
            mac_address=self._mac_address,
            room=self._room,
            alias=self._alias,
            ip_address="",
            version=VERSION_STRING,
        )

        if not await self._send_message(connect_msg):
            perror("SyncConnection: SatelliteConnect senden fehlgeschlagen")
            return False

        # Antwort empfangen
        try:
            response = await asyncio.wait_for(
                self._read_message(),
                timeout=self._config.connect_timeout_seconds,
            )
        except asyncio.TimeoutError:
            perror("SyncConnection: Timeout beim Warten auf Server-Antwort")
            return False

        if response is None:
            perror("SyncConnection: Keine Antwort vom Server")
            return False

        # Antwort verarbeiten
        if response.class_name == "SatelliteDenied":
            data = response.data
            reason = "Unbekannt"
            if isinstance(data, dict):
                reason = data.get("reason", "Unbekannt")
            elif isinstance(data, SatelliteDenied):
                reason = data.reason
            pwarn(f"SyncConnection: Verbindung abgelehnt: {reason}")
            return False

        if response.class_name != "SatelliteAccepted":
            perror(f"SyncConnection: Unerwartete Antwort: {response.class_name}")
            return False

        # SatelliteAccepted verarbeiten
        data = response.data
        if isinstance(data, dict):
            self._satellite_id = data.get("satellite_id", "")
        elif isinstance(data, SatelliteAccepted):
            self._satellite_id = data.satellite_id
        else:
            perror(f"SyncConnection: Ungueltiges SatelliteAccepted: {type(data)}")
            return False

        pinfo(f"SyncConnection: Handshake erfolgreich, ID: {self._satellite_id}")
        return True

    # =========================================================================
    # Empfangs-Loop
    # =========================================================================

    async def _receive_loop(self) -> None:
        """Empfaengt Nachrichten vom Server."""
        while self._running:
            try:
                message = await self._read_message()
                if message is None:
                    if self._running:
                        pwarn("SyncConnection: Verbindung zum Server verloren")
                        asyncio.create_task(self._handle_disconnect("Verbindung verloren"))
                    break

                await self._handle_message(message)

            except asyncio.CancelledError:
                break
            except Exception as e:
                if self._running:
                    perror(f"SyncConnection: Empfangsfehler: {e}")
                    asyncio.create_task(self._handle_disconnect(f"Fehler: {e}"))
                break

    async def _handle_message(self, message: ProtocolMessage) -> None:
        """Verarbeitet eine empfangene Nachricht."""
        class_name = message.class_name

        # PING beantworten
        if class_name == "TRXIPING":
            await self._send_pong()
            return

        # PONG ignorieren
        if class_name == "TRXIPONG":
            return

        # An Callback weiterleiten
        if self._on_message:
            try:
                await self._on_message(message)
            except Exception as e:
                perror(f"SyncConnection: Message-Callback Fehler: {e}")

    async def _keepalive_loop(self) -> None:
        """Sendet regelmaessige Pings."""
        while self._running:
            try:
                await asyncio.sleep(30.0)
                if not self._running:
                    break
                await self._send_ping()
            except asyncio.CancelledError:
                break
            except Exception:
                pass

    async def _handle_disconnect(self, reason: str) -> None:
        """Behandelt einen Verbindungsabbruch."""
        was_connected = self._connected
        self._connected = False
        self._running = False

        await self._close_socket()

        if was_connected and self._on_disconnected:
            try:
                await self._on_disconnected(reason)
            except Exception as e:
                perror(f"SyncConnection: on_disconnected Fehler: {e}")

    # =========================================================================
    # Low-Level I/O
    # =========================================================================

    async def _send_message(self, message: object, flags: ProtocolFlags = ProtocolFlags.PICKLE) -> bool:
        """Sendet eine Nachricht ueber den Command-Socket."""
        if not self._writer:
            return False

        try:
            data = self._protocol.serialize(message, flags)
            self._writer.write(data)
            await self._writer.drain()
            return True
        except Exception as e:
            perror(f"SyncConnection: Sende-Fehler: {e}")
            return False

    async def _send_ping(self) -> None:
        """Sendet einen Ping."""
        if self._writer:
            try:
                self._writer.write(COMMAND_PING)
                await self._writer.drain()
            except Exception:
                pass

    async def _send_pong(self) -> None:
        """Sendet einen Pong."""
        if self._writer:
            try:
                self._writer.write(COMMAND_PONG)
                await self._writer.drain()
            except Exception:
                pass

    async def _read_message(self) -> ProtocolMessage | None:
        """Liest eine Nachricht vom Command-Socket."""
        if not self._reader:
            return None

        try:
            header_size = MAGIC_LENGTH + 2 + 8 + 4 + 16 + 2  # 36 bytes

            # Prüfe auf Hard-coded Befehle (8 bytes)
            peek_data = await self._reader.readexactly(8)

            if self._protocol.is_hard_command(peek_data):
                return self._protocol.deserialize(peek_data)

            # Kein Hard-coded Befehl — lese Rest des Headers
            remaining_header = await self._reader.readexactly(header_size - 8)
            header = peek_data + remaining_header

            if header[:MAGIC_LENGTH] != MAGIC:
                perror(f"SyncConnection: Ungueltige Magic: {header[:MAGIC_LENGTH]}")
                return None

            # Parse Klassennamen-Laenge
            class_name_length = struct.unpack(">H", header[34:36])[0]

            # Lese Klassenname
            class_name = await self._reader.readexactly(class_name_length)

            # Lese Datenlaenge (4 bytes)
            data_length_bytes = await self._reader.readexactly(4)
            data_length = struct.unpack(">I", data_length_bytes)[0]

            # Lese Daten
            data = await self._reader.readexactly(data_length) if data_length > 0 else b""

            # Baue vollstaendige Nachricht zusammen
            full_message = header + class_name + data_length_bytes + data
            return self._protocol.deserialize(full_message)

        except (ConnectionResetError, asyncio.IncompleteReadError):
            return None
        except Exception as e:
            perror(f"SyncConnection: Lese-Fehler: {e}")
            return None

    async def _close_socket(self) -> None:
        """Schliesst den Socket."""
        if self._writer:
            try:
                self._writer.close()
                await self._writer.wait_closed()
            except Exception:
                pass

        self._reader = None
        self._writer = None