1import { PgliteClient } from "@effect/sql-pglite"
2import { assert, describe, layer } from "@effect/vitest"
3import { Deferred, Effect } from "effect"
4import { TestClock } from "effect/testing"
5
6const ClientLayer = PgliteClient.layer({})
7
8const setup = (table: string) =>
9 Effect.gen(function*() {
10 const sql = yield* PgliteClient.PgliteClient
11 yield* sql.unsafe(`CREATE TABLE IF NOT EXISTS ${table} (id SERIAL PRIMARY KEY, name TEXT)`)
12 yield* sql.unsafe(`TRUNCATE TABLE ${table} RESTART IDENTITY`)
13 return sql
14 })
15
16describe("PgliteClient transactions", () => {
17 layer(ClientLayer, { timeout: "30 seconds" })((it) => {
18 it.effect("withTransaction commit", () =>
19 Effect.gen(function*() {
20 const sql = yield* setup("tx_commit")
21 yield* sql.withTransaction(sql.unsafe(`INSERT INTO tx_commit (name) VALUES ('hello')`))
22 const rows = yield* sql.unsafe<{ name: string }>(`SELECT name FROM tx_commit`)
23 assert.deepStrictEqual(rows, [{ name: "hello" }])
24 }))
25
26 it.effect("withTransaction rollback", () =>
27 Effect.gen(function*() {
28 const sql = yield* setup("tx_rollback")
29 yield* sql.unsafe(`INSERT INTO tx_rollback (name) VALUES ('hello')`).pipe(
30 Effect.andThen(Effect.fail("boom")),
31 sql.withTransaction,
32 Effect.ignore
33 )
34 const rows = yield* sql.unsafe(`SELECT * FROM tx_rollback`)
35 assert.deepStrictEqual(rows, [])
36 }))
37
38 it.effect("nested transaction commits both", () =>
39 Effect.gen(function*() {
40 const sql = yield* setup("tx_nested_commit")
41 const stmt = sql.unsafe(`INSERT INTO tx_nested_commit (name) VALUES ('hello')`)
42 yield* stmt.pipe(Effect.andThen(() => stmt.pipe(sql.withTransaction)), sql.withTransaction)
43 const rows = yield* sql.unsafe<{ total: number }>(
44 `SELECT count(*)::int AS total FROM tx_nested_commit`
45 )
46 assert.strictEqual(rows.at(0)?.total, 2)
47 }))
48
49 it.effect("nested transaction rollback via savepoint", () =>
50 Effect.gen(function*() {
51 const sql = yield* setup("tx_nested_rollback")
52 const stmt = sql.unsafe(`INSERT INTO tx_nested_rollback (name) VALUES ('hello')`)
53 yield* stmt.pipe(
54 Effect.andThen(() => stmt.pipe(Effect.andThen(Effect.fail("boom")), sql.withTransaction, Effect.ignore)),
55 sql.withTransaction
56 )
57 const rows = yield* sql.unsafe<{ total: number }>(
58 `SELECT count(*)::int AS total FROM tx_nested_rollback`
59 )
60 assert.strictEqual(rows.at(0)?.total, 1)
61 }))
62
63 it.effect("releases completed nested transaction locks", () =>
64 Effect.gen(function*() {
65 const sql = yield* PgliteClient.PgliteClient
66 const locks = sql`SELECT count(*)::integer AS count FROM pg_locks
67 WHERE pid = pg_backend_pid() AND locktype = 'transactionid'`
68 yield* sql.withTransaction(Effect.gen(function*() {
69 yield* sql`CREATE TEMP TABLE savepoint_locks (value INTEGER) ON COMMIT DROP`
70
71 yield* sql.withTransaction(sql`INSERT INTO savepoint_locks VALUES (1)`)
72 assert.deepStrictEqual(yield* locks, [{ count: 1 }])
73
74 const error = yield* sql.withTransaction(
75 sql`INSERT INTO savepoint_locks VALUES (2)`.pipe(Effect.andThen(Effect.fail("rollback")))
76 ).pipe(Effect.flip)
77 assert.strictEqual(error, "rollback")
78 assert.deepStrictEqual(yield* locks, [{ count: 1 }])
79
80 assert.deepStrictEqual(yield* sql`SELECT value FROM savepoint_locks`, [{ value: 1 }])
81 }))
82 }))
83
84 it.effect("preserves successful concurrent nested transactions", () =>
85 Effect.gen(function*() {
86 const sql = yield* setup("tx_nested_concurrent")
87 const firstStarted = yield* Deferred.make<void>()
88 const firstInserted = yield* Deferred.make<void>()
89
90 yield* sql.withTransaction(
91 Effect.all([
92 sql.withTransaction(
93 Effect.gen(function*() {
94 yield* Deferred.succeed(firstStarted, undefined)
95 yield* Effect.sleep("100 millis")
96 yield* sql.unsafe(`INSERT INTO tx_nested_concurrent (name) VALUES ('first')`)
97 yield* Deferred.succeed(firstInserted, undefined)
98 })
99 ),
100 Deferred.await(firstStarted).pipe(
101 Effect.andThen(sql.withTransaction(
102 Deferred.await(firstInserted).pipe(
103 Effect.andThen(Effect.fail("rollback"))
104 )
105 ))
106 )
107 ], { concurrency: "unbounded" }).pipe(Effect.catch(() => Effect.void))
108 )
109
110 const rows = yield* sql.unsafe<{ name: string }>(`SELECT name FROM tx_nested_concurrent`)
111 assert.deepStrictEqual(rows, [{ name: "first" }])
112 }).pipe(TestClock.withLive))
113 })
114})