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 hashlib
import hmac
import json
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, StreamingResponse
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.briefing import format_briefing_text, ntfy_configured, send_ntfy
from app.models import (
    KANBAN_COLUMNS,
    column_entry_times,
    get_board_state,
    get_events,
    get_kv,
    get_metrics,
    get_notes,
    get_stats,
    init_db,
    log_event,
    move_issue,
    ping_db,
    reconcile_board_state,
    set_column_order,
    set_kv,
    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"]

EVENT_ICONS = {
    "created": "➕",
    "moved": "➡️",
    "closed": "✅",
    "reopened": "🔁",
    "commented": "💬",
}


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

async def _briefing_scheduler() -> None:
    """Send the daily briefing to ntfy at KANBANTE_BRIEFING_TIME.

    Checks once a minute; the kv store remembers the last day sent so
    restarts never double-send.
    """
    while True:
        try:
            settings = get_settings()
            if settings.briefing_time and ntfy_configured(settings):
                now = datetime.now()
                today = now.strftime("%Y-%m-%d")
                if (
                    now.strftime("%H:%M") >= settings.briefing_time
                    and get_kv("briefing_last_sent") != today
                ):
                    data = await _briefing_data()
                    if await send_ntfy(settings, format_briefing_text(data)):
                        set_kv("briefing_last_sent", today)
                        logger.info("Daily briefing sent to ntfy (%s)", today)
        except asyncio.CancelledError:
            raise
        except Exception as exc:  # never let the scheduler die silently
            logger.warning("Briefing scheduler error: %s", exc)
        await asyncio.sleep(60)


@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."
        )
    scheduler = asyncio.create_task(_briefing_scheduler())
    yield
    scheduler.cancel()
    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):
        # /api/health: liveness for Docker. /webhooks/gitea: authenticated
        # by its HMAC signature, not by Basic auth (Gitea cannot log in).
        if request.url.path in ("/api/health", "/webhooks/gitea"):
            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)


# ── Live updates (SSE hub) ─────────────────────────────────

class LiveHub:
    """Fan-out of Gitea webhook events to connected browsers (SSE)."""

    def __init__(self) -> None:
        self._subscribers: set[asyncio.Queue] = set()

    def subscribe(self) -> asyncio.Queue:
        queue: asyncio.Queue = asyncio.Queue(maxsize=50)
        self._subscribers.add(queue)
        return queue

    def unsubscribe(self, queue: asyncio.Queue) -> None:
        self._subscribers.discard(queue)

    async def broadcast(self, payload: dict) -> None:
        for queue in list(self._subscribers):
            try:
                queue.put_nowait(payload)
            except asyncio.QueueFull:
                pass  # slow client: it will resync on its next load

    @property
    def subscriber_count(self) -> int:
        return len(self._subscribers)


hub = LiveHub()


# ── 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:
    """True for issues, False for pull requests.

    Gitea's issues endpoint also returns PRs. A PR carries a
    ``pull_request`` *object*; a plain issue carries the same key
    with value ``null`` (verified against Gitea 1.27.3). Testing key
    presence therefore classifies every real issue as a PR and
    empties the board — test the value's truthiness instead.
    """
    return not item.get("pull_request")


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,
        metrics=get_metrics(window_days=30),
    )


@app.get("/board/{owner}/{repo}", response_class=HTMLResponse)
async def board(request: Request, owner: str, repo: str, milestone: str = ""):
    """Kanban board for a specific repo (optionally one milestone)."""
    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.
    changes = reconcile_board_state(full_name, open_numbers, closed_numbers)
    for number in changes["closed_numbers"]:
        log_event(full_name, number, "closed", to_column="done", source="gitea")

    # Milestone view: filter the board to one milestone (sprint-style),
    # with its deadline and completion progress.
    milestones: list[dict] = []
    selected_milestone = None
    milestone_progress = None
    display_issues = open_issues
    try:
        milestones = await gitea.get_milestones(owner, repo)
    except Exception as exc:  # milestones are decoration, never fatal
        logger.debug("milestones failed for %s: %s", full_name, exc)
        milestones = []
    if milestone:
        def _in_milestone(issue: dict) -> bool:
            ms = issue.get("milestone") or {}
            return bool(ms) and (
                str(ms.get("id")) == milestone or ms.get("title") == milestone
            )

        display_issues = [i for i in open_issues if _in_milestone(i)]
        selected_milestone = next(
            (
                m for m in milestones
                if str(m.get("id")) == milestone or m.get("title") == milestone
            ),
            None,
        )
        if selected_milestone is None and display_issues:
            selected_milestone = display_issues[0].get("milestone")
        scoped = [i for i in issues if _in_milestone(i)]
        total = len(scoped)
        closed_count = sum(1 for i in scoped if i.get("state") == "closed")
        milestone_progress = {
            "total": total,
            "closed": closed_count,
            "pct": round(100 * closed_count / total) if total else 0,
            "due_on": (selected_milestone or {}).get("due_on"),
            "title": (selected_milestone or {}).get("title", milestone),
        }

    board_state = get_board_state(full_name)
    columns = _merge_issues_with_board(display_issues, board_state)

    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)

    settings = get_settings()
    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,
        ages=_card_ages(full_name, display_issues, board_state),
        wip_limits=settings.wip_limits or {},
        aging_days=settings.aging_days,
        milestones=milestones,
        selected_milestone=selected_milestone,
        milestone_progress=milestone_progress,
        milestone_param=milestone,
    )


def _card_ages(
    full_name: str, open_issues: list[dict], board: dict[str, list[dict]]
) -> dict[int, dict]:
    """Days each open card has spent in its current column.

    Source of truth, in order: the event journal (latest entry),
    board_state.updated_at (pre-journal data), the issue's own
    creation date.
    """
    entered = column_entry_times(full_name)
    updated = {
        card["issue_number"]: card["updated_at"]
        for cards in board.values()
        for card in cards
    }
    now = datetime.now(timezone.utc)
    ages: dict[int, dict] = {}
    for issue in open_issues:
        number = issue["number"]
        raw = entered.get(number) or updated.get(number) or issue.get("created_at")
        since = _parse_datetime(raw)
        if since is None:
            continue
        if since.tzinfo is None:
            since = since.replace(tzinfo=timezone.utc)
        ages[number] = {"days": max(0, (now - since).days), "since": raw}
    return ages


async def _collect_focus_groups() -> list[dict]:
    """In-progress / in-review cards across all active repos."""
    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 groups


@app.get("/focus", response_class=HTMLResponse)
async def focus(request: Request):
    """Cross-project view: everything in progress or in review."""
    return render(
        request,
        "focus.html",
        groups=await _collect_focus_groups(),
        column_labels=COLUMN_LABELS,
        column_colors=COLUMN_COLORS,
    )


def _group_issue_results(issues: list[dict]) -> list[dict]:
    """Group search results by repo, annotated with the board column."""
    by_repo: dict[str, list[dict]] = {}
    for issue in issues:
        full_name = (issue.get("repository") or {}).get("full_name", "?")
        by_repo.setdefault(full_name, []).append(issue)
    groups = []
    for full_name in sorted(by_repo):
        state = get_board_state(full_name) if "/" in full_name else {}
        column_of = {
            card["issue_number"]: column
            for column, cards in state.items()
            for card in cards
        }
        rows = []
        for issue in by_repo[full_name]:
            if issue.get("state") == "open":
                column = column_of.get(issue["number"], "backlog")
            else:
                column = "done"
            rows.append({"issue": issue, "column": column})
        groups.append({"full_name": full_name, "rows": rows})
    return groups


@app.get("/search", response_class=HTMLResponse)
async def search_page(request: Request, q: str = ""):
    """Global issue search across every repo (Gitea search API)."""
    q = q.strip()
    groups: list[dict] = []
    total = 0
    if q:
        results = await gitea.search_issues(q=q, state="all")
        total = len(results)
        groups = _group_issue_results(results)
    return render(
        request,
        "search.html",
        q=q,
        groups=groups,
        total=total,
        column_labels=COLUMN_LABELS,
        column_colors=COLUMN_COLORS,
    )


@app.get("/my", response_class=HTMLResponse)
async def my_issues(request: Request):
    """Open issues assigned to the Gitea token's owner, all repos."""
    user = None
    groups: list[dict] = []
    total = 0
    try:
        user = await gitea.get_user()
    except Exception as exc:
        logger.debug("get_user failed for /my: %s", exc)
    if user and user.get("login"):
        results = await gitea.search_issues(
            state="open", assigned_by=user["login"]
        )
        total = len(results)
        groups = _group_issue_results(results)
    return render(
        request,
        "my.html",
        user=user,
        groups=groups,
        total=total,
        column_labels=COLUMN_LABELS,
        column_colors=COLUMN_COLORS,
    )


# ── Briefing ───────────────────────────────────────────────

async def _briefing_data() -> dict:
    """Everything the daily briefing needs, gathered in one dict."""
    settings = get_settings()
    focus_groups = await _collect_focus_groups()

    repos = await gitea.search_repos()
    active, _i, _a = classify_repos(repos)
    soon = datetime.now(timezone.utc) + timedelta(days=7)
    aging: list[dict] = []
    milestones_due: list[dict] = []

    async def _per_repo(repo: dict):
        full_name = repo["full_name"]
        owner, _, name = full_name.partition("/")
        items = await gitea.get_repo_issues(owner, name, state="open")
        open_issues = [i for i in items if _is_issue(i)]
        ages = _card_ages(full_name, open_issues, get_board_state(full_name))
        board = get_board_state(full_name)
        column_of = {
            card["issue_number"]: column
            for column, cards in board.items()
            for card in cards
        }
        repo_aging = [
            {
                "repo": full_name,
                "number": issue["number"],
                "title": issue["title"],
                "days": ages[issue["number"]]["days"],
                "column_label": COLUMN_LABELS[
                    column_of.get(issue["number"], "backlog")
                ],
            }
            for issue in open_issues
            if issue["number"] in ages
            and ages[issue["number"]]["days"] >= settings.aging_days
        ]
        repo_milestones = []
        try:
            for ms in await gitea.get_milestones(owner, name):
                due = _parse_datetime(ms.get("due_on"))
                if due and due <= soon and (ms.get("open_issues") or 0) > 0:
                    repo_milestones.append(
                        {
                            "repo": full_name,
                            "title": ms.get("title", ""),
                            "due_on": str(ms["due_on"])[:10],
                            "open": ms.get("open_issues") or 0,
                        }
                    )
        except Exception as exc:
            logger.debug("briefing milestones failed for %s: %s", full_name, exc)
        return repo_aging, repo_milestones

    results = await asyncio.gather(
        *(_per_repo(r) for r in active[:100]), return_exceptions=True
    )
    for result in results:
        if isinstance(result, Exception):
            logger.warning("briefing: repo fetch failed: %s", result)
            continue
        repo_aging, repo_milestones = result
        aging.extend(repo_aging)
        milestones_due.extend(repo_milestones)
    aging.sort(key=lambda c: -c["days"])

    metrics_24h = get_metrics(window_days=1)
    metrics_30d = get_metrics(window_days=30)
    return {
        "date": datetime.now().strftime("%d/%m/%Y"),
        "focus_total": sum(len(g["cards"]) for g in focus_groups),
        "focus_groups": [
            {
                "full_name": g["full_name"],
                "count": len(g["cards"]),
                "titles": [c["issue"]["title"] for c in g["cards"]],
            }
            for g in focus_groups
        ],
        "aging": aging[:10],
        "milestones_due": milestones_due[:5],
        "activity_24h": {
            **metrics_24h,
            "avg_cycle_days": metrics_30d["avg_cycle_days"],
        },
    }


@app.get("/briefing", response_class=HTMLResponse)
async def briefing_page(request: Request):
    """The daily briefing, readable on the web before it is pushed."""
    settings = get_settings()
    data = await _briefing_data()
    return render(
        request,
        "briefing.html",
        data=data,
        text=format_briefing_text(data),
        ntfy_configured=ntfy_configured(settings),
        briefing_time=settings.briefing_time,
        last_sent=get_kv("briefing_last_sent"),
    )


@app.get("/api/briefing")
async def api_briefing():
    data = await _briefing_data()
    return {"ok": True, "briefing": data, "text": format_briefing_text(data)}


@app.post("/api/briefing/send")
async def api_briefing_send(request: Request, csrf_token: str = Form("")):
    """Send the briefing to the configured ntfy topic, right now."""
    if not _csrf_valid(request, csrf_token):
        return _csrf_error()
    settings = get_settings()
    if not ntfy_configured(settings):
        return JSONResponse(
            {
                "ok": False,
                "error": "ntfy is not configured "
                         "(KANBANTE_NTFY_URL / KANBANTE_NTFY_TOPIC)",
            },
            status_code=400,
        )
    data = await _briefing_data()
    sent = await send_ntfy(settings, format_briefing_text(data))
    if not sent:
        return JSONResponse(
            {"ok": False, "error": "ntfy server rejected or unreachable"},
            status_code=502,
        )
    set_kv("briefing_last_sent", datetime.now().strftime("%Y-%m-%d"))
    return {"ok": True, "sent": True}


# ── Webhooks + live stream ─────────────────────────────────

def _apply_external_transition(
    full_name: str, number: int, target_column: str, event_type: str
) -> str:
    """Apply a close/reopen that happened in Gitea itself.

    Idempotent by construction: when Kanbante's own API made the
    change, the local row is already in the target state and this is
    a no-op — so our own webhook echo never double-logs.
    """
    board = get_board_state(full_name)
    current = None
    for column, cards in board.items():
        if any(c["issue_number"] == number for c in cards):
            current = column
            break
    if current is None or current == target_column:
        return "noop"
    if event_type == "reopened" and current != "done":
        return "noop"  # locally open already: nothing truthful to apply
    move_issue(full_name, number, target_column)
    if event_type == "closed":
        log_event(full_name, number, "closed", to_column="done", source="webhook")
    else:
        log_event(
            full_name, number, "reopened",
            from_column="done", to_column=target_column, source="webhook",
        )
    return event_type


@app.post("/webhooks/gitea")
async def gitea_webhook(request: Request):
    """Receive Gitea webhooks: verify HMAC, apply external close/reopen,
    invalidate the read cache and push a live event to browsers."""
    settings = get_settings()
    raw = await request.body()
    if settings.webhook_secret:
        signature = request.headers.get("x-gitea-signature", "")
        expected = hmac.new(
            settings.webhook_secret.encode(), raw, hashlib.sha256
        ).hexdigest()
        if not signature or not hmac.compare_digest(signature, expected):
            return JSONResponse(
                {"ok": False, "error": "Invalid webhook signature"},
                status_code=401,
            )
    try:
        payload = json.loads(raw or b"{}")
    except ValueError:
        return JSONResponse(
            {"ok": False, "error": "Invalid JSON payload"}, status_code=400
        )

    event = request.headers.get("x-gitea-event", "")
    action = payload.get("action", "")
    full_name = (payload.get("repository") or {}).get("full_name", "")
    issue = payload.get("issue") or {}
    number = issue.get("number")

    handled = "ignored"
    if event == "issues" and full_name and number:
        if action == "closed":
            handled = _apply_external_transition(full_name, number, "done", "closed")
        elif action == "reopened":
            handled = _apply_external_transition(
                full_name, number, "backlog", "reopened"
            )

    gitea.invalidate_cache()
    await hub.broadcast(
        {"repo": full_name, "event": event, "action": action, "issue": number}
    )
    return {"ok": True, "handled": handled}


async def sse_event_stream(request: Request):
    """Yield SSE frames for one subscriber until it disconnects.

    Split out of the route so it can be unit-tested directly:
    test clients cannot reliably tear down an infinite HTTP stream.
    """
    queue = hub.subscribe()
    try:
        yield "retry: 5000\n\n"
        while True:
            try:
                payload = await asyncio.wait_for(queue.get(), timeout=25)
            except asyncio.TimeoutError:
                if await request.is_disconnected():
                    break
                yield ": heartbeat\n\n"
                continue
            if await request.is_disconnected():
                break
            yield f"event: gitea\ndata: {json.dumps(payload)}\n\n"
    finally:
        hub.unsubscribe(queue)


@app.get("/api/stream")
async def live_stream(request: Request):
    """Server-Sent Events: browsers reload softly on Gitea activity."""
    return StreamingResponse(
        sse_event_stream(request),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
    )


@app.get("/activity", response_class=HTMLResponse)
async def activity(request: Request):
    """Activity feed: the latest entries of the event journal."""
    events = get_events(limit=100)
    return render(
        request,
        "activity.html",
        events=events,
        event_icons=EVENT_ICONS,
        column_labels=COLUMN_LABELS,
    )


# ── 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)

    if from_column != to_column:
        log_event(
            repo_full_name, issue_number, "moved",
            from_column or None, to_column,
        )

    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"
                log_event(repo_full_name, issue_number, "closed", to_column="done")
            elif from_column == "done":
                await gitea.set_issue_state(owner, repo, issue_number, "open")
                result["gitea_sync"] = "reopened"
                log_event(
                    repo_full_name, issue_number, "reopened",
                    from_column="done", to_column=to_column,
                )
        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"

    # Mirror intermediate columns to Gitea labels (kanban:<column>) so
    # the workflow state is visible in Gitea itself, not just here.
    result["label_sync"] = "skipped"
    if settings.sync_labels and from_column != to_column:
        try:
            desired = (
                f"kanban:{to_column}"
                if to_column in ("todo", "in_progress", "review")
                else None
            )
            issue = await gitea.get_issue(owner, repo, issue_number)
            current = [
                label for label in issue.get("labels") or []
                if str(label.get("name", "")).startswith("kanban:")
            ]
            for label in current:
                if label.get("name") != desired:
                    await gitea.remove_issue_label(
                        owner, repo, issue_number, label["id"]
                    )
            if desired and all(l.get("name") != desired for l in current):
                label = await gitea.ensure_label(
                    owner, repo, desired, COLUMN_COLORS[to_column]
                )
                await gitea.add_issue_label(owner, repo, issue_number, label["id"])
            result["label_sync"] = "ok"
        except Exception as exc:
            logger.warning(
                "Label sync failed for %s#%s: %s", repo_full_name, issue_number, exc
            )
            result["label_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)
    log_event(repo_full_name, issue["number"], "created", to_column=column)
    return JSONResponse(
        {
            "ok": True,
            "issue": {
                "number": issue["number"],
                "title": issue.get("title", title),
                "html_url": issue.get("html_url", ""),
            },
        },
        status_code=201,
    )


def _issue_payload(issue: dict) -> dict:
    assignees = issue.get("assignees") or []
    if not assignees and issue.get("assignee"):
        assignees = [issue["assignee"]]
    milestone = issue.get("milestone") or {}
    return {
        "number": issue.get("number"),
        "title": issue.get("title", ""),
        "body": issue.get("body") or "",
        "state": issue.get("state", "open"),
        "html_url": issue.get("html_url", ""),
        "created_at": issue.get("created_at", ""),
        "updated_at": issue.get("updated_at", ""),
        "due_date": issue.get("due_date"),
        "labels": [
            {"id": l.get("id"), "name": l.get("name", ""), "color": l.get("color", "")}
            for l in issue.get("labels") or []
        ],
        "assignees": [
            {"login": a.get("login", ""), "avatar_url": a.get("avatar_url", "")}
            for a in assignees
        ],
        "milestone": milestone.get("title"),
    }


def _comment_payload(comment: dict) -> dict:
    user = comment.get("user") or {}
    return {
        "id": comment.get("id"),
        "body": comment.get("body") or "",
        "created_at": comment.get("created_at", ""),
        "updated_at": comment.get("updated_at", ""),
        "html_url": comment.get("html_url", ""),
        "user": {
            "login": user.get("login", "?"),
            "avatar_url": user.get("avatar_url", ""),
        },
    }


@app.get("/api/issues/{owner}/{repo}/{number}")
async def api_issue_detail(owner: str, repo: str, number: int):
    """Issue detail + comments for the card panel (proxied from Gitea)."""
    try:
        issue = await gitea.get_issue(owner, repo, number)
        comments = await gitea.get_issue_comments(owner, repo, number)
    except httpx.HTTPStatusError as exc:
        if exc.response.status_code == 404:
            return JSONResponse(
                {"ok": False, "error": "Issue not found"}, status_code=404
            )
        raise
    return {
        "ok": True,
        "issue": _issue_payload(issue),
        "comments": [_comment_payload(c) for c in comments],
    }


@app.post("/api/issues/{owner}/{repo}/{number}/comments")
async def api_issue_comment(
    request: Request,
    owner: str,
    repo: str,
    number: int,
    body: str = Form(...),
    csrf_token: str = Form(""),
):
    """Post a comment on the issue in Gitea."""
    if not _csrf_valid(request, csrf_token):
        return _csrf_error()
    body = body.strip()
    if not body:
        return JSONResponse(
            {"ok": False, "error": "Comment must not be empty"}, status_code=400
        )
    if len(body) > 10_000:
        return JSONResponse(
            {"ok": False, "error": "Comment too large (max 10000 chars)"},
            status_code=413,
        )
    comment = await gitea.create_comment(owner, repo, number, body)
    log_event(f"{owner}/{repo}", number, "commented")
    return JSONResponse(
        {"ok": True, "comment": _comment_payload(comment)}, status_code=201
    )


@app.get("/api/events")
async def api_events(repo: str = "", limit: int = 100):
    """Latest journal entries (all repos, or one via ?repo=owner/name)."""
    events = get_events(repo or None, max(1, min(limit, 500)))
    return {"ok": True, "events": events}


@app.get("/api/metrics")
async def api_metrics(repo: str = "", window: int = 30):
    """Throughput / cycle-time metrics from the journal."""
    return {"ok": True, "metrics": get_metrics(repo or None, window)}


# ── Label management ───────────────────────────────────────

@app.get("/api/repos/{owner}/{repo}/labels")
async def api_list_labels(owner: str, repo: str):
    """All label definitions of a repo."""
    labels = await gitea.get_labels(owner, repo)
    return {
        "ok": True,
        "labels": [
            {"id": l.get("id"), "name": l.get("name", ""), "color": l.get("color", "")}
            for l in labels
        ],
    }


@app.post("/api/repos/{owner}/{repo}/labels")
async def api_create_label(
    request: Request,
    owner: str,
    repo: str,
    name: str = Form(...),
    color: str = Form("7c5cfc"),
    csrf_token: str = Form(""),
):
    """Create a label definition in the repo."""
    if not _csrf_valid(request, csrf_token):
        return _csrf_error()
    name = name.strip()
    color = color.strip().lstrip("#")
    if not name or len(name) > 50:
        return JSONResponse(
            {"ok": False, "error": "Label name must be 1-50 chars"},
            status_code=400,
        )
    if len(color) != 6 or any(c not in "0123456789abcdefABCDEF" for c in color):
        return JSONResponse(
            {"ok": False, "error": "Color must be a 6-digit hex value"},
            status_code=400,
        )
    label = await gitea.create_label(owner, repo, name, color)
    gitea.forget_label_cache(owner, repo)
    return JSONResponse(
        {
            "ok": True,
            "label": {
                "id": label.get("id"),
                "name": label.get("name", name),
                "color": label.get("color", color),
            },
        },
        status_code=201,
    )


@app.delete("/api/repos/{owner}/{repo}/labels/{label_id}")
async def api_delete_label(
    request: Request, owner: str, repo: str, label_id: int
):
    """Delete a label definition (CSRF via the X-CSRF-Token header)."""
    if not _csrf_valid(request, ""):
        return _csrf_error()
    await gitea.delete_label(owner, repo, label_id)
    return {"ok": True, "deleted": label_id}


@app.post("/api/issues/{owner}/{repo}/{number}/labels")
async def api_toggle_issue_label(
    request: Request,
    owner: str,
    repo: str,
    number: int,
    label_id: int = Form(...),
    action: str = Form(...),
    csrf_token: str = Form(""),
):
    """Add or remove one label on one issue; returns the new label set."""
    if not _csrf_valid(request, csrf_token):
        return _csrf_error()
    if action not in ("add", "remove"):
        return JSONResponse(
            {"ok": False, "error": "action must be 'add' or 'remove'"},
            status_code=400,
        )
    if action == "add":
        await gitea.add_issue_label(owner, repo, number, label_id)
    else:
        await gitea.remove_issue_label(owner, repo, number, label_id)
    issue = await gitea.get_issue(owner, repo, number)
    return {
        "ok": True,
        "labels": [
            {"id": l.get("id"), "name": l.get("name", ""), "color": l.get("color", "")}
            for l in issue.get("labels") or []
        ],
    }


# ── MCP server (read-only tools over JSON-RPC) ─────────────

MCP_TOOLS = [
    {
        "name": "list_projects",
        "description": "List the projects (Gitea repos) Kanbante tracks, grouped by activity.",
        "inputSchema": {"type": "object", "properties": {}},
    },
    {
        "name": "get_board",
        "description": "Get the kanban board of one repo: columns with their cards.",
        "inputSchema": {
            "type": "object",
            "properties": {
                "owner": {"type": "string"},
                "repo": {"type": "string"},
            },
            "required": ["owner", "repo"],
        },
    },
    {
        "name": "get_issue",
        "description": "Get one issue with its body and comments.",
        "inputSchema": {
            "type": "object",
            "properties": {
                "owner": {"type": "string"},
                "repo": {"type": "string"},
                "number": {"type": "integer"},
            },
            "required": ["owner", "repo", "number"],
        },
    },
    {
        "name": "get_focus",
        "description": "Cards in progress or in review across all active projects.",
        "inputSchema": {"type": "object", "properties": {}},
    },
    {
        "name": "search_issues",
        "description": "Search issues by text across all repos.",
        "inputSchema": {
            "type": "object",
            "properties": {
                "query": {"type": "string"},
                "state": {"type": "string", "enum": ["open", "closed", "all"]},
            },
            "required": ["query"],
        },
    },
    {
        "name": "get_activity",
        "description": "Latest entries of the board event journal.",
        "inputSchema": {
            "type": "object",
            "properties": {"limit": {"type": "integer"}},
        },
    },
    {
        "name": "get_metrics",
        "description": "Throughput and cycle-time metrics from the journal.",
        "inputSchema": {
            "type": "object",
            "properties": {"repo": {"type": "string"}},
        },
    },
]


async def _mcp_call_tool(name: str, args: dict) -> str:
    """Execute one read-only MCP tool and return text content."""
    if name == "list_projects":
        repos = await gitea.search_repos()
        active, inactive, archived = classify_repos(repos)
        lines = [f"Active projects ({len(active)}):"]
        lines += [
            f"- {r['full_name']} ({r.get('open_issues_count') or 0} open issues)"
            for r in active
        ]
        lines.append(
            f"Inactive: {len(inactive)} - Archived: {len(archived)}"
        )
        return "\n".join(lines)
    if name == "get_board":
        owner, repo = args["owner"], args["repo"]
        items = await gitea.get_repo_issues(owner, repo, state="all")
        issues = [i for i in items if _is_issue(i)]
        open_issues = [i for i in issues if i.get("state") == "open"]
        columns = _merge_issues_with_board(open_issues, get_board_state(f"{owner}/{repo}"))
        return json.dumps(
            {
                column: [
                    {
                        "number": i["number"],
                        "title": i["title"],
                        "labels": [l.get("name") for l in i.get("labels") or []],
                    }
                    for i in cards
                ]
                for column, cards in columns.items()
            },
            ensure_ascii=False,
        )
    if name == "get_issue":
        detail = await api_issue_detail(args["owner"], args["repo"], int(args["number"]))
        return json.dumps(detail, ensure_ascii=False, default=str)
    if name == "get_focus":
        groups = await _collect_focus_groups()
        return json.dumps(
            [
                {
                    "repo": g["full_name"],
                    "cards": [
                        {
                            "number": c["issue"]["number"],
                            "title": c["issue"]["title"],
                            "column": c["column"],
                        }
                        for c in g["cards"]
                    ],
                }
                for g in groups
            ],
            ensure_ascii=False,
        )
    if name == "search_issues":
        results = await gitea.search_issues(
            q=args["query"], state=args.get("state", "open")
        )
        return json.dumps(
            [
                {
                    "repo": (i.get("repository") or {}).get("full_name"),
                    "number": i["number"],
                    "title": i["title"],
                    "state": i["state"],
                }
                for i in results
            ],
            ensure_ascii=False,
        )
    if name == "get_activity":
        return json.dumps(
            get_events(limit=int(args.get("limit", 50))), ensure_ascii=False
        )
    if name == "get_metrics":
        return json.dumps(get_metrics(args.get("repo") or None), ensure_ascii=False)
    raise ValueError(f"Unknown tool: {name}")


@app.get("/mcp")
async def mcp_info():
    """Discoverability for the MCP endpoint (the protocol is POST)."""
    return {
        "name": "kanbante",
        "version": APP_VERSION,
        "transport": "streamable-http (JSON-RPC 2.0 over POST)",
        "read_only": True,
        "tools": [t["name"] for t in MCP_TOOLS],
    }


@app.post("/mcp")
async def mcp_endpoint(request: Request):
    """Minimal MCP server: initialize / tools/list / tools/call / ping.

    Read-only by design (Bruno's rule: cross-platform reads first,
    writes go through approval flows elsewhere).
    """
    try:
        payload = await request.json()
    except ValueError:
        return JSONResponse(
            {"jsonrpc": "2.0", "id": None,
             "error": {"code": -32700, "message": "Parse error"}},
            status_code=400,
        )
    if isinstance(payload, list):  # batch: not supported by this server
        return JSONResponse(
            {"jsonrpc": "2.0", "id": None,
             "error": {"code": -32600, "message": "Batch requests not supported"}},
            status_code=400,
        )
    method = payload.get("method", "")
    rpc_id = payload.get("id")
    if rpc_id is None:  # notification: acknowledge silently
        return Response(status_code=202)
    params = payload.get("params") or {}

    def _result(result):
        return {"jsonrpc": "2.0", "id": rpc_id, "result": result}

    def _error(code, message):
        return {"jsonrpc": "2.0", "id": rpc_id,
                "error": {"code": code, "message": message}}

    if method == "initialize":
        return _result(
            {
                "protocolVersion": params.get("protocolVersion", "2024-11-05"),
                "capabilities": {"tools": {}},
                "serverInfo": {"name": "kanbante", "version": APP_VERSION},
            }
        )
    if method == "ping":
        return _result({})
    if method == "tools/list":
        return _result({"tools": MCP_TOOLS})
    if method == "tools/call":
        tool_name = params.get("name", "")
        arguments = params.get("arguments") or {}
        try:
            text = await _mcp_call_tool(tool_name, arguments)
        except KeyError as exc:
            return _result(
                {"content": [{"type": "text", "text": f"Missing argument: {exc}"}],
                 "isError": True}
            )
        except Exception as exc:
            return _result(
                {"content": [{"type": "text", "text": f"Tool error: {exc}"}],
                 "isError": True}
            )
        return _result({"content": [{"type": "text", "text": text}], "isError": False})
    return _error(-32601, f"Method not found: {method}")


@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,
    )