|
| 1 | +import { createHash } from 'node:crypto'; |
| 2 | +import { createClient } from 'redis'; |
| 3 | + |
| 4 | +import { appendErrorEventMirror } from './error-event-mirror'; |
| 5 | +import { queryRowsWithPgSystemContext } from './pg-context'; |
| 6 | + |
| 7 | +const REDIS_ERROR_STREAM_KEY = 'errors:ingest'; |
| 8 | +const REDIS_CLIENT_KEY = '__agentifui_fastify_error_events_redis_client__'; |
| 9 | +const CONTEXT_REDACT_PATTERN = |
| 10 | + /(password|secret|token|key|authorization|cookie)/i; |
| 11 | +const MAX_CONTEXT_DEPTH = 3; |
| 12 | + |
| 13 | +type FrontendErrorSeverity = 'info' | 'warn' | 'error' | 'critical'; |
| 14 | + |
| 15 | +function sanitizeContextValue(value: unknown, depth: number): unknown { |
| 16 | + if (depth > MAX_CONTEXT_DEPTH) { |
| 17 | + return '[truncated]'; |
| 18 | + } |
| 19 | + |
| 20 | + if (Array.isArray(value)) { |
| 21 | + return value |
| 22 | + .slice(0, 50) |
| 23 | + .map(item => sanitizeContextValue(item, depth + 1)); |
| 24 | + } |
| 25 | + |
| 26 | + if (!value || typeof value !== 'object') { |
| 27 | + return value; |
| 28 | + } |
| 29 | + |
| 30 | + const source = value as Record<string, unknown>; |
| 31 | + const sanitized: Record<string, unknown> = {}; |
| 32 | + for (const [key, inner] of Object.entries(source)) { |
| 33 | + if (CONTEXT_REDACT_PATTERN.test(key)) { |
| 34 | + sanitized[key] = '[redacted]'; |
| 35 | + continue; |
| 36 | + } |
| 37 | + sanitized[key] = sanitizeContextValue(inner, depth + 1); |
| 38 | + } |
| 39 | + return sanitized; |
| 40 | +} |
| 41 | + |
| 42 | +function sanitizeContext( |
| 43 | + context: Record<string, unknown> | undefined |
| 44 | +): Record<string, unknown> { |
| 45 | + if (!context) { |
| 46 | + return {}; |
| 47 | + } |
| 48 | + return sanitizeContextValue(context, 0) as Record<string, unknown>; |
| 49 | +} |
| 50 | + |
| 51 | +function normalizeText(value: string | undefined): string { |
| 52 | + return (value || '').trim().slice(0, 2000); |
| 53 | +} |
| 54 | + |
| 55 | +function buildFingerprint(input: { |
| 56 | + code: string; |
| 57 | + route?: string; |
| 58 | + method?: string; |
| 59 | + userMessage: string; |
| 60 | +}): string { |
| 61 | + const normalizedMessage = input.userMessage |
| 62 | + .toLowerCase() |
| 63 | + .replace(/[0-9a-f]{8}-[0-9a-f-]{27,36}/gi, '{uuid}') |
| 64 | + .replace(/\b\d{6,}\b/g, '{num}') |
| 65 | + .slice(0, 400); |
| 66 | + const raw = [ |
| 67 | + input.code, |
| 68 | + 'frontend', |
| 69 | + (input.route || '').toLowerCase(), |
| 70 | + (input.method || '').toUpperCase(), |
| 71 | + normalizedMessage, |
| 72 | + ].join('|'); |
| 73 | + |
| 74 | + return createHash('sha256').update(raw).digest('hex'); |
| 75 | +} |
| 76 | + |
| 77 | +function resolveRedisUrl(): string { |
| 78 | + const fromPrimary = process.env.REDIS_URL?.trim(); |
| 79 | + if (fromPrimary) { |
| 80 | + return fromPrimary; |
| 81 | + } |
| 82 | + |
| 83 | + const host = process.env.REDIS_HOST?.trim(); |
| 84 | + if (!host) { |
| 85 | + throw new Error( |
| 86 | + 'REDIS_URL (or REDIS_HOST) is required for frontend error event publishing' |
| 87 | + ); |
| 88 | + } |
| 89 | + |
| 90 | + const port = process.env.REDIS_PORT?.trim() || '6379'; |
| 91 | + const db = process.env.REDIS_DB?.trim() || '0'; |
| 92 | + const password = process.env.REDIS_PASSWORD?.trim(); |
| 93 | + if (password) { |
| 94 | + return `redis://:${encodeURIComponent(password)}@${host}:${port}/${db}`; |
| 95 | + } |
| 96 | + return `redis://${host}:${port}/${db}`; |
| 97 | +} |
| 98 | + |
| 99 | +function getRedisPrefix(): string { |
| 100 | + const prefix = process.env.REDIS_PREFIX?.trim() || 'agentifui'; |
| 101 | + return prefix.replace(/:+$/g, ''); |
| 102 | +} |
| 103 | + |
| 104 | +function withRedisPrefix(key: string): string { |
| 105 | + const normalizedKey = key.trim().replace(/^:+|:+$/g, ''); |
| 106 | + return `${getRedisPrefix()}:${normalizedKey}`; |
| 107 | +} |
| 108 | + |
| 109 | +async function getRedisClient() { |
| 110 | + const globalState = globalThis as unknown as Record<string, unknown>; |
| 111 | + let client = globalState[REDIS_CLIENT_KEY] as |
| 112 | + | ReturnType<typeof createClient> |
| 113 | + | undefined; |
| 114 | + if (!client) { |
| 115 | + client = createClient({ |
| 116 | + url: resolveRedisUrl(), |
| 117 | + socket: { |
| 118 | + connectTimeout: Number(process.env.REDIS_CONNECT_TIMEOUT_MS || 5000), |
| 119 | + }, |
| 120 | + pingInterval: Number(process.env.REDIS_PING_INTERVAL_MS || 10000), |
| 121 | + }); |
| 122 | + client.on('error', error => { |
| 123 | + console.warn('[FastifyFrontendErrorEvents] redis client error:', error); |
| 124 | + }); |
| 125 | + globalState[REDIS_CLIENT_KEY] = client; |
| 126 | + } |
| 127 | + |
| 128 | + if (!client.isOpen) { |
| 129 | + await client.connect(); |
| 130 | + } |
| 131 | + return client; |
| 132 | +} |
| 133 | + |
| 134 | +async function publishToRedisStream(input: { |
| 135 | + code: string; |
| 136 | + severity: FrontendErrorSeverity; |
| 137 | + requestId: string; |
| 138 | + fingerprint: string; |
| 139 | +}): Promise<void> { |
| 140 | + try { |
| 141 | + const client = await getRedisClient(); |
| 142 | + await client.xAdd( |
| 143 | + withRedisPrefix(REDIS_ERROR_STREAM_KEY), |
| 144 | + '*', |
| 145 | + { |
| 146 | + code: input.code, |
| 147 | + source: 'frontend', |
| 148 | + severity: input.severity, |
| 149 | + request_id: input.requestId, |
| 150 | + fingerprint: input.fingerprint, |
| 151 | + }, |
| 152 | + { |
| 153 | + TRIM: { |
| 154 | + strategy: 'MAXLEN', |
| 155 | + strategyModifier: '~', |
| 156 | + threshold: 20000, |
| 157 | + }, |
| 158 | + } |
| 159 | + ); |
| 160 | + } catch (error) { |
| 161 | + console.warn( |
| 162 | + '[FastifyFrontendErrorEvents] Redis stream publish failed:', |
| 163 | + error instanceof Error ? error.message : String(error) |
| 164 | + ); |
| 165 | + } |
| 166 | +} |
| 167 | + |
| 168 | +export async function recordFrontendErrorEvent(input: { |
| 169 | + code: string; |
| 170 | + severity: FrontendErrorSeverity; |
| 171 | + retryable: boolean; |
| 172 | + userMessage: string; |
| 173 | + developerMessage?: string; |
| 174 | + requestId: string; |
| 175 | + traceId?: string; |
| 176 | + actorUserId?: string; |
| 177 | + httpStatus?: number; |
| 178 | + method?: string; |
| 179 | + route?: string; |
| 180 | + context?: Record<string, unknown>; |
| 181 | +}): Promise<void> { |
| 182 | + const userMessage = normalizeText(input.userMessage) || 'Unknown error'; |
| 183 | + const developerMessage = normalizeText(input.developerMessage); |
| 184 | + const method = normalizeText(input.method).toUpperCase() || null; |
| 185 | + const route = normalizeText(input.route) || null; |
| 186 | + const traceId = normalizeText(input.traceId) || null; |
| 187 | + const actorUserId = normalizeText(input.actorUserId) || null; |
| 188 | + const contextJson = sanitizeContext(input.context); |
| 189 | + const fingerprint = buildFingerprint({ |
| 190 | + code: input.code, |
| 191 | + route: route || undefined, |
| 192 | + method: method || undefined, |
| 193 | + userMessage, |
| 194 | + }); |
| 195 | + |
| 196 | + await appendErrorEventMirror({ |
| 197 | + runtime: 'fastify', |
| 198 | + fingerprint, |
| 199 | + code: input.code, |
| 200 | + source: 'frontend', |
| 201 | + severity: input.severity, |
| 202 | + retryable: input.retryable, |
| 203 | + userMessage, |
| 204 | + developerMessage: developerMessage || null, |
| 205 | + httpStatus: input.httpStatus || null, |
| 206 | + method, |
| 207 | + route, |
| 208 | + requestId: input.requestId, |
| 209 | + traceId, |
| 210 | + actorUserId, |
| 211 | + contextJson, |
| 212 | + }); |
| 213 | + |
| 214 | + await publishToRedisStream({ |
| 215 | + code: input.code, |
| 216 | + severity: input.severity, |
| 217 | + requestId: input.requestId, |
| 218 | + fingerprint, |
| 219 | + }); |
| 220 | + |
| 221 | + await queryRowsWithPgSystemContext( |
| 222 | + ` |
| 223 | + INSERT INTO error_events ( |
| 224 | + fingerprint, |
| 225 | + code, |
| 226 | + source, |
| 227 | + severity, |
| 228 | + retryable, |
| 229 | + user_message, |
| 230 | + developer_message, |
| 231 | + http_status, |
| 232 | + method, |
| 233 | + route, |
| 234 | + request_id, |
| 235 | + trace_id, |
| 236 | + actor_user_id, |
| 237 | + context_json, |
| 238 | + first_seen_at, |
| 239 | + last_seen_at, |
| 240 | + occurrence_count, |
| 241 | + created_at, |
| 242 | + updated_at |
| 243 | + ) VALUES ( |
| 244 | + $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, |
| 245 | + $11, $12, $13, $14::jsonb, NOW(), NOW(), 1, NOW(), NOW() |
| 246 | + ) |
| 247 | + ON CONFLICT (fingerprint) DO UPDATE SET |
| 248 | + severity = EXCLUDED.severity, |
| 249 | + retryable = EXCLUDED.retryable, |
| 250 | + user_message = EXCLUDED.user_message, |
| 251 | + developer_message = COALESCE(EXCLUDED.developer_message, error_events.developer_message), |
| 252 | + http_status = COALESCE(EXCLUDED.http_status, error_events.http_status), |
| 253 | + method = COALESCE(EXCLUDED.method, error_events.method), |
| 254 | + route = COALESCE(EXCLUDED.route, error_events.route), |
| 255 | + request_id = EXCLUDED.request_id, |
| 256 | + trace_id = COALESCE(EXCLUDED.trace_id, error_events.trace_id), |
| 257 | + actor_user_id = COALESCE(EXCLUDED.actor_user_id, error_events.actor_user_id), |
| 258 | + context_json = EXCLUDED.context_json, |
| 259 | + last_seen_at = NOW(), |
| 260 | + occurrence_count = error_events.occurrence_count + 1, |
| 261 | + updated_at = NOW() |
| 262 | + `, |
| 263 | + [ |
| 264 | + fingerprint, |
| 265 | + input.code, |
| 266 | + 'frontend', |
| 267 | + input.severity, |
| 268 | + input.retryable, |
| 269 | + userMessage, |
| 270 | + developerMessage || null, |
| 271 | + input.httpStatus || null, |
| 272 | + method, |
| 273 | + route, |
| 274 | + input.requestId, |
| 275 | + traceId, |
| 276 | + actorUserId, |
| 277 | + JSON.stringify(contextJson), |
| 278 | + ] |
| 279 | + ); |
| 280 | +} |
0 commit comments