path: root/sb_builder/mainwindow.py

# sb-builder - main window
# Copyright (C) 2026 Danilo M. <danix@danix.xyz>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 2 as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.

"""Main window: tabs, status bar, connection polling, VM control, notifications."""

import time

from PyQt6.QtCore import QMetaType, QObject, QSize, Qt, QTimer, pyqtSignal, pyqtSlot
from PyQt6.QtDBus import QDBusArgument, QDBusConnection, QDBusInterface, QDBusMessage
from PyQt6.QtGui import QIcon, QKeySequence, QShortcut
from PyQt6.QtWidgets import (QLabel, QMainWindow, QMessageBox, QPushButton, QStyle, QTabBar,
                             QTabWidget, QToolBar)

from . import VERSION, config
from .dashboard import Dashboard
from .fmt import job_title
from .hints_tab import HINT_JOBS, HintResultDialog, HintsTab
from .jobs import JobsTab
from .packages import PackagesTab
from .results import ResultsTab
from .remote import DOWN, MISSING

NOTIFY = "org.freedesktop.Notifications"
NOTIFY_PATH = "/org/freedesktop/Notifications"
# ribbon entries: label, freedesktop icon name, Qt fallback icon
RIBBON = (("Dashboard", "go-home", QStyle.StandardPixmap.SP_DirHomeIcon),
          ("Packages", "package-x-generic", QStyle.StandardPixmap.SP_FileDialogDetailedView),
          ("Hints", "document-edit", QStyle.StandardPixmap.SP_FileDialogContentsView),
          ("Results", "view-list-text", QStyle.StandardPixmap.SP_MessageBoxWarning),
          ("Jobs", "system-run", QStyle.StandardPixmap.SP_MediaPlay))
# tabs as a distinct, centered row; colors from the palette so the theme applies
TAB_STYLE = """
QTabWidget::tab-bar { alignment: center; }
QTabWidget::pane { border-top: 2px solid palette(highlight); top: -2px; }
QTabBar::tab { padding: 6px 20px; margin: 0 3px; border: 1px solid palette(mid);
               border-bottom: none; border-top-left-radius: 8px; border-top-right-radius: 8px;
               background: palette(button); color: palette(button-text); }
QTabBar::tab:selected { background: palette(highlight); color: palette(highlighted-text); }
QTabBar::tab:hover:!selected { background: palette(midlight); }
"""
END_EVENT = {"ok": "finished", "failed": "failed", "killed": "killed", "lost": "lost"}
MAX_RETRY_MS = 30000


class Notifier(QObject):
    """freedesktop notifications over D-Bus; clicked(job id) on the default action."""

    clicked = pyqtSignal(str)

    def __init__(self, parent=None):
        super().__init__(parent)
        bus = QDBusConnection.sessionBus()
        self.iface = QDBusInterface(NOTIFY, NOTIFY_PATH, NOTIFY, bus)
        bus.connect(NOTIFY, NOTIFY_PATH, NOTIFY, "ActionInvoked", self._action)
        self.ids = {}  # notification id -> job id

    def send(self, title, body, jid=""):
        reply = self.iface.call(
            "Notify", "sb-builder", QDBusArgument(0, QMetaType.Type.UInt.value), "", title, body,
            QDBusArgument(["default", "Open"], QMetaType.Type.QStringList.value), {}, -1)
        if reply.type() == QDBusMessage.MessageType.ReplyMessage:
            self.ids[reply.arguments()[0]] = jid

    @pyqtSlot("uint", "QString")
    def _action(self, nid, _key):
        if nid in self.ids:
            self.clicked.emit(self.ids.pop(nid))


class MainWindow(QMainWindow):
    def __init__(self, cfg, remote, notifier=None):
        super().__init__()
        self.cfg, self.remote = cfg, remote
        self.notifier = notifier or Notifier(self)
        self.notifier.clicked.connect(self.open_job)
        self.jobs_tab = None
        self.packages_tab = None
        self.results_tab = None
        self.hints_tab = None
        self.hint_dialogs = []  # open HintResultDialogs, kept alive while shown
        self.seen = None        # job id -> status at the last poll; None before the first
        self.online = None      # None until the first answer
        self.overlays = []
        self.vm_action = None   # "start" or "resume" offered on the VM button
        self.vm_woken = False   # we started the VM: notify once ssh answers
        self.running_id = None
        self.setWindowTitle("sb-builder")

        self.tabs = QTabWidget(tabsClosable=True)
        self.tabs.setStyleSheet(TAB_STYLE)
        self.tabs.tabCloseRequested.connect(self._close_tab)
        self.tabs.currentChanged.connect(lambda _i: self.poll_stats())
        self.dashboard = Dashboard()
        self.dashboard.refresh_btn.clicked.connect(self.poll_stats)
        self.dashboard.tiles["Packages"].link(self.open_packages)
        self.dashboard.jobs.link(self.open_jobs)
        self.dashboard.result_clicked.connect(self.open_results)
        self.tabs.addTab(self.dashboard, "Dashboard")
        self.tabs.tabBar().setTabButton(0, QTabBar.ButtonPosition.RightSide, None)
        self.setCentralWidget(self.tabs)

        file_menu = self.menuBar().addMenu("&File")
        file_menu.addAction("&Reload config", self.reload_config)
        file_menu.addAction("&Quit", self.close).setShortcut("Ctrl+Q")
        # window-wide: QTabWidget's own Ctrl+PgUp/PgDn needs focus on the tabs
        self.tab_next = QShortcut(QKeySequence("Ctrl+PgDown"), self, lambda: self._cycle_tab(1))
        self.tab_prev = QShortcut(QKeySequence("Ctrl+PgUp"), self, lambda: self._cycle_tab(-1))
        self.ribbon = QToolBar("Screens", movable=False, floatable=False)
        self.ribbon.setToolButtonStyle(Qt.ToolButtonStyle.ToolButtonTextUnderIcon)
        self.ribbon.setIconSize(QSize(32, 32))
        self.addToolBar(Qt.ToolBarArea.TopToolBarArea, self.ribbon)
        slots = {"Dashboard": lambda: self.tabs.setCurrentWidget(self.dashboard),
                 "Packages": self.open_packages, "Hints": lambda: self.open_hints(),
                 "Results": lambda: self.open_results(),
                 "Jobs": self.open_jobs}
        for label, name, fallback in RIBBON:
            icon = QIcon.fromTheme(name, self.style().standardIcon(fallback))
            self.ribbon.addAction(icon, label, slots[label])

        self.conn = QLabel("● connecting")
        self.key = QLabel()
        self.key.hide()
        self.vm_btn = QPushButton()
        self.vm_btn.clicked.connect(lambda: self._vm_do(self.vm_action))
        self.vm_btn.hide()
        self.job_btn = QPushButton(flat=True)
        self.job_btn.clicked.connect(lambda: self.open_job(self.running_id))
        self.job_btn.hide()
        for w in (self.job_btn, self.key, self.vm_btn, self.conn):
            self.statusBar().addPermanentWidget(w)

        self.jobs_timer = QTimer(self, timeout=self.poll_jobs)
        self.stats_timer = QTimer(self, timeout=self.poll_slow)
        self._start_timers()
        QTimer.singleShot(0, self.poll_jobs)

    # ── polling ─────────────────────────────────────────────────────────────

    def _start_timers(self):
        self.jobs_timer.start(self.cfg["refresh"]["jobs"] * 1000)
        self.stats_timer.start(self.cfg["refresh"]["dashboard"] * 1000)

    def poll_jobs(self):
        self.remote.call(self._got_jobs, "jobs")

    def poll_slow(self):
        self.poll_stats()
        if self.online:
            self.remote.call(self._got_key, "gpg-status")

    def _got_key(self, data, err):
        if err:
            return self._failed(err)
        self.key.setVisible(data["unlocked"] is not None)
        self.key.setText("🔓 key unlocked" if data["unlocked"] else "🔒 key locked")
        self.key.setToolTip(f"package signing key {data['key']}")

    def poll_stats(self):
        if self.online and self.tabs.currentWidget() is self.dashboard:
            self.remote.call(self._got_stats, "stats")

    def _got_jobs(self, data, err):
        if err:
            return self._failed(err)
        self._up()
        jobs = data["jobs"]
        self._notify_ends(jobs)
        self.dashboard.show_jobs(jobs)
        if self.jobs_tab:
            self.jobs_tab.set_jobs(jobs)
        run = next((j for j in jobs if j["status"] == "running"), None)
        self.running_id = run and run["id"]
        self.job_btn.setVisible(run is not None)
        if run:
            self.job_btn.setText(f"▶ {job_title(run)}")

    def _got_stats(self, data, err):
        if err:
            return self._failed(err)
        self.dashboard.show_stats(data)

    def _got_version(self, data, err):
        if err:
            return self._failed(err)
        self.overlays = data.get("overlays", [])
        if self.jobs_tab:
            self.jobs_tab.set_overlays(self.overlays)
        if data["version"] != VERSION:
            self.conn.setText(f"● helper {data['version']}, GUI {VERSION}: run ./install.sh --remote")

    def _up(self):
        if self.online:
            return
        self.online = True
        self.conn.setText("● connected")
        self.vm_btn.hide()
        self.dashboard.set_stale(None)
        self.jobs_timer.setInterval(self.cfg["refresh"]["jobs"] * 1000)
        self.remote.call(self._got_version, "version")
        self.poll_slow()
        if self.vm_woken:
            self.vm_woken = False
            self.notify("vm-up", f"{self.cfg['vm']['domain']} is up")

    def _failed(self, err):
        if err == MISSING:
            self.conn.setText("● sbb-remote missing: run ./install.sh --remote")
            return
        if err != DOWN:
            self.statusBar().showMessage(err, 10000)
            return
        if self.online is not False:
            self.online = False
            self.dashboard.set_stale(time.time())
        base = self.cfg["refresh"]["jobs"] * 1000
        self.jobs_timer.setInterval(min(self.jobs_timer.interval() * 2, max(MAX_RETRY_MS, base)))
        self._check_vm()

    # ── VM control ──────────────────────────────────────────────────────────

    def _check_vm(self):
        if not self.cfg["vm"]["domain"]:
            self.conn.setText("● down")
            return
        self.remote.virsh(self._got_domstate, "domstate", self.cfg["vm"]["domain"])

    def _got_domstate(self, rc, state):
        if self.online:
            return  # ssh came back while virsh ran
        self.vm_btn.hide()
        if rc != 0:
            self.conn.setText("● down")
        elif state == "shut off":
            self.conn.setText("● VM off")
            if self.cfg["vm"]["autostart"]:
                self._vm_do("start")
            else:
                self._offer("Start VM", "start")
        elif state == "paused":
            self.conn.setText("● VM paused")
            self._offer("Resume VM", "resume")
        elif state == "running":
            self.conn.setText("● VM running, ssh unreachable")
        else:
            self.conn.setText(f"● VM {state}")

    def _offer(self, label, action):
        self.vm_action = action
        self.vm_btn.setText(label)
        self.vm_btn.show()

    def _vm_do(self, action):
        self.vm_btn.hide()
        self.vm_woken = True
        self.conn.setText(f"● VM {action} requested")
        self.remote.virsh(
            lambda rc, out: rc and self.statusBar().showMessage(f"virsh {action}: {out}", 10000),
            action, self.cfg["vm"]["domain"])

    # ── notifications ───────────────────────────────────────────────────────

    def notify(self, event, body, jid=""):
        n = self.cfg["notify"]
        if n["enabled"] and event in n["events"]:
            self.notifier.send(f"sb-builder: {event}", body, jid)

    def _notify_ends(self, jobs):
        prev, self.seen = self.seen, {j["id"]: j["status"] for j in jobs}
        if prev is None:
            return  # first poll: old jobs are history, not news
        for j in jobs:
            event = END_EVENT.get(j["status"])
            if event and prev.get(j["id"], "running") != j["status"]:
                self.notify(event, job_title(j), j["id"])
                if j["kind"] in ("slackrepo", "publish"):  # build state changed
                    self._reload(self.packages_tab, self.results_tab)
                if j["kind"] in HINT_JOBS and j["status"] in ("ok", "failed"):
                    self._reload(self.packages_tab, self.hints_tab)
                    self.remote.call(lambda d, err: d and self.show_hint_result(d), "job-result", j["id"])

    @staticmethod
    def _reload(*tabs):
        for tab in tabs:
            if tab:
                tab.reload()

    def show_hint_result(self, result):
        if not result["ok"] and not result["failed"]:
            return  # nothing changed, nothing failed: no news
        d = HintResultDialog(self.remote, result, self)
        d.build_requested.connect(
            lambda pkgs: self.open_jobs().start("slackrepo", "update", "--preview", *pkgs))
        d.changed.connect(lambda: self._reload(self.packages_tab, self.hints_tab))
        d.finished.connect(lambda _r: self.hint_dialogs.remove(d))
        self.hint_dialogs.append(d)
        d.show()

    # ── tabs ────────────────────────────────────────────────────────────────

    def open_jobs(self):
        if self.jobs_tab is None:
            self.jobs_tab = JobsTab(self.remote, self.cfg["builds"]["publish"])
            self.jobs_tab.refresh_requested.connect(self.poll_jobs)
            self.jobs_tab.set_overlays(self.overlays)
            self.tabs.addTab(self.jobs_tab, "Jobs")
            self.poll_jobs()
        self.tabs.setCurrentWidget(self.jobs_tab)
        return self.jobs_tab

    def _cycle_tab(self, step):
        self.tabs.setCurrentIndex((self.tabs.currentIndex() + step) % self.tabs.count())

    def open_packages(self):
        if self.packages_tab is None:
            self.packages_tab = PackagesTab(self.remote)
            self.packages_tab.start_requested.connect(lambda args: self.open_jobs().start(*args))
            self.packages_tab.hints_requested.connect(self.open_hints)
            self.tabs.addTab(self.packages_tab, "Packages")
            self.packages_tab.reload()
        self.tabs.setCurrentWidget(self.packages_tab)
        return self.packages_tab

    def open_hints(self, name=None):
        if self.hints_tab is None:
            self.hints_tab = HintsTab(self.remote, self.cfg)
            self.hints_tab.start_requested.connect(lambda args: self.open_jobs().start(*args))
            self.hints_tab.result_ready.connect(self.show_hint_result)
            self.tabs.addTab(self.hints_tab, "Hints")
            self.hints_tab.reload()
        if name:
            self.hints_tab.select(name)
        self.tabs.setCurrentWidget(self.hints_tab)
        return self.hints_tab

    def open_results(self, result=None):
        if self.results_tab is None:
            self.results_tab = ResultsTab(self.remote)
            self.tabs.addTab(self.results_tab, "Results")
            self.results_tab.reload()
        if result:
            self.results_tab.filter_to(result)
        self.tabs.setCurrentWidget(self.results_tab)
        return self.results_tab

    def open_job(self, jid):
        self.show()
        self.raise_()
        self.activateWindow()
        if jid:
            self.open_jobs().select(jid)

    def _close_tab(self, i):
        w = self.tabs.widget(i)
        if w is self.dashboard:
            return
        self.tabs.removeTab(i)
        if w is self.jobs_tab:
            self.jobs_tab = None
            w.stop()
        if w is self.packages_tab:
            self.packages_tab = None
        if w is self.results_tab:
            self.results_tab = None
        if w is self.hints_tab:
            self.hints_tab = None
        w.deleteLater()

    def reload_config(self):
        cfg, err = config.load()
        if err:
            QMessageBox.warning(self, "Config error", f"{err}\nUsing defaults.")
        self.cfg.clear()
        self.cfg.update(cfg)  # Remote holds the same dict
        self._start_timers()