extended verify tool with reply count fixing

bin/cli-verify

@@ -14,10 +14,10 @@ util.onshutdown([
   () => mongoose.disconnect()
 ]);
 
-async function database(program) {
+async function database({fix = false, limit = Infinity, batch = 1000}) {
   try {
     for (const verification of databaseVerifications) {
-      await verification(program);
+      await verification({fix, limit, batch});
     }
   } catch (err) {
     console.error(`Failed to process all the ${databaseVerifications.length} verifications`, err);
@@ -36,6 +36,7 @@ program
   .command('db')
   .description('verifies the database integrity')
   .option('-f, --fix', 'fix the problems found with database inconsistencies')
+  .option('-l, --limit [size]', 'limit the amount of documents to process in a single pass, this will ensure only a maximum number of batch operations are issued [default: inf]', parseInt)
   .option('-b, --batch [size]', 'batch size to process verifications and repairs of documents [default: 1000]', parseInt)
   .action(database);
 

bin/verifications/database/comments.js

@@ -1,16 +1,17 @@
 const CommentModel = require('../../../models/comment');
 const ActionsService = require('../../../services/actions');
-const {arrayJoinBy} = require('../../../graph/loaders/util');
+const {arrayJoinBy, singleJoinBy} = require('../../../graph/loaders/util');
 const sc = require('snake-case');
+const debug = require('debug')('talk:cli:verify');
 
 const getBatch = async (limit, offset) => CommentModel
   .find({})
-  .select({'id': 1, 'action_counts': 1})
+  .select({'id': 1, 'action_counts': 1, 'reply_count': 1})
   .limit(limit)
   .skip(offset)
   .sort('created_at');
 
-module.exports = async ({fix = false, batch = 1000}) => {
+module.exports = async ({fix, limit, batch}) => {
   let operations = [];
 
   // Count how many comments there are to process.
@@ -20,7 +21,7 @@ module.exports = async ({fix = false, batch = 1000}) => {
   let comments = [];
   let commentIDs = [];
 
-  console.log(`Processing ${totalCount} comments...`);
+  debug(`Processing ${totalCount} comments...`);
 
   // Keep processing documents until there are is none left.
   while (offset < totalCount) {
@@ -29,6 +30,31 @@ module.exports = async ({fix = false, batch = 1000}) => {
     comments = await getBatch(batch, offset);
     commentIDs = comments.map(({id}) => id);
 
+    // Get their reply counts.
+    let allReplyCounts = await CommentModel
+      .aggregate([
+        {
+          $match: {
+            parent_id: {
+              $in: commentIDs,
+            },
+            status: {
+              $in: ['NONE', 'ACCEPTED']
+            }
+          }
+        },
+        {
+          $group: {
+            _id: '$parent_id',
+            count: {
+              $sum: 1
+            }
+          }
+        }
+      ])
+      .then(singleJoinBy(commentIDs, '_id'))
+      .then((results) => results.map((result) => result ? result.count : 0));
+
     // Get their action summaries.
     let allActionSummaries = await ActionsService
       .getActionSummaries(commentIDs)
@@ -38,17 +64,26 @@ module.exports = async ({fix = false, batch = 1000}) => {
     for (let i = 0; i < comments.length; i++) {
       let comment = comments[i];
       let actionSummaries = allActionSummaries[i];
+      let replyCount = allReplyCounts[i];
 
       // And check to see if the action summaries we just computed match what is
       // currently set for the comments.
       let commentOperations = [];
 
+      // If the reply count needs to be updated, then update it!
+      if (comment.reply_count !== replyCount) {
+        commentOperations.push({
+          reply_count: replyCount,
+        });
+      }
+
       // First we process all the group id's.
       for (let actionSummary of actionSummaries) {
         if (actionSummary.group_id === null) {
           continue;
         }
 
+        // And we generate the group id.
         const ACTION_TYPE = sc(actionSummary.action_type.toLowerCase());
         const GROUP_ID = sc(actionSummary.group_id.toLowerCase());
 
@@ -56,6 +91,8 @@ module.exports = async ({fix = false, batch = 1000}) => {
           continue;
         }
 
+        // And we add a new batch operation if the action summary is associated
+        // with a group.
         const ACTION_COUNT_FIELD = `${ACTION_TYPE}_${GROUP_ID}`;
 
         // Check that the action summaries match the cached counts.
@@ -82,7 +119,7 @@ module.exports = async ({fix = false, batch = 1000}) => {
         return acc;
       }, {});
 
-      Object.keys(groupedActionSummaries).forEach((ACTION_COUNT_FIELD) => {
+      for (const ACTION_COUNT_FIELD of Object.keys(groupedActionSummaries)) {
         const count = groupedActionSummaries[ACTION_COUNT_FIELD];
 
         // Check that the action summaries match the cached counts.
@@ -93,7 +130,7 @@ module.exports = async ({fix = false, batch = 1000}) => {
             [`action_counts.${ACTION_COUNT_FIELD}`]: count,
           });
         }
-      });
+      }
 
       // If this comment has action summaries that should be updated, then
       // perform an update!
@@ -111,29 +148,44 @@ module.exports = async ({fix = false, batch = 1000}) => {
       }
     }
 
-    console.log(`Processed batch of ${comments.length} comments.`);
+    debug(`Processed batch of ${comments.length} comments.`);
+
+    if (operations.length >= limit) {
+      debug(`Queued operations are ${operations.length}, reached limit of ${limit}, not processing any more.`);
+
+      if (operations.length > limit) {
+        debug(`${operations.length - limit} operations have been truncated to enforce the limit`);
+      }
+
+      break;
+    }
 
     offset += batch;
   }
 
   const OPERATIONS_LENGTH = operations.length;
 
-  console.log(`Processed all ${totalCount} comments.`);
-  console.log(`${OPERATIONS_LENGTH} documents need fixing.`);
+  if (limit < Infinity && offset + comments.length < totalCount) {
+    debug(`Processed ${offset + comments.length}/${totalCount} comments because we reached the update limit of ${limit}.`);
+    debug(`Fixing ${OPERATIONS_LENGTH} documents.`);
+  } else {
+    debug(`Processed all ${totalCount} comments.`);
+    debug(`${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...`);
+      debug(`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.`);
+        debug(`Fixed batch of ${result.modifiedCount} documents.`);
       }
 
-      console.log(`Fixed all ${OPERATIONS_LENGTH} documents.`);
+      debug(`Applied all ${OPERATIONS_LENGTH} fixes.`);
     } else {
       console.warn('Skipping fixing, --fix was not enabled, pass --fix to fix these errors');
     }

models/comment.js

@@ -71,7 +71,10 @@ const CommentSchema = new Schema({
   parent_id: String,
 
   // The number of replies to this comment directly.
-  reply_count: Number,
+  reply_count: {
+    type: Number,
+    default: 0,
+  },
 
   // Counts to store related to actions taken on the given comment.
   action_counts: {