Вы можете использовать $ graphLookup для группировки последовательных документов в массив. Для поиска требуется коллекция, и в вашем случае это может быть представление .
Представление объединяет документы в предыдущих и следующих парах, используя оператор $ zip :
db.createView("events-view", "original_collection", [
{ $sort: { timestamp: 1 } },
{ $group: { _id: null, docs: { $push: "$$ROOT" } } },
{ $project: {
pair: { $zip: {
inputs:[ { $concatArrays: [ [false], "$docs" ]} , "$docs" ]
} }
} },
{ $unwind: "$pair" },
{ $project: {
prev: { $arrayElemAt: [ "$pair", 0 ] },
next: { $arrayElemAt: [ "$pair", 1 ] }
} },
{ $project: {
_id: "$prev._id",
prev: 1,
next: 1,
sameType: { $eq: ["$prev.type", "$next.type"] }
} },
]);
Это должно выглядеть следующим образом:
{
"prev" : false,
"next" : {
"_id" : ObjectId("5d1b68d708f3870049d9cc37"),
"controllerId" : "80058c2b-9525-4f7f-8e26-faea4ad92b15",
"id" : "76800453-e72e-410b-accc-cf47cd2773a1",
"type" : "controller_connection_status",
"timestamp" : 1562077399832.0
},
"sameType" : false
},
{
"prev" : {
"_id" : ObjectId("5d1b68d708f3870049d9cc37"),
"controllerId" : "80058c2b-9525-4f7f-8e26-faea4ad92b15",
"id" : "76800453-e72e-410b-accc-cf47cd2773a1",
"type" : "controller_connection_status",
"timestamp" : 1562077399832.0
},
"next" : {
"_id" : ObjectId("5d1b68db08f3870049d9cc39"),
"controllerId" : "80058c2b-9525-4f7f-8e26-faea4ad92b15",
"id" : "fabd883c-6971-4977-b3fc-31679c2b85dd",
"type" : "controller_connection_status",
"timestamp" : 1562077402916.0
},
"_id" : ObjectId("5d1b68d708f3870049d9cc37"),
"sameType" : true
},
{
"prev" : {
"_id" : ObjectId("5d1b68db08f3870049d9cc39"),
"controllerId" : "80058c2b-9525-4f7f-8e26-faea4ad92b15",
"id" : "fabd883c-6971-4977-b3fc-31679c2b85dd",
"type" : "controller_connection_status",
"timestamp" : 1562077402916.0
},
"next" : {
"_id" : ObjectId("5d1b68db08f3870049d9cc3a"),
"controllerId" : "80058c2b-9525-4f7f-8e26-faea4ad92b15",
"siteId" : "226168be-866c-11e8-adc0-fa7ae01bbebc",
"id" : "98decbea-8288-4df5-807d-14e90f929df2",
"type" : "controller_added",
"timestamp" : 1562077402920.0
},
"_id" : ObjectId("5d1b68db08f3870049d9cc39"),
"sameType" : false
},
etc...
Затем вы можете запросить документы группировки представлений по типу и последней отметке времени, пока не будет выполнено условие "sameType". Самая длинная цепочка документов - это количество, которое вы ищете:
db.getCollection("events-view").aggregate([
{ $graphLookup: {
from: "events-view",
startWith: "$next._id",
connectFromField: "next._id",
connectToField: "_id",
restrictSearchWithMatch: { "sameType": true },
as: "chain"
} },
{ $project: {
_id: "$next._id",
type: "$next.type",
chain: { $concatArrays: [ [{ next: "$next" }], "$chain" ] }
} },
{ $addFields: {
chainLength: { $size: "$chain" },
timestamp: { $max: { $map: {
input: "$chain",
in: "$$this.next.timestamp"
} } }
} },
{ $group: {
_id: {type: "$type", timestamp: "$timestamp"},
count: {$max: "$chainLength"}
} },
{ $sort: { "_id.timestamp": 1 } },
{ $project: {
_id: 0,
timestamp: "$_id.timestamp",
type: "$_id.type",
count: 1
} }
])
Следует отметить, что запрос будет довольно медленным. Чем длиннее цепи, тем хуже производительность. Также имейте в виду, что этап $ graphLookup должен находиться в пределах 100 МБ памяти. Для больших коллекций вы должны установить опцию allowDiskUse на true.