1import { DatabaseReclaimError, type DatabaseStorageStatus } from "@t3tools/contracts";
2import * as Clock from "effect/Clock";
3import * as Context from "effect/Context";
4import * as Duration from "effect/Duration";
5import * as Effect from "effect/Effect";
6import * as Equal from "effect/Equal";
7import * as FileSystem from "effect/FileSystem";
8import * as Layer from "effect/Layer";
9import * as Option from "effect/Option";
10import * as Path from "effect/Path";
11import * as Schedule from "effect/Schedule";
12import * as Stream from "effect/Stream";
13import * as SubscriptionRef from "effect/SubscriptionRef";
14import * as SqlClient from "effect/sql/SqlClient";
15
16import { resolveHostCliCommand } from "../cli/invocation.ts";
17import * as ServerConfig from "../config.ts";
18import * as ProjectionMaintenance from "../orchestration-v2/ProjectionMaintenance.ts";
19import * as DatabaseMaintenance from "./DatabaseMaintenance.ts";
20import { forkParked } from "../serverActivation.ts";
21
22// Bound each reclaim pass to one minute of churn.
23const RECLAIM_PAGES_PER_PASS = 1024;
24const COMPACTION_INTERVAL = Duration.hours(6);
25
26export class DatabaseStorage extends Context.Service<
27 DatabaseStorage,
28 {
29 /** Return free pages to the filesystem on databases with incremental auto_vacuum. */
30 readonly reclaim: Effect.Effect<void>;
31 /** Current file accounting, re-published after each maintenance pass. Null until first read. */
32 readonly status: Stream.Stream<DatabaseStorageStatus | null>;
33 /** Checks free space, then stops the server; its shutdown runs the offline repair. */
34 readonly requestReclaim: Effect.Effect<void, DatabaseReclaimError>;
35 }
36>()("t3/persistence/DatabaseStorage") {}
37
38/** Stops this server so its desktop app or background service starts it again; null when run by hand. */
39export const ServerRestart = Context.Reference<Effect.Effect<void> | null>(
40 "t3/persistence/DatabaseStorage/ServerRestart",
41 { defaultValue: () => null },
42);
43
44const formatGiB = (bytes: number) => `${(bytes / 1024 ** 3).toFixed(1)} GiB`;
45
46const make = Effect.gen(function* () {
47 const sql = yield* SqlClient.SqlClient;
48 const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2;
49 const config = yield* ServerConfig.ServerConfig;
50 const fs = yield* FileSystem.FileSystem;
51 const path = yield* Path.Path;
52 const statusRef = yield* SubscriptionRef.make<DatabaseStorageStatus | null>(null);
53 const restart = yield* ServerRestart;
54 const reclaimFiles = DatabaseMaintenance.reclaimFiles(path, config.dbPath);
55 let reclaimFailure = Option.getOrUndefined(
56 yield* fs.readFileString(reclaimFiles.failure).pipe(Effect.option),
57 );
58
59 // The notice shows host-shell quoting verbatim; desktop needs its launcher because t3 is absent from PATH.
60 const { command: compactCommand } = yield* resolveHostCliCommand("doctor", [
61 "--base-dir",
62 config.baseDir,
63 ]);
64 // doctor --base-dir targets userdata, not implicit dev homes; suppress status to avoid touching the live install.
65 const doctorTargetsThisDatabase =
66 path.resolve(config.dbPath) ===
67 path.resolve(config.baseDir, "userdata", path.basename(config.dbPath));
68 // Only VACUUM changes these, and doctor refuses to VACUUM while a server has the database open.
69 let layout:
70 | {
71 readonly pageSize: number;
72 readonly autoVacuum: number;
73 readonly incrementalVacuum: boolean;
74 }
75 | undefined;
76 const readLayout = Effect.suspend(() =>
77 layout !== undefined
78 ? Effect.succeed(layout)
79 : Effect.gen(function* () {
80 const size = yield* sql<{ readonly page_size: number }>`PRAGMA page_size`;
81 const mode = yield* sql<{ readonly auto_vacuum: number }>`PRAGMA auto_vacuum`;
82 const autoVacuum = mode[0]?.auto_vacuum ?? 0;
83 layout = {
84 pageSize: size[0]?.page_size ?? 0,
85 autoVacuum,
86 incrementalVacuum: autoVacuum === 2,
87 };
88 return layout;
89 }),
90 );
91 const readPages = Effect.gen(function* () {
92 const { pageSize, incrementalVacuum } = yield* readLayout;
93 const pages = yield* sql<{ readonly page_count: number }>`PRAGMA page_count`;
94 const free = yield* sql<{ readonly freelist_count: number }>`PRAGMA freelist_count`;
95 return {
96 databaseBytes: (pages[0]?.page_count ?? 0) * pageSize,
97 reclaimableBytes: (free[0]?.freelist_count ?? 0) * pageSize,
98 incrementalVacuum,
99 };
100 });
101 const publishStatus = Effect.gen(function* () {
102 if (!doctorTargetsThisDatabase) return;
103 const { databaseBytes, reclaimableBytes, incrementalVacuum } = yield* readPages;
104 const next: DatabaseStorageStatus = {
105 databaseBytes,
106 reclaimableBytes,
107 incrementalVacuumEnabled: incrementalVacuum,
108 compactCommand,
109 canReclaim: restart !== null,
110 ...(reclaimFailure === undefined ? {} : { reclaimFailure }),
111 };
112 // Every set reaches each open Storage page; most passes change nothing.
113 if (!Equal.equals(yield* SubscriptionRef.get(statusRef), next)) {
114 yield* SubscriptionRef.set(statusRef, next);
115 }
116 }).pipe(
117 Effect.catch((cause) => Effect.logWarning("Database storage inspection failed", { cause })),
118 );
119
120 const reclaim = Effect.suspend(() => {
121 let page = 0;
122 return Effect.gen(function* () {
123 if (!(yield* readLayout).incrementalVacuum) return;
124 // The driver steps once per statement; incremental_vacuum frees one page per step. Yield between pages.
125 for (; page < RECLAIM_PAGES_PER_PASS; page++) {
126 const free = yield* sql<{ readonly freelist_count: number }>`PRAGMA freelist_count`;
127 if ((free[0]?.freelist_count ?? 0) === 0) break;
128 yield* sql`PRAGMA incremental_vacuum(1)`;
129 yield* Effect.yieldNow;
130 }
131 }).pipe(
132 Effect.ensuring(Effect.suspend(() => Effect.annotateCurrentSpan("db.reclaimed_pages", page))),
133 Effect.catch((cause) => Effect.logWarning("Database space reclamation failed", { cause })),
134 // The pass owns the span; tracing each page's two PRAGMAs floods the server trace.
135 Effect.withTracerEnabled(false),
136 Effect.withSpan("DatabaseStorage.reclaim"),
137 );
138 });
139
140 // Even a pass that deletes nothing scans the event store. Persist completion
141 // so restarting does not trigger another scan before the six-hour interval.
142 const compactedAtPath = path.join(path.dirname(config.dbPath), "compacted-at");
143 const compact = Effect.gen(function* () {
144 let sliceStarted = yield* Clock.monotonicTimeNanos;
145 const betweenPages = Effect.gen(function* () {
146 const elapsed = (yield* Clock.monotonicTimeNanos) - sliceStarted;
147 if (elapsed < 25_000_000n) return;
148 // Keep background maintenance at 10% duty, including pages that exceed the budget.
149 yield* Effect.sleep((Number(elapsed) / 1_000_000) * 9);
150 sliceStarted = yield* Clock.monotonicTimeNanos;
151 });
152 const result = yield* maintenance.pruneEventStore(betweenPages);
153 const completedAt = yield* Clock.currentTimeMillis;
154 yield* fs.writeFileString(compactedAtPath, String(completedAt)).pipe(Effect.ignore);
155 if (result.deletedEventCount > 0 || result.deletedReceiptCount > 0) {
156 yield* Effect.logInfo("Compacted orchestration history", result);
157 }
158 }).pipe(
159 Effect.catch((cause) =>
160 Effect.logWarning("Orchestration history compaction failed", { cause }),
161 ),
162 );
163 const lastCompactedAt = yield* fs.readFileString(compactedAtPath).pipe(
164 Effect.map(Number),
165 Effect.orElseSucceed(() => Number.NaN),
166 );
167 const intervalMs = Duration.toMillis(COMPACTION_INTERVAL);
168 const untilDueMs = Number.isFinite(lastCompactedAt)
169 ? lastCompactedAt + intervalMs - (yield* Clock.currentTimeMillis)
170 : 0;
171 const firstCompactionDelay = Duration.millis(
172 Math.min(intervalMs, Math.max(Duration.toMillis(Duration.minutes(1)), untilDueMs)),
173 );
174
175 yield* forkParked(
176 Effect.sleep(firstCompactionDelay).pipe(
177 Effect.andThen(
178 compact.pipe(
179 Effect.andThen(publishStatus),
180 Effect.repeat(Schedule.spaced(COMPACTION_INTERVAL)),
181 ),
182 ),
183 ),
184 );
185 yield* forkParked(
186 publishStatus.pipe(
187 Effect.andThen(Effect.sleep("1 minute")),
188 Effect.andThen(
189 reclaim.pipe(Effect.andThen(publishStatus), Effect.repeat(Schedule.spaced("1 minute"))),
190 ),
191 ),
192 );
193
194 const requestReclaim = Effect.gen(function* () {
195 if (restart === null || !doctorTargetsThisDatabase) {
196 return yield* new DatabaseReclaimError({
197 reason: `Stop T3 Code on this machine, then run: ${compactCommand}`,
198 });
199 }
200 // Only databases without automatic shrinking need it; a stale dialog can ask after another client's repair.
201 if ((yield* readLayout).autoVacuum !== 0) {
202 return yield* new DatabaseReclaimError({
203 reason: "This database already shrinks on its own. Nothing to reclaim.",
204 });
205 }
206 const { databaseBytes, reclaimableBytes } = yield* readPages;
207 const shortfall = yield* DatabaseMaintenance.reclaimSpaceShortfall(
208 config.dbPath,
209 databaseBytes - reclaimableBytes,
210 ).pipe(
211 Effect.provideService(FileSystem.FileSystem, fs),
212 Effect.provideService(Path.Path, path),
213 );
214 if (shortfall !== null) {
215 return yield* new DatabaseReclaimError({
216 reason: `Reclaiming needs ${formatGiB(shortfall.neededBytes)} of free disk space on this machine. ${formatGiB(shortfall.freeBytes)} is free.`,
217 });
218 }
219 yield* fs.remove(reclaimFiles.failure, { force: true });
220 reclaimFailure = undefined;
221 yield* publishStatus;
222 const requestedAt = yield* Clock.currentTimeMillis;
223 // One step: a request interrupted between them would repair on some unrelated later exit.
224 yield* Effect.uninterruptible(
225 fs.writeFileString(reclaimFiles.request, String(requestedAt)).pipe(
226 // Reply before stopping: the restart closes this connection.
227 Effect.andThen(restart.pipe(Effect.delay("500 millis"), Effect.forkDetach)),
228 ),
229 );
230 yield* Effect.logInfo("Restarting to reclaim database space");
231 }).pipe(
232 Effect.mapError((cause) =>
233 cause._tag === "DatabaseReclaimError"
234 ? cause
235 : new DatabaseReclaimError({ reason: "Could not prepare the repair. Try again." }),
236 ),
237 Effect.withSpan("DatabaseStorage.requestReclaim"),
238 );
239
240 return DatabaseStorage.of({
241 reclaim,
242 status: SubscriptionRef.changes(statusRef),
243 requestReclaim,
244 });
245});
246
247export const layer = Layer.effect(DatabaseStorage, make);