MongoDB多集合数据采样：获取符合条件的单条最新聊天消息

ChatsUsers 集合用于控制消息过滤（比如用户清空自身聊天记录时仅对该用户生效），原理是通过指定 start_message_id（当前时间戳），返回 message 集合中 create_date 大于等于该值的消息。

相关数据结构：

ChatsUsers 结构 #

type ChatsUsers struct {
    ID             string `json:"id" bson:"id"`
    ChatID         string `json:"chat_id" bson:"chat_id"`
    UserID         string `json:"user_id" bson:"user_id"`
    StartMessageID int64  `json:"start_message_id,omitempty" bson:"start_message_id"`
    EndMessageID   int64  `json:"end_message_id,omitempty" bson:"end_message_id"`
}

Message 结构 #

type Message struct {
    ID         string `json:"id" bson:"id"`
    ChatID     string `json:"chat_id" bson:"chat_id"`
    FromID     string `json:"from_id" bson:"from_id"`
    CreateDate int64  `json:"create_date" bson:"create_date"`
    Body       string `json:"body" bson:"body"`
    UpdateAt   int64  `json:"update_at" bson:"update_at"`
    // 其他字段
}

Chat 结构 #

type Chat struct {
    ID           string   `json:"id" bson:"id"`
    Participants []string `json:"participants" bson:"participants"`
    LastMessage  *Message `json:"last_message,omitempty" bson:"last_message"`
    // 其他字段
}

需求：获取每个Chat中，last_message.create_date ≥ chatsusers.start_message_id 的单条最新消息，但现有聚合查询无法正确筛选单条消息，且$gte条件不生效。

可以通过以下聚合管道实现需求，核心是关联ChatsUsers过滤规则，再按Chat筛选出符合条件的最新消息：

db.chat.aggregate([
  // 1. 关联当前用户的ChatsUsers过滤规则，同时筛选符合时间条件的记录
  {
    $lookup: {
      from: "chatsusers",
      localField: "id",
      foreignField: "chat_id",
      let: { chatLastMsgDate: "$last_message.create_date" },
      pipeline: [
        // 替换为目标用户ID
        { $match: { user_id: "target_user_id" } },
        // 过滤满足last_message.create_date >= start_message_id的规则
        { $match: { $expr: { $gte: ["$$chatLastMsgDate", "$start_message_id"] } } }
      ],
      as: "valid_rules"
    }
  },
  // 2. 只保留有有效过滤规则的Chat（即满足时间条件的Chat）
  { $match: { valid_rules: { $ne: [] } } },
  // 3. 提取基础字段，暂用Chat的last_message作为候选
  {
    $project: {
      _id: 0,
      chat_id: "$id",
      participants: "$participants",
      candidate_msg: "$last_message",
      start_msg_id: { $arrayElemAt: ["$valid_rules.start_message_id", 0] }
    }
  },
  // 4. （可选）如果Chat的last_message不保证实时更新，直接从Message集合取最新符合条件的消息
  {
    $lookup: {
      from: "message",
      localField: "chat_id",
      foreignField: "chat_id",
      let: { start_id: "$start_msg_id" },
      pipeline: [
        { $match: { $expr: { $gte: ["$create_date", "$$start_id"] } } },
        { $sort: { create_date: -1 } },
        { $limit: 1 }
      ],
      as: "latest_valid_msg"
    }
  },
  // 5. （可选）替换为实际最新消息
  {
    $project: {
      chat_id: 1,
      participants: 1,
      latest_valid_message: { $ifNull: [{ $arrayElemAt: ["$latest_valid_msg", 0] }, "$candidate_msg"] }
    }
  }
])

关键说明 #

• 关联过滤规则：通过$lookup关联用户对应的ChatsUsers记录，并用$expr直接在关联管道内判断时间条件，提前过滤无效规则。
• 筛选有效Chat：剔除没有匹配到有效规则的Chat，确保后续只处理符合条件的记录。
• 可选的消息校验：如果你的业务中Chat的last_message字段不能保证是实时最新的，步骤4和5会直接从Message集合按create_date倒序取第一条符合条件的消息，确保结果准确。如果last_message始终是最新的，可以跳过这两步。

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