MongoDB：关联productRecord计算总价并写入saleRecords的查询实现

背景信息 #

产品集合结构 #

productRecord 集合存储产品基础信息，文档结构如下：

{
  "_id": "ObjectId(产品唯一ID)",
  "productName": "产品名称",
  "productPrice": 199.99, // 产品单价（数值类型）
  "productQuantity": 100 // 库存数量
}

待处理销售数据 #

收到的单次销售数据格式如下：

var sale = {
  "id": "SALE20240501001", // 销售记录唯一ID
  "products": [
     {"productId": "60d21b4667d0d8992e610c85", "qty": 2},
     {"productId": "60d21b4667d0d8992e610c86", "qty": 1},
     {"productId": "60d21b4667d0d8992e610c87", "qty": 5}
   ]
}

需求 #

需要将上述销售数据写入 saleRecords 集合，最终文档必须包含：

• id：原销售数据的记录ID
• products：销售的产品列表（需关联对应产品的单价，可保留原结构或补充单价）
• totalsaleprice：本次销售的总金额（所有产品「销售数量 × 对应产品单价」的总和）

实现方案 #

方案1：单条销售数据快速处理（聚合+插入） #

适合一次性处理单条销售记录、无需事务保证的场景：

// 提取所有涉及的产品ID（转换为ObjectId匹配集合内的_id类型）
const targetProductIds = sale.products.map(item => ObjectId(item.productId));

// 临时存储当前销售的产品数量信息
db.temp_single_sale.insertMany(sale.products);

// 通过聚合管道关联产品数据并计算总价
const saleDoc = db.productRecord.aggregate([
  // 匹配本次销售涉及的产品
  { $match: { _id: { $in: targetProductIds } } },
  // 关联临时存储的销售产品数量
  { $lookup: {
      from: "temp_single_sale",
      localField: "_id",
      foreignField: "productId",
      as: "saleDetail"
    }
  },
  { $unwind: "$saleDetail" },
  // 计算单个产品的销售金额
  { $addFields: {
      subtotal: { $multiply: ["$productPrice", "$saleDetail.qty"] }
    }
  },
  // 聚合计算总价并整理产品列表
  { $group: {
      _id: null,
      totalsaleprice: { $sum: "$subtotal" },
      products: { $push: {
          productId: "$_id",
          qty: "$saleDetail.qty",
          productPrice: "$productPrice"
        }
      }
    }
  },
  // 构造最终销售记录结构
  { $project: {
      _id: 0,
      id: sale.id,
      products: 1,
      totalsaleprice: 1
    }
  }
]).next(); // 获取聚合结果的第一条文档

// 将结果插入saleRecords集合
if (saleDoc) {
  db.saleRecords.insertOne(saleDoc);
}

// 清理临时集合
db.temp_single_sale.drop();

方案2：带库存扣减的原子性处理（事务） #

如果业务要求销售记录写入与库存扣减必须同时成功/失败，使用MongoDB事务保证数据一致性：

// 开启会话与事务
const session = db.getMongo().startSession();
session.startTransaction();

try {
  const dbName = "your_database_name";
  const saleColl = session.getDatabase(dbName).saleRecords;
  const productColl = session.getDatabase(dbName).productRecord;

  // 批量查询本次销售涉及的产品信息
  const productList = productColl.find({
    _id: { $in: sale.products.map(item => ObjectId(item.productId)) }
  }).toArray();

  // 计算总价并整理销售产品详情
  let totalSalePrice = 0;
  const formattedProducts = sale.products.map(saleItem => {
    const matchedProduct = productList.find(p => p._id.toString() === saleItem.productId);
    const subtotal = matchedProduct.productPrice * saleItem.qty;
    totalSalePrice += subtotal;
    return {
      productId: saleItem.productId,
      qty: saleItem.qty,
      productPrice: matchedProduct.productPrice
    };
  });

  // 插入销售记录
  saleColl.insertOne({
    id: sale.id,
    products: formattedProducts,
    totalsaleprice: totalSalePrice
  });

  // 批量扣减对应产品的库存
  sale.products.forEach(saleItem => {
    productColl.updateOne(
      { _id: ObjectId(saleItem.productId) },
      { $inc: { productQuantity: -saleItem.qty } }
    );
  });

  // 提交事务
  session.commitTransaction();
  print("销售记录写入与库存扣减完成");
} catch (error) {
  // 事务失败回滚
  session.abortTransaction();
  print(`操作失败：${error.message}`);
  throw error;
} finally {
  // 结束会话
  session.endSession();
}

方案3：批量销售数据处理（$merge聚合阶段） #

如果需要批量导入多条销售记录，先将销售数据存入临时集合，再通过聚合关联计算后写入saleRecords：

// 1. 批量插入销售数据到临时集合（示例为单条，批量场景用insertMany）
db.temp_sales.insertOne(sale);

// 2. 聚合计算并合并到saleRecords
db.temp_sales.aggregate([
  // 拆分产品数组为单条文档
  { $unwind: "$products" },
  // 关联productRecord获取产品单价
  { $lookup: {
      from: "productRecord",
      localField: "products.productId",
      foreignField: "_id",
      as: "productInfo"
    }
  },
  { $unwind: "$productInfo" },
  // 补充产品单价并计算单个产品销售金额
  { $addFields: {
      "products.productPrice": "$productInfo.productPrice",
      "products.subtotal": { $multiply: ["$productInfo.productPrice", "$products.qty"] }
    }
  },
  // 按销售记录ID聚合，计算总价并重组产品列表
  { $group: {
      _id: "$id",
      products: { $push: "$products" },
      totalsaleprice: { $sum: "$products.subtotal" }
    }
  },
  // 调整输出结构
  { $project: {
      _id: 0,
      id: "$_id",
      products: 1,
      totalsaleprice: 1
    }
  },
  // 将结果合并到saleRecords集合（存在则替换，不存在则插入）
  { $merge: {
      into: "saleRecords",
      on: "id",
      whenMatched: "replace",
      whenNotMatched: "insert"
    }
  }
]);

// 3. 清理临时集合
db.temp_sales.drop();

内容的提问来源于stack exchange，提问作者Sahil Shrestha