mirror of
https://github.com/wassname/talk.git
synced 2026-08-09 12:30:42 +08:00
[next] Story Mutations (#2055)
* feat: added mutations for removing + scraping stories - added mutation for removing stories - added mutation for scraping stories - renamed all delete* to remove* to avoid reserved keyword collision * fix: linting * feat: implemented createStory * feat: added support for createStory * feat: improved Story deletion - Deletes comments on the deleted Story - Deletes actions of comments that were deleted via new `root_item_id` parameter for Comment Action's - Added logging around critical components of the remove process * feat: implemented mergeStory - Added new merge function for root actions - Added new merge function for story comments - Added new merge function for comment status counts - Added new remove stories function * fix: fixed story creation method * fix: added action count merging * fix: removed outdated file * fix: removed queue that got duplicated via rebase * fix: prevent error when action counts are empty
This commit is contained in:
Wyatt Johnson
@@ -0,0 +1,61 @@
|
||||
import Queue, { Job, Queue as QueueType } from "bull";
|
||||
import logger from "talk-server/logger";
|
||||
|
||||
export interface TaskOptions<T, U = any> {
|
||||
jobName: string;
|
||||
jobProcessor: (job: Job<T>) => Promise<U>;
|
||||
queue: Queue.QueueOptions;
|
||||
}
|
||||
|
||||
export default class Task<T, U = any> {
|
||||
private options: TaskOptions<T, U>;
|
||||
private queue: QueueType<T>;
|
||||
constructor(options: TaskOptions<T, U>) {
|
||||
this.queue = new Queue(options.jobName, options.queue);
|
||||
this.options = options;
|
||||
|
||||
// Sets up and attaches the job processor to the queue.
|
||||
this.setupAndAttachProcessor();
|
||||
}
|
||||
|
||||
/**
|
||||
* Add will add the job to the queue to get processed. It's not needed to
|
||||
* handle the job after it has been created.
|
||||
*
|
||||
* @param data the data for the job to add.
|
||||
*/
|
||||
public async add(data: T) {
|
||||
const job = await this.queue.add(data, {
|
||||
// We always remove the job when it's complete, no need to fill up Redis
|
||||
// with completed entries if we don't need to.
|
||||
removeOnComplete: true,
|
||||
});
|
||||
|
||||
logger.trace(
|
||||
{ job_id: job.id, job_name: this.options.jobName },
|
||||
"added job to queue"
|
||||
);
|
||||
return job;
|
||||
}
|
||||
private setupAndAttachProcessor() {
|
||||
this.queue.process(async (job: Job<T>) => {
|
||||
logger.trace(
|
||||
{ job_id: job.id, job_name: this.options.jobName },
|
||||
"processing job from queue"
|
||||
);
|
||||
|
||||
// Send the job off to the job processor to be handled.
|
||||
const promise: U = await this.options.jobProcessor(job);
|
||||
logger.trace(
|
||||
{ job_id: job.id, job_name: this.options.jobName },
|
||||
"processing completed"
|
||||
);
|
||||
return promise;
|
||||
});
|
||||
|
||||
logger.trace(
|
||||
{ job_name: this.options.jobName },
|
||||
"registered processor for job type"
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,68 @@
|
||||
import Queue from "bull";
|
||||
import { Db } from "mongodb";
|
||||
|
||||
import { Config } from "talk-common/config";
|
||||
import Task from "talk-server/queue/Task";
|
||||
import { createMailerTask, Mailer } from "talk-server/queue/tasks/mailer";
|
||||
import {
|
||||
createScraperTask,
|
||||
ScraperData,
|
||||
} from "talk-server/queue/tasks/scraper";
|
||||
import { createRedisClient } from "talk-server/services/redis";
|
||||
import TenantCache from "talk-server/services/tenant/cache";
|
||||
|
||||
const createQueueOptions = (config: Config): Queue.QueueOptions => {
|
||||
const client = createRedisClient(config);
|
||||
const subscriber = createRedisClient(config);
|
||||
const blockingClient = createRedisClient(config);
|
||||
|
||||
// Return the options that can be used by the Queue.
|
||||
return {
|
||||
// Here, we are reusing the clients based on the requested types. This way,
|
||||
// any time we need a specific client, we get to use one of the ones that
|
||||
// already have been created.
|
||||
createClient: type => {
|
||||
switch (type) {
|
||||
case "subscriber":
|
||||
return subscriber;
|
||||
case "client":
|
||||
return client;
|
||||
case "bclient":
|
||||
return blockingClient;
|
||||
}
|
||||
},
|
||||
|
||||
// Because bull uses atomic operations across separate keys, we need to add
|
||||
// a prefix to the keys to help the Redis cluster place all those elements
|
||||
// together to support the atomic operations. See:
|
||||
// https://redis.io/topics/cluster-tutorial
|
||||
prefix: "{queue}",
|
||||
};
|
||||
};
|
||||
|
||||
export interface QueueOptions {
|
||||
mongo: Db;
|
||||
config: Config;
|
||||
tenantCache: TenantCache;
|
||||
}
|
||||
|
||||
export interface TaskQueue {
|
||||
mailer: Mailer;
|
||||
scraper: Task<ScraperData>;
|
||||
}
|
||||
|
||||
export function createQueue(options: QueueOptions): TaskQueue {
|
||||
// Create the processor queue options. This holds references to the Redis
|
||||
// clients that are shared per queue.
|
||||
const queueOptions = createQueueOptions(options.config);
|
||||
|
||||
// Attach process functions to the various tasks in the queue.
|
||||
const mailer = createMailerTask(queueOptions, options);
|
||||
const scraper = createScraperTask(queueOptions, options);
|
||||
|
||||
// Return the tasks + client.
|
||||
return {
|
||||
mailer,
|
||||
scraper,
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,63 @@
|
||||
import Queue, { Job, Queue as QueueType } from "bull";
|
||||
import logger from "talk-server/logger";
|
||||
|
||||
export interface TaskOptions<T, U = any> {
|
||||
jobName: string;
|
||||
jobProcessor: (job: Job<T>) => Promise<U>;
|
||||
queue: Queue.QueueOptions;
|
||||
}
|
||||
|
||||
export default class Task<T, U = any> {
|
||||
private options: TaskOptions<T, U>;
|
||||
private queue: QueueType<T>;
|
||||
|
||||
constructor(options: TaskOptions<T, U>) {
|
||||
this.queue = new Queue(options.jobName, options.queue);
|
||||
this.options = options;
|
||||
|
||||
// Sets up and attaches the job processor to the queue.
|
||||
this.setupAndAttachProcessor();
|
||||
}
|
||||
|
||||
/**
|
||||
* Add will add the job to the queue to get processed. It's not needed to
|
||||
* handle the job after it has been created.
|
||||
*
|
||||
* @param data the data for the job to add.
|
||||
*/
|
||||
public async add(data: T) {
|
||||
const job = await this.queue.add(data, {
|
||||
// We always remove the job when it's complete, no need to fill up Redis
|
||||
// with completed entries if we don't need to.
|
||||
removeOnComplete: true,
|
||||
});
|
||||
|
||||
logger.trace(
|
||||
{ job_id: job.id, job_name: this.options.jobName },
|
||||
"added job to queue"
|
||||
);
|
||||
return job;
|
||||
}
|
||||
|
||||
private setupAndAttachProcessor() {
|
||||
this.queue.process(async (job: Job<T>) => {
|
||||
logger.trace(
|
||||
{ job_id: job.id, job_name: this.options.jobName },
|
||||
"processing job from queue"
|
||||
);
|
||||
|
||||
// Send the job off to the job processor to be handled.
|
||||
const promise: U = await this.options.jobProcessor(job);
|
||||
logger.trace(
|
||||
{ job_id: job.id, job_name: this.options.jobName },
|
||||
"processing completed"
|
||||
);
|
||||
return promise;
|
||||
});
|
||||
|
||||
logger.trace(
|
||||
{ job_name: this.options.jobName },
|
||||
"registered processor for job type"
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,43 @@
|
||||
import nunjucks from "nunjucks";
|
||||
import path from "path";
|
||||
|
||||
import { Config } from "talk-common/config";
|
||||
|
||||
/**
|
||||
* templateDirectory is the directory containing the email templates.
|
||||
*/
|
||||
const templateDirectory = path.join(__dirname, "templates");
|
||||
|
||||
export interface GenerateHTMLOptions {
|
||||
/**
|
||||
* name is the name of the template to render.
|
||||
*/
|
||||
name: string;
|
||||
context: any;
|
||||
}
|
||||
|
||||
export default class MailerContent {
|
||||
private env: nunjucks.Environment;
|
||||
|
||||
constructor(config: Config) {
|
||||
// Configure the nunjucks environment.
|
||||
this.env = new nunjucks.Environment(
|
||||
new nunjucks.FileSystemLoader(templateDirectory),
|
||||
{
|
||||
// When we aren't in production mode, reload the templates.
|
||||
watch: config.get("env") !== "production",
|
||||
}
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* generateHTML will generate the HTML for a template and optionally cache
|
||||
* the compiled template based on the configured environment.
|
||||
*
|
||||
* @param options configuration for generating HTML based on the email
|
||||
* template.
|
||||
*/
|
||||
public generateHTML(options: GenerateHTMLOptions): string {
|
||||
return this.env.render(options.name + ".html", options.context);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,256 @@
|
||||
import Queue, { Job } from "bull";
|
||||
import htmlToText from "html-to-text";
|
||||
import Joi from "joi";
|
||||
import { Db } from "mongodb";
|
||||
import { createTransport } from "nodemailer";
|
||||
|
||||
import { Config } from "talk-common/config";
|
||||
import logger from "talk-server/logger";
|
||||
import Task from "talk-server/queue/Task";
|
||||
import MailerContent from "talk-server/queue/tasks/mailer/content";
|
||||
import TenantCache from "talk-server/services/tenant/cache";
|
||||
import { TenantCacheAdapter } from "talk-server/services/tenant/cache/adapter";
|
||||
|
||||
const JOB_NAME = "mailer";
|
||||
|
||||
export interface MailProcessorOptions {
|
||||
config: Config;
|
||||
mongo: Db;
|
||||
tenantCache: TenantCache;
|
||||
}
|
||||
|
||||
export interface MailerData {
|
||||
message: {
|
||||
to: string;
|
||||
subject: string;
|
||||
html: string;
|
||||
};
|
||||
tenantID: string;
|
||||
}
|
||||
|
||||
const MailerDataSchema = Joi.object().keys({
|
||||
message: Joi.object().keys({
|
||||
to: Joi.string(),
|
||||
subject: Joi.string(),
|
||||
html: Joi.string(),
|
||||
}),
|
||||
tenantID: Joi.string(),
|
||||
});
|
||||
|
||||
const createJobProcessor = (options: MailProcessorOptions) => {
|
||||
const { tenantCache } = options;
|
||||
|
||||
// Create the cache adapter that will handle invalidating the email transport
|
||||
// when the tenant experiences a change.
|
||||
const cache = new TenantCacheAdapter<ReturnType<typeof createTransport>>(
|
||||
tenantCache
|
||||
);
|
||||
|
||||
return async (job: Job<MailerData>) => {
|
||||
const { value, error: err } = Joi.validate(job.data, MailerDataSchema, {
|
||||
stripUnknown: true,
|
||||
presence: "required",
|
||||
abortEarly: false,
|
||||
});
|
||||
if (err) {
|
||||
logger.error(
|
||||
{
|
||||
job_id: job.id,
|
||||
job_name: JOB_NAME,
|
||||
err,
|
||||
},
|
||||
"job data did not match expected schema"
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
// Pull the data out of the validated model.
|
||||
const { message, tenantID } = value;
|
||||
|
||||
// Get the referenced tenant so we know who to send it from.
|
||||
const tenant = await tenantCache.retrieveByID(tenantID);
|
||||
if (!tenant) {
|
||||
logger.error(
|
||||
{
|
||||
job_id: job.id,
|
||||
job_name: JOB_NAME,
|
||||
tenant_id: tenantID,
|
||||
},
|
||||
"referenced tenant was not found"
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
if (!tenant.email.enabled) {
|
||||
logger.error(
|
||||
{
|
||||
job_id: job.id,
|
||||
job_name: JOB_NAME,
|
||||
tenant_id: tenantID,
|
||||
},
|
||||
"not sending email, it was disabled"
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
if (!tenant.email.smtpURI) {
|
||||
logger.error(
|
||||
{
|
||||
job_id: job.id,
|
||||
job_name: JOB_NAME,
|
||||
tenant_id: tenantID,
|
||||
},
|
||||
"email was enabled but the smtpURI configuration was missing"
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
if (!tenant.email.fromAddress) {
|
||||
// TODO: possibly have fallback email address?
|
||||
logger.error(
|
||||
{
|
||||
job_id: job.id,
|
||||
job_name: JOB_NAME,
|
||||
tenant_id: tenantID,
|
||||
},
|
||||
"email was enabled but the fromAddress configuration was missing"
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
let transport = cache.get(tenantID);
|
||||
if (!transport) {
|
||||
// Create the transport based on the smtp uri.
|
||||
transport = createTransport(tenant.email.smtpURI);
|
||||
|
||||
// Set the transport back into the cache.
|
||||
cache.set(tenantID, transport);
|
||||
|
||||
logger.debug(
|
||||
{
|
||||
job_id: job.id,
|
||||
job_name: JOB_NAME,
|
||||
tenant_id: tenantID,
|
||||
},
|
||||
"transport was not cached"
|
||||
);
|
||||
} else {
|
||||
logger.debug(
|
||||
{
|
||||
job_id: job.id,
|
||||
job_name: JOB_NAME,
|
||||
tenant_id: tenantID,
|
||||
},
|
||||
"transport was cached"
|
||||
);
|
||||
}
|
||||
|
||||
logger.debug(
|
||||
{
|
||||
job_id: job.id,
|
||||
job_name: JOB_NAME,
|
||||
tenant_id: tenantID,
|
||||
},
|
||||
"starting to send the email"
|
||||
);
|
||||
|
||||
// Send the mail message.
|
||||
await transport.sendMail({
|
||||
...message,
|
||||
// Generate the text content of the message from the HTML.
|
||||
text: htmlToText.fromString(message.html),
|
||||
from: tenant.email.fromAddress,
|
||||
});
|
||||
|
||||
logger.debug(
|
||||
{
|
||||
job_id: job.id,
|
||||
job_name: JOB_NAME,
|
||||
tenant_id: tenantID,
|
||||
},
|
||||
"sent the email"
|
||||
);
|
||||
};
|
||||
};
|
||||
|
||||
export interface MailerInput {
|
||||
message: {
|
||||
to: string;
|
||||
subject: string;
|
||||
};
|
||||
template: {
|
||||
name: string;
|
||||
context: object;
|
||||
};
|
||||
tenantID: string;
|
||||
}
|
||||
|
||||
export class Mailer {
|
||||
private task: Task<MailerData>;
|
||||
private content: MailerContent;
|
||||
private tenantCache: TenantCache;
|
||||
|
||||
constructor(queue: Queue.QueueOptions, options: MailProcessorOptions) {
|
||||
this.task = new Task<MailerData>({
|
||||
jobName: JOB_NAME,
|
||||
jobProcessor: createJobProcessor(options),
|
||||
queue,
|
||||
});
|
||||
this.content = new MailerContent(options.config);
|
||||
this.tenantCache = options.tenantCache;
|
||||
}
|
||||
|
||||
public async add({ template, ...rest }: MailerInput) {
|
||||
const { tenantID } = rest;
|
||||
|
||||
// All email templates require the tenant in order to insert the footer, so
|
||||
// load it from the tenant cache here.
|
||||
const tenant = await this.tenantCache.retrieveByID(tenantID);
|
||||
if (!tenant) {
|
||||
logger.error(
|
||||
{
|
||||
job_name: JOB_NAME,
|
||||
tenant_id: tenantID,
|
||||
},
|
||||
"referenced tenant was not found"
|
||||
);
|
||||
// TODO: (wyattjoh) maybe throw an error here?
|
||||
return;
|
||||
}
|
||||
|
||||
if (!tenant.email.enabled) {
|
||||
logger.error(
|
||||
{
|
||||
job_name: JOB_NAME,
|
||||
tenant_id: tenantID,
|
||||
},
|
||||
"not adding email, it was disabled"
|
||||
);
|
||||
// TODO: (wyattjoh) maybe throw an error here?
|
||||
return;
|
||||
}
|
||||
|
||||
// Generate the HTML for the email template.
|
||||
const html = this.content.generateHTML({
|
||||
...template,
|
||||
context: {
|
||||
...template.context,
|
||||
tenant,
|
||||
},
|
||||
});
|
||||
|
||||
// Return the job that'll add the email to the queue to be processed later.
|
||||
return this.task.add({
|
||||
...rest,
|
||||
message: {
|
||||
...rest.message,
|
||||
html,
|
||||
},
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
export const createMailerTask = (
|
||||
queue: Queue.QueueOptions,
|
||||
options: MailProcessorOptions
|
||||
) => new Mailer(queue, options);
|
||||
@@ -0,0 +1,14 @@
|
||||
<!doctype html>
|
||||
<html lang="en">
|
||||
<head></head>
|
||||
<body>
|
||||
<h1>Forgot Password</h1>
|
||||
<p>We received a request to reset your password on <a href="{{ tenant.organizationURL }}">{{ tenant.organizationName }}</a>.</p>
|
||||
<p>Please follow the link below to reset your password:</p>
|
||||
<p><a href="{{ resetURL }}">Click here to reset your password</a></p>
|
||||
<p><i>If you did not request this change, you can ignore this email.</i></p>
|
||||
<footer>
|
||||
<p>Sent by <a href="{{ tenant.organizationURL }}">{{ tenant.organizationName }}</a></p>
|
||||
</footer>
|
||||
</body>
|
||||
</html>
|
||||
@@ -0,0 +1,55 @@
|
||||
import Queue, { Job } from "bull";
|
||||
import { Db } from "mongodb";
|
||||
|
||||
import logger from "talk-server/logger";
|
||||
import Task from "talk-server/queue/Task";
|
||||
import { scrape } from "talk-server/services/stories/scraper";
|
||||
|
||||
const JOB_NAME = "scraper";
|
||||
|
||||
export interface ScrapeProcessorOptions {
|
||||
mongo: Db;
|
||||
}
|
||||
|
||||
export interface ScraperData {
|
||||
storyID: string;
|
||||
storyURL: string;
|
||||
tenantID: string;
|
||||
}
|
||||
|
||||
const createJobProcessor = ({ mongo }: ScrapeProcessorOptions) => async (
|
||||
job: Job<ScraperData>
|
||||
) => {
|
||||
// Pull out the job data.
|
||||
const { storyID, storyURL, tenantID } = job.data;
|
||||
|
||||
const log = logger.child({
|
||||
job_id: job.id,
|
||||
job_name: JOB_NAME,
|
||||
story_id: storyID,
|
||||
story_url: storyURL,
|
||||
tenant_id: tenantID,
|
||||
});
|
||||
|
||||
log.debug("starting to scrape the story");
|
||||
|
||||
try {
|
||||
await scrape(mongo, tenantID, storyID, storyURL);
|
||||
log.debug("scraped the story");
|
||||
} catch (err) {
|
||||
log.error({ err }, "could not scrape the story");
|
||||
|
||||
throw err;
|
||||
}
|
||||
};
|
||||
|
||||
export function createScraperTask(
|
||||
queue: Queue.QueueOptions,
|
||||
options: ScrapeProcessorOptions
|
||||
) {
|
||||
return new Task({
|
||||
jobName: JOB_NAME,
|
||||
jobProcessor: createJobProcessor(options),
|
||||
queue,
|
||||
});
|
||||
}
|
||||
Reference in New Issue
Block a user