fix: restore backend code after stale checkout

src/gitea_proxy.py

@@ -0,0 +1,37 @@
+import os
+from typing import Any
+
+import httpx
+
+GITEA_URL = os.getenv("GITEA_URL", "http://127.0.0.1:3000").rstrip("/")
+GITEA_TOKEN = os.getenv("GITEA_TOKEN", "")
+
+
+def _auth() -> dict[str, str]:
+    headers: dict[str, str] = {"Accept": "application/json"}
+    if GITEA_TOKEN:
+        headers["Authorization"] = f"token {GITEA_TOKEN}"
+    return headers
+
+
+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()
+
+
+async def current_user() -> dict:
+    return await fetch("user")
+
+
+async def repos() -> list[dict]:
+    return await fetch("user/repos?limit=50")
+
+
+async def issues() -> list[dict]:
+    return await fetch("user/issues?limit=50&type=all")
+
+
+async def pull_requests() -> list[dict]:
+    return await fetch("user/pulls?limit=50")

src/main.py

@@ -0,0 +1,56 @@
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import JSONResponse
+from fastapi.staticfiles import StaticFiles
+
+from src.gitea_proxy import current_user, repos, issues, pull_requests
+from src.models import User, Repo, Issue, PullRequest
+from src.suggestion_engine import compute
+from src.views import router as frontend_router
+
+app = FastAPI(title="Stackchain Dashboard")
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=False,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+app.mount("/static", StaticFiles(directory="frontend/static"), name="static")
+app.include_router(frontend_router)
+
+
+@app.get("/api/v1/context")
+async def context() -> JSONResponse:
+    user_data = await current_user()
+    user_model = User(id=user_data["id"], login=user_data["login"], full_name=user_data.get("full_name", ""), email=user_data.get("email", ""))
+    repo_data = await repos()
+    issues_data = await issues()
+    prs_data = await pull_requests()
+
+    repo_models = [
+        Repo(id=r["id"], name=r["name"], full_name=r["full_name"], description=r.get("description", ""), url=r["html_url"], updated_at=r.get("updated_at", ""))
+        for r in repo_data[:50]
+        if isinstance(r, dict)
+    ]
+    issue_models = [
+        Issue(id=i["id"], number=i["number"], title=i["title"], state=i["state"], labels=[l.get("name", "") for l in i.get("labels", [])], assignees=[a.get("login", "") for a in i.get("assignees", [])], url=i["html_url"])
+        for i in issues_data[:50]
+        if isinstance(i, dict)
+    ]
+    pr_models = [
+        PullRequest(id=p["id"], number=p["number"], title=p["title"], state=p["state"], user=p.get("user", {}).get("login", ""), url=p["html_url"])
+        for p in prs_data[:50]
+        if isinstance(p, dict)
+    ]
+    ctx = compute(user_model, repo_models, issue_models, pr_models)
+    return JSONResponse({
+        "user": ctx.user.dict(),
+        "repos": [r.dict() for r in ctx.repos],
+        "issues": [i.dict() for i in ctx.issues],
+        "pull_requests": [p.dict() for p in ctx.pull_requests],
+        "view": ctx.view,
+        "deltas": [d.dict() for d in ctx.deltas],
+    })

src/models.py

@@ -0,0 +1,52 @@
+from pydantic import BaseModel
+
+
+class User(BaseModel):
+    id: int
+    login: str
+    full_name: str = ""
+    email: str = ""
+
+
+class Repo(BaseModel):
+    id: int
+    name: str
+    full_name: str
+    description: str = ""
+    url: str
+    updated_at: str = ""
+
+
+class Issue(BaseModel):
+    id: int
+    number: int
+    title: str
+    state: str
+    labels: list[str] = []
+    assignees: list[str] = []
+    url: str
+
+
+class PullRequest(BaseModel):
+    id: int
+    number: int
+    title: str
+    state: str
+    user: str
+    url: str
+
+
+class ViewContext(BaseModel):
+    user: User
+    repos: list[Repo] = []
+    issues: list[Issue] = []
+    pull_requests: list[PullRequest] = []
+    view: str = "dashboard"
+    deltas: list[dict] = []
+
+
+class SuggestionDelta(BaseModel):
+    panel: str
+    action: str
+    target: str = ""
+    priority: str = "medium"

src/suggestion_engine.py

@@ -0,0 +1,36 @@
+from src.models import Issue, PullRequest, Repo, SuggestionDelta, User, ViewContext
+
+
+def _load(user: User, repos: list[Repo], issues: list[Issue], prs: list[PullRequest], view: str) -> ViewContext:
+    return ViewContext(user=user, repos=repos, issues=issues, pull_requests=prs, view=view)
+
+
+def _suggest_review(prs: list[PullRequest]) -> list[SuggestionDelta]:
+    deltas: list[SuggestionDelta] = []
+    for pr in prs:
+        deltas.append(SuggestionDelta(panel="review", action="review", target=str(pr.number), priority="high"))
+    return deltas[:5]
+
+
+def _suggest_triage(issues: list[Issue]) -> list[SuggestionDelta]:
+    deltas: list[SuggestionDelta] = []
+    for issue in issues:
+        if issue.state != "open":
+            continue
+        priority = "high" if any(label.lower() in {"p0", "priority-high"} for label in issue.labels) else "medium"
+        deltas.append(SuggestionDelta(panel="triage", action="open", target=str(issue.number), priority=priority))
+    return deltas[:8]
+
+
+def _suggest_focus(repos: list[Repo], prs: list[PullRequest], issues: list[Issue]) -> list[SuggestionDelta]:
+    if len(prs) > len(issues):
+        return [SuggestionDelta(panel="layout", action="highlight", target="review", priority="medium")]
+    if len(issues) > len(prs):
+        return [SuggestionDelta(panel="layout", action="highlight", target="issues", priority="medium")]
+    return [SuggestionDelta(panel="layout", action="highlight", target="dashboard", priority="low")]
+
+
+def compute(user: User, repos: list[Repo], issues: list[Issue], prs: list[PullRequest], view: str = "dashboard") -> ViewContext:
+    ctx = _load(user, repos, issues, prs, view)
+    ctx.deltas = _suggest_review(prs) + _suggest_triage(issues) + _suggest_focus(repos, prs, issues)
+    return ctx

src/views.py

@@ -0,0 +1,9 @@
+from fastapi import APIRouter
+from fastapi.responses import HTMLResponse
+
+router = APIRouter()
+
+
+@router.get("/", response_class=HTMLResponse)
+async def dashboard() -> str:
+    return open("frontend/index.html").read()

tests/test_suggestion_engine.py

@@ -0,0 +1,16 @@
+from src.models import Issue, PullRequest, Repo, User
+from src.suggestion_engine import compute
+
+
+def test_suggest_review_when_more_prs_than_issues():
+    user = User(id=1, login="timmy", full_name="Timmy", email="timmy@example.com")
+    prs = [PullRequest(id=1, number=1, title="p", state="open", user="timmy", url="http://example.com")]
+    ctx = compute(user, [], [], prs)
+    assert [d for d in getattr(ctx, "deltas", []) if d.panel == "review"]
+
+
+def test_suggest_triage_when_open_issues():
+    user = User(id=1, login="timmy", full_name="Timmy", email="timmy@example.com")
+    issues = [Issue(id=1, number=1, title="t", state="open", labels=["P0"], assignees=[], url="http://example.com")]
+    ctx = compute(user, [], issues, [])
+    assert [d for d in getattr(ctx, "deltas", []) if d.panel == "triage"]