From e87067f2fb7fc680d00968af53b0c7fd5aeeb62f Mon Sep 17 00:00:00 2001 From: diegosouzapw Date: Sat, 14 Feb 2026 20:06:44 -0300 Subject: [PATCH] feat(pipeline): wire 7 backend modules into request pipeline MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Batch 1 — Pipeline Wiring: - server-init.js: initialize compliance audit_log, run expired log cleanup, log server.start - chat.js: wire circuitBreaker (provider resilience), modelAvailability (TTL cooldowns), requestTelemetry (7-phase lifecycle), requestId, costRules (budget check/record), compliance audit logging. All wiring is non-breaking with try/catch guards. - proxy.js: replace bare fetch() with fetchWithTimeout (5s timeout on /api/settings), add X-Request-Id header for end-to-end tracing - 307/307 tests pass, build succeeds --- src/proxy.js | 23 ++++-- src/server-init.js | 23 ++++++ src/sse/handlers/chat.js | 147 ++++++++++++++++++++++++++++++++------- 3 files changed, 162 insertions(+), 31 deletions(-) diff --git a/src/proxy.js b/src/proxy.js index be3ed65821..d720a22ff6 100644 --- a/src/proxy.js +++ b/src/proxy.js @@ -1,5 +1,7 @@ import { NextResponse } from "next/server"; import { jwtVerify } from "jose"; +import { fetchWithTimeout } from "./shared/utils/fetchTimeout.js"; +import { generateRequestId } from "./shared/utils/requestId.js"; // FASE-01: Fail-fast — no hardcoded fallback. Server must have JWT_SECRET configured. if (!process.env.JWT_SECRET) { @@ -11,11 +13,16 @@ const SECRET = new TextEncoder().encode(process.env.JWT_SECRET); export async function proxy(request) { const { pathname } = request.nextUrl; + // Pipeline: Add request ID header for end-to-end tracing + const requestId = generateRequestId(); + const response = NextResponse.next(); + response.headers.set("X-Request-Id", requestId); + // Protect all dashboard routes (except onboarding) if (pathname.startsWith("/dashboard")) { // Always allow onboarding — it has its own setupComplete guard if (pathname.startsWith("/dashboard/onboarding")) { - return NextResponse.next(); + return response; } const token = request.cookies.get("auth_token")?.value; @@ -23,12 +30,13 @@ export async function proxy(request) { if (token) { try { await jwtVerify(token, SECRET); - return NextResponse.next(); + return response; } catch (err) { // FASE-01: Log auth errors instead of silently redirecting console.error("[Middleware] auth_error: JWT verification failed:", err.message, { path: pathname, tokenPresent: true, + requestId, }); return NextResponse.redirect(new URL("/login", request.url)); } @@ -36,22 +44,24 @@ export async function proxy(request) { const origin = request.nextUrl.origin; try { - const res = await fetch(`${origin}/api/settings`); + // Pipeline: Use fetchWithTimeout instead of bare fetch + const res = await fetchWithTimeout(`${origin}/api/settings`, { timeoutMs: 5000 }); const data = await res.json(); // Skip auth if login is not required if (data.requireLogin === false) { - return NextResponse.next(); + return response; } // Skip auth if no password has been set yet (fresh install) // This prevents an unresolvable loop where requireLogin=true but no password exists if (!data.hasPassword) { - return NextResponse.next(); + return response; } } catch (err) { // FASE-01: Log settings fetch errors instead of silencing them console.error("[Middleware] settings_error: Settings fetch failed:", err.message, { path: pathname, origin, + requestId, }); // On error, require login } @@ -63,9 +73,10 @@ export async function proxy(request) { return NextResponse.redirect(new URL("/dashboard", request.url)); } - return NextResponse.next(); + return response; } export const config = { matcher: ["/", "/dashboard/:path*"], }; + diff --git a/src/server-init.js b/src/server-init.js index dff5b143cf..ba179420e2 100644 --- a/src/server-init.js +++ b/src/server-init.js @@ -1,17 +1,39 @@ // Server startup script import initializeCloudSync from "./shared/services/initializeCloudSync.js"; import { enforceSecrets } from "./shared/utils/secretsValidator.js"; +import { initAuditLog, cleanupExpiredLogs, logAuditEvent } from "./lib/compliance/index.js"; async function startServer() { // FASE-01: Validate required secrets before anything else (fail-fast) enforceSecrets(); + // Compliance: Initialize audit_log table + try { + initAuditLog(); + console.log("[COMPLIANCE] Audit log table initialized"); + } catch (err) { + console.warn("[COMPLIANCE] Could not initialize audit log:", err.message); + } + + // Compliance: One-time cleanup of expired logs + try { + const cleanup = cleanupExpiredLogs(); + if (cleanup.deletedUsage || cleanup.deletedCallLogs || cleanup.deletedAuditLogs) { + console.log("[COMPLIANCE] Expired log cleanup:", cleanup); + } + } catch (err) { + console.warn("[COMPLIANCE] Log cleanup failed:", err.message); + } + console.log("Starting server with cloud sync..."); try { // Initialize cloud sync await initializeCloudSync(); console.log("Server started with cloud sync initialized"); + + // Log server start event to audit log + logAuditEvent({ action: "server.start", details: { timestamp: new Date().toISOString() } }); } catch (error) { console.log("Error initializing cloud sync:", error); process.exit(1); @@ -23,3 +45,4 @@ startServer().catch(console.log); // Export for use as module if needed export default startServer; + diff --git a/src/sse/handlers/chat.js b/src/sse/handlers/chat.js index efb549a121..2a8a58e365 100644 --- a/src/sse/handlers/chat.js +++ b/src/sse/handlers/chat.js @@ -25,21 +25,36 @@ import { logProxyEvent } from "../../lib/proxyLogger.js"; import { logTranslationEvent } from "../../lib/translatorEvents.js"; import { sanitizeRequest } from "../../shared/utils/inputSanitizer.js"; +// Pipeline integration — wired modules +import { getCircuitBreaker, CircuitBreakerOpenError } from "../../shared/utils/circuitBreaker.js"; +import { isModelAvailable, setModelUnavailable } from "../../domain/modelAvailability.js"; +import { RequestTelemetry, recordTelemetry } from "../../shared/utils/requestTelemetry.js"; +import { generateRequestId } from "../../shared/utils/requestId.js"; +import { checkBudget, recordCost } from "../../domain/costRules.js"; +import { logAuditEvent } from "../../lib/compliance/index.js"; + /** * Handle chat completion request * Supports: OpenAI, Claude, Gemini, OpenAI Responses API formats * Format detection and translation handled by translator */ export async function handleChat(request, clientRawRequest = null) { + // Pipeline: Start request telemetry + const reqId = generateRequestId(); + const telemetry = new RequestTelemetry(reqId); + let body; try { + telemetry.startPhase("parse"); body = await request.json(); + telemetry.endPhase(); } catch { log.warn("CHAT", "Invalid JSON body"); return errorResponse(HTTP_STATUS.BAD_REQUEST, "Invalid JSON body"); } // FASE-01: Input sanitization — prompt injection detection & PII redaction + telemetry.startPhase("validate"); const sanitizeResult = sanitizeRequest(body, log); if (sanitizeResult.blocked) { log.warn("SANITIZER", "Request blocked due to prompt injection", { @@ -50,6 +65,7 @@ export async function handleChat(request, clientRawRequest = null) { if (sanitizeResult.modified && sanitizeResult.sanitizedBody) { body = sanitizeResult.sanitizedBody; } + telemetry.endPhase(); // Build clientRawRequest for logging (if not provided) if (!clientRawRequest) { @@ -109,7 +125,23 @@ export async function handleChat(request, clientRawRequest = null) { return errorResponse(HTTP_STATUS.BAD_REQUEST, "Missing model"); } + // Pipeline: Budget check (if API key has budget limits) + telemetry.startPhase("policy"); + if (apiKeyInfo?.id) { + try { + const budgetOk = checkBudget(apiKeyInfo.id); + if (!budgetOk.allowed) { + log.warn("BUDGET", `API key ${apiKeyInfo.id} exceeded budget: ${budgetOk.reason}`); + return errorResponse(429, budgetOk.reason || "Budget limit exceeded"); + } + } catch { + // Budget check is best-effort — don't block on errors + } + } + telemetry.endPhase(); + // Check if model is a combo (has multiple models with fallback) + telemetry.startPhase("resolve"); const combo = await getCombo(modelStr); if (combo) { log.info( @@ -118,10 +150,18 @@ export async function handleChat(request, clientRawRequest = null) { ); // Pre-check function: skip models where all accounts are in cooldown - const isModelAvailable = async (modelString) => { + // Uses modelAvailability module for TTL-based cooldowns + const checkModelAvailable = async (modelString) => { const parsed = parseModel(modelString); const provider = parsed.provider; if (!provider) return true; // can't determine provider, let it try + + // Check domain-level availability (cooldown) + if (!isModelAvailable(provider, parsed.model || modelString)) { + log.debug("AVAILABILITY", `${provider}/${parsed.model} in cooldown, skipping`); + return false; + } + const creds = await getProviderCredentials(provider); if (!creds || creds.allRateLimited) return false; return true; @@ -132,21 +172,29 @@ export async function handleChat(request, clientRawRequest = null) { getSettings().catch(() => ({})), getCombos().catch(() => []), ]); + telemetry.endPhase(); - return handleComboChat({ + const response = await handleComboChat({ body, combo, handleSingleModel: (b, m) => - handleSingleModelChat(b, m, clientRawRequest, request, combo.name, apiKeyInfo), - isModelAvailable, + handleSingleModelChat(b, m, clientRawRequest, request, combo.name, apiKeyInfo, telemetry), + isModelAvailable: checkModelAvailable, log, settings, allCombos, }); + + // Record telemetry + recordTelemetry(telemetry); + return response; } + telemetry.endPhase(); // Single model request - return handleSingleModelChat(body, modelStr, clientRawRequest, request, null, apiKeyInfo); + const response = await handleSingleModelChat(body, modelStr, clientRawRequest, request, null, apiKeyInfo, telemetry); + recordTelemetry(telemetry); + return response; } /** @@ -162,7 +210,8 @@ async function handleSingleModelChat( clientRawRequest = null, request = null, comboName = null, - apiKeyInfo = null + apiKeyInfo = null, + telemetry = null ) { // 1. Resolve model → provider/model (or return error) const modelInfo = await getModelInfo(modelStr); @@ -192,6 +241,23 @@ async function handleSingleModelChat( log.info("ROUTING", `Provider: ${provider}, Model: ${model}`); } + // Pipeline: Check model availability (TTL cooldown) + if (!isModelAvailable(provider, model)) { + log.warn("AVAILABILITY", `${provider}/${model} is in cooldown, rejecting request`); + return unavailableResponse(HTTP_STATUS.SERVICE_UNAVAILABLE, `Model ${provider}/${model} is temporarily unavailable (cooldown)`, 30); + } + + // Pipeline: Check circuit breaker for this provider + const breaker = getCircuitBreaker(provider, { + failureThreshold: 5, + resetTimeout: 30000, + onStateChange: (name, from, to) => log.info("CIRCUIT", `${name}: ${from} → ${to}`), + }); + if (!breaker.canExecute()) { + log.warn("CIRCUIT", `Circuit breaker OPEN for ${provider}, rejecting request`); + return unavailableResponse(HTTP_STATUS.SERVICE_UNAVAILABLE, `Provider ${provider} circuit breaker is open`, 30); + } + const userAgent = request?.headers?.get("user-agent") || ""; // 2. Credential retry loop @@ -214,31 +280,62 @@ async function handleSingleModelChat( const proxyInfo = await safeResolveProxy(credentials.connectionId); const proxyStartTime = Date.now(); - // 3. Execute chat via core - const result = await runWithProxyContext(proxyInfo?.proxy || null, () => - handleChatCore({ - body: { ...body, model: `${provider}/${model}` }, - modelInfo: { provider, model }, - credentials: refreshedCredentials, log, clientRawRequest, - connectionId: credentials.connectionId, apiKeyInfo, userAgent, comboName, - onCredentialsRefreshed: async (newCreds) => { - await updateProviderCredentials(credentials.connectionId, { - accessToken: newCreds.accessToken, refreshToken: newCreds.refreshToken, - providerSpecificData: newCreds.providerSpecificData, testStatus: "active", - }); - }, - onRequestSuccess: async () => { - await clearAccountError(credentials.connectionId, credentials); - }, - }) - ); + // 3. Execute chat via core (with circuit breaker) + if (telemetry) telemetry.startPhase("connect"); + let result; + try { + result = await breaker.execute(() => + runWithProxyContext(proxyInfo?.proxy || null, () => + handleChatCore({ + body: { ...body, model: `${provider}/${model}` }, + modelInfo: { provider, model }, + credentials: refreshedCredentials, log, clientRawRequest, + connectionId: credentials.connectionId, apiKeyInfo, userAgent, comboName, + onCredentialsRefreshed: async (newCreds) => { + await updateProviderCredentials(credentials.connectionId, { + accessToken: newCreds.accessToken, refreshToken: newCreds.refreshToken, + providerSpecificData: newCreds.providerSpecificData, testStatus: "active", + }); + }, + onRequestSuccess: async () => { + await clearAccountError(credentials.connectionId, credentials); + }, + }) + ) + ); + } catch (cbErr) { + if (cbErr instanceof CircuitBreakerOpenError) { + log.warn("CIRCUIT", `${provider} circuit open during retry: ${cbErr.message}`); + return unavailableResponse(HTTP_STATUS.SERVICE_UNAVAILABLE, `Provider ${provider} circuit breaker is open`, Math.ceil(cbErr.retryAfterMs / 1000)); + } + throw cbErr; + } + if (telemetry) telemetry.endPhase(); const proxyLatency = Date.now() - proxyStartTime; // 4. Log proxy + translation events (fire-and-forget) safeLogEvents({ result, proxyInfo, proxyLatency, provider, model, sourceFormat, targetFormat, credentials, comboName, clientRawRequest }); - if (result.success) return result.response; + if (result.success) { + // Pipeline: Record cost on success + if (apiKeyInfo?.id) { + try { + const usage = result.usage || {}; + const estimatedCost = ((usage.prompt_tokens || 0) + (usage.completion_tokens || 0)) * 0.000001; // rough estimate + if (estimatedCost > 0) recordCost(apiKeyInfo.id, estimatedCost); + } catch {} + } + if (telemetry) telemetry.startPhase("finalize"); + if (telemetry) telemetry.endPhase(); + return result.response; + } + + // Pipeline: Mark model unavailable on repeated failures (429, 503) + if (result.status === 429 || result.status === 503) { + setModelUnavailable(provider, model, 60000, `HTTP ${result.status}`); + log.info("AVAILABILITY", `${provider}/${model} marked unavailable for 60s (HTTP ${result.status})`); + } // 5. Fallback to next account const { shouldFallback } = await markAccountUnavailable(