MongoDB: معالجة المعاملات: ACID واتساق المستندات المتعددة

// ❌ Counterexample: Transfer without a transaction
async function transfer(fromUserId, toUserId, amount) {
  await User.updateOne({ _id: fromUserId }, { $inc: { balance: -amount } });
  // System Crash!
  await User.updateOne({ _id: toUserId }, { $inc: { balance: amount } });
  // The user's balance was deducted, but the recipient did not receive the payment
}

// === MongoDB 4.0+ Multi-document transactions ===
const session = db.getMongo().startSession();

session.startTransaction();
try {
  // 1. Deduct
  db.users.updateOne(
    { _id: fromUserId },
    { $inc: { balance: -amount } },
    { session }
  );

  // 2. Add
  db.users.updateOne(
    { _id: toUserId },
    { $inc: { balance: amount } },
    { session }
  );

  // 3. Commit Transaction
  await session.commitTransaction();
} catch (err) {
  // 4. Rollback
  await session.abortTransaction();
  throw err;
} finally {
  session.endSession();
}

session.startTransaction({
  writeConcern: {
    w: 'majority',         // Confirmed by a majority of nodes
    j: true,               // Write to disk journal
    wtimeout: 5000         // 5 Timeout in seconds
  }
});

session.startTransaction({
  readConcern: {
    level: 'majority'      // Read Submitted Data
  },
  writeConcern: { w: 'majority' }
});

session.startTransaction({
  readPreference: 'primary'             // Read-Only Primary Node
});

session.startTransaction({
  readPreference: 'secondary'           // Read from a child node
});

session.startTransaction({
  readPreference: 'secondaryPreferred'  // Priority Node
});

// === mongoose Transaction Encapsulation ===
async function withTransaction(callback) {
  const session = await mongoose.startSession();
  session.startTransaction();
  try {
    const result = await callback(session);
    await session.commitTransaction();
    return result;
  } catch (err) {
    await session.abortTransaction();
    throw err;
  } finally {
    session.endSession();
  }
}

// === Usage: Transfer ===
async function transfer(fromUserId, toUserId, amount) {
  return withTransaction(async (session) => {
    const fromUser = await User.findById(fromUserId).session(session);
    if (fromUser.balance < amount) {
      throw new Error('Insufficient balance');
    }

    await User.updateOne(
      { _id: fromUserId },
      { $inc: { balance: -amount } },
      { session }
    );

    await User.updateOne(
      { _id: toUserId },
      { $inc: { balance: amount } },
      { session }
    );

    await TransactionLog.create([{
      fromUserId,
      toUserId,
      amount,
      createdAt: new Date()
    }], { session });

    return { success: true };
  });
}

// Alice's ShopHub Financial System: Transaction encountered WriteConflict automatic retry
async function withRetryTransaction(callback, maxRetries = 3) {
  let lastError;
  for (let i = 0; i < maxRetries; i++) {
    const session = await mongoose.startSession();
    session.startTransaction({
      readConcern: { level: 'snapshot' },
      writeConcern: { w: 'majority' }
    });
    try {
      const result = await callback(session);
      await session.commitTransaction();
      return result;
    } catch (err) {
      await session.abortTransaction();
      lastError = err;
      if (err.errorLabels && err.errorLabels.includes('TransientTransactionError')) {
        console.log(`Retry ${i + 1}/${maxRetries} due to WriteConflict`);
        continue;
      }
      throw err;
    } finally {
      session.endSession();
    }
  }
  throw lastError;
}

// Usage
await withRetryTransaction(async (session) => {
  await User.updateOne({ _id: fromId }, { $inc: { balance: -100 } }, { session });
  await User.updateOne({ _id: toId }, { $inc: { balance: 100 } }, { session });
});

// === Causal Consistency:Ensure the Correct Order of Operations ===
const session = db.getMongo().startSession();
session.startTransaction({
  readConcern: { level: 'majority' },
  writeConcern: { w: 'majority' }
});

// Operation 1:Read the current balance
const account = db.accounts.findOne({ userId: 'user_001' }, { session });
// Operation 2:Write Based on Read Results
db.accounts.updateOne(
  { userId: 'user_001' },
  { $set: { balance: account.balance - 100 } },
  { session }
);
// Guarantee:Operation 2 What you see is the operation 1 Subsequent Status

// Scene: Order placement process (Order + Inventory Deduction + Wallet Deduction + Logging), Fully Atomic
// Introduction: A replica set is required. Transactions must be in progress

// Initialize Data
db.products.insertOne({ sku: 'PHONE-001', stock: 10, price: 599 });
db.users.insertOne({ _id: 'user_001', balance: 1000 });
db.transaction_logs.createIndex({ userId: 1, createdAt: -1 });

// Complete Transaction Functions
async function placeOrder(userId, items) {
  const session = db.getMongo().startSession();
  session.startTransaction({
    readConcern: { level: 'snapshot' },
    writeConcern: { w: 'majority' }
  });

  try {
    // 1. Calculate the total amount + Check Inventory (Atomic Read)
    let total = 0;
    for (const item of items) {
      const product = db.products.findOne(
        { sku: item.sku, stock: { $gte: item.qty } },
        { session }
      );
      if (!product) {
        throw new Error(`Out of Stock: ${item.sku}`);
      }
      total += product.price * item.qty;
    }

    // 2. Check User Balance
    const user = db.users.findOne({ _id: userId }, { session });
    if (user.balance < total) {
      throw new Error('Insufficient balance');
    }

    // 3. Inventory Deduction (Conditional, Preventing Overselling)
    for (const item of items) {
      const result = db.products.updateOne(
        { sku: item.sku, stock: { $gte: item.qty } },
        { $inc: { stock: -item.qty } },
        { session }
      );
      if (result.modifiedCount === 0) {
        throw new Error(`Inventory deduction failed: ${item.sku}`);
      }
    }

    // 4. Deduct from the user's balance
    db.users.updateOne(
      { _id: userId, balance: { $gte: total } },
      { $inc: { balance: -total } },
      { session }
    );

    // 5. Create an Order
    const orderResult = db.orders.insertOne({
      userId,
      items,
      total,
      status: 'paid',
      createdAt: new Date()
    }, { session });

    // 6. Record the transaction log
    db.transaction_logs.insertOne({
      userId,
      orderId: orderResult.insertedId,
      amount: total,
      type: 'purchase',
      createdAt: new Date()
    }, { session });

    // 7. Commit Transaction
    session.commitTransaction();
    return { success: true, orderId: orderResult.insertedId };

  } catch (err) {
    // Any failure → Roll Back All
    session.abortTransaction();
    return { success: false, error: err.message };
  } finally {
    session.endSession();
  }
}

// Execute: Place an Order
placeOrder('user_001', [
  { sku: 'PHONE-001', qty: 1 }
]);

// Testing Rollback Scenarios: Deliberately Creating Errors
placeOrder('user_001', [
  { sku: 'NONEXIST', qty: 1 }  // The product does not exist.
]);
// Throw an exception → Transaction Rollback → Inventory, balance, order, all logs remain unchanged

// Verifying Atomicity:
// db.products.findOne({ sku: 'PHONE-001' }) → stock: 10 (Not deducted)
// db.users.findOne({ _id: 'user_001' }) → balance: 1000 (Not deducted)

// Scene:ShopHub E-commerce transaction system with automatic retry
const mongoose = require('mongoose');

// غلاف معاملات مع إعادة المحاولة التلقائية
async function withTransactionRetry(callback, maxRetries = 3) {
  let lastError;
  
  for (let attempt = 1; attempt <= maxRetries; attempt++) {
    const session = await mongoose.startSession();
    session.startTransaction({
      readConcern: { level: 'snapshot' },
      writeConcern: { w: 'majority', j: true }
    });

    try {
      const result = await callback(session);
      await session.commitTransaction();
      return result;
    } catch (err) {
      await session.abortTransaction();
      lastError = err;

      // إعادة المحاولة فقط لأخطاء TransientTransactionError
      if (err.errorLabels && err.errorLabels.includes('TransientTransactionError')) {
        console.log(`إعادة المحاولة ${attempt}/${maxRetries} بسبب تعارض الكتابة`);
        continue;
      }
      throw err;
    } finally {
      session.endSession();
    }
  }
  
  throw lastError;
}

// معاملة طلب كاملة
async function placeOrder(userId, items) {
  return withTransactionRetry(async (session) => {
    let total = 0;

    // 1. التحقق من المخزون وحساب الإجمالي
    for (const item of items) {
      const product = await Product.findOne(
        { _id: item.productId, stock: { $gte: item.quantity } },
        null,
        { session }
      );
      
      if (!product) {
        throw new Error(`المخزون غير كافٍ: ${item.productId}`);
      }
      total += product.price * item.quantity;
    }

    // 2. التحقق من رصيد المستخدم
    const user = await User.findById(userId).session(session);
    if (user.balance < total) {
      throw new Error('الرصيد غير كافٍ');
    }

    // 3. خصم المخزون
    for (const item of items) {
      await Product.updateOne(
        { _id: item.productId, stock: { $gte: item.quantity } },
        { $inc: { stock: -item.quantity } },
        { session }
      );
    }

    // 4. خصم الرصيد
    await User.updateOne(
      { _id: userId },
      { $inc: { balance: -total } },
      { session }
    );

    // 5. إنشاء الطلب
    const order = await Order.create([{
      userId,
      items: items.map(i => ({ productId: i.productId, quantity: i.quantity })),
      total,
      status: 'paid',
      createdAt: new Date()
    }], { session });

    // 6. إنشاء سجل المعاملة
    await TransactionLog.create([{
      userId,
      orderId: order[0]._id,
      amount: total,
      type: 'purchase'
    }], { session });

    return { success: true, orderId: order[0]._id, total };
  });
}

// استخدام النظام
try {
  const result = await placeOrder('user_001', [
    { productId: 'prod_001', quantity: 2 },
    { productId: 'prod_002', quantity: 1 }
  ]);
  console.log('تم الطلب بنجاح:', result);
} catch (err) {
  console.error('فشل الطلب:', err.message);
}