Trixy7/trixy_core/sync/service.py

# -*- coding: utf-8 -*-
"""
SyncService — Standalone-seitiger Sync-Orchestrator.

Probe-Loop: Alle probe_interval_seconds prüfen ob Server erreichbar.
Bei Verbindung: Initialer Sync, danach periodischer Sync alle sync_interval_seconds.
"""

from __future__ import annotations

import asyncio
from datetime import datetime, timezone
from typing import TYPE_CHECKING

from trixy_core.sync.connection import SyncConnection, SyncConnectionConfig
from trixy_core.sync.models import SyncItem, _now_iso
from trixy_core.network.cmd.sync import SyncPush, SyncRequest, SyncResponse, SyncAck
from trixy_core.network.protocol import ProtocolMessage
from trixy_core.network.client_connection import get_mac_address
from trixy_core.utils.debug import pinfo, pdebug, perror, pwarn

if TYPE_CHECKING:
    from trixy_core.config.datasets.standalone import SyncConfig
    from trixy_core.standalone import StandaloneApplication


class SyncService:
    """
    Orchestriert die Datensynchronisation zwischen Standalone und Server.

    - Probe-Loop: Regelmaessig versuchen, den Server zu erreichen
    - Bei Verbindung: Handshake → initialer Sync → periodischer Sync
    - Bei Verbindungsverlust: Zurueck zum Probe-Loop
    """

    def __init__(self, application: StandaloneApplication, config: SyncConfig) -> None:
        """
        Initialisiert den SyncService.

        Args:
            application: Die Standalone-Anwendung
            config: Sync-Konfiguration
        """
        self._app = application
        self._config = config
        self._mac_address = get_mac_address()

        # Verbindung
        conn_config = SyncConnectionConfig(
            host=config.server_host,
            port=config.server_port,
            probe_interval_seconds=config.probe_interval_seconds,
            sync_interval_seconds=config.sync_interval_seconds,
            connect_timeout_seconds=config.connect_timeout_seconds,
        )
        self._connection = SyncConnection(
            config=conn_config,
            room=application.room,
            alias=application.alias,
            mac_address=self._mac_address,
        )
        self._connection.on_connected(self._on_connected)
        self._connection.on_disconnected(self._on_disconnected)
        self._connection.on_message(self._on_message)

        # Zustand
        self._running = False
        self._probe_task: asyncio.Task | None = None
        self._sync_task: asyncio.Task | None = None
        self._sync_in_progress = False

        # Fuer Antwort-Handling
        self._pending_response: asyncio.Future[SyncResponse] | None = None

    async def start(self) -> None:
        """Startet den SyncService (Probe-Loop)."""
        if not self._config.server_host:
            pwarn("SyncService: Kein Server-Host konfiguriert, Sync deaktiviert")
            return

        self._running = True
        self._probe_task = asyncio.create_task(self._probe_loop())
        pinfo(f"SyncService: Gestartet (Probe alle {self._config.probe_interval_seconds}s zu {self._config.server_host}:{self._config.server_port})")

    async def stop(self) -> None:
        """Stoppt den SyncService."""
        self._running = False

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

        # Verbindung trennen
        if self._connection.is_connected:
            await self._connection.disconnect("SyncService gestoppt")

        pinfo("SyncService: Gestoppt")

    async def sync_now(self) -> bool:
        """
        Fuehrt einen manuellen Sync durch (z.B. per Sprachbefehl).

        Returns:
            True bei Erfolg
        """
        if not self._connection.is_connected:
            pwarn("SyncService: Nicht verbunden, Sync nicht moeglich")
            return False

        if self._sync_in_progress:
            pdebug("SyncService: Sync bereits aktiv")
            return False

        return await self._perform_sync()

    # =========================================================================
    # Probe-Loop
    # =========================================================================

    async def _probe_loop(self) -> None:
        """Versucht regelmaessig, den Server zu erreichen."""
        while self._running:
            try:
                if not self._connection.is_connected:
                    pdebug(f"SyncService: Probe zu {self._config.server_host}:{self._config.server_port}...")
                    connected = await self._connection.connect()
                    if connected:
                        # Initialer Sync wird in _on_connected ausgeloest
                        # Periodischen Sync starten
                        self._sync_task = asyncio.create_task(self._periodic_sync_loop())

                await asyncio.sleep(self._config.probe_interval_seconds)

            except asyncio.CancelledError:
                break
            except Exception as e:
                perror(f"SyncService: Probe-Fehler: {e}")
                await asyncio.sleep(self._config.probe_interval_seconds)

    async def _periodic_sync_loop(self) -> None:
        """Fuehrt periodische Syncs durch solange verbunden."""
        while self._running and self._connection.is_connected:
            try:
                await asyncio.sleep(self._config.sync_interval_seconds)

                if not self._running or not self._connection.is_connected:
                    break

                await self._perform_sync()

            except asyncio.CancelledError:
                break
            except Exception as e:
                perror(f"SyncService: Periodischer Sync Fehler: {e}")

    # =========================================================================
    # Sync-Ablauf
    # =========================================================================

    async def _perform_sync(self) -> bool:
        """
        Fuehrt einen vollstaendigen Sync-Zyklus durch.

        1. Lokale Aenderungen an Server pushen
        2. Server-Aenderungen anfragen
        3. Server-Items lokal mergen
        4. last_sync_time aktualisieren
        """
        store = self._app.sync_store
        if store is None:
            perror("SyncService: Kein SyncStore verfuegbar")
            return False

        self._sync_in_progress = True
        try:
            last_sync = store.get_last_sync_time()
            pdebug(f"SyncService: Starte Sync (seit: {last_sync or 'nie'})")

            # 1. Lokale Aenderungen pushen
            local_changes = store.get_changes_since(last_sync) if last_sync else []
            if local_changes:
                push_msg = SyncPush(
                    items=[item.to_dict() for item in local_changes]
                )
                if not await self._connection.send_command(push_msg):
                    perror("SyncService: SyncPush senden fehlgeschlagen")
                    return False
                pdebug(f"SyncService: {len(local_changes)} lokale Aenderungen gepusht")

            # 2. Server-Aenderungen anfragen
            request_msg = SyncRequest(since=last_sync)

            # Future fuer Antwort erstellen
            loop = asyncio.get_running_loop()
            self._pending_response = loop.create_future()

            if not await self._connection.send_command(request_msg):
                perror("SyncService: SyncRequest senden fehlgeschlagen")
                self._pending_response = None
                return False

            # 3. Auf Antwort warten (mit Timeout)
            try:
                response = await asyncio.wait_for(self._pending_response, timeout=30.0)
            except asyncio.TimeoutError:
                perror("SyncService: Timeout beim Warten auf SyncResponse")
                self._pending_response = None
                return False
            finally:
                self._pending_response = None

            # 4. Server-Items lokal mergen
            merged_count = 0
            for item_dict in response.items:
                item = SyncItem.from_dict(item_dict)
                if store.merge_incoming(item):
                    merged_count += 1

            # 5. last_sync_time aktualisieren
            server_time = response.server_time or _now_iso()
            store.set_last_sync_time(server_time)

            pinfo(f"SyncService: Sync abgeschlossen ({len(local_changes)} gepusht, {merged_count} gemerged)")

            # 6. Event emittieren
            await self._app.events.emit("sync_completed", {
                "pushed_count": len(local_changes),
                "merged_count": merged_count,
                "server_time": server_time,
            })

            return True

        except Exception as e:
            perror(f"SyncService: Sync-Fehler: {e}")
            return False
        finally:
            self._sync_in_progress = False

    # =========================================================================
    # Callbacks
    # =========================================================================

    async def _on_connected(self) -> None:
        """Callback bei Verbindungsaufbau."""
        pinfo("SyncService: Mit Server verbunden, starte initialen Sync...")
        await self._perform_sync()

    async def _on_disconnected(self, reason: str) -> None:
        """Callback bei Verbindungsabbruch."""
        pinfo(f"SyncService: Verbindung verloren: {reason}")

        # Periodischen Sync stoppen
        if self._sync_task and not self._sync_task.done():
            self._sync_task.cancel()
            try:
                await self._sync_task
            except asyncio.CancelledError:
                pass

    async def _on_message(self, message: ProtocolMessage) -> None:
        """Callback fuer eingehende Nachrichten vom Server."""
        class_name = message.class_name

        if class_name == "SyncResponse":
            # Antwort auf SyncRequest
            data = message.data
            if isinstance(data, SyncResponse):
                response = data
            elif isinstance(data, dict):
                response = SyncResponse(
                    items=data.get("items", []),
                    server_time=data.get("server_time", ""),
                )
            else:
                perror(f"SyncService: Ungueltiges SyncResponse-Format: {type(data)}")
                return

            if self._pending_response and not self._pending_response.done():
                self._pending_response.set_result(response)

        elif class_name == "SyncAck":
            pdebug("SyncService: SyncAck empfangen")

        elif class_name == "SyncPush":
            # Server pusht Aenderungen (z.B. nach Aenderung durch anderen Satellite)
            store = self._app.sync_store
            if store is None:
                return

            data = message.data
            items: list[dict] = []
            if isinstance(data, SyncPush):
                items = data.items
            elif isinstance(data, dict):
                items = data.get("items", [])

            merged_count = 0
            for item_dict in items:
                item = SyncItem.from_dict(item_dict)
                if store.merge_incoming(item):
                    merged_count += 1

            if merged_count > 0:
                pdebug(f"SyncService: {merged_count} Server-Push Items gemerged")
                await self._app.events.emit("sync_push_received", {
                    "merged_count": merged_count,
                })