Improved MongoDB change stream.
Ylian Saint-Hilaire committed
May 29, 2019 at 14:36 UTC
3c0e7c9700d31862bfcd03df82a1ece1f8ec88b8
6 files changed
+76
-29
db.js
+38
-12
@@ -231,19 +231,45 @@ module.exports.CreateDB = function (parent, func) {
231
232
// Setup the changeStream on the MongoDB main collection if possible
233
if (parent.args.mongodbchangestream == true) {
234
- obj.fileChangeStream = obj.file.watch([{ $match: { 'fullDocument.type': { $in: ['node', 'mesh', 'user'] } } }], { fullDocument: 'updateLookup' });
234
+ obj.fileChangeStream = obj.file.watch( [ { $match: { $or: [{ 'fullDocument.type': { $in: ['node', 'mesh', 'user'] } }, { 'operationType': 'delete' }] } } ], { fullDocument: 'updateLookup' });
235
obj.fileChangeStream.on('change', function (change) {
236
- switch (change.fullDocument.type) {
237
- case 'node': { dbNodeChange(change); break; } // A node has changed
238
- case 'mesh': { dbMeshChange(change); break; } // A device group has changed
239
- case 'user': { dbUserChange(change); break; } // A user account has changed
236
+ if (change.operationType == 'update') {
237
+ switch (change.fullDocument.type) {
238
+ case 'node': { dbNodeChange(change, false); break; } // A node has changed
239
+ case 'mesh': { dbMeshChange(change, false); break; } // A device group has changed
240
+ case 'user': { dbUserChange(change, false); break; } // A user account has changed
241
+ }
242
+ } else if (change.operationType == 'insert') {
243
+ switch (change.fullDocument.type) {
244
+ case 'node': { dbNodeChange(change, true); break; } // A node has added
245
+ case 'mesh': { dbMeshChange(change, true); break; } // A device group has created
246
+ case 'user': { dbUserChange(change, true); break; } // A user account has created
247
+ }
248
+ } else if (change.operationType == 'delete') {
249
+ var splitId = change.documentKey._id.split('/');
250
+ switch (splitId[0]) {
251
+ case 'node': {
252
+ //Not Good: Problem here is that we don't know what meshid the node belonged to before the delete.
253
+ //parent.DispatchEvent(['*', node.meshid], obj, { etype: 'node', action: 'removenode', nodeid: change.documentKey._id, domain: splitId[1] });
254
+ break;
255
+ }
256
+ case 'mesh': {
257
+ parent.DispatchEvent(['*', node.meshid], obj, { etype: 'mesh', action: 'deletemesh', meshid: change.documentKey._id, domain: splitId[1] });
258
+ break;
259
+ }
260
+ case 'user': {
261
+ //Not Good: This is not a perfect user removal because we don't know what groups the user was in.
262
+ //parent.DispatchEvent(['*', 'server-users'], obj, { etype: 'user', action: 'accountremove', userid: change.documentKey._id, domain: splitId[1], username: splitId[2] });
263
+ break;
264
+ }
265
+ }
266
}
267
});
268
obj.changeStream = true;
269
}
270
271
// Setup MongoDB events collection and indexes
246
- obj.eventsfile = db.collection('events'); // Collection containing all events
272
+ obj.eventsfile = db.collection('events'); // Collection containing all events
273
obj.eventsfile.indexes(function (err, indexes) {
274
// Check if we need to reset indexes
275
var indexesByName = {}, indexCount = 0;
@@ -773,16 +799,16 @@ module.exports.CreateDB = function (parent, func) {
799
function padNumber(number, digits) { return Array(Math.max(digits - String(number).length + 1, 0)).join(0) + number; }
800
801
// Called when a node has changed
776
- function dbNodeChange(nodeChange) {
802
+ function dbNodeChange(nodeChange, added) {
803
const node = nodeChange.fullDocument;
804
if (node.intelamt && node.intelamt.pass) { delete node.intelamt.pass; } // Remove the Intel AMT password before eventing this.
779
- parent.DispatchEvent(['*', node.meshid], obj, { etype: 'node', action: 'changenode', node: node, nodeid: node._id, domain: node.domain, nolog: 1 });
805
+ parent.DispatchEvent(['*', node.meshid], obj, { etype: 'node', action: (added ? 'addnode' : 'changenode'), node: node, nodeid: node._id, domain: node.domain, nolog: 1 });
806
}
807
808
// Called when a device group has changed
783
- function dbMeshChange(meshChange) {
809
+ function dbMeshChange(meshChange, added) {
810
const mesh = meshChange.fullDocument;
785
- mesh.action = 'meshchange';
811
+ if (mesh.deleted) { mesh.action = 'deletemesh'; } else { mesh.action = (added ? 'createmesh' : 'meshchange'); }
812
mesh.meshid = mesh._id;
813
mesh.nolog = 1;
814
delete mesh.type;
@@ -791,9 +817,9 @@ module.exports.CreateDB = function (parent, func) {
817
}
818
819
// Called when a user account has changed
794
- function dbUserChange(userChange) {
820
+ function dbUserChange(userChange, added) {
821
const user = userChange.fullDocument;
796
- parent.DispatchEvent(['*', 'server-users', user._id], obj, { etype: 'user', username: user.name, account: parent.webserver.CloneSafeUser(user), action: 'accountchange', domain: user.domain, nolog: 1 });
822
+ parent.DispatchEvent(['*', 'server-users', user._id], obj, { etype: 'user', username: user.name, account: parent.webserver.CloneSafeUser(user), action: (added ? 'accountcreate' : 'accountchange'), domain: user.domain, nolog: 1 });
823
}
824
825
return obj;
meshuser.js
+22
-11
@@ -1038,13 +1038,15 @@ module.exports.CreateMeshUser = function (parent, db, ws, req, args, domain, use
1038
newuser.hash = hash;
1039
db.SetUser(newuser);
1040
1041
- var targets = ['*', 'server-users'];
1041
+ var event, targets = ['*', 'server-users'];
1042
if (newuser.groups) { for (var i in newuser.groups) { targets.push('server-users:' + i); } }
1043
if (newuser.email == null) {
1044
- parent.parent.DispatchEvent(targets, obj, { etype: 'user', username: newuser.name, account: parent.CloneSafeUser(newuser), action: 'accountcreate', msg: 'Account created, username is ' + newuser.name, domain: domain.id });
1044
+ event = { etype: 'user', username: newuser.name, account: parent.CloneSafeUser(newuser), action: 'accountcreate', msg: 'Account created, username is ' + newuser.name, domain: domain.id };
1045
} else {
1046
- parent.parent.DispatchEvent(targets, obj, { etype: 'user', username: newuser.name, account: parent.CloneSafeUser(newuser), action: 'accountcreate', msg: 'Account created, email is ' + newuser.email, domain: domain.id });
1046
+ event = { etype: 'user', username: newuser.name, account: parent.CloneSafeUser(newuser), action: 'accountcreate', msg: 'Account created, email is ' + newuser.email, domain: domain.id };
1047
}
1048
+ if (parent.db.changeStream) { event.noact = 1; } // If DB change stream is active, don't use this event to create the user. Another event will come.
1049
+ parent.parent.DispatchEvent(targets, obj, event);
1050
}, newuser);
1051
}
1052
}
@@ -1094,13 +1096,15 @@ module.exports.CreateMeshUser = function (parent, db, ws, req, args, domain, use
1096
newuser.hash = hash;
1097
db.SetUser(newuser);
1098
1097
- var targets = ['*', 'server-users'];
1099
+ var event, targets = ['*', 'server-users'];
1100
if (newuser.groups) { for (var i in newuser.groups) { targets.push('server-users:' + i); } }
1101
if (command.email == null) {
1100
- parent.parent.DispatchEvent(targets, obj, { etype: 'user', username: newusername, account: parent.CloneSafeUser(newuser), action: 'accountcreate', msg: 'Account created, username is ' + command.user, domain: domain.id });
1102
+ event = { etype: 'user', username: newusername, account: parent.CloneSafeUser(newuser), action: 'accountcreate', msg: 'Account created, username is ' + command.user, domain: domain.id };
1103
} else {
1102
- parent.parent.DispatchEvent(targets, obj, { etype: 'user', username: newusername, account: parent.CloneSafeUser(newuser), action: 'accountcreate', msg: 'Account created, email is ' + command.email, domain: domain.id });
1104
+ event = { etype: 'user', username: newusername, account: parent.CloneSafeUser(newuser), action: 'accountcreate', msg: 'Account created, email is ' + command.email, domain: domain.id };
1105
}
1106
+ if (parent.db.changeStream) { event.noact = 1; } // If DB change stream is active, don't use this event to create the user. Another event will come.
1107
+ parent.parent.DispatchEvent(targets, obj, event);
1108
}, 0);
1109
}
1110
});
@@ -1390,7 +1394,9 @@ module.exports.CreateMeshUser = function (parent, db, ws, req, args, domain, use
1394
user.links[meshid] = { rights: 0xFFFFFFFF };
1395
user.subscriptions = parent.subscribe(user._id, ws);
1396
db.SetUser(user);
1393
- parent.parent.DispatchEvent(['*', meshid, user._id], obj, { etype: 'mesh', username: user.name, meshid: meshid, name: command.meshname, mtype: command.meshtype, desc: command.desc, action: 'createmesh', links: links, msg: 'Mesh created: ' + command.meshname, domain: domain.id });
1397
+ var event = { etype: 'mesh', username: user.name, meshid: meshid, name: command.meshname, mtype: command.meshtype, desc: command.desc, action: 'createmesh', links: links, msg: 'Mesh created: ' + command.meshname, domain: domain.id };
1398
+ if (db.changeStream) { event.noact = 1; } // If DB change stream is active, don't use this event to create the mesh. Another event will come.
1399
+ parent.parent.DispatchEvent(['*', meshid, user._id], obj, event);
1400
});
1401
}
1402
break;
@@ -1408,7 +1414,9 @@ module.exports.CreateMeshUser = function (parent, db, ws, req, args, domain, use
1414
if ((command.meshid.split('/').length != 3) || (command.meshid.split('/')[1] != domain.id)) return; // Invalid domain, operation only valid for current domain
1415
1416
// Fire the removal event first, because after this, the event will not route
1411
- parent.parent.DispatchEvent(['*', command.meshid], obj, { etype: 'mesh', username: user.name, meshid: command.meshid, name: command.meshname, action: 'deletemesh', msg: 'Mesh deleted: ' + command.meshname, domain: domain.id });
1417
+ var event = { etype: 'mesh', username: user.name, meshid: command.meshid, name: command.meshname, action: 'deletemesh', msg: 'Mesh deleted: ' + command.meshname, domain: domain.id };
1418
+ if (db.changeStream) { event.noact = 1; } // If DB change stream is active, don't use this event to remove the mesh. Another event will come.
1419
+ parent.parent.DispatchEvent(['*', command.meshid], obj, event);
1420
1421
// Remove all user links to this mesh
1422
for (i in meshes) {
@@ -1722,12 +1730,15 @@ module.exports.CreateMeshUser = function (parent, db, ws, req, args, domain, use
1730
db.RemoveAllNodeEvents(node._id); // Remove all events for this node
1731
db.removeAllPowerEventsForNode(node._id); // Remove all power events for this node
1732
db.Get('ra' + obj.dbNodeKey, function (err, nodes) {
1725
- if ((nodes != null) && (nodes.length == 1)) { db.Remove('da' + nodes[0].daid); } // Remove diagnostic agent to real agent link
1726
- db.Remove('ra' + node._id); // Remove real agent to diagnostic agent link
1733
+ if ((nodes != null) && (nodes.length == 1)) { db.Remove('da' + nodes[0].daid); } // Remove diagnostic agent to real agent link
1734
+ db.Remove('ra' + node._id); // Remove real agent to diagnostic agent link
1735
});
1736
1737
// Event node deletion
1730
- parent.parent.DispatchEvent(['*', node.meshid], obj, { etype: 'node', username: user.name, action: 'removenode', nodeid: node._id, msg: 'Removed device ' + node.name + ' from group ' + mesh.name, domain: domain.id });
1738
+ var event = { etype: 'node', username: user.name, action: 'removenode', nodeid: node._id, msg: 'Removed device ' + node.name + ' from group ' + mesh.name, domain: domain.id };
1739
+ // TODO: We can't use the changeStream for node delete because we will not know the meshid the device was in.
1740
+ //if (db.changeStream) { event.noact = 1; } // If DB change stream is active, don't use this event to remove the node. Another event will come.
1741
+ parent.parent.DispatchEvent(['*', node.meshid], obj, event);
1742
1743
// Disconnect all connections if needed
1744
var state = parent.parent.GetConnectivityState(nodeid);
package.json
+1
-1
@@ -1,6 +1,6 @@
1
{
2
"name": "meshcentral",
3
- "version": "0.3.5-j",
3
+ "version": "0.3.5-k",
4
"keywords": [
5
"Remote Management",
6
"Intel AMT",
views/default-mobile.handlebars
+1
@@ -881,6 +881,7 @@
881
case 'addnode': {
882
var node = message.event.node;
883
if (!meshes[node.meshid]) break; // This is a node for a mesh we don't know. Happens when we are site administrator, we get all messages.
884
+ if (getNodeFromId(node._id) != null) break; // This node is already known.
885
node.namel = node.name.toLowerCase();
886
if (node.rname) { node.rnamel = node.rname.toLowerCase(); } else { node.rnamel = node.namel; }
887
node.meshnamel = meshes[node.meshid].name.toLowerCase();
views/default.handlebars
+2
-1
@@ -1690,7 +1690,7 @@
1690
}
1691
case 'createmesh': {
1692
// A new mesh was created
1693
- if (message.event.links[userinfo._id] != null) { // Check if this is a mesh create for a mesh we own. If site administrator, we get all messages so need to ignore some.
1693
+ if ((meshes[message.event.meshid] == null) && (message.event.links[userinfo._id] != null)) { // Check if this is a mesh create for a mesh we own. If site administrator, we get all messages so need to ignore some.
1694
meshes[message.event.meshid] = { _id: message.event.meshid, name: message.event.name, mtype: message.event.mtype, desc: message.event.desc, links: message.event.links };
1695
masterUpdate(4 + 128);
1696
meshserver.send({ action: 'files' });
@@ -1758,6 +1758,7 @@
1758
case 'addnode': {
1759
var node = message.event.node;
1760
if (!meshes[node.meshid]) break; // This is a node for a mesh we don't know. Happens when we are site administrator, we get all messages.
1761
+ if (getNodeFromId(node._id) != null) break; // This node is already known.
1762
node.namel = node.name.toLowerCase();
1763
if (node.rname) { node.rnamel = node.rname.toLowerCase(); } else { node.rnamel = node.namel; }
1764
node.meshnamel = meshes[node.meshid].name.toLowerCase();
webserver.js
+12
-4
@@ -309,7 +309,9 @@ module.exports.CreateWebServer = function (parent, db, args, certificates) {
309
if (usercount == 0) { user.siteadmin = 0xFFFFFFFF; /*if (domain.newaccounts === 2) { delete domain.newaccounts; }*/ } // If this is the first user, give the account site admin.
310
obj.users[user._id] = user;
311
obj.db.SetUser(user);
312
- obj.parent.DispatchEvent(['*', 'server-users'], obj, { etype: 'user', userid: userid, username: username, account: obj.CloneSafeUser(user), action: 'accountcreate', msg: 'Account created, name is ' + name, domain: domain.id });
312
+ var event = { etype: 'user', userid: userid, username: username, account: obj.CloneSafeUser(user), action: 'accountcreate', msg: 'Account created, name is ' + name, domain: domain.id };
313
+ if (obj.db.changeStream) { event.noact = 1; } // If DB change stream is active, don't use this event to create the user. Another event will come.
314
+ obj.parent.DispatchEvent(['*', 'server-users'], obj, event);
315
return fn(null, user._id);
316
} else {
317
// This is an existing user
@@ -363,7 +365,9 @@ module.exports.CreateWebServer = function (parent, db, args, certificates) {
365
if (usercount == 0) { user.siteadmin = 0xFFFFFFFF; /*if (domain.newaccounts === 2) { delete domain.newaccounts; }*/ } // If this is the first user, give the account site admin.
366
obj.users[user._id] = user;
367
obj.db.SetUser(user);
366
- obj.parent.DispatchEvent(['*', 'server-users'], obj, { etype: 'user', username: user.name, account: obj.CloneSafeUser(user), action: 'accountcreate', msg: 'Account created, name is ' + name, domain: domain.id });
368
+ var event = { etype: 'user', username: user.name, account: obj.CloneSafeUser(user), action: 'accountcreate', msg: 'Account created, name is ' + name, domain: domain.id };
369
+ if (obj.db.changeStream) { event.noact = 1; } // If DB change stream is active, don't use this event to create the user. Another event will come.
370
+ obj.parent.DispatchEvent(['*', 'server-users'], obj, event);
371
return fn(null, user._id);
372
} else {
373
// This is an existing user
@@ -814,7 +818,9 @@ module.exports.CreateWebServer = function (parent, db, args, certificates) {
818
// Send the verification email
819
if ((obj.parent.mailserver != null) && (domain.auth != 'sspi') && (domain.auth != 'ldap') && (obj.common.validateEmail(user.email, 1, 256) == true)) { obj.parent.mailserver.sendAccountCheckMail(domain, user.name, user.email); }
820
}, 0);
817
- obj.parent.DispatchEvent(['*', 'server-users'], obj, { etype: 'user', username: user.name, account: obj.CloneSafeUser(user), action: 'accountcreate', msg: 'Account created, email is ' + req.body.email, domain: domain.id });
821
+ var event = { etype: 'user', username: user.name, account: obj.CloneSafeUser(user), action: 'accountcreate', msg: 'Account created, email is ' + req.body.email, domain: domain.id };
822
+ if (obj.db.changeStream) { event.noact = 1; } // If DB change stream is active, don't use this event to create the user. Another event will come.
823
+ obj.parent.DispatchEvent(['*', 'server-users'], obj, event);
824
}
825
res.redirect(domain.url);
826
}
@@ -1239,7 +1245,9 @@ module.exports.CreateWebServer = function (parent, db, args, certificates) {
1245
if (usercount == 0) { user2.siteadmin = 0xFFFFFFFF; } // If this is the first user, give the account site admin.
1246
obj.users[req.session.userid] = user2;
1247
obj.db.SetUser(user2);
1242
- obj.parent.DispatchEvent(['*', 'server-users'], obj, { etype: 'user', username: req.connection.user, account: obj.CloneSafeUser(user2), action: 'accountcreate', msg: 'Domain account created, user ' + req.connection.user, domain: domain.id });
1248
+ var event = { etype: 'user', username: req.connection.user, account: obj.CloneSafeUser(user2), action: 'accountcreate', msg: 'Domain account created, user ' + req.connection.user, domain: domain.id };
1249
+ if (obj.db.changeStream) { event.noact = 1; } // If DB change stream is active, don't use this event to create the user. Another event will come.
1250
+ obj.parent.DispatchEvent(['*', 'server-users'], obj, event);
1251
}
1252
}
1253
}