import { Pool } from 'pg'; import { globalRequestStream, type RequestEvent } from './request-stream.js'; /** * RequestLogger: Handles logging requests to database and emitting SSE events */ export class RequestLogger { constructor(private db: Pool) {} /** * Log a completion request to request_tracking table * Also emits event for real-time SSE subscribers */ async logRequest( requestId: string, caller: string, taskType: string | undefined, model: string, status: 'approved' | 'warning' | 'pending_review' | 'rejected' | 'error', tokensIn: number, tokensOut: number, costUsd: number, latencyMs: number, confidenceScore?: number, fallbackUsed?: boolean, errorMessage?: string ): Promise { const now = new Date(); const epochSeconds = Math.floor(now.getTime() / 1000); try { // Write to database await this.db.query( ` INSERT INTO request_tracking ( request_id, caller_id, task_type, model, status, confidence_score, tokens_in, tokens_out, cost_usd, latency_ms, fallback_used, error_message, created_at ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13) `, [ requestId, caller, taskType || null, model, status, confidenceScore || null, tokensIn, tokensOut, costUsd, latencyMs, fallbackUsed || false, errorMessage || null, now ] ); // Emit SSE event for real-time subscribers const event: RequestEvent = { request_id: requestId, caller, task_type: taskType, model, status, confidence_score: confidenceScore, tokens_in: tokensIn, tokens_out: tokensOut, cost_usd: costUsd, latency_ms: latencyMs, fallback_used: fallbackUsed || false, error_message: errorMessage, timestamp: epochSeconds }; globalRequestStream.emitRequest(event); } catch (error) { console.error('Error logging request:', error); // Don't throw - logging failure shouldn't break request processing } } /** * Get recent requests from request_tracking * Used by /api/dashboard/requests endpoint */ async getRecentRequests( limit: number = 100, offsetHours: number = 24 ): Promise< Array<{ request_id: string; caller: string; task_type?: string; model: string; status: string; confidence_score?: number; tokens_in: number; tokens_out: number; cost_usd: number; latency_ms: number; fallback_used: boolean; error_message?: string; created_at: string; }> > { const result = await this.db.query( ` SELECT request_id, caller_id as caller, task_type, model, status, confidence_score, tokens_in, tokens_out, cost_usd, latency_ms, fallback_used, error_message, created_at FROM request_tracking WHERE created_at > NOW() - MAKE_INTERVAL(hours => $1) ORDER BY created_at DESC LIMIT $2 `, [offsetHours, limit] ); return result.rows.map((row: any) => ({ request_id: row.request_id, caller: row.caller, task_type: row.task_type, model: row.model, status: row.status, confidence_score: row.confidence_score, tokens_in: row.tokens_in, tokens_out: row.tokens_out, cost_usd: row.cost_usd, latency_ms: row.latency_ms, fallback_used: row.fallback_used, error_message: row.error_message, created_at: row.created_at })); } /** * Get aggregated metrics for dashboard */ async getMetrics(bucketMinutes: number = 60): Promise<{ total_requests: number; total_cost: number; avg_latency: number; success_rate: number; avg_confidence: number; fallback_percentage: number; top_callers: Array<{ caller: string; count: number }>; top_models: Array<{ model: string; count: number }>; recent_errors: Array<{ request_id: string; caller: string; error_message: string; created_at: string; }>; }> { const metricsResult = await this.db.query( ` SELECT COUNT(*) as total_requests, SUM(cost_usd) as total_cost, AVG(latency_ms) as avg_latency, SUM(CASE WHEN status = 'approved' THEN 1 ELSE 0 END)::FLOAT / COUNT(*) as success_rate, AVG(confidence_score) as avg_confidence, SUM(CASE WHEN fallback_used = true THEN 1 ELSE 0 END)::FLOAT / COUNT(*) as fallback_percentage FROM request_tracking WHERE created_at > NOW() - MAKE_INTERVAL(mins => $1) `, [bucketMinutes] ); const topCallersResult = await this.db.query( ` SELECT caller_id as caller, COUNT(*) as count FROM request_tracking WHERE created_at > NOW() - MAKE_INTERVAL(mins => $1) GROUP BY caller_id ORDER BY count DESC LIMIT 5 `, [bucketMinutes] ); const topModelsResult = await this.db.query( ` SELECT model, COUNT(*) as count FROM request_tracking WHERE created_at > NOW() - MAKE_INTERVAL(mins => $1) GROUP BY model ORDER BY count DESC LIMIT 5 `, [bucketMinutes] ); const recentErrorsResult = await this.db.query( ` SELECT request_id, caller_id as caller, error_message, created_at FROM request_tracking WHERE status IN ('rejected', 'error') AND created_at > NOW() - ($1 * INTERVAL '1 minute') ORDER BY created_at DESC LIMIT 10 `, [bucketMinutes] ); const metrics = metricsResult.rows[0]; return { total_requests: parseInt(metrics.total_requests) || 0, total_cost: parseFloat(metrics.total_cost) || 0, avg_latency: Math.round(parseFloat(metrics.avg_latency) || 0), success_rate: parseFloat(metrics.success_rate) || 0, avg_confidence: parseFloat(metrics.avg_confidence) || 0, fallback_percentage: parseFloat(metrics.fallback_percentage) || 0, top_callers: topCallersResult.rows.map((row: any) => ({ caller: row.caller, count: parseInt(row.count) })), top_models: topModelsResult.rows.map((row: any) => ({ model: row.model, count: parseInt(row.count) })), recent_errors: recentErrorsResult.rows.map((row: any) => ({ request_id: row.request_id, caller: row.caller, error_message: row.error_message, created_at: row.created_at })) }; } } export const createRequestLogger = (db: Pool): RequestLogger => { return new RequestLogger(db); };