clover's git

the-knife

1	/**
2	* Native PostgreSQL sessions built on the `PgProtocol` wire codec.
3	*
4	* **Details**
5	*
6	* Sessions support queries, streaming, `LISTEN`/`NOTIFY`, cancellation, and
7	* exclusive ownership for transactions.
8	*
9	* @since 4.0.0
10	*/
11	import type * as Arr from "effect/Array"
12	import * as Cause from "effect/Cause"
13	import * as Channel from "effect/Channel"
14	import * as Context from "effect/Context"
15	import * as Deferred from "effect/Deferred"
16	import * as Duration from "effect/Duration"
17	import * as Effect from "effect/Effect"
18	import * as Exit from "effect/Exit"
19	import * as Queue from "effect/Queue"
20	import * as Redacted from "effect/Redacted"
21	import * as EffectResult from "effect/Result"
22	import * as Scope from "effect/Scope"
23	import * as Semaphore from "effect/Semaphore"
24	import { AuthenticationError, ConnectionError, SqlError, type SqlErrorReason, UnknownError } from "effect/sql/SqlError"
25	import * as Stream from "effect/Stream"
26	import { Buffer } from "node:buffer"
27	import { randomBytes } from "node:crypto"
28	import * as Net from "node:net"
29	import type { Duplex } from "node:stream"
30	import * as Tls from "node:tls"
31	import type { ConnectionOptions } from "node:tls"
32	import { type ConnectionInternals, internalsKey } from "./internal/connection.ts"
33	import * as PasswordInternal from "./internal/password.ts"
34	import { classifySqlState, validateChannelName } from "./internal/sqlError.ts"
35	import * as PgAuth from "./PgAuth.ts"
36	import * as PgProtocol from "./PgProtocol.ts"
37	import * as PgTypes from "./PgTypes.ts"
38
39	/**
40	* The runtime type identifier for `PgConnection`.
41	*
42	* @category type IDs
43	* @since 4.0.0
44	*/
45	export const TypeId: TypeId = "~@effect/sql-pg/PgConnection"
46
47	/**
48	* The type-level identifier for `PgConnection`.
49	*
50	* @category type IDs
51	* @since 4.0.0
52	*/
53	export type TypeId = "~@effect/sql-pg/PgConnection"
54
55	/**
56	* Connection settings for a PostgreSQL session.
57	*
58	* **Details**
59	*
60	* A `url` is parsed as a libpq URI (`postgres://` or `postgresql://`);
61	* explicit fields win over anything the URL carries. A `stream` factory wins
62	* over `host`, `port`, and `path`. A `path` is used verbatim as the unix
63	* socket path, while a `host` beginning with `/` is treated as a socket
64	* directory and expands to `${host}/.s.PGSQL.${port}`.
65	*
66	* URL modes `sslmode=prefer` and `sslmode=allow` try TLS first, falling back
67	* to plaintext only when the server answers `SSLRequest` with `N`. Unlike
68	* libpq, `allow` also tries TLS first. Certificate verification stays enabled
69	* unless explicitly disabled through `ssl` options.
70	*
71	* Prepared statements are enabled by default and limited by
72	* `preparedStatementCacheSize`. Disable them for statement-mode poolers or
73	* workloads that generate unique SQL. Streams always use unnamed statements.
74	*
75	* With `multiplex` enabled, unpinned queries from multiple fibers are
76	* pipelined. Transactions, streams, and listeners remain exclusive. An
77	* unpinned multiplexed connection cannot be interrupted because cancellation
78	* could affect another fiber's query.
79	*
80	* @category models
81	* @since 4.0.0
82	*/
83	export interface Config {
84	readonly url?: Redacted.Redacted | undefined
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	readonly connectTimeout?: Duration.Input | undefined
98	/**
99	* Overrides `startupParameters.application_name`, the URL's `application_name`,
100	* and the default `"@effect/sql-pg"`, in that order.
101	*/
102	readonly applicationName?: string | undefined
103	/**
104	* Session defaults sent in every physical connection's startup packet.
105	* Names are lowercased; `user`, `database`, `replication`, and `options` are
106	* reserved. `client_encoding` only accepts UTF8 / UTF-8 (case-insensitive).
107	* Empty names and NUL bytes fail before connecting; PostgreSQL validates
108	* other settings. Do not set the same GUC here and in `startupOptions`.
109	*/
110	readonly startupParameters?: Readonly<Record<string, string>> | undefined
111	/**
112	* Opaque PostgreSQL startup options, overriding the URL's `options` parameter.
113	* Forwarded without parsing `-c` flags or checking for duplicate GUCs.
114	*/
115	readonly startupOptions?: string | undefined
116	readonly stream?: (() => Duplex) | undefined
117	readonly types?: PgTypes.Registry | undefined
118	readonly multiplex?: boolean | undefined
119	readonly prepare?: boolean | undefined
120	readonly preparedStatementCacheSize?: number | undefined
121	/** Maximum backend message size in bytes. Defaults to 16 MiB. */
122	readonly maxMessageSize?: number | undefined
123	}
124
125	/**
126	* Builds the prepared-statement cache a session should use, or `undefined`
127	* when `prepare` is off or the cache is sized to nothing.
128	*/
129	const preparedCacheFor = (config: Config): PreparedCache | undefined => {
130	if (config.prepare === false) return undefined
131	const max = config.preparedStatementCacheSize ?? defaultPreparedStatements
132	return max > 0 ? new PreparedCache(max) : undefined
133	}
134
135	/**
136	* An object result row keyed by column name.
137	*
138	* @category models
139	* @since 4.0.0
140	*/
141	export interface Row {
142	readonly [column: string]: unknown
143	}
144
145	/**
146	* Metadata for one result column.
147	*
148	* @category models
149	* @since 4.0.0
150	*/
151	export interface Field {
152	readonly name: string
153	readonly dataTypeId: number
154	}
155
156	/**
157	* The result of a query.
158	*
159	* @category models
160	* @since 4.0.0
161	*/
162	export interface Result {
163	readonly command: string
164	readonly rowCount: number
165	readonly oid: number | null
166	readonly rows: ReadonlyArray<Row>
167	readonly fields: ReadonlyArray<Field>
168	}
169
170	/**
171	* A `NOTIFY` message received while listening on a channel.
172	*
173	* @category models
174	* @since 4.0.0
175	*/
176	export interface Notification {
177	readonly processId: number
178	readonly channel: string
179	readonly payload: string
180	}
181
182	/**
183	* A connected and authenticated PostgreSQL session.
184	*
185	* @category models
186	* @since 4.0.0
187	*/
188	export interface PgConnection {
189	readonly [TypeId]: TypeId
190	readonly config: Config
191	readonly processId: number
192	/**
193	* Reserves the session for exclusive use until the scope closes. Pinning is
194	* reentrant. Calls through the returned connection skip the
195	* ownership queue, while calls through the original connection wait.
196	*/
197	readonly pin: Effect.Effect<PgConnection, never, Scope.Scope>
198	/**
199	* Runs a query and returns rows keyed by column name. Pass `false` to skip
200	* the prepared statement cache.
201	*
202	* **Details**
203	*
204	* On interruption, the connection drains to `ReadyForQuery` and sends a
205	* `CancelRequest` if needed. Unless the backend confirms cancellation with
206	* `57014`, a pool retires the session before its next checkout. A retained
207	* checkout and an unpooled session remain exposed to a late cancel.
208	* `statement_timeout` also reports `57014` and can be mistaken for
209	* confirmation.
210	*/
211	readonly query: (
212	sql: string,
213	params?: ReadonlyArray<unknown>,
214	prepare?: boolean
215	) => Effect.Effect<Result, SqlError>
216	/**
217	* Runs a query and returns positional rows. Pass `false` to skip the
218	* prepared statement cache. Interruption behaves as for `query`.
219	*/
220	readonly queryValues: (
221	sql: string,
222	params?: ReadonlyArray<unknown>,
223	prepare?: boolean
224	) => Effect.Effect<ReadonlyArray<ReadonlyArray<unknown>>, SqlError>
225	/**
226	* Streams rows without collecting the full result. The session is pinned for
227	* the lifetime of the stream. An early abort behaves like an interrupted
228	* `query`.
229	*/
230	readonly stream: (
231	sql: string,
232	params?: ReadonlyArray<unknown>
233	) => Stream.Stream<Row, SqlError>
234	/**
235	* Registers a channel listener and returns its notification queue after
236	* PostgreSQL confirms `LISTEN`. The session stays pinned until the scope
237	* closes, when it runs `UNLISTEN` and shuts down the queue. PostgreSQL
238	* registration errors fail the acquiring effect.
239	* Connection failures after registration fail the queue with the original
240	* `SqlError`. Intentional scope closure interrupts consumers.
241	*/
242	readonly listen: (
243	channel: string
244	) => Effect.Effect<Queue.Dequeue<Notification, SqlError>, SqlError, Scope.Scope>
245	/**
246	* Attempts to cancel the active query through a side connection. This is a
247	* no-op for an unpinned multiplexed connection because the active
248	* query may belong to another fiber. The effect never fails.
249	*/
250	readonly interrupt: Effect.Effect<void>
251	}
252
253	/**
254	* The service tag for `PgConnection`.
255	*
256	* @category services
257	* @since 4.0.0
258	*/
259	export const PgConnection = Context.Service<PgConnection>("@effect/sql-pg/PgConnection")
260
261	/**
262	* Connects and authenticates a single PostgreSQL session.
263	*
264	* **Details**
265	*
266	* Password resolution, transport, optional `SSLRequest`, startup, and
267	* authentication run under `connectTimeout` (5 seconds by default). The effect
268	* resolves when the backend sends `ReadyForQuery`. Closing the scope sends
269	* `Terminate` and ends the socket.
270	*
271	* Use `sslmode=require` or explicit `ssl: true` to require encryption.
272	* Unix sockets and custom streams should set `ssl.servername` explicitly.
273	*
274	* @category constructors
275	* @since 4.0.0
276	*/
277	export const make = (options: Config): Effect.Effect<PgConnection, SqlError, Scope.Scope> =>
278	Effect.flatMap(resolveConfig(options), (config) =>
279	Effect.acquireRelease(
280	Effect.gen(function*() {
281	const password = yield* PasswordInternal.resolve(config.password)
282	const session = yield* connect(config, password)
283	return new PgConnectionImpl(options, config, session, options.types)
284	}),
285	(connection) => Effect.sync(() => connection.closeUnsafe()),
286	{ interruptible: true }
287	).pipe(
288	Effect.timeoutOrElse({
289	duration: config.connectTimeout,
290	orElse: () =>
291	Effect.fail(
292	new SqlError({
293	reason: new ConnectionError({
294	cause: new Error("Connection timed out"),
295	message: "PgConnection: Connection timed out",
296	operation: "connect"
297	})
298	})
299	)
300	})
301	))
302
303	interface Session {
304	readonly socket: Duplex
305	readonly encrypted: boolean
306	readonly parser: PgProtocol.Parser<unknown>
307	readonly processId: number
308	readonly secretKey: number
309	}
310
311	/**
312	* The active protocol consumer: the state machine of the statement currently
313	* on the wire. Messages the pump does not handle itself are forwarded here.
314	*/
315	interface Consumer {
316	readonly onMessage: (message: PgProtocol.BackendMessage<unknown>) => void
317	readonly onBatchEnd?: (() => void) | undefined
318	readonly onFatal: (error: SqlError) => void
319	}
320
321	/** One statement waiting in, or travelling through, a multiplexed pipeline. */
322	interface PipelineEntry {
323	readonly plan: Plan
324	readonly deferred: Deferred.Deferred<QueryOutput, SqlError>
325	readonly machine: QueryMachine
326	abandoned: boolean
327	}
328
329	const abortDrainTimeoutMillis = 5000
330	const abortDrainGraceMillis = 10
331	/** PostgreSQL `query_canceled` SQLSTATE. */
332	const queryCanceledCode = "57014"
333	const cancelRequestTimeoutMillis = 5000
334	/** How many statements a multiplexed session keeps on the wire at once. */
335	const maxPipelineDepth = 128
336	const streamPauseThreshold = 512
337
338	class PgConnectionImpl implements PgConnection {
339	readonly [TypeId]: TypeId = TypeId
340	readonly config: Config
341	readonly processId: number
342	readonly session: Session
343	/** The custom codec registry, or `undefined` for the builtin catalogue. */
344	readonly registry: PgTypes.Registry | undefined
345	readonly bindEncoder: BindEncoder
346	/** The statements this session has named, or `undefined` when disabled. */
347	readonly prepared: PreparedCache | undefined
348	readonly resolved: ResolvedConfig
349	readonly multiplex: boolean
350	/** Serializes statements: one in-flight extended-query cycle. */
351	readonly wire = Semaphore.makeUnsafe(1)
352	/** Exclusive-ownership queue used by `pin` and unpinned statements. */
353	readonly owner = Semaphore.makeUnsafe(1)
354	pinned = false
355	consumer: Consumer | undefined
356	deadWith: SqlError | undefined
357	closed = false
358	cancelPending = false
359	readonly channels = new Map<string, Set<Queue.Queue<Notification, SqlError>>>()
360	readonly retireHooks = new Set<() => void>()
361	/** Queued but not yet written; drained into `pipelineInFlight` on flush. */
362	readonly pipelinePending: Array<PipelineEntry> = []
363	/** Written and awaiting their `ReadyForQuery`, oldest first. */
364	readonly pipelineInFlight: Array<PipelineEntry> = []
365	pipelineHead = 0
366	pipelineFlushScheduled = false
367	readonly pipelineIdleWaiters = new Set<() => void>()
368	readonly pinnedView: PgConnection
369	readonly [internalsKey]: ConnectionInternals
370
371	constructor(config: Config, resolved: ResolvedConfig, session: Session, registry: PgTypes.Registry | undefined) {
372	this.config = config
373	this.resolved = resolved
374	this.session = session
375	this.processId = session.processId
376	this.registry = registry
377	this.bindEncoder = makeBindEncoder(registry)
378	this.prepared = preparedCacheFor(config)
379	this.multiplex = config.multiplex ?? false
380	// Before the pinned view, which copies it.
381	this[internalsKey] = {
382	base: this,
383	deadError: () => this.deadWith,
384	retireHooks: this.retireHooks
385	}
386	this.pinnedView = new PinnedPgConnection(this)
387	session.socket.on("data", this.onData)
388	session.socket.on("error", this.onSocketError)
389	session.socket.on("close", this.onSocketClose)
390	}
391
392	private readonly onData = (chunk: Uint8Array): void => {
393	try {
394	this.session.parser.pushEach(chunk, this.dispatch)
395	} catch (cause) {
396	// A field reader that threw reports the row it could not decode; anything
397	// else came out of the framing itself.
398	return this.fatal(
399	cause instanceof PgProtocol.ParseError
400	? connectionQueryError(cause, "PgConnection: Failed to parse server messages")
401	: queryError(cause, "PgConnection: Failed to decode row")
402	)
403	}
404	if (this.deadWith === undefined) this.consumer?.onBatchEnd?.()
405	}
406
407	private readonly onSocketError = (cause: Error): void =>
408	this.fatal(connectionQueryError(cause, "PgConnection: Socket error"))
409
410	private readonly onSocketClose = (): void =>
411	this.fatal(connectionQueryError(new Error("Connection closed"), "PgConnection: Connection closed"))
412
413	private readonly dispatch = (message: PgProtocol.BackendMessage<unknown>): void => {
414	if (this.deadWith !== undefined) return
415	switch (message._tag) {
416	case "ErrorResponse":
417	if (message.fields.code === queryCanceledCode) this.cancelPending = false
418	break
419	case "ReadyForQuery":
420	if (this.cancelPending) {
421	this.cancelPending = false
422	this.retire()
423	}
424	break
425	case "NotificationResponse": {
426	const queues = this.channels.get(message.channel)
427	if (queues !== undefined) {
428	const notification: Notification = {
429	processId: message.pid,
430	channel: message.channel,
431	payload: message.payload
432	}
433	for (const queue of queues) Queue.offerUnsafe(queue, notification)
434	}
435	return
436	}
437	case "ParameterStatus":
438	case "NoticeResponse":
439	return
440	}
441	if (this.consumer !== undefined) return this.consumer.onMessage(message)
442	if (message._tag === "ErrorResponse") {
443	return this.fatal(
444	new SqlError({
445	reason: classifyFields(message.fields, "PgConnection: The server reported an error", "query")
446	})
447	)
448	}
449	this.fatal(
450	connectionQueryError(
451	new Error(`Unexpected ${message._tag} while idle`),
452	`PgConnection: Unexpected ${message._tag} while idle`
453	)
454	)
455	}
456
457	/** Marks the session dead: optionally destroys the socket, fails the active statement
458	* and every listen queue, and notifies pool hooks unless the session's own
459	* scope is being released. */
460	fatal(error: SqlError, destroySocket = true): void {
461	if (this.deadWith !== undefined) return
462	this.deadWith = error
463	if (destroySocket) this.session.socket.destroy()
464	const consumer = this.consumer
465	this.consumer = undefined
466	this.retire()
467	consumer?.onFatal(error)
468	const sets = Array.from(this.channels.values())
469	this.channels.clear()
470	const cause = this.closed ? Cause.interrupt() : Cause.fail(error)
471	for (const set of sets) {
472	for (const queue of set) Queue.failCauseUnsafe(queue, cause)
473	}
474	}
475
476	private retire(): void {
477	if (this.closed) return
478	for (const hook of this.retireHooks) hook()
479	}
480
481	closeUnsafe(): void {
482	this.closed = true
483	let destroySocket = true
484	if (this.deadWith === undefined && this.session.socket.writable) {
485	try {
486	this.session.socket.end(PgProtocol.encodeTerminate())
487	destroySocket = false
488	} catch {
489	// Fall back to the fatal path's immediate destroy below.
490	}
491	}
492	this.fatal(
493	new SqlError({
494	reason: new ConnectionError({
495	cause: new Error("Connection is closed"),
496	message: "PgConnection: Connection is closed",
497	operation: "query"
498	})
499	}),
500	destroySocket
501	)
502	}
503
504	/** Plans one execution. `cache` is `undefined` to force the unnamed path. */
505	readonly encodeQuery = (
506	sql: string,
507	params: ReadonlyArray<unknown>,
508	cache: PreparedCache | undefined
509	): Plan => encodeQuery(sql, params, this.registry, this.bindEncoder, cache)
510
511	/**
512	* Routes backend messages to the oldest statement still on the wire.
513	*
514	* Every cycle carries its own `Sync`, so the backend answers them in order
515	* and finishes each with a `ReadyForQuery`. That boundary is what advances
516	* the queue, and it is also why a statement the backend rejects only skips
517	* the rest of its own cycle: the ones queued behind it still run.
518	*/
519	private readonly pipelineConsumer: Consumer = {
520	onMessage: (message) => {
521	const entry = this.pipelineInFlight[this.pipelineHead]
522	if (entry === undefined) {
523	return this.fatal(connectionQueryError(
524	new Error(`Unexpected ${message._tag} without an in-flight query`),
525	`PgConnection: Unexpected ${message._tag} without an in-flight query`
526	))
527	}
528	entry.machine.onMessage(message)
529	},
530	onBatchEnd: () => this.flushPipeline(),
531	onFatal: (error) => {
532	const entries = [
533	...this.pipelineInFlight.slice(this.pipelineHead),
534	...this.pipelinePending
535	]
536	this.pipelineInFlight.length = 0
537	this.pipelineHead = 0
538	this.pipelinePending.length = 0
539	this.pipelineFlushScheduled = false
540	for (const entry of entries) {
541	if (!entry.abandoned) Deferred.doneUnsafe(entry.deferred, Effect.fail(error))
542	}
543	this.notifyPipelineIdle()
544	}
545	}
546
547	private readonly pipelineDepth = (): number => this.pipelineInFlight.length - this.pipelineHead
548
549	private readonly pipelineIsIdle = (): boolean => this.pipelineDepth() === 0 && this.pipelinePending.length === 0
550
551	private readonly notifyPipelineIdle = (): void => {
552	if (!this.pipelineIsIdle()) return
553	this.consumer = undefined
554	this.session.parser.readField = undefined
555	const waiters = Array.from(this.pipelineIdleWaiters)
556	this.pipelineIdleWaiters.clear()
557	for (const waiter of waiters) waiter()
558	}
559
560	/** Resolves once nothing is on the wire, so `pin` can take the connection. */
561	readonly waitPipelineIdle: Effect.Effect<void> = Effect.callback((resume) => {
562	if (this.pipelineIsIdle()) return resume(Effect.void)
563	const waiter = () => resume(Effect.void)
564	this.pipelineIdleWaiters.add(waiter)
565	return Effect.sync(() => this.pipelineIdleWaiters.delete(waiter))
566	})
567
568	private readonly finishPipelineEntry = (
569	entry: PipelineEntry,
570	result: Effect.Effect<QueryOutput, SqlError>
571	): void => {
572	if (this.pipelineInFlight[this.pipelineHead] !== entry) {
573	return this.fatal(connectionQueryError(
574	new Error("A pipelined query completed out of order"),
575	"PgConnection: A pipelined query completed out of order"
576	))
577	}
578	this.pipelineHead++
579	// The next statement's rows can be in this same chunk, so its reader has to
580	// be in place before the parser reads on.
581	this.session.parser.readField = this.pipelineInFlight[this.pipelineHead]?.machine.readField
582	if (!entry.abandoned) Deferred.doneUnsafe(entry.deferred, result)
583	if (this.pipelineHead === this.pipelineInFlight.length) {
584	this.pipelineInFlight.length = 0
585	this.pipelineHead = 0
586	}
587	}
588
589	/** Writes everything queued since the last flush as one batch. */
590	private readonly flushPipeline = (): void => {
591	this.pipelineFlushScheduled = false
592	if (this.deadWith !== undefined) return
593	let capacity = maxPipelineDepth - this.pipelineDepth()
594	if (capacity <= 0) return
595	const wasEmpty = this.pipelineDepth() === 0
596	const batch: Array<PipelineEntry> = []
597	let index = 0
598	while (capacity > 0 && index < this.pipelinePending.length) {
599	const entry = this.pipelinePending[index++]
600	if (entry.abandoned) {
601	if (entry.plan.parses && entry.plan.prepared !== undefined) {
602	entry.plan.prepared.parsing = false
603	}
604	continue
605	}
606	batch.push(entry)
607	this.pipelineInFlight.push(entry)
608	capacity--
609	}
610	this.pipelinePending.splice(0, index)
611	if (batch.length === 0) {
612	this.notifyPipelineIdle()
613	return
614	}
615	if (wasEmpty) this.session.parser.readField = batch[0].machine.readField
616	const frame = batch.length === 1
617	? batch[0].plan.frame
618	: concat(batch.map((entry) => entry.plan.frame))
619	try {
620	this.session.socket.write(frame)
621	} catch (cause) {
622	this.fatal(connectionQueryError(cause, "PgConnection: Failed to write query batch"))
623	}
624	}
625
626	private readonly schedulePipelineFlush = (): void => {
627	if (this.pipelineFlushScheduled || this.pipelineDepth() >= maxPipelineDepth) return
628	this.pipelineFlushScheduled = true
629	// A microtask is what lets fibers submitted in the same tick share a write.
630	queueMicrotask(this.flushPipeline)
631	}
632
633	/**
634	* One cycle on a multiplexed connection. The `owner` permit is only taken to
635	* queue the statement, so a pin still orders against submissions without
636	* serializing them.
637	*/
638	private readonly pipelineCycle = (
639	sql: string,
640	params: ReadonlyArray<unknown>,
641	wantRows: boolean,
642	cache: PreparedCache | undefined
643	): Effect.Effect<QueryOutput, SqlError> =>
644	Effect.flatMap(
645	this.owner.withPermit(Effect.suspend(() => {
646	if (this.deadWith !== undefined) return Effect.fail(this.deadWith)
647	let plan: Plan
648	try {
649	plan = this.encodeQuery(sql, params, cache)
650	} catch (cause) {
651	return Effect.fail(queryError(cause, "PgConnection: Failed to encode query"))
652	}
653	const deferred = Deferred.makeUnsafe<QueryOutput, SqlError>()
654	let entry: PipelineEntry
655	const machine = new QueryMachine(this, plan, wantRows, (result) => this.finishPipelineEntry(entry, result))
656	entry = { plan, deferred, machine, abandoned: false }
657	this.pipelinePending.push(entry)
658	this.consumer = this.pipelineConsumer
659	this.schedulePipelineFlush()
660	return Effect.succeed(entry)
661	})),
662	(entry) => {
663	const awaited = Deferred.await(entry.deferred).pipe(
664	Effect.onInterrupt(() =>
665	Effect.sync(() => {
666	entry.abandoned = true
667	})
668	)
669	)
670	if (cache === undefined || entry.plan.parses) return awaited
671	return retryStale(entry.plan, cache, awaited, () => this.pipelineCycle(sql, params, wantRows, undefined))
672	}
673	)
674
675	/**
676	* One extended-query cycle for a caller that shares the session with others:
677	* the fibers inside a transaction, and a stream running beside them. The
678	* wire permit is what serializes them.
679	*/
680	readonly cycle = (
681	sql: string,
682	params: ReadonlyArray<unknown>,
683	wantRows: boolean,
684	cache: PreparedCache | undefined
685	): Effect.Effect<QueryOutput, SqlError> => this.wire.withPermit(this.cycleOwned(sql, params, wantRows, cache))
686
687	/**
688	* Runs one cycle for a caller that owns the session. No wire permit is needed
689	* because competing pinned work and streams must acquire `owner` first.
690	*/
691	readonly cycleOwned = (
692	sql: string,
693	params: ReadonlyArray<unknown>,
694	wantRows: boolean,
695	cache: PreparedCache | undefined
696	): Effect.Effect<QueryOutput, SqlError> => Effect.suspend(() => this.attempt(sql, params, wantRows, cache))
697
698	/**
699	* Runs one cycle. A reused statement the backend has since dropped, or whose
700	* plan no longer matches its columns, is parsed again on a second attempt;
701	* that attempt skips the cache, so it cannot loop.
702	*/
703	private readonly attempt = (
704	sql: string,
705	params: ReadonlyArray<unknown>,
706	wantRows: boolean,
707	cache: PreparedCache | undefined
708	): Effect.Effect<QueryOutput, SqlError> => {
709	if (this.deadWith !== undefined) return Effect.fail(this.deadWith)
710	let plan: Plan
711	try {
712	plan = this.encodeQuery(sql, params, cache)
713	} catch (cause) {
714	return Effect.fail(queryError(cause, "PgConnection: Failed to encode query"))
715	}
716	const run = runQuery(this, plan, wantRows)
717	if (cache === undefined || plan.parses) return run
718	return retryStale(plan, cache, run, () => this.attempt(sql, params, wantRows, undefined))
719	}
720
721	/** Sends a `CancelRequest` for this session on a side connection. */
722	readonly cancel: Effect.Effect<void> = Effect.suspend(() => {
723	if (this.deadWith !== undefined) return Effect.void
724	this.cancelPending = true
725	if (this.consumer === undefined) this.retire()
726	return sendCancelRequest(this.resolved, this.session)
727	})
728
729	readonly pin: Effect.Effect<PgConnection, never, Scope.Scope> = Effect.suspend(() => {
730	const reserve = this[internalsKey].reserve
731	return reserve === undefined ? this.pinExclusive : Effect.andThen(reserve, this.pinExclusive)
732	})
733
734	private readonly pinExclusive: Effect.Effect<PgConnection, never, Scope.Scope> = Effect.acquireRelease(
735	Effect.flatMap(this.owner.take(1), () =>
736	// Holding `owner` stops new submissions; a pipeline already on the wire
737	// still has to drain before this fiber owns the connection.
738	Effect.as(
739	Effect.tap(
740	Effect.onInterrupt(this.waitPipelineIdle, () => this.owner.release(1)),
741	() =>
742	Effect.sync(() => {
743	this.pinned = true
744	})
745	),
746	this.pinnedView
747	)),
748	() =>
749	Effect.suspend(() => {
750	this.pinned = false
751	return this.owner.release(1)
752	})
753	)
754
755	private readonly run = (
756	sql: string,
757	params: ReadonlyArray<unknown>,
758	wantRows: boolean,
759	prepare: boolean
760	): Effect.Effect<QueryOutput, SqlError> => {
761	const cache = prepare ? this.prepared : undefined
762	return this.multiplex
763	? this.pipelineCycle(sql, params, wantRows, cache)
764	: this.owner.withPermit(this.cycleOwned(sql, params, wantRows, cache))
765	}
766
767	readonly query = (sql: string, params?: ReadonlyArray<unknown>, prepare = true): Effect.Effect<Result, SqlError> =>
768	Effect.map(this.run(sql, params ?? emptyParams, true, prepare), takeResult)
769
770	readonly queryValues = (
771	sql: string,
772	params?: ReadonlyArray<unknown>,
773	prepare = true
774	): Effect.Effect<ReadonlyArray<ReadonlyArray<unknown>>, SqlError> =>
775	Effect.map(this.run(sql, params ?? emptyParams, false, prepare), takeValues)
776
777	readonly stream = (sql: string, params?: ReadonlyArray<unknown>): Stream.Stream<Row, SqlError> =>
778	streamRows(this, this.pin, sql, params ?? emptyParams)
779
780	readonly listen = (
781	channel: string
782	): Effect.Effect<Queue.Dequeue<Notification, SqlError>, SqlError, Scope.Scope> =>
783	listenChannel(this, this.pin, channel)
784
785	readonly interrupt: Effect.Effect<void> = Effect.suspend(() =>
786	this.multiplex && !this.pinned ? Effect.void : this.cancel
787	)
788	}
789
790	/**
791	* The view of a session returned by `pin`: statements skip the ownership
792	* queue and re-pinning is a no-op, making `pin` reentrant for `stream` and
793	* `listen` running inside a transaction.
794	*/
795	class PinnedPgConnection implements PgConnection {
796	readonly [TypeId]: TypeId = TypeId
797	readonly base: PgConnectionImpl
798	readonly [internalsKey]: ConnectionInternals
799	readonly pin: Effect.Effect<PgConnection, never, Scope.Scope>
800	readonly interrupt: Effect.Effect<void>
801
802	constructor(base: PgConnectionImpl) {
803	this.base = base
804	this[internalsKey] = base[internalsKey]
805	this.pin = Effect.succeed(this)
806	this.interrupt = base.cancel
807	}
808
809	get config(): Config {
810	return this.base.config
811	}
812
813	get processId(): number {
814	return this.base.processId
815	}
816
817	readonly query = (sql: string, params?: ReadonlyArray<unknown>, prepare = true): Effect.Effect<Result, SqlError> =>
818	Effect.map(this.base.cycle(sql, params ?? emptyParams, true, prepare ? this.base.prepared : undefined), takeResult)
819
820	readonly queryValues = (
821	sql: string,
822	params?: ReadonlyArray<unknown>,
823	prepare = true
824	): Effect.Effect<ReadonlyArray<ReadonlyArray<unknown>>, SqlError> =>
825	Effect.map(this.base.cycle(sql, params ?? emptyParams, false, prepare ? this.base.prepared : undefined), takeValues)
826
827	readonly stream = (sql: string, params?: ReadonlyArray<unknown>): Stream.Stream<Row, SqlError> =>
828	streamRows(this.base, this.pin, sql, params ?? emptyParams)
829
830	readonly listen = (
831	channel: string
832	): Effect.Effect<Queue.Dequeue<Notification, SqlError>, SqlError, Scope.Scope> =>
833	listenChannel(this.base, this.pin, channel)
834	}
835
836	interface QueryOutput {
837	readonly result: Result
838	readonly values: ReadonlyArray<ReadonlyArray<unknown>>
839	}
840
841	const INT32_MIN = -2147483648
842	const INT32_MAX = 2147483647
843
844	const inferredParameter = (oid: number, value: unknown): PgTypes.Parameter => ({
845	[PgTypes.ParameterTypeId]: PgTypes.ParameterTypeId,
846	oid,
847	value
848	})
849
850	const inferScalar = (value: unknown): PgTypes.Parameter => {
851	if (PgTypes.isParameter(value)) return value
852	if (value === null || value === undefined) return inferredParameter(0, null)
853	switch (typeof value) {
854	case "boolean":
855	return inferredParameter(PgTypes.OID.bool, value)
856	case "bigint":
857	return inferredParameter(PgTypes.OID.int8, value)
858	case "number":
859	if (Number.isInteger(value)) {
860	if (value >= INT32_MIN && value <= INT32_MAX) {
861	return inferredParameter(PgTypes.OID.int4, value)
862	}
863	if (Number.isSafeInteger(value)) {
864	return inferredParameter(PgTypes.OID.int8, BigInt(value))
865	}
866	}
867	return inferredParameter(PgTypes.OID.float8, value)
868	case "string":
869	// Bound with no concrete type, as a text-format literal, so the backend
870	// derives the type from the statement: a string works against a bigint
871	// or timestamp column the way it did with the text-protocol drivers.
872	return inferredParameter(0, value)
873	}
874	if (value instanceof Date) return inferredParameter(PgTypes.OID.timestamptz, value)
875	if (value instanceof Uint8Array) return inferredParameter(PgTypes.OID.bytea, value)
876	if (value instanceof Int8Array) {
877	return inferredParameter(
878	PgTypes.OID.bytea,
879	new Uint8Array(value.buffer, value.byteOffset, value.byteLength)
880	)
881	}
882	throw new PgTypes.CodecError({ message: `Cannot infer a PostgreSQL type for ${String(value)}` })
883	}
884
885	const inferParameter = (value: unknown, registry: PgTypes.Registry | undefined): PgTypes.Parameter => {
886	if (!Array.isArray(value)) return inferScalar(value)
887	if (value.length === 0) {
888	throw new PgTypes.CodecError({ message: "Cannot infer the type of an empty array; use PgTypes.array" })
889	}
890	let elementOid: number | undefined
891	const values: Array<unknown> = new Array(value.length)
892	for (let index = 0; index < value.length; index++) {
893	const element = value[index]
894	if (Array.isArray(element)) {
895	throw new PgTypes.CodecError({ message: "Nested array parameters are not supported" })
896	}
897	const parameter = inferScalar(element)
898	// A scalar string binds untyped, but an array names its element type.
899	let oid = parameter.oid
900	if (oid === 0) {
901	if (parameter.value === null) {
902	values[index] = null
903	continue
904	}
905	oid = PgTypes.OID.text
906	}
907	if (Array.isArray(parameter.value)) {
908	throw new PgTypes.CodecError({ message: "Nested array parameters are not supported" })
909	}
910	if (elementOid === undefined) elementOid = oid
911	else if (elementOid !== oid) {
912	throw new PgTypes.CodecError({ message: "Array parameter elements must have the same inferred OID" })
913	}
914	values[index] = parameter.value
915	}
916	if (elementOid === undefined) {
917	throw new PgTypes.CodecError({ message: "Cannot infer the type of an array containing only null values" })
918	}
919	const arrayOid = PgTypes.arrayOidFor(elementOid, registry)
920	if (arrayOid === undefined) {
921	throw new PgTypes.CodecError({ message: `No array type known for element OID ${elementOid}` })
922	}
923	return inferredParameter(arrayOid, values)
924	}
925
926	/**
927	* Joins the parts of a frame into the buffer that goes on the wire.
928	*
929	* The buffer comes from Node's pool and can be passed directly to
930	* `socket.write`. Every byte is initialized before the buffer is returned.
931	*/
932	const concat = (chunks: ReadonlyArray<Uint8Array>): Uint8Array => {
933	let length = 0
934	for (let index = 0; index < chunks.length; index++) length += chunks[index].length
935	const output = Buffer.allocUnsafe(length)
936	let offset = 0
937	for (let index = 0; index < chunks.length; index++) {
938	output.set(chunks[index], offset)
939	offset += chunks[index].length
940	}
941	return output
942	}
943
944	const emptyParams: ReadonlyArray<unknown> = []
945
946	const takeResult = (output: QueryOutput): Result => output.result
947	const takeValues = (output: QueryOutput): ReadonlyArray<ReadonlyArray<unknown>> => output.values
948
949	/**
950	* The tail of every extended-query frame. `Describe` names the unnamed portal,
951	* `Execute` runs it without a row limit, and `Sync` closes the cycle; none of
952	* the three carries per-query state, so the bytes are encoded once.
953	*/
954	const describeExecuteSync: Uint8Array = concat([
955	PgProtocol.encodeDescribe({ target: "portal", name: "" }),
956	PgProtocol.encodeExecute({ portal: "", maxRows: 0 }),
957	PgProtocol.encodeSync()
958	])
959
960	/** The same tail for a statement whose columns are already known. */
961	const executeSync: Uint8Array = concat([
962	PgProtocol.encodeExecute({ portal: "", maxRows: 0 }),
963	PgProtocol.encodeSync()
964	])
965
966	/** The default number of statements a connection keeps prepared. */
967	const defaultPreparedStatements = 100
968
969	/**
970	* A statement the backend has parsed and holds under a name.
971	*
972	* `ready` turns true once `ParseComplete` confirms the backend has it and the
973	* first execution has reported its columns. Until then the entry is treated as
974	* a miss, which is safe because one cycle runs at a time. A `ready` entry with
975	* no `description` describes a statement that returns no rows.
976	*/
977	interface Prepared {
978	readonly name: string
979	readonly key: string
980	ready: boolean
981	/** A cycle carrying this statement's `Parse` is on the wire. */
982	parsing: boolean
983	description: Description | undefined
984	}
985
986	/** One planned execution: the bytes to write and what to expect back. */
987	interface Plan {
988	readonly frame: Uint8Array
989	/** `CloseComplete` messages to consume before the cycle proper. */
990	readonly closes: number
991	/** Whether the frame carries a `Parse`. */
992	readonly parses: boolean
993	/** Whether the frame carries a `Describe`, so the columns arrive on the wire. */
994	readonly describes: boolean
995	/** The statement being filled in or reused, if this execution names one. */
996	readonly prepared: Prepared | undefined
997	/** Set when the columns were already known, so no `RowDescription` is coming. */
998	readonly description: Description | undefined
999	/** Set by the cycle when the backend rejected the name or the cached plan. */
1000	stale: boolean
1001	}
1002
1003	/** The unnamed path: `Parse` / `Bind` / `Describe` / `Execute` / `Sync`. */
1004	const encodeUnnamed = (
1005	sql: string,
1006	parameters: ReadonlyArray<PgTypes.Parameter>,
1007	parameterTypes: ReadonlyArray<number>,
1008	encodeBind: BindEncoder
1009	): Plan => {
1010	const parse = PgProtocol.encodeParse({ name: "", query: sql, parameterTypes })
1011	if (EffectResult.isFailure(parse)) throw parse.failure
1012	const bind = encodeBind({ portal: "", statement: "", parameters })
1013	if (EffectResult.isFailure(bind)) throw bind.failure
1014	return {
1015	frame: concat([parse.success, bind.success, describeExecuteSync]),
1016	closes: 0,
1017	parses: true,
1018	describes: true,
1019	prepared: undefined,
1020	description: undefined,
1021	stale: false
1022	}
1023	}
1024
1025	/**
1026	* Builds one extended-query cycle as a single buffer, so a statement costs one
1027	* socket write.
1028	*
1029	* A statement the backend already holds under a name needs only
1030	* `Bind` / `Execute` / `Sync`: no `Parse` for the backend to plan and no
1031	* `Describe`, because the columns came back the first time and are cached with
1032	* the name. Statements evicted from the cache ride along as `Close` messages
1033	* rather than paying a round trip of their own.
1034	*/
1035	const encodeQuery = (
1036	sql: string,
1037	params: ReadonlyArray<unknown>,
1038	registry: PgTypes.Registry | undefined,
1039	encodeBind: BindEncoder,
1040	cache: PreparedCache | undefined
1041	): Plan => {
1042	const count = params.length
1043	const parameters: Array<PgTypes.Parameter> = new Array(count)
1044	const parameterTypes: Array<number> = new Array(count)
1045	for (let index = 0; index < count; index++) {
1046	const parameter = inferParameter(params[index], registry)
1047	parameters[index] = parameter
1048	parameterTypes[index] = parameter.oid
1049	}
1050
1051	if (cache === undefined) {
1052	return encodeUnnamed(sql, parameters, parameterTypes, encodeBind)
1053	}
1054
1055	const prepared = cache.get(sql, parameterTypes)
1056	if (!prepared.ready && prepared.parsing) {
1057	// Another cycle is already on the wire carrying this name's `Parse`.
1058	// Naming it again would collide, and reusing it would mean binding to
1059	// columns nobody has seen yet, so this execution goes unnamed.
1060	return encodeUnnamed(sql, parameters, parameterTypes, encodeBind)
1061	}
1062	const bind = encodeBind({ portal: "", statement: prepared.name, parameters })
1063	if (EffectResult.isFailure(bind)) throw bind.failure
1064	const parse = prepared.ready ? undefined : PgProtocol.encodeParse({ name: prepared.name, query: sql, parameterTypes })
1065	if (parse !== undefined && EffectResult.isFailure(parse)) throw parse.failure
1066	const closeFrames = cache.takeCloses()
1067
1068	if (parse === undefined) {
1069	return {
1070	frame: closeFrames === undefined
1071	? concat([bind.success, executeSync])
1072	: concat([closeFrames.frames, bind.success, executeSync]),
1073	closes: closeFrames?.count ?? 0,
1074	parses: false,
1075	describes: false,
1076	prepared,
1077	description: prepared.description,
1078	stale: false
1079	}
1080	}
1081
1082	prepared.parsing = true
1083	return {
1084	frame: closeFrames === undefined
1085	? concat([parse.success, bind.success, describeExecuteSync])
1086	: concat([closeFrames.frames, parse.success, bind.success, describeExecuteSync]),
1087	closes: closeFrames?.count ?? 0,
1088	parses: true,
1089	describes: true,
1090	prepared,
1091	description: undefined,
1092	stale: false
1093	}
1094	}
1095
1096	/**
1097	* The statements one connection has prepared, keyed by SQL text and the
1098	* parameter OIDs inferred for it: the same text with differently typed
1099	* parameters is a different statement to the backend.
1100	*
1101	* The cache is bounded and evicts least-recently-used. An evicted statement is
1102	* closed on the backend, but its `Close` waits for the next statement to go
1103	* out rather than taking a round trip of its own.
1104	*/
1105	class PreparedCache {
1106	readonly max: number
1107	private readonly statements = new Map<string, Prepared>()
1108	private readonly namespace = randomBytes(8).toString("hex")
1109	private closes: Array<Uint8Array> | undefined
1110	private counter = 0
1111
1112	constructor(max: number) {
1113	this.max = max
1114	}
1115
1116	get(sql: string, parameterTypes: ReadonlyArray<number>): Prepared {
1117	const key = parameterTypes.length === 0 ? sql : `${sql}\u0000${parameterTypes.join(",")}`
1118	const found = this.statements.get(key)
1119	if (found !== undefined) {
1120	// Re-insert to move it to the end: `Map` iterates in insertion order, so
1121	// the first key is the least recently used one.
1122	this.statements.delete(key)
1123	this.statements.set(key, found)
1124	this.trim(found)
1125	return found
1126	}
1127	const prepared: Prepared = {
1128	name: `effect_${this.namespace}_${++this.counter}`,
1129	key,
1130	ready: false,
1131	parsing: false,
1132	description: undefined
1133	}
1134	this.statements.set(key, prepared)
1135	this.trim(prepared)
1136	return prepared
1137	}
1138
1139	/**
1140	* A statement still being parsed cannot be closed yet: its `Parse` may not
1141	* have reached the socket. Let the cache exceed its bound temporarily, then
1142	* discard abandoned entries and close excess ready statements on access.
1143	*/
1144	private trim(protectedStatement?: Prepared): void {
1145	while (this.statements.size > this.max) {
1146	let evicted: Prepared | undefined
1147	for (const prepared of this.statements.values()) {
1148	if (prepared !== protectedStatement && (prepared.ready || !prepared.parsing)) {
1149	evicted = prepared
1150	break
1151	}
1152	}
1153	if (evicted === undefined) return
1154	this.statements.delete(evicted.key)
1155	if (evicted.ready) this.close(evicted.name)
1156	}
1157	}
1158
1159	/**
1160	* Drops a statement the backend no longer holds, or whose plan went stale.
1161	*
1162	* The name is closed either way. A plan that went stale is still held under
1163	* it, and re-parsing would collide; a name the backend has already lost
1164	* ignores the `Close`, which Postgres treats as a success.
1165	*/
1166	evict(prepared: Prepared): void {
1167	if (prepared.parsing) return
1168	if (this.statements.delete(prepared.key) && prepared.ready) this.close(prepared.name)
1169	}
1170
1171	/**
1172	* Drops a statement whose parsing cycle failed before its columns arrived.
1173	*
1174	* A `Parse` that completed before the error outlives the failed cycle, so
1175	* the backend may hold the name while the entry can never become ready. The
1176	* name is closed either way: closing one the backend never registered is a
1177	* no-op.
1178	*/
1179	evictFailed(prepared: Prepared): void {
1180	this.statements.delete(prepared.key)
1181	this.close(prepared.name)
1182	}
1183
1184	private close(name: string): void {
1185	;(this.closes ??= []).push(PgProtocol.encodeClose({ target: "statement", name }))
1186	}
1187
1188	takeCloses(): { readonly frames: Uint8Array; readonly count: number } | undefined {
1189	const closes = this.closes
1190	if (closes === undefined) return undefined
1191	this.closes = undefined
1192	return { frames: closes.length === 1 ? closes[0] : concat(closes), count: closes.length }
1193	}
1194	}
1195
1196	/**
1197	* SQLSTATEs that mean "this name is no longer usable": the backend lost the
1198	* statement, or the plan behind it no longer matches the columns it was
1199	* prepared for. Both are recovered by parsing the statement again.
1200	*/
1201	const isStalePreparedStatement = (code: string | undefined): boolean => code === "26000" || code === "0A000"
1202
1203	type BindEncoder = (options: {
1204	readonly portal: string
1205	readonly statement: string
1206	readonly parameters: ReadonlyArray<PgTypes.Parameter>
1207	}) => EffectResult.Result<Uint8Array, PgProtocol.EncodeError | PgTypes.CodecError>
1208
1209	/**
1210	* The default builtin-catalogue encoder, shared by every connection without a
1211	* custom registry. Passing `PgTypes.writeParameter` itself engages
1212	* `makeBindEncoder`'s unsafe fast path.
1213	*/
1214	const defaultBindEncoder: BindEncoder = PgProtocol.makeBindEncoder(PgTypes.writeParameter, PgTypes.isTextFormat)
1215
1216	/** One bind encoder per registry, since building one allocates a closure. */
1217	const makeBindEncoder = (registry: PgTypes.Registry | undefined): BindEncoder =>
1218	registry === undefined
1219	? defaultBindEncoder
1220	: PgProtocol.makeBindEncoder(
1221	(sink: PgProtocol.ValueSink, parameter: PgTypes.Parameter) => PgTypes.writeParameter(sink, parameter, registry),
1222	PgTypes.isTextFormat
1223	)
1224
1225	const queryError = (cause: unknown, message: string): SqlError =>
1226	new SqlError({ reason: new UnknownError({ cause, message, operation: "query" }) })
1227
1228	/**
1229	* Retries a cycle whose named statement the backend no longer honors: the
1230	* statement is dropped from the cache and the retry runs unnamed, so it cannot
1231	* loop.
1232	*/
1233	const retryStale = (
1234	plan: Plan,
1235	cache: PreparedCache,
1236	run: Effect.Effect<QueryOutput, SqlError>,
1237	rerun: () => Effect.Effect<QueryOutput, SqlError>
1238	): Effect.Effect<QueryOutput, SqlError> =>
1239	Effect.catchCause(run, (cause) => {
1240	if (!plan.stale) return Effect.failCause(cause)
1241	cache.evict(plan.prepared!)
1242	return rerun()
1243	})
1244
1245	const connectionQueryError = (cause: unknown, message: string): SqlError =>
1246	new SqlError({ reason: new ConnectionError({ cause, message, operation: "query" }) })
1247
1248	const escapeIdentifier = (identifier: string): string => `"${identifier.replaceAll("\"", "\"\"")}"`
1249
1250	type QueryPhase = "close" | "parse" | "bind" | "describe" | "rows" | "complete" | "error"
1251
1252	/**
1253	* Splits a command tag such as `SELECT 3` or `INSERT 0 1`. Reading the two
1254	* spaces directly keeps a completed statement from allocating the parts array
1255	* that splitting on every space would.
1256	*/
1257	const parseCommandTag = (tag: string): { command: string; rowCount: number; oid: number | null } => {
1258	const firstSpace = tag.indexOf(" ")
1259	if (firstSpace < 0) return { command: tag, oid: null, rowCount: 0 }
1260	const command = tag.slice(0, firstSpace)
1261	const secondSpace = tag.indexOf(" ", firstSpace + 1)
1262	if (command === "INSERT" && secondSpace > 0) {
1263	return {
1264	command,
1265	oid: Number(tag.slice(firstSpace + 1, secondSpace)),
1266	rowCount: Number(tag.slice(secondSpace + 1))
1267	}
1268	}
1269	const last = tag.slice((secondSpace < 0 ? firstSpace : secondSpace) + 1)
1270	return { command, oid: null, rowCount: isDigits(last) ? Number(last) : 0 }
1271	}
1272
1273	const isDigits = (value: string): boolean => {
1274	if (value.length === 0) return false
1275	for (let index = 0; index < value.length; index++) {
1276	const code = value.charCodeAt(index)
1277	if (code < 48 || code > 57) return false
1278	}
1279	return true
1280	}
1281
1282	/**
1283	* Tracks one query cycle while the connection routes backend messages to it.
1284	* Each pipelined cycle keeps its own parser state and completes once.
1285	*/
1286	class QueryMachine implements Consumer {
1287	private readonly conn: PgConnectionImpl
1288	private readonly plan: Plan
1289	private readonly finish: (effect: Effect.Effect<QueryOutput, SqlError>) => void
1290	/** Installed on the parser while this machine is at the head of the queue. */
1291	readonly readField: PgProtocol.FieldReader<unknown> | undefined
1292	/** Object rows, or `undefined` when the caller asked for positional ones. */
1293	private readonly rows: Array<Row> | undefined
1294	private readonly values: Array<ReadonlyArray<unknown>> | undefined
1295
1296	private closes: number
1297	private phase: QueryPhase
1298	private fieldCount: number
1299	private resultFields: ReadonlyArray<Field>
1300	private rowBuilder: RowBuilder | undefined
1301	private command = ""
1302	private rowCount = 0
1303	private oid: number | null = null
1304	private failure: SqlError | undefined
1305	private done = false
1306	private aborted = false
1307	private drainDone: (() => void) | undefined
1308
1309	constructor(
1310	conn: PgConnectionImpl,
1311	plan: Plan,
1312	wantRows: boolean,
1313	finish: (effect: Effect.Effect<QueryOutput, SqlError>) => void
1314	) {
1315	this.conn = conn
1316	this.plan = plan
1317	this.finish = finish
1318	this.rows = wantRows ? [] : undefined
1319	this.values = wantRows ? undefined : []
1320	this.closes = plan.closes
1321	this.phase = plan.closes > 0 ? "close" : plan.parses ? "parse" : "bind"
1322	const description = plan.description
1323	this.readField = description?.readField
1324	this.fieldCount = description?.resultFields.length ?? 0
1325	this.resultFields = description?.resultFields ?? emptyFields
1326	this.rowBuilder = description?.rowBuilder
1327	}
1328
1329	isDone(): boolean {
1330	return this.done
1331	}
1332
1333	abort(onDrained: () => void): void {
1334	if (this.done) return onDrained()
1335	this.aborted = true
1336	this.drainDone = onDrained
1337	}
1338
1339	private complete(effect: Effect.Effect<QueryOutput, SqlError>): void {
1340	if (this.done) return
1341	this.done = true
1342	// Whether it parsed or not, this cycle no longer holds the name: success
1343	// has already marked it ready, failure leaves it to be parsed again.
1344	const prepared = this.plan.prepared
1345	if (prepared !== undefined) prepared.parsing = false
1346	this.finish(effect)
1347	}
1348
1349	private failDesync(message: string): void {
1350	this.conn.fatal(connectionQueryError(new Error(message), `PgConnection: ${message}`))
1351	}
1352
1353	onFatal(error: SqlError): void {
1354	if (this.done) return
1355	this.done = true
1356	const prepared = this.plan.prepared
1357	if (prepared !== undefined) prepared.parsing = false
1358	if (this.aborted) this.drainDone?.()
1359	else this.finish(Effect.fail(error))
1360	}
1361
1362	onMessage(message: PgProtocol.BackendMessage<unknown>): void {
1363	if (this.aborted) {
1364	if (message._tag === "ReadyForQuery") {
1365	this.done = true
1366	this.drainDone?.()
1367	}
1368	return
1369	}
1370	if (this.phase === "error") {
1371	if (message._tag === "ReadyForQuery") return this.complete(Effect.fail(this.failure!))
1372	return this.failDesync(`Unexpected ${message._tag} after ErrorResponse`)
1373	}
1374	switch (message._tag) {
1375	case "CloseComplete":
1376	if (this.phase !== "close") return this.failDesync(`Unexpected CloseComplete during ${this.phase}`)
1377	if (--this.closes === 0) this.phase = this.plan.parses ? "parse" : "bind"
1378	return
1379	case "ParseComplete":
1380	if (this.phase !== "parse") return this.failDesync(`Unexpected ParseComplete during ${this.phase}`)
1381	this.phase = "bind"
1382	return
1383	case "BindComplete":
1384	if (this.phase !== "bind") return this.failDesync(`Unexpected BindComplete during ${this.phase}`)
1385	this.phase = this.plan.describes ? "describe" : "rows"
1386	return
1387	case "RowDescription": {
1388	if (this.phase !== "describe") return this.failDesync(`Unexpected RowDescription during ${this.phase}`)
1389	const description = describe(message.fields, this.conn.registry)
1390	if (EffectResult.isFailure(description)) {
1391	return this.conn.fatal(queryError(description.failure, "PgConnection: Failed to decode row"))
1392	}
1393	this.fieldCount = message.fields.length
1394	this.resultFields = description.success.resultFields
1395	this.rowBuilder = description.success.rowBuilder
1396	// `pushEach` hands this description over before it reads the rows
1397	// behind it, including rows that arrived in the same chunk.
1398	this.conn.session.parser.readField = description.success.readField
1399	const prepared = this.plan.prepared
1400	if (prepared !== undefined) {
1401	prepared.description = description.success
1402	prepared.ready = true
1403	}
1404	this.phase = "rows"
1405	return
1406	}
1407	case "NoData": {
1408	if (this.phase !== "describe") return this.failDesync(`Unexpected NoData during ${this.phase}`)
1409	const prepared = this.plan.prepared
1410	if (prepared !== undefined) {
1411	prepared.description = undefined
1412	prepared.ready = true
1413	}
1414	this.phase = "rows"
1415	return
1416	}
1417	case "DataRow": {
1418	if (this.phase !== "rows" || this.rowBuilder === undefined) {
1419	return this.failDesync(`Unexpected DataRow during ${this.phase}`)
1420	}
1421	const rowValues = message.values
1422	if (rowValues.length !== this.fieldCount) {
1423	return this.failDesync(`DataRow has ${rowValues.length} values for ${this.fieldCount} fields`)
1424	}
1425	if (this.rows !== undefined) this.rows.push(this.rowBuilder(rowValues))
1426	else this.values!.push(rowValues)
1427	return
1428	}
1429	case "CommandComplete": {
1430	if (this.phase !== "rows") return this.failDesync(`Unexpected CommandComplete during ${this.phase}`)
1431	const parsed = parseCommandTag(message.commandTag)
1432	this.command = parsed.command
1433	this.rowCount = parsed.rowCount
1434	this.oid = parsed.oid
1435	this.phase = "complete"
1436	return
1437	}
1438	case "EmptyQueryResponse":
1439	if (this.phase !== "rows") return this.failDesync(`Unexpected EmptyQueryResponse during ${this.phase}`)
1440	this.phase = "complete"
1441	return
1442	case "ErrorResponse": {
1443	if (isStalePreparedStatement(message.fields.code)) this.plan.stale = true
1444	// A cycle that carried this statement's `Parse` but failed before its
1445	// columns arrived leaves an entry that can never become ready while
1446	// the backend may still hold the name: the `Parse` outlives the
1447	// failed cycle. Drop the entry and close the name, so the next
1448	// execution parses fresh under a new one.
1449	const prepared = this.plan.prepared
1450	if (this.plan.parses && prepared !== undefined && !prepared.ready) {
1451	this.conn.prepared?.evictFailed(prepared)
1452	}
1453	// Every phase drains the same way: the backend skips the rest of the
1454	// cycle and sends `ReadyForQuery` after the `Sync` that closes it, so
1455	// a statement it refused to parse leaves the session usable.
1456	this.failure = new SqlError({
1457	reason: classifyFields(message.fields, "PgConnection: Query failed", "query")
1458	})
1459	this.phase = "error"
1460	return
1461	}
1462	case "ReadyForQuery":
1463	if (this.phase !== "complete") return this.failDesync(`Unexpected ReadyForQuery during ${this.phase}`)
1464	return this.complete(Effect.succeed({
1465	result: {
1466	command: this.command,
1467	rowCount: this.rowCount,
1468	oid: this.oid,
1469	rows: this.rows ?? emptyRows,
1470	fields: this.resultFields
1471	},
1472	values: this.values ?? emptyValues
1473	}))
1474	case "CopyInResponse":
1475	case "CopyOutResponse":
1476	case "CopyBothResponse":
1477	case "CopyData":
1478	case "CopyDone":
1479	return this.failDesync(`Unexpected ${message._tag}; COPY is not supported`)
1480	default:
1481	return this.failDesync(`Unexpected ${message._tag} during ${this.phase}`)
1482	}
1483	}
1484	}
1485
1486	const emptyRows: ReadonlyArray<Row> = []
1487	const emptyValues: ReadonlyArray<ReadonlyArray<unknown>> = []
1488
1489	/** Drains an aborted statement, sending a cancel after a short grace period and failing if it stalls. */
1490	const drainAborted = (
1491	conn: PgConnectionImpl,
1492	timeoutMessage: string,
1493	abort: (onDrained: () => void) => void
1494	): Effect.Effect<void> => {
1495	const drained = Deferred.makeUnsafe<void>()
1496	const timeout = setTimeout(
1497	() => conn.fatal(connectionQueryError(new Error(timeoutMessage), `PgConnection: ${timeoutMessage}`)),
1498	abortDrainTimeoutMillis
1499	)
1500	abort(() => {
1501	clearTimeout(timeout)
1502	Deferred.doneUnsafe(drained, Effect.void)
1503	})
1504	const grace = Effect.callback<void>((resume) => {
1505	const timer = setTimeout(() => resume(Effect.void), abortDrainGraceMillis)
1506	return Effect.sync(() => clearTimeout(timer))
1507	})
1508	const cancelAndDrain = Effect.andThen(conn.cancel, Deferred.await(drained))
1509	return Effect.andThen(
1510	Effect.raceFirst(Deferred.await(drained), grace),
1511	Effect.suspend(() => Deferred.isDoneUnsafe(drained) ? Effect.void : cancelAndDrain)
1512	)
1513	}
1514
1515	/** One cycle on a connection that carries a single statement at a time. */
1516	const runQuery = (
1517	conn: PgConnectionImpl,
1518	plan: Plan,
1519	wantRows: boolean
1520	): Effect.Effect<QueryOutput, SqlError> =>
1521	Effect.callback<QueryOutput, SqlError>((resume) => {
1522	if (conn.deadWith !== undefined) {
1523	resume(Effect.fail(conn.deadWith))
1524	return
1525	}
1526	const machine = new QueryMachine(conn, plan, wantRows, (effect) => {
1527	conn.consumer = undefined
1528	resume(effect)
1529	})
1530	conn.consumer = machine
1531	// A reused statement has no `RowDescription` coming, so its reader has to
1532	// be in place before the rows are.
1533	conn.session.parser.readField = machine.readField
1534	try {
1535	conn.session.socket.write(plan.frame)
1536	} catch (cause) {
1537	conn.fatal(connectionQueryError(cause, "PgConnection: Failed to write query"))
1538	}
1539
1540	return Effect.suspend(() => {
1541	if (machine.isDone()) return Effect.void
1542	return drainAborted(conn, "Query cancellation timed out", (onDrained) =>
1543	machine.abort(() => {
1544	conn.consumer = undefined
1545	onDrained()
1546	}))
1547	})
1548	})
1549
1550	/** Turns one decoded row into an object keyed by column name. */
1551	type RowBuilder = (rowValues: ReadonlyArray<unknown>) => Row
1552
1553	/** Everything a query needs from one `RowDescription`. */
1554	interface Description {
1555	readonly readField: PgProtocol.FieldReader<unknown>
1556	readonly rowBuilder: RowBuilder
1557	readonly resultFields: ReadonlyArray<Field>
1558	}
1559
1560	const emptyFields: ReadonlyArray<Field> = []
1561
1562	/** Derives the per-column readers and the row constructor for a description. */
1563	const describe = (
1564	fields: ReadonlyArray<PgProtocol.FieldDescription>,
1565	registry: PgTypes.Registry | undefined
1566	): EffectResult.Result<Description, PgTypes.CodecError> => {
1567	const reader = PgTypes.makeFieldReader(fields, registry)
1568	if (EffectResult.isFailure(reader)) return EffectResult.fail(reader.failure)
1569	const resultFields: Array<Field> = new Array(fields.length)
1570	for (let index = 0; index < fields.length; index++) {
1571	resultFields[index] = { name: fields[index].name, dataTypeId: fields[index].dataTypeOid }
1572	}
1573	return EffectResult.succeed({
1574	readField: reader.success,
1575	rowBuilder: makeRowBuilder(fields),
1576	resultFields
1577	})
1578	}
1579
1580	/**
1581	* Builds the row constructor for one `RowDescription`.
1582	*
1583	* Assignment is an order of magnitude faster than `Object.defineProperty` and
1584	* stores the same own, enumerable, writable, configurable property - except
1585	* for `__proto__`, which assignment routes to the prototype setter instead. A
1586	* description carrying that column name falls back to the slow spelling.
1587	*/
1588	const makeRowBuilder = (fields: ReadonlyArray<PgProtocol.FieldDescription>): RowBuilder => {
1589	const names: Array<string> = new Array(fields.length)
1590	let hasProto = false
1591	for (let index = 0; index < fields.length; index++) {
1592	const name = fields[index].name
1593	names[index] = name
1594	if (name === "__proto__") hasProto = true
1595	}
1596	if (hasProto) {
1597	return (rowValues) => {
1598	const row: Record<string, unknown> = {}
1599	for (let index = 0; index < names.length; index++) {
1600	Object.defineProperty(row, names[index], {
1601	value: rowValues[index],
1602	enumerable: true,
1603	configurable: true,
1604	writable: true
1605	})
1606	}
1607	return row
1608	}
1609	}
1610	return (rowValues) => {
1611	const row: Record<string, unknown> = {}
1612	for (let index = 0; index < names.length; index++) {
1613	row[names[index]] = rowValues[index]
1614	}
1615	return row
1616	}
1617	}
1618
1619	const streamRows = (
1620	conn: PgConnectionImpl,
1621	pin: Effect.Effect<PgConnection, never, Scope.Scope>,
1622	sql: string,
1623	params: ReadonlyArray<unknown>
1624	): Stream.Stream<Row, SqlError> =>
1625	Stream.fromChannel(Channel.fromTransform(Effect.fnUntraced(function*(_, scope) {
1626	yield* Scope.provide(pin, scope)
1627	yield* Scope.provide(
1628	Effect.acquireRelease(conn.wire.take(1), () => conn.wire.release(1)),
1629	scope
1630	)
1631	if (conn.deadWith !== undefined) return yield* conn.deadWith
1632	// Streams stay on the unnamed path: a stream pays its setup once over the
1633	// whole result, so naming the statement buys little and would need the
1634	// stale-plan retry to unwind rows already delivered.
1635	const plan = yield* Effect.try({
1636	try: () => conn.encodeQuery(sql, params, undefined),
1637	catch: (cause) => queryError(cause, "PgConnection: Failed to encode query")
1638	})
1639	const frame = plan.frame
1640
1641	const socket = conn.session.socket
1642	const parser = conn.session.parser
1643	let phase: QueryPhase = "parse"
1644	let fieldCount = 0
1645	let rowBuilder: RowBuilder | undefined
1646	let buffer: Array<Row> = []
1647	let failure: SqlError | undefined
1648	let finished = false
1649	let done = false
1650	let aborted = false
1651	let paused = false
1652	let pending:
1653	| ((effect: Effect.Effect<Arr.NonEmptyReadonlyArray<Row>, SqlError | Cause.Done>) => void)
1654	| undefined
1655	let drainDone: (() => void) | undefined
1656
1657	const setPaused = (value: boolean): void => {
1658	if (paused === value) return
1659	paused = value
1660	if (value) socket.pause()
1661	else socket.resume()
1662	}
1663
1664	const deliver = (): void => {
1665	if (pending === undefined) return
1666	const resume = pending
1667	if (buffer.length > 0) {
1668	const chunk = buffer as Arr.NonEmptyArray<Row>
1669	buffer = []
1670	pending = undefined
1671	resume(Effect.succeed(chunk))
1672	} else if (finished) {
1673	pending = undefined
1674	resume(failure !== undefined ? Effect.fail(failure) : Cause.done())
1675	}
1676	}
1677
1678	const onFatal = (error: SqlError): void => {
1679	if (done) return
1680	done = true
1681	finished = true
1682	if (failure === undefined) failure = error
1683	drainDone?.()
1684	deliver()
1685	}
1686	// `conn.fatal` notifies the registered consumer; the direct `onFatal` call
1687	// covers the case where the connection was already dead.
1688	const failFatal = (error: SqlError): void => {
1689	conn.fatal(error)
1690	onFatal(error)
1691	}
1692	const failDesync = (message: string): void =>
1693	failFatal(connectionQueryError(new Error(message), `PgConnection: ${message}`))
1694
1695	const onMessage = (message: PgProtocol.BackendMessage<unknown>): void => {
1696	if (aborted) {
1697	if (message._tag === "ReadyForQuery") {
1698	done = true
1699	conn.consumer = undefined
1700	drainDone?.()
1701	}
1702	return
1703	}
1704	if (phase === "error") {
1705	switch (message._tag) {
1706	case "ReadyForQuery":
1707	finished = true
1708	done = true
1709	conn.consumer = undefined
1710	deliver()
1711	return
1712	default:
1713	return failDesync(`Unexpected ${message._tag} after ErrorResponse`)
1714	}
1715	}
1716	switch (message._tag) {
1717	case "ParseComplete":
1718	if (phase !== "parse") return failDesync(`Unexpected ParseComplete during ${phase}`)
1719	phase = "bind"
1720	return
1721	case "BindComplete":
1722	if (phase !== "bind") return failDesync(`Unexpected BindComplete during ${phase}`)
1723	phase = "describe"
1724	return
1725	case "RowDescription": {
1726	if (phase !== "describe") return failDesync(`Unexpected RowDescription during ${phase}`)
1727	const description = describe(message.fields, conn.registry)
1728	if (EffectResult.isFailure(description)) {
1729	return failFatal(queryError(description.failure, "PgConnection: Failed to decode row"))
1730	}
1731	fieldCount = message.fields.length
1732	rowBuilder = description.success.rowBuilder
1733	parser.readField = description.success.readField
1734	phase = "rows"
1735	return
1736	}
1737	case "NoData":
1738	if (phase !== "describe") return failDesync(`Unexpected NoData during ${phase}`)
1739	phase = "rows"
1740	return
1741	case "DataRow": {
1742	if (phase !== "rows" || rowBuilder === undefined) {
1743	return failDesync(`Unexpected DataRow during ${phase}`)
1744	}
1745	const rowValues = message.values
1746	if (rowValues.length !== fieldCount) {
1747	return failDesync(`DataRow has ${rowValues.length} values for ${fieldCount} fields`)
1748	}
1749	buffer.push(rowBuilder(rowValues))
1750	return
1751	}
1752	case "CommandComplete":
1753	if (phase !== "rows") return failDesync(`Unexpected CommandComplete during ${phase}`)
1754	phase = "complete"
1755	return
1756	case "EmptyQueryResponse":
1757	if (phase !== "rows") return failDesync(`Unexpected EmptyQueryResponse during ${phase}`)
1758	phase = "complete"
1759	return
1760	case "ErrorResponse": {
1761	failure = new SqlError({
1762	reason: classifyFields(message.fields, "PgConnection: Query failed", "query")
1763	})
1764	phase = "error"
1765	return
1766	}
1767	case "ReadyForQuery":
1768	if (phase !== "complete") return failDesync(`Unexpected ReadyForQuery during ${phase}`)
1769	finished = true
1770	done = true
1771	conn.consumer = undefined
1772	deliver()
1773	return
1774	case "CopyInResponse":
1775	case "CopyOutResponse":
1776	case "CopyBothResponse":
1777	case "CopyData":
1778	case "CopyDone":
1779	return failDesync(`Unexpected ${message._tag}; COPY is not supported`)
1780	default:
1781	return failDesync(`Unexpected ${message._tag} during ${phase}`)
1782	}
1783	}
1784
1785	const onBatchEnd = (): void => {
1786	if (done || aborted) return
1787	if (pending !== undefined) deliver()
1788	else if (buffer.length >= streamPauseThreshold) setPaused(true)
1789	}
1790
1791	yield* Scope.addFinalizer(
1792	scope,
1793	Effect.suspend(() => {
1794	if (done) return Effect.void
1795	aborted = true
1796	const drain = drainAborted(conn, "Stream cancellation timed out", (onDrained) => {
1797	drainDone = onDrained
1798	})
1799	setPaused(false)
1800	return drain
1801	})
1802	)
1803
1804	conn.consumer = { onMessage, onBatchEnd, onFatal }
1805	parser.readField = undefined
1806	// @effect-diagnostics-next-line tryCatchInEffectGen:off
1807	try {
1808	socket.write(frame)
1809	} catch (cause) {
1810	failFatal(connectionQueryError(cause, "PgConnection: Failed to write query"))
1811	}
1812
1813	// @effect-diagnostics-next-line returnEffectInGen:off
1814	return Effect.callback<Arr.NonEmptyReadonlyArray<Row>, SqlError | Cause.Done>((resume) => {
1815	pending = resume
1816	setPaused(false)
1817	deliver()
1818	if (pending === undefined) return
1819	return Effect.sync(() => {
1820	if (pending === resume) pending = undefined
1821	})
1822	})
1823	})))
1824
1825	const listenChannel = (
1826	conn: PgConnectionImpl,
1827	pin: Effect.Effect<PgConnection, never, Scope.Scope>,
1828	channel: string
1829	): Effect.Effect<Queue.Dequeue<Notification, SqlError>, SqlError, Scope.Scope> =>
1830	Effect.uninterruptibleMask((restore) =>
1831	Effect.gen(function*() {
1832	const channelError = validateChannelName(channel, "listen")
1833	if (channelError !== undefined) return yield* Effect.fail(channelError)
1834	const parentScope = yield* Scope.Scope
1835	const scope = yield* Scope.fork(parentScope)
1836	return yield* restore(Effect.gen(function*() {
1837	const pinned = yield* Scope.provide(pin, scope)
1838	if (conn.deadWith !== undefined) return yield* conn.deadWith
1839	const queue = yield* Queue.unbounded<Notification, SqlError>()
1840	const identifier = escapeIdentifier(channel)
1841	let queues = conn.channels.get(channel)
1842	if (queues === undefined) {
1843	queues = new Set()
1844	conn.channels.set(channel, queues)
1845	}
1846	queues.add(queue)
1847	yield* Scope.addFinalizer(
1848	scope,
1849	Effect.suspend(() => {
1850	const current = conn.channels.get(channel)
1851	if (current === undefined) return Queue.shutdown(queue)
1852	current.delete(queue)
1853	if (current.size > 0) return Queue.shutdown(queue)
1854	conn.channels.delete(channel)
1855	const unlisten = conn.deadWith === undefined
1856	? Effect.ignore(pinned.query(`UNLISTEN ${identifier}`))
1857	: Effect.void
1858	return Effect.andThen(unlisten, Queue.shutdown(queue))
1859	})
1860	)
1861	yield* pinned.query(`LISTEN ${identifier}`)
1862	return queue
1863	})).pipe(
1864	Effect.tapCause((cause) => Scope.close(scope, Exit.failCause(cause)))
1865	)
1866	})
1867	)
1868
1869	const sendCancelRequest = (config: ResolvedConfig, session: Session): Effect.Effect<void> =>
1870	Effect.callback<void>((resume) => {
1871	let done = false
1872	let socket: Duplex
1873	let timer: ReturnType<typeof setTimeout> | undefined
1874	const finish = (): void => {
1875	if (done) return
1876	done = true
1877	if (timer !== undefined) clearTimeout(timer)
1878	socket?.destroy()
1879	resume(Effect.void)
1880	}
1881	const frame = PgProtocol.encodeCancelRequest({ pid: session.processId, secret: session.secretKey })
1882	// After the frame is written the server processes the request and closes
1883	// the connection, which lands in the `close` handler.
1884	const send = (): void => {
1885	socket.write(frame)
1886	}
1887	const begin = (): void => {
1888	if (config.ssl === false) return send()
1889	socket.once("data", (chunk: Uint8Array) => {
1890	if (done) return
1891	// A TLS session's cancel secret must stay encrypted.
1892	if (chunk.length === 1 && chunk[0] === 0x4e && config.sslOptional && !session.encrypted) return send()
1893	if (chunk.length !== 1 || chunk[0] !== 0x53) return finish()
1894	const raw = socket
1895	raw.off("error", finish)
1896	raw.off("close", finish)
1897	socket = Tls.connect({
1898	host: config.host,
1899	servername: Net.isIP(config.host) === 0 ? config.host : undefined,
1900	...(typeof config.ssl === "object" ? config.ssl : {}),
1901	socket: raw as Net.Socket
1902	})
1903	socket.on("error", finish)
1904	socket.on("close", finish)
1905	socket.once("secureConnect", send)
1906	})
1907	socket.write(PgProtocol.encodeSslRequest())
1908	}
1909	try {
1910	socket = config.stream !== undefined
1911	? config.stream()
1912	: config.path !== undefined
1913	? Net.connect({ path: config.path })
1914	: Net.connect({ host: config.host, port: config.port, noDelay: true })
1915	} catch {
1916	resume(Effect.void)
1917	return
1918	}
1919	timer = setTimeout(finish, cancelRequestTimeoutMillis)
1920	socket.on("error", finish)
1921	socket.on("close", finish)
1922	if (config.stream !== undefined) begin()
1923	else socket.once("connect", begin)
1924	return Effect.sync(finish)
1925	})
1926
1927	const connect = (config: ResolvedConfig, resolvedPassword: string | undefined): Effect.Effect<Session, SqlError> =>
1928	Effect.callback<Session, SqlError>((resume) => {
1929	let done = false
1930	let socket: Duplex
1931	let encrypted = false
1932	let parser: PgProtocol.Parser<unknown> | undefined
1933	let sslErrorParser: PgProtocol.Parser | undefined
1934	let scram: PgAuth.ScramState | undefined
1935	let processId = 0
1936	let secretKey = 0
1937
1938	const fail = (reason: SqlErrorReason): void => {
1939	if (done) return
1940	done = true
1941	socket?.destroy()
1942	resume(Effect.fail(new SqlError({ reason })))
1943	}
1944	const failConnect = (cause: unknown, message: string): void =>
1945	fail(new ConnectionError({ cause, message, operation: "connect" }))
1946	const failAuth = (cause: unknown, message: string): void =>
1947	fail(new AuthenticationError({ cause, message, operation: "connect" }))
1948
1949	const onError = (cause: Error) => failConnect(cause, "PgConnection: Failed to connect")
1950	const onClose = () =>
1951	failConnect(new Error("Connection closed unexpectedly"), "PgConnection: Connection closed during startup")
1952
1953	const password = (): string | undefined => {
1954	if (resolvedPassword === undefined) {
1955	failAuth(
1956	new Error("The server requested password authentication"),
1957	"PgConnection: No password configured"
1958	)
1959	return undefined
1960	}
1961	return resolvedPassword
1962	}
1963
1964	const handleMessage = (message: PgProtocol.BackendMessage<unknown>): void => {
1965	switch (message._tag) {
1966	case "AuthenticationOk":
1967	case "NoticeResponse":
1968	case "NegotiateProtocolVersion":
1969	case "ParameterStatus":
1970	return
1971	case "AuthenticationCleartextPassword": {
1972	const secret = password()
1973	if (secret === undefined) return
1974	socket.write(PgProtocol.encodePasswordMessage({ password: secret }))
1975	return
1976	}
1977	case "AuthenticationMD5Password": {
1978	const secret = password()
1979	if (secret === undefined) return
1980	const hashed = PgAuth.md5Password({ user: config.username, password: secret, salt: message.salt })
1981	if (EffectResult.isFailure(hashed)) {
1982	return failAuth(hashed.failure, "PgConnection: MD5 authentication failed")
1983	}
1984	socket.write(PgProtocol.encodePasswordMessage({ password: hashed.success }))
1985	return
1986	}
1987	case "AuthenticationSASL": {
1988	if (!message.mechanisms.includes(PgAuth.SCRAM_SHA_256)) {
1989	return failAuth(
1990	new Error(`Unsupported SASL mechanisms: ${message.mechanisms.join(", ")}`),
1991	`PgConnection: Only ${PgAuth.SCRAM_SHA_256} is supported`
1992	)
1993	}
1994	const secret = password()
1995	if (secret === undefined) return
1996	const init = PgAuth.scramInit({ password: secret, nonce: randomBytes(18).toString("base64") })
1997	if (EffectResult.isFailure(init)) {
1998	return failAuth(init.failure, "PgConnection: SCRAM authentication failed")
1999	}
2000	scram = init.success.state
2001	socket.write(PgProtocol.encodeSASLInitialResponse({
2002	mechanism: PgAuth.SCRAM_SHA_256,
2003	initialResponse: init.success.response
2004	}))
2005	return
2006	}
2007	case "AuthenticationSASLContinue": {
2008	if (scram === undefined || scram._tag !== "ScramFirst") {
2009	return failConnect(new Error("Unexpected AuthenticationSASLContinue"), "PgConnection: Protocol desync")
2010	}
2011	const next = PgAuth.scramContinue(scram, message.data)
2012	if (EffectResult.isFailure(next)) {
2013	return failAuth(next.failure, "PgConnection: SCRAM authentication failed")
2014	}
2015	scram = next.success.state
2016	socket.write(PgProtocol.encodeSASLResponse({ data: next.success.response }))
2017	return
2018	}
2019	case "AuthenticationSASLFinal": {
2020	if (scram === undefined || scram._tag !== "ScramFinal") {
2021	return failConnect(new Error("Unexpected AuthenticationSASLFinal"), "PgConnection: Protocol desync")
2022	}
2023	const verified = PgAuth.scramFinish(scram, message.data)
2024	if (EffectResult.isFailure(verified)) {
2025	return failAuth(verified.failure, "PgConnection: SCRAM server verification failed")
2026	}
2027	scram = undefined
2028	return
2029	}
2030	case "AuthenticationUnsupported":
2031	return failAuth(
2032	new Error(`Authentication method ${message.method} is not supported`),
2033	"PgConnection: Unsupported authentication method"
2034	)
2035	case "BackendKeyData":
2036	processId = message.pid
2037	secretKey = message.secret
2038	return
2039	case "ErrorResponse":
2040	return fail(classifyFields(message.fields, "PgConnection: Failed to connect", "connect"))
2041	case "ReadyForQuery":
2042	if (scram !== undefined) {
2043	return failAuth(
2044	new Error("The server completed authentication without proving its identity"),
2045	"PgConnection: SCRAM exchange did not complete"
2046	)
2047	}
2048	done = true
2049	socket.off("data", onData)
2050	socket.off("error", onError)
2051	socket.off("close", onClose)
2052	socket.on("error", ignoreError)
2053	resume(Effect.succeed({ socket, encrypted, parser: parser!, processId, secretKey }))
2054	return
2055	default:
2056	return failConnect(
2057	new Error(`Unexpected ${message._tag} message during startup`),
2058	"PgConnection: Protocol desync"
2059	)
2060	}
2061	}
2062
2063	const onData = (chunk: Uint8Array): void => {
2064	let messages: ReadonlyArray<PgProtocol.BackendMessage<unknown>>
2065	try {
2066	messages = parser!.push(chunk)
2067	} catch (cause) {
2068	return failConnect(cause, "PgConnection: Failed to parse server response")
2069	}
2070	for (const message of messages) {
2071	if (done) return
2072	handleMessage(message)
2073	}
2074	}
2075
2076	const startup = (): void => {
2077	parser = PgProtocol.makeParser<unknown>({ maxMessageSize: config.maxMessageSize })
2078	socket.on("data", onData)
2079	socket.write(PgProtocol.encodeStartupMessage(config.startupParameters))
2080	}
2081
2082	const onSslResponse = (chunk: Uint8Array): void => {
2083	if (done) return
2084	if (sslErrorParser !== undefined || chunk[0] === 0x45) {
2085	sslErrorParser ??= PgProtocol.makeParser({ maxMessageSize: config.maxMessageSize })
2086	let messages: ReadonlyArray<PgProtocol.BackendMessage>
2087	try {
2088	messages = sslErrorParser.push(chunk)
2089	} catch (cause) {
2090	return failConnect(cause, "PgConnection: Failed to parse SSLRequest error response")
2091	}
2092	if (messages.length === 0) return
2093	socket.off("data", onSslResponse)
2094	const message = messages[0]
2095	if (messages.length !== 1 || message._tag !== "ErrorResponse") {
2096	return failConnect(
2097	new Error("Expected one ErrorResponse after SSLRequest"),
2098	"PgConnection: Invalid SSLRequest response"
2099	)
2100	}
2101	return fail(classifyFields(message.fields, "PgConnection: Failed to negotiate TLS", "connect"))
2102	}
2103	socket.off("data", onSslResponse)
2104	if (chunk.length !== 1) {
2105	return failConnect(
2106	new Error(`Received ${chunk.length} bytes in response to SSLRequest`),
2107	"PgConnection: Invalid SSLRequest response"
2108	)
2109	}
2110	const response = PgProtocol.decodeSslResponse(chunk[0])
2111	if (EffectResult.isFailure(response)) {
2112	return failConnect(response.failure, "PgConnection: Invalid SSLRequest response")
2113	}
2114	if (response.success === "N") {
2115	if (config.sslOptional) return startup()
2116	return failConnect(new Error("The server does not support TLS"), "PgConnection: Server refused TLS")
2117	}
2118	const raw = socket
2119	raw.off("error", onError)
2120	raw.off("close", onClose)
2121	socket = Tls.connect({
2122	host: config.host,
2123	servername: Net.isIP(config.host) === 0 ? config.host : undefined,
2124	...(typeof config.ssl === "object" ? config.ssl : {}),
2125	socket: raw as Net.Socket
2126	})
2127	socket.on("error", onError)
2128	socket.on("close", onClose)
2129	socket.once("secureConnect", () => {
2130	encrypted = true
2131	startup()
2132	})
2133	}
2134
2135	const begin = (): void => {
2136	if (config.ssl === false) return startup()
2137	socket.on("data", onSslResponse)
2138	socket.write(PgProtocol.encodeSslRequest())
2139	}
2140
2141	try {
2142	socket = config.stream !== undefined
2143	? config.stream()
2144	: config.path !== undefined
2145	? Net.connect({ path: config.path })
2146	// A statement is one write and then a wait for its answer, so Nagle has
2147	// nothing to coalesce and only holds the write back.
2148	: Net.connect({ host: config.host, port: config.port, noDelay: true })
2149	} catch (cause) {
2150	resume(Effect.fail(
2151	new SqlError({
2152	reason: new ConnectionError({ cause, message: "PgConnection: Failed to connect", operation: "connect" })
2153	})
2154	))
2155	return
2156	}
2157	socket.on("error", onError)
2158	socket.on("close", onClose)
2159	if (config.stream !== undefined) {
2160	begin()
2161	} else {
2162	socket.once("connect", begin)
2163	}
2164
2165	return Effect.sync(() => {
2166	if (done) return
2167	done = true
2168	socket.destroy()
2169	})
2170	})
2171
2172	const ignoreError = (_: Error) => {}
2173
2174	interface ResolvedConfig {
2175	readonly host: string
2176	readonly port: number
2177	readonly path: string | undefined
2178	readonly ssl: boolean | ConnectionOptions
2179	readonly sslOptional: boolean
2180	readonly username: string
2181	readonly password: Redacted.Redacted | Effect.Effect<Redacted.Redacted> | undefined
2182	readonly connectTimeout: Duration.Duration
2183	readonly startupParameters: PgProtocol.StartupParameters
2184	readonly stream: (() => Duplex) | undefined
2185	readonly maxMessageSize: number | undefined
2186	}
2187
2188	const configError = (message: string, cause?: unknown): SqlError =>
2189	new SqlError({
2190	reason: new ConnectionError({
2191	cause: cause ?? new Error(message),
2192	message: `PgConnection: ${message}`,
2193	operation: "connect"
2194	})
2195	})
2196
2197	const resolveConfig = (options: Config): Effect.Effect<ResolvedConfig, SqlError> =>
2198	Effect.suspend(() => {
2199	const parsed: EffectResult.Result<UrlConfig, SqlError> = options.url !== undefined
2200	? parseUrl(Redacted.value(options.url))
2201	: EffectResult.succeed({})
2202	if (EffectResult.isFailure(parsed)) return Effect.fail(parsed.failure)
2203	const url = parsed.success
2204	const host = options.host ?? url.host ?? "localhost"
2205	const port = options.port ?? url.port ?? 5432
2206	const username = options.username ?? url.username ?? process.env.USER ?? process.env.USERNAME
2207	if (username === undefined) {
2208	return Effect.fail(configError("No username configured"))
2209	}
2210	const named: Record<string, string> = Object.create(null)
2211	for (const [name, value] of Object.entries(options.startupParameters ?? {})) {
2212	const key = name.toLowerCase()
2213	if (key === "user" || key === "database" || key === "replication" || key === "options") {
2214	return Effect.fail(configError(`Reserved startup parameter: "${name}"`))
2215	}
2216	if (key === "client_encoding") {
2217	const encoding = value.toUpperCase()
2218	if (encoding !== "UTF8" && encoding !== "UTF-8") {
2219	return Effect.fail(configError("Startup parameter client_encoding must be UTF8 or UTF-8"))
2220	}
2221	named[key] = "UTF8"
2222	} else {
2223	named[key] = value
2224	}
2225	if (name === "" || name.includes("\0") || value.includes("\0")) {
2226	return Effect.fail(
2227	configError("Startup parameter names must be nonempty and names/values must not contain NUL")
2228	)
2229	}
2230	}
2231	const startupParameters: PgProtocol.StartupParameters = {
2232	...named,
2233	user: username,
2234	database: options.database ?? url.database,
2235	application_name: options.applicationName ?? named.application_name ?? url.applicationName ?? "@effect/sql-pg",
2236	options: options.startupOptions ?? url.options
2237	}
2238	for (const [name, value] of Object.entries(startupParameters)) {
2239	if (value?.includes("\0")) {
2240	return Effect.fail(configError(`Startup parameter "${name}" must not contain NUL`))
2241	}
2242	}
2243	return Effect.succeed<ResolvedConfig>({
2244	host,
2245	port,
2246	path: options.path ?? (host.startsWith("/") ? `${host}/.s.PGSQL.${port}` : undefined),
2247	ssl: options.ssl ?? (url.ssl === "prefer" ? true : url.ssl ?? false),
2248	sslOptional: options.ssl === undefined && url.ssl === "prefer",
2249	username,
2250	password: options.password ?? (url.password !== undefined ? Redacted.make(url.password) : undefined),
2251	connectTimeout: Duration.fromInputUnsafe(options.connectTimeout ?? url.connectTimeout ?? Duration.seconds(5)),
2252	startupParameters,
2253	stream: options.stream,
2254	maxMessageSize: options.maxMessageSize
2255	})
2256	})
2257
2258	interface UrlConfig {
2259	host?: string | undefined
2260	port?: number | undefined
2261	database?: string | undefined
2262	username?: string | undefined
2263	password?: string | undefined
2264	applicationName?: string | undefined
2265	options?: string | undefined
2266	connectTimeout?: Duration.Duration | undefined
2267	ssl?: boolean | "prefer" | undefined
2268	}
2269
2270	const decodeComponent = (value: string, what: string): EffectResult.Result<string, SqlError> => {
2271	try {
2272	return EffectResult.succeed(decodeURIComponent(value))
2273	} catch {
2274	return EffectResult.fail(configError(`Invalid percent-encoding in URL ${what}`))
2275	}
2276	}
2277
2278	const parsePort = (value: string, what: string): EffectResult.Result<number, SqlError> => {
2279	const port = Number(value)
2280	return !Number.isInteger(port) || port < 1 || port > 65535
2281	? EffectResult.fail(configError(`Invalid port in URL ${what}: "${value}"`))
2282	: EffectResult.succeed(port)
2283	}
2284
2285	const parseUrl = (raw: string): EffectResult.Result<UrlConfig, SqlError> => {
2286	let url: URL
2287	try {
2288	url = new URL(raw)
2289	} catch (cause) {
2290	return EffectResult.fail(configError("Invalid connection URL", cause))
2291	}
2292	if (url.protocol !== "postgres:" && url.protocol !== "postgresql:") {
2293	return EffectResult.fail(configError(`Unsupported connection URL protocol: "${url.protocol}"`))
2294	}
2295
2296	const config: UrlConfig = {}
2297	if (url.hostname !== "") {
2298	if (url.hostname.startsWith("[") && url.hostname.endsWith("]")) {
2299	config.host = url.hostname.slice(1, -1)
2300	} else {
2301	const host = decodeComponent(url.hostname, "host")
2302	if (EffectResult.isFailure(host)) return EffectResult.fail(host.failure)
2303	config.host = host.success
2304	}
2305	}
2306	if (url.port !== "") {
2307	const port = parsePort(url.port, "authority")
2308	if (EffectResult.isFailure(port)) return EffectResult.fail(port.failure)
2309	config.port = port.success
2310	}
2311	if (url.username !== "") {
2312	const username = decodeComponent(url.username, "username")
2313	if (EffectResult.isFailure(username)) return EffectResult.fail(username.failure)
2314	config.username = username.success
2315	}
2316	if (url.password !== "") {
2317	const password = decodeComponent(url.password, "password")
2318	if (EffectResult.isFailure(password)) return EffectResult.fail(password.failure)
2319	config.password = password.success
2320	}
2321	const database = decodeComponent(url.pathname.replace(/^\//, ""), "database")
2322	if (EffectResult.isFailure(database)) return EffectResult.fail(database.failure)
2323	if (database.success !== "") config.database = database.success
2324
2325	for (const [key, value] of url.searchParams) {
2326	switch (key) {
2327	case "host":
2328	config.host = value
2329	break
2330	case "port": {
2331	const port = parsePort(value, "port parameter")
2332	if (EffectResult.isFailure(port)) return EffectResult.fail(port.failure)
2333	config.port = port.success
2334	break
2335	}
2336	case "user":
2337	config.username = value
2338	break
2339	case "password":
2340	config.password = value
2341	break
2342	case "dbname":
2343	config.database = value
2344	break
2345	case "application_name":
2346	config.applicationName = value
2347	break
2348	case "options":
2349	config.options = value
2350	break
2351	case "connect_timeout": {
2352	const seconds = Number(value)
2353	if (!Number.isInteger(seconds) || seconds < 0) {
2354	return EffectResult.fail(configError(`Invalid connect_timeout in URL: "${value}"`))
2355	}
2356	config.connectTimeout = seconds === 0 ? Duration.infinity : Duration.seconds(seconds)
2357	break
2358	}
2359	case "sslmode":
2360	switch (value) {
2361	case "disable":
2362	config.ssl = false
2363	break
2364	case "require":
2365	case "verify-ca":
2366	case "verify-full":
2367	config.ssl = true
2368	break
2369	case "prefer":
2370	case "allow":
2371	config.ssl = "prefer"
2372	break
2373	default:
2374	return EffectResult.fail(configError(`Unrecognized sslmode in URL: "${value}"`))
2375	}
2376	break
2377	// Unknown query parameters are ignored, matching libpq.
2378	}
2379	}
2380	return EffectResult.succeed(config)
2381	}
2382
2383	const classifyFields = (
2384	fields: PgProtocol.ErrorFields,
2385	message: string,
2386	operation: string
2387	): SqlErrorReason => {
2388	const cause = Object.assign(new Error(fields.message ?? "Unknown PostgreSQL error"), fields)
2389	return classifySqlState(fields.code, fields.constraint, { cause, message, operation })
2390	}