feat: notifications (#1947)

* feat: notifications
* feat: watchers

* improvements

* handle page move for watchers

* make watchers non-blocking

* more

apps/server/src/integrations/queue/constants/queue.constants.ts

@@ -7,6 +7,7 @@ export enum QueueName {
   SEARCH_QUEUE = '{search-queue}',
   AI_QUEUE = '{ai-queue}',
   HISTORY_QUEUE = '{history-queue}',
+  NOTIFICATION_QUEUE = '{notification-queue}',
 }
 
 export enum QueueJob {
@@ -19,6 +20,7 @@ export enum QueueJob {
   DELETE_USER_AVATARS = 'delete-user-avatars',
 
   PAGE_BACKLINKS = 'page-backlinks',
+  ADD_PAGE_WATCHERS = 'add-page-watchers',
 
   STRIPE_SEATS_SYNC = 'sync-stripe-seats',
   TRIAL_ENDED = 'trial-ended',
@@ -61,4 +63,8 @@ export enum QueueJob {
   DELETE_PAGE_EMBEDDINGS = 'delete-page-embeddings',
 
   PAGE_HISTORY = 'page-history',
+
+  COMMENT_NOTIFICATION = 'comment-notification',
+  COMMENT_RESOLVED_NOTIFICATION = 'comment-resolved-notification',
+  PAGE_MENTION_NOTIFICATION = 'page-mention-notification',
 }

apps/server/src/integrations/queue/constants/queue.interface.ts

@@ -7,10 +7,56 @@ export interface IPageBacklinkJob {
   mentions: MentionNode[];
 }
 
+export interface IAddPageWatchersJob {
+  userIds: string[];
+  pageId: string;
+  spaceId: string;
+  workspaceId: string;
+}
+
 export interface IStripeSeatsSyncJob {
   workspaceId: string;
 }
 
 export interface IPageHistoryJob {
   pageId: string;
-}
+}
+
+export interface INotificationCreateJob {
+  userId: string;
+  workspaceId: string;
+  type: string;
+  actorId?: string;
+  pageId?: string;
+  spaceId?: string;
+  commentId?: string;
+  data?: Record<string, unknown>;
+}
+
+export interface ICommentNotificationJob {
+  commentId: string;
+  parentCommentId?: string;
+  pageId: string;
+  spaceId: string;
+  workspaceId: string;
+  actorId: string;
+  mentionedUserIds: string[];
+  notifyWatchers: boolean;
+}
+
+export interface ICommentResolvedNotificationJob {
+  commentId: string;
+  commentCreatorId: string;
+  pageId: string;
+  spaceId: string;
+  workspaceId: string;
+  actorId: string;
+}
+
+export interface IPageMentionNotificationJob {
+  userMentions: { userId: string; mentionId: string; creatorId: string }[];
+  oldMentionedUserIds: string[];
+  pageId: string;
+  spaceId: string;
+  workspaceId: string;
+}

apps/server/src/integrations/queue/processors/backlinks.processor.ts

@@ -1,135 +0,0 @@
-import { Logger, OnModuleDestroy } from '@nestjs/common';
-import { OnWorkerEvent, Processor, WorkerHost } from '@nestjs/bullmq';
-import { Job } from 'bullmq';
-import { QueueJob, QueueName } from '../constants';
-import { IPageBacklinkJob } from '../constants/queue.interface';
-import { InjectKysely } from 'nestjs-kysely';
-import { KyselyDB } from '@docmost/db/types/kysely.types';
-import { BacklinkRepo } from '@docmost/db/repos/backlink/backlink.repo';
-import { executeTx } from '@docmost/db/utils';
-
-@Processor(QueueName.GENERAL_QUEUE)
-export class BacklinksProcessor extends WorkerHost implements OnModuleDestroy {
-  private readonly logger = new Logger(BacklinksProcessor.name);
-  constructor(
-    @InjectKysely() private readonly db: KyselyDB,
-    private readonly backlinkRepo: BacklinkRepo,
-  ) {
-    super();
-  }
-
-  async process(job: Job<IPageBacklinkJob, void>): Promise<void> {
-    try {
-      const { pageId, mentions, workspaceId } = job.data;
-
-      switch (job.name) {
-        case QueueJob.PAGE_BACKLINKS:
-          {
-            await executeTx(this.db, async (trx) => {
-              const existingBacklinks = await trx
-                .selectFrom('backlinks')
-                .select('targetPageId')
-                .where('sourcePageId', '=', pageId)
-                .execute();
-
-              if (existingBacklinks.length === 0 && mentions.length === 0) {
-                return;
-              }
-
-              const existingTargetPageIds = existingBacklinks.map(
-                (backlink) => backlink.targetPageId,
-              );
-
-              const targetPageIds = mentions
-                .filter((mention) => mention.entityId !== pageId)
-                .map((mention) => mention.entityId);
-
-              // make sure target pages belong to the same workspace
-              let validTargetPages = [];
-              if (targetPageIds.length > 0) {
-                validTargetPages = await trx
-                  .selectFrom('pages')
-                  .select('id')
-                  .where('id', 'in', targetPageIds)
-                  .where('workspaceId', '=', workspaceId)
-                  .execute();
-              }
-
-              const validTargetPageIds = validTargetPages.map(
-                (page) => page.id,
-              );
-
-              // new backlinks
-              const backlinksToAdd = validTargetPageIds.filter(
-                (id) => !existingTargetPageIds.includes(id),
-              );
-
-              // stale backlinks
-              const backlinksToRemove = existingTargetPageIds.filter(
-                (existingId) => !validTargetPageIds.includes(existingId),
-              );
-
-              // add new backlinks
-              if (backlinksToAdd.length > 0) {
-                const newBacklinks = backlinksToAdd.map((targetPageId) => ({
-                  sourcePageId: pageId,
-                  targetPageId: targetPageId,
-                  workspaceId: workspaceId,
-                }));
-
-                await this.backlinkRepo.insertBacklink(newBacklinks, trx);
-                this.logger.debug(
-                  `Added ${newBacklinks.length} new backlinks to ${pageId}`,
-                );
-              }
-
-              // remove stale backlinks
-              if (backlinksToRemove.length > 0) {
-                await this.db
-                  .deleteFrom('backlinks')
-                  .where('sourcePageId', '=', pageId)
-                  .where('targetPageId', 'in', backlinksToRemove)
-                  .execute();
-
-                this.logger.debug(
-                  `Removed ${backlinksToRemove.length} outdated backlinks from ${pageId}.`,
-                );
-              }
-            });
-          }
-          break;
-      }
-    } catch (err) {
-      throw err;
-    }
-  }
-
-  @OnWorkerEvent('active')
-  onActive(job: Job) {
-    if (job.name === QueueJob.PAGE_BACKLINKS) {
-      this.logger.debug(`Processing ${job.name} job`);
-    }
-  }
-
-  @OnWorkerEvent('failed')
-  onError(job: Job) {
-    if (job.name === QueueJob.PAGE_BACKLINKS) {
-      this.logger.error(
-        `Error processing ${job.name} job. Reason: ${job.failedReason}`,
-      );
-    }
-  }
-
-  @OnWorkerEvent('completed')
-  onCompleted(job: Job) {
-    if (job.name === QueueJob.PAGE_BACKLINKS) {
-      this.logger.debug(`Completed ${job.name} job`);
-    }
-  }
-
-  async onModuleDestroy(): Promise<void> {
-    if (this.worker) {
-      await this.worker.close();
-    }
-  }
-}

apps/server/src/integrations/queue/processors/general-queue.processor.ts

@@ -0,0 +1,87 @@
+import { Logger, OnModuleDestroy } from '@nestjs/common';
+import { OnWorkerEvent, Processor, WorkerHost } from '@nestjs/bullmq';
+import { Job } from 'bullmq';
+import { QueueJob, QueueName } from '../constants';
+import {
+  IAddPageWatchersJob,
+  IPageBacklinkJob,
+} from '../constants/queue.interface';
+import { InjectKysely } from 'nestjs-kysely';
+import { KyselyDB } from '@docmost/db/types/kysely.types';
+import { BacklinkRepo } from '@docmost/db/repos/backlink/backlink.repo';
+import {
+  WatcherRepo,
+  WatcherType,
+} from '@docmost/db/repos/watcher/watcher.repo';
+import { InsertableWatcher } from '@docmost/db/types/entity.types';
+import { processBacklinks } from '../tasks/backlinks.task';
+
+@Processor(QueueName.GENERAL_QUEUE)
+export class GeneralQueueProcessor
+  extends WorkerHost
+  implements OnModuleDestroy
+{
+  private readonly logger = new Logger(GeneralQueueProcessor.name);
+  constructor(
+    @InjectKysely() private readonly db: KyselyDB,
+    private readonly backlinkRepo: BacklinkRepo,
+    private readonly watcherRepo: WatcherRepo,
+  ) {
+    super();
+  }
+
+  async process(job: Job): Promise<void> {
+    try {
+      switch (job.name) {
+        case QueueJob.ADD_PAGE_WATCHERS: {
+          const { userIds, pageId, spaceId, workspaceId } =
+            job.data as IAddPageWatchersJob;
+          const watchers: InsertableWatcher[] = userIds.map((userId) => ({
+            userId,
+            pageId,
+            spaceId,
+            workspaceId,
+            type: WatcherType.PAGE,
+            addedById: userId,
+          }));
+          await this.watcherRepo.insertMany(watchers);
+          break;
+        }
+
+        case QueueJob.PAGE_BACKLINKS: {
+          await processBacklinks(
+            this.db,
+            this.backlinkRepo,
+            job.data as IPageBacklinkJob,
+          );
+          break;
+        }
+      }
+    } catch (err) {
+      throw err;
+    }
+  }
+
+  @OnWorkerEvent('active')
+  onActive(job: Job) {
+    this.logger.debug(`Processing ${job.name} job`);
+  }
+
+  @OnWorkerEvent('failed')
+  onError(job: Job) {
+    this.logger.error(
+      `Error processing ${job.name} job. Reason: ${job.failedReason}`,
+    );
+  }
+
+  @OnWorkerEvent('completed')
+  onCompleted(job: Job) {
+    this.logger.debug(`Completed ${job.name} job`);
+  }
+
+  async onModuleDestroy(): Promise<void> {
+    if (this.worker) {
+      await this.worker.close();
+    }
+  }
+}

apps/server/src/integrations/queue/queue.module.ts

@@ -3,7 +3,7 @@ import { BullModule } from '@nestjs/bullmq';
 import { EnvironmentService } from '../environment/environment.service';
 import { createRetryStrategy, parseRedisUrl } from '../../common/helpers';
 import { QueueName } from './constants';
-import { BacklinksProcessor } from './processors/backlinks.processor';
+import { GeneralQueueProcessor } from './processors/general-queue.processor';
 
 @Global()
 @Module({
@@ -81,8 +81,11 @@ import { BacklinksProcessor } from './processors/backlinks.processor';
         attempts: 2,
       },
     }),
+    BullModule.registerQueue({
+      name: QueueName.NOTIFICATION_QUEUE,
+    }),
   ],
   exports: [BullModule],
-  providers: [BacklinksProcessor],
+  providers: [GeneralQueueProcessor],
 })
 export class QueueModule {}

apps/server/src/integrations/queue/tasks/backlinks.task.ts

@@ -0,0 +1,80 @@
+import { Logger } from '@nestjs/common';
+import { KyselyDB } from '@docmost/db/types/kysely.types';
+import { BacklinkRepo } from '@docmost/db/repos/backlink/backlink.repo';
+import { IPageBacklinkJob } from '../constants/queue.interface';
+import { executeTx } from '@docmost/db/utils';
+
+const logger = new Logger('BacklinksTask');
+
+export async function processBacklinks(
+  db: KyselyDB,
+  backlinkRepo: BacklinkRepo,
+  data: IPageBacklinkJob,
+): Promise<void> {
+  const { pageId, mentions, workspaceId } = data;
+
+  await executeTx(db, async (trx) => {
+    const existingBacklinks = await trx
+      .selectFrom('backlinks')
+      .select('targetPageId')
+      .where('sourcePageId', '=', pageId)
+      .execute();
+
+    if (existingBacklinks.length === 0 && mentions.length === 0) {
+      return;
+    }
+
+    const existingTargetPageIds = existingBacklinks.map(
+      (backlink) => backlink.targetPageId,
+    );
+
+    const targetPageIds = mentions
+      .filter((mention) => mention.entityId !== pageId)
+      .map((mention) => mention.entityId);
+
+    let validTargetPages = [];
+    if (targetPageIds.length > 0) {
+      validTargetPages = await trx
+        .selectFrom('pages')
+        .select('id')
+        .where('id', 'in', targetPageIds)
+        .where('workspaceId', '=', workspaceId)
+        .execute();
+    }
+
+    const validTargetPageIds = validTargetPages.map((page) => page.id);
+
+    const backlinksToAdd = validTargetPageIds.filter(
+      (id) => !existingTargetPageIds.includes(id),
+    );
+
+    const backlinksToRemove = existingTargetPageIds.filter(
+      (existingId) => !validTargetPageIds.includes(existingId),
+    );
+
+    if (backlinksToAdd.length > 0) {
+      const newBacklinks = backlinksToAdd.map((targetPageId) => ({
+        sourcePageId: pageId,
+        targetPageId: targetPageId,
+        workspaceId: workspaceId,
+      }));
+
+      await backlinkRepo.insertBacklink(newBacklinks, trx);
+      logger.debug(
+        `Added ${newBacklinks.length} new backlinks to ${pageId}`,
+      );
+    }
+
+    if (backlinksToRemove.length > 0) {
+      await db
+        .deleteFrom('backlinks')
+        .where('sourcePageId', '=', pageId)
+        .where('targetPageId', 'in', backlinksToRemove)
+        .execute();
+
+      logger.debug(
+        `Removed ${backlinksToRemove.length} outdated backlinks from ${pageId}.`,
+      );
+    }
+  });
+}