added group_id support to action counts, cli verifier

bin/cli

@@ -16,6 +16,7 @@ program
   .command('token', 'work with the access tokens')
   .command('users', 'work with the application auth')
   .command('migration', 'provides utilities for migrating the database')
+  .command('verify', 'provides utilities for performing data verification')
   .command(
     'plugins',
     'provides utilities for interacting with the plugin system'

bin/cli-verify

@@ -0,0 +1,48 @@
+#!/usr/bin/env node
+
+/**
+ * Module dependencies.
+ */
+
+const program = require('./commander');
+const mongoose = require('../services/mongoose');
+const util = require('./util');
+const databaseVerifications = require('./verifications/database');
+
+// Register the shutdown criteria.
+util.onshutdown([
+  () => mongoose.disconnect()
+]);
+
+async function database(program) {
+  try {
+    for (const verification of databaseVerifications) {
+      await verification(program);
+    }
+  } catch (err) {
+    console.error(`Failed to process all the ${databaseVerifications.length} verifications`, err);
+    util.shutdown(1);
+    return;
+  }
+
+  util.shutdown();
+}
+
+//==============================================================================
+// Setting up the program command line arguments.
+//==============================================================================
+
+program
+  .command('db')
+  .description('verifies the database integrity')
+  .option('-f, --fix', 'fix the problems found with database inconsistencies')
+  .option('-b, --batch [size]', 'batch size to process verifications and repairs of documents [default: 1000]', parseInt)
+  .action(database);
+
+program.parse(process.argv);
+
+// If there is no command listed, output help.
+if (!process.argv.slice(2).length) {
+  program.outputHelp();
+  util.shutdown();
+}

bin/verifications/database/comments.js

@@ -0,0 +1,141 @@
+const CommentModel = require('../../../models/comment');
+const ActionsService = require('../../../services/actions');
+const {arrayJoinBy} = require('../../../graph/loaders/util');
+const sc = require('snake-case');
+
+const getBatch = async (limit, offset) => CommentModel
+  .find({})
+  .select({'id': 1, 'action_counts': 1})
+  .limit(limit)
+  .skip(offset)
+  .sort('created_at');
+
+module.exports = async ({fix = false, batch = 1000}) => {
+  let operations = [];
+
+  // Count how many comments there are to process.
+  const totalCount = await CommentModel.count();
+
+  let offset = 0;
+  let comments = [];
+  let commentIDs = [];
+
+  console.log(`Processing ${totalCount} comments...`);
+
+  // Keep processing documents until there are is none left.
+  while (offset < totalCount) {
+
+    // Get a batch of comments.
+    comments = await getBatch(batch, offset);
+    commentIDs = comments.map(({id}) => id);
+
+    // Get their action summaries.
+    let allActionSummaries = await ActionsService
+      .getActionSummaries(commentIDs)
+      .then(arrayJoinBy(commentIDs, 'item_id'));
+
+    // Loop over the comments, with their action summaries.
+    for (let i = 0; i < comments.length; i++) {
+      let comment = comments[i];
+      let actionSummaries = allActionSummaries[i];
+
+      // And check to see if the action summaries we just computed match what is
+      // currently set for the comments.
+      let commentOperations = [];
+
+      // First we process all the group id's.
+      for (let actionSummary of actionSummaries) {
+        if (actionSummary.group_id === null) {
+          continue;
+        }
+
+        const ACTION_TYPE = sc(actionSummary.action_type.toLowerCase());
+        const GROUP_ID = sc(actionSummary.group_id.toLowerCase());
+
+        if (GROUP_ID.length <= 0) {
+          continue;
+        }
+
+        const ACTION_COUNT_FIELD = `${ACTION_TYPE}_${GROUP_ID}`;
+
+        // Check that the action summaries match the cached counts.
+        if (!comment.action_counts || !(ACTION_COUNT_FIELD in comment.action_counts) || comment.action_counts[ACTION_COUNT_FIELD] !== actionSummary.count) {
+
+          // Batch updates for those changes.
+          commentOperations.push({
+            [`action_counts.${ACTION_COUNT_FIELD}`]: actionSummary.count,
+          });
+        }
+      }
+
+      // Group all the action summaries together from all the different group
+      // ids.
+      let groupedActionSummaries = actionSummaries.reduce((acc, actionSummary) => {
+        const ACTION_TYPE = sc(actionSummary.action_type.toLowerCase());
+
+        if (!(ACTION_TYPE in acc)) {
+          acc[ACTION_TYPE] = 0;
+        }
+
+        acc[ACTION_TYPE] += actionSummary.count;
+
+        return acc;
+      }, {});
+
+      Object.keys(groupedActionSummaries).forEach((ACTION_COUNT_FIELD) => {
+        const count = groupedActionSummaries[ACTION_COUNT_FIELD];
+
+        // Check that the action summaries match the cached counts.
+        if (!comment.action_counts || !(ACTION_COUNT_FIELD in comment.action_counts) || comment.action_counts[ACTION_COUNT_FIELD] !== count) {
+
+            // Batch updates for those changes.
+          commentOperations.push({
+            [`action_counts.${ACTION_COUNT_FIELD}`]: count,
+          });
+        }
+      });
+
+      // If this comment has action summaries that should be updated, then
+      // perform an update!
+      if (commentOperations.length > 0) {
+        operations.push({
+          updateOne: {
+            filter: {
+              id: comment.id
+            },
+            update: {
+              $set: Object.assign({}, ...commentOperations),
+            },
+          },
+        });
+      }
+    }
+
+    console.log(`Processed batch of ${comments.length} comments.`);
+
+    offset += batch;
+  }
+
+  const OPERATIONS_LENGTH = operations.length;
+
+  console.log(`Processed all ${totalCount} comments.`);
+  console.log(`${OPERATIONS_LENGTH} documents need fixing.`);
+
+  // If fix was enabled, execute the batch writes.
+  if (OPERATIONS_LENGTH > 0) {
+    if (fix) {
+      console.log(`Fixing ${OPERATIONS_LENGTH} documents...`);
+
+      while (operations.length) {
+        let batchOperations = operations.splice(0, batch);
+        let result = await CommentModel.collection.bulkWrite(batchOperations);
+
+        console.log(`Fixed batch of ${result.modifiedCount} documents.`);
+      }
+
+      console.log(`Fixed all ${OPERATIONS_LENGTH} documents.`);
+    } else {
+      console.warn('Skipping fixing, --fix was not enabled, pass --fix to fix these errors');
+    }
+  }
+};

bin/verifications/database/index.js

@@ -0,0 +1,12 @@
+// This will import all the verifications that should be run by the:
+//
+//   cli verify database
+//
+// command. They exist in the form:
+//
+//   async ({fix = false, batch = 1000}) => {}
+//
+// where their options are derrived.
+module.exports = [
+  require('./comments'),
+];

services/comments.js

@@ -4,6 +4,7 @@ const ActionModel = require('../models/action');
 const ActionsService = require('./actions');
 const SettingsService = require('./settings');
 
+const sc = require('snake-case');
 const errors = require('../errors');
 const events = require('./events');
 const {
@@ -300,25 +301,37 @@ module.exports = class CommentsService {
 // Event Hooks
 //==============================================================================
 
+const incrActionCounts = async (action, value) => {
+  const ACTION_TYPE = sc(action.action_type.toLowerCase());
+
+  const update = {
+    [`action_counts.${ACTION_TYPE}`]: value,
+  };
+
+  if (action.group_id !== null && action.group_id.length > 0) {
+    const GROUP_ID = sc(action.group_id.toLowerCase());
+
+    update[`action_counts.${ACTION_TYPE}_${GROUP_ID}`] = value;
+  }
+
+  try {
+    await CommentModel.update({
+      id: action.item_id,
+    }, {
+      $inc: update,
+    });
+  } catch (err) {
+    console.error(`Can't mutate the action_counts.${ACTION_TYPE}:`, err);
+  }
+};
+
 // When a new action is created, modify the comment.
 events.on(ACTIONS_NEW, async (action) => {
   if (!action || action.item_type !== 'COMMENTS') {
     return;
   }
 
-  const ACTION_TYPE = action.action_type.toLowerCase();
-
-  try {
-    await CommentModel.update({
-      id: action.item_id,
-    }, {
-      $inc: {
-        [`action_counts.${ACTION_TYPE}`]: 1,
-      }
-    });
-  } catch (err) {
-    console.error(`Can't increment the action_counts.${ACTION_TYPE}:`, err);
-  }
+  return incrActionCounts(action, 1);
 });
 
 // When an action is deleted, modify the comment.
@@ -327,17 +340,5 @@ events.on(ACTIONS_DELETE, async (action) => {
     return;
   }
 
-  const ACTION_TYPE = action.action_type.toLowerCase();
-
-  try {
-    await CommentModel.update({
-      id: action.item_id,
-    }, {
-      $inc: {
-        [`action_counts.${ACTION_TYPE}`]: -1,
-      }
-    });
-  } catch (err) {
-    console.error(`Can't decrement the action_counts.${ACTION_TYPE}:`, err);
-  }
+  return incrActionCounts(action, -1);
 });