Merge pull request #4448 from Infisical/ENG-3586

feat(notifications): User Notifications

backend/src/services/notification/notification-queue.ts

@@ -0,0 +1,39 @@
+import { QueueJobs, TQueueServiceFactory } from "@app/queue";
+
+import { TCreateUserNotificationDTO } from "./notification-types";
+import { TUserNotificationDALFactory } from "./user-notification-dal";
+
+type TNotificationQueueServiceFactoryDep = {
+  userNotificationDAL: Pick<TUserNotificationDALFactory, "batchInsert">;
+  queueService: TQueueServiceFactory;
+};
+
+export type TNotificationQueueServiceFactory = {
+  pushUserNotifications: (data: TCreateUserNotificationDTO[]) => Promise<void>;
+};
+
+export const notificationQueueServiceFactory = async ({
+  userNotificationDAL,
+  queueService
+}: TNotificationQueueServiceFactoryDep): Promise<TNotificationQueueServiceFactory> => {
+  const pushUserNotifications = async (data: TCreateUserNotificationDTO[]) => {
+    await queueService.queuePg(QueueJobs.UserNotification, { notifications: data });
+  };
+
+  await queueService.startPg(
+    QueueJobs.UserNotification,
+    async ([job]) => {
+      const { notifications } = job.data as { notifications: TCreateUserNotificationDTO[] };
+      await userNotificationDAL.batchInsert(notifications);
+    },
+    {
+      batchSize: 1,
+      workerCount: 2,
+      pollingIntervalSeconds: 1
+    }
+  );
+
+  return {
+    pushUserNotifications
+  };
+};

backend/src/services/notification/notification-service.ts

@@ -0,0 +1,81 @@
+import { NotFoundError, UnauthorizedError } from "@app/lib/errors";
+
+import { TNotificationQueueServiceFactory } from "./notification-queue";
+import { TCreateUserNotificationDTO } from "./notification-types";
+import { TUserNotificationDALFactory } from "./user-notification-dal";
+
+type TNotificationServiceFactoryDep = {
+  notificationQueue: TNotificationQueueServiceFactory;
+  userNotificationDAL: TUserNotificationDALFactory;
+};
+
+export type TNotificationServiceFactory = ReturnType<typeof notificationServiceFactory>;
+
+export const notificationServiceFactory = ({
+  notificationQueue,
+  userNotificationDAL
+}: TNotificationServiceFactoryDep) => {
+  const listUserNotifications = async ({ userId, orgId }: { userId: string; orgId: string }) => {
+    const now = new Date();
+    const threeMonthsAgo = new Date();
+    threeMonthsAgo.setMonth(threeMonthsAgo.getMonth() - 3);
+
+    const notifications = await userNotificationDAL.find({
+      userId,
+      orgId,
+      startDate: threeMonthsAgo.toISOString(),
+      endDate: now.toISOString()
+    });
+
+    return notifications;
+  };
+
+  const createUserNotifications = async (data: TCreateUserNotificationDTO[]) => {
+    return notificationQueue.pushUserNotifications(data);
+  };
+
+  const deleteUserNotification = async ({ userId, notificationId }: { userId: string; notificationId: string }) => {
+    if (!userId) throw new UnauthorizedError({ message: "Invalid userId" });
+
+    const deletedNotifications = await userNotificationDAL.delete({ id: notificationId, userId });
+
+    if (deletedNotifications.length <= 0) throw new NotFoundError({ message: "Notification not found" });
+
+    return deletedNotifications[0];
+  };
+
+  const markUserNotificationsAsRead = async ({ userId, orgId }: { userId: string; orgId: string }) => {
+    await userNotificationDAL.markAllNotificationsAsRead(userId, orgId);
+  };
+
+  const updateUserNotification = async ({
+    userId,
+    notificationId,
+    isRead
+  }: {
+    userId: string;
+    notificationId: string;
+    isRead: boolean;
+  }) => {
+    const [updatedNotification] = await userNotificationDAL.update(
+      {
+        id: notificationId,
+        userId
+      },
+      {
+        isRead
+      }
+    );
+
+    if (!updatedNotification) throw new NotFoundError({ message: "Notification not found" });
+    return updatedNotification;
+  };
+
+  return {
+    listUserNotifications,
+    createUserNotifications,
+    deleteUserNotification,
+    markUserNotificationsAsRead,
+    updateUserNotification
+  };
+};

backend/src/services/notification/notification-types.ts

@@ -0,0 +1,15 @@
+export enum NotificationType {
+  ACCESS_APPROVAL_REQUEST = "access-approval-request",
+  ACCESS_APPROVAL_REQUEST_UPDATED = "access-approval-request-updated"
+}
+
+export interface TCreateUserNotificationDTO {
+  userId: string;
+  // Adding an orgId will make the notification only show up when a user is in a certain org. Otherwise, it shows up in all orgs.
+  // Keep in mind that org-scoped links for a notification will break if orgId is missing and the user is in the wrong org
+  orgId?: string;
+  type: NotificationType;
+  title: string;
+  body?: string;
+  link?: string;
+}

backend/src/services/notification/user-notification-dal.ts

@@ -0,0 +1,130 @@
+import knex from "knex";
+
+import { TDbClient } from "@app/db";
+import { TableName } from "@app/db/schemas";
+import { DatabaseError, GatewayTimeoutError } from "@app/lib/errors";
+import { ormify, selectAllTableCols } from "@app/lib/knex";
+import { logger } from "@app/lib/logger";
+import { QueueName } from "@app/queue";
+
+export type TUserNotificationDALFactory = ReturnType<typeof userNotificationDALFactory>;
+
+const QUERY_TIMEOUT_MS = 10 * 60 * 1000; // 10 minutes
+const PRUNE_BATCH_SIZE = 10000;
+const MAX_RETRY_ON_FAILURE = 3;
+
+export const userNotificationDALFactory = (db: TDbClient) => {
+  const notificationOrm = ormify(db, TableName.UserNotifications);
+
+  const find = async (
+    {
+      userId,
+      orgId,
+      startDate,
+      endDate,
+      limit = 1000,
+      offset = 0
+    }: {
+      userId: string;
+      orgId: string;
+      startDate: string;
+      endDate: string;
+      limit?: number;
+      offset?: number;
+    },
+    tx?: knex.Knex
+  ) => {
+    try {
+      const docs = await (tx || db.replicaNode())(TableName.UserNotifications)
+        .where(`${TableName.UserNotifications}.userId`, userId)
+        .andWhere((qb) => {
+          void qb
+            .where(`${TableName.UserNotifications}.orgId`, orgId)
+            .orWhereNull(`${TableName.UserNotifications}.orgId`);
+        })
+        .whereRaw(`"${TableName.UserNotifications}"."createdAt" >= ?::timestamptz`, [startDate])
+        .andWhereRaw(`"${TableName.UserNotifications}"."createdAt" < ?::timestamptz`, [endDate])
+        .select(selectAllTableCols(TableName.UserNotifications))
+        .limit(limit)
+        .offset(offset)
+        .orderBy(`${TableName.UserNotifications}.createdAt`, "desc")
+        .timeout(1000 * 120); // 2 minutes timeout
+
+      return docs;
+    } catch (error) {
+      if (error instanceof knex.KnexTimeoutError) {
+        throw new GatewayTimeoutError({
+          error,
+          message: "Failed to fetch notifications due to timeout."
+        });
+      }
+
+      throw new DatabaseError({ error });
+    }
+  };
+
+  // delete all notifications older than 3 months
+  const pruneNotifications = async () => {
+    const threeMonthsAgo = new Date();
+    threeMonthsAgo.setMonth(threeMonthsAgo.getMonth() - 3);
+    let deletedNotificationIds: { id: string }[] = [];
+    let numberOfRetryOnFailure = 0;
+
+    logger.info(`${QueueName.DailyResourceCleanUp}: prune notifications started`);
+    do {
+      try {
+        // eslint-disable-next-line no-await-in-loop
+        deletedNotificationIds = await db.transaction(async (trx) => {
+          await trx.raw(`SET statement_timeout = ${QUERY_TIMEOUT_MS}`);
+
+          const findExpiredNotificationSubQuery = trx(TableName.UserNotifications)
+            .where("createdAt", "<", threeMonthsAgo)
+            .orderBy(`${TableName.UserNotifications}.createdAt`, "desc")
+            .select("id")
+            .limit(PRUNE_BATCH_SIZE);
+
+          // eslint-disable-next-line no-await-in-loop
+          const results = await trx(TableName.UserNotifications)
+            .whereIn("id", findExpiredNotificationSubQuery)
+            .del()
+            .returning("id");
+
+          return results;
+        });
+
+        numberOfRetryOnFailure = 0;
+      } catch (error) {
+        numberOfRetryOnFailure += 1;
+        deletedNotificationIds = [];
+        logger.error(error, "Failed to delete notification on pruning. Retrying...");
+      } finally {
+        // eslint-disable-next-line no-await-in-loop
+        await new Promise((resolve) => {
+          setTimeout(resolve, 10);
+        });
+      }
+    } while (
+      deletedNotificationIds.length > 0 ||
+      (numberOfRetryOnFailure > 0 && numberOfRetryOnFailure < MAX_RETRY_ON_FAILURE)
+    );
+
+    if (numberOfRetryOnFailure >= MAX_RETRY_ON_FAILURE) {
+      logger.error(
+        `${QueueName.DailyResourceCleanUp}: prune notifications completed with persistent errors after ${MAX_RETRY_ON_FAILURE} retries. Some notifications might not have been pruned.`
+      );
+    } else {
+      logger.info(`${QueueName.DailyResourceCleanUp}: prune notifications completed`);
+    }
+  };
+
+  const markAllNotificationsAsRead = async (userId: string, orgId: string) => {
+    await db(TableName.UserNotifications)
+      .where({ userId })
+      .andWhere((qb) => {
+        void qb.where({ orgId }).orWhereNull("orgId");
+      })
+      .update({ isRead: true });
+  };
+
+  return { ...notificationOrm, pruneNotifications, find, markAllNotificationsAsRead };
+};

backend/src/services/resource-cleanup/resource-cleanup-queue.ts

@@ -3,6 +3,7 @@ import { TSnapshotDALFactory } from "@app/ee/services/secret-snapshot/snapshot-d
 import { getConfig } from "@app/lib/config/env";
 import { logger } from "@app/lib/logger";
 import { QueueJobs, QueueName, TQueueServiceFactory } from "@app/queue";
+import { TUserNotificationDALFactory } from "@app/services/notification/user-notification-dal";
 
 import { TIdentityAccessTokenDALFactory } from "../identity-access-token/identity-access-token-dal";
 import { TIdentityUaClientSecretDALFactory } from "../identity-ua/identity-ua-client-secret-dal";
@@ -25,6 +26,7 @@ type TDailyResourceCleanUpQueueServiceFactoryDep = {
   serviceTokenService: Pick<TServiceTokenServiceFactory, "notifyExpiringTokens">;
   queueService: TQueueServiceFactory;
   orgService: TOrgServiceFactory;
+  userNotificationDAL: Pick<TUserNotificationDALFactory, "pruneNotifications">;
 };
 
 export type TDailyResourceCleanUpQueueServiceFactory = ReturnType<typeof dailyResourceCleanUpQueueServiceFactory>;
@@ -40,7 +42,8 @@ export const dailyResourceCleanUpQueueServiceFactory = ({
   secretVersionV2DAL,
   identityUniversalAuthClientSecretDAL,
   serviceTokenService,
-  orgService
+  orgService,
+  userNotificationDAL
 }: TDailyResourceCleanUpQueueServiceFactoryDep) => {
   const appCfg = getConfig();
 
@@ -78,6 +81,7 @@ export const dailyResourceCleanUpQueueServiceFactory = ({
           await serviceTokenService.notifyExpiringTokens();
           await orgService.notifyInvitedUsers();
           await auditLogDAL.pruneAuditLog();
+          await userNotificationDAL.pruneNotifications();
           logger.info(`${QueueName.DailyResourceCleanUp}: queue task completed`);
         } catch (error) {
           logger.error(error, `${QueueName.DailyResourceCleanUp}: resource cleanup failed`);