diff --git a/packages/core/src/database/v1-migration.bun.ts b/packages/core/src/database/v1-migration.bun.ts index 74212df1354..193d454831d 100644 --- a/packages/core/src/database/v1-migration.bun.ts +++ b/packages/core/src/database/v1-migration.bun.ts @@ -481,7 +481,7 @@ export function status(): Effect.Effect { if (runtimeState.status === "error") return runtimeState if (state?.phase === "completed") return { status: "completed" as const } return { status: "required" as const } - }).pipe(Effect.orDie) + }) } export const layer = Layer.effectDiscard( @@ -521,76 +521,75 @@ export function run(options: Options = {}): Effect.Effect - Effect.gen(function* () { - while (true) { - yield* tx.run(sql` + if (state === undefined) + yield* db + .transaction((tx) => + Effect.gen(function* () { + while (true) { + yield* tx.run(sql` DELETE FROM event WHERE rowid IN (SELECT rowid FROM event LIMIT ${EVENT_DELETE_BATCH_SIZE}) `) - const deleted = (yield* tx.get<{ value: number }>(sql`SELECT changes() AS value`))?.value ?? 0 - if (deleted < EVENT_DELETE_BATCH_SIZE) break - yield* Effect.yieldNow - } - yield* tx - .insert(KVTable) - .values({ key: MIGRATION_STATE_KEY, value: { phase: "sessions" } }) - .run() - }), - ) - .pipe(Effect.orDie) - const sourceTotal = yield* countNextSessions(nextPath(options, global.data)) - const legacyTotal = (yield* db.get<{ value: number }>(sql`SELECT COUNT(*) AS value FROM session`))?.value ?? 0 - const cursor = state?.phase === "sessions" ? state.cursor : undefined - const migrated = - cursor !== undefined - ? ((yield* db.get<{ value: number }>(sql`SELECT COUNT(*) AS value FROM session WHERE id >= ${cursor}`)) - ?.value ?? 0) - : 0 - const denominator = sourceTotal + legacyTotal - updateProgress({ label: "Migrating sessions", numerator: migrated, denominator }) - yield* importNextDatabase(db, nextPath(options, global.data), (completed) => { - updateProgress({ label: "Migrating sessions", numerator: migrated + completed, denominator }) - }) - updateProgress({ label: "Migrating sessions", numerator: migrated + sourceTotal, denominator }) - const projects = new Set( - (yield* db.all<{ id: string }>(sql`SELECT id FROM project`)).map((project) => project.id), - ) - while (true) { - const state = yield* readState(db) - const cursorValue = state?.phase === "sessions" ? state.cursor : undefined - const nextID = yield* db.get<{ id: string; project_id: string }>( - cursorValue === undefined - ? sql`SELECT id, project_id FROM session ORDER BY id DESC LIMIT 1` - : sql`SELECT id, project_id FROM session WHERE id < ${cursorValue} ORDER BY id DESC LIMIT 1`, + const deleted = (yield* tx.get<{ value: number }>(sql`SELECT changes() AS value`))?.value ?? 0 + if (deleted < EVENT_DELETE_BATCH_SIZE) break + yield* Effect.yieldNow + } + yield* tx + .insert(KVTable) + .values({ key: MIGRATION_STATE_KEY, value: { phase: "sessions" } }) + .run() + }), ) - if (!nextID) break - yield* db - .transaction((tx) => - Effect.gen(function* () { - yield* tx - .insert(KVTable) - .values({ key: MIGRATION_STATE_KEY, value: { phase: "sessions", cursor: nextID.id } }) - .onConflictDoUpdate({ - target: KVTable.key, - set: { value: { phase: "sessions", cursor: nextID.id }, time_updated: Date.now() }, - }) - .run() - const projectID = projects.has(nextID.project_id) ? nextID.project_id : Project.ID.global - if (projectID !== nextID.project_id) - yield* Effect.logWarning("Reassigned V1 session with missing project", { - sessionID: nextID.id, - projectID: nextID.project_id, - }) - yield* tx.run(sql` + .pipe(Effect.orDie) + const sourceTotal = yield* countNextSessions(nextPath(options, global.data)) + const legacyTotal = (yield* db.get<{ value: number }>(sql`SELECT COUNT(*) AS value FROM session`))?.value ?? 0 + const cursor = state?.phase === "sessions" ? state.cursor : undefined + const migrated = + cursor !== undefined + ? ((yield* db.get<{ value: number }>(sql`SELECT COUNT(*) AS value FROM session WHERE id >= ${cursor}`)) + ?.value ?? 0) + : 0 + const denominator = sourceTotal + legacyTotal + updateProgress({ label: "Migrating sessions", numerator: migrated, denominator }) + yield* importNextDatabase(db, nextPath(options, global.data), (completed) => { + updateProgress({ label: "Migrating sessions", numerator: migrated + completed, denominator }) + }) + updateProgress({ label: "Migrating sessions", numerator: migrated + sourceTotal, denominator }) + const projects = new Set( + (yield* db.all<{ id: string }>(sql`SELECT id FROM project`)).map((project) => project.id), + ) + while (true) { + const state = yield* readState(db) + const cursorValue = state?.phase === "sessions" ? state.cursor : undefined + const nextID = yield* db.get<{ id: string; project_id: string }>( + cursorValue === undefined + ? sql`SELECT id, project_id FROM session ORDER BY id DESC LIMIT 1` + : sql`SELECT id, project_id FROM session WHERE id < ${cursorValue} ORDER BY id DESC LIMIT 1`, + ) + if (!nextID) break + yield* db + .transaction((tx) => + Effect.gen(function* () { + yield* tx + .insert(KVTable) + .values({ key: MIGRATION_STATE_KEY, value: { phase: "sessions", cursor: nextID.id } }) + .onConflictDoUpdate({ + target: KVTable.key, + set: { value: { phase: "sessions", cursor: nextID.id }, time_updated: Date.now() }, + }) + .run() + const projectID = projects.has(nextID.project_id) ? nextID.project_id : Project.ID.global + if (projectID !== nextID.project_id) + yield* Effect.logWarning("Reassigned V1 session with missing project", { + sessionID: nextID.id, + projectID: nextID.project_id, + }) + yield* tx.run(sql` INSERT OR IGNORE INTO session_v2 ( id, project_id, workspace_id, parent_id, slug, directory, path, title, version, share_url, summary_additions, summary_deletions, summary_files, summary_diffs, metadata, cost, @@ -605,81 +604,79 @@ export function run(options: Options = {}): Effect.Effect( - sql`SELECT id, session_id, time_created, time_updated, data FROM message WHERE session_id = ${next.id}`, - ) - const sourceParts = yield* tx.all( - sql`SELECT id, message_id, session_id, time_created, time_updated, data FROM part WHERE session_id = ${next.id}`, - ) - const transformed = transformSession({ session: next, messages: sourceMessages, parts: sourceParts }) - yield* Effect.forEach(transformed.warnings, (warning) => - Effect.logWarning("Skipped V1 migration row", warning), - ) - yield* tx.delete(SessionMessageTable).where(eq(SessionMessageTable.session_id, next.id)).run() - yield* Effect.forEach(transformed.messages, (message) => - tx - .insert(SessionMessageTable) - .values({ - id: SessionMessage.ID.make(message.id), - session_id: SessionSchema.ID.make(message.session_id), - type: message.type, - seq: message.seq, - time_created: message.time_created, - time_updated: message.time_updated, - data: sql`${JSON.stringify(message.data)}`, - }) - .run(), - ) - yield* tx - .update(SessionTable) - .set({ ...transformed.session, time_updated: next.time_updated }) - .where(eq(SessionTable.id, next.id)) - .run() - yield* tx - .insert(EventSequenceTable) - .values({ aggregate_id: next.id, seq: transformed.watermark }) - .onConflictDoUpdate({ - target: EventSequenceTable.aggregate_id, - set: { seq: transformed.watermark, owner_id: null }, + const next = yield* tx + .select() + .from(SessionTable) + .where(eq(SessionTable.id, SessionSchema.ID.make(nextID.id))) + .get() + if (!next) return yield* Effect.die(new Error(`Failed to copy V1 session ${nextID.id}`)) + const sourceMessages = yield* tx.all( + sql`SELECT id, session_id, time_created, time_updated, data FROM message WHERE session_id = ${next.id}`, + ) + const sourceParts = yield* tx.all( + sql`SELECT id, message_id, session_id, time_created, time_updated, data FROM part WHERE session_id = ${next.id}`, + ) + const transformed = transformSession({ session: next, messages: sourceMessages, parts: sourceParts }) + yield* Effect.forEach(transformed.warnings, (warning) => + Effect.logWarning("Skipped V1 migration row", warning), + ) + yield* tx.delete(SessionMessageTable).where(eq(SessionMessageTable.session_id, next.id)).run() + yield* Effect.forEach(transformed.messages, (message) => + tx + .insert(SessionMessageTable) + .values({ + id: SessionMessage.ID.make(message.id), + session_id: SessionSchema.ID.make(message.session_id), + type: message.type, + seq: message.seq, + time_created: message.time_created, + time_updated: message.time_updated, + data: sql`${JSON.stringify(message.data)}`, }) - .run() - }), - ) - .pipe(Effect.orDie) - if (runtimeState.status === "running") - runtimeState = { - status: "running", - progress: { - label: "Migrating sessions", - numerator: (runtimeState.progress.numerator ?? 0) + 1, - denominator, - }, - } - yield* Effect.yieldNow - } - yield* db - .transaction((tx) => - Effect.gen(function* () { + .run(), + ) yield* tx - .insert(KVTable) - .values({ key: MIGRATION_STATE_KEY, value: { phase: "completed" } }) + .update(SessionTable) + .set({ ...transformed.session, time_updated: next.time_updated }) + .where(eq(SessionTable.id, next.id)) + .run() + yield* tx + .insert(EventSequenceTable) + .values({ aggregate_id: next.id, seq: transformed.watermark }) .onConflictDoUpdate({ - target: KVTable.key, - set: { value: { phase: "completed" }, time_updated: Date.now() }, + target: EventSequenceTable.aggregate_id, + set: { seq: transformed.watermark, owner_id: null }, }) .run() }), ) .pipe(Effect.orDie) - return { status: "completed" as const } - }) - return yield* migrate + if (runtimeState.status === "running") + runtimeState = { + status: "running", + progress: { + label: "Migrating sessions", + numerator: (runtimeState.progress.numerator ?? 0) + 1, + denominator, + }, + } + yield* Effect.yieldNow + } + yield* db + .transaction((tx) => + Effect.gen(function* () { + yield* tx + .insert(KVTable) + .values({ key: MIGRATION_STATE_KEY, value: { phase: "completed" } }) + .onConflictDoUpdate({ + target: KVTable.key, + set: { value: { phase: "completed" }, time_updated: Date.now() }, + }) + .run() + }), + ) + .pipe(Effect.orDie) + return { status: "completed" as const } }).pipe(Effect.orDie), ) } @@ -708,7 +705,7 @@ function countNextSessions(sourcePath: string | undefined) { if (!isNextDatabase(source)) return 0 return source.query<{ value: number }, []>("SELECT COUNT(*) AS value FROM session").get()?.value ?? 0 }), - ).pipe(Effect.orElseSucceed(() => 0)) + ) } function importNextDatabase(