1/**
2 * Cloudflare D1 client implementation for Effect SQL, backed by a Workers `D1Database` binding.
3 *
4 * This module adapts a Cloudflare D1 database binding into both the
5 * D1-specific `D1Client` service and the generic Effect `SqlClient` service.
6 * It uses the SQLite statement compiler, caches prepared statements, maps D1
7 * failures to `SqlError`, and provides direct or config-backed layers.
8 * Transactions, streaming queries, and `updateValues` are not supported by this
9 * driver.
10 *
11 * @stability unstable
12 * @since 4.0.0
13 */
14import type { D1Database, D1PreparedStatement, D1Result } from "@cloudflare/workers-types"
15import * as Cache from "effect/Cache"
16import * as Config from "effect/Config"
17import * as Context from "effect/Context"
18import * as Duration from "effect/Duration"
19import * as Effect from "effect/Effect"
20import { identity } from "effect/Function"
21import * as Layer from "effect/Layer"
22import * as Reactivity from "effect/reactivity/Reactivity"
23import type * as Scope from "effect/Scope"
24import * as Client from "effect/sql/SqlClient"
25import type { Connection } from "effect/sql/SqlConnection"
26import { SqlError, UnknownError } from "effect/sql/SqlError"
27import * as Statement from "effect/sql/Statement"
28import * as Stream from "effect/Stream"
29
30const ATTR_DB_SYSTEM_NAME = "db.system.name"
31const ATTR_DB_OPERATION_NAME = "db.operation.name"
32const ATTR_DB_QUERY_TEXT = "db.query.text"
33
34const classifyError = (cause: unknown, message: string, operation: string) =>
35 new UnknownError({ cause, message, operation })
36
37/**
38 * Unique runtime identifier used to tag `D1Client` values.
39 *
40 * @stability unstable
41 * @category type IDs
42 * @since 4.0.0
43 */
44export const TypeId: TypeId = "~@effect/sql-d1/D1Client"
45
46/**
47 * Type-level literal for the `D1Client` runtime identifier.
48 *
49 * @stability unstable
50 * @category type IDs
51 * @since 4.0.0
52 */
53export type TypeId = "~@effect/sql-d1/D1Client"
54
55/**
56 * Cloudflare D1 SQL client service, extending `SqlClient` with its D1 configuration and no `updateValues` support.
57 *
58 * @stability unstable
59 * @category services
60 * @since 4.0.0
61 */
62export interface D1Client extends Client.SqlClient {
63 readonly [TypeId]: TypeId
64 readonly config: D1ClientConfig
65
66 /**
67 * Executes SQL statements as a single atomic D1 batch and returns their row results in order.
68 *
69 * **When to use**
70 *
71 * Use when you have a fixed collection of statements that should run in one
72 * request and roll back together if any statement fails.
73 *
74 * **Gotchas**
75 *
76 * Each statement uses the query and result name transformations from the
77 * client that created it. Mixing clients can produce differently shaped row
78 * results within the same batch.
79 *
80 * @since 4.0.0
81 */
82 readonly batch: <const Statements extends ReadonlyArray<Statement.Statement<any>>>(
83 statements: Statements
84 ) => Effect.Effect<
85 {
86 readonly [K in keyof Statements]: Effect.Success<Statements[K]>
87 },
88 SqlError
89 >
90
91 /** Not supported in d1 */
92 readonly updateValues: never
93}
94
95/**
96 * Service tag for the Cloudflare D1 SQL client.
97 *
98 * **When to use**
99 *
100 * Use to access or provide a Cloudflare D1 SQL client through the Effect
101 * context.
102 *
103 * @stability unstable
104 * @category services
105 * @since 4.0.0
106 */
107export const D1Client = Context.Service<D1Client>("@effect/sql-d1/D1Client")
108
109/**
110 * Configuration for a Cloudflare D1 client, including the `D1Database`, prepared statement cache settings, span attributes, and query/result name transforms.
111 *
112 * @stability unstable
113 * @category models
114 * @since 4.0.0
115 */
116export interface D1ClientConfig {
117 readonly db: D1Database
118 readonly prepareCacheSize?: number | undefined
119 readonly prepareCacheTTL?: Duration.Input | undefined
120 readonly spanAttributes?: Record<string, unknown> | undefined
121
122 readonly transformResultNames?: ((str: string) => string) | undefined
123 readonly transformQueryNames?: ((str: string) => string) | undefined
124}
125
126type TransformRows = <A extends object>(rows: ReadonlyArray<A>) => ReadonlyArray<A>
127
128type BatchResults<Statements extends ReadonlyArray<Statement.Statement<any>>> = {
129 readonly [K in keyof Statements]: Effect.Success<Statements[K]>
130}
131
132interface StatementWithTransformRows extends Statement.Statement<any> {
133 readonly transformRows: TransformRows | undefined
134}
135
136const makeBatch = (options: {
137 readonly db: D1Database
138 readonly prepareCache: Cache.Cache<string, D1PreparedStatement, SqlError>
139 readonly spanAttributes: ReadonlyArray<readonly [string, unknown]>
140 readonly getClient: () => D1Client
141}): D1Client["batch"] =>
142<const Statements extends ReadonlyArray<Statement.Statement<any>>>(
143 statements: Statements
144) => {
145 if (statements.length === 0) {
146 return Effect.succeed([] as unknown as BatchResults<Statements>)
147 }
148 return Effect.useSpan(
149 "sql.execute",
150 { kind: "client" },
151 (span) =>
152 Effect.withFiber(Effect.fnUntraced(function*(fiber) {
153 const transformer = fiber.getRef(Statement.CurrentTransformer)
154 const prepared: Array<D1PreparedStatement> = []
155 const transforms: Array<TransformRows | undefined> = []
156 const queryTexts: Array<string> = []
157
158 for (const original of statements) {
159 const statement = transformer === undefined
160 ? original
161 : yield* transformer(original, options.getClient(), fiber, span)
162 const [sql, params] = statement.compile()
163 queryTexts.push(sql)
164 transforms.push((statement as StatementWithTransformRows).transformRows)
165 prepared.push((yield* Cache.get(options.prepareCache, sql)).bind(...params))
166 }
167
168 for (const [key, value] of options.spanAttributes) {
169 span.attribute(key, value)
170 }
171 span.attribute(ATTR_DB_OPERATION_NAME, "batch")
172 span.attribute(ATTR_DB_QUERY_TEXT, queryTexts.join("; "))
173
174 // D1 batches execute on the binding directly and intentionally cannot participate in SqlClient transactions.
175 const responses = yield* Effect.tryPromise({
176 try: () =>
177 options.db.batch<Record<string, unknown>>(prepared).then((responses) => {
178 for (const response of responses) {
179 if (response.error) {
180 throw response.error
181 }
182 }
183 return responses
184 }),
185 catch: (cause) => new SqlError({ reason: classifyError(cause, "Failed to execute batch", "execute") })
186 })
187
188 const results = responses.map((response, index) => {
189 const rows = response.results || []
190 const transformRows = transforms[index]
191 return transformRows ? transformRows(rows) : rows
192 })
193 return results as BatchResults<Statements>
194 }))
195 )
196}
197
198/**
199 * Creates a scoped Cloudflare D1 SQL client. Prepared statements are cached, while transactions and streaming queries are not supported by this driver.
200 *
201 * @stability unstable
202 * @category constructors
203 * @since 4.0.0
204 */
205export const make = (
206 options: D1ClientConfig
207): Effect.Effect<D1Client, never, Scope.Scope | Reactivity.Reactivity> =>
208 Effect.gen(function*() {
209 const compiler = Statement.makeCompilerSqlite(options.transformQueryNames)
210 const transformRows = options.transformResultNames ?
211 Statement.defaultTransforms(options.transformResultNames).array :
212 undefined
213 const spanAttributes: Array<readonly [string, unknown]> = [
214 ...(options.spanAttributes ? Object.entries(options.spanAttributes) : []),
215 [ATTR_DB_SYSTEM_NAME, "sqlite"]
216 ]
217
218 const makeConnection = Effect.gen(function*() {
219 const db = options.db
220
221 const prepareCache = yield* Cache.make({
222 capacity: options.prepareCacheSize ?? 200,
223 timeToLive: options.prepareCacheTTL ?? Duration.minutes(10),
224 lookup: (sql: string) =>
225 Effect.try({
226 try: () => db.prepare(sql),
227 catch: (cause) => new SqlError({ reason: classifyError(cause, "Failed to prepare statement", "prepare") })
228 })
229 })
230
231 const runStatementRaw = (
232 statement: D1PreparedStatement,
233 params: ReadonlyArray<unknown> = []
234 ): Effect.Effect<D1Result, SqlError, never> =>
235 Effect.tryPromise({
236 try: async () => {
237 const response = await statement.bind(...params).all()
238 if (response.error) {
239 throw response.error
240 }
241 return response
242 },
243 catch: (cause) => new SqlError({ reason: classifyError(cause, "Failed to execute statement", "execute") })
244 })
245
246 const runStatement = (
247 statement: D1PreparedStatement,
248 params: ReadonlyArray<unknown> = []
249 ): Effect.Effect<ReadonlyArray<any>, SqlError, never> =>
250 Effect.map(runStatementRaw(statement, params), (response) => response.results || [])
251
252 const runRaw = (
253 sql: string,
254 params: ReadonlyArray<unknown> = []
255 ) => runStatementRaw(db.prepare(sql), params)
256
257 const runCached = (
258 sql: string,
259 params: ReadonlyArray<unknown> = []
260 ) => Effect.flatMap(Cache.get(prepareCache, sql), (s) => runStatement(s, params))
261
262 const runUncached = (
263 sql: string,
264 params: ReadonlyArray<unknown> = []
265 ) => runStatement(db.prepare(sql), params)
266
267 const runValues = (
268 sql: string,
269 params: ReadonlyArray<unknown>
270 ) =>
271 Effect.flatMap(
272 Cache.get(prepareCache, sql),
273 (statement) =>
274 Effect.tryPromise({
275 try: () => {
276 return statement.bind(...params).raw() as Promise<
277 ReadonlyArray<
278 ReadonlyArray<unknown>
279 >
280 >
281 },
282 catch: (cause) => new SqlError({ reason: classifyError(cause, "Failed to execute statement", "execute") })
283 })
284 )
285
286 const runValuesUncached = (
287 sql: string,
288 params: ReadonlyArray<unknown>
289 ) =>
290 Effect.tryPromise({
291 try: () => {
292 return db.prepare(sql).bind(...params).raw() as Promise<
293 ReadonlyArray<
294 ReadonlyArray<unknown>
295 >
296 >
297 },
298 catch: (cause) => new SqlError({ reason: classifyError(cause, "Failed to execute statement", "execute") })
299 })
300
301 const connection = identity<Connection>({
302 execute(sql, params, transformRows) {
303 return transformRows
304 ? Effect.map(runCached(sql, params), transformRows)
305 : runCached(sql, params)
306 },
307 executeRaw(sql, params) {
308 return runRaw(sql, params)
309 },
310 executeValues(sql, params) {
311 return runValues(sql, params)
312 },
313 executeValuesUnprepared(sql, params) {
314 return runValuesUncached(sql, params)
315 },
316 executeUnprepared(sql, params, transformRows) {
317 return transformRows
318 ? Effect.map(runUncached(sql, params), transformRows)
319 : runUncached(sql, params)
320 },
321 executeStream(_sql, _params) {
322 return Stream.die("executeStream not implemented")
323 }
324 })
325 return { connection, prepareCache } as const
326 })
327
328 const { connection, prepareCache } = yield* makeConnection
329 const acquirer = Effect.succeed(connection)
330 const transactionAcquirer = Effect.die("transactions are not supported in D1")
331
332 let client!: D1Client
333 client = Object.assign(
334 (yield* Client.make({
335 acquirer,
336 compiler,
337 transactionAcquirer,
338 spanAttributes,
339 transformRows
340 })) as D1Client,
341 {
342 [TypeId]: TypeId as TypeId,
343 config: options,
344 batch: makeBatch({
345 db: options.db,
346 prepareCache,
347 spanAttributes,
348 getClient: () => client
349 })
350 }
351 )
352
353 if (options.transformQueryNames !== undefined || transformRows !== undefined) {
354 const clientWithoutTransformsBase = yield* Client.make({
355 acquirer: Effect.succeed(connection),
356 compiler: compiler.withoutTransform,
357 transactionAcquirer,
358 spanAttributes,
359 transformRows: undefined
360 })
361 let clientWithoutTransforms!: D1Client
362 clientWithoutTransforms = Object.assign(clientWithoutTransformsBase as D1Client, {
363 [TypeId]: TypeId as TypeId,
364 config: options,
365 batch: makeBatch({
366 db: options.db,
367 prepareCache,
368 spanAttributes,
369 getClient: () => clientWithoutTransforms
370 }),
371 withoutTransforms: () => clientWithoutTransforms
372 })
373 Object.assign(client, {
374 withoutTransforms: () => clientWithoutTransforms
375 })
376 }
377
378 return client
379 })
380
381/**
382 * Creates a layer from a `Config`-wrapped D1 client configuration, providing both `D1Client` and `SqlClient`.
383 *
384 * @stability unstable
385 * @category layers
386 * @since 4.0.0
387 */
388export const layerConfig = (
389 config: Config.Wrap<D1ClientConfig>
390): Layer.Layer<D1Client | Client.SqlClient, Config.ConfigError> =>
391 Layer.effectContext(
392 Config.unwrap(config).pipe(
393 Effect.flatMap(make),
394 Effect.map((client) =>
395 Context.make(D1Client, client).pipe(
396 Context.add(Client.SqlClient, client)
397 )
398 )
399 )
400 ).pipe(Layer.provide(Reactivity.layer))
401
402/**
403 * Creates a layer from a concrete D1 client configuration, providing both `D1Client` and `SqlClient`.
404 *
405 * @stability unstable
406 * @category layers
407 * @since 4.0.0
408 */
409export const layer = (
410 config: D1ClientConfig
411): Layer.Layer<D1Client | Client.SqlClient, Config.ConfigError> =>
412 Layer.effectContext(
413 Effect.map(make(config), (client) =>
414 Context.make(D1Client, client).pipe(
415 Context.add(Client.SqlClient, client)
416 ))
417 ).pipe(Layer.provide(Reactivity.layer))