Merge branch 'master' into permalink-enh

.eslintignore

@@ -2,35 +2,4 @@
 dist
 docs
 node_modules
-plugins/*
 public
-
-!plugins/talk-plugin-akismet
-!plugins/talk-plugin-auth
-!plugins/talk-plugin-author-menu
-!plugins/talk-plugin-comment-content
-!plugins/talk-plugin-deep-reply-count
-!plugins/talk-plugin-facebook-auth
-!plugins/talk-plugin-featured-comments
-!plugins/talk-plugin-flag-details
-!plugins/talk-plugin-google-auth
-!plugins/talk-plugin-ignore-user
-!plugins/talk-plugin-like
-!plugins/talk-plugin-love
-!plugins/talk-plugin-member-since
-!plugins/talk-plugin-mod
-!plugins/talk-plugin-moderation-actions
-!plugins/talk-plugin-offtopic
-!plugins/talk-plugin-permalink
-!plugins/talk-plugin-profile-settings
-!plugins/talk-plugin-remember-sort
-!plugins/talk-plugin-respect
-!plugins/talk-plugin-sort-most-liked
-!plugins/talk-plugin-sort-most-loved
-!plugins/talk-plugin-sort-most-replied
-!plugins/talk-plugin-sort-most-respected
-!plugins/talk-plugin-sort-newest
-!plugins/talk-plugin-sort-oldest
-!plugins/talk-plugin-subscriber
-!plugins/talk-plugin-toxic-comments
-!plugins/talk-plugin-viewing-options

.gitignore

@@ -23,37 +23,39 @@ browserstack.err
 
 plugins.json
 plugins/*
+
 !plugins/talk-plugin-akismet
-!plugins/talk-plugin-facebook-auth
-!plugins/talk-plugin-google-auth
 !plugins/talk-plugin-auth
-!plugins/talk-plugin-respect
-!plugins/talk-plugin-offtopic
-!plugins/talk-plugin-like
-!plugins/talk-plugin-mod
-!plugins/talk-plugin-love
-!plugins/talk-plugin-viewing-options
+!plugins/talk-plugin-author-menu
 !plugins/talk-plugin-comment-content
-!plugins/talk-plugin-permalink
+!plugins/talk-plugin-deep-reply-count
+!plugins/talk-plugin-facebook-auth
 !plugins/talk-plugin-featured-comments
-!plugins/talk-plugin-toxic-comments
-!plugins/talk-plugin-sort-newest
-!plugins/talk-plugin-sort-oldest
-!plugins/talk-plugin-sort-most-replied
+!plugins/talk-plugin-flag-details
+!plugins/talk-plugin-google-auth
+!plugins/talk-plugin-ignore-user
+!plugins/talk-plugin-like
+!plugins/talk-plugin-love
+!plugins/talk-plugin-member-since
+!plugins/talk-plugin-mod
+!plugins/talk-plugin-moderation-actions
+!plugins/talk-plugin-notifications
+!plugins/talk-plugin-notifications-category-reply
+!plugins/talk-plugin-offtopic
+!plugins/talk-plugin-permalink
+!plugins/talk-plugin-profile-settings
+!plugins/talk-plugin-remember-sort
+!plugins/talk-plugin-respect
+!plugins/talk-plugin-slack-notifications
 !plugins/talk-plugin-sort-most-liked
 !plugins/talk-plugin-sort-most-loved
+!plugins/talk-plugin-sort-most-replied
 !plugins/talk-plugin-sort-most-respected
-!plugins/talk-plugin-author-menu
-!plugins/talk-plugin-member-since
-!plugins/talk-plugin-ignore-user
-!plugins/talk-plugin-moderation-actions
-!plugins/talk-plugin-toxic-comments
-!plugins/talk-plugin-remember-sort
-!plugins/talk-plugin-deep-reply-count
+!plugins/talk-plugin-sort-newest
+!plugins/talk-plugin-sort-oldest
 !plugins/talk-plugin-subscriber
-!plugins/talk-plugin-flag-details
-!plugins/talk-plugin-slack-notifications
-!plugins/talk-plugin-profile-settings
+!plugins/talk-plugin-toxic-comments
+!plugins/talk-plugin-viewing-options
 
 **/node_modules/*
 yarn-error.log

app.js

@@ -1,7 +1,6 @@
 const express = require('express');
 const morgan = require('morgan');
 const path = require('path');
-const uuid = require('uuid');
 const merge = require('lodash/merge');
 const helmet = require('helmet');
 const plugins = require('./services/plugins');
@@ -13,13 +12,6 @@ const { ENABLE_TRACING, APOLLO_ENGINE_KEY, PORT } = require('./config');
 
 const app = express();
 
-// Request Identity Middleware
-app.use((req, res, next) => {
-  req.id = uuid.v4();
-
-  next();
-});
-
 //==============================================================================
 // PLUGIN PRE APPLICATION MIDDLEWARE
 //==============================================================================

bin/cli-jobs

@@ -6,8 +6,7 @@
 
 const util = require('./util');
 const program = require('commander');
-const scraper = require('../services/scraper');
-const mailer = require('../services/mailer');
+const jobs = require('../jobs');
 const mongoose = require('../services/mongoose');
 const kue = require('../services/kue');
 
@@ -21,11 +20,8 @@ function processJobs() {
   // started.
   util.onshutdown([() => kue.Task.shutdown()]);
 
-  // Start the scraper processor.
-  scraper.process();
-
-  // Start the mail processor.
-  mailer.process();
+  // Start the jobs processor.
+  jobs.process();
 }
 
 /**

client/coral-framework/lib/action.js

@@ -0,0 +1,21 @@
+export default class Action {
+  listeners = [];
+
+  listen = cb => {
+    this.listeners.push(cb);
+  };
+
+  unlisten = cb => {
+    this.listeners = this.listeners.filter(i => i !== cb);
+  };
+
+  event = { listen: this.listen, unlisten: this.unlisten };
+
+  call(...args) {
+    this.listeners.forEach(cb => cb(...args));
+  }
+
+  asEvent() {
+    return this.event;
+  }
+}

client/coral-framework/reducers/config.js

@@ -1,9 +1,15 @@
 import { MERGE_CONFIG } from '../constants/config';
+import { LOGOUT } from '../constants/auth';
 
 const initialState = {};
 
 export default function config(state = initialState, action) {
   switch (action.type) {
+    case LOGOUT:
+      return {
+        ...state,
+        auth_token: null,
+      };
     case MERGE_CONFIG:
       return {
         ...state,

docs/_docs/04-03-additional-plugins.md

@@ -118,3 +118,34 @@ Configuration:
 
 - `TALK_AKISMET_API_KEY` (**required**) - The Akismet API key located on your account page.
 - `TALK_AKISMET_SITE` (**required**) - The URL where you are embedding the comment stream on to provide context to Akismet. If you're hosting talk on https://talk.mynews.org/, and your news site is https://mynews.org/, then you should set this parameter to `https://mynews.org/`
+
+## talk-plugin-notifications
+
+Source: [plugins/talk-plugin-notifications](https://github.com/coralproject/talk/tree/master/plugins/talk-plugin-notifications){:target="_blank"}
+
+Enables the Notification system for sending out enabled email notifications to
+users when they interact with Talk. By itself, this plugin will not send
+anything. You need to enable one of the `talk-plugin-notifications-category-*` plugins.
+
+**Note that all `talk-plugin-notifications-*` plugins must be registered
+*before* this plugin in order to work. For example:**
+
+```js
+{
+  "server": [
+    // ...
+    "talk-plugin-notifications-category-reply",
+    "talk-plugin-notifications",
+    // ...
+  ]
+}
+```
+{:.no-copy}
+
+### talk-plugin-notifications-category-reply
+{:.param}
+
+Source: [plugins/talk-plugin-notifications-category-reply](https://github.com/coralproject/talk/tree/master/plugins/talk-plugin-notifications-category-reply){:target="_blank"}
+
+Replies made to each user will trigger an email to be sent with the notification
+details if enabled.

docs/_sass/talk.scss

@@ -264,7 +264,7 @@ pre {
 .toc {
   a {
     @extend .coral-link;
-    border-bottom: none;
+    border-bottom: none !important;
   }
 }
 

graph/connectors.js

@@ -1,9 +1,23 @@
 const debug = require('debug')('talk:graph:connectors');
 const merge = require('lodash/merge');
 
+// Config.
+const config = require('../config');
+
+// Secrets.
+const secrets = require('../secrets');
+
 // Errors.
 const errors = require('../errors');
 
+// Graph.
+const { getBroker } = require('./subscriptions/broker');
+const { getPubsub } = require('./subscriptions/pubsub');
+const resolvers = require('./resolvers');
+const mutators = require('./mutators');
+const loaders = require('./loaders');
+const schema = require('./schema');
+
 // Models.
 const Action = require('../models/action');
 const Asset = require('../models/asset');
@@ -28,7 +42,6 @@ const Moderation = require('../services/moderation');
 const Mongoose = require('../services/mongoose');
 const Passport = require('../services/passport');
 const Plugins = require('../services/plugins');
-const Pubsub = require('../services/pubsub');
 const Redis = require('../services/redis');
 const Regex = require('../services/regex');
 const Scraper = require('../services/scraper');
@@ -40,9 +53,11 @@ const Tokens = require('../services/tokens');
 const Users = require('../services/users');
 const Wordlist = require('../services/wordlist');
 
-// Connector.
-const connectors = {
+// Connectors.
+const defaultConnectors = {
   errors,
+  config,
+  secrets,
   models: {
     Action,
     Asset,
@@ -67,7 +82,6 @@ const connectors = {
     Mongoose,
     Passport,
     Plugins,
-    Pubsub,
     Redis,
     Regex,
     Scraper,
@@ -79,14 +93,23 @@ const connectors = {
     Users,
     Wordlist,
   },
+  graph: {
+    subscriptions: { getBroker, getPubsub },
+    resolvers,
+    mutators,
+    loaders,
+    schema,
+  },
 };
 
-module.exports = Plugins.get('server', 'connectors').reduce(
+const connectors = Plugins.get('server', 'connectors').reduce(
   (defaultConnectors, { plugin, connectors: pluginConnectors }) => {
     debug(`adding plugin '${plugin.name}'`);
 
     // Merge in the plugin connectors.
     return merge(defaultConnectors, pluginConnectors);
   },
-  connectors
+  defaultConnectors
 );
+
+module.exports = connectors;

graph/context.js

@@ -1,12 +1,13 @@
 const loaders = require('./loaders');
 const mutators = require('./mutators');
-const uuid = require('uuid');
-const merge = require('lodash/merge');
+const uuid = require('uuid/v4');
 const connectors = require('./connectors');
-
+const { get, merge } = require('lodash');
 const plugins = require('../services/plugins');
-const pubsub = require('../services/pubsub');
+const { getBroker } = require('./subscriptions/broker');
 const debug = require('debug')('talk:graph:context');
+const { createLogger } = require('../services/logging');
+const { graphql } = require('graphql');
 
 /**
  * Contains the array of plugins that provide context to the server, these top
@@ -45,15 +46,16 @@ const decorateContextPlugins = (context, contextPlugins) => {
  * Stores the request context.
  */
 class Context {
-  constructor(parent) {
+  constructor(ctx) {
     // Generate a new context id for the request if the parent doesn't provide
     // one.
-    this.id = parent.id || uuid.v4();
+    this.id = ctx.id || uuid.v4();
+
+    // Attach a logger or create one.
+    this.log = ctx.log || createLogger('context', this.id);
 
     // Load the current logged in user to `user`, otherwise this will be null.
-    if (parent.user) {
-      this.user = parent.user;
-    }
+    this.user = get(ctx, 'user');
 
     // Attach the connectors.
     this.connectors = connectors;
@@ -68,14 +70,48 @@ class Context {
     this.plugins = decorateContextPlugins(this, contextPlugins);
 
     // Bind the publish/subscribe to the context.
-    this.pubsub = pubsub.getClient();
+    this.pubsub = getBroker();
 
     // Bind the parent context.
-    this.parent = parent;
+    this.parent = ctx;
   }
 
   /**
+   * graphql will execute a graph request for the current context.
    *
+   * @param {String} requestString  A GraphQL language formatted string
+   *    representing the requested operation.
+   * @param {Object} variableValues  A mapping of variable name to runtime value
+   *    to use for all variables defined in the requestString.
+   * @param {Object} rootValue The value provided as the first argument to
+   *    resolver functions on the top level type (e.g. the query object type).
+   * @param {String} operationName The name of the operation to use if
+   *    requestString contains multiple possible operations. Can be omitted if
+   *    requestString contains only one operation.
+   * @returns {Promise}
+   */
+  async graphql(
+    requestString,
+    variableValues = {},
+    rootValue = {},
+    operationName = undefined
+  ) {
+    // Perform the graph request directly using the graphql client.
+    return graphql(
+      // Use the connected graph schema.
+      this.connectors.graph.schema,
+      requestString,
+      rootValue,
+      // Use this, the context as the context.
+      this,
+      variableValues,
+      operationName
+    );
+  }
+
+  /**
+   * forSystem returns a system context object that can be used for internal
+   * operations.
    */
   static forSystem() {
     const { models: { User } } = connectors;
@@ -87,4 +123,15 @@ class Context {
   }
 }
 
+// Attach the Context to the connectors.
+connectors.graph.Context = Context;
+
+// Connect the connect based plugins after the server has started.
+plugins.defer('server', 'connect', ({ plugin, connect }) => {
+  debug(`connecting plugin to connectors '${plugin.name}'`);
+
+  // Pass the connectors down to the connect plugin.
+  connect(connectors);
+});
+
 module.exports = Context;

graph/index.js

@@ -2,6 +2,7 @@ const schema = require('./schema');
 const Context = require('./context');
 const { createSubscriptionManager } = require('./subscriptions');
 const { ENABLE_TRACING } = require('../config');
+const connectors = require('./connectors');
 
 module.exports = {
   createGraphOptions: req => ({
@@ -10,11 +11,12 @@ module.exports = {
 
     // Load in the new context here, this will create the loaders + mutators for
     // the lifespan of this request.
-    context: new Context(req),
+    context: new Context(req.context),
 
     // Tracing request options, needed for Apollo Engine.
     tracing: ENABLE_TRACING,
     cacheControl: ENABLE_TRACING,
   }),
   createSubscriptionManager,
+  connectors,
 };

graph/loaders/users.js

@@ -1,12 +1,7 @@
 const DataLoader = require('dataloader');
-
 const util = require('./util');
-
 const { SEARCH_OTHER_USERS } = require('../../perms/constants');
-
-const UsersService = require('../../services/users');
 const { escapeRegExp } = require('../../services/regex');
-const UserModel = require('../../models/user');
 
 const mergeState = (query, state) => {
   const { status } = state;
@@ -51,17 +46,14 @@ const mergeState = (query, state) => {
   }
 };
 
-const genUserByIDs = async (context, ids) => {
+const genUserByIDs = async (ctx, ids) => {
   if (!ids || ids.length === 0) {
     return [];
   }
 
-  if (ids.length === 1) {
-    const user = await UsersService.findById(ids[0]);
-    return [user];
-  }
+  const { connectors: { models: { User } } } = ctx;
 
-  return UsersService.findByIdArray(ids).then(util.singleJoinBy(ids, 'id'));
+  return User.find({ id: { $in: ids } }).then(util.singleJoinBy(ids, 'id'));
 };
 
 /**
@@ -71,10 +63,10 @@ const genUserByIDs = async (context, ids) => {
  * @param  {Object} query     query terms to apply to the users query
  */
 const getUsersByQuery = async (
-  { user },
+  { user, connectors: { models: { User } } },
   { limit, cursor, value = '', state, action_type, sortOrder }
 ) => {
-  let query = UserModel.find();
+  let query = User.find();
 
   if (action_type || state || value.length > 0) {
     if (!user || !user.can(SEARCH_OTHER_USERS)) {
@@ -182,8 +174,11 @@ const getUsersByQuery = async (
  * @return {Promise}          resolves to the counts of the users from the
  *                            query
  */
-const getCountByQuery = async ({ user }, { action_type, state }) => {
-  let query = UserModel.find();
+const getCountByQuery = async (
+  { user, connectors: { models: { User } } },
+  { action_type, state }
+) => {
+  const query = User.find();
 
   if (action_type || state) {
     if (!user || !user.can(SEARCH_OTHER_USERS)) {
@@ -203,7 +198,7 @@ const getCountByQuery = async ({ user }, { action_type, state }) => {
     }
   }
 
-  return UserModel.find(query).count();
+  return query.count();
 };
 
 /**

graph/resolvers/action.js

@@ -1,4 +1,4 @@
-const { SEARCH_OTHER_USERS } = require('../../perms/constants');
+const { decorateUserField } = require('./util');
 
 const Action = {
   __resolveType({ action_type }) {
@@ -11,14 +11,8 @@ const Action = {
         return undefined;
     }
   },
-
-  // This will load the user for the specific action. We'll limit this to the
-  // admin users only or the current logged in user.
-  user({ user_id }, _, { loaders: { Users }, user }) {
-    if (user && (user.can(SEARCH_OTHER_USERS) || user_id === user.id)) {
-      return Users.getByID.load(user_id);
-    }
-  },
 };
 
+decorateUserField(Action, 'user', 'user_id');
+
 module.exports = Action;

graph/resolvers/comment.js

@@ -1,4 +1,6 @@
-const { decorateWithTags } = require('./util');
+const { property } = require('lodash');
+const { SEARCH_ACTIONS } = require('../../perms/constants');
+const { decorateWithTags, decorateWithPermissionCheck } = require('./util');
 
 const Comment = {
   hasParent({ parent_id }) {
@@ -29,15 +31,8 @@ const Comment = {
 
     return Comments.getByQuery(query);
   },
-  replyCount({ reply_count }) {
-    // A simple remap from the underlying database model to the graph model.
-    return reply_count;
-  },
-  actions({ id }, _, { user, loaders: { Actions } }) {
-    if (!user || !user.can('SEARCH_ACTIONS')) {
-      return null;
-    }
-
+  replyCount: property('reply_count'),
+  actions({ id }, _, { loaders: { Actions } }) {
     return Actions.getByID.load(id);
   },
   action_summaries(comment, _, { loaders: { Actions } }) {
@@ -65,4 +60,9 @@ const Comment = {
 // Decorate the Comment type resolver with a tags field.
 decorateWithTags(Comment);
 
+// Protect direct action access.
+decorateWithPermissionCheck(Comment, {
+  actions: [SEARCH_ACTIONS],
+});
+
 module.exports = Comment;

graph/resolvers/flag_action.js

@@ -1,18 +1,11 @@
-const FlagAction = {
-  // Stored in the metadata, extract and return.
-  message({ metadata: { message } }) {
-    return message;
-  },
-  reason({ group_id }) {
-    return group_id;
-  },
-  user({ user_id }, _, { loaders: { Users } }) {
-    if (!user_id) {
-      return null;
-    }
+const { decorateUserField } = require('./util');
+const { property } = require('lodash');
 
-    return Users.getByID.load(user_id);
-  },
+const FlagAction = {
+  message: property('metadata.message'),
+  reason: property('group_id'),
 };
 
+decorateUserField(FlagAction, 'user', 'user_id');
+
 module.exports = FlagAction;

graph/resolvers/flag_action_summary.js

@@ -1,7 +1,7 @@
+const { property } = require('lodash');
+
 const FlagActionSummary = {
-  reason({ group_id }) {
-    return group_id;
-  },
+  reason: property('group_id'),
 };
 
 module.exports = FlagActionSummary;

graph/resolvers/index.js

@@ -1,4 +1,4 @@
-const _ = require('lodash');
+const { merge } = require('lodash');
 const debug = require('debug')('talk:graph:resolvers');
 
 const Action = require('./action');
@@ -70,7 +70,7 @@ resolvers = plugins
   .reduce((acc, { plugin, resolvers }) => {
     debug(`added plugin '${plugin.name}'`);
 
-    return _.merge(acc, resolvers);
+    return merge(acc, resolvers);
   }, resolvers);
 
 module.exports = resolvers;

graph/resolvers/root_query.js

@@ -1,3 +1,4 @@
+const { decorateWithPermissionCheck } = require('./util');
 const {
   SEARCH_ASSETS,
   SEARCH_OTHERS_COMMENTS,
@@ -5,11 +6,7 @@ const {
 } = require('../../perms/constants');
 
 const RootQuery = {
-  assets(_, { query }, { loaders: { Assets }, user }) {
-    if (user == null || !user.can(SEARCH_ASSETS)) {
-      return null;
-    }
-
+  assets(_, { query }, { loaders: { Assets } }) {
     return Assets.getByQuery(query);
   },
   asset(_, query, { loaders: { Assets } }) {
@@ -33,11 +30,7 @@ const RootQuery = {
     return Comments.get.load(id);
   },
 
-  async commentCount(_, { query }, { user, loaders: { Comments, Assets } }) {
-    if (user == null || !user.can(SEARCH_OTHERS_COMMENTS)) {
-      return null;
-    }
-
+  async commentCount(_, { query }, { loaders: { Comments, Assets } }) {
     const { asset_url, asset_id } = query;
     if (
       (!asset_id || asset_id.length === 0) &&
@@ -53,11 +46,7 @@ const RootQuery = {
     return Comments.getCountByQuery(query);
   },
 
-  async userCount(_, { query }, { user, loaders: { Users } }) {
-    if (user == null || !user.can(SEARCH_OTHER_USERS)) {
-      return null;
-    }
-
+  async userCount(_, { query }, { loaders: { Users } }) {
     return Users.getCountByQuery(query);
   },
 
@@ -72,23 +61,40 @@ const RootQuery = {
   },
 
   // this returns an arbitrary user
-  user(_, { id }, { user, loaders: { Users } }) {
-    if (user == null || !user.can(SEARCH_OTHER_USERS)) {
-      return null;
-    }
-
+  user(_, { id }, { loaders: { Users } }) {
     return Users.getByID.load(id);
   },
 
   // This endpoint is used for loading the user moderation queues (users whose username has been flagged),
   // so hide it in the event that we aren't an admin.
-  users(_, { query }, { user, loaders: { Users } }) {
-    if (user == null || !user.can(SEARCH_OTHER_USERS)) {
-      return null;
-    }
-
+  users(_, { query }, { loaders: { Users } }) {
     return Users.getByQuery(query);
   },
 };
 
+// Protect some query fields that are privileged.
+decorateWithPermissionCheck(RootQuery, {
+  assets: [SEARCH_ASSETS],
+  users: [SEARCH_OTHER_USERS],
+  userCount: [SEARCH_OTHER_USERS],
+  commentCount: [SEARCH_OTHERS_COMMENTS],
+});
+
+// Protect the user field so only users who have permission to look up another
+// user may do so as well as a user looking up themselves.
+decorateWithPermissionCheck(
+  RootQuery,
+  {
+    user: [SEARCH_OTHER_USERS],
+  },
+  (obj, { id }, { user }) => {
+    if (user && user.id === id) {
+      return true;
+    }
+
+    // We don't return false because we want to fallthrough to the permission
+    // check if the custom check fails.
+  }
+);
+
 module.exports = RootQuery;

graph/resolvers/subscription.js

@@ -1,40 +1,23 @@
-const Subscription = {
-  commentAdded(comment) {
-    return comment;
-  },
-  commentEdited(comment) {
-    return comment;
-  },
-  commentAccepted(comment) {
-    return comment;
-  },
-  commentRejected(comment) {
-    return comment;
-  },
-  commentReset(comment) {
-    return comment;
-  },
-  commentFlagged(comment) {
-    return comment;
-  },
-  userBanned(user) {
-    return user;
-  },
-  userSuspended(user) {
-    return user;
-  },
-  usernameApproved(user) {
-    return user;
-  },
-  usernameRejected(user) {
-    return user;
-  },
-  usernameFlagged(user) {
-    return user;
-  },
-  usernameChanged(payload) {
-    return payload;
-  },
-};
+const Subscription = {};
+
+// All of the subscription endpoints need to have an object to serialize when
+// pushing out via PubSub, this simply ensures that all these entries will
+// return the root object from the subscription to the PubSub framework.
+[
+  'commentAdded',
+  'commentEdited',
+  'commentAccepted',
+  'commentRejected',
+  'commentReset',
+  'commentFlagged',
+  'userBanned',
+  'userSuspended',
+  'usernameApproved',
+  'usernameRejected',
+  'usernameFlagged',
+  'usernameChanged',
+].forEach(field => {
+  Subscription[field] = obj => obj;
+});
 
 module.exports = Subscription;

graph/resolvers/tag_link.js

@@ -1,11 +1,7 @@
-const { SEARCH_OTHER_USERS } = require('../../perms/constants');
+const { decorateUserField } = require('./util');
 
-const TagLink = {
-  assigned_by({ assigned_by }, _, { user, loaders: { Users } }) {
-    if (user && user.can(SEARCH_OTHER_USERS) && assigned_by != null) {
-      return Users.getByID.load(assigned_by);
-    }
-  },
-};
+const TagLink = {};
+
+decorateUserField(TagLink, 'assigned_by');
 
 module.exports = TagLink;

graph/resolvers/user.js

@@ -1,4 +1,8 @@
-const { decorateWithTags } = require('./util');
+const {
+  decorateWithTags,
+  decorateWithPermissionCheck,
+  checkSelfField,
+} = require('./util');
 const KarmaService = require('../../services/karma');
 const {
   SEARCH_ACTIONS,
@@ -7,86 +11,65 @@ const {
   VIEW_USER_ROLE,
   LIST_OWN_TOKENS,
   VIEW_USER_STATUS,
+  VIEW_USER_EMAIL,
 } = require('../../perms/constants');
+const { property } = require('lodash');
 
 const User = {
   action_summaries(user, _, { loaders: { Actions } }) {
     return Actions.getSummariesByItem.load(user);
   },
-  actions({ id }, _, { user, loaders: { Actions } }) {
-    // Only return the actions if the user is not an admin.
-    if (user && user.can(SEARCH_ACTIONS)) {
-      return Actions.getByID.load(id);
-    }
+  actions({ id }, _, { loaders: { Actions } }) {
+    return Actions.getByID.load(id);
   },
-  comments({ id }, { query }, { loaders: { Comments }, user }) {
-    // If there is no user, or there is a user, but they are requesting someone
-    // else's comments, and they aren't allowed, don't return then anything!
-    if (!user || (user.id !== id && !user.can(SEARCH_OTHERS_COMMENTS))) {
-      return null;
-    }
-
+  comments({ id }, { query }, { loaders: { Comments } }) {
     // Set the author id on the query.
     query.author_id = id;
 
     return Comments.getByQuery(query);
   },
-  profiles({ profiles }, _, { user }) {
-    // if the user is not an admin, do not return the profiles
-    if (user && user.can(SEARCH_OTHER_USERS)) {
-      return profiles;
-    }
-
-    return null;
-  },
-  tokens({ id, tokens }, args, { user }) {
-    if (!user || (user.id !== id && !user.can(LIST_OWN_TOKENS))) {
-      return null;
-    }
-
-    return tokens;
-  },
-  ignoredUsers({ id }, args, { user, loaders: { Users } }) {
-    // Only allow a logged in user that is either the current user or is a staff
-    // member to access the ignoredUsers of a given user.
-    if (!user || (user.id !== id && !user.can(SEARCH_OTHER_USERS))) {
-      return null;
-    }
 
+  ignoredUsers({ ignoresUsers }, args, { user, loaders: { Users } }) {
     // Return nothing if there is nothing to query for.
     if (!user.ignoresUsers || user.ignoresUsers.length <= 0) {
       return [];
     }
 
-    return Users.getByID.loadMany(user.ignoresUsers);
-  },
-  role({ id, role }, _, { user }) {
-    // If the user is not an admin, only return the current user's roles.
-    if (user && (user.can(VIEW_USER_ROLE) || user.id === id)) {
-      return role;
-    }
-
-    return null;
+    return Users.getByID.loadMany(ignoresUsers);
   },
 
   // Extract the reliability from the user metadata if they have permission.
-  reliable(user, _, { user: requestingUser }) {
-    if (requestingUser && requestingUser.can(SEARCH_ACTIONS)) {
-      return KarmaService.model(user);
-    }
-  },
+  reliable: user => KarmaService.model(user),
 
-  state(user, args, ctx) {
-    if (
-      ctx.user &&
-      (ctx.user.id === user.id || ctx.user.can(VIEW_USER_STATUS))
-    ) {
-      return user;
-    }
-  },
+  // The state requires the whole user object to make decisions.
+  state: user => user,
+
+  // Get the first email on the user.
+  email: property('firstEmail'),
 };
 
 // Decorate the User type resolver with a tags field.
 decorateWithTags(User);
 
+// decorate the fields on the User resolver with a permission check where the
+// current user can also get their own properties.
+decorateWithPermissionCheck(
+  User,
+  {
+    actions: [SEARCH_ACTIONS],
+    email: [VIEW_USER_EMAIL],
+    state: [VIEW_USER_STATUS],
+    role: [VIEW_USER_ROLE],
+    ignoredUsers: [SEARCH_OTHER_USERS],
+    tokens: [LIST_OWN_TOKENS],
+    profiles: [SEARCH_OTHER_USERS],
+    comments: [SEARCH_OTHERS_COMMENTS],
+  },
+  checkSelfField('id')
+);
+
+// Decorate the fields on the User resolver where the current user has no impact
+// on the resolvability of a field.
+decorateWithPermissionCheck(User, { reliable: [SEARCH_ACTIONS] });
+
 module.exports = User;

graph/resolvers/user_state.js

@@ -1,14 +1,12 @@
+const { decorateWithPermissionCheck, checkSelfField } = require('./util');
 const { VIEW_USER_STATUS } = require('../../perms/constants');
 
-const UserState = {
-  status: (user, args, ctx) => {
-    if (
-      ctx.user &&
-      (ctx.user.id === user.id || ctx.user.can(VIEW_USER_STATUS))
-    ) {
-      return user.status;
-    }
-  },
-};
+const UserState = {};
+
+decorateWithPermissionCheck(
+  UserState,
+  { status: [VIEW_USER_STATUS] },
+  checkSelfField('id')
+);
 
 module.exports = UserState;

graph/resolvers/util.js

@@ -2,59 +2,167 @@ const {
   ADD_COMMENT_TAG,
   SEARCH_OTHER_USERS,
 } = require('../../perms/constants');
-const property = require('lodash/property');
+const { property, isBoolean } = require('lodash');
 
 /**
- * Decorates the typeResolver with the tags field.
+ * getResolver will get the resolver from the typeResolver or apply the default
+ * resolver.
+ *
+ * @param {Object} typeResolver the type resolver
+ * @param {String} field the field name of the resolver we're getting
  */
-const decorateWithTags = typeResolver => {
-  typeResolver.tags = ({ tags = [] }, _, { user }) => {
-    if (user && user.can(ADD_COMMENT_TAG)) {
-      return tags;
+const getResolver = (typeResolver, field) => {
+  if (field in typeResolver) {
+    return typeResolver[field];
+  }
+
+  return property(field);
+};
+
+/**
+ *
+ * @param {Object} typeResolver the type resolver
+ * @param {String} field the name of the field being wrapped
+ * @param {Function} customCheck the function that can return a boolean
+ *                               indicating the resolution status
+ * @param {Function} skipFieldResolver the optional skip resolver that can be used
+ *                                 skip out from the oldFieldResolver.
+ */
+const wrapCheck = (
+  typeResolver,
+  field,
+  customCheck,
+  skipFieldResolver = getResolver(typeResolver, field)
+) => {
+  // Cache the old field resolver. In the event that the check does not return
+  // with a boolean, we'll use this.
+  const oldFieldResolver = getResolver(typeResolver, field);
+
+  // Override the field resolver on the type resolver with this wrapped
+  // function.
+  typeResolver[field] = (obj, args, ctx, info) => {
+    const decision = customCheck(obj, args, ctx, info);
+    if (isBoolean(decision)) {
+      if (decision) {
+        // The custom check returns a boolean true, so we should execute the
+        // underlying field resolver (which may just be the old resolver).
+        return skipFieldResolver(obj, args, ctx, info);
+      }
+
+      // The custom check returns a boolean false, then we should return null,
+      // because the check explicity said that we weren't allowed to access that
+      // field.
+      return null;
     }
 
-    return tags.filter(t => t.tag.permissions.public);
+    // The custom check yielded no decision, so we should just fall back to the
+    // oldFieldResolver.
+    return oldFieldResolver(obj, args, ctx, info);
   };
 };
 
+/**
+ * checkPermissions checks that the current user has all the required
+ * permissions.
+ *
+ * @param {Object} ctx graph context
+ * @param {Array<String>} permissions permissions that the user must have
+ */
+const checkPermissions = (ctx, permissions) =>
+  !ctx.user || !ctx.user.can(...permissions);
+
+/**
+ * wrapCheckPermissions will wrap a specific field with a permission check.
+ *
+ * @param {Object} typeResolver the type resolver
+ * @param {String} field the field name of the resolver we're wrapping
+ * @param {Array<String>} permissions array of permissions to check against
+ * @param {Function} fieldResolver base resolver for the field
+ */
+const wrapCheckPermissions = (
+  typeResolver,
+  field,
+  permissions,
+  skipFieldResolver = getResolver(typeResolver, field)
+) =>
+  wrapCheck(
+    typeResolver,
+    field,
+    (obj, args, ctx) => !checkPermissions(ctx, permissions),
+    skipFieldResolver
+  );
+
 /**
  * decorateWithPermissionCheck will decorate the field resolver with
  * permission checks.
  *
  * @param {Object} typeResolver the type resolver
  * @param {Object} protect the object with field -> Array<String> of permissions
+ * @param {Function} customCheck a function that can return a boolean based on a
+ *                               custom check
  */
-const decorateWithPermissionCheck = (typeResolver, protect) => {
+const decorateWithPermissionCheck = (
+  typeResolver,
+  protect,
+  customCheck = null
+) => {
   for (const [field, permissions] of Object.entries(protect)) {
-    let fieldResolver = property(field);
-    if (field in typeResolver) {
-      fieldResolver = typeResolver[field];
+    const baseFieldResolver = getResolver(typeResolver, field);
+    wrapCheckPermissions(typeResolver, field, permissions, baseFieldResolver);
+
+    if (customCheck !== null) {
+      wrapCheck(typeResolver, field, customCheck, baseFieldResolver);
     }
-
-    typeResolver[field] = (obj, args, ctx, info) => {
-      if (!ctx.user || !ctx.user.can(...permissions)) {
-        return null;
-      }
-
-      return fieldResolver(obj, args, ctx, info);
-    };
   }
 };
 
 /**
- * decorateUserField will decorate the user field accesses with correct
- * permission checks.
+ * checkSelf will check if the current object is the same as the current user.
+ *
+ * @param {String} referenceField the field for the user id to check.
+ */
+const checkSelfField = referenceField => (obj, args, ctx) => {
+  if (
+    ctx.user &&
+    obj[referenceField] !== null &&
+    ctx.user.id === obj[referenceField]
+  ) {
+    return true;
+  }
+};
+
+/**
+ * wrapCheckSelf wraps a typeResolver with a check for self (if the type is
+ * referencing the current user).
  *
  * @param {Object} typeResolver the type resolver
  * @param {String} field the field to decorate
+ * @param {String} referenceField the field to pull the user id from.
+ * @param {Function} fieldResolver base resolver for the field
  */
-const decorateUserField = (typeResolver, field) => {
+const wrapCheckSelf = (
+  typeResolver,
+  field,
+  referenceField = field,
+  fieldResolver = getResolver(typeResolver, field)
+) =>
+  wrapCheck(typeResolver, field, checkSelfField(referenceField), fieldResolver);
+
+/**
+ * decorateUserField will decorate the user field accesses with correct
+ * permission checks and will load the user.
+ *
+ * @param {Object} typeResolver the type resolver
+ * @param {String} field the field to decorate
+ * @param {String} referenceField the field to pull the user id from.
+ */
+const decorateUserField = (typeResolver, field, referenceField = field) => {
   // The default resolver for the user decorator is loading the user by id.
   let fieldResolver = (obj, args, ctx) => {
-    if (!obj[field]) {
+    if (!obj[referenceField]) {
       return null;
     }
-    return ctx.loaders.Users.getByID.load(obj[field]);
+    return ctx.loaders.Users.getByID.load(obj[referenceField]);
   };
 
   // The resolver can be overridden however. This decorator will simply wrap the
@@ -63,16 +171,39 @@ const decorateUserField = (typeResolver, field) => {
     fieldResolver = typeResolver[field];
   }
 
-  typeResolver[field] = (obj, args, ctx, info) => {
-    if (
-      !ctx.user ||
-      obj[field] === null ||
-      (ctx.user.id !== obj[field] && !ctx.user.can(SEARCH_OTHER_USERS))
-    ) {
-      return null;
+  // Wrap the current fieldResolver with the resolver which will return the
+  // user.
+  wrapCheckPermissions(
+    typeResolver,
+    field,
+    [SEARCH_OTHER_USERS],
+    fieldResolver
+  );
+
+  // Wrap the checked resolver with a check to see if the current user is the
+  // one being retrieved, in which case we should allow it.
+  wrapCheckSelf(typeResolver, field, referenceField, fieldResolver);
+};
+
+/**
+ * decorateWithTags decorates a typeResolver with a tags getter that will
+ * sanitize the tag output for users without permission to see non-public
+ * tags.
+ *
+ * @param {Object} typeResolver base type resolver
+ * @param {Function} fieldResolver the field resolver that gets the tags
+ */
+const decorateWithTags = (
+  typeResolver,
+  fieldResolver = getResolver(typeResolver, 'tags')
+) => {
+  typeResolver.tags = async (obj, args, ctx, info) => {
+    const tags = await fieldResolver(obj, args, ctx, info);
+    if (checkPermissions(ctx, [ADD_COMMENT_TAG])) {
+      return tags;
     }
 
-    return fieldResolver(obj, args, ctx, info);
+    return tags.filter(t => t.tag.permissions.public);
   };
 };
 
@@ -80,4 +211,5 @@ module.exports = {
   decorateUserField,
   decorateWithTags,
   decorateWithPermissionCheck,
+  checkSelfField,
 };

graph/subscriptions/broker.js

@@ -0,0 +1,48 @@
+const { EventEmitter2 } = require('eventemitter2');
+const debug = require('debug')('talk:graph:subscriptions:broker');
+const { getPubsub } = require('./pubsub');
+
+/**
+ * Broker acts as a pubsub client adapter. Any calls to publish will push into
+ * the PubSub client and the local event emitter.
+ */
+class Broker extends EventEmitter2 {
+  constructor(pubsub) {
+    // Create the underlying event emitter.
+    super({
+      wildcard: true, // Allow wildcard listeners.
+      maxListeners: 0, // Disable maximum for listeners.
+    });
+
+    this.pubsub = pubsub;
+  }
+
+  /**
+   * Publishes the event out to the pubsub system and to the broker.
+   *
+   * @param {String} event the name of the event to publish
+   * @param {Any} args the
+   */
+  publish(event, ...args) {
+    debug(`publish:${event}`);
+    this.pubsub.publish(event, ...args);
+    this.emit(event, ...args);
+  }
+}
+
+let client = null;
+const getBroker = () => {
+  if (client !== null) {
+    return client;
+  }
+
+  // Create the new Broker to manage events being published out to
+  // the pubsub so that we may intercept.
+  client = new Broker(getPubsub());
+
+  debug('created');
+
+  return client;
+};
+
+module.exports.getBroker = getBroker;

graph/subscriptions/index.js

@@ -2,18 +2,18 @@ const { SubscriptionManager } = require('graphql-subscriptions');
 const { SubscriptionServer } = require('subscriptions-transport-ws');
 const debug = require('debug')('talk:graph:subscriptions');
 
-const pubsub = require('../services/pubsub');
-const schema = require('./schema');
-const Context = require('./context');
-const plugins = require('../services/plugins');
+const { getPubsub } = require('./pubsub');
+const schema = require('../schema');
+const Context = require('../context');
+const plugins = require('../../services/plugins');
 
-const { deserializeUser } = require('../services/subscriptions');
+const { deserializeUser } = require('../../services/subscriptions');
 const setupFunctions = require('./setupFunctions');
 
 const ms = require('ms');
-const { KEEP_ALIVE } = require('../config');
+const { KEEP_ALIVE } = require('../../config');
 
-const { BASE_PATH } = require('../url');
+const { BASE_PATH } = require('../../url');
 
 // Collect all the plugin hooks that should be executed onConnect and
 // onDisconnect.
@@ -99,7 +99,7 @@ const createSubscriptionManager = server =>
     {
       subscriptionManager: new SubscriptionManager({
         schema,
-        pubsub: pubsub.getClient(),
+        pubsub: getPubsub(),
         setupFunctions,
       }),
       onConnect,

graph/subscriptions/pubsub.js

@@ -0,0 +1,29 @@
+const { RedisPubSub } = require('graphql-redis-subscriptions');
+const { createClient: createRedisClient } = require('../../services/redis');
+const debug = require('debug')('talk:graph:subscriptions:pubsub');
+
+/**
+ * getPubsub returns the pubsub singleton for this instance.
+ */
+let pubsub = null;
+const getPubsub = () => {
+  if (pubsub !== null) {
+    return pubsub;
+  }
+
+  // Create the publisher and subscriber redis clients.
+  const publisher = createRedisClient();
+  const subscriber = createRedisClient();
+
+  // Create the new PubSub client, we only need one per instance of Talk.
+  pubsub = new RedisPubSub({
+    publisher,
+    subscriber,
+  });
+
+  debug('created');
+
+  return pubsub;
+};
+
+module.exports.getPubsub = getPubsub;

graph/subscriptions/setupFunctions.js

@@ -11,11 +11,11 @@ const {
   SUBSCRIBE_ALL_USERNAME_APPROVED,
   SUBSCRIBE_ALL_USERNAME_FLAGGED,
   SUBSCRIBE_ALL_USERNAME_CHANGED,
-} = require('../perms/constants');
+} = require('../../perms/constants');
 
 const merge = require('lodash/merge');
 const debug = require('debug')('talk:graph:setupFunctions');
-const plugins = require('../services/plugins');
+const plugins = require('../../services/plugins');
 
 const setupFunctions = {
   commentAdded: (options, args, comment, context) => {

graph/typeDefs.graphql

@@ -63,7 +63,7 @@ type Token {
 }
 
 type UserProfile {
-  # the id is an identifier for the user profile (email, facebook id, etc)
+  # The id is an identifier for the user profile (email, facebook id, etc)
   id: String!
 
   # name of the provider attached to the authentication mode
@@ -184,13 +184,17 @@ type User {
   # Actions completed on the parent.
   actions: [Action!]
 
-  # the current roles of the user.
+  # The current roles of the user.
   role: USER_ROLES
 
-  # the current profiles of the user.
+  # The current profiles of the user.
   profiles: [UserProfile]
 
-  # the tags on the user
+  # The primary email address of the user. Only accessible to Administrators or
+  # the current user.
+  email: String
+
+  # The tags on the user.
   tags: [TagLink!]
 
   # ignored users.
@@ -421,7 +425,7 @@ input CommentCountQuery {
   # The URL that the asset is located on.
   asset_url: String
 
-  # the parent of the comment that we want to retrieve.
+  # The parent of the comment that we want to retrieve.
   parent_id: ID
 
   # comments returned will only be ones which have at least one action of this
@@ -479,13 +483,13 @@ type Comment {
   # The body history of the comment.
   body_history: [CommentBodyHistory!]!
 
-  # the tags on the comment
+  # The tags on the comment
   tags: [TagLink!]
 
-  # the user who authored the comment.
+  # The user who authored the comment.
   user: User
 
-  # the replies that were made to the comment.
+  # The replies that were made to the comment.
   replies(query: RepliesQuery = {}): CommentConnection!
 
   # replyCount is the number of replies with a depth of 1. Only direct replies
@@ -765,7 +769,7 @@ type Settings {
   infoBoxContent: String
 
   # questionBoxEnable will enable the Question Box's content to be visible above
-  # the comment box.
+  # The comment box.
   questionBoxEnable: Boolean
 
   # questionBoxContent is the content of the Question Box.
@@ -858,7 +862,7 @@ type Asset {
   # The date that the asset was created.
   created_at: Date
 
-  # the tags on the asset
+  # The tags on the asset
   tags: [TagLink!]
 
   # The author(s) of the asset.
@@ -1091,7 +1095,7 @@ input AssetSettingsInput {
   moderation: MODERATION_MODE
 
   # questionBoxEnable will enable the Question Boxs' content to be visible above
-  # the comment box.
+  # The comment box.
   questionBoxEnable: Boolean
 
   # questionBoxContent is the content of the Question Box.
@@ -1167,7 +1171,7 @@ input ModifyTagInput {
   item_type: TAGGABLE_ITEM_TYPE!
 
   # asset_id is used when the item_type is `COMMENTS`, the is needed to rectify
-  # the settings to get the asset specific tags/settings.
+  # The settings to get the asset specific tags/settings.
   asset_id: ID
 }
 
@@ -1225,7 +1229,7 @@ input UpdateSettingsInput {
   infoBoxContent: String
 
   # questionBoxEnable will enable the Question Box's content to be visible above
-  # the comment box.
+  # The comment box.
   questionBoxEnable: Boolean
 
   # questionBoxContent is the content of the Question Box.

jobs/index.js

@@ -0,0 +1,5 @@
+const jobs = [require('./mailer')];
+
+const process = () => jobs.forEach(job => job());
+
+module.exports = { process };

jobs/mailer.js

@@ -0,0 +1,147 @@
+const { task } = require('../services/mailer');
+const nodemailer = require('nodemailer');
+const debug = require('debug')('talk:jobs:mailer');
+const Context = require('../graph/context');
+const { get } = require('lodash');
+
+const {
+  SMTP_HOST,
+  SMTP_USERNAME,
+  SMTP_PORT,
+  SMTP_PASSWORD,
+  SMTP_FROM_ADDRESS,
+} = require('../config');
+
+// parseSMTPPort will return the port for SMTP.
+const parseSMTPPort = () => {
+  if (!SMTP_PORT) {
+    return 25;
+  }
+
+  try {
+    return parseInt(SMTP_PORT);
+  } catch (e) {
+    throw new Error('TALK_SMTP_PORT is not an integer');
+  }
+};
+
+// createTransport will create a new transport.
+const createTransport = () => {
+  const options = {
+    host: SMTP_HOST,
+  };
+
+  if (SMTP_USERNAME && SMTP_PASSWORD) {
+    options.auth = {
+      user: SMTP_USERNAME,
+      pass: SMTP_PASSWORD,
+    };
+  }
+
+  // Get the SMTP port.
+  options.port = parseSMTPPort();
+
+  return nodemailer.createTransport(options);
+};
+
+// sharedTransport is the transport singleton.
+let sharedTransport;
+
+// getTransport will retrieve the mailer transport singleton.
+const getTransport = () => {
+  if (sharedTransport) {
+    return sharedTransport;
+  }
+
+  // enabled is true when the required configuration is available. When testing
+  // is enabled, we will be simulating that emails are being sent, because in a
+  // production system, emails should and would be sent.
+
+  // If the transport details aren't available, we will return null as the
+  // transport.
+  if (!SMTP_HOST || !SMTP_FROM_ADDRESS) {
+    return null;
+  }
+
+  // Create the transport.
+  sharedTransport = createTransport();
+
+  return sharedTransport;
+};
+
+// getEmailAddress will retrieve the email address to send the message to from
+// the job data.
+const getEmailAddress = async ({ email, user }) => {
+  // If the message has a specific email already to sent it to, just assign
+  // that to the message. If the email does not have an email, and instead has
+  // a user id, then we should lookup the user with the graph and get their
+  // email.
+  if (email) {
+    return email;
+  } else {
+    // Get the user to send the message to.
+    const ctx = Context.forSystem();
+
+    const { data, errors } = await ctx.graphql(
+      `
+        query GetUserEmail($user: ID!) {
+          user(id: $user) {
+            email
+          }
+        }
+      `,
+      { user }
+    );
+    if (errors) {
+      throw errors;
+    }
+
+    const email = get(data, 'user.email');
+    if (!email) {
+      throw errors.ErrMissingEmail;
+    }
+
+    return email;
+  }
+};
+
+// processJob will handle new jobs sent via this queue.
+const processJob = transport => async ({ id, data }, done) => {
+  const { message } = data;
+
+  // Get the email address from the job data.
+  message.to = await getEmailAddress(data);
+
+  debug(`Starting to send mail for Job[${id}]`);
+
+  // Actually send the email.
+  transport.sendMail(message, err => {
+    if (err) {
+      debug(`Failed to send mail for Job[${id}]:`, err);
+      return done(err);
+    }
+
+    debug(`Finished sending mail for Job[${id}]`);
+    return done();
+  });
+};
+
+/**
+ * Start the queue processor for the mailer job.
+ */
+module.exports = () => {
+  // Get a transport.
+  const transport = getTransport();
+  if (transport === null) {
+    console.warn(
+      new Error(
+        'sending email is not enabled because required configuration is not available'
+      )
+    );
+    return;
+  }
+
+  debug(`Now processing ${task.name} jobs`);
+
+  return task.process(processJob(transport));
+};

jobs/scraper.js

@@ -0,0 +1,76 @@
+const Asset = require('../models/asset');
+const scraper = require('../services/scraper');
+const Assets = require('../services/assets');
+const debug = require('debug')('talk:jobs:scraper');
+const metascraper = require('metascraper');
+
+/**
+ * Scrapes the given asset for metadata.
+ */
+async function scrape(asset) {
+  return metascraper.scrapeUrl(
+    asset.url,
+    Object.assign({}, metascraper.RULES, {
+      section: $ => $('meta[property="article:section"]').attr('content'),
+      modified: $ => $('meta[property="article:modified"]').attr('content'),
+    })
+  );
+}
+
+/**
+ * Updates an Asset based on scraped asset metadata.
+ */
+function update(id, meta) {
+  return Asset.update(
+    { id },
+    {
+      $set: {
+        title: meta.title || '',
+        description: meta.description || '',
+        image: meta.image ? meta.image : '',
+        author: meta.author || '',
+        publication_date: meta.date || '',
+        modified_date: meta.modified || '',
+        section: meta.section || '',
+        scraped: new Date(),
+      },
+    }
+  );
+}
+
+module.exports = () => {
+  debug(`Now processing ${scraper.task.name} jobs`);
+
+  scraper.task.process(async (job, done) => {
+    debug(`Starting on Job[${job.id}] for Asset[${job.data.asset_id}]`);
+
+    try {
+      // Find the asset, or complain that it doesn't exist.
+      const asset = await Assets.findById(job.data.asset_id);
+      if (!asset) {
+        return done(new Error('asset not found'));
+      }
+
+      // Scrape the metadata from the asset.
+      const meta = await scrape(asset);
+
+      debug(
+        `Scraped ${JSON.stringify(meta)} on Job[${job.id}] for Asset[${
+          job.data.asset_id
+        }]`
+      );
+
+      // Assign the metadata retrieved for the asset to the db.
+      await update(job.data.asset_id, meta);
+    } catch (err) {
+      debug(
+        `Failed to scrape on Job[${job.id}] for Asset[${job.data.asset_id}]:`,
+        err
+      );
+      return done(err);
+    }
+
+    debug(`Finished on Job[${job.id}] for Asset[${job.data.asset_id}]`);
+    done();
+  });
+};

middleware/context.js

@@ -0,0 +1,8 @@
+const Context = require('../graph/context');
+
+// Attach a new context to the request.
+module.exports = (req, res, next) => {
+  req.context = new Context(req);
+
+  next();
+};

models/setting.js

@@ -99,6 +99,12 @@ const SettingSchema = new Schema(
       default: 30 * 1000,
     },
     tags: [TagSchema],
+
+    // Additional metadata to let plugins write settings.
+    metadata: {
+      default: {},
+      type: Object,
+    },
   },
   {
     timestamps: {

package.json

@@ -78,6 +78,7 @@
     "bcryptjs": "^2.4.3",
     "bowser": "^1.7.2",
     "brotli-webpack-plugin": "^0.5.0",
+    "bunyan": "^1.8.12",
     "cli-table": "^0.3.1",
     "clipboard": "^1.7.1",
     "colors": "^1.1.2",
@@ -116,13 +117,14 @@
     "hammerjs": "^2.0.8",
     "helmet": "3.8.2",
     "history": "^3.0.0",
+    "hjson": "^3.1.1",
+    "hjson-loader": "^1.0.0",
     "immutability-helper": "^2.2.0",
     "imports-loader": "^0.7.1",
     "inquirer": "^3.2.2",
     "inquirer-autocomplete-prompt": "^0.12.1",
     "ioredis": "3.1.4",
     "joi": "^13.0.0",
-    "json-loader": "^0.5.7",
     "jsonwebtoken": "^8.0.0",
     "jwt-decode": "^2.2.0",
     "keymaster": "^1.6.2",
@@ -136,7 +138,7 @@
     "minimist": "^1.2.0",
     "moment": "^2.18.1",
     "mongoose": "^4.12.3",
-    "morgan": "1.9.0",
+    "morgan": "^1.9.0",
     "ms": "^2.0.0",
     "murmurhash-js": "^1.0.0",
     "node-emoji": "^1.8.1",

perms/constants/query.js

@@ -9,4 +9,5 @@ module.exports = {
   VIEW_PROTECTED_SETTINGS: 'VIEW_PROTECTED_SETTINGS',
   LIST_OWN_TOKENS: 'LIST_OWN_TOKENS',
   VIEW_USER_ROLE: 'VIEW_USER_ROLE',
+  VIEW_USER_EMAIL: 'VIEW_USER_EMAIL',
 };

perms/reducers/query.js

@@ -8,11 +8,11 @@ module.exports = (user, perm) => {
     case types.SEARCH_ACTIONS:
     case types.SEARCH_NON_NULL_OR_ACCEPTED_COMMENTS:
     case types.SEARCH_OTHERS_COMMENTS:
-      return check(user, ['ADMIN', 'MODERATOR']);
     case types.SEARCH_COMMENT_STATUS_HISTORY:
     case types.VIEW_USER_STATUS:
     case types.VIEW_PROTECTED_SETTINGS:
     case types.VIEW_USER_ROLE:
+    case types.VIEW_USER_EMAIL:
       return check(user, ['ADMIN', 'MODERATOR']);
     case types.LIST_OWN_TOKENS:
       return check(user, ['ADMIN']);

plugin-api/beta/client/hocs/index.js

@@ -6,6 +6,7 @@ export {
   withEmit,
   excludeIf,
   withFragments,
+  withMutation,
   withForgotPassword,
   withSignIn,
   withSignUp,

plugin-api/client/lib/index.js

@@ -0,0 +1 @@
+export { default as Action } from 'coral-framework/lib/action';

plugins.js

@@ -4,6 +4,7 @@ const resolve = require('resolve');
 const debug = require('debug')('talk:plugins');
 const Joi = require('joi');
 const amp = require('app-module-path');
+const hjson = require('hjson');
 const pkg = require('./package.json');
 
 const PLUGINS_JSON = process.env.TALK_PLUGINS_JSON;
@@ -33,7 +34,9 @@ try {
     pluginsPath = defaultPlugins;
   }
 
-  plugins = require(pluginsPath);
+  // Load/parse the plugin content using hjson.
+  const pluginContent = fs.readFileSync(pluginsPath, 'utf8');
+  plugins = hjson.parse(pluginContent);
 } catch (err) {
   if (err.code === 'ENOENT') {
     console.error(
@@ -73,6 +76,7 @@ const hookSchemas = {
     onConnect: Joi.func(),
     onDisconnect: Joi.func(),
   }),
+  connect: Joi.func().maxArity(1),
 };
 
 /**
@@ -293,14 +297,42 @@ class PluginManager {
         plugins[section]
       );
     }
+
+    this.deferredHooks = [];
+    this.ranDeferredHooks = false;
   }
 
   /**
    * Utility function which combines the Plugins.section and PluginSection.hook
    * calls.
    */
-  get(section, hook) {
-    return this.section(section).hook(hook);
+  get(sectionName, hookName) {
+    return this.section(sectionName).hook(hookName);
+  }
+
+  /**
+   * Utility function which combines the Plugins.section and PluginSection.hook
+   * calls and runs them when the `runDeferred` is called.
+   */
+  defer(sectionName, hookName, callback) {
+    const plugins = this.section(sectionName).hook(hookName);
+
+    // If we've already ran the callbacks, then we should run it immediately.
+    if (this.ranDeferredHooks) {
+      plugins.forEach(callback);
+    } else {
+      this.deferredHooks.push({ plugins, callback });
+    }
+  }
+
+  /**
+   * Calls all deferred hooks.
+   */
+  runDeferred() {
+    this.deferredHooks.forEach(({ plugins, callback }) =>
+      plugins.forEach(callback)
+    );
+    this.ranDeferredHooks = true;
   }
 
   /**

plugins/talk-plugin-facebook-auth/server/router.js

@@ -1,24 +1,31 @@
 module.exports = router => {
-  const { passport, HandleAuthPopupCallback } = require('services/passport');
-
   /**
    * Facebook auth endpoint, this will redirect the user immediately to Facebook
    * for authorization.
    */
-  router.get(
-    '/api/v1/auth/facebook',
-    passport.authenticate('facebook', {
+  router.get('/api/v1/auth/facebook', (req, res, next) => {
+    const {
+      connectors: { services: { Passport: { passport } } },
+    } = req.context;
+
+    return passport.authenticate('facebook', {
       display: 'popup',
       authType: 'rerequest',
       scope: ['public_profile'],
-    })
-  );
+    })(req, res, next);
+  });
 
   /**
    * Facebook callback endpoint, this will send the user a HTML page designed to
    * send back the user credentials upon successful login.
    */
   router.get('/api/v1/auth/facebook/callback', (req, res, next) => {
+    const {
+      connectors: {
+        services: { Passport: { passport, HandleAuthPopupCallback } },
+      },
+    } = req.context;
+
     // Perform the facebook login flow and pass the data back through the opener.
     passport.authenticate(
       'facebook',

plugins/talk-plugin-notifications-category-reply/.eslintrc.json

@@ -0,0 +1,3 @@
+{
+  "extends": "@coralproject/eslint-config-talk"
+}

plugins/talk-plugin-notifications-category-reply/client/.eslintrc.json

@@ -0,0 +1,3 @@
+{
+  "extends": "@coralproject/eslint-config-talk/client"
+}

plugins/talk-plugin-notifications-category-reply/client/containers/Toggle.js

@@ -0,0 +1,69 @@
+import React from 'react';
+import PropTypes from 'prop-types';
+import { compose, gql } from 'react-apollo';
+import Toggle from 'talk-plugin-notifications/client/components/Toggle';
+import { t } from 'plugin-api/beta/client/services';
+import { withFragments } from 'plugin-api/beta/client/hocs';
+
+class ToggleContainer extends React.Component {
+  constructor(props) {
+    super(props);
+    props.setTurnOffInputFragment({ onReply: false });
+
+    if (this.getOnReplySetting()) {
+      props.indicateOn();
+    }
+  }
+
+  componentWillReceiveProps(nextProps) {
+    const prevSetting = this.getOnReplySetting(this.props);
+    const nextSetting = this.getOnReplySetting(nextProps);
+    if (prevSetting && !nextSetting) {
+      nextProps.indicateOff();
+    } else if (!prevSetting && nextSetting) {
+      nextProps.indicateOn();
+    }
+  }
+
+  getOnReplySetting = (props = this.props) =>
+    props.root.me.notificationSettings.onReply;
+
+  toggle = () => {
+    this.props.updateNotificationSettings({
+      onReply: !this.getOnReplySetting(),
+    });
+  };
+
+  render() {
+    return (
+      <Toggle checked={this.getOnReplySetting()} onChange={this.toggle}>
+        {t('talk-plugin-notifications-category-reply.toggle_description')}
+      </Toggle>
+    );
+  }
+}
+
+ToggleContainer.propTypes = {
+  data: PropTypes.object,
+  root: PropTypes.object,
+  indicateOn: PropTypes.func.isRequired,
+  indicateOff: PropTypes.func.isRequired,
+  setTurnOffInputFragment: PropTypes.func.isRequired,
+  updateNotificationSettings: PropTypes.func.isRequired,
+};
+
+const enhance = compose(
+  withFragments({
+    root: gql`
+      fragment TalkNotificationsCategoryReply_Toggle_root on RootQuery {
+        me {
+          notificationSettings {
+            onReply
+          }
+        }
+      }
+    `,
+  })
+);
+
+export default enhance(ToggleContainer);

plugins/talk-plugin-notifications-category-reply/client/graphql.js

@@ -0,0 +1,33 @@
+import { gql } from 'react-apollo';
+
+export default {
+  mutations: {
+    UpdateNotificationSettings: ({
+      variables: { input },
+      state: { auth: { user: { id } } },
+    }) => ({
+      update: proxy => {
+        if (input.onReply === undefined) {
+          return;
+        }
+
+        const fragment = gql`
+          fragment TalkNotificationsCategoryReply_User_Fragment on User {
+            notificationSettings {
+              onReply
+            }
+          }
+        `;
+        const fragmentId = `User_${id}`;
+        const data = {
+          __typename: 'User',
+          notificationSettings: {
+            __typename: 'NotificationSettings',
+            onReply: input.onReply,
+          },
+        };
+        proxy.writeFragment({ fragment, id: fragmentId, data });
+      },
+    }),
+  },
+};

plugins/talk-plugin-notifications-category-reply/client/index.js

@@ -0,0 +1,11 @@
+import Toggle from './containers/Toggle';
+import translations from './translations.yml';
+import graphql from './graphql';
+
+export default {
+  slots: {
+    notificationSettings: [Toggle],
+  },
+  translations,
+  ...graphql,
+};

plugins/talk-plugin-notifications-category-reply/client/translations.yml

@@ -0,0 +1,3 @@
+en:
+  talk-plugin-notifications-category-reply:
+    toggle_description: My comment receives a reply

plugins/talk-plugin-notifications-category-reply/index.js

@@ -0,0 +1,124 @@
+const { graphql } = require('graphql');
+const { get } = require('lodash');
+const path = require('path');
+
+const handle = async (ctx, comment) => {
+  const { connectors: { graph: { schema } } } = ctx;
+
+  // Check to see if this is a reply to an existing comment.
+  const parentID = get(comment, 'parent_id', null);
+  if (parentID === null) {
+    ctx.log.debug('could not get parent comment id');
+    return;
+  }
+
+  // Execute the graph request.
+  const reply = await graphql(
+    schema,
+    `
+      query GetAuthorUserMetadata($comment_id: ID!) {
+        comment(id: $comment_id) {
+          id
+          user {
+            id
+            notificationSettings {
+              onReply
+            }
+          }
+        }
+      }
+    `,
+    {},
+    ctx,
+    { comment_id: parentID }
+  );
+  if (reply.errors) {
+    ctx.log.error({ err: reply.errors }, 'could not query for author metadata');
+    return;
+  }
+
+  // Check if the user has notifications enabled.
+  const enabled = get(
+    reply,
+    'data.comment.user.notificationSettings.onReply',
+    false
+  );
+  if (!enabled) {
+    return;
+  }
+
+  const userID = get(reply, 'data.comment.user.id', null);
+  if (!userID) {
+    ctx.log.debug('could not get parent comment user id');
+    return;
+  }
+
+  // Check to see if this is yourself replying to yourself, if that's the case
+  // don't send a notification.
+  if (userID === get(comment, 'author_id')) {
+    ctx.log.debug('user id of parent comment is the same as the new comment');
+    return;
+  }
+
+  // The user does have notifications for replied comments enabled, queue the
+  // notification to be sent.
+  return { userID, date: comment.created_at, context: comment.id };
+};
+
+const hydrate = async (ctx, category, context) => {
+  const { connectors: { graph: { schema } } } = ctx;
+
+  const reply = await graphql(
+    schema,
+    `
+      query GetNotificationData($context: ID!) {
+        comment(id: $context) {
+          id
+          asset {
+            title
+            url
+          }
+          user {
+            username
+          }
+        }
+      }
+    `,
+    {},
+    ctx,
+    { context }
+  );
+  if (reply.errors) {
+    throw reply.errors;
+  }
+
+  const comment = get(reply, 'data.comment');
+  const headline = get(comment, 'asset.title', null);
+  const replier = get(comment, 'user.username', null);
+  const assetURL = get(comment, 'asset.url', null);
+  const permalink = `${assetURL}?commentId=${comment.id}`;
+
+  return [headline, replier, permalink];
+};
+
+const handler = { handle, category: 'reply', event: 'commentAdded', hydrate };
+
+module.exports = {
+  typeDefs: `
+    type NotificationSettings {
+      onReply: Boolean!
+    }
+
+    input NotificationSettingsInput {
+      onReply: Boolean
+    }
+  `,
+  resolvers: {
+    NotificationSettings: {
+      // onReply returns false by default if not specified.
+      onReply: settings => get(settings, 'onReply', false),
+    },
+  },
+  translations: path.join(__dirname, 'translations.yml'),
+  notifications: [handler],
+};

plugins/talk-plugin-notifications-category-reply/translations.yml

@@ -0,0 +1,6 @@
+en:
+  talk-plugin-notifications:
+    categories:
+      reply:
+        subject: "Some has replied to your comment on [{0}]"
+        body: "{0}\n{1} replied to your comment: {2}"

plugins/talk-plugin-notifications/.eslintrc.json

@@ -0,0 +1,3 @@
+{
+  "extends": "@coralproject/eslint-config-talk"
+}

plugins/talk-plugin-notifications/client/.eslintrc.json

@@ -0,0 +1,3 @@
+{
+  "extends": "@coralproject/eslint-config-talk/client"
+}

plugins/talk-plugin-notifications/client/components/Settings.css

@@ -0,0 +1,27 @@
+.root {
+  margin-bottom: 20px;
+}
+
+.innerSettings {
+  padding-left: 12px;
+}
+
+.subtitle {
+  margin: 0;
+  margin-bottom: 8px;
+}
+
+.turnOffButton {
+  padding: 2px 0;
+  color: #2099d6;
+  border-bottom: 1px solid #2099d6;
+
+  &:disabled {
+    color: #e5e5e5;
+    border-bottom: 1px solid #e5e5e5;
+  }
+}
+
+.notifcationSettingsSlot {
+  margin-bottom: 3px;
+}

plugins/talk-plugin-notifications/client/components/Settings.js

@@ -0,0 +1,68 @@
+import React from 'react';
+import PropTypes from 'prop-types';
+import { IfSlotIsNotEmpty } from 'plugin-api/beta/client/components';
+import { Slot } from 'plugin-api/beta/client/components';
+import { t } from 'plugin-api/beta/client/services';
+import styles from './Settings.css';
+import { BareButton } from 'plugin-api/beta/client/components/ui';
+
+class Settings extends React.Component {
+  childFactory = el => {
+    const pluginName = el.type.talkPluginName;
+    const props = {
+      indicateOn: () => this.props.indicateOn(pluginName),
+      indicateOff: () => this.props.indicateOff(pluginName),
+    };
+    return React.cloneElement(el, props);
+  };
+
+  render() {
+    const {
+      root,
+      setTurnOffInputFragment,
+      updateNotificationSettings,
+      turnOffAll,
+      turnOffButtonDisabled,
+    } = this.props;
+
+    return (
+      <IfSlotIsNotEmpty slot="notificationSettings" queryData={{ root }}>
+        <div className={styles.root}>
+          <h3>{t('talk-plugin-notifications.settings_title')}</h3>
+          <h4 className={styles.subtitle}>
+            {t('talk-plugin-notifications.settings_subtitle')}
+          </h4>
+          <div className={styles.innerSettings}>
+            <Slot
+              className={styles.notifcationSettingsSlot}
+              fill="notificationSettings"
+              queryData={{ root }}
+              childFactory={this.childFactory}
+              setTurnOffInputFragment={setTurnOffInputFragment}
+              updateNotificationSettings={updateNotificationSettings}
+            />
+            <BareButton
+              className={styles.turnOffButton}
+              onClick={turnOffAll}
+              disabled={turnOffButtonDisabled}
+            >
+              {t('talk-plugin-notifications.turn_off_all')}
+            </BareButton>
+          </div>
+        </div>
+      </IfSlotIsNotEmpty>
+    );
+  }
+}
+
+Settings.propTypes = {
+  root: PropTypes.object,
+  indicateOn: PropTypes.func.isRequired,
+  indicateOff: PropTypes.func.isRequired,
+  setTurnOffInputFragment: PropTypes.func.isRequired,
+  updateNotificationSettings: PropTypes.func.isRequired,
+  turnOffAll: PropTypes.func.isRequired,
+  turnOffButtonDisabled: PropTypes.bool.isRequired,
+};
+
+export default Settings;

plugins/talk-plugin-notifications/client/components/Toggle.css

@@ -0,0 +1,6 @@
+.title {
+  display: inline-block;
+  width: 270px;
+  cursor: pointer;
+  user-select: none;
+}

plugins/talk-plugin-notifications/client/components/Toggle.js

@@ -0,0 +1,29 @@
+import React from 'react';
+import PropTypes from 'prop-types';
+import { Checkbox } from 'plugin-api/beta/client/components/ui';
+import styles from './Toggle.css';
+import uuid from 'uuid/v4';
+
+class Toggle extends React.Component {
+  id = uuid();
+
+  render() {
+    const { checked, onChange, children } = this.props;
+    return (
+      <div className={styles.toggle}>
+        <label htmlFor={this.id} className={styles.title}>
+          {children}
+        </label>
+        <Checkbox checked={checked} onChange={onChange} id={this.id} />
+      </div>
+    );
+  }
+}
+
+Toggle.propTypes = {
+  checked: PropTypes.bool,
+  onChange: PropTypes.func,
+  children: PropTypes.node,
+};
+
+export default Toggle;

plugins/talk-plugin-notifications/client/containers/Settings.js

@@ -0,0 +1,68 @@
+import React from 'react';
+import PropTypes from 'prop-types';
+import { compose, gql } from 'react-apollo';
+import Settings from '../components/Settings';
+import { withFragments } from 'plugin-api/beta/client/hocs';
+import { getSlotFragmentSpreads } from 'plugin-api/beta/client/utils';
+import { withUpdateNotificationSettings } from '../mutations';
+
+const slots = ['notificationSettings'];
+
+class SettingsContainer extends React.Component {
+  state = {
+    hasNotifications: [],
+    turnOffInput: {},
+  };
+
+  indicateOn = plugin =>
+    this.setState({
+      hasNotifications: this.state.hasNotifications.concat(plugin),
+    });
+  indicateOff = plugin =>
+    this.setState({
+      hasNotifications: this.state.hasNotifications.filter(i => i !== plugin),
+    });
+  setTurnOffInputFragment = fragment =>
+    this.setState({
+      turnOffInput: { ...this.state.turnOffInput, ...fragment },
+    });
+
+  turnOffAll = () => {
+    this.props.updateNotificationSettings(this.state.turnOffInput);
+  };
+
+  render() {
+    return (
+      <Settings
+        data={this.props.data}
+        root={this.props.root}
+        indicateOn={this.indicateOn}
+        indicateOff={this.indicateOff}
+        setTurnOffInputFragment={this.setTurnOffInputFragment}
+        updateNotificationSettings={this.props.updateNotificationSettings}
+        turnOffAll={this.turnOffAll}
+        turnOffButtonDisabled={this.state.hasNotifications.length === 0}
+      />
+    );
+  }
+}
+
+SettingsContainer.propTypes = {
+  data: PropTypes.object,
+  root: PropTypes.object,
+  updateNotificationSettings: PropTypes.func.isRequired,
+};
+
+const enhance = compose(
+  withFragments({
+    root: gql`
+      fragment TalkNotifications_Settings_root on RootQuery {
+        __typename
+        ${getSlotFragmentSpreads(slots, 'root')}
+      }
+    `,
+  }),
+  withUpdateNotificationSettings
+);
+
+export default enhance(SettingsContainer);

plugins/talk-plugin-notifications/client/graphql.js

@@ -0,0 +1,23 @@
+import { gql } from 'react-apollo';
+
+export default {
+  fragments: {
+    UpdateNotificationSettingsResponse: gql`
+      fragment Talk_UpdateNotificationSettingsResponse on UpdateNotificationSettingsResponse {
+        errors {
+          translation_key
+        }
+      }
+    `,
+  },
+  mutations: {
+    UpdateNotificationSettings: () => ({
+      optimisticResponse: {
+        updateNotificationSettings: {
+          __typename: 'UpdateNotificationSettingsResponse',
+          errors: null,
+        },
+      },
+    }),
+  },
+};

plugins/talk-plugin-notifications/client/index.js

@@ -0,0 +1,11 @@
+import Settings from './containers/Settings';
+import translations from './translations.yml';
+import graphql from './graphql';
+
+export default {
+  slots: {
+    profileSettings: [Settings],
+  },
+  translations,
+  ...graphql,
+};

plugins/talk-plugin-notifications/client/mutations.js

@@ -0,0 +1,23 @@
+import { withMutation } from 'plugin-api/beta/client/hocs';
+import { gql } from 'react-apollo';
+
+export const withUpdateNotificationSettings = withMutation(
+  gql`
+    mutation UpdateNotificationSettings($input: NotificationSettingsInput!) {
+      updateNotificationSettings(input: $input) {
+        ...UpdateNotificationSettingsResponse
+      }
+    }
+  `,
+  {
+    props: ({ mutate }) => ({
+      updateNotificationSettings: input => {
+        return mutate({
+          variables: {
+            input,
+          },
+        });
+      },
+    }),
+  }
+);

plugins/talk-plugin-notifications/client/translations.yml

@@ -0,0 +1,5 @@
+en:
+  talk-plugin-notifications:
+    settings_title: Notifications
+    settings_subtitle: Receive notifications when
+    turn_off_all: I do not want to receive notifications

plugins/talk-plugin-notifications/index.js

@@ -0,0 +1 @@
+module.exports = require('./server');

plugins/talk-plugin-notifications/package.json

@@ -0,0 +1,11 @@
+{
+  "name": "@coralproject/talk-plugin-notifications",
+  "version": "1.0.0",
+  "description": "Adds notification support for Talk",
+  "main": "index.js",
+  "license": "Apache-2.0",
+  "private": false,
+  "dependencies": {
+    "linkifyjs": "^2.1.5"
+  }
+}

plugins/talk-plugin-notifications/server/NotificationManager.js

@@ -0,0 +1,158 @@
+const { groupBy, forEach } = require('lodash');
+const debug = require('debug')('talk-plugin-notifications');
+const uuid = require('uuid/v4');
+const { UNSUBSCRIBE_SUBJECT } = require('./config');
+
+class NotificationManager {
+  constructor(context) {
+    this.context = context;
+    this.registry = [];
+  }
+
+  /**
+   * register will include the notification handlers on the manager.
+   *
+   * @param {Array<Object>} handlers notification handlers to register
+   */
+  register(...handlers) {
+    this.registry.push(...handlers);
+  }
+
+  /**
+   * attach will setup the notifications by walking the registry and loading all
+   * the notification types onto the handler.
+   *
+   * @param {Object} broker the event emitter for the Talk events
+   */
+  attach(broker) {
+    const events = groupBy(this.registry, 'event');
+
+    forEach(events, (handlers, event) => {
+      debug(
+        `will now notify the [${handlers
+          .map(({ category }) => category)
+          .join(', ')}] handlers when the '${event}' event is emitted`
+      );
+      broker.on(event, this.handle(handlers));
+    });
+  }
+
+  /**
+   * handle will wrap a notification handler and attach it to the notification
+   * stream system.
+   *
+   * @param {Object} handler a notification handler
+   */
+  handle(handlers) {
+    return async (...args) =>
+      Promise.all(
+        handlers.map(async handler => {
+          // Grab the handler reference.
+          const { handle } = handler;
+
+          // Create a system context to send down.
+          const ctx = this.context.forSystem();
+
+          try {
+            // Attempt to create a notification out of it.
+            const notification = await handle(ctx, ...args);
+            if (!notification) {
+              return;
+            }
+
+            // Extract the notification details.
+            const { userID, date, context } = notification;
+
+            // Send the notification.
+            return this.send(ctx, userID, date, handler, context);
+          } catch (err) {
+            ctx.log.error({ err }, 'could not handle the event');
+            return;
+          }
+        })
+      );
+  }
+
+  async send(ctx, userID, date, handler, context) {
+    const {
+      connectors: {
+        secrets: { jwt },
+        config: { JWT_ISSUER, JWT_AUDIENCE },
+        services: { Mailer, I18n: { t } },
+      },
+      loaders: { Settings },
+    } = ctx;
+    const { category } = handler;
+
+    try {
+      // Get the settings.
+      const { organizationName = null } = await Settings.load(
+        'organizationName'
+      );
+      if (organizationName === null) {
+        ctx.log.debug(
+          'could not send the notification, organization name not in settings'
+        );
+        return;
+      }
+
+      // unsubscribeToken is the token used to perform the one-click
+      // unsubscribe.
+      const unsubscribeToken = jwt.sign({
+        jti: uuid(),
+        iss: JWT_ISSUER,
+        aud: JWT_AUDIENCE,
+        sub: UNSUBSCRIBE_SUBJECT,
+        user: userID,
+      });
+
+      // Compose the subject for the email.
+      const subject = t(
+        `talk-plugin-notifications.categories.${category}.subject`,
+        organizationName
+      );
+
+      // Load the content into the comment.
+      const body = await this.getBody(ctx, handler, context);
+
+      // Send the notification to the user.
+      const task = await Mailer.send({
+        template: 'notification',
+        locals: { body, organizationName, unsubscribeToken },
+        subject,
+        user: userID,
+      });
+
+      ctx.log.debug(`Sent the notification for Job.ID[${task.id}]`);
+    } catch (err) {
+      ctx.log.error(
+        { err, message: err.message },
+        'could not send the notification, an error occurred'
+      );
+      return;
+    }
+  }
+
+  /**
+   * getBody will return the body for the notification payload.
+   *
+   * @param {Object} ctx the graph context
+   * @param {Object} handler the notification handler
+   * @param {Mixed} context the notification context
+   */
+  async getBody(ctx, handler, context) {
+    const { connectors: { services: { I18n: { t } } } } = ctx;
+    const { category } = handler;
+
+    // Get the body replacement variables for the translation key.
+    const replacements = await handler.hydrate(ctx, category, context);
+
+    // Generate the body.
+    return t(
+      `talk-plugin-notifications.categories.${category}.body`,
+      ...replacements
+    );
+  }
+}
+
+module.exports = NotificationManager;

plugins/talk-plugin-notifications/server/config.js

@@ -0,0 +1,3 @@
+module.exports = {
+  UNSUBSCRIBE_SUBJECT: 'nunsub',
+};

plugins/talk-plugin-notifications/server/connect.js

@@ -0,0 +1,72 @@
+const debug = require('debug')('talk-plugin-notifications');
+const path = require('path');
+const linkify = require('linkifyjs/html');
+const NotificationManager = require('./NotificationManager');
+
+module.exports = connectors => {
+  const {
+    graph: { subscriptions: { getBroker }, Context },
+    services: { Mailer, Plugins },
+  } = connectors;
+
+  // Setup the mailer. Other plugins registered before this one can replace the
+  // notification template by passing the same name + format for the template
+  // registration.
+  Mailer.templates.register(
+    path.join(__dirname, 'emails', 'notification.html.ejs'),
+    'notification',
+    'html'
+  );
+  Mailer.templates.register(
+    path.join(__dirname, 'emails', 'notification.txt.ejs'),
+    'notification',
+    'txt'
+  );
+
+  // Register the mail helpers. You can register your own helpers by calling
+  // this function in another plugin.
+  Mailer.registerHelpers({ linkify });
+
+  // Get the handle for the broker to attach to notifications.
+  const broker = getBroker();
+
+  // Create a NotificationManager to handle notifications.
+  const manager = new NotificationManager(Context);
+
+  // Get all the notification handlers. Additional plugins registered before
+  // this one can expose a `notifications` hook, that contains an array of
+  // notification handlers.
+  //
+  // A notification handler has the following form:
+  //
+  // {
+  //   event     // the graph event to listen for
+  //   handle    // the function called when the event is fired. It is called with
+  //             // the (ctx, arg1, arg2, ...) where arg1, arg2 are args from the
+  //             // event.
+  //   category  // the name representing the notification type (like 'reply')
+  //   hydrate   // returns the replacement parameters (in order!) to be used
+  //             // in the translation.
+  // }
+  //
+  const notificationHandlers = Plugins.get('server', 'notifications').reduce(
+    (handlers, { plugin, notifications }) => {
+      debug(
+        `registered the ${
+          plugin.name
+        } plugin for notifications ${notifications.map(
+          ({ category }) => category
+        )}`
+      );
+      handlers.push(...notifications);
+      return handlers;
+    },
+    []
+  );
+
+  // Attach all the notification handlers.
+  manager.register(...notificationHandlers);
+
+  // Attach the broker to the manager so it can listen for the events.
+  manager.attach(broker);
+};

plugins/talk-plugin-notifications/server/emails/notification.html.ejs

@@ -0,0 +1,3 @@
+<p><%= linkify(body, {nl2br: true}) %></p>
+<p><%= t('talk-plugin-notifications.templates.footer', organizationName) %></p>
+<p><a href="<%= BASE_URL %>account/unsubscribe-notifications#<%= unsubscribeToken %>" target="_blank"><%= t('talk-plugin-notifications.templates.links.unsubscribe') %></a></p>

plugins/talk-plugin-notifications/server/emails/notification.txt.ejs

@@ -0,0 +1,7 @@
+<%= body %>
+
+<%= t('talk-plugin-notifications.templates.footer', organizationName) %>
+
+<%= t('talk-plugin-notifications.templates.links.unsubscribe') %>
+
+    <%= BASE_URL %>account/unsubscribe-notifications#<%= unsubscribeToken %>

plugins/talk-plugin-notifications/server/index.js

@@ -0,0 +1,16 @@
+const path = require('path');
+const connect = require('./connect');
+const typeDefs = require('./typeDefs');
+const resolvers = require('./resolvers');
+const router = require('./router');
+const mutators = require('./mutators');
+const translations = path.join(__dirname, 'translations.yml');
+
+module.exports = {
+  translations,
+  typeDefs,
+  resolvers,
+  mutators,
+  connect,
+  router,
+};

plugins/talk-plugin-notifications/server/mutators.js

@@ -0,0 +1,46 @@
+const { reduce, isNull, isEmpty } = require('lodash');
+
+/**
+ * Reduce the settings to dotize the settings.
+ */
+function reduceSettings(newSettings, newValue, key) {
+  if (!isNull(newValue)) {
+    newSettings[`metadata.notifications.settings.${key}`] = newValue;
+  }
+
+  return newSettings;
+}
+
+/**
+ * Update the user notification settings.
+ */
+async function updateNotificationSettings(ctx, settings) {
+  const { connectors: { models: { User } }, user } = ctx;
+
+  // Generate the settings set object, and just exit if we haven't changed
+  // anything.
+  const $set = reduce(settings, reduceSettings, {});
+  if (isEmpty($set)) {
+    return;
+  }
+
+  // Update the user.
+  return User.updateOne({ id: user.id }, { $set });
+}
+
+module.exports = ctx => {
+  let mutators = {
+    User: {
+      updateNotificationSettings: () =>
+        Promise.reject(ctx.connectors.errors.ErrNotAuthorized),
+    },
+  };
+
+  if (ctx.user) {
+    // TODO: check to see if the user is verified?
+    mutators.User.updateNotificationSettings = settings =>
+      updateNotificationSettings(ctx, settings);
+  }
+
+  return mutators;
+};

plugins/talk-plugin-notifications/server/resolvers.js

@@ -0,0 +1,19 @@
+const { get } = require('lodash');
+
+module.exports = {
+  User: {
+    notificationSettings(user, args, { user: currentUser }) {
+      if (
+        currentUser &&
+        (currentUser.id === user.id || currentUser.can('VIEW_USER_STATUS'))
+      ) {
+        return get(user, 'metadata.notifications.settings', {});
+      }
+    },
+  },
+  RootMutation: {
+    async updateNotificationSettings(obj, { input }, { mutators: { User } }) {
+      await User.updateNotificationSettings(input);
+    },
+  },
+};

plugins/talk-plugin-notifications/server/router.js

@@ -0,0 +1,95 @@
+const path = require('path');
+const { UNSUBSCRIBE_SUBJECT } = require('./config');
+const { get, isEmpty, reduce } = require('lodash');
+
+module.exports = router => {
+  router.get('/account/unsubscribe-notifications', (req, res) => {
+    res.render(path.join(__dirname, 'views/unsubscribe-notifications'));
+  });
+
+  /**
+   * Verifies that the token is valid.
+   */
+  const verifyToken = (req, res, next) => {
+    const {
+      connectors: { secrets: { jwt }, config: { JWT_ISSUER, JWT_AUDIENCE } },
+    } = req.context;
+    const { token: tokenString = '' } = req.body;
+    if (!tokenString) {
+      return res.status(400).end();
+    }
+
+    jwt.verify(
+      tokenString,
+      {
+        issuer: JWT_ISSUER,
+        subject: UNSUBSCRIBE_SUBJECT,
+        audience: JWT_AUDIENCE,
+      },
+      (err, token) => {
+        if (err) {
+          return res.status(400).end();
+        }
+
+        req.token = token;
+        next();
+      }
+    );
+  };
+
+  // Verifies that a token is valid.
+  router.post(
+    '/api/v1/account/unsubscribe-notifications/verify',
+    verifyToken,
+    (req, res) => {
+      res.status(204).end();
+    }
+  );
+
+  router.post(
+    '/api/v1/account/unsubscribe-notifications',
+    verifyToken,
+    async (req, res, next) => {
+      const { connectors: { models: { User } } } = req.context;
+      const { user: userID } = req.token;
+
+      try {
+        const user = await User.findOne({ id: userID });
+        if (!user) {
+          return res.status(400).end();
+        }
+
+        // Get the notification settings.
+        const settings = get(user, 'metadata.notifications.settings', {});
+
+        // If they have no notification settings set to true, then we're done.
+        if (isEmpty(settings)) {
+          return res.status(204).end();
+        }
+
+        const update = reduce(
+          settings,
+          (updates, value, key) => {
+            if (value) {
+              updates[`metadata.notifications.settings.${key}`] = false;
+            }
+
+            return updates;
+          },
+          {}
+        );
+
+        if (isEmpty(update)) {
+          return res.status(204).end();
+        }
+
+        // Save the user.
+        await User.update({ id: userID }, { $set: update });
+
+        res.status(204).end();
+      } catch (err) {
+        res.status(400).end();
+      }
+    }
+  );
+};

plugins/talk-plugin-notifications/server/translations.yml

@@ -0,0 +1,12 @@
+en:
+  talk-plugin-notifications:
+    templates:
+      footer: "You received this notification because you are a commenter on {0} and you opted in to receive notifications."
+      links:
+        unsubscribe: "Unsubscribe from comment notifications"
+    unsubscribe_page:
+      unsubscribe: "Unsubscribe from comment notifications"
+      click_to_confirm: "Click below to confirm that you would like to unsubscribe from all notifications"
+      confirm: "Confirm"
+      are_unsubscribed: "You are now unsubscribed from all notifications."
+      token_invalid: "Unsubscribe link is invalid, click the link from a more recent email or visit a comment stream and login to change your notification preferences"

plugins/talk-plugin-notifications/server/typeDefs.graphql

@@ -0,0 +1,21 @@
+# NotificationSettings stores all the preferences related to notifications.
+type NotificationSettings { }
+
+type User {
+    notificationSettings: NotificationSettings
+}
+
+type UpdateNotificationSettingsResponse implements Response {
+  # An array of errors relating to the mutation that occurred.
+  errors: [UserError!]
+}
+
+input NotificationSettingsInput {
+
+}
+
+type RootMutation {
+    # updateNotificationSettings will update the current user's notification
+    # settings.
+    updateNotificationSettings(input: NotificationSettingsInput!): UpdateNotificationSettingsResponse
+}

plugins/talk-plugin-notifications/server/typeDefs.js

@@ -0,0 +1,7 @@
+const fs = require('fs');
+const path = require('path');
+
+module.exports = fs.readFileSync(
+  path.join(__dirname, 'typeDefs.graphql'),
+  'utf8'
+);

plugins/talk-plugin-notifications/server/views/unsubscribe-notifications.ejs

@@ -0,0 +1,64 @@
+<!DOCTYPE html>
+<html>
+  <head>
+    <meta name="viewport" content="initial-scale=1, maximum-scale=1">
+    <title><%= t('talk-plugin-notifications.unsubscribe_page.unsubscribe') %></title>
+    <link rel="stylesheet" href="https://code.getmdl.io/1.2.1/material.indigo-pink.min.css">
+    <link rel="stylesheet" href="<%= BASE_PATH %>public/css/admin.css">
+    <%- include(root + '/partials/head') %>
+  </head>
+  <body class="confirm-email-page">
+    <div id="root">
+      <div class="error-console container"><%= t('talk-plugin-notifications.unsubscribe_page.token_invalid') %></div>
+      <div id="success" style="display:none;" class="legend container"><%= t('talk-plugin-notifications.unsubscribe_page.are_unsubscribed') %></div>
+      <form id="unsubscribe-form" class="container">
+        <legend class="legend"><%= t('talk-plugin-notifications.unsubscribe_page.click_to_confirm') %></legend>
+        <button type="submit"><%= t('talk-plugin-notifications.unsubscribe_page.confirm') %></button>
+      </form>
+    </div>
+    <script src="https://ajax.googleapis.com/ajax/libs/jquery/3.1.1/jquery.min.js"></script>
+    <script type="text/javascript">
+      $(function() {
+        var submitting = false;
+        var payload = JSON.stringify({token: location.hash.replace('#', '')});
+        function handleSubmit(e) {
+          e.preventDefault();
+
+          if (submitting) {
+              return;
+          }
+
+          submitting = true;
+          $('.error-console').removeClass('active');
+
+          $.ajax({
+            url: '<%= BASE_PATH %>api/v1/account/unsubscribe-notifications',
+            contentType: 'application/json',
+            method: 'POST',
+            data: payload,
+          }).then(function (success) {
+            $('#unsubscribe-form').fadeOut(function () {
+                $('#success').fadeIn();
+            });
+          }).catch(function () {
+            submitting = false;
+            $('.error-console').addClass('active');
+          });
+        }
+
+        $.ajax({
+          url: '<%= BASE_PATH %>api/v1/account/unsubscribe-notifications/verify',
+          contentType: 'application/json',
+          method: 'POST',
+          data: payload,
+        })
+        .then(function () {
+          $('#unsubscribe-form').fadeIn().on('submit', handleSubmit);
+        })
+        .catch(function () {
+          $('.error-console').addClass('active');
+        });
+      });
+    </script>
+  </body>
+</html>

plugins/talk-plugin-slack-notifications/server/config.js

@@ -5,7 +5,9 @@ const config = {
 
 if (process.env.NODE_ENV !== 'test' && !config.SLACK_WEBHOOK_URL) {
   // TODO this error should point users to Talk's Slack app once that's in place
-  throw new Error('Please set the TALK_SLACK_WEBHOOK_URL environment variable to use the slack-notifications plugin.');
+  throw new Error(
+    'Please set the TALK_SLACK_WEBHOOK_URL environment variable to use the slack-notifications plugin.'
+  );
 }
 
 module.exports = config;

plugins/talk-plugin-slack-notifications/server/hooks.js

@@ -1,5 +1,5 @@
 const fetch = require('node-fetch');
-const {SLACK_WEBHOOK_URL, SLACK_WEBHOOK_TIMEOUT} = require('./config');
+const { SLACK_WEBHOOK_URL, SLACK_WEBHOOK_TIMEOUT } = require('./config');
 const debug = require('debug')('talk:plugin:slack-notifications');
 
 // We don't add the hooks during _test_ as the Slack API is not available.
@@ -10,14 +10,9 @@ if (process.env.NODE_ENV === 'test') {
 module.exports = {
   RootMutation: {
     createComment: {
-      async post(_, {input}, context, _info, result) {
+      async post(root, args, context, info, result) {
         debug(`Posting notification to Slack webhook: ${SLACK_WEBHOOK_URL}`);
-        const {
-          comment: {
-            body: text,
-            created_at: createdAt
-          }
-        } = result;
+        const { comment: { body: text, created_at: createdAt } } = result;
         const username = context.user.username;
         process.nextTick(async () => {
           const response = await fetch(SLACK_WEBHOOK_URL, {
@@ -27,16 +22,22 @@ module.exports = {
             },
             timeout: SLACK_WEBHOOK_TIMEOUT,
             body: JSON.stringify({
-              attachments: [{
-                text: text,
-                footer: `Comment by ${username}`,
-                ts: Math.floor(Date.parse(createdAt) / 1000),
-              }]
+              attachments: [
+                {
+                  text: text,
+                  footer: `Comment by ${username}`,
+                  ts: Math.floor(Date.parse(createdAt) / 1000),
+                },
+              ],
             }),
           });
           if (!response.ok) {
             const responseText = await response.text();
-            console.trace(`Posting to Slack failed with HTTP code ${response.status} and body '${responseText}'`);
+            console.trace(
+              `Posting to Slack failed with HTTP code ${
+                response.status
+              } and body '${responseText}'`
+            );
           }
         });
         return result;

routes/index.js

@@ -1,7 +1,6 @@
 const SetupService = require('../services/setup');
 const authentication = require('../middleware/authentication');
 const cookieParser = require('cookie-parser');
-const debug = require('debug')('talk:routes');
 const enabled = require('debug').enabled;
 const errors = require('../errors');
 const express = require('express');
@@ -14,6 +13,7 @@ const staticMiddleware = require('express-static-gzip');
 const { DISABLE_STATIC_SERVER } = require('../config');
 const { passport } = require('../services/passport');
 const { MOUNT_PATH } = require('../url');
+const context = require('../middleware/context');
 
 const router = express.Router();
 
@@ -101,9 +101,12 @@ plugins.get('server', 'passport').forEach(plugin => {
 // Setup the PassportJS Middleware.
 router.use(passport.initialize());
 
+// Setup the Graph Context on the router.
+router.use(authentication, context);
+
 // Attach the authentication middleware, this will be responsible for decoding
 // (if present) the JWT on the request.
-router.use('/api', authentication, require('./api'));
+router.use('/api', require('./api'));
 
 //==============================================================================
 // DEVELOPMENT ROUTES
@@ -136,17 +139,8 @@ if (process.env.NODE_ENV !== 'production') {
   });
 }
 
-//==============================================================================
-// PLUGIN ROUTES
-//==============================================================================
-
-// Inject server route plugins.
-plugins.get('server', 'router').forEach(plugin => {
-  debug(`added plugin '${plugin.plugin.name}'`);
-
-  // Pass the root router to the plugin to mount it's routes.
-  plugin.router(router);
-});
+// Mount the plugin routes.
+router.use(require('./plugins'));
 
 //==============================================================================
 // ERROR HANDLING

routes/plugins.js

@@ -0,0 +1,23 @@
+const express = require('express');
+const debug = require('debug')('talk:routes:plugins');
+const plugins = require('../services/plugins');
+const staticTemplate = require('../middleware/staticTemplate');
+
+const router = express.Router();
+
+// Routes mounted from plugins won't have access to our internal partials
+// directory, so we should make that available.
+router.use(staticTemplate, (req, res, next) => {
+  res.locals.root = res.app.get('views');
+  next();
+});
+
+// Inject server route plugins.
+plugins.get('server', 'router').forEach(plugin => {
+  debug(`added plugin '${plugin.plugin.name}'`);
+
+  // Pass the root router to the plugin to mount it's routes.
+  plugin.router(router);
+});
+
+module.exports = router;

serve.js

@@ -2,10 +2,10 @@ const app = require('./app');
 const debug = require('debug')('talk:cli:serve');
 const errors = require('./errors');
 const { createServer } = require('http');
-const scraper = require('./services/scraper');
-const mailer = require('./services/mailer');
+const jobs = require('./jobs');
 const MigrationService = require('./services/migration');
 const SetupService = require('./services/setup');
+const PluginsService = require('./services/plugins');
 const kue = require('./services/kue');
 const mongoose = require('./services/mongoose');
 const cache = require('./services/cache');
@@ -78,7 +78,10 @@ async function onListening() {
 /**
  * Start the app.
  */
-async function serve({ jobs = false, websockets = false } = {}) {
+async function serve({ jobs: processJobs = false, websockets = false } = {}) {
+  // Run the deferred plugins.
+  PluginsService.runDeferred();
+
   // Start the cache instance.
   await cache.init();
 
@@ -132,19 +135,16 @@ async function serve({ jobs = false, websockets = false } = {}) {
   });
 
   // Enable job processing on the thread if enabled.
-  if (jobs) {
-    // Start the scraper processor.
-    scraper.process();
-
+  if (processJobs) {
     // Start the mail processor.
-    mailer.process();
+    jobs.process();
   }
 
   // Define a safe shutdown function to call in the event we need to shutdown
   // because the node hooks are below which will interrupt the shutdown process.
   // Shutdown the mongoose connection, the app server, and the scraper.
   util.onshutdown([
-    () => (jobs ? kue.Task.shutdown() : null),
+    () => (processJobs ? kue.Task.shutdown() : null),
     () => mongoose.disconnect(),
     () => server.close(),
   ]);

services/actions.js

@@ -3,8 +3,34 @@ const CommentModel = require('../models/comment');
 const UserModel = require('../models/user');
 const _ = require('lodash');
 const errors = require('../errors');
-const events = require('./events');
-const { ACTIONS_NEW, ACTIONS_DELETE } = require('./events/constants');
+
+const incrActionCounts = async (action, value) => {
+  const ACTION_TYPE = action.action_type.toLowerCase();
+
+  const query = { id: action.item_id };
+  const update = {
+    [`action_counts.${ACTION_TYPE}`]: value,
+  };
+
+  if (action.group_id && action.group_id.length > 0) {
+    const GROUP_ID = action.group_id.toLowerCase();
+
+    update[`action_counts.${ACTION_TYPE}_${GROUP_ID}`] = value;
+  }
+
+  switch (action.item_type) {
+    case 'USERS':
+      return UserModel.update(query, {
+        $inc: update,
+      });
+    case 'COMMENTS':
+      return CommentModel.update(query, {
+        $inc: update,
+      });
+    default:
+      return;
+  }
+};
 
 /**
  * findOnlyOneAndUpdate will perform a fondOneAndUpdate on the mongo collection
@@ -58,14 +84,12 @@ module.exports = class ActionsService {
   /**
    * Inserts an action.
    *
-   * @param {String} item_id  identifier of the item (uuid)
-   * @param {String} user_id  user id of the action (uuid)
    * @param {String} action   the new action to the item
    * @return {Promise}
    */
   static async create(action) {
-    // Actions are made unique by using a query that can be reproducable, i.e.,
-    // not containing user inputable values.
+    // Actions are made unique by using a query that can be reproducible, i.e.,
+    // not containing user modifiable values.
     let foundAction = await findOnlyOneAndUpdate(
       {
         action_type: action.action_type,
@@ -80,8 +104,7 @@ module.exports = class ActionsService {
       }
     );
 
-    // Emit that there was a new action created.
-    await events.emitAsync(ACTIONS_NEW, foundAction);
+    await incrActionCounts(action, 1);
 
     return foundAction;
   }
@@ -119,19 +142,6 @@ module.exports = class ActionsService {
     });
   }
 
-  /**
-   * Finds all comments for a specific action.
-   *
-   * @param {String} action_type type of action
-   * @param {String} item_type type of item the action is on
-   */
-  static findByType(action_type, item_type) {
-    return ActionModel.find({
-      action_type: action_type,
-      item_type: item_type,
-    });
-  }
-
   /**
    * delete will remove the record from the collection if it exists. Otherwise
    * it will do nothing. This will then return the deleted action.
@@ -148,8 +158,7 @@ module.exports = class ActionsService {
       return;
     }
 
-    // Emit that the action was deleted.
-    await events.emitAsync(ACTIONS_DELETE, action);
+    await incrActionCounts(action, -1);
 
     return action;
   }
@@ -170,68 +179,3 @@ module.exports = class ActionsService {
     );
   }
 };
-
-const incrActionCounts = async (action, value) => {
-  const ACTION_TYPE = action.action_type.toLowerCase();
-
-  const update = {
-    [`action_counts.${ACTION_TYPE}`]: value,
-  };
-
-  if (action.group_id && action.group_id.length > 0) {
-    const GROUP_ID = 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

@@ -1,22 +1,36 @@
 const CommentModel = require('../models/comment');
 
 const debug = require('debug')('talk:services:comments');
-const ActionsService = require('./actions');
 const SettingsService = require('./settings');
 
 const cloneDeep = require('lodash/cloneDeep');
 const errors = require('../errors');
-const events = require('./events');
 const merge = require('lodash/merge');
-const { COMMENTS_NEW, COMMENTS_EDIT } = require('./events/constants');
 
-module.exports = class CommentsService {
+const incrReplyCount = async (comment, value) => {
+  try {
+    await CommentModel.update(
+      {
+        id: comment.parent_id,
+      },
+      {
+        $inc: {
+          reply_count: value,
+        },
+      }
+    );
+  } catch (err) {
+    console.error("Can't mutate the reply count:", err);
+  }
+};
+
+module.exports = {
   /**
    * Creates a new Comment that came from a public source.
    * @param  {Object} input either a single comment or an array of comments.
    * @return {Promise}
    */
-  static async publicCreate(input) {
+  publicCreate: async input => {
     // Extract the parent_id from the comment, if there is one.
     const { status = 'NONE', parent_id = null } = input;
     const created_at = new Date();
@@ -54,27 +68,10 @@ module.exports = class CommentsService {
     );
 
     // Emit that the comment was created!
-    await events.emitAsync(COMMENTS_NEW, comment);
+    await incrReplyCount(comment, 1);
 
     return comment;
-  }
-
-  /**
-   * lastUnmoderatedStatus will retrieve the last status before this one.
-   *
-   * @param {Object} comment the comment to get the last status of
-   */
-  static lastUnmoderatedStatus(comment) {
-    const UNMODERATED_STATUSES = ['NONE', 'PREMOD'];
-
-    for (let i = comment.status_history.length - 1; i >= 0; i--) {
-      const { type } = comment.status_history[i];
-
-      if (UNMODERATED_STATUSES.includes(type)) {
-        return type;
-      }
-    }
-  }
+  },
 
   /**
    * Edit a Comment.
@@ -84,7 +81,7 @@ module.exports = class CommentsService {
    * @param {String} body       the new Comment body
    * @param {String} status     the new Comment status
    */
-  static async edit({ id, author_id, body, status }) {
+  edit: async ({ id, author_id, body, status }) => {
     const EDITABLE_STATUSES = ['NONE', 'PREMOD', 'ACCEPTED'];
     const created_at = new Date();
 
@@ -125,7 +122,7 @@ module.exports = class CommentsService {
 
     if (originalComment == null) {
       // Try to get the comment.
-      const comment = await CommentsService.findById(id);
+      const comment = await CommentModel.findOne({ id });
       if (comment == null) {
         debug('rejecting comment edit because comment was not found');
         throw errors.ErrNotFound;
@@ -169,86 +166,8 @@ module.exports = class CommentsService {
       created_at,
     });
 
-    await events.emitAsync(COMMENTS_EDIT, originalComment, editedComment);
-
     return editedComment;
-  }
-
-  /**
-   * Finds a comment by the id.
-   * @param {String} id  identifier of comment (uuid)
-   * @return {Promise}
-   */
-  static findById(id) {
-    return CommentModel.findOne({ id });
-  }
-
-  /**
-   * Finds ALL the comments by the asset_id.
-   * @param {String} asset_id  identifier of the asset which owns this comment (uuid)
-   * @return {Promise}
-   */
-  static findByAssetId(asset_id) {
-    return CommentModel.find({
-      asset_id,
-    });
-  }
-
-  /**
-   * findByAssetIdWithStatuses finds all the comments where the asset id matches
-   * what's provided and the status is one of the ones listed in the statuses
-   * array.
-   * @param {String} asset_id      the asset id to search by
-   * @param {Array}  [statuses=[]] the array of statuses to search by
-   * @return {Promise}             resolves to an array of comments
-   */
-  static findByAssetIdWithStatuses(asset_id, statuses = []) {
-    return CommentModel.find({
-      asset_id,
-      status: {
-        $in: statuses,
-      },
-    });
-  }
-
-  /**
-   * Find comments by an action that was performed on them.
-   * @param {String} action_type the type of action that was performed on the comment
-   * @return {Promise}
-   */
-  static findByActionType(action_type) {
-    return ActionsService.findCommentsIdByActionType(
-      action_type,
-      'COMMENTS'
-    ).then(actions =>
-      CommentModel.find({
-        id: {
-          $in: actions.map(a => a.item_id),
-        },
-      })
-    );
-  }
-
-  /**
-   * Find comment id's where the action type matches the argument.
-   * @param {String} action_type the type of action that was performed on the comment
-   * @return {Promise}
-   */
-  static findIdsByActionType(action_type) {
-    return ActionsService.findCommentsIdByActionType(
-      action_type,
-      'COMMENTS'
-    ).then(actions => actions.map(a => a.item_id));
-  }
-
-  /**
-   * Find comments by current status
-   * @param {String} status status of the comment to search for
-   * @return {Promise} resovles to comment array
-   */
-  static findByStatus(status = 'NONE') {
-    return CommentModel.find({ status });
-  }
+  },
 
   /**
    * Pushes a new status in for the user.
@@ -258,7 +177,7 @@ module.exports = class CommentsService {
    *                             moderation action
    * @return {Promise}
    */
-  static async pushStatus(id, status, assigned_by = null) {
+  pushStatus: async (id, status, assigned_by = null) => {
     const created_at = new Date();
     const originalComment = await CommentModel.findOneAndUpdate(
       { id },
@@ -286,83 +205,16 @@ module.exports = class CommentsService {
     });
     editedComment.status = status;
 
-    // Emit that the comment was edited, and pass the original comment and the
-    // edited comment.
-    await events.emitAsync(COMMENTS_EDIT, originalComment, editedComment);
+    // If the comment was visible before, and now it isn't, decrement the count;
+    if (originalComment.visible && !editedComment.visible) {
+      await incrReplyCount(editedComment, -1);
+    }
+
+    // If the comment was not visible before, and now it is, increment the count.
+    if (!originalComment.visible && editedComment.visible) {
+      await incrReplyCount(editedComment, 1);
+    }
 
     return editedComment;
-  }
-
-  /**
-   * Add an action to the comment.
-   * @param {String} item_id  identifier of the comment  (uuid)
-   * @param {String} user_id  user id of the action (uuid)
-   * @param {String} action the new action to the comment
-   * @return {Promise}
-   */
-  static addAction(item_id, user_id, action_type, metadata = {}) {
-    return ActionsService.create({
-      item_id,
-      item_type: 'COMMENTS',
-      user_id,
-      action_type,
-      metadata,
-    });
-  }
+  },
 };
-
-//==============================================================================
-// Event Hooks
-//==============================================================================
-
-const incrReplyCount = async (comment, value) => {
-  try {
-    await CommentModel.update(
-      {
-        id: comment.parent_id,
-      },
-      {
-        $inc: {
-          reply_count: value,
-        },
-      }
-    );
-  } catch (err) {
-    console.error("Can't mutate the reply count:", err);
-  }
-};
-
-// When a comment is created, if it is a reply, increment the reply count on the
-// parent's document.
-events.on(COMMENTS_NEW, async comment => {
-  if (
-    !comment || // Check that the comment is defined.
-    (!comment.parent_id || comment.parent_id.length === 0) || // Check that the comment has a parent (is a reply).
-    !(comment.status === 'NONE' || comment.status === 'ACCEPTED') // Check that the comment is visible.
-  ) {
-    return;
-  }
-
-  return incrReplyCount(comment, 1);
-});
-
-// When a comment is edited, if the visability changed publicly, then modify the
-// comment.
-events.on(COMMENTS_EDIT, async (originalComment, editedComment) => {
-  if (
-    !editedComment || // Check that the comment is defined.
-    (!editedComment.parent_id || editedComment.parent_id.length === 0) // Check that the comment has a parent (is a reply).
-  ) {
-    return;
-  }
-
-  // If the comment was visible before, and now it isn't, decrement the count;
-  if (originalComment.visible && !editedComment.visible) {
-    return incrReplyCount(editedComment, -1);
-  }
-
-  // If the comment was not visible before, and now it is, increment the count.
-  if (!originalComment.visible && editedComment.visible) {
-    return incrReplyCount(editedComment, 1);
-  }
-});

services/events/constants.js

@@ -1,10 +0,0 @@
-module.exports = {
-  USERS_NEW: 'USERS_NEW',
-  ACTIONS_DELETE: 'ACTIONS_DELETE',
-  ACTIONS_NEW: 'ACTIONS_NEW',
-  COMMENTS_NEW: 'COMMENTS_NEW',
-  COMMENTS_EDIT: 'COMMENTS_EDIT',
-  USERS_SUSPENSION_CHANGE: 'USERS_SUSPENSION_CHANGE',
-  USERS_BAN_CHANGE: 'USERS_BAN_CHANGE',
-  USERS_USERNAME_STATUS_CHANGE: 'USERS_USERNAME_STATUS_CHANGE',
-};

services/events/index.js

@@ -1,36 +0,0 @@
-const { EventEmitter2 } = require('eventemitter2');
-const constants = require('./constants');
-const debug = require('debug')('talk:services:events');
-const enabled = require('debug').enabled('talk:services:events');
-
-const emitter = new EventEmitter2({
-  wildcard: true,
-});
-
-// If event debugging is enabled, bind the debugger to all events being emitted
-// and log a debug message.
-if (enabled) {
-  emitter.onAny(function(event) {
-    debug(
-      `[${event}] ${arguments.length - 1} argument${
-        arguments.length - 1 === 1 ? '' : 's'
-      }`
-    );
-  });
-}
-
-// Allow any number of listeners to attach to this.
-emitter.setMaxListeners(0);
-
-// The default error handler.
-emitter.on('error', err => {
-  console.error('events error:', err);
-});
-
-emitter.on('newListener', event => {
-  if (!(event in constants)) {
-    throw new Error(`Event[${event}] not a valid event name`);
-  }
-});
-
-module.exports = emitter;

services/jwt.js

@@ -101,9 +101,12 @@ class Secret {
     jwt.verify(
       token,
       this.verifiyingKey,
-      Object.assign({}, options, {
-        algorithms: [this.algorithm],
-      }),
+      omitBy(
+        merge({}, options, {
+          algorithms: [this.algorithm],
+        }),
+        isUndefined
+      ),
       callback
     );
   }

services/logging.js

@@ -0,0 +1,16 @@
+const { version } = require('../package.json');
+const Logger = require('bunyan');
+const uuid = require('uuid/v1');
+
+// Create the logging instance that all logger's are branched from.
+function createLogger(name, id = uuid()) {
+  return new Logger({
+    src: true,
+    name,
+    id,
+    version,
+    serializers: { req: Logger.stdSerializers.req },
+  });
+}
+
+module.exports = { createLogger };

services/mailer.js

@@ -1,148 +0,0 @@
-const debug = require('debug')('talk:services:mailer');
-const nodemailer = require('nodemailer');
-const kue = require('./kue');
-const i18n = require('./i18n');
-const path = require('path');
-const fs = require('fs-extra');
-const _ = require('lodash');
-const { TEMPLATE_LOCALS } = require('../middleware/staticTemplate');
-
-const {
-  SMTP_HOST,
-  SMTP_USERNAME,
-  SMTP_PORT,
-  SMTP_PASSWORD,
-  SMTP_FROM_ADDRESS,
-  EMAIL_SUBJECT_PREFIX,
-} = require('../config');
-
-// load all the templates as strings
-const templates = {
-  data: {},
-};
-
-// load the templates per request during development
-templates.render = async (name, format = 'txt', context) => {
-  if (process.env.NODE_ENV === 'production') {
-    // If we are in production mode, check the view cache.
-    const view = _.get(templates.data, [name, format], null);
-    if (view !== null) {
-      return view(context);
-    }
-  }
-
-  const filename = path.join(
-    __dirname,
-    'email',
-    [name, format, 'ejs'].join('.')
-  );
-  const file = await fs.readFile(filename, 'utf8');
-  const view = _.template(file);
-
-  if (process.env.NODE_ENV === 'production') {
-    // If we are in production mode, fill the view cache.
-    _.set(templates.data, [name, format], view);
-  }
-
-  return view(context);
-};
-
-const mailer = {};
-
-// enabled is true when the required configuration is available. When testing
-// is enabled, we will be simulating that emails are being sent, because in a
-// production system, emails should and would be sent.
-mailer.enabled =
-  Boolean(SMTP_HOST && SMTP_USERNAME && SMTP_PASSWORD && SMTP_FROM_ADDRESS) ||
-  process.env.NODE_ENV === 'test';
-
-if (mailer.enabled) {
-  const options = {
-    host: SMTP_HOST,
-    auth: {
-      user: SMTP_USERNAME,
-      pass: SMTP_PASSWORD,
-    },
-  };
-
-  if (SMTP_PORT) {
-    try {
-      options.port = parseInt(SMTP_PORT);
-    } catch (e) {
-      throw new Error('TALK_SMTP_PORT is not an integer');
-    }
-  } else {
-    options.port = 25;
-  }
-
-  mailer.transport = nodemailer.createTransport(options);
-}
-
-/**
- * Create the new Task kue.
- */
-mailer.task = new kue.Task({
-  name: 'mailer',
-});
-
-/**
- * send will create a new message and send it.
- */
-mailer.send = async options => {
-  if (!mailer.enabled) {
-    const err = new Error(
-      'sending email is not enabled because required configuration is not available'
-    );
-    console.warn(err);
-    return;
-  }
-
-  // Create the new locals object and attach the static locals and the i18n
-  // framework.
-  const locals = _.merge({}, options.locals, TEMPLATE_LOCALS, { t: i18n.t });
-
-  // Render the templates.
-  const [html, text] = await Promise.all(
-    ['html', 'txt'].map(fmt => {
-      return templates.render(options.template, fmt, locals);
-    })
-  );
-
-  // Create the job to send the email later.
-  return mailer.task.create({
-    title: 'Mail',
-    message: {
-      to: options.to,
-      subject: `${EMAIL_SUBJECT_PREFIX} ${options.subject}`,
-      text,
-      html,
-    },
-  });
-};
-
-/**
- * Start the queue processor for the mailer job.
- */
-mailer.process = () => {
-  debug(`Now processing ${mailer.task.name} jobs`);
-
-  return mailer.task.process(({ id, data }, done) => {
-    debug(`Starting to send mail for Job[${id}]`);
-
-    // Set the `from` field.
-    data.message.from = SMTP_FROM_ADDRESS;
-
-    // Actually send the email.
-    mailer.transport.sendMail(data.message, err => {
-      if (err) {
-        debug(`Failed to send mail for Job[${id}]:`, err);
-        return done(err);
-      }
-
-      debug(`Finished sending mail for Job[${id}]`);
-      return done();
-    });
-  });
-};
-
-module.exports = mailer;

services/mailer/index.js

@@ -0,0 +1,112 @@
+const kue = require('../kue');
+const i18n = require('../i18n');
+const { get, merge, isObject } = require('lodash');
+const { TEMPLATE_LOCALS } = require('../../middleware/staticTemplate');
+const debug = require('debug')('talk:services:mailer');
+const { SMTP_FROM_ADDRESS, EMAIL_SUBJECT_PREFIX } = require('../../config');
+const templates = require('./templates');
+
+// createRecipient will extract the recipient details from the options.
+const createRecipient = options => {
+  // Try to get the email if it was explicitly provided.
+  const email = get(options, 'email');
+  if (email) {
+    return { email };
+  }
+
+  // Try to get the user if the email wasn't explicitly provided.
+  const user = isObject(options.user) ? get(options.user, 'id') : options.user;
+  if (user) {
+    return { user };
+  }
+
+  // If we don't have a user or a email, we can't send an email.
+  throw new Error('user/email not provided');
+};
+
+// createMessage creates a message payload to send a email to a user.
+const createMessage = async options => {
+  // Create the new locals object and attach the static locals and the i18n
+  // framework.
+  const locals = merge({}, mailer.helpers, options.locals, TEMPLATE_LOCALS, {
+    t: i18n.t,
+  });
+
+  // Render the templates.
+  const [html, text] = await Promise.all(
+    ['html', 'txt'].map(fmt => {
+      return mailer.templates.render(options.template, fmt, locals);
+    })
+  );
+
+  const subject = EMAIL_SUBJECT_PREFIX
+    ? `${EMAIL_SUBJECT_PREFIX} ${options.subject}`
+    : options.subject;
+
+  return {
+    html,
+    text,
+    subject,
+    from: SMTP_FROM_ADDRESS,
+  };
+};
+
+const mailer = { templates, helpers: {} };
+
+/**
+ * Create the new Task kue.
+ */
+mailer.task = new kue.Task({
+  name: 'mailer',
+});
+
+/**
+ * registerHelpers will register the helpers on the mailer.
+ *
+ * @param {Object} helpers the helpers in object form that should be used by the
+ *                         mailer.
+ */
+mailer.registerHelpers = helpers => {
+  mailer.helpers = merge(mailer.helpers, helpers);
+};
+
+/**
+ * queue will add the message to the sending queue.
+ *
+ * @param {Object} message the message to be sent
+ * @param {Object} recipient the recipient to send it to
+ */
+mailer.queue = async (message, recipient) => {
+  debug('Creating Job');
+
+  // Create the job to send the email later.
+  const job = await mailer.task.create(
+    merge(
+      {
+        title: 'Mail',
+        message,
+      },
+      recipient
+    )
+  );
+
+  debug(`Created Job[${job.id}]`);
+
+  return job;
+};
+
+/**
+ * send will prepare the message and queue the message to be sent.
+ */
+mailer.send = async options => {
+  // Create the recipient to sent the message to.
+  const recipient = createRecipient(options);
+
+  // Create the message to send.
+  const message = await createMessage(options);
+
+  // Create the job to send the message.
+  return mailer.queue(message, recipient);
+};
+
+module.exports = mailer;

services/mailer/templates.js

@@ -0,0 +1,58 @@
+const path = require('path');
+const fs = require('fs-extra');
+const { get, set, template } = require('lodash');
+
+// load all the templates as strings
+const templates = {
+  cache: {},
+  registered: {},
+};
+
+// Registers a template with the given filename and format.
+templates.register = async (filename, name, format) => {
+  // Check to see if this template was already registered.
+  if (get(templates.registered, [name, format], null) !== null) {
+    return;
+  }
+
+  const file = await fs.readFile(filename, 'utf8');
+  const view = template(file);
+
+  set(templates.registered, [name, format], view);
+};
+
+// load the templates per request during development
+templates.render = async (name, format = 'txt', context) => {
+  // Check to see if the template is a registered template (provided by a plugin
+  // ) and prefer that first.
+  let view = get(templates.registered, [name, format], null);
+  if (view !== null) {
+    return view(context);
+  }
+
+  if (process.env.NODE_ENV === 'production') {
+    // If we are in production mode, check the view cache.
+    const view = get(templates.cache, [name, format], null);
+    if (view !== null) {
+      return view(context);
+    }
+  }
+
+  // Template was not registered and was not cached. Let's try and find it!
+  const filename = path.join(
+    __dirname,
+    'templates',
+    [name, format, 'ejs'].join('.')
+  );
+  const file = await fs.readFile(filename, 'utf8');
+  view = template(file);
+
+  if (process.env.NODE_ENV === 'production') {
+    // If we are in production mode, fill the view cache.
+    set(templates.cache, [name, format], view);
+  }
+
+  return view(context);
+};
+
+module.exports = templates;

services/pubsub.js

@@ -1,24 +0,0 @@
-const { RedisPubSub } = require('graphql-redis-subscriptions');
-const { createClient } = require('./redis');
-
-/**
- * getClient returns the pubsub singleton for this instance.
- */
-let pubsub = null;
-const getClient = () => {
-  if (pubsub !== null) {
-    return pubsub;
-  }
-
-  // Create the new PubSub client, we only need one per instance of Talk.
-  pubsub = new RedisPubSub({
-    publisher: createClient(),
-    subscriber: createClient(),
-  });
-
-  return pubsub;
-};
-
-module.exports = {
-  getClient,
-};

services/scraper.js

@@ -1,9 +1,5 @@
 const kue = require('./kue');
 const debug = require('debug')('talk:services:scraper');
-const AssetModel = require('../models/asset');
-const AssetsService = require('./assets');
-
-const metascraper = require('metascraper');
 
 /**
  * Exposes a service object to allow operations to execute against the scraper.
@@ -20,93 +16,17 @@ const scraper = {
   /**
    * Creates a new scraper job and scrapes the url when it gets processed.
    */
-  create(asset) {
+  async create(asset) {
     debug(`Creating job for Asset[${asset.id}]`);
 
-    return scraper.task
-      .create({
-        title: `Scrape for asset ${asset.id}`,
-        asset_id: asset.id,
-      })
-      .then(job => {
-        debug(`Created Job[${job.id}] for Asset[${asset.id}]`);
-
-        return job;
-      });
-  },
-
-  /**
-   * Scrapes the given asset for metadata.
-   */
-  async scrape(asset) {
-    return metascraper.scrapeUrl(
-      asset.url,
-      Object.assign({}, metascraper.RULES, {
-        section: $ => $('meta[property="article:section"]').attr('content'),
-        modified: $ => $('meta[property="article:modified"]').attr('content'),
-      })
-    );
-  },
-
-  /**
-   * Updates an Asset based on scraped asset metadata.
-   */
-  update(id, meta) {
-    return AssetModel.update(
-      { id },
-      {
-        $set: {
-          title: meta.title || '',
-          description: meta.description || '',
-          image: meta.image ? meta.image : '',
-          author: meta.author || '',
-          publication_date: meta.date || '',
-          modified_date: meta.modified || '',
-          section: meta.section || '',
-          scraped: new Date(),
-        },
-      }
-    );
-  },
-
-  /**
-   * Start the queue processor for the scraper job.
-   */
-  process() {
-    debug(`Now processing ${scraper.task.name} jobs`);
-
-    scraper.task.process(async (job, done) => {
-      debug(`Starting on Job[${job.id}] for Asset[${job.data.asset_id}]`);
-
-      try {
-        // Find the asset, or complain that it doesn't exist.
-        const asset = await AssetsService.findById(job.data.asset_id);
-        if (!asset) {
-          return done(new Error('asset not found'));
-        }
-
-        // Scrape the metadata from the asset.
-        const meta = await scraper.scrape(asset);
-
-        debug(
-          `Scraped ${JSON.stringify(meta)} on Job[${job.id}] for Asset[${
-            job.data.asset_id
-          }]`
-        );
-
-        // Assign the metadata retrieved for the asset to the db.
-        await scraper.update(job.data.asset_id, meta);
-      } catch (err) {
-        debug(
-          `Failed to scrape on Job[${job.id}] for Asset[${job.data.asset_id}]:`,
-          err
-        );
-        return done(err);
-      }
-
-      debug(`Finished on Job[${job.id}] for Asset[${job.data.asset_id}]`);
-      done();
+    const job = await scraper.task.create({
+      title: `Scrape for asset ${asset.id}`,
+      asset_id: asset.id,
     });
+
+    debug(`Created Job[${job.id}] for Asset[${asset.id}]`);
+
+    return job;
   },
 };
 

services/users.js

@@ -1,28 +1,13 @@
 const uuid = require('uuid');
 const bcrypt = require('bcryptjs');
 const errors = require('../errors');
-const some = require('lodash/some');
-const merge = require('lodash/merge');
-
-const {
-  USERS_NEW,
-  USERS_SUSPENSION_CHANGE,
-  USERS_BAN_CHANGE,
-  USERS_USERNAME_STATUS_CHANGE,
-} = require('./events/constants');
-const events = require('./events');
-
+const { some, merge } = require('lodash');
 const { ROOT_URL } = require('../config');
-
 const { jwt: JWT_SECRET } = require('../secrets');
-
 const debug = require('debug')('talk:services:users');
-
 const UserModel = require('../models/user');
-
 const RECAPTCHA_WINDOW = '10m'; // 10 minutes.
 const RECAPTCHA_INCORRECT_TRIGGER = 5; // after 5 incorrect attempts, recaptcha will be required.
-
 const ActionsService = require('./actions');
 const mailer = require('./mailer');
 const i18n = require('./i18n');
@@ -125,12 +110,16 @@ class UsersService {
       );
     }
 
-    // Emit that the user username status was changed.
-    await events.emitAsync(USERS_SUSPENSION_CHANGE, user, {
-      until,
-      message,
-      assignedBy,
-    });
+    // Check to see if the user was suspended now and is currently suspended.
+    if (user.suspended && message && message.length > 0) {
+      await UsersService.sendEmail(user, {
+        template: 'plain',
+        locals: {
+          body: message,
+        },
+        subject: 'Your account has been suspended',
+      });
+    }
 
     return user;
   }
@@ -174,12 +163,16 @@ class UsersService {
       throw new Error('ban status change edit failed for an unknown reason');
     }
 
-    // Emit that the user ban status was changed.
-    await events.emitAsync(USERS_BAN_CHANGE, user, {
-      status,
-      assignedBy,
-      message,
-    });
+    // Check to see if the user was banned now and is currently banned.
+    if (user.banned && status && message && message.length > 0) {
+      await UsersService.sendEmail(user, {
+        template: 'plain',
+        locals: {
+          body: message,
+        },
+        subject: 'Your account has been banned',
+      });
+    }
 
     return user;
   }
@@ -223,12 +216,6 @@ class UsersService {
       );
     }
 
-    // Emit that the user username status was changed.
-    await events.emitAsync(USERS_USERNAME_STATUS_CHANGE, user, {
-      status,
-      assignedBy,
-    });
-
     return user;
   }
 
@@ -286,9 +273,6 @@ class UsersService {
         throw new Error('edit username failed for an unexpected reason');
       }
 
-      // Emit that the user username status was changed.
-      await events.emitAsync(USERS_USERNAME_STATUS_CHANGE, user, toStatus);
-
       return user;
     } catch (err) {
       if (err.code === 11000) {
@@ -404,16 +388,14 @@ class UsersService {
     // Save the user in the database.
     await user.save();
 
-    // Emit that the user was created.
-    await events.emitAsync(USERS_NEW, user);
-
     return user;
   }
 
   /**
    * sendEmailConfirmation sends a confirmation email to the user.
+   *
    * @param {String}     user  the user to send the email to
-   * @param {String}     email   the email for the user to send the email to
+   * @param {String}     email the email for the user to send the email to
    */
   static async sendEmailConfirmation(user, email, redirectURI = ROOT_URL) {
     let token = await UsersService.createEmailConfirmToken(
@@ -430,21 +412,14 @@ class UsersService {
         email,
       },
       subject: i18n.t('email.confirm.subject'),
-      to: email,
+      email,
     });
   }
 
   static async sendEmail(user, options) {
-    const email = user.firstEmail;
-    if (!email) {
-      // Rather than throwing an error here, we'll
-      console.warn(new Error('user does not have an email'));
-      return;
-    }
-
     return mailer.send(
       merge({}, options, {
-        to: email,
+        user: user.id,
       })
     );
   }
@@ -578,9 +553,6 @@ class UsersService {
       throw err;
     }
 
-    // Emit that the user was created.
-    await events.emitAsync(USERS_NEW, user);
-
     return user;
   }
 
@@ -952,38 +924,6 @@ class UsersService {
 
 module.exports = UsersService;
 
-events.on(USERS_BAN_CHANGE, async (user, { status, message }) => {
-  // Check to see if the user was banned now and is currently banned.
-  if (user.banned && status && message && message.length > 0) {
-    await UsersService.sendEmail(user, {
-      template: 'plain',
-      locals: {
-        body: message,
-      },
-      subject: 'Your account has been banned',
-    });
-  }
-});
-
-events.on(USERS_SUSPENSION_CHANGE, async (user, { until, message }) => {
-  // Check to see if the user was suspended now and is currently suspended.
-  if (
-    user.suspended &&
-    until !== null &&
-    until > Date.now() &&
-    message &&
-    message.length > 0
-  ) {
-    await UsersService.sendEmail(user, {
-      template: 'plain',
-      locals: {
-        body: message,
-      },
-      subject: 'Your account has been suspended',
-    });
-  }
-});
-
 // Extract all the tokenUserNotFound plugins so we can integrate with other
 // providers.
 let tokenUserNotFoundHooks = null;

test/server/graph/context.js

@@ -39,7 +39,7 @@ describe('graph.Context', () => {
     });
 
     it('creates a context without a user', done => {
-      expect(c).to.not.have.property('user');
+      expect(c.user).to.be.falsy;
 
       done();
     });

test/server/graph/mutations/addTag.js

@@ -3,6 +3,7 @@ const { graphql } = require('graphql');
 const schema = require('../../../../graph/schema');
 const Context = require('../../../../graph/context');
 const AssetModel = require('../../../../models/asset');
+const CommentModel = require('../../../../models/comment');
 const UserModel = require('../../../../models/user');
 const SettingsService = require('../../../../services/settings');
 const CommentsService = require('../../../../services/comments');
@@ -50,7 +51,7 @@ describe('graph.mutations.addTag', () => {
 
     expect(res.errors).to.be.empty;
 
-    let { tags } = await CommentsService.findById(comment.id);
+    let { tags } = await CommentModel.findOne({ id: comment.id });
     expect(tags).to.have.length(1);
   });
 

test/server/graph/mutations/createComment.js

@@ -3,12 +3,12 @@ const { graphql } = require('graphql');
 const schema = require('../../../../graph/schema');
 const Context = require('../../../../graph/context');
 
-const UserModel = require('../../../../models/user');
-const AssetModel = require('../../../../models/asset');
 const ActionModel = require('../../../../models/action');
+const AssetModel = require('../../../../models/asset');
+const CommentModel = require('../../../../models/comment');
+const UserModel = require('../../../../models/user');
 
 const SettingsService = require('../../../../services/settings');
-const CommentsService = require('../../../../services/comments');
 
 const { expect } = require('chai');
 
@@ -336,9 +336,9 @@ describe('graph.mutations.createComment', () => {
           expect(data.createComment).to.have.property('comment').not.null;
           expect(data.createComment).to.have.property('errors').null;
 
-          const { tags } = await CommentsService.findById(
-            data.createComment.comment.id
-          );
+          const { tags } = await CommentModel.findOne({
+            id: data.createComment.comment.id,
+          });
           if (tag) {
             expect(tags).to.have.length(1);
             expect(tags[0].tag.name).to.have.equal(tag);