claude-code-monitor/backend/app/crud.py

"""CRUD operations for database access."""

from sqlalchemy.orm import Session
from sqlalchemy import desc, func, or_
from typing import List, Optional, Tuple
import time
import json

from app.database import Event, Session as SessionModel, Statistics, ToolUsage
from app.schemas import EventCreate


def create_event(db: Session, event: EventCreate) -> Event:
    """
    Create a new event and update statistics.

    Args:
        db: Database session
        event: Event data to create

    Returns:
        Created event object
    """
    # Convert tool_input to JSON string if it's a dict
    tool_input_str = None
    if event.tool_input is not None:
        if isinstance(event.tool_input, str):
            tool_input_str = event.tool_input
        else:
            tool_input_str = json.dumps(event.tool_input)

    # Generate description if not provided
    description = event.description
    if not description:
        description = _generate_description(event)

    # Create event
    db_event = Event(
        session_id=event.session_id,
        event_type=event.event_type,
        tool_name=event.tool_name,
        tool_input=tool_input_str,
        tool_output=event.tool_output,
        success=event.success,
        description=description,
        timestamp=event.timestamp,
    )
    db.add(db_event)

    # Update or create session
    session = db.query(SessionModel).filter(SessionModel.session_id == event.session_id).first()
    if not session:
        session = SessionModel(
            session_id=event.session_id,
            started_at=event.timestamp,
            event_count=1,
        )
        db.add(session)
    else:
        session.event_count += 1
        if event.event_type == "SessionEnd":
            session.ended_at = event.timestamp

    # Update tool usage for tool events
    if event.tool_name:
        tool_usage = db.query(ToolUsage).filter(ToolUsage.tool_name == event.tool_name).first()
        if not tool_usage:
            tool_usage = ToolUsage(
                tool_name=event.tool_name,
                usage_count=1,
                first_used=event.timestamp,
                last_used=event.timestamp,
            )
            db.add(tool_usage)
        else:
            tool_usage.usage_count += 1
            tool_usage.last_used = event.timestamp

    # Update statistics
    _update_statistics(db)

    db.commit()
    db.refresh(db_event)
    return db_event


def get_events(
    db: Session,
    skip: int = 0,
    limit: int = 50,
    event_type: Optional[str] = None,
    session_id: Optional[str] = None,
    tool_name: Optional[str] = None,
) -> Tuple[List[Event], int]:
    """
    Get paginated list of events with optional filtering.

    Args:
        db: Database session
        skip: Number of records to skip (offset)
        limit: Maximum number of records to return
        event_type: Filter by event type
        session_id: Filter by session ID
        tool_name: Filter by tool name

    Returns:
        Tuple of (events list, total count)
    """
    query = db.query(Event)

    # Apply filters
    if event_type:
        # Support filtering by multiple event types (comma-separated)
        if "," in event_type:
            event_types = [et.strip() for et in event_type.split(",")]
            query = query.filter(Event.event_type.in_(event_types))
        else:
            query = query.filter(Event.event_type == event_type)

    if session_id:
        query = query.filter(Event.session_id == session_id)

    if tool_name:
        query = query.filter(Event.tool_name == tool_name)

    # Get total count
    total = query.count()

    # Order by timestamp descending (newest first) and apply pagination
    events = query.order_by(desc(Event.timestamp)).offset(skip).limit(limit).all()

    return events, total


def get_event_by_id(db: Session, event_id: int) -> Optional[Event]:
    """Get a single event by ID."""
    return db.query(Event).filter(Event.id == event_id).first()


def delete_all_events(db: Session) -> None:
    """Delete all events and reset statistics (for Reset Stats functionality)."""
    # Delete all events
    db.query(Event).delete()

    # Delete all sessions
    db.query(SessionModel).delete()

    # Delete all tool usage
    db.query(ToolUsage).delete()

    # Reset statistics
    stats = db.query(Statistics).first()
    if stats:
        stats.total_events = 0
        stats.total_tools_used = 0
        stats.total_agents = 0
        stats.total_sessions = 0
        stats.last_updated = time.time()

    db.commit()


def get_statistics(db: Session) -> Statistics:
    """Get current statistics."""
    stats = db.query(Statistics).first()
    if not stats:
        # Initialize if not exists
        stats = Statistics(
            id=1,
            total_events=0,
            total_tools_used=0,
            total_agents=0,
            total_sessions=0,
            last_updated=time.time(),
        )
        db.add(stats)
        db.commit()
        db.refresh(stats)
    return stats


def get_tool_usage(db: Session) -> List[ToolUsage]:
    """Get tool usage statistics."""
    return db.query(ToolUsage).order_by(desc(ToolUsage.usage_count)).all()


def get_sessions(db: Session, limit: int = 100) -> List[SessionModel]:
    """Get list of sessions."""
    return db.query(SessionModel).order_by(desc(SessionModel.started_at)).limit(limit).all()


# Helper functions
def _update_statistics(db: Session) -> None:
    """Update cached statistics."""
    stats = db.query(Statistics).first()
    if not stats:
        stats = Statistics(id=1)
        db.add(stats)

    # Count events
    stats.total_events = db.query(Event).count()

    # Count unique tools
    stats.total_tools_used = db.query(ToolUsage).count()

    # Count agents (SubagentStop events)
    stats.total_agents = db.query(Event).filter(Event.event_type == "SubagentStop").count()

    # Count sessions
    stats.total_sessions = db.query(SessionModel).count()

    # Update timestamp
    stats.last_updated = time.time()


def _generate_description(event: EventCreate) -> str:
    """Generate a human-readable description for an event."""
    if event.description:
        return event.description

    # Generate description based on event type
    if event.event_type == "PreToolUse":
        if event.tool_name:
            return f"Preparing to use {event.tool_name}"
        return "Preparing to use tool"

    elif event.event_type == "PostToolUse":
        if event.tool_name:
            status = "successfully" if event.success else "with errors"
            return f"Executed {event.tool_name} {status}"
        return "Tool execution completed"

    elif event.event_type == "SessionStart":
        return "Session started"

    elif event.event_type == "SessionEnd":
        return "Session ended"

    elif event.event_type == "SubagentStop":
        return "Agent completed task"

    elif event.event_type == "UserPromptSubmit":
        return "User submitted prompt"

    else:
        return f"{event.event_type} event"