1#!/usr/bin/env node
2
3/**
4 * Rebuild an isolated dev database from a pruned snapshot of the real
5 * ~/.t3 database, then run this checkout's migrations against it.
6 *
7 * `vp run migrate-dev-db` from a worktree:
8 * 1. Nukes `<worktree>/.t3/userdata/statev2.sqlite`.
9 * 2. Copies a slice of the real db (`~/.t3/userdata/statev2.sqlite`,
10 * attached read-only): the schema, and every row except those of
11 * deleted, archived, and settled threads, which are nearly all of a
12 * long-lived database. It then prunes that to the most recently updated
13 * projects and, per project, the most recent threads that have fully
14 * stopped, with
15 * their forks and subagents. Working, settled, and archived threads, and
16 * threads with pending recovery, are skipped, and scheduled tasks and
17 * queued effects are dropped, so the dev server never adopts live work.
18 * Auth sessions, pairing links, command receipts, and provider
19 * runtime rows are dropped — pair a fresh browser against dev.
20 * 3. Runs migrations on the result. Because the clone carries the real
21 * `effect_sql_migrations` table, this proves a new migration applies
22 * on top of the real applied set, and the slot check below catches
23 * the silent failure where two branches claim the same
24 * `Migrations/NNN_` id (the second one's CREATE TABLE is skipped).
25 *
26 * The event log (`orchestration_events`) is pruned per stream while
27 * `sqlite_sequence` and the projection cursors carry over untouched, so new events keep appending
28 * after the old high-water mark and projection cursors never rewind.
29 */
30
31import * as NodeRuntime from "@effect/platform-node/NodeRuntime";
32import * as NodeServices from "@effect/platform-node/NodeServices";
33import * as NodeOS from "node:os";
34import * as NodeURL from "node:url";
35import { resolveWorktreeT3Home } from "@t3tools/shared/devHome";
36import * as Console from "effect/Console";
37import * as Effect from "effect/Effect";
38import * as FileSystem from "effect/FileSystem";
39import * as Layer from "effect/Layer";
40import * as Option from "effect/Option";
41import * as Path from "effect/Path";
42import * as Schema from "effect/Schema";
43import * as SqlClient from "effect/sql/SqlClient";
44import { Command, Flag } from "effect/cli";
45
46import * as ProjectionStore from "../src/orchestration-v2/ProjectionStore.ts";
47import { migrationManifest, runMigrations } from "../src/persistence/Migrations.ts";
48import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient";
49
50export class MigrateDevDbNotInWorktreeError extends Schema.TaggedError<MigrateDevDbNotInWorktreeError>()(
51 "MigrateDevDbNotInWorktreeError",
52 {},
53) {
54 override get message(): string {
55 return "Not inside a linked git worktree. Pass --base-dir to target an isolated .t3 directory.";
56 }
57}
58
59export class MigrateDevDbSharedHomeError extends Schema.TaggedError<MigrateDevDbSharedHomeError>()(
60 "MigrateDevDbSharedHomeError",
61 {},
62) {
63 override get message(): string {
64 return "Refusing to rebuild the shared ~/.t3 database. Use an isolated --base-dir.";
65 }
66}
67
68export class MigrateDevDbSourceMissingError extends Schema.TaggedError<MigrateDevDbSourceMissingError>()(
69 "MigrateDevDbSourceMissingError",
70 {
71 sourcePath: Schema.String,
72 },
73) {
74 override get message(): string {
75 return `Source database does not exist at '${this.sourcePath}'.`;
76 }
77}
78
79export class MigrateDevDbSourceIsDestinationError extends Schema.TaggedError<MigrateDevDbSourceIsDestinationError>()(
80 "MigrateDevDbSourceIsDestinationError",
81 {
82 sourcePath: Schema.String,
83 },
84) {
85 override get message(): string {
86 return `Source database '${this.sourcePath}' resolves to a path this command rewrites. Pick a different --source or --base-dir.`;
87 }
88}
89
90export class MigrateDevDbServerRunningError extends Schema.TaggedError<MigrateDevDbServerRunningError>()(
91 "MigrateDevDbServerRunningError",
92 {
93 databasePath: Schema.String,
94 pid: Schema.Number,
95 },
96) {
97 override get message(): string {
98 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.`;
99 }
100}
101
102export class MigrateDevDbDestinationBusyError extends Schema.TaggedError<MigrateDevDbDestinationBusyError>()(
103 "MigrateDevDbDestinationBusyError",
104 {
105 databasePath: Schema.String,
106 reason: Schema.Literals(["write-locked", "wal-held"]),
107 cause: Schema.optional(Schema.Defect()),
108 },
109) {
110 override get message(): string {
111 const detail =
112 this.reason === "write-locked"
113 ? "the database is write-locked"
114 : "another connection is holding its WAL";
115 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.`;
116 }
117}
118
119/**
120 * Two branches claimed the same Migrations/NNN_ slot: the id was already
121 * recorded under a different name, so this checkout's migration was
122 * silently skipped and its schema changes never applied.
123 */
124export class MigrateDevDbSlotCollisionError extends Schema.TaggedError<MigrateDevDbSlotCollisionError>()(
125 "MigrateDevDbSlotCollisionError",
126 {
127 slot: Schema.Number,
128 codeName: Schema.String,
129 appliedName: Schema.String,
130 },
131) {
132 override get message(): string {
133 return `Migration slot collision at ${this.slot}: this checkout registers '${this.codeName}' but the database already applied '${this.appliedName}' in that slot. Renumber the new migration to a free slot.`;
134 }
135}
136
137export class MigrateDevDbPhaseError extends Schema.TaggedError<MigrateDevDbPhaseError>()(
138 "MigrateDevDbPhaseError",
139 {
140 phase: Schema.Literals(["snapshot", "prune", "compact", "migrate", "verify"]),
141 databasePath: Schema.String,
142 cause: Schema.Defect(),
143 },
144) {
145 override get message(): string {
146 return `migrate-dev-db failed during ${this.phase} on '${this.databasePath}'.`;
147 }
148}
149
150export interface RunMigrateDevDbInput {
151 /** Isolated .t3 directory. Defaults to `<worktree>/.t3` of the cwd. */
152 readonly baseDir?: string | undefined;
153 /** Source database. Defaults to `~/.t3/userdata/statev2.sqlite`. */
154 readonly source?: string | undefined;
155 readonly projects: number;
156 readonly threadsPerProject: number;
157}
158
159export interface RunMigrateDevDbOptions {
160 /** Overridable for tests; the directory writes must never target. */
161 readonly sharedHome?: string | undefined;
162}
163
164interface KeptProject {
165 readonly title: string;
166 readonly threads: number;
167}
168
169const removeDatabaseFiles = Effect.fn("removeDatabaseFiles")(function* (databasePath: string) {
170 const fs = yield* FileSystem.FileSystem;
171 for (const suffix of ["", "-wal", "-shm"]) {
172 yield* fs.remove(`${databasePath}${suffix}`).pipe(Effect.orElseSucceed(() => undefined));
173 }
174});
175
176/** The slice of server-runtime.json this script cares about. */
177const ServerRuntimeState = Schema.fromJsonString(Schema.Struct({ pid: Schema.Number }));
178const decodeServerRuntimeState = Schema.decodeEffect(ServerRuntimeState);
179
180const isProcessAlive = (pid: number): boolean => {
181 try {
182 process.kill(pid, 0);
183 return true;
184 } catch (error) {
185 // EPERM means the process exists but belongs to someone else.
186 return (error as NodeJS.ErrnoException).code === "EPERM";
187 }
188};
189
190/** Liveness probe for a running dev server. The server writes its pid to
191 * server-runtime.json next to the database, which also catches an idle
192 * server holding an open-but-inactive connection. The SQL probes below back
193 * that up: BEGIN IMMEDIATE fails while a writer is active, and
194 * wal_checkpoint(TRUNCATE) reports busy while another connection holds the
195 * WAL. A leftover -shm alone is not a signal — read-only connections cannot
196 * clean it up on close. */
197const ensureNotInUse = Effect.fn("ensureDevDbNotInUse")(function* (databasePath: string) {
198 const fs = yield* FileSystem.FileSystem;
199 const path = yield* Path.Path;
200
201 const runtimeStatePath = path.join(path.dirname(databasePath), "server-runtime.json");
202 const runtimeState = yield* fs.readFileString(runtimeStatePath).pipe(
203 Effect.flatMap(decodeServerRuntimeState),
204 // A missing or malformed descriptor is not a liveness signal.
205 Effect.option,
206 );
207 if (Option.isSome(runtimeState) && isProcessAlive(runtimeState.value.pid)) {
208 return yield* new MigrateDevDbServerRunningError({
209 databasePath,
210 pid: runtimeState.value.pid,
211 });
212 }
213
214 if (!(yield* fs.exists(databasePath))) {
215 return;
216 }
217 const checkpoint = yield* Effect.gen(function* () {
218 const sql = yield* SqlClient.SqlClient;
219 yield* sql.unsafe("PRAGMA busy_timeout = 0").unprepared;
220 yield* sql.unsafe("BEGIN IMMEDIATE").unprepared;
221 yield* sql.unsafe("ROLLBACK").unprepared;
222 return yield* sql.unsafe<{ busy: number }>("PRAGMA wal_checkpoint(TRUNCATE)").unprepared;
223 }).pipe(
224 Effect.provide(NodeSqliteClient.layer({ filename: databasePath })),
225 Effect.mapError(
226 (cause) =>
227 new MigrateDevDbDestinationBusyError({
228 databasePath,
229 reason: "write-locked",
230 cause,
231 }),
232 ),
233 );
234 if (checkpoint[0] !== undefined && Number(checkpoint[0].busy) !== 0) {
235 return yield* new MigrateDevDbDestinationBusyError({
236 databasePath,
237 reason: "wal-held",
238 });
239 }
240});
241
242const RECOVERY_KINDS: ReadonlyArray<ProjectionStore.ProjectionRecoveryKind> = [
243 "queued-runs",
244 "runtime",
245 "subagent-results",
246 "delegated-completions",
247];
248
249/** Tables the prune empties. The copy skips their rows. */
250const CLEARED_TABLES: ReadonlyArray<string> = [
251 // Pending work the dev server would otherwise pick up and run.
252 "scheduled_tasks",
253 "orchestration_v2_effect_outbox",
254 "orchestration_v2_thread_launch_workflows",
255 "orchestration_command_receipts",
256 "provider_session_runtime",
257 "auth_sessions",
258 "auth_pairing_links",
259];
260
261/** Shared between threads; the prune keeps the ones a kept thread is bound to. */
262const PROVIDER_SESSIONS_TABLE = "orchestration_v2_projection_provider_sessions";
263
264/** Whether the thread aliased `root` is shown: not deleted, archived, or settled. */
265const rootIsVisible = (sql: SqlClient.SqlClient) => sql`
266 root.deleted_at IS NULL
267 AND json_extract(root.payload_json, '$.deletedAt') IS NULL
268 AND json_extract(root.payload_json, '$.archivedAt') IS NULL
269 AND json_extract(root.payload_json, '$.settledAt') IS NULL
270 AND json_extract(root.payload_json, '$.settledOverride') IS NOT 'settled'`;
271
272/**
273 * Copies the source into the empty snapshot, without the rows of thread families whose root is
274 * hidden. The prune drops those anyway, and they hold nearly all of a long-lived database's
275 * events, so copying them is what used to make this take a full-size copy of the real db.
276 * A source from before the V2 thread tables has no visibility to filter on, so it is copied
277 * whole. One transaction reads the source, so the copy is consistent while the real server writes.
278 */
279const copySourceSlice = Effect.fn("copyDevDbSourceSlice")(function* (sourcePath: string) {
280 const sql = yield* SqlClient.SqlClient;
281 yield* sql`ATTACH DATABASE ${`${NodeURL.pathToFileURL(sourcePath).href}?mode=ro`} AS src`;
282 const schema = yield* sql<{ type: string; name: string; sql: string }>`
283 SELECT type, name, sql FROM src.sqlite_master
284 WHERE sql IS NOT NULL AND name NOT LIKE 'sqlite_%'`;
285 const threadColumns = yield* sql<{ name: string; notnull: number }>`
286 SELECT m.name, c."notnull" FROM src.sqlite_master m, pragma_table_info(m.name, 'src') c
287 WHERE m.type = 'table' AND c.name = 'thread_id'`;
288 const threadIdNullable = new Map(threadColumns.map((row) => [row.name, row.notnull === 0]));
289 const tables = schema.filter((entry) => entry.type === "table");
290 const filtered = tables.some((table) => table.name === "orchestration_v2_projection_threads");
291
292 yield* sql.withTransaction(
293 Effect.gen(function* () {
294 for (const table of tables) yield* sql.unsafe(table.sql).unprepared;
295 if (filtered) {
296 yield* sql`CREATE TEMP TABLE copied_threads (thread_id TEXT PRIMARY KEY)`;
297 yield* sql`INSERT OR IGNORE INTO copied_threads
298 SELECT family.thread_id FROM src.orchestration_v2_projection_threads family
299 JOIN src.orchestration_v2_projection_threads root ON root.thread_id =
300 COALESCE(json_extract(family.payload_json, '$.lineage.rootThreadId'), family.thread_id)
301 WHERE ${rootIsVisible(sql)}`;
302 }
303 for (const { name } of tables) {
304 if (CLEARED_TABLES.includes(name)) continue;
305 const nullable = threadIdNullable.get(name);
306 const filter = !filtered
307 ? ""
308 : name === "orchestration_events"
309 ? // Two ranges rather than `<>`, so SQLite seeks the (aggregate_kind, stream_id)
310 // index instead of scanning every event.
311 `WHERE (aggregate_kind = 'thread' AND stream_id IN (SELECT thread_id FROM copied_threads))
312 OR aggregate_kind < 'thread' OR aggregate_kind > 'thread'`
313 : nullable === undefined || name === PROVIDER_SESSIONS_TABLE
314 ? ""
315 : `WHERE thread_id IN (SELECT thread_id FROM copied_threads)${nullable ? " OR thread_id IS NULL" : ""}`;
316 yield* sql.unsafe(`INSERT INTO main."${name}" SELECT * FROM src."${name}" ${filter}`)
317 .unprepared;
318 }
319 // The source's AUTOINCREMENT high-water marks, so new events append after them.
320 yield* sql`DELETE FROM main.sqlite_sequence`;
321 yield* sql`INSERT INTO main.sqlite_sequence SELECT name, seq FROM src.sqlite_sequence`;
322 for (const entry of schema) {
323 if (entry.type !== "table") yield* sql.unsafe(entry.sql).unprepared;
324 }
325 }),
326 );
327 yield* sql`DETACH DATABASE src`;
328});
329
330const pruneSnapshot = Effect.fn("pruneDevDbSnapshot")(function* (input: RunMigrateDevDbInput) {
331 const sql = yield* SqlClient.SqlClient;
332
333 // Threads the server would resume or start work on by itself. Its own
334 // recovery queries find most of them; usage-limit recovery also depends on
335 // settings, so any thread whose latest run failed or that has a recovery
336 // choice counts as live too.
337 const projections = yield* ProjectionStore.ProjectionStoreV2;
338 const recoveryThreadIds = yield* Effect.forEach(RECOVERY_KINDS, (kind) =>
339 projections.getRecoveryThreadIds(kind),
340 );
341 yield* sql`CREATE TEMP TABLE live_threads (thread_id TEXT PRIMARY KEY)`;
342 for (const threadId of new Set(recoveryThreadIds.flat())) {
343 yield* sql`INSERT INTO live_threads (thread_id) VALUES (${threadId})`;
344 }
345 yield* sql`INSERT OR IGNORE INTO live_threads (thread_id)
346 SELECT r.thread_id FROM orchestration_v2_projection_runs r
347 WHERE r.status = 'failed' AND r.ordinal = (
348 SELECT MAX(latest.ordinal) FROM orchestration_v2_projection_runs latest
349 WHERE latest.thread_id = r.thread_id)
350 UNION
351 SELECT thread_id FROM orchestration_v2_projection_threads
352 WHERE json_extract(payload_json, '$.limitRecovery') IS NOT NULL`;
353
354 // Forks and subagents read history and results through their lineage, so
355 // a thread family is cloned or dropped as a whole. A family is stopped when
356 // its root is visible and unsettled and none of its threads is live.
357 yield* sql`CREATE TEMP TABLE thread_families AS
358 SELECT thread_id, updated_at,
359 COALESCE(json_extract(payload_json, '$.lineage.rootThreadId'), thread_id) AS root_id
360 FROM orchestration_v2_projection_threads`;
361
362 yield* sql`CREATE TEMP TABLE stopped_families AS
363 SELECT f.root_id, root.project_id, MAX(f.updated_at) AS updated_at
364 FROM thread_families f
365 JOIN orchestration_v2_projection_threads root ON root.thread_id = f.root_id
366 WHERE ${rootIsVisible(sql)}
367 GROUP BY f.root_id, root.project_id
368 HAVING SUM(f.thread_id IN (SELECT thread_id FROM live_threads)) = 0`;
369
370 // Projects with clonable threads outrank empty-but-recent ones: the point
371 // of the exercise is thread data, not the project list.
372 yield* sql`CREATE TEMP TABLE kept_projects AS
373 SELECT p.project_id
374 FROM projection_projects p
375 LEFT JOIN (
376 SELECT project_id, MAX(updated_at) AS last_stopped_at
377 FROM stopped_families
378 GROUP BY project_id
379 ) q ON q.project_id = p.project_id
380 WHERE p.deleted_at IS NULL
381 ORDER BY (q.last_stopped_at IS NULL) ASC,
382 COALESCE(q.last_stopped_at, p.updated_at) DESC
383 LIMIT ${input.projects}`;
384
385 yield* sql`CREATE TEMP TABLE kept_threads AS
386 SELECT f.thread_id FROM thread_families f
387 WHERE f.root_id IN (
388 SELECT root_id FROM (
389 SELECT
390 sf.root_id,
391 ROW_NUMBER() OVER (
392 PARTITION BY sf.project_id
393 ORDER BY sf.updated_at DESC
394 ) AS recency_rank
395 FROM stopped_families sf
396 JOIN kept_projects kp ON kp.project_id = sf.project_id
397 )
398 WHERE recency_rank <= ${input.threadsPerProject}
399 )`;
400
401 // Every V2 table and every V1 table the lazy importer reads is keyed by
402 // thread_id, so one sweep covers both and new tables need no change here.
403 // Provider sessions are shared between threads and pruned by binding below.
404 const threadTables = yield* sql<{ name: string }>`
405 SELECT m.name FROM sqlite_master m, pragma_table_info(m.name) c
406 WHERE m.type = 'table' AND c.name = 'thread_id'
407 AND m.name <> ${PROVIDER_SESSIONS_TABLE}`;
408
409 yield* sql.withTransaction(
410 Effect.gen(function* () {
411 yield* sql`DELETE FROM projection_projects
412 WHERE project_id NOT IN (SELECT project_id FROM kept_projects)`;
413 yield* sql`DELETE FROM ${sql(PROVIDER_SESSIONS_TABLE)}
414 WHERE COALESCE(thread_id, '') NOT IN (SELECT thread_id FROM kept_threads)
415 AND provider_session_id NOT IN (
416 SELECT provider_session_id FROM orchestration_v2_projection_provider_session_bindings
417 WHERE thread_id IN (SELECT thread_id FROM kept_threads)
418 )`;
419 for (const { name } of threadTables) {
420 yield* sql.unsafe(
421 `DELETE FROM "${name}" WHERE thread_id NOT IN (SELECT thread_id FROM kept_threads)`,
422 ).unprepared;
423 }
424 yield* sql`DELETE FROM orchestration_v2_projection_context_transfers
425 WHERE source_thread_id NOT IN (SELECT thread_id FROM kept_threads)
426 OR target_thread_id NOT IN (SELECT thread_id FROM kept_threads)`;
427 yield* sql`DELETE FROM orchestration_events
428 WHERE (aggregate_kind = 'thread'
429 AND stream_id NOT IN (SELECT thread_id FROM kept_threads))
430 OR (aggregate_kind = 'project'
431 AND stream_id NOT IN (SELECT project_id FROM kept_projects))`;
432 for (const table of CLEARED_TABLES) {
433 yield* sql`DELETE FROM ${sql(table)}`;
434 }
435 }),
436 );
437
438 const keptProjects = yield* sql<{ title: string; threads: number }>`
439 SELECT
440 p.title,
441 (SELECT COUNT(*) FROM orchestration_v2_projection_threads t
442 WHERE t.project_id = p.project_id) AS threads
443 FROM projection_projects p
444 ORDER BY p.updated_at DESC`;
445 const [events] = yield* sql<{ count: number }>`
446 SELECT COUNT(*) AS count FROM orchestration_events`;
447
448 return {
449 projects: keptProjects as ReadonlyArray<KeptProject>,
450 eventCount: events?.count ?? 0,
451 };
452});
453
454/** Compare this checkout's migration registry against what the cloned
455 * database recorded: same slot under a different name means the migration
456 * was skipped, not applied. */
457const verifyMigrationSlots = Effect.fn("verifyMigrationSlots")(function* () {
458 const sql = yield* SqlClient.SqlClient;
459 const applied = yield* sql<{ migration_id: number; name: string }>`
460 SELECT migration_id, name FROM effect_sql_migrations`;
461 const appliedById = new Map(applied.map((row) => [Number(row.migration_id), row.name]));
462 for (const [slot, codeName] of migrationManifest) {
463 const appliedName = appliedById.get(slot);
464 if (appliedName !== undefined && appliedName !== codeName) {
465 return yield* new MigrateDevDbSlotCollisionError({ slot, codeName, appliedName });
466 }
467 }
468});
469
470export const runMigrateDevDb = Effect.fn("runMigrateDevDb")(function* (
471 input: RunMigrateDevDbInput,
472 options: RunMigrateDevDbOptions = {},
473) {
474 // SQLite treats a negative LIMIT as "no limit", which would clone
475 // everything. The CLI flags validate this too; this covers direct callers.
476 if (input.projects < 1 || input.threadsPerProject < 0) {
477 return yield* Effect.die("projects must be >= 1 and threadsPerProject >= 0");
478 }
479 const fs = yield* FileSystem.FileSystem;
480 const path = yield* Path.Path;
481
482 const sharedHome = path.resolve(options.sharedHome ?? path.join(NodeOS.homedir(), ".t3"));
483 const sourcePath = path.resolve(
484 input.source ?? path.join(sharedHome, "userdata", "statev2.sqlite"),
485 );
486
487 const baseDir =
488 input.baseDir !== undefined
489 ? path.resolve(input.baseDir)
490 : yield* resolveWorktreeT3Home(process.cwd());
491 if (baseDir === undefined) {
492 return yield* new MigrateDevDbNotInWorktreeError();
493 }
494 const stateDir = path.join(baseDir, "userdata");
495 const databasePath = path.join(stateDir, "statev2.sqlite");
496 const snapshotPath = `${databasePath}.migrate-dev-db-tmp`;
497
498 if (!(yield* fs.exists(sourcePath))) {
499 return yield* new MigrateDevDbSourceMissingError({ sourcePath });
500 }
501 const [canonicalBaseDir, canonicalSharedHome] = yield* Effect.all([
502 fs.realPath(baseDir).pipe(Effect.orElseSucceed(() => baseDir)),
503 fs.realPath(sharedHome).pipe(Effect.orElseSucceed(() => sharedHome)),
504 ]);
505 if (canonicalBaseDir === canonicalSharedHome) {
506 return yield* new MigrateDevDbSharedHomeError();
507 }
508 // The destination db and snapshot both get deleted below; a --source that
509 // resolves to either (e.g. a leftover snapshot file) would be destroyed
510 // before it is ever read.
511 const canonicalSourcePath = yield* fs
512 .realPath(sourcePath)
513 .pipe(Effect.orElseSucceed(() => sourcePath));
514 for (const destination of [databasePath, snapshotPath]) {
515 const canonicalDestination = yield* fs
516 .realPath(destination)
517 .pipe(Effect.orElseSucceed(() => destination));
518 if (canonicalSourcePath === canonicalDestination) {
519 return yield* new MigrateDevDbSourceIsDestinationError({ sourcePath });
520 }
521 }
522
523 yield* fs.makeDirectory(stateDir, { recursive: true });
524 yield* ensureNotInUse(databasePath);
525
526 const wrapPhase =
527 (phase: MigrateDevDbPhaseError["phase"], phaseDatabasePath: string) =>
528 <A, E, R>(effect: Effect.Effect<A, E, R>) =>
529 effect.pipe(
530 Effect.mapError(
531 (cause) => new MigrateDevDbPhaseError({ phase, databasePath: phaseDatabasePath, cause }),
532 ),
533 );
534
535 yield* removeDatabaseFiles(snapshotPath);
536 // Make sure the snapshot is removed even when a phase fails partway through.
537 const { executedMigrations, pruned } = yield* Effect.gen(function* () {
538 yield* Console.log(`Copying visible threads from ${sourcePath} (read-only)...`);
539 yield* copySourceSlice(sourcePath).pipe(
540 Effect.provide(NodeSqliteClient.layer({ filename: snapshotPath })),
541 wrapPhase("snapshot", sourcePath),
542 );
543
544 // Migrate before pruning: a source older than this checkout would
545 // otherwise crash the prune queries on columns that don't exist yet.
546 yield* Console.log("Running migrations on the snapshot...");
547 const executed = yield* Effect.gen(function* () {
548 const sql = yield* SqlClient.SqlClient;
549 // Mirror server boot (persistence/Sqlite.ts).
550 yield* sql.unsafe("PRAGMA foreign_keys = ON").unprepared;
551 return yield* runMigrations();
552 }).pipe(
553 Effect.provide(NodeSqliteClient.layer({ filename: snapshotPath })),
554 wrapPhase("migrate", snapshotPath),
555 );
556
557 // Verify while the snapshot is still the only thing touched: a slot
558 // collision must abort before the old worktree db gets replaced with a
559 // schema whose colliding migration was silently skipped.
560 yield* verifyMigrationSlots().pipe(
561 Effect.provide(NodeSqliteClient.layer({ filename: snapshotPath })),
562 Effect.catchTags({
563 SqlError: (cause) =>
564 Effect.fail(
565 new MigrateDevDbPhaseError({ phase: "verify", databasePath: snapshotPath, cause }),
566 ),
567 }),
568 );
569
570 yield* Console.log(
571 `Pruning to ${input.projects} projects, ${input.threadsPerProject} stopped threads each...`,
572 );
573 const result = yield* pruneSnapshot(input).pipe(
574 Effect.provide(
575 ProjectionStore.layer.pipe(
576 Layer.provideMerge(NodeSqliteClient.layer({ filename: snapshotPath })),
577 ),
578 ),
579 wrapPhase("prune", snapshotPath),
580 );
581
582 yield* Console.log(`Compacting into ${databasePath}...`);
583 // Re-check right before the swap: a dev server started while the
584 // snapshot was migrating and pruning must not lose its database.
585 yield* ensureNotInUse(databasePath);
586 yield* removeDatabaseFiles(databasePath);
587 yield* Effect.gen(function* () {
588 const sql = yield* SqlClient.SqlClient;
589 yield* sql`VACUUM INTO ${databasePath}`;
590 }).pipe(
591 Effect.provide(NodeSqliteClient.layer({ filename: snapshotPath })),
592 wrapPhase("compact", databasePath),
593 );
594 return { executedMigrations: executed, pruned: result };
595 }).pipe(Effect.ensuring(removeDatabaseFiles(snapshotPath)));
596 yield* fs.chmod(databasePath, 0o600);
597
598 yield* Effect.gen(function* () {
599 const sql = yield* SqlClient.SqlClient;
600 // WAL does not survive VACUUM INTO; set it so first `vp run dev` finds
601 // the database exactly as server boot would have left it.
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.",
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}