wassname/talk
1
0
Fork 0
mirror of https://github.com/wassname/talk.git synced 2026-07-31 12:50:48 +08:00
Code Issues Packages Projects Releases Wiki Activity

added action counts for users

Browse Source
This commit is contained in:
Wyatt Johnson
2017-11-13 15:46:49 -07:00
parent 401d8a423d
commit 9f823c0b11
7 changed files with 278 additions and 129 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bin/verifications/database/action_counts.js
+185
View File
@@ -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');
}
}
}
};
bin/verifications/database/comments.js → bin/verifications/database/comment_replies.js
+1 -64
View File
@@ -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) {
bin/verifications/database/index.js
+3 -2
View File
@@ -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'),
];
graph/loaders/users.js
+25 -16
View File
@@ -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
models/user.js
+6
View File
@@ -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],
services/actions.js
+58 -2
View File
@@ -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);
});
services/comments.js
-45
View File
@@ -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({