impactflow_discovery/app/tracking.py

"""Activity-log helpers + middleware.

Middleware fires after every request and writes one ActivityLog row when
the request was authenticated and succeeded (status < 400). We classify the
verb (GET -> view / POST -> create / etc.) and bucket the URL path into a
coarse resource name so the dashboard can group "everything about
conversations" without parsing URLs at query time.
"""
import json
import uuid
from datetime import datetime, timezone
from typing import Optional

from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware

from app.database import AsyncSessionLocal
from app.models import ActivityLog

_METHOD_ACTION = {
    "GET": "view",
    "POST": "create",
    "PUT": "update",
    "PATCH": "update",
    "DELETE": "delete",
}

# Paths whose activity is not worth recording (noisy / no user signal).
_SKIP_PATH_PREFIXES = (
    "/health",
    "/static",
    "/favicon.ico",
    "/openapi.json",
    "/docs",
    "/redoc",
    "/api/auth/login",
    "/api/auth/callback",
)


def _method_to_action(method: str) -> str:
    return _METHOD_ACTION.get(method.upper(), "other")


def _path_to_resource(path: str) -> str:
    parts = [p for p in path.strip("/").split("/") if p]
    if not parts:
        return "root"
    if parts[0] == "api" and len(parts) > 1:
        return parts[1]
    return parts[0]


async def log_activity(
    *,
    user_id: str,
    action: str,
    resource: str,
    resource_id: Optional[str] = None,
    metadata: Optional[dict] = None,
    source: str = "web",
    ip_address: Optional[str] = None,
    user_agent: Optional[str] = None,
) -> None:
    """Write one row. Uses its own session because the request's session is
    already closed by the time tracking middleware sees the response."""
    row = ActivityLog(
        id=str(uuid.uuid4()),
        user_id=user_id,
        action=action,
        resource=resource,
        resource_id=resource_id,
        metadata_json=json.dumps(metadata) if metadata else None,
        source=source,
        ip_address=ip_address,
        user_agent=(user_agent or "")[:1000] or None,
        created_at=datetime.now(timezone.utc),
    )
    async with AsyncSessionLocal() as db:
        db.add(row)
        await db.commit()


class ActivityTrackingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        response = await call_next(request)

        # Only log authenticated, successful, non-noisy requests.
        if response.status_code >= 400:
            return response
        path = request.url.path
        if any(path.startswith(p) for p in _SKIP_PATH_PREFIXES):
            return response
        user = getattr(request.state, "user", None)
        if user is None:
            return response

        source = "mcp" if request.headers.get("x-api-key") else "web"
        try:
            await log_activity(
                user_id=user.id,
                action=_method_to_action(request.method),
                resource=_path_to_resource(path),
                source=source,
                ip_address=request.client.host if request.client else None,
                user_agent=request.headers.get("user-agent"),
            )
        except Exception:
            # Tracking failures must never break the underlying request.
            pass
        return response