feat(notifications): User Notifications

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

@@ -0,0 +1,43 @@
+import { QueueJobs, QueueName, TQueueServiceFactory } from "@app/queue";
+
+import { TCreateUserNotificationDTO } from "./notification-types";
+import { TUserNotificationDALFactory } from "./user-notification-dal";
+
+type TNotificationQueueServiceFactoryDep = {
+  userNotificationDAL: Pick<TUserNotificationDALFactory, "create">;
+  queueService: TQueueServiceFactory;
+};
+
+export type TNotificationQueueServiceFactory = {
+  pushUserNotification: (data: TCreateUserNotificationDTO) => Promise<void>;
+};
+
+export const notificationQueueServiceFactory = async ({
+  userNotificationDAL,
+  queueService
+}: TNotificationQueueServiceFactoryDep): Promise<TNotificationQueueServiceFactory> => {
+  const pushUserNotification = async (data: TCreateUserNotificationDTO) => {
+    await queueService.queue<QueueName.UserNotification>(QueueName.UserNotification, QueueJobs.UserNotification, data, {
+      removeOnFail: {
+        count: 3
+      },
+      removeOnComplete: true
+    });
+  };
+
+  queueService.start(QueueName.UserNotification, async (job) => {
+    const { userId, type, title, body, link } = job.data;
+
+    await userNotificationDAL.create({
+      userId,
+      type,
+      title,
+      body,
+      link
+    });
+  });
+
+  return {
+    pushUserNotification
+  };
+};

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

@@ -0,0 +1,70 @@
+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 }: { userId: string }) => {
+    const now = new Date();
+    const threeMonthsAgo = new Date();
+    threeMonthsAgo.setMonth(threeMonthsAgo.getMonth() - 3);
+
+    const notifications = await userNotificationDAL.find({
+      userId,
+      startDate: threeMonthsAgo.toISOString(),
+      endDate: now.toISOString()
+    });
+
+    return notifications;
+  };
+
+  const createUserNotification = async (data: TCreateUserNotificationDTO) => {
+    return notificationQueue.pushUserNotification(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 }: { userId: string }) => {
+    await userNotificationDAL.markAllNotificationsAsRead(userId);
+  };
+
+  const markUserNotificationAsRead = async ({ userId, notificationId }: { userId: string; notificationId: string }) => {
+    await userNotificationDAL.update(
+      {
+        id: notificationId,
+        userId,
+        isRead: false
+      },
+      {
+        isRead: true
+      }
+    );
+  };
+
+  return {
+    listUserNotifications,
+    createUserNotification,
+    deleteUserNotification,
+    markUserNotificationsAsRead,
+    markUserNotificationAsRead
+  };
+};

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

@@ -0,0 +1,22 @@
+export enum NotificationType {
+  ACCESS_APPROVAL_REQUEST = "access-approval-request"
+}
+
+export interface TUserNotification {
+  id: string;
+  userId: string;
+  type: string;
+  title: string;
+  body?: string | null;
+  link?: string | null;
+  isRead: boolean;
+  createdAt: Date;
+}
+
+export interface TCreateUserNotificationDTO {
+  userId: string;
+  type: NotificationType;
+  title: string;
+  body?: string;
+  link?: string;
+}

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

@@ -0,0 +1,118 @@
+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,
+      startDate,
+      endDate,
+      limit = 10000,
+      offset = 0
+    }: {
+      userId: 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)
+        .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) => {
+    await db(TableName.UserNotifications).where({ userId }).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`);