sync
This commit is contained in:
@@ -0,0 +1,147 @@
|
||||
import type { Argv } from "yargs"
|
||||
import { cmd } from "./cmd"
|
||||
import { bootstrap } from "../bootstrap"
|
||||
import { UI } from "../ui"
|
||||
import { db } from "../../storage/db"
|
||||
import { ProjectTable } from "../../project/project.sql"
|
||||
import { Project } from "../../project/project"
|
||||
import {
|
||||
SessionTable,
|
||||
MessageTable,
|
||||
PartTable,
|
||||
SessionDiffTable,
|
||||
TodoTable,
|
||||
PermissionTable,
|
||||
} from "../../session/session.sql"
|
||||
import { Session } from "../../session"
|
||||
import { SessionShareTable, ShareTable } from "../../share/share.sql"
|
||||
import path from "path"
|
||||
import fs from "fs/promises"
|
||||
|
||||
export const DatabaseCommand = cmd({
|
||||
command: "database",
|
||||
describe: "database management commands",
|
||||
builder: (yargs) => yargs.command(ExportCommand).demandCommand(),
|
||||
async handler() {},
|
||||
})
|
||||
|
||||
const ExportCommand = cmd({
|
||||
command: "export",
|
||||
describe: "export database to JSON files",
|
||||
builder: (yargs: Argv) => {
|
||||
return yargs.option("output", {
|
||||
alias: ["o"],
|
||||
describe: "output directory",
|
||||
type: "string",
|
||||
demandOption: true,
|
||||
})
|
||||
},
|
||||
handler: async (args) => {
|
||||
await bootstrap(process.cwd(), async () => {
|
||||
const outDir = path.resolve(args.output)
|
||||
await fs.mkdir(outDir, { recursive: true })
|
||||
|
||||
const stats = {
|
||||
projects: 0,
|
||||
sessions: 0,
|
||||
messages: 0,
|
||||
parts: 0,
|
||||
diffs: 0,
|
||||
todos: 0,
|
||||
permissions: 0,
|
||||
sessionShares: 0,
|
||||
shares: 0,
|
||||
}
|
||||
|
||||
// Export projects
|
||||
const projectDir = path.join(outDir, "project")
|
||||
await fs.mkdir(projectDir, { recursive: true })
|
||||
for (const row of db().select().from(ProjectTable).all()) {
|
||||
const project = Project.fromRow(row)
|
||||
await Bun.write(path.join(projectDir, `${row.id}.json`), JSON.stringify(project, null, 2))
|
||||
stats.projects++
|
||||
}
|
||||
|
||||
// Export sessions (organized by projectID)
|
||||
const sessionDir = path.join(outDir, "session")
|
||||
for (const row of db().select().from(SessionTable).all()) {
|
||||
const dir = path.join(sessionDir, row.projectID)
|
||||
await fs.mkdir(dir, { recursive: true })
|
||||
await Bun.write(path.join(dir, `${row.id}.json`), JSON.stringify(Session.fromRow(row), null, 2))
|
||||
stats.sessions++
|
||||
}
|
||||
|
||||
// Export messages (organized by sessionID)
|
||||
const messageDir = path.join(outDir, "message")
|
||||
for (const row of db().select().from(MessageTable).all()) {
|
||||
const dir = path.join(messageDir, row.sessionID)
|
||||
await fs.mkdir(dir, { recursive: true })
|
||||
await Bun.write(path.join(dir, `${row.id}.json`), JSON.stringify(row.data, null, 2))
|
||||
stats.messages++
|
||||
}
|
||||
|
||||
// Export parts (organized by messageID)
|
||||
const partDir = path.join(outDir, "part")
|
||||
for (const row of db().select().from(PartTable).all()) {
|
||||
const dir = path.join(partDir, row.messageID)
|
||||
await fs.mkdir(dir, { recursive: true })
|
||||
await Bun.write(path.join(dir, `${row.id}.json`), JSON.stringify(row.data, null, 2))
|
||||
stats.parts++
|
||||
}
|
||||
|
||||
// Export session diffs
|
||||
const diffDir = path.join(outDir, "session_diff")
|
||||
await fs.mkdir(diffDir, { recursive: true })
|
||||
for (const row of db().select().from(SessionDiffTable).all()) {
|
||||
await Bun.write(path.join(diffDir, `${row.sessionID}.json`), JSON.stringify(row.data, null, 2))
|
||||
stats.diffs++
|
||||
}
|
||||
|
||||
// Export todos
|
||||
const todoDir = path.join(outDir, "todo")
|
||||
await fs.mkdir(todoDir, { recursive: true })
|
||||
for (const row of db().select().from(TodoTable).all()) {
|
||||
await Bun.write(path.join(todoDir, `${row.sessionID}.json`), JSON.stringify(row.data, null, 2))
|
||||
stats.todos++
|
||||
}
|
||||
|
||||
// Export permissions
|
||||
const permDir = path.join(outDir, "permission")
|
||||
await fs.mkdir(permDir, { recursive: true })
|
||||
for (const row of db().select().from(PermissionTable).all()) {
|
||||
await Bun.write(path.join(permDir, `${row.projectID}.json`), JSON.stringify(row.data, null, 2))
|
||||
stats.permissions++
|
||||
}
|
||||
|
||||
// Export session shares
|
||||
const sessionShareDir = path.join(outDir, "session_share")
|
||||
await fs.mkdir(sessionShareDir, { recursive: true })
|
||||
for (const row of db().select().from(SessionShareTable).all()) {
|
||||
await Bun.write(path.join(sessionShareDir, `${row.sessionID}.json`), JSON.stringify(row.data, null, 2))
|
||||
stats.sessionShares++
|
||||
}
|
||||
|
||||
// Export shares
|
||||
const shareDir = path.join(outDir, "share")
|
||||
await fs.mkdir(shareDir, { recursive: true })
|
||||
for (const row of db().select().from(ShareTable).all()) {
|
||||
await Bun.write(path.join(shareDir, `${row.sessionID}.json`), JSON.stringify(row.data, null, 2))
|
||||
stats.shares++
|
||||
}
|
||||
|
||||
// Create migration marker so this can be imported back
|
||||
await Bun.write(path.join(outDir, "migration"), Date.now().toString())
|
||||
|
||||
UI.println(`Exported to ${outDir}:`)
|
||||
UI.println(` ${stats.projects} projects`)
|
||||
UI.println(` ${stats.sessions} sessions`)
|
||||
UI.println(` ${stats.messages} messages`)
|
||||
UI.println(` ${stats.parts} parts`)
|
||||
UI.println(` ${stats.diffs} session diffs`)
|
||||
UI.println(` ${stats.todos} todos`)
|
||||
UI.println(` ${stats.permissions} permissions`)
|
||||
UI.println(` ${stats.sessionShares} session shares`)
|
||||
UI.println(` ${stats.shares} shares`)
|
||||
})
|
||||
},
|
||||
})
|
||||
@@ -2,7 +2,8 @@ import type { Argv } from "yargs"
|
||||
import { Session } from "../../session"
|
||||
import { cmd } from "./cmd"
|
||||
import { bootstrap } from "../bootstrap"
|
||||
import { Storage } from "../../storage/storage"
|
||||
import { db } from "../../storage/db"
|
||||
import { SessionTable, MessageTable, PartTable } from "../../session/session.sql"
|
||||
import { Instance } from "../../project/instance"
|
||||
import { EOL } from "os"
|
||||
|
||||
@@ -81,13 +82,31 @@ export const ImportCommand = cmd({
|
||||
return
|
||||
}
|
||||
|
||||
await Storage.write(["session", Instance.project.id, exportData.info.id], exportData.info)
|
||||
db().insert(SessionTable).values(Session.toRow(exportData.info)).onConflictDoNothing().run()
|
||||
|
||||
for (const msg of exportData.messages) {
|
||||
await Storage.write(["message", exportData.info.id, msg.info.id], msg.info)
|
||||
db()
|
||||
.insert(MessageTable)
|
||||
.values({
|
||||
id: msg.info.id,
|
||||
sessionID: exportData.info.id,
|
||||
createdAt: msg.info.time?.created ?? Date.now(),
|
||||
data: msg.info,
|
||||
})
|
||||
.onConflictDoNothing()
|
||||
.run()
|
||||
|
||||
for (const part of msg.parts) {
|
||||
await Storage.write(["part", msg.info.id, part.id], part)
|
||||
db()
|
||||
.insert(PartTable)
|
||||
.values({
|
||||
id: part.id,
|
||||
messageID: msg.info.id,
|
||||
sessionID: exportData.info.id,
|
||||
data: part,
|
||||
})
|
||||
.onConflictDoNothing()
|
||||
.run()
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -2,7 +2,8 @@ import type { Argv } from "yargs"
|
||||
import { cmd } from "./cmd"
|
||||
import { Session } from "../../session"
|
||||
import { bootstrap } from "../bootstrap"
|
||||
import { Storage } from "../../storage/storage"
|
||||
import { db } from "../../storage/db"
|
||||
import { SessionTable } from "../../session/session.sql"
|
||||
import { Project } from "../../project/project"
|
||||
import { Instance } from "../../project/instance"
|
||||
|
||||
@@ -83,25 +84,8 @@ async function getCurrentProject(): Promise<Project.Info> {
|
||||
}
|
||||
|
||||
async function getAllSessions(): Promise<Session.Info[]> {
|
||||
const sessions: Session.Info[] = []
|
||||
|
||||
const projectKeys = await Storage.list(["project"])
|
||||
const projects = await Promise.all(projectKeys.map((key) => Storage.read<Project.Info>(key)))
|
||||
|
||||
for (const project of projects) {
|
||||
if (!project) continue
|
||||
|
||||
const sessionKeys = await Storage.list(["session", project.id])
|
||||
const projectSessions = await Promise.all(sessionKeys.map((key) => Storage.read<Session.Info>(key)))
|
||||
|
||||
for (const session of projectSessions) {
|
||||
if (session) {
|
||||
sessions.push(session)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return sessions
|
||||
const rows = db().select().from(SessionTable).all()
|
||||
return rows.map((row) => Session.fromRow(row))
|
||||
}
|
||||
|
||||
export async function aggregateSessionStats(days?: number, projectFilter?: string): Promise<SessionStats> {
|
||||
|
||||
@@ -26,6 +26,7 @@ import { EOL } from "os"
|
||||
import { WebCommand } from "./cli/cmd/web"
|
||||
import { PrCommand } from "./cli/cmd/pr"
|
||||
import { SessionCommand } from "./cli/cmd/session"
|
||||
import { DatabaseCommand } from "./cli/cmd/database"
|
||||
|
||||
process.on("unhandledRejection", (e) => {
|
||||
Log.Default.error("rejection", {
|
||||
@@ -97,6 +98,7 @@ const cli = yargs(hideBin(process.argv))
|
||||
.command(GithubCommand)
|
||||
.command(PrCommand)
|
||||
.command(SessionCommand)
|
||||
.command(DatabaseCommand)
|
||||
.fail((msg, err) => {
|
||||
if (
|
||||
msg?.startsWith("Unknown argument") ||
|
||||
|
||||
@@ -3,7 +3,9 @@ import { BusEvent } from "@/bus/bus-event"
|
||||
import { Config } from "@/config/config"
|
||||
import { Identifier } from "@/id/id"
|
||||
import { Instance } from "@/project/instance"
|
||||
import { Storage } from "@/storage/storage"
|
||||
import { db } from "@/storage/db"
|
||||
import { PermissionTable } from "@/session/session.sql"
|
||||
import { eq } from "drizzle-orm"
|
||||
import { fn } from "@/util/fn"
|
||||
import { Log } from "@/util/log"
|
||||
import { Wildcard } from "@/util/wildcard"
|
||||
@@ -105,9 +107,10 @@ export namespace PermissionNext {
|
||||
),
|
||||
}
|
||||
|
||||
const state = Instance.state(async () => {
|
||||
const state = Instance.state(() => {
|
||||
const projectID = Instance.project.id
|
||||
const stored = await Storage.read<Ruleset>(["permission", projectID]).catch(() => [] as Ruleset)
|
||||
const row = db().select().from(PermissionTable).where(eq(PermissionTable.projectID, projectID)).get()
|
||||
const stored = row?.data ?? ([] as Ruleset)
|
||||
|
||||
const pending: Record<
|
||||
string,
|
||||
@@ -222,7 +225,8 @@ export namespace PermissionNext {
|
||||
|
||||
// TODO: we don't save the permission ruleset to disk yet until there's
|
||||
// UI to manage it
|
||||
// await Storage.write(["permission", Instance.project.id], s.approved)
|
||||
// db().insert(PermissionTable).values({ projectID: Instance.project.id, data: s.approved })
|
||||
// .onConflictDoUpdate({ target: PermissionTable.projectID, set: { data: s.approved } }).run()
|
||||
return
|
||||
}
|
||||
},
|
||||
@@ -275,6 +279,7 @@ export namespace PermissionNext {
|
||||
}
|
||||
|
||||
export async function list() {
|
||||
return state().then((x) => Object.values(x.pending).map((x) => x.info))
|
||||
const s = await state()
|
||||
return Object.values(s.pending).map((x) => x.info)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,14 @@
|
||||
import { integer, sqliteTable, text } from "drizzle-orm/sqlite-core"
|
||||
|
||||
export const ProjectTable = sqliteTable("project", {
|
||||
id: text("id").primaryKey(),
|
||||
worktree: text("worktree").notNull(),
|
||||
vcs: text("vcs"),
|
||||
name: text("name"),
|
||||
icon_url: text("icon_url"),
|
||||
icon_color: text("icon_color"),
|
||||
time_created: integer("time_created").notNull(),
|
||||
time_updated: integer("time_updated").notNull(),
|
||||
time_initialized: integer("time_initialized"),
|
||||
sandboxes: text("sandboxes", { mode: "json" }).notNull().$type<string[]>(),
|
||||
})
|
||||
@@ -3,10 +3,12 @@ import fs from "fs/promises"
|
||||
import { Filesystem } from "../util/filesystem"
|
||||
import path from "path"
|
||||
import { $ } from "bun"
|
||||
import { Storage } from "../storage/storage"
|
||||
import { db } from "../storage/db"
|
||||
import { ProjectTable } from "./project.sql"
|
||||
import { SessionTable } from "../session/session.sql"
|
||||
import { eq } from "drizzle-orm"
|
||||
import { Log } from "../util/log"
|
||||
import { Flag } from "@/flag/flag"
|
||||
import { Session } from "../session"
|
||||
import { work } from "../util/queue"
|
||||
import { fn } from "@opencode-ai/util/fn"
|
||||
import { BusEvent } from "@/bus/bus-event"
|
||||
@@ -50,6 +52,28 @@ export namespace Project {
|
||||
Updated: BusEvent.define("project.updated", Info),
|
||||
}
|
||||
|
||||
type Row = typeof ProjectTable.$inferSelect
|
||||
|
||||
export function fromRow(row: Row): Info {
|
||||
const icon =
|
||||
row.icon_url || row.icon_color
|
||||
? { url: row.icon_url ?? undefined, color: row.icon_color ?? undefined }
|
||||
: undefined
|
||||
return {
|
||||
id: row.id,
|
||||
worktree: row.worktree,
|
||||
vcs: row.vcs as Info["vcs"],
|
||||
name: row.name ?? undefined,
|
||||
icon,
|
||||
time: {
|
||||
created: row.time_created,
|
||||
updated: row.time_updated,
|
||||
initialized: row.time_initialized ?? undefined,
|
||||
},
|
||||
sandboxes: row.sandboxes,
|
||||
}
|
||||
}
|
||||
|
||||
export async function fromDirectory(directory: string) {
|
||||
log.info("fromDirectory", { directory })
|
||||
|
||||
@@ -175,9 +199,10 @@ export namespace Project {
|
||||
}
|
||||
})
|
||||
|
||||
let existing = await Storage.read<Info>(["project", id]).catch(() => undefined)
|
||||
if (!existing) {
|
||||
existing = {
|
||||
const row = db().select().from(ProjectTable).where(eq(ProjectTable.id, id)).get()
|
||||
const existing = await iife(async () => {
|
||||
if (row) return fromRow(row)
|
||||
const fresh: Info = {
|
||||
id,
|
||||
worktree,
|
||||
vcs: vcs as Info["vcs"],
|
||||
@@ -190,10 +215,8 @@ export namespace Project {
|
||||
if (id !== "global") {
|
||||
await migrateFromGlobal(id, worktree)
|
||||
}
|
||||
}
|
||||
|
||||
// migrate old projects before sandboxes
|
||||
if (!existing.sandboxes) existing.sandboxes = []
|
||||
return fresh
|
||||
})
|
||||
|
||||
if (Flag.OPENCODE_EXPERIMENTAL_ICON_DISCOVERY) discover(existing)
|
||||
|
||||
@@ -208,7 +231,29 @@ export namespace Project {
|
||||
}
|
||||
if (sandbox !== result.worktree && !result.sandboxes.includes(sandbox)) result.sandboxes.push(sandbox)
|
||||
result.sandboxes = result.sandboxes.filter((x) => existsSync(x))
|
||||
await Storage.write<Info>(["project", id], result)
|
||||
const insert = {
|
||||
id: result.id,
|
||||
worktree: result.worktree,
|
||||
vcs: result.vcs,
|
||||
name: result.name,
|
||||
icon_url: result.icon?.url,
|
||||
icon_color: result.icon?.color,
|
||||
time_created: result.time.created,
|
||||
time_updated: result.time.updated,
|
||||
time_initialized: result.time.initialized,
|
||||
sandboxes: result.sandboxes,
|
||||
}
|
||||
const updateSet = {
|
||||
worktree: result.worktree,
|
||||
vcs: result.vcs,
|
||||
name: result.name,
|
||||
icon_url: result.icon?.url,
|
||||
icon_color: result.icon?.color,
|
||||
time_updated: result.time.updated,
|
||||
time_initialized: result.time.initialized,
|
||||
sandboxes: result.sandboxes,
|
||||
}
|
||||
db().insert(ProjectTable).values(insert).onConflictDoUpdate({ target: ProjectTable.id, set: updateSet }).run()
|
||||
GlobalBus.emit("event", {
|
||||
payload: {
|
||||
type: Event.Updated.type,
|
||||
@@ -249,42 +294,47 @@ export namespace Project {
|
||||
}
|
||||
|
||||
async function migrateFromGlobal(newProjectID: string, worktree: string) {
|
||||
const globalProject = await Storage.read<Info>(["project", "global"]).catch(() => undefined)
|
||||
if (!globalProject) return
|
||||
const globalRow = db().select().from(ProjectTable).where(eq(ProjectTable.id, "global")).get()
|
||||
if (!globalRow) return
|
||||
|
||||
const globalSessions = await Storage.list(["session", "global"]).catch(() => [])
|
||||
const globalSessions = db().select().from(SessionTable).where(eq(SessionTable.projectID, "global")).all()
|
||||
if (globalSessions.length === 0) return
|
||||
|
||||
log.info("migrating sessions from global", { newProjectID, worktree, count: globalSessions.length })
|
||||
|
||||
await work(10, globalSessions, async (key) => {
|
||||
const sessionID = key[key.length - 1]
|
||||
const session = await Storage.read<Session.Info>(key).catch(() => undefined)
|
||||
if (!session) return
|
||||
if (session.directory && session.directory !== worktree) return
|
||||
await work(10, globalSessions, async (row) => {
|
||||
// Skip sessions that belong to a different directory
|
||||
if (row.directory && row.directory !== worktree) return
|
||||
|
||||
session.projectID = newProjectID
|
||||
log.info("migrating session", { sessionID, from: "global", to: newProjectID })
|
||||
await Storage.write(["session", newProjectID, sessionID], session)
|
||||
await Storage.remove(key)
|
||||
log.info("migrating session", { sessionID: row.id, from: "global", to: newProjectID })
|
||||
db().update(SessionTable).set({ projectID: newProjectID }).where(eq(SessionTable.id, row.id)).run()
|
||||
}).catch((error) => {
|
||||
log.error("failed to migrate sessions from global to project", { error, projectId: newProjectID })
|
||||
})
|
||||
}
|
||||
|
||||
export async function setInitialized(projectID: string) {
|
||||
await Storage.update<Info>(["project", projectID], (draft) => {
|
||||
draft.time.initialized = Date.now()
|
||||
})
|
||||
export function setInitialized(projectID: string) {
|
||||
db()
|
||||
.update(ProjectTable)
|
||||
.set({
|
||||
time_initialized: Date.now(),
|
||||
})
|
||||
.where(eq(ProjectTable.id, projectID))
|
||||
.run()
|
||||
}
|
||||
|
||||
export async function list() {
|
||||
const keys = await Storage.list(["project"])
|
||||
const projects = await Promise.all(keys.map((x) => Storage.read<Info>(x)))
|
||||
return projects.map((project) => ({
|
||||
...project,
|
||||
sandboxes: project.sandboxes?.filter((x) => existsSync(x)),
|
||||
}))
|
||||
export function list() {
|
||||
return db()
|
||||
.select()
|
||||
.from(ProjectTable)
|
||||
.all()
|
||||
.map((row) => fromRow(row))
|
||||
}
|
||||
|
||||
export function get(projectID: string): Info | undefined {
|
||||
const row = db().select().from(ProjectTable).where(eq(ProjectTable.id, projectID)).get()
|
||||
if (!row) return undefined
|
||||
return fromRow(row)
|
||||
}
|
||||
|
||||
export const update = fn(
|
||||
@@ -295,43 +345,35 @@ export namespace Project {
|
||||
commands: Info.shape.commands.optional(),
|
||||
}),
|
||||
async (input) => {
|
||||
const result = await Storage.update<Info>(["project", input.projectID], (draft) => {
|
||||
if (input.name !== undefined) draft.name = input.name
|
||||
if (input.icon !== undefined) {
|
||||
draft.icon = {
|
||||
...draft.icon,
|
||||
}
|
||||
if (input.icon.url !== undefined) draft.icon.url = input.icon.url
|
||||
if (input.icon.override !== undefined) draft.icon.override = input.icon.override || undefined
|
||||
if (input.icon.color !== undefined) draft.icon.color = input.icon.color
|
||||
}
|
||||
|
||||
if (input.commands?.start !== undefined) {
|
||||
const start = input.commands.start || undefined
|
||||
draft.commands = {
|
||||
...(draft.commands ?? {}),
|
||||
}
|
||||
draft.commands.start = start
|
||||
if (!draft.commands.start) draft.commands = undefined
|
||||
}
|
||||
|
||||
draft.time.updated = Date.now()
|
||||
})
|
||||
const result = db()
|
||||
.update(ProjectTable)
|
||||
.set({
|
||||
name: input.name,
|
||||
icon_url: input.icon?.url,
|
||||
icon_color: input.icon?.color,
|
||||
time_updated: Date.now(),
|
||||
})
|
||||
.where(eq(ProjectTable.id, input.projectID))
|
||||
.returning()
|
||||
.get()
|
||||
if (!result) throw new Error(`Project not found: ${input.projectID}`)
|
||||
const data = fromRow(result)
|
||||
GlobalBus.emit("event", {
|
||||
payload: {
|
||||
type: Event.Updated.type,
|
||||
properties: result,
|
||||
properties: data,
|
||||
},
|
||||
})
|
||||
return result
|
||||
return data
|
||||
},
|
||||
)
|
||||
|
||||
export async function sandboxes(projectID: string) {
|
||||
const project = await Storage.read<Info>(["project", projectID]).catch(() => undefined)
|
||||
if (!project?.sandboxes) return []
|
||||
const row = db().select().from(ProjectTable).where(eq(ProjectTable.id, projectID)).get()
|
||||
if (!row) return []
|
||||
const data = fromRow(row)
|
||||
const valid: string[] = []
|
||||
for (const dir of project.sandboxes) {
|
||||
for (const dir of data.sandboxes) {
|
||||
const stat = await fs.stat(dir).catch(() => undefined)
|
||||
if (stat?.isDirectory()) valid.push(dir)
|
||||
}
|
||||
@@ -339,33 +381,45 @@ export namespace Project {
|
||||
}
|
||||
|
||||
export async function addSandbox(projectID: string, directory: string) {
|
||||
const result = await Storage.update<Info>(["project", projectID], (draft) => {
|
||||
const sandboxes = draft.sandboxes ?? []
|
||||
if (!sandboxes.includes(directory)) sandboxes.push(directory)
|
||||
draft.sandboxes = sandboxes
|
||||
draft.time.updated = Date.now()
|
||||
})
|
||||
const row = db().select().from(ProjectTable).where(eq(ProjectTable.id, projectID)).get()
|
||||
if (!row) throw new Error(`Project not found: ${projectID}`)
|
||||
const sandboxes = [...row.sandboxes]
|
||||
if (!sandboxes.includes(directory)) sandboxes.push(directory)
|
||||
const result = db()
|
||||
.update(ProjectTable)
|
||||
.set({ sandboxes, time_updated: Date.now() })
|
||||
.where(eq(ProjectTable.id, projectID))
|
||||
.returning()
|
||||
.get()
|
||||
if (!result) throw new Error(`Project not found: ${projectID}`)
|
||||
const data = fromRow(result)
|
||||
GlobalBus.emit("event", {
|
||||
payload: {
|
||||
type: Event.Updated.type,
|
||||
properties: result,
|
||||
properties: data,
|
||||
},
|
||||
})
|
||||
return result
|
||||
return data
|
||||
}
|
||||
|
||||
export async function removeSandbox(projectID: string, directory: string) {
|
||||
const result = await Storage.update<Info>(["project", projectID], (draft) => {
|
||||
const sandboxes = draft.sandboxes ?? []
|
||||
draft.sandboxes = sandboxes.filter((sandbox) => sandbox !== directory)
|
||||
draft.time.updated = Date.now()
|
||||
})
|
||||
const row = db().select().from(ProjectTable).where(eq(ProjectTable.id, projectID)).get()
|
||||
if (!row) throw new Error(`Project not found: ${projectID}`)
|
||||
const sandboxes = row.sandboxes.filter((s: string) => s !== directory)
|
||||
const result = db()
|
||||
.update(ProjectTable)
|
||||
.set({ sandboxes, time_updated: Date.now() })
|
||||
.where(eq(ProjectTable.id, projectID))
|
||||
.returning()
|
||||
.get()
|
||||
if (!result) throw new Error(`Project not found: ${projectID}`)
|
||||
const data = fromRow(result)
|
||||
GlobalBus.emit("event", {
|
||||
payload: {
|
||||
type: Event.Updated.type,
|
||||
properties: result,
|
||||
properties: data,
|
||||
},
|
||||
})
|
||||
return result
|
||||
return data
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
import { resolver } from "hono-openapi"
|
||||
import z from "zod"
|
||||
import { Storage } from "../storage/storage"
|
||||
import { NotFoundError } from "../storage/db"
|
||||
|
||||
export const ERRORS = {
|
||||
400: {
|
||||
@@ -25,7 +25,7 @@ export const ERRORS = {
|
||||
description: "Not found",
|
||||
content: {
|
||||
"application/json": {
|
||||
schema: resolver(Storage.NotFoundError.Schema),
|
||||
schema: resolver(NotFoundError.Schema),
|
||||
},
|
||||
},
|
||||
},
|
||||
|
||||
@@ -3,7 +3,7 @@ import { describeRoute, validator, resolver } from "hono-openapi"
|
||||
import { upgradeWebSocket } from "hono/bun"
|
||||
import z from "zod"
|
||||
import { Pty } from "@/pty"
|
||||
import { Storage } from "../../storage/storage"
|
||||
import { NotFoundError } from "../../storage/db"
|
||||
import { errors } from "../error"
|
||||
import { lazy } from "../../util/lazy"
|
||||
|
||||
@@ -76,7 +76,7 @@ export const PtyRoutes = lazy(() =>
|
||||
async (c) => {
|
||||
const info = Pty.get(c.req.valid("param").ptyID)
|
||||
if (!info) {
|
||||
throw new Storage.NotFoundError({ message: "Session not found" })
|
||||
throw new NotFoundError({ message: "Session not found" })
|
||||
}
|
||||
return c.json(info)
|
||||
},
|
||||
|
||||
@@ -31,7 +31,7 @@ import { ExperimentalRoutes } from "./routes/experimental"
|
||||
import { ProviderRoutes } from "./routes/provider"
|
||||
import { lazy } from "../util/lazy"
|
||||
import { InstanceBootstrap } from "../project/bootstrap"
|
||||
import { Storage } from "../storage/storage"
|
||||
import { NotFoundError } from "../storage/db"
|
||||
import type { ContentfulStatusCode } from "hono/utils/http-status"
|
||||
import { websocket } from "hono/bun"
|
||||
import { HTTPException } from "hono/http-exception"
|
||||
@@ -65,7 +65,7 @@ export namespace Server {
|
||||
})
|
||||
if (err instanceof NamedError) {
|
||||
let status: ContentfulStatusCode
|
||||
if (err instanceof Storage.NotFoundError) status = 404
|
||||
if (err instanceof NotFoundError) status = 404
|
||||
else if (err instanceof Provider.ModelNotFoundError) status = 400
|
||||
else if (err.name.startsWith("Worktree")) status = 400
|
||||
else status = 500
|
||||
|
||||
@@ -10,7 +10,10 @@ import { Flag } from "../flag/flag"
|
||||
import { Identifier } from "../id/id"
|
||||
import { Installation } from "../installation"
|
||||
|
||||
import { Storage } from "../storage/storage"
|
||||
import { db, NotFoundError } from "../storage/db"
|
||||
import { SessionTable, MessageTable, PartTable, SessionDiffTable } from "./session.sql"
|
||||
import { ShareTable } from "../share/share.sql"
|
||||
import { eq } from "drizzle-orm"
|
||||
import { Log } from "../util/log"
|
||||
import { MessageV2 } from "./message-v2"
|
||||
import { Instance } from "../project/instance"
|
||||
@@ -39,6 +42,75 @@ export namespace Session {
|
||||
).test(title)
|
||||
}
|
||||
|
||||
type SessionRow = typeof SessionTable.$inferSelect
|
||||
|
||||
export function fromRow(row: SessionRow): Info {
|
||||
const summary =
|
||||
row.summary_additions !== null || row.summary_deletions !== null || row.summary_files !== null
|
||||
? {
|
||||
additions: row.summary_additions ?? 0,
|
||||
deletions: row.summary_deletions ?? 0,
|
||||
files: row.summary_files ?? 0,
|
||||
diffs: row.summary_diffs ?? undefined,
|
||||
}
|
||||
: undefined
|
||||
const share = row.share_url ? { url: row.share_url } : undefined
|
||||
const revert =
|
||||
row.revert_messageID !== null
|
||||
? {
|
||||
messageID: row.revert_messageID,
|
||||
partID: row.revert_partID ?? undefined,
|
||||
snapshot: row.revert_snapshot ?? undefined,
|
||||
diff: row.revert_diff ?? undefined,
|
||||
}
|
||||
: undefined
|
||||
return {
|
||||
id: row.id,
|
||||
slug: row.slug,
|
||||
projectID: row.projectID,
|
||||
directory: row.directory,
|
||||
parentID: row.parentID ?? undefined,
|
||||
title: row.title,
|
||||
version: row.version,
|
||||
summary,
|
||||
share,
|
||||
revert,
|
||||
permission: row.permission ?? undefined,
|
||||
time: {
|
||||
created: row.time_created,
|
||||
updated: row.time_updated,
|
||||
compacting: row.time_compacting ?? undefined,
|
||||
archived: row.time_archived ?? undefined,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
export function toRow(info: Info) {
|
||||
return {
|
||||
id: info.id,
|
||||
projectID: info.projectID,
|
||||
parentID: info.parentID,
|
||||
slug: info.slug,
|
||||
directory: info.directory,
|
||||
title: info.title,
|
||||
version: info.version,
|
||||
share_url: info.share?.url,
|
||||
summary_additions: info.summary?.additions,
|
||||
summary_deletions: info.summary?.deletions,
|
||||
summary_files: info.summary?.files,
|
||||
summary_diffs: info.summary?.diffs,
|
||||
revert_messageID: info.revert?.messageID ?? null,
|
||||
revert_partID: info.revert?.partID ?? null,
|
||||
revert_snapshot: info.revert?.snapshot ?? null,
|
||||
revert_diff: info.revert?.diff ?? null,
|
||||
permission: info.permission,
|
||||
time_created: info.time.created,
|
||||
time_updated: info.time.updated,
|
||||
time_compacting: info.time.compacting,
|
||||
time_archived: info.time.archived,
|
||||
}
|
||||
}
|
||||
|
||||
export const Info = z
|
||||
.object({
|
||||
id: Identifier.schema("session"),
|
||||
@@ -211,7 +283,7 @@ export namespace Session {
|
||||
},
|
||||
}
|
||||
log.info("created", result)
|
||||
await Storage.write(["session", Instance.project.id, result.id], result)
|
||||
db().insert(SessionTable).values(toRow(result)).run()
|
||||
Bus.publish(Event.Created, {
|
||||
info: result,
|
||||
})
|
||||
@@ -240,12 +312,14 @@ export namespace Session {
|
||||
}
|
||||
|
||||
export const get = fn(Identifier.schema("session"), async (id) => {
|
||||
const read = await Storage.read<Info>(["session", Instance.project.id, id])
|
||||
return read as Info
|
||||
const row = db().select().from(SessionTable).where(eq(SessionTable.id, id)).get()
|
||||
if (!row) throw new NotFoundError({ message: `Session not found: ${id}` })
|
||||
return fromRow(row)
|
||||
})
|
||||
|
||||
export const getShare = fn(Identifier.schema("session"), async (id) => {
|
||||
return Storage.read<ShareInfo>(["share", id])
|
||||
const row = db().select().from(ShareTable).where(eq(ShareTable.sessionID, id)).get()
|
||||
return row?.data
|
||||
})
|
||||
|
||||
export const share = fn(Identifier.schema("session"), async (id) => {
|
||||
@@ -280,23 +354,24 @@ export namespace Session {
|
||||
)
|
||||
})
|
||||
|
||||
export async function update(id: string, editor: (session: Info) => void, options?: { touch?: boolean }) {
|
||||
const project = Instance.project
|
||||
const result = await Storage.update<Info>(["session", project.id, id], (draft) => {
|
||||
editor(draft)
|
||||
if (options?.touch !== false) {
|
||||
draft.time.updated = Date.now()
|
||||
}
|
||||
})
|
||||
export function update(id: string, editor: (session: Info) => void, options?: { touch?: boolean }) {
|
||||
const row = db().select().from(SessionTable).where(eq(SessionTable.id, id)).get()
|
||||
if (!row) throw new Error(`Session not found: ${id}`)
|
||||
const data = fromRow(row)
|
||||
editor(data)
|
||||
if (options?.touch !== false) {
|
||||
data.time.updated = Date.now()
|
||||
}
|
||||
db().update(SessionTable).set(toRow(data)).where(eq(SessionTable.id, id)).run()
|
||||
Bus.publish(Event.Updated, {
|
||||
info: result,
|
||||
info: data,
|
||||
})
|
||||
return result
|
||||
return data
|
||||
}
|
||||
|
||||
export const diff = fn(Identifier.schema("session"), async (sessionID) => {
|
||||
const diffs = await Storage.read<Snapshot.FileDiff[]>(["session_diff", sessionID])
|
||||
return diffs ?? []
|
||||
const row = db().select().from(SessionDiffTable).where(eq(SessionDiffTable.sessionID, sessionID)).get()
|
||||
return row?.data ?? []
|
||||
})
|
||||
|
||||
export const messages = fn(
|
||||
@@ -315,22 +390,17 @@ export namespace Session {
|
||||
},
|
||||
)
|
||||
|
||||
export async function* list() {
|
||||
export function* list() {
|
||||
const project = Instance.project
|
||||
for (const item of await Storage.list(["session", project.id])) {
|
||||
yield Storage.read<Info>(item)
|
||||
const rows = db().select().from(SessionTable).where(eq(SessionTable.projectID, project.id)).all()
|
||||
for (const row of rows) {
|
||||
yield fromRow(row)
|
||||
}
|
||||
}
|
||||
|
||||
export const children = fn(Identifier.schema("session"), async (parentID) => {
|
||||
const project = Instance.project
|
||||
const result = [] as Session.Info[]
|
||||
for (const item of await Storage.list(["session", project.id])) {
|
||||
const session = await Storage.read<Info>(item)
|
||||
if (session.parentID !== parentID) continue
|
||||
result.push(session)
|
||||
}
|
||||
return result
|
||||
const rows = db().select().from(SessionTable).where(eq(SessionTable.parentID, parentID)).all()
|
||||
return rows.map((row) => fromRow(row))
|
||||
})
|
||||
|
||||
export const remove = fn(Identifier.schema("session"), async (sessionID) => {
|
||||
@@ -341,13 +411,8 @@ export namespace Session {
|
||||
await remove(child.id)
|
||||
}
|
||||
await unshare(sessionID).catch(() => {})
|
||||
for (const msg of await Storage.list(["message", sessionID])) {
|
||||
for (const part of await Storage.list(["part", msg.at(-1)!])) {
|
||||
await Storage.remove(part)
|
||||
}
|
||||
await Storage.remove(msg)
|
||||
}
|
||||
await Storage.remove(["session", project.id, sessionID])
|
||||
// CASCADE delete handles messages and parts automatically
|
||||
db().delete(SessionTable).where(eq(SessionTable.id, sessionID)).run()
|
||||
Bus.publish(Event.Deleted, {
|
||||
info: session,
|
||||
})
|
||||
@@ -357,7 +422,17 @@ export namespace Session {
|
||||
})
|
||||
|
||||
export const updateMessage = fn(MessageV2.Info, async (msg) => {
|
||||
await Storage.write(["message", msg.sessionID, msg.id], msg)
|
||||
const createdAt = msg.role === "user" ? msg.time.created : msg.time.created
|
||||
db()
|
||||
.insert(MessageTable)
|
||||
.values({
|
||||
id: msg.id,
|
||||
sessionID: msg.sessionID,
|
||||
createdAt,
|
||||
data: msg,
|
||||
})
|
||||
.onConflictDoUpdate({ target: MessageTable.id, set: { data: msg } })
|
||||
.run()
|
||||
Bus.publish(MessageV2.Event.Updated, {
|
||||
info: msg,
|
||||
})
|
||||
@@ -370,7 +445,8 @@ export namespace Session {
|
||||
messageID: Identifier.schema("message"),
|
||||
}),
|
||||
async (input) => {
|
||||
await Storage.remove(["message", input.sessionID, input.messageID])
|
||||
// CASCADE delete handles parts automatically
|
||||
db().delete(MessageTable).where(eq(MessageTable.id, input.messageID)).run()
|
||||
Bus.publish(MessageV2.Event.Removed, {
|
||||
sessionID: input.sessionID,
|
||||
messageID: input.messageID,
|
||||
@@ -386,7 +462,7 @@ export namespace Session {
|
||||
partID: Identifier.schema("part"),
|
||||
}),
|
||||
async (input) => {
|
||||
await Storage.remove(["part", input.messageID, input.partID])
|
||||
db().delete(PartTable).where(eq(PartTable.id, input.partID)).run()
|
||||
Bus.publish(MessageV2.Event.PartRemoved, {
|
||||
sessionID: input.sessionID,
|
||||
messageID: input.messageID,
|
||||
@@ -411,7 +487,16 @@ export namespace Session {
|
||||
export const updatePart = fn(UpdatePartInput, async (input) => {
|
||||
const part = "delta" in input ? input.part : input
|
||||
const delta = "delta" in input ? input.delta : undefined
|
||||
await Storage.write(["part", part.messageID, part.id], part)
|
||||
db()
|
||||
.insert(PartTable)
|
||||
.values({
|
||||
id: part.id,
|
||||
messageID: part.messageID,
|
||||
sessionID: part.sessionID,
|
||||
data: part,
|
||||
})
|
||||
.onConflictDoUpdate({ target: PartTable.id, set: { data: part } })
|
||||
.run()
|
||||
Bus.publish(MessageV2.Event.PartUpdated, {
|
||||
part,
|
||||
delta,
|
||||
|
||||
@@ -6,7 +6,9 @@ import { Identifier } from "../id/id"
|
||||
import { LSP } from "../lsp"
|
||||
import { Snapshot } from "@/snapshot"
|
||||
import { fn } from "@/util/fn"
|
||||
import { Storage } from "@/storage/storage"
|
||||
import { db } from "@/storage/db"
|
||||
import { MessageTable, PartTable } from "./session.sql"
|
||||
import { eq, desc } from "drizzle-orm"
|
||||
import { ProviderTransform } from "@/provider/transform"
|
||||
import { STATUS_CODES } from "http"
|
||||
import { iife } from "@/util/iife"
|
||||
@@ -607,21 +609,23 @@ export namespace MessageV2 {
|
||||
}
|
||||
|
||||
export const stream = fn(Identifier.schema("session"), async function* (sessionID) {
|
||||
const list = await Array.fromAsync(await Storage.list(["message", sessionID]))
|
||||
for (let i = list.length - 1; i >= 0; i--) {
|
||||
yield await get({
|
||||
sessionID,
|
||||
messageID: list[i][2],
|
||||
})
|
||||
const rows = db()
|
||||
.select()
|
||||
.from(MessageTable)
|
||||
.where(eq(MessageTable.sessionID, sessionID))
|
||||
.orderBy(desc(MessageTable.createdAt))
|
||||
.all()
|
||||
for (const row of rows) {
|
||||
yield {
|
||||
info: row.data,
|
||||
parts: await parts(row.id),
|
||||
}
|
||||
}
|
||||
})
|
||||
|
||||
export const parts = fn(Identifier.schema("message"), async (messageID) => {
|
||||
const result = [] as MessageV2.Part[]
|
||||
for (const item of await Storage.list(["part", messageID])) {
|
||||
const read = await Storage.read<MessageV2.Part>(item)
|
||||
result.push(read)
|
||||
}
|
||||
const rows = db().select().from(PartTable).where(eq(PartTable.messageID, messageID)).all()
|
||||
const result = rows.map((row) => row.data)
|
||||
result.sort((a, b) => (a.id > b.id ? 1 : -1))
|
||||
return result
|
||||
})
|
||||
@@ -632,8 +636,10 @@ export namespace MessageV2 {
|
||||
messageID: Identifier.schema("message"),
|
||||
}),
|
||||
async (input) => {
|
||||
const row = db().select().from(MessageTable).where(eq(MessageTable.id, input.messageID)).get()
|
||||
if (!row) throw new Error(`Message not found: ${input.messageID}`)
|
||||
return {
|
||||
info: await Storage.read<MessageV2.Info>(["message", input.sessionID, input.messageID]),
|
||||
info: row.data,
|
||||
parts: await parts(input.messageID),
|
||||
}
|
||||
},
|
||||
|
||||
@@ -5,7 +5,9 @@ import { MessageV2 } from "./message-v2"
|
||||
import { Session } from "."
|
||||
import { Log } from "../util/log"
|
||||
import { splitWhen } from "remeda"
|
||||
import { Storage } from "../storage/storage"
|
||||
import { db } from "../storage/db"
|
||||
import { SessionDiffTable, MessageTable, PartTable } from "./session.sql"
|
||||
import { eq } from "drizzle-orm"
|
||||
import { Bus } from "../bus"
|
||||
import { SessionPrompt } from "./prompt"
|
||||
import { SessionSummary } from "./summary"
|
||||
@@ -60,7 +62,11 @@ export namespace SessionRevert {
|
||||
if (revert.snapshot) revert.diff = await Snapshot.diff(revert.snapshot)
|
||||
const rangeMessages = all.filter((msg) => msg.info.id >= revert!.messageID)
|
||||
const diffs = await SessionSummary.computeDiff({ messages: rangeMessages })
|
||||
await Storage.write(["session_diff", input.sessionID], diffs)
|
||||
db()
|
||||
.insert(SessionDiffTable)
|
||||
.values({ sessionID: input.sessionID, data: diffs })
|
||||
.onConflictDoUpdate({ target: SessionDiffTable.sessionID, set: { data: diffs } })
|
||||
.run()
|
||||
Bus.publish(Session.Event.Diff, {
|
||||
sessionID: input.sessionID,
|
||||
diff: diffs,
|
||||
@@ -97,7 +103,7 @@ export namespace SessionRevert {
|
||||
const [preserve, remove] = splitWhen(msgs, (x) => x.info.id === messageID)
|
||||
msgs = preserve
|
||||
for (const msg of remove) {
|
||||
await Storage.remove(["message", sessionID, msg.info.id])
|
||||
db().delete(MessageTable).where(eq(MessageTable.id, msg.info.id)).run()
|
||||
await Bus.publish(MessageV2.Event.Removed, { sessionID: sessionID, messageID: msg.info.id })
|
||||
}
|
||||
const last = preserve.at(-1)
|
||||
@@ -106,7 +112,7 @@ export namespace SessionRevert {
|
||||
const [preserveParts, removeParts] = splitWhen(last.parts, (x) => x.id === partID)
|
||||
last.parts = preserveParts
|
||||
for (const part of removeParts) {
|
||||
await Storage.remove(["part", last.info.id, part.id])
|
||||
db().delete(PartTable).where(eq(PartTable.id, part.id)).run()
|
||||
await Bus.publish(MessageV2.Event.PartRemoved, {
|
||||
sessionID: sessionID,
|
||||
messageID: last.info.id,
|
||||
|
||||
@@ -0,0 +1,83 @@
|
||||
import { sqliteTable, text, integer, index } from "drizzle-orm/sqlite-core"
|
||||
import { ProjectTable } from "../project/project.sql"
|
||||
import type { MessageV2 } from "./message-v2"
|
||||
import type { Snapshot } from "@/snapshot"
|
||||
import type { Todo } from "./todo"
|
||||
import type { PermissionNext } from "@/permission/next"
|
||||
|
||||
export const SessionTable = sqliteTable(
|
||||
"session",
|
||||
{
|
||||
id: text("id").primaryKey(),
|
||||
projectID: text("project_id")
|
||||
.notNull()
|
||||
.references(() => ProjectTable.id, { onDelete: "cascade" }),
|
||||
parentID: text("parent_id"),
|
||||
slug: text("slug").notNull(),
|
||||
directory: text("directory").notNull(),
|
||||
title: text("title").notNull(),
|
||||
version: text("version").notNull(),
|
||||
share_url: text("share_url"),
|
||||
summary_additions: integer("summary_additions"),
|
||||
summary_deletions: integer("summary_deletions"),
|
||||
summary_files: integer("summary_files"),
|
||||
summary_diffs: text("summary_diffs", { mode: "json" }).$type<Snapshot.FileDiff[]>(),
|
||||
revert_messageID: text("revert_message_id"),
|
||||
revert_partID: text("revert_part_id"),
|
||||
revert_snapshot: text("revert_snapshot"),
|
||||
revert_diff: text("revert_diff"),
|
||||
permission: text("permission", { mode: "json" }).$type<PermissionNext.Ruleset>(),
|
||||
time_created: integer("time_created").notNull(),
|
||||
time_updated: integer("time_updated").notNull(),
|
||||
time_compacting: integer("time_compacting"),
|
||||
time_archived: integer("time_archived"),
|
||||
},
|
||||
(table) => [index("session_project_idx").on(table.projectID), index("session_parent_idx").on(table.parentID)],
|
||||
)
|
||||
|
||||
export const MessageTable = sqliteTable(
|
||||
"message",
|
||||
{
|
||||
id: text("id").primaryKey(),
|
||||
sessionID: text("session_id")
|
||||
.notNull()
|
||||
.references(() => SessionTable.id, { onDelete: "cascade" }),
|
||||
createdAt: integer("created_at").notNull(),
|
||||
data: text("data", { mode: "json" }).notNull().$type<MessageV2.Info>(),
|
||||
},
|
||||
(table) => [index("message_session_idx").on(table.sessionID)],
|
||||
)
|
||||
|
||||
export const PartTable = sqliteTable(
|
||||
"part",
|
||||
{
|
||||
id: text("id").primaryKey(),
|
||||
messageID: text("message_id")
|
||||
.notNull()
|
||||
.references(() => MessageTable.id, { onDelete: "cascade" }),
|
||||
sessionID: text("session_id").notNull(),
|
||||
data: text("data", { mode: "json" }).notNull().$type<MessageV2.Part>(),
|
||||
},
|
||||
(table) => [index("part_message_idx").on(table.messageID), index("part_session_idx").on(table.sessionID)],
|
||||
)
|
||||
|
||||
export const SessionDiffTable = sqliteTable("session_diff", {
|
||||
sessionID: text("session_id")
|
||||
.primaryKey()
|
||||
.references(() => SessionTable.id, { onDelete: "cascade" }),
|
||||
data: text("data", { mode: "json" }).notNull().$type<Snapshot.FileDiff[]>(),
|
||||
})
|
||||
|
||||
export const TodoTable = sqliteTable("todo", {
|
||||
sessionID: text("session_id")
|
||||
.primaryKey()
|
||||
.references(() => SessionTable.id, { onDelete: "cascade" }),
|
||||
data: text("data", { mode: "json" }).notNull().$type<Todo.Info[]>(),
|
||||
})
|
||||
|
||||
export const PermissionTable = sqliteTable("permission", {
|
||||
projectID: text("project_id")
|
||||
.primaryKey()
|
||||
.references(() => ProjectTable.id, { onDelete: "cascade" }),
|
||||
data: text("data", { mode: "json" }).notNull().$type<PermissionNext.Ruleset>(),
|
||||
})
|
||||
@@ -11,7 +11,9 @@ import { Snapshot } from "@/snapshot"
|
||||
import { Log } from "@/util/log"
|
||||
import path from "path"
|
||||
import { Instance } from "@/project/instance"
|
||||
import { Storage } from "@/storage/storage"
|
||||
import { db } from "@/storage/db"
|
||||
import { SessionDiffTable } from "./session.sql"
|
||||
import { eq } from "drizzle-orm"
|
||||
import { Bus } from "@/bus"
|
||||
|
||||
import { LLM } from "./llm"
|
||||
@@ -54,7 +56,11 @@ export namespace SessionSummary {
|
||||
files: diffs.length,
|
||||
}
|
||||
})
|
||||
await Storage.write(["session_diff", input.sessionID], diffs)
|
||||
db()
|
||||
.insert(SessionDiffTable)
|
||||
.values({ sessionID: input.sessionID, data: diffs })
|
||||
.onConflictDoUpdate({ target: SessionDiffTable.sessionID, set: { data: diffs } })
|
||||
.run()
|
||||
Bus.publish(Session.Event.Diff, {
|
||||
sessionID: input.sessionID,
|
||||
diff: diffs,
|
||||
@@ -116,7 +122,8 @@ export namespace SessionSummary {
|
||||
messageID: Identifier.schema("message").optional(),
|
||||
}),
|
||||
async (input) => {
|
||||
return Storage.read<Snapshot.FileDiff[]>(["session_diff", input.sessionID]).catch(() => [])
|
||||
const row = db().select().from(SessionDiffTable).where(eq(SessionDiffTable.sessionID, input.sessionID)).get()
|
||||
return row?.data ?? []
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
@@ -1,7 +1,9 @@
|
||||
import { BusEvent } from "@/bus/bus-event"
|
||||
import { Bus } from "@/bus"
|
||||
import z from "zod"
|
||||
import { Storage } from "../storage/storage"
|
||||
import { db } from "../storage/db"
|
||||
import { TodoTable } from "./session.sql"
|
||||
import { eq } from "drizzle-orm"
|
||||
|
||||
export namespace Todo {
|
||||
export const Info = z
|
||||
@@ -24,14 +26,17 @@ export namespace Todo {
|
||||
),
|
||||
}
|
||||
|
||||
export async function update(input: { sessionID: string; todos: Info[] }) {
|
||||
await Storage.write(["todo", input.sessionID], input.todos)
|
||||
export function update(input: { sessionID: string; todos: Info[] }) {
|
||||
db()
|
||||
.insert(TodoTable)
|
||||
.values({ sessionID: input.sessionID, data: input.todos })
|
||||
.onConflictDoUpdate({ target: TodoTable.sessionID, set: { data: input.todos } })
|
||||
.run()
|
||||
Bus.publish(Event.Updated, input)
|
||||
}
|
||||
|
||||
export async function get(sessionID: string) {
|
||||
return Storage.read<Info[]>(["todo", sessionID])
|
||||
.then((x) => x || [])
|
||||
.catch(() => [])
|
||||
export function get(sessionID: string) {
|
||||
const row = db().select().from(TodoTable).where(eq(TodoTable.sessionID, sessionID)).get()
|
||||
return row?.data ?? []
|
||||
}
|
||||
}
|
||||
|
||||
@@ -4,7 +4,9 @@ import { ulid } from "ulid"
|
||||
import { Provider } from "@/provider/provider"
|
||||
import { Session } from "@/session"
|
||||
import { MessageV2 } from "@/session/message-v2"
|
||||
import { Storage } from "@/storage/storage"
|
||||
import { db } from "@/storage/db"
|
||||
import { SessionShareTable } from "./share.sql"
|
||||
import { eq } from "drizzle-orm"
|
||||
import { Log } from "@/util/log"
|
||||
import type * as SDK from "@opencode-ai/sdk/v2"
|
||||
|
||||
@@ -77,17 +79,18 @@ export namespace ShareNext {
|
||||
})
|
||||
.then((x) => x.json())
|
||||
.then((x) => x as { id: string; url: string; secret: string })
|
||||
await Storage.write(["session_share", sessionID], result)
|
||||
db()
|
||||
.insert(SessionShareTable)
|
||||
.values({ sessionID, data: result })
|
||||
.onConflictDoUpdate({ target: SessionShareTable.sessionID, set: { data: result } })
|
||||
.run()
|
||||
fullSync(sessionID)
|
||||
return result
|
||||
}
|
||||
|
||||
function get(sessionID: string) {
|
||||
return Storage.read<{
|
||||
id: string
|
||||
secret: string
|
||||
url: string
|
||||
}>(["session_share", sessionID])
|
||||
const row = db().select().from(SessionShareTable).where(eq(SessionShareTable.sessionID, sessionID)).get()
|
||||
return row?.data
|
||||
}
|
||||
|
||||
type Data =
|
||||
@@ -132,7 +135,7 @@ export namespace ShareNext {
|
||||
const queued = queue.get(sessionID)
|
||||
if (!queued) return
|
||||
queue.delete(sessionID)
|
||||
const share = await get(sessionID).catch(() => undefined)
|
||||
const share = get(sessionID)
|
||||
if (!share) return
|
||||
|
||||
await fetch(`${await url()}/api/share/${share.id}/sync`, {
|
||||
@@ -152,7 +155,7 @@ export namespace ShareNext {
|
||||
export async function remove(sessionID: string) {
|
||||
if (disabled) return
|
||||
log.info("removing share", { sessionID })
|
||||
const share = await get(sessionID)
|
||||
const share = get(sessionID)
|
||||
if (!share) return
|
||||
await fetch(`${await url()}/api/share/${share.id}`, {
|
||||
method: "DELETE",
|
||||
@@ -163,7 +166,7 @@ export namespace ShareNext {
|
||||
secret: share.secret,
|
||||
}),
|
||||
})
|
||||
await Storage.remove(["session_share", sessionID])
|
||||
db().delete(SessionShareTable).where(eq(SessionShareTable.sessionID, sessionID)).run()
|
||||
}
|
||||
|
||||
async function fullSync(sessionID: string) {
|
||||
|
||||
@@ -0,0 +1,19 @@
|
||||
import { sqliteTable, text } from "drizzle-orm/sqlite-core"
|
||||
import { SessionTable } from "../session/session.sql"
|
||||
import type { Session } from "../session"
|
||||
|
||||
export const SessionShareTable = sqliteTable("session_share", {
|
||||
sessionID: text("session_id")
|
||||
.primaryKey()
|
||||
.references(() => SessionTable.id, { onDelete: "cascade" }),
|
||||
data: text("data", { mode: "json" }).notNull().$type<{
|
||||
id: string
|
||||
secret: string
|
||||
url: string
|
||||
}>(),
|
||||
})
|
||||
|
||||
export const ShareTable = sqliteTable("share", {
|
||||
sessionID: text("session_id").primaryKey(),
|
||||
data: text("data", { mode: "json" }).notNull().$type<Session.ShareInfo>(),
|
||||
})
|
||||
Vendored
+4
@@ -0,0 +1,4 @@
|
||||
declare module "*.sql" {
|
||||
const content: string
|
||||
export default content
|
||||
}
|
||||
@@ -0,0 +1,87 @@
|
||||
import { Database } from "bun:sqlite"
|
||||
import { drizzle } from "drizzle-orm/bun-sqlite"
|
||||
import { lazy } from "../util/lazy"
|
||||
import { Global } from "../global"
|
||||
import { Log } from "../util/log"
|
||||
import { migrations } from "./migrations.generated"
|
||||
import { migrateFromJson } from "./json-migration"
|
||||
import { NamedError } from "@opencode-ai/util/error"
|
||||
import z from "zod"
|
||||
import path from "path"
|
||||
|
||||
export const NotFoundError = NamedError.create(
|
||||
"NotFoundError",
|
||||
z.object({
|
||||
message: z.string(),
|
||||
}),
|
||||
)
|
||||
|
||||
const log = Log.create({ service: "db" })
|
||||
|
||||
export type DB = ReturnType<typeof drizzle>
|
||||
|
||||
const connection = lazy(() => {
|
||||
const dbPath = path.join(Global.Path.data, "opencode.db")
|
||||
log.info("opening database", { path: dbPath })
|
||||
|
||||
const sqlite = new Database(dbPath, { create: true })
|
||||
|
||||
sqlite.run("PRAGMA journal_mode = WAL")
|
||||
sqlite.run("PRAGMA synchronous = NORMAL")
|
||||
sqlite.run("PRAGMA busy_timeout = 5000")
|
||||
sqlite.run("PRAGMA cache_size = -64000")
|
||||
sqlite.run("PRAGMA foreign_keys = ON")
|
||||
|
||||
migrate(sqlite)
|
||||
|
||||
// Run JSON migration asynchronously after schema is ready
|
||||
migrateFromJson(sqlite).catch((e) => log.error("json migration failed", { error: e }))
|
||||
|
||||
return drizzle(sqlite)
|
||||
})
|
||||
|
||||
function migrate(sqlite: Database) {
|
||||
sqlite.exec(`
|
||||
CREATE TABLE IF NOT EXISTS _migrations (
|
||||
name TEXT PRIMARY KEY,
|
||||
applied_at INTEGER NOT NULL
|
||||
)
|
||||
`)
|
||||
|
||||
const applied = new Set(
|
||||
sqlite
|
||||
.query<{ name: string }, []>("SELECT name FROM _migrations")
|
||||
.all()
|
||||
.map((r) => r.name),
|
||||
)
|
||||
|
||||
for (const migration of migrations) {
|
||||
if (applied.has(migration.name)) continue
|
||||
log.info("applying migration", { name: migration.name })
|
||||
|
||||
// Split by statement breakpoint and execute each statement
|
||||
// Use IF NOT EXISTS variants to handle partial migrations
|
||||
const statements = migration.sql.split("--> statement-breakpoint")
|
||||
for (const stmt of statements) {
|
||||
const trimmed = stmt.trim()
|
||||
if (!trimmed) continue
|
||||
|
||||
try {
|
||||
sqlite.exec(trimmed)
|
||||
} catch (e: any) {
|
||||
// Ignore "already exists" errors for idempotency
|
||||
if (e?.message?.includes("already exists")) {
|
||||
log.info("skipping existing object", { statement: trimmed.slice(0, 50) })
|
||||
continue
|
||||
}
|
||||
throw e
|
||||
}
|
||||
}
|
||||
|
||||
sqlite.run("INSERT INTO _migrations (name, applied_at) VALUES (?, ?)", [migration.name, Date.now()])
|
||||
}
|
||||
}
|
||||
|
||||
export function db() {
|
||||
return connection()
|
||||
}
|
||||
@@ -0,0 +1,300 @@
|
||||
import { Database } from "bun:sqlite"
|
||||
import { drizzle } from "drizzle-orm/bun-sqlite"
|
||||
import { eq } from "drizzle-orm"
|
||||
import { Global } from "../global"
|
||||
import { Log } from "../util/log"
|
||||
import { ProjectTable } from "../project/project.sql"
|
||||
import {
|
||||
SessionTable,
|
||||
MessageTable,
|
||||
PartTable,
|
||||
SessionDiffTable,
|
||||
TodoTable,
|
||||
PermissionTable,
|
||||
} from "../session/session.sql"
|
||||
import { SessionShareTable, ShareTable } from "../share/share.sql"
|
||||
import path from "path"
|
||||
|
||||
const log = Log.create({ service: "json-migration" })
|
||||
|
||||
export async function migrateFromJson(sqlite: Database, customStorageDir?: string) {
|
||||
const storageDir = customStorageDir ?? path.join(Global.Path.data, "storage")
|
||||
const migrationMarker = path.join(storageDir, "sqlite-migrated")
|
||||
|
||||
if (await Bun.file(migrationMarker).exists()) {
|
||||
log.info("json migration already completed")
|
||||
return
|
||||
}
|
||||
|
||||
if (!(await Bun.file(path.join(storageDir, "migration")).exists())) {
|
||||
log.info("no json storage found, skipping migration")
|
||||
await Bun.write(migrationMarker, Date.now().toString())
|
||||
return
|
||||
}
|
||||
|
||||
log.info("starting json to sqlite migration", { storageDir })
|
||||
|
||||
const db = drizzle(sqlite)
|
||||
const stats = {
|
||||
projects: 0,
|
||||
sessions: 0,
|
||||
messages: 0,
|
||||
parts: 0,
|
||||
diffs: 0,
|
||||
todos: 0,
|
||||
permissions: 0,
|
||||
shares: 0,
|
||||
errors: [] as string[],
|
||||
}
|
||||
|
||||
// Migrate projects first (no FK deps)
|
||||
const projectGlob = new Bun.Glob("project/*.json")
|
||||
for await (const file of projectGlob.scan({ cwd: storageDir, absolute: true })) {
|
||||
try {
|
||||
const data = await Bun.file(file).json()
|
||||
if (!data.id) {
|
||||
stats.errors.push(`project missing id: ${file}`)
|
||||
continue
|
||||
}
|
||||
db.insert(ProjectTable)
|
||||
.values({
|
||||
id: data.id,
|
||||
worktree: data.worktree ?? "/",
|
||||
vcs: data.vcs,
|
||||
name: data.name ?? undefined,
|
||||
icon_url: data.icon?.url,
|
||||
icon_color: data.icon?.color,
|
||||
time_created: data.time?.created ?? Date.now(),
|
||||
time_updated: data.time?.updated ?? Date.now(),
|
||||
time_initialized: data.time?.initialized,
|
||||
sandboxes: data.sandboxes ?? [],
|
||||
})
|
||||
.onConflictDoNothing()
|
||||
.run()
|
||||
stats.projects++
|
||||
} catch (e) {
|
||||
stats.errors.push(`failed to migrate project ${file}: ${e}`)
|
||||
}
|
||||
}
|
||||
log.info("migrated projects", { count: stats.projects })
|
||||
|
||||
// Migrate sessions (depends on projects)
|
||||
const sessionGlob = new Bun.Glob("session/*/*.json")
|
||||
for await (const file of sessionGlob.scan({ cwd: storageDir, absolute: true })) {
|
||||
try {
|
||||
const data = await Bun.file(file).json()
|
||||
if (!data.id || !data.projectID) {
|
||||
stats.errors.push(`session missing id or projectID: ${file}`)
|
||||
continue
|
||||
}
|
||||
// Check if project exists (skip orphaned sessions)
|
||||
const project = db.select().from(ProjectTable).where(eq(ProjectTable.id, data.projectID)).get()
|
||||
if (!project) {
|
||||
log.warn("skipping orphaned session", { sessionID: data.id, projectID: data.projectID })
|
||||
continue
|
||||
}
|
||||
db.insert(SessionTable)
|
||||
.values({
|
||||
id: data.id,
|
||||
projectID: data.projectID,
|
||||
parentID: data.parentID ?? null,
|
||||
slug: data.slug ?? "",
|
||||
directory: data.directory ?? "",
|
||||
title: data.title ?? "",
|
||||
version: data.version ?? "",
|
||||
share_url: data.share?.url ?? null,
|
||||
summary_additions: data.summary?.additions ?? null,
|
||||
summary_deletions: data.summary?.deletions ?? null,
|
||||
summary_files: data.summary?.files ?? null,
|
||||
summary_diffs: data.summary?.diffs ?? null,
|
||||
revert_messageID: data.revert?.messageID ?? null,
|
||||
revert_partID: data.revert?.partID ?? null,
|
||||
revert_snapshot: data.revert?.snapshot ?? null,
|
||||
revert_diff: data.revert?.diff ?? null,
|
||||
permission: data.permission ?? null,
|
||||
time_created: data.time?.created ?? Date.now(),
|
||||
time_updated: data.time?.updated ?? Date.now(),
|
||||
time_compacting: data.time?.compacting ?? null,
|
||||
time_archived: data.time?.archived ?? null,
|
||||
})
|
||||
.onConflictDoNothing()
|
||||
.run()
|
||||
stats.sessions++
|
||||
} catch (e) {
|
||||
stats.errors.push(`failed to migrate session ${file}: ${e}`)
|
||||
}
|
||||
}
|
||||
log.info("migrated sessions", { count: stats.sessions })
|
||||
|
||||
// Migrate messages (depends on sessions)
|
||||
const messageGlob = new Bun.Glob("message/*/*.json")
|
||||
for await (const file of messageGlob.scan({ cwd: storageDir, absolute: true })) {
|
||||
try {
|
||||
const data = await Bun.file(file).json()
|
||||
if (!data.id || !data.sessionID) {
|
||||
stats.errors.push(`message missing id or sessionID: ${file}`)
|
||||
continue
|
||||
}
|
||||
// Check if session exists
|
||||
const session = db.select().from(SessionTable).where(eq(SessionTable.id, data.sessionID)).get()
|
||||
if (!session) {
|
||||
log.warn("skipping orphaned message", { messageID: data.id, sessionID: data.sessionID })
|
||||
continue
|
||||
}
|
||||
db.insert(MessageTable)
|
||||
.values({
|
||||
id: data.id,
|
||||
sessionID: data.sessionID,
|
||||
createdAt: data.time?.created ?? Date.now(),
|
||||
data,
|
||||
})
|
||||
.onConflictDoNothing()
|
||||
.run()
|
||||
stats.messages++
|
||||
} catch (e) {
|
||||
stats.errors.push(`failed to migrate message ${file}: ${e}`)
|
||||
}
|
||||
}
|
||||
log.info("migrated messages", { count: stats.messages })
|
||||
|
||||
// Migrate parts (depends on messages)
|
||||
const partGlob = new Bun.Glob("part/*/*.json")
|
||||
for await (const file of partGlob.scan({ cwd: storageDir, absolute: true })) {
|
||||
try {
|
||||
const data = await Bun.file(file).json()
|
||||
if (!data.id || !data.messageID || !data.sessionID) {
|
||||
stats.errors.push(`part missing id, messageID, or sessionID: ${file}`)
|
||||
continue
|
||||
}
|
||||
// Check if message exists
|
||||
const message = db.select().from(MessageTable).where(eq(MessageTable.id, data.messageID)).get()
|
||||
if (!message) {
|
||||
log.warn("skipping orphaned part", { partID: data.id, messageID: data.messageID })
|
||||
continue
|
||||
}
|
||||
db.insert(PartTable)
|
||||
.values({
|
||||
id: data.id,
|
||||
messageID: data.messageID,
|
||||
sessionID: data.sessionID,
|
||||
data,
|
||||
})
|
||||
.onConflictDoNothing()
|
||||
.run()
|
||||
stats.parts++
|
||||
} catch (e) {
|
||||
stats.errors.push(`failed to migrate part ${file}: ${e}`)
|
||||
}
|
||||
}
|
||||
log.info("migrated parts", { count: stats.parts })
|
||||
|
||||
// Migrate session diffs
|
||||
const diffGlob = new Bun.Glob("session_diff/*.json")
|
||||
for await (const file of diffGlob.scan({ cwd: storageDir, absolute: true })) {
|
||||
try {
|
||||
const data = await Bun.file(file).json()
|
||||
const sessionID = path.basename(file, ".json")
|
||||
// Check if session exists
|
||||
const session = db.select().from(SessionTable).where(eq(SessionTable.id, sessionID)).get()
|
||||
if (!session) {
|
||||
log.warn("skipping orphaned session_diff", { sessionID })
|
||||
continue
|
||||
}
|
||||
db.insert(SessionDiffTable).values({ sessionID, data }).onConflictDoNothing().run()
|
||||
stats.diffs++
|
||||
} catch (e) {
|
||||
stats.errors.push(`failed to migrate session_diff ${file}: ${e}`)
|
||||
}
|
||||
}
|
||||
log.info("migrated session diffs", { count: stats.diffs })
|
||||
|
||||
// Migrate todos
|
||||
const todoGlob = new Bun.Glob("todo/*.json")
|
||||
for await (const file of todoGlob.scan({ cwd: storageDir, absolute: true })) {
|
||||
try {
|
||||
const data = await Bun.file(file).json()
|
||||
const sessionID = path.basename(file, ".json")
|
||||
const session = db.select().from(SessionTable).where(eq(SessionTable.id, sessionID)).get()
|
||||
if (!session) {
|
||||
log.warn("skipping orphaned todo", { sessionID })
|
||||
continue
|
||||
}
|
||||
db.insert(TodoTable).values({ sessionID, data }).onConflictDoNothing().run()
|
||||
stats.todos++
|
||||
} catch (e) {
|
||||
stats.errors.push(`failed to migrate todo ${file}: ${e}`)
|
||||
}
|
||||
}
|
||||
log.info("migrated todos", { count: stats.todos })
|
||||
|
||||
// Migrate permissions
|
||||
const permGlob = new Bun.Glob("permission/*.json")
|
||||
for await (const file of permGlob.scan({ cwd: storageDir, absolute: true })) {
|
||||
try {
|
||||
const data = await Bun.file(file).json()
|
||||
const projectID = path.basename(file, ".json")
|
||||
const project = db.select().from(ProjectTable).where(eq(ProjectTable.id, projectID)).get()
|
||||
if (!project) {
|
||||
log.warn("skipping orphaned permission", { projectID })
|
||||
continue
|
||||
}
|
||||
db.insert(PermissionTable).values({ projectID, data }).onConflictDoNothing().run()
|
||||
stats.permissions++
|
||||
} catch (e) {
|
||||
stats.errors.push(`failed to migrate permission ${file}: ${e}`)
|
||||
}
|
||||
}
|
||||
log.info("migrated permissions", { count: stats.permissions })
|
||||
|
||||
// Migrate session shares
|
||||
const shareGlob = new Bun.Glob("session_share/*.json")
|
||||
for await (const file of shareGlob.scan({ cwd: storageDir, absolute: true })) {
|
||||
try {
|
||||
const data = await Bun.file(file).json()
|
||||
const sessionID = path.basename(file, ".json")
|
||||
const session = db.select().from(SessionTable).where(eq(SessionTable.id, sessionID)).get()
|
||||
if (!session) {
|
||||
log.warn("skipping orphaned session_share", { sessionID })
|
||||
continue
|
||||
}
|
||||
db.insert(SessionShareTable).values({ sessionID, data }).onConflictDoNothing().run()
|
||||
stats.shares++
|
||||
} catch (e) {
|
||||
stats.errors.push(`failed to migrate session_share ${file}: ${e}`)
|
||||
}
|
||||
}
|
||||
log.info("migrated session shares", { count: stats.shares })
|
||||
|
||||
// Migrate shares (downloaded shared sessions, no FK)
|
||||
const share2Glob = new Bun.Glob("share/*.json")
|
||||
for await (const file of share2Glob.scan({ cwd: storageDir, absolute: true })) {
|
||||
try {
|
||||
const data = await Bun.file(file).json()
|
||||
const sessionID = path.basename(file, ".json")
|
||||
db.insert(ShareTable).values({ sessionID, data }).onConflictDoNothing().run()
|
||||
} catch (e) {
|
||||
stats.errors.push(`failed to migrate share ${file}: ${e}`)
|
||||
}
|
||||
}
|
||||
|
||||
// Mark migration complete
|
||||
await Bun.write(migrationMarker, Date.now().toString())
|
||||
|
||||
log.info("json migration complete", {
|
||||
projects: stats.projects,
|
||||
sessions: stats.sessions,
|
||||
messages: stats.messages,
|
||||
parts: stats.parts,
|
||||
diffs: stats.diffs,
|
||||
todos: stats.todos,
|
||||
permissions: stats.permissions,
|
||||
shares: stats.shares,
|
||||
errorCount: stats.errors.length,
|
||||
})
|
||||
|
||||
if (stats.errors.length > 0) {
|
||||
log.warn("migration errors", { errors: stats.errors.slice(0, 20) })
|
||||
}
|
||||
|
||||
return stats
|
||||
}
|
||||
@@ -0,0 +1,6 @@
|
||||
// Auto-generated - do not edit
|
||||
import m0 from "../../migration/0000_magical_strong_guy.sql" with { type: "text" }
|
||||
|
||||
export const migrations = [
|
||||
{ name: "0000_magical_strong_guy", sql: m0 },
|
||||
]
|
||||
@@ -1,227 +0,0 @@
|
||||
import { Log } from "../util/log"
|
||||
import path from "path"
|
||||
import fs from "fs/promises"
|
||||
import { Global } from "../global"
|
||||
import { Filesystem } from "../util/filesystem"
|
||||
import { lazy } from "../util/lazy"
|
||||
import { Lock } from "../util/lock"
|
||||
import { $ } from "bun"
|
||||
import { NamedError } from "@opencode-ai/util/error"
|
||||
import z from "zod"
|
||||
|
||||
export namespace Storage {
|
||||
const log = Log.create({ service: "storage" })
|
||||
|
||||
type Migration = (dir: string) => Promise<void>
|
||||
|
||||
export const NotFoundError = NamedError.create(
|
||||
"NotFoundError",
|
||||
z.object({
|
||||
message: z.string(),
|
||||
}),
|
||||
)
|
||||
|
||||
const MIGRATIONS: Migration[] = [
|
||||
async (dir) => {
|
||||
const project = path.resolve(dir, "../project")
|
||||
if (!(await Filesystem.isDir(project))) return
|
||||
for await (const projectDir of new Bun.Glob("*").scan({
|
||||
cwd: project,
|
||||
onlyFiles: false,
|
||||
})) {
|
||||
log.info(`migrating project ${projectDir}`)
|
||||
let projectID = projectDir
|
||||
const fullProjectDir = path.join(project, projectDir)
|
||||
let worktree = "/"
|
||||
|
||||
if (projectID !== "global") {
|
||||
for await (const msgFile of new Bun.Glob("storage/session/message/*/*.json").scan({
|
||||
cwd: path.join(project, projectDir),
|
||||
absolute: true,
|
||||
})) {
|
||||
const json = await Bun.file(msgFile).json()
|
||||
worktree = json.path?.root
|
||||
if (worktree) break
|
||||
}
|
||||
if (!worktree) continue
|
||||
if (!(await Filesystem.isDir(worktree))) continue
|
||||
const [id] = await $`git rev-list --max-parents=0 --all`
|
||||
.quiet()
|
||||
.nothrow()
|
||||
.cwd(worktree)
|
||||
.text()
|
||||
.then((x) =>
|
||||
x
|
||||
.split("\n")
|
||||
.filter(Boolean)
|
||||
.map((x) => x.trim())
|
||||
.toSorted(),
|
||||
)
|
||||
if (!id) continue
|
||||
projectID = id
|
||||
|
||||
await Bun.write(
|
||||
path.join(dir, "project", projectID + ".json"),
|
||||
JSON.stringify({
|
||||
id,
|
||||
vcs: "git",
|
||||
worktree,
|
||||
time: {
|
||||
created: Date.now(),
|
||||
initialized: Date.now(),
|
||||
},
|
||||
}),
|
||||
)
|
||||
|
||||
log.info(`migrating sessions for project ${projectID}`)
|
||||
for await (const sessionFile of new Bun.Glob("storage/session/info/*.json").scan({
|
||||
cwd: fullProjectDir,
|
||||
absolute: true,
|
||||
})) {
|
||||
const dest = path.join(dir, "session", projectID, path.basename(sessionFile))
|
||||
log.info("copying", {
|
||||
sessionFile,
|
||||
dest,
|
||||
})
|
||||
const session = await Bun.file(sessionFile).json()
|
||||
await Bun.write(dest, JSON.stringify(session))
|
||||
log.info(`migrating messages for session ${session.id}`)
|
||||
for await (const msgFile of new Bun.Glob(`storage/session/message/${session.id}/*.json`).scan({
|
||||
cwd: fullProjectDir,
|
||||
absolute: true,
|
||||
})) {
|
||||
const dest = path.join(dir, "message", session.id, path.basename(msgFile))
|
||||
log.info("copying", {
|
||||
msgFile,
|
||||
dest,
|
||||
})
|
||||
const message = await Bun.file(msgFile).json()
|
||||
await Bun.write(dest, JSON.stringify(message))
|
||||
|
||||
log.info(`migrating parts for message ${message.id}`)
|
||||
for await (const partFile of new Bun.Glob(`storage/session/part/${session.id}/${message.id}/*.json`).scan(
|
||||
{
|
||||
cwd: fullProjectDir,
|
||||
absolute: true,
|
||||
},
|
||||
)) {
|
||||
const dest = path.join(dir, "part", message.id, path.basename(partFile))
|
||||
const part = await Bun.file(partFile).json()
|
||||
log.info("copying", {
|
||||
partFile,
|
||||
dest,
|
||||
})
|
||||
await Bun.write(dest, JSON.stringify(part))
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
async (dir) => {
|
||||
for await (const item of new Bun.Glob("session/*/*.json").scan({
|
||||
cwd: dir,
|
||||
absolute: true,
|
||||
})) {
|
||||
const session = await Bun.file(item).json()
|
||||
if (!session.projectID) continue
|
||||
if (!session.summary?.diffs) continue
|
||||
const { diffs } = session.summary
|
||||
await Bun.file(path.join(dir, "session_diff", session.id + ".json")).write(JSON.stringify(diffs))
|
||||
await Bun.file(path.join(dir, "session", session.projectID, session.id + ".json")).write(
|
||||
JSON.stringify({
|
||||
...session,
|
||||
summary: {
|
||||
additions: diffs.reduce((sum: any, x: any) => sum + x.additions, 0),
|
||||
deletions: diffs.reduce((sum: any, x: any) => sum + x.deletions, 0),
|
||||
},
|
||||
}),
|
||||
)
|
||||
}
|
||||
},
|
||||
]
|
||||
|
||||
const state = lazy(async () => {
|
||||
const dir = path.join(Global.Path.data, "storage")
|
||||
const migration = await Bun.file(path.join(dir, "migration"))
|
||||
.json()
|
||||
.then((x) => parseInt(x))
|
||||
.catch(() => 0)
|
||||
for (let index = migration; index < MIGRATIONS.length; index++) {
|
||||
log.info("running migration", { index })
|
||||
const migration = MIGRATIONS[index]
|
||||
await migration(dir).catch(() => log.error("failed to run migration", { index }))
|
||||
await Bun.write(path.join(dir, "migration"), (index + 1).toString())
|
||||
}
|
||||
return {
|
||||
dir,
|
||||
}
|
||||
})
|
||||
|
||||
export async function remove(key: string[]) {
|
||||
const dir = await state().then((x) => x.dir)
|
||||
const target = path.join(dir, ...key) + ".json"
|
||||
return withErrorHandling(async () => {
|
||||
await fs.unlink(target).catch(() => {})
|
||||
})
|
||||
}
|
||||
|
||||
export async function read<T>(key: string[]) {
|
||||
const dir = await state().then((x) => x.dir)
|
||||
const target = path.join(dir, ...key) + ".json"
|
||||
return withErrorHandling(async () => {
|
||||
using _ = await Lock.read(target)
|
||||
const result = await Bun.file(target).json()
|
||||
return result as T
|
||||
})
|
||||
}
|
||||
|
||||
export async function update<T>(key: string[], fn: (draft: T) => void) {
|
||||
const dir = await state().then((x) => x.dir)
|
||||
const target = path.join(dir, ...key) + ".json"
|
||||
return withErrorHandling(async () => {
|
||||
using _ = await Lock.write(target)
|
||||
const content = await Bun.file(target).json()
|
||||
fn(content)
|
||||
await Bun.write(target, JSON.stringify(content, null, 2))
|
||||
return content as T
|
||||
})
|
||||
}
|
||||
|
||||
export async function write<T>(key: string[], content: T) {
|
||||
const dir = await state().then((x) => x.dir)
|
||||
const target = path.join(dir, ...key) + ".json"
|
||||
return withErrorHandling(async () => {
|
||||
using _ = await Lock.write(target)
|
||||
await Bun.write(target, JSON.stringify(content, null, 2))
|
||||
})
|
||||
}
|
||||
|
||||
async function withErrorHandling<T>(body: () => Promise<T>) {
|
||||
return body().catch((e) => {
|
||||
if (!(e instanceof Error)) throw e
|
||||
const errnoException = e as NodeJS.ErrnoException
|
||||
if (errnoException.code === "ENOENT") {
|
||||
throw new NotFoundError({ message: `Resource not found: ${errnoException.path}` })
|
||||
}
|
||||
throw e
|
||||
})
|
||||
}
|
||||
|
||||
const glob = new Bun.Glob("**/*")
|
||||
export async function list(prefix: string[]) {
|
||||
const dir = await state().then((x) => x.dir)
|
||||
try {
|
||||
const result = await Array.fromAsync(
|
||||
glob.scan({
|
||||
cwd: path.join(dir, ...prefix),
|
||||
onlyFiles: true,
|
||||
}),
|
||||
).then((results) => results.map((x) => [...prefix, ...x.slice(0, -5).split(path.sep)]))
|
||||
result.sort()
|
||||
return result
|
||||
} catch {
|
||||
return []
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -4,9 +4,14 @@ export function lazy<T>(fn: () => T) {
|
||||
|
||||
const result = (): T => {
|
||||
if (loaded) return value as T
|
||||
loaded = true
|
||||
value = fn()
|
||||
return value as T
|
||||
try {
|
||||
value = fn()
|
||||
loaded = true
|
||||
return value as T
|
||||
} catch (e) {
|
||||
// Don't mark as loaded if initialization failed
|
||||
throw e
|
||||
}
|
||||
}
|
||||
|
||||
result.reset = () => {
|
||||
|
||||
@@ -7,7 +7,9 @@ import { Global } from "../global"
|
||||
import { Instance } from "../project/instance"
|
||||
import { InstanceBootstrap } from "../project/bootstrap"
|
||||
import { Project } from "../project/project"
|
||||
import { Storage } from "../storage/storage"
|
||||
import { db } from "../storage/db"
|
||||
import { ProjectTable } from "../project/project.sql"
|
||||
import { eq } from "drizzle-orm"
|
||||
import { fn } from "../util/fn"
|
||||
import { Log } from "../util/log"
|
||||
import { BusEvent } from "@/bus/bus-event"
|
||||
@@ -318,7 +320,8 @@ export namespace Worktree {
|
||||
},
|
||||
})
|
||||
|
||||
const project = await Storage.read<Project.Info>(["project", projectID]).catch(() => undefined)
|
||||
const row = db().select().from(ProjectTable).where(eq(ProjectTable.id, projectID)).get()
|
||||
const project = row ? Project.fromRow(row) : undefined
|
||||
const startup = project?.commands?.start?.trim() ?? ""
|
||||
|
||||
const run = async (cmd: string, kind: "project" | "worktree") => {
|
||||
|
||||
Reference in New Issue
Block a user