leocrm/app/routes/errors.py

"""Error logging endpoint — accepts frontend errors and logs them.

No auth required so errors can be logged even during logout.
Rate-limited to 10 requests per minute per IP (simple in-memory implementation).
"""

from __future__ import annotations

import time
import logging
from collections import defaultdict, deque
from typing import Any

from fastapi import APIRouter, Request, Response, status
from pydantic import BaseModel, Field

logger = logging.getLogger(__name__)

router = APIRouter(prefix="/api/v1/errors", tags=["errors"])

# -- Simple in-memory rate limiter (10 req/min per IP) --
RATE_LIMIT = 10  # max requests
RATE_WINDOW = 60  # seconds
_ip_requests: dict[str, deque[float]] = defaultdict(deque)


def _is_rate_limited(client_ip: str) -> bool:
    """Return True if the IP has exceeded the rate limit."""
    now = time.monotonic()
    dq = _ip_requests[client_ip]

    # Remove timestamps outside the window
    while dq and now - dq[0] > RATE_WINDOW:
        dq.popleft()

    if len(dq) >= RATE_LIMIT:
        return True

    dq.append(now)
    return False


# -- Request schema --

class ErrorReport(BaseModel):
    timestamp: str | None = None
    message: str = Field(..., max_length=2000)
    stack: str | None = Field(None, max_length=10000)
    context: dict[str, Any] | None = None
    url: str | None = Field(None, max_length=500)
    userAgent: str | None = Field(None, max_length=500)


@router.post("", status_code=status.HTTP_204_NO_CONTENT)
async def report_error(error: ErrorReport, request: Request) -> Response:
    """Log a frontend error. No auth required. Rate-limited per IP."""
    client_ip = request.client.host if request.client else "unknown"

    if _is_rate_limited(client_ip):
        return Response(status_code=status.HTTP_429_TOO_MANY_REQUESTS)

    # Log with structured info
    logger.error(
        "Frontend error reported: %s",
        error.message,
        extra={
            "error_timestamp": error.timestamp,
            "error_message": error.message,
            "error_stack": error.stack,
            "error_context": error.context,
            "error_url": error.url,
            "error_user_agent": error.userAgent,
            "client_ip": client_ip,
        },
    )

    # If forgejo_error_reporter plugin is active, forward error
    try:
        from app.plugins.builtins.forgejo_error_reporter.service import report_error_to_forgejo
        entry = {
            "message": error.message,
            "stack": error.stack,
            "url": error.url,
            "userAgent": error.userAgent,
            "timestamp": error.timestamp,
            "context": error.context,
        }
        await report_error_to_forgejo(entry)
    except Exception:
        pass  # Plugin not active or error in reporting

    return Response(status_code=status.HTTP_204_NO_CONTENT)