Skip to content

Commit dcbd244

Browse files
kitlangtonkitlangton
committed
refactor(v2): use effect sqlite session storage
1 parent dc55ece commit dcbd244

7 files changed

Lines changed: 183 additions & 124 deletions

File tree

‎bun.lock‎

Lines changed: 2 additions & 0 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎packages/effect-drizzle-sqlite/src/effect-sqlite/migrator.ts‎

Lines changed: 20 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,8 @@
11
/* oxlint-disable */
2-
import type { MigrationConfig } from "drizzle-orm/migrator"
2+
import type { MigrationConfig, MigrationFromJournalConfig, MigrationsJournal } from "drizzle-orm/migrator"
33
import { readMigrationFiles } from "drizzle-orm/migrator"
44
import type { AnyRelations } from "drizzle-orm/relations"
5+
import crypto from "node:crypto"
56
import { migrate as coreMigrate } from "../sqlite-core/effect/session"
67
import type { EffectSQLiteDatabase } from "./driver"
78

@@ -12,3 +13,21 @@ export function migrate<TRelations extends AnyRelations>(
1213
const migrations = readMigrationFiles(config)
1314
return coreMigrate(migrations, db.session, config)
1415
}
16+
17+
export function migrateFromJournal<TRelations extends AnyRelations>(
18+
db: EffectSQLiteDatabase<TRelations>,
19+
journal: MigrationsJournal,
20+
config: Omit<MigrationFromJournalConfig, "migrationsJournal"> = {},
21+
) {
22+
return coreMigrate(
23+
journal.map((migration) => ({
24+
sql: migration.sql.split("--> statement-breakpoint"),
25+
bps: true,
26+
folderMillis: migration.timestamp,
27+
hash: crypto.createHash("sha256").update(migration.sql).digest("hex"),
28+
name: migration.name,
29+
})),
30+
db.session,
31+
{ migrationsFolder: "", migrationsTable: config.migrationsTable },
32+
)
33+
}

‎packages/effect-drizzle-sqlite/src/index.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
export { EffectLogger } from "drizzle-orm/effect-core"
22
export * from "./effect-sqlite/driver"
33
export * from "./effect-sqlite/session"
4-
export { migrate } from "./effect-sqlite/migrator"
4+
export { migrate, migrateFromJournal } from "./effect-sqlite/migrator"
55

66
export * as EffectDrizzleSqlite from "."

‎packages/opencode/package.json‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -96,12 +96,14 @@
9696
"@clack/prompts": "1.0.0-alpha.1",
9797
"@effect/opentelemetry": "catalog:",
9898
"@effect/platform-node": "catalog:",
99+
"@effect/sql-sqlite-bun": "catalog:",
99100
"@gitlab/opencode-gitlab-auth": "1.3.3",
100101
"@lydell/node-pty": "catalog:",
101102
"@modelcontextprotocol/sdk": "1.27.1",
102103
"@octokit/graphql": "9.0.2",
103104
"@octokit/rest": "catalog:",
104105
"@openauthjs/openauth": "catalog:",
106+
"@opencode-ai/effect-drizzle-sqlite": "workspace:*",
105107
"@opencode-ai/llm": "workspace:*",
106108
"@opencode-ai/plugin": "workspace:*",
107109
"@opencode-ai/script": "workspace:*",

‎packages/opencode/src/storage/db.ts‎

Lines changed: 23 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
import { type SQLiteBunDatabase } from "drizzle-orm/bun-sqlite"
22
import { migrate } from "drizzle-orm/bun-sqlite/migrator"
3+
import type { MigrationsJournal } from "drizzle-orm/migrator"
34
import { type SQLiteTransaction } from "drizzle-orm/sqlite-core"
45
export * from "drizzle-orm"
56
import { RuntimeFlags } from "@/effect/runtime-flags"
@@ -47,13 +48,19 @@ export type Transaction = SQLiteTransaction<"sync", void>
4748

4849
type Client = ReturnType<typeof init>
4950

50-
type Journal = { sql: string; timestamp: number; name: string }[]
51-
52-
// Drizzle's migrate overloads trigger expensive variance checks here; narrow to the journal overload we actually use.
53-
const migrateFromJournal = migrate as unknown as (db: SQLiteBunDatabase, entries: Journal) => void
51+
export type Journal = MigrationsJournal
5452

5553
function applyMigrations(db: SQLiteBunDatabase, entries: Journal) {
56-
migrateFromJournal(db, entries)
54+
migrate(db, entries)
55+
}
56+
57+
export function migrationJournal(flags: Pick<DatabaseFlags, "skipMigrations"> = readRuntimeFlags()) {
58+
const entries =
59+
typeof OPENCODE_MIGRATIONS !== "undefined"
60+
? OPENCODE_MIGRATIONS
61+
: migrations(path.join(import.meta.dirname, "../../migration"))
62+
if (!flags.skipMigrations) return entries
63+
return entries.map((item) => ({ ...item, sql: "select 1;" }))
5764
}
5865

5966
function time(tag: string) {
@@ -74,17 +81,17 @@ function migrations(dir: string): Journal {
7481
.filter((entry) => entry.isDirectory())
7582
.map((entry) => entry.name)
7683

77-
const sql = dirs
84+
const sql: Journal = dirs
7885
.map((name) => {
7986
const file = path.join(dir, name, "migration.sql")
80-
if (!existsSync(file)) return
87+
if (!existsSync(file)) return undefined
8188
return {
8289
sql: readFileSync(file, "utf-8"),
8390
timestamp: time(name),
8491
name,
8592
}
8693
})
87-
.filter(Boolean) as Journal
94+
.filter((entry) => entry !== undefined)
8895

8996
return sql.sort((a, b) => a.timestamp - b.timestamp)
9097
}
@@ -94,7 +101,7 @@ let loaded = false
94101

95102
export const Client = Object.assign(
96103
(flags: DatabaseFlags = readRuntimeFlags()): Client => {
97-
if (loaded) return client as Client
104+
if (loaded && client) return client
98105

99106
const dbPath = getPath(flags)
100107
log.info("opening database", { path: dbPath })
@@ -109,20 +116,12 @@ export const Client = Object.assign(
109116
db.run("PRAGMA wal_checkpoint(PASSIVE)")
110117

111118
// Apply schema migrations
112-
const entries =
113-
typeof OPENCODE_MIGRATIONS !== "undefined"
114-
? OPENCODE_MIGRATIONS
115-
: migrations(path.join(import.meta.dirname, "../../migration"))
119+
const entries = migrationJournal(flags)
116120
if (entries.length > 0) {
117121
log.info("applying migrations", {
118122
count: entries.length,
119123
mode: typeof OPENCODE_MIGRATIONS !== "undefined" ? "bundled" : "dev",
120124
})
121-
if (flags.skipMigrations) {
122-
for (const item of entries) {
123-
item.sql = "select 1;"
124-
}
125-
}
126125
applyMigrations(db, entries)
127126
}
128127

@@ -159,19 +158,19 @@ export function use<T>(callback: (trx: TxOrDb) => T): T {
159158
if (err instanceof LocalContext.NotFound) {
160159
const effects: (() => void | Promise<void>)[] = []
161160
const result = ctx.provide({ effects, tx: Client() }, () => callback(Client()))
162-
for (const effect of effects) effect()
161+
for (const effect of effects) void effect()
163162
return result
164163
}
165164
throw err
166165
}
167166
}
168167

169-
export function effect(fn: () => any | Promise<any>) {
168+
export function effect(fn: () => void | Promise<void>) {
170169
const bound = EffectBridge.bind(fn)
171170
try {
172171
ctx.use().effects.push(bound)
173172
} catch {
174-
bound()
173+
void bound()
175174
}
176175
}
177176

@@ -190,7 +189,9 @@ export function transaction<T>(
190189
const effects: (() => void | Promise<void>)[] = []
191190
const txCallback = EffectBridge.bind((tx: TxOrDb) => ctx.provide({ tx, effects }, () => callback(tx)))
192191
const result = Client().transaction(txCallback, { behavior: options?.behavior })
193-
for (const effect of effects) effect()
192+
for (const effect of effects) void effect()
193+
// Drizzle's transaction type does not preserve our NotPromise<T> constraint through the callback wrapper.
194+
// oxlint-disable-next-line typescript-eslint/no-unsafe-type-assertion
194195
return result as NotPromise<T>
195196
}
196197
throw err

‎packages/opencode/src/v2/storage/session-sql.ts‎

Lines changed: 99 additions & 74 deletions
Original file line numberDiff line numberDiff line change
@@ -1,87 +1,110 @@
11
import { SessionMessageTable, SessionTable } from "@/session/session.sql"
2-
import { and, asc, Database, desc, eq, gt, gte, isNull, like, lt, or, type SQL } from "@/storage/db"
2+
import { and, asc, Database as LegacyDatabase, desc, eq, gt, gte, isNull, like, lt, or, type SQL } from "@/storage/db"
3+
import { SqliteClient } from "@effect/sql-sqlite-bun"
34
import { SessionMessage } from "@opencode-ai/core/session-message"
4-
import { Effect, Layer, Schema } from "effect"
5+
import { EffectDrizzleSqlite } from "@opencode-ai/effect-drizzle-sqlite"
6+
import { Context, Effect, Layer, Schema } from "effect"
57
import { SessionStorage } from "./session"
68

79
const decodeMessage = Schema.decodeUnknownSync(SessionMessage.Message)
810
const decodeSessionRow = Schema.decodeUnknownSync(SessionStorage.SessionRow)
11+
const makeDatabase = EffectDrizzleSqlite.makeWithDefaults()
12+
type DatabaseShape = Effect.Success<typeof makeDatabase>
13+
14+
export class Database extends Context.Service<Database, DatabaseShape>()("@opencode/v2/session/StorageSql/Database") {}
15+
16+
export const databaseLayer = Layer.unwrap(
17+
Effect.sync(() => {
18+
const filename = LegacyDatabase.getPath()
19+
return Layer.effect(
20+
Database,
21+
Effect.gen(function* () {
22+
const db = yield* makeDatabase
23+
yield* db.run("PRAGMA journal_mode = WAL")
24+
yield* db.run("PRAGMA synchronous = NORMAL")
25+
yield* db.run("PRAGMA busy_timeout = 5000")
26+
yield* db.run("PRAGMA cache_size = -64000")
27+
yield* db.run("PRAGMA foreign_keys = ON")
28+
yield* db.run("PRAGMA wal_checkpoint(PASSIVE)")
29+
yield* EffectDrizzleSqlite.migrateFromJournal(db, LegacyDatabase.migrationJournal())
30+
return db
31+
}),
32+
).pipe(Layer.provide(SqliteClient.layer({ filename, disableWAL: filename === ":memory:" })))
33+
}),
34+
)
935

1036
export const layer = Layer.effect(
1137
SessionStorage.Service,
1238
Effect.gen(function* () {
13-
const get: SessionStorage.Interface["get"] = Effect.fn("SessionStorageSql.get")((sessionID) =>
14-
attempt(() =>
15-
Database.use((db) => db.select().from(SessionTable).where(eq(SessionTable.id, sessionID)).get()),
16-
).pipe(Effect.map((row) => (row ? fromSessionRow(row) : undefined))),
17-
)
39+
const db = yield* Database
1840

19-
const list: SessionStorage.Interface["list"] = Effect.fn("SessionStorageSql.list")((input) =>
20-
attempt(() => {
21-
const direction = input.cursor?.direction ?? "next"
22-
const order = SessionStorage.pageOrder(input.order ?? "desc", direction)
23-
const sortColumn = SessionTable.time_updated
24-
const conditions: SQL[] = []
25-
if (input.directory) conditions.push(eq(SessionTable.directory, input.directory))
26-
if (input.path)
27-
conditions.push(or(eq(SessionTable.path, input.path), like(SessionTable.path, `${input.path}/%`))!)
28-
if (input.workspaceID) conditions.push(eq(SessionTable.workspace_id, input.workspaceID))
29-
if (input.roots) conditions.push(isNull(SessionTable.parent_id))
30-
if (input.start) conditions.push(gte(sortColumn, input.start))
31-
if (input.search) conditions.push(like(SessionTable.title, `%${input.search}%`))
32-
if (input.cursor) conditions.push(sessionCursorBoundary(input.cursor, order))
33-
34-
return Database.use((db) => {
35-
const query = db
36-
.select()
37-
.from(SessionTable)
38-
.where(conditions.length > 0 ? and(...conditions) : undefined)
39-
.orderBy(
40-
order === "asc" ? asc(sortColumn) : desc(sortColumn),
41-
order === "asc" ? asc(SessionTable.id) : desc(SessionTable.id),
42-
)
43-
const rows = input.limit === undefined ? query.all() : query.limit(input.limit).all()
44-
return direction === "previous" ? rows.toReversed() : rows
45-
})
46-
}).pipe(Effect.map((rows) => rows.map(fromSessionRow))),
47-
)
41+
const get: SessionStorage.Interface["get"] = Effect.fn("SessionStorageSql.get")(function* (sessionID) {
42+
const row = yield* attempt(db.select().from(SessionTable).where(eq(SessionTable.id, sessionID)).get())
43+
return row ? fromSessionRow(row) : undefined
44+
})
4845

49-
const messages: SessionStorage.Interface["messages"] = Effect.fn("SessionStorageSql.messages")((input) =>
50-
attempt(() => {
51-
const direction = input.cursor?.direction ?? "next"
52-
const order = SessionStorage.pageOrder(input.order ?? "desc", direction)
53-
const boundary = input.cursor ? messageCursorBoundary(input.cursor, order) : undefined
54-
const where = boundary
55-
? and(eq(SessionMessageTable.session_id, input.sessionID), boundary)
56-
: eq(SessionMessageTable.session_id, input.sessionID)
57-
58-
return Database.use((db) => {
59-
const query = db
60-
.select()
61-
.from(SessionMessageTable)
62-
.where(where)
63-
.orderBy(
64-
order === "asc" ? asc(SessionMessageTable.time_created) : desc(SessionMessageTable.time_created),
65-
order === "asc" ? asc(SessionMessageTable.id) : desc(SessionMessageTable.id),
66-
)
67-
const rows = input.limit === undefined ? query.all() : query.limit(input.limit).all()
68-
return direction === "previous" ? rows.toReversed() : rows
69-
})
70-
}).pipe(Effect.map((rows) => rows.map((row) => decodeMessage({ ...row.data, id: row.id, type: row.type })))),
71-
)
46+
const list: SessionStorage.Interface["list"] = Effect.fn("SessionStorageSql.list")(function* (input) {
47+
const direction = input.cursor?.direction ?? "next"
48+
const order = SessionStorage.pageOrder(input.order ?? "desc", direction)
49+
const sortColumn = SessionTable.time_updated
50+
const conditions: SQL[] = []
51+
if (input.directory) conditions.push(eq(SessionTable.directory, input.directory))
52+
if (input.path)
53+
conditions.push(or(eq(SessionTable.path, input.path), like(SessionTable.path, `${input.path}/%`))!)
54+
if (input.workspaceID) conditions.push(eq(SessionTable.workspace_id, input.workspaceID))
55+
if (input.roots) conditions.push(isNull(SessionTable.parent_id))
56+
if (input.start) conditions.push(gte(sortColumn, input.start))
57+
if (input.search) conditions.push(like(SessionTable.title, `%${input.search}%`))
58+
if (input.cursor) conditions.push(sessionCursorBoundary(input.cursor, order))
59+
60+
const query = db
61+
.select()
62+
.from(SessionTable)
63+
.where(conditions.length > 0 ? and(...conditions) : undefined)
64+
.orderBy(
65+
order === "asc" ? asc(sortColumn) : desc(sortColumn),
66+
order === "asc" ? asc(SessionTable.id) : desc(SessionTable.id),
67+
)
68+
const rows = yield* attempt(input.limit === undefined ? query : query.limit(input.limit))
69+
return (direction === "previous" ? rows.toReversed() : rows).map(fromSessionRow)
70+
})
71+
72+
const messages: SessionStorage.Interface["messages"] = Effect.fn("SessionStorageSql.messages")(function* (input) {
73+
const direction = input.cursor?.direction ?? "next"
74+
const order = SessionStorage.pageOrder(input.order ?? "desc", direction)
75+
const boundary = input.cursor ? messageCursorBoundary(input.cursor, order) : undefined
76+
const where = boundary
77+
? and(eq(SessionMessageTable.session_id, input.sessionID), boundary)
78+
: eq(SessionMessageTable.session_id, input.sessionID)
79+
80+
const query = db
81+
.select()
82+
.from(SessionMessageTable)
83+
.where(where)
84+
.orderBy(
85+
order === "asc" ? asc(SessionMessageTable.time_created) : desc(SessionMessageTable.time_created),
86+
order === "asc" ? asc(SessionMessageTable.id) : desc(SessionMessageTable.id),
87+
)
88+
const rows = yield* attempt(input.limit === undefined ? query : query.limit(input.limit))
89+
return (direction === "previous" ? rows.toReversed() : rows).map((row) =>
90+
decodeMessage({ ...row.data, id: row.id, type: row.type }),
91+
)
92+
})
7293

7394
const context: SessionStorage.Interface["context"] = Effect.fn("SessionStorageSql.context")((sessionID) =>
74-
attempt(() =>
75-
Database.use((db) => {
76-
const compaction = db
95+
Effect.gen(function* () {
96+
const compaction = yield* attempt(
97+
db
7798
.select()
7899
.from(SessionMessageTable)
79100
.where(and(eq(SessionMessageTable.session_id, sessionID), eq(SessionMessageTable.type, "compaction")))
80101
.orderBy(desc(SessionMessageTable.time_created), desc(SessionMessageTable.id))
81102
.limit(1)
82-
.get()
103+
.get(),
104+
)
83105

84-
return db
106+
const rows = yield* attempt(
107+
db
85108
.select()
86109
.from(SessionMessageTable)
87110
.where(
@@ -98,23 +121,25 @@ export const layer = Layer.effect(
98121
: undefined,
99122
),
100123
)
101-
.orderBy(asc(SessionMessageTable.time_created), asc(SessionMessageTable.id))
102-
.all()
103-
}),
104-
).pipe(Effect.map((rows) => rows.map((row) => decodeMessage({ ...row.data, id: row.id, type: row.type })))),
124+
.orderBy(asc(SessionMessageTable.time_created), asc(SessionMessageTable.id)),
125+
)
126+
127+
return rows.map((row) => decodeMessage({ ...row.data, id: row.id, type: row.type }))
128+
}),
105129
)
106130

107131
return SessionStorage.Service.of({ get, list, messages, context })
108132
}),
109133
)
110134

111-
export const defaultLayer = layer
135+
export const defaultLayer = layer.pipe(Layer.provide(databaseLayer.pipe(Layer.orDie)))
112136

113-
function attempt<A>(body: () => A) {
114-
return Effect.try({
115-
try: body,
116-
catch: (cause) => new SessionStorage.StorageError({ message: "Session storage SQL operation failed", cause }),
117-
})
137+
function attempt<A, E, R>(effect: Effect.Effect<A, E, R>) {
138+
return effect.pipe(
139+
Effect.mapError(
140+
(cause) => new SessionStorage.StorageError({ message: "Session storage SQL operation failed", cause }),
141+
),
142+
)
118143
}
119144

120145
function sessionCursorBoundary(cursor: SessionStorage.SessionCursor, order: SessionStorage.SortOrder) {

0 commit comments

Comments
 (0)