feat: add native Oracle transactions

Author: ryanrasti

docs/ARCHITECTURE.md

@@ -57,10 +57,12 @@ place, in one language.
 ### `Driver` vs `Database` vs `Connection`
 
 - `Driver` is the low-level connection layer (`PgDriver`, `PgliteDriver`,
-  `SqliteDriver`, `DoSqliteDriver`). It exposes `execute(sql)`,
-  `runInSingleConnection(fn)`, `close()`, and a `dialect`. Each lives at its
-  own entry point (`typegres/drivers/*`) so a bundle only ever resolves the
-  optional peer it actually imports.
+  `SqliteDriver`, `DoSqliteDriver`, `OracleDriver`). It exposes `execute(sql)`,
+  `runInTransaction(options, fn)`, `close()`, and a `dialect`. Each driver owns
+  connection pinning plus its database's transaction protocol and supplies the
+  transaction-bound executor to `fn`. Drivers live at separate entry points
+  (`typegres/drivers/*`) so a bundle only ever resolves the optional peer it
+  actually imports.
 - `Database` is the schema handle: provenance identity and the `Table`
   factory, no driver of its own. `typegres()` constructs one synchronously,
   so table classes can be declared at module load without a top-level await.

src/database.ts

@@ -1,9 +1,15 @@
-import { type Driver, isSyncDriver, type QueryResult } from "./drivers/types";
+import {
+  type Driver,
+  isSyncDriver,
+  type QueryResult,
+  type TransactionIsolation,
+  type TransactionOptions,
+} from "./drivers/types";
 import type { Fromable, RowType, RowTypeToTsType } from "./builder/query";
 import { QueryBuilder, hydrateRows } from "./builder/query";
 import { deserializeRows } from "./util";
 import type { Sql } from "./builder/sql";
-import { compile, sql, Ident } from "./builder/sql";
+import { Ident } from "./builder/sql";
 import { Table, type TableBase, type TableOptions } from "./table";
 import { Values } from "./builder/values";
 import { InsertBuilder } from "./builder/insert";
@@ -15,10 +21,7 @@ import { PgExecutor } from "./live/pg/executor";
 import { StatementExecutor, type Executor } from "./executor";
 import type { DialectName } from "./builder/sql";
 
-export type TransactionIsolation = "read committed" | "repeatable read" | "serializable";
-export type TransactionOptions = {
-  isolation?: TransactionIsolation;
-};
+export type { TransactionIsolation, TransactionOptions } from "./drivers/types";
 
 // Postgres isolation levels are totally ordered. A nested call asking for
 // weaker-or-equal isolation than the active txn flattens harmlessly (caller
@@ -30,10 +33,10 @@ export type TransactionOptions = {
 // `default_transaction_isolation`). We can't prove what level we got, so
 // any *explicit* nested request inside an ambient txn must throw — the
 // alternative would silently downgrade the caller's expectation.
-const ISOLATION: { [K in TransactionIsolation]: { rank: number; begin: Sql } } = {
-  "read committed": { rank: 0, begin: sql`BEGIN ISOLATION LEVEL READ COMMITTED` },
-  "repeatable read": { rank: 1, begin: sql`BEGIN ISOLATION LEVEL REPEATABLE READ` },
-  "serializable":    { rank: 2, begin: sql`BEGIN ISOLATION LEVEL SERIALIZABLE` },
+const ISOLATION: { [K in TransactionIsolation]: { rank: number } } = {
+  "read committed": { rank: 0 },
+  "repeatable read": { rank: 1 },
+  "serializable": { rank: 2 },
 };
 
 // Provenance identity, no driver and no dialect of its own. Construction
@@ -323,61 +326,38 @@ export class Connection<C = undefined> {
       }
       return fn(this);
     }
+    const driver = this.driver;
     const bus = this.#bus;
-    return this.driver.runInSingleConnection(async (execute) => {
-      const driver = this.driver;
-      let txExecutor: Executor;
-      if (this.database.dialect === "postgres") {
-        txExecutor = new PgExecutor(this.database, execute, true);
-      } else if (this.database.dialect === "sqlite") {
-        if (!isSyncDriver(driver)) {
-          throw new Error("unreachable: sqlite Connection without a SyncDriver");
-        }
-        // Bound and pooled are the same channel on sqlite's one handle —
-        // checked, not assumed; the bound executor differs only in event
-        // timing (commit-deferred flush).
-        if (execute !== driver.executeSync) {
-          throw new Error(
-            "sync driver must pass its executeSync to runInSingleConnection — one handle, one channel",
-          );
-        }
-        if (!bus) {
-          throw new Error("sqlite Connection is missing its live bus");
-        }
-        txExecutor = new SqliteLiveExecutor(this.database, driver, bus, true);
-      } else {
-        txExecutor = new StatementExecutor(this.database, execute, true);
-      }
-      const tx = new Connection<C>(this.database, this.driver, txExecutor, opts?.isolation);
-      // Drivers with a native transaction protocol (Durable Objects) own
-      // commit/rollback; everyone else gets BEGIN/COMMIT/ROLLBACK SQL.
-      if (driver.runInTransaction) {
-        try {
-          const result = await driver.runInTransaction(() => fn(tx));
-          txExecutor.onCommit();
-          return result;
-        } catch (e) {
-          txExecutor.onRollback();
-          throw e;
-        }
-      }
-      const runSql = async (s: Sql) => execute(compile(s, { database: this.database }));
-      await runSql(opts?.isolation ? ISOLATION[opts.isolation].begin : sql`BEGIN`);
-      try {
-        const result = await fn(tx);
-        await runSql(sql`COMMIT`);
-        txExecutor.onCommit();
-        return result;
-      } catch (e) {
-        try {
-          await runSql(sql`ROLLBACK`);
-        } catch (rollbackErr) {
-          console.error("ROLLBACK failed after transaction error:", rollbackErr);
+    let txExecutor: Executor | undefined;
+    try {
+      const result = await driver.runInTransaction(opts ?? {}, async (execute) => {
+        if (this.database.dialect === "postgres") {
+          txExecutor = new PgExecutor(this.database, execute, true);
+        } else if (this.database.dialect === "sqlite") {
+          if (!isSyncDriver(driver)) {
+            throw new Error("unreachable: sqlite Connection without a SyncDriver");
+          }
+          if (execute !== driver.executeSync) {
+            throw new Error(
+              "sync driver must pass its executeSync to the transaction — one handle, one channel",
+            );
+          }
+          if (!bus) {
+            throw new Error("sqlite Connection is missing its live bus");
+          }
+          txExecutor = new SqliteLiveExecutor(this.database, driver, bus, true);
+        } else {
+          txExecutor = new StatementExecutor(this.database, execute, true);
         }
-        txExecutor.onRollback();
-        throw e;
-      }
-    });
+        const tx = new Connection<C>(this.database, driver, txExecutor, opts?.isolation);
+        return fn(tx);
+      });
+      txExecutor?.onCommit();
+      return result;
+    } catch (e) {
+      txExecutor?.onRollback();
+      throw e;
+    }
   }
 
   async close(): Promise<void> {

src/drivers/do.ts

@@ -1,5 +1,5 @@
 import type { CompiledSql } from "../builder/sql";
-import type { ExecuteFn, ExecuteSyncFn, QueryResult, SyncDriver } from "./types";
+import type { ExecuteFn, ExecuteSyncFn, QueryResult, SyncDriver, TransactionOptions } from "./types";
 import { normalizeRow } from "./shared-sqlite";
 import { stripMatchedOuterParens } from "./shared";
 
@@ -51,12 +51,10 @@ export class DoSqliteDriver implements SyncDriver {
   };
 
   // storage.transaction() commits on resolution, rolls back on throw.
-  runInTransaction = <T>(cb: () => Promise<T>): Promise<T> => this.storage.transaction(cb);
-
-  // One handle: the single-connection execute IS executeSync (callers
-  // assert this identity — see Connection.transaction).
-  runInSingleConnection = <T>(cb: (execute: ExecuteSyncFn) => Promise<T>): Promise<T> =>
-    cb(this.executeSync);
+  runInTransaction = <T>(
+    _opts: TransactionOptions,
+    cb: (execute: ExecuteSyncFn) => Promise<T>,
+  ): Promise<T> => this.storage.transaction(() => cb(this.executeSync));
 
   close = (): Promise<void> => Promise.resolve();
 }

src/drivers/oracle-transaction.test.ts

@@ -0,0 +1,83 @@
+import { afterAll, beforeAll, describe, expect, test } from "vitest";
+import { sql } from "../builder/sql";
+import { Database, type Connection } from "../database";
+import { Varchar2 } from "../types/oracle";
+import { OracleDriver } from "./oracle";
+import { requireOraclePoolAttributes } from "./oracle-url";
+
+const enabled = process.env["ORACLE_URL"] !== undefined;
+const db = new Database();
+
+class TransactionRows extends db.Table("oracle_transaction_rows") {
+  id = Varchar2.column({ nonNull: true });
+  value = Varchar2.column({ nonNull: true });
+}
+
+describe.skipIf(!enabled)("Oracle transactions", () => {
+  let conn: Connection;
+
+  beforeAll(async () => {
+    conn = db.connect(await OracleDriver.create(requireOraclePoolAttributes()));
+    try {
+      await conn.execute(sql`DROP TABLE ${db.scopedIdent("oracle_transaction_rows")} PURGE`);
+    } catch {
+      // The table does not exist on the first run.
+    }
+    await conn.execute(sql`
+      CREATE TABLE ${db.scopedIdent("oracle_transaction_rows")} (
+        ${db.scopedIdent("id")} VARCHAR2(36) PRIMARY KEY,
+        ${db.scopedIdent("value")} VARCHAR2(100) NOT NULL
+      )
+    `);
+  });
+
+  afterAll(async () => {
+    await conn.execute(sql`DROP TABLE ${db.scopedIdent("oracle_transaction_rows")} PURGE`);
+    await conn.close();
+  });
+
+  test("commits successful transactions", async () => {
+    const result = await conn.transaction(async (tx) => {
+      await TransactionRows.insert({ id: "commit", value: "visible" }).execute(tx);
+      return "committed";
+    });
+
+    expect(result).toBe("committed");
+    expect(await TransactionRows.from()
+      .where(({ oracle_transaction_rows: row }) => row.id.eq("commit"))
+      .select(({ oracle_transaction_rows: row }) => ({ value: row.value }))
+      .execute()).toEqual([{ value: "visible" }]);
+  });
+
+  test("rolls back failed transactions", async () => {
+    await expect(conn.transaction(async (tx) => {
+      await TransactionRows.insert({ id: "rollback", value: "hidden" }).execute(tx);
+      throw new Error("rollback requested");
+    })).rejects.toThrow("rollback requested");
+
+    expect(await TransactionRows.from()
+      .where(({ oracle_transaction_rows: row }) => row.id.eq("rollback"))
+      .select(({ oracle_transaction_rows: row }) => ({ id: row.id }))
+      .execute()).toEqual([]);
+  });
+
+  test("pins one Oracle session and flattens nested transactions", async () => {
+    await conn.transaction(async (tx) => {
+      const first = await tx.execute(sql`
+        SELECT SYS_CONTEXT('USERENV', 'SID') AS ${db.scopedIdent("sid")} FROM DUAL
+      `);
+      await tx.transaction(async (nested) => {
+        const second = await nested.execute(sql`
+          SELECT SYS_CONTEXT('USERENV', 'SID') AS ${db.scopedIdent("sid")} FROM DUAL
+        `);
+        expect(second.rows[0]?.["sid"]).toBe(first.rows[0]?.["sid"]);
+        await TransactionRows.insert({ id: "nested", value: "committed" }).execute(nested);
+      });
+    });
+
+    expect(await TransactionRows.from()
+      .where(({ oracle_transaction_rows: row }) => row.id.eq("nested"))
+      .select(({ oracle_transaction_rows: row }) => ({ value: row.value }))
+      .execute()).toEqual([{ value: "committed" }]);
+  });
+});

src/drivers/oracle.test.ts

@@ -20,7 +20,7 @@ test("oracle Connection constructs without a live engine", async () => {
   const conn = db.connect({
     dialect: "oracle",
     execute: async () => ({ rows: [{ v: "1" }] }),
-    runInSingleConnection: async () => {
+    runInTransaction: async () => {
       throw new Error("unused");
     },
     close: async () => {},

src/drivers/oracle.ts

@@ -1,8 +1,9 @@
 import type { CompiledSql } from "../builder/sql";
 import type { DialectName } from "../builder/sql";
 import oracledb from "oracledb";
-import type { Driver, ExecuteFn, QueryResult } from "./types";
+import type { Driver, ExecuteFn, QueryResult, TransactionOptions } from "./types";
 import { stripMatchedOuterParens } from "./shared";
+import { runTransaction } from "./transaction";
 
 // node-oracledb adapter (thin mode — no Instant Client). Optional peer,
 // imported statically because this module only loads when the caller
@@ -50,29 +51,35 @@ export class OracleDriver implements Driver {
 
   private constructor(private pool: oracledb.Pool) {}
 
-  async execute({ text, values }: CompiledSql): Promise<QueryResult> {
-    const conn = await this.pool.getConnection();
-    try {
+  private executor(conn: oracledb.Connection, autoCommit: boolean): ExecuteFn {
+    return async ({ text, values }) => {
       const result = await conn.execute(stripMatchedOuterParens(text), oracleBinds(values), {
         outFormat: oracledb.OUT_FORMAT_OBJECT,
-        autoCommit: true,
+        autoCommit,
       });
       return { rows: normalizeRows(result.rows) };
+    };
+  }
+
+  async execute(compiled: CompiledSql): Promise<QueryResult> {
+    const conn = await this.pool.getConnection();
+    try {
+      return await this.executor(conn, true)(compiled);
     } finally {
       await conn.close();
     }
   }
 
-  async runInSingleConnection<T>(cb: (execute: ExecuteFn) => Promise<T>): Promise<T> {
+  async runInTransaction<T>(
+    _opts: TransactionOptions,
+    cb: (execute: ExecuteFn) => Promise<T>,
+  ): Promise<T> {
     const conn = await this.pool.getConnection();
     try {
-      return await cb(async ({ text, values }) => {
-        const result = await conn.execute(stripMatchedOuterParens(text), oracleBinds(values), {
-          outFormat: oracledb.OUT_FORMAT_OBJECT,
-          autoCommit: false,
-        });
-        return { rows: normalizeRows(result.rows) };
-      });
+      return await runTransaction({
+        commit: () => conn.commit(),
+        rollback: () => conn.rollback(),
+      }, () => cb(this.executor(conn, false)));
     } finally {
       await conn.close();
     }

src/drivers/pg.ts

@@ -1,7 +1,8 @@
 import type { CompiledSql } from "../builder/sql";
 import type { DialectName } from "../builder/sql";
 import pgLib from "pg";
-import type { Driver, ExecuteFn, QueryResult } from "./types";
+import type { Driver, ExecuteFn, QueryResult, TransactionOptions } from "./types";
+import { postgresBeginSql, runSqlTransaction } from "./transaction";
 
 // pg adapter — returns raw text strings (no driver-side deserialization).
 // `pg` is an *optional* peer dep (see package.json#peerDependenciesMeta),
@@ -31,10 +32,14 @@ export class PgDriver implements Driver {
     return this.pool.query(text, values as unknown[]);
   }
 
-  async runInSingleConnection<T>(cb: (execute: ExecuteFn) => Promise<T>): Promise<T> {
+  async runInTransaction<T>(
+    opts: TransactionOptions,
+    cb: (execute: ExecuteFn) => Promise<T>,
+  ): Promise<T> {
     const client = await this.pool.connect();
+    const execute: ExecuteFn = ({ text, values }) => client.query(text, values as unknown[]);
     try {
-      return await cb(({ text, values }) => client.query(text, values as unknown[]));
+      return await runSqlTransaction(execute, postgresBeginSql(opts), () => cb(execute));
     } finally {
       client.release();
     }

src/drivers/pglite.ts

@@ -1,7 +1,8 @@
 import { PGlite } from "@electric-sql/pglite";
 import type { CompiledSql } from "../builder/sql";
 import type { DialectName } from "../builder/sql";
-import type { Driver, ExecuteFn, QueryResult } from "./types";
+import type { Driver, ExecuteFn, QueryResult, TransactionOptions } from "./types";
+import { postgresBeginSql, runSqlTransaction } from "./transaction";
 
 // pglite adapter — returns raw text strings (no driver-side deserialization).
 // `@electric-sql/pglite` is an optional peer, imported statically because
@@ -35,8 +36,12 @@ export class PgliteDriver implements Driver {
     return this.db.query(text, values as unknown[], { parsers: this.parsers }) as Promise<QueryResult>;
   }
 
-  async runInSingleConnection<T>(cb: (execute: ExecuteFn) => Promise<T>): Promise<T> {
-    return cb(this.execute.bind(this));
+  async runInTransaction<T>(
+    opts: TransactionOptions,
+    cb: (execute: ExecuteFn) => Promise<T>,
+  ): Promise<T> {
+    const execute = this.execute.bind(this);
+    return runSqlTransaction(execute, postgresBeginSql(opts), () => cb(execute));
   }
 
   async close(): Promise<void> {