chore: Move notification event writing to model layer (#8754)

* Move notification event writing to model layer
fix: Bulk notification action does not reflect on other clients

* Add missing locks

* fixes

server/commands/notificationUpdater.test.ts

@@ -1,182 +0,0 @@
-import { NotificationEventType } from "@shared/types";
-import { Event } from "@server/models";
-import {
-  buildUser,
-  buildNotification,
-  buildDocument,
-  buildCollection,
-} from "@server/test/factories";
-import { withAPIContext } from "@server/test/support";
-import notificationUpdater from "./notificationUpdater";
-
-describe("notificationUpdater", () => {
-  it("should mark the notification as viewed", async () => {
-    const user = await buildUser();
-    const actor = await buildUser({
-      teamId: user.teamId,
-    });
-    const collection = await buildCollection({
-      teamId: user.teamId,
-      createdById: actor.id,
-    });
-    const document = await buildDocument({
-      teamId: user.teamId,
-      collectionId: collection.id,
-      createdById: actor.id,
-    });
-    const notification = await buildNotification({
-      actorId: actor.id,
-      event: NotificationEventType.UpdateDocument,
-      userId: user.id,
-      teamId: user.teamId,
-      documentId: document.id,
-      collectionId: collection.id,
-    });
-
-    expect(notification.archivedAt).toBe(null);
-    expect(notification.viewedAt).toBe(null);
-
-    await withAPIContext(user, (ctx) =>
-      notificationUpdater(ctx, {
-        notification,
-        viewedAt: new Date(),
-      })
-    );
-    const event = await Event.findLatest({
-      teamId: user.teamId,
-    });
-
-    expect(notification.viewedAt).not.toBe(null);
-    expect(notification.archivedAt).toBe(null);
-    expect(event!.name).toEqual("notifications.update");
-    expect(event!.modelId).toEqual(notification.id);
-  });
-
-  it("should mark the notification as unseen", async () => {
-    const user = await buildUser();
-    const actor = await buildUser({
-      teamId: user.teamId,
-    });
-    const collection = await buildCollection({
-      teamId: user.teamId,
-      createdById: actor.id,
-    });
-    const document = await buildDocument({
-      teamId: user.teamId,
-      collectionId: collection.id,
-      createdById: actor.id,
-    });
-    const notification = await buildNotification({
-      actorId: actor.id,
-      event: NotificationEventType.UpdateDocument,
-      userId: user.id,
-      teamId: user.teamId,
-      documentId: document.id,
-      collectionId: collection.id,
-      viewedAt: new Date(),
-    });
-
-    expect(notification.archivedAt).toBe(null);
-    expect(notification.viewedAt).not.toBe(null);
-
-    await withAPIContext(user, (ctx) =>
-      notificationUpdater(ctx, {
-        notification,
-        viewedAt: null,
-      })
-    );
-    const event = await Event.findLatest({
-      teamId: user.teamId,
-    });
-
-    expect(notification.viewedAt).toBe(null);
-    expect(notification.archivedAt).toBe(null);
-    expect(event!.name).toEqual("notifications.update");
-    expect(event!.modelId).toEqual(notification.id);
-  });
-
-  it("should archive the notification", async () => {
-    const user = await buildUser();
-    const actor = await buildUser({
-      teamId: user.teamId,
-    });
-    const collection = await buildCollection({
-      teamId: user.teamId,
-      createdById: actor.id,
-    });
-    const document = await buildDocument({
-      teamId: user.teamId,
-      collectionId: collection.id,
-      createdById: actor.id,
-    });
-    const notification = await buildNotification({
-      actorId: actor.id,
-      event: NotificationEventType.UpdateDocument,
-      userId: user.id,
-      teamId: user.teamId,
-      documentId: document.id,
-      collectionId: collection.id,
-    });
-
-    expect(notification.archivedAt).toBe(null);
-    expect(notification.viewedAt).toBe(null);
-
-    await withAPIContext(user, (ctx) =>
-      notificationUpdater(ctx, {
-        notification,
-        archivedAt: new Date(),
-      })
-    );
-    const event = await Event.findLatest({
-      teamId: user.teamId,
-    });
-
-    expect(notification.viewedAt).toBe(null);
-    expect(notification.archivedAt).not.toBe(null);
-    expect(event!.name).toEqual("notifications.update");
-    expect(event!.modelId).toEqual(notification.id);
-  });
-
-  it("should unarchive the notification", async () => {
-    const user = await buildUser();
-    const actor = await buildUser({
-      teamId: user.teamId,
-    });
-    const collection = await buildCollection({
-      teamId: user.teamId,
-      createdById: actor.id,
-    });
-    const document = await buildDocument({
-      teamId: user.teamId,
-      collectionId: collection.id,
-      createdById: actor.id,
-    });
-    const notification = await buildNotification({
-      actorId: actor.id,
-      event: NotificationEventType.UpdateDocument,
-      userId: user.id,
-      teamId: user.teamId,
-      documentId: document.id,
-      collectionId: collection.id,
-      archivedAt: new Date(),
-    });
-
-    expect(notification.archivedAt).not.toBe(null);
-    expect(notification.viewedAt).toBe(null);
-
-    await withAPIContext(user, (ctx) =>
-      notificationUpdater(ctx, {
-        notification,
-        archivedAt: null,
-      })
-    );
-    const event = await Event.findLatest({
-      teamId: user.teamId,
-    });
-
-    expect(notification.viewedAt).toBe(null);
-    expect(notification.archivedAt).toBeNull();
-    expect(event!.name).toEqual("notifications.update");
-    expect(event!.modelId).toEqual(notification.id);
-  });
-});

server/commands/notificationUpdater.ts

@@ -1,53 +0,0 @@
-import isUndefined from "lodash/isUndefined";
-import { Event, Notification } from "@server/models";
-import { APIContext } from "@server/types";
-
-type Props = {
-  /** Notification to be updated */
-  notification: Notification;
-  /** Time at which notification was viewed */
-  viewedAt?: Date | null;
-  /** Time at which notification was archived */
-  archivedAt?: Date | null;
-};
-
-/**
- * This command updates notification properties.
- *
- * @param ctx The originating request context
- * @param Props The properties of the notification to update
- * @returns Notification The updated notification
- */
-export default async function notificationUpdater(
-  ctx: APIContext,
-  { notification, viewedAt, archivedAt }: Props
-): Promise<Notification> {
-  const { transaction } = ctx.state;
-
-  if (!isUndefined(viewedAt)) {
-    notification.viewedAt = viewedAt;
-  }
-  if (!isUndefined(archivedAt)) {
-    notification.archivedAt = archivedAt;
-  }
-  const changed = notification.changed();
-  if (changed) {
-    await notification.save({ transaction });
-
-    await Event.createFromContext(
-      ctx,
-      {
-        name: "notifications.update",
-        userId: notification.userId,
-        modelId: notification.id,
-        documentId: notification.documentId,
-      },
-      {
-        actorId: notification.userId,
-        teamId: notification.teamId,
-      }
-    );
-  }
-
-  return notification;
-}

server/models/base/Model.ts

@@ -280,6 +280,7 @@ class Model<
    *
    * @param query The query options.
    * @param callback The function to call for each batch of results
+   * @return The total number of results processed.
    */
   static async findAllInBatches<T extends Model>(
     query: Replace<FindOptions<T>, "limit", "batchLimit"> & {
@@ -287,7 +288,8 @@ class Model<
       totalLimit?: number;
     },
     callback: (results: Array<T>, query: FindOptions<T>) => Promise<void>
-  ) {
+  ): Promise<number> {
+    let total = 0;
     const mappedQuery = {
       ...query,
       offset: query.offset ?? 0,
@@ -299,12 +301,15 @@ class Model<
     do {
       // @ts-expect-error this T
       results = await this.findAll<T>(mappedQuery);
+      total += results.length;
       await callback(results, mappedQuery);
       mappedQuery.offset += mappedQuery.limit;
     } while (
       results.length >= mappedQuery.limit &&
       (mappedQuery.totalLimit ?? Infinity) > mappedQuery.offset
     );
+
+    return total;
   }
 
   /**

server/routes/api/notifications/notifications.ts

@@ -1,10 +1,10 @@
 import Router from "koa-router";
-import { isNil } from "lodash";
+import isEmpty from "lodash/isEmpty";
+import isNil from "lodash/isNil";
 import isNull from "lodash/isNull";
 import isUndefined from "lodash/isUndefined";
 import { WhereOptions, Op } from "sequelize";
 import { NotificationEventType } from "@shared/types";
-import notificationUpdater from "@server/commands/notificationUpdater";
 import env from "@server/env";
 import { AuthenticationError } from "@server/errors";
 import auth from "@server/middlewares/authentication";
@@ -29,6 +29,7 @@ const pixel = Buffer.from(
 const handleUnsubscribe = async (
   ctx: APIContext<T.NotificationsUnsubscribeReq>
 ) => {
+  const { transaction } = ctx.state;
   const eventType = (ctx.input.body.eventType ??
     ctx.input.query.eventType) as NotificationEventType;
   const userId = (ctx.input.body.userId ?? ctx.input.query.userId) as string;
@@ -46,6 +47,8 @@ const handleUnsubscribe = async (
 
   const user = await User.scope("withTeam").findByPk(userId, {
     rejectOnEmpty: true,
+    lock: transaction.LOCK.UPDATE,
+    transaction,
   });
 
   user.setNotificationEventType(eventType, false);
@@ -145,17 +148,21 @@ router.get(
   transaction(),
   async (ctx: APIContext<T.NotificationsPixelReq>) => {
     const { id, token } = ctx.input.query;
-    const notification = await Notification.unscoped().findByPk(id);
+    const { transaction } = ctx.state;
+
+    const notification = await Notification.unscoped().findByPk(id, {
+      lock: transaction.LOCK.UPDATE,
+      rejectOnEmpty: true,
+      transaction,
+    });
 
     if (!notification || !safeEqual(token, notification.pixelToken)) {
       throw AuthenticationError();
     }
 
     if (!notification.viewedAt) {
-      await notificationUpdater(ctx, {
-        notification,
-        viewedAt: new Date(),
-      });
+      notification.viewedAt = new Date();
+      await notification.saveWithCtx(ctx);
     }
 
     ctx.response.set("Content-Type", "image/gif");
@@ -171,15 +178,25 @@ router.post(
   async (ctx: APIContext<T.NotificationsUpdateReq>) => {
     const { id, viewedAt, archivedAt } = ctx.input.body;
     const { user } = ctx.state.auth;
+    const { transaction } = ctx.state;
 
-    const notification = await Notification.findByPk(id);
+    const notification = await Notification.findByPk(id, {
+      lock: {
+        level: transaction.LOCK.UPDATE,
+        of: Notification,
+      },
+      rejectOnEmpty: true,
+      transaction,
+    });
     authorize(user, "update", notification);
 
-    await notificationUpdater(ctx, {
-      notification,
-      viewedAt,
-      archivedAt,
-    });
+    if (!isUndefined(viewedAt)) {
+      notification.viewedAt = viewedAt;
+    }
+    if (!isUndefined(archivedAt)) {
+      notification.archivedAt = archivedAt;
+    }
+    await notification.saveWithCtx(ctx);
 
     ctx.body = {
       data: await presentNotification(ctx, notification),
@@ -196,7 +213,7 @@ router.post(
     const { viewedAt, archivedAt } = ctx.input.body;
     const { user } = ctx.state.auth;
 
-    const values: { [x: string]: any } = {};
+    const values: Partial<Notification> = {};
     let where: WhereOptions<Notification> = {
       teamId: user.teamId,
       userId: user.id,
@@ -216,7 +233,19 @@ router.post(
       };
     }
 
-    const [total] = await Notification.update(values, { where });
+    let total = 0;
+    if (!isEmpty(values)) {
+      total = await Notification.findAllInBatches(
+        { where },
+        async (results) => {
+          await Promise.all(
+            results.map((notification) =>
+              notification.updateWithCtx(ctx, values)
+            )
+          );
+        }
+      );
+    }
 
     ctx.body = {
       success: true,