diff --git a/app/briefing.py b/app/briefing.py new file mode 100644 index 0000000..459c93f --- /dev/null +++ b/app/briefing.py @@ -0,0 +1,71 @@ +"""Daily briefing: formatting and ntfy delivery. + +The data itself is gathered in app.main (it needs the board helpers); +this module only turns it into a readable text and pushes it to an +ntfy topic. Kept dependency-light so it is unit-testable alone. +""" + +from __future__ import annotations + +import httpx + +from app.config import Settings + + +def ntfy_configured(settings: Settings) -> bool: + return bool(settings.ntfy_url and settings.ntfy_topic) + + +def format_briefing_text(data: dict) -> str: + """Plain-text briefing (ntfy body, also shown on /briefing).""" + lines: list[str] = [f"πŸ“‹ Kanbante β€” briefing du {data['date']}"] + activity = data["activity_24h"] + lines.append( + f"\n⚑ DerniΓ¨res 24 h : {activity['closed']} fermΓ©e(s), " + f"{activity['created']} créée(s), {activity['moved']} dΓ©placement(s)." + ) + if activity.get("avg_cycle_days") is not None: + lines.append(f"⏱️ Temps de cycle moyen (30 j) : {activity['avg_cycle_days']} j.") + + lines.append(f"\n🎯 En cours / en rΓ©vision : {data['focus_total']} carte(s)") + for group in data["focus_groups"][:6]: + lines.append(f" β€’ {group['full_name']} : {group['count']}") + for title in group["titles"][:3]: + lines.append(f" – {title}") + + if data["aging"]: + lines.append("\n⏳ Cartes qui vieillissent :") + for card in data["aging"][:8]: + lines.append( + f" β€’ {card['repo']} #{card['number']} Β« {card['title']} Β» " + f"β€” {card['days']} j en {card['column_label']}" + ) + + if data["milestones_due"]: + lines.append("\nπŸ“… Milestones proches :") + for ms in data["milestones_due"][:5]: + lines.append( + f" β€’ {ms['repo']} Β« {ms['title']} Β» β€” Γ©chΓ©ance " + f"{ms['due_on']} ({ms['open']} ouverte(s))" + ) + + if data["focus_total"] == 0 and not data["aging"]: + lines.append("\nπŸŽ‰ Rien d'urgent : aucun encours, aucune carte vieillissante.") + return "\n".join(lines) + + +async def send_ntfy(settings: Settings, text: str) -> bool: + """POST the briefing to the configured ntfy topic. True on 2xx.""" + if not ntfy_configured(settings): + return False + url = f"{settings.ntfy_url.rstrip('/')}/{settings.ntfy_topic}" + headers = { + # ntfy headers must stay ASCII; the body carries the accents. + "Title": "Kanbante - briefing du jour", + "Tags": "clipboard,chart_with_upwards_trend", + } + if settings.ntfy_token: + headers["Authorization"] = f"Bearer {settings.ntfy_token}" + async with httpx.AsyncClient(timeout=15.0) as client: + response = await client.post(url, content=text.encode("utf-8"), headers=headers) + return response.status_code < 300 diff --git a/app/config.py b/app/config.py index 8a558fc..2b2a0b8 100644 --- a/app/config.py +++ b/app/config.py @@ -28,6 +28,14 @@ Environment variables: X-Gitea-Signature HMAC is enforced. KANBANTE_SYNC_LABELS "1" (default) = mirror intermediate columns to Gitea labels named kanban:. + KANBANTE_NTFY_URL Base URL of an ntfy server (e.g. + https://ntfy.sh or self-hosted). + KANBANTE_NTFY_TOPIC ntfy topic for the daily briefing. + KANBANTE_NTFY_TOKEN Optional Bearer token for ntfy. + KANBANTE_BRIEFING_TIME "HH:MM" (server local time) at which + the daily briefing is sent to ntfy. + Empty (default) = no automatic send, + the /briefing page stays available. """ import os @@ -36,7 +44,7 @@ from pathlib import Path BASE_DIR = Path(__file__).resolve().parent.parent -APP_VERSION = "0.4.0" +APP_VERSION = "0.5.0" def _env_bool(name: str, default: bool) -> bool: @@ -88,6 +96,10 @@ class Settings: aging_days: int = 7 webhook_secret: str = "" sync_labels: bool = True + ntfy_url: str = "" + ntfy_topic: str = "" + ntfy_token: str = "" + briefing_time: str = "" @property def auth_enabled(self) -> bool: @@ -113,4 +125,8 @@ def get_settings() -> Settings: aging_days=_env_int("KANBANTE_AGING_DAYS", 7), webhook_secret=os.getenv("KANBANTE_WEBHOOK_SECRET", ""), sync_labels=_env_bool("KANBANTE_SYNC_LABELS", True), + ntfy_url=os.getenv("KANBANTE_NTFY_URL", ""), + ntfy_topic=os.getenv("KANBANTE_NTFY_TOPIC", ""), + ntfy_token=os.getenv("KANBANTE_NTFY_TOKEN", ""), + briefing_time=os.getenv("KANBANTE_BRIEFING_TIME", ""), ) diff --git a/app/gitea_client.py b/app/gitea_client.py index 64ed967..8909784 100644 --- a/app/gitea_client.py +++ b/app/gitea_client.py @@ -269,6 +269,37 @@ class GiteaClient: ) r.raise_for_status() + async def delete_label(self, owner: str, repo: str, label_id: int) -> None: + """Delete a label definition from the repo (all issues lose it).""" + r = await self.client.delete(f"/repos/{owner}/{repo}/labels/{label_id}") + r.raise_for_status() + self.forget_label_cache(owner, repo) + + def forget_label_cache(self, owner: str, repo: str) -> None: + cache = self.__dict__.get("_label_cache") + if cache: + cache.pop(f"{owner}/{repo}", None) + + # ── Issue search (global + assigned) ─────────────────── + + async def search_issues( + self, + q: str = "", + state: str = "open", + assigned_by: str | None = None, + ) -> list[dict]: + """Search issues across every repo visible to the token. + + Uses Gitea's /repos/issues/search (type=issues excludes PRs). + Results carry their repository in issue["repository"]. + """ + params: dict[str, Any] = {"state": state, "type": "issues"} + if q: + params["q"] = q + if assigned_by: + params["assigned_by"] = assigned_by + return await self._get_all_pages("/repos/issues/search", **params) + # ── Milestones ───────────────────────────────────────── async def get_milestones( diff --git a/app/main.py b/app/main.py index 73c1c32..8fb8676 100644 --- a/app/main.py +++ b/app/main.py @@ -34,11 +34,13 @@ from starlette.middleware.base import BaseHTTPMiddleware from app.config import APP_VERSION, get_settings from app.gitea_client import GiteaClient +from app.briefing import format_briefing_text, ntfy_configured, send_ntfy from app.models import ( KANBAN_COLUMNS, column_entry_times, get_board_state, get_events, + get_kv, get_metrics, get_notes, get_stats, @@ -48,6 +50,7 @@ from app.models import ( ping_db, reconcile_board_state, set_column_order, + set_kv, set_notes, ) @@ -86,6 +89,33 @@ EVENT_ICONS = { # ── App setup ────────────────────────────────────────────── +async def _briefing_scheduler() -> None: + """Send the daily briefing to ntfy at KANBANTE_BRIEFING_TIME. + + Checks once a minute; the kv store remembers the last day sent so + restarts never double-send. + """ + while True: + try: + settings = get_settings() + if settings.briefing_time and ntfy_configured(settings): + now = datetime.now() + today = now.strftime("%Y-%m-%d") + if ( + now.strftime("%H:%M") >= settings.briefing_time + and get_kv("briefing_last_sent") != today + ): + data = await _briefing_data() + if await send_ntfy(settings, format_briefing_text(data)): + set_kv("briefing_last_sent", today) + logger.info("Daily briefing sent to ntfy (%s)", today) + except asyncio.CancelledError: + raise + except Exception as exc: # never let the scheduler die silently + logger.warning("Briefing scheduler error: %s", exc) + await asyncio.sleep(60) + + @asynccontextmanager async def lifespan(app: FastAPI): init_db() @@ -102,7 +132,9 @@ async def lifespan(app: FastAPI): logger.warning( "GITEA_TOKEN is not set: only public Gitea content will be visible." ) + scheduler = asyncio.create_task(_briefing_scheduler()) yield + scheduler.cancel() await gitea.aclose() @@ -471,9 +503,8 @@ def _card_ages( return ages -@app.get("/focus", response_class=HTMLResponse) -async def focus(request: Request): - """Cross-project view: everything in progress or in review.""" +async def _collect_focus_groups() -> list[dict]: + """In-progress / in-review cards across all active repos.""" repos = await gitea.search_repos() active, _inactive, _archived = classify_repos(repos) active = active[:100] @@ -500,15 +531,232 @@ async def focus(request: Request): ] if cards: groups.append({"full_name": full_name, "cards": cards}) + return groups + + +@app.get("/focus", response_class=HTMLResponse) +async def focus(request: Request): + """Cross-project view: everything in progress or in review.""" return render( request, "focus.html", - groups=groups, + groups=await _collect_focus_groups(), column_labels=COLUMN_LABELS, column_colors=COLUMN_COLORS, ) +def _group_issue_results(issues: list[dict]) -> list[dict]: + """Group search results by repo, annotated with the board column.""" + by_repo: dict[str, list[dict]] = {} + for issue in issues: + full_name = (issue.get("repository") or {}).get("full_name", "?") + by_repo.setdefault(full_name, []).append(issue) + groups = [] + for full_name in sorted(by_repo): + state = get_board_state(full_name) if "/" in full_name else {} + column_of = { + card["issue_number"]: column + for column, cards in state.items() + for card in cards + } + rows = [] + for issue in by_repo[full_name]: + if issue.get("state") == "open": + column = column_of.get(issue["number"], "backlog") + else: + column = "done" + rows.append({"issue": issue, "column": column}) + groups.append({"full_name": full_name, "rows": rows}) + return groups + + +@app.get("/search", response_class=HTMLResponse) +async def search_page(request: Request, q: str = ""): + """Global issue search across every repo (Gitea search API).""" + q = q.strip() + groups: list[dict] = [] + total = 0 + if q: + results = await gitea.search_issues(q=q, state="all") + total = len(results) + groups = _group_issue_results(results) + return render( + request, + "search.html", + q=q, + groups=groups, + total=total, + column_labels=COLUMN_LABELS, + column_colors=COLUMN_COLORS, + ) + + +@app.get("/my", response_class=HTMLResponse) +async def my_issues(request: Request): + """Open issues assigned to the Gitea token's owner, all repos.""" + user = None + groups: list[dict] = [] + total = 0 + try: + user = await gitea.get_user() + except Exception as exc: + logger.debug("get_user failed for /my: %s", exc) + if user and user.get("login"): + results = await gitea.search_issues( + state="open", assigned_by=user["login"] + ) + total = len(results) + groups = _group_issue_results(results) + return render( + request, + "my.html", + user=user, + groups=groups, + total=total, + column_labels=COLUMN_LABELS, + column_colors=COLUMN_COLORS, + ) + + +# ── Briefing ─────────────────────────────────────────────── + +async def _briefing_data() -> dict: + """Everything the daily briefing needs, gathered in one dict.""" + settings = get_settings() + focus_groups = await _collect_focus_groups() + + repos = await gitea.search_repos() + active, _i, _a = classify_repos(repos) + soon = datetime.now(timezone.utc) + timedelta(days=7) + aging: list[dict] = [] + milestones_due: list[dict] = [] + + async def _per_repo(repo: dict): + full_name = repo["full_name"] + owner, _, name = full_name.partition("/") + items = await gitea.get_repo_issues(owner, name, state="open") + open_issues = [i for i in items if _is_issue(i)] + ages = _card_ages(full_name, open_issues, get_board_state(full_name)) + board = get_board_state(full_name) + column_of = { + card["issue_number"]: column + for column, cards in board.items() + for card in cards + } + repo_aging = [ + { + "repo": full_name, + "number": issue["number"], + "title": issue["title"], + "days": ages[issue["number"]]["days"], + "column_label": COLUMN_LABELS[ + column_of.get(issue["number"], "backlog") + ], + } + for issue in open_issues + if issue["number"] in ages + and ages[issue["number"]]["days"] >= settings.aging_days + ] + repo_milestones = [] + try: + for ms in await gitea.get_milestones(owner, name): + due = _parse_datetime(ms.get("due_on")) + if due and due <= soon and (ms.get("open_issues") or 0) > 0: + repo_milestones.append( + { + "repo": full_name, + "title": ms.get("title", ""), + "due_on": str(ms["due_on"])[:10], + "open": ms.get("open_issues") or 0, + } + ) + except Exception as exc: + logger.debug("briefing milestones failed for %s: %s", full_name, exc) + return repo_aging, repo_milestones + + results = await asyncio.gather( + *(_per_repo(r) for r in active[:100]), return_exceptions=True + ) + for result in results: + if isinstance(result, Exception): + logger.warning("briefing: repo fetch failed: %s", result) + continue + repo_aging, repo_milestones = result + aging.extend(repo_aging) + milestones_due.extend(repo_milestones) + aging.sort(key=lambda c: -c["days"]) + + metrics_24h = get_metrics(window_days=1) + metrics_30d = get_metrics(window_days=30) + return { + "date": datetime.now().strftime("%d/%m/%Y"), + "focus_total": sum(len(g["cards"]) for g in focus_groups), + "focus_groups": [ + { + "full_name": g["full_name"], + "count": len(g["cards"]), + "titles": [c["issue"]["title"] for c in g["cards"]], + } + for g in focus_groups + ], + "aging": aging[:10], + "milestones_due": milestones_due[:5], + "activity_24h": { + **metrics_24h, + "avg_cycle_days": metrics_30d["avg_cycle_days"], + }, + } + + +@app.get("/briefing", response_class=HTMLResponse) +async def briefing_page(request: Request): + """The daily briefing, readable on the web before it is pushed.""" + settings = get_settings() + data = await _briefing_data() + return render( + request, + "briefing.html", + data=data, + text=format_briefing_text(data), + ntfy_configured=ntfy_configured(settings), + briefing_time=settings.briefing_time, + last_sent=get_kv("briefing_last_sent"), + ) + + +@app.get("/api/briefing") +async def api_briefing(): + data = await _briefing_data() + return {"ok": True, "briefing": data, "text": format_briefing_text(data)} + + +@app.post("/api/briefing/send") +async def api_briefing_send(request: Request, csrf_token: str = Form("")): + """Send the briefing to the configured ntfy topic, right now.""" + if not _csrf_valid(request, csrf_token): + return _csrf_error() + settings = get_settings() + if not ntfy_configured(settings): + return JSONResponse( + { + "ok": False, + "error": "ntfy is not configured " + "(KANBANTE_NTFY_URL / KANBANTE_NTFY_TOPIC)", + }, + status_code=400, + ) + data = await _briefing_data() + sent = await send_ntfy(settings, format_briefing_text(data)) + if not sent: + return JSONResponse( + {"ok": False, "error": "ntfy server rejected or unreachable"}, + status_code=502, + ) + set_kv("briefing_last_sent", datetime.now().strftime("%Y-%m-%d")) + return {"ok": True, "sent": True} + + # ── Webhooks + live stream ───────────────────────────────── def _apply_external_transition( @@ -802,7 +1050,7 @@ def _issue_payload(issue: dict) -> dict: "updated_at": issue.get("updated_at", ""), "due_date": issue.get("due_date"), "labels": [ - {"name": l.get("name", ""), "color": l.get("color", "")} + {"id": l.get("id"), "name": l.get("name", ""), "color": l.get("color", "")} for l in issue.get("labels") or [] ], "assignees": [ @@ -889,6 +1137,330 @@ async def api_metrics(repo: str = "", window: int = 30): return {"ok": True, "metrics": get_metrics(repo or None, window)} +# ── Label management ─────────────────────────────────────── + +@app.get("/api/repos/{owner}/{repo}/labels") +async def api_list_labels(owner: str, repo: str): + """All label definitions of a repo.""" + labels = await gitea.get_labels(owner, repo) + return { + "ok": True, + "labels": [ + {"id": l.get("id"), "name": l.get("name", ""), "color": l.get("color", "")} + for l in labels + ], + } + + +@app.post("/api/repos/{owner}/{repo}/labels") +async def api_create_label( + request: Request, + owner: str, + repo: str, + name: str = Form(...), + color: str = Form("7c5cfc"), + csrf_token: str = Form(""), +): + """Create a label definition in the repo.""" + if not _csrf_valid(request, csrf_token): + return _csrf_error() + name = name.strip() + color = color.strip().lstrip("#") + if not name or len(name) > 50: + return JSONResponse( + {"ok": False, "error": "Label name must be 1-50 chars"}, + status_code=400, + ) + if len(color) != 6 or any(c not in "0123456789abcdefABCDEF" for c in color): + return JSONResponse( + {"ok": False, "error": "Color must be a 6-digit hex value"}, + status_code=400, + ) + label = await gitea.create_label(owner, repo, name, color) + gitea.forget_label_cache(owner, repo) + return JSONResponse( + { + "ok": True, + "label": { + "id": label.get("id"), + "name": label.get("name", name), + "color": label.get("color", color), + }, + }, + status_code=201, + ) + + +@app.delete("/api/repos/{owner}/{repo}/labels/{label_id}") +async def api_delete_label( + request: Request, owner: str, repo: str, label_id: int +): + """Delete a label definition (CSRF via the X-CSRF-Token header).""" + if not _csrf_valid(request, ""): + return _csrf_error() + await gitea.delete_label(owner, repo, label_id) + return {"ok": True, "deleted": label_id} + + +@app.post("/api/issues/{owner}/{repo}/{number}/labels") +async def api_toggle_issue_label( + request: Request, + owner: str, + repo: str, + number: int, + label_id: int = Form(...), + action: str = Form(...), + csrf_token: str = Form(""), +): + """Add or remove one label on one issue; returns the new label set.""" + if not _csrf_valid(request, csrf_token): + return _csrf_error() + if action not in ("add", "remove"): + return JSONResponse( + {"ok": False, "error": "action must be 'add' or 'remove'"}, + status_code=400, + ) + if action == "add": + await gitea.add_issue_label(owner, repo, number, label_id) + else: + await gitea.remove_issue_label(owner, repo, number, label_id) + issue = await gitea.get_issue(owner, repo, number) + return { + "ok": True, + "labels": [ + {"id": l.get("id"), "name": l.get("name", ""), "color": l.get("color", "")} + for l in issue.get("labels") or [] + ], + } + + +# ── MCP server (read-only tools over JSON-RPC) ───────────── + +MCP_TOOLS = [ + { + "name": "list_projects", + "description": "List the projects (Gitea repos) Kanbante tracks, grouped by activity.", + "inputSchema": {"type": "object", "properties": {}}, + }, + { + "name": "get_board", + "description": "Get the kanban board of one repo: columns with their cards.", + "inputSchema": { + "type": "object", + "properties": { + "owner": {"type": "string"}, + "repo": {"type": "string"}, + }, + "required": ["owner", "repo"], + }, + }, + { + "name": "get_issue", + "description": "Get one issue with its body and comments.", + "inputSchema": { + "type": "object", + "properties": { + "owner": {"type": "string"}, + "repo": {"type": "string"}, + "number": {"type": "integer"}, + }, + "required": ["owner", "repo", "number"], + }, + }, + { + "name": "get_focus", + "description": "Cards in progress or in review across all active projects.", + "inputSchema": {"type": "object", "properties": {}}, + }, + { + "name": "search_issues", + "description": "Search issues by text across all repos.", + "inputSchema": { + "type": "object", + "properties": { + "query": {"type": "string"}, + "state": {"type": "string", "enum": ["open", "closed", "all"]}, + }, + "required": ["query"], + }, + }, + { + "name": "get_activity", + "description": "Latest entries of the board event journal.", + "inputSchema": { + "type": "object", + "properties": {"limit": {"type": "integer"}}, + }, + }, + { + "name": "get_metrics", + "description": "Throughput and cycle-time metrics from the journal.", + "inputSchema": { + "type": "object", + "properties": {"repo": {"type": "string"}}, + }, + }, +] + + +async def _mcp_call_tool(name: str, args: dict) -> str: + """Execute one read-only MCP tool and return text content.""" + if name == "list_projects": + repos = await gitea.search_repos() + active, inactive, archived = classify_repos(repos) + lines = [f"Active projects ({len(active)}):"] + lines += [ + f"- {r['full_name']} ({r.get('open_issues_count') or 0} open issues)" + for r in active + ] + lines.append( + f"Inactive: {len(inactive)} - Archived: {len(archived)}" + ) + return "\n".join(lines) + if name == "get_board": + owner, repo = args["owner"], args["repo"] + items = await gitea.get_repo_issues(owner, repo, state="all") + issues = [i for i in items if _is_issue(i)] + open_issues = [i for i in issues if i.get("state") == "open"] + columns = _merge_issues_with_board(open_issues, get_board_state(f"{owner}/{repo}")) + return json.dumps( + { + column: [ + { + "number": i["number"], + "title": i["title"], + "labels": [l.get("name") for l in i.get("labels") or []], + } + for i in cards + ] + for column, cards in columns.items() + }, + ensure_ascii=False, + ) + if name == "get_issue": + detail = await api_issue_detail(args["owner"], args["repo"], int(args["number"])) + return json.dumps(detail, ensure_ascii=False, default=str) + if name == "get_focus": + groups = await _collect_focus_groups() + return json.dumps( + [ + { + "repo": g["full_name"], + "cards": [ + { + "number": c["issue"]["number"], + "title": c["issue"]["title"], + "column": c["column"], + } + for c in g["cards"] + ], + } + for g in groups + ], + ensure_ascii=False, + ) + if name == "search_issues": + results = await gitea.search_issues( + q=args["query"], state=args.get("state", "open") + ) + return json.dumps( + [ + { + "repo": (i.get("repository") or {}).get("full_name"), + "number": i["number"], + "title": i["title"], + "state": i["state"], + } + for i in results + ], + ensure_ascii=False, + ) + if name == "get_activity": + return json.dumps( + get_events(limit=int(args.get("limit", 50))), ensure_ascii=False + ) + if name == "get_metrics": + return json.dumps(get_metrics(args.get("repo") or None), ensure_ascii=False) + raise ValueError(f"Unknown tool: {name}") + + +@app.get("/mcp") +async def mcp_info(): + """Discoverability for the MCP endpoint (the protocol is POST).""" + return { + "name": "kanbante", + "version": APP_VERSION, + "transport": "streamable-http (JSON-RPC 2.0 over POST)", + "read_only": True, + "tools": [t["name"] for t in MCP_TOOLS], + } + + +@app.post("/mcp") +async def mcp_endpoint(request: Request): + """Minimal MCP server: initialize / tools/list / tools/call / ping. + + Read-only by design (Bruno's rule: cross-platform reads first, + writes go through approval flows elsewhere). + """ + try: + payload = await request.json() + except ValueError: + return JSONResponse( + {"jsonrpc": "2.0", "id": None, + "error": {"code": -32700, "message": "Parse error"}}, + status_code=400, + ) + if isinstance(payload, list): # batch: not supported by this server + return JSONResponse( + {"jsonrpc": "2.0", "id": None, + "error": {"code": -32600, "message": "Batch requests not supported"}}, + status_code=400, + ) + method = payload.get("method", "") + rpc_id = payload.get("id") + if rpc_id is None: # notification: acknowledge silently + return Response(status_code=202) + params = payload.get("params") or {} + + def _result(result): + return {"jsonrpc": "2.0", "id": rpc_id, "result": result} + + def _error(code, message): + return {"jsonrpc": "2.0", "id": rpc_id, + "error": {"code": code, "message": message}} + + if method == "initialize": + return _result( + { + "protocolVersion": params.get("protocolVersion", "2024-11-05"), + "capabilities": {"tools": {}}, + "serverInfo": {"name": "kanbante", "version": APP_VERSION}, + } + ) + if method == "ping": + return _result({}) + if method == "tools/list": + return _result({"tools": MCP_TOOLS}) + if method == "tools/call": + tool_name = params.get("name", "") + arguments = params.get("arguments") or {} + try: + text = await _mcp_call_tool(tool_name, arguments) + except KeyError as exc: + return _result( + {"content": [{"type": "text", "text": f"Missing argument: {exc}"}], + "isError": True} + ) + except Exception as exc: + return _result( + {"content": [{"type": "text", "text": f"Tool error: {exc}"}], + "isError": True} + ) + return _result({"content": [{"type": "text", "text": text}], "isError": False}) + return _error(-32601, f"Method not found: {method}") + + @app.post("/api/notes") async def api_notes( request: Request, diff --git a/app/models.py b/app/models.py index 72cb6d9..d1314da 100644 --- a/app/models.py +++ b/app/models.py @@ -79,6 +79,11 @@ def init_db() -> None: CREATE INDEX IF NOT EXISTS idx_board_repo ON board_state(repo_full_name, column_name); + CREATE TABLE IF NOT EXISTS kv_store ( + key TEXT PRIMARY KEY, + value TEXT NOT NULL + ); + CREATE INDEX IF NOT EXISTS idx_events_repo ON board_events(repo_full_name, issue_number, created_at); """ @@ -234,6 +239,23 @@ def reconcile_board_state( return changed +def get_kv(key: str) -> str | None: + with get_db() as conn: + row = conn.execute( + "SELECT value FROM kv_store WHERE key = ?", (key,) + ).fetchone() + return row["value"] if row else None + + +def set_kv(key: str, value: str) -> None: + with get_db() as conn: + conn.execute( + "INSERT INTO kv_store (key, value) VALUES (?, ?) " + "ON CONFLICT(key) DO UPDATE SET value = excluded.value", + (key, value), + ) + + # ── Event journal ────────────────────────────────────────── EVENT_TYPES = ("created", "moved", "closed", "reopened", "commented")