"""``GET /roadmap`` — pulls open issues from GitHub so the FE can render a "what's next" page without a second source of truth. Design notes: - Reads the GitHub Issues API anonymously by default. Anonymous calls cap at 60 requests/hour/IP; we cache responses for 5 minutes so even a busy multi-user session stays comfortably under quota. A ``GITHUB_TOKEN`` env var, if present, is sent as a Bearer credential and lifts the cap to 5000 req/h. - Repo coordinates come from ``GITHUB_REPO_OWNER`` and ``GITHUB_REPO_NAME`` (or the comma-separated ``GITHUB_REPOS`` list for the multi-repo case). If neither is set, the endpoint returns a friendly empty envelope so the FE can render a "set GitHub repo in config" hint instead of erroring. - Filter: by default we fetch issues with the ``roadmap`` label so the in-app page doesn't double as a public tracker for every internal bug. Override with ``?label=...``. - The endpoint is authenticated (any role) so it doesn't leak the issue list to anonymous visitors; this is the same posture the rest of /api takes. """ from __future__ import annotations import os import time from typing import Any import requests from fastapi import APIRouter, Depends, HTTPException, Query from src.api.deps import forbid_readers from src.api.dto.common import ApiModel from src.lib.auth.models import User router = APIRouter(prefix="/roadmap", tags=["roadmap"]) _CACHE_TTL_SEC = 5 * 60 # 5 minutes _CACHE: dict[str, tuple[float, list[dict[str, Any]]]] = {} # ---------- DTOs ---------------------------------------------------------- class RoadmapItem(ApiModel): """One GitHub Issue projected onto the roadmap row shape. ``status`` is derived from the issue's state + labels: - state=open + label "in-progress" → "in_progress" - state=open + label "planned" → "planned" - state=open → "idea" - state=closed → "done" """ id: int number: int title: str body: str status: str category: str | None = None labels: list[str] url: str repo: str created_at: str updated_at: str comments: int class RoadmapResponse(ApiModel): """Wrapper for ``GET /roadmap``.""" items: list[RoadmapItem] repos: list[str] fetched_at: float """Unix timestamp of the underlying GitHub fetch (cached or fresh).""" error: str | None = None """Set when GitHub was unreachable or no repo is configured. Items may still be empty even when error is None (e.g. no issues with the label).""" # ---------- Helpers ------------------------------------------------------- def _repos_from_env() -> list[str]: """Return ``owner/name`` slugs in priority order. Reads ``GITHUB_REPOS`` (comma-separated) first; falls back to the singular ``GITHUB_REPO_OWNER`` / ``GITHUB_REPO_NAME`` pair so single- repo deployments stay simple. """ multi = os.environ.get("GITHUB_REPOS", "").strip() if multi: return [s.strip() for s in multi.split(",") if s.strip()] owner = os.environ.get("GITHUB_REPO_OWNER", "").strip() name = os.environ.get("GITHUB_REPO_NAME", "").strip() if owner and name: return [f"{owner}/{name}"] return [] def _derive_status(issue: dict[str, Any]) -> str: """Map issue state + labels → roadmap status enum.""" label_names = {(lb.get("name") or "").lower() for lb in issue.get("labels", [])} if issue.get("state") == "closed": return "done" if "in-progress" in label_names or "in progress" in label_names: return "in_progress" if "planned" in label_names: return "planned" return "idea" def _derive_category(issue: dict[str, Any]) -> str | None: """Pick the first labels[].name that looks like a category tag. Treats anything starting with ``cat:`` or in a known set (ux/cost/performance/hardening/feature/bug) as the category. """ known = {"ux", "cost", "performance", "hardening", "feature", "bug", "docs"} for lb in issue.get("labels", []): name = (lb.get("name") or "").lower() if name.startswith("cat:"): return name[len("cat:") :] if name in known: return name return None def _fetch_repo_issues(repo: str, label: str) -> list[dict[str, Any]]: """One blocking GET against the GitHub Issues API. Cached by ``(repo, label)`` via ``_CACHE`` — the caller hits the cache, not us.""" headers = {"Accept": "application/vnd.github+json"} # Accept either GITHUB_TOKEN or GH_TOKEN (the gh CLI convention) so # devs who already have one of them set don't need to duplicate. token = os.environ.get("GITHUB_TOKEN", "").strip() or os.environ.get("GH_TOKEN", "").strip() if token: headers["Authorization"] = f"Bearer {token}" params = { "labels": label, "state": "all", "per_page": 100, "sort": "updated", "direction": "desc", } url = f"https://api.github.com/repos/{repo}/issues" resp = requests.get(url, headers=headers, params=params, timeout=10) resp.raise_for_status() raw = resp.json() # The /issues endpoint also returns pull requests; filter them out. return [it for it in raw if "pull_request" not in it] def _cached_repo_issues(repo: str, label: str) -> list[dict[str, Any]]: """Wrap ``_fetch_repo_issues`` with a 5-minute TTL cache.""" key = f"{repo}#{label}" now = time.time() entry = _CACHE.get(key) if entry is not None and entry[0] > now: return entry[1] fresh = _fetch_repo_issues(repo, label) _CACHE[key] = (now + _CACHE_TTL_SEC, fresh) return fresh def _project(issue: dict[str, Any], repo: str) -> RoadmapItem: return RoadmapItem( id=int(issue["id"]), number=int(issue["number"]), title=str(issue.get("title", "")), body=str(issue.get("body") or ""), status=_derive_status(issue), category=_derive_category(issue), labels=[(lb.get("name") or "") for lb in issue.get("labels", [])], url=str(issue.get("html_url", "")), repo=repo, created_at=str(issue.get("created_at", "")), updated_at=str(issue.get("updated_at", "")), comments=int(issue.get("comments", 0)), ) # ---------- Endpoint ------------------------------------------------------ @router.get("", response_model=RoadmapResponse) def get_roadmap( label: str = Query(default="roadmap"), _: User = Depends(forbid_readers), ) -> RoadmapResponse: """Pull GitHub Issues labeled ``label`` from every configured repo. Returns a friendly envelope even when GitHub is unreachable or no repo is configured, so the FE can render a degraded view rather than a broken page. """ repos = _repos_from_env() if not repos: return RoadmapResponse( items=[], repos=[], fetched_at=time.time(), error=( "No GitHub repo configured. Set GITHUB_REPO_OWNER + " "GITHUB_REPO_NAME (or GITHUB_REPOS=owner/repo,owner/repo2) in " "the BE environment." ), ) items: list[RoadmapItem] = [] fetch_errors: list[str] = [] for repo in repos: try: raw_issues = _cached_repo_issues(repo, label) except requests.HTTPError as e: fetch_errors.append(f"{repo}: HTTP {e.response.status_code}") continue except requests.RequestException as e: fetch_errors.append(f"{repo}: {type(e).__name__}") continue for issue in raw_issues: items.append(_project(issue, repo)) return RoadmapResponse( items=items, repos=repos, fetched_at=time.time(), error="; ".join(fetch_errors) if fetch_errors else None, )