kadence/desktop/kadence_desktop/worker.py

"""The network thread.

Every HTTP call happens here, on a worker thread, so a slow or dead backend can
never freeze the window — the UI keeps ticking its countdown from the last known
state and shows a "reconnecting" chip instead.

Commands go in through a queued slot and come back out as signals; the GUI side
never touches `Api` directly.
"""

from __future__ import annotations

import logging
from typing import Any

from PyQt6.QtCore import QObject, QTimer, pyqtSignal, pyqtSlot

from kadence_desktop.api import Api, ApiError

log = logging.getLogger(__name__)


class ApiWorker(QObject):
    stateChanged = pyqtSignal(dict)
    policyChanged = pyqtSignal(dict)
    tasksChanged = pyqtSignal(dict)
    connectionChanged = pyqtSignal(bool, str)
    commandFailed = pyqtSignal(str, str)
    commandDone = pyqtSignal(str, dict)

    def __init__(self, api: Api, poll_seconds: int = 5) -> None:
        super().__init__()
        self._api = api
        self._poll_seconds = poll_seconds
        self._timer: QTimer | None = None
        self._connected: bool | None = None
        self._slow_tick = 0

    # ---- lifecycle ---------------------------------------------------------

    @pyqtSlot()
    def start(self) -> None:
        """Called once the worker is on its own thread."""
        self._timer = QTimer(self)
        self._timer.timeout.connect(self.poll)
        self._timer.start(max(1, self._poll_seconds) * 1000)
        self.poll()
        self.refresh_tasks()

    @pyqtSlot(int)
    def set_interval(self, seconds: int) -> None:
        if self._timer:
            self._timer.start(max(1, seconds) * 1000)

    @pyqtSlot(str, str)
    def configure(self, base_url: str, token: str) -> None:
        self._api.configure(base_url, token)
        self._connected = None
        self.poll()

    # ---- polling -----------------------------------------------------------

    @pyqtSlot()
    def poll(self) -> None:
        try:
            state = self._api.timer_state()
            policy = self._api.blocker_policy()
        except ApiError as exc:
            self._set_connected(False, str(exc))
            return
        self._set_connected(True, "")
        self.stateChanged.emit(state)
        self.policyChanged.emit(policy)

        # The task list changes far more slowly than the clock does, and all the
        # app wants from it now is which one is pinned.
        self._slow_tick += 1
        if self._slow_tick % 12 == 0:
            self.refresh_tasks()

    def _set_connected(self, connected: bool, detail: str) -> None:
        if connected != self._connected:
            self._connected = connected
            self.connectionChanged.emit(connected, detail)

    @pyqtSlot()
    def refresh_tasks(self) -> None:
        try:
            self.tasksChanged.emit(self._api.tasks())
        except ApiError as exc:
            log.debug("tasks: %s", exc)

    # ---- commands ----------------------------------------------------------

    @pyqtSlot(str, dict)
    def command(self, name: str, args: dict[str, Any]) -> None:
        """Run one API action by name and re-poll so the UI lands on the truth."""
        handlers = {
            "start": lambda: self._api.start(args.get("kind"), args.get("task_id"), args.get("minutes")),
            "pause": self._api.pause,
            "resume": self._api.resume,
            "complete": lambda: self._api.complete(bool(args.get("auto_next", True))),
            "abandon": lambda: self._api.abandon(args.get("reason")),
            "extend": lambda: self._api.extend(int(args.get("minutes", 5))),
            "interrupt": self._api.interrupt,
            "refresh_sources": self._api.refresh_tasks,
            "send_report": self._api.send_report,
            "patch_settings": lambda: self._api.patch_settings(dict(args.get("patch", {}))),
            "report_block": lambda: self._api.report_block(
                str(args.get("kind", "app")),
                str(args.get("target", "")),
                str(args.get("action", "blocked")),
                args.get("details"),
            ),
        }
        handler = handlers.get(name)
        if handler is None:
            self.commandFailed.emit(name, f"unknown command {name!r}")
            return
        try:
            result = handler()
        except ApiError as exc:
            self.commandFailed.emit(name, str(exc))
            return
        except Exception as exc:  # pragma: no cover
            log.exception("command %s blew up", name)
            self.commandFailed.emit(name, str(exc))
            return

        self.commandDone.emit(name, result if isinstance(result, dict) else {})
        if name == "refresh_sources":
            self.refresh_tasks()
        if name != "report_block":
            self.poll()