feat: add conversation search and usage dashboard
This commit is contained in:
shiyue
859
lib/usage-statistics.js
Normal file
859
lib/usage-statistics.js
Normal file
@@ -0,0 +1,859 @@
|
||||
'use strict';
|
||||
|
||||
const fsp = require('fs/promises');
|
||||
const path = require('path');
|
||||
|
||||
const SCHEMA_VERSION = 1;
|
||||
const CACHE_VERSION = 1;
|
||||
const DEFAULT_MAX_FILE_BYTES = 32 * 1024 * 1024;
|
||||
const DEFAULT_MAX_RANGE_DAYS = 370;
|
||||
const DEFAULT_RECENT_LIMIT = 50;
|
||||
const DEFAULT_MCP_TOOL_LIMIT = 200;
|
||||
const DEFAULT_SKILL_LIMIT = 100;
|
||||
const DEFAULT_MCP_DETAIL_LIMIT = 250;
|
||||
const BUILD_BATCH_SIZE = 1;
|
||||
const UPSERT_DEBOUNCE_MS = 300;
|
||||
const PERSIST_DEBOUNCE_MS = 600;
|
||||
const SESSION_ID_RE = /^[a-zA-Z0-9-]+$/;
|
||||
|
||||
class UsageStatisticsError extends Error {
|
||||
constructor(code, message) {
|
||||
super(message);
|
||||
this.name = 'UsageStatisticsError';
|
||||
this.code = code;
|
||||
}
|
||||
}
|
||||
|
||||
function isObject(value) {
|
||||
return !!value && typeof value === 'object' && !Array.isArray(value);
|
||||
}
|
||||
|
||||
function clampInteger(value, fallback, min, max) {
|
||||
const number = Number.parseInt(String(value ?? ''), 10);
|
||||
if (!Number.isFinite(number)) return fallback;
|
||||
return Math.max(min, Math.min(max, number));
|
||||
}
|
||||
|
||||
function cleanDisplayText(value, maxChars = 240) {
|
||||
if (typeof value !== 'string') return '';
|
||||
return value
|
||||
.replace(/[\u0000-\u001f\u007f]+/g, ' ')
|
||||
.replace(/\s+/g, ' ')
|
||||
.trim()
|
||||
.slice(0, maxChars);
|
||||
}
|
||||
|
||||
function sanitizeSessionId(value) {
|
||||
return String(value || '').replace(/[^a-zA-Z0-9-]/g, '');
|
||||
}
|
||||
|
||||
function legalSessionId(value) {
|
||||
return !!value && SESSION_ID_RE.test(String(value));
|
||||
}
|
||||
|
||||
function safeTimestamp(value) {
|
||||
if (!value) return null;
|
||||
const timestamp = new Date(value).getTime();
|
||||
return Number.isFinite(timestamp) ? timestamp : null;
|
||||
}
|
||||
|
||||
function safeIso(value, fallbackMs = null) {
|
||||
const timestamp = safeTimestamp(value);
|
||||
if (timestamp !== null) return new Date(timestamp).toISOString();
|
||||
return Number.isFinite(fallbackMs) ? new Date(fallbackMs).toISOString() : null;
|
||||
}
|
||||
|
||||
function basenameFromCwd(value) {
|
||||
const cwd = cleanDisplayText(value, 1200);
|
||||
return cwd ? path.basename(cwd.replace(/[\\/]+$/, '')) : '';
|
||||
}
|
||||
|
||||
function normalizeAgent(value) {
|
||||
return cleanDisplayText(String(value || 'codex'), 40).toLowerCase() || 'codex';
|
||||
}
|
||||
|
||||
function parseMaybeObject(value) {
|
||||
if (isObject(value)) return value;
|
||||
if (typeof value !== 'string') return null;
|
||||
try {
|
||||
const parsed = JSON.parse(value);
|
||||
return isObject(parsed) ? parsed : null;
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
function normalizeMcpStatus(value, done = false) {
|
||||
const status = String(value || '').trim().toLowerCase().replace(/[\s_-]+/g, '');
|
||||
if (status === 'failed' || status === 'error') return 'failed';
|
||||
if (status === 'completed' || status === 'complete' || status === 'succeeded' || status === 'success') {
|
||||
return 'completed';
|
||||
}
|
||||
if (status === 'cancelled' || status === 'canceled') return 'cancelled';
|
||||
if (status === 'inprogress' || status === 'running' || status === 'pending') return 'in_progress';
|
||||
return done ? 'completed' : 'in_progress';
|
||||
}
|
||||
|
||||
function splitMcpSubtitle(value) {
|
||||
const subtitle = cleanDisplayText(value, 320);
|
||||
const index = subtitle.indexOf('.');
|
||||
if (index <= 0 || index >= subtitle.length - 1) return null;
|
||||
return {
|
||||
server: subtitle.slice(0, index),
|
||||
tool: subtitle.slice(index + 1),
|
||||
};
|
||||
}
|
||||
|
||||
function extractMcpToolCall(toolCall) {
|
||||
if (!isObject(toolCall)) return null;
|
||||
const meta = isObject(toolCall.meta) ? toolCall.meta : {};
|
||||
const compactKind = String(toolCall.kind || meta.kind || '').toLowerCase().replace(/[\s_-]+/g, '');
|
||||
const compactName = String(toolCall.name || '').toLowerCase().replace(/[\s_-]+/g, '');
|
||||
const input = parseMaybeObject(toolCall.input) || {};
|
||||
const looksLikeMcp = compactKind === 'mcptoolcall'
|
||||
|| compactName === 'mcptoolcall'
|
||||
|| (typeof toolCall.name === 'string' && toolCall.name.startsWith('mcp__'))
|
||||
|| (!!input.server && !!input.tool);
|
||||
if (!looksLikeMcp) return null;
|
||||
|
||||
let server = cleanDisplayText(String(input.server || ''), 160);
|
||||
let tool = cleanDisplayText(String(input.tool || ''), 200);
|
||||
if ((!server || !tool) && typeof toolCall.name === 'string' && toolCall.name.startsWith('mcp__')) {
|
||||
const parts = toolCall.name.split('__');
|
||||
if (parts.length >= 3) {
|
||||
server ||= cleanDisplayText(parts[1], 160);
|
||||
tool ||= cleanDisplayText(parts.slice(2).join('__'), 200);
|
||||
}
|
||||
}
|
||||
if (!server || !tool) {
|
||||
const subtitle = splitMcpSubtitle(meta.subtitle);
|
||||
if (subtitle) {
|
||||
server ||= subtitle.server;
|
||||
tool ||= subtitle.tool;
|
||||
}
|
||||
}
|
||||
if (!server || !tool) return null;
|
||||
return {
|
||||
server,
|
||||
tool,
|
||||
status: normalizeMcpStatus(meta.status || toolCall.status, !!toolCall.done),
|
||||
};
|
||||
}
|
||||
|
||||
function extractSkillMention(mention) {
|
||||
if (!isObject(mention) || String(mention.kind || '').toLowerCase() !== 'skill') return null;
|
||||
const name = cleanDisplayText(String(mention.name || mention.title || mention.label || ''), 160).replace(/^\$/, '');
|
||||
if (!name) return null;
|
||||
return {
|
||||
name,
|
||||
label: cleanDisplayText(String(mention.label || mention.title || `$${name}`), 180) || `$${name}`,
|
||||
};
|
||||
}
|
||||
|
||||
function fileFingerprint(stat) {
|
||||
return {
|
||||
dev: Number(stat?.dev) || 0,
|
||||
ino: Number(stat?.ino) || 0,
|
||||
size: Number(stat?.size) || 0,
|
||||
mtimeMs: Number(stat?.mtimeMs) || 0,
|
||||
};
|
||||
}
|
||||
|
||||
function sameFingerprint(left, right) {
|
||||
return !!left
|
||||
&& !!right
|
||||
&& left.dev === right.dev
|
||||
&& left.ino === right.ino
|
||||
&& left.size === right.size
|
||||
&& left.mtimeMs === right.mtimeMs;
|
||||
}
|
||||
|
||||
function createUsageDocument(session, stat, fallbackId) {
|
||||
const source = isObject(session) ? session : {};
|
||||
const fallback = sanitizeSessionId(fallbackId);
|
||||
const sourceId = sanitizeSessionId(source.id);
|
||||
const sessionId = legalSessionId(fallback) ? fallback : sourceId;
|
||||
if (!legalSessionId(sessionId)) return null;
|
||||
|
||||
const created = safeIso(source.created || source.createdAt, stat?.birthtimeMs || stat?.mtimeMs);
|
||||
const updated = safeIso(source.updated || source.updatedAt, stat?.mtimeMs) || created;
|
||||
const cwd = cleanDisplayText(String(source.cwd || ''), 1200);
|
||||
const title = cleanDisplayText(String(source.title || '未命名会话'), 240) || '未命名会话';
|
||||
const events = [];
|
||||
const messages = Array.isArray(source.messages) ? source.messages : [];
|
||||
|
||||
for (let messageIndex = 0; messageIndex < messages.length; messageIndex += 1) {
|
||||
const message = messages[messageIndex];
|
||||
if (!isObject(message)) continue;
|
||||
const timestampMs = safeTimestamp(message.timestamp || message.created || message.createdAt);
|
||||
if (timestampMs === null) continue;
|
||||
const timestamp = new Date(timestampMs).toISOString();
|
||||
const role = String(message.role || '').toLowerCase();
|
||||
|
||||
if (role === 'user') {
|
||||
events.push({
|
||||
type: 'message',
|
||||
timestamp,
|
||||
channel: message.crossConversation ? 'cross_conversation' : 'direct',
|
||||
});
|
||||
for (const mention of Array.isArray(message.composerMentions) ? message.composerMentions : []) {
|
||||
const skill = extractSkillMention(mention);
|
||||
if (!skill) continue;
|
||||
events.push({ type: 'skill', timestamp, name: skill.name, label: skill.label });
|
||||
}
|
||||
}
|
||||
|
||||
if (role === 'assistant') {
|
||||
for (const toolCall of Array.isArray(message.toolCalls) ? message.toolCalls : []) {
|
||||
const mcp = extractMcpToolCall(toolCall);
|
||||
if (!mcp) continue;
|
||||
events.push({
|
||||
type: 'mcp',
|
||||
timestamp,
|
||||
server: mcp.server,
|
||||
tool: mcp.tool,
|
||||
status: mcp.status,
|
||||
messageIndex,
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return {
|
||||
sessionId,
|
||||
title,
|
||||
cwd,
|
||||
projectName: cleanDisplayText(String(source.projectName || source.project || basenameFromCwd(cwd)), 240),
|
||||
agent: normalizeAgent(source.agent),
|
||||
created,
|
||||
updated,
|
||||
sourceMessageCount: messages.length,
|
||||
events,
|
||||
};
|
||||
}
|
||||
|
||||
function validateTimeZone(value) {
|
||||
const timeZone = cleanDisplayText(String(value || 'UTC'), 80) || 'UTC';
|
||||
try {
|
||||
new Intl.DateTimeFormat('en-US', { timeZone }).format(0);
|
||||
} catch {
|
||||
throw new UsageStatisticsError('invalid_time_zone', '无效的时区');
|
||||
}
|
||||
return timeZone;
|
||||
}
|
||||
|
||||
function normalizeUsageRange(params = {}, maxRangeDays = DEFAULT_MAX_RANGE_DAYS) {
|
||||
const fromMs = safeTimestamp(params.from);
|
||||
const toMs = safeTimestamp(params.to);
|
||||
if (fromMs === null || toMs === null) {
|
||||
throw new UsageStatisticsError('invalid_range', '统计时间范围无效');
|
||||
}
|
||||
if (fromMs >= toMs) {
|
||||
throw new UsageStatisticsError('invalid_range', '统计开始时间必须早于结束时间');
|
||||
}
|
||||
const maxMs = Math.max(1, maxRangeDays) * 24 * 60 * 60 * 1000;
|
||||
if (toMs - fromMs > maxMs) {
|
||||
throw new UsageStatisticsError('range_too_large', `统计时间范围不能超过 ${maxRangeDays} 天`);
|
||||
}
|
||||
return {
|
||||
fromMs,
|
||||
toMs,
|
||||
from: new Date(fromMs).toISOString(),
|
||||
to: new Date(toMs).toISOString(),
|
||||
timeZone: validateTimeZone(params.timeZone),
|
||||
bucket: 'day',
|
||||
};
|
||||
}
|
||||
|
||||
function createDayKeyFormatter(timeZone) {
|
||||
const formatter = new Intl.DateTimeFormat('en-US', {
|
||||
timeZone,
|
||||
year: 'numeric',
|
||||
month: '2-digit',
|
||||
day: '2-digit',
|
||||
});
|
||||
return (timestampMs) => {
|
||||
const parts = formatter.formatToParts(new Date(timestampMs));
|
||||
const values = {};
|
||||
for (const part of parts) {
|
||||
if (part.type !== 'literal') values[part.type] = part.value;
|
||||
}
|
||||
return `${values.year}-${values.month}-${values.day}`;
|
||||
};
|
||||
}
|
||||
|
||||
function createTrendBucket(date) {
|
||||
return {
|
||||
date,
|
||||
newSessions: 0,
|
||||
messages: 0,
|
||||
directMessages: 0,
|
||||
crossConversationMessages: 0,
|
||||
mcpCalls: 0,
|
||||
mcpFailures: 0,
|
||||
skillMentions: 0,
|
||||
};
|
||||
}
|
||||
|
||||
function createTrendBuckets(range, dayKey) {
|
||||
const keys = new Set([dayKey(range.fromMs), dayKey(range.toMs - 1)]);
|
||||
const halfDayMs = 12 * 60 * 60 * 1000;
|
||||
for (let cursor = range.fromMs; cursor < range.toMs; cursor += halfDayMs) {
|
||||
keys.add(dayKey(cursor));
|
||||
}
|
||||
const map = new Map();
|
||||
Array.from(keys).sort().forEach((key) => map.set(key, createTrendBucket(key)));
|
||||
return map;
|
||||
}
|
||||
|
||||
function eventInRange(timestamp, range) {
|
||||
const value = safeTimestamp(timestamp);
|
||||
return value !== null && value >= range.fromMs && value < range.toMs ? value : null;
|
||||
}
|
||||
|
||||
function aggregateUsageStatistics(documents, params = {}, meta = {}) {
|
||||
const range = normalizeUsageRange(params, meta.maxRangeDays || DEFAULT_MAX_RANGE_DAYS);
|
||||
const recentLimit = clampInteger(params.recentLimit, DEFAULT_RECENT_LIMIT, 1, 100);
|
||||
const mcpToolLimit = clampInteger(params.mcpToolLimit, DEFAULT_MCP_TOOL_LIMIT, 1, 500);
|
||||
const skillLimit = clampInteger(params.skillLimit, DEFAULT_SKILL_LIMIT, 1, 500);
|
||||
const mcpDetailLimit = clampInteger(params.mcpDetailLimit, DEFAULT_MCP_DETAIL_LIMIT, 1, 500);
|
||||
const dayKey = createDayKeyFormatter(range.timeZone);
|
||||
const trendMap = createTrendBuckets(range, dayKey);
|
||||
const overview = {
|
||||
newSessions: 0,
|
||||
messages: 0,
|
||||
directMessages: 0,
|
||||
crossConversationMessages: 0,
|
||||
mcpCalls: 0,
|
||||
mcpFailures: 0,
|
||||
skillMentions: 0,
|
||||
};
|
||||
const mcpStatus = { completed: 0, failed: 0, other: 0 };
|
||||
const mcpTools = new Map();
|
||||
const skills = new Map();
|
||||
const mcpRecentCalls = [];
|
||||
const recentSessions = [];
|
||||
let retainedMessages = 0;
|
||||
|
||||
for (const doc of documents || []) {
|
||||
if (!doc || !doc.sessionId) continue;
|
||||
retainedMessages += Math.max(0, Number(doc.sourceMessageCount || 0));
|
||||
const sessionSummary = {
|
||||
sessionId: doc.sessionId,
|
||||
title: doc.title || '未命名会话',
|
||||
agent: doc.agent || 'codex',
|
||||
cwd: doc.cwd || '',
|
||||
projectName: doc.projectName || '',
|
||||
created: doc.created || null,
|
||||
lastActivity: null,
|
||||
newSession: false,
|
||||
messages: 0,
|
||||
directMessages: 0,
|
||||
crossConversationMessages: 0,
|
||||
mcpCalls: 0,
|
||||
mcpFailures: 0,
|
||||
skillMentions: 0,
|
||||
};
|
||||
|
||||
const createdMs = eventInRange(doc.created, range);
|
||||
if (createdMs !== null) {
|
||||
overview.newSessions += 1;
|
||||
sessionSummary.newSession = true;
|
||||
sessionSummary.lastActivity = new Date(createdMs).toISOString();
|
||||
const bucket = trendMap.get(dayKey(createdMs));
|
||||
if (bucket) bucket.newSessions += 1;
|
||||
}
|
||||
|
||||
for (const event of Array.isArray(doc.events) ? doc.events : []) {
|
||||
const timestampMs = eventInRange(event.timestamp, range);
|
||||
if (timestampMs === null) continue;
|
||||
const timestamp = new Date(timestampMs).toISOString();
|
||||
if (!sessionSummary.lastActivity || timestamp > sessionSummary.lastActivity) {
|
||||
sessionSummary.lastActivity = timestamp;
|
||||
}
|
||||
const bucket = trendMap.get(dayKey(timestampMs));
|
||||
if (!bucket) continue;
|
||||
|
||||
if (event.type === 'message') {
|
||||
overview.messages += 1;
|
||||
sessionSummary.messages += 1;
|
||||
bucket.messages += 1;
|
||||
if (event.channel === 'cross_conversation') {
|
||||
overview.crossConversationMessages += 1;
|
||||
sessionSummary.crossConversationMessages += 1;
|
||||
bucket.crossConversationMessages += 1;
|
||||
} else {
|
||||
overview.directMessages += 1;
|
||||
sessionSummary.directMessages += 1;
|
||||
bucket.directMessages += 1;
|
||||
}
|
||||
} else if (event.type === 'mcp') {
|
||||
overview.mcpCalls += 1;
|
||||
sessionSummary.mcpCalls += 1;
|
||||
bucket.mcpCalls += 1;
|
||||
const failed = event.status === 'failed';
|
||||
if (failed) {
|
||||
overview.mcpFailures += 1;
|
||||
sessionSummary.mcpFailures += 1;
|
||||
bucket.mcpFailures += 1;
|
||||
mcpStatus.failed += 1;
|
||||
} else if (event.status === 'completed') {
|
||||
mcpStatus.completed += 1;
|
||||
} else {
|
||||
mcpStatus.other += 1;
|
||||
}
|
||||
const key = `${event.server}\u0000${event.tool}`;
|
||||
const current = mcpTools.get(key) || {
|
||||
key: `${event.server}/${event.tool}`,
|
||||
server: event.server,
|
||||
tool: event.tool,
|
||||
calls: 0,
|
||||
completed: 0,
|
||||
failed: 0,
|
||||
other: 0,
|
||||
successRate: 0,
|
||||
lastUsedAt: null,
|
||||
};
|
||||
current.calls += 1;
|
||||
if (failed) current.failed += 1;
|
||||
else if (event.status === 'completed') current.completed += 1;
|
||||
else current.other += 1;
|
||||
if (!current.lastUsedAt || timestamp > current.lastUsedAt) current.lastUsedAt = timestamp;
|
||||
mcpTools.set(key, current);
|
||||
mcpRecentCalls.push({
|
||||
timestamp,
|
||||
server: event.server,
|
||||
tool: event.tool,
|
||||
status: event.status,
|
||||
sessionId: doc.sessionId,
|
||||
sessionTitle: doc.title || '未命名会话',
|
||||
agent: doc.agent || 'codex',
|
||||
messageIndex: Number.isFinite(Number(event.messageIndex)) ? Number(event.messageIndex) : null,
|
||||
});
|
||||
} else if (event.type === 'skill') {
|
||||
overview.skillMentions += 1;
|
||||
sessionSummary.skillMentions += 1;
|
||||
bucket.skillMentions += 1;
|
||||
const name = cleanDisplayText(String(event.name || ''), 160);
|
||||
if (!name) continue;
|
||||
const current = skills.get(name) || {
|
||||
name,
|
||||
label: cleanDisplayText(String(event.label || `$${name}`), 180) || `$${name}`,
|
||||
uses: 0,
|
||||
lastUsedAt: null,
|
||||
};
|
||||
current.uses += 1;
|
||||
if (!current.lastUsedAt || timestamp > current.lastUsedAt) current.lastUsedAt = timestamp;
|
||||
skills.set(name, current);
|
||||
}
|
||||
}
|
||||
|
||||
if (sessionSummary.lastActivity) recentSessions.push(sessionSummary);
|
||||
}
|
||||
|
||||
const mcpToolRows = Array.from(mcpTools.values());
|
||||
for (const row of mcpToolRows) {
|
||||
row.successRate = row.calls > 0 ? Math.round((row.completed / row.calls) * 1000) / 10 : 0;
|
||||
}
|
||||
mcpToolRows.sort((a, b) => b.calls - a.calls
|
||||
|| b.failed - a.failed
|
||||
|| a.server.localeCompare(b.server)
|
||||
|| a.tool.localeCompare(b.tool));
|
||||
const skillRows = Array.from(skills.values()).sort((a, b) => b.uses - a.uses || a.name.localeCompare(b.name));
|
||||
recentSessions.sort((a, b) => String(b.lastActivity).localeCompare(String(a.lastActivity)) || a.sessionId.localeCompare(b.sessionId));
|
||||
mcpRecentCalls.sort((a, b) => b.timestamp.localeCompare(a.timestamp)
|
||||
|| a.server.localeCompare(b.server)
|
||||
|| a.tool.localeCompare(b.tool)
|
||||
|| a.sessionId.localeCompare(b.sessionId));
|
||||
const returnedMcpTools = mcpToolRows.slice(0, mcpToolLimit);
|
||||
const returnedMcpToolKeys = new Set(returnedMcpTools.map((row) => `${row.server}\u0000${row.tool}`));
|
||||
const returnedMcpRecentCalls = mcpRecentCalls
|
||||
.filter((call) => returnedMcpToolKeys.has(`${call.server}\u0000${call.tool}`))
|
||||
.slice(0, mcpDetailLimit);
|
||||
const returnedDetailCounts = new Map();
|
||||
for (const call of returnedMcpRecentCalls) {
|
||||
const key = `${call.server}\u0000${call.tool}`;
|
||||
returnedDetailCounts.set(key, (returnedDetailCounts.get(key) || 0) + 1);
|
||||
}
|
||||
for (const row of returnedMcpTools) {
|
||||
row.detailCallsReturned = returnedDetailCounts.get(`${row.server}\u0000${row.tool}`) || 0;
|
||||
}
|
||||
const returnedSkills = skillRows.slice(0, skillLimit);
|
||||
|
||||
return {
|
||||
schemaVersion: SCHEMA_VERSION,
|
||||
generatedAt: new Date().toISOString(),
|
||||
range: {
|
||||
from: range.from,
|
||||
to: range.to,
|
||||
timeZone: range.timeZone,
|
||||
bucket: range.bucket,
|
||||
semantics: '[from,to)',
|
||||
},
|
||||
coverage: {
|
||||
scope: 'retained_sessions',
|
||||
indexedSessions: Number(meta.indexedSessions ?? (documents || []).length) || 0,
|
||||
trackedFiles: Number(meta.trackedFiles ?? (documents || []).length) || 0,
|
||||
retainedMessages,
|
||||
skippedFiles: Number(meta.skippedFiles || 0),
|
||||
failedFiles: Number(meta.failedFiles || 0),
|
||||
distinctMcpTools: mcpToolRows.length,
|
||||
returnedMcpTools: returnedMcpTools.length,
|
||||
distinctSkills: skillRows.length,
|
||||
returnedSkills: returnedSkills.length,
|
||||
lastBuiltAt: meta.lastBuiltAt || null,
|
||||
mcpTimestampBasis: 'assistant_message',
|
||||
skillBasis: 'explicit_composer_mention',
|
||||
},
|
||||
overview,
|
||||
trend: Array.from(trendMap.values()),
|
||||
mcpStatus,
|
||||
limits: {
|
||||
mcpTools: mcpToolLimit,
|
||||
skills: skillLimit,
|
||||
mcpRecentCalls: mcpDetailLimit,
|
||||
recentSessions: recentLimit,
|
||||
},
|
||||
mcpTools: returnedMcpTools,
|
||||
mcpRecentCalls: returnedMcpRecentCalls,
|
||||
skills: returnedSkills,
|
||||
recentSessions: recentSessions.slice(0, recentLimit),
|
||||
};
|
||||
}
|
||||
|
||||
function safeError(error) {
|
||||
return cleanDisplayText(error?.message || String(error || 'unknown error'), 500);
|
||||
}
|
||||
|
||||
function sleepImmediate() {
|
||||
return new Promise((resolve) => setImmediate(resolve));
|
||||
}
|
||||
|
||||
function timerUnref(timer) {
|
||||
if (timer && typeof timer.unref === 'function') timer.unref();
|
||||
}
|
||||
|
||||
async function writeJsonAtomic(filePath, value) {
|
||||
await fsp.mkdir(path.dirname(filePath), { recursive: true });
|
||||
const tempFile = `${filePath}.${process.pid}.${Date.now()}.tmp`;
|
||||
try {
|
||||
await fsp.writeFile(tempFile, JSON.stringify(value), 'utf8');
|
||||
await fsp.rename(tempFile, filePath);
|
||||
} catch (error) {
|
||||
await fsp.unlink(tempFile).catch(() => {});
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
function createUsageStatisticsIndex(options = {}) {
|
||||
const sessionsDir = path.resolve(options.sessionsDir || path.join(process.cwd(), 'sessions'));
|
||||
const cacheFile = path.resolve(options.cacheFile || path.join(sessionsDir, '_usage', 'index-v1.json'));
|
||||
const maxFileBytes = clampInteger(options.maxFileBytes, DEFAULT_MAX_FILE_BYTES, 64 * 1024, 1024 * 1024 * 1024);
|
||||
const maxRangeDays = clampInteger(options.maxRangeDays, DEFAULT_MAX_RANGE_DAYS, 1, 3660);
|
||||
const logger = options.logger || null;
|
||||
const documents = new Map();
|
||||
const fileMeta = new Map();
|
||||
const upsertTimers = new Map();
|
||||
let state = 'idle';
|
||||
let initialized = false;
|
||||
let buildPromise = null;
|
||||
let persistPromise = null;
|
||||
let persistTimer = null;
|
||||
let dirty = false;
|
||||
let skippedFiles = 0;
|
||||
let failedFiles = 0;
|
||||
let lastBuiltAt = null;
|
||||
let lastPersistedAt = null;
|
||||
let lastError = null;
|
||||
|
||||
function log(level, event, data = {}) {
|
||||
try {
|
||||
if (logger && typeof logger[level] === 'function') logger[level](event, data);
|
||||
else if (logger && typeof logger.log === 'function') logger.log(event, data);
|
||||
} catch {}
|
||||
}
|
||||
|
||||
function status() {
|
||||
return {
|
||||
state,
|
||||
ready: state === 'ready',
|
||||
building: state === 'building',
|
||||
initialized,
|
||||
error: lastError,
|
||||
sessionsDir,
|
||||
cacheFile,
|
||||
indexedSessions: documents.size,
|
||||
trackedFiles: fileMeta.size,
|
||||
pendingUpserts: upsertTimers.size,
|
||||
skippedFiles,
|
||||
failedFiles,
|
||||
lastBuiltAt,
|
||||
lastPersistedAt,
|
||||
};
|
||||
}
|
||||
|
||||
function serializeCache() {
|
||||
return {
|
||||
version: CACHE_VERSION,
|
||||
generatedAt: new Date().toISOString(),
|
||||
files: Object.fromEntries(fileMeta.entries()),
|
||||
documents: Array.from(documents.values()),
|
||||
};
|
||||
}
|
||||
|
||||
async function loadCache() {
|
||||
try {
|
||||
const stat = await fsp.stat(cacheFile);
|
||||
if (!stat.isFile() || stat.size <= 0 || stat.size > 256 * 1024 * 1024) return;
|
||||
const parsed = JSON.parse(await fsp.readFile(cacheFile, 'utf8'));
|
||||
if (!isObject(parsed) || parsed.version !== CACHE_VERSION || !Array.isArray(parsed.documents)) return;
|
||||
documents.clear();
|
||||
fileMeta.clear();
|
||||
for (const doc of parsed.documents) {
|
||||
if (doc && legalSessionId(doc.sessionId)) documents.set(doc.sessionId, doc);
|
||||
}
|
||||
if (isObject(parsed.files)) {
|
||||
for (const [sessionId, fingerprint] of Object.entries(parsed.files)) {
|
||||
if (legalSessionId(sessionId) && isObject(fingerprint)) fileMeta.set(sessionId, fingerprint);
|
||||
}
|
||||
}
|
||||
lastPersistedAt = safeIso(parsed.generatedAt);
|
||||
} catch (error) {
|
||||
if (error?.code !== 'ENOENT') {
|
||||
dirty = true;
|
||||
log('warn', 'usage_statistics_cache_load_failed', { error: safeError(error) });
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async function scanSessionFiles() {
|
||||
const scanned = new Map();
|
||||
let entries = [];
|
||||
try {
|
||||
entries = await fsp.readdir(sessionsDir, { withFileTypes: true });
|
||||
} catch (error) {
|
||||
if (error?.code === 'ENOENT') return scanned;
|
||||
throw error;
|
||||
}
|
||||
for (const entry of entries) {
|
||||
if (!entry.isFile() || !entry.name.endsWith('.json')) continue;
|
||||
const sessionId = entry.name.slice(0, -5);
|
||||
if (!legalSessionId(sessionId)) continue;
|
||||
const filePath = path.join(sessionsDir, entry.name);
|
||||
try {
|
||||
const stat = await fsp.stat(filePath);
|
||||
if (!stat.isFile() || stat.size <= 0 || stat.size > maxFileBytes) {
|
||||
skippedFiles += 1;
|
||||
continue;
|
||||
}
|
||||
scanned.set(sessionId, { filePath, stat, fingerprint: fileFingerprint(stat) });
|
||||
} catch (error) {
|
||||
failedFiles += 1;
|
||||
log('warn', 'usage_statistics_file_stat_failed', { sessionId: sessionId.slice(0, 8), error: safeError(error) });
|
||||
}
|
||||
}
|
||||
return scanned;
|
||||
}
|
||||
|
||||
async function readAndIndexFile(sessionId, filePath, stat, fingerprint) {
|
||||
try {
|
||||
const raw = await fsp.readFile(filePath, 'utf8');
|
||||
if (Buffer.byteLength(raw) > maxFileBytes) {
|
||||
skippedFiles += 1;
|
||||
return false;
|
||||
}
|
||||
const doc = createUsageDocument(JSON.parse(raw), stat, sessionId);
|
||||
if (!doc) throw new Error('invalid session document');
|
||||
documents.set(sessionId, doc);
|
||||
fileMeta.set(sessionId, fingerprint || fileFingerprint(stat));
|
||||
dirty = true;
|
||||
return true;
|
||||
} catch (error) {
|
||||
failedFiles += 1;
|
||||
documents.delete(sessionId);
|
||||
fileMeta.delete(sessionId);
|
||||
dirty = true;
|
||||
log('warn', 'usage_statistics_file_index_failed', { sessionId: sessionId.slice(0, 8), error: safeError(error) });
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
async function persistNow() {
|
||||
if (persistPromise) return persistPromise;
|
||||
persistPromise = (async () => {
|
||||
while (dirty) {
|
||||
dirty = false;
|
||||
await writeJsonAtomic(cacheFile, serializeCache());
|
||||
lastPersistedAt = new Date().toISOString();
|
||||
}
|
||||
})().finally(() => {
|
||||
persistPromise = null;
|
||||
});
|
||||
return persistPromise;
|
||||
}
|
||||
|
||||
function schedulePersist() {
|
||||
dirty = true;
|
||||
if (persistTimer) clearTimeout(persistTimer);
|
||||
persistTimer = setTimeout(() => {
|
||||
persistTimer = null;
|
||||
persistNow().catch((error) => {
|
||||
state = 'error';
|
||||
lastError = safeError(error);
|
||||
log('error', 'usage_statistics_cache_persist_failed', { error: lastError });
|
||||
});
|
||||
}, PERSIST_DEBOUNCE_MS);
|
||||
timerUnref(persistTimer);
|
||||
}
|
||||
|
||||
async function rebuildChangedFiles() {
|
||||
state = 'building';
|
||||
lastError = null;
|
||||
skippedFiles = 0;
|
||||
failedFiles = 0;
|
||||
await loadCache();
|
||||
const scanned = await scanSessionFiles();
|
||||
for (const sessionId of Array.from(documents.keys())) {
|
||||
if (!scanned.has(sessionId)) {
|
||||
documents.delete(sessionId);
|
||||
fileMeta.delete(sessionId);
|
||||
dirty = true;
|
||||
}
|
||||
}
|
||||
const changed = [];
|
||||
for (const [sessionId, entry] of scanned.entries()) {
|
||||
if (!sameFingerprint(fileMeta.get(sessionId), entry.fingerprint) || !documents.has(sessionId)) {
|
||||
changed.push([sessionId, entry]);
|
||||
}
|
||||
}
|
||||
for (let index = 0; index < changed.length; index += 1) {
|
||||
const [sessionId, entry] = changed[index];
|
||||
await readAndIndexFile(sessionId, entry.filePath, entry.stat, entry.fingerprint);
|
||||
if ((index + 1) % BUILD_BATCH_SIZE === 0) await sleepImmediate();
|
||||
}
|
||||
if (dirty) await persistNow();
|
||||
lastBuiltAt = new Date().toISOString();
|
||||
initialized = true;
|
||||
state = 'ready';
|
||||
return status();
|
||||
}
|
||||
|
||||
function initialize() {
|
||||
if (buildPromise) return buildPromise;
|
||||
if (initialized && state === 'ready') return Promise.resolve(status());
|
||||
buildPromise = rebuildChangedFiles()
|
||||
.catch((error) => {
|
||||
state = 'error';
|
||||
lastError = safeError(error);
|
||||
log('error', 'usage_statistics_initialize_failed', { error: lastError });
|
||||
return status();
|
||||
})
|
||||
.finally(() => {
|
||||
buildPromise = null;
|
||||
});
|
||||
return buildPromise;
|
||||
}
|
||||
|
||||
async function upsertNow(sessionId) {
|
||||
const id = sanitizeSessionId(sessionId);
|
||||
if (!legalSessionId(id)) return false;
|
||||
const filePath = path.join(sessionsDir, `${id}.json`);
|
||||
try {
|
||||
const stat = await fsp.stat(filePath);
|
||||
if (!stat.isFile() || stat.size <= 0 || stat.size > maxFileBytes) {
|
||||
const existed = documents.delete(id) || fileMeta.delete(id);
|
||||
fileMeta.delete(id);
|
||||
if (existed) schedulePersist();
|
||||
return false;
|
||||
}
|
||||
const changed = !sameFingerprint(fileMeta.get(id), fileFingerprint(stat)) || !documents.has(id);
|
||||
if (!changed) return true;
|
||||
const result = await readAndIndexFile(id, filePath, stat, fileFingerprint(stat));
|
||||
if (result) schedulePersist();
|
||||
return result;
|
||||
} catch (error) {
|
||||
if (error?.code === 'ENOENT') return remove(id);
|
||||
failedFiles += 1;
|
||||
log('warn', 'usage_statistics_upsert_failed', { sessionId: id.slice(0, 8), error: safeError(error) });
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
function scheduleUpsert(sessionId) {
|
||||
if (!initialized && !buildPromise) return false;
|
||||
const id = sanitizeSessionId(sessionId);
|
||||
if (!legalSessionId(id)) return false;
|
||||
const existing = upsertTimers.get(id);
|
||||
if (existing) clearTimeout(existing);
|
||||
const timer = setTimeout(() => {
|
||||
upsertTimers.delete(id);
|
||||
upsertNow(id).catch((error) => {
|
||||
lastError = safeError(error);
|
||||
log('error', 'usage_statistics_schedule_upsert_failed', { sessionId: id.slice(0, 8), error: lastError });
|
||||
});
|
||||
}, UPSERT_DEBOUNCE_MS);
|
||||
timerUnref(timer);
|
||||
upsertTimers.set(id, timer);
|
||||
return true;
|
||||
}
|
||||
|
||||
function remove(sessionId) {
|
||||
if (!initialized && !buildPromise) return false;
|
||||
const id = sanitizeSessionId(sessionId);
|
||||
if (!legalSessionId(id)) return false;
|
||||
const timer = upsertTimers.get(id);
|
||||
if (timer) clearTimeout(timer);
|
||||
upsertTimers.delete(id);
|
||||
const existed = documents.delete(id) || fileMeta.delete(id);
|
||||
fileMeta.delete(id);
|
||||
if (existed) schedulePersist();
|
||||
return existed;
|
||||
}
|
||||
|
||||
async function flush() {
|
||||
if (buildPromise) await buildPromise;
|
||||
for (const id of Array.from(upsertTimers.keys())) {
|
||||
const timer = upsertTimers.get(id);
|
||||
if (timer) clearTimeout(timer);
|
||||
upsertTimers.delete(id);
|
||||
await upsertNow(id);
|
||||
}
|
||||
if (persistTimer) {
|
||||
clearTimeout(persistTimer);
|
||||
persistTimer = null;
|
||||
}
|
||||
if (dirty) await persistNow();
|
||||
if (persistPromise) await persistPromise;
|
||||
return status();
|
||||
}
|
||||
|
||||
function query(params = {}) {
|
||||
if (!initialized || state !== 'ready') {
|
||||
throw new UsageStatisticsError('index_unavailable', '统计索引暂不可用');
|
||||
}
|
||||
return aggregateUsageStatistics(Array.from(documents.values()), params, {
|
||||
...status(),
|
||||
maxRangeDays,
|
||||
});
|
||||
}
|
||||
|
||||
return {
|
||||
initialize,
|
||||
scheduleUpsert,
|
||||
remove,
|
||||
flush,
|
||||
query,
|
||||
status,
|
||||
};
|
||||
}
|
||||
|
||||
module.exports = {
|
||||
SCHEMA_VERSION,
|
||||
UsageStatisticsError,
|
||||
createUsageStatisticsIndex,
|
||||
createUsageDocument,
|
||||
aggregateUsageStatistics,
|
||||
normalizeUsageRange,
|
||||
extractMcpToolCall,
|
||||
extractSkillMention,
|
||||
};
|
||||
Reference in New Issue
Block a user