1/**
2 * PostgreSQL support for Effect SQL, backed by the native wire protocol client.
3 *
4 * @since 4.0.0
5 */
6import type * as Arr from "effect/Array"
7import * as Config from "effect/Config"
8import * as Context from "effect/Context"
9import type * as Duration from "effect/Duration"
10import * as Effect from "effect/Effect"
11import * as Layer from "effect/Layer"
12import type * as Queue from "effect/Queue"
13import * as Reactivity from "effect/reactivity/Reactivity"
14import type * as Redacted from "effect/Redacted"
15import type * as Scope from "effect/Scope"
16import * as Client from "effect/sql/SqlClient"
17import type { Borrower, Connection } from "effect/sql/SqlConnection"
18import { SqlError, UnknownError } from "effect/sql/SqlError"
19import type { Custom, Fragment } from "effect/sql/Statement"
20import * as Statement from "effect/sql/Statement"
21import * as Stream from "effect/Stream"
22import type { Duplex } from "node:stream"
23import type { ConnectionOptions } from "node:tls"
24import { validateChannelName } from "./internal/sqlError.ts"
25import * as PgConnection from "./PgConnection.ts"
26import * as PgPool from "./PgPool.ts"
27import * as PgTypes from "./PgTypes.ts"
28
29/**
30 * The runtime type identifier for `PgClient`.
31 *
32 * @category type IDs
33 * @since 4.0.0
34 */
35export const TypeId: TypeId = "~@effect/sql-pg/PgClient"
36
37/**
38 * The type-level identifier for `PgClient`.
39 *
40 * @category type IDs
41 * @since 4.0.0
42 */
43export type TypeId = "~@effect/sql-pg/PgClient"
44
45/**
46 * A PostgreSQL `SqlClient` with JSON and `LISTEN`/`NOTIFY` helpers.
47 *
48 * @category services
49 * @since 4.0.0
50 */
51export interface PgClient extends Client.SqlClient {
52 readonly [TypeId]: TypeId
53 readonly config: PgClientConfig
54 readonly json: (_: unknown) => Fragment
55 /**
56 * Registers a channel listener and returns its notification queue after
57 * PostgreSQL confirms `LISTEN`. The listener holds a connection until the
58 * scope closes.
59 * Connection failures after registration fail the queue with the original
60 * `SqlError`. Intentional scope closure interrupts consumers.
61 */
62 readonly listen: (
63 channel: string
64 ) => Effect.Effect<Queue.Dequeue<PgConnection.Notification, SqlError>, SqlError, Scope.Scope>
65 readonly notify: (channel: string, payload: string) => Effect.Effect<void, SqlError>
66}
67
68/**
69 * The service tag for `PgClient`.
70 *
71 * @category services
72 * @since 4.0.0
73 */
74export const PgClient = Context.Service<PgClient>("@effect/sql-pg/PgClient")
75
76/**
77 * Connection and query settings for a PostgreSQL client.
78 *
79 * @category models
80 * @since 4.0.0
81 */
82export interface PgClientConfig {
83 readonly url?: Redacted.Redacted | undefined
84
85 readonly host?: string | undefined
86 readonly port?: number | undefined
87 readonly path?: string | undefined
88 readonly ssl?: boolean | ConnectionOptions | undefined
89 readonly database?: string | undefined
90 readonly username?: string | undefined
91 /**
92 * A static password or an Effect evaluated for each connection attempt.
93 * Providers must handle typed errors and require no services.
94 * {@link Effect.orDie} converts typed errors to defects, not retryable SQL errors.
95 */
96 readonly password?: Redacted.Redacted | Effect.Effect<Redacted.Redacted> | undefined
97
98 readonly connectTimeout?: Duration.Input | undefined
99
100 readonly stream?: (() => Duplex) | undefined
101
102 /**
103 * Overrides `startupParameters.application_name`, the URL's `application_name`,
104 * and the default `"@effect/sql-pg"`, in that order.
105 */
106 readonly applicationName?: string | undefined
107 /**
108 * Session defaults sent in every physical connection's startup packet.
109 * Names are lowercased; `user`, `database`, `replication`, and `options` are
110 * reserved. `client_encoding` only accepts UTF8 / UTF-8 (case-insensitive).
111 * Empty names and NUL bytes fail before connecting; PostgreSQL validates
112 * other settings. Do not set the same GUC here and in `startupOptions`.
113 */
114 readonly startupParameters?: Readonly<Record<string, string>> | undefined
115 /**
116 * Opaque PostgreSQL startup options, overriding the URL's `options` parameter.
117 * Forwarded without parsing `-c` flags or checking for duplicate GUCs.
118 */
119 readonly startupOptions?: string | undefined
120 readonly spanAttributes?: Record<string, unknown> | undefined
121
122 readonly transformResultNames?: ((str: string) => string) | undefined
123 readonly transformQueryNames?: ((str: string) => string) | undefined
124 readonly transformJson?: boolean | undefined
125 readonly types?: PgTypes.Registry | undefined
126 /**
127 * Pipelines queries from multiple fibers on each pooled connection.
128 * Transactions, streams, and listeners still reserve a connection.
129 */
130 readonly multiplex?: boolean | undefined
131 /**
132 * How many statements may share one connection when `multiplex` is on.
133 * Defaults to `32`. Higher values trade tail latency for throughput: the
134 * statements sharing a connection are pipelined into one write, and they
135 * also queue behind the slowest of them.
136 */
137 readonly multiplexConcurrency?: number | undefined
138 /**
139 * Caches prepared statements by name. Enabled by default. Disable it for
140 * poolers that cannot preserve named statements between queries.
141 */
142 readonly prepare?: boolean | undefined
143 /** How many statements a connection keeps prepared. Defaults to `100`. */
144 readonly preparedStatementCacheSize?: number | undefined
145 /** Maximum backend message size in bytes. Defaults to 16 MiB. */
146 readonly maxMessageSize?: number | undefined
147}
148
149/**
150 * PostgreSQL client settings with connection pool limits and timeouts.
151 *
152 * @category models
153 * @since 4.0.0
154 */
155export interface PgPoolConfig extends PgClientConfig {
156 readonly idleTimeout?: Duration.Input | undefined
157
158 readonly maxConnections?: number | undefined
159 readonly minConnections?: number | undefined
160 readonly connectionTTL?: Duration.Input | undefined
161}
162
163/**
164 * Creates a scoped PostgreSQL client backed by a connection pool.
165 *
166 * @category constructors
167 * @since 4.0.0
168 */
169export const make = (options: PgPoolConfig): Effect.Effect<PgClient, SqlError, Scope.Scope | Reactivity.Reactivity> =>
170 Effect.flatMap(PgPool.make(options), (pool) =>
171 makeImpl({
172 acquirer: Effect.map(pool.get, makeConnection),
173 borrower: (f) => pool.use((connection) => f(makeConnection(connection))),
174 transactionAcquirer: Effect.map(pool.reserve, makeConnection),
175 listenAcquirer: pool.reserve,
176 config: options
177 }))
178
179/**
180 * Creates a scoped PostgreSQL client backed by one connection.
181 *
182 * @category constructors
183 * @since 4.0.0
184 */
185export const makeClient = (
186 options: PgClientConfig & {
187 /**
188 * Opens a separate connection for each stream and listener when enabled.
189 */
190 readonly acquireForStream?: boolean | undefined
191 }
192): Effect.Effect<PgClient, SqlError, Scope.Scope | Reactivity.Reactivity> =>
193 Effect.flatMap(PgConnection.make(options), (connection) =>
194 makeImpl({
195 acquirer: Effect.succeed(makeConnection(
196 connection,
197 options.acquireForStream ? PgConnection.make(options) : undefined
198 )),
199 transactionAcquirer: Effect.map(connection.pin, makeConnection),
200 listenAcquirer: options.acquireForStream ? PgConnection.make(options) : connection.pin,
201 config: options
202 }))
203
204/**
205 * Builds the shared SQL facade around native connection acquirers.
206 */
207const makeImpl = Effect.fnUntraced(function*(
208 options: {
209 readonly acquirer: Effect.Effect<Connection, SqlError, Scope.Scope>
210 readonly borrower?: Borrower | undefined
211 readonly transactionAcquirer: Effect.Effect<Connection, SqlError, Scope.Scope>
212 readonly listenAcquirer: Effect.Effect<PgConnection.PgConnection, SqlError, Scope.Scope>
213 readonly config: PgClientConfig
214 }
215): Effect.fn.Return<PgClient, SqlError, Scope.Scope | Reactivity.Reactivity> {
216 const config = options.config
217 const compiler = makeCompiler(
218 config.transformQueryNames,
219 config.transformJson
220 )
221 const transformRows = config.transformResultNames ?
222 Statement.defaultTransforms(
223 config.transformResultNames,
224 config.transformJson
225 ).array :
226 undefined
227
228 const listen = (
229 channel: string
230 ): Effect.Effect<Queue.Dequeue<PgConnection.Notification, SqlError>, SqlError, Scope.Scope> =>
231 Effect.flatMap(options.listenAcquirer, (connection) => connection.listen(channel))
232
233 return Object.assign(
234 yield* Client.make({
235 acquirer: options.acquirer,
236 borrower: options.borrower,
237 // Postgres prepares transaction control like anything else, and a client
238 // with preparation turned off falls back to the unnamed path anyway.
239 prepareTransactionControls: true,
240 commit,
241 releaseSavepoint: (name) => `RELEASE SAVEPOINT ${name}`,
242 transactionAcquirer: options.transactionAcquirer,
243 compiler,
244 spanAttributes: [
245 ...(config.spanAttributes ? Object.entries(config.spanAttributes) : []),
246 [ATTR_DB_SYSTEM_NAME, "postgresql"],
247 [ATTR_DB_NAMESPACE, config.database ?? config.username ?? "postgres"],
248 [ATTR_SERVER_ADDRESS, config.host ?? "localhost"],
249 [ATTR_SERVER_PORT, config.port ?? 5432]
250 ],
251 transformRows
252 }),
253 {
254 [TypeId]: TypeId as TypeId,
255 config,
256 json: (_: unknown) => Statement.fragment([PgJson(_)]),
257 listen,
258 notify: (channel: string, payload: string) => {
259 const channelError = validateChannelName(channel, "notify")
260 return channelError !== undefined ?
261 Effect.fail(channelError) :
262 Effect.asVoid(Effect.scoped(Effect.flatMap(
263 options.acquirer,
264 (conn) => conn.executeRaw(`SELECT pg_notify($1, $2)`, [channel, payload])
265 )))
266 }
267 }
268 )
269})
270
271class ConnectionImpl implements Connection {
272 readonly connection: PgConnection.PgConnection
273 readonly streamAcquirer: Effect.Effect<PgConnection.PgConnection, SqlError, Scope.Scope> | undefined
274
275 constructor(
276 connection: PgConnection.PgConnection,
277 streamAcquirer?: Effect.Effect<PgConnection.PgConnection, SqlError, Scope.Scope> | undefined
278 ) {
279 this.connection = connection
280 this.streamAcquirer = streamAcquirer
281 }
282
283 private run(query: string, params: ReadonlyArray<unknown>, prepare = true) {
284 return Effect.map(this.connection.query(query, params, prepare), (result) => result.rows)
285 }
286
287 execute(
288 sql: string,
289 params: ReadonlyArray<unknown>,
290 transformRows: (<A extends object>(row: ReadonlyArray<A>) => ReadonlyArray<A>) | undefined
291 ) {
292 return transformRows
293 ? Effect.map(this.run(sql, params), transformRows)
294 : this.run(sql, params)
295 }
296 executeRaw(sql: string, params: ReadonlyArray<unknown>) {
297 return this.connection.query(sql, params)
298 }
299 executeWithoutTransform(sql: string, params: ReadonlyArray<unknown>) {
300 return this.run(sql, params)
301 }
302 executeValues(sql: string, params: ReadonlyArray<unknown>) {
303 return this.connection.queryValues(sql, params)
304 }
305 executeValuesUnprepared(sql: string, params: ReadonlyArray<unknown>) {
306 return this.connection.queryValues(sql, params, false)
307 }
308 executeUnprepared(
309 sql: string,
310 params: ReadonlyArray<unknown>,
311 transformRows: (<A extends object>(row: ReadonlyArray<A>) => ReadonlyArray<A>) | undefined
312 ) {
313 const operation = this.run(sql, params, false)
314 return transformRows
315 ? Effect.map(operation, transformRows)
316 : operation
317 }
318 executeStream(
319 sql: string,
320 params: ReadonlyArray<unknown>,
321 transformRows: (<A extends object>(row: ReadonlyArray<A>) => ReadonlyArray<A>) | undefined
322 ) {
323 const stream = this.streamAcquirer === undefined
324 ? this.connection.stream(sql, params)
325 : Stream.unwrap(Effect.map(this.streamAcquirer, (connection) => connection.stream(sql, params)))
326 return transformRows
327 ? Stream.mapArray(stream, (rows) => transformRows(rows) as Arr.NonEmptyReadonlyArray<PgConnection.Row>)
328 : stream
329 }
330}
331
332/** PostgreSQL returns `ROLLBACK` for `COMMIT` when a transaction is aborted. */
333const commit = (conn: Connection): Effect.Effect<void, SqlError> =>
334 Effect.flatMap(
335 conn.executeRaw("COMMIT", []) as Effect.Effect<PgConnection.Result, SqlError>,
336 (result) =>
337 result.command === "ROLLBACK"
338 ? Effect.fail(
339 new SqlError({
340 reason: new UnknownError({
341 cause: new Error("COMMIT rolled back an aborted transaction"),
342 message: "PgClient: COMMIT rolled back an aborted transaction",
343 operation: "commit"
344 })
345 })
346 )
347 : Effect.void
348 )
349
350const makeConnection = (
351 connection: PgConnection.PgConnection,
352 streamAcquirer?: Effect.Effect<PgConnection.PgConnection, SqlError, Scope.Scope> | undefined
353): Connection => new ConnectionImpl(connection, streamAcquirer)
354
355/**
356 * Provides both `PgClient` and `SqlClient` from an acquisition effect.
357 *
358 * @category layers
359 * @since 4.0.0
360 */
361export const layerFrom = <E, R>(
362 acquire: Effect.Effect<PgClient, E, R>
363): Layer.Layer<PgClient | Client.SqlClient, E, Exclude<R, Scope.Scope | Reactivity.Reactivity>> =>
364 Layer.effectContext(
365 Effect.map(acquire, (client) =>
366 Context.make(PgClient, client).pipe(
367 Context.add(Client.SqlClient, client)
368 ))
369 ).pipe(Layer.provide(Reactivity.layer)) as any
370
371/**
372 * Creates a client layer from wrapped pool configuration.
373 *
374 * @category layers
375 * @since 4.0.0
376 */
377export const layerConfig = (
378 config: Config.Wrap<PgPoolConfig>
379): Layer.Layer<PgClient | Client.SqlClient, Config.ConfigError | SqlError> =>
380 layerFrom(Effect.flatMap(
381 Config.unwrap(config),
382 make
383 ))
384
385/**
386 * Creates a client layer from pool configuration.
387 *
388 * @category layers
389 * @since 4.0.0
390 */
391export const layer = (
392 config: PgPoolConfig
393): Layer.Layer<PgClient | Client.SqlClient, SqlError> => layerFrom(make(config))
394
395/**
396 * Creates the PostgreSQL statement compiler.
397 *
398 * @category constructors
399 * @since 4.0.0
400 */
401export const makeCompiler = (
402 transform?: (_: string) => string,
403 transformJson = true
404): Statement.Compiler => {
405 const transformValue = transformJson && transform
406 ? Statement.defaultTransforms(transform).value
407 : undefined
408
409 return Statement.makeCompiler<PgCustom>({
410 dialect: "pg",
411 placeholder(_) {
412 return `$${_}`
413 },
414 onIdentifier: transform ?
415 function(value, withoutTransform) {
416 return withoutTransform ? escape(value) : escape(transform(value))
417 } :
418 escape,
419 onRecordUpdate(placeholders, valueAlias, valueColumns, values, returning) {
420 return [
421 `(values ${placeholders}) AS ${valueAlias}${valueColumns}${returning ? ` RETURNING ${returning[0]}` : ""}`,
422 returning ?
423 values.flat().concat(returning[1]) :
424 values.flat()
425 ]
426 },
427 onCustom(type, placeholder, withoutTransform) {
428 switch (type.kind) {
429 case "PgJson": {
430 const value = withoutTransform || transformValue === undefined
431 ? type.paramA
432 : transformValue(type.paramA)
433 return [
434 placeholder(undefined),
435 [PgTypes.jsonb(value)]
436 ]
437 }
438 }
439 }
440 })
441}
442
443const escape = Statement.defaultEscape("\"")
444
445/**
446 * PostgreSQL-specific statement fragments.
447 *
448 * @category models
449 * @since 4.0.0
450 */
451export type PgCustom = PgJson
452
453/**
454 * @category models
455 * @since 4.0.0
456 */
457interface PgJson extends Custom<"PgJson", unknown> {}
458/**
459 * @category constructors
460 * @since 4.0.0
461 */
462const PgJson = Statement.custom<PgJson>("PgJson")
463
464const ATTR_DB_SYSTEM_NAME = "db.system.name"
465const ATTR_DB_NAMESPACE = "db.namespace"
466const ATTR_SERVER_ADDRESS = "server.address"
467const ATTR_SERVER_PORT = "server.port"