1/**
2 * Durable per-file scan cache.
3 *
4 * Transcripts are append-only and a file that has not changed can never yield
5 * different usage, so parsed records are keyed by `(size, mtime)` and reused.
6 * Without this every server restart re-parses the whole window: roughly 3.5s
7 * for a 30-day scan here, against ~11ms to reload this cache.
8 *
9 * Caching *per file* rather than per day is deliberate. It is timezone
10 * independent, so changing the reporting zone does not invalidate anything, and
11 * it keeps cross-file de-duplication exact: cached entries are de-duplicated
12 * within their own file only, and the aggregator still applies the global
13 * dedupe pass over the small surviving key set.
14 *
15 * @module usageScanCache
16 */
17import type { UsageProviderKind } from "@t3tools/contracts";
18import type {
19 TranscriptUsageFormat,
20 UsageRecord,
21 UsageSpeed,
22} from "@t3tools/provider-core/server/usage";
23import * as Schema from "effect/Schema";
24
25import { GUARD_LENGTH, type TranscriptParsePosition } from "./usageTranscriptReader.ts";
26
27// v2: Codex fork-copy suppression changed what a file parses to, so v1
28// entries would keep serving double-counted records forever.
29// v3: entries carry the parse position and reducer state so a grown file
30// re-parses only its appended bytes instead of starting over.
31// v4: records carry Claude fast mode, which v3 rows never captured.
32// v5: Codex records carry their service tier. v4 rows store speed the same
33// way, so v4 entries still load; see `decodeScanCache` for v4 stateful entries.
34const USAGE_SCAN_CACHE_VERSION = 5 as const;
35const SPEED_COMPATIBLE_SINCE_VERSION = 4;
36
37/**
38 * Each cache version writes its own file in the state directory. An older
39 * server sharing that directory cannot read a newer cache and would replace
40 * it, dropping saved usage for deleted transcripts. Separate files keep both.
41 * A v5 server reads the legacy (v4) file once, when its own file is missing.
42 */
43export const SCAN_CACHE_FILE_NAME = "usage-scan-cache-v5.json";
44export const LEGACY_SCAN_CACHE_FILE_NAME = "usage-scan-cache.json";
45
46/** Serialised as the index into this list. */
47const SPEEDS: readonly UsageSpeed[] = ["standard", "fast", "ultrafast"];
48
49export interface CachedFile {
50 readonly size: number;
51 readonly mtimeMs: number;
52 readonly provider: UsageProviderKind;
53 /** Records from newline-terminated lines, up to `position.resumeOffset`. */
54 readonly records: readonly UsageRecord[];
55 /**
56 * Records from a trailing segment the writer had not newline-terminated at
57 * parse time. Kept apart from `records` because an incremental parse
58 * re-reads that segment and would otherwise double count it.
59 */
60 readonly tailRecords: readonly UsageRecord[];
61 readonly position: TranscriptParsePosition;
62}
63
64export type ScanCache = Map<string, CachedFile>;
65
66/**
67 * Row layout for the serialised form. Positional and interned rather than
68 * object-per-record: on a 30-day window that is the difference between a file
69 * measured in tens of megabytes and one under six.
70 */
71type SerializedRecord = readonly [
72 timestampMs: number,
73 modelIndex: number,
74 sessionIndex: number,
75 uncachedInputTokens: number,
76 cachedInputTokens: number,
77 cacheCreationTokens: number,
78 outputTokens: number,
79 reasoningTokens: number,
80 dedupeKey: string | null,
81 reportedCostUsd: number | null,
82 speed: number,
83 /** Optional trailing field keeps existing cache rows readable. */
84 rateModelIndex?: number | null,
85];
86
87interface SerializedFile {
88 readonly s: number;
89 readonly m: number;
90 readonly p: UsageProviderKind;
91 readonly r: readonly SerializedRecord[];
92 /** Tail records; see `CachedFile.tailRecords`. */
93 readonly t: readonly SerializedRecord[];
94 /** Parse position: resume offset, guard length, guard hash. */
95 readonly o: number;
96 readonly gl: number;
97 readonly gh: number;
98 /** The format's encoded reducer state at `o`; `null` for stateless formats. */
99 readonly cs: unknown;
100}
101
102interface SerializedCache {
103 readonly version: number;
104 readonly models: readonly string[];
105 readonly sessions: readonly string[];
106 readonly files: Readonly<Record<string, SerializedFile>>;
107}
108
109/** Model and session strings, each stored once and referenced by index. */
110interface InternTables {
111 readonly models: string[];
112 readonly sessions: string[];
113 readonly modelIndex: Map<string, number>;
114 readonly sessionIndex: Map<string, number>;
115}
116
117function makeInternTables(): InternTables {
118 return { models: [], sessions: [], modelIndex: new Map(), sessionIndex: new Map() };
119}
120
121function intern(table: string[], index: Map<string, number>, value: string): number {
122 const existing = index.get(value);
123 if (existing !== undefined) return existing;
124 const next = table.length;
125 table.push(value);
126 index.set(value, next);
127 return next;
128}
129
130function serializeFile(entry: CachedFile, tables: InternTables): SerializedFile {
131 const serializeRecord = (record: UsageRecord): SerializedRecord => [
132 record.timestampMs,
133 intern(tables.models, tables.modelIndex, record.model),
134 intern(tables.sessions, tables.sessionIndex, record.sessionId),
135 record.totals.uncachedInputTokens,
136 record.totals.cachedInputTokens,
137 record.totals.cacheCreationTokens,
138 record.totals.outputTokens,
139 record.totals.reasoningTokens,
140 record.dedupeKey,
141 record.reportedCostUsd,
142 SPEEDS.indexOf(record.speed),
143 record.rateModel === undefined
144 ? null
145 : intern(tables.models, tables.modelIndex, record.rateModel),
146 ];
147 return {
148 s: entry.size,
149 m: entry.mtimeMs,
150 p: entry.provider,
151 r: entry.records.map(serializeRecord),
152 t: entry.tailRecords.map(serializeRecord),
153 o: entry.position.resumeOffset,
154 gl: entry.position.guardLength,
155 gh: entry.position.guardHash,
156 cs: entry.position.state,
157 };
158}
159
160/** Serialises the cache, interning the repeated model and session strings. */
161export function encodeScanCache(cache: ScanCache): SerializedCache {
162 const tables = makeInternTables();
163 const files: Record<string, SerializedFile> = {};
164 for (const [path, entry] of cache) files[path] = serializeFile(entry, tables);
165 return {
166 version: USAGE_SCAN_CACHE_VERSION,
167 models: tables.models,
168 sessions: tables.sessions,
169 files,
170 };
171}
172
173/**
174 * Returns a function that serialises the cache to JSON text, re-encoding only
175 * the entries that changed since its last call. Call it once per persist.
176 *
177 * Writes the same document as `encodeScanCache`. Most entries never change
178 * between scans, and encoding all of them made each persist cost close to a
179 * second on a large cache. Entries are replaced, never mutated, when their file
180 * changes, so an entry's JSON is memoised by identity. The intern tables only
181 * grow, so a memoised entry's indexes stay valid; a pruned entry can leave an
182 * unused string behind until the next process start.
183 */
184export function makeScanCacheWriter(): (
185 cache: ScanCache,
186 extra: Readonly<Record<string, unknown>>,
187) => string {
188 const tables = makeInternTables();
189 const fragments = new WeakMap<CachedFile, string>();
190 return (cache, extra) => {
191 const files: string[] = [];
192 for (const [path, entry] of cache) {
193 let fragment = fragments.get(entry);
194 if (fragment === undefined) {
195 fragment = JSON.stringify(serializeFile(entry, tables));
196 fragments.set(entry, fragment);
197 }
198 files.push(`${JSON.stringify(path)}:${fragment}`);
199 }
200 // Encoded after the files, which may have added to the intern tables.
201 const head = JSON.stringify({
202 ...extra,
203 version: USAGE_SCAN_CACHE_VERSION,
204 models: tables.models,
205 sessions: tables.sessions,
206 });
207 return `${head.slice(0, -1)},"files":{${files.join(",")}}}`;
208 };
209}
210
211function isRecordArray(value: unknown): value is readonly unknown[] {
212 return Array.isArray(value);
213}
214
215/**
216 * Rebuilds the cache from a parsed document.
217 *
218 * Anything malformed yields an empty cache rather than an error: a corrupt
219 * cache should cost one cold scan, never a broken page. Entries of a provider
220 * without a transcript format in `formats` are dropped.
221 */
222export function decodeScanCache(
223 document: unknown,
224 formats: ReadonlyMap<UsageProviderKind, TranscriptUsageFormat<unknown>>,
225): ScanCache {
226 const cache: ScanCache = new Map();
227 const isValidState = makeStateValidators(formats);
228 if (typeof document !== "object" || document === null) return cache;
229
230 const root = document as Partial<SerializedCache>;
231 const version = root.version;
232 if (
233 typeof version !== "number" ||
234 version < SPEED_COMPATIBLE_SINCE_VERSION ||
235 version > USAGE_SCAN_CACHE_VERSION
236 ) {
237 return cache;
238 }
239 if (!isRecordArray(root.models) || !isRecordArray(root.sessions)) return cache;
240 if (typeof root.files !== "object" || root.files === null) return cache;
241
242 // The intern tables must be all strings: a numeric entry would pass the
243 // undefined guard below, land in a record's model, and crash the aggregate
244 // at lookupRate. A corrupt table rejects the whole cache.
245 if (!root.models.every((value) => typeof value === "string")) return cache;
246 if (!root.sessions.every((value) => typeof value === "string")) return cache;
247 const models = root.models as readonly string[];
248 const sessions = root.sessions as readonly string[];
249
250 // Any corrupt row disqualifies the whole entry. Keeping the survivors
251 // under the original (size, mtime) would read as a valid warm hit and the
252 // file would never be re-parsed, silently losing the dropped rows' usage.
253 const decodeRecords = (
254 rows: readonly unknown[],
255 provider: UsageProviderKind,
256 ): UsageRecord[] | null => {
257 const records: UsageRecord[] = [];
258 for (const row of rows) {
259 if (!isRecordArray(row) || row.length < 11) return null;
260 const [
261 timestampMs,
262 modelIndex,
263 sessionIndex,
264 uncached,
265 cached,
266 cacheCreation,
267 output,
268 reasoning,
269 dedupeKey,
270 reportedCostUsd,
271 speedIndex,
272 rateModelIndex,
273 ] = row as SerializedRecord;
274 const speed = typeof speedIndex === "number" ? SPEEDS[speedIndex] : undefined;
275
276 const model = typeof modelIndex === "number" ? models[modelIndex] : undefined;
277 const rateModel = typeof rateModelIndex === "number" ? models[rateModelIndex] : undefined;
278 if (
279 typeof timestampMs !== "number" ||
280 !Number.isFinite(timestampMs) ||
281 model === undefined ||
282 (rateModelIndex != null && rateModel === undefined) ||
283 !Number.isFinite(uncached) ||
284 !Number.isFinite(cached) ||
285 !Number.isFinite(cacheCreation) ||
286 !Number.isFinite(output) ||
287 !Number.isFinite(reasoning) ||
288 speed === undefined
289 ) {
290 return null;
291 }
292
293 records.push({
294 provider,
295 timestampMs,
296 model,
297 ...(rateModel === undefined ? {} : { rateModel }),
298 sessionId: (typeof sessionIndex === "number" ? sessions[sessionIndex] : undefined) ?? "",
299 totals: {
300 uncachedInputTokens: uncached,
301 cachedInputTokens: cached,
302 cacheCreationTokens: cacheCreation,
303 outputTokens: output,
304 reasoningTokens: reasoning,
305 },
306 reportedCostUsd: typeof reportedCostUsd === "number" ? reportedCostUsd : null,
307 speed,
308 dedupeKey: typeof dedupeKey === "string" ? dedupeKey : null,
309 });
310 }
311 return records;
312 };
313
314 for (const [path, raw] of Object.entries(root.files)) {
315 if (typeof raw !== "object" || raw === null) continue;
316 const entry = raw as Partial<SerializedFile>;
317 if (typeof entry.s !== "number" || typeof entry.m !== "number") continue;
318 if (entry.p === undefined) continue;
319 const format = formats.get(entry.p);
320 if (format === undefined) continue;
321 if (!isRecordArray(entry.r) || !isRecordArray(entry.t)) continue;
322 // Position fields feed byte offsets and a Buffer allocation in the reader,
323 // so anything outside their real ranges must reject the entry: a bogus
324 // guard length would otherwise fail every parse of the file, silently
325 // dropping its usage instead of costing the documented cold re-parse.
326 if (
327 typeof entry.o !== "number" ||
328 !Number.isSafeInteger(entry.o) ||
329 entry.o < 0 ||
330 typeof entry.gl !== "number" ||
331 !Number.isSafeInteger(entry.gl) ||
332 entry.gl < 0 ||
333 entry.gl > GUARD_LENGTH ||
334 entry.gl > entry.o ||
335 typeof entry.gh !== "number" ||
336 !Number.isFinite(entry.gh)
337 ) {
338 continue;
339 }
340 // v4 records of stateful formats (Codex) predate service tiers, so they
341 // all priced as standard. Keep them, because the rollout may be gone, but
342 // make a live rollout re-parse whole: no file has size -1, and a zero
343 // position cannot resume.
344 const legacy = format.state !== undefined && version < USAGE_SCAN_CACHE_VERSION;
345 // A corrupt state disqualifies the entry: resuming with it would attach
346 // appended usage to the wrong model or replay fork-copied history.
347 if (!legacy && !isValidState.get(entry.p)?.(entry.cs)) continue;
348
349 const provider = entry.p;
350 const records = decodeRecords(entry.r, provider);
351 const tailRecords = decodeRecords(entry.t, provider);
352 if (records === null || tailRecords === null) continue;
353
354 cache.set(path, {
355 size: legacy ? -1 : entry.s,
356 mtimeMs: entry.m,
357 provider,
358 records,
359 tailRecords,
360 position: legacy
361 ? { resumeOffset: 0, guardLength: 0, guardHash: 0, state: null }
362 : {
363 resumeOffset: entry.o,
364 guardLength: entry.gl,
365 guardHash: entry.gh,
366 state: entry.cs,
367 },
368 });
369 }
370
371 return cache;
372}
373
374/**
375 * Per provider, whether a persisted state is valid: `null`, or one the
376 * format's schema accepts.
377 */
378function makeStateValidators(
379 formats: ReadonlyMap<UsageProviderKind, TranscriptUsageFormat<unknown>>,
380): ReadonlyMap<UsageProviderKind, (value: unknown) => boolean> {
381 return new Map(
382 [...formats].map(([provider, format]) => {
383 const isState = format.state === undefined ? undefined : Schema.is(format.state.schema);
384 return [
385 provider,
386 (value: unknown) => value === null || (isState !== undefined && isState(value)),
387 ] as const;
388 }),
389 );
390}
391
392/** Keeps saved usage after transcript cleanup, until the reporting retention expires. */
393export function pruneScanCache(cache: ScanCache, retentionCutoffMs: number): number {
394 let removed = 0;
395 for (const [path, entry] of cache) {
396 if (entry.mtimeMs < retentionCutoffMs) {
397 cache.delete(path);
398 removed += 1;
399 }
400 }
401 return removed;
402}
403
404/**
405 * Within-file de-duplication, applied before an entry is cached.
406 *
407 * Callers stitching an incremental parse together pass one `seen` set across
408 * the line and tail record batches so the whole file stays deduplicated as a
409 * unit; the set is mutated in place.
410 */
411export function dedupeWithinFile(
412 records: readonly UsageRecord[],
413 seen: Set<string> = new Set(),
414): readonly UsageRecord[] {
415 const kept: UsageRecord[] = [];
416 for (const record of records) {
417 if (record.dedupeKey !== null) {
418 if (seen.has(record.dedupeKey)) continue;
419 seen.add(record.dedupeKey);
420 }
421 kept.push(record);
422 }
423 return kept;
424}