1#!/usr/bin/env node
2
3import * as NodeRuntime from "@effect/platform-node/NodeRuntime";
4import * as NodeServices from "@effect/platform-node/NodeServices";
5import * as NodeOS from "node:os";
6import * as NodeURL from "node:url";
7import { resolveWorktreeT3Home } from "@t3tools/shared/devHome";
8import * as Console from "effect/Console";
9import * as Effect from "effect/Effect";
10import * as FileSystem from "effect/FileSystem";
11import * as Layer from "effect/Layer";
12import * as Option from "effect/Option";
13import * as Path from "effect/Path";
14import * as Schema from "effect/Schema";
15import * as SqlClient from "effect/sql/SqlClient";
16import { Command, Flag } from "effect/cli";
17
18import * as ProjectionStore from "../src/orchestration-v2/ProjectionStore.ts";
19import { rebuildMessageSearch } from "../src/orchestration-v2/messageSearchIndex.ts";
20import { runMigrations } from "../src/persistence/Migrations.ts";
21import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient";
22
23export class MigrateDevDbNotInWorktreeError extends Schema.TaggedError<MigrateDevDbNotInWorktreeError>()(
24 "MigrateDevDbNotInWorktreeError",
25 {},
26) {
27 override get message(): string {
28 return "Not inside a linked git worktree. Pass --base-dir to target an isolated .t3 directory.";
29 }
30}
31
32export class MigrateDevDbSharedHomeError extends Schema.TaggedError<MigrateDevDbSharedHomeError>()(
33 "MigrateDevDbSharedHomeError",
34 {},
35) {
36 override get message(): string {
37 return "Refusing to rebuild the shared ~/.t3 database. Use an isolated --base-dir.";
38 }
39}
40
41export class MigrateDevDbSourceMissingError extends Schema.TaggedError<MigrateDevDbSourceMissingError>()(
42 "MigrateDevDbSourceMissingError",
43 {
44 sourcePath: Schema.String,
45 },
46) {
47 override get message(): string {
48 return `Source database does not exist at '${this.sourcePath}'.`;
49 }
50}
51
52export class MigrateDevDbSourceIsDestinationError extends Schema.TaggedError<MigrateDevDbSourceIsDestinationError>()(
53 "MigrateDevDbSourceIsDestinationError",
54 {
55 sourcePath: Schema.String,
56 },
57) {
58 override get message(): string {
59 return `Source database '${this.sourcePath}' resolves to a path this command rewrites. Pick a different --source or --base-dir.`;
60 }
61}
62
63export class MigrateDevDbLegacySourceError extends Schema.TaggedError<MigrateDevDbLegacySourceError>()(
64 "MigrateDevDbLegacySourceError",
65 {
66 sourcePath: Schema.String,
67 legacyThreads: Schema.Number,
68 },
69) {
70 override get message(): string {
71 return `Source database '${this.sourcePath}' holds ${this.legacyThreads} legacy threads the server has not imported into V2 yet, which this command cannot carry over. Start a server once on a T3 home holding a copy of it so it imports them, then pass the userdata/statev2.sqlite that server leaves behind as --source.`;
72 }
73}
74
75export class MigrateDevDbServerRunningError extends Schema.TaggedError<MigrateDevDbServerRunningError>()(
76 "MigrateDevDbServerRunningError",
77 {
78 databasePath: Schema.String,
79 pid: Schema.Number,
80 },
81) {
82 override get message(): string {
83 return `Dev database at '${this.databasePath}' is open by a running server (pid ${this.pid} per server-runtime.json). Stop that server first; if that pid is not actually a T3 server (stale descriptor, reused pid), delete the server-runtime.json next to the database and retry.`;
84 }
85}
86
87export class MigrateDevDbDestinationBusyError extends Schema.TaggedError<MigrateDevDbDestinationBusyError>()(
88 "MigrateDevDbDestinationBusyError",
89 {
90 databasePath: Schema.String,
91 reason: Schema.Literals(["write-locked", "wal-held"]),
92 cause: Schema.optional(Schema.Defect()),
93 },
94) {
95 override get message(): string {
96 const detail =
97 this.reason === "write-locked"
98 ? "the database is write-locked"
99 : "another connection is holding its WAL";
100 return `Dev database at '${this.databasePath}' looks in use (${detail}). Stop the dev server first; if none is running, delete the -wal/-shm files next to it.`;
101 }
102}
103
104export class MigrateDevDbPhaseError extends Schema.TaggedError<MigrateDevDbPhaseError>()(
105 "MigrateDevDbPhaseError",
106 {
107 phase: Schema.Literals(["snapshot", "prune", "compact", "migrate"]),
108 databasePath: Schema.String,
109 cause: Schema.Defect(),
110 },
111) {
112 override get message(): string {
113 return `migrate-dev-db failed during ${this.phase} on '${this.databasePath}'.`;
114 }
115}
116
117export interface RunMigrateDevDbInput {
118 readonly baseDir?: string | undefined;
119 readonly source?: string | undefined;
120 readonly projects: number;
121 readonly threadsPerProject: number;
122}
123
124export interface RunMigrateDevDbOptions {
125 /** Test override for the directory writes must never target. */
126 readonly sharedHome?: string | undefined;
127}
128
129interface KeptProject {
130 readonly title: string;
131 readonly threads: number;
132}
133
134const removeDatabaseFiles = Effect.fn("removeDatabaseFiles")(function* (databasePath: string) {
135 const fs = yield* FileSystem.FileSystem;
136 for (const suffix of ["", "-wal", "-shm"]) {
137 yield* fs.remove(`${databasePath}${suffix}`).pipe(Effect.orElseSucceed(() => undefined));
138 }
139});
140
141const ServerRuntimeState = Schema.fromJsonString(Schema.Struct({ pid: Schema.Number }));
142const decodeServerRuntimeState = Schema.decodeEffect(ServerRuntimeState);
143
144const isProcessAlive = (pid: number): boolean => {
145 try {
146 process.kill(pid, 0);
147 return true;
148 } catch (error) {
149 // EPERM means the process exists but belongs to someone else.
150 return (error as NodeJS.ErrnoException).code === "EPERM";
151 }
152};
153
154// The runtime PID catches idle servers; leftover -shm files can come from read-only connections.
155const ensureNotInUse = Effect.fn("ensureDevDbNotInUse")(function* (databasePath: string) {
156 const fs = yield* FileSystem.FileSystem;
157 const path = yield* Path.Path;
158
159 const runtimeStatePath = path.join(path.dirname(databasePath), "server-runtime.json");
160 const runtimeState = yield* fs
161 .readFileString(runtimeStatePath)
162 .pipe(Effect.flatMap(decodeServerRuntimeState), Effect.option);
163 if (Option.isSome(runtimeState) && isProcessAlive(runtimeState.value.pid)) {
164 return yield* new MigrateDevDbServerRunningError({
165 databasePath,
166 pid: runtimeState.value.pid,
167 });
168 }
169
170 if (!(yield* fs.exists(databasePath))) {
171 return;
172 }
173 const checkpoint = yield* Effect.gen(function* () {
174 const sql = yield* SqlClient.SqlClient;
175 yield* sql.unsafe("PRAGMA busy_timeout = 0").unprepared;
176 yield* sql.unsafe("BEGIN IMMEDIATE").unprepared;
177 yield* sql.unsafe("ROLLBACK").unprepared;
178 return yield* sql.unsafe<{ busy: number }>("PRAGMA wal_checkpoint(TRUNCATE)").unprepared;
179 }).pipe(
180 Effect.provide(NodeSqliteClient.layer({ filename: databasePath })),
181 Effect.mapError(
182 (cause) =>
183 new MigrateDevDbDestinationBusyError({
184 databasePath,
185 reason: "write-locked",
186 cause,
187 }),
188 ),
189 );
190 if (checkpoint[0] !== undefined && Number(checkpoint[0].busy) !== 0) {
191 return yield* new MigrateDevDbDestinationBusyError({
192 databasePath,
193 reason: "wal-held",
194 });
195 }
196});
197
198const RECOVERY_KINDS: ReadonlyArray<ProjectionStore.ProjectionRecoveryKind> = [
199 "queued-runs",
200 "runtime",
201 "subagent-results",
202 "delegated-completions",
203];
204
205const CLEARED_TABLES: ReadonlyArray<string> = [
206 "scheduled_tasks",
207 "orchestration_v2_effect_outbox",
208 "orchestration_v2_thread_launch_workflows",
209 "orchestration_command_receipts",
210 "provider_session_runtime",
211 "auth_sessions",
212 "auth_pairing_links",
213 "peer_environment_links",
214];
215
216const PROVIDER_SESSIONS_TABLE = "orchestration_v2_projection_provider_sessions";
217const FTS5_SHADOW_SUFFIXES = ["_data", "_idx", "_content", "_docsize", "_config"];
218
219// Settled threads stay: they are most of real history, and the UI shows them.
220const rootIsVisible = (sql: SqlClient.SqlClient) => sql`
221 root.deleted_at IS NULL
222 AND json_extract(root.payload_json, '$.deletedAt') IS NULL
223 AND json_extract(root.payload_json, '$.archivedAt') IS NULL`;
224
225// One read transaction keeps the copy, and its legacy-thread guard, consistent with concurrent
226// server writes. The server imports V1 threads at boot, and the copy keeps only threads with V2
227// rows, so unimported ones would vanish silently (see LegacyV1ThreadImporter.pendingThreadCount).
228const copySourceSlice = Effect.fn("copyDevDbSourceSlice")(function* (sourcePath: string) {
229 const sql = yield* SqlClient.SqlClient;
230 yield* sql`ATTACH DATABASE ${`${NodeURL.pathToFileURL(sourcePath).href}?mode=ro`} AS src`;
231
232 yield* sql.withTransaction(
233 Effect.gen(function* () {
234 const schema = yield* sql<{ type: string; name: string; sql: string }>`
235 SELECT type, name, sql FROM src.sqlite_master
236 WHERE sql IS NOT NULL AND name NOT LIKE 'sqlite_%'`;
237 const threadColumns = yield* sql<{ name: string; notnull: number }>`
238 SELECT m.name, c."notnull" FROM src.sqlite_master m, pragma_table_info(m.name, 'src') c
239 WHERE m.type = 'table' AND c.name = 'thread_id'`;
240 const threadIdNullable = new Map(threadColumns.map((row) => [row.name, row.notnull === 0]));
241 // FTS5 makes its own shadow tables and contentless rows cannot be copied:
242 // create the virtual table, skip its shadow tables and rows, and rebuild the
243 // index once the messages and triggers are in.
244 const virtualTables = schema
245 .filter((entry) => entry.type === "table" && /^CREATE VIRTUAL TABLE/i.test(entry.sql))
246 .map((entry) => entry.name);
247 const isShadowTable = (name: string) =>
248 virtualTables.some((virtual) =>
249 FTS5_SHADOW_SUFFIXES.some((suffix) => name === `${virtual}${suffix}`),
250 );
251 const tables = schema.filter((entry) => entry.type === "table" && !isShadowTable(entry.name));
252 const filtered = tables.some((table) => table.name === "orchestration_v2_projection_threads");
253 // A row that points into a cleared table cannot be copied, so whatever references one is
254 // cleared with it, however far down the chain a newer migration added it.
255 const references = yield* sql<{ name: string; parent: string }>`
256 SELECT m.name, f."table" AS parent FROM src.sqlite_master m, pragma_foreign_key_list(m.name, 'src') f
257 WHERE m.type = 'table'`;
258 const cleared = new Set(CLEARED_TABLES);
259 for (let grew = true; grew;) {
260 grew = false;
261 for (const { name, parent } of references) {
262 if (cleared.has(parent) && !cleared.has(name)) {
263 cleared.add(name);
264 grew = true;
265 }
266 }
267 }
268
269 if (tables.some((table) => table.name === "projection_threads")) {
270 // The V2 tables, application_event_version, and the partial index arrive in
271 // the same migration. Without the hint SQLite scans each thread's stream.
272 const [pending] = filtered
273 ? yield* sql<{ count: number }>`
274 SELECT COUNT(*) AS count FROM src.projection_threads AS thread
275 WHERE NOT EXISTS (
276 SELECT 1
277 FROM src.orchestration_events AS event INDEXED BY orchestration_events_v2_created_threads_idx
278 WHERE event.application_event_version = 2
279 AND event.aggregate_kind = 'thread'
280 AND event.stream_id = thread.thread_id
281 AND event.event_type = 'thread.created'
282 )`
283 : yield* sql<{ count: number }>`SELECT COUNT(*) AS count FROM src.projection_threads`;
284 const legacyThreads = Number(pending?.count ?? 0);
285 if (legacyThreads > 0) {
286 return yield* new MigrateDevDbLegacySourceError({ sourcePath, legacyThreads });
287 }
288 }
289
290 for (const table of tables) yield* sql.unsafe(table.sql).unprepared;
291 if (filtered) {
292 yield* sql`CREATE TEMP TABLE copied_threads (thread_id TEXT PRIMARY KEY)`;
293 yield* sql`INSERT OR IGNORE INTO copied_threads
294 SELECT family.thread_id FROM src.orchestration_v2_projection_threads family
295 JOIN src.orchestration_v2_projection_threads root ON root.thread_id =
296 COALESCE(json_extract(family.payload_json, '$.lineage.rootThreadId'), family.thread_id)
297 WHERE ${rootIsVisible(sql)}`;
298 }
299 for (const { name } of tables) {
300 if (cleared.has(name) || virtualTables.includes(name)) continue;
301 const nullable = threadIdNullable.get(name);
302 const filter = !filtered
303 ? ""
304 : name === "orchestration_events"
305 ? // Two ranges let SQLite seek the (aggregate_kind, stream_id) index instead of scanning.
306 `WHERE (aggregate_kind = 'thread' AND stream_id IN (SELECT thread_id FROM copied_threads))
307 OR aggregate_kind < 'thread' OR aggregate_kind > 'thread'`
308 : nullable === undefined || name === PROVIDER_SESSIONS_TABLE
309 ? ""
310 : `WHERE thread_id IN (SELECT thread_id FROM copied_threads)${nullable ? " OR thread_id IS NULL" : ""}`;
311 yield* sql.unsafe(`INSERT INTO main."${name}" SELECT * FROM src."${name}" ${filter}`)
312 .unprepared;
313 }
314 yield* sql`DELETE FROM main.sqlite_sequence`;
315 yield* sql`INSERT INTO main.sqlite_sequence SELECT name, seq FROM src.sqlite_sequence`;
316 for (const entry of schema) {
317 if (entry.type !== "table") yield* sql.unsafe(entry.sql).unprepared;
318 }
319 yield* rebuildMessageSearch(sql);
320 }),
321 );
322 yield* sql`DETACH DATABASE src`;
323});
324
325const pruneSnapshot = Effect.fn("pruneDevDbSnapshot")(function* (input: RunMigrateDevDbInput) {
326 const sql = yield* SqlClient.SqlClient;
327
328 // Usage-limit recovery also depends on settings; exclude latest-run failures and recovery choices.
329 const projections = yield* ProjectionStore.ProjectionStoreV2;
330 const recoveryThreadIds = yield* Effect.forEach(RECOVERY_KINDS, (kind) =>
331 projections.getRecoveryThreadIds(kind),
332 );
333 yield* sql`CREATE TEMP TABLE live_threads (thread_id TEXT PRIMARY KEY)`;
334 for (const threadId of new Set(recoveryThreadIds.flat())) {
335 yield* sql`INSERT INTO live_threads (thread_id) VALUES (${threadId})`;
336 }
337 yield* sql`INSERT OR IGNORE INTO live_threads (thread_id)
338 SELECT r.thread_id FROM orchestration_v2_projection_runs r
339 WHERE r.status = 'failed' AND r.ordinal = (
340 SELECT MAX(latest.ordinal) FROM orchestration_v2_projection_runs latest
341 WHERE latest.thread_id = r.thread_id)
342 UNION
343 SELECT thread_id FROM orchestration_v2_projection_threads
344 WHERE json_extract(payload_json, '$.limitRecovery') IS NOT NULL`;
345
346 // Clone or drop whole thread families to preserve fork and subagent lineage reads.
347 yield* sql`CREATE TEMP TABLE thread_families AS
348 SELECT thread_id, updated_at,
349 COALESCE(json_extract(payload_json, '$.lineage.rootThreadId'), thread_id) AS root_id
350 FROM orchestration_v2_projection_threads`;
351
352 yield* sql`CREATE TEMP TABLE stopped_families AS
353 SELECT f.root_id, root.project_id, MAX(f.updated_at) AS updated_at, COUNT(*) AS size
354 FROM thread_families f
355 JOIN orchestration_v2_projection_threads root ON root.thread_id = f.root_id
356 WHERE ${rootIsVisible(sql)}
357 GROUP BY f.root_id, root.project_id
358 HAVING SUM(f.thread_id IN (SELECT thread_id FROM live_threads)) = 0`;
359
360 yield* sql`CREATE TEMP TABLE kept_projects AS
361 SELECT p.project_id
362 FROM projection_projects p
363 LEFT JOIN (
364 SELECT project_id, MAX(updated_at) AS last_stopped_at
365 FROM stopped_families
366 GROUP BY project_id
367 ) q ON q.project_id = p.project_id
368 WHERE p.deleted_at IS NULL
369 ORDER BY (q.last_stopped_at IS NULL) ASC,
370 COALESCE(q.last_stopped_at, p.updated_at) DESC
371 LIMIT ${input.projects}`;
372
373 yield* sql`CREATE TEMP TABLE kept_threads AS
374 SELECT f.thread_id FROM thread_families f
375 WHERE f.root_id IN (
376 SELECT root_id FROM (
377 SELECT
378 sf.root_id,
379 ROW_NUMBER() OVER (
380 PARTITION BY sf.project_id
381 ORDER BY sf.updated_at DESC
382 ) AS recency_rank
383 FROM stopped_families sf
384 JOIN kept_projects kp ON kp.project_id = sf.project_id
385 -- One orchestrator family can hold hundreds of subagents and most of
386 -- the database, so a family larger than the limit is skipped.
387 WHERE sf.size <= ${input.threadsPerProject}
388 )
389 WHERE recency_rank <= ${input.threadsPerProject}
390 )`;
391
392 const threadTables = yield* sql<{ name: string }>`
393 SELECT m.name FROM sqlite_master m, pragma_table_info(m.name) c
394 WHERE m.type = 'table' AND c.name = 'thread_id'
395 AND m.name <> ${PROVIDER_SESSIONS_TABLE}`;
396
397 yield* sql.withTransaction(
398 Effect.gen(function* () {
399 yield* sql`DELETE FROM projection_projects
400 WHERE project_id NOT IN (SELECT project_id FROM kept_projects)`;
401 yield* sql`DELETE FROM ${sql(PROVIDER_SESSIONS_TABLE)}
402 WHERE COALESCE(thread_id, '') NOT IN (SELECT thread_id FROM kept_threads)
403 AND provider_session_id NOT IN (
404 SELECT provider_session_id FROM orchestration_v2_projection_provider_session_bindings
405 WHERE thread_id IN (SELECT thread_id FROM kept_threads)
406 )`;
407 for (const { name } of threadTables) {
408 yield* sql.unsafe(
409 `DELETE FROM "${name}" WHERE thread_id NOT IN (SELECT thread_id FROM kept_threads)`,
410 ).unprepared;
411 }
412 yield* sql`DELETE FROM orchestration_v2_projection_context_transfers
413 WHERE source_thread_id NOT IN (SELECT thread_id FROM kept_threads)
414 OR target_thread_id NOT IN (SELECT thread_id FROM kept_threads)`;
415 yield* sql`DELETE FROM orchestration_events
416 WHERE (aggregate_kind = 'thread'
417 AND stream_id NOT IN (SELECT thread_id FROM kept_threads))
418 OR (aggregate_kind = 'project'
419 AND stream_id NOT IN (SELECT project_id FROM kept_projects))`;
420 yield* sql`UPDATE orchestration_v2_projection_threads
421 SET payload_json = json_set(payload_json, '$.pullRequests', json((
422 SELECT json_group_array(json_remove(link.value, '$.watch') ORDER BY link.key)
423 FROM json_each(payload_json, '$.pullRequests') AS link)))
424 WHERE EXISTS (
425 SELECT 1 FROM json_each(payload_json, '$.pullRequests') AS link
426 WHERE json_extract(link.value, '$.watch') IS NOT NULL)`;
427 // Copied threads keep their real worktree paths, where a settle script would run.
428 yield* sql`UPDATE projection_projects
429 SET scripts_json = (
430 SELECT json_group_array(json_remove(script.value, '$.runOnSettle') ORDER BY script.key)
431 FROM json_each(scripts_json) AS script)
432 WHERE EXISTS (
433 SELECT 1 FROM json_each(scripts_json) AS script
434 WHERE json_extract(script.value, '$.runOnSettle') IS NOT NULL)`;
435 for (const table of CLEARED_TABLES) {
436 yield* sql`DELETE FROM ${sql(table)}`;
437 }
438 }),
439 );
440
441 const keptProjects = yield* sql<{ title: string; threads: number }>`
442 SELECT
443 p.title,
444 (SELECT COUNT(*) FROM orchestration_v2_projection_threads t
445 WHERE t.project_id = p.project_id) AS threads
446 FROM projection_projects p
447 ORDER BY p.updated_at DESC`;
448 const [events] = yield* sql<{ count: number }>`
449 SELECT COUNT(*) AS count FROM orchestration_events`;
450
451 return {
452 projects: keptProjects as ReadonlyArray<KeptProject>,
453 eventCount: events?.count ?? 0,
454 };
455});
456
457// Peer links preview builds recorded 68-70 before main took those slots for other migrations.
458// A snapshot with exactly that history drops the preview's schema so main's 68+ can run.
459const PEER_LINKS_PREVIEW_HISTORY = [
460 "68:PeerLinks",
461 "69:PeerLinkFence",
462 "70:WaitingLinkRequestIndex",
463];
464
465const forgetPeerLinksPreviewHistory = Effect.fn("forgetPeerLinksPreviewHistory")(function* () {
466 const sql = yield* SqlClient.SqlClient;
467 const tables = yield* sql`
468 SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'effect_sql_migrations'`;
469 if (tables.length === 0) return;
470 const tail = yield* sql<{ migration_id: number; name: string }>`
471 SELECT migration_id, name FROM effect_sql_migrations WHERE migration_id >= 68
472 ORDER BY migration_id`;
473 const history = tail.map((row) => `${row.migration_id}:${row.name}`);
474 if (history.join() !== PEER_LINKS_PREVIEW_HISTORY.join()) return;
475 yield* sql.withTransaction(
476 Effect.gen(function* () {
477 yield* sql`DROP INDEX IF EXISTS orchestration_v2_projection_turn_items_waiting_link_idx`;
478 yield* sql`DROP TABLE IF EXISTS peer_environment_links`;
479 yield* sql`DELETE FROM effect_sql_migrations WHERE migration_id >= 68`;
480 }),
481 );
482});
483
484export const runMigrateDevDb = Effect.fn("runMigrateDevDb")(function* (
485 input: RunMigrateDevDbInput,
486 options: RunMigrateDevDbOptions = {},
487) {
488 // Negative SQLite LIMITs mean unlimited; validate direct callers as well as CLI flags.
489 if (input.projects < 1 || input.threadsPerProject < 0) {
490 return yield* Effect.die("projects must be >= 1 and threadsPerProject >= 0");
491 }
492 const fs = yield* FileSystem.FileSystem;
493 const path = yield* Path.Path;
494
495 const sharedHome = path.resolve(options.sharedHome ?? path.join(NodeOS.homedir(), ".t3"));
496 const sourcePath = path.resolve(
497 input.source ?? path.join(sharedHome, "userdata", "statev2.sqlite"),
498 );
499
500 const baseDir =
501 input.baseDir !== undefined
502 ? path.resolve(input.baseDir)
503 : yield* resolveWorktreeT3Home(process.cwd());
504 if (baseDir === undefined) {
505 return yield* new MigrateDevDbNotInWorktreeError();
506 }
507 const stateDir = path.join(baseDir, "userdata");
508 const databasePath = path.join(stateDir, "statev2.sqlite");
509 const snapshotPath = `${databasePath}.migrate-dev-db-tmp`;
510
511 if (!(yield* fs.exists(sourcePath))) {
512 return yield* new MigrateDevDbSourceMissingError({ sourcePath });
513 }
514 const [canonicalBaseDir, canonicalSharedHome] = yield* Effect.all([
515 fs.realPath(baseDir).pipe(Effect.orElseSucceed(() => baseDir)),
516 fs.realPath(sharedHome).pipe(Effect.orElseSucceed(() => sharedHome)),
517 ]);
518 if (canonicalBaseDir === canonicalSharedHome) {
519 return yield* new MigrateDevDbSharedHomeError();
520 }
521 // Both destination and snapshot are deleted below; reject sources resolving to either.
522 const canonicalSourcePath = yield* fs
523 .realPath(sourcePath)
524 .pipe(Effect.orElseSucceed(() => sourcePath));
525 for (const destination of [databasePath, snapshotPath]) {
526 const canonicalDestination = yield* fs
527 .realPath(destination)
528 .pipe(Effect.orElseSucceed(() => destination));
529 if (canonicalSourcePath === canonicalDestination) {
530 return yield* new MigrateDevDbSourceIsDestinationError({ sourcePath });
531 }
532 }
533
534 yield* fs.makeDirectory(stateDir, { recursive: true });
535
536 const wrapPhase =
537 (phase: MigrateDevDbPhaseError["phase"], phaseDatabasePath: string) =>
538 <A, E, R>(effect: Effect.Effect<A, E, R>) =>
539 effect.pipe(
540 Effect.mapError(
541 (cause) => new MigrateDevDbPhaseError({ phase, databasePath: phaseDatabasePath, cause }),
542 ),
543 );
544
545 yield* removeDatabaseFiles(snapshotPath);
546 const { executedMigrations, pruned } = yield* Effect.gen(function* () {
547 yield* Console.log(`Copying visible threads from ${sourcePath} (read-only)...`);
548 yield* copySourceSlice(sourcePath).pipe(
549 Effect.provide(NodeSqliteClient.layer({ filename: snapshotPath })),
550 Effect.catchTags({
551 SqlError: (cause) =>
552 Effect.fail(
553 new MigrateDevDbPhaseError({ phase: "snapshot", databasePath: sourcePath, cause }),
554 ),
555 }),
556 );
557 // Checkpoints the destination, so it runs after the copy's legacy guard.
558 yield* ensureNotInUse(databasePath);
559
560 // Migrate first so prune queries can use columns missing from older sources.
561 yield* Console.log("Running migrations on the snapshot...");
562 const executed = yield* Effect.gen(function* () {
563 const sql = yield* SqlClient.SqlClient;
564 yield* sql.unsafe("PRAGMA foreign_keys = ON").unprepared;
565 yield* forgetPeerLinksPreviewHistory();
566 return yield* runMigrations();
567 }).pipe(
568 Effect.provide(NodeSqliteClient.layer({ filename: snapshotPath })),
569 wrapPhase("migrate", snapshotPath),
570 );
571
572 yield* Console.log(
573 `Pruning to ${input.projects} projects, ${input.threadsPerProject} stopped threads each...`,
574 );
575 const result = yield* pruneSnapshot(input).pipe(
576 Effect.provide(
577 ProjectionStore.layer.pipe(
578 Layer.provideMerge(NodeSqliteClient.layer({ filename: snapshotPath })),
579 ),
580 ),
581 wrapPhase("prune", snapshotPath),
582 );
583
584 yield* Console.log(`Compacting into ${databasePath}...`);
585 // A server may have started during pruning; re-check before replacing its database.
586 yield* ensureNotInUse(databasePath);
587 yield* removeDatabaseFiles(databasePath);
588 yield* Effect.gen(function* () {
589 const sql = yield* SqlClient.SqlClient;
590 yield* sql`VACUUM INTO ${databasePath}`;
591 }).pipe(
592 Effect.provide(NodeSqliteClient.layer({ filename: snapshotPath })),
593 wrapPhase("compact", databasePath),
594 );
595 return { executedMigrations: executed, pruned: result };
596 }).pipe(Effect.ensuring(removeDatabaseFiles(snapshotPath)));
597 yield* fs.chmod(databasePath, 0o600);
598
599 yield* Effect.gen(function* () {
600 const sql = yield* SqlClient.SqlClient;
601 // VACUUM INTO loses WAL mode; restore the server boot setting.
602 yield* sql.unsafe("PRAGMA journal_mode = WAL").unprepared;
603 }).pipe(
604 Effect.provide(NodeSqliteClient.layer({ filename: databasePath })),
605 wrapPhase("compact", databasePath),
606 );
607
608 const size = (yield* fs.stat(databasePath)).size;
609 return {
610 databasePath,
611 sizeBytes: Number(size),
612 projects: pruned.projects,
613 eventCount: pruned.eventCount,
614 executedMigrations: executedMigrations.map(([id, name]) => `${id}_${name}`),
615 };
616});
617
618const formatSize = (bytes: number): string =>
619 bytes >= 1024 * 1024
620 ? `${(bytes / (1024 * 1024)).toFixed(1)} MB`
621 : `${(bytes / 1024).toFixed(0)} KB`;
622
623export const migrateDevDbCommand = Command.make(
624 "migrate-dev-db",
625 {
626 projects: Flag.Int("projects").pipe(
627 Flag.withDefault(5),
628 Flag.withDescription("How many recently updated projects to keep."),
629 ),
630 threadsPerProject: Flag.Int("threads-per-project").pipe(
631 Flag.withDefault(10),
632 Flag.withDescription(
633 "How many recent stopped threads, with their forks and subagents, to keep per project. A family with more threads than this is skipped.",
634 ),
635 ),
636 baseDir: Flag.String("base-dir").pipe(
637 Flag.optional,
638 Flag.withDescription("Isolated .t3 directory. Defaults to the current worktree's .t3."),
639 ),
640 source: Flag.String("source").pipe(
641 Flag.optional,
642 Flag.withDescription("Source database. Defaults to ~/.t3/userdata/statev2.sqlite."),
643 ),
644 },
645 ({ projects, threadsPerProject, baseDir, source }) =>
646 Effect.gen(function* () {
647 const result = yield* runMigrateDevDb({
648 projects,
649 threadsPerProject,
650 baseDir: Option.getOrUndefined(baseDir),
651 source: Option.getOrUndefined(source),
652 });
653 yield* Console.log("");
654 yield* Console.log(
655 `Dev database ready: ${result.databasePath} (${formatSize(result.sizeBytes)})`,
656 );
657 for (const project of result.projects) {
658 yield* Console.log(` ${project.title}: ${project.threads} threads`);
659 }
660 yield* Console.log(` ${result.eventCount} orchestration events kept`);
661 yield* Console.log(
662 result.executedMigrations.length === 0
663 ? " Migrations: already current (no new migrations in this checkout)"
664 : ` Migrations applied: ${result.executedMigrations.join(", ")}`,
665 );
666 }),
667).pipe(
668 Command.withDescription(
669 "Rebuild the worktree dev database from a pruned snapshot of the real ~/.t3 data, then run migrations.",
670 ),
671);
672
673if (import.meta.main) {
674 Command.run(migrateDevDbCommand, { version: "0.0.0" }).pipe(
675 Effect.provide(NodeServices.layer),
676 NodeRuntime.runMain,
677 );
678}