1/**
2 * Pools of native `PgConnection` sessions.
3 *
4 * @since 4.0.0
5 */
6import * as Clock from "effect/Clock"
7import * as Context from "effect/Context"
8import * as Duration from "effect/Duration"
9import * as Effect from "effect/Effect"
10import * as Pool from "effect/Pool"
11import type * as Scope from "effect/Scope"
12import type { SqlError } from "effect/sql/SqlError"
13import { connectionInternals } from "./internal/connection.ts"
14import * as PgConnection from "./PgConnection.ts"
15
16const defaultMultiplexConcurrency = 32
17
18/**
19 * The runtime type identifier for `PgPool`.
20 *
21 * @category type IDs
22 * @since 4.0.0
23 */
24export const TypeId: TypeId = "~@effect/sql-pg/PgPool"
25
26/**
27 * The type-level identifier for `PgPool`.
28 *
29 * @category type IDs
30 * @since 4.0.0
31 */
32export type TypeId = "~@effect/sql-pg/PgPool"
33
34/**
35 * Connection and sizing settings for a PostgreSQL session pool.
36 *
37 * **Details**
38 *
39 * The defaults are 0 to 10 connections and a 60-second idle timeout.
40 * `connectionTTL` replaces connections that exceed the configured lifetime.
41 * Every connection is used at least once, so a TTL of zero disables reuse.
42 *
43 * With `multiplex` enabled, fibers may share pooled connections for pipelined
44 * queries. Reserved connections remain exclusive. Without multiplexing, every
45 * checkout is exclusive.
46 *
47 * @category models
48 * @since 4.0.0
49 */
50export interface Config extends PgConnection.Config {
51 readonly idleTimeout?: Duration.Input | undefined
52 readonly maxConnections?: number | undefined
53 readonly minConnections?: number | undefined
54 readonly connectionTTL?: Duration.Input | undefined
55 /**
56 * How many statements may share one connection when `multiplex` is on.
57 * Defaults to `32`. Statements are pipelined into one write, so a higher
58 * number means fewer round trips, but it also means a slow statement holds
59 * up more of the statements queued behind it.
60 */
61 readonly multiplexConcurrency?: number | undefined
62}
63
64/**
65 * A PostgreSQL session pool.
66 *
67 * @category models
68 * @since 4.0.0
69 */
70export interface PgPool {
71 readonly [TypeId]: TypeId
72 readonly config: Config
73 /**
74 * Checks out a session until the scope closes. Without multiplexing the
75 * checkout is exclusive. With multiplexing the
76 * session may be shared with other fibers, so multi-statement work should
77 * use `reserve` instead.
78 */
79 readonly get: Effect.Effect<PgConnection.PgConnection, SqlError, Scope.Scope>
80 /**
81 * Checks out a session for exclusive use until the scope closes. Use this for
82 * transactions and listeners on a multiplexed pool.
83 */
84 readonly reserve: Effect.Effect<PgConnection.PgConnection, SqlError, Scope.Scope>
85 /**
86 * Removes a session so the pool can replace it. Fatal protocol and socket
87 * errors invalidate sessions automatically.
88 */
89 /**
90 * Lends a session for the duration of one effect and takes it back on any
91 * exit, without opening a scope for it.
92 *
93 * **Details**
94 *
95 * For work that finishes with the effect that runs it. A lease that has to
96 * outlive its effect - a stream, a transaction - takes `get` or `reserve`.
97 */
98 readonly use: <A, E, R>(
99 f: (connection: PgConnection.PgConnection) => Effect.Effect<A, E, R>
100 ) => Effect.Effect<A, E | SqlError, R>
101 readonly invalidate: (connection: PgConnection.PgConnection) => Effect.Effect<void>
102}
103
104/**
105 * The service tag for `PgPool`.
106 *
107 * @category services
108 * @since 4.0.0
109 */
110export const PgPool = Context.Service<PgPool>("@effect/sql-pg/PgPool")
111
112/**
113 * Creates a scoped PostgreSQL session pool.
114 *
115 * **Details**
116 *
117 * Connections are opened lazily up to `maxConnections` and released down to
118 * `minConnections` after `idleTimeout` without use. Closing the scope shuts
119 * the pool down and releases every session.
120 *
121 * @category constructors
122 * @since 4.0.0
123 */
124export const make = Effect.fnUntraced(function*(options: Config): Effect.fn.Return<PgPool, SqlError, Scope.Scope> {
125 const clock = yield* Clock.Clock
126 const runFork = Effect.runForkWith(yield* Effect.context())
127
128 const multiplex = options.multiplex ?? false
129 const connectionTTL = options.connectionTTL !== undefined
130 ? Duration.toMillis(Duration.fromInputUnsafe(options.connectionTTL))
131 : undefined
132 const deadConnections = new Set<PgConnection.PgConnection>()
133 const createdAt = new WeakMap<PgConnection.PgConnection, number>()
134 const checkedOut = new WeakSet<PgConnection.PgConnection>()
135
136 // Assigned below, once the pool exists. `acquire` only runs when the pool
137 // opens a connection, which is always after that.
138 let pool: Pool.Pool<PgConnection.PgConnection, SqlError>
139 const acquire = Effect.tap(PgConnection.make(options), (connection) =>
140 Effect.sync(() => {
141 createdAt.set(connection, clock.currentTimeMillisUnsafe())
142 const internals = connectionInternals(connection)
143 internals.retireHooks.add(() => {
144 deadConnections.add(connection)
145 // `deadConnections` is only read by the next checkout, and a checkout
146 // already waiting for this connection would never get that far. Tell
147 // the pool now so it can replace the connection and admit them.
148 runFork(Pool.invalidate(pool, connection))
149 })
150 // Pinning a shared session has to take it out of circulation, or a
151 // second checkout lands on the connection its own stream is holding.
152 if (multiplex) internals.reserve = Pool.reserve(pool, connection)
153 }))
154
155 const maxConnections = options.maxConnections ?? 10
156 pool = yield* Pool.makeWithTTL({
157 acquire,
158 min: options.minConnections ?? 0,
159 max: maxConnections,
160 concurrency: multiplex
161 ? Math.max(1, options.multiplexConcurrency ?? defaultMultiplexConcurrency)
162 : 1,
163 timeToLive: options.idleTimeout ?? Duration.seconds(60),
164 timeToLiveStrategy: "usage"
165 })
166
167 const expired = (connection: PgConnection.PgConnection): boolean => {
168 if (connectionInternals(connection).deadError() !== undefined) return true
169 if (connectionTTL === undefined) return false
170 if (!checkedOut.has(connection)) {
171 checkedOut.add(connection)
172 return false
173 }
174 const openedAt = createdAt.get(connection)
175 return openedAt !== undefined && clock.currentTimeMillisUnsafe() - openedAt >= connectionTTL
176 }
177
178 // A checkout runs per statement, so the case where nothing has died and the
179 // first connection is usable stays a plain flatMap; retrying is the
180 // exception and pays for the loop.
181 const retry: Effect.Effect<PgConnection.PgConnection, SqlError, Scope.Scope> = Effect.gen(function*() {
182 while (true) {
183 if (deadConnections.size > 0) {
184 const dead = Array.from(deadConnections)
185 deadConnections.clear()
186 yield* Effect.forEach(dead, (connection) => Pool.invalidate(pool, connection), { discard: true })
187 }
188 const connection = yield* Pool.get(pool)
189 if (expired(connection)) {
190 yield* Pool.invalidate(pool, connection)
191 continue
192 }
193 return connection
194 }
195 })
196
197 const get: Effect.Effect<PgConnection.PgConnection, SqlError, Scope.Scope> = Effect.suspend(() =>
198 deadConnections.size > 0 ? retry : Effect.flatMap(Pool.get(pool), (connection) =>
199 expired(connection)
200 ? Effect.andThen(Pool.invalidate(pool, connection), retry)
201 : Effect.succeed(connection))
202 )
203
204 // `pin` reserves the pool item itself, so this needs no help.
205 const reserve = Effect.flatMap(get, (connection) => connection.pin)
206
207 // `Pool.use` cannot pre-check a session. Use it only when no TTL applies and
208 // no connection awaits retirement. Checking inside its callback can deadlock
209 // a size-one pool while acquiring a replacement.
210 const use = <A, E, R>(
211 f: (connection: PgConnection.PgConnection) => Effect.Effect<A, E, R>
212 ): Effect.Effect<A, E | SqlError, R> =>
213 Effect.suspend(() =>
214 connectionTTL === undefined && deadConnections.size === 0
215 ? Pool.use(pool, f)
216 : Effect.scoped(Effect.flatMap(get, f))
217 )
218
219 const pgPool: PgPool = {
220 [TypeId]: TypeId,
221 config: options,
222 get,
223 reserve,
224 use,
225 invalidate: (connection) => Pool.invalidate(pool, connectionInternals(connection).base as PgConnection.PgConnection)
226 }
227 return pgPool
228})