1import { PgClient } from "@effect/sql-pg"
2import { assert, expect, it } from "@effect/vitest"
3import { Cause, DateTime, Deferred, Effect, Fiber, Option, Queue, Schedule, Schema, Stream, String } from "effect"
4import * as Reactivity from "effect/reactivity/Reactivity"
5import { Model } from "effect/schema"
6import { SqlClient, SqlError, SqlModel } from "effect/sql"
7import * as Statement from "effect/sql/Statement"
8import { TestClock } from "effect/testing"
9import { PgContainer } from "./utils.ts"
10
11const compilerTransform = PgClient.makeCompiler(String.camelToSnake)
12const transformsNested = Statement.defaultTransforms(String.snakeToCamel)
13const transforms = Statement.defaultTransforms(String.snakeToCamel, false)
14
15it.layer(PgContainer.layerClient, { timeout: "30 seconds" })("PgClient", (it) => {
16 it.effect("round trips Model.DateTimeInsertFromDate through a repository", () =>
17 Effect.gen(function*() {
18 class Entry extends Model.Class<Entry>("Entry")({
19 id: Schema.Int.pipe(Model.FieldExcept(["insert"])),
20 created_at: Model.DateTimeInsertFromDate
21 }) {}
22
23 const sql = yield* PgClient.PgClient
24 const repo = yield* SqlModel.makeRepository(Entry, {
25 tableName: "model_timestamp",
26 idColumn: "id",
27 spanPrefix: "EntryRepository"
28 })
29 const instant = new Date("2024-05-06T07:08:09.123Z")
30 yield* sql.withTransaction(Effect.gen(function*() {
31 yield* sql`SET LOCAL TIME ZONE 'Europe/Berlin'`
32 yield* sql`CREATE TEMP TABLE model_timestamp (id SERIAL PRIMARY KEY, created_at TIMESTAMPTZ) ON COMMIT DROP`
33 const inserted = yield* repo.insert(
34 Entry.insert.make({ created_at: Model.Override(DateTime.makeUnsafe(instant)) })
35 )
36 assert.strictEqual(DateTime.toEpochMillis(inserted.created_at), instant.getTime())
37 const selected = yield* repo.findById(inserted.id)
38 assert.deepStrictEqual(selected, inserted)
39 assert.strictEqual(DateTime.toEpochMillis(selected.created_at), instant.getTime())
40 }))
41 }))
42
43 it.effect("insert helper", () =>
44 Effect.gen(function*() {
45 const sql = yield* PgClient.PgClient
46 const [query, params] = sql`INSERT INTO people ${sql.insert({ name: "Tim", age: 10 })}`.compile()
47 expect(query).toEqual(`INSERT INTO people ("name","age") VALUES ($1,$2)`)
48 expect(params).toEqual(["Tim", 10])
49 }))
50
51 it.effect("updateValues helper", () =>
52 Effect.gen(function*() {
53 const sql = yield* PgClient.PgClient
54 const [query, params] = sql`UPDATE people SET name = data.name FROM ${
55 sql.updateValues(
56 [{ name: "Tim" }, { name: "John" }],
57 "data"
58 )
59 }`.compile()
60 expect(query).toEqual(
61 `UPDATE people SET name = data.name FROM (values ($1),($2)) AS data("name")`
62 )
63 expect(params).toEqual(["Tim", "John"])
64 }))
65
66 it.effect("updateValues helper returning", () =>
67 Effect.gen(function*() {
68 const sql = yield* PgClient.PgClient
69 const [query, params] = sql`UPDATE people SET name = data.name FROM ${
70 sql.updateValues(
71 [{ name: "Tim" }, { name: "John" }],
72 "data"
73 ).returning("*")
74 }`.compile()
75 expect(query).toEqual(
76 `UPDATE people SET name = data.name FROM (values ($1),($2)) AS data("name") RETURNING *`
77 )
78 expect(params).toEqual(["Tim", "John"])
79 }))
80
81 it.effect("update helper", () =>
82 Effect.gen(function*() {
83 const sql = yield* PgClient.PgClient
84 let result = sql`UPDATE people SET ${sql.update({ name: "Tim" })}`.compile()
85 expect(result[0]).toEqual(`UPDATE people SET "name" = $1`)
86 expect(result[1]).toEqual(["Tim"])
87
88 result = sql`UPDATE people SET ${sql.update({ name: "Tim", age: 10 }, ["age"])}`.compile()
89 expect(result[0]).toEqual(`UPDATE people SET "name" = $1`)
90 expect(result[1]).toEqual(["Tim"])
91 }))
92
93 it.effect("update helper returning", () =>
94 Effect.gen(function*() {
95 const sql = yield* PgClient.PgClient
96 const result = sql`UPDATE people SET ${sql.update({ name: "Tim" }).returning("*")}`.compile()
97 expect(result[0]).toEqual(`UPDATE people SET "name" = $1 RETURNING *`)
98 expect(result[1]).toEqual(["Tim"])
99 }))
100
101 it.effect("array helper", () =>
102 Effect.gen(function*() {
103 const sql = yield* PgClient.PgClient
104 const [query, params] = sql`SELECT * FROM ${sql("people")} WHERE id IN ${sql.in([1, 2, "string"])}`.compile()
105 expect(query).toEqual(`SELECT * FROM "people" WHERE id IN ($1,$2,$3)`)
106 expect(params).toEqual([1, 2, "string"])
107 }))
108
109 it.effect("array helper with column", () =>
110 Effect.gen(function*() {
111 const sql = yield* PgClient.PgClient
112 let result = sql`SELECT * FROM ${sql("people")} WHERE ${sql.in("id", [1, 2, "string"])}`.compile()
113 expect(result[0]).toEqual(`SELECT * FROM "people" WHERE "id" IN ($1,$2,$3)`)
114 expect(result[1]).toEqual([1, 2, "string"])
115
116 result = sql`SELECT * FROM ${sql("people")} WHERE ${sql.in("id", [])}`.compile()
117 expect(result[0]).toEqual(`SELECT * FROM "people" WHERE 1=0`)
118 expect(result[1]).toEqual([])
119 }))
120
121 it.effect("and", () =>
122 Effect.gen(function*() {
123 const sql = yield* PgClient.PgClient
124 const now = new Date()
125 const result = sql`SELECT * FROM ${sql("people")} WHERE ${
126 sql.and([
127 sql.in("name", ["Tim", "John"]),
128 sql`created_at < ${now}`
129 ])
130 }`.compile()
131 expect(result[0]).toEqual(`SELECT * FROM "people" WHERE ("name" IN ($1,$2) AND created_at < $3)`)
132 expect(result[1]).toEqual(["Tim", "John", now])
133 }))
134
135 it("transform nested", () => {
136 assert.deepEqual(
137 transformsNested.array([
138 {
139 a_key: 1,
140 nested: [{ b_key: 2 }],
141 arr_primitive: [1, "2", true]
142 }
143 ]) as any,
144 [
145 {
146 aKey: 1,
147 nested: [{ bKey: 2 }],
148 arrPrimitive: [1, "2", true]
149 }
150 ]
151 )
152 })
153
154 it("transform non nested", () => {
155 assert.deepEqual(
156 transforms.array([
157 {
158 a_key: 1,
159 nested: [{ b_key: 2 }],
160 arr_primitive: [1, "2", true]
161 }
162 ]) as any,
163 [
164 {
165 aKey: 1,
166 nested: [{ b_key: 2 }],
167 arrPrimitive: [1, "2", true]
168 }
169 ]
170 )
171
172 assert.deepEqual(
173 transforms.array([
174 {
175 json_field: {
176 test_value: [1, true, null, "text"],
177 test_nested: {
178 test_value: [1, true, null, "text"]
179 }
180 }
181 }
182 ]) as any,
183 [
184 {
185 jsonField: {
186 test_value: [1, true, null, "text"],
187 test_nested: {
188 test_value: [1, true, null, "text"]
189 }
190 }
191 }
192 ]
193 )
194 })
195
196 it.effect("insert fragments", () =>
197 Effect.gen(function*() {
198 const sql = yield* PgClient.PgClient
199 const [query, params] = sql`INSERT INTO people ${
200 sql.insert({
201 name: "Tim",
202 age: 10,
203 json: sql.json({ a: 1 })
204 })
205 }`.compile()
206 assert.strictEqual(
207 query,
208 "INSERT INTO people (\"name\",\"age\",\"json\") VALUES ($1,$2,$3)"
209 )
210 assert.lengthOf(params, 3)
211 }))
212
213 it.effect("update fragments", () =>
214 Effect.gen(function*() {
215 const sql = yield* PgClient.PgClient
216 const now = new Date()
217 const [query, params] = sql`UPDATE people SET json = data.json FROM ${
218 sql.updateValues(
219 [{ json: sql.json({ a: 1 }) }, { json: sql.json({ b: 1 }) }],
220 "data"
221 )
222 } WHERE created_at > ${now}`.compile()
223 assert.strictEqual(
224 query,
225 `UPDATE people SET json = data.json FROM (values ($1),($2)) AS data("json") WHERE created_at > $3`
226 )
227 assert.lengthOf(params, 3)
228 }))
229
230 it.effect("onDialect", () =>
231 Effect.gen(function*() {
232 const sql = yield* PgClient.PgClient
233 assert.strictEqual(
234 sql.onDialect({
235 sqlite: () => "A",
236 pg: () => "B",
237 mysql: () => "C",
238 mssql: () => "D",
239 clickhouse: () => "E"
240 }),
241 "B"
242 )
243 assert.strictEqual(
244 sql.onDialectOrElse({
245 orElse: () => "A",
246 pg: () => "B"
247 }),
248 "B"
249 )
250 }))
251
252 it.effect("identifier transform", () =>
253 Effect.gen(function*() {
254 const sql = yield* PgClient.PgClient
255 const [query] = compilerTransform.compile(
256 sql`SELECT * from ${sql("peopleTest")}`,
257 false
258 )
259 expect(query).toEqual(`SELECT * from "people_test"`)
260 }))
261
262 it.effect("jsonb", () =>
263 Effect.gen(function*() {
264 const sql = yield* PgClient.PgClient
265 const rows = yield* sql<{ json: unknown }>`select ${sql.json({ testValue: 123 })}::jsonb as json`
266 expect(rows[0].json).toEqual({ testValue: 123 })
267 }))
268
269 it.effect("reads scalar enums as strings without registering their OIDs", () =>
270 Effect.gen(function*() {
271 const sql = yield* PgClient.PgClient
272 yield* sql.withTransaction(Effect.gen(function*() {
273 yield* sql`CREATE TYPE capability_scalar AS ENUM ('use_key', 'manage', 'gérer')`
274 const rows = yield* sql`
275 SELECT 'use_key'::capability_scalar AS capability,
276 ${"gérer"}::capability_scalar AS bound,
277 NULL::capability_scalar AS nullable
278 `
279 yield* sql`DROP TYPE capability_scalar`
280 assert.deepStrictEqual(rows, [{ capability: "use_key", bound: "gérer", nullable: null }])
281 }))
282 }))
283
284 it.effect("reads bytea as Uint8Array", () =>
285 Effect.gen(function*() {
286 const sql = yield* PgClient.PgClient
287 const payload = new Uint8Array([0, 1, 254, 255])
288 const rows = yield* sql`SELECT ${payload}::bytea AS payload`
289 assert.instanceOf(rows[0].payload, Uint8Array)
290 assert.deepStrictEqual(rows, [{ payload }])
291 }))
292
293 it.effect("stream", () =>
294 Effect.gen(function*() {
295 const sql = yield* SqlClient.SqlClient
296 const rows = yield* sql`SELECT generate_series(1, 3)`.stream.pipe(
297 Stream.runCollect
298 )
299 expect(rows).toEqual([
300 { "generate_series": 1 },
301 { "generate_series": 2 },
302 { "generate_series": 3 }
303 ])
304 }))
305
306 it.effect("releases completed nested transaction locks", () =>
307 Effect.gen(function*() {
308 const sql = yield* PgClient.PgClient
309 const locks = sql`SELECT count(*)::integer AS count FROM pg_locks
310 WHERE pid = pg_backend_pid() AND locktype = 'transactionid'`
311 yield* sql.withTransaction(Effect.gen(function*() {
312 yield* sql`CREATE TEMP TABLE savepoint_locks (value INTEGER) ON COMMIT DROP`
313
314 yield* sql.withTransaction(sql`INSERT INTO savepoint_locks VALUES (1)`)
315 assert.deepStrictEqual(yield* locks, [{ count: 1 }])
316
317 const error = yield* sql.withTransaction(
318 sql`INSERT INTO savepoint_locks VALUES (2)`.pipe(Effect.andThen(Effect.fail("rollback")))
319 ).pipe(Effect.flip)
320 assert.strictEqual(error, "rollback")
321 assert.deepStrictEqual(yield* locks, [{ count: 1 }])
322
323 assert.deepStrictEqual(yield* sql`SELECT value FROM savepoint_locks`, [{ value: 1 }])
324 }))
325 }))
326
327 it.effect("preserves successful concurrent nested transactions", () =>
328 Effect.gen(function*() {
329 const sql = yield* PgClient.PgClient
330 const firstStarted = yield* Deferred.make<void>()
331 const firstInserted = yield* Deferred.make<void>()
332
333 const rows = yield* sql.withTransaction(
334 Effect.gen(function*() {
335 yield* sql`CREATE TEMP TABLE nested_transactions (value TEXT) ON COMMIT DROP`
336 yield* Effect.all([
337 sql.withTransaction(
338 Effect.gen(function*() {
339 yield* Deferred.succeed(firstStarted, undefined)
340 yield* Effect.sleep("100 millis")
341 yield* sql`INSERT INTO nested_transactions VALUES ('first')`
342 yield* Deferred.succeed(firstInserted, undefined)
343 })
344 ),
345 Deferred.await(firstStarted).pipe(
346 Effect.andThen(sql.withTransaction(
347 Deferred.await(firstInserted).pipe(
348 Effect.andThen(Effect.fail("rollback"))
349 )
350 ))
351 )
352 ], { concurrency: "unbounded" }).pipe(Effect.catch(() => Effect.void))
353 return yield* sql<{ value: string }>`SELECT value FROM nested_transactions`
354 })
355 )
356
357 assert.deepStrictEqual(rows, [{ value: "first" }])
358 }).pipe(TestClock.withLive))
359
360 it.effect("fails a transaction whose COMMIT rolls back after a caught error", () =>
361 Effect.gen(function*() {
362 const sql = yield* PgClient.PgClient
363 const cause = yield* sql.withTransaction(Effect.gen(function*() {
364 yield* Effect.ignore(sql`SELECT 1 / 0`)
365 })).pipe(Effect.sandbox, Effect.flip)
366
367 assert.isTrue(Cause.hasDies(cause))
368 assert.isFalse(Cause.hasFails(cause))
369 const defect = Cause.squash(cause)
370 assert.instanceOf(defect, SqlError.SqlError)
371 assert.strictEqual(defect.reason._tag, "UnknownError")
372 assert.strictEqual(defect.reason.operation, "commit")
373 }))
374})
375
376it.layer(PgContainer.layerMakeClientUnprepared, { timeout: "30 seconds" })(
377 "PgClient.makeClient without preparation",
378 (it) => {
379 it.effect("fails an aborted COMMIT and reuses the connection", () =>
380 Effect.gen(function*() {
381 const sql = yield* PgClient.PgClient
382 const cause = yield* sql.withTransaction(Effect.gen(function*() {
383 yield* Effect.ignore(sql`SELECT 1 / 0`)
384 })).pipe(Effect.sandbox, Effect.flip)
385
386 assert.isTrue(Cause.hasDies(cause))
387 assert.isFalse(Cause.hasFails(cause))
388 const defect = Cause.squash(cause)
389 assert.instanceOf(defect, SqlError.SqlError)
390 assert.strictEqual(defect.reason._tag, "UnknownError")
391 assert.strictEqual(defect.reason.operation, "commit")
392
393 const rows = yield* sql.withTransaction(sql<{ value: number }>`SELECT 1 AS value`)
394 assert.deepStrictEqual(rows, [{ value: 1 }])
395 }))
396 }
397)
398
399it.layer(PgContainer.layerMakeClient, { timeout: "30 seconds" })("PgClient.makeClient", (it) => {
400 it.effect("connects before executing queries", () =>
401 Effect.gen(function*() {
402 const sql = yield* PgClient.PgClient
403 const rows = yield* sql<{ value: number }>`SELECT 1 AS value`
404 assert.deepStrictEqual(rows, [{ value: 1 }])
405 }))
406
407 it.effect("skips prepared statements for unprepared executions only", () =>
408 Effect.gen(function*() {
409 const sql = yield* PgClient.PgClient
410 const rows = sql<{ unprepared_row: number }>`SELECT ${1}::int4 AS unprepared_row`
411 const values = sql`SELECT ${2}::int4 AS unprepared_values`
412 const preparedStatements = sql<{ statement: string }>`
413 SELECT statement FROM pg_prepared_statements
414 WHERE statement IN (
415 'SELECT $1::int4 AS unprepared_row',
416 'SELECT $1::int4 AS unprepared_values'
417 )
418 ORDER BY statement
419 `
420
421 assert.deepStrictEqual(yield* rows.unprepared, [{ unprepared_row: 1 }])
422 assert.deepStrictEqual(yield* values.valuesUnprepared, [[2]])
423 assert.deepStrictEqual(yield* preparedStatements, [])
424
425 assert.deepStrictEqual(yield* rows, [{ unprepared_row: 1 }])
426 assert.deepStrictEqual(yield* values.values, [[2]])
427 assert.deepStrictEqual(yield* preparedStatements, [
428 { statement: "SELECT $1::int4 AS unprepared_row" },
429 { statement: "SELECT $1::int4 AS unprepared_values" }
430 ])
431 }))
432
433 it.effect("pins the primary connection for streams by default", () =>
434 Effect.gen(function*() {
435 const sql = yield* PgClient.PgClient
436 const streamStarted = yield* Deferred.make<void>()
437 const releaseStream = yield* Deferred.make<void>()
438 const streamFiber = yield* sql<{ value: number }>`SELECT generate_series(1, 2) AS value`.stream.pipe(
439 Stream.tap(() =>
440 Effect.andThen(
441 Deferred.succeed(streamStarted, undefined),
442 Deferred.await(releaseStream)
443 )
444 ),
445 Stream.runCollect,
446 Effect.forkScoped
447 )
448
449 yield* Deferred.await(streamStarted)
450 const rows = yield* sql<{ value: number }>`SELECT 1 AS value`.pipe(
451 Effect.timeoutOption("100 millis")
452 )
453 yield* Deferred.succeed(releaseStream, undefined)
454 yield* Fiber.join(streamFiber)
455
456 assert.isTrue(Option.isNone(rows))
457 }).pipe(TestClock.withLive))
458
459 it.effect("serializes transactions on its single connection", () =>
460 Effect.gen(function*() {
461 const sql = yield* PgClient.PgClient
462 const firstBodyStarted = yield* Deferred.make<void>()
463 const releaseFirstBody = yield* Deferred.make<void>()
464 const secondBodyStarted = yield* Deferred.make<void>()
465
466 const first = yield* sql.withTransaction(Effect.gen(function*() {
467 yield* Deferred.succeed(firstBodyStarted, undefined)
468 yield* Deferred.await(releaseFirstBody)
469 })).pipe(Effect.forkScoped)
470 yield* Deferred.await(firstBodyStarted)
471 const second = yield* sql.withTransaction(
472 Deferred.succeed(secondBodyStarted, undefined)
473 ).pipe(Effect.forkScoped)
474
475 const overlap = yield* Deferred.await(secondBodyStarted).pipe(
476 Effect.timeoutOption("100 millis"),
477 TestClock.withLive
478 )
479 yield* Deferred.succeed(releaseFirstBody, undefined)
480 yield* Fiber.join(first)
481 yield* Fiber.join(second)
482
483 assert.isTrue(Option.isNone(overlap))
484 }))
485})
486
487it.layer(PgContainer.layerMakeClientAcquireForStream, { timeout: "30 seconds" })(
488 "PgClient.makeClient acquireForStream",
489 (it) => {
490 it.effect("runs queries while a stream is active", () =>
491 Effect.gen(function*() {
492 const sql = yield* PgClient.PgClient
493 const streamStarted = yield* Deferred.make<void>()
494 const releaseStream = yield* Deferred.make<void>()
495 const streamFiber = yield* sql<{ value: number }>`SELECT generate_series(1, 2) AS value`.stream.pipe(
496 Stream.tap(() =>
497 Effect.andThen(
498 Deferred.succeed(streamStarted, undefined),
499 Deferred.await(releaseStream)
500 )
501 ),
502 Stream.runCollect,
503 Effect.forkScoped
504 )
505
506 yield* Deferred.await(streamStarted)
507 const rows = yield* sql<{ value: number }>`SELECT 1 AS value`.pipe(
508 Effect.timeoutOrElse({
509 duration: "3 seconds",
510 orElse: () => Effect.fail(new Error("query timed out while stream was active"))
511 })
512 )
513 yield* Deferred.succeed(releaseStream, undefined)
514 yield* Fiber.join(streamFiber)
515
516 assert.deepStrictEqual(rows, [{ value: 1 }])
517 }).pipe(TestClock.withLive))
518
519 it.effect("uses a new session for streams", () =>
520 Effect.gen(function*() {
521 const sql = yield* PgClient.PgClient
522 yield* sql`SET application_name = 'sticky-primary'`
523
524 const rows = yield* sql<{ applicationName: string }>`
525 SELECT current_setting('application_name') AS "applicationName"
526 `.stream.pipe(Stream.runCollect)
527
528 assert.deepStrictEqual(Array.from(rows), [{ applicationName: "side-default" }])
529 }))
530 }
531)
532
533it.effect("PgClient.makeClient surfaces transport creation failures", () =>
534 Effect.gen(function*() {
535 const error = yield* Effect.flip(PgClient.makeClient({
536 username: "test",
537 stream: () => {
538 throw new Error("connection failed")
539 }
540 }))
541
542 assert.strictEqual(error.reason._tag, "ConnectionError")
543 assert.strictEqual(error.reason.operation, "connect")
544 }).pipe(
545 Effect.scoped,
546 Effect.provide(Reactivity.layer)
547 ))
548
549it.layer(PgContainer.layerClientWithTransforms, { timeout: "30 seconds" })("PgClient transforms", (it) => {
550 it.effect("insert helper", () =>
551 Effect.gen(function*() {
552 const sql = yield* PgClient.PgClient
553 const [query, params] = sql`INSERT INTO people ${sql.insert({ firstName: "Tim", age: 10 })}`.compile()
554 expect(query).toEqual(`INSERT INTO people ("first_name","age") VALUES ($1,$2)`)
555 expect(params).toEqual(["Tim", 10])
556 }))
557
558 it.effect("insert helper withoutTransforms", () =>
559 Effect.gen(function*() {
560 const sql = (yield* PgClient.PgClient).withoutTransforms()
561 const [query, params] = sql`INSERT INTO people ${sql.insert({ first_name: "Tim", age: 10 })}`.compile()
562 expect(query).toEqual(`INSERT INTO people ("first_name","age") VALUES ($1,$2)`)
563 expect(params).toEqual(["Tim", 10])
564 }))
565
566 it.effect("rejects multi-statement queries", () =>
567 Effect.gen(function*() {
568 const sql = yield* SqlClient.SqlClient
569
570 const error = yield* Effect.flip(sql`
571 CREATE TABLE test_multi (id TEXT PRIMARY KEY, name TEXT);
572 SELECT 1;
573 `)
574
575 assert.strictEqual(error.reason._tag, "SqlSyntaxError")
576 }))
577
578 it.effect("interruption", () =>
579 Effect.gen(function*() {
580 const sql = yield* SqlClient.SqlClient
581 const conn = yield* sql.reserve
582 yield* conn.executeRaw("select pg_sleep(1000)", []).pipe(
583 Effect.timeoutOption("50 millis"),
584 TestClock.withLive
585 )
586 const value = yield* conn.executeValues("select 1", [])
587 expect(value).toEqual([[1]])
588 }))
589
590 it.effect("retains the supplied config", () =>
591 Effect.gen(function*() {
592 const sql = yield* PgClient.PgClient
593
594 assert.isDefined(sql.config.url)
595 expect(sql.config.transformResultNames).toEqual(String.snakeToCamel)
596 expect(sql.config.transformQueryNames).toEqual(String.camelToSnake)
597 }))
598})
599
600// Each listener uses one of two pool connections.
601it.layer(PgContainer.layerClientForListen, { timeout: "30 seconds", concurrent: false })("PgClient listen", (it) => {
602 it.effect("keeps queries available while a listener reserves one connection", () =>
603 Effect.gen(function*() {
604 const sql = yield* PgClient.PgClient
605 const channel = "pool_connection_listen"
606
607 const payloads = yield* sql.listen(channel)
608
609 const rows = yield* sql<{ value: number }>`SELECT 1 as value`.pipe(
610 Effect.timeoutOrElse({
611 duration: "3 seconds",
612 orElse: () => Effect.fail(new Error("query timed out while listener was active"))
613 })
614 )
615 expect(rows).toEqual([{ value: 1 }])
616
617 yield* sql.notify(channel, "payload")
618 const payload = yield* Queue.take(payloads).pipe(
619 Effect.timeoutOrElse({
620 duration: "3 seconds",
621 orElse: () => Effect.fail(new Error("listener did not receive notification in time"))
622 })
623 )
624 expect(payload.payload).toEqual("payload")
625 }).pipe(TestClock.withLive), { timeout: 20_000 })
626
627 it.effect("notify sends payload", () =>
628 Effect.gen(function*() {
629 const sql = yield* PgClient.PgClient
630 const channel = "pool_connection_notify"
631
632 const payloads = yield* sql.listen(channel)
633 yield* sql.notify(channel, "payload")
634
635 const payload = yield* Queue.take(payloads).pipe(
636 Effect.timeoutOrElse({
637 duration: "3 seconds",
638 orElse: () => Effect.fail(new Error("listener did not receive notification in time"))
639 })
640 )
641 expect(payload.payload).toEqual("payload")
642 }).pipe(TestClock.withLive), { timeout: 20_000 })
643
644 it.effect("retries a failed listener and receives notifications on a new connection", () =>
645 Effect.gen(function*() {
646 const sql = yield* PgClient.PgClient
647 const channel = "retry_listener"
648 const registered = yield* Queue.unbounded<void>()
649 const consumer = yield* Stream.unwrap(Effect.gen(function*() {
650 const notifications = yield* sql.listen(channel)
651 yield* Queue.offer(registered, undefined)
652 return Stream.fromQueue(notifications)
653 })).pipe(
654 Stream.retry(Schedule.recurs(1)),
655 Stream.runHead,
656 Effect.forkScoped
657 )
658
659 yield* Queue.take(registered)
660 const [listener] = yield* sql<{ pid: number }>`
661 SELECT pid FROM pg_stat_activity WHERE query = ${`LISTEN "${channel}"`}
662 `
663 assert.isDefined(listener)
664 yield* sql`SELECT pg_terminate_backend(${listener.pid})`
665
666 // Wait for registration; PostgreSQL does not replay missed notifications.
667 yield* Queue.take(registered)
668 const [replacement] = yield* sql<{ pid: number }>`
669 SELECT pid FROM pg_stat_activity
670 WHERE query = ${`LISTEN "${channel}"`} AND pid <> ${listener.pid}
671 `
672 assert.isDefined(replacement)
673 yield* sql.notify(channel, "after reconnect")
674 const notification = Option.getOrThrow(yield* Fiber.join(consumer))
675 assert.strictEqual(notification.channel, channel)
676 assert.strictEqual(notification.payload, "after reconnect")
677 }), { timeout: 20_000 })
678
679 it.effect("listen rejects channel names longer than 63 UTF-8 bytes", () =>
680 Effect.gen(function*() {
681 const sql = yield* PgClient.PgClient
682 const error = yield* Effect.flip(sql.listen("é".repeat(32)))
683
684 assert.strictEqual(error.message, "PostgreSQL channel names must not exceed 63 UTF-8 bytes")
685 assert.strictEqual(error.reason.operation, "listen")
686 }))
687
688 it.effect("notify rejects channel names longer than 63 UTF-8 bytes", () =>
689 Effect.gen(function*() {
690 const sql = yield* PgClient.PgClient
691 const error = yield* Effect.flip(sql.notify("é".repeat(32), "payload"))
692
693 assert.strictEqual(error.message, "PostgreSQL channel names must not exceed 63 UTF-8 bytes")
694 assert.strictEqual(error.reason.operation, "notify")
695 }))
696})