MongoDB单聚合查询实现按Chat/用户筛选未读消息

需求描述 #

需要通过单条MongoDB聚合查询，按指定Chat ID列表和用户ID筛选未读消息。未读消息判定规则：

• chats_users.max_read_date 小于 message.create_date
• message.from_id 不等于当前用户ID

当前实现 #

目前采用循环查询方式：先循环查询chats_users集合获取指定Chat ID和用户ID的记录，再循环查询message集合筛选符合条件的消息。Go代码如下：

func (r *Mongo) UnreadMessageCount(ctx context.Context, chats []*Chat, uid string) (map[string]int64, error) {

    match := bson.A{}
    for _, chat := range chats {
        match = append(match, chat.ID)
    }

    chatsUsersList := make([]*domain.ChatsUsers, 0)
    for _, ch := range chats {
        chu, err := r.FindChatUser(ctx, ch.ID, uid)
        if err != nil {
            l.Error().Err(err).Msg("failed to find chat user")
            return nil, err
        }
        chatsUsersList = append(chatsUsersList, chu)
    }

    list := make([]*domain.Message, 0)

    for _, ch := range chats {
        for _, chu := range chatsUsersList {
            if chu.ChatID == ch.ID {
                filter := bson.D{
                    // search for messages by active chat IDs
                    { Key: "chat_id", Value: ch.ID},
                    // add filtering: that the message has not yet been read,
                    // and that the messages we select are not written by the current user.
                    {Key: "$and",
                        Value: bson.A{
                            bson.D{
                                {
                                    Key: "create_date", Value: bson.D{ bson.D{
                                    // $gt Matches values that are greater than the specified value.
                                    { Key: "$gt", Value: chu.MaxReadDate}}
                                },
                                // $ne Matches all values that are not equal to the specified value.
                                {Key: "from_id", Value: bson.M{"$ne": uid}}
                            },
                        },
                    },
                }

                cursor, err := r.colMessage.Find(ctx, filter)
                ....

                var res []*domain.Message
                ....
            }
        }
    }

    // it doesn't make sense to use an array of messages, we need to create a map,
    // which will have the chat ID and the number of unread messages in it.
    messages := make(map[string]int64)
    for _, msg := range list {
        messages[msg.ChatID]++
    }

    return messages, nil
}

集合结构 #

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"` // 系统内对应用户ID
    MaxReadDate int64 `json:"max_read_date" bson:"max_read_date"` // 最后读消息时间
}

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"`
}

尝试的聚合查询 #

第一次尝试（未成功） #

db.message.aggregate([
    {
        $match: {
            $expr: {
                $and: [
                    {
                        chat_id: {
                            $in: ["ad0a3405-1a16-48f9-93e6-51b17a7283e2"]
                        }
                    },
                    {
                        from_id: {
                            $ne: "63f5002735bb916dab3f2b1d"
                        }
                    },
                    {
                        $gt: ["$create_date", { $max: "$chats_users.max_read_date" }]
                    }
                ]
            }
        }
    },
    {
        $lookup: {
            from: "chatsusers",
            localField: "chat_id",
            foreignField: "chat_id",
            as: "chats_users"
        }
    },
    {
        $unwind: "$chats_users"
    },
    {
        $group: {
            _id: "$chat_id",
            messages: {
                $push: {
                    id: "$id",
                    chat_id: "$chat_id",
                    from_id: "$from_id",
                    create_date: "$create_date",
                    type: "$type",
                    media: "$media",
                    body: "$body",
                    update_at: "$update_at",
                    modifications: "$modifications",
                    viewed: "$viewed"
                }
            },
            max_read_date: { $max: "$chats_users.max_read_date" }
        }
    }
]);

第二次尝试（未成功） #

db.chats_users.aggregate([
    {
        $match: {
            user_id: "63f5002735bb916dab3f2b1d",
            chat_id: { $in: ["ad0a3405-1a16-48f9-93e6-51b17a7283e2"] },
        }
    },
    {
        $lookup: {
            from: "message",
            let: { chat_id: "$chat_id", max_read_date: "$max_read_date" },
            pipeline: [
                {
                    $match: {
                        $expr: {
                            $and: [
                                { $eq: ["$chat_id", "$$chat_id"] },
                                { $ne: ["$from_id", "63f5002735bb916dab3f2b1d"] },
                                { $gt: ["$create_date", "$$max_read_date"] },
                            ]
                        }
                    }
                }
            ],
            as: "messages"
        }
    },
    {
        $project: {
            chat_id: 1,
            message_count: { $size: "$messages" }
        }
    }
])

预期结果 #

获取符合条件的4条测试数据，或按Chat ID统计未读消息数量。

请问如何修正聚合查询，实现按指定Chat ID和用户ID筛选符合条件的未读消息？

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