# -*- coding: utf-8 -*- """ SyncHandler — Server-seitige Verarbeitung von Sync-Nachrichten. Registriert sich als Message-Handler bei NetworkService. Verarbeitet SyncPush und SyncRequest von Standalone-Clients. """ from __future__ import annotations from typing import TYPE_CHECKING 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.utils.debug import pinfo, pdebug, perror if TYPE_CHECKING: from trixy_core.server import ServerApplication from trixy_core.satellite.satellite import Satellite class SyncHandler: """ Server-seitiger Handler fuer Sync-Nachrichten. Verarbeitet: - SyncPush: Items vom Standalone empfangen und lokal mergen - SyncRequest: Aenderungen seit Zeitpunkt X zuruecksenden """ def __init__(self, application: ServerApplication) -> None: """ Initialisiert den SyncHandler. Args: application: Die Server-Anwendung """ self._app = application async def on_message(self, satellite: Satellite, message: ProtocolMessage) -> None: """ Verarbeitet eine eingehende Sync-Nachricht. Wird als Message-Handler bei NetworkService registriert. Args: satellite: Der sendende Satellite message: Die empfangene Nachricht """ class_name = message.class_name if class_name == "SyncPush": await self._handle_sync_push(satellite, message) elif class_name == "SyncRequest": await self._handle_sync_request(satellite, message) async def _handle_sync_push(self, satellite: Satellite, message: ProtocolMessage) -> None: """Verarbeitet empfangene SyncPush-Nachrichten.""" store = self._app.sync_store if store is None: perror("SyncHandler: Kein SyncStore verfuegbar") return data = message.data items: list[dict] = [] if isinstance(data, SyncPush): items = data.items elif isinstance(data, dict): items = data.get("items", []) else: perror(f"SyncHandler: Ungueltiges SyncPush-Format: {type(data)}") return merged_count = 0 for item_dict in items: item = SyncItem.from_dict(item_dict) if store.merge_incoming(item): merged_count += 1 # Bestaetigung senden ack = SyncAck(accepted_count=merged_count) try: await self._app.network.send_to_satellite(satellite.id, ack) except Exception as e: perror(f"SyncHandler: SyncAck senden fehlgeschlagen: {e}") if merged_count > 0: pdebug(f"SyncHandler: {merged_count} Items von {satellite.alias} gemerged") # Event emittieren (damit Server-Plugins reagieren koennen) await self._app.events.emit("sync_item_received", { "satellite_id": satellite.id, "merged_count": merged_count, }) async def _handle_sync_request(self, satellite: Satellite, message: ProtocolMessage) -> None: """Verarbeitet SyncRequest-Nachrichten.""" store = self._app.sync_store if store is None: perror("SyncHandler: Kein SyncStore verfuegbar") return data = message.data since = "" if isinstance(data, SyncRequest): since = data.since elif isinstance(data, dict): since = data.get("since", "") # Aenderungen seit 'since' sammeln if since: changes = store.get_changes_since(since) else: # Erster Sync: Alle Items senden changes = store.get_changes_since("") # Antwort senden server_time = _now_iso() response = SyncResponse( items=[item.to_dict() for item in changes], server_time=server_time, ) try: await self._app.network.send_to_satellite(satellite.id, response) except Exception as e: perror(f"SyncHandler: SyncResponse senden fehlgeschlagen: {e}") pdebug(f"SyncHandler: {len(changes)} Items an {satellite.alias} gesendet (seit: {since or 'Anfang'})")