// Flawed Design: Orders and order items are stored in two separate collections, requiring transactions to ensure consistency.
const session = client.startSession();
session.startTransaction();
try {
    await db.t_orders.insertOne(
        { orderId: "ORD001", userId: "U001", total: 299 },
        { session }
    );
    await db.t_order_items.insertMany([
        { orderId: "ORD001", productId: "P001", quantity: 1, price: 199 },
        { orderId: "ORD001", productId: "P002", quantity: 1, price: 100 }
    ], { session });
    await session.commitTransaction();
} catch (err) {
    await session.abortTransaction();
    throw err;
} finally {
    await session.endSession();
}

// Correct Design: Using embedded documents, operations on a single document are inherently atomic and do not require transactions.
await db.t_orders.insertOne({
    orderId: "ORD001",
    userId: "U001",
    total: 299,
    items: [
        { productId: "P001", quantity: 1, price: 199 },
        { productId: "P002", quantity: 1, price: 100 }
    ]
});

from pymongo import MongoClient
from pymongo.read_concern import ReadConcern
from pymongo.write_concern import WriteConcern
import logging

logger = logging.getLogger(__name__)


def transfer_funds(db, session, from_account, to_account, amount):
    "Fund Transfer Business Logic"
    accounts = db.accounts

    # Deduct Funds (CAS Balance Check)
    result = accounts.update_one(
        {"_id": from_account, "balance": {"$gte": amount}},
        {"$inc": {"balance": -amount}},
        session=session
    )
    if result.matched_count != 1:
        # Distinguish Between Non-existent Accounts and Insufficient Balance
        account = accounts.find_one({"_id": from_account}, session=session)
        if not account:
            raise Exception(f"Deduction account does not exist: {from_account}")
        raise Exception(
            f"Insufficient balance: {from_account}, current balance {account['balance']}, required deduction {amount}"
        )

    # Credit Funds (Must Verify Target Account Exists)
    result = accounts.update_one(
        {"_id": to_account},
        {"$inc": {"balance": amount}},
        session=session
    )
    if result.matched_count != 1:
        raise Exception(f"Credit account does not exist: {to_account}")


# Use the with_transaction Callback API (Automatically Retries TransientTransactionError)
client = MongoClient("mongodb://...")
db = client.bank

try:
    with client.start_session() as session:
        session.with_transaction(
            lambda s: transfer_funds(db, s, "A001", "A002", 100),
            read_concern=ReadConcern(level="snapshot"),
            write_concern=WriteConcern(w="majority", j=True)
        )
    logger.info("Transfer successful: A001 -> A002, amount 100")
except Exception as e:
    logger.error(f"Transfer failed: {e}")
    raise

const { MongoClient } = require('mongodb');

async function transferFunds(client, fromAccount, toAccount, amount) {
    const session = client.startSession();

    try {
        // Use the withTransaction Callback API (Automatically Retries TransientTransactionError)
        await session.withTransaction(async () => {
            const accounts = client.db('bank').collection('accounts');

            // Deduct Funds (CAS Balance Check)
            const debitResult = await accounts.updateOne(
                { _id: fromAccount, balance: { $gte: amount } },
                { $inc: { balance: -amount } },
                { session }
            );

            if (debitResult.matchedCount !== 1) {
                // Distinguish Between "Account Does Not Exist" and "Insufficient Balance"
                const account = await accounts.findOne(
                    { _id: fromAccount },
                    { session }
                );
                if (!account) {
                    throw new Error(`Deduction account does not exist: ${fromAccount}`);
                }
                throw new Error(
                    `Insufficient balance: ${fromAccount}, current balance ${account.balance}, required deduction ${amount}`
                );
            }

            // Credit Funds (Must Verify Target Account Exists, Otherwise Funds Will Be Lost)
            const creditResult = await accounts.updateOne(
                { _id: toAccount },
                { $inc: { balance: amount } },
                { session }
            );

            if (creditResult.matchedCount !== 1) {
                throw new Error(`Credit account does not exist: ${toAccount}`);
            }

        }, {
            readConcern: { level: 'snapshot' },
            writeConcern: { w: 'majority', j: true }
        });

        console.log(`Transfer successful: ${fromAccount} -> ${toAccount}, amount ${amount}`);
    } finally {
        await session.endSession();
    }
}

 // Error: A time-consuming external call is included within the transaction.
 session.startTransaction();
 await collection.updateOne({...}, {...}, { session });
 const result = await callExternalAPI();  // External call with uncontrollable latency
 await collection.updateOne({...}, {...}, { session });
 session.commitTransaction();
 
 // Correct: External calls are placed outside the transaction.
 const externalResult = await callExternalAPI();  // Complete the external call first
 
 session.startTransaction();
 await collection.updateOne({...}, {...}, { session });
 await collection.updateOne({...}, {...}, { session });
 session.commitTransaction();

// Optimistic Locking Implementation
async function updateWithOptimisticLock(collection, id, updateFn, maxRetries = 5) {
    for (let i = 0; i < maxRetries; i++) {
        // Read the current document and its version number
        const doc = await collection.findOne({ _id: id });
        if (!doc) {
            throw new Error(`Document does not exist: ${id}`);
        }
        const currentVersion = doc.version || 0;

        // Apply business logic (Business exceptions are thrown directly without retry)
        const updates = updateFn(doc);

        // Conditional update with version number
        const result = await collection.updateOne(
            { _id: id, version: currentVersion },
            {
                $set: updates,
                $inc: { version: 1 }
            }
        );

        // Use matchedCount to determine whether CAS is hit, not modifiedCount.
        if (result.matchedCount === 1) {
            return true;  // Update successful
        }

        // Version conflict, retry after exponential backoff + jitter
        const backoff = Math.random() * Math.pow(2, i) * 10;
        await new Promise(resolve => setTimeout(resolve, backoff));
    }
    throw new Error(`Optimistic locking update failed, maximum retry count exceeded: ${maxRetries}`);
}

// Usage example: Inventory deduction
await updateWithOptimisticLock(
    db.t_products,
    "P001",
    (product) => {
        if (product.stock < 1) throw new Error('Insufficient stock');
        return { stock: product.stock - 1 };
    }
);

 // Default: local (may read data that is not persisted)
 db.t_orders.find({ orderId: "ORD001" })
 
 // Recommended: majority (reads persisted data)
 db.t_orders.find({ orderId: "ORD001" }).readConcern("majority")
 
 // Within a transaction: snapshot (snapshot isolation)
 session.startTransaction({
     readConcern: { level: "snapshot" },
     writeConcern: { w: "majority" }
 });
 
 // Connection string configuration method (globally effective)
 // mongodb://user:pwd@host1,host2,host3/db?readConcernLevel=majority

while (retries < maxRetries) {
    try {
        session.startTransaction();
        // ... Business Logic ...
        session.commitTransaction();
        break;  // Exit successfully
    } catch (error) {
        session.abortTransaction();
        if (error.hasErrorLabel("TransientTransactionError") && retries < maxRetries) {
            retries++;
            continue;  // Retry
        }
        throw error;  // Non-transient error, throw it
    }
}

// 1. View the current active transactions
db.currentOp({ "transaction": { $exists: true } })

// 2. Check the transaction execution time
// Focus on the transaction.timeActiveMicros field
// Investigate the cause for transactions that exceed 1 second.

// 3. Monitor transaction conflicts
// Focus on the transactions metric in serverStatus
db.serverStatus().transactions