mirror of
https://github.com/diegosouzapw/OmniRoute.git
synced 2026-07-09 17:28:51 +00:00
fix(live-ws): bridge sidecar events to dashboard (#4004)
Integrated into release/v3.8.27 — repair LiveWS sidecar (startup, same-origin /live-ws, main→sidecar compression.completed bridge, early-msg queue). Fixed the cookie-parse regex (\s) + added a focused unit test; baseline bumped for the non-blocking chatCore bridge.
This commit is contained in:
parent
0bc670e414
commit
02302131fb
10 changed files with 525 additions and 54 deletions
|
|
@ -36,7 +36,7 @@
|
|||
"open-sse/executors/muse-spark-web.ts": 1284,
|
||||
"open-sse/executors/perplexity-web.ts": 1013,
|
||||
"open-sse/handlers/audioSpeech.ts": 965,
|
||||
"open-sse/handlers/chatCore.ts": 5830,
|
||||
"open-sse/handlers/chatCore.ts": 5851,
|
||||
"open-sse/handlers/imageGeneration.ts": 3777,
|
||||
"open-sse/handlers/responseSanitizer.ts": 1103,
|
||||
"open-sse/handlers/search.ts": 1442,
|
||||
|
|
@ -156,5 +156,6 @@
|
|||
"_rebaseline_2026_06_15_3941_provider_request_capture": "PR #3941 own growth: chatCore.ts 5823->5830 (+7 by check-file-size counting = run executor attempts inside the unified provider request capture scope), antigravity.ts 1649->1664 (+15) and codex.ts 1439->1447 (+8) = bridge hand-written upstream transports that bypass normal fetch/BaseExecutor capture. Cohesive logging-fidelity refactor; not extractable without hiding the actual dispatch boundary.",
|
||||
"_rebaseline_2026_06_16_3958_qwen_body_check": "PR #3958 own growth: validation.ts 4407->4428 (+21 = validateQwenWebProvider now parses the /api/v2/user 200 body and requires a real user object, since Qwen returns HTTP 200 even for invalid tokens — fixes the validation false-positive, #3931). Cohesive with the existing qwen-web validation branch; not separately extractable.",
|
||||
"_rebaseline_2026_06_16_4001_perplexity_diff_block": "PR #4001 own growth: perplexity-web.ts 939->1013 (+74 = parse the schematized API's RFC-6902 diff_block JSON-patch frames — applyMarkdownDiff + isAnswerTextUsage primary-usage lock + stop only on COMPLETED, not on a still-PENDING final flag — so streamed answers aren't empty, #3938 follow-up). Cohesive single-executor SSE-parsing logic; not separately extractable.",
|
||||
"_rebaseline_2026_06_16_4005_openai_dynamic_models": "PR #4005 own growth: models/route.ts 2494->2512 (+18 = openai model-discovery derives {customBaseUrl}/v1/models from providerSpecificData.baseUrl, SSRF-guarded via safeOutboundFetch+public-only) and pricing.ts 1529->1581 (+52 = pure-data pricing rows closing $0 gaps for registry-exposed ids: openai gpt-5.4/-mini/-nano, gpt-4.1, gpt-4o-2024-11-20, o3 + codex(cx) gpt-5.4-{xhigh,high,medium,low}, gpt-5.3-codex-spark). Cohesive; pricing is data, route change mirrors the anthropic-compat discovery path."
|
||||
"_rebaseline_2026_06_16_4005_openai_dynamic_models": "PR #4005 own growth: models/route.ts 2494->2512 (+18 = openai model-discovery derives {customBaseUrl}/v1/models from providerSpecificData.baseUrl, SSRF-guarded via safeOutboundFetch+public-only) and pricing.ts 1529->1581 (+52 = pure-data pricing rows closing $0 gaps for registry-exposed ids: openai gpt-5.4/-mini/-nano, gpt-4.1, gpt-4o-2024-11-20, o3 + codex(cx) gpt-5.4-{xhigh,high,medium,low}, gpt-5.3-codex-spark). Cohesive; pricing is data, route change mirrors the anthropic-compat discovery path.",
|
||||
"_rebaseline_2026_06_16_4004_livews_bridge": "PR #4004 own growth: chatCore.ts 5830->5851 (+21 = forwardDashboardEventToLiveWs — a best-effort, non-blocking, timeout-bounded POST that bridges compression.completed events from the main process to the LiveWS sidecar so the dashboard updates under a reverse proxy). Cohesive fire-and-forget beacon at the existing compression emit site; not extractable. Structural shrink of chatCore.ts tracked in #3501."
|
||||
}
|
||||
|
|
|
|||
|
|
@ -445,6 +445,24 @@ function extractMemoryTextFromRequestBody(
|
|||
return "";
|
||||
}
|
||||
|
||||
async function forwardDashboardEventToLiveWs(event: string, payload: unknown): Promise<void> {
|
||||
const port = process.env.LIVE_WS_PORT || "20129";
|
||||
const controller = new AbortController();
|
||||
const timeout = setTimeout(() => controller.abort(), 1_500);
|
||||
try {
|
||||
await fetch(`http://127.0.0.1:${port}/__omniroute_event`, {
|
||||
method: "POST",
|
||||
headers: { "content-type": "application/json" },
|
||||
body: JSON.stringify({ event, payload, timestamp: Date.now() }),
|
||||
signal: controller.signal,
|
||||
});
|
||||
} catch {
|
||||
// Best-effort sidecar bridge; do not affect the chat hot path.
|
||||
} finally {
|
||||
clearTimeout(timeout);
|
||||
}
|
||||
}
|
||||
|
||||
async function maybeSyncClaudeExtraUsageState({
|
||||
provider,
|
||||
connectionId,
|
||||
|
|
@ -2654,7 +2672,7 @@ export async function handleChatCore({
|
|||
// Guard: only emit when compression actually ran and produced stats.
|
||||
if (result.compressed && result.stats) {
|
||||
try {
|
||||
emit("compression.completed", {
|
||||
const compressionCompletedPayload = {
|
||||
requestId: traceId,
|
||||
comboId: result.stats.compressionComboId ?? null,
|
||||
mode,
|
||||
|
|
@ -2665,7 +2683,12 @@ export async function handleChatCore({
|
|||
validationWarnings: result.stats.validationWarnings,
|
||||
fallbackApplied: result.stats.fallbackApplied,
|
||||
timestamp: Date.now(),
|
||||
});
|
||||
};
|
||||
emit("compression.completed", compressionCompletedPayload);
|
||||
void forwardDashboardEventToLiveWs(
|
||||
"compression.completed",
|
||||
compressionCompletedPayload
|
||||
);
|
||||
} catch (_emitErr) {
|
||||
// never propagate into the hot path — but log like the sibling
|
||||
// fire-and-forget blocks so a throwing event bus isn't fully silent.
|
||||
|
|
|
|||
|
|
@ -244,6 +244,7 @@
|
|||
"react-reconciler": "^0.33.0",
|
||||
"recharts": "^3.8.1",
|
||||
"selfsigned": "^5.5.0",
|
||||
"socks": "^2.8.7",
|
||||
"sql.js": "^1.14.1",
|
||||
"sqlite-vec": "^0.1.9",
|
||||
"tsx": "^4.22.3",
|
||||
|
|
|
|||
|
|
@ -4,15 +4,17 @@
|
|||
* This script starts the live dashboard WebSocket server as a separate
|
||||
* process alongside the Next.js app. Run it with:
|
||||
*
|
||||
* node scripts/start-ws-server.js
|
||||
*
|
||||
* Or use the built-in auto-start in src/server/ws/liveServer.ts.
|
||||
* node scripts/start-ws-server.mjs
|
||||
*
|
||||
* Environment variables:
|
||||
* LIVE_WS_PORT — WebSocket server port (default: 20129)
|
||||
* LIVE_WS_HOST — WebSocket server host (default: 127.0.0.1)
|
||||
* OMNIROUTE_DISABLE_LIVE_WS — Set to "1" or "true" to disable
|
||||
*/
|
||||
|
||||
import { spawnSync } from "node:child_process";
|
||||
import { fileURLToPath } from "node:url";
|
||||
|
||||
if (
|
||||
process.env.OMNIROUTE_DISABLE_LIVE_WS === "1" ||
|
||||
process.env.OMNIROUTE_DISABLE_LIVE_WS === "true"
|
||||
|
|
@ -21,23 +23,39 @@ if (
|
|||
process.exit(0);
|
||||
}
|
||||
|
||||
// Register tsx to handle TypeScript imports
|
||||
import { register } from "node:module";
|
||||
import { pathToFileURL } from "node:url";
|
||||
const BOOTSTRAPPED_ENV = "OMNIROUTE_LIVE_WS_BOOTSTRAPPED";
|
||||
|
||||
register("tsx", pathToFileURL("./"));
|
||||
if (process.env[BOOTSTRAPPED_ENV] !== "1") {
|
||||
const result = spawnSync(process.execPath, ["--import", "tsx", fileURLToPath(import.meta.url)], {
|
||||
stdio: "inherit",
|
||||
env: {
|
||||
...process.env,
|
||||
[BOOTSTRAPPED_ENV]: "1",
|
||||
// Prevent liveServer.ts from auto-starting on import; this script owns
|
||||
// process startup so errors propagate to the supervisor/CLI caller.
|
||||
OMNIROUTE_ENABLE_LIVE_WS: "0",
|
||||
},
|
||||
});
|
||||
|
||||
const { startLiveDashboardServer } = await import("../src/server/ws/liveServer");
|
||||
if (result.signal) {
|
||||
process.kill(process.pid, result.signal);
|
||||
}
|
||||
|
||||
process.exit(result.status ?? 1);
|
||||
}
|
||||
|
||||
const { startLiveDashboardServer } = await import("../src/server/ws/liveServer.ts");
|
||||
|
||||
const port = parseInt(process.env.LIVE_WS_PORT || "20129", 10);
|
||||
const host = process.env.LIVE_WS_HOST || "127.0.0.1";
|
||||
|
||||
console.log(`[LiveWS] Starting dashboard WebSocket server on port ${port}...`);
|
||||
console.log(`[LiveWS] Starting dashboard WebSocket server on ${host}:${port}...`);
|
||||
|
||||
startLiveDashboardServer(port)
|
||||
.then((server) => {
|
||||
console.log(`[LiveWS] Dashboard WebSocket server listening on ws://0.0.0.0:${port}`);
|
||||
console.log(`[LiveWS] Connect via: ws://localhost:${port}?token=<api-key>`);
|
||||
console.log(`[LiveWS] Channels: requests, combo, credentials`);
|
||||
startLiveDashboardServer(port, host)
|
||||
.then(() => {
|
||||
console.log(`[LiveWS] Dashboard WebSocket server listening on ws://${host}:${port}`);
|
||||
console.log("[LiveWS] Connect via: ws://localhost:%d?token=<api-key>", port);
|
||||
console.log("[LiveWS] Channels: requests, combo, credentials");
|
||||
})
|
||||
.catch((err) => {
|
||||
console.error("[LiveWS] Failed to start:", err);
|
||||
|
|
|
|||
|
|
@ -17,7 +17,17 @@ import type { DashboardChannel, DashboardEventName } from "@/lib/events/types";
|
|||
// ── Config ────────────────────────────────────────────────────────────────
|
||||
|
||||
const WS_RECONNECT_DELAYS = [1000, 2000, 4000, 8000, 16000, 30000];
|
||||
const DEFAULT_WS_URL = `ws://${typeof window !== "undefined" ? window.location.hostname : "localhost"}:20129`;
|
||||
function getDefaultWsUrl(): string {
|
||||
if (typeof window === "undefined") return "ws://localhost:20129";
|
||||
const protocol = window.location.protocol === "https:" ? "wss:" : "ws:";
|
||||
const { hostname } = window.location;
|
||||
if (hostname === "localhost" || hostname === "127.0.0.1") {
|
||||
return `${protocol}//${hostname}:20129`;
|
||||
}
|
||||
return `${protocol}//${window.location.host}/live-ws`;
|
||||
}
|
||||
|
||||
const DEFAULT_WS_URL = getDefaultWsUrl();
|
||||
|
||||
// ── Types ─────────────────────────────────────────────────────────────────
|
||||
|
||||
|
|
|
|||
|
|
@ -440,3 +440,32 @@ export function getCompressionAnalyticsSummary(since?: string): CompressionAnaly
|
|||
},
|
||||
};
|
||||
}
|
||||
|
||||
export interface LatestCompressionAnalyticsRun {
|
||||
id: number;
|
||||
timestamp: string;
|
||||
combo_id: string | null;
|
||||
compression_combo_id: string | null;
|
||||
mode: string;
|
||||
original_tokens: number;
|
||||
compressed_tokens: number;
|
||||
tokens_saved: number;
|
||||
duration_ms: number | null;
|
||||
request_id: string | null;
|
||||
engine: string | null;
|
||||
validation_fallback: number | null;
|
||||
}
|
||||
|
||||
export function getLatestCompressionAnalyticsRun(): LatestCompressionAnalyticsRun | undefined {
|
||||
const db = getDbInstance();
|
||||
return db
|
||||
.prepare(
|
||||
`SELECT id, timestamp, combo_id, compression_combo_id, mode,
|
||||
original_tokens, compressed_tokens, tokens_saved, duration_ms,
|
||||
request_id, engine, validation_fallback
|
||||
FROM compression_analytics
|
||||
ORDER BY timestamp DESC, id DESC
|
||||
LIMIT 1`
|
||||
)
|
||||
.get() as LatestCompressionAnalyticsRun | undefined;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -14,7 +14,8 @@
|
|||
*/
|
||||
|
||||
import { WebSocketServer, WebSocket } from "ws";
|
||||
import { createServer } from "http";
|
||||
import { jwtVerify } from "jose";
|
||||
import { createServer, type IncomingMessage, type ServerResponse } from "http";
|
||||
import { randomUUID } from "crypto";
|
||||
|
||||
// ── Types ─────────────────────────────────────────────────────────────────
|
||||
|
|
@ -38,6 +39,8 @@ const HEARTBEAT_INTERVAL_MS = 15_000;
|
|||
const HEARTBEAT_TIMEOUT_MS = 35_000;
|
||||
const MAX_CLIENTS = 500;
|
||||
const MAX_EVENTS_PER_SECOND = 100;
|
||||
const MAX_PENDING_MESSAGES_PER_CLIENT = 32;
|
||||
const MAX_PENDING_MESSAGE_BYTES = 16_384;
|
||||
|
||||
/**
|
||||
* Origins allowed to open a WebSocket. Defaults to the loopback dashboard
|
||||
|
|
@ -91,6 +94,20 @@ const BACKLOG_MAX = 500;
|
|||
|
||||
// ── Auth ──────────────────────────────────────────────────────────────────
|
||||
|
||||
function toWebHeaders(headers: import("http").IncomingMessage["headers"]): Headers {
|
||||
const webHeaders = new Headers();
|
||||
|
||||
for (const [name, value] of Object.entries(headers)) {
|
||||
if (typeof value === "string") {
|
||||
webHeaders.set(name, value);
|
||||
} else if (Array.isArray(value)) {
|
||||
webHeaders.set(name, name.toLowerCase() === "cookie" ? value.join("; ") : value.join(", "));
|
||||
}
|
||||
}
|
||||
|
||||
return webHeaders;
|
||||
}
|
||||
|
||||
async function authorizeConnection(request: import("http").IncomingMessage): Promise<WsAuthResult> {
|
||||
const sessionId = randomUUID().slice(0, 8);
|
||||
|
||||
|
|
@ -99,16 +116,25 @@ async function authorizeConnection(request: import("http").IncomingMessage): Pro
|
|||
// headers — a single screenshot of the URL bar exposes the API key.
|
||||
const token = extractBearerToken(request) || extractAltTokenHeader(request);
|
||||
|
||||
// Browser WebSocket clients cannot set custom Authorization headers. When
|
||||
// LiveWS is exposed same-origin through a reverse proxy, accept the existing
|
||||
// dashboard session cookie before falling back to API-key authentication. Keep
|
||||
// the check local to this sidecar so it does not import Next.js-only modules.
|
||||
if (!token) {
|
||||
if (await isDashboardCookieAuthenticated(request)) {
|
||||
return { authorized: true, sessionId };
|
||||
}
|
||||
return { authorized: false, sessionId, error: "Missing token" };
|
||||
}
|
||||
|
||||
try {
|
||||
// Validate API key via the existing auth system
|
||||
const { extractApiKey, isValidApiKey } = await import("../services/auth");
|
||||
const apiKey = extractApiKey({ headers: { authorization: `Bearer ${token}` } } as any);
|
||||
// Validate API key via the existing auth system.
|
||||
const { extractApiKey, isValidApiKey } = await import("../../sse/services/auth.ts");
|
||||
const apiKey = extractApiKey({ headers: { authorization: `Bearer ${token}` } } as any, {
|
||||
allowUrl: false,
|
||||
});
|
||||
|
||||
if (!apiKey || !isValidApiKey(apiKey)) {
|
||||
if (!apiKey || !(await isValidApiKey(apiKey))) {
|
||||
return { authorized: false, sessionId, error: "Invalid API key" };
|
||||
}
|
||||
|
||||
|
|
@ -124,6 +150,36 @@ function extractAltTokenHeader(request: import("http").IncomingMessage): string
|
|||
return typeof raw === "string" ? raw : null;
|
||||
}
|
||||
|
||||
export function getCookieValueFromHeader(
|
||||
headers: import("http").IncomingHttpHeaders,
|
||||
name: string
|
||||
): string | null {
|
||||
const raw = headers.cookie;
|
||||
const cookieHeader = Array.isArray(raw) ? raw.join("; ") : raw;
|
||||
if (!cookieHeader) return null;
|
||||
const escaped = name.replace(/[.*+?^${}()|[\]\\]/g, "\\$&");
|
||||
// NOTE: \\s (not \s) — this is a plain template literal, so \s would collapse to a
|
||||
// literal "s" and the pattern would only match auth_token when it is the FIRST cookie.
|
||||
// Browsers serialize the Cookie header as "a=1; b=2", so the leading-cookie case
|
||||
// (auth_token preceded by another cookie) must match too (#4004 same-origin proxy auth).
|
||||
const match = cookieHeader.match(new RegExp(`(?:^|;\\s*)${escaped}=([^;]*)`));
|
||||
return match ? decodeURIComponent(match[1]) : null;
|
||||
}
|
||||
|
||||
async function isDashboardCookieAuthenticated(
|
||||
request: import("http").IncomingMessage
|
||||
): Promise<boolean> {
|
||||
const token = getCookieValueFromHeader(request.headers, "auth_token");
|
||||
if (!token || !process.env.JWT_SECRET) return false;
|
||||
try {
|
||||
const secret = new TextEncoder().encode(process.env.JWT_SECRET);
|
||||
await jwtVerify(token, secret);
|
||||
return true;
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
function extractBearerToken(request: import("http").IncomingMessage): string | null {
|
||||
const auth = request.headers["authorization"];
|
||||
if (!auth || typeof auth !== "string") return null;
|
||||
|
|
@ -202,35 +258,148 @@ function sendTo(ws: WebSocket, msg: WsServerMessage | Record<string, unknown>):
|
|||
|
||||
// ── Event Bus → WebSocket Bridge ──────────────────────────────────────────
|
||||
|
||||
function publishDashboardEvent(
|
||||
event: DashboardEventName,
|
||||
payload: unknown,
|
||||
timestamp = Date.now()
|
||||
): boolean {
|
||||
const channel = getChannelForEvent(event);
|
||||
if (!channel) return false;
|
||||
|
||||
// Store in backlog so clients that subscribe just after a run still receive it.
|
||||
eventHistoryBacklog.push({ event, payload, timestamp });
|
||||
if (eventHistoryBacklog.length > BACKLOG_MAX) {
|
||||
eventHistoryBacklog.shift();
|
||||
}
|
||||
|
||||
const msg: WsEventMessage = {
|
||||
type: "event",
|
||||
channel,
|
||||
event,
|
||||
data: payload,
|
||||
};
|
||||
|
||||
for (const [clientId, client] of clients) {
|
||||
if (client.ws.readyState !== WebSocket.OPEN) {
|
||||
clients.delete(clientId);
|
||||
continue;
|
||||
}
|
||||
if (client.subscribedChannels.has(channel)) {
|
||||
sendTo(client.ws, msg);
|
||||
}
|
||||
}
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
function subscribeToEventBus(): () => void {
|
||||
return onAny((event: DashboardEventName, payload: unknown) => {
|
||||
const channel = getChannelForEvent(event);
|
||||
if (!channel) return;
|
||||
publishDashboardEvent(event, payload);
|
||||
});
|
||||
}
|
||||
|
||||
// Store in backlog
|
||||
eventHistoryBacklog.push({ event, payload, timestamp: Date.now() });
|
||||
if (eventHistoryBacklog.length > BACKLOG_MAX) {
|
||||
eventHistoryBacklog.shift();
|
||||
}
|
||||
function isLoopbackRequest(req: IncomingMessage): boolean {
|
||||
const addr = req.socket.remoteAddress;
|
||||
return addr === "127.0.0.1" || addr === "::1" || addr === "::ffff:127.0.0.1";
|
||||
}
|
||||
|
||||
// Forward to subscribed clients
|
||||
const msg: WsEventMessage = {
|
||||
type: "event",
|
||||
channel,
|
||||
event,
|
||||
data: payload,
|
||||
};
|
||||
function handleInternalEventRequest(req: IncomingMessage, res: ServerResponse): void {
|
||||
if (req.method !== "POST" || req.url !== "/__omniroute_event") {
|
||||
res.writeHead(404).end();
|
||||
return;
|
||||
}
|
||||
if (!isLoopbackRequest(req)) {
|
||||
res.writeHead(403, { "content-type": "application/json" }).end(JSON.stringify({ ok: false }));
|
||||
return;
|
||||
}
|
||||
|
||||
for (const [clientId, client] of clients) {
|
||||
if (client.ws.readyState !== WebSocket.OPEN) {
|
||||
clients.delete(clientId);
|
||||
continue;
|
||||
}
|
||||
if (client.subscribedChannels.has(channel)) {
|
||||
sendTo(client.ws, msg);
|
||||
}
|
||||
let body = "";
|
||||
req.setEncoding("utf8");
|
||||
req.on("data", (chunk) => {
|
||||
body += chunk;
|
||||
if (body.length > 1_000_000) {
|
||||
req.destroy(new Error("Internal event payload too large"));
|
||||
}
|
||||
});
|
||||
req.on("error", () => {
|
||||
if (!res.headersSent) res.writeHead(400).end();
|
||||
});
|
||||
req.on("end", () => {
|
||||
try {
|
||||
const parsed = JSON.parse(body || "{}");
|
||||
const event = parsed.event as DashboardEventName;
|
||||
if (!Object.values(CHANNEL_EVENTS).some((events) => events.includes(event))) {
|
||||
res
|
||||
.writeHead(400, { "content-type": "application/json" })
|
||||
.end(JSON.stringify({ ok: false }));
|
||||
return;
|
||||
}
|
||||
const ok = publishDashboardEvent(
|
||||
event,
|
||||
parsed.payload,
|
||||
Number(parsed.timestamp) || Date.now()
|
||||
);
|
||||
res
|
||||
.writeHead(ok ? 202 : 400, { "content-type": "application/json" })
|
||||
.end(JSON.stringify({ ok }));
|
||||
} catch {
|
||||
res.writeHead(400, { "content-type": "application/json" }).end(JSON.stringify({ ok: false }));
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
async function seedLatestCompressionRunFromDb(): Promise<void> {
|
||||
try {
|
||||
const { getLatestCompressionAnalyticsRun } = await import("@/lib/db/compressionAnalytics");
|
||||
const row = getLatestCompressionAnalyticsRun();
|
||||
if (!row) return;
|
||||
|
||||
const originalTokens = Number(row.original_tokens) || 0;
|
||||
const compressedTokens = Number(row.compressed_tokens) || 0;
|
||||
const savingsPercent =
|
||||
originalTokens > 0
|
||||
? Math.round(((originalTokens - compressedTokens) / originalTokens) * 100)
|
||||
: 0;
|
||||
const timestamp = Number.isFinite(Date.parse(row.timestamp))
|
||||
? Date.parse(row.timestamp)
|
||||
: Date.now();
|
||||
|
||||
publishDashboardEvent(
|
||||
"compression.completed",
|
||||
{
|
||||
requestId: row.request_id || `analytics-${row.id}`,
|
||||
comboId: row.compression_combo_id || row.combo_id || null,
|
||||
mode: row.mode,
|
||||
originalTokens,
|
||||
compressedTokens,
|
||||
savingsPercent,
|
||||
engineBreakdown: [
|
||||
{
|
||||
engine: row.engine || row.mode || "compression",
|
||||
originalTokens,
|
||||
compressedTokens,
|
||||
savingsPercent,
|
||||
techniquesUsed: [],
|
||||
rulesApplied: [],
|
||||
durationMs: row.duration_ms ?? undefined,
|
||||
},
|
||||
],
|
||||
validationWarnings: [],
|
||||
fallbackApplied: Boolean(row.validation_fallback),
|
||||
timestamp,
|
||||
},
|
||||
timestamp
|
||||
);
|
||||
console.log(
|
||||
"[LiveWS] Seeded latest compression run from analytics: %s",
|
||||
row.request_id || row.id
|
||||
);
|
||||
} catch (err) {
|
||||
console.warn(
|
||||
"[LiveWS] Could not seed compression analytics backlog: %s",
|
||||
err instanceof Error ? err.message : String(err)
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
// ── Heartbeat ─────────────────────────────────────────────────────────────
|
||||
|
|
@ -270,13 +439,39 @@ export async function startLiveDashboardServer(
|
|||
port = DEFAULT_PORT,
|
||||
host = DEFAULT_HOST
|
||||
): Promise<import("http").Server> {
|
||||
const server = createServer();
|
||||
const server = createServer((req, res) => {
|
||||
handleInternalEventRequest(req, res);
|
||||
});
|
||||
const wss = new WebSocketServer({ server });
|
||||
|
||||
// Subscribe to EventBus
|
||||
const unsubscribe = subscribeToEventBus();
|
||||
await seedLatestCompressionRunFromDb();
|
||||
|
||||
wss.on("connection", async (ws, request) => {
|
||||
const pendingMessages: string[] = [];
|
||||
let activeClientId: string | null = null;
|
||||
|
||||
// Clients can send the subscribe frame immediately after the WS open event,
|
||||
// while dashboard cookie/API-key auth is still resolving. Queue those early
|
||||
// messages so the first subscribe is not dropped.
|
||||
ws.on("message", (data) => {
|
||||
const raw = data.toString();
|
||||
if (!activeClientId) {
|
||||
if (
|
||||
pendingMessages.length >= MAX_PENDING_MESSAGES_PER_CLIENT ||
|
||||
raw.length > MAX_PENDING_MESSAGE_BYTES
|
||||
) {
|
||||
sendTo(ws, { type: "error", code: "RATE_LIMITED", message: "Too many early messages" });
|
||||
ws.close(4008, "Too many early messages");
|
||||
return;
|
||||
}
|
||||
pendingMessages.push(raw);
|
||||
return;
|
||||
}
|
||||
handleMessage(activeClientId, raw);
|
||||
});
|
||||
|
||||
// Origin check — browsers always send Origin on the WS upgrade; reject
|
||||
// unknown origins to stop drive-by cross-origin WebSocket from a victim
|
||||
// page. Non-browser clients (CLI / MCP) omit Origin and are accepted
|
||||
|
|
@ -305,6 +500,7 @@ export async function startLiveDashboardServer(
|
|||
}
|
||||
|
||||
const clientId = auth.sessionId;
|
||||
activeClientId = clientId;
|
||||
const client: ClientState = {
|
||||
ws,
|
||||
sessionId: clientId,
|
||||
|
|
@ -327,10 +523,10 @@ export async function startLiveDashboardServer(
|
|||
clients.size
|
||||
);
|
||||
|
||||
// Handle messages
|
||||
ws.on("message", (data) => {
|
||||
handleMessage(clientId, data.toString());
|
||||
});
|
||||
// Replay any subscribe/ping frames sent while auth was still pending.
|
||||
for (const raw of pendingMessages.splice(0)) {
|
||||
handleMessage(clientId, raw);
|
||||
}
|
||||
|
||||
// Handle close
|
||||
ws.on("close", () => {
|
||||
|
|
|
|||
|
|
@ -8,7 +8,7 @@
|
|||
|
||||
export interface WsSubscribeMessage {
|
||||
type: "subscribe";
|
||||
channels: Array<"requests" | "combo" | "credentials">;
|
||||
channels: Array<"requests" | "combo" | "credentials" | "compression">;
|
||||
}
|
||||
|
||||
export interface WsPingMessage {
|
||||
|
|
@ -21,7 +21,7 @@ export type WsClientMessage = WsSubscribeMessage | WsPingMessage;
|
|||
|
||||
export interface WsEventMessage {
|
||||
type: "event";
|
||||
channel: "requests" | "combo" | "credentials";
|
||||
channel: "requests" | "combo" | "credentials" | "compression";
|
||||
event: string;
|
||||
data: unknown;
|
||||
}
|
||||
|
|
@ -35,7 +35,7 @@ export interface WsWelcomeMessage {
|
|||
version: string;
|
||||
sessionId: string;
|
||||
serverTime: number;
|
||||
channels: Array<"requests" | "combo" | "credentials">;
|
||||
channels: Array<"requests" | "combo" | "credentials" | "compression">;
|
||||
/** Number of buffered events since last reconnect */
|
||||
backlog: number;
|
||||
}
|
||||
|
|
|
|||
151
tests/unit/cli/live-ws-startup.test.ts
Normal file
151
tests/unit/cli/live-ws-startup.test.ts
Normal file
|
|
@ -0,0 +1,151 @@
|
|||
import assert from "node:assert/strict";
|
||||
import { spawn, type ChildProcessWithoutNullStreams } from "node:child_process";
|
||||
import { SignJWT } from "jose";
|
||||
import net from "node:net";
|
||||
import test from "node:test";
|
||||
import WebSocket from "ws";
|
||||
|
||||
function getFreePort(): Promise<number> {
|
||||
return new Promise((resolve, reject) => {
|
||||
const server = net.createServer();
|
||||
server.once("error", reject);
|
||||
server.listen(0, "127.0.0.1", () => {
|
||||
const address = server.address();
|
||||
server.close(() => {
|
||||
if (address && typeof address === "object") resolve(address.port);
|
||||
else reject(new Error("Failed to allocate a local port"));
|
||||
});
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
function terminateTree(child: ChildProcessWithoutNullStreams): void {
|
||||
if (!child.pid) return;
|
||||
|
||||
try {
|
||||
process.kill(-child.pid, "SIGTERM");
|
||||
} catch {
|
||||
child.kill("SIGTERM");
|
||||
}
|
||||
}
|
||||
|
||||
function waitForStartup(
|
||||
child: ChildProcessWithoutNullStreams,
|
||||
getOutput: () => string
|
||||
): Promise<void> {
|
||||
return new Promise((resolve, reject) => {
|
||||
const timeout = setTimeout(() => {
|
||||
reject(new Error(`LiveWS startup timed out. Output:\n${getOutput()}`));
|
||||
}, 8_000);
|
||||
|
||||
const onData = () => {
|
||||
const output = getOutput();
|
||||
if (output.includes("Dashboard WebSocket server listening")) {
|
||||
cleanup();
|
||||
resolve();
|
||||
}
|
||||
};
|
||||
|
||||
const onExit = (code: number | null, signal: NodeJS.Signals | null) => {
|
||||
cleanup();
|
||||
reject(
|
||||
new Error(`LiveWS exited before listening: code=${code} signal=${signal}\n${getOutput()}`)
|
||||
);
|
||||
};
|
||||
|
||||
const cleanup = () => {
|
||||
clearTimeout(timeout);
|
||||
child.stdout.off("data", onData);
|
||||
child.stderr.off("data", onData);
|
||||
child.off("exit", onExit);
|
||||
};
|
||||
|
||||
child.stdout.on("data", onData);
|
||||
child.stderr.on("data", onData);
|
||||
child.once("exit", onExit);
|
||||
onData();
|
||||
});
|
||||
}
|
||||
|
||||
test(
|
||||
"LiveWS startup script boots on current Node and accepts API-key WebSocket clients",
|
||||
{ timeout: 15_000 },
|
||||
async () => {
|
||||
const port = await getFreePort();
|
||||
const apiKey = "test-live-ws-key";
|
||||
const jwtSecret = "test-live-ws-jwt-secret";
|
||||
const origin = "http://localhost";
|
||||
let output = "";
|
||||
|
||||
const child = spawn(process.execPath, ["scripts/start-ws-server.mjs"], {
|
||||
cwd: process.cwd(),
|
||||
detached: process.platform !== "win32",
|
||||
env: {
|
||||
...process.env,
|
||||
NODE_ENV: "test",
|
||||
OMNIROUTE_API_KEY: apiKey,
|
||||
JWT_SECRET: jwtSecret,
|
||||
LIVE_WS_HOST: "127.0.0.1",
|
||||
LIVE_WS_PORT: String(port),
|
||||
LIVE_WS_ALLOWED_ORIGINS: origin,
|
||||
},
|
||||
});
|
||||
|
||||
child.stdout.setEncoding("utf8");
|
||||
child.stderr.setEncoding("utf8");
|
||||
child.stdout.on("data", (chunk) => {
|
||||
output += chunk;
|
||||
});
|
||||
child.stderr.on("data", (chunk) => {
|
||||
output += chunk;
|
||||
});
|
||||
|
||||
try {
|
||||
await waitForStartup(child, () => output);
|
||||
|
||||
assert.doesNotMatch(output, /tsx must be loaded with --import/i);
|
||||
assert.doesNotMatch(output, /EADDRINUSE/i);
|
||||
|
||||
async function expectLiveWsOpen(headers: Record<string, string>): Promise<void> {
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
const timeout = setTimeout(() => {
|
||||
reject(new Error(`Timed out waiting for LiveWS connection. Output:\n${output}`));
|
||||
}, 4_000);
|
||||
|
||||
const ws = new WebSocket(`ws://127.0.0.1:${port}/live-ws`, { headers });
|
||||
|
||||
ws.once("open", () => {
|
||||
clearTimeout(timeout);
|
||||
ws.close(1000);
|
||||
resolve();
|
||||
});
|
||||
|
||||
ws.once("error", (error) => {
|
||||
clearTimeout(timeout);
|
||||
reject(new Error(`LiveWS client failed: ${error.message}. Output:\n${output}`));
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
await expectLiveWsOpen({
|
||||
Authorization: `Bearer ${apiKey}`,
|
||||
Origin: origin,
|
||||
});
|
||||
|
||||
const dashboardToken = await new SignJWT({ authenticated: true })
|
||||
.setProtectedHeader({ alg: "HS256" })
|
||||
.setExpirationTime("5m")
|
||||
.sign(new TextEncoder().encode(jwtSecret));
|
||||
|
||||
// Send auth_token preceded by another cookie (the real browser case: "a=1; auth_token=…").
|
||||
// Guards #4004's cookie-parse regex: a literal-"s" bug (\s vs \\s) only matched auth_token
|
||||
// when it was the FIRST cookie, silently breaking same-origin reverse-proxy auth otherwise.
|
||||
await expectLiveWsOpen({
|
||||
Cookie: `omni_pref=dark; auth_token=${dashboardToken}`,
|
||||
Origin: origin,
|
||||
});
|
||||
} finally {
|
||||
terminateTree(child);
|
||||
}
|
||||
}
|
||||
);
|
||||
42
tests/unit/live-ws-cookie-parse.test.ts
Normal file
42
tests/unit/live-ws-cookie-parse.test.ts
Normal file
|
|
@ -0,0 +1,42 @@
|
|||
// Regression for #4004: liveServer's cookie parser used an UNTAGGED template literal
|
||||
// `(?:^|;\s*)` — \s collapsed to a literal "s", so auth_token only matched when it was
|
||||
// the FIRST cookie. Browsers serialize "a=1; auth_token=…", so the same-origin
|
||||
// reverse-proxy dashboard auth silently failed whenever any cookie preceded auth_token.
|
||||
// Keep the server from auto-starting on import.
|
||||
process.env.OMNIROUTE_ENABLE_LIVE_WS = "0";
|
||||
|
||||
import test from "node:test";
|
||||
import assert from "node:assert/strict";
|
||||
import { getCookieValueFromHeader } from "@/server/ws/liveServer";
|
||||
|
||||
test("getCookieValueFromHeader reads auth_token when it is the only cookie", () => {
|
||||
assert.equal(getCookieValueFromHeader({ cookie: "auth_token=abc123" }, "auth_token"), "abc123");
|
||||
});
|
||||
|
||||
test("getCookieValueFromHeader reads auth_token when preceded by other cookies (#4004)", () => {
|
||||
// The standard browser "; " separator — the case the \s-vs-\\s bug broke.
|
||||
assert.equal(
|
||||
getCookieValueFromHeader({ cookie: "omni_pref=dark; auth_token=abc123" }, "auth_token"),
|
||||
"abc123"
|
||||
);
|
||||
assert.equal(
|
||||
getCookieValueFromHeader({ cookie: "a=1; b=2; auth_token=xyz" }, "auth_token"),
|
||||
"xyz"
|
||||
);
|
||||
});
|
||||
|
||||
test("getCookieValueFromHeader handles a no-space separator too", () => {
|
||||
assert.equal(getCookieValueFromHeader({ cookie: "a=1;auth_token=tok" }, "auth_token"), "tok");
|
||||
});
|
||||
|
||||
test("getCookieValueFromHeader returns null when the cookie is absent", () => {
|
||||
assert.equal(getCookieValueFromHeader({ cookie: "other=1; foo=2" }, "auth_token"), null);
|
||||
assert.equal(getCookieValueFromHeader({}, "auth_token"), null);
|
||||
});
|
||||
|
||||
test("getCookieValueFromHeader URL-decodes the value", () => {
|
||||
assert.equal(
|
||||
getCookieValueFromHeader({ cookie: "x=1; auth_token=a%20b" }, "auth_token"),
|
||||
"a b"
|
||||
);
|
||||
});
|
||||
Loading…
Add table
Add a link
Reference in a new issue