From 631f5832a1d25159d2e5ae1b8a0b080eaef00b0e Mon Sep 17 00:00:00 2001 From: redshift <213178690+1ftredsh@users.noreply.github.com> Date: Thu, 26 Mar 2026 17:22:03 +0000 Subject: [PATCH] checkpoint --- bun.lock | 4 +- package.json | 2 +- src/daemon/http/index.ts | 188 +++++++++++++++++---------------------- src/daemon/usage.ts | 135 ---------------------------- 4 files changed, 87 insertions(+), 242 deletions(-) delete mode 100644 src/daemon/usage.ts diff --git a/bun.lock b/bun.lock index f12b173..633aa85 100644 --- a/bun.lock +++ b/bun.lock @@ -6,7 +6,7 @@ "name": "routstrd", "dependencies": { "@cashu/cashu-ts": "^3.1.1", - "@routstr/sdk": "^0.2.4", + "@routstr/sdk": "^0.2.5", "applesauce-core": "^5.1.0", "applesauce-relay": "^5.1.0", "commander": "^14.0.2", @@ -30,7 +30,7 @@ "@noble/hashes": ["@noble/hashes@2.0.1", "", {}, "sha512-XlOlEbQcE9fmuXxrVTXCTlG2nlRXa9Rj3rr5Ue/+tX+nmkgbX720YHh0VR3hBF9xDvwnb8D2shVGOwNx+ulArw=="], - "@routstr/sdk": ["@routstr/sdk@0.2.4", "", { "dependencies": { "@cashu/cashu-ts": "^3.1.1", "applesauce-core": "^5.1.0", "applesauce-relay": "^5.1.0", "rxjs": "^7.8.1", "zustand": "^5.0.5" }, "optionalDependencies": { "better-sqlite3": "^11.7.2" }, "peerDependencies": { "typescript": ">=5.0.0" } }, "sha512-QN4ZjvzVc61MjVtCO4eKR8PKcoJomuBK2gv6oJW03IAo/ok++F1cTATWY9Qi2qhHEokmbMQXEAF7AggUWJsfyw=="], + "@routstr/sdk": ["@routstr/sdk@0.2.5", "", { "dependencies": { "@cashu/cashu-ts": "^3.1.1", "applesauce-core": "^5.1.0", "applesauce-relay": "^5.1.0", "rxjs": "^7.8.1", "zustand": "^5.0.5" }, "optionalDependencies": { "better-sqlite3": "^11.7.2" }, "peerDependencies": { "typescript": ">=5.0.0" } }, "sha512-ejITpKDF7cpVaCIMOfWblVgSEc6vkDZTwM7wZWzGaNKVOmw+K/o0hJsAboY3gv9+Dit9PbGE3CaD/wI3CqyMmQ=="], "@scure/base": ["@scure/base@2.0.0", "", {}, "sha512-3E1kpuZginKkek01ovG8krQ0Z44E3DHPjc5S2rjJw9lZn3KSQOs8S7wqikF/AH7iRanHypj85uGyxk0XAyC37w=="], diff --git a/package.json b/package.json index 93f47e3..4169d5c 100644 --- a/package.json +++ b/package.json @@ -23,7 +23,7 @@ }, "dependencies": { "@cashu/cashu-ts": "^3.1.1", - "@routstr/sdk": "^0.2.4", + "@routstr/sdk": "^0.2.5", "applesauce-core": "^5.1.0", "applesauce-relay": "^5.1.0", "commander": "^14.0.2", diff --git a/src/daemon/http/index.ts b/src/daemon/http/index.ts index d233a11..7c8e14a 100644 --- a/src/daemon/http/index.ts +++ b/src/daemon/http/index.ts @@ -1,47 +1,12 @@ import { type IncomingMessage, type ServerResponse } from "http"; import { Readable } from "stream"; -import { ReadableStream as WebReadableStream } from "stream/web"; -import { routeRequests, InsufficientBalanceError } from "@routstr/sdk"; -import { createSSEParserTransform } from "../sse"; import { - createUsageTracker, - extractResponseId, - extractUsageFromResponseBody, - resolveUsageBaseUrl, -} from "../usage"; -import type { UsageData } from "../types"; + routeRequests, + InsufficientBalanceError, + getDefaultUsageTrackingDriver, +} from "@routstr/sdk"; import { logger } from "../../utils/logger"; -/** - * Extracts the client ID from an incoming request by looking up the API key - * in the store's clientIds list. - */ -function getClientIdFromRequest( - req: IncomingMessage, - store: { getState(): any }, -): string | undefined { - const authHeader = req.headers.authorization; - - if (!authHeader || !authHeader.startsWith("Bearer ")) { - return undefined; - } - - const apiKey = authHeader.slice(7); // Remove "Bearer " prefix - - if (!apiKey.startsWith("sk-")) { - return undefined; - } - - const state = store.getState(); - const clientIds = state.clientIds || []; - - const matchingClient = (clientIds as { clientId: string; apiKey: string }[]).find( - (c) => c.apiKey === apiKey, - ); - - return matchingClient?.clientId; -} - async function readBody(req: IncomingMessage): Promise { return new Promise((resolve, reject) => { let data = ""; @@ -75,8 +40,6 @@ export function createDaemonRequestHandler(deps: { parseBalances: (output: string) => Record; mode?: "xcashu" | "lazyrefund" | "apikeys"; }) { - const usageTracker = createUsageTracker(deps.store); - return async function handler(req: IncomingMessage, res: ServerResponse) { const host = req.headers.host || "localhost"; const url = new URL(req.url || "/", `http://${host}`); @@ -327,9 +290,30 @@ export function createDaemonRequestHandler(deps: { if (req.method === "GET" && url.pathname === "/usage") { try { - const output = usageTracker.listRecent(parseLimit(url.searchParams.get("limit"))); + const usageDriver = getDefaultUsageTrackingDriver(); + const limit = parseLimit(url.searchParams.get("limit")); + const entries = await usageDriver.list({ limit }); + const totalEntries = await usageDriver.count(); + const totalSatsCost = ( + await usageDriver.list() + ).reduce((sum, entry) => sum + (entry.satsCost || 0), 0); + const recentSatsCost = entries.reduce( + (sum, entry) => sum + (entry.satsCost || 0), + 0, + ); + res.writeHead(200, { "Content-Type": "application/json" }); - res.end(JSON.stringify({ output })); + res.end( + JSON.stringify({ + output: { + entries, + totalEntries, + totalSatsCost, + recentSatsCost, + limit, + }, + }), + ); } catch (error) { res.writeHead(500, { "Content-Type": "application/json" }); res.end(JSON.stringify({ error: String(error) })); @@ -350,12 +334,37 @@ export function createDaemonRequestHandler(deps: { return; } - const output = usageTracker.listForTimestamp( - timestamp, - parseLimit(url.searchParams.get("limit")), + const usageDriver = getDefaultUsageTrackingDriver(); + const limit = parseLimit(url.searchParams.get("limit")); + const allMatching = await usageDriver.list(); + const requestIdPrefix = `gen-${timestamp}-`; + const filtered = allMatching.filter((entry) => + entry.requestId.startsWith(requestIdPrefix), ); + const entries = filtered.slice(0, limit); + const totalEntries = filtered.length; + const totalSatsCost = filtered.reduce( + (sum, entry) => sum + (entry.satsCost || 0), + 0, + ); + const recentSatsCost = entries.reduce( + (sum, entry) => sum + (entry.satsCost || 0), + 0, + ); + res.writeHead(200, { "Content-Type": "application/json" }); - res.end(JSON.stringify({ output })); + res.end( + JSON.stringify({ + output: { + entries, + totalEntries, + totalSatsCost, + recentSatsCost, + limit, + timestamp, + }, + }), + ); } catch (error) { res.writeHead(500, { "Content-Type": "application/json" }); res.end(JSON.stringify({ error: String(error) })); @@ -415,9 +424,11 @@ export function createDaemonRequestHandler(deps: { }); const isStream = bodyObj.stream === true; - const requestId = response.headers.get("x-routstr-request-id") || undefined; + const requestId = + (response as any).requestId || + response.headers.get("x-routstr-request-id") || + undefined; logger.log("Request ID, ", requestId, " with path: ", url.pathname); - const usageBaseUrl = resolveUsageBaseUrl(response, forcedProvider); if (isStream) { res.statusCode = response.status; @@ -426,67 +437,36 @@ export function createDaemonRequestHandler(deps: { }); const body = response.body; - if (body) { - let capturedUsage: UsageData | null = null; - let capturedResponseId: string | undefined; - const nodeReadable = Readable.fromWeb( - body as unknown as WebReadableStream, - ); - const sseParser = createSSEParserTransform( - (usage) => { - capturedUsage = usage; - }, - (responseId) => { - capturedResponseId = responseId; - }, - ); - nodeReadable.pipe(sseParser).pipe(res); - - res.on("finish", () => { - if (capturedUsage) { - const usageRequestId = capturedResponseId || requestId || "unknown"; - usageTracker.append({ - id: - usageRequestId === "unknown" - ? `req-${Date.now()}-${modelId}` - : usageRequestId, - timestamp: Date.now(), - modelId, - baseUrl: usageBaseUrl, - requestId: usageRequestId, - client: getClientIdFromRequest(req, deps.store), - ...capturedUsage, - }); - logger.log( - "Streaming request usage:", - JSON.stringify(capturedUsage), - ); - } - }); - } else { + if (!body) { res.end(); + return; } + + const nodeReadable = Readable.fromWeb(body as any); + await new Promise((resolve, reject) => { + let settled = false; + const finish = () => { + if (settled) return; + settled = true; + resolve(); + }; + const fail = (err: unknown) => { + if (settled) return; + settled = true; + reject(err); + }; + + res.once("finish", finish); + res.once("close", finish); + res.once("error", fail); + nodeReadable.once("error", fail); + + nodeReadable.pipe(res); + }); return; } const responseBody = await response.json(); - const nonStreamUsage = extractUsageFromResponseBody(responseBody); - if (nonStreamUsage) { - const responseRequestId = - extractResponseId(responseBody) || requestId || "unknown"; - usageTracker.append({ - id: - responseRequestId === "unknown" - ? `req-${Date.now()}-${modelId}` - : responseRequestId, - timestamp: Date.now(), - modelId, - baseUrl: usageBaseUrl, - requestId: responseRequestId, - client: getClientIdFromRequest(req, deps.store), - ...nonStreamUsage, - }); - } res.writeHead(response.status, { "Content-Type": "application/json", }); diff --git a/src/daemon/usage.ts b/src/daemon/usage.ts deleted file mode 100644 index ed639cf..0000000 --- a/src/daemon/usage.ts +++ /dev/null @@ -1,135 +0,0 @@ -import type { UsageData, UsageTrackingEntry } from "./types"; -import { logger } from "../utils/logger"; - -export function extractUsageFromResponseBody(body: unknown): UsageData | null { - if (!body || typeof body !== "object") return null; - const usage = (body as { usage?: Record }).usage; - if (!usage || typeof usage !== "object") return null; - - const promptTokens = Number(usage.prompt_tokens ?? 0); - const completionTokens = Number(usage.completion_tokens ?? 0); - const totalTokens = Number(usage.total_tokens ?? 0); - const costValue = usage.cost; - - let cost = 0; - let satsCost = 0; - - if (typeof costValue === "number") { - cost = costValue; - } else if (costValue && typeof costValue === "object") { - const costObj = costValue as Record; - const totalUsd = costObj.total_usd; - const totalMsats = costObj.total_msats; - - cost = typeof totalUsd === "number" ? totalUsd : 0; - satsCost = typeof totalMsats === "number" ? totalMsats / 1000 : 0; - } - - if ( - promptTokens === 0 && - completionTokens === 0 && - totalTokens === 0 && - cost === 0 && - satsCost === 0 - ) { - return null; - } - - return { - promptTokens, - completionTokens, - totalTokens, - cost, - satsCost, - }; -} - -export function resolveUsageBaseUrl(response: Response, fallback?: string): string { - const responseWithBaseUrl = response as Response & { baseUrl?: unknown }; - if (typeof responseWithBaseUrl.baseUrl === "string") { - return responseWithBaseUrl.baseUrl; - } - - try { - if (response.url) { - const parsed = new URL(response.url); - return `${parsed.protocol}//${parsed.host}`; - } - } catch { - // Ignore URL parsing failures. - } - - return fallback || "unknown"; -} - -export function extractResponseId(body: unknown): string | undefined { - if (!body || typeof body !== "object") return undefined; - const id = (body as { id?: unknown }).id; - if (typeof id !== "string") return undefined; - const trimmed = id.trim(); - return trimmed.length > 0 ? trimmed : undefined; -} - -export function createUsageTracker(store: { getState(): any }) { - const append = (entry: UsageTrackingEntry): void => { - const state = store.getState(); - const nextUsage = [...(state.usageTracking || []), entry]; - state.setUsageTracking(nextUsage); - logger.log("Usage tracking saved:", JSON.stringify(entry)); - }; - - const listRecent = (limit: number) => { - const usageTracking = - ((store.getState().usageTracking || []) as UsageTrackingEntry[]) || []; - const recent = usageTracking.slice(-limit).reverse(); - const totalSatsCost = usageTracking.reduce( - (sum, entry) => sum + (entry.satsCost || 0), - 0, - ); - const recentSatsCost = recent.reduce( - (sum, entry) => sum + (entry.satsCost || 0), - 0, - ); - - return { - entries: recent, - totalEntries: usageTracking.length, - totalSatsCost, - recentSatsCost, - limit, - }; - }; - - const listForTimestamp = (timestamp: string, limit: number) => { - const usageTracking = - ((store.getState().usageTracking || []) as UsageTrackingEntry[]) || []; - const requestIdPrefix = `gen-${timestamp}-`; - const filteredUsage = usageTracking.filter((entry) => - entry.requestId.startsWith(requestIdPrefix), - ); - const recent = filteredUsage.slice(-limit).reverse(); - const totalSatsCost = filteredUsage.reduce( - (sum, entry) => sum + (entry.satsCost || 0), - 0, - ); - const recentSatsCost = recent.reduce( - (sum, entry) => sum + (entry.satsCost || 0), - 0, - ); - - return { - entries: recent, - totalEntries: filteredUsage.length, - totalSatsCost, - recentSatsCost, - limit, - timestamp, - }; - }; - - return { - append, - listRecent, - listForTimestamp, - }; -}