1/**
2 * Main SQL client service for tagged-template queries.
3 *
4 * `SqlClient` combines the tagged-template statement constructor with
5 * connection acquisition, dialect compilation, transactions, row transforms,
6 * tracing, and reactive query helpers. Driver integrations build this service
7 * from their connection and compiler pieces.
8 *
9 * @stability unstable
10 * @since 4.0.0
11 */
12import type * as Cause from "../Cause.ts"
13import { Clock } from "../Clock.ts"
14import * as Context from "../Context.ts"
15import * as Effect from "../Effect.ts"
16import * as Exit from "../Exit.ts"
17import { identity } from "../Function.ts"
18import * as Option from "../Option.ts"
19import { hasProperty } from "../Predicate.ts"
20import type * as Queue from "../Queue.ts"
21import { Reactivity } from "../reactivity/Reactivity.ts"
22import type { ReadonlyRecord } from "../Record.ts"
23import * as Scope from "../Scope.ts"
24import * as Semaphore from "../Semaphore.ts"
25import * as Stream from "../Stream.ts"
26import * as Tracer from "../Tracer.ts"
27import type { NoInfer } from "../Types.ts"
28import type * as Connection from "./SqlConnection.ts"
29import { ConnectionError, SqlError } from "./SqlError.ts"
30import type { Compiler, Constructor } from "./Statement.ts"
31import * as Statement from "./Statement.ts"
32
33const TypeId = "~effect/sql/SqlClient"
34
35/**
36 * SQL client service interface, combining the statement constructor API with
37 * connection reservation, transaction handling, and reactive query helpers.
38 *
39 * @stability unstable
40 * @category models
41 * @since 4.0.0
42 */
43export interface SqlClient extends Constructor {
44 readonly [TypeId]: typeof TypeId
45
46 /**
47 * Copy of the client for safeql etc.
48 */
49 readonly safe: this
50
51 /**
52 * Copy of the client without transformations.
53 */
54 readonly withoutTransforms: () => this
55
56 readonly reserve: Effect.Effect<Connection.Connection, SqlError, Scope.Scope>
57
58 /**
59 * With the given effect, ensure all sql queries are run in a transaction.
60 */
61 readonly withTransaction: <R, E, A>(
62 self: Effect.Effect<A, E, R>
63 ) => Effect.Effect<A, E | SqlError, R>
64
65 /**
66 * The transaction service for this client.
67 */
68 readonly transactionService: Context.Service<TransactionConnection, TransactionConnection.Service>
69
70 /**
71 * Use the Reactivity service to create a reactive query.
72 */
73 readonly reactive: <A, E, R>(
74 keys: ReadonlyArray<unknown> | ReadonlyRecord<string, ReadonlyArray<unknown>>,
75 effect: Effect.Effect<A, E, R>
76 ) => Stream.Stream<A, E, R>
77
78 /**
79 * Use the Reactivity service to create a reactive
80 * query.
81 */
82 readonly reactiveMailbox: <A, E, R>(
83 keys: ReadonlyArray<unknown> | ReadonlyRecord<string, ReadonlyArray<unknown>>,
84 effect: Effect.Effect<A, E, R>
85 ) => Effect.Effect<Queue.Dequeue<A, E>, never, R | Scope.Scope>
86}
87
88/**
89 * Service tag for the active SQL client service.
90 *
91 * **When to use**
92 *
93 * Use to access or provide the SQL client used to build statements, stream
94 * rows, reserve connections, and run transactions.
95 *
96 * @stability unstable
97 * @category services
98 * @since 4.0.0
99 */
100export const SqlClient = Context.Service<SqlClient>("effect/sql/SqlClient")
101
102/**
103 * Namespace containing types associated with the `SqlClient` service.
104 *
105 * @stability unstable
106 * @since 4.0.0
107 */
108export declare namespace SqlClient {
109 /**
110 * Options used to construct a `SqlClient`, including connection acquirers,
111 * the SQL compiler, transaction SQL, row transformation, tracing attributes,
112 * and optional reactive query integration.
113 *
114 * @stability unstable
115 * @category options
116 * @since 4.0.0
117 */
118 export interface MakeOptions {
119 readonly acquirer: Connection.Acquirer
120 /**
121 * Lends a connection for one statement instead of leasing one into a
122 * scope. A client that can do this saves the scope and finalizer per
123 * statement; `stream`, transactions, and `reserve` keep the `acquirer`,
124 * whose lease outlives the effect that starts it.
125 */
126 readonly borrower?: Connection.Borrower | undefined
127 readonly compiler: Compiler
128 readonly transactionAcquirer?: Connection.Acquirer
129 readonly spanAttributes: ReadonlyArray<readonly [string, unknown]>
130 readonly transactionService?: Context.Service<TransactionConnection, TransactionConnection.Service>
131 /**
132 * Whether the transaction control statements - `BEGIN`, `COMMIT`,
133 * `ROLLBACK`, and the savepoint pair - can be prepared like any other
134 * statement.
135 *
136 * **Details**
137 *
138 * They run on every transaction and never change, so a database that can
139 * prepare them stops parsing them again each time. Off by default,
140 * because several databases refuse to prepare transaction control at all;
141 * a driver has to say that its own does not.
142 */
143 readonly prepareTransactionControls?: boolean | undefined
144 readonly beginTransaction?: string | undefined
145 readonly rollback?: string | undefined
146 /** Commit SQL, or an effect that commits on the connection and checks the result. */
147 readonly commit?: string | ((conn: Connection.Connection) => Effect.Effect<void, SqlError>) | undefined
148 /** Cleanup on the same connection when COMMIT fails. Omit when the driver already ends the transaction. */
149 readonly onCommitFailure?: ((conn: Connection.Connection) => Effect.Effect<void, SqlError>) | undefined
150 readonly savepoint?: ((name: string) => string) | undefined
151 /**
152 * SQL to release a nested savepoint after success or a successful rollback.
153 * Omit to leave savepoints unreleased, including for dialects such as MSSQL
154 * that do not support releasing savepoints.
155 */
156 readonly releaseSavepoint?: ((name: string) => string) | undefined
157 readonly rollbackSavepoint?: ((name: string) => string) | undefined
158 readonly transformRows?: (<A extends object>(row: ReadonlyArray<A>) => ReadonlyArray<A>) | undefined
159 readonly reactiveQueue?: <A, E, R>(
160 keys: ReadonlyArray<unknown> | ReadonlyRecord<string, ReadonlyArray<unknown>>,
161 effect: Effect.Effect<A, E, R>
162 ) => Effect.Effect<Queue.Dequeue<A, E>, never, R | Scope.Scope>
163 }
164}
165
166let clientIdCounter = 0
167let transactionSemaphoreIdCounter = 0
168
169/**
170 * Constructs a `SqlClient` from connection acquirers, a compiler, transaction
171 * commands, tracing attributes, optional row transforms, and reactive query
172 * integration.
173 *
174 * @stability unstable
175 * @category constructors
176 * @since 4.0.0
177 */
178export const make = Effect.fnUntraced(function*(options: SqlClient.MakeOptions) {
179 const transactionService = options.transactionService ?? TransactionConnection(clientIdCounter++)
180 const getConnection = Effect.flatMap(
181 Effect.serviceOption(transactionService),
182 Option.match({
183 onNone: () => options.acquirer,
184 onSome: ([conn]) => Effect.succeed(conn)
185 })
186 )
187
188 const beginTransaction = options.beginTransaction ?? "BEGIN"
189 const commit = options.commit ?? "COMMIT"
190 const savepoint = options.savepoint ?? ((name: string) => `SAVEPOINT ${name}`)
191 const releaseSavepoint = options.releaseSavepoint
192 const rollback = options.rollback ?? "ROLLBACK"
193 const rollbackSavepoint = options.rollbackSavepoint ?? ((name: string) => `ROLLBACK TO SAVEPOINT ${name}`)
194 const transactionAcquirer = options.transactionAcquirer ?? options.acquirer
195 const control = options.prepareTransactionControls === true
196 ? (conn: Connection.Connection, sql: string) => conn.execute(sql, [], undefined)
197 : (conn: Connection.Connection, sql: string) => conn.executeUnprepared(sql, [], undefined)
198 const withTransaction = makeWithTransaction({
199 transactionService,
200 spanAttributes: options.spanAttributes,
201 acquireConnection: Effect.flatMap(
202 Scope.make(),
203 (scope) =>
204 Scope.provide(transactionAcquirer!, scope).pipe(
205 Effect.onError((cause) => Scope.close(scope, Exit.failCause(cause))),
206 Effect.map((conn) => [scope, conn] as const)
207 )
208 ),
209 begin: (conn) => control(conn, beginTransaction),
210 savepoint: (conn, id) => control(conn, savepoint(`effect_sql_${id}`)),
211 releaseSavepoint: releaseSavepoint
212 ? (conn, id) => control(conn, releaseSavepoint(`effect_sql_${id}`))
213 : undefined,
214 commit: typeof commit === "string" ? (conn) => control(conn, commit) : commit,
215 onCommitFailure: options.onCommitFailure,
216 rollback: (conn) => control(conn, rollback),
217 rollbackSavepoint: (conn, id) => control(conn, rollbackSavepoint(`effect_sql_${id}`))
218 })
219
220 const reactivity = yield* Reactivity
221 // A statement inside a transaction has to run on that transaction's
222 // connection, so borrowing is only for statements that reach the pool.
223 const borrower: Connection.Borrower | undefined = options.borrower === undefined ? undefined : (f) =>
224 Effect.flatMap(
225 Effect.serviceOption(transactionService),
226 Option.match({
227 onNone: () => options.borrower!(f),
228 onSome: ([conn]) => f(conn)
229 })
230 )
231
232 const client: SqlClient = Object.assign(
233 Statement.make(getConnection, options.compiler, options.spanAttributes, options.transformRows, borrower),
234 {
235 [TypeId]: TypeId as typeof TypeId,
236 safe: undefined as any,
237 withTransaction,
238 transactionService,
239 reserve: transactionAcquirer,
240 withoutTransforms(): any {
241 if (options.transformRows === undefined) {
242 return this
243 }
244 const statement = Statement.make(
245 getConnection,
246 options.compiler.withoutTransform,
247 options.spanAttributes,
248 undefined,
249 borrower
250 )
251 const client = Object.assign(statement, {
252 ...this,
253 ...statement
254 })
255 ;(client as any).safe = client
256 ;(client as any).withoutTransforms = () => client
257 return client
258 },
259 reactive: options.reactiveQueue ?
260 <A, E, R>(
261 keys: ReadonlyArray<unknown> | ReadonlyRecord<string, ReadonlyArray<unknown>>,
262 effect: Effect.Effect<A, E, R>
263 ) =>
264 options.reactiveQueue!(keys, effect).pipe(
265 Effect.map(Stream.fromQueue),
266 Stream.unwrap
267 ) :
268 reactivity.stream,
269 reactiveMailbox: options.reactiveQueue ?? reactivity.query
270 }
271 )
272 ;(client as any).safe = client
273
274 return client
275})
276
277/**
278 * Builds the acquirers and failed-COMMIT cleanup for a SQLite client that
279 * shares one connection through a semaphore.
280 *
281 * **Details**
282 *
283 * SQLite can keep a transaction open after COMMIT fails, for example on a
284 * deferred foreign key violation. `onCommitFailure` rolls it back before the
285 * connection is reused. If that ROLLBACK fails, the connection is rejected, and
286 * each acquisition retries the ROLLBACK once while holding the semaphore until
287 * it succeeds. The database is never reopened, and transactions started with a
288 * plain `BEGIN` are only touched after a failed cleanup.
289 *
290 * Pass `isTransaction` when the driver can report whether a transaction is
291 * open. Without it, ROLLBACK always runs and SQLite's "no transaction is
292 * active" error counts as clean.
293 *
294 * @stability unstable
295 * @category constructors
296 * @since 4.0.0
297 */
298export const makeSqliteAcquirers = <C extends Connection.Connection>(options: {
299 readonly connection: Effect.Effect<C, SqlError>
300 readonly semaphore: Semaphore.Semaphore
301 readonly isTransaction?: ((conn: C) => boolean) | undefined
302}): {
303 readonly acquirer: Effect.Effect<C, SqlError>
304 readonly transactionAcquirer: Effect.Effect<C, SqlError, Scope.Scope>
305 readonly onCommitFailure: (conn: Connection.Connection) => Effect.Effect<void, SqlError>
306} => {
307 const { isTransaction, semaphore } = options
308 let poisoned: SqlError | undefined
309 const poison = (cause: Cause.Cause<SqlError>) =>
310 poisoned = new SqlError({
311 reason: new ConnectionError({
312 message: "SQLite connection cannot be reused after failed COMMIT cleanup",
313 operation: "rollback",
314 cause
315 })
316 })
317
318 const rollback = (conn: C): Effect.Effect<void, SqlError> => {
319 if (isTransaction) {
320 return isTransaction(conn) ? Effect.asVoid(conn.executeUnprepared("ROLLBACK", [], undefined)) : Effect.void
321 }
322 return conn.executeUnprepared("ROLLBACK", [], undefined).pipe(
323 Effect.asVoid,
324 Effect.catchIf(isNoTransactionError, () => Effect.void)
325 )
326 }
327
328 const available = Effect.flatMap(options.connection, (conn) => {
329 if (poisoned === undefined) return Effect.succeed(conn)
330 return Effect.matchCauseEffect(rollback(conn), {
331 onFailure: (cause) => Effect.fail(poison(cause)),
332 onSuccess: () => {
333 poisoned = undefined
334 return Effect.succeed(conn)
335 }
336 })
337 })
338
339 return {
340 acquirer: semaphore.withPermits(1)(available),
341 transactionAcquirer: Effect.uninterruptibleMask((restore) =>
342 Effect.flatMap(Effect.scope, (scope) =>
343 restore(semaphore.take(1)).pipe(
344 Effect.andThen(Scope.addFinalizer(scope, semaphore.release(1))),
345 Effect.andThen(available)
346 ))
347 ),
348 onCommitFailure: (conn) => Effect.tapCause(rollback(conn as C), (cause) => Effect.sync(() => poison(cause)))
349 }
350}
351
352const isNoTransactionError = (error: SqlError): boolean => {
353 const cause = error.reason.cause
354 const message = typeof cause === "string"
355 ? cause
356 : hasProperty(cause, "message") && typeof cause.message === "string"
357 ? cause.message
358 : ""
359 return /no transaction is active/i.test(message)
360}
361
362/**
363 * Builds a transaction wrapper that begins top-level transactions, uses
364 * savepoints for nested transactions, commits on success, and rolls back on
365 * failure or interruption. Releases nested savepoints when `releaseSavepoint`
366 * is provided.
367 *
368 * @stability unstable
369 * @category transactions
370 * @since 4.0.0
371 */
372export const makeWithTransaction = <I, S>(options: {
373 readonly transactionService: Context.Key<I, readonly [conn: S, counter: number]>
374 readonly spanAttributes: ReadonlyArray<readonly [string, unknown]>
375 readonly acquireConnection: Effect.Effect<readonly [Scope.Closeable | undefined, S], SqlError>
376 readonly begin: (conn: NoInfer<S>) => Effect.Effect<void, SqlError>
377 readonly savepoint: (conn: NoInfer<S>, id: number) => Effect.Effect<void, SqlError>
378 /**
379 * Releases a nested savepoint after success or a successful rollback.
380 * Omit to leave savepoints unreleased, including for dialects such as MSSQL
381 * that do not support releasing savepoints.
382 */
383 readonly releaseSavepoint?: ((conn: NoInfer<S>, id: number) => Effect.Effect<void, SqlError>) | undefined
384 readonly commit: (conn: NoInfer<S>) => Effect.Effect<void, SqlError>
385 /** Driver-specific recovery before the connection is released after a failed COMMIT. */
386 readonly onCommitFailure?: ((conn: NoInfer<S>) => Effect.Effect<void, SqlError>) | undefined
387 readonly rollback: (conn: NoInfer<S>) => Effect.Effect<void, SqlError>
388 readonly rollbackSavepoint: (conn: NoInfer<S>, id: number) => Effect.Effect<void, SqlError>
389}) => {
390 const transactionSemaphore = Context.Service<Semaphore.Semaphore>(
391 `effect/sql/SqlClient/TransactionSemaphore/${transactionSemaphoreIdCounter++}`
392 )
393 return <R, E, A>(effect: Effect.Effect<A, E, R>): Effect.Effect<A, E | SqlError, R> =>
394 Effect.uninterruptibleMask((restore) =>
395 Effect.useSpan(
396 "sql.transaction",
397 { kind: "client" },
398 (span) =>
399 Effect.withFiber<A, E | SqlError, R>((fiber) => {
400 for (const [key, value] of options.spanAttributes) {
401 span.attribute(key, value)
402 }
403 const services = fiber.context
404 const clock = fiber.getRef(Clock)
405 const connOption = Context.getOption(services, options.transactionService)
406 const conn = connOption._tag === "Some"
407 ? Effect.succeed([undefined, connOption.value[0]] as const)
408 : options.acquireConnection
409 const id = connOption._tag === "Some" ? connOption.value[1] + 1 : 0
410 const transaction = Effect.flatMap(
411 conn,
412 (
413 [scope, conn]
414 ) =>
415 (id === 0 ? options.begin(conn) : options.savepoint(conn, id)).pipe(
416 Effect.flatMap(() =>
417 Effect.onExitPrimitive(
418 Effect.provideContext(
419 restore(effect),
420 services.pipe(
421 Context.add(options.transactionService, [conn, id]),
422 Context.add(transactionSemaphore, Semaphore.makeUnsafe(1)),
423 Context.add(Tracer.ParentSpan, span)
424 )
425 ),
426 (exit) => {
427 let effect: Effect.Effect<void>
428 if (Exit.isSuccess(exit)) {
429 if (id === 0) {
430 span.event("db.transaction.commit", clock.currentTimeNanosUnsafe())
431 const onCommitFailure = options.onCommitFailure
432 effect = Effect.orDie(options.commit(conn))
433 if (onCommitFailure) {
434 effect = Effect.onError(effect, () => Effect.orDie(onCommitFailure(conn)))
435 }
436 } else {
437 span.event("db.transaction.savepoint", clock.currentTimeNanosUnsafe())
438 effect = Effect.void
439 }
440 } else {
441 span.event("db.transaction.rollback", clock.currentTimeNanosUnsafe())
442 effect = Effect.orDie(
443 id > 0
444 ? options.rollbackSavepoint(conn, id)
445 : options.rollback(conn)
446 )
447 }
448 if (id > 0 && options.releaseSavepoint) {
449 effect = Effect.andThen(effect, Effect.orDie(options.releaseSavepoint(conn, id)))
450 }
451 return Effect.flatMap(effect, () => exit)
452 },
453 true
454 )
455 ),
456 scope ? (eff) => Effect.onExitPrimitive(eff, (exit) => Scope.close(scope, exit), true) : identity
457 )
458 )
459 return id === 0
460 ? transaction
461 : Context.getUnsafe(services, transactionSemaphore).withPermit(transaction)
462 })
463 )
464 )
465}
466
467/**
468 * Phantom identifier for the scoped transaction connection service associated
469 * with a SQL client.
470 *
471 * @stability unstable
472 * @category models
473 * @since 4.0.0
474 */
475export interface TransactionConnection {
476 readonly _: unique symbol
477}
478
479/**
480 * Namespace containing types associated with transaction connection services.
481 *
482 * @stability unstable
483 * @since 4.0.0
484 */
485export declare namespace TransactionConnection {
486 /**
487 * Service payload stored during a transaction, containing the active
488 * connection and nested transaction depth.
489 *
490 * @stability unstable
491 * @category services
492 * @since 4.0.0
493 */
494 export type Service = readonly [conn: Connection.Connection, depth: number]
495}
496
497/**
498 * Creates a unique context service tag for the active transaction connection of
499 * a specific SQL client.
500 *
501 * @stability unstable
502 * @category services
503 * @since 4.0.0
504 */
505export const TransactionConnection = (
506 clientId: number
507): Context.Service<TransactionConnection, TransactionConnection.Service> =>
508 Context.Service(`effect/sql/SqlClient/TransactionConnection/${clientId}`)
509
510/**
511 * Context reference used by SQL integrations to opt in to safe integer
512 * handling; defaults to `false`.
513 *
514 * @stability unstable
515 * @category services
516 * @since 4.0.0
517 */
518export const SafeIntegers = Context.Reference<boolean>("effect/sql/SqlClient/SafeIntegers", {
519 defaultValue: () => false
520})