from __future__ import annotations import base64 import json from uuid import UUID from fastapi import APIRouter, Depends, HTTPException, Query from sqlalchemy.orm import Session, joinedload from app.core.dependencies import get_current_user from app.database.session import get_db from app.models.commit import Commit from app.models.document import Document from app.models.document_version import DocumentVersion from app.models.knowledge_item import KnowledgeItem from app.models.proposal import Proposal, ProposalStatus from app.models.review import Review from app.models.user import User from app.models.workflow_run import WorkflowRun, WorkflowStatus from app.models.workspace import Workspace router = APIRouter( prefix="/activity", tags=["Activity"], ) # ------------------------------------------------------------------ # Cursor helpers # ------------------------------------------------------------------ def _encode_cursor(timestamp: str, event_id: str) -> str: """Encode a compound cursor as a URL-safe base64 string.""" payload = json.dumps({"t": timestamp, "id": event_id}) return base64.urlsafe_b64encode(payload.encode()).decode() def _decode_cursor(cursor: str) -> tuple[str, str]: """Decode a compound cursor. Returns (timestamp, event_id).""" try: payload = json.loads(base64.urlsafe_b64decode(cursor.encode())) return payload["t"], payload["id"] except (json.JSONDecodeError, KeyError, Exception): raise HTTPException( status_code=400, detail="Invalid cursor format.", ) # ------------------------------------------------------------------ # Activity endpoint # ------------------------------------------------------------------ @router.get("") def list_activity( workspace_id: UUID, limit: int = 50, cursor: str | None = None, current_user: User = Depends(get_current_user), db: Session = Depends(get_db), ): """ Aggregated activity feed for a workspace with compound cursor pagination. Events are synthesized from WorkflowRun, DocumentVersion, Proposal, and Commit tables. Ordered by (timestamp DESC, event_id DESC). Pagination uses an opaque cursor encoding (timestamp, event_id). """ workspace = ( db.query(Workspace) .filter( Workspace.id == workspace_id, Workspace.created_by == current_user.id, ) .first() ) if workspace is None: raise HTTPException( status_code=403, detail="You do not have access to this workspace.", ) # Decode cursor if provided cursor_ts: str | None = None cursor_id: str | None = None if cursor: cursor_ts, cursor_id = _decode_cursor(cursor) # Fetch enough rows per table to satisfy the merged limit without overfetching fetch_limit = min(max(limit * 2, 20), 100) events: list[dict] = [] # ------------------------------------------------------------------ # Workflow events (eagerly load document_version in 1 query) # ------------------------------------------------------------------ workflows = ( db.query(WorkflowRun) .options(joinedload(WorkflowRun.document_version)) .filter(WorkflowRun.workspace_id == workspace_id) .order_by(WorkflowRun.started_at.desc()) .limit(fetch_limit) .all() ) for w in workflows: dv = w.document_version filename = dv.filename if dv else "Unknown" events.append({ "id": f"wf-start-{w.id}", "type": "workflow_started", "message": f"Workflow started for {filename}", "timestamp": w.started_at.isoformat() if w.started_at else None, "metadata": { "workflow_id": str(w.id), "document_id": str(dv.document_id) if dv else None, "filename": filename, "status": w.status.value, }, }) if w.status == WorkflowStatus.COMPLETED and w.completed_at: events.append({ "id": f"wf-complete-{w.id}", "type": "workflow_completed", "message": f"Workflow completed for {filename}", "timestamp": w.completed_at.isoformat(), "metadata": { "workflow_id": str(w.id), "filename": filename, }, }) if w.status == WorkflowStatus.WAITING_FOR_REVIEW: events.append({ "id": f"wf-review-{w.id}", "type": "review_requested", "message": f"Human review requested for {filename}", "timestamp": w.started_at.isoformat() if w.started_at else None, "metadata": { "workflow_id": str(w.id), "filename": filename, }, }) # ------------------------------------------------------------------ # Document upload events (direct join on documents) # ------------------------------------------------------------------ versions = ( db.query(DocumentVersion) .join(Document, DocumentVersion.document_id == Document.id) .filter(Document.workspace_id == workspace_id) .order_by(DocumentVersion.uploaded_at.desc()) .limit(fetch_limit) .all() ) for v in versions: events.append({ "id": f"doc-upload-{v.id}", "type": "document_uploaded", "message": f"Document uploaded: {v.filename}", "timestamp": v.uploaded_at.isoformat() if v.uploaded_at else None, "metadata": { "document_id": str(v.document_id), "version_id": str(v.id), "filename": v.filename, }, }) # ------------------------------------------------------------------ # Proposal events (eagerly load knowledge_item -> document_version) # ------------------------------------------------------------------ proposals = ( db.query(Proposal) .options( joinedload(Proposal.knowledge_item).joinedload(KnowledgeItem.document_version) ) .filter(Proposal.workspace_id == workspace_id) .order_by(Proposal.created_at.desc()) .limit(fetch_limit) .all() ) for p in proposals: p_filename = None ki = p.knowledge_item if ki and ki.document_version: p_filename = ki.document_version.filename events.append({ "id": f"proposal-{p.id}", "type": "proposal_created", "message": f"Proposal created: {p.summary[:80]}", "timestamp": p.created_at.isoformat() if p.created_at else None, "metadata": { "proposal_id": str(p.id), "proposal_type": p.proposal_type.value, "status": p.status.value, "filename": p_filename, }, }) if p.status == ProposalStatus.APPROVED and p.reviewed_at: events.append({ "id": f"proposal-approved-{p.id}", "type": "proposal_approved", "message": f"Proposal approved: {p.summary[:80]}", "timestamp": p.reviewed_at.isoformat(), "metadata": { "proposal_id": str(p.id), "proposal_type": p.proposal_type.value, "filename": p_filename, }, }) if p.status == ProposalStatus.REJECTED and p.reviewed_at: events.append({ "id": f"proposal-rejected-{p.id}", "type": "proposal_rejected", "message": f"Proposal rejected: {p.summary[:80]}", "timestamp": p.reviewed_at.isoformat(), "metadata": { "proposal_id": str(p.id), "proposal_type": p.proposal_type.value, "filename": p_filename, }, }) if p.status == ProposalStatus.ARCHIVED and p.reviewed_at: events.append({ "id": f"proposal-archived-{p.id}", "type": "proposal_archived", "message": f"Proposal archived: {p.summary[:80]}", "timestamp": p.reviewed_at.isoformat(), "metadata": { "proposal_id": str(p.id), "proposal_type": p.proposal_type.value, "filename": p_filename, }, }) # ------------------------------------------------------------------ # Commit events (eagerly load proposal -> knowledge_item -> document_version) # ------------------------------------------------------------------ commits = ( db.query(Commit) .options( joinedload(Commit.proposal) .joinedload(Proposal.knowledge_item) .joinedload(KnowledgeItem.document_version) ) .filter(Commit.workspace_id == workspace_id) .order_by(Commit.committed_at.desc()) .limit(fetch_limit) .all() ) for c in commits: c_filename = None if c.proposal and c.proposal.knowledge_item and c.proposal.knowledge_item.document_version: c_filename = c.proposal.knowledge_item.document_version.filename events.append({ "id": f"commit-{c.id}", "type": "commit_created", "message": f"Knowledge committed: {c.message[:80]}", "timestamp": c.committed_at.isoformat() if c.committed_at else None, "metadata": { "commit_id": str(c.id), "proposal_id": str(c.proposal_id) if c.proposal_id else None, "filename": c_filename, }, }) # ------------------------------------------------------------------ # Global sort: (timestamp DESC, event_id DESC) # ------------------------------------------------------------------ events.sort( key=lambda e: (e["timestamp"] or "", e["id"]), reverse=True, ) # ------------------------------------------------------------------ # Apply cursor filter (compound: timestamp + event_id) # ------------------------------------------------------------------ if cursor_ts and cursor_id: events = [ e for e in events if (e["timestamp"] or "") < cursor_ts or ( (e["timestamp"] or "") == cursor_ts and e["id"] < cursor_id ) ] # ------------------------------------------------------------------ # Paginate # ------------------------------------------------------------------ has_more = len(events) > limit result_events = events[:limit] next_cursor = None if has_more and result_events: last = result_events[-1] next_cursor = _encode_cursor(last["timestamp"] or "", last["id"]) return { "events": result_events, "has_more": has_more, "next_cursor": next_cursor, }