1// @effect-diagnostics globalDate:off
2/**
3 * Folds parsed transcript records into `(day, hourStart?, provider, model)`
4 * buckets.
5 *
6 * `Intl.DateTimeFormat` is the only reliable way to resolve a wall-clock day in
7 * an arbitrary IANA zone, and it takes a `Date`. That is why the raw `Date`
8 * construction is allowed here; nothing in this module reads the clock.
9 *
10 * Pure, so the bucketing and de-duplication rules are testable without touching
11 * the filesystem or the network.
12 *
13 * @module usageAggregation
14 */
15import type {
16 UsageBucket,
17 UsageCategoryCost,
18 UsageDay,
19 UsageResolution,
20 UsageTokenTotals,
21} from "@t3tools/contracts";
22
23import { EMPTY_TOTALS, type UsageRecord } from "@t3tools/provider-core/server/usage";
24import { cacheSavingsUsd, priceUsage, type RateTable } from "./usagePricing.ts";
25
26/**
27 * Formats an instant as a `YYYY-MM-DD` day in `timeZone`.
28 *
29 * `en-CA` yields ISO-ordered parts, which is why it is used here rather than
30 * assembling the day from `Date` getters (those are host-local only).
31 */
32function makeDayFormatter(timeZone: string): (timestampMs: number) => string {
33 let format: Intl.DateTimeFormat;
34 try {
35 format = new Intl.DateTimeFormat("en-CA", {
36 timeZone,
37 year: "numeric",
38 month: "2-digit",
39 day: "2-digit",
40 });
41 } catch {
42 // An unknown zone should degrade to UTC rather than fail the whole scan.
43 format = new Intl.DateTimeFormat("en-CA", {
44 timeZone: "UTC",
45 year: "numeric",
46 month: "2-digit",
47 day: "2-digit",
48 });
49 }
50 // Formatting once per quarter-hour slot keeps `Intl` (about 2µs a call) off
51 // the per-record path. A slot whose two ends fall on one day lies wholly in
52 // that day. Named zones put midnight on a quarter hour, so their slots never
53 // split; a fixed offset such as `+00:01` can, and records in a split slot are
54 // formatted exactly. `null` marks a split slot.
55 const days = new Map<number, string | null>();
56 return (timestampMs) => {
57 const slot = Math.floor(timestampMs / QUARTER_HOUR_MS);
58 let day = days.get(slot);
59 if (day === undefined) {
60 const first = format.format(new Date(slot * QUARTER_HOUR_MS));
61 const last = format.format(new Date((slot + 1) * QUARTER_HOUR_MS - 1));
62 day = first === last ? first : null;
63 days.set(slot, day);
64 }
65 return day ?? format.format(new Date(timestampMs));
66 };
67}
68
69const QUARTER_HOUR_MS = 15 * 60 * 1000;
70const HOUR_MS = 60 * 60 * 1000;
71
72interface MutableBucket {
73 totals: { -readonly [K in keyof UsageTokenTotals]: number };
74 costUsd: number;
75 cacheSavingsUsd: number;
76 categoryCostUsd: UsageCategoryCost | null;
77 fastCostUsd: number;
78 ultrafastCostUsd: number;
79 speedPremiumUsd: number;
80 records: number;
81 unpricedRecords: number;
82 providerReportedRecords: number;
83 sessions: Set<string>;
84}
85
86export interface AggregateOptions {
87 readonly timeZone: string;
88 readonly sinceDay: string;
89 readonly untilDay: string;
90 readonly rates: RateTable;
91 readonly priceOverrides?: RateTable;
92 /** From {@link resolveModelAliases}. Mapped records bucket and price as their target. */
93 readonly modelAliases?: ReadonlyMap<string, string>;
94 readonly resolution?: UsageResolution;
95 readonly sinceTimeMs?: number;
96 readonly untilTimeMs?: number;
97}
98
99/**
100 * Resolves user model mappings to their final target, so `a -> b` and
101 * `b -> c` both land on `c`. A model whose chain enters a loop is left
102 * unmapped.
103 */
104export function resolveModelAliases(
105 aliases: Readonly<Record<string, string>>,
106): ReadonlyMap<string, string> {
107 const resolved = new Map<string, string>();
108 for (const model of Object.keys(aliases)) {
109 const seen = new Set([model]);
110 let target = aliases[model]!;
111 while (Object.hasOwn(aliases, target) && !seen.has(target)) {
112 seen.add(target);
113 target = aliases[target]!;
114 }
115 // Stopping on a mapped model means the chain entered a loop.
116 if (!Object.hasOwn(aliases, target)) resolved.set(model, target);
117 }
118 return resolved;
119}
120
121export interface AggregateResult {
122 readonly buckets: readonly UsageBucket[];
123 /** Records dropped because an earlier record carried the same dedupe key. */
124 readonly duplicatesDropped: number;
125 /** Records whose day fell outside the requested window. */
126 readonly outOfWindow: number;
127}
128
129/**
130 * Accumulates records across many files.
131 *
132 * De-duplication is global across the whole scan, not per file: Claude Code
133 * copies a message's records forward when a session is resumed or forked, so
134 * the same `dedupeKey` legitimately appears in several transcripts.
135 */
136export class UsageAggregator {
137 readonly #buckets = new Map<string, MutableBucket>();
138 readonly #seen = new Set<string>();
139 readonly #toDay: (timestampMs: number) => string;
140 readonly #hourlyWindow: { readonly sinceTimeMs: number; readonly untilTimeMs: number } | null;
141 readonly #options: AggregateOptions;
142 #lastBucket: {
143 readonly day: string;
144 readonly hourIndex: number;
145 readonly provider: string;
146 readonly model: string;
147 readonly source: string;
148 readonly bucket: MutableBucket;
149 } | null = null;
150 #duplicatesDropped = 0;
151 #outOfWindow = 0;
152
153 constructor(options: AggregateOptions) {
154 this.#options = options;
155 this.#toDay = makeDayFormatter(options.timeZone);
156 if (options.resolution === "hour") {
157 if (options.sinceTimeMs === undefined || options.untilTimeMs === undefined) {
158 throw new Error("Hourly usage aggregation requires exact time bounds");
159 }
160 this.#hourlyWindow = {
161 sinceTimeMs: options.sinceTimeMs,
162 untilTimeMs: options.untilTimeMs,
163 };
164 } else {
165 this.#hourlyWindow = null;
166 }
167 }
168
169 /**
170 * Folds one record in. Returns whether it actually contributed, so callers
171 * can derive per-window facts (distinct sessions, for one) from the records
172 * that landed rather than everything the mtime prefilter happened to admit.
173 */
174 add(input: UsageRecord, sourcePath?: string): boolean {
175 const record = this.#mapModel(input);
176 if (record.dedupeKey !== null) {
177 if (this.#seen.has(record.dedupeKey)) {
178 this.#duplicatesDropped += 1;
179 return false;
180 }
181 this.#seen.add(record.dedupeKey);
182 }
183
184 if (
185 this.#hourlyWindow !== null &&
186 (record.timestampMs < this.#hourlyWindow.sinceTimeMs ||
187 record.timestampMs >= this.#hourlyWindow.untilTimeMs)
188 ) {
189 this.#outOfWindow += 1;
190 return false;
191 }
192
193 const day = this.#toDay(record.timestampMs);
194 if (
195 this.#hourlyWindow === null &&
196 (day < this.#options.sinceDay || day > this.#options.untilDay)
197 ) {
198 this.#outOfWindow += 1;
199 return false;
200 }
201
202 const hourIndex =
203 this.#hourlyWindow === null
204 ? -1
205 : Math.floor((record.timestampMs - this.#hourlyWindow.sinceTimeMs) / HOUR_MS);
206 const bucket = this.#bucketFor(day, hourIndex, record.provider, record.model, sourcePath ?? "");
207
208 const priced = priceUsage(this.#options.rates, record, this.#options.priceOverrides);
209
210 const totals = bucket.totals;
211 totals.uncachedInputTokens += record.totals.uncachedInputTokens;
212 totals.cachedInputTokens += record.totals.cachedInputTokens;
213 totals.cacheCreationTokens += record.totals.cacheCreationTokens;
214 totals.outputTokens += record.totals.outputTokens;
215 totals.reasoningTokens += record.totals.reasoningTokens;
216 bucket.costUsd += priced.costUsd;
217 if (priced.categoryCostUsd !== null) {
218 const sum = bucket.categoryCostUsd;
219 const add = priced.categoryCostUsd;
220 bucket.categoryCostUsd =
221 sum === null
222 ? add
223 : {
224 input: sum.input + add.input,
225 cacheRead: sum.cacheRead + add.cacheRead,
226 cacheWrite: sum.cacheWrite + add.cacheWrite,
227 output: sum.output + add.output,
228 };
229 }
230 if (record.speed === "fast") bucket.fastCostUsd += priced.costUsd;
231 if (record.speed === "ultrafast") bucket.ultrafastCostUsd += priced.costUsd;
232 bucket.speedPremiumUsd += priced.speedPremiumUsd;
233 bucket.cacheSavingsUsd += cacheSavingsUsd(
234 this.#options.rates,
235 record,
236 this.#options.priceOverrides,
237 );
238 bucket.records += 1;
239 if (priced.costSource === "unpriced") bucket.unpricedRecords += 1;
240 if (priced.costSource === "providerReported") bucket.providerReportedRecords += 1;
241 if (record.sessionId.length > 0) bucket.sessions.add(record.sessionId);
242 return true;
243 }
244
245 /** The target's own rate applies, so a provider-specific `rateModel` is dropped. */
246 #mapModel(record: UsageRecord): UsageRecord {
247 const model = this.#options.modelAliases?.get(record.model);
248 if (model === undefined) return record;
249 const { rateModel: _rateModel, ...rest } = record;
250 return { ...rest, model };
251 }
252
253 /**
254 * Records arrive file by file in time order, so most land in the bucket the
255 * previous record used. Checking that first skips building and hashing a
256 * key string per record.
257 */
258 #bucketFor(
259 day: string,
260 hourIndex: number,
261 provider: string,
262 model: string,
263 source: string,
264 ): MutableBucket {
265 const last = this.#lastBucket;
266 if (
267 last !== null &&
268 last.day === day &&
269 last.hourIndex === hourIndex &&
270 last.provider === provider &&
271 last.model === model &&
272 last.source === source
273 ) {
274 return last.bucket;
275 }
276 const window = this.#hourlyWindow;
277 const hourStart =
278 window === null ? "" : new Date(window.sinceTimeMs + hourIndex * HOUR_MS).toISOString();
279 const key = `${day}\u0000${hourStart}\u0000${provider}\u0000${model}\u0000${source}`;
280 let bucket = this.#buckets.get(key);
281 if (bucket === undefined) {
282 bucket = {
283 totals: { ...EMPTY_TOTALS },
284 costUsd: 0,
285 cacheSavingsUsd: 0,
286 categoryCostUsd: null,
287 fastCostUsd: 0,
288 ultrafastCostUsd: 0,
289 speedPremiumUsd: 0,
290 records: 0,
291 unpricedRecords: 0,
292 providerReportedRecords: 0,
293 sessions: new Set<string>(),
294 };
295 this.#buckets.set(key, bucket);
296 }
297 this.#lastBucket = { day, hourIndex, provider, model, source, bucket };
298 return bucket;
299 }
300
301 finish(): AggregateResult {
302 const buckets: UsageBucket[] = [];
303 for (const [key, bucket] of this.#buckets) {
304 const [day = "", hourStart = "", provider = "", model = "", sourcePath = ""] =
305 key.split("\u0000");
306 const category = bucket.categoryCostUsd;
307 const fastCostUsd = roundUsd(bucket.fastCostUsd);
308 const ultrafastCostUsd = roundUsd(bucket.ultrafastCostUsd);
309 const speedPremiumUsd = roundUsd(bucket.speedPremiumUsd);
310 buckets.push({
311 day: day as UsageDay,
312 ...(hourStart === "" ? {} : { hourStart }),
313 provider: provider as UsageBucket["provider"],
314 model,
315 ...(sourcePath === "" ? {} : { sourcePath }),
316 totals: { ...bucket.totals },
317 costUsd: bucket.costUsd,
318 cacheSavingsUsd: bucket.cacheSavingsUsd,
319 // Zero and unknown figures are omitted to keep payloads small.
320 ...(category === null
321 ? {}
322 : {
323 categoryCostUsd: {
324 input: roundUsd(category.input),
325 cacheRead: roundUsd(category.cacheRead),
326 cacheWrite: roundUsd(category.cacheWrite),
327 output: roundUsd(category.output),
328 },
329 }),
330 ...(fastCostUsd === 0 ? {} : { fastCostUsd }),
331 ...(ultrafastCostUsd === 0 ? {} : { ultrafastCostUsd }),
332 ...(speedPremiumUsd === 0 ? {} : { speedPremiumUsd }),
333 costSource: resolveCostSource(bucket),
334 records: bucket.records,
335 unpricedRecords: bucket.unpricedRecords,
336 sessions: bucket.sessions.size,
337 });
338 }
339 // Stable ordering keeps payloads diffable and snapshot tests meaningful.
340 buckets.sort(
341 (a, b) =>
342 a.day.localeCompare(b.day) ||
343 (a.hourStart ?? "").localeCompare(b.hourStart ?? "") ||
344 a.provider.localeCompare(b.provider) ||
345 a.model.localeCompare(b.model),
346 );
347
348 return {
349 buckets,
350 duplicatesDropped: this.#duplicatesDropped,
351 outOfWindow: this.#outOfWindow,
352 };
353 }
354}
355
356/**
357 * Rounds to micro-dollars. The split and speed figures need no more precision,
358 * and shorter numbers keep them cheap on the wire.
359 */
360function roundUsd(value: number): number {
361 return Math.round(value * 1e6) / 1e6;
362}
363
364/**
365 * A bucket mixes records from one model, but their cost provenance can differ
366 * when only some records carried a reported cost. The weakest provenance in the
367 * bucket wins so the UI never overstates confidence.
368 */
369function resolveCostSource(bucket: MutableBucket): UsageBucket["costSource"] {
370 if (bucket.unpricedRecords === bucket.records) return "unpriced";
371 if (bucket.providerReportedRecords === bucket.records) return "providerReported";
372 return "modelPriced";
373}