[next] Email (#2261)

* feat: suspending, banning, now propogation

* feat: added email rendering + localization support

* fix: fix related to lib

* refactor: moved juicer to queue task

* refactor: cleanup of job processor

* refactor: improved error messaging around failed email

* feat: initial forgot passwor impl

* fix: fixed rebase errors

* feat: send back Content-Language header with requests

* feat: added ban email

* feat: implemented forgotten password API

* fix: linting

* feat: support more emails

* fix: promise patches

* feat: initial confirm email API

* feat: added rate limiting

* feat: added URL support

* feat: added email docs

* fix: updated docs

* chore: documentation review

* fix: fixed build bug

* feat: implement forgot password in auth popup

* test: add tests + fixes

* chore: rename StatelessComponent to FunctionComponent

* fix: types and test fixes

* chore: upgrade deps

* fix: THANK YOU TESTS FOR SAVING MY A**

* chore: reorder imports

* chore: remove obsolete !

* feat: implement accounts bundle

* refactor: review suggestion

* fix: rebase upgrade error

* fix: rebase bug

* feat: reset password link support

* test: add tests for account password reset page

* fix: remove redirect uri

* fix: revert local state changes

src/core/server/queue/Task.ts

@@ -6,6 +6,7 @@ import logger from "talk-server/logger";
 export interface TaskOptions<T, U = any> {
   jobName: string;
   jobProcessor: (job: Job<T>) => Promise<U>;
+  jobOptions?: Queue.JobOptions;
   queue: Queue.QueueOptions;
 }
 
@@ -14,10 +15,39 @@ export default class Task<T, U = any> {
   private queue: QueueType<T>;
   private log: Logger;
 
-  constructor(options: TaskOptions<T, U>) {
-    this.queue = new Queue(options.jobName, options.queue);
-    this.options = options;
-    this.log = logger.child({ jobName: options.jobName });
+  constructor({
+    jobName,
+    jobProcessor,
+    jobOptions = {},
+    queue,
+  }: TaskOptions<T, U>) {
+    this.log = logger.child({ jobName });
+    this.queue = new Queue(jobName, queue);
+    this.options = {
+      jobName,
+      jobProcessor,
+      jobOptions: {
+        // 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,
+
+        // By default, configure jobs to use an exponential backoff
+        // strategy starting at a 10 second delay.
+        backoff: {
+          type: "exponential",
+          delay: 10000,
+        },
+
+        // Be default, try all jobs at least 5 times.
+        attempts: 5,
+
+        // Add the custom job options if they exist.
+        ...jobOptions,
+      },
+      queue,
+    };
+
+    // TODO: (wyattjoh) attach event handlers to the queue for metrics via: https://github.com/OptimalBits/bull/blob/develop/REFERENCE.md#events
   }
 
   /**
@@ -27,11 +57,8 @@ export default class Task<T, U = any> {
    * @param data the data for the job to add.
    */
   public async add(data: T): Promise<Queue.Job<T> | undefined> {
-    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,
-    });
+    // Create the job.
+    const job = await this.queue.add(data, this.options.jobOptions);
 
     this.log.trace({ jobID: job.id }, "added job to queue");
     return job;
@@ -47,10 +74,15 @@ export default class Task<T, U = any> {
 
       log.trace("processing job from queue");
 
-      // Send the job off to the job processor to be handled.
-      const promise: U = await this.options.jobProcessor(job);
-      log.trace("processing completed");
-      return promise;
+      try {
+        // Send the job off to the job processor to be handled.
+        const promise: U = await this.options.jobProcessor(job);
+        log.trace("processing completed");
+        return promise;
+      } catch (err) {
+        log.error({ err }, "job failed to process");
+        throw err;
+      }
     });
 
     this.log.trace("registered processor for job type");