1// @effect-diagnostics nodeBuiltinImport:off
2import * as NodeFS from "node:fs";
3import * as NodeOS from "node:os";
4import * as NodePath from "node:path";
5import type * as NodeWorkerThreads from "node:worker_threads";
6
7import * as NodeCrypto from "@effect/platform-node/NodeCrypto";
8import * as NodeServices from "@effect/platform-node/NodeServices";
9import { assert, it } from "@effect/vitest";
10import * as HostProcess from "@t3tools/shared/HostProcess";
11import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient";
12import * as Effect from "effect/Effect";
13import * as FileSystem from "effect/FileSystem";
14import * as Layer from "effect/Layer";
15import * as SqlClient from "effect/sql/SqlClient";
16import * as SqlError from "effect/sql/SqlError";
17import { vi } from "vite-plus/test";
18
19import * as ServerConfig from "../config.ts";
20import * as ProcessRunner from "../processRunner.ts";
21import * as ServerOwnership from "../serverOwnership.ts";
22import { migrationManifest, runMigrations } from "./Migrations.ts";
23import * as SqlitePersistence from "./Sqlite.ts";
24
25const workerData = vi.hoisted((): Array<unknown> => []);
26
27vi.mock("node:worker_threads", async (importOriginal) => {
28 const actual = await importOriginal<typeof NodeWorkerThreads>();
29 class RecordedWorker extends actual.Worker {
30 constructor(...args: ConstructorParameters<typeof actual.Worker>) {
31 super(...args);
32 workerData.push(args[1]?.workerData);
33 }
34 }
35 return { ...actual, Worker: RecordedWorker };
36});
37
38// Linux names an open, already-deleted file "<path> (deleted)" in /proc/self/fd.
39// SQLite names its temp files "etilqs_*" and, without TMPDIR, puts them in
40// /var/tmp while os.tmpdir() is /tmp, so match the name, not the directory.
41const openDeletedTempFiles = () =>
42 NodeFS.readdirSync("/proc/self/fd").flatMap((fd) => {
43 try {
44 const target = NodeFS.readlinkSync(`/proc/self/fd/${fd}`);
45 return target.includes("/etilqs_") && target.endsWith(" (deleted)") ? [target] : [];
46 } catch {
47 return [];
48 }
49 });
50
51it.effect.skipIf(HostProcess.Platform.defaultValue() !== "linux")(
52 "keeps a write transaction's statement journal off the temp directory",
53 () => {
54 const tempDir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-sqlite-temp-"));
55 const persistence = SqlitePersistence.layerFromPath(
56 NodePath.join(tempDir, "state.sqlite"),
57 ).pipe(Layer.provide(NodeServices.layer));
58
59 return Effect.gen(function* () {
60 const sql = yield* SqlClient.SqlClient;
61 yield* sql`CREATE TABLE journal_probe(id INTEGER PRIMARY KEY, payload TEXT NOT NULL)`;
62 for (let id = 0; id < 64; id++) {
63 yield* sql`INSERT INTO journal_probe VALUES (${id}, ${"x".repeat(4096)})`;
64 }
65 const before = new Set(openDeletedTempFiles());
66 const opened = yield* sql.withTransaction(
67 Effect.gen(function* () {
68 // A multi-row update can fail halfway, so SQLite journals every page
69 // it changes, here well past the 64 KiB it holds in memory by default.
70 yield* sql`UPDATE journal_probe SET payload = payload || 'y'`;
71 return openDeletedTempFiles().filter((target) => !before.has(target));
72 }),
73 );
74 assert.deepEqual(opened, []);
75 }).pipe(
76 Effect.provide(persistence),
77 Effect.ensuring(Effect.sync(() => NodeFS.rmSync(tempDir, { recursive: true, force: true }))),
78 );
79 },
80);
81
82it.effect("drops planner statistics a database already has", () => {
83 const tempDir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-sqlite-stats-"));
84 const dbPath = NodePath.join(tempDir, "state.sqlite");
85 const persistence = () =>
86 SqlitePersistence.layerFromPath(dbPath).pipe(Layer.provide(NodeServices.layer));
87
88 return Effect.gen(function* () {
89 yield* Effect.gen(function* () {
90 const sql = yield* SqlClient.SqlClient;
91 yield* sql`CREATE TABLE walk_probe(node_id TEXT PRIMARY KEY, parent_node_id TEXT)`;
92 yield* sql`INSERT INTO walk_probe VALUES ('node-1', NULL), ('node-2', 'node-1')`;
93 yield* sql`ANALYZE walk_probe`;
94 }).pipe(Effect.provide(persistence()));
95
96 yield* Effect.gen(function* () {
97 const sql = yield* SqlClient.SqlClient;
98 const stats = yield* sql<{ readonly count: number }>`
99 SELECT COUNT(*) AS count FROM sqlite_stat1
100 `;
101 assert.equal(stats[0]?.count, 0);
102 // A connection still using the old stats scans the table at each step.
103 const plan = yield* sql<{ readonly detail: string }>`
104 EXPLAIN QUERY PLAN
105 WITH RECURSIVE retained(node_id) AS (
106 SELECT 'node-2'
107 UNION
108 SELECT parent.parent_node_id FROM walk_probe AS parent
109 INNER JOIN retained ON parent.node_id = retained.node_id
110 )
111 SELECT node_id FROM retained
112 `;
113 assert.include(
114 plan.map((row) => row.detail),
115 "SEARCH parent USING INDEX sqlite_autoindex_walk_probe_1 (node_id=?)",
116 );
117 }).pipe(Effect.provide(persistence()));
118 }).pipe(
119 Effect.ensuring(Effect.sync(() => NodeFS.rmSync(tempDir, { recursive: true, force: true }))),
120 );
121});
122
123it.effect("upgrades an existing database on a read-write worker connection", () => {
124 const tempDir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-sqlite-upgrade-"));
125 const dbPath = NodePath.join(tempDir, "state.sqlite");
126
127 return Effect.gen(function* () {
128 yield* Effect.gen(function* () {
129 const sql = yield* SqlClient.SqlClient;
130 yield* sql`PRAGMA journal_mode = WAL;`;
131 yield* runMigrations({ toMigrationInclusive: 81 });
132 }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: dbPath })));
133 const startedBefore = workerData.length;
134
135 yield* Effect.gen(function* () {
136 const sql = yield* SqlClient.SqlClient;
137 const rows = yield* sql<{ readonly migration_id: number; readonly name: string }>`
138 SELECT migration_id, name FROM effect_sql_migrations ORDER BY migration_id
139 `;
140 assert.deepStrictEqual(
141 rows.map((row) => [row.migration_id, row.name] as const),
142 migrationManifest,
143 );
144 }).pipe(
145 Effect.provide(
146 SqlitePersistence.layerFromPath(dbPath).pipe(Layer.provide(NodeServices.layer)),
147 ),
148 );
149 assert.deepInclude(workerData.slice(startedBefore), {
150 filename: dbPath,
151 readOnly: false,
152 requireWal: false,
153 allowExtension: false,
154 prepareCacheSize: 200,
155 });
156 }).pipe(
157 Effect.ensuring(Effect.sync(() => NodeFS.rmSync(tempDir, { recursive: true, force: true }))),
158 );
159});
160
161it.effect("names a database file that is not SQLite", () => {
162 const tempDir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-sqlite-notadb-"));
163 const dbPath = NodePath.join(tempDir, "state.sqlite");
164 NodeFS.writeFileSync(dbPath, "not a db\n");
165
166 return Effect.gen(function* () {
167 const opened = yield* Effect.void.pipe(
168 Effect.provide(
169 SqlitePersistence.layerFromPath(dbPath).pipe(Layer.provide(NodeServices.layer)),
170 ),
171 Effect.result,
172 );
173
174 assert.equal(opened._tag, "Failure");
175 if (opened._tag === "Failure") {
176 assert.isTrue(SqlError.isSqlError(opened.failure));
177 assert.include(opened.failure.message, dbPath);
178 }
179 }).pipe(
180 Effect.ensuring(Effect.sync(() => NodeFS.rmSync(tempDir, { recursive: true, force: true }))),
181 );
182});
183
184const ownT3Home = Effect.fn(function* () {
185 const { serverRuntimeStatePath } = yield* ServerConfig.ServerConfig;
186 // An empty home, so no boot service installed on this machine is read.
187 const home = yield* (yield* FileSystem.FileSystem).makeTempDirectoryScoped();
188 yield* ServerOwnership.acquireServerOwnership(serverRuntimeStatePath).pipe(
189 Effect.provide(ProcessRunner.layer),
190 Effect.provideService(HostProcess.HomeDirectory, home),
191 Effect.provideService(HostProcess.Environment, {}),
192 );
193});
194
195const migrateTo = (toMigrationInclusive?: number) =>
196 Effect.gen(function* () {
197 const { dbPath } = yield* ServerConfig.ServerConfig;
198 yield* runMigrations({ toMigrationInclusive }).pipe(
199 Effect.provide(NodeSqliteClient.layer({ filename: dbPath })),
200 );
201 });
202
203const latestRecordedMigrationId = Effect.gen(function* () {
204 const { dbPath } = yield* ServerConfig.ServerConfig;
205 const rows = yield* Effect.gen(function* () {
206 const sql = yield* SqlClient.SqlClient;
207 return yield* sql<{ readonly id: number }>`
208 SELECT MAX(migration_id) AS id FROM effect_sql_migrations
209 `;
210 }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: dbPath })));
211 return rows[0]?.id;
212});
213
214const layerT3Home = ServerConfig.layerTest(process.cwd(), { prefix: "t3-sqlite-cli-" });
215
216it.layer(Layer.mergeAll(NodeServices.layer, NodeCrypto.layer))("layerCliConfig", (it) => {
217 it.effect("does not migrate a T3 home a running server owns", () =>
218 Effect.gen(function* () {
219 const recorded = Math.max(...migrationManifest.map(([id]) => id)) - 1;
220 yield* migrateTo(recorded);
221 yield* ownT3Home();
222
223 const opened = yield* Effect.void.pipe(
224 Effect.provide(SqlitePersistence.layerCliConfig),
225 Effect.result,
226 );
227
228 assert.equal(yield* latestRecordedMigrationId, recorded);
229 assert.equal(opened._tag, "Failure");
230 if (opened._tag === "Failure") {
231 assert.instanceOf(opened.failure, SqlitePersistence.ServerOwnsDatabaseError);
232 }
233 }).pipe(Effect.scoped, Effect.provide(layerT3Home)),
234 );
235
236 it.effect("opens a current T3 home while a running server owns it", () =>
237 Effect.gen(function* () {
238 yield* migrateTo();
239 yield* ownT3Home();
240
241 yield* Effect.void.pipe(Effect.provide(SqlitePersistence.layerCliConfig));
242 }).pipe(Effect.scoped, Effect.provide(layerT3Home)),
243 );
244});