如何在Mongoose聚合操作中复用Pipeline阶段？

我在使用Mongoose的Aggregate方法构建查询管道时，遇到了这样的场景：获取照片列表和照片详情时，大部分聚合阶段是相同的，只需要增减少量阶段（比如列表需要sort、limit，详情需要在match里加_id条件且不需要sort/limit）。请问怎么避免代码重复，保证复用性？

原场景代码示例 #

获取照片列表

PhotoModel.aggregate()
          .match({ deleted: false })
          .sort({ createdAt: -1 })
          .limit(10)
          .lookup({
             from: 'users',
             localField: 'userId',
             foreignField: '_id',        
             as: 'createdBy',
          })
          .unwind('$createdBy')      
          .lookup({
             from: 'conversations',
             localField: '_id',
             foreignField: 'conversationId',       
             as: 'conversations',
          })
          .unwind({
             path: '$conversations',
             preserveNullAndEmptyArrays: true,
          })
          .project({
             _id: 0,
             id: '$_id',
             photoDescription: 1,
             photoUrl: 1,
             createdBy: '$createdBy',
             conversations: '$conversations',
             createdAt: 1,
          });

获取照片详情

PhotoModel.aggregate()
          .match({ 
              deleted: false, 
              _id: req.params.id   // 这里新增了_id条件
          })

          // 详情场景不需要sort和limit，所以移除了这两个阶段

          .lookup({
             from: 'users',
             localField: 'userId',
             foreignField: '_id',        
             as: 'createdBy',
          })
          .unwind('$createdBy')      
          .lookup({
             from: 'conversations',
             localField: '_id',
             foreignField: 'conversationId',       
             as: 'conversations',
          })
          .unwind({
             path: '$conversations',
             preserveNullAndEmptyArrays: true,
          })
          .project({
             _id: 0,
             id: '$_id',
             photoDescription: 1,
             photoUrl: 1,
             createdBy: '$createdBy',
             conversations: '$conversations',
             createdAt: 1,
          });

解决方案 #

方法1：抽离公共阶段为数组，动态拼接

把所有重复的聚合阶段（lookup、unwind、project等）抽成独立数组，不同场景只需拼接专属阶段和公共数组即可。

// 定义公共聚合阶段，只维护一次
const commonPhotoStages = [
  {
    $lookup: {
      from: 'users',
      localField: 'userId',
      foreignField: '_id',        
      as: 'createdBy',
    }
  },
  { $unwind: '$createdBy' },
  {
    $lookup: {
      from: 'conversations',
      localField: '_id',
      foreignField: 'conversationId',       
      as: 'conversations',
    }
  },
  {
    $unwind: {
      path: '$conversations',
      preserveNullAndEmptyArrays: true,
    }
  },
  {
    $project: {
      _id: 0,
      id: '$_id',
      photoDescription: 1,
      photoUrl: 1,
      createdBy: '$createdBy',
      conversations: '$conversations',
      createdAt: 1,
    }
  }
];

// 照片列表专属管道
const photoListPipeline = [
  { $match: { deleted: false } },
  { $sort: { createdAt: -1 } },
  { $limit: 10 },
  ...commonPhotoStages // 合并公共阶段
];

// 照片详情专属管道（接受id参数）
const getPhotoDetailPipeline = (photoId) => [
  { $match: { deleted: false, _id: photoId } },
  ...commonPhotoStages // 合并公共阶段
];

// 使用方式
PhotoModel.aggregate(photoListPipeline); // 列表查询
PhotoModel.aggregate(getPhotoDetailPipeline(req.params.id)); // 详情查询

方法2：封装可配置的生成函数

如果需要更灵活的场景适配（比如新增搜索条件、动态调整字段），可以封装一个函数，通过参数控制管道结构。

function buildPhotoPipeline(options = {}) {
  const { match = {}, includeSortLimit = false } = options;
  
  // 合并基础match条件和传入的自定义条件
  const matchStage = {
    $match: {
      deleted: false,
      ...match
    }
  };

  const pipeline = [matchStage];

  // 根据配置添加排序和分页
  if (includeSortLimit) {
    pipeline.push(
      { $sort: { createdAt: -1 } },
      { $limit: 10 }
    );
  }

  // 追加公共阶段
  pipeline.push(...commonPhotoStages);

  return pipeline;
}

// 使用方式
PhotoModel.aggregate(buildPhotoPipeline({ includeSortLimit: true })); // 列表查询
PhotoModel.aggregate(buildPhotoPipeline({ match: { _id: req.params.id } })); // 详情查询

方法3：利用Mongoose链式调用特性

如果习惯链式写法，可以封装一个返回基础聚合实例的函数，不同场景在实例上继续扩展阶段（注意每次调用都要生成新实例，避免相互影响）。

function getBasePhotoAggregate() {
  return PhotoModel.aggregate()
    .lookup({
      from: 'users',
      localField: 'userId',
      foreignField: '_id',        
      as: 'createdBy',
    })
    .unwind('$createdBy')
    .lookup({
      from: 'conversations',
      localField: '_id',
      foreignField: 'conversationId',       
      as: 'conversations',
    })
    .unwind({
      path: '$conversations',
      preserveNullAndEmptyArrays: true,
    })
    .project({
      _id: 0,
      id: '$_id',
      photoDescription: 1,
      photoUrl: 1,
      createdBy: '$createdBy',
      conversations: '$conversations',
      createdAt: 1,
    });
}

// 列表查询
const photoListAggregate = getBasePhotoAggregate()
  .match({ deleted: false })
  .sort({ createdAt: -1 })
  .limit(10);

// 详情查询
const photoDetailAggregate = getBasePhotoAggregate()
  .match({ deleted: false, _id: req.params.id });

// 执行查询
photoListAggregate.exec();
photoDetailAggregate.exec();

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