feat: Typesense search driver (EE) (#1664)

* feat: typesense driver (EE) - WIP

* feat: typesense driver (EE) - WIP

* feat: typesense

* sync

* fix

apps/server/src/database/database.module.ts

@@ -25,6 +25,7 @@ import { MigrationService } from '@docmost/db/services/migration.service';
 import { UserTokenRepo } from './repos/user-token/user-token.repo';
 import { BacklinkRepo } from '@docmost/db/repos/backlink/backlink.repo';
 import { ShareRepo } from '@docmost/db/repos/share/share.repo';
+import { PageListener } from '@docmost/db/listeners/page.listener';
 
 // https://github.com/brianc/node-postgres/issues/811
 types.setTypeParser(types.builtins.INT8, (val) => Number(val));
@@ -75,7 +76,8 @@ types.setTypeParser(types.builtins.INT8, (val) => Number(val));
     AttachmentRepo,
     UserTokenRepo,
     BacklinkRepo,
-    ShareRepo
+    ShareRepo,
+    PageListener,
   ],
   exports: [
     WorkspaceRepo,
@@ -90,7 +92,7 @@ types.setTypeParser(types.builtins.INT8, (val) => Number(val));
     AttachmentRepo,
     UserTokenRepo,
     BacklinkRepo,
-    ShareRepo
+    ShareRepo,
   ],
 })
 export class DatabaseModule

apps/server/src/database/listeners/page.listener.ts

@@ -0,0 +1,49 @@
+import { Injectable, Logger } from '@nestjs/common';
+import { OnEvent } from '@nestjs/event-emitter';
+import { EventName } from '../../common/events/event.contants';
+import { InjectQueue } from '@nestjs/bullmq';
+import { QueueJob, QueueName } from '../../integrations/queue/constants';
+import { Queue } from 'bullmq';
+
+export class PageEvent {
+  pageIds: string[];
+}
+
+@Injectable()
+export class PageListener {
+  private readonly logger = new Logger(PageListener.name);
+
+  constructor(
+    @InjectQueue(QueueName.SEARCH_QUEUE) private searchQueue: Queue,
+  ) {}
+
+  @OnEvent(EventName.PAGE_CREATED)
+  async handlePageCreated(event: PageEvent) {
+    const { pageIds } = event;
+    await this.searchQueue.add(QueueJob.PAGE_CREATED, { pageIds });
+  }
+
+  @OnEvent(EventName.PAGE_UPDATED)
+  async handlePageUpdated(event: PageEvent) {
+    const { pageIds } = event;
+    await this.searchQueue.add(QueueJob.PAGE_UPDATED, { pageIds });
+  }
+
+  @OnEvent(EventName.PAGE_DELETED)
+  async handlePageDeleted(event: PageEvent) {
+    const { pageIds } = event;
+    await this.searchQueue.add(QueueJob.PAGE_DELETED, { pageIds });
+  }
+
+  @OnEvent(EventName.PAGE_SOFT_DELETED)
+  async handlePageSoftDeleted(event: PageEvent) {
+    const { pageIds } = event;
+    await this.searchQueue.add(QueueJob.PAGE_SOFT_DELETED, { pageIds });
+  }
+
+  @OnEvent(EventName.PAGE_RESTORED)
+  async handlePageRestored(event: PageEvent) {
+    const { pageIds } = event;
+    await this.searchQueue.add(QueueJob.PAGE_RESTORED, { pageIds });
+  }
+}

apps/server/src/database/repos/page/page.repo.ts

@@ -14,32 +14,17 @@ import { ExpressionBuilder, sql } from 'kysely';
 import { DB } from '@docmost/db/types/db';
 import { jsonArrayFrom, jsonObjectFrom } from 'kysely/helpers/postgres';
 import { SpaceMemberRepo } from '@docmost/db/repos/space/space-member.repo';
+import { EventEmitter2 } from '@nestjs/event-emitter';
+import { EventName } from '../../../common/events/event.contants';
 
 @Injectable()
 export class PageRepo {
   constructor(
     @InjectKysely() private readonly db: KyselyDB,
     private spaceMemberRepo: SpaceMemberRepo,
+    private eventEmitter: EventEmitter2,
   ) {}
 
-  withHasChildren(eb: ExpressionBuilder<DB, 'pages'>) {
-    return eb
-      .selectFrom('pages as child')
-      .select((eb) =>
-        eb
-          .case()
-          .when(eb.fn.countAll(), '>', 0)
-          .then(true)
-          .else(false)
-          .end()
-          .as('count'),
-      )
-      .whereRef('child.parentPageId', '=', 'pages.id')
-      .where('child.deletedAt', 'is', null)
-      .limit(1)
-      .as('hasChildren');
-  }
-
   private baseFields: Array<keyof Page> = [
     'id',
     'slugId',
@@ -63,6 +48,7 @@ export class PageRepo {
     pageId: string,
     opts?: {
       includeContent?: boolean;
+      includeTextContent?: boolean;
       includeYdoc?: boolean;
       includeSpace?: boolean;
       includeCreator?: boolean;
@@ -80,6 +66,7 @@ export class PageRepo {
       .select(this.baseFields)
       .$if(opts?.includeContent, (qb) => qb.select('content'))
       .$if(opts?.includeYdoc, (qb) => qb.select('ydoc'))
+      .$if(opts?.includeTextContent, (qb) => qb.select('textContent'))
       .$if(opts?.includeHasChildren, (qb) =>
         qb.select((eb) => this.withHasChildren(eb)),
       );
@@ -126,7 +113,7 @@ export class PageRepo {
     pageIds: string[],
     trx?: KyselyTransaction,
   ) {
-    return dbOrTx(this.db, trx)
+    const result = await dbOrTx(this.db, trx)
       .updateTable('pages')
       .set({ ...updatePageData, updatedAt: new Date() })
       .where(
@@ -135,6 +122,12 @@ export class PageRepo {
         pageIds,
       )
       .executeTakeFirst();
+
+    this.eventEmitter.emit(EventName.PAGE_UPDATED, {
+      pageIds: pageIds,
+    });
+
+    return result;
   }
 
   async insertPage(
@@ -142,11 +135,17 @@ export class PageRepo {
     trx?: KyselyTransaction,
   ): Promise<Page> {
     const db = dbOrTx(this.db, trx);
-    return db
+    const result = await db
       .insertInto('pages')
       .values(insertablePage)
       .returning(this.baseFields)
       .executeTakeFirst();
+
+    this.eventEmitter.emit(EventName.PAGE_CREATED, {
+      pageIds: [result.id],
+    });
+
+    return result;
   }
 
   async deletePage(pageId: string): Promise<void> {
@@ -196,6 +195,9 @@ export class PageRepo {
 
         await trx.deleteFrom('shares').where('pageId', 'in', pageIds).execute();
       });
+      this.eventEmitter.emit(EventName.PAGE_SOFT_DELETED, {
+        pageIds: pageIds,
+      });
     }
   }
 
@@ -259,6 +261,9 @@ export class PageRepo {
         .where('id', '=', pageId)
         .execute();
     }
+    this.eventEmitter.emit(EventName.PAGE_RESTORED, {
+      pageIds: pageIds,
+    });
   }
 
   async getRecentPagesInSpace(spaceId: string, pagination: PaginationOptions) {
@@ -379,6 +384,24 @@ export class PageRepo {
     ).as('contributors');
   }
 
+  withHasChildren(eb: ExpressionBuilder<DB, 'pages'>) {
+    return eb
+      .selectFrom('pages as child')
+      .select((eb) =>
+        eb
+          .case()
+          .when(eb.fn.countAll(), '>', 0)
+          .then(true)
+          .else(false)
+          .end()
+          .as('count'),
+      )
+      .whereRef('child.parentPageId', '=', 'pages.id')
+      .where('child.deletedAt', 'is', null)
+      .limit(1)
+      .as('hasChildren');
+  }
+
   async getPageAndDescendants(
     parentPageId: string,
     opts: { includeContent: boolean },