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