Merge branch 'develop' of github.com:codervisor/devlog into develop

Author: tikazyq

apps/web/app/api/chat-sessions/[sessionId]/events/route.ts

@@ -0,0 +1,70 @@
+/**
+ * Chat Session Events API Endpoint
+ * 
+ * GET /api/chat-sessions/[sessionId]/events - Get session events
+ */
+
+import { NextRequest, NextResponse } from 'next/server';
+import { getPrismaClient } from '@codervisor/devlog-core/server';
+
+// Mark route as dynamic
+export const dynamic = 'force-dynamic';
+
+/**
+ * GET /api/chat-sessions/:sessionId/events - Get events for a chat session
+ * 
+ * Returns all agent events associated with the specified chat session,
+ * ordered chronologically.
+ */
+export async function GET(
+  request: NextRequest,
+  { params }: { params: { sessionId: string } }
+) {
+  try {
+    const { sessionId } = params;
+
+    // Validate UUID format
+    const uuidRegex = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i;
+    if (!uuidRegex.test(sessionId)) {
+      return NextResponse.json(
+        { error: 'Invalid session ID format' },
+        { status: 400 }
+      );
+    }
+
+    // Get Prisma client
+    const prisma = getPrismaClient();
+
+    // Fetch events for the session
+    const events = await prisma.agentEvent.findMany({
+      where: { sessionId },
+      orderBy: { timestamp: 'asc' },
+      include: {
+        session: {
+          include: {
+            workspace: {
+              include: {
+                machine: true,
+                project: true,
+              },
+            },
+          },
+        },
+      },
+    });
+
+    return NextResponse.json({
+      sessionId,
+      events,
+      count: events.length,
+    });
+  } catch (error) {
+    console.error('[GET /api/chat-sessions/:sessionId/events] Error:', error);
+    return NextResponse.json(
+      {
+        error: error instanceof Error ? error.message : 'Failed to get session events',
+      },
+      { status: 500 }
+    );
+  }
+}

apps/web/app/api/chat-sessions/route.ts

@@ -0,0 +1,68 @@
+/**
+ * Chat Session API Endpoint
+ * 
+ * POST /api/chat-sessions - Create/update chat session
+ */
+
+import { NextRequest, NextResponse } from 'next/server';
+import { getPrismaClient } from '@codervisor/devlog-core/server';
+import { ChatSessionCreateSchema } from '@/schemas/hierarchy';
+import { ApiValidator } from '@/schemas/validation';
+
+// Mark route as dynamic
+export const dynamic = 'force-dynamic';
+
+/**
+ * POST /api/chat-sessions - Create/update chat session
+ * 
+ * Creates a new chat session or updates an existing one based on sessionId.
+ * Supports updating message count, token count, and end time.
+ */
+export async function POST(request: NextRequest) {
+  try {
+    // Validate request body
+    const validation = await ApiValidator.validateJsonBody(
+      request,
+      ChatSessionCreateSchema
+    );
+
+    if (!validation.success) {
+      return validation.response;
+    }
+
+    const data = validation.data;
+
+    // Get Prisma client
+    const prisma = getPrismaClient();
+
+    // Upsert chat session
+    const session = await prisma.chatSession.upsert({
+      where: { sessionId: data.sessionId },
+      create: {
+        sessionId: data.sessionId,
+        workspaceId: data.workspaceId,
+        agentType: data.agentType,
+        modelId: data.modelId,
+        startedAt: data.startedAt,
+        endedAt: data.endedAt,
+        messageCount: data.messageCount,
+        totalTokens: data.totalTokens,
+      },
+      update: {
+        endedAt: data.endedAt,
+        messageCount: data.messageCount,
+        totalTokens: data.totalTokens,
+      },
+    });
+
+    return NextResponse.json(session, { status: 200 });
+  } catch (error) {
+    console.error('[POST /api/chat-sessions] Error:', error);
+    return NextResponse.json(
+      {
+        error: error instanceof Error ? error.message : 'Failed to upsert session',
+      },
+      { status: 500 }
+    );
+  }
+}

apps/web/app/api/events/batch/route.ts

@@ -0,0 +1,95 @@
+/**
+ * Batch Event Creation API Endpoint
+ * 
+ * POST /api/events/batch - Batch create agent events
+ */
+
+import { NextRequest, NextResponse } from 'next/server';
+import { getPrismaClient } from '@codervisor/devlog-core/server';
+import { BatchEventsCreateSchema } from '@/schemas/hierarchy';
+import { ApiValidator } from '@/schemas/validation';
+
+// Mark route as dynamic
+export const dynamic = 'force-dynamic';
+
+/**
+ * POST /api/events/batch - Batch create events
+ * 
+ * Creates multiple agent events in a single transaction.
+ * Maximum 1000 events per request for performance.
+ */
+export async function POST(request: NextRequest) {
+  try {
+    // Validate request body
+    const validation = await ApiValidator.validateJsonBody(
+      request,
+      BatchEventsCreateSchema
+    );
+
+    if (!validation.success) {
+      return validation.response;
+    }
+
+    const events = validation.data;
+
+    if (events.length === 0) {
+      return NextResponse.json(
+        { error: 'At least one event is required' },
+        { status: 400 }
+      );
+    }
+
+    // Get Prisma client
+    const prisma = getPrismaClient();
+
+    // Use createMany for better performance
+    const result = await prisma.agentEvent.createMany({
+      data: events.map((event) => ({
+        timestamp: event.timestamp,
+        eventType: event.eventType,
+        agentId: event.agentId,
+        agentVersion: event.agentVersion,
+        sessionId: event.sessionId,
+        projectId: event.projectId,
+        context: event.context,
+        data: event.data,
+        metrics: event.metrics,
+        parentEventId: event.parentEventId,
+        relatedEventIds: event.relatedEventIds,
+        tags: event.tags,
+        severity: event.severity,
+      })),
+      skipDuplicates: true, // Skip events with duplicate IDs
+    });
+
+    return NextResponse.json(
+      {
+        created: result.count,
+        requested: events.length,
+      },
+      { status: 201 }
+    );
+  } catch (error) {
+    console.error('[POST /api/events/batch] Error:', error);
+
+    // Handle specific Prisma errors
+    if (error instanceof Error) {
+      if (error.message.includes('Foreign key constraint')) {
+        return NextResponse.json(
+          {
+            error: 'Invalid reference: session or project not found',
+            details: error.message,
+          },
+          { status: 400 }
+        );
+      }
+    }
+
+    return NextResponse.json(
+      {
+        error: error instanceof Error ? error.message : 'Failed to create events',
+      },
+      { status: 500 }
+    );
+  }
+}

apps/web/app/api/events/stream/route.ts

@@ -2,56 +2,142 @@
  * Server-Sent Events (SSE) endpoint for real-time updates
  * 
  * Provides a persistent connection that streams updates about:
- * - New agent sessions
- * - Session status changes
  * - New agent events
+ * - Session status changes
  * - Dashboard metrics updates
+ * 
+ * Supports hierarchy-based filtering:
+ * - projectId: Filter events by project
+ * - machineId: Filter events by machine
+ * - workspaceId: Filter events by workspace
  */
 
 import { NextRequest } from 'next/server';
-import { EventBroadcaster } from '@/lib/realtime/event-broadcaster';
+import { getPrismaClient } from '@codervisor/devlog-core/server';
 
 // Mark this route as dynamic to prevent static generation
 export const dynamic = 'force-dynamic';
 export const runtime = 'nodejs';
 
 // Keep-alive interval in milliseconds
 const KEEP_ALIVE_INTERVAL = 30000; // 30 seconds
+// Polling interval for new events
+const POLL_INTERVAL = 5000; // 5 seconds
 
 export async function GET(request: NextRequest) {
-  const broadcaster = EventBroadcaster.getInstance();
+  // Parse filter parameters
+  const { searchParams } = new URL(request.url);
+  const projectId = searchParams.get('projectId');
+  const machineId = searchParams.get('machineId');
+  const workspaceId = searchParams.get('workspaceId');
+
+  // Build filter for events
+  const filters = {
+    projectId: projectId ? parseInt(projectId, 10) : undefined,
+    machineId: machineId ? parseInt(machineId, 10) : undefined,
+    workspaceId: workspaceId ? parseInt(workspaceId, 10) : undefined,
+  };
 
   // Create a readable stream for SSE
   const stream = new ReadableStream({
     start(controller) {
       const encoder = new TextEncoder();
+      let lastTimestamp = new Date();
 
       // Send initial connection message
       const connectionMessage = `event: connected\ndata: ${JSON.stringify({ 
         timestamp: new Date().toISOString(),
-        clientCount: broadcaster.getClientCount() + 1 
+        filters,
       })}\n\n`;
       controller.enqueue(encoder.encode(connectionMessage));
 
-      // Add this client to the broadcaster
-      broadcaster.addClient(controller);
-
       // Set up keep-alive heartbeat
       const keepAliveInterval = setInterval(() => {
         try {
           const heartbeat = `: heartbeat ${Date.now()}\n\n`;
           controller.enqueue(encoder.encode(heartbeat));
         } catch (error) {
-          console.error('Error sending heartbeat:', error);
+          console.error('[SSE] Error sending heartbeat:', error);
           clearInterval(keepAliveInterval);
-          broadcaster.removeClient(controller);
+          clearInterval(pollInterval);
         }
       }, KEEP_ALIVE_INTERVAL);
 
+      // Poll for new events
+      const pollInterval = setInterval(async () => {
+        try {
+          const prisma = getPrismaClient();
+          
+          // Build where clause based on filters
+          const where: any = {
+            timestamp: {
+              gte: lastTimestamp,
+            },
+          };
+
+          if (filters.projectId) {
+            where.projectId = filters.projectId;
+          }
+
+          if (filters.machineId) {
+            where.session = {
+              workspace: {
+                machineId: filters.machineId,
+              },
+            };
+          }
+
+          if (filters.workspaceId) {
+            where.session = {
+              ...where.session,
+              workspaceId: filters.workspaceId,
+            };
+          }
+
+          // Fetch new events
+          const events = await prisma.agentEvent.findMany({
+            where,
+            orderBy: { timestamp: 'desc' },
+            take: 50,
+            include: {
+              session: {
+                include: {
+                  workspace: {
+                    include: {
+                      machine: true,
+                      project: true,
+                    },
+                  },
+                },
+              },
+            },
+          });
+
+          if (events.length > 0) {
+            // Update last timestamp
+            lastTimestamp = new Date(events[0].timestamp);
+
+            // Send events to client
+            const message = `event: events\ndata: ${JSON.stringify({
+              type: 'events',
+              data: events,
+            })}\n\n`;
+            controller.enqueue(encoder.encode(message));
+          }
+        } catch (error) {
+          console.error('[SSE] Error polling events:', error);
+          const errorMessage = `event: error\ndata: ${JSON.stringify({
+            type: 'error',
+            error: error instanceof Error ? error.message : 'Unknown error',
+          })}\n\n`;
+          controller.enqueue(encoder.encode(errorMessage));
+        }
+      }, POLL_INTERVAL);
+
       // Clean up when client disconnects
       request.signal.addEventListener('abort', () => {
         clearInterval(keepAliveInterval);
-        broadcaster.removeClient(controller);
+        clearInterval(pollInterval);
         try {
           controller.close();
         } catch (error) {