如何在MongoDB 4.0中通过聚合实现线程消息分组与未读统计？

问题说明 #

需要针对聊天应用的message_collection集合编写聚合查询，实现两个核心需求：

• 统计每个会话线程的最新消息发送时间戳
• 判断该线程是否存在未读消息（未读定义：消息的statuses为空数组，或数组元素没有readInMilli字段）

示例集合数据 #

[
    {
      "thread" : "war conference",
      "message" : "stop",
      "sentInMilli": 1696858510000, // war conference最早消息
      "statuses": [ {"readInMilli": 1696858547857} ]
    },
    {
      "thread" : "economic conference",
      "message" : "welcome on economic conference",
      "sentInMilli": 1696858520000, // economic conference最新消息
      "statuses": [ {"readInMilli": 1696858547857} ]
    },
    {
      "thread" : "war conference",
      "message" : "the war",
      "sentInMilli": 1696858530000, // war conference最新消息
      "statuses": [ ]
    }
]

期望聚合结果 #

[
  {
    "thread": "war conference",
    "timeInMilliOfNewestMessage": 1696858530000,
    "containsUnreadMessages": true
  },
  {
    "thread": "economic conference",
    "timeInMilliOfNewestMessage": 1696858520000,
    "containsUnreadMessages": false
  }
]

当前错误查询及报错 #

原尝试的聚合查询误用了语法，导致报错：

db.getCollection('message_collection').aggregate(
    [
        {
            "$match": { } // 自定义过滤条件
        },
        {
            "$group": {
                "_id": "thread", 
                "timeInMilliOfNewestMessage": {"$max": "$sentInMilli"},
                "containsUnreadMessages": { 
                    "$count": {
                        "statuses": {
                            "$elemMatch": {
                                "readInMilli": { "$exists": false }
                            } 
                        }
                    }
                }
            }
        }
    ]
)

执行后报错：

Unrecognized expression '$elemMatch'

解决方案 #

错误原因 #

1. $elemMatch是查询过滤操作符，不能直接在$group阶段的累加器中使用
2. $count是聚合阶段（用于统计文档总数），不是累加器函数，你错误地将其当作字段计算函数使用

正确聚合查询（兼容MongoDB 4.0） #

db.getCollection('message_collection').aggregate([
    // 保留你的自定义match过滤条件
    { "$match": {} },
    // 标记单条消息是否为未读
    {
        "$addFields": {
            "isUnread": {
                "$or": [
                    // 情况1：statuses是空数组
                    { "$eq": ["$statuses", []] },
                    // 情况2：statuses数组中没有带readInMilli的元素
                    { "$not": { "$exists": { "$arrayElemAt": ["$statuses.readInMilli", 0] } } }
                ]
            }
        }
    },
    // 按线程分组汇总
    {
        "$group": {
            "_id": "$thread",
            "timeInMilliOfNewestMessage": { "$max": "$sentInMilli" },
            // 只要分组内有一条未读消息，结果就是true
            "containsUnreadMessages": { "$max": "$isUnread" }
        }
    },
    // 调整输出字段格式，匹配期望结果
    {
        "$project": {
            "_id": 0,
            "thread": "$_id",
            "timeInMilliOfNewestMessage": 1,
            "containsUnreadMessages": 1
        }
    }
])

逻辑补充 #

如果你的statuses数组可能包含多个用户的已读状态（比如单条消息被多个用户查看），需要判断是否存在任意一个未读状态，可将isUnread的判断逻辑修改为：

"isUnread": {
    "$anyElementTrue": {
        "$map": {
            "input": "$statuses",
            "as": "status",
            "in": { "$not": { "$exists": "$$status.readInMilli" } }
        }
    }
}

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