mirror of
https://github.com/anomalyco/opencode.git
synced 2026-08-30 13:22:00 +00:00
refactor(core): simplify V1 migration effects (#45685)
This commit is contained in:
Kit Langton
GitHub
parent
134cdda333
commit
facd7ff452
1 changed files with 129 additions and 132 deletions
|
|
@ -481,7 +481,7 @@ export function status(): Effect.Effect<Status, never, Database.Service> {
|
|||
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<RunResult, never, Data
|
|||
const state = yield* readState(db)
|
||||
if (state?.phase === "completed") return { status: "completed" as const }
|
||||
if (!(yield* hasLegacySessions(db))) return { status: "completed" as const }
|
||||
const migrate = Effect.gen(function* () {
|
||||
const now = Date.now()
|
||||
yield* db.run(sql`
|
||||
const now = Date.now()
|
||||
yield* db.run(sql`
|
||||
INSERT OR IGNORE INTO project (id, worktree, time_created, time_updated, sandboxes)
|
||||
VALUES (${Project.ID.global}, ${path.parse(global.data).root}, ${now}, ${now}, '[]')
|
||||
`)
|
||||
if (state === undefined)
|
||||
yield* db
|
||||
.transaction((tx) =>
|
||||
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<RunResult, never, Data
|
|||
FROM session
|
||||
WHERE id = ${nextID.id}
|
||||
`)
|
||||
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<SourceMessage>(
|
||||
sql`SELECT id, session_id, time_created, time_updated, data FROM message WHERE session_id = ${next.id}`,
|
||||
)
|
||||
const sourceParts = yield* tx.all<SourcePart>(
|
||||
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<SourceMessage>(
|
||||
sql`SELECT id, session_id, time_created, time_updated, data FROM message WHERE session_id = ${next.id}`,
|
||||
)
|
||||
const sourceParts = yield* tx.all<SourcePart>(
|
||||
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(
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue