From 23e63c7f6e4558f69c3459a503a626758ed4f3e5 Mon Sep 17 00:00:00 2001 From: Spark Date: Sun, 11 Oct 2026 10:23:50 -0400 Subject: [PATCH] =?UTF-8?q?feat(palier-2):=20webhooks=20Gitea=20sign=C3=A9?= =?UTF-8?q?s,=20flux=20SSE,=20=C3=A9tiquettes,=20milestones?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - POST /webhooks/gitea: HMAC-SHA256 si KANBANTE_WEBHOOK_SECRET, applique fermetures/réouvertures externes de façon idempotente (l'écho de nos propres appels est un no-op, journal source=webhook), invalide le cache, diffuse sur le hub SSE - GET /api/stream (SSE) + LiveHub; générateur testable séparément - synchro d'étiquettes kanban:* à chaque déplacement (ensure/add/ remove granulaires, étiquettes utilisateur intactes, flag KANBANTE_SYNC_LABELS, échec rapporté label_sync sans casser le move) - board ?milestone=: filtre, échéance, progression; client: get_milestones, labels, invalidate_cache public - dashboard: métriques 30 j du journal exposées au template --- app/config.py | 11 ++- app/gitea_client.py | 63 ++++++++++++ app/main.py | 235 ++++++++++++++++++++++++++++++++++++++++++-- 3 files changed, 302 insertions(+), 7 deletions(-) diff --git a/app/config.py b/app/config.py index 25bd331..8a558fc 100644 --- a/app/config.py +++ b/app/config.py @@ -23,6 +23,11 @@ Environment variables: warns, moves are never blocked. KANBANTE_AGING_DAYS Days in the same column before a card is flagged as aging (default 7; "stale" at 2x). + KANBANTE_WEBHOOK_SECRET Shared secret for the Gitea webhook + (POST /webhooks/gitea). When set, the + X-Gitea-Signature HMAC is enforced. + KANBANTE_SYNC_LABELS "1" (default) = mirror intermediate + columns to Gitea labels named kanban:. """ import os @@ -31,7 +36,7 @@ from pathlib import Path BASE_DIR = Path(__file__).resolve().parent.parent -APP_VERSION = "0.3.0" +APP_VERSION = "0.4.0" def _env_bool(name: str, default: bool) -> bool: @@ -81,6 +86,8 @@ class Settings: sync_gitea: bool = True wip_limits: dict | None = None aging_days: int = 7 + webhook_secret: str = "" + sync_labels: bool = True @property def auth_enabled(self) -> bool: @@ -104,4 +111,6 @@ def get_settings() -> Settings: sync_gitea=_env_bool("KANBANTE_SYNC_GITEA", True), wip_limits=_env_wip_limits("KANBANTE_WIP_LIMITS"), aging_days=_env_int("KANBANTE_AGING_DAYS", 7), + webhook_secret=os.getenv("KANBANTE_WEBHOOK_SECRET", ""), + sync_labels=_env_bool("KANBANTE_SYNC_LABELS", True), ) diff --git a/app/gitea_client.py b/app/gitea_client.py index 4db55be..64ed967 100644 --- a/app/gitea_client.py +++ b/app/gitea_client.py @@ -120,6 +120,10 @@ class GiteaClient: def _cache_clear(self) -> None: self._cache.clear() + def invalidate_cache(self) -> None: + """Drop every cached list response (called on webhook events).""" + self._cache.clear() + # ── API ──────────────────────────────────────────────── async def ping(self) -> bool: @@ -221,6 +225,65 @@ class GiteaClient: r.raise_for_status() return r.json() + # ── Labels (kanban:* column mirroring) ───────────────── + + async def get_labels(self, owner: str, repo: str) -> list[dict]: + return await self._get_all_pages(f"/repos/{owner}/{repo}/labels") + + async def create_label( + self, owner: str, repo: str, name: str, color: str + ) -> dict: + r = await self.client.post( + f"/repos/{owner}/{repo}/labels", + json={"name": name, "color": color.lstrip("#")}, + ) + r.raise_for_status() + return r.json() + + async def ensure_label( + self, owner: str, repo: str, name: str, color: str + ) -> dict: + """Return the repo label with this name, creating it if needed.""" + key = f"{owner}/{repo}" + cache = self.__dict__.setdefault("_label_cache", {}) + if key not in cache: + cache[key] = {l["name"]: l for l in await self.get_labels(owner, repo)} + if name not in cache[key]: + cache[key][name] = await self.create_label(owner, repo, name, color) + return cache[key][name] + + async def add_issue_label( + self, owner: str, repo: str, number: int, label_id: int + ) -> None: + r = await self.client.post( + f"/repos/{owner}/{repo}/issues/{number}/labels", + json={"labels": [label_id]}, + ) + r.raise_for_status() + + async def remove_issue_label( + self, owner: str, repo: str, number: int, label_id: int + ) -> None: + r = await self.client.delete( + f"/repos/{owner}/{repo}/issues/{number}/labels/{label_id}" + ) + r.raise_for_status() + + # ── Milestones ───────────────────────────────────────── + + async def get_milestones( + self, owner: str, repo: str, state: str = "open" + ) -> list[dict]: + key = f"milestones:{owner}/{repo}:{state}" + cached = self._cache_get(key) + if cached is not None: + return cached + milestones = await self._get_all_pages( + f"/repos/{owner}/{repo}/milestones", state=state + ) + self._cache_put(key, milestones) + return milestones + async def create_issue( self, owner: str, repo: str, title: str, body: str = "" ) -> dict: diff --git a/app/main.py b/app/main.py index 0af7536..73c1c32 100644 --- a/app/main.py +++ b/app/main.py @@ -16,6 +16,9 @@ FastAPI backend serving HTML + JSON API. import asyncio import base64 +import hashlib +import hmac +import json import logging import secrets from contextlib import asynccontextmanager @@ -24,7 +27,7 @@ from pathlib import Path import httpx from fastapi import FastAPI, Form, Request -from fastapi.responses import HTMLResponse, JSONResponse, Response +from fastapi.responses import HTMLResponse, JSONResponse, Response, StreamingResponse from fastapi.staticfiles import StaticFiles from jinja2 import Environment, FileSystemLoader, select_autoescape from starlette.middleware.base import BaseHTTPMiddleware @@ -119,7 +122,9 @@ class AuthMiddleware(BaseHTTPMiddleware): """ async def dispatch(self, request: Request, call_next): - if request.url.path == "/api/health": + # /api/health: liveness for Docker. /webhooks/gitea: authenticated + # by its HMAC signature, not by Basic auth (Gitea cannot log in). + if request.url.path in ("/api/health", "/webhooks/gitea"): return await call_next(request) settings = get_settings() if not settings.auth_enabled: @@ -158,6 +163,37 @@ class AuthMiddleware(BaseHTTPMiddleware): app.add_middleware(AuthMiddleware) +# ── Live updates (SSE hub) ───────────────────────────────── + +class LiveHub: + """Fan-out of Gitea webhook events to connected browsers (SSE).""" + + def __init__(self) -> None: + self._subscribers: set[asyncio.Queue] = set() + + def subscribe(self) -> asyncio.Queue: + queue: asyncio.Queue = asyncio.Queue(maxsize=50) + self._subscribers.add(queue) + return queue + + def unsubscribe(self, queue: asyncio.Queue) -> None: + self._subscribers.discard(queue) + + async def broadcast(self, payload: dict) -> None: + for queue in list(self._subscribers): + try: + queue.put_nowait(payload) + except asyncio.QueueFull: + pass # slow client: it will resync on its next load + + @property + def subscriber_count(self) -> int: + return len(self._subscribers) + + +hub = LiveHub() + + # ── Templates ────────────────────────────────────────────── _tpl_env = Environment( @@ -311,12 +347,13 @@ async def dashboard(request: Request): COLUMN_LABELS=COLUMN_LABELS, COLUMN_COLORS=COLUMN_COLORS, active_window_days=get_settings().active_window_days, + metrics=get_metrics(window_days=30), ) @app.get("/board/{owner}/{repo}", response_class=HTMLResponse) -async def board(request: Request, owner: str, repo: str): - """Kanban board for a specific repo.""" +async def board(request: Request, owner: str, repo: str, milestone: str = ""): + """Kanban board for a specific repo (optionally one milestone).""" full_name = f"{owner}/{repo}" all_items = await gitea.get_repo_issues(owner, repo, state="all") issues = [i for i in all_items if _is_issue(i)] @@ -329,8 +366,47 @@ async def board(request: Request, owner: str, repo: str): for number in changes["closed_numbers"]: log_event(full_name, number, "closed", to_column="done", source="gitea") + # Milestone view: filter the board to one milestone (sprint-style), + # with its deadline and completion progress. + milestones: list[dict] = [] + selected_milestone = None + milestone_progress = None + display_issues = open_issues + try: + milestones = await gitea.get_milestones(owner, repo) + except Exception as exc: # milestones are decoration, never fatal + logger.debug("milestones failed for %s: %s", full_name, exc) + milestones = [] + if milestone: + def _in_milestone(issue: dict) -> bool: + ms = issue.get("milestone") or {} + return bool(ms) and ( + str(ms.get("id")) == milestone or ms.get("title") == milestone + ) + + display_issues = [i for i in open_issues if _in_milestone(i)] + selected_milestone = next( + ( + m for m in milestones + if str(m.get("id")) == milestone or m.get("title") == milestone + ), + None, + ) + if selected_milestone is None and display_issues: + selected_milestone = display_issues[0].get("milestone") + scoped = [i for i in issues if _in_milestone(i)] + total = len(scoped) + closed_count = sum(1 for i in scoped if i.get("state") == "closed") + milestone_progress = { + "total": total, + "closed": closed_count, + "pct": round(100 * closed_count / total) if total else 0, + "due_on": (selected_milestone or {}).get("due_on"), + "title": (selected_milestone or {}).get("title", milestone), + } + board_state = get_board_state(full_name) - columns = _merge_issues_with_board(open_issues, board_state) + columns = _merge_issues_with_board(display_issues, board_state) labels: dict[str, dict] = {} for issue in open_issues: @@ -356,9 +432,13 @@ async def board(request: Request, owner: str, repo: str): labels=sorted(labels.values(), key=lambda l: l["name"].lower()), notes=get_notes(full_name) or "", repo_info=repo_info, - ages=_card_ages(full_name, open_issues, board_state), + ages=_card_ages(full_name, display_issues, board_state), wip_limits=settings.wip_limits or {}, aging_days=settings.aging_days, + milestones=milestones, + selected_milestone=selected_milestone, + milestone_progress=milestone_progress, + milestone_param=milestone, ) @@ -429,6 +509,117 @@ async def focus(request: Request): ) +# ── Webhooks + live stream ───────────────────────────────── + +def _apply_external_transition( + full_name: str, number: int, target_column: str, event_type: str +) -> str: + """Apply a close/reopen that happened in Gitea itself. + + Idempotent by construction: when Kanbante's own API made the + change, the local row is already in the target state and this is + a no-op — so our own webhook echo never double-logs. + """ + board = get_board_state(full_name) + current = None + for column, cards in board.items(): + if any(c["issue_number"] == number for c in cards): + current = column + break + if current is None or current == target_column: + return "noop" + if event_type == "reopened" and current != "done": + return "noop" # locally open already: nothing truthful to apply + move_issue(full_name, number, target_column) + if event_type == "closed": + log_event(full_name, number, "closed", to_column="done", source="webhook") + else: + log_event( + full_name, number, "reopened", + from_column="done", to_column=target_column, source="webhook", + ) + return event_type + + +@app.post("/webhooks/gitea") +async def gitea_webhook(request: Request): + """Receive Gitea webhooks: verify HMAC, apply external close/reopen, + invalidate the read cache and push a live event to browsers.""" + settings = get_settings() + raw = await request.body() + if settings.webhook_secret: + signature = request.headers.get("x-gitea-signature", "") + expected = hmac.new( + settings.webhook_secret.encode(), raw, hashlib.sha256 + ).hexdigest() + if not signature or not hmac.compare_digest(signature, expected): + return JSONResponse( + {"ok": False, "error": "Invalid webhook signature"}, + status_code=401, + ) + try: + payload = json.loads(raw or b"{}") + except ValueError: + return JSONResponse( + {"ok": False, "error": "Invalid JSON payload"}, status_code=400 + ) + + event = request.headers.get("x-gitea-event", "") + action = payload.get("action", "") + full_name = (payload.get("repository") or {}).get("full_name", "") + issue = payload.get("issue") or {} + number = issue.get("number") + + handled = "ignored" + if event == "issues" and full_name and number: + if action == "closed": + handled = _apply_external_transition(full_name, number, "done", "closed") + elif action == "reopened": + handled = _apply_external_transition( + full_name, number, "backlog", "reopened" + ) + + gitea.invalidate_cache() + await hub.broadcast( + {"repo": full_name, "event": event, "action": action, "issue": number} + ) + return {"ok": True, "handled": handled} + + +async def sse_event_stream(request: Request): + """Yield SSE frames for one subscriber until it disconnects. + + Split out of the route so it can be unit-tested directly: + test clients cannot reliably tear down an infinite HTTP stream. + """ + queue = hub.subscribe() + try: + yield "retry: 5000\n\n" + while True: + try: + payload = await asyncio.wait_for(queue.get(), timeout=25) + except asyncio.TimeoutError: + if await request.is_disconnected(): + break + yield ": heartbeat\n\n" + continue + if await request.is_disconnected(): + break + yield f"event: gitea\ndata: {json.dumps(payload)}\n\n" + finally: + hub.unsubscribe(queue) + + +@app.get("/api/stream") +async def live_stream(request: Request): + """Server-Sent Events: browsers reload softly on Gitea activity.""" + return StreamingResponse( + sse_event_stream(request), + media_type="text/event-stream", + headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}, + ) + + @app.get("/activity", response_class=HTMLResponse) async def activity(request: Request): """Activity feed: the latest entries of the event journal.""" @@ -517,6 +708,38 @@ async def api_move( exc, ) result["gitea_sync"] = "failed" + + # Mirror intermediate columns to Gitea labels (kanban:) so + # the workflow state is visible in Gitea itself, not just here. + result["label_sync"] = "skipped" + if settings.sync_labels and from_column != to_column: + try: + desired = ( + f"kanban:{to_column}" + if to_column in ("todo", "in_progress", "review") + else None + ) + issue = await gitea.get_issue(owner, repo, issue_number) + current = [ + label for label in issue.get("labels") or [] + if str(label.get("name", "")).startswith("kanban:") + ] + for label in current: + if label.get("name") != desired: + await gitea.remove_issue_label( + owner, repo, issue_number, label["id"] + ) + if desired and all(l.get("name") != desired for l in current): + label = await gitea.ensure_label( + owner, repo, desired, COLUMN_COLORS[to_column] + ) + await gitea.add_issue_label(owner, repo, issue_number, label["id"]) + result["label_sync"] = "ok" + except Exception as exc: + logger.warning( + "Label sync failed for %s#%s: %s", repo_full_name, issue_number, exc + ) + result["label_sync"] = "failed" return result