1/**
2 * Connects Effect SQL to SQLite storage inside Cloudflare Durable Objects.
3 *
4 * Provides `SqliteClient` and the generic `SqlClient` service. Pass `db` for
5 * queries only, or `storage` for transactions and migrations. SQLite blobs are
6 * returned as `Uint8Array`; `updateValues` is unsupported.
7 *
8 * The outer transaction holds the connection semaphore until storage completes.
9 * Nested `withTransaction` calls reuse that connection and call
10 * `storage.transaction()` without emitting transaction SQL. Child failures and
11 * interruptions roll back the child; uncaught failures also roll back the parent.
12 *
13 * Concurrent sibling transactions are unsupported. Fibers inheriting the
14 * transaction context must finish their transaction work before the enclosing
15 * transaction exits; later use can bypass the connection semaphore.
16 *
17 * @stability unstable
18 * @since 4.0.0
19 */
20import type { DurableObjectStorage, SqlStorage } from "@cloudflare/workers-types"
21import * as Cause from "effect/Cause"
22import * as Config from "effect/Config"
23import * as Context from "effect/Context"
24import * as Effect from "effect/Effect"
25import * as Exit from "effect/Exit"
26import * as Fiber from "effect/Fiber"
27import { identity } from "effect/Function"
28import * as Layer from "effect/Layer"
29import * as Reactivity from "effect/reactivity/Reactivity"
30import * as Rec from "effect/Record"
31import * as Scheduler from "effect/Scheduler"
32import * as Scope from "effect/Scope"
33import * as Semaphore from "effect/Semaphore"
34import * as Client from "effect/sql/SqlClient"
35import type { Connection } from "effect/sql/SqlConnection"
36import { classifySqliteError, SqlError, UnknownError } from "effect/sql/SqlError"
37import * as Statement from "effect/sql/Statement"
38import * as Stream from "effect/Stream"
39
40const ATTR_DB_SYSTEM_NAME = "db.system.name"
41
42const classifyError = (cause: unknown, message: string, operation: string) =>
43 classifySqliteError(cause, { message, operation })
44
45/**
46 * Runtime type identifier used to mark Cloudflare Durable Object `SqliteClient` values.
47 *
48 * @stability unstable
49 * @category type IDs
50 * @since 4.0.0
51 */
52export const TypeId: TypeId = "~@effect/sql-sqlite-do/SqliteClient"
53
54/**
55 * Type-level identifier used to mark Cloudflare Durable Object `SqliteClient` values.
56 *
57 * @stability unstable
58 * @category type IDs
59 * @since 4.0.0
60 */
61export type TypeId = "~@effect/sql-sqlite-do/SqliteClient"
62
63/**
64 * Cloudflare Durable Object SQLite client service, extending `SqlClient` with its configuration. `updateValues` is not supported.
65 *
66 * @stability unstable
67 * @category services
68 * @since 4.0.0
69 */
70export interface SqliteClient extends Client.SqlClient {
71 readonly [TypeId]: TypeId
72 readonly config: SqliteClientConfig
73
74 /** Not supported in sqlite */
75 readonly updateValues: never
76}
77
78/**
79 * Service tag for the Cloudflare Durable Object SQLite client service.
80 *
81 * **When to use**
82 *
83 * Use to access or provide a Durable Object SQLite client through the Effect
84 * context.
85 *
86 * @stability unstable
87 * @category services
88 * @since 4.0.0
89 */
90export const SqliteClient = Context.Service<SqliteClient>("@effect/sql-sqlite-do/SqliteClient")
91
92const SqliteTransaction = Context.Service<Client.TransactionConnection, Client.TransactionConnection.Service>(
93 "@effect/sql-sqlite-do/SqliteClient/SqliteTransaction"
94)
95
96/**
97 * Configuration for a Cloudflare Durable Object SQLite client, including either a `SqlStorage` handle or the full `DurableObjectStorage` for transaction support, span attributes, and query/result name transforms.
98 *
99 * @stability unstable
100 * @category models
101 * @since 4.0.0
102 */
103export interface SqliteClientConfig {
104 readonly db?: SqlStorage | undefined
105 readonly storage?: DurableObjectStorage | undefined
106 readonly spanAttributes?: Record<string, unknown> | undefined
107
108 readonly transformResultNames?: ((str: string) => string) | undefined
109 readonly transformQueryNames?: ((str: string) => string) | undefined
110}
111
112const unsupportedTransaction = (message: string, operation: string) =>
113 new SqlError({
114 reason: new UnknownError({
115 cause: new Error(message),
116 message,
117 operation
118 })
119 })
120
121const makeUnsupportedWithTransaction =
122 (message: string): Client.SqlClient["withTransaction"] =>
123 <R, E, A>(_effect: Effect.Effect<A, E, R>): Effect.Effect<A, E | SqlError, R> =>
124 Effect.fail(unsupportedTransaction(message, "transaction"))
125
126const makeStorageBackedWithTransaction = (
127 storage: DurableObjectStorage,
128 connection: Connection,
129 semaphore: Semaphore.Semaphore
130): Client.SqlClient["withTransaction"] =>
131<R, E, A>(effect: Effect.Effect<A, E, R>): Effect.Effect<A, E | SqlError, R> =>
132 Effect.withFiber((fiber) => {
133 const services = fiber.context
134 const connOption = Context.getOption(services, SqliteTransaction)
135 if (connOption._tag === "Some" && connOption.value[0] !== connection) {
136 return Effect.fail(
137 unsupportedTransaction(
138 "Transactions cannot use a connection from a different SQLite client",
139 "transaction"
140 )
141 )
142 }
143 const depth = connOption._tag === "Some" ? connOption.value[1] + 1 : 0
144
145 const effectWithTxn = Effect.provideContext(
146 effect,
147 Context.add(services, SqliteTransaction, [connection, depth] as const)
148 )
149
150 const transaction = Effect.callback<A, E | SqlError, R>((resume) => {
151 let interrupted = false
152 const promise = storage.transaction((txn) =>
153 new Promise<void>((resolve) => {
154 if (interrupted) return resolve()
155 resume(Effect.onExit(effectWithTxn, (exit) => {
156 if (Exit.isFailure(exit)) {
157 txn.rollback()
158 }
159 // Throwing from a child callback can abort its parent.
160 resolve()
161 return Effect.flatten(Effect.promise(() => promise))
162 }))
163 })
164 ).then(
165 () => Exit.void,
166 (cause) => {
167 const exit = Exit.fail(new SqlError({ reason: classifyError(cause, "Failed transaction", "transaction") }))
168 // Report rejection before the transaction callback starts; later resumes are ignored.
169 resume(exit)
170 return exit
171 }
172 )
173 return Effect.suspend(() => {
174 interrupted = true
175 return Effect.asVoid(Effect.promise(() => promise))
176 })
177 })
178 return connOption._tag === "Some" ? transaction : semaphore.withPermits(1)(transaction)
179 }).pipe(
180 // The storage transaction closes the input gate, blocking dispatcher tasks until it completes.
181 Effect.provideService(Scheduler.PreventSchedulerYield, true)
182 )
183
184/**
185 * Creates a scoped Cloudflare Durable Object SQLite client around Durable Object SQLite storage, serializing access and converting returned `ArrayBuffer` values to `Uint8Array`.
186 *
187 * @stability unstable
188 * @category constructors
189 * @since 4.0.0
190 */
191export const make = (
192 options: SqliteClientConfig
193): Effect.Effect<SqliteClient, never, Scope.Scope | Reactivity.Reactivity> =>
194 Effect.gen(function*() {
195 const compiler = Statement.makeCompilerSqlite(options.transformQueryNames)
196 const transformRows = options.transformResultNames
197 ? Statement.defaultTransforms(options.transformResultNames).array
198 : undefined
199 const db = options.storage?.sql ?? options.db
200
201 if (db === undefined) {
202 return yield* Effect.die("SqliteClient.make requires either a Durable Object storage or sql storage")
203 }
204 const sqlStorage = db
205
206 const makeConnection = Effect.gen(function*() {
207 function* runIterator(
208 sql: string,
209 params: ReadonlyArray<unknown> = []
210 ) {
211 const cursor = sqlStorage.exec(sql, ...params)
212 const columns = cursor.columnNames
213 for (const result of cursor.raw()) {
214 const obj: any = {}
215 for (let i = 0; i < columns.length; i++) {
216 const value = result[i]
217 Rec.assignProperty(obj, columns[i], value instanceof ArrayBuffer ? new Uint8Array(value) : value)
218 }
219 yield obj
220 }
221 }
222
223 const runStatement = (
224 sql: string,
225 params: ReadonlyArray<unknown> = []
226 ): Effect.Effect<ReadonlyArray<any>, SqlError, never> =>
227 Effect.try({
228 try: () => Array.from(runIterator(sql, params)),
229 catch: (cause) => new SqlError({ reason: classifyError(cause, "Failed to execute statement", "execute") })
230 })
231
232 const runValues = (
233 sql: string,
234 params: ReadonlyArray<unknown> = []
235 ): Effect.Effect<ReadonlyArray<any>, SqlError, never> =>
236 Effect.try({
237 try: () =>
238 Array.from(sqlStorage.exec(sql, ...params).raw(), (row) => {
239 for (let i = 0; i < row.length; i++) {
240 const value = row[i]
241 if (value instanceof ArrayBuffer) {
242 row[i] = new Uint8Array(value) as any
243 }
244 }
245 return row
246 }),
247 catch: (cause) => new SqlError({ reason: classifyError(cause, "Failed to execute statement", "execute") })
248 })
249
250 return identity<Connection>({
251 execute(sql, params, transformRows) {
252 return transformRows
253 ? Effect.map(runStatement(sql, params), transformRows)
254 : runStatement(sql, params)
255 },
256 executeRaw(sql, params) {
257 return runStatement(sql, params)
258 },
259 executeValues(sql, params) {
260 return runValues(sql, params)
261 },
262 executeValuesUnprepared(sql, params) {
263 return runValues(sql, params)
264 },
265 executeUnprepared(sql, params, transformRows) {
266 return transformRows
267 ? Effect.map(runStatement(sql, params), transformRows)
268 : runStatement(sql, params)
269 },
270 executeStream(sql, params, transformRows) {
271 return Stream.suspend(() => {
272 const iterator = runIterator(sql, params)
273 return Stream.fromIteratorSucceed(iterator, 128)
274 }).pipe(
275 Stream.catchCauseFilter(Cause.findDefect, (defect) =>
276 Stream.fail(
277 new SqlError({
278 reason: classifyError(defect, "Failed to execute statement", "execute")
279 })
280 )),
281 transformRows
282 ? Stream.mapArray((chunk) => transformRows(chunk) as any)
283 : identity
284 )
285 }
286 })
287 })
288
289 const semaphore = yield* Semaphore.make(1)
290 const connection = yield* makeConnection
291
292 const acquirer = semaphore.withPermits(1)(Effect.succeed(connection))
293 const transactionAcquirer = Effect.uninterruptibleMask((restore) => {
294 const fiber = Fiber.getCurrent()!
295 const scope = Context.getUnsafe(fiber.context, Scope.Scope)
296 return Effect.as(
297 Effect.tap(
298 restore(semaphore.take(1)),
299 () => Scope.addFinalizer(scope, semaphore.release(1))
300 ),
301 connection
302 )
303 })
304
305 const client = (yield* Client.make({
306 acquirer,
307 compiler,
308 transactionAcquirer,
309 transactionService: SqliteTransaction,
310 spanAttributes: [
311 ...(options.spanAttributes ? Object.entries(options.spanAttributes) : []),
312 [ATTR_DB_SYSTEM_NAME, "sqlite"]
313 ],
314 transformRows
315 })) as SqliteClient
316
317 return Object.assign(client, {
318 [TypeId]: TypeId as TypeId,
319 config: options,
320 withTransaction: options.storage
321 ? makeStorageBackedWithTransaction(options.storage, connection, semaphore)
322 : makeUnsupportedWithTransaction(
323 "Transactions require Durable Object storage; pass ctx.storage as the storage option"
324 )
325 })
326 })
327
328/**
329 * Creates a layer from a `Config`-wrapped Durable Object SQLite client configuration, providing both `SqliteClient` and `SqlClient`.
330 *
331 * @stability unstable
332 * @category layers
333 * @since 4.0.0
334 */
335export const layerConfig = (
336 config: Config.Wrap<SqliteClientConfig>
337): Layer.Layer<SqliteClient | Client.SqlClient, Config.ConfigError> =>
338 Layer.effectContext(
339 Config.unwrap(config).pipe(
340 Effect.flatMap(make),
341 Effect.map((client) =>
342 Context.make(SqliteClient, client).pipe(
343 Context.add(Client.SqlClient, client)
344 )
345 )
346 )
347 ).pipe(Layer.provide(Reactivity.layer))
348
349/**
350 * Creates a layer from a concrete Durable Object SQLite client configuration, providing both `SqliteClient` and `SqlClient`.
351 *
352 * @stability unstable
353 * @category layers
354 * @since 4.0.0
355 */
356export const layer = (
357 config: SqliteClientConfig
358): Layer.Layer<SqliteClient | Client.SqlClient> =>
359 Layer.effectContext(
360 Effect.map(make(config), (client) =>
361 Context.make(SqliteClient, client).pipe(
362 Context.add(Client.SqlClient, client)
363 ))
364 ).pipe(Layer.provide(Reactivity.layer))