feat(EE): AI vector search (#1691)

* WIP

* AI module - init

* WIP

* sync

* WIP

* refactor naming

* new columns

* sync

* sync

* fix search bug

* stream response

* WIP

* feat embeddings sync

* refine

* Add workspaceId to page events

* refine

* WIP

* add translation string

* sync

* reset ai answer on query change

* hide AI search in cloud

* capture streaming error

* sync

apps/server/src/database/helpers/helpers.ts

@@ -0,0 +1,22 @@
+import { sql } from 'kysely';
+import { KyselyDB } from '@docmost/db/types/kysely.types';
+
+export async function isPageEmbeddingsTableExists(db: KyselyDB) {
+  return tableExists({ db, tableName: 'page_embeddings' });
+}
+
+export async function tableExists(opts: {
+  db: KyselyDB;
+  tableName: string;
+}): Promise<boolean> {
+  const { db, tableName } = opts;
+  const result = await sql<{ exists: boolean }>`
+      SELECT EXISTS (
+        SELECT 1 FROM information_schema.tables
+        WHERE table_schema = COALESCE(current_schema(), 'public')
+        AND table_name = ${tableName}
+      ) as exists
+    `.execute(db);
+
+  return result.rows[0]?.exists ?? false;
+}

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

@@ -4,9 +4,11 @@ import { EventName } from '../../common/events/event.contants';
 import { InjectQueue } from '@nestjs/bullmq';
 import { QueueJob, QueueName } from '../../integrations/queue/constants';
 import { Queue } from 'bullmq';
+import { EnvironmentService } from '../../integrations/environment/environment.service';
 
 export class PageEvent {
   pageIds: string[];
+  workspaceId: string;
 }
 
 @Injectable()
@@ -14,36 +16,65 @@ export class PageListener {
   private readonly logger = new Logger(PageListener.name);
 
   constructor(
+    private readonly environmentService: EnvironmentService,
     @InjectQueue(QueueName.SEARCH_QUEUE) private searchQueue: Queue,
+    @InjectQueue(QueueName.AI_QUEUE) private aiQueue: Queue,
   ) {}
 
   @OnEvent(EventName.PAGE_CREATED)
   async handlePageCreated(event: PageEvent) {
-    const { pageIds } = event;
-    await this.searchQueue.add(QueueJob.PAGE_CREATED, { pageIds });
+    const { pageIds, workspaceId } = event;
+    if (this.isTypesense()) {
+      await this.searchQueue.add(QueueJob.PAGE_CREATED, {
+        pageIds,
+      });
+    }
+
+    await this.aiQueue.add(QueueJob.PAGE_CREATED, { pageIds, workspaceId });
   }
 
   @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 });
+    const { pageIds, workspaceId } = event;
+    if (this.isTypesense()) {
+      await this.searchQueue.add(QueueJob.PAGE_DELETED, { pageIds });
+    }
+
+    await this.aiQueue.add(QueueJob.PAGE_DELETED, { pageIds, workspaceId });
   }
 
   @OnEvent(EventName.PAGE_SOFT_DELETED)
   async handlePageSoftDeleted(event: PageEvent) {
-    const { pageIds } = event;
-    await this.searchQueue.add(QueueJob.PAGE_SOFT_DELETED, { pageIds });
+    const { pageIds, workspaceId } = event;
+
+    if (this.isTypesense()) {
+      await this.searchQueue.add(QueueJob.PAGE_SOFT_DELETED, { pageIds });
+    }
+
+    await this.aiQueue.add(QueueJob.PAGE_SOFT_DELETED, {
+      pageIds,
+      workspaceId,
+    });
   }
 
   @OnEvent(EventName.PAGE_RESTORED)
   async handlePageRestored(event: PageEvent) {
-    const { pageIds } = event;
-    await this.searchQueue.add(QueueJob.PAGE_RESTORED, { pageIds });
+    const { pageIds, workspaceId } = event;
+    if (this.isTypesense()) {
+      await this.searchQueue.add(QueueJob.PAGE_RESTORED, { pageIds });
+    }
+
+    await this.aiQueue.add(QueueJob.PAGE_RESTORED, { pageIds, workspaceId });
+  }
+
+  isTypesense(): boolean {
+    return this.environmentService.getSearchDriver() === 'typesense';
   }
 }

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

@@ -0,0 +1,36 @@
+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';
+import { EnvironmentService } from '../../integrations/environment/environment.service';
+
+export class SpaceEvent {
+  spaceId: string;
+}
+
+@Injectable()
+export class SpaceListener {
+  private readonly logger = new Logger(SpaceListener.name);
+
+  constructor(
+    private readonly environmentService: EnvironmentService,
+    @InjectQueue(QueueName.SEARCH_QUEUE) private searchQueue: Queue,
+    @InjectQueue(QueueName.AI_QUEUE) private aiQueue: Queue,
+  ) {}
+
+  @OnEvent(EventName.SPACE_DELETED)
+  async handleSpaceDeleted(event: SpaceEvent) {
+    const { spaceId } = event;
+    if (this.isTypesense()) {
+      await this.searchQueue.add(QueueJob.SPACE_DELETED, { spaceId });
+    }
+
+    await this.aiQueue.add(QueueJob.SPACE_DELETED, { spaceId });
+  }
+
+  isTypesense(): boolean {
+    return this.environmentService.getSearchDriver() === 'typesense';
+  }
+}

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

@@ -0,0 +1,36 @@
+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';
+import { EnvironmentService } from '../../integrations/environment/environment.service';
+
+export class WorkspaceEvent {
+  workspaceId: string;
+}
+
+@Injectable()
+export class WorkspaceListener {
+  private readonly logger = new Logger(WorkspaceListener.name);
+
+  constructor(
+    private readonly environmentService: EnvironmentService,
+    @InjectQueue(QueueName.SEARCH_QUEUE) private searchQueue: Queue,
+    @InjectQueue(QueueName.AI_QUEUE) private aiQueue: Queue,
+  ) {}
+
+  @OnEvent(EventName.WORKSPACE_DELETED)
+  async handlePageDeleted(event: WorkspaceEvent) {
+    const { workspaceId } = event;
+    if (this.isTypesense()) {
+      await this.searchQueue.add(QueueJob.WORKSPACE_DELETED, { workspaceId });
+    }
+
+    await this.aiQueue.add(QueueJob.WORKSPACE_DELETED, { workspaceId });
+  }
+
+  isTypesense(): boolean {
+    return this.environmentService.getSearchDriver() === 'typesense';
+  }
+}

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

@@ -125,6 +125,7 @@ export class PageRepo {
 
     this.eventEmitter.emit(EventName.PAGE_UPDATED, {
       pageIds: pageIds,
+      workspaceId: updatePageData.workspaceId,
     });
 
     return result;
@@ -143,6 +144,7 @@ export class PageRepo {
 
     this.eventEmitter.emit(EventName.PAGE_CREATED, {
       pageIds: [result.id],
+      workspaceId: result.workspaceId,
     });
 
     return result;
@@ -160,7 +162,11 @@ export class PageRepo {
     await query.execute();
   }
 
-  async removePage(pageId: string, deletedById: string): Promise<void> {
+  async removePage(
+    pageId: string,
+    deletedById: string,
+    workspaceId: string,
+  ): Promise<void> {
     const currentDate = new Date();
 
     const descendants = await this.db
@@ -195,13 +201,15 @@ export class PageRepo {
 
         await trx.deleteFrom('shares').where('pageId', 'in', pageIds).execute();
       });
+
       this.eventEmitter.emit(EventName.PAGE_SOFT_DELETED, {
         pageIds: pageIds,
+        workspaceId,
       });
     }
   }
 
-  async restorePage(pageId: string): Promise<void> {
+  async restorePage(pageId: string, workspaceId: string): Promise<void> {
     // First, check if the page being restored has a deleted parent
     const pageToRestore = await this.db
       .selectFrom('pages')
@@ -263,6 +271,7 @@ export class PageRepo {
     }
     this.eventEmitter.emit(EventName.PAGE_RESTORED, {
       pageIds: pageIds,
+      workspaceId: workspaceId,
     });
   }
 

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

@@ -12,10 +12,15 @@ import { PaginationOptions } from '../../pagination/pagination-options';
 import { executeWithPagination } from '@docmost/db/pagination/pagination';
 import { DB } from '@docmost/db/types/db';
 import { validate as isValidUUID } from 'uuid';
+import { EventEmitter2 } from '@nestjs/event-emitter';
+import { EventName } from '../../../common/events/event.contants';
 
 @Injectable()
 export class SpaceRepo {
-  constructor(@InjectKysely() private readonly db: KyselyDB) {}
+  constructor(
+    @InjectKysely() private readonly db: KyselyDB,
+    private eventEmitter: EventEmitter2,
+  ) {}
 
   async findById(
     spaceId: string,
@@ -110,7 +115,11 @@ export class SpaceRepo {
 
     if (pagination.query) {
       query = query.where((eb) =>
-        eb(sql`f_unaccent(name)`, 'ilike', sql`f_unaccent(${'%' + pagination.query + '%'})`).or(
+        eb(
+          sql`f_unaccent(name)`,
+          'ilike',
+          sql`f_unaccent(${'%' + pagination.query + '%'})`,
+        ).or(
           sql`f_unaccent(description)`,
           'ilike',
           sql`f_unaccent(${'%' + pagination.query + '%'})`,
@@ -155,5 +164,9 @@ export class SpaceRepo {
       .where('id', '=', spaceId)
       .where('workspaceId', '=', workspaceId)
       .execute();
+
+    this.eventEmitter.emit(EventName.SPACE_DELETED, {
+      spaceId,
+    });
   }
 }

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

@@ -175,4 +175,22 @@ export class WorkspaceRepo {
       .returning(this.baseFields)
       .executeTakeFirst();
   }
+
+  async updateAiSettings(
+    workspaceId: string,
+    prefKey: string,
+    prefValue: string | boolean,
+  ) {
+    return this.db
+      .updateTable('workspaces')
+      .set({
+        settings: sql`COALESCE(settings, '{}'::jsonb)
+                || jsonb_build_object('ai', COALESCE(settings->'ai', '{}'::jsonb) 
+                || jsonb_build_object('${sql.raw(prefKey)}', ${sql.lit(prefValue)}))`,
+        updatedAt: new Date(),
+      })
+      .where('id', '=', workspaceId)
+      .returning(this.baseFields)
+      .executeTakeFirst();
+  }
 }

apps/server/src/database/types/db.interface.ts

@@ -0,0 +1,47 @@
+import {
+  ApiKeys,
+  Attachments,
+  AuthAccounts,
+  AuthProviders,
+  Backlinks,
+  Billing,
+  Comments,
+  FileTasks,
+  Groups,
+  GroupUsers,
+  PageHistory,
+  Pages,
+  Shares,
+  SpaceMembers,
+  Spaces,
+  UserMfa,
+  Users,
+  UserTokens,
+  WorkspaceInvitations,
+  Workspaces,
+} from '@docmost/db/types/db';
+import { PageEmbeddings } from '@docmost/db/types/embeddings.types';
+
+export interface DbInterface {
+  attachments: Attachments;
+  authAccounts: AuthAccounts;
+  authProviders: AuthProviders;
+  backlinks: Backlinks;
+  billing: Billing;
+  comments: Comments;
+  fileTasks: FileTasks;
+  groups: Groups;
+  groupUsers: GroupUsers;
+  pageEmbeddings: PageEmbeddings;
+  pageHistory: PageHistory;
+  pages: Pages;
+  shares: Shares;
+  spaceMembers: SpaceMembers;
+  spaces: Spaces;
+  userMfa: UserMfa;
+  users: Users;
+  userTokens: UserTokens;
+  workspaceInvitations: WorkspaceInvitations;
+  workspaces: Workspaces;
+  apiKeys: ApiKeys;
+}

apps/server/src/database/types/embeddings.types.ts

@@ -0,0 +1,20 @@
+import { Json, Timestamp, Generated } from '@docmost/db/types/db';
+
+// embeddings type
+export interface PageEmbeddings {
+  id: Generated<string>;
+  pageId: string;
+  spaceId: string;
+  modelName: string;
+  modelDimensions: number;
+  workspaceId: string;
+  attachmentId: string;
+  embedding: number[];
+  chunkIndex: Generated<number>;
+  chunkStart: Generated<number>;
+  chunkLength: Generated<number>;
+  metadata: Generated<Json>;
+  createdAt: Generated<Timestamp>;
+  updatedAt: Generated<Timestamp>;
+  deletedAt: Timestamp | null;
+}

apps/server/src/database/types/entity.types.ts

@@ -21,6 +21,7 @@ import {
   UserMfa as _UserMFA,
   ApiKeys,
 } from './db';
+import { PageEmbeddings } from '@docmost/db/types/embeddings.types';
 
 // Workspace
 export type Workspace = Selectable<Workspaces>;
@@ -125,3 +126,8 @@ export type UpdatableUserMFA = Updateable<Omit<_UserMFA, 'id'>>;
 export type ApiKey = Selectable<ApiKeys>;
 export type InsertableApiKey = Insertable<ApiKeys>;
 export type UpdatableApiKey = Updateable<Omit<ApiKeys, 'id'>>;
+
+// Page Embedding
+export type PageEmbedding = Selectable<PageEmbeddings>;
+export type InsertablePageEmbedding = Insertable<PageEmbeddings>;
+export type UpdatablePageEmbedding = Updateable<Omit<PageEmbeddings, 'id'>>;

apps/server/src/database/types/kysely.types.ts

@@ -1,5 +1,5 @@
-import { DB } from './db';
 import { Kysely, Transaction } from 'kysely';
+import { DbInterface } from '@docmost/db/types/db.interface';
 
-export type KyselyDB = Kysely<DB>;
-export type KyselyTransaction = Transaction<DB>;
+export type KyselyDB = Kysely<DbInterface>;
+export type KyselyTransaction = Transaction<DbInterface>;