如何用JavaScript实现一个支持事务的数据操作层？

IDBTransaction

readonly

readwrite

class IndexedDBService {
    constructor(dbName, version, storeName) {
        this.dbName = dbName;
        this.version = version;
        this.storeName = storeName;
        this.db = null;
    }

    async openDB() {
        return new Promise((resolve, reject) => {
            if (this.db) {
                resolve(this.db);
                return;
            }

            const request = indexedDB.open(this.dbName, this.version);

            request.onupgradeneeded = (event) => {
                const db = event.target.result;
                if (!db.objectStoreNames.contains(this.storeName)) {
                    db.createObjectStore(this.storeName, { keyPath: 'id', autoIncrement: true });
                }
            };

            request.onsuccess = (event) => {
                this.db = event.target.result;
                resolve(this.db);
            };

            request.onerror = (event) => {
                console.error("IndexedDB error:", event.target.errorCode);
                reject(new Error("Failed to open IndexedDB"));
            };
        });
    }

    async executeTransaction(operations, mode = 'readwrite') {
        const db = await this.openDB();
        return new Promise((resolve, reject) => {
            const transaction = db.transaction(this.storeName, mode);
            const store = transaction.objectStore(this.storeName);

            // 执行所有操作
            try {
                operations(store);
            } catch (error) {
                transaction.abort(); // 如果操作代码本身有同步错误，立即中止
                reject(error);
                return;
            }

            transaction.oncomplete = () => {
                resolve("Transaction completed successfully.");
            };

            transaction.onerror = (event) => {
                console.error("Transaction error:", event.target.error);
                reject(event.target.error);
            };

            transaction.onabort = () => {
                console.warn("Transaction aborted.");
                reject(new Error("Transaction aborted."));
            };
        });
    }

    async addData(data) {
        return this.executeTransaction((store) => {
            store.add(data);
        });
    }

    async updateData(data) {
        return this.executeTransaction((store) => {
            store.put(data);
        });
    }

    async deleteData(id) {
        return this.executeTransaction((store) => {
            store.delete(id);
        });
    }

    async getById(id) {
        return this.executeTransaction((store) => {
            const request = store.get(id);
            request.onsuccess = (event) => {
                // 这里需要一种机制将结果传递出去，通常是Promise的resolve
                // 但executeTransaction的operations回调是同步的，所以需要改造一下
                // 为了简化，这里假设getById不需要在同一个事务内做其他操作
            };
        }, 'readonly');
    }
}

// 示例使用：
const service = new IndexedDBService('MyAppData', 1, 'users');

async function performComplexOperation() {
    try {
        await service.executeTransaction((store) => {
            // 假设用户ID为101的用户购买了商品
            // 1. 更新用户余额
            store.put({ id: 101, name: 'Alice', balance: 50 }); // 模拟扣款
            // 2. 记录购买历史
            store.add({ id: Date.now(), userId: 101, item: 'Book', price: 50 });
            // 如果其中任何一步操作失败（比如key重复），整个事务都会回滚
        });
        console.log("Complex operation (purchase) successful!");
    } catch (error) {
        console.error("Complex operation failed:", error.message);
    }
}

// performComplexOperation();

executeTransaction

pg

mysql2

sequelize

prisma

// 假设我们有一个数据库连接池，比如使用'pg'库
// const { Pool } = require('pg');
// const pool = new Pool({ connectionString: '...' });

class DatabaseService {
    constructor(pool) {
        this.pool = pool;
    }

    // 这是一个通用的事务执行器
    async withTransaction(callback) {
        const client = await this.pool.connect(); // 从连接池获取一个客户端
        try {
            await client.query('BEGIN'); // 启动事务
            const result = await callback(client); // 执行业务逻辑，传入客户端
            await client.query('COMMIT'); // 提交事务
            return result;
        } catch (error) {
            await client.query('ROLLBACK'); // 出现错误时回滚
            throw error; // 重新抛出错误，让上层捕获
        } finally {
            client.release(); // 释放客户端回连接池
        }
    }
}

// 假设我们有User和Product的Repository
class UserRepository {
    constructor(dbClient) {
        this.dbClient = dbClient; // 这个client是在事务中传递进来的
    }

    async createUser(name, email) {
        const res = await this.dbClient.query(
            'INSERT INTO users(name, email) VALUES($1, $2) RETURNING *',
            [name, email]
        );
        return res.rows[0];
    }

    async updateUserBalance(userId, amount) {
        const res = await this.dbClient.query(
            'UPDATE users SET balance = balance + $1 WHERE id = $2 RETURNING *',
            [amount, userId]
        );
        if (res.rowCount === 0) throw new Error('User not found or balance update failed.');
        return res.rows[0];
    }
}

class ProductRepository {
    constructor(dbClient) {
        this.dbClient = dbClient;
    }

    async decreaseStock(productId, quantity) {
        const res = await this.dbClient.query(
            'UPDATE products SET stock = stock - $1 WHERE id = $2 AND stock >= $1 RETURNING *',
            [quantity, productId]
        );
        if (res.rowCount === 0) throw new Error('Product not found or insufficient stock.');
        return res.rows[0];
    }
}

// 示例使用：
// const dbService = new DatabaseService(pool);

async function processOrder(userId, productId, quantity) {
    try {
        const orderResult = await dbService.withTransaction(async (client) => {
            const userRepository = new UserRepository(client);
            const productRepository = new ProductRepository(client);

            // 1. 减少商品库存
            const updatedProduct = await productRepository.decreaseStock(productId, quantity);
            console.log(`Product ${updatedProduct.name} stock decreased.`);

            // 2. 更新用户余额（这里简化为增加，实际可能是扣除）
            const updatedUser = await userRepository.updateUserBalance(userId, -updatedProduct.price * quantity); // 假设扣款
            console.log(`User ${updatedUser.name} balance updated.`);

            // 3. 记录订单
            // await client.query('INSERT INTO orders(...) VALUES(...)');
            return { user: updatedUser, product: updatedProduct };
        });
        console.log("Order processed successfully:", orderResult);
    } catch (error) {
        console.error("Order processing failed, transaction rolled back:", error.message);
    }
}

// processOrder(1, 101, 2);

withTransaction

db.transaction()

IDBTransaction

Object Store

add

put

delete

get

transaction.abort()

oncomplete

transaction.abort()

onabort

onerror

readwrite

Object Store

Object Store

readwrite

BEGIN

COMMIT

ROLLBACK

DatabaseService

Repository