feat(palier-2): webhooks Gitea signés, flux SSE, étiquettes, milestones

- 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=<titre|id>: filtre, échéance, progression;
  client: get_milestones, labels, invalidate_cache public
- dashboard: métriques 30 j du journal exposées au template
This commit is contained in:
Spark
2026-10-11 10:23:50 -04:00
parent 3cac4fe9c3
commit 23e63c7f6e
3 changed files with 302 additions and 7 deletions
+10 -1
View File
@@ -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:<col>.
"""
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),
)
+63
View File
@@ -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:
+229 -6
View File
@@ -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:<column>) 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