1import { PgConnection, PgTypes } from "@effect/sql-pg"
2import { assert, it } from "@effect/vitest"
3import { Cause, Deferred, Effect, Exit, Fiber, Queue, Redacted, Scope, Stream } from "effect"
4import type { SqlError } from "effect/sql/SqlError"
5import { PgContainer } from "./utils.ts"
6
7const makeConnection = (options?: PgConnection.Config) =>
8 Effect.gen(function*() {
9 const container = yield* PgContainer
10 return yield* PgConnection.make({
11 url: Redacted.make(container.getConnectionUri()),
12 ...options
13 })
14 })
15
16const assertInterruptedOnClose = (
17 scope: Scope.Closeable,
18 notifications: Queue.Dequeue<PgConnection.Notification, SqlError>
19) =>
20 Effect.gen(function*() {
21 const consumer = yield* Effect.forkScoped(Queue.take(notifications))
22 yield* Scope.close(scope, Exit.void)
23 const exit = yield* Fiber.await(consumer)
24 assert.isTrue(Exit.isFailure(exit) && Cause.hasInterruptsOnly(exit.cause))
25 })
26
27it.layer(PgContainer.layer, { timeout: "30 seconds" })("PgConnection", (it) => {
28 it.effect("interrupts notification consumers when the connection scope closes", () =>
29 Effect.gen(function*() {
30 const scope = yield* Scope.fork(yield* Scope.Scope)
31 const connection = yield* Scope.provide(makeConnection(), scope)
32 const notifications = yield* connection.listen("closed_listener")
33 yield* assertInterruptedOnClose(scope, notifications)
34 }))
35
36 it.effect("interrupts notification consumers when the listener scope closes", () =>
37 Effect.gen(function*() {
38 const scope = yield* Scope.fork(yield* Scope.Scope)
39 const connection = yield* makeConnection()
40 const notifications = yield* Scope.provide(connection.listen("closed_listener"), scope)
41 yield* assertInterruptedOnClose(scope, notifications)
42 }))
43
44 it.effect("connects through ReadyForQuery", () =>
45 Effect.gen(function*() {
46 const connection = yield* makeConnection()
47 assert.isAbove(connection.processId, 0)
48 }))
49
50 it.effect("runs unnamed extended queries with inferred binary parameters", () =>
51 Effect.gen(function*() {
52 const connection = yield* makeConnection()
53 const instant = new Date("2024-05-06T07:08:09.123Z")
54 const result = yield* connection.query(
55 "SELECT $1::int4 AS i, $2::float8 AS f, $3::int8 AS b, $4::bool AS flag, " +
56 "$5::text AS text, $6::bytea AS bytes, $7::timestamptz AS instant, " +
57 "$8::int4[] AS numbers, $9::text AS nil, $10::jsonb AS jsonb",
58 [
59 1,
60 1.5,
61 BigInt(2),
62 true,
63 "hello",
64 new Int8Array([1, -1]),
65 instant,
66 [1, null, 3],
67 null,
68 PgTypes.jsonb({ nested: true })
69 ]
70 )
71 assert.strictEqual(result.command, "SELECT")
72 assert.strictEqual(result.rowCount, 1)
73 assert.strictEqual(result.oid, null)
74 assert.deepStrictEqual(result.rows, [{
75 i: 1,
76 f: 1.5,
77 b: BigInt(2),
78 flag: true,
79 text: "hello",
80 bytes: new Uint8Array([1, 255]),
81 instant,
82 numbers: [1, null, 3],
83 nil: null,
84 jsonb: { nested: true }
85 }])
86 assert.deepStrictEqual(result.fields.map((field) => field.name), [
87 "i",
88 "f",
89 "b",
90 "flag",
91 "text",
92 "bytes",
93 "instant",
94 "numbers",
95 "nil",
96 "jsonb"
97 ])
98 }))
99
100 it.effect("applies the session TimeZone only to Dates bound into timestamp columns", () =>
101 Effect.gen(function*() {
102 const connection = yield* makeConnection()
103 const instant = new Date("2024-05-06T07:08:09.123Z")
104 yield* connection.query(
105 "CREATE TEMP TABLE timestamp_timezone (bare timestamp, explicit timestamp, zoned timestamptz)"
106 )
107 for (
108 const [zone, expectedBare] of [
109 ["UTC", instant],
110 ["Europe/Berlin", new Date("2024-05-06T09:08:09.123Z")]
111 ] as const
112 ) {
113 yield* connection.query(`SET TIME ZONE '${zone}'`)
114 yield* connection.query("TRUNCATE timestamp_timezone")
115 yield* connection.query(
116 "INSERT INTO timestamp_timezone VALUES ($1, $2, $3)",
117 [instant, PgTypes.timestamp(instant), instant]
118 )
119 const result = yield* connection.query("SELECT * FROM timestamp_timezone")
120 assert.deepStrictEqual(result.rows, [{ bare: expectedBare, explicit: instant, zoned: instant }])
121 }
122 }))
123
124 it.effect("returns queryValues in RowDescription order", () =>
125 Effect.gen(function*() {
126 const connection = yield* makeConnection()
127 const rows = yield* connection.queryValues(
128 "SELECT $1::int8 AS second, $2::text AS first",
129 [PgTypes.int8(BigInt(7)), "value"]
130 )
131 assert.deepStrictEqual(rows, [[BigInt(7), "value"]])
132 }))
133
134 it.effect("drains query errors and keeps the connection usable", () =>
135 Effect.gen(function*() {
136 const connection = yield* makeConnection()
137 yield* connection.query("CREATE TEMP TABLE pg_connection_unique (value int4 UNIQUE)")
138 yield* connection.query("INSERT INTO pg_connection_unique VALUES ($1)", [1])
139 const error = yield* Effect.flip(connection.query("INSERT INTO pg_connection_unique VALUES ($1)", [1]))
140 assert.strictEqual(error.reason._tag, "UniqueViolation")
141 if (error.reason._tag === "UniqueViolation") {
142 assert.include(error.reason.constraint, "pg_connection_unique")
143 }
144 const result = yield* connection.query("SELECT count(*)::int8 AS count FROM pg_connection_unique")
145 assert.deepStrictEqual(result.rows, [{ count: BigInt(1) }])
146 }))
147
148 it.effect("binds untyped strings and integers beyond int4", () =>
149 Effect.gen(function*() {
150 const connection = yield* makeConnection()
151 yield* connection.query("CREATE TEMP TABLE inference_widening (id bigint, at timestamptz)")
152 yield* connection.query(
153 "INSERT INTO inference_widening (id, at) VALUES ($1, $2)",
154 ["12345678901234", "2024-05-06T07:08:09Z"]
155 )
156 yield* connection.query("INSERT INTO inference_widening (id) VALUES ($1)", [1755000000000])
157 const rows = yield* connection.queryValues("SELECT id FROM inference_widening ORDER BY id")
158 assert.deepStrictEqual(rows.map((row) => row[0]), [BigInt(1755000000000), BigInt("12345678901234")])
159 }))
160
161 it.effect("reparses a statement whose first execution failed at bind", () =>
162 Effect.gen(function*() {
163 const connection = yield* makeConnection()
164 const missing = yield* Effect.flip(connection.query("select $1::regclass", ["prepared_bind_failure"]))
165 assert.strictEqual(missing.reason._tag, "SqlSyntaxError")
166 yield* connection.query("CREATE TEMP TABLE prepared_bind_failure (id int)")
167 const found = yield* connection.query("select $1::regclass AS rel", ["prepared_bind_failure"])
168 assert.strictEqual(found.rows.length, 1)
169 }))
170
171 it.effect("fails with AuthenticationError on a bad password", () =>
172 Effect.gen(function*() {
173 const container = yield* PgContainer
174 const error = yield* Effect.flip(PgConnection.make({
175 host: container.getHost(),
176 port: container.getMappedPort(5432),
177 username: container.getUsername(),
178 password: Redacted.make("definitely-wrong"),
179 database: container.getDatabase()
180 }))
181 assert.strictEqual(error._tag, "SqlError")
182 assert.strictEqual(error.reason._tag, "AuthenticationError")
183 }))
184
185 it.effect("fails when the server refuses TLS", () =>
186 Effect.gen(function*() {
187 const container = yield* PgContainer
188 const error = yield* Effect.flip(PgConnection.make({
189 host: container.getHost(),
190 port: container.getMappedPort(5432),
191 username: container.getUsername(),
192 password: Redacted.make(container.getPassword()),
193 database: container.getDatabase(),
194 ssl: true
195 }))
196 assert.strictEqual(error.reason._tag, "ConnectionError")
197 assert.include(error.reason.message, "refused TLS")
198 }))
199 it.effect("reuses a prepared statement and keeps its columns", () =>
200 Effect.gen(function*() {
201 const connection = yield* makeConnection()
202 for (let index = 0; index < 3; index++) {
203 const result = yield* connection.query("SELECT $1::int4 AS value, $2::text AS label", [index, "x"])
204 assert.deepStrictEqual(result.rows, [{ value: index, label: "x" }])
205 assert.deepStrictEqual(result.fields, [
206 { name: "value", dataTypeId: 23 },
207 { name: "label", dataTypeId: 25 }
208 ])
209 }
210 const prepared = yield* connection.query("SELECT count(*)::int4 AS n FROM pg_prepared_statements")
211 assert.deepStrictEqual(prepared.rows[0], { n: 2 })
212 }))
213
214 it.effect("prepares the same text once per parameter signature", () =>
215 Effect.gen(function*() {
216 const connection = yield* makeConnection()
217 assert.deepStrictEqual((yield* connection.query("SELECT $1 AS v", [1])).rows, [{ v: 1 }])
218 assert.deepStrictEqual((yield* connection.query("SELECT $1 AS v", [BigInt(2)])).rows, [{ v: BigInt(2) }])
219 const types = yield* connection.query(
220 "SELECT parameter_types::text AS t FROM pg_prepared_statements WHERE statement = 'SELECT $1 AS v' ORDER BY t"
221 )
222 assert.deepStrictEqual(types.rows, [{ t: "{bigint}" }, { t: "{integer}" }])
223 }))
224
225 it.effect("re-parses a statement whose result type changed", () =>
226 Effect.gen(function*() {
227 const connection = yield* makeConnection()
228 yield* connection.query("CREATE TEMP TABLE shifting (a int)")
229 yield* connection.query("INSERT INTO shifting VALUES (1)")
230 assert.deepStrictEqual((yield* connection.query("SELECT * FROM shifting")).rows, [{ a: 1 }])
231 yield* connection.query("ALTER TABLE shifting ADD COLUMN b text DEFAULT 'added'")
232 const after = yield* connection.query("SELECT * FROM shifting")
233 assert.deepStrictEqual(after.rows, [{ a: 1, b: "added" }])
234 assert.deepStrictEqual(after.fields, [
235 { name: "a", dataTypeId: 23 },
236 { name: "b", dataTypeId: 25 }
237 ])
238 }))
239
240 it.effect("re-parses a statement the backend no longer holds", () =>
241 Effect.gen(function*() {
242 const connection = yield* makeConnection()
243 assert.deepStrictEqual((yield* connection.query("SELECT $1::int4 AS n", [7])).rows, [{ n: 7 }])
244 yield* connection.query("DEALLOCATE ALL")
245 assert.deepStrictEqual((yield* connection.query("SELECT $1::int4 AS n", [8])).rows, [{ n: 8 }])
246 }))
247
248 it.effect("closes statements it evicts from a full cache", () =>
249 Effect.gen(function*() {
250 const connection = yield* makeConnection({ preparedStatementCacheSize: 3 })
251 for (let round = 0; round < 2; round++) {
252 for (let index = 0; index < 8; index++) {
253 const result = yield* connection.query(`SELECT ${index}::int4 AS n, $1::text AS t`, ["v"])
254 assert.deepStrictEqual(result.rows, [{ n: index, t: "v" }])
255 }
256 }
257 const held = yield* connection.query("SELECT count(*)::int4 AS n FROM pg_prepared_statements")
258 assert.deepStrictEqual(held.rows[0], { n: 3 })
259 }))
260
261 it.effect("closes statements parsed while a pipeline overfills the cache", () =>
262 Effect.gen(function*() {
263 const connection = yield* makeConnection({ preparedStatementCacheSize: 1, multiplex: true })
264 yield* Effect.all(
265 Array.from({ length: 4 }, (_, index) => connection.query(`SELECT ${index}::int4 AS n`)),
266 { concurrency: "unbounded" }
267 )
268 const held = yield* connection.query("SELECT count(*)::int4 AS n FROM pg_prepared_statements")
269 assert.deepStrictEqual(held.rows[0], { n: 1 })
270 }))
271
272 it.effect("parses every statement when prepare is off", () =>
273 Effect.gen(function*() {
274 const connection = yield* makeConnection({ prepare: false })
275 for (let index = 0; index < 3; index++) {
276 assert.deepStrictEqual((yield* connection.query("SELECT $1::int4 AS n", [index])).rows, [{ n: index }])
277 }
278 const held = yield* connection.query("SELECT count(*)::int4 AS n FROM pg_prepared_statements")
279 assert.deepStrictEqual(held.rows[0], { n: 0 })
280 }))
281
282 it.effect("stays usable after a statement the server refuses to parse", () =>
283 Effect.gen(function*() {
284 const connection = yield* makeConnection()
285 const missing = yield* Effect.flip(connection.query("SELECT * FROM does_not_exist"))
286 assert.strictEqual(missing.reason._tag, "SqlSyntaxError")
287 const syntax = yield* Effect.flip(connection.query("SELECT FROM WHERE ????"))
288 assert.strictEqual(syntax.reason._tag, "SqlSyntaxError")
289 assert.deepStrictEqual((yield* connection.query("SELECT 42::int4 AS answer")).rows, [{ answer: 42 }])
290 }))
291 it.effect("closes a statement whose plan it had to throw away", () =>
292 Effect.gen(function*() {
293 const connection = yield* makeConnection()
294 yield* connection.query("CREATE TEMP TABLE churn (a int)")
295 for (let round = 0; round < 4; round++) {
296 yield* connection.query("SELECT * FROM churn")
297 yield* connection.query(`ALTER TABLE churn ADD COLUMN c${round} text`)
298 }
299 yield* connection.query("SELECT * FROM churn")
300 // Every re-parse after a stale plan has to close the name it replaced,
301 // or each DDL change strands a statement on the backend.
302 const held = yield* connection.query(
303 "SELECT count(*)::int4 AS n FROM pg_prepared_statements WHERE statement = 'SELECT * FROM churn'"
304 )
305 assert.deepStrictEqual(held.rows[0], { n: 1 })
306 }))
307 it.effect("re-parses a statement first prepared in a rolled back transaction", () =>
308 Effect.gen(function*() {
309 const connection = yield* makeConnection()
310 yield* connection.query("CREATE TEMP TABLE rolled_back (a int)")
311
312 // Postgres drops a statement prepared inside a transaction it rolls
313 // back, so the cached name is gone even though the session lives on.
314 yield* connection.query("BEGIN")
315 yield* connection.query("INSERT INTO rolled_back VALUES ($1::int4)", [1])
316 yield* connection.query("ROLLBACK")
317
318 yield* connection.query("INSERT INTO rolled_back VALUES ($1::int4)", [2])
319 yield* connection.query("INSERT INTO rolled_back VALUES ($1::int4)", [3])
320 assert.deepStrictEqual(
321 (yield* connection.query("SELECT a FROM rolled_back ORDER BY a")).rows,
322 [{ a: 2 }, { a: 3 }]
323 )
324 }))
325 it.effect("isolates errors between pipelined queries", () =>
326 Effect.gen(function*() {
327 const connection = yield* makeConnection({ multiplex: true })
328 const exits = yield* Effect.all([
329 Effect.exit(connection.query("SELECT $1::int4 AS first", [1])),
330 Effect.exit(connection.query("SELECT missing_pipeline_column")),
331 Effect.exit(connection.query("SELECT $1::int4 AS third", [3]))
332 ], { concurrency: "unbounded" })
333
334 assert.strictEqual(exits[0]._tag, "Success")
335 assert.strictEqual(exits[1]._tag, "Failure")
336 assert.strictEqual(exits[2]._tag, "Success")
337 if (exits[0]._tag === "Success") assert.deepStrictEqual(exits[0].value.rows, [{ first: 1 }])
338 if (exits[2]._tag === "Success") assert.deepStrictEqual(exits[2].value.rows, [{ third: 3 }])
339 }))
340
341 it.effect("drains an interrupted pipelined query without cancelling its neighbors", () =>
342 Effect.gen(function*() {
343 const connection = yield* makeConnection({ multiplex: true })
344 const first = yield* connection.query("SELECT pg_sleep(0.05), 1 AS first").pipe(Effect.forkScoped)
345 const second = yield* connection.query("SELECT 2 AS second").pipe(Effect.forkScoped)
346 yield* Effect.yieldNow
347 yield* Fiber.interrupt(first)
348
349 const result = yield* Fiber.join(second)
350 assert.deepStrictEqual(result.rows, [{ second: 2 }])
351 assert.deepStrictEqual((yield* connection.query("SELECT 3 AS after")).rows, [{ after: 3 }])
352 }))
353
354 it.effect("places pin requests between earlier and later pipelined queries", () =>
355 Effect.gen(function*() {
356 const connection = yield* makeConnection({ multiplex: true })
357 yield* connection.query(
358 "CREATE TEMP TABLE pipeline_pin_order (position int GENERATED ALWAYS AS IDENTITY, value int4)"
359 )
360 const first = yield* connection.query(
361 "INSERT INTO pipeline_pin_order (value) SELECT 1 FROM pg_sleep(0.05)"
362 ).pipe(Effect.forkScoped)
363 yield* Effect.yieldNow
364
365 const pinned = yield* Deferred.make<void>()
366 const release = yield* Deferred.make<void>()
367 const pinFiber = yield* Effect.scoped(Effect.gen(function*() {
368 const exclusive = yield* connection.pin
369 yield* exclusive.query("INSERT INTO pipeline_pin_order (value) VALUES (2)")
370 yield* Deferred.succeed(pinned, undefined)
371 yield* Deferred.await(release)
372 })).pipe(Effect.forkScoped)
373 yield* Effect.yieldNow
374 const third = yield* connection.query("INSERT INTO pipeline_pin_order (value) VALUES (3)").pipe(
375 Effect.forkScoped
376 )
377
378 yield* Deferred.await(pinned)
379 assert.isUndefined(third.pollUnsafe())
380 yield* Deferred.succeed(release, undefined)
381 yield* Fiber.join(first)
382 yield* Fiber.join(pinFiber)
383 yield* Fiber.join(third)
384
385 const result = yield* connection.query("SELECT value FROM pipeline_pin_order ORDER BY position")
386 assert.deepStrictEqual(result.rows, [{ value: 1 }, { value: 2 }, { value: 3 }])
387 }))
388
389 it.effect("reuses prepared statements across a pipeline", () =>
390 Effect.gen(function*() {
391 const connection = yield* makeConnection({ multiplex: true })
392 for (let round = 0; round < 3; round++) {
393 const results = yield* Effect.all(
394 Array.from({ length: 8 }, (_, index) => connection.query("SELECT $1::int4 AS n", [index])),
395 { concurrency: "unbounded" }
396 )
397 results.forEach((result, index) => assert.deepStrictEqual(result.rows, [{ n: index }]))
398 }
399 const held = yield* connection.query(
400 "SELECT count(*)::int4 AS n FROM pg_prepared_statements WHERE statement = 'SELECT $1::int4 AS n'"
401 )
402 assert.deepStrictEqual(held.rows[0], { n: 1 })
403 }))
404 it.effect("orders ordinary statements against a stream on the same session", () =>
405 Effect.gen(function*() {
406 const connection = yield* makeConnection()
407 // An ordinary statement holds `owner` rather than the wire permit, so
408 // this is what says it still cannot interleave with a stream, which
409 // reaches the session through a pin.
410 const streamed = yield* Effect.forkScoped(
411 Stream.runCollect(connection.stream("SELECT generate_series(1, 200) AS n"))
412 )
413 const queried = yield* Effect.forkScoped(
414 Effect.all(
415 Array.from({ length: 12 }, (_, index) => connection.query("SELECT $1::int4 AS n", [index])),
416 { concurrency: "unbounded" }
417 )
418 )
419 const rows = yield* Fiber.join(streamed)
420 assert.strictEqual(Array.from(rows).length, 200)
421 const results = yield* Fiber.join(queried)
422 results.forEach((result, index) => assert.deepStrictEqual(result.rows, [{ n: index }]))
423 assert.deepStrictEqual((yield* connection.query("SELECT 1 AS ok")).rows, [{ ok: 1 }])
424 }))
425})