feat: Document subscriptions (#3834)

Co-authored-by: Tom Moor <tom.moor@gmail.com>

server/queues/processors/NotificationsProcessor.ts

@@ -1,4 +1,7 @@
+import { uniqBy } from "lodash";
 import { Op } from "sequelize";
+import subscriptionCreator from "@server/commands/subscriptionCreator";
+import { sequelize } from "@server/database/sequelize";
 import CollectionNotificationEmail from "@server/emails/templates/CollectionNotificationEmail";
 import DocumentNotificationEmail from "@server/emails/templates/DocumentNotificationEmail";
 import Logger from "@server/logging/Logger";
@@ -9,12 +12,14 @@ import {
   Collection,
   User,
   NotificationSetting,
+  Subscription,
 } from "@server/models";
+import { can } from "@server/policies";
 import {
-  DocumentEvent,
   CollectionEvent,
   RevisionEvent,
   Event,
+  DocumentEvent,
 } from "@server/types";
 import BaseProcessor from "./BaseProcessor";
 
@@ -44,141 +49,217 @@ export default class NotificationsProcessor extends BaseProcessor {
     if (event.data?.source === "import") {
       return;
     }
+
     const [collection, document, team] = await Promise.all([
       Collection.findByPk(event.collectionId),
       Document.findByPk(event.documentId),
       Team.findByPk(event.teamId),
     ]);
+
     if (!document || !team || !collection) {
       return;
     }
-    const notificationSettings = await NotificationSetting.findAll({
-      where: {
-        userId: {
-          [Op.ne]: document.lastModifiedById,
-        },
-        teamId: document.teamId,
-        event:
-          event.name === "documents.publish"
-            ? "documents.publish"
-            : "documents.update",
-      },
-      include: [
-        {
-          model: User,
-          required: true,
-          as: "user",
-        },
-      ],
-    });
-    const eventName =
-      event.name === "documents.publish" ? "published" : "updated";
 
-    for (const setting of notificationSettings) {
-      // Suppress notifications for suspended users
-      if (setting.user.isSuspended) {
-        continue;
+    await this.createDocumentSubscriptions(document, event);
+
+    const recipients = await this.getDocumentNotificationRecipients(
+      document,
+      event.name === "documents.publish"
+        ? "documents.publish"
+        : "documents.update"
+    );
+
+    for (const recipient of recipients) {
+      const notify = await this.shouldNotify(document, recipient.user);
+
+      if (notify) {
+        await DocumentNotificationEmail.schedule({
+          to: recipient.user.email,
+          eventName:
+            event.name === "documents.publish" ? "published" : "updated",
+          documentId: document.id,
+          teamUrl: team.url,
+          actorName: document.updatedBy.name,
+          collectionName: collection.name,
+          unsubscribeUrl: recipient.unsubscribeUrl,
+        });
       }
-
-      // For document updates we only want to send notifications if
-      // the document has been edited by the user with this notification setting
-      // This could be replaced with ability to "follow" in the future
-      if (
-        eventName === "updated" &&
-        !document.collaboratorIds.includes(setting.userId)
-      ) {
-        continue;
-      }
-
-      // Check the user has access to the collection this document is in. Just
-      // because they were a collaborator once doesn't mean they still are.
-      const collectionIds = await setting.user.collectionIds();
-
-      if (!collectionIds.includes(document.collectionId)) {
-        continue;
-      }
-
-      // If this user has viewed the document since the last update was made
-      // then we can avoid sending them a useless notification, yay.
-      const view = await View.findOne({
-        where: {
-          userId: setting.userId,
-          documentId: event.documentId,
-          updatedAt: {
-            [Op.gt]: document.updatedAt,
-          },
-        },
-      });
-
-      if (view) {
-        Logger.info(
-          "processor",
-          `suppressing notification to ${setting.userId} because update viewed`
-        );
-        continue;
-      }
-
-      if (!setting.user.email) {
-        continue;
-      }
-
-      await DocumentNotificationEmail.schedule({
-        to: setting.user.email,
-        eventName,
-        documentId: document.id,
-        teamUrl: team.url,
-        actorName: document.updatedBy.name,
-        collectionName: collection.name,
-        unsubscribeUrl: setting.unsubscribeUrl,
-      });
     }
   }
 
   async collectionCreated(event: CollectionEvent) {
-    const collection = await Collection.findByPk(event.collectionId, {
-      include: [
-        {
-          model: User,
-          required: true,
-          as: "user",
-        },
-      ],
+    const collection = await Collection.scope("withUser").findByPk(
+      event.collectionId
+    );
+
+    if (!collection || !collection.permission) {
+      return;
+    }
+
+    const recipients = await this.getCollectionNotificationRecipients(
+      collection,
+      event.name
+    );
+
+    for (const recipient of recipients) {
+      // Suppress notifications for suspended users
+      if (recipient.user.isSuspended || !recipient.user.email) {
+        continue;
+      }
+
+      await CollectionNotificationEmail.schedule({
+        to: recipient.user.email,
+        eventName: "created",
+        collectionId: collection.id,
+        unsubscribeUrl: recipient.unsubscribeUrl,
+      });
+    }
+  }
+
+  /**
+   * Create any new subscriptions that might be missing for collaborators in the
+   * document on publish and revision creation. This does mean that there is a
+   * short period of time where the user is not subscribed after editing until a
+   * revision is created.
+   *
+   * @param document The document to create subscriptions for
+   * @param event The event that triggered the subscription creation
+   */
+  private createDocumentSubscriptions = async (
+    document: Document,
+    event: DocumentEvent | RevisionEvent
+  ): Promise<void> => {
+    await sequelize.transaction(async (transaction) => {
+      const users = await document.collaborators({ transaction });
+
+      for (const user of users) {
+        if (user && can(user, "subscribe", document)) {
+          await subscriptionCreator({
+            user,
+            documentId: document.id,
+            event: "documents.update",
+            resubscribe: false,
+            transaction,
+            ip: event.ip,
+          });
+        }
+      }
     });
-    if (!collection) {
-      return;
-    }
-    if (!collection.permission) {
-      return;
-    }
-    const notificationSettings = await NotificationSetting.findAll({
+  };
+
+  /**
+   * Get the recipients of a notification for a collection event.
+   *
+   * @param collection The collection to get recipients for
+   * @param eventName The event name
+   * @returns A list of recipients
+   */
+  private getCollectionNotificationRecipients = async (
+    collection: Collection,
+    eventName: string
+  ): Promise<NotificationSetting[]> => {
+    // First find all the users that have notifications enabled for this event
+    // type at all and aren't the one that performed the action.
+    const recipients = await NotificationSetting.scope("withUser").findAll({
       where: {
         userId: {
           [Op.ne]: collection.createdById,
         },
         teamId: collection.teamId,
-        event: event.name,
+        event: eventName,
       },
-      include: [
-        {
-          model: User,
-          required: true,
-          as: "user",
-        },
-      ],
     });
 
-    for (const setting of notificationSettings) {
-      // Suppress notifications for suspended users
-      if (setting.user.isSuspended || !setting.user.email) {
-        continue;
-      }
+    // Ensure we only have one recipient per user as a safety measure
+    return uniqBy(recipients, "userId");
+  };
 
-      await CollectionNotificationEmail.schedule({
-        to: setting.user.email,
-        eventName: "created",
-        collectionId: collection.id,
-        unsubscribeUrl: setting.unsubscribeUrl,
+  /**
+   * Get the recipients of a notification for a document event.
+   *
+   * @param document The document to get recipients for
+   * @param eventName The event name
+   * @returns A list of recipients
+   */
+  private getDocumentNotificationRecipients = async (
+    document: Document,
+    eventName: string
+  ): Promise<NotificationSetting[]> => {
+    // First find all the users that have notifications enabled for this event
+    // type at all and aren't the one that performed the action.
+    let recipients = await NotificationSetting.scope("withUser").findAll({
+      where: {
+        userId: {
+          [Op.ne]: document.lastModifiedById,
+        },
+        teamId: document.teamId,
+        event: eventName,
+      },
+    });
+
+    // If the event is a revision creation we can filter further to only those
+    // that have a subscription to the document…
+    if (eventName === "documents.update") {
+      const subscriptions = await Subscription.findAll({
+        attributes: ["userId"],
+        where: {
+          userId: recipients.map((recipient) => recipient.user.id),
+          documentId: document.id,
+          event: eventName,
+        },
       });
+
+      const subscribedUserIds = subscriptions.map(
+        (subscription) => subscription.userId
+      );
+
+      recipients = recipients.filter((recipient) =>
+        subscribedUserIds.includes(recipient.user.id)
+      );
     }
-  }
+
+    // Ensure we only have one recipient per user as a safety measure
+    return uniqBy(recipients, "userId");
+  };
+
+  private shouldNotify = async (
+    document: Document,
+    user: User
+  ): Promise<boolean> => {
+    // Suppress notifications for suspended and users with no email address
+    if (user.isSuspended || !user.email) {
+      return false;
+    }
+
+    // Check the recipient has access to the collection this document is in. Just
+    // because they are subscribed doesn't meant they still have access to read
+    // the document.
+    const collectionIds = await user.collectionIds();
+
+    if (!collectionIds.includes(document.collectionId)) {
+      return false;
+    }
+
+    // If this recipient has viewed the document since the last update was made
+    // then we can avoid sending them a useless notification, yay.
+    const view = await View.findOne({
+      where: {
+        userId: user.id,
+        documentId: document.id,
+        updatedAt: {
+          [Op.gt]: document.updatedAt,
+        },
+      },
+    });
+
+    if (view) {
+      Logger.info(
+        "processor",
+        `suppressing notification to ${user.id} because update viewed`
+      );
+      return false;
+    }
+
+    return true;
+  };
 }