All files / api/src/middlewares sse-events.ts

42.42% Statements 42/99
31.81% Branches 7/22
100% Functions 2/2
42.42% Lines 42/99

Press n or j to go to the next uncovered block, b, p or k for the previous block.

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152  1x                 1x 25x 25x 25x   25x     25x 25x 25x     25x 25x 2x 2x 25x 2x 2x                                                                     2x 25x         2x 2x 2x 2x   2x     2x             2x             2x 2x 2x 2x             2x             2x 2x 2x 2x             2x         2x         2x         2x         2x         2x         2x 2x  
import type { AppEnv } from '@api/types/hono';
import { emitSSEEvent } from '@api/utils/sse-event-manager';
import type { SSEEventType } from '@shared/types/sse';
import type { MiddlewareHandler } from 'hono';
 
/**
 * SSE Events Middleware
 * Automatically emits SSE events when mutations occur
 * Maps HTTP methods and routes to SSE event types
 */
export const sseEventsMiddleware: MiddlewareHandler<AppEnv> = async (
  c,
  next,
) => {
  // Execute the route handler
  await next();
 
  // Only emit events for successful mutations (POST, PATCH, DELETE)
  const method = c.req.method;
  const path = c.req.path;
  const status = c.res.status;
 
  // Only process successful mutations
  if (
    (method === 'POST' || method === 'PATCH' || method === 'DELETE') &&
    status >= 200 &&
    status < 300
  ) {
    const eventType = mapRouteToEventType(method, path);
    if (eventType) {
      // Extract resource ID from response or URL if available
      let resourceId: string | undefined;
 
      try {
        const responseClone = c.res.clone();
        const responseBody = (await responseClone.json()) as Record<
          string,
          unknown
        >;
        resourceId =
          typeof responseBody?.id === 'string' ? responseBody.id : undefined;
      } catch {
        // Response might not be JSON or might be empty (DELETE)
        // Try to extract from URL
        const pathParts = path.split('/');
        // Look for UUID pattern in path
        resourceId = pathParts.find((part) =>
          /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i.test(
            part,
          ),
        );
      }
 
      // Get user ID from JWT payload or use 'default'
      const jwtPayload = c.get('jwtPayload');
      const userId = jwtPayload?.sub || 'default';
 
      // Emit the event
      emitSSEEvent(eventType, {
        resourceId,
        userId: String(userId),
        timestamp: Date.now(),
      });
    }
  }
};
 
/**
 * Map HTTP method and route to SSE event type
 */
function mapRouteToEventType(
  method: string,
  path: string,
): SSEEventType | null {
  // Normalize path to handle base paths
  const normalizedPath = path.replace(/^\/v1\/super-agents/, '');
 
  // Agent events
  if (normalizedPath.startsWith('/agents')) {
    if (method === 'POST') return 'agent:created';
    if (method === 'PATCH') return 'agent:updated';
    if (method === 'DELETE') return 'agent:deleted';
  }
 
  // Skill events
  if (normalizedPath.startsWith('/skills')) {
    if (method === 'POST') return 'skill:created';
    if (method === 'PATCH') return 'skill:updated';
    if (method === 'DELETE') return 'skill:deleted';
  }
 
  // Model events
  if (
    normalizedPath.match(/\/skills\/[^/]+\/models/) ||
    normalizedPath.startsWith('/models')
  ) {
    if (method === 'POST') return 'model:created';
    if (method === 'PATCH') return 'model:updated';
    if (method === 'DELETE') return 'model:deleted';
  }
 
  // Evaluation events
  if (normalizedPath.match(/\/skills\/[^/]+\/evaluations/)) {
    if (method === 'POST') return 'evaluation:created';
    if (method === 'PATCH') return 'evaluation:updated';
    if (method === 'DELETE') return 'evaluation:deleted';
  }
 
  // AI Provider events
  if (
    normalizedPath.startsWith('/ai-providers') ||
    normalizedPath.startsWith('/ai-providers')
  ) {
    if (method === 'POST') return 'ai-provider:created';
    if (method === 'PATCH') return 'ai-provider:updated';
    if (method === 'DELETE') return 'ai-provider:deleted';
  }
 
  // Log events (only creation)
  if (normalizedPath.startsWith('/observability/logs') && method === 'POST') {
    return 'log:created';
  }
 
  // Feedback events
  if (normalizedPath.startsWith('/feedbacks') && method === 'POST') {
    return 'feedback:created';
  }
 
  // Improved response events
  if (normalizedPath.startsWith('/improved-responses') && method === 'POST') {
    return 'improved-response:created';
  }
 
  // Skill optimization events
  if (normalizedPath.match(/\/skills\/[^/]+\/arms/)) {
    if (method === 'POST' || method === 'PATCH')
      return 'skill-optimization:arm-updated';
  }
 
  if (normalizedPath.match(/\/skills\/[^/]+\/clusters/)) {
    if (method === 'POST' || method === 'PATCH')
      return 'skill-optimization:cluster-updated';
  }
 
  if (normalizedPath.match(/\/skills\/[^/]+\/evaluation-runs/)) {
    if (method === 'POST') return 'skill-optimization:evaluation-run-created';
    if (method === 'PATCH') return 'skill-optimization:evaluation-run-updated';
  }
 
  return null;
}