mirror of
https://github.com/anomalyco/opencode.git
synced 2026-08-11 12:10:01 -04:00
Compare commits
3 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 4d57e54327 | |||
| 3bbc3fc267 | |||
| 0f56ebdb28 |
@@ -25,6 +25,7 @@
|
|||||||
},
|
},
|
||||||
"imports": {
|
"imports": {
|
||||||
"#sqlite": {
|
"#sqlite": {
|
||||||
|
"workerd": "./src/database/sqlite.workerd.ts",
|
||||||
"bun": "./src/database/sqlite.bun.ts",
|
"bun": "./src/database/sqlite.bun.ts",
|
||||||
"node": "./src/database/sqlite.node.ts",
|
"node": "./src/database/sqlite.node.ts",
|
||||||
"default": "./src/database/sqlite.bun.ts"
|
"default": "./src/database/sqlite.bun.ts"
|
||||||
|
|||||||
@@ -1,8 +1,9 @@
|
|||||||
export * as Database from "./database"
|
export * as Database from "./database"
|
||||||
|
|
||||||
import { EffectDrizzleSqlite } from "@opencode-ai/effect-drizzle-sqlite"
|
import { EffectDrizzleSqlite } from "@opencode-ai/effect-drizzle-sqlite"
|
||||||
import { sqliteLayer } from "#sqlite"
|
import { sqliteLayer, supportsForeignKeyToggle, supportsTuningPragmas } from "#sqlite"
|
||||||
import { Context, Effect, Layer, Schema } from "effect"
|
import { Context, Effect, Layer, Schema } from "effect"
|
||||||
|
import type { SqlClient } from "effect/unstable/sql"
|
||||||
import { Global } from "@opencode-ai/util/global"
|
import { Global } from "@opencode-ai/util/global"
|
||||||
import { isAbsolute, join } from "path"
|
import { isAbsolute, join } from "path"
|
||||||
import { DatabaseMigration } from "./migration"
|
import { DatabaseMigration } from "./migration"
|
||||||
@@ -27,12 +28,15 @@ const databaseLayer = Layer.effect(
|
|||||||
Effect.gen(function* () {
|
Effect.gen(function* () {
|
||||||
const db = yield* makeDatabase
|
const db = yield* makeDatabase
|
||||||
|
|
||||||
yield* db.run("PRAGMA journal_mode = WAL")
|
if (supportsTuningPragmas) {
|
||||||
yield* db.run("PRAGMA synchronous = NORMAL")
|
yield* db.run("PRAGMA journal_mode = WAL")
|
||||||
yield* db.run("PRAGMA busy_timeout = 5000")
|
yield* db.run("PRAGMA synchronous = NORMAL")
|
||||||
yield* db.run("PRAGMA cache_size = -64000")
|
yield* db.run("PRAGMA busy_timeout = 5000")
|
||||||
yield* db.run("PRAGMA foreign_keys = ON")
|
yield* db.run("PRAGMA cache_size = -64000")
|
||||||
yield* db.run("PRAGMA wal_checkpoint(PASSIVE)")
|
yield* db.run("PRAGMA wal_checkpoint(PASSIVE)")
|
||||||
|
}
|
||||||
|
// Durable Object SQLite always enforces foreign keys and rejects the pragma.
|
||||||
|
if (supportsForeignKeyToggle) yield* db.run("PRAGMA foreign_keys = ON")
|
||||||
yield* DatabaseMigration.apply(db)
|
yield* DatabaseMigration.apply(db)
|
||||||
|
|
||||||
return { db }
|
return { db }
|
||||||
@@ -43,14 +47,19 @@ export function layer(options: Options = { path: ":memory:" }) {
|
|||||||
return Layer.unwrap(
|
return Layer.unwrap(
|
||||||
Effect.gen(function* () {
|
Effect.gen(function* () {
|
||||||
const global = yield* Global.Service
|
const global = yield* Global.Service
|
||||||
const provide = (filename: string) => databaseLayer.pipe(Layer.provide(sqliteLayer({ filename })))
|
|
||||||
const filename = options.path ?? ":memory:"
|
const filename = options.path ?? ":memory:"
|
||||||
if (filename === ":memory:" || isAbsolute(filename)) return provide(filename)
|
if (filename === ":memory:" || isAbsolute(filename)) return layerWith(sqliteLayer({ filename }))
|
||||||
return provide(join(global.data, filename))
|
return layerWith(sqliteLayer({ filename: join(global.data, filename) }))
|
||||||
}),
|
}),
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Builds the database service over an already-configured SqlClient layer for
|
||||||
|
// runtimes that receive database storage instead of opening a filesystem path.
|
||||||
|
export function layerWith(sqlite: Layer.Layer<SqlClient.SqlClient>) {
|
||||||
|
return databaseLayer.pipe(Layer.provide(sqlite))
|
||||||
|
}
|
||||||
|
|
||||||
export function configured(options?: Options) {
|
export function configured(options?: Options) {
|
||||||
return makeGlobalNode({ service: Service, layer: layer(options), deps: [Global.node] })
|
return makeGlobalNode({ service: Service, layer: layer(options), deps: [Global.node] })
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -2,6 +2,7 @@ export * as DatabaseMigration from "./migration"
|
|||||||
|
|
||||||
import { sql } from "drizzle-orm"
|
import { sql } from "drizzle-orm"
|
||||||
import { Effect, Semaphore } from "effect"
|
import { Effect, Semaphore } from "effect"
|
||||||
|
import { supportsForeignKeyToggle } from "#sqlite"
|
||||||
import type { EffectDrizzleSqlite } from "@opencode-ai/effect-drizzle-sqlite"
|
import type { EffectDrizzleSqlite } from "@opencode-ai/effect-drizzle-sqlite"
|
||||||
import { migrations } from "./migration.gen"
|
import { migrations } from "./migration.gen"
|
||||||
import schema from "./schema.gen"
|
import schema from "./schema.gen"
|
||||||
@@ -20,8 +21,10 @@ export type Migration = {
|
|||||||
export function apply(db: Database) {
|
export function apply(db: Database) {
|
||||||
return lock.withPermit(
|
return lock.withPermit(
|
||||||
Effect.gen(function* () {
|
Effect.gen(function* () {
|
||||||
|
// OpenCode owns the unprefixed table namespace. Embedders sharing this
|
||||||
|
// database may own underscore-prefixed tables, which bootstrap ignores.
|
||||||
const tables = yield* db.all<{ name: string }>(
|
const tables = yield* db.all<{ name: string }>(
|
||||||
sql`SELECT name FROM sqlite_master WHERE type = 'table' AND name NOT LIKE 'sqlite_%'`,
|
sql`SELECT name FROM sqlite_master WHERE type = 'table' AND name NOT LIKE 'sqlite_%' AND substr(name, 1, 1) <> '_'`,
|
||||||
)
|
)
|
||||||
if (tables.some((table) => table.name === "session" || table.name === "session_v2"))
|
if (tables.some((table) => table.name === "session" || table.name === "session_v2"))
|
||||||
return yield* applyOnly(db, migrations)
|
return yield* applyOnly(db, migrations)
|
||||||
@@ -103,9 +106,15 @@ export function applyOnly(db: Database, input: Migration[]) {
|
|||||||
})
|
})
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
yield* db.run(sql`PRAGMA foreign_keys = OFF`)
|
// Durable Object SQLite rejects the foreign_keys toggle; the closest
|
||||||
|
// allowlisted relaxation is deferring enforcement to transaction commit.
|
||||||
|
const relaxForeignKeys = supportsForeignKeyToggle
|
||||||
|
? db.run(sql`PRAGMA foreign_keys = OFF`)
|
||||||
|
: db.run(sql`PRAGMA defer_foreign_keys = ON`)
|
||||||
|
const restoreForeignKeys = supportsForeignKeyToggle ? db.run(sql`PRAGMA foreign_keys = ON`) : Effect.void
|
||||||
|
yield* relaxForeignKeys
|
||||||
yield* apply.pipe(
|
yield* apply.pipe(
|
||||||
Effect.ensuring(db.run(sql`PRAGMA foreign_keys = ON`).pipe(Effect.orDie)),
|
Effect.ensuring(restoreForeignKeys.pipe(Effect.orDie)),
|
||||||
Effect.tapError((error) =>
|
Effect.tapError((error) =>
|
||||||
Effect.logError("database migration failed", {
|
Effect.logError("database migration failed", {
|
||||||
migration: migration.id,
|
migration: migration.id,
|
||||||
|
|||||||
@@ -13,6 +13,11 @@ const ATTR_DB_SYSTEM_NAME = "db.system.name"
|
|||||||
const TypeId = "~@opencode-ai/core/database/SqliteBun" as const
|
const TypeId = "~@opencode-ai/core/database/SqliteBun" as const
|
||||||
type TypeId = typeof TypeId
|
type TypeId = typeof TypeId
|
||||||
|
|
||||||
|
export const supportsTuningPragmas = true
|
||||||
|
|
||||||
|
// Foreign keys default OFF and can be toggled per connection.
|
||||||
|
export const supportsForeignKeyToggle = true
|
||||||
|
|
||||||
interface SqliteClient extends SqlClient.SqlClient {
|
interface SqliteClient extends SqlClient.SqlClient {
|
||||||
readonly [TypeId]: TypeId
|
readonly [TypeId]: TypeId
|
||||||
readonly config: Config
|
readonly config: Config
|
||||||
|
|||||||
@@ -13,6 +13,11 @@ const ATTR_DB_SYSTEM_NAME = "db.system.name"
|
|||||||
const TypeId = "~@opencode-ai/core/database/SqliteNode" as const
|
const TypeId = "~@opencode-ai/core/database/SqliteNode" as const
|
||||||
type TypeId = typeof TypeId
|
type TypeId = typeof TypeId
|
||||||
|
|
||||||
|
export const supportsTuningPragmas = true
|
||||||
|
|
||||||
|
// Foreign keys default OFF and can be toggled per connection.
|
||||||
|
export const supportsForeignKeyToggle = true
|
||||||
|
|
||||||
interface SqliteClient extends SqlClient.SqlClient {
|
interface SqliteClient extends SqlClient.SqlClient {
|
||||||
readonly [TypeId]: TypeId
|
readonly [TypeId]: TypeId
|
||||||
readonly config: Config
|
readonly config: Config
|
||||||
|
|||||||
@@ -0,0 +1,254 @@
|
|||||||
|
import { drizzle } from "drizzle-orm/durable-sqlite"
|
||||||
|
import { Context, Effect, Exit, Fiber, Layer, Scope, Semaphore, Stream } from "effect"
|
||||||
|
import { identity } from "effect/Function"
|
||||||
|
import { Reactivity } from "effect/unstable/reactivity"
|
||||||
|
import { SqlClient, Statement } from "effect/unstable/sql"
|
||||||
|
import type { Connection } from "effect/unstable/sql/SqlConnection"
|
||||||
|
import { classifySqliteError, SqlError, UnknownError } from "effect/unstable/sql/SqlError"
|
||||||
|
import { Sqlite } from "./sqlite"
|
||||||
|
|
||||||
|
const ATTR_DB_SYSTEM_NAME = "db.system.name"
|
||||||
|
|
||||||
|
const TypeId = "~@opencode-ai/core/database/SqliteWorkerd" as const
|
||||||
|
type TypeId = typeof TypeId
|
||||||
|
|
||||||
|
// Durable Object SQLite only allowlists introspection pragmas; journal_mode,
|
||||||
|
// synchronous, busy_timeout, cache_size, and wal_checkpoint all throw, and
|
||||||
|
// foreign keys are already enforced by default (SQLITE_DEFAULT_FOREIGN_KEYS=1).
|
||||||
|
export const supportsTuningPragmas = false
|
||||||
|
|
||||||
|
// Durable Object SQLite rejects `PRAGMA foreign_keys`: enforcement is always
|
||||||
|
// on (SQLITE_DEFAULT_FOREIGN_KEYS=1) and only `defer_foreign_keys` is
|
||||||
|
// allowlisted for migrations that must relax checking inside a transaction.
|
||||||
|
export const supportsForeignKeyToggle = false
|
||||||
|
|
||||||
|
// Minimal structural types for the Durable Object storage API so this adapter
|
||||||
|
// does not depend on @cloudflare/workers-types (whose ambient globals conflict
|
||||||
|
// with @types/bun). Shapes match the SqlStorage and DurableObjectStorage docs.
|
||||||
|
type SqlStorageValue = ArrayBuffer | string | number | null
|
||||||
|
|
||||||
|
interface SqlStorageCursor {
|
||||||
|
readonly columnNames: Array<string>
|
||||||
|
raw(): IterableIterator<Array<SqlStorageValue>>
|
||||||
|
toArray(): Array<Record<string, SqlStorageValue>>
|
||||||
|
}
|
||||||
|
|
||||||
|
export interface SqlStorage {
|
||||||
|
exec(query: string, ...bindings: Array<unknown>): SqlStorageCursor
|
||||||
|
}
|
||||||
|
|
||||||
|
export interface DurableObjectStorage {
|
||||||
|
readonly sql: SqlStorage
|
||||||
|
transaction<T>(closure: (txn: { rollback(): void }) => Promise<T>): Promise<T>
|
||||||
|
transactionSync<T>(closure: () => T): T
|
||||||
|
}
|
||||||
|
|
||||||
|
interface SqliteClient extends SqlClient.SqlClient {
|
||||||
|
readonly [TypeId]: TypeId
|
||||||
|
readonly config: Config
|
||||||
|
readonly updateValues: never
|
||||||
|
}
|
||||||
|
|
||||||
|
interface Config {
|
||||||
|
readonly storage: DurableObjectStorage
|
||||||
|
readonly spanAttributes?: Record<string, unknown>
|
||||||
|
readonly transformResultNames?: (str: string) => string
|
||||||
|
readonly transformQueryNames?: (str: string) => string
|
||||||
|
}
|
||||||
|
|
||||||
|
// sql.exec() rejects BEGIN/COMMIT/SAVEPOINT, so SqlClient.make's default
|
||||||
|
// transaction SQL can never run. withTransaction is replaced below with a
|
||||||
|
// DurableObjectStorage.transaction-backed implementation; this service only
|
||||||
|
// tracks the active transaction connection for statements and nesting checks.
|
||||||
|
const WorkerdTransaction = Context.Service<SqlClient.TransactionConnection, SqlClient.TransactionConnection.Service>(
|
||||||
|
"@opencode-ai/core/database/SqliteWorkerdTransaction",
|
||||||
|
)
|
||||||
|
|
||||||
|
const transactionError = (message: string) =>
|
||||||
|
new SqlError({
|
||||||
|
reason: new UnknownError({ cause: new Error(message), message, operation: "transaction" }),
|
||||||
|
})
|
||||||
|
|
||||||
|
const makeWithTransaction =
|
||||||
|
(
|
||||||
|
storage: DurableObjectStorage,
|
||||||
|
connection: Connection,
|
||||||
|
semaphore: Semaphore.Semaphore,
|
||||||
|
): SqlClient.SqlClient["withTransaction"] =>
|
||||||
|
<A, E, R>(effect: Effect.Effect<A, E, R>): Effect.Effect<A, E | SqlError, R> =>
|
||||||
|
Effect.withFiber((fiber) => {
|
||||||
|
const services = fiber.context
|
||||||
|
if (Context.getOption(services, WorkerdTransaction)._tag === "Some")
|
||||||
|
return Effect.fail(
|
||||||
|
transactionError("Nested transactions are not supported by Cloudflare Durable Object SQLite storage"),
|
||||||
|
)
|
||||||
|
const effectWithTxn = Effect.provideContext(
|
||||||
|
effect,
|
||||||
|
Context.add(services, WorkerdTransaction, [connection, 0] as const),
|
||||||
|
)
|
||||||
|
return semaphore.withPermits(1)(
|
||||||
|
Effect.callback((resume) => {
|
||||||
|
let interrupted = false
|
||||||
|
const promise = storage
|
||||||
|
.transaction(
|
||||||
|
(txn) =>
|
||||||
|
new Promise<void>((resolve) => {
|
||||||
|
if (interrupted) return resolve()
|
||||||
|
resume(
|
||||||
|
Effect.onExit(effectWithTxn, (exit) => {
|
||||||
|
if (Exit.isFailure(exit)) txn.rollback()
|
||||||
|
resolve()
|
||||||
|
// wait for the transaction to complete
|
||||||
|
return Effect.promise(() => promise)
|
||||||
|
}),
|
||||||
|
)
|
||||||
|
}),
|
||||||
|
)
|
||||||
|
.catch((cause) =>
|
||||||
|
resume(
|
||||||
|
Effect.fail(
|
||||||
|
new SqlError({
|
||||||
|
reason: classifySqliteError(cause, { message: "Failed transaction", operation: "transaction" }),
|
||||||
|
}),
|
||||||
|
),
|
||||||
|
),
|
||||||
|
)
|
||||||
|
return Effect.suspend(() => {
|
||||||
|
interrupted = true
|
||||||
|
return Effect.promise(() => promise)
|
||||||
|
})
|
||||||
|
}),
|
||||||
|
)
|
||||||
|
})
|
||||||
|
|
||||||
|
const make = (options: Config) =>
|
||||||
|
Effect.gen(function* () {
|
||||||
|
const native = (yield* Sqlite.Native) as DurableObjectStorage
|
||||||
|
|
||||||
|
const compiler = Statement.makeCompilerSqlite(options.transformQueryNames)
|
||||||
|
const transformRows = options.transformResultNames
|
||||||
|
? Statement.defaultTransforms(options.transformResultNames).array
|
||||||
|
: undefined
|
||||||
|
|
||||||
|
// SqlClient.SafeIntegers is ignored: Durable Object SQLite has no bigint
|
||||||
|
// mode and always returns integers as numbers. Blobs come back as
|
||||||
|
// ArrayBuffer and are normalized to Uint8Array to match the other adapters.
|
||||||
|
function* runIterator(query: string, params: ReadonlyArray<unknown> = []) {
|
||||||
|
const cursor = native.sql.exec(query, ...params)
|
||||||
|
const columns = cursor.columnNames
|
||||||
|
for (const row of cursor.raw()) {
|
||||||
|
const record: Record<string, unknown> = {}
|
||||||
|
for (let i = 0; i < columns.length; i++) {
|
||||||
|
const value = row[i]
|
||||||
|
record[columns[i]] = value instanceof ArrayBuffer ? new Uint8Array(value) : value
|
||||||
|
}
|
||||||
|
yield record
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
const run = (query: string, params: ReadonlyArray<unknown> = []) =>
|
||||||
|
Effect.try({
|
||||||
|
try: () => Array.from(runIterator(query, params)),
|
||||||
|
catch: (cause) =>
|
||||||
|
new SqlError({
|
||||||
|
reason: classifySqliteError(cause, { message: "Failed to execute statement", operation: "execute" }),
|
||||||
|
}),
|
||||||
|
})
|
||||||
|
|
||||||
|
const runValues = (query: string, params: ReadonlyArray<unknown> = []) =>
|
||||||
|
Effect.try({
|
||||||
|
try: () =>
|
||||||
|
Array.from(native.sql.exec(query, ...params).raw(), (row) =>
|
||||||
|
row.map((value) => (value instanceof ArrayBuffer ? new Uint8Array(value) : value)),
|
||||||
|
),
|
||||||
|
catch: (cause) =>
|
||||||
|
new SqlError({
|
||||||
|
reason: classifySqliteError(cause, { message: "Failed to execute statement", operation: "execute" }),
|
||||||
|
}),
|
||||||
|
})
|
||||||
|
|
||||||
|
const connection = identity<Connection>({
|
||||||
|
execute(query, params, transformRows) {
|
||||||
|
return transformRows ? Effect.map(run(query, params), transformRows) : run(query, params)
|
||||||
|
},
|
||||||
|
executeRaw(query, params) {
|
||||||
|
return run(query, params)
|
||||||
|
},
|
||||||
|
executeValues(query, params) {
|
||||||
|
return runValues(query, params)
|
||||||
|
},
|
||||||
|
executeValuesUnprepared(query, params) {
|
||||||
|
return runValues(query, params)
|
||||||
|
},
|
||||||
|
executeUnprepared(query, params, transformRows) {
|
||||||
|
return this.execute(query, params, transformRows)
|
||||||
|
},
|
||||||
|
executeStream() {
|
||||||
|
return Stream.die("executeStream not implemented")
|
||||||
|
},
|
||||||
|
})
|
||||||
|
|
||||||
|
const semaphore = yield* Semaphore.make(1)
|
||||||
|
const acquirer = semaphore.withPermits(1)(Effect.succeed(connection))
|
||||||
|
const transactionAcquirer = Effect.uninterruptibleMask((restore) => {
|
||||||
|
const fiber = Fiber.getCurrent()!
|
||||||
|
const scope = Context.getUnsafe(fiber.context, Scope.Scope)
|
||||||
|
return Effect.as(
|
||||||
|
Effect.tap(restore(semaphore.take(1)), () => Scope.addFinalizer(scope, semaphore.release(1))),
|
||||||
|
connection,
|
||||||
|
)
|
||||||
|
})
|
||||||
|
|
||||||
|
const client = Object.assign(
|
||||||
|
(yield* SqlClient.make({
|
||||||
|
acquirer,
|
||||||
|
compiler,
|
||||||
|
transactionAcquirer,
|
||||||
|
transactionService: WorkerdTransaction,
|
||||||
|
spanAttributes: [
|
||||||
|
...(options.spanAttributes ? Object.entries(options.spanAttributes) : []),
|
||||||
|
[ATTR_DB_SYSTEM_NAME, "sqlite"],
|
||||||
|
],
|
||||||
|
transformRows,
|
||||||
|
})) as SqliteClient,
|
||||||
|
{
|
||||||
|
[TypeId]: TypeId,
|
||||||
|
config: options,
|
||||||
|
withTransaction: makeWithTransaction(native, connection, semaphore),
|
||||||
|
// Durable Object SQLite rejects BEGIN/COMMIT/SAVEPOINT; consumers such
|
||||||
|
// as the drizzle session must route through withTransaction instead.
|
||||||
|
transactionStatements: false,
|
||||||
|
},
|
||||||
|
)
|
||||||
|
|
||||||
|
return client
|
||||||
|
})
|
||||||
|
|
||||||
|
// Defends against the shared path-based Database.layer, which passes a
|
||||||
|
// filename instead of storage when resolved under the workerd condition.
|
||||||
|
const nativeLayer = (config: Config) =>
|
||||||
|
config.storage
|
||||||
|
? Layer.succeed(Sqlite.Native, config.storage)
|
||||||
|
: Layer.effect(
|
||||||
|
Sqlite.Native,
|
||||||
|
Effect.die(
|
||||||
|
"workerd sqlite cannot open a database from a path; use Database.layerWith(sqliteLayer({ storage }))",
|
||||||
|
),
|
||||||
|
)
|
||||||
|
|
||||||
|
const clientLayer = (config: Config) => Layer.effect(SqlClient.SqlClient, make(config))
|
||||||
|
|
||||||
|
const drizzleLayer = Layer.effect(
|
||||||
|
Sqlite.Drizzle,
|
||||||
|
Effect.gen(function* () {
|
||||||
|
const native = (yield* Sqlite.Native) as DurableObjectStorage
|
||||||
|
return drizzle(native) as unknown as Sqlite.DrizzleClient
|
||||||
|
}),
|
||||||
|
)
|
||||||
|
|
||||||
|
export const sqliteLayer = (config: Config) => {
|
||||||
|
const native = nativeLayer(config)
|
||||||
|
return Layer.merge(native, Layer.merge(clientLayer(config), drizzleLayer).pipe(Layer.provide(native))).pipe(
|
||||||
|
Layer.provide(Reactivity.layer),
|
||||||
|
)
|
||||||
|
}
|
||||||
@@ -84,6 +84,19 @@ describe("DatabaseMigration", () => {
|
|||||||
).rejects.toThrow("Database is not empty and has no session table")
|
).rejects.toThrow("Database is not empty and has no session table")
|
||||||
})
|
})
|
||||||
|
|
||||||
|
test("bootstraps alongside underscore-prefixed embedder tables", async () => {
|
||||||
|
await run(
|
||||||
|
Effect.gen(function* () {
|
||||||
|
const db = yield* makeDb
|
||||||
|
yield* db.run(sql`CREATE TABLE _embedder_state (id text PRIMARY KEY)`)
|
||||||
|
yield* DatabaseMigration.apply(db)
|
||||||
|
expect(yield* db.get(sql`SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'session_v2'`)).toEqual(
|
||||||
|
{ name: "session_v2" },
|
||||||
|
)
|
||||||
|
}),
|
||||||
|
)
|
||||||
|
})
|
||||||
|
|
||||||
test("applies generic migrations once and records their order", async () => {
|
test("applies generic migrations once and records their order", async () => {
|
||||||
await run(
|
await run(
|
||||||
Effect.gen(function* () {
|
Effect.gen(function* () {
|
||||||
|
|||||||
@@ -0,0 +1,137 @@
|
|||||||
|
import { describe, expect, test } from "bun:test"
|
||||||
|
import { Database } from "bun:sqlite"
|
||||||
|
import { Effect, Layer } from "effect"
|
||||||
|
import { SqlClient } from "effect/unstable/sql"
|
||||||
|
import { SqlError } from "effect/unstable/sql/SqlError"
|
||||||
|
import { sqliteLayer } from "@opencode-ai/core/database/sqlite.workerd"
|
||||||
|
import type { DurableObjectStorage } from "@opencode-ai/core/database/sqlite.workerd"
|
||||||
|
import { tempGlobalLayer } from "./fixture/global"
|
||||||
|
|
||||||
|
// Emulates the Durable Object storage API over bun:sqlite so the adapter can
|
||||||
|
// be verified without workerd or Cloudflare runtime dependencies.
|
||||||
|
const makeFakeStorage = () => {
|
||||||
|
const native = new Database(":memory:")
|
||||||
|
const toSqlStorageValue = (value: unknown) => {
|
||||||
|
if (!(value instanceof Uint8Array)) return value as ArrayBuffer | string | number | null
|
||||||
|
const buffer = new ArrayBuffer(value.byteLength)
|
||||||
|
new Uint8Array(buffer).set(value)
|
||||||
|
return buffer
|
||||||
|
}
|
||||||
|
const storage: DurableObjectStorage = {
|
||||||
|
sql: {
|
||||||
|
exec(query: string, ...bindings: Array<unknown>) {
|
||||||
|
const statement = native.query(query)
|
||||||
|
const rows = (statement.values(...(bindings as never[])) ?? []).map((row) => row.map(toSqlStorageValue))
|
||||||
|
const columnNames = statement.columnNames
|
||||||
|
return {
|
||||||
|
columnNames,
|
||||||
|
raw: () => rows[Symbol.iterator](),
|
||||||
|
toArray: () => rows.map((row) => Object.fromEntries(columnNames.map((name, i) => [name, row[i]]))),
|
||||||
|
}
|
||||||
|
},
|
||||||
|
},
|
||||||
|
transaction<T>(closure: (txn: { rollback(): void }) => Promise<T>): Promise<T> {
|
||||||
|
native.run("BEGIN")
|
||||||
|
let rolledBack = false
|
||||||
|
return closure({ rollback: () => (rolledBack = true) }).then(
|
||||||
|
(result) => {
|
||||||
|
native.run(rolledBack ? "ROLLBACK" : "COMMIT")
|
||||||
|
return result
|
||||||
|
},
|
||||||
|
(error) => {
|
||||||
|
native.run("ROLLBACK")
|
||||||
|
throw error
|
||||||
|
},
|
||||||
|
)
|
||||||
|
},
|
||||||
|
transactionSync<T>(closure: () => T): T {
|
||||||
|
return native.transaction(closure)()
|
||||||
|
},
|
||||||
|
}
|
||||||
|
return storage
|
||||||
|
}
|
||||||
|
|
||||||
|
const run = <A, E>(storage: DurableObjectStorage, effect: Effect.Effect<A, E, SqlClient.SqlClient>) =>
|
||||||
|
Effect.runPromise(effect.pipe(Effect.provide(sqliteLayer({ storage })), Effect.scoped))
|
||||||
|
|
||||||
|
describe("sqlite.workerd", () => {
|
||||||
|
test("executes statements with bindings and maps rows to records", async () => {
|
||||||
|
const rows = await run(
|
||||||
|
makeFakeStorage(),
|
||||||
|
Effect.gen(function* () {
|
||||||
|
const sql = yield* SqlClient.SqlClient
|
||||||
|
yield* sql`CREATE TABLE item (id INTEGER PRIMARY KEY, name TEXT NOT NULL)`
|
||||||
|
yield* sql`INSERT INTO item (id, name) VALUES (${1}, ${"one"}), (${2}, ${"two"})`
|
||||||
|
return yield* sql<{ id: number; name: string }>`SELECT id, name FROM item ORDER BY id`
|
||||||
|
}),
|
||||||
|
)
|
||||||
|
expect(rows).toEqual([
|
||||||
|
{ id: 1, name: "one" },
|
||||||
|
{ id: 2, name: "two" },
|
||||||
|
])
|
||||||
|
})
|
||||||
|
|
||||||
|
test("normalizes ArrayBuffer blob values to Uint8Array", async () => {
|
||||||
|
const rows = await run(
|
||||||
|
makeFakeStorage(),
|
||||||
|
Effect.gen(function* () {
|
||||||
|
const sql = yield* SqlClient.SqlClient
|
||||||
|
yield* sql`CREATE TABLE blob (data BLOB NOT NULL)`
|
||||||
|
yield* sql`INSERT INTO blob (data) VALUES (${new Uint8Array([1, 2, 3])})`
|
||||||
|
return yield* sql<{ data: Uint8Array }>`SELECT data FROM blob`
|
||||||
|
}),
|
||||||
|
)
|
||||||
|
expect(rows[0].data).toBeInstanceOf(Uint8Array)
|
||||||
|
expect(Array.from(rows[0].data)).toEqual([1, 2, 3])
|
||||||
|
})
|
||||||
|
|
||||||
|
test("withTransaction commits on success and rolls back on failure", async () => {
|
||||||
|
const storage = makeFakeStorage()
|
||||||
|
const count = await run(
|
||||||
|
storage,
|
||||||
|
Effect.gen(function* () {
|
||||||
|
const sql = yield* SqlClient.SqlClient
|
||||||
|
yield* sql`CREATE TABLE t (value TEXT NOT NULL)`
|
||||||
|
yield* sql.withTransaction(sql`INSERT INTO t (value) VALUES (${"kept"})`)
|
||||||
|
yield* sql
|
||||||
|
.withTransaction(
|
||||||
|
Effect.gen(function* () {
|
||||||
|
yield* sql`INSERT INTO t (value) VALUES (${"discarded"})`
|
||||||
|
return yield* Effect.fail("rollback")
|
||||||
|
}),
|
||||||
|
)
|
||||||
|
.pipe(Effect.ignore)
|
||||||
|
return yield* sql<{ count: number }>`SELECT count(*) AS count FROM t`
|
||||||
|
}),
|
||||||
|
)
|
||||||
|
expect(count[0].count).toBe(1)
|
||||||
|
})
|
||||||
|
|
||||||
|
test("nested withTransaction fails with SqlError", async () => {
|
||||||
|
const error = await run(
|
||||||
|
makeFakeStorage(),
|
||||||
|
Effect.gen(function* () {
|
||||||
|
const sql = yield* SqlClient.SqlClient
|
||||||
|
yield* sql`CREATE TABLE t (value TEXT NOT NULL)`
|
||||||
|
return yield* sql
|
||||||
|
.withTransaction(sql.withTransaction(sql`INSERT INTO t (value) VALUES (${"nested"})`))
|
||||||
|
.pipe(Effect.flip)
|
||||||
|
}),
|
||||||
|
)
|
||||||
|
expect(error).toBeInstanceOf(SqlError)
|
||||||
|
})
|
||||||
|
|
||||||
|
test("boots the full database layer with migrations over injected storage", async () => {
|
||||||
|
const storage = makeFakeStorage()
|
||||||
|
const core = await import("@opencode-ai/core/database/database")
|
||||||
|
await Effect.runPromise(
|
||||||
|
Effect.scoped(Layer.build(core.Database.layerWith(sqliteLayer({ storage })).pipe(Layer.provide(tempGlobalLayer)))),
|
||||||
|
)
|
||||||
|
const names = storage.sql
|
||||||
|
.exec("SELECT name FROM sqlite_master WHERE type = 'table' ORDER BY name")
|
||||||
|
.toArray()
|
||||||
|
.map((row) => row.name)
|
||||||
|
expect(names).toContain("migration")
|
||||||
|
expect(names).toContain("session_v2")
|
||||||
|
})
|
||||||
|
})
|
||||||
Reference in New Issue
Block a user