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