fix(cli-proxy): durable analytics history persistence and dedupe migration

This commit is contained in:
Tam Nhu Tran committed 2026-04-26 16:44:39 -04:00
1 parent 286b5781a7
commit 69b75c0449
5 files changed
+525 -117

No files matched your search

+174 -86
View File
@@ -12,30 +12,43 @@ import * as fs from 'fs';
import * as path from 'path';
import { fetchCliproxyUsageRaw } from '../../cliproxy/stats-fetcher';
import {
transformCliproxyToDailyUsage,
transformCliproxyToHourlyUsage,
transformCliproxyToMonthlyUsage,
buildCliproxyUsageHistoryAggregates,
extractCliproxyUsageHistoryDetails,
mergeCliproxyUsageHistoryDetails,
pruneCliproxyUsageHistoryDetails,
type CliproxyUsageHistoryDetail,
} from './cliproxy-usage-transformer';
import type { DailyUsage, HourlyUsage, MonthlyUsage } from './types';
import { getCcsDir } from '../../utils/config-manager';
import { ok, info, warn } from '../../utils/ui';
// ---------------------------------------------------------------------------
// Snapshot format
// ---------------------------------------------------------------------------
interface CliproxyUsageSnapshot {
version: number;
timestamp: number;
details: CliproxyUsageHistoryDetail[];
daily: DailyUsage[];
hourly: HourlyUsage[];
monthly: MonthlyUsage[];
}
type LegacyCliproxyUsageSnapshot = {
version?: number;
timestamp?: number;
daily?: DailyUsage[];
hourly?: HourlyUsage[];
monthly?: MonthlyUsage[];
};
type FetchCliproxyUsageRaw = typeof fetchCliproxyUsageRaw;
const SNAPSHOT_VERSION = 2;
const SNAPSHOT_VERSION = 3;
const SNAPSHOT_MAX_AGE_MS = 7 * 24 * 60 * 60 * 1000;
const HISTORY_RETENTION_DAYS = Math.max(
30,
parseInt(process.env.CCS_CLIPROXY_HISTORY_RETENTION_DAYS ?? '365', 10) || 365
);
const HISTORY_RETENTION_MS = HISTORY_RETENTION_DAYS * 24 * 60 * 60 * 1000;
const MAX_WRITE_ATTEMPTS = 3;
/** Sync interval in ms, configurable via CCS_CLIPROXY_SYNC_INTERVAL env var (default: 5 min) */
const SYNC_INTERVAL_MS = Math.max(
@@ -43,14 +56,9 @@ const SYNC_INTERVAL_MS = Math.max(
parseInt(process.env.CCS_CLIPROXY_SYNC_INTERVAL ?? '300000', 10) || 300_000
);
// Module-level interval ID
let syncIntervalId: ReturnType<typeof setInterval> | null = null;
let snapshotTimestampOrdinal = 0;
// ---------------------------------------------------------------------------
// Cache directory helpers
// ---------------------------------------------------------------------------
function getCliproxyCacheDir(): string {
return path.join(getCcsDir(), 'cache', 'cliproxy-usage');
}
@@ -71,6 +79,88 @@ function getSnapshotTimestamp(): number {
return Date.now() + snapshotTimestampOrdinal / 1000;
}
function buildHourlyTimestamp(hour: string): string {
return `${hour.replace(' ', 'T')}:00.000Z`;
}
function buildDailyTimestamp(date: string): string {
return `${date}T00:00:00.000Z`;
}
function buildLegacyHistoryDetails(
snapshot: LegacyCliproxyUsageSnapshot
): CliproxyUsageHistoryDetail[] {
const details: CliproxyUsageHistoryDetail[] = [];
const coveredDailyKeys = new Set<string>();
for (const hour of snapshot.hourly ?? []) {
for (const breakdown of hour.modelBreakdowns) {
details.push({
model: breakdown.modelName,
timestamp: buildHourlyTimestamp(hour.hour),
source: hour.source,
authIndex: 'legacy-hourly',
inputTokens: breakdown.inputTokens,
outputTokens: breakdown.outputTokens,
cacheReadTokens: breakdown.cacheReadTokens,
failed: false,
});
coveredDailyKeys.add(`${hour.hour.slice(0, 10)}|${breakdown.modelName}`);
}
}
for (const day of snapshot.daily ?? []) {
for (const breakdown of day.modelBreakdowns) {
const key = `${day.date}|${breakdown.modelName}`;
if (coveredDailyKeys.has(key)) {
continue;
}
details.push({
model: breakdown.modelName,
timestamp: buildDailyTimestamp(day.date),
source: day.source,
authIndex: 'legacy-daily',
inputTokens: breakdown.inputTokens,
outputTokens: breakdown.outputTokens,
cacheReadTokens: breakdown.cacheReadTokens,
failed: false,
});
}
}
return details;
}
function migrateLegacySnapshot(
snapshot: LegacyCliproxyUsageSnapshot,
emitWarnings: boolean
): CliproxyUsageSnapshot | null {
const details = buildLegacyHistoryDetails(snapshot);
if (details.length === 0) {
if (emitWarnings) {
console.log(
info('CLIProxy legacy snapshot had no migratable history, will refresh on next sync')
);
}
return null;
}
if (emitWarnings) {
console.log(info('Loaded legacy CLIProxy snapshot into historical format'));
}
const { daily, hourly, monthly } = buildCliproxyUsageHistoryAggregates(details);
return {
version: SNAPSHOT_VERSION,
timestamp: Number.isFinite(snapshot.timestamp) ? Number(snapshot.timestamp) : Date.now(),
details,
daily,
hourly,
monthly,
};
}
function readSnapshot(emitWarnings = true): CliproxyUsageSnapshot | null {
try {
const snapshotPath = getLatestSnapshotPath();
@@ -79,23 +169,34 @@ function readSnapshot(emitWarnings = true): CliproxyUsageSnapshot | null {
}
const raw = fs.readFileSync(snapshotPath, 'utf-8');
const snapshot = JSON.parse(raw) as CliproxyUsageSnapshot;
const snapshot = JSON.parse(raw) as CliproxyUsageSnapshot | LegacyCliproxyUsageSnapshot;
if (snapshot.version !== SNAPSHOT_VERSION) {
if (emitWarnings) {
console.log(info('CLIProxy snapshot version mismatch, will refresh on next sync'));
if (snapshot.version === SNAPSHOT_VERSION) {
if (!Number.isFinite(snapshot.timestamp)) {
if (emitWarnings) {
console.log(info('CLIProxy snapshot timestamp invalid, will refresh on next sync'));
}
return null;
}
return null;
if (!Array.isArray((snapshot as CliproxyUsageSnapshot).details)) {
if (emitWarnings) {
console.log(info('CLIProxy snapshot details missing, will refresh on next sync'));
}
return null;
}
return snapshot as CliproxyUsageSnapshot;
}
if (!Number.isFinite(snapshot.timestamp)) {
if (emitWarnings) {
console.log(info('CLIProxy snapshot timestamp invalid, will refresh on next sync'));
}
return null;
if (snapshot.version === 2) {
return migrateLegacySnapshot(snapshot as LegacyCliproxyUsageSnapshot, emitWarnings);
}
return snapshot;
if (emitWarnings) {
console.log(info('CLIProxy snapshot version mismatch, will refresh on next sync'));
}
return null;
} catch (err) {
if (emitWarnings) {
console.log(warn('Failed to read CLIProxy snapshot:') + ` ${(err as Error).message}`);
@@ -104,14 +205,55 @@ function readSnapshot(emitWarnings = true): CliproxyUsageSnapshot | null {
}
}
// ---------------------------------------------------------------------------
// Load cached data
// ---------------------------------------------------------------------------
function buildSnapshot(
baseDetails: CliproxyUsageHistoryDetail[],
incomingDetails: CliproxyUsageHistoryDetail[]
): CliproxyUsageSnapshot {
const mergedDetails = pruneCliproxyUsageHistoryDetails(
mergeCliproxyUsageHistoryDetails(baseDetails, incomingDetails),
Date.now() - HISTORY_RETENTION_MS
);
const { daily, hourly, monthly } = buildCliproxyUsageHistoryAggregates(mergedDetails);
return {
version: SNAPSHOT_VERSION,
timestamp: getSnapshotTimestamp(),
details: mergedDetails,
daily,
hourly,
monthly,
};
}
async function writeSnapshotWithMerge(
incomingDetails: CliproxyUsageHistoryDetail[]
): Promise<void> {
ensureCliproxyCacheDir();
const snapshotPath = getLatestSnapshotPath();
for (let attempt = 0; attempt < MAX_WRITE_ATTEMPTS; attempt++) {
const baseSnapshot = readSnapshot(false);
const baseTimestamp = baseSnapshot?.timestamp ?? -Infinity;
const snapshot = buildSnapshot(baseSnapshot?.details ?? [], incomingDetails);
const tempFile = `${snapshotPath}.${process.pid}.${snapshot.timestamp}.tmp`;
fs.writeFileSync(tempFile, JSON.stringify(snapshot), 'utf-8');
const latestSnapshot = readSnapshot(false);
const latestTimestamp = latestSnapshot?.timestamp ?? -Infinity;
if (latestTimestamp > baseTimestamp) {
fs.rmSync(tempFile, { force: true });
continue;
}
fs.renameSync(tempFile, snapshotPath);
console.log(ok('CLIProxy usage snapshot updated'));
return;
}
throw new Error('Failed to write CLIProxy snapshot after repeated overlap retries');
}
/**
* Read the latest CLIProxy usage snapshot from disk.
* Returns empty arrays on failure (file not found, parse error, version mismatch).
*/
export async function loadCachedCliproxyData(): Promise<{
daily: DailyUsage[];
hourly: HourlyUsage[];
@@ -132,18 +274,9 @@ export async function loadCachedCliproxyData(): Promise<{
return { daily: snapshot.daily, hourly: snapshot.hourly, monthly: snapshot.monthly };
}
// ---------------------------------------------------------------------------
// Sync
// ---------------------------------------------------------------------------
/**
* Fetch latest CLIProxy usage data and persist a snapshot to disk.
* Non-fatal: logs warning and returns early if CLIProxy is unavailable.
*/
export async function syncCliproxyUsage(
fetchRaw: FetchCliproxyUsageRaw = fetchCliproxyUsageRaw
): Promise<void> {
const syncStartedAt = getSnapshotTimestamp();
const raw = await fetchRaw();
if (raw === null) {
@@ -152,53 +285,12 @@ export async function syncCliproxyUsage(
}
try {
ensureCliproxyCacheDir();
const daily = transformCliproxyToDailyUsage(raw);
const hourly = transformCliproxyToHourlyUsage(raw);
const monthly = transformCliproxyToMonthlyUsage(raw);
const snapshot: CliproxyUsageSnapshot = {
version: SNAPSHOT_VERSION,
timestamp: syncStartedAt,
daily,
hourly,
monthly,
};
const snapshotPath = getLatestSnapshotPath();
const tempFile = `${snapshotPath}.${process.pid}.${syncStartedAt}.tmp`;
const currentSnapshot = readSnapshot(false);
if (currentSnapshot && currentSnapshot.timestamp > snapshot.timestamp) {
console.log(info('Skipping stale CLIProxy snapshot write'));
return;
}
fs.writeFileSync(tempFile, JSON.stringify(snapshot), 'utf-8');
const latestSnapshot = readSnapshot(false);
if (latestSnapshot && latestSnapshot.timestamp > snapshot.timestamp) {
fs.rmSync(tempFile, { force: true });
console.log(info('Skipping stale CLIProxy snapshot write'));
return;
}
fs.renameSync(tempFile, snapshotPath);
console.log(ok('CLIProxy usage snapshot updated'));
await writeSnapshotWithMerge(extractCliproxyUsageHistoryDetails(raw));
} catch (err) {
// Non-fatal - stale snapshot will continue to be served
console.log(warn('Failed to write CLIProxy snapshot:') + ` ${(err as Error).message}`);
}
}
// ---------------------------------------------------------------------------
// Interval management
// ---------------------------------------------------------------------------
/**
* Start periodic CLIProxy usage sync (every 5 minutes).
* Performs an immediate sync on startup.
*/
export function startCliproxySync(syncNow: () => Promise<void> = () => syncCliproxyUsage()): void {
if (syncIntervalId !== null) {
return;
@@ -207,7 +299,6 @@ export function startCliproxySync(syncNow: () => Promise<void> = () => syncClipr
const intervalMin = Math.round(SYNC_INTERVAL_MS / 60_000);
console.log(info(`Starting CLIProxy usage sync (interval: ${intervalMin} min)`));
// Fire-and-forget initial sync
void syncNow();
syncIntervalId = setInterval(() => {
@@ -215,9 +306,6 @@ export function startCliproxySync(syncNow: () => Promise<void> = () => syncClipr
}, SYNC_INTERVAL_MS);
}
/**
* Stop periodic CLIProxy usage sync.
*/
export function stopCliproxySync(): void {
if (syncIntervalId !== null) {
clearInterval(syncIntervalId);
@@ -13,10 +13,16 @@ import type { ModelBreakdown, DailyUsage, HourlyUsage, MonthlyUsage } from './ty
// INTERNAL HELPERS
// ============================================================================
/** Flat entry pairing a model name with its request detail */
interface FlatDetail {
/** Persisted request detail used to rebuild historical CLIProxy analytics buckets */
export interface CliproxyUsageHistoryDetail {
model: string;
detail: CliproxyRequestDetail;
timestamp: string;
source: string;
authIndex: string;
inputTokens: number;
outputTokens: number;
cacheReadTokens: number;
failed: boolean;
}
/** Accumulator for token counts per model per time bucket */
@@ -36,6 +42,22 @@ function buildModelBreakdown(modelName: string, acc: ModelAccumulator): ModelBre
return { modelName, inputTokens, outputTokens, cacheCreationTokens: 0, cacheReadTokens, cost };
}
function createHistoryDetail(
model: string,
detail: CliproxyRequestDetail
): CliproxyUsageHistoryDetail {
return {
model,
timestamp: detail.timestamp,
source: detail.source,
authIndex: String(detail.auth_index),
inputTokens: detail.tokens?.input_tokens ?? 0,
outputTokens: detail.tokens?.output_tokens ?? 0,
cacheReadTokens: detail.tokens?.cached_tokens ?? 0,
failed: detail.failed,
};
}
// ============================================================================
// FLATTEN
// ============================================================================
@@ -51,14 +73,16 @@ function hasTrackedUsage(detail: CliproxyRequestDetail): boolean {
/**
* Flatten the nested response.usage.apis[provider].models[model].details[]
* structure into a flat array. Failed requests are retained only when they
* still report tracked token usage that analytics can account for.
* structure into normalized history details. Failed requests are retained only
* when they still report tracked token usage that analytics can account for.
*/
export function flattenCliproxyDetails(response: CliproxyUsageApiResponse): FlatDetail[] {
export function extractCliproxyUsageHistoryDetails(
response: CliproxyUsageApiResponse
): CliproxyUsageHistoryDetail[] {
const apis = response?.usage?.apis;
if (!apis) return [];
const results: FlatDetail[] = [];
const results: CliproxyUsageHistoryDetail[] = [];
for (const providerData of Object.values(apis)) {
const models = providerData?.models;
if (!models) continue;
@@ -67,20 +91,89 @@ export function flattenCliproxyDetails(response: CliproxyUsageApiResponse): Flat
if (!details) continue;
for (const detail of details) {
if (detail.failed && !hasTrackedUsage(detail)) continue;
results.push({ model, detail });
results.push(createHistoryDetail(model, detail));
}
}
}
return results;
}
function createHistorySignature(detail: CliproxyUsageHistoryDetail): string {
return [
detail.model,
detail.timestamp,
detail.source,
detail.authIndex,
detail.inputTokens,
detail.outputTokens,
detail.cacheReadTokens,
detail.failed ? '1' : '0',
].join('|');
}
export function mergeCliproxyUsageHistoryDetails(
existing: CliproxyUsageHistoryDetail[],
incoming: CliproxyUsageHistoryDetail[]
): CliproxyUsageHistoryDetail[] {
const existingCounts = new Map<string, { detail: CliproxyUsageHistoryDetail; count: number }>();
for (const detail of existing) {
const signature = createHistorySignature(detail);
const entry = existingCounts.get(signature);
if (entry) {
entry.count += 1;
} else {
existingCounts.set(signature, { detail, count: 1 });
}
}
const incomingCounts = new Map<string, { detail: CliproxyUsageHistoryDetail; count: number }>();
for (const detail of incoming) {
const signature = createHistorySignature(detail);
const entry = incomingCounts.get(signature);
if (entry) {
entry.count += 1;
} else {
incomingCounts.set(signature, { detail, count: 1 });
}
}
for (const [signature, incomingEntry] of incomingCounts) {
const existingEntry = existingCounts.get(signature);
if (!existingEntry || incomingEntry.count > existingEntry.count) {
existingCounts.set(signature, {
detail: incomingEntry.detail,
count: incomingEntry.count,
});
}
}
const merged: CliproxyUsageHistoryDetail[] = [];
for (const { detail, count } of existingCounts.values()) {
for (let index = 0; index < count; index++) {
merged.push({ ...detail });
}
}
return merged.sort((a, b) => a.timestamp.localeCompare(b.timestamp));
}
export function pruneCliproxyUsageHistoryDetails(
details: CliproxyUsageHistoryDetail[],
oldestTimestamp: number
): CliproxyUsageHistoryDetail[] {
return details.filter((detail) => {
const timestamp = Date.parse(detail.timestamp);
return Number.isFinite(timestamp) && timestamp >= oldestTimestamp;
});
}
// ============================================================================
// GENERIC AGGREGATOR
// ============================================================================
/** Group flat details by a time key extractor, return sorted DailyUsage-like records */
function aggregateByKey<T>(
flat: FlatDetail[],
flat: CliproxyUsageHistoryDetail[],
keyFn: (timestamp: string) => string,
buildRecord: (key: string, breakdowns: ModelBreakdown[], requestCount: number) => T,
sortFn: (a: T, b: T) => number
@@ -89,18 +182,18 @@ function aggregateByKey<T>(
const buckets = new Map<string, Map<string, ModelAccumulator>>();
const requestCounts = new Map<string, number>();
for (const { model, detail } of flat) {
for (const detail of flat) {
const key = keyFn(detail.timestamp);
if (!buckets.has(key)) buckets.set(key, new Map());
requestCounts.set(key, (requestCounts.get(key) ?? 0) + 1);
const modelMap = buckets.get(key) as Map<string, ModelAccumulator>;
if (!modelMap.has(model)) {
modelMap.set(model, { inputTokens: 0, outputTokens: 0, cacheReadTokens: 0 });
if (!modelMap.has(detail.model)) {
modelMap.set(detail.model, { inputTokens: 0, outputTokens: 0, cacheReadTokens: 0 });
}
const acc = modelMap.get(model) as ModelAccumulator;
acc.inputTokens += detail.tokens?.input_tokens ?? 0;
acc.outputTokens += detail.tokens?.output_tokens ?? 0;
acc.cacheReadTokens += detail.tokens?.cached_tokens ?? 0;
const acc = modelMap.get(detail.model) as ModelAccumulator;
acc.inputTokens += detail.inputTokens;
acc.outputTokens += detail.outputTokens;
acc.cacheReadTokens += detail.cacheReadTokens;
}
const records: T[] = [];
@@ -125,7 +218,7 @@ function sumField(breakdowns: ModelBreakdown[], field: keyof ModelBreakdown): nu
/** Transform CLIProxy usage response into DailyUsage array (sorted descending by date) */
export function transformCliproxyToDailyUsage(response: CliproxyUsageApiResponse): DailyUsage[] {
const flat = flattenCliproxyDetails(response);
const flat = extractCliproxyUsageHistoryDetails(response);
return aggregateByKey(
flat,
(ts) => ts.slice(0, 10),
@@ -150,7 +243,7 @@ export function transformCliproxyToDailyUsage(response: CliproxyUsageApiResponse
/** Transform CLIProxy usage response into HourlyUsage array (sorted descending by hour) */
export function transformCliproxyToHourlyUsage(response: CliproxyUsageApiResponse): HourlyUsage[] {
const flat = flattenCliproxyDetails(response);
const flat = extractCliproxyUsageHistoryDetails(response);
return aggregateByKey(
flat,
(ts) => {
@@ -182,7 +275,7 @@ export function transformCliproxyToHourlyUsage(response: CliproxyUsageApiRespons
export function transformCliproxyToMonthlyUsage(
response: CliproxyUsageApiResponse
): MonthlyUsage[] {
const flat = flattenCliproxyDetails(response);
const flat = extractCliproxyUsageHistoryDetails(response);
return aggregateByKey(
flat,
(ts) => ts.slice(0, 7),
@@ -200,3 +293,73 @@ export function transformCliproxyToMonthlyUsage(
(a, b) => b.month.localeCompare(a.month)
);
}
export function buildCliproxyUsageHistoryAggregates(details: CliproxyUsageHistoryDetail[]): {
daily: DailyUsage[];
hourly: HourlyUsage[];
monthly: MonthlyUsage[];
} {
return {
daily: aggregateByKey(
details,
(timestamp) => timestamp.slice(0, 10),
(date, breakdowns) => {
const totalCost = sumField(breakdowns, 'cost');
return {
date,
source: 'cliproxy',
inputTokens: sumField(breakdowns, 'inputTokens'),
outputTokens: sumField(breakdowns, 'outputTokens'),
cacheCreationTokens: 0,
cacheReadTokens: sumField(breakdowns, 'cacheReadTokens'),
cost: totalCost,
totalCost,
modelsUsed: breakdowns.map((breakdown) => breakdown.modelName),
modelBreakdowns: breakdowns,
};
},
(a, b) => b.date.localeCompare(a.date)
),
hourly: aggregateByKey(
details,
(timestamp) => {
const date = timestamp.slice(0, 10);
const hour = timestamp.slice(11, 13) || '00';
return `${date} ${hour}:00`;
},
(hour, breakdowns, requestCount) => {
const totalCost = sumField(breakdowns, 'cost');
return {
hour,
source: 'cliproxy',
inputTokens: sumField(breakdowns, 'inputTokens'),
outputTokens: sumField(breakdowns, 'outputTokens'),
cacheCreationTokens: 0,
cacheReadTokens: sumField(breakdowns, 'cacheReadTokens'),
cost: totalCost,
totalCost,
modelsUsed: breakdowns.map((breakdown) => breakdown.modelName),
modelBreakdowns: breakdowns,
requestCount,
};
},
(a, b) => b.hour.localeCompare(a.hour)
),
monthly: aggregateByKey(
details,
(timestamp) => timestamp.slice(0, 7),
(month, breakdowns) => ({
month,
source: 'cliproxy',
inputTokens: sumField(breakdowns, 'inputTokens'),
outputTokens: sumField(breakdowns, 'outputTokens'),
cacheCreationTokens: 0,
cacheReadTokens: sumField(breakdowns, 'cacheReadTokens'),
totalCost: sumField(breakdowns, 'cost'),
modelsUsed: breakdowns.map((breakdown) => breakdown.modelName),
modelBreakdowns: breakdowns,
}),
(a, b) => b.month.localeCompare(a.month)
),
};
}