video-sync/desktop-client/sync_client.py

import asyncio
import json
import ssl
import time

import certifi
import websockets
from PyQt6.QtCore import QThread, pyqtSignal


class SyncClientThread(QThread):
    # Signals from WebSocket -> PyQt UI
    connected = pyqtSignal()
    disconnected = pyqtSignal()
    connection_error = pyqtSignal(str)
    room_joined = pyqtSignal(dict)
    room_rejoined = pyqtSignal(dict)
    room_error = pyqtSignal(str)
    file_check_needed = pyqtSignal(dict)  # msg
    users_updated = pyqtSignal(list)
    chat_message = pyqtSignal(str, str, int)  # author, text, timestamp
    system_message = pyqtSignal(str)
    sync_event = pyqtSignal(dict)
    ready_check_updated = pyqtSignal(dict)
    fight_reaction = pyqtSignal(dict)
    fight_poll_updated = pyqtSignal(dict)
    latency_updated = pyqtSignal(int)  # latency in ms

    def __init__(self, url="ws://localhost:3000/ws"):
        super().__init__()
        self.url = url
        self.ws = None
        self.loop = None
        self.running = False
        self._ping_sent_at = 0

    @staticmethod
    def _create_ssl_context():
        """Build a verified TLS context from the CA bundle shipped with the app."""
        return ssl.create_default_context(cafile=certifi.where())

    @staticmethod
    def _certificate_error(exc):
        """Return a nested certificate verification error, if present."""
        while exc is not None:
            if isinstance(exc, ssl.SSLCertVerificationError):
                return exc
            exc = exc.__cause__ or exc.__context__
        return None

    def run(self):
        """Runs strictly within the newly created QThread"""
        self.running = True
        self.loop = asyncio.new_event_loop()
        asyncio.set_event_loop(self.loop)
        try:
            self.loop.run_until_complete(self._connect_and_listen())
        finally:
            self.ws = None
            self.running = False
            self.loop.close()

    def stop(self):
        self.running = False
        if self.ws and self.loop:
            asyncio.run_coroutine_threadsafe(self.ws.close(), self.loop)
        self.quit()
        self.wait()

    def send_message(self, message: dict):
        """Called safely from the main PyQt thread to send data out"""
        if self.ws and self.loop:
            json_str = json.dumps(message)
            asyncio.run_coroutine_threadsafe(self.ws.send(json_str), self.loop)

    async def _ping_loop(self, ws):
        """Sends WebSocket protocol-level pings every 5s to measure latency."""
        while self.running:
            try:
                pong = await ws.ping()
                sent_at = time.time()
                await asyncio.wait_for(pong, timeout=5)
                latency_ms = int((time.time() - sent_at) * 1000)
                self.latency_updated.emit(latency_ms)
            except Exception:
                break
            await asyncio.sleep(5)

    async def _connect_and_listen(self):
        ssl_context = (
            self._create_ssl_context() if self.url.startswith("wss://") else None
        )

        while self.running:
            try:
                connect_options = {"ssl": ssl_context} if ssl_context else {}
                async with websockets.connect(self.url, **connect_options) as ws:
                    self.ws = ws
                    self.connected.emit()

                    ping_task = asyncio.create_task(self._ping_loop(ws))
                    try:
                        async for message in ws:
                            if not self.running:
                                break
                            self._handle_message(json.loads(message))
                    except websockets.ConnectionClosed:
                        pass
                    finally:
                        ping_task.cancel()

            except Exception as e:
                certificate_error = self._certificate_error(e)
                if certificate_error:
                    verification_reason = getattr(
                        certificate_error, "verify_message", str(certificate_error)
                    )
                    message = (
                        "Secure connection failed because the server certificate could not be "
                        f"verified ({verification_reason}). "
                        "Check that your computer's date and time are correct, then update or "
                        "restart VideoSync."
                    )
                    print(f"WebSocket TLS Error: {certificate_error}")
                    self.connection_error.emit(message)
                    self.running = False
                    break
                print(f"WebSocket Error: {e}")

            finally:
                self.ws = None
                self.disconnected.emit()

            if self.running:
                # Reconnect backoff
                await asyncio.sleep(2)

    def _handle_message(self, msg: dict):
        t = msg.get("type")

        if t == "room_created":
            self.room_joined.emit(msg)

        elif t == "room_joined":
            self.room_joined.emit(msg)

        elif t == "room_rejoined":
            self.room_rejoined.emit(msg)

        elif t == "error":
            self.room_error.emit(msg.get("message", "Unknown error"))

        elif t == "room_file_check":
            self.file_check_needed.emit(msg)

        elif t in ["user_joined", "user_left", "user_rejoined"]:
            self.users_updated.emit(msg.get("users", []))

        elif t == "chat":
            self.chat_message.emit(
                msg.get("username", "Unknown"),
                msg.get("message", ""),
                msg.get("timestamp", 0),
            )

        elif t == "sync":
            self.sync_event.emit(msg)

        elif t == "ready_check":
            self.ready_check_updated.emit(msg)

        elif t == "reaction":
            self.fight_reaction.emit(msg)

        elif t == "fight_poll":
            self.fight_poll_updated.emit(msg)

        elif t == "reaction_error":
            self.system_message.emit(msg.get("message", "Reaction unavailable"))