From a9abd558958a587441da38716ce2a47e31203b08 Mon Sep 17 00:00:00 2001 From: Wyatt Johnson Date: Thu, 17 Aug 2017 15:32:47 -0600 Subject: [PATCH] added group_id support to action counts, cli verifier --- bin/cli | 1 + bin/cli-verify | 48 +++++++++ bin/verifications/database/comments.js | 141 +++++++++++++++++++++++++ bin/verifications/database/index.js | 12 +++ services/comments.js | 53 +++++----- 5 files changed, 229 insertions(+), 26 deletions(-) create mode 100755 bin/cli-verify create mode 100644 bin/verifications/database/comments.js create mode 100644 bin/verifications/database/index.js diff --git a/bin/cli b/bin/cli index 52d67c9af..4eee09f70 100755 --- a/bin/cli +++ b/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' diff --git a/bin/cli-verify b/bin/cli-verify new file mode 100755 index 000000000..be4420a7b --- /dev/null +++ b/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(); +} diff --git a/bin/verifications/database/comments.js b/bin/verifications/database/comments.js new file mode 100644 index 000000000..1f9a81951 --- /dev/null +++ b/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'); + } + } +}; diff --git a/bin/verifications/database/index.js b/bin/verifications/database/index.js new file mode 100644 index 000000000..3dafeb3bf --- /dev/null +++ b/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'), +]; diff --git a/services/comments.js b/services/comments.js index 4c57907e4..8db14966b 100644 --- a/services/comments.js +++ b/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); });