stackchain-dashboard/src/main.py

import asyncio
import math
from pathlib import Path

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 activity_events, 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")
CONTEXT_TIMEOUT_SECONDS = 5.0
EVENT_STREAM_TIMEOUT_SECONDS = 5.0
READINESS_TIMEOUT_SECONDS = 5.0
FRONTEND_DIR = Path(__file__).resolve().parent.parent / "frontend"

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=False,
    allow_methods=["*"],
    allow_headers=["*"],
)

app.mount("/static", StaticFiles(directory=FRONTEND_DIR), name="static")
app.include_router(frontend_router)


@app.get("/healthz")
def health() -> dict[str, str]:
    """Return process liveness without depending on Gitea."""
    return {"status": "ok", "service": "stackchain-dashboard"}


@app.get("/readyz")
async def readiness():
    """Return readiness after verifying the configured Gitea connection."""
    try:
        user = await asyncio.wait_for(
            current_user(), timeout=READINESS_TIMEOUT_SECONDS
        )
    except Exception as exc:
        timed_out = isinstance(exc, TimeoutError)
        error_message = (
            f"Gitea readiness check timed out after {READINESS_TIMEOUT_SECONDS:g}s"
            if timed_out
            else str(exc)
        )
        return JSONResponse(
            {
                "status": "not_ready",
                "service": "stackchain-dashboard",
                "error": error_message,
            },
            status_code=503,
            headers={
                "Retry-After": str(max(1, math.ceil(READINESS_TIMEOUT_SECONDS)))
            } if timed_out else None,
        )
    return {
        "status": "ready",
        "service": "stackchain-dashboard",
        "gitea_user": user["login"],
    }


@app.get("/api/v1/context")
async def context() -> JSONResponse:
    try:
        user_data, repo_data, issues_data, prs_data = await asyncio.wait_for(
            asyncio.gather(current_user(), repos(), issues(), pull_requests()),
            timeout=CONTEXT_TIMEOUT_SECONDS,
        )
    except Exception as e:
        error_message = (
            f"Gitea context request timed out after {CONTEXT_TIMEOUT_SECONDS:g}s"
            if isinstance(e, TimeoutError)
            else str(e)
        )
        return JSONResponse({
            "user": {"id": None, "login": "timmy", "full_name": "Timmy Jr", "email": ""},
            "repos": [],
            "issues": [],
            "pull_requests": [],
            "view": "dashboard",
            "deltas": [
                {"panel": "auth", "action": "connect", "target": "stackchain-dashboard backend", "priority": "high"},
                {"panel": "gitea", "action": "set GITEA_URL + GITEA_TOKEN", "target": "/root/stackchain-dashboard/.env or systemd env", "priority": "high"},
            ],
            "error": error_message,
        }, 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)
    ]
    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") or [])], assignees=[a.get("login", "") for a in (i.get("assignees") or [])], url=i["html_url"])
        for i in (issues_data or [])[:50]
        if isinstance(i, dict)
    ]
    pr_models = [
        PullRequest(id=p["id"], number=p["number"], title=p["title"], state=p["state"], user=(p.get("user") or {}).get("login", ""), url=p["html_url"])
        for p in (prs_data or [])[:50]
        if isinstance(p, dict)
    ]
    ctx = compute(user_model, repo_models, issue_models, pr_models)
    return JSONResponse(ctx.model_dump())


@app.get("/api/v1/events")
async def event_stream():
    try:
        return await asyncio.wait_for(
            activity_events(), timeout=EVENT_STREAM_TIMEOUT_SECONDS
        )
    except TimeoutError:
        return JSONResponse(
            {
                "error": (
                    "Gitea event stream request timed out after "
                    f"{EVENT_STREAM_TIMEOUT_SECONDS:g}s"
                )
            },
            status_code=503,
            headers={
                "Retry-After": str(
                    max(1, math.ceil(EVENT_STREAM_TIMEOUT_SECONDS))
                )
            },
        )
    except Exception:
        return JSONResponse(
            {"error": "Gitea event stream is temporarily unavailable"},
            status_code=503,
            headers={
                "Retry-After": str(
                    max(1, math.ceil(EVENT_STREAM_TIMEOUT_SECONDS))
                )
            },
        )