ceos/opencode
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

refactor(v2): share effect sqlite database layer

Browse source
This commit is contained in:
Kit Langton 2026-05-20 20:41:57 -04:00
commit b3d6f93148
6 changed files with 98 additions and 112 deletions
Download patch file Download diff file Expand all files Collapse all files

21
packages/effect-drizzle-sqlite/src/effect-sqlite/migrator.ts
View file

@ -1,8 +1,7 @@
/* oxlint-disable */
import type { MigrationConfig, MigrationFromJournalConfig, MigrationsJournal } from "drizzle-orm/migrator"
import type { MigrationConfig } from "drizzle-orm/migrator"
import { readMigrationFiles } from "drizzle-orm/migrator"
import type { AnyRelations } from "drizzle-orm/relations"
import crypto from "node:crypto"
import { migrate as coreMigrate } from "../sqlite-core/effect/session"
import type { EffectSQLiteDatabase } from "./driver"
@ -13,21 +12,3 @@ export function migrate<TRelations extends AnyRelations>(
const migrations = readMigrationFiles(config)
return coreMigrate(migrations, db.session, config)
}
export function migrateFromJournal<TRelations extends AnyRelations>(
db: EffectSQLiteDatabase<TRelations>,
journal: MigrationsJournal,
config: Omit<MigrationFromJournalConfig, "migrationsJournal"> = {},
) {
return coreMigrate(
journal.map((migration) => ({
sql: migration.sql.split("--> statement-breakpoint"),
bps: true,
folderMillis: migration.timestamp,
hash: crypto.createHash("sha256").update(migration.sql).digest("hex"),
name: migration.name,
})),
db.session,
{ migrationsFolder: "", migrationsTable: config.migrationsTable },
)
}

2
packages/effect-drizzle-sqlite/src/index.ts
View file

@ -1,6 +1,6 @@
export { EffectLogger } from "drizzle-orm/effect-core"
export * from "./effect-sqlite/driver"
export * from "./effect-sqlite/session"
export { migrate, migrateFromJournal } from "./effect-sqlite/migrator"
export { migrate } from "./effect-sqlite/migrator"
export * as EffectDrizzleSqlite from "."

21
packages/opencode/src/storage/db.ts
View file

@ -48,21 +48,12 @@ export type Transaction = SQLiteTransaction<"sync", void>
type Client = ReturnType<typeof init>
export type Journal = MigrationsJournal
type Journal = MigrationsJournal
function applyMigrations(db: SQLiteBunDatabase, entries: Journal) {
migrate(db, entries)
}
export function migrationJournal(flags: Pick<DatabaseFlags, "skipMigrations"> = readRuntimeFlags()) {
const entries =
typeof OPENCODE_MIGRATIONS !== "undefined"
? OPENCODE_MIGRATIONS
: migrations(path.join(import.meta.dirname, "../../migration"))
if (!flags.skipMigrations) return entries
return entries.map((item) => ({ ...item, sql: "select 1;" }))
}
function time(tag: string) {
const match = /^(\d{4})(\d{2})(\d{2})(\d{2})(\d{2})(\d{2})/.exec(tag)
if (!match) return 0
@ -116,12 +107,20 @@ export const Client = Object.assign(
db.run("PRAGMA wal_checkpoint(PASSIVE)")
// Apply schema migrations
const entries = migrationJournal(flags)
const entries =
typeof OPENCODE_MIGRATIONS !== "undefined"
? OPENCODE_MIGRATIONS
: migrations(path.join(import.meta.dirname, "../../migration"))
if (entries.length > 0) {
log.info("applying migrations", {
count: entries.length,
mode: typeof OPENCODE_MIGRATIONS !== "undefined" ? "bundled" : "dev",
})
if (flags.skipMigrations) {
for (const item of entries) {
item.sql = "select 1;"
}
}
applyMigrations(db, entries)
}

39
packages/opencode/src/v2/storage/database.ts Normal file
View file

@ -0,0 +1,39 @@
import { Database as LegacyDatabase } from "@/storage/db"
import { SqliteClient } from "@effect/sql-sqlite-bun"
import { EffectDrizzleSqlite } from "@opencode-ai/effect-drizzle-sqlite"
import { Context, Effect, Layer } from "effect"
import path from "path"
const makeDatabase = EffectDrizzleSqlite.makeWithDefaults()
type DatabaseShape = Effect.Success<typeof makeDatabase>
export class Service extends Context.Service<Service, DatabaseShape>()("@opencode/v2/storage/Database") {}
export const layer = Layer.unwrap(
Effect.sync(() => {
const filename = LegacyDatabase.getPath()
return Layer.effect(
Service,
Effect.gen(function* () {
LegacyDatabase.Client()
const db = yield* makeDatabase
yield* db.run("PRAGMA journal_mode = WAL")
yield* db.run("PRAGMA synchronous = NORMAL")
yield* db.run("PRAGMA busy_timeout = 5000")
yield* db.run("PRAGMA cache_size = -64000")
yield* db.run("PRAGMA foreign_keys = ON")
yield* db.run("PRAGMA wal_checkpoint(PASSIVE)")
if (filename === ":memory:") {
yield* EffectDrizzleSqlite.migrate(db, {
migrationsFolder: path.join(import.meta.dirname, "../../../migration"),
})
}
return db
}),
).pipe(Layer.provide(SqliteClient.layer({ filename, disableWAL: filename === ":memory:" })))
}),
)
export const defaultLayer = layer
export * as StorageDatabase from "./database"

112
packages/opencode/src/v2/storage/session-sql.ts
View file

@ -1,47 +1,25 @@
import { SessionMessageTable, SessionTable } from "@/session/session.sql"
import { and, asc, Database as LegacyDatabase, desc, eq, gt, gte, isNull, like, lt, or, type SQL } from "@/storage/db"
import { SqliteClient } from "@effect/sql-sqlite-bun"
import { and, asc, desc, eq, gt, gte, isNull, like, lt, or, type SQL } from "@/storage/db"
import { SessionMessage } from "@opencode-ai/core/session-message"
import { EffectDrizzleSqlite } from "@opencode-ai/effect-drizzle-sqlite"
import { Context, Effect, Layer, Schema } from "effect"
import { Effect, Layer, Schema } from "effect"
import { StorageDatabase } from "./database"
import { SessionStorage } from "./session"
const decodeMessage = Schema.decodeUnknownSync(SessionMessage.Message)
const decodeSessionRow = Schema.decodeUnknownSync(SessionStorage.SessionRow)
const makeDatabase = EffectDrizzleSqlite.makeWithDefaults()
type DatabaseShape = Effect.Success<typeof makeDatabase>
export class Database extends Context.Service<Database, DatabaseShape>()("@opencode/v2/session/StorageSql/Database") {}
export const databaseLayer = Layer.unwrap(
Effect.sync(() => {
const filename = LegacyDatabase.getPath()
return Layer.effect(
Database,
Effect.gen(function* () {
const db = yield* makeDatabase
yield* db.run("PRAGMA journal_mode = WAL")
yield* db.run("PRAGMA synchronous = NORMAL")
yield* db.run("PRAGMA busy_timeout = 5000")
yield* db.run("PRAGMA cache_size = -64000")
yield* db.run("PRAGMA foreign_keys = ON")
yield* db.run("PRAGMA wal_checkpoint(PASSIVE)")
yield* EffectDrizzleSqlite.migrateFromJournal(db, LegacyDatabase.migrationJournal())
return db
}),
).pipe(Layer.provide(SqliteClient.layer({ filename, disableWAL: filename === ":memory:" })))
}),
const mapStorageError = Effect.mapError(
(cause) => new SessionStorage.StorageError({ message: "Session storage SQL operation failed", cause }),
)
export const layer = Layer.effect(
SessionStorage.Service,
Effect.gen(function* () {
const db = yield* Database
const db = yield* StorageDatabase.Service
const get: SessionStorage.Interface["get"] = Effect.fn("SessionStorageSql.get")(function* (sessionID) {
const row = yield* attempt(db.select().from(SessionTable).where(eq(SessionTable.id, sessionID)).get())
const row = yield* db.select().from(SessionTable).where(eq(SessionTable.id, sessionID)).get()
return row ? fromSessionRow(row) : undefined
})
}, mapStorageError)
const list: SessionStorage.Interface["list"] = Effect.fn("SessionStorageSql.list")(function* (input) {
const direction = input.cursor?.direction ?? "next"
@ -65,9 +43,9 @@ export const layer = Layer.effect(
order === "asc" ? asc(sortColumn) : desc(sortColumn),
order === "asc" ? asc(SessionTable.id) : desc(SessionTable.id),
)
const rows = yield* attempt(input.limit === undefined ? query : query.limit(input.limit))
const rows = yield* input.limit === undefined ? query : query.limit(input.limit)
return (direction === "previous" ? rows.toReversed() : rows).map(fromSessionRow)
})
}, mapStorageError)
const messages: SessionStorage.Interface["messages"] = Effect.fn("SessionStorageSql.messages")(function* (input) {
const direction = input.cursor?.direction ?? "next"
@ -85,62 +63,50 @@ export const layer = Layer.effect(
order === "asc" ? asc(SessionMessageTable.time_created) : desc(SessionMessageTable.time_created),
order === "asc" ? asc(SessionMessageTable.id) : desc(SessionMessageTable.id),
)
const rows = yield* attempt(input.limit === undefined ? query : query.limit(input.limit))
const rows = yield* input.limit === undefined ? query : query.limit(input.limit)
return (direction === "previous" ? rows.toReversed() : rows).map((row) =>
decodeMessage({ ...row.data, id: row.id, type: row.type }),
)
})
}, mapStorageError)
const context: SessionStorage.Interface["context"] = Effect.fn("SessionStorageSql.context")((sessionID) =>
Effect.gen(function* () {
const compaction = yield* attempt(
db
.select()
.from(SessionMessageTable)
.where(and(eq(SessionMessageTable.session_id, sessionID), eq(SessionMessageTable.type, "compaction")))
.orderBy(desc(SessionMessageTable.time_created), desc(SessionMessageTable.id))
.limit(1)
.get(),
)
const compaction = yield* db
.select()
.from(SessionMessageTable)
.where(and(eq(SessionMessageTable.session_id, sessionID), eq(SessionMessageTable.type, "compaction")))
.orderBy(desc(SessionMessageTable.time_created), desc(SessionMessageTable.id))
.limit(1)
.get()
const rows = yield* attempt(
db
.select()
.from(SessionMessageTable)
.where(
and(
eq(SessionMessageTable.session_id, sessionID),
compaction
? or(
gt(SessionMessageTable.time_created, compaction.time_created),
and(
eq(SessionMessageTable.time_created, compaction.time_created),
gte(SessionMessageTable.id, compaction.id),
),
)
: undefined,
),
)
.orderBy(asc(SessionMessageTable.time_created), asc(SessionMessageTable.id)),
)
const rows = yield* db
.select()
.from(SessionMessageTable)
.where(
and(
eq(SessionMessageTable.session_id, sessionID),
compaction
? or(
gt(SessionMessageTable.time_created, compaction.time_created),
and(
eq(SessionMessageTable.time_created, compaction.time_created),
gte(SessionMessageTable.id, compaction.id),
),
)
: undefined,
),
)
.orderBy(asc(SessionMessageTable.time_created), asc(SessionMessageTable.id))
return rows.map((row) => decodeMessage({ ...row.data, id: row.id, type: row.type }))
}),
}).pipe(mapStorageError),
)
return SessionStorage.Service.of({ get, list, messages, context })
}),
)
export const defaultLayer = layer.pipe(Layer.provide(databaseLayer.pipe(Layer.orDie)))
function attempt<A, E, R>(effect: Effect.Effect<A, E, R>) {
return effect.pipe(
Effect.mapError(
(cause) => new SessionStorage.StorageError({ message: "Session storage SQL operation failed", cause }),
),
)
}
export const defaultLayer = layer.pipe(Layer.provide(StorageDatabase.defaultLayer.pipe(Layer.orDie)))
function sessionCursorBoundary(cursor: SessionStorage.SessionCursor, order: SessionStorage.SortOrder) {
if (order === "asc")

13
packages/opencode/test/v2/session-storage.test.ts
View file

@ -4,6 +4,7 @@ import { ProjectTable } from "@/project/project.sql"
import { SessionID } from "@/session/schema"
import { SessionMessageTable, SessionTable } from "@/session/session.sql"
import { SessionStorage } from "@/v2/storage/session"
import { StorageDatabase } from "@/v2/storage/database"
import { SessionStorageMemory } from "@/v2/storage/session-memory"
import { SessionStorageSql } from "@/v2/storage/session-sql"
import { EventV2 } from "@opencode-ai/core/event"
@ -156,9 +157,9 @@ function sessionStorageContract<R, E>(name: string, layer: Layer.Layer<SessionSt
)
}
const sqlLayer = SessionStorageSql.layer.pipe(Layer.provideMerge(SessionStorageSql.databaseLayer))
const sqlLayer = SessionStorageSql.layer.pipe(Layer.provideMerge(StorageDatabase.defaultLayer))
const sqlSeeds: Seeds<SessionStorageSql.Database> = {
const sqlSeeds: Seeds<StorageDatabase.Service> = {
reset: resetSqlSeeds(),
project: seedProject(),
session: seedSession,
@ -192,7 +193,7 @@ sessionStorageContract("SessionStorageMemory", memoryLayer, memorySeeds)
function seedProject() {
return Effect.gen(function* () {
const db = yield* SessionStorageSql.Database
const db = yield* StorageDatabase.Service
yield* db
.insert(ProjectTable)
.values({
@ -210,7 +211,7 @@ function seedProject() {
function resetSqlSeeds() {
return Effect.gen(function* () {
const db = yield* SessionStorageSql.Database
const db = yield* StorageDatabase.Service
yield* db.delete(SessionMessageTable).where(eq(SessionMessageTable.session_id, sessionA)).run().pipe(Effect.orDie)
yield* db
.delete(SessionTable)
@ -230,7 +231,7 @@ function resetSqlSeeds() {
function seedSession(input: { id: SessionID; title: string; directory?: string; path: string; updated: number }) {
return Effect.gen(function* () {
const db = yield* SessionStorageSql.Database
const db = yield* StorageDatabase.Service
yield* db
.insert(SessionTable)
.values({
@ -320,7 +321,7 @@ function seedMessage(
data: typeof SessionMessageTable.$inferInsert.data,
) {
return Effect.gen(function* () {
const db = yield* SessionStorageSql.Database
const db = yield* StorageDatabase.Service
yield* db
.insert(SessionMessageTable)
.values({