diff --git a/open-sse/handlers/chatCore/nonStreamingHandler.js b/open-sse/handlers/chatCore/nonStreamingHandler.js index d1610058..0053e9b1 100644 --- a/open-sse/handlers/chatCore/nonStreamingHandler.js +++ b/open-sse/handlers/chatCore/nonStreamingHandler.js @@ -167,7 +167,13 @@ export async function handleNonStreamingResponse({ providerResponse, provider, m } reqLogger.logProviderResponse(providerResponse.status, providerResponse.statusText, providerResponse.headers, responseBody); - if (onRequestSuccess) await onRequestSuccess(); + if (onRequestSuccess) { + Promise.resolve() + .then(onRequestSuccess) + .catch(err => { + console.error("[ChatCore] onRequestSuccess failed:", err?.message || err); + }); + } // Decloak tool_use names once on raw Claude body, before any translation (INPUT side) responseBody = decloakToolNames(responseBody, toolNameMap); diff --git a/open-sse/handlers/chatCore/streamingHandler.js b/open-sse/handlers/chatCore/streamingHandler.js index aa907cd5..d91bb3a2 100644 --- a/open-sse/handlers/chatCore/streamingHandler.js +++ b/open-sse/handlers/chatCore/streamingHandler.js @@ -44,7 +44,13 @@ function buildTransformStream({ provider, sourceFormat, targetFormat, userAgent, * Handle streaming response — pipe provider SSE through transform stream to client. */ export function handleStreamingResponse({ providerResponse, provider, model, sourceFormat, targetFormat, userAgent, body, stream, translatedBody, finalBody, requestStartTime, connectionId, apiKey, clientRawRequest, onRequestSuccess, reqLogger, toolNameMap, streamController, onStreamComplete }) { - if (onRequestSuccess) onRequestSuccess(); + if (onRequestSuccess) { + Promise.resolve() + .then(onRequestSuccess) + .catch(err => { + console.error("[ChatCore] onRequestSuccess failed:", err?.message || err); + }); + } const transformStream = buildTransformStream({ provider, sourceFormat, targetFormat, userAgent, reqLogger, toolNameMap, model, connectionId, body, onStreamComplete, apiKey }); diff --git a/src/app/(dashboard)/dashboard/console-log/ConsoleLogClient.js b/src/app/(dashboard)/dashboard/console-log/ConsoleLogClient.js index 6373015d..01a8a757 100644 --- a/src/app/(dashboard)/dashboard/console-log/ConsoleLogClient.js +++ b/src/app/(dashboard)/dashboard/console-log/ConsoleLogClient.js @@ -47,6 +47,11 @@ export default function ConsoleLogClient() { const next = [...prev, msg.line]; return next.length > CONSOLE_LOG_CONFIG.maxLines ? next.slice(-CONSOLE_LOG_CONFIG.maxLines) : next; }); + } else if (msg.type === "lines") { + setLogs((prev) => { + const next = [...prev, ...msg.lines]; + return next.length > CONSOLE_LOG_CONFIG.maxLines ? next.slice(-CONSOLE_LOG_CONFIG.maxLines) : next; + }); } else if (msg.type === "clear") { setLogs([]); } diff --git a/src/app/api/translator/console-logs/stream/route.js b/src/app/api/translator/console-logs/stream/route.js index e90f11f7..5eb3ae37 100644 --- a/src/app/api/translator/console-logs/stream/route.js +++ b/src/app/api/translator/console-logs/stream/route.js @@ -7,13 +7,14 @@ initConsoleLogCapture(); export async function GET(request) { const encoder = new TextEncoder(); const emitter = getConsoleEmitter(); - const state = { closed: false, send: null, sendClear: null, keepalive: null }; + const state = { closed: false, send: null, sendLines: null, sendClear: null, keepalive: null }; // Idempotent: safe to call from request.signal abort, cancel(), or enqueue failure. const cleanup = () => { if (state.closed) return; state.closed = true; if (state.send) emitter.off("line", state.send); + if (state.sendLines) emitter.off("lines", state.sendLines); if (state.sendClear) emitter.off("clear", state.sendClear); if (state.keepalive) clearInterval(state.keepalive); }; @@ -40,6 +41,15 @@ export async function GET(request) { } }; + state.sendLines = (lines) => { + if (state.closed || !Array.isArray(lines) || lines.length === 0) return; + try { + controller.enqueue(encoder.encode(`data: ${JSON.stringify({ type: "lines", lines })}\n\n`)); + } catch { + cleanup(); + } + }; + // Notify client when cleared state.sendClear = () => { if (state.closed) return; @@ -51,6 +61,7 @@ export async function GET(request) { }; emitter.on("line", state.send); + emitter.on("lines", state.sendLines); emitter.on("clear", state.sendClear); // Keepalive ping every 25s diff --git a/src/lib/consoleLogBuffer.js b/src/lib/consoleLogBuffer.js index afff411f..9ace30fb 100644 --- a/src/lib/consoleLogBuffer.js +++ b/src/lib/consoleLogBuffer.js @@ -21,6 +21,26 @@ if (!state.emitter) { state.emitter.setMaxListeners(50); } +if (!state.pendingLines) state.pendingLines = []; +if (!state.flushTimer) state.flushTimer = null; + +const FLUSH_INTERVAL_MS = 100; +const MAX_BATCH_LINES = 50; + +function flushPendingLines() { + state.flushTimer = null; + if (!state.pendingLines.length) return; + + const lines = state.pendingLines.splice(0, state.pendingLines.length); + state.emitter.emit("lines", lines); +} + +function scheduleFlush() { + if (state.flushTimer) return; + state.flushTimer = setTimeout(flushPendingLines, FLUSH_INTERVAL_MS); + state.flushTimer?.unref?.(); +} + function toLogLine(level, args) { return args.map(formatArg).join(" "); } @@ -48,7 +68,16 @@ function appendLine(line) { if (state.logs.length > maxLines) { state.logs = state.logs.slice(-maxLines); } - state.emitter.emit("line", line); + state.pendingLines.push(line); + if (state.pendingLines.length >= MAX_BATCH_LINES) { + if (state.flushTimer) { + clearTimeout(state.flushTimer); + state.flushTimer = null; + } + flushPendingLines(); + } else { + scheduleFlush(); + } } export function initConsoleLogCapture() { diff --git a/src/lib/db/repos/usageRepo.js b/src/lib/db/repos/usageRepo.js index 8bd632ad..c6b6cb56 100644 --- a/src/lib/db/repos/usageRepo.js +++ b/src/lib/db/repos/usageRepo.js @@ -18,15 +18,27 @@ if (!global._statsEmitter) { if (!global._pendingTimers) global._pendingTimers = {}; if (!global._recentRing) global._recentRing = { items: [], initialized: false }; if (!global._connectionMapCache) global._connectionMapCache = { map: {}, ts: 0 }; +if (!global._statsEmitTimers) global._statsEmitTimers = { pending: null, update: null }; const pendingRequests = global._pendingRequests; const lastErrorProvider = global._lastErrorProvider; const pendingTimers = global._pendingTimers; const recentRing = global._recentRing; const connCache = global._connectionMapCache; +const statsEmitTimers = global._statsEmitTimers; export const statsEmitter = global._statsEmitter; +function scheduleStatsEvent(event, delayMs = 150) { + const key = event === "update" ? "update" : "pending"; + if (statsEmitTimers[key]) return; + statsEmitTimers[key] = setTimeout(() => { + statsEmitTimers[key] = null; + statsEmitter.emit(event); + }, delayMs); + statsEmitTimers[key]?.unref?.(); +} + function getLocalDateKey(timestamp) { const d = timestamp ? new Date(timestamp) : new Date(); return `${d.getFullYear()}-${String(d.getMonth() + 1).padStart(2, "0")}-${String(d.getDate()).padStart(2, "0")}`; @@ -178,7 +190,7 @@ export function trackPendingRequest(model, provider, connectionId, started, erro if (connectionId && pendingRequests.byAccount[connectionId]?.[modelKey] > 0) { pendingRequests.byAccount[connectionId][modelKey] = 0; } - statsEmitter.emit("pending"); + scheduleStatsEvent("pending"); }, PENDING_TIMEOUT_MS); } else { clearTimeout(pendingTimers[timerKey]); @@ -192,7 +204,7 @@ export function trackPendingRequest(model, provider, connectionId, started, erro const t = new Date().toLocaleTimeString("en-US", { hour12: false, hour: "2-digit", minute: "2-digit", second: "2-digit" }); console.log(`[${t}] [PENDING] ${started ? "START" : "END"}${error ? " (ERROR)" : ""} | provider=${provider} | model=${model}`); - statsEmitter.emit("pending"); + scheduleStatsEvent("pending"); } export async function getActiveRequests() { @@ -251,9 +263,35 @@ export async function saveRequestUsage(entry) { const promptTokens = tokens.prompt_tokens || tokens.input_tokens || 0; const completionTokens = tokens.completion_tokens || tokens.output_tokens || 0; + let inserted = false; + // All 3 writes (history insert, daily upsert, lifetime counter) in ONE transaction. // better-sqlite3 is sync → no JS yield mid-transaction → no race in same process. db.transaction(() => { + const existing = db.get( + `SELECT id, endpoint FROM usageHistory + WHERE timestamp = ? + AND COALESCE(provider, '') = COALESCE(?, '') + AND COALESCE(model, '') = COALESCE(?, '') + AND COALESCE(connectionId, '') = COALESCE(?, '') + AND COALESCE(apiKey, '') = COALESCE(?, '') + AND promptTokens = ? + AND completionTokens = ? + ORDER BY id DESC LIMIT 1`, + [ + entry.timestamp, entry.provider || null, entry.model || null, + entry.connectionId || null, entry.apiKey || null, + promptTokens, completionTokens, + ] + ); + + if (existing) { + if (!existing.endpoint && entry.endpoint) { + db.run(`UPDATE usageHistory SET endpoint = ? WHERE id = ?`, [entry.endpoint, existing.id]); + } + return; + } + db.run( `INSERT INTO usageHistory(timestamp, provider, model, connectionId, apiKey, endpoint, promptTokens, completionTokens, cost, status, tokens, meta) VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, [ @@ -277,10 +315,13 @@ export async function saveRequestUsage(entry) { const cur = db.get(`SELECT value FROM _meta WHERE key = 'totalRequestsLifetime'`); const next = (cur ? parseInt(cur.value, 10) : 0) + 1; db.run(`INSERT INTO _meta(key, value) VALUES('totalRequestsLifetime', ?) ON CONFLICT(key) DO UPDATE SET value = excluded.value`, [String(next)]); + inserted = true; }); - pushToRing(entry); - statsEmitter.emit("update"); + if (inserted) { + pushToRing(entry); + scheduleStatsEvent("update", 250); + } } catch (e) { console.error("Failed to save usage stats:", e); } diff --git a/src/sse/utils/logger.js b/src/sse/utils/logger.js index d8127086..bb47ba5d 100644 --- a/src/sse/utils/logger.js +++ b/src/sse/utils/logger.js @@ -7,7 +7,7 @@ const LOG_LEVELS = { ERROR: 3 }; -const LEVEL = LOG_LEVELS.DEBUG; +const LEVEL = LOG_LEVELS[process.env.LOG_LEVEL?.toUpperCase?.()] ?? LOG_LEVELS.INFO; function formatTime() { return new Date().toLocaleTimeString("en-US", { hour12: false }); @@ -72,4 +72,3 @@ export function maskKey(key) { if (!key || key.length < 8) return "***"; return `${key.slice(0, 4)}...${key.slice(-4)}`; } -