1import * as HostProcess from "@t3tools/shared/HostProcess";
2import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient";
3import * as Context from "effect/Context";
4import * as Effect from "effect/Effect";
5import * as FileSystem from "effect/FileSystem";
6import * as Layer from "effect/Layer";
7import * as Option from "effect/Option";
8import * as Path from "effect/Path";
9import * as Schedule from "effect/Schedule";
10import * as Schema from "effect/Schema";
11// @effect-diagnostics-next-line nodeBuiltinImport:off -- Effect has no free-space query.
12import * as NodeFSP from "node:fs/promises";
13import * as NodeOS from "node:os";
14import * as SqlClient from "effect/sql/SqlClient";
15import { isSqlError } from "effect/sql/SqlError";
16
17import * as ProjectionMaintenance from "../orchestration-v2/ProjectionMaintenance.ts";
18import { rebuildMessageSearch } from "../orchestration-v2/messageSearchIndex.ts";
19
20export class DatabaseMissingError extends Schema.TaggedError<DatabaseMissingError>()(
21 "DatabaseMissingError",
22 { path: Schema.String },
23) {
24 override get message() {
25 return "No existing T3 Code database was found in this data directory.";
26 }
27}
28
29export class DatabaseInUseError extends Schema.TaggedError<DatabaseInUseError>()(
30 "DatabaseInUseError",
31 {
32 path: Schema.String,
33 cause: Schema.Defect(),
34 pid: Schema.optional(Schema.Int),
35 serviceManaged: Schema.optional(Schema.Boolean),
36 },
37) {
38 override get message() {
39 if (this.pid !== undefined) {
40 return `Stop T3 Code server process ${this.pid}${this.serviceManaged ? "" : ", which the background service did not start,"} before repairing the database.`;
41 }
42 return "Stop the T3 Code server and desktop app, including any version before 0.1.0, before repairing the database.";
43 }
44}
45
46export class DatabaseInspectError extends Schema.TaggedError<DatabaseInspectError>()(
47 "DatabaseInspectError",
48 { path: Schema.String, cause: Schema.Defect() },
49) {
50 override get message() {
51 return "Could not read the database. Check that this user can open it.";
52 }
53}
54
55export class DatabaseCompactError extends Schema.TaggedError<DatabaseCompactError>()(
56 "DatabaseCompactError",
57 { path: Schema.String, cause: Schema.Defect() },
58) {
59 override get message() {
60 return "Database compaction failed. Check that no other process has the database open and enough disk space is available.";
61 }
62}
63
64export class LegacyDatabaseRemoveError extends Schema.TaggedError<LegacyDatabaseRemoveError>()(
65 "LegacyDatabaseRemoveError",
66 { path: Schema.String, cause: Schema.Defect() },
67) {
68 override get message() {
69 return "Could not delete the database from before 0.1.0. Check that this user can write to the data directory.";
70 }
71}
72
73export class DiskSpaceCheckError extends Schema.TaggedError<DiskSpaceCheckError>()(
74 "DiskSpaceCheckError",
75 { path: Schema.String, cause: Schema.Defect() },
76) {
77 override get message() {
78 return "Could not check free disk space.";
79 }
80}
81
82export type DatabaseCompactStep = "prune-events" | "vacuum" | "rebuild-search" | "checkpoint";
83
84export class DatabaseMaintenance extends Context.Service<
85 DatabaseMaintenance,
86 {
87 readonly inspect: (databasePath: string) => Effect.Effect<
88 {
89 readonly databaseBytes: number;
90 readonly reclaimableBytes: number;
91 readonly incrementalVacuumEnabled: boolean;
92 /** Size of the database from before 0.1.0 still kept beside it, or null. */
93 readonly legacyDatabaseBytes: number | null;
94 },
95 DatabaseMissingError | DatabaseInspectError
96 >;
97 /**
98 * Compacts history, then VACUUMs into incremental auto_vacuum. Refuses while a server has it open.
99 * Calls `onStep` as each step starts; VACUUM time grows with the whole database size.
100 */
101 readonly compact: (
102 databasePath: string,
103 onStep?: (step: DatabaseCompactStep) => Effect.Effect<void>,
104 ) => Effect.Effect<
105 {
106 readonly beforeBytes: number;
107 readonly afterBytes: number;
108 readonly deletedEventCount: number;
109 readonly deletedReceiptCount: number;
110 },
111 DatabaseMissingError | DatabaseInUseError | DatabaseCompactError
112 >;
113 /** Deletes only the pre-0.1.0 database; earlier versions still use it. */
114 readonly removeLegacyDatabase: (
115 databasePath: string,
116 ) => Effect.Effect<void, DatabaseInUseError | LegacyDatabaseRemoveError>;
117 }
118>()("t3/persistence/DatabaseMaintenance") {}
119
120const isDomainError = Schema.is(Schema.Union([DatabaseMissingError, DatabaseInUseError]));
121
122const readStorage = Effect.gen(function* () {
123 const sql = yield* SqlClient.SqlClient;
124 const pages = yield* sql<{ readonly page_count: number }>`PRAGMA page_count`;
125 const free = yield* sql<{ readonly freelist_count: number }>`PRAGMA freelist_count`;
126 const size = yield* sql<{ readonly page_size: number }>`PRAGMA page_size`;
127 const mode = yield* sql<{ readonly auto_vacuum: number }>`PRAGMA auto_vacuum`;
128 const pageSize = size[0]?.page_size ?? 0;
129 return {
130 databaseBytes: (pages[0]?.page_count ?? 0) * pageSize,
131 reclaimableBytes: (free[0]?.freelist_count ?? 0) * pageSize,
132 incrementalVacuumEnabled: mode[0]?.auto_vacuum === 2,
133 };
134});
135
136/**
137 * Holds an exclusive lock on the connection's database. In WAL mode this is
138 * refused while any other connection has the file open, even an idle one,
139 * which is how a running server or desktop app is detected.
140 */
141const lockExclusively = Effect.fn("DatabaseMaintenance.lockExclusively")(function* (
142 databasePath: string,
143) {
144 const sql = yield* SqlClient.SqlClient;
145 yield* sql`PRAGMA busy_timeout = 0`;
146 yield* sql`PRAGMA locking_mode = EXCLUSIVE`;
147 yield* sql`BEGIN EXCLUSIVE`.pipe(
148 Effect.mapError((cause) =>
149 isSqlError(cause) && cause.reason._tag === "LockTimeoutError"
150 ? new DatabaseInUseError({ path: databasePath, cause })
151 : cause,
152 ),
153 );
154 yield* sql`COMMIT`;
155});
156
157const make = Effect.gen(function* () {
158 const fs = yield* FileSystem.FileSystem;
159 const path = yield* Path.Path;
160 const legacyPath = (databasePath: string) =>
161 path.join(path.dirname(databasePath), "state.sqlite");
162
163 const locate = Effect.fn("DatabaseMaintenance.locate")(function* (databasePath: string) {
164 if (!(yield* fs.exists(databasePath))) {
165 return yield* new DatabaseMissingError({ path: databasePath });
166 }
167 });
168
169 const inspect = Effect.fn("DatabaseMaintenance.inspect")(
170 function* (databasePath: string) {
171 yield* locate(databasePath);
172 const storage = yield* readStorage.pipe(
173 Effect.provide(NodeSqliteClient.layer({ filename: databasePath, readonly: true })),
174 );
175 const legacy = legacyPath(databasePath);
176 const legacyDatabaseBytes = (yield* fs.exists(legacy))
177 ? Number((yield* fs.stat(legacy)).size)
178 : null;
179 return { ...storage, legacyDatabaseBytes };
180 },
181 (effect, databasePath) =>
182 effect.pipe(
183 Effect.mapError((cause) =>
184 isDomainError(cause) ? cause : new DatabaseInspectError({ path: databasePath, cause }),
185 ),
186 ),
187 );
188
189 const compact = Effect.fn("DatabaseMaintenance.compact")(
190 function* (
191 databasePath: string,
192 onStep: (step: DatabaseCompactStep) => Effect.Effect<void> = () => Effect.void,
193 ) {
194 yield* locate(databasePath);
195 const beforeBytes = Number((yield* fs.stat(databasePath)).size);
196 const result = yield* Effect.gen(function* () {
197 const sql = yield* SqlClient.SqlClient;
198 yield* lockExclusively(databasePath);
199 const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2;
200 yield* onStep("prune-events");
201 const summary = yield* maintenance.pruneEventStore();
202 yield* onStep("vacuum");
203 // VACUUM is required to enable incremental auto_vacuum on existing databases.
204 yield* sql`PRAGMA auto_vacuum = INCREMENTAL`;
205 yield* sql`VACUUM`;
206 yield* onStep("rebuild-search");
207 yield* rebuildMessageSearch(sql);
208 yield* onStep("checkpoint");
209 yield* sql`PRAGMA wal_checkpoint(TRUNCATE)`;
210 return summary;
211 }).pipe(
212 Effect.provide(
213 ProjectionMaintenance.layer.pipe(
214 Layer.provideMerge(NodeSqliteClient.layer({ filename: databasePath })),
215 ),
216 ),
217 );
218 const afterBytes = Number((yield* fs.stat(databasePath)).size);
219 return {
220 beforeBytes,
221 afterBytes,
222 deletedEventCount: result.deletedEventCount,
223 deletedReceiptCount: result.deletedReceiptCount,
224 };
225 },
226 (effect, databasePath) =>
227 effect.pipe(
228 Effect.mapError((cause) =>
229 isDomainError(cause) ? cause : new DatabaseCompactError({ path: databasePath, cause }),
230 ),
231 ),
232 );
233
234 const removeLegacyDatabase = Effect.fn("DatabaseMaintenance.removeLegacyDatabase")(
235 function* (databasePath: string) {
236 const legacy = legacyPath(databasePath);
237 if (!(yield* fs.exists(legacy))) return;
238 // Delete the database before WAL to preserve commits if unlink fails.
239 const removeFiles = Effect.forEach(
240 ["", "-wal", "-shm"],
241 (suffix) => fs.remove(legacy + suffix, { force: true }),
242 { discard: true },
243 );
244 const connection = NodeSqliteClient.layer({ filename: legacy });
245 if ((yield* HostProcess.Platform) === "win32") {
246 // Windows blocks unlink while any SQLite handle is open, including our own.
247 yield* lockExclusively(legacy).pipe(Effect.provide(connection));
248 yield* removeFiles;
249 } else {
250 yield* lockExclusively(legacy).pipe(
251 Effect.andThen(removeFiles),
252 Effect.provide(connection),
253 );
254 }
255 },
256 (effect, databasePath) =>
257 effect.pipe(
258 Effect.mapError((cause) =>
259 cause._tag === "DatabaseInUseError"
260 ? cause
261 : new LegacyDatabaseRemoveError({ path: legacyPath(databasePath), cause }),
262 ),
263 ),
264 );
265
266 return DatabaseMaintenance.of({ inspect, compact, removeLegacyDatabase });
267});
268
269export const layer = Layer.effect(DatabaseMaintenance, make);
270
271/** Settings' Reclaim writes the request; the failure stays for Settings until the next request. */
272export const reclaimFiles = (path: Path.Path, databasePath: string) => ({
273 request: path.join(path.dirname(databasePath), "reclaim-requested"),
274 failure: path.join(path.dirname(databasePath), "reclaim-failed"),
275});
276
277/** Runs a repair that Settings requested. The server calls this on exit, once its database is closed. */
278export const runRequestedReclaim = Effect.fn("DatabaseMaintenance.runRequestedReclaim")(
279 function* (databasePath: string) {
280 const fs = yield* FileSystem.FileSystem;
281 const files = reclaimFiles(yield* Path.Path, databasePath);
282 if (!(yield* fs.exists(files.request))) return;
283 // One attempt per request, so a repair killed midway (it rolls back) cannot delay later exits.
284 yield* fs.remove(files.request, { force: true });
285 yield* Effect.logInfo("Reclaiming database space before the restart. This can take minutes.");
286 const maintenance = yield* DatabaseMaintenance;
287 yield* maintenance.compact(databasePath).pipe(
288 // Worker threads and agent processes can hold the file for a moment after the server stops.
289 Effect.retry({
290 times: 10,
291 while: (error) => error._tag === "DatabaseInUseError",
292 schedule: Schedule.spaced("1 second"),
293 }),
294 Effect.matchEffect({
295 onSuccess: (result) =>
296 fs
297 .remove(files.failure, { force: true })
298 .pipe(Effect.andThen(Effect.logInfo("Reclaimed database space", result))),
299 onFailure: (error) =>
300 fs
301 .writeFileString(
302 files.failure,
303 error._tag === "DatabaseInUseError"
304 ? "Another program kept the database open. Close other T3 Code apps and t3 commands on this machine, then try again."
305 : error.message,
306 )
307 .pipe(
308 Effect.andThen(Effect.logWarning("Database space reclamation failed", { error })),
309 ),
310 }),
311 );
312 },
313 Effect.catchCause((cause) => Effect.logWarning("Database space reclamation failed", { cause })),
314);
315
316// SQLite's search order for its temporary files (unixTempFileDir in os_unix.c).
317const sqliteTempDirectory = Effect.gen(function* () {
318 if ((yield* HostProcess.Platform) === "win32") return NodeOS.tmpdir();
319 const fs = yield* FileSystem.FileSystem;
320 const env = yield* HostProcess.Environment;
321 for (const directory of [env.SQLITE_TMPDIR, env.TMPDIR, "/var/tmp", "/usr/tmp", "/tmp"]) {
322 if (directory === undefined || directory === "") continue;
323 const info = yield* fs.stat(directory).pipe(Effect.option);
324 if (Option.isNone(info) || info.value.type !== "Directory") continue;
325 // SQLite skips directories it cannot write to and search: osAccess(zDir, 03).
326 const usable = yield* Effect.tryPromise({
327 try: () => NodeFSP.access(directory, NodeFSP.constants.W_OK | NodeFSP.constants.X_OK),
328 catch: (cause) => new DiskSpaceCheckError({ path: directory, cause }),
329 }).pipe(
330 Effect.as(true),
331 Effect.orElseSucceed(() => false),
332 );
333 if (usable) return directory;
334 }
335 return ".";
336});
337
338const freeBytes = (directory: string) =>
339 Effect.tryPromise({
340 try: () => NodeFSP.statfs(directory),
341 catch: (cause) => new DiskSpaceCheckError({ path: directory, cause }),
342 }).pipe(Effect.map((stats) => stats.bavail * stats.bsize));
343
344/**
345 * VACUUM copies the live pages to a temporary file, then writes them again to the WAL.
346 * Returns the free space the repair needs when either file system has less, else null.
347 */
348export const reclaimSpaceShortfall = Effect.fn("DatabaseMaintenance.reclaimSpaceShortfall")(
349 function* (databasePath: string, liveBytes: number) {
350 const fs = yield* FileSystem.FileSystem;
351 const path = yield* Path.Path;
352 const databaseDirectory = path.dirname(databasePath);
353 const tempDirectory = yield* sqliteTempDirectory;
354 const copyBytes = Math.ceil(liveBytes * 1.1);
355 const [databaseFree, tempFree, databaseDevice, tempDevice] = yield* Effect.all([
356 freeBytes(databaseDirectory),
357 freeBytes(tempDirectory),
358 fs.stat(databaseDirectory).pipe(Effect.map((info) => info.dev)),
359 fs.stat(tempDirectory).pipe(Effect.map((info) => info.dev)),
360 ]);
361 if (databaseDevice === tempDevice) {
362 return databaseFree < 2 * copyBytes
363 ? { neededBytes: 2 * copyBytes, freeBytes: databaseFree }
364 : null;
365 }
366 if (databaseFree < copyBytes) return { neededBytes: copyBytes, freeBytes: databaseFree };
367 return tempFree < copyBytes ? { neededBytes: copyBytes, freeBytes: tempFree } : null;
368 },
369);