diff --git a/bin/cli-jobs b/bin/cli-jobs index 672a5a70d..a9599213d 100755 --- a/bin/cli-jobs +++ b/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(); } /** diff --git a/graph/loaders/users.js b/graph/loaders/users.js index 7b797130d..2c1e9fbac 100644 --- a/graph/loaders/users.js +++ b/graph/loaders/users.js @@ -1,11 +1,7 @@ const DataLoader = require('dataloader'); - const util = require('./util'); - const { SEARCH_OTHER_USERS } = require('../../perms/constants'); - const { escapeRegExp } = require('../../services/regex'); -const UserModel = require('../../models/user'); const mergeState = (query, state) => { const { status } = state; @@ -50,14 +46,14 @@ const mergeState = (query, state) => { } }; -const genUserByIDs = async (context, ids) => { +const genUserByIDs = async (ctx, ids) => { if (!ids || ids.length === 0) { return []; } - return UserModel.find({ id: { $in: ids } }).then( - util.singleJoinBy(ids, 'id') - ); + const { connectors: { models: { User } } } = ctx; + + return User.find({ id: { $in: ids } }).then(util.singleJoinBy(ids, 'id')); }; /** @@ -67,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)) { @@ -178,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)) { @@ -199,7 +198,7 @@ const getCountByQuery = async ({ user }, { action_type, state }) => { } } - return UserModel.find(query).count(); + return query.count(); }; /** diff --git a/jobs/index.js b/jobs/index.js new file mode 100644 index 000000000..b64e59362 --- /dev/null +++ b/jobs/index.js @@ -0,0 +1,5 @@ +const jobs = [require('./mailer')]; + +const process = () => jobs.forEach(job => job()); + +module.exports = { process }; diff --git a/jobs/mailer.js b/jobs/mailer.js new file mode 100644 index 000000000..aa771bdeb --- /dev/null +++ b/jobs/mailer.js @@ -0,0 +1,62 @@ +const mailer = require('../services/mailer'); +const debug = require('debug')('talk:jobs:mailer'); +const Context = require('../graph/context'); + +/** + * Start the queue processor for the mailer job. + */ +module.exports = () => { + debug(`Now processing ${mailer.task.name} jobs`); + + return mailer.task.process(async ({ id, data }, done) => { + const { email, user, message } = data; + + debug(`Starting to send mail for Job[${id}]`); + + // 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) { + message.to = email; + } else { + // Get the user to send the message to. + const ctx = Context.forSystem(); + + try { + const { data, errors } = await ctx.graphql( + ` + query GetUserEmail($user: ID!) { + user(id: $user) { + email + } + } + `, + { user } + ); + if (errors) { + return done(errors); + } + const email = get(data, 'user.email'); + if (!email) { + return done(errors.ErrMissingEmail); + } + + message.to = email; + } catch (err) { + return done(err); + } + } + + // Actually send the email. + mailer.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(); + }); + }); +}; diff --git a/jobs/scraper.js b/jobs/scraper.js new file mode 100644 index 000000000..424304ba8 --- /dev/null +++ b/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(); + }); +}; diff --git a/plugins/talk-plugin-facebook-auth/server/router.js b/plugins/talk-plugin-facebook-auth/server/router.js index 53b6c7e56..fd6253fbe 100644 --- a/plugins/talk-plugin-facebook-auth/server/router.js +++ b/plugins/talk-plugin-facebook-auth/server/router.js @@ -1,26 +1,28 @@ module.exports = router => { - const { passport, HandleAuthPopupCallback } = require('services/passport'); - /** - * Facebook auth endpoint, this will redirect the user immediatly to facebook + * 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 } } } = 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 sucesfull login. + * send back the user credentials upon successful login. */ router.get('/api/v1/auth/facebook/callback', (req, res, next) => { + const { connectors: { services: { Passport } } } = req.context; + const { HandleAuthPopupCallback } = Passport; + // Perform the facebook login flow and pass the data back through the opener. - passport.authenticate( + Passport.authenticate( 'facebook', { session: false }, HandleAuthPopupCallback(req, res, next) diff --git a/serve.js b/serve.js index 0e3552bab..c5a1c8807 100644 --- a/serve.js +++ b/serve.js @@ -2,8 +2,7 @@ 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'); @@ -79,7 +78,7 @@ async function onListening() { /** * Start the app. */ -async function serve({ jobs = false, websockets = false } = {}) { +async function serve({ jobs: processJobs = false, websockets = false } = {}) { // Run the deffered plugins. PluginsService.runDeferred(); @@ -136,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(), ]); diff --git a/services/mailer.js b/services/mailer.js index 0ca88fa94..efdaf5d32 100644 --- a/services/mailer.js +++ b/services/mailer.js @@ -1,4 +1,3 @@ -const debug = require('debug')('talk:services:mailer'); const nodemailer = require('nodemailer'); const kue = require('./kue'); const i18n = require('./i18n'); @@ -6,7 +5,7 @@ const path = require('path'); const fs = require('fs-extra'); const { get, set, merge, template, isObject } = require('lodash'); const { TEMPLATE_LOCALS } = require('../middleware/staticTemplate'); -const Context = require('../graph/context'); +const debug = require('debug')('talk:services:mailer'); const { SMTP_HOST, @@ -137,7 +136,7 @@ mailer.send = async options => { const email = get(options, 'email'); const user = isObject(options.user) ? get(options.user, 'id') : options.user; if (!user && !email) { - throw new Error('user not provided'); + throw new Error('user/email not provided'); } // Create the new locals object and attach the static locals and the i18n @@ -153,8 +152,10 @@ mailer.send = async options => { }) ); + debug('Creating job for mail'); + // Create the job to send the email later. - return mailer.task.create({ + const job = await mailer.task.create({ title: 'Mail', user, email, @@ -165,65 +166,10 @@ mailer.send = async options => { html, }, }); -}; -/** - * Start the queue processor for the mailer job. - */ -mailer.process = () => { - debug(`Now processing ${mailer.task.name} jobs`); + debug(`Created Job[${job.id}]`); - return mailer.task.process(async ({ id, data }, done) => { - const { email, user, message } = data; - - debug(`Starting to send mail for Job[${id}]`); - - // 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) { - message.to = email; - } else { - // Get the user to send the message to. - const ctx = Context.forSystem(); - - try { - const { data, errors } = await ctx.graphql( - ` - query GetUserEmail($user: ID!) { - user(id: $user) { - email - } - } - `, - { user } - ); - if (errors) { - return done(errors); - } - const email = get(data, 'user.email'); - if (!email) { - return done(errors.ErrMissingEmail); - } - - message.to = email; - } catch (err) { - return done(err); - } - } - - // Actually send the email. - mailer.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(); - }); - }); + return job; }; module.exports = mailer; diff --git a/services/passport.js b/services/passport.js index 59215dd54..803da5344 100644 --- a/services/passport.js +++ b/services/passport.js @@ -441,6 +441,8 @@ passport.use( debug(`hasRecaptcha=${hasRecaptcha}, recaptchaPassed=${recaptchaPassed}`); + console.log(UsersService); + // If the request didn't have a recaptcha, check to see if we did need one by // checking the rate limit against failed attempts on this email // address/login. diff --git a/services/scraper.js b/services/scraper.js index 36f935876..daba96a99 100644 --- a/services/scraper.js +++ b/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; }, }; diff --git a/services/users.js b/services/users.js index 945f56fd5..f5319122c 100644 --- a/services/users.js +++ b/services/users.js @@ -1,20 +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 { 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');