1// @effect-diagnostics nodeBuiltinImport:off
2import * as NodeChildProcess from "node:child_process";
3import * as NodeFS from "node:fs";
4import * as NodeOS from "node:os";
5import * as NodePath from "node:path";
6import * as NodeSqlite from "node:sqlite";
7
8import * as NodeServices from "@effect/platform-node/NodeServices";
9import { assert, it } from "@effect/vitest";
10import { ThreadId } from "@t3tools/contracts";
11import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient";
12import * as Effect from "effect/Effect";
13import * as Layer from "effect/Layer";
14import * as Predicate from "effect/Predicate";
15import * as SqlClient from "effect/sql/SqlClient";
16
17import * as ServerConfig from "../config.ts";
18import * as SqlitePersistence from "./Sqlite.ts";
19import { runMigrations } from "./Migrations.ts";
20import { initializeV2Database } from "./initializeV2Database.ts";
21import * as EventStore from "../orchestration-v2/EventStore.ts";
22import * as ProjectionStore from "../orchestration-v2/ProjectionStore.ts";
23import * as EventSink from "../orchestration-v2/EventSink.ts";
24import * as LegacyV1ThreadImporter from "../orchestration-v2/legacy/LegacyV1ThreadImporter.ts";
25
26it.effect(
27 "snapshots V1, imports transcripts lazily, and preserves both databases across switches",
28 () => {
29 const directory = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-v1-v2-"));
30 const sourcePath = NodePath.join(directory, "state.sqlite");
31 const destinationPath = NodePath.join(directory, "statev2.sqlite");
32 const threadId = ThreadId.make("legacy-thread");
33 const seed = Effect.gen(function* () {
34 const sql = yield* SqlClient.SqlClient;
35 yield* runMigrations({ toMigrationInclusive: 52 });
36 yield* sql`INSERT INTO projection_projects (project_id, title, workspace_root, scripts_json, created_at, updated_at)
37 VALUES ('project', 'Project', '/tmp/project', '[]', '2026-01-01T00:00:00.000Z', '2026-01-01T00:00:00.000Z')`;
38 yield* sql`INSERT INTO projection_threads (thread_id, project_id, title, model_selection_json, runtime_mode, interaction_mode, created_at, updated_at)
39 VALUES (${threadId}, 'project', 'V1 thread', '{"instanceId":"codex","model":"gpt-5.4"}', 'full-access', 'default', '2026-01-01T00:00:00.000Z', '2026-01-01T00:00:00.000Z')`;
40 for (let index = 0; index < 6; index++) {
41 yield* sql`INSERT INTO projection_thread_messages (message_id, thread_id, role, text, is_streaming, created_at, updated_at)
42 VALUES (${`message-${index}`}, ${threadId}, ${index % 2 ? "assistant" : "user"}, ${`Text ${index}`}, 0, ${`2026-01-0${index + 1}T00:00:00.000Z`}, ${`2026-01-0${index + 1}T00:00:00.000Z`})`;
43 }
44 }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: sourcePath })));
45
46 return Effect.gen(function* () {
47 yield* seed;
48 const original = NodeFS.readFileSync(sourcePath);
49 const config = yield* ServerConfig.ServerConfig;
50 const layerDatabase = SqlitePersistence.layerConfig.pipe(
51 Layer.provide(ServerConfig.layer({ ...config, dbPath: destinationPath })),
52 );
53 const layerStores = Layer.mergeAll(EventStore.layer, ProjectionStore.layer).pipe(
54 Layer.provideMerge(layerDatabase),
55 );
56 const layerSink = EventSink.layer.pipe(Layer.provide(layerStores));
57 const layerImporter = LegacyV1ThreadImporter.layer.pipe(
58 Layer.provideMerge(Layer.mergeAll(layerStores, layerSink)),
59 );
60 yield* Effect.gen(function* () {
61 const sql = yield* SqlClient.SqlClient;
62 const legacy = yield* LegacyV1ThreadImporter.LegacyV1ThreadImporter;
63 yield* legacy.reconcileShells;
64 const projections = yield* ProjectionStore.ProjectionStoreV2;
65 const shell = yield* projections.getThreadProjection(threadId);
66 assert.equal(shell.thread.id, threadId);
67 assert.deepEqual(
68 shell.messages.map((message) => message.text),
69 ["Text 4", "Text 5"],
70 );
71 const pending =
72 yield* sql`SELECT transcript_imported_at FROM orchestration_v2_legacy_imports`;
73 assert.equal(pending[0]?.transcript_imported_at, null);
74 yield* legacy.ensureTranscript(threadId);
75 const transcript = yield* projections.getThreadProjection(threadId);
76 assert.deepEqual(
77 transcript.messages.map((message) => message.text),
78 ["Text 0", "Text 1", "Text 2", "Text 3", "Text 4", "Text 5"],
79 );
80 const imported =
81 yield* sql`SELECT imported_message_count, transcript_imported_at FROM orchestration_v2_legacy_imports`;
82 assert.equal(imported[0]?.imported_message_count, 6);
83 assert.isNotNull(imported[0]?.transcript_imported_at);
84 yield* sql`CREATE TABLE v2_work (text TEXT)`;
85 yield* sql`INSERT INTO v2_work VALUES ('Keep V2 work')`;
86 }).pipe(Effect.provide(layerImporter));
87 assert.deepEqual(NodeFS.readFileSync(sourcePath), original);
88 const v1 = new NodeSqlite.DatabaseSync(sourcePath);
89 try {
90 assert.equal(
91 v1.prepare("SELECT MAX(migration_id) AS id FROM effect_sql_migrations").get()?.id,
92 52,
93 );
94 assert.equal(
95 v1
96 .prepare(
97 "SELECT count(*) AS count FROM sqlite_master WHERE name = 'orchestration_v2_legacy_imports'",
98 )
99 .get()?.count,
100 0,
101 );
102 v1.exec("UPDATE projection_threads SET title = 'Continued in V1'");
103 } finally {
104 v1.close();
105 }
106 yield* Effect.gen(function* () {
107 const sql = yield* SqlClient.SqlClient;
108 assert.equal((yield* sql`SELECT text FROM v2_work`)[0]?.text, "Keep V2 work");
109 assert.equal((yield* sql`SELECT title FROM projection_threads`)[0]?.title, "V1 thread");
110 }).pipe(Effect.provide(layerDatabase));
111 }).pipe(
112 Effect.provide(
113 ServerConfig.layerTest(directory, directory).pipe(Layer.provideMerge(NodeServices.layer)),
114 ),
115 Effect.ensuring(
116 Effect.sync(() => NodeFS.rmSync(directory, { recursive: true, force: true })),
117 ),
118 );
119 },
120);
121
122it.effect("includes committed WAL data and does not publish a failed snapshot", () => {
123 const directory = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-v2-snapshot-"));
124 const sourcePath = NodePath.join(directory, "state.sqlite");
125 const destinationPath = NodePath.join(directory, "statev2.sqlite");
126 return Effect.gen(function* () {
127 NodeFS.writeFileSync(sourcePath, "invalid SQLite");
128 const failure = yield* Effect.flip(initializeV2Database(destinationPath));
129 assert.isFalse(Predicate.isTagged(failure.cause, "V2DatabaseImportError"));
130 assert.isFalse(NodeFS.existsSync(destinationPath));
131 NodeFS.unlinkSync(sourcePath);
132 const source = new NodeSqlite.DatabaseSync(sourcePath);
133 try {
134 source.exec(
135 "PRAGMA journal_mode=WAL; CREATE TABLE messages(text TEXT); INSERT INTO messages VALUES ('committed'); BEGIN; INSERT INTO messages VALUES ('uncommitted');",
136 );
137 yield* initializeV2Database(destinationPath);
138 const copy = new NodeSqlite.DatabaseSync(destinationPath, { readOnly: true });
139 try {
140 assert.deepEqual(
141 copy
142 .prepare("SELECT text FROM messages")
143 .all()
144 .map((row) => row.text),
145 ["committed"],
146 );
147 } finally {
148 copy.close();
149 }
150 source.exec("ROLLBACK");
151 } finally {
152 source.close();
153 }
154 }).pipe(
155 Effect.provide(NodeServices.layer),
156 Effect.ensuring(Effect.sync(() => NodeFS.rmSync(directory, { recursive: true, force: true }))),
157 );
158});
159
160it.effect("starts fresh without V1 and never imports over existing V2 state", () => {
161 const directory = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-v2-fresh-"));
162 const destinationPath = NodePath.join(directory, "userdata", "statev2.sqlite");
163 return Effect.gen(function* () {
164 const config = yield* ServerConfig.ServerConfig;
165 const layerDatabase = SqlitePersistence.layerConfig.pipe(
166 Layer.provide(ServerConfig.layer({ ...config, dbPath: destinationPath })),
167 );
168 yield* Effect.gen(function* () {
169 const sql = yield* SqlClient.SqlClient;
170 yield* sql`CREATE TABLE v2_work (text TEXT)`;
171 yield* sql`INSERT INTO v2_work VALUES ('fresh V2 work')`;
172 }).pipe(Effect.provide(layerDatabase));
173 const sourcePath = NodePath.join(NodePath.dirname(destinationPath), "state.sqlite");
174 assert.isFalse(NodeFS.existsSync(sourcePath));
175 NodeFS.writeFileSync(sourcePath, "This source must never be opened once V2 exists");
176 yield* initializeV2Database(destinationPath);
177 yield* Effect.gen(function* () {
178 const sql = yield* SqlClient.SqlClient;
179 assert.equal((yield* sql`SELECT text FROM v2_work`)[0]?.text, "fresh V2 work");
180 }).pipe(Effect.provide(layerDatabase));
181 }).pipe(
182 Effect.provide(
183 ServerConfig.layerTest(directory, directory).pipe(Layer.provideMerge(NodeServices.layer)),
184 ),
185 Effect.ensuring(Effect.sync(() => NodeFS.rmSync(directory, { recursive: true, force: true }))),
186 );
187});
188
189it.effect("removes copies left by a killed start, but not one that is still running", () => {
190 const directory = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-v2-abandoned-"));
191 const killed = NodeChildProcess.spawnSync(process.execPath, ["-e", ""]).pid;
192 const running = `.v2-import-${process.pid}-Kq81Zx`;
193 const unrelated = ".v2-import-not-a-copy";
194 return Effect.gen(function* () {
195 for (const name of [`.v2-import-${killed}-Wm3nPa`, ".v2-import-Q3xZ9a", running]) {
196 NodeFS.mkdirSync(NodePath.join(directory, name));
197 NodeFS.writeFileSync(NodePath.join(directory, name, "snapshot.sqlite"), "partial copy");
198 }
199 NodeFS.mkdirSync(NodePath.join(directory, unrelated));
200 const destinationPath = NodePath.join(directory, "statev2.sqlite");
201 NodeFS.writeFileSync(destinationPath, "existing V2 database");
202 yield* initializeV2Database(destinationPath);
203 assert.deepEqual(
204 NodeFS.readdirSync(directory).sort(),
205 [running, unrelated, "statev2.sqlite"].sort(),
206 );
207 assert.equal(NodeFS.readFileSync(destinationPath, "utf8"), "existing V2 database");
208 }).pipe(
209 Effect.provide(NodeServices.layer),
210 Effect.ensuring(Effect.sync(() => NodeFS.rmSync(directory, { recursive: true, force: true }))),
211 );
212});