1/**
2 * Usage history for OpenCode, read from its SQLite databases and legacy JSON
3 * message store under each data directory.
4 *
5 * @module provider-opencode/server/usage
6 */
7
8import { expandHomePath } from "@t3tools/provider-core/server/pathExpansion";
9import {
10 totalTokens,
11 type ProviderUsageReader,
12 type UsageRecord,
13} from "@t3tools/provider-core/server/usage";
14import * as HostProcess from "@t3tools/shared/HostProcess";
15import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient";
16import * as Effect from "effect/Effect";
17import * as FileSystem from "effect/FileSystem";
18import * as Option from "effect/Option";
19import * as Path from "effect/Path";
20import type * as PlatformError from "effect/PlatformError";
21import * as SqlClient from "effect/sql/SqlClient";
22
23import type { OpenCodeSettings } from "../settings.ts";
24
25function object(value: unknown): Record<string, unknown> {
26 return typeof value === "object" && value !== null && !Array.isArray(value)
27 ? (value as Record<string, unknown>)
28 : {};
29}
30
31function tokens(value: unknown): number {
32 return typeof value === "number" && Number.isFinite(value) && value > 0 ? Math.trunc(value) : 0;
33}
34
35function text(value: unknown): string {
36 return typeof value === "string" ? value : "";
37}
38
39/** OpenCode stores uncached input and reasoning separately from input/output. */
40function parseOpenCodeMessage(
41 source: string,
42 fallback: {
43 readonly id?: string;
44 readonly sessionId?: string;
45 readonly timestampMs?: number;
46 } = {},
47): UsageRecord | null {
48 let parsed: unknown;
49 try {
50 parsed = JSON.parse(source);
51 } catch {
52 return null;
53 }
54 const message = object(parsed);
55 if (message.role !== undefined && message.role !== "assistant") return null;
56 const usage = object(message.tokens);
57 const cache = object(usage.cache);
58 const modelReference = object(message.model);
59 const model = text(modelReference.id) || text(modelReference.modelID) || text(message.modelID);
60 const timestampMs = object(message.time).created ?? fallback.timestampMs;
61 if (!model || typeof timestampMs !== "number" || !Number.isFinite(timestampMs)) return null;
62 const reasoningTokens = tokens(usage.reasoning);
63 const totals = {
64 uncachedInputTokens: tokens(usage.input),
65 cachedInputTokens: tokens(cache.read),
66 cacheCreationTokens: tokens(cache.write),
67 outputTokens: tokens(usage.output) + reasoningTokens,
68 reasoningTokens,
69 };
70 if (totalTokens(totals) === 0) return null;
71 const id = fallback.id || text(message.id);
72 const cost = message.cost;
73 return {
74 provider: "opencode",
75 timestampMs,
76 model,
77 sessionId: fallback.sessionId || text(message.sessionID),
78 totals,
79 // OpenCode writes zero for models without a known rate, including paid
80 // subscription models. Let the shared price table estimate those records.
81 reportedCostUsd: typeof cost === "number" && Number.isFinite(cost) && cost > 0 ? cost : null,
82 speed: "standard",
83 dedupeKey: id ? `opencode:${id}` : null,
84 };
85}
86
87export interface OpenCodeUsageReadResult {
88 readonly files: readonly { readonly path: string; readonly records: readonly UsageRecord[] }[];
89 readonly missing: boolean;
90 readonly error: boolean;
91}
92
93const isNotFound = (cause: PlatformError.PlatformError) => cause.reason._tag === "NotFound";
94const MESSAGE_PAGE_SIZE = 64;
95
96/** Whether the path is itself a symlink, without following it. */
97const isSymbolicLink = Effect.fnUntraced(function* (path: string) {
98 const fileSystem = yield* FileSystem.FileSystem;
99 return yield* Effect.isSuccess(fileSystem.readLink(path));
100});
101
102/** Stat of a regular file or directory; a symlink reports as itself and is never followed. */
103const entryInfo = Effect.fn("entryInfo")(function* (path: string) {
104 const fileSystem = yield* FileSystem.FileSystem;
105 if (yield* isSymbolicLink(path))
106 return { type: "SymbolicLink" as const, mtime: Option.none<Date>() };
107 return yield* fileSystem.stat(path);
108});
109
110/**
111 * Stats run in the libuv thread pool, so a few at a time cut the walk's wall
112 * time without lengthening any event-loop turn.
113 */
114const LEGACY_STAT_CONCURRENCY = 4;
115
116/** Entries stat'd before any is processed, so one huge directory is not held in memory at once. */
117const LEGACY_STAT_BATCH = 256;
118
119/** Reads current SQLite and pre-migration JSON stores without modifying either. */
120export const readOpenCodeUsage = Effect.fn("readOpenCodeUsage")(function* (
121 root: string,
122 sinceMs: number,
123): Effect.fn.Return<OpenCodeUsageReadResult, never, FileSystem.FileSystem | Path.Path> {
124 const fileSystem = yield* FileSystem.FileSystem;
125 const path = yield* Path.Path;
126 const files: { path: string; records: UsageRecord[] }[] = [];
127 const seen = new Set<string>();
128 let found = false;
129 let error = false;
130 const append = (records: UsageRecord[], record: UsageRecord | null) => {
131 if (record === null || record.timestampMs < sinceMs) return;
132 if (record.dedupeKey !== null) {
133 if (seen.has(record.dedupeKey)) return;
134 seen.add(record.dedupeKey);
135 }
136 records.push(record);
137 };
138
139 const databases = yield* fileSystem.readDirectory(root).pipe(
140 Effect.flatMap((names) =>
141 Effect.filter(
142 names.filter((name) => /^opencode(?:-[a-zA-Z0-9_-]+)?\.db$/.test(name)),
143 // One database removed between listing and stat must not hide the rest.
144 (name) =>
145 entryInfo(path.join(root, name)).pipe(
146 Effect.map((info) => info.type === "File"),
147 Effect.catchTags({
148 PlatformError: (cause) =>
149 isNotFound(cause) ? Effect.succeed(false) : Effect.fail(cause),
150 }),
151 ),
152 ),
153 ),
154 Effect.map((names) =>
155 names.sort((a, b) =>
156 a === "opencode.db" ? -1 : b === "opencode.db" ? 1 : a.localeCompare(b),
157 ),
158 ),
159 Effect.catchTags({
160 PlatformError: (cause) => {
161 if (!isNotFound(cause)) error = true;
162 return Effect.succeed([]);
163 },
164 }),
165 );
166 for (const name of databases) {
167 found = true;
168 const file = { path: path.join(root, name), records: [] as UsageRecord[] };
169 files.push(file);
170 yield* Effect.gen(function* () {
171 const sql = yield* SqlClient.SqlClient;
172 // A busy live provider should fail this source promptly rather than
173 // stalling the server while SQLite waits for its writer.
174 yield* sql.unsafe("PRAGMA busy_timeout = 100");
175 const tables = new Set(
176 (yield* sql.unsafe<{ readonly name: unknown }>(
177 "SELECT name FROM sqlite_master WHERE type = 'table'",
178 )).map((row) => row.name),
179 );
180 if (!tables.has("message") && !tables.has("session_message")) error = true;
181 for (const table of ["message", "session_message"] as const) {
182 if (!tables.has(table)) continue;
183 const columns = new Set(
184 (yield* sql.unsafe<{ readonly name: unknown }>(`PRAGMA table_info(${table})`)).map(
185 (row) => row.name,
186 ),
187 );
188 const timestamp = columns.has("time_created") ? "time_created" : "NULL";
189 const predicates = table === "session_message" ? ["type = 'assistant'"] : [];
190 if (timestamp !== "NULL") predicates.push("time_created >= ?");
191 predicates.push("rowid > ?", "rowid <= ?");
192 const [range] = yield* sql.unsafe<{
193 readonly first_row_id: number | null;
194 readonly last_row_id: number | null;
195 }>(
196 `SELECT (SELECT MIN(rowid) FROM ${table}) AS first_row_id, (SELECT MAX(rowid) FROM ${table}) AS last_row_id`,
197 );
198 const firstRowId = range?.first_row_id;
199 const lastRowId = range?.last_row_id;
200 if (typeof firstRowId !== "number" || typeof lastRowId !== "number") continue;
201 // SqlClient materializes each result with StatementSync.all(), and the
202 // window can exclude most rows. Each statement covers MESSAGE_PAGE_SIZE
203 // rowids, however few match, and rows appended mid-scan are left out.
204 // Untraced, as a large history runs thousands of them.
205 for (let after = firstRowId - 1; after < lastRowId; after += MESSAGE_PAGE_SIZE) {
206 const rows = yield* sql
207 .unsafe<{
208 readonly id: unknown;
209 readonly session_id: unknown;
210 readonly data: unknown;
211 readonly created: unknown;
212 }>(
213 `SELECT id, session_id, data, ${timestamp} AS created
214 FROM ${table} WHERE ${predicates.join(" AND ")} ORDER BY rowid`,
215 [
216 ...(timestamp === "NULL" ? [] : [sinceMs]),
217 after,
218 Math.min(after + MESSAGE_PAGE_SIZE, lastRowId),
219 ],
220 )
221 .pipe(Effect.withTracerEnabled(false));
222 for (const row of rows) {
223 append(
224 file.records,
225 parseOpenCodeMessage(text(row.data), {
226 id: text(row.id),
227 sessionId: text(row.session_id),
228 ...(typeof row.created === "number" ? { timestampMs: row.created } : {}),
229 }),
230 );
231 }
232 yield* Effect.yieldNow;
233 }
234 }
235 }).pipe(
236 Effect.provide(NodeSqliteClient.layer({ filename: file.path, readonly: true })),
237 Effect.catchTags({
238 SqlError: () => {
239 error = true;
240 return Effect.void;
241 },
242 }),
243 );
244 }
245
246 // Do not follow symlinks, including cycles. Database records win over their
247 // old JSON copies when OpenCode has migrated a store in place.
248 const directories = [path.join(root, "storage", "message")];
249 while (directories.length > 0) {
250 const directory = directories.pop()!;
251 yield* Effect.gen(function* () {
252 const names = yield* fileSystem.readDirectory(directory);
253 for (let start = 0; start < names.length; start += LEGACY_STAT_BATCH) {
254 const batch = names.slice(start, start + LEGACY_STAT_BATCH);
255 const entries = batch.map((name) => path.join(directory, name));
256 // `stat` follows symlinks and `readLink` fails, expensively, for
257 // everything else. Stat a batch, then ask whether an entry is a symlink
258 // only where the answer changes the result.
259 const infos = yield* Effect.forEach(
260 entries,
261 (entry) =>
262 fileSystem.stat(entry).pipe(
263 Effect.catchTags({
264 // A symlink whose target cannot be stat'd is skipped, not an error.
265 PlatformError: (cause) =>
266 isNotFound(cause)
267 ? Effect.succeed(null)
268 : isSymbolicLink(entry).pipe(
269 Effect.flatMap((link) =>
270 link ? Effect.succeed(null) : Effect.fail(cause),
271 ),
272 ),
273 }),
274 Effect.exit,
275 ),
276 { concurrency: LEGACY_STAT_CONCURRENCY },
277 );
278 for (const [index, name] of batch.entries()) {
279 const entry = entries[index]!;
280 const info = yield* infos[index]!;
281 if (info?.type === "Directory") {
282 if (!(yield* isSymbolicLink(entry))) directories.push(entry);
283 } else if (info?.type === "File" && name.endsWith(".json")) {
284 const id = name.slice(0, -5);
285 // A message cannot be created after its file was last written, so a
286 // file untouched since the window opened holds nothing in range.
287 const skip =
288 seen.has(`opencode:${id}`) ||
289 Option.exists(info.mtime, (mtime) => mtime.getTime() < sinceMs);
290 // A skipped file only matters as proof that the store exists.
291 if (found && skip) continue;
292 if (yield* isSymbolicLink(entry)) continue;
293 found = true;
294 if (skip) continue;
295 const file = { path: entry, records: [] as UsageRecord[] };
296 files.push(file);
297 yield* fileSystem.readFileString(entry).pipe(
298 Effect.map((source) => append(file.records, parseOpenCodeMessage(source, { id }))),
299 Effect.catchTags({
300 PlatformError: (cause) => {
301 if (!isNotFound(cause)) error = true;
302 return Effect.void;
303 },
304 }),
305 );
306 }
307 }
308 }
309 }).pipe(
310 Effect.catchTags({
311 PlatformError: (cause) => {
312 if (!isNotFound(cause)) error = true;
313 return Effect.void;
314 },
315 }),
316 );
317 }
318 return { files, missing: !found && !error, error };
319});
320
321/**
322 * The data directories to read: `OPENCODE_DATA_DIR` (comma-separated) or the
323 * XDG default, canonicalized so aliases count once.
324 */
325const resolveOpenCodeDataDirs = Effect.fn("resolveOpenCodeDataDirs")(function* () {
326 const fileSystem = yield* FileSystem.FileSystem;
327 const path = yield* Path.Path;
328 const environment = yield* HostProcess.Environment;
329 const homeDirectory = yield* HostProcess.HomeDirectory;
330 const roots = environment["OPENCODE_DATA_DIR"]
331 ?.split(",")
332 .map((value) => value.trim())
333 .filter(Boolean);
334 const dataHome = environment["XDG_DATA_HOME"]?.trim();
335 const defaults = [
336 path.join(
337 dataHome && path.isAbsolute(dataHome)
338 ? dataHome
339 : path.join(homeDirectory, ".local", "share"),
340 "opencode",
341 ),
342 ];
343 const canonical = new Set<string>();
344 for (const root of roots?.length ? roots : defaults) {
345 const resolved = path.resolve(expandHomePath(root, homeDirectory));
346 canonical.add(yield* fileSystem.realPath(resolved).pipe(Effect.orElseSucceed(() => resolved)));
347 }
348 return [...canonical];
349});
350
351export type OpenCodeUsageReaderEnv = FileSystem.FileSystem | Path.Path;
352
353export const openCodeUsageReader: ProviderUsageReader<OpenCodeSettings, OpenCodeUsageReaderEnv> = {
354 kind: "scan",
355 provider: "opencode",
356 scan: Effect.fn("openCodeUsageReader.scan")(function* ({ windowStartMs }) {
357 const roots = yield* resolveOpenCodeDataDirs();
358 return yield* Effect.forEach(
359 roots,
360 (dir) =>
361 readOpenCodeUsage(dir, windowStartMs).pipe(
362 Effect.map((result) => ({
363 dir,
364 files: result.missing && !result.error ? null : result.files,
365 status: result.error ? ("partial" as const) : ("ok" as const),
366 ...(result.error ? { message: "Some OpenCode history could not be read." } : {}),
367 })),
368 ),
369 { concurrency: "unbounded" },
370 );
371 }),
372};