MongoDB聚合优化：消除循环简化提速供货门店商品统计

现有一个复杂度较高的供货门店统计函数，需要做逻辑简化和执行效率提升。
函数最终输出主供货门店（primaryStores）列表，以及各供货门店向指定门店组供应的商品数量，统计规则为：仅当门店与对应主供货门店存在已生效的签约连接时，才统计对应商品（部分门店可能仅与某一个供货门店签约，未与其他供货门店签约）。

• 传入门店ID列表后，逐门店循环执行查询：
  1. 查询当前门店已签约的所有生效主供货门店，提取供货门店ID
  2. 查询当前门店下的所有商品，提取商品ID、商品与门店的关联记录ID
  3. 执行聚合查询：匹配关联到其他门店的目标商品，关联查询对应门店信息并过滤出已建立签约连接的供货门店，通过投影、分组操作统计各供货门店的商品供应总数、对应商品列表，将每个主门店关联的供货门店数据存入allFeederStores数组
• 逐门店循环查询结束后，需要通过reduce函数合并数组中重复的供货门店记录、拼接商品统计值，才能得到所有主门店的商品供货来源总览
• 核心待解决问题：移除开头的for循环，避免逐门店单独查询，实现批量校验两个门店间是否存在签约连接的逻辑

async getFeederStores(storeIDs): Promise<any[]> {

  let allFeederStores = [];
  for (let storeID of storeIDs) {
    // 查询当前门店所有已签约的主供货门店连接
    const linkedPrimaryStores = await this.linkedStoreModel.aggregate([
      {
        $match: {
          main_store: storeID,
          primary_store_signed_by: { $ne: null },
        },
      },
    ]);

    // 提取主供货门店ID列表
    const primaryStoreIDs = linkedPrimaryStores.map((linkedPrimaryStore) => {
      return linkedPrimaryStore.primary_store;
    });

    // 查询当前门店下的所有商品
    const products = await this.productStoreModel
      .find({
        store: storeID,
      })
      .select("product")
      .lean();

    // 提取商品ID列表
    const productIDs = products.map((product) => {
      return product.product;
    });

    // 提取商品-门店关联记录ID列表
    const productStoreIDs = products.map((product) => {
      return product._id;
    });

    const linkedStores = await this.productStoreModel.aggregate([
      {
        // 匹配其他门店关联的同商品记录
        $match: {
          product: {
            $in: productIDs,
          },
          _id: {
            $nin: productStoreIDs,
          },
        },
      },
      // 关联查询门店信息
      {
        $lookup: {
          from: "stores",
          localField: "store",
          foreignField: "_id",
          as: "store",
        },
      },
      // 展开门店信息数组
      {
        $unwind: "$store",
      },
      // 过滤出和当前门店有签约连接的主供货门店
      {
        $match: {
          "store._id": { $in: primaryStoreIDs },
        },
      },
      // 投影提取需要的字段
      {
        $project: {
          "store._id": 1,
          "store.name": 1,
          product_store: "$_id",
        },
      },
      // 按供货门店分组统计商品总数和商品列表
      {
        $group: {
          _id: "$store._id",
          name: { $first: "$store.name" },
          total_products: { $sum: 1 },
          products: {
            $push: {
              product_store: "$product_store ",
            },
          },
        },
      },
    ]);
    for (let primaryStore of linkedPrimaryStores) {
      allFeederStores.push(primaryStore);
    }
  }

  // 合并重复的供货门店记录，拼接统计值
  let concatenatedPrimaryStores = allFeederStores.reduce((accum, cv) => {
    const index = accum.findIndex((item) => item._id === cv._id);
    if (index === -1) {
      accum.push(cv);
    } else {
      accum[index].total_products += ", " + cv.total_products;
      accum[index].products += ", " + cv.products;
    }
    return accum;
  }, []);

  return concatenatedPrimaryStores;
}

样例规则说明：MainStore1仅与PrimaryStore1存在签约连接，与PrimaryStore2未签约，因此该门店来自PrimaryStore2的商品不计入统计；MainStore2与PrimaryStore1、PrimaryStore2均存在签约连接，因此两个供货门店的商品均计入统计。

db={
  "storesModel": [
    {
      "_id": "Main1",
      "name": "Main Store 1",
    },
    {
      "_id": "Main2",
      "name": "Main Store 2",
    },
    {
      "_id": "Primary1",
      "name": "Primary Store 1",
    },
    {
      "_id": "Primary2",
      "name": "Primary Store 2",
    },
  ],
  "linkedStoreModel": [
    {
      "_id": "LS1",
      "main_store": "Main1",
      "primary_store_signed_by": "Bob",
      "primary_store": "Primary1"
    },
        {
      "_id": "LS2",
      "main_store": "Main2",
      "primary_store_signed_by": "Bill",
      "primary_store": "Primary1"
    },
    {
      "_id": "LS3",
      "main_store": "Main1",
      "primary_store_signed_by": null,
      "primary_store": "Primary2"
    },
    {
      "_id": "LS4",
      "main_store": "Main2",
      "primary_store_signed_by": "Betty",
      "primary_store": "Primary2"
    }
  ],
  "productStoreModel": [
    {
      "_id": "PS1",
      "store": "Main1",
      "product": "Product1"
    },
    {
      "_id": "PS2",
      "store": "Main2",
      "product": "Product1"
    },
   {
      "_id": "PS3",
      "store": "Main1",
      "product": "Primary2"
    },
    {
      "_id": "PS4",
      "store": "Main2",
      "product": "Primary2"
    },
  ]
}

concatenatedPrimaryStores:
[
  {  
    primaryStoreName: "Primary1",
    total_products: "2",
    products: [/* 对应商品数组 */]
  },
  {  
    primaryStoreName: "Primary2",
    total_products: "1",
    products: [/* 对应商品数组 */]
  }
]

核心思路是把逐门店循环的逻辑全部下沉到MongoDB聚合管道中，通过$lookup关联多集合数据，在数据库层面完成签约关系校验、商品匹配、分组统计，全程只需要1次数据库查询，完全去掉应用层的循环和reduce合并逻辑。

优化后的实现代码：

async getFeederStores(storeIDs): Promise<any[]> {
  return this.productStoreModel.aggregate([
    // 1. 先筛选出传入门店列表对应的所有商品关联记录
    {
      $match: {
        store: { $in: storeIDs }
      }
    },
    // 2. 按商品ID分组，收集每个商品关联的所有门店、记录ID
    {
      $group: {
        _id: "$product",
        mainStoreRecords: {
          $push: {
            storeId: "$store",
            recordId: "$_id"
          }
        }
      }
    },
    // 3. 关联查询同一个商品在其他门店的关联记录（排除当前传入门店组自身的记录）
    {
      $lookup: {
        from: "productStoreModel",
        let: { productId: "$_id", excludeRecords: "$mainStoreRecords.recordId" },
        pipeline: [
          {
            $match: {
              $expr: {
                $and: [
                  { $eq: ["$product", "$$productId"] },
                  { $not: { $in: ["$_id", "$$excludeRecords"] } }
                ]
              }
            }
          }
        ],
        as: "feederStoreRecords"
      }
    },
    // 过滤掉没有其他门店供货的商品
    { $match: { "feederStoreRecords.0": { $exists: true } } },
    // 4. 展开供货门店记录，同时关联对应的主门店信息
    { $unwind: "$feederStoreRecords" },
    { $unwind: "$mainStoreRecords" },
    // 5. 关联查询主门店和供货门店之间的生效签约关系
    {
      $lookup: {
        from: "linkedStoreModel",
        let: { 
          mainStoreId: "$mainStoreRecords.storeId", 
          feederStoreId: "$feederStoreRecords.store" 
        },
        pipeline: [
          {
            $match: {
              $expr: {
                $and: [
                  { $eq: ["$main_store", "$$mainStoreId"] },
                  { $eq: ["$primary_store", "$$feederStoreId"] },
                  { $ne: ["$primary_store_signed_by", null] }
                ]
              }
            }
          }
        ],
        as: "validSignConnection"
      }
    },
    // 只保留存在生效签约关系的记录
    { $match: { "validSignConnection.0": { $exists: true } } },
    // 6. 关联查询供货门店的基础信息
    {
      $lookup: {
        from: "storesModel",
        localField: "feederStoreRecords.store",
        foreignField: "_id",
        as: "feederStoreInfo"
      }
    },
    { $unwind: "$feederStoreInfo" },
    // 7. 按供货门店分组，统计商品总数和商品列表
    {
      $group: {
        _id: "$feederStoreInfo._id",
        primaryStoreName: { $first: "$feederStoreInfo.name" },
        total_products: { $sum: 1 },
        products: {
          $push: {
            product_store: "$feederStoreRecords._id"
          }
        }
      }
    },
    // 去掉不需要的_id字段，格式化输出
    {
      $project: {
        _id: 0,
        primaryStoreName: 1,
        total_products: { $toString: "$total_products" },
        products: 1
      }
    }
  ])
}

优化点说明：

• 数据库查询次数从原来的3*N+1次（N为传入门店数量）降到1次，大幅减少网络IO开销
• 去掉应用层的for循环和reduce合并逻辑，所有数据计算、过滤、统计都在MongoDB引擎层面完成，执行效率更高
• 签约关系校验通过$lookup的自定义管道实现，批量完成所有门店对的签约状态校验，不需要逐门店查询
• 修复了原有代码中total_products和products用字符串拼接的bug，直接输出正确的数值和数组结构

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