1// @effect-diagnostics nodeBuiltinImport:off
2/**
3 * Raw filesystem access for transcript scanning.
4 *
5 * Isolated here so the rest of the usage code stays on Effect's `FileSystem`.
6 * The direct `node:fs` streaming is deliberate: a cold 30-day window is ~1.4 GB
7 * across ~1,500 files, and buffer-level streaming is roughly an order of
8 * magnitude cheaper than materialising each file. The equivalent Effect stream
9 * pipeline is idiomatic but not fast enough to sit behind a page load.
10 *
11 * Transcripts are append-only, so a parse also reports the byte position it
12 * stopped at. A later scan of the same file resumes from that position and
13 * parses only the appended bytes, which is what keeps a warm scan cheap while a
14 * session is actively writing a multi-hundred-megabyte rollout.
15 *
16 * @module usageTranscriptReader
17 */
18import * as NodeFSP from "node:fs/promises";
19import * as NodePath from "node:path";
20import * as NodeStringDecoder from "node:string_decoder";
21
22import type {
23 SelectedFields,
24 TranscriptUsageFormat,
25 UsageRecord,
26} from "@t3tools/provider-core/server/usage";
27import * as Option from "effect/Option";
28import * as Schema from "effect/Schema";
29
30import { createTranscriptJsonReader } from "../project/AgentSessionJson.ts";
31
32export interface TranscriptFile {
33 readonly path: string;
34 readonly size: number;
35 readonly mtimeMs: number;
36}
37
38/**
39 * Where a parse stopped, with enough state to continue from there.
40 *
41 * The guard hash fingerprints the bytes immediately before `resumeOffset`. A
42 * resume only proceeds when those bytes still match: transcripts are
43 * append-only by design, but a rotated or rewritten file silently mis-parsed
44 * from the middle would corrupt usage totals. The window is a cheap tripwire
45 * for those realistic failure shapes, all of which disturb the file's tail at
46 * that exact offset; it deliberately does not hash the whole prefix, which
47 * would cost the full re-read the resume exists to avoid.
48 */
49export interface TranscriptParsePosition {
50 /** Byte offset just past the last newline-terminated line consumed. */
51 readonly resumeOffset: number;
52 /** Length of the fingerprinted window ending at `resumeOffset`. */
53 readonly guardLength: number;
54 /** FNV-1a hash of that window. */
55 readonly guardHash: number;
56 /**
57 * The format's reducer state as of `resumeOffset`, encoded by its schema;
58 * `null` for stateless formats.
59 */
60 readonly state: unknown;
61}
62
63export interface TranscriptParseResult {
64 /** Records from newline-terminated lines at or after the parse start. */
65 readonly records: readonly UsageRecord[];
66 /**
67 * Records from a trailing segment the writer has not newline-terminated yet.
68 * Kept out of `records` because `position` deliberately excludes that
69 * segment: the next scan re-reads it once the writer finishes the line.
70 */
71 readonly tailRecords: readonly UsageRecord[];
72 readonly position: TranscriptParsePosition;
73 /** Whether the parse continued from `resumeFrom` rather than byte 0. */
74 readonly resumed: boolean;
75}
76
77/** 64 bytes of JSONL tail is ample to distinguish a replaced file. */
78export const GUARD_LENGTH = 64;
79// Native parsing is faster for common 1–4 MiB context/tool records. Above
80// 8 MiB, project usage without allocating the whole record. This switches
81// readers; it never discards a record because of its size.
82const STREAMING_THRESHOLD_BYTES = 8 * 1024 * 1024;
83const NEWLINE = 0x0a;
84/** `stat` calls one transcript walk keeps in flight. The libuv pool has 4 threads. */
85const STAT_CONCURRENCY = 32;
86const CARRIAGE_RETURN = 0x0d;
87
88function selectUsageFields(fields: SelectedFields) {
89 return (path: ReadonlyArray<string | number | null>): boolean => {
90 let selected: true | SelectedFields = fields;
91 for (const key of path) {
92 if (selected === true) return true;
93 if (typeof key !== "string" || !Object.hasOwn(selected, key)) return false;
94 selected = selected[key]!;
95 }
96 return true;
97 };
98}
99
100function fnv1a(buffer: Buffer): number {
101 let hash = 0x811c9dc5;
102 for (let index = 0; index < buffer.length; index += 1) {
103 hash ^= buffer[index]!;
104 hash = Math.imul(hash, 0x01000193);
105 }
106 return hash >>> 0;
107}
108
109/**
110 * Lists `.jsonl` transcripts under `root` last modified at or after `sinceMs`.
111 *
112 * Unreadable directories and files are counted in `failedPaths` rather than
113 * failing the page, so the source can report incomplete usage. Files that
114 * vanish between `readdir` and `stat` are ordinary rotation and not counted.
115 *
116 * `fileName` restricts the walk to a single basename (Grok's `updates.jsonl`).
117 * Grok sessions also ship multi-megabyte `chat_history` and `events` logs that
118 * never carry usage, so the basename filter keeps a cold scan off those files.
119 *
120 * Directories are listed depth-first, one at a time, then the candidates are
121 * stat'd by a fixed pool of workers: a warm scan stats thousands of files, and
122 * one at a time each waits its own trip through the thread pool. Results keep
123 * `readdir` order, which the aggregator's first-seen dedupe relies on.
124 */
125export async function listTranscriptFiles(
126 root: string,
127 sinceMs: number,
128 options?: { readonly fileName?: string },
129): Promise<{ readonly files: readonly TranscriptFile[]; readonly failedPaths: number }> {
130 const fileName = options?.fileName;
131 const candidates: string[] = [];
132 let failedPaths = 0;
133 const walk = async (dir: string): Promise<void> => {
134 let entries;
135 try {
136 entries = await NodeFSP.readdir(dir, { withFileTypes: true });
137 } catch {
138 failedPaths += 1;
139 return;
140 }
141 for (const entry of entries) {
142 const child = NodePath.join(dir, entry.name);
143 if (entry.isDirectory()) await walk(child);
144 else if (fileName !== undefined ? entry.name === fileName : entry.name.endsWith(".jsonl")) {
145 candidates.push(child);
146 }
147 }
148 };
149 await walk(root);
150
151 const found: Array<TranscriptFile | undefined> = Array.from({ length: candidates.length });
152 // Each worker pulls the next candidate from one shared iterator.
153 const queue = candidates.entries();
154 const statQueued = async (): Promise<void> => {
155 for (const [index, path] of queue) {
156 try {
157 const stats = await NodeFSP.stat(path);
158 if (stats.mtimeMs >= sinceMs) {
159 found[index] = { path, size: stats.size, mtimeMs: stats.mtimeMs };
160 }
161 } catch (error) {
162 if (!(error instanceof Error && "code" in error && error.code === "ENOENT")) {
163 failedPaths += 1;
164 }
165 }
166 }
167 };
168 await Promise.all(
169 Array.from({ length: Math.min(STAT_CONCURRENCY, candidates.length) }, statQueued),
170 );
171 return { files: found.filter((file) => file !== undefined), failedPaths };
172}
173
174/**
175 * Filesystem identity of a directory, as `device:inode`.
176 *
177 * Used to tell "two servers reading the same transcript directory" apart from
178 * "two machines whose hostname and home path happen to match". Returns an empty
179 * string when the directory cannot be stat'd.
180 */
181export async function readDirectoryVolumeId(path: string): Promise<string> {
182 try {
183 const stats = await NodeFSP.stat(path);
184 return `${stats.dev}:${stats.ino}`;
185 } catch {
186 return "";
187 }
188}
189
190async function guardMatches(
191 handle: NodeFSP.FileHandle,
192 position: TranscriptParsePosition,
193): Promise<boolean> {
194 if (position.guardLength <= 0 || position.guardLength > GUARD_LENGTH) return false;
195 try {
196 const window = Buffer.alloc(position.guardLength);
197 const { bytesRead } = await handle.read(
198 window,
199 0,
200 position.guardLength,
201 position.resumeOffset - position.guardLength,
202 );
203 return bytesRead === position.guardLength && fnv1a(window) === position.guardHash;
204 } catch {
205 return false;
206 }
207}
208
209/**
210 * Builds the byte-level test for a format's `usageMarkers`: whether a line
211 * contains any of them.
212 *
213 * One marker is a single `Buffer` search. Several would be one pass over the
214 * line each, so they are matched together as one alternation over the line's
215 * latin1 view instead: latin1 maps every byte to one character without
216 * validating it, so an ASCII marker matches exactly where its bytes do.
217 */
218export function transcriptLineGate(markers: readonly [string, ...string[]]) {
219 if (markers.length === 1) {
220 const marker = Buffer.from(markers[0]);
221 return (line: Buffer) => line.includes(marker);
222 }
223 const pattern = new RegExp(
224 markers.map((marker) => marker.replace(/[.*+?^${}()|[\]\\]/g, "\\$&")).join("|"),
225 );
226 return (line: Buffer) => pattern.test(line.toString("latin1"));
227}
228
229/**
230 * Streams one transcript and returns the usage records it contains, or `null`
231 * when the file could not be read.
232 *
233 * The distinction matters to the caller's cache: a genuinely empty transcript
234 * is a stable fact worth memoising, while a transient read failure memoised
235 * under the same `(size, mtime)` key would silently drop that file's usage
236 * until the file next changes.
237 *
238 * With `resumeFrom`, parsing continues from that position when its guard bytes
239 * still match, so only appended lines are read; otherwise the whole file is
240 * re-parsed from the start and `resumed` reports `false`.
241 *
242 * A stateful format resumes with the state saved at that position; a saved
243 * state its schema rejects re-parses the file whole.
244 */
245export async function readTranscriptRecords<State>(
246 filePath: string,
247 format: TranscriptUsageFormat<State>,
248 resumeFrom?: TranscriptParsePosition,
249 options?: { readonly streamingThresholdBytes?: number },
250): Promise<TranscriptParseResult | null> {
251 const streamingThresholdBytes = options?.streamingThresholdBytes ?? STREAMING_THRESHOLD_BYTES;
252 let handle: NodeFSP.FileHandle;
253 try {
254 handle = await NodeFSP.open(filePath, "r");
255 } catch {
256 return null;
257 }
258
259 try {
260 const codec = format.state;
261 const decodeState = codec === undefined ? undefined : Schema.decodeUnknownOption(codec.schema);
262 const encodeState = codec === undefined ? undefined : Schema.encodeSync(codec.schema);
263 // Stateless formats never read their state argument.
264 let state = codec === undefined ? (undefined as State) : codec.initial();
265 let resumed = false;
266 let start = 0;
267 const savedState =
268 resumeFrom === undefined || decodeState === undefined || resumeFrom.state === null
269 ? Option.none()
270 : decodeState(resumeFrom.state);
271 if (
272 resumeFrom !== undefined &&
273 resumeFrom.resumeOffset > 0 &&
274 (codec === undefined || Option.isSome(savedState)) &&
275 (await guardMatches(handle, resumeFrom))
276 ) {
277 if (Option.isSome(savedState)) state = savedState.value;
278 start = resumeFrom.resumeOffset;
279 resumed = true;
280 }
281
282 const toLineString = (lineBuffer: Buffer): string => {
283 const content =
284 lineBuffer.length > 0 && lineBuffer[lineBuffer.length - 1] === CARRIAGE_RETURN
285 ? lineBuffer.subarray(0, -1)
286 : lineBuffer;
287 return content.toString("utf8");
288 };
289
290 // Gate on the raw bytes so the lines that fail it, which is most of them,
291 // are never decoded.
292 const mightCarryUsage = transcriptLineGate(format.usageMarkers);
293 const parseLine = (lineBuffer: Buffer, lineState: State, out: UsageRecord[]): void => {
294 if (!mightCarryUsage(lineBuffer)) return;
295 for (const record of format.parseLine(toLineString(lineBuffer), lineState)) out.push(record);
296 };
297
298 const records: UsageRecord[] = [];
299 // Byte offsets remain independent of UTF-8 decoding. Only complete lines
300 // commit the resume point; an unfinished tail is replayed on the next scan.
301 let resumeOffset = start;
302 let scanOffset = start;
303 let pendingChunks: Buffer[] = [];
304 let pendingBytes = 0;
305 let streaming: ReturnType<typeof createTranscriptJsonReader> | undefined;
306 let decoder: NodeStringDecoder.StringDecoder | undefined;
307 const selectPath = selectUsageFields(format.selectFields);
308
309 const append = (segment: Buffer) => {
310 if (!streaming && pendingBytes + segment.length <= streamingThresholdBytes) {
311 if (segment.length > 0) pendingChunks.push(segment);
312 pendingBytes += segment.length;
313 return;
314 }
315 if (!streaming) {
316 // Usage has no import-history budget: retain all selected usage fields,
317 // regardless of the size of the surrounding unselected tool content.
318 streaming = createTranscriptJsonReader(() => {}, selectPath, { maxDepth: Infinity });
319 decoder = new NodeStringDecoder.StringDecoder("utf8");
320 for (const pending of pendingChunks) streaming.write(decoder.write(pending));
321 pendingChunks = [];
322 pendingBytes = 0;
323 }
324 streaming.write(decoder!.write(segment));
325 };
326 const finish = (lineState: State, out: UsageRecord[]) => {
327 if (streaming) {
328 streaming.write(decoder!.end());
329 const projected = streaming.finish();
330 for (const record of format.parseProjected(projected, lineState)) out.push(record);
331 } else if (pendingBytes > 0) {
332 const line =
333 pendingChunks.length === 1
334 ? pendingChunks[0]!
335 : Buffer.concat(pendingChunks, pendingBytes);
336 parseLine(line, lineState, out);
337 }
338 pendingChunks = [];
339 pendingBytes = 0;
340 streaming = undefined;
341 decoder = undefined;
342 };
343 const stream = handle.createReadStream({
344 start,
345 autoClose: false,
346 highWaterMark: 256 * 1024,
347 }) as AsyncIterable<Buffer>;
348 for await (const chunk of stream) {
349 let lineStart = 0;
350 while (lineStart < chunk.length) {
351 const newlineIndex = chunk.indexOf(NEWLINE, lineStart);
352 if (newlineIndex === -1) {
353 append(chunk.subarray(lineStart));
354 break;
355 }
356 // Most lines fit in the current chunk. Avoid buffering/streaming
357 // machinery on this hot path.
358 if (!streaming && pendingBytes === 0) {
359 parseLine(chunk.subarray(lineStart, newlineIndex), state, records);
360 } else {
361 append(chunk.subarray(lineStart, newlineIndex));
362 finish(state, records);
363 }
364 lineStart = newlineIndex + 1;
365 resumeOffset = scanOffset + lineStart;
366 }
367 scanOffset += chunk.length;
368 }
369
370 // The unfinished tail parses against a copy, so the saved state stays at
371 // `resumeOffset` and the next scan replays the tail from it.
372 const encodedState = encodeState === undefined ? null : encodeState(state);
373 const tailRecords: UsageRecord[] = [];
374 finish(
375 decodeState === undefined ? state : Option.getOrThrow(decodeState(encodedState)),
376 tailRecords,
377 );
378
379 const guardLength = Math.min(GUARD_LENGTH, resumeOffset);
380 let guardHash = 0;
381 if (guardLength > 0) {
382 const window = Buffer.alloc(guardLength);
383 await handle.read(window, 0, guardLength, resumeOffset - guardLength);
384 guardHash = fnv1a(window);
385 }
386
387 return {
388 records,
389 tailRecords,
390 position: {
391 resumeOffset,
392 guardLength,
393 guardHash,
394 state: encodedState,
395 },
396 resumed,
397 };
398 } catch {
399 return null;
400 } finally {
401 await handle.close().catch(() => undefined);
402 }
403}