feat: unify live dashboard snapshot (#131)

src/gitea_proxy.py

@@ -9,6 +9,7 @@ GITEA_URL = os.getenv("GITEA_URL", "http://127.0.0.1:3000").rstrip("/")
 GITEA_TOKEN = os.getenv("GITEA_TOKEN", "")
 REVIEW_DIFF_MAX_BYTES = 64 * 1024
 REVIEW_DIFF_MAX_LINES = 400
+_client: httpx.AsyncClient | None = None
 
 
 def _auth() -> dict[str, str]:
@@ -18,30 +19,47 @@ def _auth() -> dict[str, str]:
     return headers
 
 
+def start_client(**kwargs) -> httpx.AsyncClient:
+    """Create the application-lifetime Gitea transport."""
+    global _client
+    _client = httpx.AsyncClient(base_url=GITEA_URL, timeout=10, **kwargs)
+    return _client
+
+
+def _get_client() -> httpx.AsyncClient:
+    if _client is None or _client.is_closed:
+        return start_client()
+    return _client
+
+
+async def stop_client() -> None:
+    global _client
+    if _client is not None and not _client.is_closed:
+        await _client.aclose()
+
+
 async def fetch(path: str) -> Any:
-    async with httpx.AsyncClient(base_url=GITEA_URL, timeout=10) as client:
-        r = await client.get(f"/api/v1/{path}", headers=_auth())
-        r.raise_for_status()
-        return r.json()
+    r = await _get_client().get(f"/api/v1/{path}", headers=_auth())
+    r.raise_for_status()
+    return r.json()
 
 
 async def fetch_text(path: str, max_bytes: int) -> tuple[str, bool]:
     chunks: list[bytes] = []
     size = 0
     truncated = False
-    async with httpx.AsyncClient(base_url=GITEA_URL, timeout=10) as client:
-        async with client.stream(
-            "GET", f"/api/v1/{path}", headers={**_auth(), "Accept": "text/plain"}
-        ) as response:
-            response.raise_for_status()
-            async for chunk in response.aiter_bytes():
-                remaining = max_bytes - size
-                if len(chunk) > remaining:
-                    chunks.append(chunk[:remaining])
-                    truncated = True
-                    break
-                chunks.append(chunk)
-                size += len(chunk)
+    async with _get_client().stream(
+        "GET", f"/api/v1/{path}", headers={**_auth(), "Accept": "text/plain"}
+    ) as response:
+        response.raise_for_status()
+        async for chunk in response.aiter_bytes():
+            remaining = max_bytes - size
+            if len(chunk) > remaining:
+                chunks.append(chunk[:remaining])
+                truncated = True
+                break
+            chunks.append(chunk)
+            size += len(chunk)
     return b"".join(chunks).decode("utf-8", errors="replace"), truncated
 
 
@@ -100,11 +118,11 @@ async def issues() -> list[dict]:
 
 
 async def pull_requests() -> list[dict]:
-    assigned = await fetch(
-        "repos/issues/search?state=open&assigned=true&type=pulls&limit=50"
-    )
-    review_requested = await fetch(
-        "repos/issues/search?state=open&review_requested=true&type=pulls&limit=50"
+    assigned, review_requested = await asyncio.gather(
+        fetch("repos/issues/search?state=open&assigned=true&type=pulls&limit=50"),
+        fetch(
+            "repos/issues/search?state=open&review_requested=true&type=pulls&limit=50"
+        ),
     )
     merged: dict[int, dict] = {}
     for reason, pulls in (
@@ -201,8 +219,9 @@ async def pull_review_detail(repository: str, number: int) -> dict:
     }
 
 
-async def activity_events() -> list[dict]:
-    user = await current_user()
+async def activity_events(user: dict | None = None) -> list[dict]:
+    if user is None:
+        user = await current_user()
     events = await fetch(f"users/{user['login']}/activities/feeds?limit=20")
     if events is None:
         events = []

src/main.py

@@ -1,5 +1,6 @@
 import asyncio
 import math
+from contextlib import asynccontextmanager
 from pathlib import Path
 
 from fastapi import FastAPI, HTTPException
@@ -7,6 +8,7 @@ from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import JSONResponse
 from fastapi.staticfiles import StaticFiles
 
+from src import gitea_proxy
 from src.gitea_proxy import (
     activity_events,
     current_user,
@@ -20,12 +22,22 @@ from src.models import Issue, PullRequest, Repo, User
 from src.suggestion_engine import compute
 from src.views import router as frontend_router
 
-app = FastAPI(title="Stackchain Dashboard")
+@asynccontextmanager
+async def lifespan(_app: FastAPI):
+    gitea_proxy.start_client()
+    try:
+        yield
+    finally:
+        await gitea_proxy.stop_client()
+
+
+app = FastAPI(title="Stackchain Dashboard", lifespan=lifespan)
 CONTEXT_TIMEOUT_SECONDS = 5.0
 EVENT_STREAM_TIMEOUT_SECONDS = 5.0
 READINESS_TIMEOUT_SECONDS = 5.0
 REVIEW_DETAIL_TIMEOUT_SECONDS = 5.0
 FRONTEND_DIR = Path(__file__).resolve().parent.parent / "frontend"
+_live_snapshot_task: asyncio.Task | None = None
 
 
 class ContextPayloadError(ValueError):
@@ -35,6 +47,52 @@ class ContextPayloadError(ValueError):
 class ReadinessPayloadError(ValueError):
     """Raised when Gitea returns a structurally invalid readiness payload."""
 
+
+def _context_payload(user_data, repo_data, issues_data, prs_data) -> dict:
+    user_model = User(
+        id=user_data["id"],
+        login=user_data["login"],
+        full_name=user_data.get("full_name") or "",
+        email=user_data.get("email") or "",
+    )
+    repo_models = [
+        Repo(
+            id=r["id"], name=r["name"], full_name=r["full_name"],
+            description=r.get("description") or "", url=r["html_url"],
+            updated_at=r.get("updated_at", ""),
+        )
+        for r in (repo_data or [])[:50]
+        if isinstance(r, dict)
+        and all(field in r for field in ("id", "name", "full_name", "html_url"))
+    ]
+    issue_models = [
+        Issue(
+            id=i["id"], number=i["number"], title=i["title"], state=i["state"],
+            labels=[label.get("name", "") for label in (i.get("labels") or []) if isinstance(label, dict)],
+            assignees=[assignee.get("login", "") for assignee in (i.get("assignees") or []) if isinstance(assignee, dict)],
+            repository=i["repository"].get("full_name", "") if isinstance(i.get("repository"), dict) else "",
+            updated_at=i.get("updated_at") or "", url=i["html_url"],
+        )
+        for i in (issues_data or [])[:50]
+        if isinstance(i, dict)
+        and all(field in i for field in ("id", "number", "title", "state", "html_url"))
+    ]
+    pr_models = [
+        PullRequest(
+            id=p["id"], number=p["number"], title=p["title"], state=p["state"],
+            user=p["user"].get("login", "") if isinstance(p.get("user"), dict) else "",
+            labels=[label.get("name", "") for label in (p.get("labels") or []) if isinstance(label, dict)],
+            assignees=[assignee.get("login", "") for assignee in (p.get("assignees") or []) if isinstance(assignee, dict)],
+            work_reasons=[reason for reason in (p.get("work_reasons") or []) if reason in ("assigned_to_me", "review_requested")],
+            repository=p["repository"].get("full_name", "") if isinstance(p.get("repository"), dict) else "",
+            updated_at=p.get("updated_at") or "", url=p["html_url"],
+        )
+        for p in (prs_data or [])[:50]
+        if isinstance(p, dict)
+        and all(field in p for field in ("id", "number", "title", "state", "html_url"))
+    ]
+    return compute(user_model, repo_models, issue_models, pr_models).model_dump()
+
 app.add_middleware(
     CORSMiddleware,
     allow_origins=["*"],
@@ -50,7 +108,7 @@ app.include_router(frontend_router)
 @app.middleware("http")
 async def prevent_live_api_caching(request, call_next):
     response = await call_next(request)
-    if request.url.path in {"/api/v1/context", "/api/v1/events"} or (
+    if request.url.path in {"/api/v1/context", "/api/v1/events", "/api/v1/live"} or (
         request.url.path.startswith("/api/v1/repos/")
         and request.url.path.endswith("/review")
     ):
@@ -145,87 +203,68 @@ async def context() -> JSONResponse:
         }, headers={
             "Retry-After": str(max(1, math.ceil(CONTEXT_TIMEOUT_SECONDS)))
         } if isinstance(e, TimeoutError) else None)
-    user_model = User(id=user_data["id"], login=user_data["login"], full_name=user_data.get("full_name") or "", email=user_data.get("email") or "")
-    repo_models = [
-        Repo(id=r["id"], name=r["name"], full_name=r["full_name"], description=r.get("description") or "", url=r["html_url"], updated_at=r.get("updated_at", ""))
-        for r in (repo_data or [])[:50]
-        if isinstance(r, dict)
-        and all(field in r for field in ("id", "name", "full_name", "html_url"))
-    ]
-    issue_models = [
-        Issue(
-            id=i["id"],
-            number=i["number"],
-            title=i["title"],
-            state=i["state"],
-            labels=[
-                label.get("name", "")
-                for label in (i.get("labels") or [])
-                if isinstance(label, dict)
-            ],
-            assignees=[
-                assignee.get("login", "")
-                for assignee in (i.get("assignees") or [])
-                if isinstance(assignee, dict)
-            ],
-            repository=(
-                i["repository"].get("full_name", "")
-                if isinstance(i.get("repository"), dict)
-                else ""
-            ),
-            updated_at=i.get("updated_at") or "",
-            url=i["html_url"],
+    return JSONResponse(_context_payload(user_data, repo_data, issues_data, prs_data))
+
+
+async def _load_context_for_user(user_data: dict) -> dict:
+    repo_data, issues_data, prs_data = await asyncio.gather(
+        repos(), issues(), pull_requests()
+    )
+    if not all(field in user_data for field in ("id", "login")):
+        raise ContextPayloadError("Gitea current-user response did not include id and login")
+    if any(
+        data is not None and not isinstance(data, list)
+        for data in (repo_data, issues_data, prs_data)
+    ):
+        raise ContextPayloadError("Gitea collection response was not a list")
+    return _context_payload(user_data, repo_data, issues_data, prs_data)
+
+
+async def _build_live_snapshot() -> dict:
+    user_data = await current_user()
+    if not isinstance(user_data, dict) or not user_data.get("login"):
+        raise ContextPayloadError("Gitea current-user response was invalid")
+    context_result, events_result = await asyncio.gather(
+        _load_context_for_user(user_data),
+        activity_events(user_data),
+        return_exceptions=True,
+    )
+    context_ok = not isinstance(context_result, BaseException)
+    events_ok = not isinstance(events_result, BaseException)
+    return {
+        "context": context_result if context_ok else None,
+        "events": events_result if events_ok else None,
+        "sections": {
+            "context": "fresh" if context_ok else "temporarily unavailable",
+            "events": "fresh" if events_ok else "temporarily unavailable",
+        },
+    }
+
+
+@app.get("/api/v1/live")
+async def live_snapshot() -> JSONResponse:
+    """Return a fresh, section-aware snapshot; join only an active identical load."""
+    global _live_snapshot_task
+    if _live_snapshot_task is None or _live_snapshot_task.done():
+        _live_snapshot_task = asyncio.create_task(_build_live_snapshot())
+    task = _live_snapshot_task
+    try:
+        result = await asyncio.wait_for(task, timeout=CONTEXT_TIMEOUT_SECONDS)
+        return JSONResponse(result)
+    except TimeoutError:
+        return JSONResponse(
+            {"error": f"Gitea live snapshot timed out after {CONTEXT_TIMEOUT_SECONDS:g}s"},
+            status_code=503,
+            headers={"Retry-After": str(max(1, math.ceil(CONTEXT_TIMEOUT_SECONDS)))},
         )
-        for i in (issues_data or [])[:50]
-        if isinstance(i, dict)
-        and all(
-            field in i
-            for field in ("id", "number", "title", "state", "html_url")
+    except Exception:
+        return JSONResponse(
+            {"error": "Gitea live snapshot is temporarily unavailable"},
+            status_code=503,
         )
-    ]
-    pr_models = [
-        PullRequest(
-            id=p["id"],
-            number=p["number"],
-            title=p["title"],
-            state=p["state"],
-            user=(
-                p["user"].get("login", "")
-                if isinstance(p.get("user"), dict)
-                else ""
-            ),
-            labels=[
-                label.get("name", "")
-                for label in (p.get("labels") or [])
-                if isinstance(label, dict)
-            ],
-            assignees=[
-                assignee.get("login", "")
-                for assignee in (p.get("assignees") or [])
-                if isinstance(assignee, dict)
-            ],
-            work_reasons=[
-                reason
-                for reason in (p.get("work_reasons") or [])
-                if reason in ("assigned_to_me", "review_requested")
-            ],
-            repository=(
-                p["repository"].get("full_name", "")
-                if isinstance(p.get("repository"), dict)
-                else ""
-            ),
-            updated_at=p.get("updated_at") or "",
-            url=p["html_url"],
-        )
-        for p in (prs_data or [])[:50]
-        if isinstance(p, dict)
-        and all(
-            field in p
-            for field in ("id", "number", "title", "state", "html_url")
-        )
-    ]
-    ctx = compute(user_model, repo_models, issue_models, pr_models)
-    return JSONResponse(ctx.model_dump())
+    finally:
+        if task.done() and _live_snapshot_task is task:
+            _live_snapshot_task = None
 
 
 @app.get("/api/v1/events")