[CORL-498, CORL-495, CORL-539, CORL-496, CORL-494] Email Notifications Support & Framework (#2498)

* chore: renamed old templates

* feat: initial notifications support

* feat: email enhancements

* fix: linting

* feat: initial digesting beheviour

* feat: added notification configuration

* feat: added unsubscribe routes

* fix: fixed failing snapshots/tests bc random ids

* feat: adjusted the save beheviour, added tests

* feat: added tests

* feat: added staff replies

* feat: renamed E-Mail to Email

* feat: enhanced cron processing

* fix: linting + updating tests

* feat: enhanced cron context

* fix: added staff replies back in

src/core/server/cron/accountDeletion.ts

@@ -1,19 +1,23 @@
-import { CronCommand, CronJob } from "cron";
 import { DateTime } from "luxon";
 import { Collection, Db } from "mongodb";
 
-import logger from "coral-server/logger";
 import { CommentAction } from "coral-server/models/action/comment";
 import { createCollection } from "coral-server/models/helpers";
 import { Story } from "coral-server/models/story";
 import { Tenant } from "coral-server/models/tenant";
 import { User } from "coral-server/models/user";
 import { MailerQueue } from "coral-server/queue/tasks/mailer";
+import TenantCache from "coral-server/services/tenant/cache";
+
+import {
+  ScheduledJob,
+  ScheduledJobCommand,
+  ScheduledJobGroup,
+} from "./scheduled";
 
 const BATCH_SIZE = 500;
 
-// TODO: extract this out to a separate file so it
-// can be re-used elsewhere
+// TODO: extract this out to a separate file so it can be re-used elsewhere
 const collections = {
   users: createCollection<User>("users"),
   comments: createCollection<Comment>("comments"),
@@ -22,80 +26,75 @@ const collections = {
   commentActions: createCollection<CommentAction>("commentActions"),
 };
 
+interface Options {
+  mongo: Db;
+  mailerQueue: MailerQueue;
+  tenantCache: TenantCache;
+}
+
+export const NAME = "Account Deletion";
+
 export function registerAccountDeletion(
-  mongo: Db,
-  mailer: MailerQueue
-): CronJob {
-  const job = new CronJob({
+  options: Options
+): ScheduledJobGroup<Options> {
+  const job = new ScheduledJob(options, {
+    name: `Twice Hourly ${NAME}`,
     cronTime: "0,30 * * * *",
-    timeZone: "America/New_York",
-    start: true,
-    runOnInit: false,
-    onTick: deleteScheduledAccounts(mongo, mailer),
+    command: deleteScheduledAccounts,
   });
 
-  if (job.running) {
-    logger.info("account deletion scheduler now running");
-  }
-
-  return job;
+  return { name: NAME, schedulers: [job] };
 }
 
-function deleteScheduledAccounts(mongo: Db, mailer: MailerQueue): CronCommand {
-  return async () => {
-    try {
-      logger.info("checking for accounts that require deletion");
+const deleteScheduledAccounts: ScheduledJobCommand<Options> = async ({
+  log,
+  mongo,
+  mailerQueue,
+  tenantCache,
+}) => {
+  // For each of the tenant's, process their users notifications.
+  for await (const tenant of tenantCache) {
+    log = log.child({ tenantID: tenant.id });
 
-      // TODO: iterate over tenants in tenant cache
-      while (true) {
-        const now = new Date();
-        const rescheduledDeletionDate = DateTime.fromJSDate(now)
-          .plus({ hours: 1 })
-          .toJSDate();
+    while (true) {
+      const now = new Date();
+      const rescheduledDeletionDate = DateTime.fromJSDate(now)
+        .plus({ hours: 1 })
+        .toJSDate();
 
-        const userResult = await collections.users(mongo).findOneAndUpdate(
-          {
-            scheduledDeletionDate: { $lte: now },
+      const { value: user } = await collections.users(mongo).findOneAndUpdate(
+        {
+          tenantID: tenant.id,
+          scheduledDeletionDate: { $lte: now },
+        },
+        {
+          $set: {
+            scheduledDeletionDate: rescheduledDeletionDate,
           },
-          {
-            $set: {
-              scheduledDeletionDate: rescheduledDeletionDate,
-            },
-          },
-          {
-            // We want to get back the user with
-            // modified scheduledDeletionDate
-            returnOriginal: false,
-          }
-        );
-
-        if (!userResult.value) {
-          logger.info("no more users were scheduled for deletion");
-          break;
+        },
+        {
+          // We want to get back the user with
+          // modified scheduledDeletionDate
+          returnOriginal: false,
         }
-
-        const userToDelete = userResult.value;
-
-        logger.info(
-          { userID: userToDelete.id, tenantID: userToDelete.tenantID },
-          `deleting user`
-        );
-
-        deleteUser(mongo, mailer, userToDelete.id, userToDelete.tenantID, now);
-      }
-    } catch (error) {
-      logger.error(
-        { error },
-        "an error occurred trying to perform scheduled account deletions"
       );
+      if (!user) {
+        log.debug("no more users were scheduled for deletion");
+        break;
+      }
+
+      log.info({ userID: user.id }, "deleting user");
+
+      await deleteUser(mongo, mailerQueue, user.id, user.tenantID, now);
     }
-  };
-}
+  }
+};
 
 async function executeBulkOperations<T>(
   collection: Collection<T>,
   operations: any[]
 ) {
+  // TODO: (wyattjoh) fix types here to support actual types when upstream changes applied
   const bulk: any = collection.initializeUnorderedBulkOp();
 
   for (const operation of operations) {
@@ -110,7 +109,11 @@ interface Batch {
   stories: any[];
 }
 
-async function deleteUserActionCounts(db: Db, userID: string) {
+async function deleteUserActionCounts(
+  mongo: Db,
+  userID: string,
+  tenantID: string
+) {
   const batch: Batch = {
     comments: [],
     stories: [],
@@ -118,24 +121,30 @@ async function deleteUserActionCounts(db: Db, userID: string) {
 
   async function processBatch() {
     await executeBulkOperations<Comment>(
-      collections.comments(db),
+      collections.comments(mongo),
       batch.comments
     );
     batch.comments = [];
 
-    await executeBulkOperations<Story>(collections.stories(db), batch.stories);
+    await executeBulkOperations<Story>(
+      collections.stories(mongo),
+      batch.stories
+    );
     batch.stories = [];
   }
 
-  const cursor = db
-    .collection("commentActions")
-    .find({ userID, actionType: "REACTION" });
+  const cursor = collections
+    .commentActions(mongo)
+    .find({ tenantID, userID, actionType: "REACTION" });
   while (await cursor.hasNext()) {
     const action = await cursor.next();
+    if (!action) {
+      continue;
+    }
 
     batch.comments.push({
       updateOne: {
-        filter: { id: action.commentID },
+        filter: { tenantID, id: action.commentID },
         update: {
           $inc: {
             "revisions.$[revisions].actionCounts.REACTION": -1,
@@ -148,7 +157,7 @@ async function deleteUserActionCounts(db: Db, userID: string) {
 
     batch.stories.push({
       updateOne: {
-        filter: { id: action.storyID },
+        filter: { tenantID, id: action.storyID },
         update: {
           $inc: {
             "commentCounts.action.REACTION": -1,
@@ -169,15 +178,20 @@ async function deleteUserActionCounts(db: Db, userID: string) {
     await processBatch();
   }
 
-  await collections.commentActions(db).deleteMany({
+  await collections.commentActions(mongo).deleteMany({
+    tenantID,
     userID,
     actionType: "REACTION",
   });
 }
 
-async function deleteUserComments(db: Db, authorID: string) {
-  await collections.comments(db).updateMany(
-    { authorID },
+async function deleteUserComments(
+  mongo: Db,
+  authorID: string,
+  tenantID: string
+) {
+  await collections.comments(mongo).updateMany(
+    { tenantID, authorID },
     {
       $set: {
         authorID: null,
@@ -190,29 +204,31 @@ async function deleteUserComments(db: Db, authorID: string) {
 }
 
 async function deleteUser(
-  db: Db,
+  mongo: Db,
   mailer: MailerQueue,
   userID: string,
   tenantID: string,
   now: Date
 ) {
-  const user = await collections.users(db).findOne({ id: userID, tenantID });
+  const user = await collections.users(mongo).findOne({ id: userID, tenantID });
   if (!user) {
-    logger.warn({ userID, tenantID }, `could not find user`);
-    return;
+    throw new Error("could not find user by ID");
   }
 
-  const tenant = await collections.tenants(db).findOne({ id: tenantID });
+  const tenant = await collections.tenants(mongo).findOne({ id: tenantID });
   if (!tenant) {
-    logger.warn({ userID, tenantID }, `could not find tenant`);
-    return;
+    throw new Error("could not find tenant by ID");
   }
 
-  await deleteUserActionCounts(db, userID);
-  await deleteUserComments(db, userID);
+  // Delete the user's action counts.
+  await deleteUserActionCounts(mongo, userID, tenantID);
 
-  collections.users(db).updateOne(
-    { id: userID },
+  // Delete the user's comments.
+  await deleteUserComments(mongo, userID, tenantID);
+
+  // Mark the user as deleted.
+  await collections.users(mongo).updateOne(
+    { tenantID, id: userID },
     {
       $set: {
         profiles: [],
@@ -224,14 +240,16 @@ async function deleteUser(
     }
   );
 
+  // If the user has an email, then send them a confirmation that their account
+  // was deleted.
   if (user.email) {
     await mailer.add({
-      tenantID: tenant.id,
+      tenantID,
       message: {
         to: user.email,
       },
       template: {
-        name: "delete-request-completed",
+        name: "account-notification/delete-request-completed",
         context: {
           organizationContactEmail: tenant.organization.contactEmail,
           organizationName: tenant.organization.name,

src/core/server/cron/index.ts

@@ -1,27 +1,40 @@
-import { CronJob } from "cron";
 import { Db } from "mongodb";
 
+import { Config } from "coral-server/config";
 import { MailerQueue } from "coral-server/queue/tasks/mailer";
+import { JWTSigningConfig } from "coral-server/services/jwt";
+import TenantCache from "coral-server/services/tenant/cache";
 
 import { registerAccountDeletion } from "./accountDeletion";
+import { registerNotificationDigesting } from "./notificationDigesting";
 
-export interface ScheduledTasks {
-  accountDeletion: ScheduledTask;
+export interface ScheduledJobGroups {
+  accountDeletion: ReturnType<typeof registerAccountDeletion>;
+  notificationDigesting: ReturnType<typeof registerNotificationDigesting>;
 }
 
-export interface ScheduledTask {
-  name: string;
-  task: CronJob;
+interface Options {
+  mongo: Db;
+  config: Config;
+  mailerQueue: MailerQueue;
+  signingConfig: JWTSigningConfig;
+  tenantCache: TenantCache;
 }
 
 export default function startScheduledTasks(
-  mongo: Db,
-  mailer: MailerQueue
-): ScheduledTasks {
-  return {
-    accountDeletion: {
-      name: "Account Deletion",
-      task: registerAccountDeletion(mongo, mailer),
-    },
+  options: Options
+): ScheduledJobGroups {
+  const tasks: ScheduledJobGroups = {
+    accountDeletion: registerAccountDeletion(options),
+    notificationDigesting: registerNotificationDigesting(options),
   };
+
+  for (const { name, schedulers } of Object.values(tasks)) {
+    for (const scheduler of schedulers) {
+      scheduler.job.start();
+      scheduler.log.debug({ jobGroupName: name }, "now started job scheduling");
+    }
+  }
+
+  return tasks;
 }

src/core/server/cron/notificationDigesting.ts

@@ -0,0 +1,131 @@
+import { Db } from "mongodb";
+import path from "path";
+
+import { Config } from "coral-server/config";
+import { GQLDIGEST_FREQUENCY } from "coral-server/graph/tenant/schema/__generated__/types";
+import { MailerQueue } from "coral-server/queue/tasks/mailer";
+import { DigestibleTemplate } from "coral-server/queue/tasks/mailer/templates";
+import { JWTSigningConfig } from "coral-server/services/jwt";
+import NotificationContext from "coral-server/services/notifications/context";
+import TenantCache from "coral-server/services/tenant/cache";
+
+import {
+  ScheduledJob,
+  ScheduledJobCommand,
+  ScheduledJobGroup,
+} from "./scheduled";
+
+interface Options {
+  mongo: Db;
+  config: Config;
+  mailerQueue: MailerQueue;
+  signingConfig: JWTSigningConfig;
+  tenantCache: TenantCache;
+}
+
+export const NAME = "Notification Digesting";
+
+export function registerNotificationDigesting(
+  options: Options
+): ScheduledJobGroup<Options> {
+  const hourly = new ScheduledJob(options, {
+    name: `Hourly ${NAME}`,
+    cronTime: "0 * * * *",
+    command: processNotificationDigesting(GQLDIGEST_FREQUENCY.HOURLY),
+  });
+
+  const daily = new ScheduledJob(options, {
+    name: `Daily ${NAME}`,
+    cronTime: "0 0 * * *",
+    command: processNotificationDigesting(GQLDIGEST_FREQUENCY.DAILY),
+  });
+
+  return {
+    name: NAME,
+    schedulers: [hourly, daily],
+  };
+}
+
+/**
+ * DigestElement represents each element that is used for the digesting
+ * operations.
+ */
+interface DigestElement {
+  template: string;
+  partial: string;
+  contexts: Array<DigestibleTemplate["context"]>;
+}
+
+const processNotificationDigesting = (
+  frequency: GQLDIGEST_FREQUENCY
+): ScheduledJobCommand<Options> => async ({
+  log,
+  mongo,
+  config,
+  signingConfig,
+  tenantCache,
+  mailerQueue,
+}) => {
+  // For each of the tenant's, process their users notifications.
+  for await (const tenant of tenantCache) {
+    // Create a notification context to handle processing notifications. Note
+    // that this will share the current date for all users processed for this
+    // Tenant, but this is OK, because we're not using this Date for the
+    // digesting operations.
+    const ctx = new NotificationContext({
+      mongo,
+      config,
+      signingConfig,
+      tenant,
+      log,
+    });
+
+    ctx.log.debug("starting digesting for tenant");
+
+    // Process all the notifications for this Tenant.
+    for await (const user of ctx.digest(frequency)) {
+      ctx.log.debug(
+        { userID: user.id, digests: user.digests.length },
+        "now processing digests for user"
+      );
+
+      // Group the digests.
+      const digests = user.digests.reduce(
+        (acc, entry) => {
+          const digest = acc.find(d => d.template === entry.template.name);
+          if (digest) {
+            digest.contexts.push(entry.template.context);
+          } else {
+            acc.push({
+              template: entry.template.name,
+              partial: path.basename(entry.template.name),
+              contexts: [entry.template.context],
+            });
+          }
+
+          return acc;
+        },
+        [] as DigestElement[]
+      );
+
+      // TODO: sort the digest template elements by the digest order.
+
+      // Add the email containing the digest information.
+      mailerQueue.add({
+        tenantID: tenant.id,
+        message: {
+          to: user.email!,
+        },
+        template: {
+          name: "notification/digest",
+          context: {
+            digests,
+            organizationName: tenant.organization.name,
+            organizationURL: tenant.organization.url,
+            unsubscribeURL: await ctx.generateUnsubscribeURL(user),
+          },
+        },
+      });
+    }
+  }
+};

src/core/server/cron/scheduled/group.ts

@@ -0,0 +1,6 @@
+import { ScheduledJob } from "./job";
+
+export interface ScheduledJobGroup<T> {
+  name: string;
+  schedulers: Array<ScheduledJob<T>>;
+}

src/core/server/cron/scheduled/index.ts

@@ -0,0 +1,2 @@
+export * from "./group";
+export * from "./job";

src/core/server/cron/scheduled/job.ts

@@ -0,0 +1,55 @@
+import { CronCommand, CronJob } from "cron";
+import now from "performance-now";
+import uuid from "uuid";
+
+import logger, { Logger } from "coral-server/logger";
+
+export type ScheduledJobCommand<T extends {}> = (
+  ctx: T & { log: Logger }
+) => Promise<void>;
+
+interface Options<T extends {}> {
+  name: string;
+  cronTime: string;
+  command: ScheduledJobCommand<T>;
+}
+
+export class ScheduledJob<T extends {} = {}> {
+  public readonly job: CronJob;
+  public readonly log: Logger;
+  public readonly context: T;
+
+  constructor(context: T, opts: Options<T>) {
+    this.context = context;
+    this.log = logger.child({
+      jobName: opts.name,
+      jobFrequency: opts.cronTime,
+    });
+    this.job = new CronJob({
+      cronTime: opts.cronTime,
+      onTick: this.command(opts.command),
+      timeZone: "America/New_York",
+      start: false,
+      runOnInit: false,
+    });
+  }
+
+  private command(command: ScheduledJobCommand<T>): CronCommand {
+    return async () => {
+      const log = this.log.child({ scheduledExecutionID: uuid.v1() });
+      log.debug("now starting scheduled job");
+      const start = now();
+      try {
+        await command({
+          ...this.context,
+          log,
+        });
+        const processingTime = Math.floor(now() - start);
+        log.debug({ processingTime }, "now finished scheduled job");
+      } catch (err) {
+        const processingTime = Math.floor(now() - start);
+        log.error({ err, processingTime }, "failed to run scheduled job");
+      }
+    };
+  }
+}