devplacepy/devplacepy/routers/bugs.py

# retoor <retoor@molodetz.nl>

import logging
from typing import Annotated

from fastapi import APIRouter, Form, Request
from fastapi.responses import HTMLResponse, JSONResponse

from devplacepy.database import build_pagination, get_table, get_users_by_uids
from devplacepy.models import BugCommentForm, BugForm, BugStatusForm
from devplacepy.responses import action_result, json_error, respond, wants_json
from devplacepy.templating import templates
from devplacepy.schemas import BugDetailOut, BugJobOut, BugsOut
from devplacepy.seo import base_seo_context, site_url, website_schema
from devplacepy.services.audit import record as audit
from devplacepy.services.gitea import runtime, store
from devplacepy.services.gitea.client import (
    STATE_ALL,
    STATE_CLOSED,
    STATE_OPEN,
    GiteaError,
)
from devplacepy.services.gitea.config import gitea_config
from devplacepy.services.jobs import queue
from devplacepy.utils import (
    create_notification,
    get_current_user,
    is_admin,
    not_found,
    require_user,
)

logger = logging.getLogger(__name__)
router = APIRouter()

PER_PAGE = 20
VALID_STATES = (STATE_OPEN, STATE_CLOSED, STATE_ALL)


def _state(raw: str) -> str:
    return raw if raw in VALID_STATES else STATE_OPEN


def _author_fields(author_uid: str | None, users_map: dict, fallback_login: str) -> dict:
    user = users_map.get(author_uid) if author_uid else None
    if user:
        return {
            "author_username": user["username"],
            "author_uid": author_uid,
            "author_avatar_seed": user["username"],
            "is_local_author": True,
        }
    return {
        "author_username": fallback_login or "developer",
        "author_uid": None,
        "author_avatar_seed": None,
        "is_local_author": False,
    }


def _issue_item(issue: dict, author_uid: str | None, users_map: dict) -> dict:
    login = (issue.get("user") or {}).get("login", "")
    return {
        "number": int(issue.get("number", 0)),
        "title": issue.get("title", ""),
        "state": issue.get("state", "open"),
        "html_url": issue.get("html_url"),
        "comments_count": int(issue.get("comments", 0)),
        "created_at": issue.get("created_at"),
        "updated_at": issue.get("updated_at"),
        **_author_fields(author_uid, users_map, login),
    }


def _comment_item(comment: dict, author_uid: str | None, users_map: dict) -> dict:
    login = (comment.get("user") or {}).get("login", "")
    return {
        "id": int(comment.get("id", 0)),
        "body": comment.get("body", ""),
        "html_url": comment.get("html_url"),
        "created_at": comment.get("created_at"),
        **_author_fields(author_uid, users_map, login),
    }


def _seo(request: Request, **extra) -> dict:
    base = site_url(request)
    return base_seo_context(
        request,
        breadcrumbs=[
            {"name": "Home", "url": "/feed"},
            {"name": "Bug Reports", "url": "/bugs"},
        ],
        schemas=[website_schema(base)],
        **extra,
    )


@router.get("", response_class=HTMLResponse)
async def bugs_page(request: Request, state: str = STATE_OPEN, page: int = 1):
    user = get_current_user(request)
    state = _state(state)
    page = max(1, page)
    seo_ctx = _seo(
        request,
        title="Bug Reports",
        description="Report bugs and track their progress on DevPlace.",
    )
    base_ctx = {
        **seo_ctx,
        "request": request,
        "user": user,
        "state": state,
        "bugs": [],
        "pagination": None,
        "configured": True,
        "error_message": None,
    }

    config = gitea_config()
    if not config.is_configured:
        base_ctx["configured"] = False
        base_ctx["error_message"] = "The bug tracker is not configured yet."
        return respond(request, "bugs.html", base_ctx, model=BugsOut)

    try:
        issues, total = await runtime.get_client().list_issues(
            state=state, page=page, limit=PER_PAGE
        )
    except GiteaError as exc:
        logger.warning("Could not load bug list: %s", exc)
        base_ctx["error_message"] = "The bug tracker is temporarily unavailable."
        return respond(request, "bugs.html", base_ctx, model=BugsOut)

    numbers = [int(issue.get("number", 0)) for issue in issues]
    author_by_number = store.author_map(numbers)
    users_map = get_users_by_uids([uid for uid in author_by_number.values() if uid])
    base_ctx["bugs"] = [
        _issue_item(issue, author_by_number.get(int(issue.get("number", 0))), users_map)
        for issue in issues
    ]
    base_ctx["pagination"] = build_pagination(page, total, PER_PAGE)
    return respond(request, "bugs.html", base_ctx, model=BugsOut)


@router.post("/create")
async def create_bug(request: Request, data: Annotated[BugForm, Form()]):
    user = require_user(request)
    if not gitea_config().is_configured:
        return json_error(503, "The bug tracker is not configured")
    title = data.title.strip()
    uid = queue.enqueue(
        "bug_create",
        {
            "author_uid": user["uid"],
            "title": title,
            "description": data.description.strip(),
        },
        "user",
        user["uid"],
        title[:64],
    )
    audit.record(
        request,
        "bug.create.request",
        user=user,
        target_type="bug",
        summary=f"{user['username']} requested a bug report: {title[:60]}",
        metadata={"title": title},
        links=[audit.job(uid)],
    )
    logger.info("Bug report enqueued by %s: %s", user["username"], title[:60])
    return JSONResponse({"uid": uid, "status_url": f"/bugs/jobs/{uid}"})


@router.get("/jobs/{uid}")
async def bug_job_status(request: Request, uid: str):
    job = queue.get_job(uid)
    if not job or job.get("kind") != "bug_create":
        raise not_found("Job not found")
    result = job.get("result", {})
    done = job.get("status") == queue.DONE
    payload = {
        "uid": job.get("uid", ""),
        "kind": job.get("kind", ""),
        "status": job.get("status", ""),
        "number": result.get("number") if done else None,
        "bug_url": result.get("bug_url") if done else None,
        "enhanced": bool(result.get("enhanced")) if done else False,
        "error": job.get("error") or None,
        "created_at": job.get("created_at"),
        "completed_at": job.get("completed_at") or None,
    }
    return JSONResponse(BugJobOut.model_validate(payload).model_dump(mode="json"))


def _tracker_unavailable(request: Request):
    if wants_json(request):
        return json_error(503, "The bug tracker is temporarily unavailable")
    seo_ctx = base_seo_context(
        request,
        title="Bug tracker unavailable - DevPlace",
        description="The bug tracker is temporarily unavailable.",
        robots="noindex",
    )
    return templates.TemplateResponse(
        request,
        "error.html",
        {
            **seo_ctx,
            "request": request,
            "error_code": 503,
            "error_message": "The bug tracker is temporarily unavailable.",
        },
        status_code=503,
    )


@router.get("/{number}", response_class=HTMLResponse)
async def bug_detail(request: Request, number: int):
    user = get_current_user(request)
    if not gitea_config().is_configured:
        raise not_found("Bug not found")
    client = runtime.get_client()
    try:
        issue = await client.get_issue(number)
    except GiteaError as exc:
        if exc.status == 404:
            raise not_found("Bug not found")
        logger.warning("Could not load bug #%s: %s", number, exc)
        return _tracker_unavailable(request)
    try:
        comments = await client.list_comments(number)
    except GiteaError as exc:
        logger.warning("Could not load comments for bug #%s: %s", number, exc)
        comments = []

    author_uid = store.author_uid_for_issue(number)
    comment_authors = store.comment_author_map(
        [int(comment.get("id", 0)) for comment in comments]
    )
    uids = [uid for uid in [author_uid, *comment_authors.values()] if uid]
    users_map = get_users_by_uids(uids)

    bug = _issue_item(issue, author_uid, users_map)
    comment_items = [
        _comment_item(
            comment, comment_authors.get(int(comment.get("id", 0))), users_map
        )
        for comment in comments
    ]
    seo_ctx = _seo(
        request,
        title=f"Bug #{number}: {issue.get('title', '')}",
        description=issue.get("title", ""),
    )
    return respond(
        request,
        "bug_detail.html",
        {
            **seo_ctx,
            "request": request,
            "user": user,
            "bug": bug,
            "body": issue.get("body", ""),
            "comments": comment_items,
            "can_comment": user is not None,
            "viewer_is_admin": is_admin(user),
        },
        model=BugDetailOut,
    )


@router.post("/{number}/comment")
async def comment_bug(
    request: Request, number: int, data: Annotated[BugCommentForm, Form()]
):
    user = require_user(request)
    if not gitea_config().is_configured:
        return json_error(503, "The bug tracker is not configured")
    client = runtime.get_client()
    body = (
        f"{data.body.strip()}\n\n---\n"
        f"*Posted by **{user['username']}** via DevPlace.*"
    )
    try:
        comment = await client.create_comment(number, body)
    except GiteaError as exc:
        logger.warning("Could not post comment on #%s: %s", number, exc)
        audit.record(
            request,
            "bug.comment",
            user=user,
            result="failure",
            target_type="bug",
            target_uid=str(number),
            summary=f"Failed to comment on bug #{number}: {exc}",
        )
        return json_error(exc.status or 502, "Could not post the comment")

    comment_id = int(comment.get("id", 0))
    store.record_comment_author(comment_id, number, user["uid"])
    _notify_admins(user, number)
    audit.record(
        request,
        "bug.comment",
        user=user,
        target_type="bug",
        target_uid=str(number),
        summary=f"{user['username']} commented on bug #{number}",
        metadata={"number": number, "comment_id": comment_id},
        links=[audit.target("bug", str(number))],
    )
    logger.info("Comment posted on bug #%s by %s", number, user["username"])
    return action_result(request, f"/bugs/{number}", data={"comment_id": comment_id})


@router.post("/{number}/status")
async def status_bug(
    request: Request, number: int, data: Annotated[BugStatusForm, Form()]
):
    user = require_user(request)
    if not is_admin(user):
        audit.record(
            request,
            "bug.status",
            user=user,
            result="denied",
            target_type="bug",
            target_uid=str(number),
            summary=f"{user['username']} denied changing status of bug #{number}",
        )
        return json_error(403, "Admins only")
    if not gitea_config().is_configured:
        return json_error(503, "The bug tracker is not configured")
    try:
        await runtime.get_client().set_state(number, data.status)
    except GiteaError as exc:
        logger.warning("Could not change status of #%s: %s", number, exc)
        audit.record(
            request,
            "bug.status",
            user=user,
            result="failure",
            target_type="bug",
            target_uid=str(number),
            summary=f"Failed to change status of bug #{number}: {exc}",
        )
        return json_error(exc.status or 502, "Could not change the status")

    audit.record(
        request,
        "bug.status",
        user=user,
        target_type="bug",
        target_uid=str(number),
        new_value=data.status,
        summary=f"{user['username']} set bug #{number} to {data.status}",
        metadata={"number": number, "state": data.status},
        links=[audit.target("bug", str(number))],
    )
    logger.info("Bug #%s set to %s by %s", number, data.status, user["username"])
    return action_result(request, f"/bugs/{number}", data={"state": data.status})


def _notify_admins(actor: dict, number: int) -> None:
    for admin in get_table("users").find(role="admin"):
        if admin["uid"] == actor["uid"]:
            continue
        create_notification(
            admin["uid"],
            "bug",
            f"{actor['username']} commented on bug #{number}",
            str(number),
            f"/bugs/{number}",
        )