获取对话中的最新消息

Question

假设我有两个Mongoose集合- conversation和message-，并且我想显示特定用户所在的对话列表，按最新消息排序，并在对话名称下方显示该消息的预览

在获得用户的对话之后，如何才能仅从每个对话中选择最新消息，并将这些消息附加到其相应的对话中？（鉴于架构看起来像这样）：

var ConversationSchema = new Schema({
    name: String,
    participants: {
        type: [{
            type: Schema.Types.ObjectId,
            ref: 'User'
        }]
    }
});

var MessageSchema = new Schema({
    conversation: {type: Schema.Types.ObjectId, ref: 'Conversation', required: true},
    text: {type: String, required: true},
    user: {type: Schema.Types.ObjectId, ref: 'User', required: true}
});

我收到的消息是我可能应该使用Mongo的“聚合”框架，但是我以前从未使用过它，因此我需要一些帮助。谢谢！

Answer 1

聚合框架确实将是您完成此任务的好伙伴。通过聚合，它可以使用过滤器，分组器，分类器，转换和其他运算符的多级管道，将集合精简为基本信息。精炼的结果集比其他技术更有效地产生。

对于上述用例，聚合由一系列特殊运算符组成，这些运算符应用于称为管道的集合：

[
    { "$match": { /* options */ } },
    { "$lookup": { /* options */ } },  
    { "$group": { /* options */ } }
]

在执行管道时，MongoDB通过管道将运算符相互传递。这里的“管道”具有Linux的含义：运算符的输出成为后续运算符的输入。每个运算符的结果都是新的文档集合。因此，Mongo如下执行前一个管道：

collection | $match | $lookup | $group => result

现在，将以上内容应用到您的任务中，您需要一个$match流水线步骤作为聚合的第一阶段，因为它允许您过滤文档流，并且仅将未经修改的匹配文档传递到下一个流水线阶段。因此，如果Conversation仅使用$match查询对象在模型 上运行聚合，则将获得特定用户所在的文档：

Conversation.aggregate([
    { "$match": { "participants.type": userId } }
]).exec(function(err, result){
    if (err) throw err;
    console.log(result);
})

您可以通过管道$lookup传递另一个运算符，在这种情况下，您需要对messages同一数据库中的集合执行左外连接的运算符，以从“已连接”集合中过滤文档以进行处理：

Conversation.aggregate([
    { "$match": { "participants.type": userId } },
    {
        "$lookup": {
            "from": "messages",
            "localfield": "_id",
            "foreignField": "conversation",
            "as": "messages"
        }
    }
]).exec(function(err, result){
    if (err) throw err;
    console.log(result);
})

这将输出控制台conversation文档中的字段以及一个名为“ messages”的新数组字段，其元素是“ joined” messages集合中的匹配文档。该$lookup阶段将这些经过重整的文档传递到下一个阶段。

现在，您已将用户conversations与一起使用messages，然后如何ONLY从每个对话中选择最新消息？

这可以通过$group管道运算符来实现，但是在将$group运算符应用于前一个管道中的文档之前，您需要展平messages数组以获得最新文档，并且$unwind运算符将为您解构数组元素。

当$unwind运算符应用于数组字段时，它将为列表中的每个元素生成一条新记录$unwind：

Conversation.aggregate([
    { "$match": { "participants.type": userId } },
    {
        "$lookup": {
            "from": "messages",
            "localfield": "_id",
            "foreignField": "conversation",
            "as": "messages"
        }
    },
    { "$unwind": "$messages" }
]).exec(function(err, result){
    if (err) throw err;
    console.log(result);
})

假设您MessageSchema有一个timestamp字段（例如createdAt），它表示发送消息的日期时间，那么您可以按该字段的降序对文档进行重新排序，以准备处理下一个管道。该$sort运营商是完美的，例如：

Conversation.aggregate([
    { "$match": { "participants.type": userId } },
    {
        "$lookup": {
            "from": "messages",
            "localfield": "_id",
            "foreignField": "conversation",
            "as": "messages"
        }
    },
    { "$unwind": "$messages" },
    { "$sort": { "messages.createdAt": -1 } }
]).exec(function(err, result){
    if (err) throw err;
    console.log(result);
})

拥有非规范化文档后，您便可以将数据分组以对其进行处理。组管道运算符类似于SQL的GROUP BY子句。在SQL中，GROUP BY除非使用任何聚合函数（称为“ 累积器”），否则不能使用。同样，您还必须在MongoDB中使用聚合函数。MongoDB _id仅使用字段标识分组表达式，在这种情况下，按_id字段将文档分组：

Conversation.aggregate([
    { "$match": { "participants.type": userId } },
    {
        "$lookup": {
            "from": "messages",
            "localfield": "_id",
            "foreignField": "conversation",
            "as": "messages"
        }
    },
    { "$unwind": "$messages" },
    { "$sort": { "messages.createdAt": -1 } }
    {
        "$group": {
            "_id": "$_id",
            "name": { "$first": "$name" },
            "participants": { "$first": "$participants" },
            "latestMessage": { "$first": "$message" }
        }
    }
]).exec(function(err, result){
    if (err) throw err;
    console.log(result);
})

在上面，我们对$first组累加器运算符很感兴趣，因为它从每个组的第一个文档返回一个值。由于前一个管道运算符按降序对文档进行排序，因此$first会给您LATEST消息。

因此，运行上述最后的聚合操作将为您提供所需的结果。假设MessageSchema时间戳记对于确定最新消息至关重要，否则它将一直执行该$unwind步骤。