1import { ArchiveError, type ArchiveStatus } from "@t3tools/contracts";
2// Allocated filesystem blocks are unavailable through Effect FileSystem.
3// @effect-diagnostics nodeBuiltinImport:off
4import * as NodeFSP from "node:fs/promises";
5import * as HostProcess from "@t3tools/shared/HostProcess";
6import * as Config from "effect/Config";
7import * as FileSystem from "effect/FileSystem";
8import * as Clock from "effect/Clock";
9import * as Context from "effect/Context";
10import * as Effect from "effect/Effect";
11import * as Layer from "effect/Layer";
12import * as Path from "effect/Path";
13import * as Redacted from "effect/Redacted";
14import * as Schema from "effect/Schema";
15import * as SqlClient from "effect/sql/SqlClient";
16import * as ServerConfig from "../config.ts";
17import * as ServerSettings from "../serverSettings.ts";
18
19const storageError = (cause: unknown) =>
20 new ArchiveError({ reason: "storage", cause: Redacted.make(cause) });
21const isMissingFile = Schema.is(Schema.Struct({ code: Schema.Literal("ENOENT") }));
22const Home = Schema.Struct({
23 homePath: Schema.optionalKey(Schema.String),
24 shadowHomePath: Schema.optionalKey(Schema.String),
25});
26const decodeHome = Schema.decodeOption(Home);
27
28export class ArchiveStorage extends Context.Service<
29 ArchiveStorage,
30 {
31 readonly inspect: Effect.Effect<ArchiveStatus["storage"], ArchiveError>;
32 readonly reclaim: Effect.Effect<void, ArchiveError>;
33 }
34>()("t3/archive/ArchiveStorage") {}
35
36export const layer = Layer.effect(
37 ArchiveStorage,
38 Effect.gen(function* () {
39 const config = yield* ServerConfig.ServerConfig;
40 const settings = yield* ServerSettings.ServerSettingsService;
41 const home = yield* HostProcess.HomeDirectory;
42 const path = yield* Path.Path;
43 const sql = yield* SqlClient.SqlClient;
44 const fs = yield* FileSystem.FileSystem;
45 const inheritedCodexHome = yield* Config.option(Config.String("CODEX_HOME")).pipe(Effect.orDie);
46 const inheritedClaudeHome = yield* Config.option(Config.String("CLAUDE_CONFIG_DIR")).pipe(
47 Effect.orDie,
48 );
49 const measure = (roots: ReadonlyArray<string>) =>
50 Effect.tryPromise({
51 try: async () => {
52 let bytes = 0;
53 let allocated = 0;
54 const visited = new Set<string>();
55 const pending = [...roots];
56 while (pending.length) {
57 const file = pending.pop()!;
58 let stat;
59 try {
60 stat = await NodeFSP.lstat(file);
61 } catch (error) {
62 if (isMissingFile(error)) continue;
63 throw error;
64 }
65 if (stat.isSymbolicLink()) continue;
66 const identity = `${stat.dev}:${stat.ino}`;
67 if (visited.has(identity)) continue;
68 visited.add(identity);
69 if (stat.isDirectory())
70 for (const name of await NodeFSP.readdir(file)) pending.push(path.join(file, name));
71 else if (stat.isFile()) {
72 bytes += stat.size;
73 allocated += stat.blocks * 512;
74 }
75 }
76 return { bytes, allocated };
77 },
78 catch: storageError,
79 });
80 const databaseFiles = Effect.gen(function* () {
81 const databases = yield* sql<{ name: string; file: string }>`PRAGMA database_list`;
82 const file = databases.find((database) => database.name === "main")?.file;
83 return file ? [file, `${file}-wal`, `${file}-shm`] : [];
84 });
85 const inspect = Effect.gen(function* () {
86 const configured = yield* settings.getSettings;
87 const roots = [config.providerLogsDir];
88 const homes = new Map<string, Set<string>>([
89 [
90 "codex",
91 new Set([
92 path.join(home, ".codex"),
93 path.join(config.stateDir, "providers/codex/shared"),
94 ]),
95 ],
96 [
97 "claudeAgent",
98 new Set([
99 path.join(home, ".claude"),
100 path.join(config.stateDir, "providers/claudeAgent/shared"),
101 ]),
102 ],
103 ]);
104 if (inheritedCodexHome._tag === "Some")
105 homes.get("codex")!.add(path.resolve(inheritedCodexHome.value));
106 if (inheritedClaudeHome._tag === "Some")
107 homes.get("claudeAgent")!.add(path.resolve(inheritedClaudeHome.value));
108 for (const [id, instance] of Object.entries(configured.providerInstances)) {
109 if (instance.driver !== "codex" && instance.driver !== "claudeAgent") continue;
110 const decoded = decodeHome(instance.config ?? {});
111 const values =
112 decoded._tag === "Some" ? [decoded.value.homePath, decoded.value.shadowHomePath] : [];
113 for (const variable of instance.environment ?? [])
114 if (variable.name === (instance.driver === "codex" ? "CODEX_HOME" : "CLAUDE_CONFIG_DIR"))
115 values.push(variable.value);
116 if (instance.driver === "codex" && id !== "codex")
117 values.push(path.join(config.stateDir, "providers", "codex", id, "shadow"));
118 for (const value of values) {
119 const raw = value?.trim();
120 if (!raw) continue;
121 const resolved = raw.startsWith("~/") ? path.join(home, raw.slice(2)) : path.resolve(raw);
122 homes.get(instance.driver)!.add(resolved);
123 }
124 }
125 for (const root of homes.get("codex")!) {
126 for (const directory of ["sessions", "archived_sessions", "sqlite", "logs", "log"])
127 roots.push(path.join(root, directory));
128 if (!(yield* fs.exists(root))) continue;
129 for (const name of yield* fs.readDirectory(root))
130 if (
131 /^(?:(?:state|logs)_\d+\.sqlite(?:-wal|-shm)?|history\.jsonl|session_index\.jsonl)$/.test(
132 name,
133 )
134 )
135 roots.push(path.join(root, name));
136 }
137 for (const root of homes.get("claudeAgent")!) roots.push(path.join(root, "projects"));
138 const database = yield* measure(yield* databaseFiles);
139 const attachments = yield* measure([config.attachmentsDir]);
140 const native = yield* measure(roots);
141 return {
142 allocatedDatabaseBytes: database.allocated,
143 attachmentsBytes: attachments.bytes,
144 nativeProviderLogBytes: native.bytes,
145 };
146 }).pipe(Effect.mapError(storageError));
147 const reclaim = Effect.gen(function* () {
148 const pending = yield* sql<{
149 pending: number;
150 }>`SELECT pending FROM transcript_archive_reclamation WHERE singleton = 1`;
151 if (!pending[0]?.pending) return;
152 const files = yield* databaseFiles;
153 const before = yield* measure(files);
154 // VACUUM runs on the existing SQLite writer, after the verified offload transaction commits.
155 yield* sql`VACUUM`;
156 const checkpoint = yield* sql<{ busy: number }>`PRAGMA wal_checkpoint(TRUNCATE)`;
157 if (checkpoint[0]?.busy) return yield* new ArchiveError({ reason: "storage" });
158 const after = yield* measure(files);
159 const now = yield* Clock.currentTimeMillis;
160 yield* sql`UPDATE transcript_archive_reclamation SET pending = 0, completed_at = ${now}, before_allocated_bytes = ${before.allocated}, after_allocated_bytes = ${after.allocated} WHERE singleton = 1`;
161 }).pipe(Effect.mapError(storageError));
162 return ArchiveStorage.of({ inspect, reclaim });
163 }),
164);