laantungir/routstrd
1
0
Fork 0
mirror of https://github.com/Routstr/routstrd.git synced 2026-08-10 04:07:00 +00:00
Code Issues Packages Projects Releases Wiki Activity

checkpoint

Browse Source
This commit is contained in:
redshift
2026-03-26 17:22:03 +00:00
parent 3ad6339822
commit 631f5832a1
4 changed files with 87 additions and 242 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bun.lock
+2 -2
View File
@@ -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=="],
package.json
+1 -1
View File
@@ -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",
src/daemon/http/index.ts
+84 -104
View File
@@ -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<string> {
return new Promise((resolve, reject) => {
let data = "";
@@ -75,8 +40,6 @@ export function createDaemonRequestHandler(deps: {
parseBalances: (output: string) => Record<string, number>;
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<void>((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",
});
src/daemon/usage.ts
-135
View File
@@ -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<string, unknown> }).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<string, unknown>;
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,
};
}