kanbante/app/main.py

"""
Kanbante — Kanban project manager integrated with Gitea.
FastAPI backend serving HTML + JSON API.

0.2.0 highlights (see CHANGELOG.md):
- Optional HTTP Basic auth (KANBANTE_AUTH_USER / _PASSWORD) + CSRF
  double-submit protection on every POST.
- Two-way sync: moving a card to "done" closes the Gitea issue,
  moving it back reopens it (KANBANTE_SYNC_GITEA=0 disables).
- Repos are classified with a rolling activity window and Gitea's real
  "archived" flag, not a hardcoded year.
- Board state is reconciled with Gitea on every board load, drag order
  is persisted, and Gitea failures render a proper 502 error page
  instead of a bare 500.
"""

import asyncio
import base64
import logging
import secrets
from contextlib import asynccontextmanager
from datetime import datetime, timedelta, timezone
from pathlib import Path

import httpx
from fastapi import FastAPI, Form, Request
from fastapi.responses import HTMLResponse, JSONResponse, Response
from fastapi.staticfiles import StaticFiles
from jinja2 import Environment, FileSystemLoader, select_autoescape
from starlette.middleware.base import BaseHTTPMiddleware

from app.config import APP_VERSION, get_settings
from app.gitea_client import GiteaClient
from app.models import (
    KANBAN_COLUMNS,
    get_board_state,
    get_notes,
    get_stats,
    init_db,
    move_issue,
    ping_db,
    reconcile_board_state,
    set_column_order,
    set_notes,
)

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("kanbante")

BASE = Path(__file__).resolve().parent.parent
CSRF_COOKIE = "kanbante_csrf"

COLUMN_LABELS = {
    "backlog": "📥 Backlog",
    "todo": "📋 À faire",
    "in_progress": "🚧 En cours",
    "review": "🔍 Révision",
    "done": "✅ Terminé",
}

COLUMN_COLORS = {
    "backlog": "#6b7280",
    "todo": "#3b82f6",
    "in_progress": "#f59e0b",
    "review": "#8b5cf6",
    "done": "#10b981",
}

FOCUS_COLUMNS = ["in_progress", "review"]


# ── App setup ──────────────────────────────────────────────

@asynccontextmanager
async def lifespan(app: FastAPI):
    init_db()
    settings = get_settings()
    if not settings.auth_enabled:
        logger.warning(
            "Authentication is DISABLED (KANBANTE_AUTH_USER / "
            "KANBANTE_AUTH_PASSWORD not set). Anyone who can reach this "
            "service can read every repo visible to the Gitea token and "
            "modify boards. Set the credentials or put Kanbante behind "
            "an authenticating reverse proxy."
        )
    if not settings.gitea_token:
        logger.warning(
            "GITEA_TOKEN is not set: only public Gitea content will be visible."
        )
    yield
    await gitea.aclose()


app = FastAPI(title="Kanbante", version=APP_VERSION, lifespan=lifespan)
app.mount("/static", StaticFiles(directory=str(BASE / "static")), name="static")
gitea = GiteaClient()


# ── Auth middleware ────────────────────────────────────────

class AuthMiddleware(BaseHTTPMiddleware):
    """HTTP Basic auth, enabled only when credentials are configured.

    /api/health stays public so Docker's HEALTHCHECK (liveness) works.
    When KANBANTE_REQUIRE_AUTH=1 and no credentials are configured, every
    other route answers 503: failing closed beats failing open.
    """

    async def dispatch(self, request: Request, call_next):
        if request.url.path == "/api/health":
            return await call_next(request)
        settings = get_settings()
        if not settings.auth_enabled:
            if settings.require_auth:
                return JSONResponse(
                    {
                        "ok": False,
                        "error": "Authentication required but "
                        "KANBANTE_AUTH_USER / KANBANTE_AUTH_PASSWORD "
                        "are not configured.",
                    },
                    status_code=503,
                )
            return await call_next(request)

        ok = False
        header = request.headers.get("authorization", "")
        if header.lower().startswith("basic "):
            try:
                decoded = base64.b64decode(header[6:]).decode("utf-8")
            except Exception:
                decoded = ""
            user, _, password = decoded.partition(":")
            ok = secrets.compare_digest(
                user, settings.auth_user
            ) and secrets.compare_digest(password, settings.auth_password)
        if not ok:
            return Response(
                "Authentication required",
                status_code=401,
                headers={"WWW-Authenticate": 'Basic realm="Kanbante"'},
            )
        return await call_next(request)


app.add_middleware(AuthMiddleware)


# ── Templates ──────────────────────────────────────────────

_tpl_env = Environment(
    loader=FileSystemLoader(str(BASE / "app" / "templates")),
    autoescape=select_autoescape(["html"]),
)


def render(
    request: Request, name: str, status_code: int = 200, **context
) -> HTMLResponse:
    """Render a Jinja2 template and manage the CSRF cookie.

    The CSRF token lives in an HttpOnly cookie; templates expose it in a
    meta tag / hidden fields (double-submit pattern) and POST handlers
    compare the submitted value with the cookie.
    """
    token = request.cookies.get(CSRF_COOKIE)
    generated = token is None
    if generated:
        token = secrets.token_urlsafe(32)
    context.setdefault("csrf_token", token)
    context.setdefault("version", APP_VERSION)
    html = _tpl_env.get_template(name).render(**context)
    response = HTMLResponse(html, status_code=status_code)
    if generated:
        response.set_cookie(CSRF_COOKIE, token, samesite="lax", httponly=True)
    return response


def _csrf_valid(request: Request, form_token: str) -> bool:
    cookie = request.cookies.get(CSRF_COOKIE, "")
    provided = form_token or request.headers.get("x-csrf-token", "")
    return bool(cookie) and bool(provided) and secrets.compare_digest(cookie, provided)


def _csrf_error() -> JSONResponse:
    return JSONResponse(
        {"ok": False, "error": "Invalid or missing CSRF token"}, status_code=403
    )


@app.exception_handler(httpx.HTTPError)
async def gitea_error_handler(request: Request, exc: httpx.HTTPError):
    """Gitea failures become a clean 502, never a bare 500."""
    logger.warning("Gitea error on %s: %s", request.url.path, exc)
    if request.url.path.startswith("/api/"):
        return JSONResponse(
            {"ok": False, "error": "Gitea is unreachable or returned an error"},
            status_code=502,
        )
    return render(
        request,
        "error.html",
        status_code=502,
        message="Kanbante n'a pas pu joindre Gitea. Vérifie que l'instance "
        "est en ligne et que le jeton est valide, puis réessaie.",
    )


# ── Helpers ────────────────────────────────────────────────

def _parse_datetime(value: str | None) -> datetime | None:
    if not value:
        return None
    try:
        return datetime.fromisoformat(value.replace("Z", "+00:00"))
    except ValueError:
        return None


def classify_repos(
    repos: list[dict], now: datetime | None = None
) -> tuple[list[dict], list[dict], list[dict]]:
    """Split repos into (active, inactive, archived).

    "Archived" is Gitea's own flag. Among the rest, a repo is active if
    it was updated within the configured rolling window — no hardcoded
    calendar year.
    """
    settings = get_settings()
    now = now or datetime.now(timezone.utc)
    cutoff = now - timedelta(days=settings.active_window_days)
    active, inactive, archived = [], [], []
    for repo in repos:
        if repo.get("archived"):
            archived.append(repo)
            continue
        updated = _parse_datetime(repo.get("updated_at"))
        if updated is not None and updated >= cutoff:
            active.append(repo)
        else:
            inactive.append(repo)
    return active, inactive, archived


def _is_issue(item: dict) -> bool:
    """Gitea's issues endpoint also returns PRs; they carry this key."""
    return "pull_request" not in item


def _merge_issues_with_board(
    issues: list[dict], board: dict[str, list[dict]]
) -> dict[str, list[dict]]:
    """Merge Gitea issues into kanban columns, honouring saved order."""
    placement: dict[int, tuple[str, float]] = {}
    for col_name, cards in board.items():
        for card in cards:
            placement[card["issue_number"]] = (col_name, card["position"])

    columns: dict[str, list[dict]] = {col: [] for col in KANBAN_COLUMNS}
    for index, issue in enumerate(issues):
        col, pos = placement.get(issue["number"], ("backlog", None))
        issue["_sort"] = (0, pos, index) if pos is not None else (1, 0.0, index)
        columns[col].append(issue)
    for col in columns.values():
        col.sort(key=lambda i: i["_sort"])
    return columns


def _split_full_name(repo_full_name: str) -> tuple[str, str] | None:
    owner, sep, repo = repo_full_name.partition("/")
    if not sep or not owner or not repo or "/" in repo:
        return None
    return owner, repo


def _parse_order(raw: str) -> list[int]:
    numbers = []
    for chunk in (raw or "").split(","):
        chunk = chunk.strip()
        if chunk.isdigit():
            numbers.append(int(chunk))
    return numbers


# ── Routes: Pages ─────────────────────────────────────────

@app.get("/", response_class=HTMLResponse)
async def dashboard(request: Request):
    """Main dashboard — all projects at a glance."""
    repos = await gitea.search_repos()
    active, inactive, archived = classify_repos(repos)
    return render(
        request,
        "dashboard.html",
        active_repos=active,
        inactive_repos=inactive,
        archived_repos=archived,
        stats=get_stats(),
        COLUMN_LABELS=COLUMN_LABELS,
        COLUMN_COLORS=COLUMN_COLORS,
        active_window_days=get_settings().active_window_days,
    )


@app.get("/board/{owner}/{repo}", response_class=HTMLResponse)
async def board(request: Request, owner: str, repo: str):
    """Kanban board for a specific repo."""
    full_name = f"{owner}/{repo}"
    all_items = await gitea.get_repo_issues(owner, repo, state="all")
    issues = [i for i in all_items if _is_issue(i)]
    open_issues = [i for i in issues if i.get("state") == "open"]
    closed_numbers = {i["number"] for i in issues if i.get("state") != "open"}
    open_numbers = {i["number"] for i in open_issues}

    # Keep local state truthful: closed in Gitea -> done, deleted -> purged.
    reconcile_board_state(full_name, open_numbers, closed_numbers)

    columns = _merge_issues_with_board(open_issues, get_board_state(full_name))

    labels: dict[str, dict] = {}
    for issue in open_issues:
        for label in issue.get("labels") or []:
            labels[label["name"]] = label

    repo_info = {}
    try:
        repo_info = await gitea.get_repo(owner, repo)
    except Exception as exc:  # non-critical decoration
        logger.debug("repo info failed for %s: %s", full_name, exc)

    return render(
        request,
        "board.html",
        owner=owner,
        repo=repo,
        full_name=full_name,
        columns=columns,
        column_labels=COLUMN_LABELS,
        column_colors=COLUMN_COLORS,
        labels=sorted(labels.values(), key=lambda l: l["name"].lower()),
        notes=get_notes(full_name) or "",
        repo_info=repo_info,
    )


@app.get("/focus", response_class=HTMLResponse)
async def focus(request: Request):
    """Cross-project view: everything in progress or in review."""
    repos = await gitea.search_repos()
    active, _inactive, _archived = classify_repos(repos)
    active = active[:100]

    async def _open_issues(repo: dict):
        owner, _, name = repo["full_name"].partition("/")
        items = await gitea.get_repo_issues(owner, name, state="open")
        return repo["full_name"], [i for i in items if _is_issue(i)]

    results = await asyncio.gather(
        *(_open_issues(r) for r in active), return_exceptions=True
    )
    groups = []
    for result in results:
        if isinstance(result, Exception):
            logger.warning("focus: repo fetch failed: %s", result)
            continue
        full_name, open_issues = result
        columns = _merge_issues_with_board(open_issues, get_board_state(full_name))
        cards = [
            {"issue": issue, "column": col}
            for col in FOCUS_COLUMNS
            for issue in columns[col]
        ]
        if cards:
            groups.append({"full_name": full_name, "cards": cards})
    return render(
        request,
        "focus.html",
        groups=groups,
        column_labels=COLUMN_LABELS,
        column_colors=COLUMN_COLORS,
    )


# ── Routes: API ────────────────────────────────────────────

@app.post("/api/move")
async def api_move(
    request: Request,
    repo_full_name: str = Form(...),
    issue_number: int = Form(...),
    to_column: str = Form(...),
    from_column: str = Form(""),
    to_order: str = Form(""),
    from_order: str = Form(""),
    csrf_token: str = Form(""),
):
    """Move an issue to a new column and persist the resulting order.

    With KANBANTE_SYNC_GITEA enabled (default), moving to "done" closes
    the issue in Gitea and moving out of "done" reopens it. A failed
    write-back is reported in the response (gitea_sync="failed") instead
    of failing the move itself.
    """
    if not _csrf_valid(request, csrf_token):
        return _csrf_error()
    if to_column not in KANBAN_COLUMNS:
        return JSONResponse(
            {"ok": False, "error": "Invalid column"}, status_code=400
        )
    parts = _split_full_name(repo_full_name)
    if parts is None:
        return JSONResponse(
            {"ok": False, "error": "Invalid repo_full_name"}, status_code=400
        )
    owner, repo = parts

    move_issue(repo_full_name, issue_number, to_column)
    if from_column in KANBAN_COLUMNS and from_column != to_column:
        remaining = [
            n for n in _parse_order(from_order) if n != issue_number
        ]
        if remaining:
            set_column_order(repo_full_name, from_column, remaining)
    order = _parse_order(to_order)
    if order:
        if issue_number not in order:
            order.append(issue_number)
        set_column_order(repo_full_name, to_column, order)

    result: dict = {"ok": True, "gitea_sync": "skipped"}
    settings = get_settings()
    if settings.sync_gitea:
        try:
            if to_column == "done":
                await gitea.set_issue_state(owner, repo, issue_number, "closed")
                result["gitea_sync"] = "closed"
            elif from_column == "done":
                await gitea.set_issue_state(owner, repo, issue_number, "open")
                result["gitea_sync"] = "reopened"
        except httpx.HTTPError as exc:
            logger.warning(
                "Gitea write-back failed for %s#%s: %s",
                repo_full_name,
                issue_number,
                exc,
            )
            result["gitea_sync"] = "failed"
    return result


@app.post("/api/issues")
async def api_create_issue(
    request: Request,
    repo_full_name: str = Form(...),
    title: str = Form(...),
    body: str = Form(""),
    column: str = Form("backlog"),
    csrf_token: str = Form(""),
):
    """Create a Gitea issue from the board and place its card."""
    if not _csrf_valid(request, csrf_token):
        return _csrf_error()
    parts = _split_full_name(repo_full_name)
    if parts is None:
        return JSONResponse(
            {"ok": False, "error": "Invalid repo_full_name"}, status_code=400
        )
    title = title.strip()
    if not title or len(title) > 255:
        return JSONResponse(
            {"ok": False, "error": "Title required (max 255 chars)"},
            status_code=400,
        )
    if column not in KANBAN_COLUMNS:
        return JSONResponse(
            {"ok": False, "error": "Invalid column"}, status_code=400
        )
    owner, repo = parts
    issue = await gitea.create_issue(owner, repo, title, body)
    move_issue(repo_full_name, issue["number"], column)
    return JSONResponse(
        {
            "ok": True,
            "issue": {
                "number": issue["number"],
                "title": issue.get("title", title),
                "html_url": issue.get("html_url", ""),
            },
        },
        status_code=201,
    )


@app.post("/api/notes")
async def api_notes(
    request: Request,
    repo_full_name: str = Form(...),
    note: str = Form(""),
    csrf_token: str = Form(""),
):
    """Save project notes."""
    if not _csrf_valid(request, csrf_token):
        return _csrf_error()
    settings = get_settings()
    if len(note) > settings.max_note_chars:
        return JSONResponse(
            {
                "ok": False,
                "error": f"Note too large (max {settings.max_note_chars} chars)",
            },
            status_code=413,
        )
    set_notes(repo_full_name, note)
    return {"ok": True}


@app.get("/api/stats")
async def api_stats():
    """Get project stats."""
    return get_stats()


@app.get("/api/projects")
async def api_projects():
    """Get all projects (JSON), classified like the dashboard."""
    repos = await gitea.search_repos()
    active, inactive, archived = classify_repos(repos)
    return {"active": active, "inactive": inactive, "archived": archived}


@app.get("/api/health")
async def health():
    """Liveness probe: the process is up. Deliberately dependency-free."""
    return {"status": "ok", "version": APP_VERSION}


@app.get("/api/readiness")
async def readiness():
    """Readiness probe: database writable and Gitea reachable."""
    db_ok = ping_db()
    gitea_ok = await gitea.ping()
    ready = db_ok and gitea_ok
    return JSONResponse(
        {
            "status": "ready" if ready else "not_ready",
            "version": APP_VERSION,
            "checks": {"database": db_ok, "gitea": gitea_ok},
        },
        status_code=200 if ready else 503,
    )