1import { PgConnection } from "@effect/sql-pg"
2import { assert, describe, it } from "@effect/vitest"
3import { Effect, Fiber, Redacted } from "effect"
4import * as TestClock from "effect/testing/TestClock"
5import { mkdtemp, rm } from "node:fs/promises"
6import * as Net from "node:net"
7import { tmpdir } from "node:os"
8import { join } from "node:path"
9import { Duplex } from "node:stream"
10import type * as Tls from "node:tls"
11import { vi } from "vitest"
12
13const tlsOptions = new WeakMap<Duplex, Tls.ConnectionOptions | undefined>()
14
15// Only simulate TLS for registered streams; other connections use native TLS.
16vi.mock("node:tls", async (importOriginal) => {
17 const original = await importOriginal<typeof Tls>()
18 return {
19 ...original,
20 connect(options: Tls.ConnectionOptions) {
21 const socket = options.socket
22 if (socket === undefined || !tlsOptions.has(socket)) {
23 return original.connect(options)
24 }
25 tlsOptions.set(socket, options)
26 queueMicrotask(() => socket.emit("secureConnect"))
27 return socket
28 }
29 }
30})
31
32const backendMessage = (tag: string, payload: Uint8Array): Buffer => {
33 const length = Buffer.allocUnsafe(4)
34 length.writeInt32BE(payload.length + 4)
35 return Buffer.concat([Buffer.from(tag), length, payload])
36}
37
38const int32 = (value: number): Buffer => {
39 const bytes = Buffer.allocUnsafe(4)
40 bytes.writeInt32BE(value)
41 return bytes
42}
43
44const int16 = (value: number): Buffer => {
45 const bytes = Buffer.allocUnsafe(2)
46 bytes.writeInt16BE(value)
47 return bytes
48}
49
50const authentication = (method: number, payload: Uint8Array = Buffer.alloc(0)): Buffer =>
51 backendMessage("R", Buffer.concat([int32(method), payload]))
52
53const authenticationOk = authentication(0)
54const backendKeyData = backendMessage("K", Buffer.concat([int32(1234), int32(5678)]))
55const readyForQuery = backendMessage("Z", Buffer.from("I"))
56const sslRequest = Buffer.concat([int32(8), int32(80877103)])
57const cancelRequest = Buffer.concat([int32(16), int32(80877102), int32(1234), int32(5678)])
58
59const makeSslStream = (reply: "S" | "N", cancel: boolean) => {
60 const writes: Array<Buffer> = []
61 const socket: Duplex = new Duplex({
62 read() {},
63 write(chunk: Buffer, _encoding, callback) {
64 writes.push(Buffer.from(chunk))
65 queueMicrotask(() => {
66 if (writes.length === 1) {
67 socket.push(Buffer.from(reply))
68 } else if (cancel) {
69 socket.destroy()
70 } else if (writes.length === 2) {
71 socket.push(Buffer.concat([authenticationOk, backendKeyData, readyForQuery]))
72 }
73 })
74 callback()
75 }
76 })
77 tlsOptions.set(socket, undefined)
78 return { socket, writes }
79}
80
81const emptyQueryResult = Buffer.concat([
82 backendMessage("1", Buffer.alloc(0)),
83 backendMessage("2", Buffer.alloc(0)),
84 backendMessage("n", Buffer.alloc(0)),
85 backendMessage("C", Buffer.from("SELECT 0\0")),
86 readyForQuery
87])
88
89const frontendTags = (message: Buffer): ReadonlyArray<string> => {
90 const tags: Array<string> = []
91 let offset = 0
92 while (offset < message.length) {
93 tags.push(String.fromCharCode(message[offset]))
94 offset += 1 + message.readInt32BE(offset + 1)
95 }
96 return tags
97}
98
99const frontendPreparedNames = (message: Buffer): ReadonlyArray<string> => {
100 const names: Array<string> = []
101 let offset = 0
102 while (offset < message.length) {
103 const tag = String.fromCharCode(message[offset])
104 const length = message.readInt32BE(offset + 1)
105 if (tag === "P") {
106 const nameStart = offset + 5
107 const nameEnd = message.indexOf(0, nameStart)
108 names.push(message.subarray(nameStart, nameEnd).toString())
109 }
110 offset += 1 + length
111 }
112 return names
113}
114
115const consumeFrontend = (
116 socket: Net.Socket,
117 onMessage: (tag: string | undefined, message: Buffer) => void
118): void => {
119 let startup = true
120 let buffered = Buffer.alloc(0)
121 socket.on("data", (chunk: Buffer) => {
122 buffered = Buffer.concat([buffered, chunk])
123 while (true) {
124 const lengthOffset = startup ? 0 : 1
125 if (buffered.length < lengthOffset + 4) return
126 const length = buffered.readInt32BE(lengthOffset)
127 const messageLength = lengthOffset + length
128 if (buffered.length < messageLength) return
129 const message = buffered.subarray(0, messageLength)
130 buffered = buffered.subarray(messageLength)
131 const tag = startup ? undefined : String.fromCharCode(message[0])
132 startup = false
133 onMessage(tag, message)
134 }
135 })
136}
137
138const startupParameters = (message: Buffer): ReadonlyMap<string, string> => {
139 const values = message.subarray(8, -1).toString().split("\0")
140 const parameters = new Map<string, string>()
141 for (let index = 0; index < values.length - 1 && values[index] !== ""; index += 2) {
142 parameters.set(values[index], values[index + 1])
143 }
144 return parameters
145}
146
147const errorResponse = (fields: ReadonlyArray<readonly [string, string]>): Buffer =>
148 backendMessage(
149 "E",
150 Buffer.concat([
151 ...fields.map(([code, value]) => Buffer.from(`${code}${value}\0`)),
152 Buffer.from([0])
153 ])
154 )
155
156const withTcpServer = (
157 onConnection: (socket: Net.Socket) => void
158) =>
159 Effect.acquireRelease(
160 Effect.tryPromise({
161 try: () =>
162 new Promise<{ readonly port: number; readonly server: Net.Server }>((resolve, reject) => {
163 const server = Net.createServer(onConnection)
164 server.once("error", reject)
165 server.listen(0, "127.0.0.1", () => {
166 server.off("error", reject)
167 const address = server.address()
168 if (address === null || typeof address === "string") {
169 reject(new Error("Test server did not bind to a TCP port"))
170 return
171 }
172 resolve({ port: address.port, server })
173 })
174 }),
175 catch: (cause) => cause
176 }),
177 ({ server }) =>
178 Effect.promise(() =>
179 new Promise<void>((resolve) => {
180 server.close(() => resolve())
181 })
182 )
183 )
184
185const withUnixServer = (
186 onConnection: (socket: Net.Socket) => void
187) =>
188 Effect.acquireRelease(
189 Effect.tryPromise({
190 try: async () => {
191 const directory = await mkdtemp(join(tmpdir(), "effect-pg-"))
192 const socketPath = join(directory, ".s.PGSQL.5432")
193 const server = Net.createServer(onConnection)
194 await new Promise<void>((resolve, reject) => {
195 server.once("error", reject)
196 server.listen(socketPath, () => {
197 server.off("error", reject)
198 resolve()
199 })
200 })
201 return { directory, server }
202 },
203 catch: (cause) => cause
204 }),
205 ({ directory, server }) =>
206 Effect.promise(async () => {
207 await new Promise<void>((resolve) => server.close(() => resolve()))
208 await rm(directory, { recursive: true, force: true })
209 })
210 )
211
212describe("PgConnection in-process server", () => {
213 it.live("forwards startup defaults and lets explicit fields override URL values", () =>
214 Effect.scoped(Effect.gen(function*() {
215 let parameters: ReadonlyMap<string, string> | undefined
216 const { port } = yield* withTcpServer((socket) => {
217 consumeFrontend(socket, (tag, message) => {
218 if (tag === undefined) {
219 parameters = startupParameters(message)
220 socket.write(Buffer.concat([authenticationOk, backendKeyData, readyForQuery]))
221 }
222 })
223 })
224 const config = {
225 host: "127.0.0.1",
226 port,
227 url: Redacted.make(
228 "postgres://test@localhost/db?application_name=url-app&options=-c%20statement_timeout%3D1234"
229 ),
230 startupParameters: { SEARCH_PATH: "public", CLIENT_ENCODING: "uTf-8", application_name: "named-app" }
231 }
232 yield* PgConnection.make(config)
233 assert.deepStrictEqual(Object.fromEntries(parameters!), {
234 user: "test",
235 database: "db",
236 application_name: "named-app",
237 search_path: "public",
238 client_encoding: "UTF8",
239 options: "-c statement_timeout=1234"
240 })
241
242 yield* PgConnection.make({ ...config, applicationName: "explicit-app", startupOptions: "-c lock_timeout=2345" })
243 assert.strictEqual(parameters!.get("application_name"), "explicit-app")
244 assert.strictEqual(parameters!.get("options"), "-c lock_timeout=2345")
245 })))
246
247 it.effect("accepts backend messages over the default limit when configured", () =>
248 Effect.gen(function*() {
249 const fieldSize = 16 * 1024 * 1024
250 const field = Buffer.alloc(fieldSize, 0x78)
251 const rowDescription = backendMessage(
252 "T",
253 Buffer.concat([
254 int16(1),
255 Buffer.from("big\0"),
256 int32(0),
257 int16(0),
258 int32(25),
259 int16(-1),
260 int32(-1),
261 int16(1)
262 ])
263 )
264 const dataRow = backendMessage("D", Buffer.concat([int16(1), int32(field.length), field]))
265 const result = Buffer.concat([
266 backendMessage("1", Buffer.alloc(0)),
267 backendMessage("2", Buffer.alloc(0)),
268 rowDescription,
269 dataRow,
270 backendMessage("C", Buffer.from("SELECT 1\0")),
271 readyForQuery
272 ])
273 let writes = 0
274 const socket: Duplex = new Duplex({
275 read() {},
276 write(_chunk: Buffer, _encoding, callback) {
277 writes++
278 queueMicrotask(() => {
279 socket.push(
280 writes === 1
281 ? Buffer.concat([authenticationOk, backendKeyData, readyForQuery])
282 : writes === 2
283 ? result
284 : emptyQueryResult
285 )
286 })
287 callback()
288 }
289 })
290 const connection = yield* PgConnection.make({
291 username: "test",
292 stream: () => socket,
293 maxMessageSize: 17 * 1024 * 1024
294 })
295
296 const oversized = yield* connection.query("SELECT repeat('x', 16777216) AS big")
297 assert.strictEqual(typeof oversized.rows[0].big, "string")
298 assert.strictEqual((oversized.rows[0].big as string).length, fieldSize)
299 assert.deepStrictEqual((yield* connection.query("SELECT 1")).rows, [])
300 }))
301
302 it.effect("flushes concurrently submitted multiplexed queries in one write", () =>
303 Effect.gen(function*() {
304 const writes: Array<Buffer> = []
305 const socket: Duplex = new Duplex({
306 read() {},
307 write(chunk: Buffer, _encoding, callback) {
308 const message = Buffer.from(chunk)
309 writes.push(message)
310 if (writes.length === 1) {
311 queueMicrotask(() => socket.push(Buffer.concat([authenticationOk, backendKeyData, readyForQuery])))
312 } else {
313 const syncs = frontendTags(message).filter((tag) => tag === "S").length
314 queueMicrotask(() => socket.push(Buffer.concat(Array.from({ length: syncs }, () => emptyQueryResult))))
315 }
316 callback()
317 }
318 })
319 const connection = yield* PgConnection.make({
320 username: "test",
321 stream: () => socket,
322 multiplex: true
323 })
324
325 yield* Effect.all([
326 connection.query("SELECT 1"),
327 connection.query("SELECT 2")
328 ], { concurrency: "unbounded" })
329
330 // Both cycles left in one write, each keeping its own Sync.
331 assert.strictEqual(writes.length, 2)
332 assert.deepStrictEqual(frontendTags(writes[1]), ["P", "B", "D", "E", "S", "P", "B", "D", "E", "S"])
333 }))
334
335 it.effect("uses distinct prepared statement names across connections", () =>
336 Effect.gen(function*() {
337 const writes: Array<Array<Buffer>> = []
338 const makeStream = () => {
339 const connectionWrites: Array<Buffer> = []
340 const socket: Duplex = new Duplex({
341 read() {},
342 write(chunk: Buffer, _encoding, callback) {
343 connectionWrites.push(Buffer.from(chunk))
344 queueMicrotask(() => {
345 socket.push(
346 connectionWrites.length === 1
347 ? Buffer.concat([authenticationOk, backendKeyData, readyForQuery])
348 : emptyQueryResult
349 )
350 })
351 callback()
352 }
353 })
354 writes.push(connectionWrites)
355 return socket
356 }
357
358 const first = yield* PgConnection.make({ username: "test", stream: makeStream })
359 const second = yield* PgConnection.make({ username: "test", stream: makeStream })
360 yield* first.query("SELECT $1", [1])
361 yield* second.query("SELECT $1, $2", [1, 2])
362
363 const firstName = frontendPreparedNames(writes[0][1])
364 const secondName = frontendPreparedNames(writes[1][1])
365 assert.strictEqual(firstName.length, 1)
366 assert.strictEqual(secondName.length, 1)
367 assert.notStrictEqual(firstName[0], secondName[0])
368 }))
369
370 it.effect("reuses a prepared name after a pending pipeline query is interrupted", () =>
371 Effect.gen(function*() {
372 const writes: Array<Buffer> = []
373 const socket: Duplex = new Duplex({
374 read() {},
375 write(chunk: Buffer, _encoding, callback) {
376 const message = Buffer.from(chunk)
377 writes.push(message)
378 if (writes.length === 1) {
379 queueMicrotask(() => socket.push(Buffer.concat([authenticationOk, backendKeyData, readyForQuery])))
380 } else {
381 queueMicrotask(() => socket.push(emptyQueryResult))
382 }
383 callback()
384 }
385 })
386 const connection = yield* PgConnection.make({
387 username: "test",
388 stream: () => socket,
389 multiplex: true
390 })
391
392 const interrupted = Effect.runFork(connection.query("SELECT 1"))
393 interrupted.interruptUnsafe()
394 yield* Effect.promise(() => new Promise<void>((resolve) => queueMicrotask(resolve)))
395 yield* connection.query("SELECT 1")
396
397 assert.strictEqual(writes.length, 2)
398 const names = frontendPreparedNames(writes[1])
399 assert.strictEqual(names.length, 1)
400 assert.match(names[0], /^effect_[0-9a-f]{16}_1$/)
401 }))
402
403 it.effect("ends a custom stream after writing Terminate", () =>
404 Effect.gen(function*() {
405 const writes: Array<Buffer> = []
406 const socket: Duplex = new Duplex({
407 read() {},
408 write(chunk: Buffer, _encoding, callback) {
409 writes.push(Buffer.from(chunk))
410 if (writes.length === 1) {
411 queueMicrotask(() => socket.push(Buffer.concat([authenticationOk, backendKeyData, readyForQuery])))
412 }
413 callback()
414 }
415 })
416
417 yield* Effect.scoped(PgConnection.make({ username: "test", stream: () => socket }))
418
419 assert.strictEqual(writes.at(-1)?.toString("hex"), "5800000004")
420 assert.isTrue(socket.writableEnded)
421 assert.isFalse(socket.destroyed)
422 }))
423
424 describe("TLS servername", () => {
425 const cases: ReadonlyArray<{
426 readonly name: string
427 readonly config: PgConnection.Config
428 readonly servername: string | undefined
429 }> = [
430 { name: "DNS host with ssl=true", config: { host: "db.example.com", ssl: true }, servername: "db.example.com" },
431 {
432 name: "DNS host with SSL options",
433 config: { host: "db.example.com", ssl: { rejectUnauthorized: false } },
434 servername: "db.example.com"
435 },
436 {
437 name: "DNS host from a URL",
438 config: { url: Redacted.make("postgres://test@db.example.com/db?sslmode=require") },
439 servername: "db.example.com"
440 },
441 {
442 name: "ssl=true overrides sslmode=disable",
443 config: { url: Redacted.make("postgres://test@db.example.com/db?sslmode=disable"), ssl: true },
444 servername: "db.example.com"
445 },
446 ...["prefer", "allow"].flatMap((sslmode) => [
447 {
448 name: `sslmode=${sslmode} enables TLS`,
449 config: { url: Redacted.make(`postgresql://test@db.example.com/db?sslmode=${sslmode}`) },
450 servername: "db.example.com"
451 },
452 {
453 name: `SSL options override sslmode=${sslmode}`,
454 config: {
455 url: Redacted.make(`postgres://test@db.example.com/db?sslmode=${sslmode}`),
456 ssl: { servername: "routing.example.com", rejectUnauthorized: false }
457 },
458 servername: "routing.example.com"
459 }
460 ]),
461 { name: "IPv4 host", config: { host: "127.0.0.1", ssl: true }, servername: undefined },
462 { name: "IPv6 host", config: { host: "::1", ssl: true }, servername: undefined },
463 ...["db.example.com", "127.0.0.1", "::1"].map((host) => ({
464 name: `explicit servername for ${host}`,
465 config: { host, ssl: { servername: "routing.example.com", rejectUnauthorized: false } },
466 servername: "routing.example.com"
467 })),
468 {
469 name: "explicit empty servername",
470 config: { host: "db.example.com", ssl: { servername: "" } },
471 servername: ""
472 }
473 ]
474
475 for (const path of ["startup", "cancel"] as const) {
476 describe(path, () => {
477 it.effect.each(cases)("$name", ({ config, servername }) =>
478 Effect.gen(function*() {
479 const streams: Array<ReturnType<typeof makeSslStream>> = []
480 const connection = yield* PgConnection.make({
481 username: "test",
482 ...config,
483 stream: () => {
484 const stream = makeSslStream("S", streams.length > 0)
485 streams.push(stream)
486 return stream.socket
487 }
488 })
489 if (path === "cancel") {
490 yield* connection.interrupt
491 }
492
493 assert.strictEqual(streams.length, path === "cancel" ? 2 : 1)
494 const stream = streams[path === "cancel" ? 1 : 0]
495 assert.deepStrictEqual(stream.writes[0], sslRequest)
496 if (path === "cancel") {
497 assert.deepStrictEqual(stream.writes[1], cancelRequest)
498 } else {
499 assert.strictEqual(stream.writes[1].readInt32BE(4), 196608)
500 assert.strictEqual(startupParameters(stream.writes[1]).get("user"), "test")
501 }
502 const options = tlsOptions.get(stream.socket)
503 assert.isDefined(options)
504 assert.strictEqual(options!.servername, servername)
505 if (typeof config.ssl === "object") {
506 assert.strictEqual(options!.rejectUnauthorized, config.ssl.rejectUnauthorized)
507 } else {
508 assert.notStrictEqual(options!.rejectUnauthorized, false)
509 }
510 }))
511 })
512 }
513 })
514
515 it.effect.each(["prefer", "allow"])(
516 "ssl=false overrides sslmode=%s with plaintext startup",
517 (sslmode) =>
518 Effect.gen(function*() {
519 const writes: Array<Buffer> = []
520 const socket: Duplex = new Duplex({
521 read() {},
522 write(chunk: Buffer, _encoding, callback) {
523 writes.push(Buffer.from(chunk))
524 if (writes.length === 1) {
525 queueMicrotask(() => socket.push(Buffer.concat([authenticationOk, backendKeyData, readyForQuery])))
526 }
527 callback()
528 }
529 })
530 tlsOptions.set(socket, undefined)
531
532 yield* PgConnection.make({
533 url: Redacted.make(`postgresql://test@db.example.com/db?sslmode=${sslmode}`),
534 ssl: false,
535 stream: () => socket
536 })
537
538 assert.strictEqual(writes.length, 1)
539 assert.strictEqual(writes[0].readInt32BE(4), 196608)
540 assert.strictEqual(startupParameters(writes[0]).get("user"), "test")
541 assert.isUndefined(tlsOptions.get(socket))
542 })
543 )
544
545 it.effect.each(["require", "verify-ca", "verify-full"])(
546 "sslmode=%s fails when the server refuses TLS",
547 (sslmode) =>
548 Effect.gen(function*() {
549 const writes: Array<Buffer> = []
550 const socket: Duplex = new Duplex({
551 read() {},
552 write(chunk: Buffer, _encoding, callback) {
553 writes.push(Buffer.from(chunk))
554 queueMicrotask(() => socket.push(Buffer.from("N")))
555 callback()
556 }
557 })
558 const error = yield* Effect.flip(PgConnection.make({
559 url: Redacted.make(`postgres://test@db.example.com/db?sslmode=${sslmode}`),
560 stream: () => socket
561 }))
562
563 assert.deepStrictEqual(writes, [Buffer.concat([int32(8), int32(80877103)])])
564 assert.strictEqual(error.reason._tag, "ConnectionError")
565 assert.strictEqual(error.reason.message, "PgConnection: Server refused TLS")
566 assert.isTrue(socket.destroyed)
567 })
568 )
569
570 describe("TLS fallback", () => {
571 it.effect.each(["prefer", "allow"])("sslmode=%s starts in plaintext after N", (sslmode) =>
572 Effect.gen(function*() {
573 const stream = makeSslStream("N", false)
574 let connections = 0
575 yield* PgConnection.make({
576 url: Redacted.make(`postgres://test@db.example.com/db?sslmode=${sslmode}`),
577 stream: () => {
578 connections++
579 return stream.socket
580 }
581 })
582
583 assert.strictEqual(connections, 1)
584 assert.strictEqual(stream.writes.length, 2)
585 assert.deepStrictEqual(stream.writes[0], sslRequest)
586 assert.strictEqual(stream.writes[1].readInt32BE(4), 196608)
587 assert.strictEqual(startupParameters(stream.writes[1]).get("user"), "test")
588 assert.isUndefined(tlsOptions.get(stream.socket))
589 }))
590
591 it.effect.each(["prefer", "allow"])(
592 "sslmode=%s allows plaintext cancellation for a plaintext session",
593 (sslmode) =>
594 Effect.gen(function*() {
595 const streams: Array<ReturnType<typeof makeSslStream>> = []
596 const connection = yield* PgConnection.make({
597 url: Redacted.make(`postgres://test@db.example.com/db?sslmode=${sslmode}`),
598 stream: () => {
599 const stream = makeSslStream("N", streams.length > 0)
600 streams.push(stream)
601 return stream.socket
602 }
603 })
604 yield* connection.interrupt
605
606 assert.strictEqual(streams.length, 2)
607 assert.isUndefined(tlsOptions.get(streams[0].socket))
608 assert.deepStrictEqual(streams[0].writes[0], sslRequest)
609 assert.strictEqual(streams[0].writes[1].readInt32BE(4), 196608)
610 assert.deepStrictEqual(streams[1].writes, [sslRequest, cancelRequest])
611 assert.isUndefined(tlsOptions.get(streams[1].socket))
612 assert.isTrue(streams[1].socket.destroyed)
613 })
614 )
615
616 it.effect.each(["prefer", "allow", "require"])(
617 "sslmode=%s refuses plaintext cancellation for a TLS session",
618 (sslmode) =>
619 Effect.gen(function*() {
620 const streams: Array<ReturnType<typeof makeSslStream>> = []
621 const connection = yield* PgConnection.make({
622 url: Redacted.make(`postgres://test@db.example.com/db?sslmode=${sslmode}`),
623 stream: () => {
624 const cancel = streams.length > 0
625 const stream = makeSslStream(cancel ? "N" : "S", cancel)
626 streams.push(stream)
627 return stream.socket
628 }
629 })
630 yield* connection.interrupt
631
632 assert.strictEqual(streams.length, 2)
633 assert.isDefined(tlsOptions.get(streams[0].socket))
634 assert.deepStrictEqual(streams[1].writes, [sslRequest])
635 assert.isUndefined(tlsOptions.get(streams[1].socket))
636 assert.isTrue(streams[1].socket.destroyed)
637 })
638 )
639
640 for (const sslmode of ["prefer", "allow"]) {
641 it.effect.each([true, { rejectUnauthorized: false }] as const)(
642 `explicit ssl=%j prevents sslmode=${sslmode} fallback`,
643 (ssl) =>
644 Effect.gen(function*() {
645 const stream = makeSslStream("N", false)
646 const error = yield* Effect.flip(PgConnection.make({
647 url: Redacted.make(`postgres://test@db.example.com/db?sslmode=${sslmode}`),
648 ssl,
649 stream: () => stream.socket
650 }))
651
652 assert.deepStrictEqual(stream.writes, [sslRequest])
653 assert.strictEqual(error.reason._tag, "ConnectionError")
654 assert.strictEqual(error.reason.message, "PgConnection: Server refused TLS")
655 assert.isUndefined(tlsOptions.get(stream.socket))
656 assert.isTrue(stream.socket.destroyed)
657 })
658 )
659 }
660 })
661
662 it.live("preserves an ErrorResponse returned for SSLRequest", () =>
663 Effect.scoped(Effect.gen(function*() {
664 const { port } = yield* withTcpServer((socket) => {
665 socket.once("data", () => {
666 socket.end(errorResponse([
667 ["S", "FATAL"],
668 ["C", "08004"],
669 ["M", "TLS is disabled by the proxy"]
670 ]))
671 })
672 })
673
674 const error = yield* Effect.flip(PgConnection.make({
675 host: "127.0.0.1",
676 port,
677 username: "test",
678 ssl: true
679 }))
680
681 assert.strictEqual(error.reason._tag, "ConnectionError")
682 assert.strictEqual(error.reason.message, "PgConnection: Failed to negotiate TLS")
683 assert.instanceOf(error.reason.cause, Error)
684 assert.strictEqual(error.reason.cause.message, "TLS is disabled by the proxy")
685 })))
686
687 it.live("rejects SCRAM authentication without server verification", () =>
688 Effect.scoped(Effect.gen(function*() {
689 const { port } = yield* withTcpServer((socket) => {
690 consumeFrontend(socket, (tag) => {
691 if (tag === undefined) {
692 socket.write(authentication(10, Buffer.from("SCRAM-SHA-256\0\0")))
693 } else if (tag === "p") {
694 socket.write(Buffer.concat([authenticationOk, backendKeyData, readyForQuery]))
695 }
696 })
697 })
698
699 const error = yield* Effect.flip(PgConnection.make({
700 host: "127.0.0.1",
701 port,
702 username: "test",
703 password: Redacted.make("secret")
704 }))
705
706 assert.strictEqual(error.reason._tag, "AuthenticationError")
707 assert.strictEqual(error.reason.message, "PgConnection: SCRAM exchange did not complete")
708 })))
709
710 it.live("rejects incomplete SCRAM when AuthenticationOk is omitted", () =>
711 Effect.scoped(Effect.gen(function*() {
712 const { port } = yield* withTcpServer((socket) => {
713 consumeFrontend(socket, (tag) => {
714 if (tag === undefined) {
715 socket.write(authentication(10, Buffer.from("SCRAM-SHA-256\0\0")))
716 } else if (tag === "p") {
717 socket.write(Buffer.concat([backendKeyData, readyForQuery]))
718 }
719 })
720 })
721
722 const error = yield* Effect.flip(PgConnection.make({
723 host: "127.0.0.1",
724 port,
725 username: "test",
726 password: Redacted.make("secret")
727 }))
728
729 assert.strictEqual(error.reason._tag, "AuthenticationError")
730 assert.strictEqual(error.reason.message, "PgConnection: SCRAM exchange did not complete")
731 })))
732
733 it.live("lets URL query host and port override the authority", () =>
734 Effect.scoped(Effect.gen(function*() {
735 const { port } = yield* withTcpServer((socket) => {
736 consumeFrontend(socket, (tag) => {
737 if (tag === undefined) {
738 socket.write(Buffer.concat([authenticationOk, backendKeyData, readyForQuery]))
739 }
740 })
741 })
742
743 const connection = yield* PgConnection.make({
744 url: Redacted.make(`postgres://test@invalid.invalid:1/db?host=127.0.0.1&port=${port}`)
745 })
746
747 assert.strictEqual(connection.processId, 1234)
748 })))
749
750 it.effect("treats URL connect_timeout=0 as no timeout", () =>
751 Effect.scoped(Effect.gen(function*() {
752 let sendReady: (() => void) | undefined
753 let signalStartup!: () => void
754 const startup = new Promise<void>((resolve) => {
755 signalStartup = resolve
756 })
757 const { port } = yield* withTcpServer((socket) => {
758 consumeFrontend(socket, (tag) => {
759 if (tag === undefined) {
760 sendReady = () => socket.write(Buffer.concat([authenticationOk, backendKeyData, readyForQuery]))
761 signalStartup()
762 }
763 })
764 })
765
766 const fiber = yield* PgConnection.make({
767 url: Redacted.make(`postgres://test@127.0.0.1:${port}/db?connect_timeout=0`)
768 }).pipe(Effect.forkScoped)
769 yield* Effect.promise(() => startup)
770 yield* TestClock.adjust("6 seconds")
771 sendReady!()
772
773 const connection = yield* Fiber.join(fiber)
774 assert.strictEqual(connection.processId, 1234)
775 })))
776
777 it.live("classifies startup SQLSTATE errors with the shared table", () =>
778 Effect.scoped(Effect.gen(function*() {
779 const { port } = yield* withTcpServer((socket) => {
780 consumeFrontend(socket, (tag) => {
781 if (tag === undefined) {
782 socket.end(errorResponse([
783 ["S", "ERROR"],
784 ["C", "23505"],
785 ["M", "duplicate key value violates unique constraint"],
786 ["n", " users_email_key "]
787 ]))
788 }
789 })
790 })
791
792 const error = yield* Effect.flip(PgConnection.make({
793 host: "127.0.0.1",
794 port,
795 username: "test"
796 }))
797
798 assert.strictEqual(error.reason._tag, "UniqueViolation")
799 if (error.reason._tag === "UniqueViolation") {
800 assert.strictEqual(error.reason.constraint, "users_email_key")
801 }
802 })))
803
804 it.live("sends Terminate when the connection scope closes", () =>
805 Effect.scoped(Effect.gen(function*() {
806 let signalTerminate!: (message: Buffer) => void
807 const terminated = new Promise<Buffer>((resolve) => {
808 signalTerminate = resolve
809 })
810 const { port } = yield* withTcpServer((socket) => {
811 consumeFrontend(socket, (tag, message) => {
812 if (tag === undefined) {
813 socket.write(Buffer.concat([authenticationOk, backendKeyData, readyForQuery]))
814 } else if (tag === "X") {
815 signalTerminate(message)
816 }
817 })
818 })
819
820 yield* Effect.scoped(PgConnection.make({ host: "127.0.0.1", port, username: "test" }))
821 const message = yield* Effect.promise(() => terminated)
822
823 assert.strictEqual(message.toString("hex"), "5800000004")
824 })))
825
826 it.live("authenticates with a cleartext password request", () =>
827 Effect.scoped(Effect.gen(function*() {
828 let password: string | undefined
829 const { port } = yield* withTcpServer((socket) => {
830 consumeFrontend(socket, (tag, message) => {
831 if (tag === undefined) {
832 socket.write(authentication(3))
833 } else if (tag === "p") {
834 password = message.subarray(5, -1).toString()
835 socket.write(Buffer.concat([authenticationOk, backendKeyData, readyForQuery]))
836 }
837 })
838 })
839
840 const connection = yield* PgConnection.make({
841 host: "127.0.0.1",
842 port,
843 username: "test",
844 password: Redacted.make("secret")
845 })
846
847 assert.strictEqual(connection.processId, 1234)
848 assert.strictEqual(password, "secret")
849 })))
850
851 it.live("resolves a password Effect for each connection", () =>
852 Effect.gen(function*() {
853 const passwords: Array<string> = []
854 const { port } = yield* withTcpServer((socket) => {
855 consumeFrontend(socket, (tag, message) => {
856 if (tag === undefined) {
857 socket.write(authentication(3))
858 } else if (tag === "p") {
859 passwords.push(message.subarray(5, -1).toString())
860 socket.write(Buffer.concat([authenticationOk, backendKeyData, readyForQuery]))
861 }
862 })
863 })
864 let calls = 0
865 const acquire = PgConnection.make({
866 host: "127.0.0.1",
867 port,
868 username: "test",
869 password: Effect.sync(() => Redacted.make(`secret-${++calls}`))
870 })
871
872 yield* Effect.scoped(acquire)
873 yield* acquire
874
875 assert.strictEqual(calls, 2)
876 assert.deepStrictEqual(passwords, ["secret-1", "secret-2"])
877 }))
878
879 it.live("authenticates with an MD5 password request", () =>
880 Effect.scoped(Effect.gen(function*() {
881 let password: string | undefined
882 const { port } = yield* withTcpServer((socket) => {
883 consumeFrontend(socket, (tag, message) => {
884 if (tag === undefined) {
885 socket.write(authentication(5, Buffer.from([1, 2, 3, 4])))
886 } else if (tag === "p") {
887 password = message.subarray(5, -1).toString()
888 socket.write(Buffer.concat([authenticationOk, backendKeyData, readyForQuery]))
889 }
890 })
891 })
892
893 const connection = yield* PgConnection.make({
894 host: "127.0.0.1",
895 port,
896 username: "test",
897 password: Redacted.make("secret")
898 })
899
900 assert.strictEqual(connection.processId, 1234)
901 assert.strictEqual(password, "md5594f15006e55e17dff918116ee00778f")
902 })))
903
904 it.live("uses a caller-supplied stream factory", () =>
905 Effect.scoped(Effect.gen(function*() {
906 let streamCalls = 0
907 const { port } = yield* withTcpServer((socket) => {
908 consumeFrontend(socket, (tag) => {
909 if (tag === undefined) {
910 socket.write(Buffer.concat([authenticationOk, backendKeyData, readyForQuery]))
911 }
912 })
913 })
914
915 const connection = yield* PgConnection.make({
916 host: "invalid.invalid",
917 port: 1,
918 username: "test",
919 stream: () => {
920 streamCalls++
921 return Net.connect({ host: "127.0.0.1", port })
922 }
923 })
924
925 assert.strictEqual(connection.processId, 1234)
926 assert.strictEqual(streamCalls, 1)
927 })))
928
929 it.live("derives a unix socket path from a slash-prefixed host", () =>
930 Effect.scoped(Effect.gen(function*() {
931 const { directory } = yield* withUnixServer((socket) => {
932 consumeFrontend(socket, (tag) => {
933 if (tag === undefined) {
934 socket.write(Buffer.concat([authenticationOk, backendKeyData, readyForQuery]))
935 }
936 })
937 })
938
939 const connection = yield* PgConnection.make({ host: directory, username: "test" })
940
941 assert.strictEqual(connection.processId, 1234)
942 })))
943
944 it.live("lets explicit fields override URL connection and startup values", () =>
945 Effect.scoped(Effect.gen(function*() {
946 let parameters: ReadonlyMap<string, string> | undefined
947 const { port } = yield* withTcpServer((socket) => {
948 consumeFrontend(socket, (tag, message) => {
949 if (tag === undefined) {
950 parameters = startupParameters(message)
951 socket.write(Buffer.concat([authenticationOk, backendKeyData, readyForQuery]))
952 }
953 })
954 })
955
956 const connection = yield* PgConnection.make({
957 url: Redacted.make("postgres://url-user@invalid.invalid:1/url-db?application_name=url-app"),
958 host: "127.0.0.1",
959 port,
960 username: "explicit-user",
961 database: "explicit-db",
962 applicationName: "explicit-app"
963 })
964
965 assert.strictEqual(connection.processId, 1234)
966 assert.strictEqual(parameters!.get("user"), "explicit-user")
967 assert.strictEqual(parameters!.get("database"), "explicit-db")
968 assert.strictEqual(parameters!.get("application_name"), "explicit-app")
969 assert.strictEqual(parameters!.get("client_encoding"), "UTF8")
970 })))
971})