Fix handling of transactions with explicit isolation level (#3)

Author: sds

src/connection.ts

@@ -1,4 +1,4 @@
-import type {CompiledQuery, DatabaseConnection, QueryResult, TransactionSettings} from 'kysely'
+import {CompiledQuery, DatabaseConnection, QueryResult, TransactionSettings} from 'kysely'
 import type {Sql} from 'postgres'
 
 import {PostgresJSDialectError} from './errors.js'
@@ -24,17 +24,19 @@ export class PostgresJSConnection implements DatabaseConnection {
 
     this.#transaction = createPostgres({...this.#config, options: {...this.#config.options, max: 1}})
 
-    const statement = `start transaction${isolationLevel ? ` ${isolationLevel}` : ''}`
+    const compiledQuery = CompiledQuery.raw(
+      isolationLevel ? `start transaction isolation level ${isolationLevel}` : 'begin',
+    )
 
-    await this.#transaction.unsafe(statement)
+    await this.executeQuery(compiledQuery)
   }
 
   async commitTransaction(): Promise<void> {
     if (!this.#transaction) {
       throw new PostgresJSDialectError('no transaction to commit!')
     }
 
-    await this.#transaction`commit`
+    await this.executeQuery(CompiledQuery.raw('commit'))
 
     this.#releaseTransaction()
   }
@@ -58,7 +60,7 @@ export class PostgresJSConnection implements DatabaseConnection {
       throw new PostgresJSDialectError('no transaction to rollback!')
     }
 
-    await this.#transaction`rollback`
+    await this.executeQuery(CompiledQuery.raw('rollback'))
 
     this.#releaseTransaction()
   }

tests/nodejs/index.test.ts

@@ -1,4 +1,4 @@
-import {DeleteResult, InsertResult, UpdateResult, sql, type Kysely, type Transaction} from 'kysely'
+import {CompiledQuery, DeleteResult, InsertResult, UpdateResult, sql, type Kysely, type Transaction} from 'kysely'
 
 import {
   CONFIGS,
@@ -19,13 +19,19 @@ import {
 
 forEach(CONFIGS).describe('PostgresJSDialect: %s', (config: TestConfig) => {
   let ctx: TestContext
+  let executedQueries: CompiledQuery[] = []
 
   before(async function () {
-    ctx = await initTest(this, config.config)
+    ctx = await initTest(this, config.config, (event) => {
+      if (event.level === 'query') {
+        executedQueries.push(event.query)
+      }
+    })
   })
 
   beforeEach(async () => {
     await insertDefaultDataSet(ctx)
+    executedQueries = []
   })
 
   afterEach(async () => {
@@ -87,6 +93,57 @@ forEach(CONFIGS).describe('PostgresJSDialect: %s', (config: TestConfig) => {
     }
   })
 
+  it('should set the transaction isolation level', async () => {
+    await ctx.db
+      .transaction()
+      .setIsolationLevel('serializable')
+      .execute(async (trx) => {
+        await trx
+          .insertInto('person')
+          .values({
+            first_name: 'Foo',
+            last_name: 'Barson',
+            gender: 'male',
+          })
+          .execute()
+      })
+
+    expect(
+      executedQueries.map((it) => ({
+        sql: it.sql,
+        parameters: it.parameters,
+      })),
+    ).to.eql([
+      {
+        sql: 'start transaction isolation level serializable',
+        parameters: [],
+      },
+      {
+        sql: 'insert into "person" ("first_name", "last_name", "gender") values ($1, $2, $3)',
+        parameters: ['Foo', 'Barson', 'male'],
+      },
+      {sql: 'commit', parameters: []},
+    ])
+  })
+
+  it('should be able to start a transaction with a single connection', async () => {
+    const result = await ctx.db.connection().execute((db) => {
+      return db.transaction().execute((trx) => {
+        return trx
+          .insertInto('person')
+          .values({
+            first_name: 'Foo',
+            last_name: 'Barson',
+            gender: 'male',
+          })
+          .returning('first_name')
+          .executeTakeFirstOrThrow()
+      })
+    })
+
+    expect(result.first_name).to.equal('Foo')
+  })
+
   it('should stream results', async () => {
     const males: unknown[] = []