Transactions & Isolation Levels

import { Prorm } from 'ts-prorm-orm';

const prorm = new Prorm({
  dialect: 'postgres',
  host: 'localhost',
  database: 'mydb',
  username: 'postgres',
  password: 'password',
});

await prorm.transaction(async (t) => {
  const from = await Account.findOne({ where: { id: 1 }, transaction: t });
  const to = await Account.findOne({ where: { id: 2 }, transaction: t });

  await from.decrement('balance', { by: 100, transaction: t });
  await to.increment('balance', { by: 100, transaction: t });
  // Reaching the end of the callback commits both updates.
});

const placed = await prorm.transaction(async (t) => {
  const created = await Order.create({ userId: 1, total: 4200 }, { transaction: t });
  await OrderItem.bulkCreate(items, { transaction: t });
  return created; // committed, then returned to the caller
});

try {
  await prorm.transaction(async (t) => {
    await Account.update({ balance: 0 }, { where: { id: 1 }, transaction: t });
    throw new Error('Business rule violated');
  });
} catch (err) {
  // The UPDATE above was rolled back; the balance is unchanged.
  console.error(err);
}

await prorm.transaction(async (t) => {
  // Model methods
  const ada = await User.create({ name: 'Ada' }, { transaction: t });
  const active = await User.findAll({ where: { active: true }, transaction: t });
  await ada.increment('loginCount', { by: 1, transaction: t });

  // Raw SQL
  await prorm.query('UPDATE settings SET value = $1 WHERE key = $2', {
    replacements: ['on', 'feature_x'],
    transaction: t,
  });
});

import { Prorm } from 'ts-prorm-orm';

const prorm = new Prorm({ dialect: 'postgres', database: 'shop', transactionContext: true });

await prorm.transaction(async () => {
  await Order.create({ userId: 1, total: 4200 });        // in the transaction
  await OrderItem.bulkCreate(items);                       // in the transaction
  await AuditLog.create({ action: 'order' }, { transaction: null }); // outside it
});

async function reserveStock(productId: number, quantity: number) {
  // Runs in whatever transaction the caller opened, or none.
  await Product.decrement('stock', { by: quantity, where: { id: productId } });
}

async function recordOrder(userId: number, productId: number, quantity: number) {
  const order = await Order.create({ userId, total: 0 });
  await OrderItem.create({ orderId: order.id, productId, quantity });
  return order;
}

// The route handler owns the transaction boundary.
async function checkout(req: { userId: number; productId: number; quantity: number }) {
  return prorm.transaction(async () => {
    await reserveStock(req.productId, req.quantity);
    return recordOrder(req.userId, req.productId, req.quantity);
  });
}

await prorm.transaction(
  async (t) => {
    await SalesReport.findAll({ transaction: t });
  },
  { isolationLevel: 'SERIALIZABLE' },
);

import { IsolationLevel } from 'ts-prorm-orm';

await prorm.transaction(
  async (t) => {
    await SalesReport.findAll({ transaction: t });
  },
  { isolationLevel: IsolationLevel.Serializable },
);

await prorm.transaction(
  async (t) => {
    /* ... */
  },
  {
    isolationLevel: 'SERIALIZABLE',
    type: 'IMMEDIATE', // SQLite BEGIN type: 'DEFERRED' | 'IMMEDIATE' | 'EXCLUSIVE'
    autocommit: false,
    lock: 'UPDATE', // apply row locking to reads in the transaction
    readOnly: true,
    deferrable: 'INITIALLY DEFERRED',
    retry: { max: 3 },
    requireAtomic: true,
  },
);

mysql has no DEFERRABLE transaction mode; drop `deferrable`.
Transaction type 'IMMEDIATE' is SQLite's BEGIN mode; postgres has no equivalent.
d1 cannot make a transaction atomic: every statement runs as its own request and is applied immediately, and a rollback cannot undo it. Use the dialect's batch() for an all-or-nothing unit, or drop requireAtomic to run best-effort.

PostgreSQL: DEFERRABLE applies only to a SERIALIZABLE READ ONLY transaction; pass { isolationLevel: 'SERIALIZABLE', readOnly: true, deferrable: true }.

sqlite cannot start a read-only transaction here; drop `readOnly` or use a read-only connection.

SELECT * FROM "accounts" WHERE "id" = ? LIMIT 1
UPDATE "accounts" SET "balance" = 400 WHERE "id" = 1
UPDATE "accounts" SET "balance" = "balance" + 100 WHERE "id" = 2
INSERT INTO "transfers" ("fromId", "toId", "amount") VALUES (?, ?, ?)

import { Op } from 'ts-prorm-orm';

await prorm.transaction(async (t) => {
  const products = await Product.findAll({
    where: { id: { [Op.in]: [1, 2] } },
    lock: true,
    transaction: t,
  });
  // Nobody else can change these rows until this callback returns.
});

async function claimJobs(workerId: string) {
  return prorm.transaction(async (t) => {
    const jobs = await Job.findAll({
      where: { status: 'queued' },
      order: [['id', 'ASC']],
      limit: 10,
      lock: { level: 'UPDATE', skipLocked: true },
      transaction: t,
    });
    for (const job of jobs) {
      await job.update({ status: 'running', workerId }, { transaction: t });
    }
    return jobs;
  });
}

await prorm.transaction(
  async (t) => {
    const account = await Account.findOne({ where: { id: 1 }, transaction: t, lock: true });
    await account.increment('balance', { by: 100, transaction: t });
  },
  { isolationLevel: 'SERIALIZABLE', retry: { max: 3 } },
);

import { Op, TransactionError } from 'ts-prorm-orm';

type CartLine = { productId: number; quantity: number };

async function placeOrder(userId: number, cart: CartLine[]) {
  const order = await prorm.transaction(
    async (t) => {
      // 1. Lock the rows whose values the writes depend on.
      const products = await Product.findAll({
        where: { id: { [Op.in]: cart.map((l) => l.productId) } },
        lock: true,
        transaction: t,
      });

      // 2. Check, then write. A throw here rolls everything back.
      let total = 0;
      for (const line of cart) {
        const product = products.find((p) => p.id === line.productId);
        if (!product || product.stock < line.quantity) {
          throw new Error(`Out of stock: product ${line.productId}`);
        }
        await product.decrement('stock', { by: line.quantity, transaction: t });
        total += product.price * line.quantity;
      }

      const created = await Order.create({ userId, total }, { transaction: t });
      await OrderItem.bulkCreate(
        cart.map((l) => ({ orderId: created.id, ...l })),
        { transaction: t },
      );

      // 3. Optional work in a savepoint: losing it must not lose the order.
      try {
        await prorm.transaction(
          async (sp) => {
            await LoyaltyPoints.create(
              { userId, points: Math.floor(total / 100) },
              { transaction: sp },
            );
          },
          { transaction: t },
        );
      } catch (err) {
        if (err instanceof TransactionError) throw err; // misuse, not a business failure
        // Points can be re-credited later; the order stands.
      }

      return created;
    },
    // 4. Two buyers racing for the last item is normal; retry the loser.
    { isolationLevel: 'SERIALIZABLE', retry: { max: 3 } },
  );

  // 5. Side effects after COMMIT, exactly once.
  await mailer.send({ to: userId, template: 'order-confirmed', orderId: order.id });
  return order;
}

const t = await prorm.transaction({ isolationLevel: 'READ COMMITTED' });

try {
  await Account.update({ balance: 500 }, { where: { id: 1 }, transaction: t });
  await Account.update({ balance: 500 }, { where: { id: 2 }, transaction: t });
  await t.commit();
} catch (err) {
  await t.rollback();
  throw err;
}

Invalid savepoint name "x; DROP TABLE t": a savepoint name must match /^[A-Za-z_][A-Za-z0-9_]*$/ (letters, digits and underscores, not starting with a digit).

Nested transactions need savepoints, and snowflake has none. Run the work in a single transaction, or restructure it so nothing nests.

await prorm.transaction(async (t) => {
  await User.create({ name: 'Ada' }, { transaction: t });

  const sp = await t.createSavepoint();
  try {
    await AuditLog.create({ action: 'risky' }, { transaction: t });
    await sp.commit(); // RELEASE SAVEPOINT — keep the audit row
  } catch (err) {
    await sp.rollback(); // ROLLBACK TO SAVEPOINT — undo just the audit row
    // The User.create above is still intact and will commit with t.
  }
});

await prorm.transaction(async (t) => {
  const sp = await t.createSavepoint();
  await Item.create({ sku: 'A-1' }, { transaction: t });

  // Undo everything since `sp` was created, but keep the transaction open.
  await t.rollback(sp);

  await Item.create({ sku: 'B-2' }, { transaction: t }); // still works
});

// Callback mode: catch around the whole call.
try {
  await prorm.transaction(async (t) => {
    await transferFunds(t);
  });
} catch (err) {
  logger.error('Transfer failed and was rolled back', err);
}

const dialect = prorm.getDialectInstance();
const caps = dialect?.transactionCapabilities?.();
if (caps && !caps.savepoints) {
  // nest nothing on this engine
}