impactflow_discovery/app/routers/activity.py

"""Activity feed + admin endpoints."""
from collections import Counter
from datetime import datetime, timedelta, timezone

from fastapi import APIRouter, Depends, Query
from pydantic import BaseModel
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.auth import get_current_user, require_admin
from app.database import get_db
from app.models import ActivityLog, User

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


class ActivityOut(BaseModel):
    id: str
    user_id: str
    action: str
    resource: str
    resource_id: str | None
    source: str
    created_at: datetime
    ip_address: str | None
    user_agent: str | None


class ActivitySummary(BaseModel):
    days: int
    total: int
    actions_per_day: dict[str, int]
    top_resources: list[tuple[str, int]]
    web_count: int
    mcp_count: int


def _to_out(row: ActivityLog) -> ActivityOut:
    return ActivityOut(
        id=row.id,
        user_id=row.user_id,
        action=row.action,
        resource=row.resource,
        resource_id=row.resource_id,
        source=row.source,
        created_at=row.created_at,
        ip_address=row.ip_address,
        user_agent=row.user_agent,
    )


@router.get("/activity", response_model=list[ActivityOut])
async def list_my_activity(
    page: int = Query(1, ge=1),
    limit: int = Query(20, ge=1, le=200),
    user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    stmt = (
        select(ActivityLog)
        .where(ActivityLog.user_id == user.id)
        .order_by(ActivityLog.created_at.desc())
        .offset((page - 1) * limit)
        .limit(limit)
    )
    rows = (await db.execute(stmt)).scalars().all()
    return [_to_out(r) for r in rows]


@router.get("/activity/summary", response_model=ActivitySummary)
async def my_activity_summary(
    days: int = Query(30, ge=1, le=365),
    user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    since = datetime.now(timezone.utc) - timedelta(days=days)
    stmt = (
        select(ActivityLog)
        .where(ActivityLog.user_id == user.id)
        .where(ActivityLog.created_at >= since)
    )
    rows = (await db.execute(stmt)).scalars().all()

    per_day: Counter[str] = Counter()
    resource_counts: Counter[str] = Counter()
    web = mcp = 0
    for r in rows:
        per_day[r.created_at.date().isoformat()] += 1
        resource_counts[r.resource] += 1
        if r.source == "mcp":
            mcp += 1
        else:
            web += 1

    return ActivitySummary(
        days=days,
        total=len(rows),
        actions_per_day=dict(per_day),
        top_resources=resource_counts.most_common(5),
        web_count=web,
        mcp_count=mcp,
    )


@router.get("/admin/activity", response_model=list[ActivityOut])
async def admin_list_activity(
    user_id: str | None = Query(None),
    page: int = Query(1, ge=1),
    limit: int = Query(50, ge=1, le=500),
    _admin: User = Depends(require_admin),
    db: AsyncSession = Depends(get_db),
):
    stmt = select(ActivityLog).order_by(ActivityLog.created_at.desc())
    if user_id:
        stmt = stmt.where(ActivityLog.user_id == user_id)
    stmt = stmt.offset((page - 1) * limit).limit(limit)
    rows = (await db.execute(stmt)).scalars().all()
    return [_to_out(r) for r in rows]


async def prune_old_activity(db: AsyncSession, days: int = 90) -> int:
    """Delete rows older than N days. Returns rows deleted."""
    cutoff = datetime.now(timezone.utc) - timedelta(days=days)
    stmt = select(func.count()).select_from(ActivityLog).where(
        ActivityLog.created_at < cutoff
    )
    count = (await db.execute(stmt)).scalar_one()
    if count:
        from sqlalchemy import delete

        await db.execute(delete(ActivityLog).where(
            ActivityLog.created_at < cutoff
        ))
        await db.commit()
    return int(count or 0)