MongoDB: 事务处理：ACID与多文档一致性

// ❌ 反例：转账无事务
async function transfer(fromUserId, toUserId, amount) {
  await User.updateOne({ _id: fromUserId }, { $inc: { balance: -amount } });
  // 系统崩溃！
  await User.updateOne({ _id: toUserId }, { $inc: { balance: amount } });
  // 用户余额扣了但对方没收到
}

// === MongoDB 4.0+ 多文档事务 ===
const session = db.getMongo().startSession();

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

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

  // 3. 提交事务
  await session.commitTransaction();
} catch (err) {
  // 4. 回滚
  await session.abortTransaction();
  throw err;
} finally {
  session.endSession();
}

session.startTransaction({
  writeConcern: {
    w: 'majority',         // 大多数节点确认
    j: true,               // 写入磁盘 journal
    wtimeout: 5000         // 5 秒超时
  }
});

session.startTransaction({
  readConcern: {
    level: 'majority'      // 读已提交的数据
  },
  writeConcern: { w: 'majority' }
});

session.startTransaction({
  readPreference: 'primary'             // 仅读主节点
});

session.startTransaction({
  readPreference: 'secondary'           // 读从节点
});

session.startTransaction({
  readPreference: 'secondaryPreferred'  // 优先从节点
});

// === mongoose 事务封装 ===
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();
  }
}

// === 使用：转账 ===
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 的 ShopHub 金融系统：事务遇到 WriteConflict 自动重试
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;
}

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

// mongoose 操作成功执行
// 数据库查询/更新结果

// === 因果一致性：保证操作顺序 ===
const session = db.getMongo().startSession();
session.startTransaction({
  readConcern: { level: 'majority' },
  writeConcern: { w: 'majority' }
});

// 操作 1：读当前余额
const account = db.accounts.findOne({ userId: 'user_001' }, { session });
// 操作 2：基于读结果写入
db.accounts.updateOne(
  { userId: 'user_001' },
  { $set: { balance: account.balance - 100 } },
  { session }
);
// 保证：操作 2 看到的是操作 1 之后的状态

// 场景：下单流程（订单 + 扣库存 + 钱包扣款 + 日志记录），全部原子性
// 前置：必须有副本集，事务才能工作

// 初始化数据
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 });

// 完整事务函数
async function placeOrder(userId, items) {
  const session = db.getMongo().startSession();
  session.startTransaction({
    readConcern: { level: 'snapshot' },
    writeConcern: { w: 'majority' }
  });

  try {
    // 1. 计算总金额 + 检查库存（原子读取）
    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(`库存不足: ${item.sku}`);
      }
      total += product.price * item.qty;
    }

    // 2. 检查用户余额
    const user = db.users.findOne({ _id: userId }, { session });
    if (user.balance < total) {
      throw new Error('余额不足');
    }

    // 3. 扣库存（带条件，防止超卖）
    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(`库存扣减失败: ${item.sku}`);
      }
    }

    // 4. 扣用户余额
    db.users.updateOne(
      { _id: userId, balance: { $gte: total } },
      { $inc: { balance: -total } },
      { session }
    );

    // 5. 创建订单
    const orderResult = db.orders.insertOne({
      userId,
      items,
      total,
      status: 'paid',
      createdAt: new Date()
    }, { session });

    // 6. 记录事务日志
    db.transaction_logs.insertOne({
      userId,
      orderId: orderResult.insertedId,
      amount: total,
      type: 'purchase',
      createdAt: new Date()
    }, { session });

    // 7. 提交事务
    session.commitTransaction();
    return { success: true, orderId: orderResult.insertedId };

  } catch (err) {
    // 任何失败 → 全部回滚
    session.abortTransaction();
    return { success: false, error: err.message };
  } finally {
    session.endSession();
  }
}

// 执行：下单
placeOrder('user_001', [
  { sku: 'PHONE-001', qty: 1 }
]);

// 测试回滚场景：故意制造错误
placeOrder('user_001', [
  { sku: 'NONEXIST', qty: 1 }  // 商品不存在
]);
// 抛出异常 → 事务回滚 → 库存、余额、订单、日志全部不变

// 验证原子性：
// db.products.findOne({ sku: 'PHONE-001' }) → stock: 10（未扣减）
// db.users.findOne({ _id: 'user_001' }) → balance: 1000（未扣款）

// === 1. 事件集合 Schema ===
const eventSchema = new mongoose.Schema({
  type: { type: String, required: true },
  payload: mongoose.Schema.Types.Mixed,
  status: { type: String, enum: ['pending', 'published', 'failed'], default: 'pending' },
  createdAt: { type: Date, default: Date.now },
  publishedAt: { type: Date, default: null }
});
const Event = mongoose.model('Event', eventSchema);

// === 2. 事务内写入业务数据 + 事件 ===
async function transferBalance(fromId, toId, amount) {
  const session = await mongoose.startSession();
  session.startTransaction();
  try {
    const from = await User.findOneAndUpdate(
      { _id: fromId, balance: { $gte: amount } },
      { $inc: { balance: -amount } },
      { session, new: true }
    );
    if (!from) throw new Error('余额不足');

    const to = await User.findOneAndUpdate(
      { _id: toId },
      { $inc: { balance: amount } },
      { session, new: true }
    );

    // 事务内写入事件（事务提交后事件才可见）
    await Event.create([{
      type: 'balance.transferred',
      payload: { from: fromId, to: toId, amount, fromBalance: from.balance, toBalance: to.balance }
    }], { session });

    await session.commitTransaction();
    return { success: true, from, to };
  } catch (err) {
    await session.abortTransaction();
    return { success: false, error: err.message };
  } finally {
    session.endSession();
  }
}

// === 3. Change Stream 监听事件集合，异步触发副作用 ===
const changeStream = Event.watch([{ $match: { operationType: 'insert' } }]);

changeStream.on('change', async (change) => {
  const event = change.fullDocument;
  try {
    switch (event.type) {
      case 'balance.transferred':
        // 发送转账通知（邮件/短信/Push）
        await sendNotification(event.payload.to, `收到转账 ${event.payload.amount} 元`);
        await sendNotification(event.payload.from, `已转出 ${event.payload.amount} 元`);
        // 更新缓存
        await cacheDel(`user:balance:${event.payload.from}`);
        await cacheDel(`user:balance:${event.payload.to}`);
        break;
    }
    // 标记事件已发布
    await Event.updateOne({ _id: event._id }, { status: 'published', publishedAt: new Date() });
  } catch (err) {
    // 标记事件发布失败，后续可重试
    await Event.updateOne({ _id: event._id }, { status: 'failed' });
    console.error('事件发布失败:', event.type, err.message);
  }
});

// === 4. 失败事件重试机制 ===
async function retryFailedEvents() {
  const failed = await Event.find({ status: 'failed' })
    .sort({ createdAt: 1 })
    .limit(100)
    .lean();
  for (const event of failed) {
    try {
      await publishEvent(event);
      await Event.updateOne({ _id: event._id }, { status: 'published', publishedAt: new Date() });
    } catch (err) {
      console.error(`重试失败: ${event._id}`, err.message);
    }
  }
}
// 定时重试：每 5 分钟扫描一次
setInterval(retryFailedEvents, 5 * 60 * 1000);