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, QTabWidget,
                             QToolBar, QToolButton)

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, NextStep, is_preview
from .packages import PackagesTab
from .results import ResultsTab
from .remote import DOWN, MISSING

NOTIFY = "org.freedesktop.Notifications"
NOTIFY_PATH = "/org/freedesktop/Notifications"
# Jobs tab icon: running, then the end status until the tab is shown
BADGE = {"running": ("system-run", QStyle.StandardPixmap.SP_MediaPlay),
         "ok": ("emblem-ok", QStyle.StandardPixmap.SP_DialogApplyButton),
         "bad": ("dialog-error", QStyle.StandardPixmap.SP_MessageBoxCritical)}
# 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.hint_dialogs = []  # open HintResultDialogs, kept alive while shown
        self.next_steps = {}    # job id -> its open NextStep dialog
        self.mine = set()       # ids of the jobs this session started: only they get next steps
        self.badge = None       # Jobs tab icon state: None, "running", "ok", "bad"
        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()
        self.tabs.setStyleSheet(TAB_STYLE)
        self.dashboard = Dashboard()
        self.dashboard.refresh.triggered.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.jobs_tab = JobsTab(self.remote)
        self.jobs_tab.refresh_requested.connect(self.poll_jobs)
        self.jobs_tab.started.connect(self.mine.add)
        self.jobs_tab.preview_requested.connect(self.show_next_step)
        self.packages_tab = PackagesTab(self.remote)
        self.packages_tab.start_requested.connect(lambda args: self.jobs_tab.start(*args))
        self.packages_tab.hints_requested.connect(self.open_hints)
        self.hints_tab = HintsTab(self.remote, self.cfg)
        self.hints_tab.start_requested.connect(lambda args: self.jobs_tab.start(*args))
        self.hints_tab.result_ready.connect(self.show_hint_result)
        self.results_tab = ResultsTab(self.remote)
        # permanent tabs (UI guidelines); each loads on first show
        for tab, label in ((self.dashboard, "Dashboard"), (self.packages_tab, "Packages"),
                           (self.hints_tab, "Hints"), (self.results_tab, "Results"),
                           (self.jobs_tab, "Jobs")):
            self.tabs.addTab(tab, label)
        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))
        # main ribbon: the current tab's actions; small ribbon under it: its view controls
        self.ribbon = QToolBar("Actions", movable=False, floatable=False)
        self.ribbon.setToolButtonStyle(Qt.ToolButtonStyle.ToolButtonTextUnderIcon)
        self.ribbon.setIconSize(QSize(32, 32))
        self.addToolBar(Qt.ToolBarArea.TopToolBarArea, self.ribbon)
        self.addToolBarBreak()
        self.view_bar = QToolBar("View", movable=False, floatable=False)
        self.view_bar.setToolButtonStyle(Qt.ToolButtonStyle.ToolButtonTextBesideIcon)
        self.view_bar.setIconSize(QSize(16, 16))
        self.addToolBar(Qt.ToolBarArea.TopToolBarArea, self.view_bar)
        self.tabs.currentChanged.connect(self._tab_changed)
        self._tab_changed()

        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)
        self.jobs_tab.set_jobs(jobs)
        run = next((j for j in jobs if j["status"] == "running"), None)
        if run:
            self._badge("running")
        elif self.badge == "running":
            self._badge(None)  # ended unseen (first poll, reconnect): nothing to report
        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", [])
        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 self.tabs.currentWidget() is not self.jobs_tab:
                    self._badge("ok" if j["status"] == "ok" else "bad")
                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)
                if j["id"] not in self.mine:
                    continue  # started elsewhere or before this session: news, not a next step
                if j["kind"] in HINT_JOBS and j["status"] in ("ok", "failed"):
                    self.remote.call(lambda d, err: d and self.show_hint_result(d), "job-result", j["id"])
                elif is_preview(j):
                    self.show_next_step(j, auto=True)

    @staticmethod
    def _reload(*tabs):
        for tab in tabs:
            if tab.loaded:  # a tab not shown yet loads fresh when it is
                tab.reload()

    def show_next_step(self, job, auto=False):
        """The preview's next step as a dialog; auto: a shadowing preview without a plan shows none."""
        if job["id"] in self.next_steps:
            self.next_steps[job["id"]].raise_()
            return
        d = NextStep(self.remote, job, self.cfg["builds"]["publish"], self)
        d.run.connect(lambda args: (self.jobs_tab.start_signed if args[0] == "slackrepo"
                                    else self.jobs_tab.start)(*args))
        d.finished.connect(lambda _r: self.next_steps.pop(job["id"], None))
        self.next_steps[job["id"]] = d
        if d.shadow and auto:
            d.loaded.connect(lambda plan: d.show() if plan else d.reject())
        else:
            d.show()

    def _badge(self, state):
        self.badge = state
        i = self.tabs.indexOf(self.jobs_tab)
        # in the text, not a color: TAB_STYLE's tab color would override setTabTextColor
        self.tabs.setTabText(i, "Jobs (running)" if state == "running" else "Jobs")
        if state is None:
            self.tabs.setTabIcon(i, QIcon())
            return
        name, fallback = BADGE[state]
        self.tabs.setTabIcon(i, QIcon.fromTheme(name, self.style().standardIcon(fallback)))

    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.jobs_tab.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 _tab_changed(self, _i=None):
        tab = self.tabs.currentWidget()
        for bar, acts in ((self.ribbon, tab.actions), (self.view_bar, tab.view_actions)):
            bar.clear()
            bar.addActions(acts)
            for a in acts:
                if a.menu():  # a menu action opens its menu on a plain click
                    bar.widgetForAction(a).setPopupMode(QToolButton.ToolButtonPopupMode.InstantPopup)
        self.view_bar.setVisible(bool(tab.view_actions))
        if not getattr(tab, "loaded", True):
            tab.reload()
        if tab is self.jobs_tab and self.badge != "running":
            self._badge(None)
        self.poll_stats()

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

    def open_jobs(self):
        self.tabs.setCurrentWidget(self.jobs_tab)
        return self.jobs_tab

    def open_packages(self):
        self.tabs.setCurrentWidget(self.packages_tab)
        return self.packages_tab

    def open_hints(self, name=None):
        self.tabs.setCurrentWidget(self.hints_tab)
        if name:
            self.hints_tab.select(name)
        return self.hints_tab

    def open_results(self, result=None):
        self.tabs.setCurrentWidget(self.results_tab)
        if result:
            self.results_tab.filter_to(result)
        return self.results_tab

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

    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()