feat(pipeline): wire 7 backend modules into request pipeline

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
2026-08-03 13:52:09 +03:00 · 2026-02-14 20:06:44 -03:00
parent a9a85fdc1b
commit e87067f2fb
3 changed files with 162 additions and 31 deletions
--- 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*"],
 };
+
--- 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;
+
--- 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(