From 9f823c0b11f8b2ed0c9da2506cb45ce09329244c Mon Sep 17 00:00:00 2001 From: Wyatt Johnson Date: Mon, 13 Nov 2017 15:46:49 -0700 Subject: [PATCH] added action counts for users --- bin/verifications/database/action_counts.js | 185 ++++++++++++++++++ .../{comments.js => comment_replies.js} | 65 +----- bin/verifications/database/index.js | 5 +- graph/loaders/users.js | 41 ++-- models/user.js | 6 + services/actions.js | 60 +++++- services/comments.js | 45 ----- 7 files changed, 278 insertions(+), 129 deletions(-) create mode 100644 bin/verifications/database/action_counts.js rename bin/verifications/database/{comments.js => comment_replies.js} (61%) diff --git a/bin/verifications/database/action_counts.js b/bin/verifications/database/action_counts.js new file mode 100644 index 000000000..fff3a0569 --- /dev/null +++ b/bin/verifications/database/action_counts.js @@ -0,0 +1,185 @@ +const UserModel = require('../../../models/user'); +const CommentModel = require('../../../models/comment'); +const ActionsService = require('../../../services/actions'); +const {arrayJoinBy} = require('../../../graph/loaders/util'); +const sc = require('snake-case'); +const debug = require('debug')('talk:cli:verify'); + +const MODELS = [ + UserModel, + CommentModel, +]; + +async function processBatch(Model, documents) { + + // Get an array of all the document id's. + const documentIDs = documents.map(({id}) => id); + + // Store all the operations on this batch in this array that we'll return + // later. + const operations = []; + + // Get the action summaries for this batch. + const totalActionSummaries = await ActionsService + .getActionSummaries(documentIDs) + .then(arrayJoinBy(documentIDs, 'item_id')); + + // Iterate over the documents. + for (let i = 0; i < documents.length; i++) { + const document = documents[i]; + const actionSummaries = totalActionSummaries[i]; + + let ops = []; + + for (const 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()); + + if (GROUP_ID.length <= 0) { + 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. + if ( + !document.action_counts || + !(ACTION_COUNT_FIELD in document.action_counts) || + document.action_counts[ACTION_COUNT_FIELD] !== actionSummary.count + ) { + + // Batch updates for those changes. + ops.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; + }, {}); + + for (const ACTION_COUNT_FIELD of Object.keys(groupedActionSummaries)) { + const count = groupedActionSummaries[ACTION_COUNT_FIELD]; + + // Check that the action summaries match the cached counts. + if ( + !document.action_counts || + !(ACTION_COUNT_FIELD in document.action_counts) || + document.action_counts[ACTION_COUNT_FIELD] !== count + ) { + + // Batch updates for those changes. + ops.push({ + [`action_counts.${ACTION_COUNT_FIELD}`]: count, + }); + } + } + + // If this comment has action summaries that should be updated, then + // perform an update! + if (ops.length > 0) { + operations.push({ + updateOne: { + filter: { + id: document.id + }, + update: { + $set: Object.assign({}, ...ops), + }, + }, + }); + } + } + + return operations; +} + +module.exports = async ({fix, batch}) => { + for (const Model of MODELS) { + const cursor = Model + .collection + .find({}) + .project({ + id: 1, + action_counts: 1 + }) + .sort({created_at: 1}); + + let operations = []; + let documents = []; + + // While there are documents to process. + while (await cursor.hasNext()) { + + // Load the document. + const document = await cursor.next(); + + // Push the document into the documents array. + documents.push(document); + + // Check to see if the length of the documents array requires us to + // process it. + if (documents.length > batch) { + + // Process this batch. + let batchOperations = await processBatch(Model, documents); + + // Push the batch operations into the model operations. + operations.push(...batchOperations); + + // Clear this batch contents. + documents = []; + } + } + + // Check to see if there are any documents left over. + if (documents.length > 0) { + + // Process this batch. + let batchOperations = await processBatch(Model, documents); + + // Push the batch operations into the model operations. + operations.push(...batchOperations); + } + + const OPERATIONS_LENGTH = operations.length; + + console.log(`action_counts.js: ${OPERATIONS_LENGTH} ${Model.collection.name} need their action counts fixed.`); + + // If fix was enabled, execute the batch writes. + if (OPERATIONS_LENGTH > 0) { + if (fix) { + debug(`action_counts.js: fixing ${OPERATIONS_LENGTH} ${Model.collection.name}...`); + + while (operations.length) { + let result = await Model.collection.bulkWrite(operations.splice(0, batch)); + + debug(`action_counts.js: fixed batch of ${result.modifiedCount} ${Model.collection.name}.`); + } + + console.log(`action_counts.js: applied all ${OPERATIONS_LENGTH} fixes to ${Model.collection.name}.`); + } else { + console.warn('Skipping fixing, --fix was not enabled, pass --fix to fix these errors'); + } + } + } +}; + diff --git a/bin/verifications/database/comments.js b/bin/verifications/database/comment_replies.js similarity index 61% rename from bin/verifications/database/comments.js rename to bin/verifications/database/comment_replies.js index 231badabb..03a3a4ac6 100644 --- a/bin/verifications/database/comments.js +++ b/bin/verifications/database/comment_replies.js @@ -1,7 +1,5 @@ const CommentModel = require('../../../models/comment'); -const ActionsService = require('../../../services/actions'); -const {arrayJoinBy, singleJoinBy} = require('../../../graph/loaders/util'); -const sc = require('snake-case'); +const {singleJoinBy} = require('../../../graph/loaders/util'); const debug = require('debug')('talk:cli:verify'); const getBatch = async (limit, offset) => CommentModel @@ -55,15 +53,9 @@ module.exports = async ({fix, limit, batch}) => { .then(singleJoinBy(commentIDs, '_id')) .then((results) => results.map((result) => result ? result.count : 0)); - // 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]; let replyCount = allReplyCounts[i]; // And check to see if the action summaries we just computed match what is @@ -77,61 +69,6 @@ module.exports = async ({fix, limit, batch}) => { }); } - // 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()); - - if (GROUP_ID.length <= 0) { - 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. - 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; - }, {}); - - for (const ACTION_COUNT_FIELD of Object.keys(groupedActionSummaries)) { - 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) { diff --git a/bin/verifications/database/index.js b/bin/verifications/database/index.js index 3dafeb3bf..2a599cfac 100644 --- a/bin/verifications/database/index.js +++ b/bin/verifications/database/index.js @@ -6,7 +6,8 @@ // // async ({fix = false, batch = 1000}) => {} // -// where their options are derrived. +// where their options are derived. module.exports = [ - require('./comments'), + require('./comment_replies'), + require('./action_counts'), ]; diff --git a/graph/loaders/users.js b/graph/loaders/users.js index 273db00f3..665690202 100644 --- a/graph/loaders/users.js +++ b/graph/loaders/users.js @@ -1,7 +1,7 @@ const DataLoader = require('dataloader'); const util = require('./util'); -const union = require('lodash/union'); +const sc = require('snake-case'); const { SEARCH_OTHER_USERS, @@ -72,7 +72,7 @@ const genUserByIDs = async (context, ids) => { * @param {Object} context graph context * @param {Object} query query terms to apply to the users query */ -const getUsersByQuery = async ({user, loaders: {Actions}}, {ids, limit, cursor, state, action_type, sortOrder}) => { +const getUsersByQuery = async ({user}, {ids, limit, cursor, state, action_type, sortOrder}) => { let query = UserModel.find(); if (action_type || state) { @@ -82,9 +82,14 @@ const getUsersByQuery = async ({user, loaders: {Actions}}, {ids, limit, cursor, if (state) { mergeState(query, state); - } else { - const userIds = await Actions.getByTypes({action_type, item_type: 'USERS'}); - ids = ids ? union(ids, userIds) : userIds; + } + + if (action_type) { + query.merge({ + [`action_counts.${sc(action_type.toLowerCase())}`]: { + $gt: 0 + } + }); } } @@ -152,21 +157,25 @@ const getUsersByQuery = async ({user, loaders: {Actions}}, {ids, limit, cursor, * @return {Promise} resolves to the counts of the users from the * query */ -const getCountByQuery = async ({loaders: {Actions}}, {action_type, state}) => { +const getCountByQuery = async ({user}, {action_type, state}) => { let query = UserModel.find(); - if (action_type) { - const userIds = await Actions.getByTypes({action_type, item_type: 'USERS'}); + if (action_type || state) { + if (!user || !user.can(SEARCH_OTHER_USERS)) { + return null; + } - query.merge({ - id: { - $in: userIds - } - }); - } + if (state) { + mergeState(query, state); + } - if (state) { - mergeState(query, state); + if (action_type) { + query.merge({ + [`action_counts.${sc(action_type.toLowerCase())}`]: { + $gt: 0 + } + }); + } } return UserModel diff --git a/models/user.js b/models/user.js index a50cffffb..5d8a41cc7 100644 --- a/models/user.js +++ b/models/user.js @@ -174,6 +174,12 @@ const UserSchema = new Schema({ // IgnoresUsers is an array of user id's that the current user is ignoring. ignoresUsers: [String], + // Counts to store related to actions taken on the given user. + action_counts: { + default: {}, + type: Object, + }, + // Tags are added by the self or by administrators. tags: [TagLinkSchema], diff --git a/services/actions.js b/services/actions.js index ba9f1fde5..adffdd134 100644 --- a/services/actions.js +++ b/services/actions.js @@ -1,4 +1,7 @@ const ActionModel = require('../models/action'); +const CommentModel = require('../models/comment'); +const UserModel = require('../models/user'); +const sc = require('snake-case'); const _ = require('lodash'); const errors = require('../errors'); const events = require('./events'); @@ -143,7 +146,7 @@ module.exports = class ActionsService { let $group = { // group unique documents by these properties, we are leveraging the - // fact that each uuid is completly unique. + // fact that each uuid is completely unique. _id: { item_id: '$item_id', action_type: '$action_type', @@ -155,7 +158,7 @@ module.exports = class ActionsService { $sum: 1 }, - // we are leveraging the fact that each uuid is completly unique and + // we are leveraging the fact that each uuid is completely unique and // just grabbing the last instance of the item type here. item_type: { $first: '$item_type' @@ -248,3 +251,56 @@ module.exports = class ActionsService { }, 'item_id'); } }; + +const incrActionCounts = async (action, value) => { + const ACTION_TYPE = sc(action.action_type.toLowerCase()); + + const update = { + [`action_counts.${ACTION_TYPE}`]: value, + }; + + if (action.group_id && action.group_id.length > 0) { + const GROUP_ID = sc(action.group_id.toLowerCase()); + + update[`action_counts.${ACTION_TYPE}_${GROUP_ID}`] = value; + } + + try { + switch (action.item_type) { + case 'USERS': + return UserModel.update({ + id: action.item_id, + }, { + $inc: update, + }); + case 'COMMENTS': + return CommentModel.update({ + id: action.item_id, + }, { + $inc: update, + }); + default: + throw new Error('Invalid item type for action summary monitoring'); + } + } 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' && action.item_type !== 'USERS')) { + return; + } + + return incrActionCounts(action, 1); +}); + +// When an action is deleted, remove the action count on the comment. +events.on(ACTIONS_DELETE, async (action) => { + if (!action || (action.item_type !== 'COMMENTS' && action.item_type !== 'USERS')) { + return; + } + + return incrActionCounts(action, -1); +}); diff --git a/services/comments.js b/services/comments.js index ec55cfb4d..086c228aa 100644 --- a/services/comments.js +++ b/services/comments.js @@ -4,13 +4,10 @@ const debug = require('debug')('talk:services:comments'); const ActionsService = require('./actions'); const SettingsService = require('./settings'); -const sc = require('snake-case'); const cloneDeep = require('lodash/cloneDeep'); const errors = require('../errors'); const events = require('./events'); const { - ACTIONS_NEW, - ACTIONS_DELETE, COMMENTS_NEW, COMMENTS_EDIT, } = require('./events/constants'); @@ -335,48 +332,6 @@ 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 && 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; - } - - return incrActionCounts(action, 1); -}); - -// When an action is deleted, remove the action count on the comment. -events.on(ACTIONS_DELETE, async (action) => { - if (!action || action.item_type !== 'COMMENTS') { - return; - } - - return incrActionCounts(action, -1); -}); - const incrReplyCount = async (comment, value) => { try { await CommentModel.update({