MongoDB聚合查询可扩展性优化：重复代码问题求解

针对你遇到的聚合查询重复代码、维护成本高的问题，结合NestJS和Express的使用场景，以下是几种实用的优化方案：

1. 模块化复用聚合阶段 #

将常用的聚合阶段（如$lookup、$match、$project）抽离成独立的函数或常量，需要时直接导入组合，避免重复编写样板代码。

示例代码（TypeScript） #

// src/common/mongo-aggregations.ts
export const lookupUser = (
  localField: string = 'userId',
  foreignField: string = '_id',
  as: string = 'user'
) => ({
  $lookup: {
    from: 'users',
    localField,
    foreignField,
    as,
  },
});

export const matchActiveDocuments = () => ({
  $match: { isActive: true },
});

在业务服务中调用：

// src/posts/posts.service.ts
import { lookupUser, matchActiveDocuments } from '../common/mongo-aggregations';

async getPostsWithAuthor() {
  return this.postModel.aggregate([
    matchActiveDocuments(),
    lookupUser('authorId', '_id', 'author'),
    { $project: { content: 1, author: { name: 1, email: 1 } } },
  ]);
}

async getCommentsWithUser() {
  return this.commentModel.aggregate([
    matchActiveDocuments(),
    lookupUser('commenterId'),
    { $project: { text: 1, user: { name: 1 } } },
  ]);
}

后续修改lookupUser的逻辑时，只需更新一处即可同步所有调用场景。

2. 工厂模式生成聚合管道 #

针对复杂的聚合逻辑，用工厂函数按需组合不同阶段，根据业务参数生成定制化的管道，提升扩展性。

示例代码 #

// src/common/aggregation-factories.ts
import { lookupUser, matchActiveDocuments } from './mongo-aggregations';

export class AggregationFactory {
  static createBasePipeline(includeActive: boolean = true) {
    const pipeline = [];
    if (includeActive) pipeline.push(matchActiveDocuments());
    return pipeline;
  }

  static createPostPipeline(options: { includeAuthor: boolean, includeStats: boolean }) {
    const pipeline = this.createBasePipeline();
    
    if (options.includeAuthor) {
      pipeline.push(lookupUser('authorId', '_id', 'author'));
    }
    if (options.includeStats) {
      pipeline.push({
        $lookup: {
          from: 'post_stats',
          localField: '_id',
          foreignField: 'postId',
          as: 'stats',
        },
        $unwind: '$stats',
      });
    }
    return pipeline;
  }
}

在服务中使用：

async getPostsWithFullDetails() {
  const pipeline = AggregationFactory.createPostPipeline({
    includeAuthor: true,
    includeStats: true,
  });
  return this.postModel.aggregate(pipeline);
}

新增需求（如添加includeTags）时，只需在工厂函数中新增分支逻辑，无需修改所有调用处。

3. NestJS专属：自定义Repository封装聚合逻辑 #

利用NestJS的MongoDB模块特性，将聚合逻辑封装在自定义Repository中，业务服务仅调用封装好的方法，隔离底层查询细节。

示例代码 #

// src/posts/posts.repository.ts
import { Injectable } from '@nestjs/common';
import { InjectModel } from '@nestjs/mongoose';
import { Model } from 'mongoose';
import { Post } from './schemas/post.schema';
import { lookupUser, matchActiveDocuments } from '../common/mongo-aggregations';

@Injectable()
export class PostsRepository {
  constructor(@InjectModel(Post.name) private postModel: Model<Post>) {}

  async findAllWithAuthor() {
    return this.postModel.aggregate([
      matchActiveDocuments(),
      lookupUser('authorId', '_id', 'author'),
    ]);
  }

  async findByIdWithStats(postId: string) {
    return this.postModel.aggregate([
      { $match: { _id: postId } },
      lookupUser('authorId'),
      {
        $lookup: {
          from: 'post_stats',
          localField: '_id',
          foreignField: 'postId',
          as: 'stats',
        },
      },
    ]);
  }
}

在业务服务中注入使用：

// src/posts/posts.service.ts
import { Injectable } from '@nestjs/common';
import { PostsRepository } from './posts.repository';

@Injectable()
export class PostsService {
  constructor(private postsRepository: PostsRepository) {}

  async getPostsWithAuthor() {
    return this.postsRepository.findAllWithAuthor();
  }
}

4. Express场景：工具函数封装聚合逻辑 #

在Express项目中，将常用聚合逻辑封装成工具函数，路由处理函数直接调用，减少重复代码。

示例代码（JavaScript） #

// utils/mongo-utils.js
exports.lookupUser = (localField = 'userId') => ({
  $lookup: {
    from: 'users',
    localField,
    foreignField: '_id',
    as: 'user'
  }
});

exports.buildPostAggregation = (includeUser = true) => {
  const pipeline = [{ $match: { isActive: true } }];
  if (includeUser) pipeline.push(exports.lookupUser('authorId'));
  return pipeline;
};

在路由中使用：

// routes/posts.js
const express = require('express');
const router = express.Router();
const Post = require('../models/Post');
const { buildPostAggregation } = require('../utils/mongo-utils');

router.get('/', async (req, res) => {
  const includeUser = req.query.includeUser === 'true';
  const pipeline = buildPostAggregation(includeUser);
  const posts = await Post.aggregate(pipeline);
  res.json(posts);
});

内容的提问来源于stack exchange，提问作者alvaro canales aguilar