From 346c0df80099fc37236aa740e55be3d550b1948b Mon Sep 17 00:00:00 2001 From: fengmk2 Date: Wed, 21 May 2025 13:19:02 +0000 Subject: [PATCH] chore(server): support disable indexer plugin (#12408) close CLOUD-220 ## Summary by CodeRabbit - **New Features** - Introduced a new service to handle indexing-related events and scheduled tasks, improving the management of document and workspace indexing. - Added support for configuring the indexer feature via the AFFINE_INDEXER_ENABLED environment variable. - **Bug Fixes** - Ensured that indexing and deletion jobs are only enqueued when the indexer feature is enabled. - **Tests** - Added comprehensive tests for the new indexing event service, covering various configuration scenarios. - Removed obsolete test related to auto-indexing scheduling. - **Chores** - Updated configuration descriptions and mappings to improve clarity and environment variable support. --- .docker/selfhost/schema.json | 2 +- .../server/src/core/doc/adapters/workspace.ts | 6 +- packages/backend/server/src/core/doc/event.ts | 36 +------- packages/backend/server/src/models/doc.ts | 27 +++++- .../plugins/indexer/__tests__/event.spec.ts | 77 +++++++++++++++++ .../src/plugins/indexer/__tests__/job.spec.ts | 7 -- .../server/src/plugins/indexer/config.ts | 1 + .../server/src/plugins/indexer/event.ts | 86 +++++++++++++++++++ .../server/src/plugins/indexer/index.ts | 2 + .../backend/server/src/plugins/indexer/job.ts | 17 ---- packages/frontend/admin/src/config.json | 3 +- 11 files changed, 197 insertions(+), 67 deletions(-) create mode 100644 packages/backend/server/src/plugins/indexer/__tests__/event.spec.ts create mode 100644 packages/backend/server/src/plugins/indexer/event.ts diff --git a/.docker/selfhost/schema.json b/.docker/selfhost/schema.json index 1916587d3..be6e33f8e 100644 --- a/.docker/selfhost/schema.json +++ b/.docker/selfhost/schema.json @@ -812,7 +812,7 @@ "properties": { "enabled": { "type": "boolean", - "description": "Enable indexer plugin\n@default true", + "description": "Enable indexer plugin\n@default true\n@environment `AFFINE_INDEXER_ENABLED`", "default": true }, "provider.type": { diff --git a/packages/backend/server/src/core/doc/adapters/workspace.ts b/packages/backend/server/src/core/doc/adapters/workspace.ts index 08b6552c9..77530ff68 100644 --- a/packages/backend/server/src/core/doc/adapters/workspace.ts +++ b/packages/backend/server/src/core/doc/adapters/workspace.ts @@ -32,13 +32,9 @@ declare global { docId: string; blob: Buffer; }; - 'doc.created': { - workspaceId: string; - docId: string; - editor?: string; - }; } } + @Injectable() export class PgWorkspaceDocStorageAdapter extends DocStorageAdapter { protected override readonly logger = new Logger( diff --git a/packages/backend/server/src/core/doc/event.ts b/packages/backend/server/src/core/doc/event.ts index 5c780e943..19ba3827b 100644 --- a/packages/backend/server/src/core/doc/event.ts +++ b/packages/backend/server/src/core/doc/event.ts @@ -1,6 +1,6 @@ import { Injectable } from '@nestjs/common'; -import { JobQueue, OnEvent } from '../../base'; +import { OnEvent } from '../../base'; import { Models } from '../../models'; import { PgWorkspaceDocStorageAdapter } from './adapters/workspace'; import { DocReader } from './reader'; @@ -10,8 +10,7 @@ export class DocEventsListener { constructor( private readonly docReader: DocReader, private readonly models: Models, - private readonly workspace: PgWorkspaceDocStorageAdapter, - private readonly queue: JobQueue + private readonly workspace: PgWorkspaceDocStorageAdapter ) {} @OnEvent('doc.snapshot.updated') @@ -29,17 +28,6 @@ export class DocEventsListener { return; } await this.models.doc.upsertMeta(workspaceId, docId, content); - await this.queue.add( - 'indexer.indexDoc', - { - workspaceId, - docId, - }, - { - jobId: `indexDoc/${workspaceId}/${docId}`, - priority: 100, - } - ); } else { // update workspace content to database const content = this.docReader.parseWorkspaceContent(blob); @@ -47,16 +35,6 @@ export class DocEventsListener { return; } await this.models.workspace.update(workspaceId, content); - await this.queue.add( - 'indexer.indexWorkspace', - { - workspaceId, - }, - { - jobId: `indexWorkspace/${workspaceId}`, - priority: 100, - } - ); } } @@ -64,16 +42,6 @@ export class DocEventsListener { async clearUserWorkspaces(payload: Events['user.deleted']) { for (const workspace of payload.ownedWorkspaces) { await this.workspace.deleteSpace(workspace); - await this.queue.add( - 'indexer.deleteWorkspace', - { - workspaceId: workspace, - }, - { - jobId: `deleteWorkspace/${workspace}`, - priority: 0, - } - ); } } } diff --git a/packages/backend/server/src/models/doc.ts b/packages/backend/server/src/models/doc.ts index 8edc30249..0f56fc3fd 100644 --- a/packages/backend/server/src/models/doc.ts +++ b/packages/backend/server/src/models/doc.ts @@ -3,11 +3,25 @@ import { Transactional } from '@nestjs-cls/transactional'; import type { Update } from '@prisma/client'; import { Prisma } from '@prisma/client'; -import { PaginationInput } from '../base'; +import { EventBus, PaginationInput } from '../base'; import { DocIsNotPublic } from '../base/error'; import { BaseModel } from './base'; import { Doc, DocRole, PublicDocMode, publicUserSelect } from './common'; +declare global { + interface Events { + 'doc.created': { + workspaceId: string; + docId: string; + editor?: string; + }; + 'doc.updated': { + workspaceId: string; + docId: string; + }; + } +} + export type DocMetaUpsertInput = Omit< Prisma.WorkspaceDocUncheckedCreateInput, 'workspaceId' | 'docId' @@ -24,6 +38,10 @@ export type DocMetaUpsertInput = Omit< */ @Injectable() export class DocModel extends BaseModel { + constructor(private readonly event: EventBus) { + super(); + } + // #region Update private updateToDocRecord(row: Update): Doc { @@ -338,7 +356,7 @@ export class DocModel extends BaseModel { docId: string, data?: DocMetaUpsertInput ) { - return await this.db.workspaceDoc.upsert({ + const doc = await this.db.workspaceDoc.upsert({ where: { workspaceId_docId: { workspaceId, @@ -354,6 +372,11 @@ export class DocModel extends BaseModel { docId, }, }); + this.event.emit('doc.updated', { + workspaceId, + docId, + }); + return doc; } /** diff --git a/packages/backend/server/src/plugins/indexer/__tests__/event.spec.ts b/packages/backend/server/src/plugins/indexer/__tests__/event.spec.ts new file mode 100644 index 000000000..7e40da45c --- /dev/null +++ b/packages/backend/server/src/plugins/indexer/__tests__/event.spec.ts @@ -0,0 +1,77 @@ +import test from 'ava'; +import Sinon from 'sinon'; + +import { createModule } from '../../../__tests__/create-module'; +import { Config } from '../../../base'; +import { IndexerModule } from '..'; +import { IndexerEvent } from '../event'; + +const module = await createModule({ + imports: [IndexerModule], +}); +const indexerEvent = module.get(IndexerEvent); +const config = module.get(Config); + +test.after.always(async () => { + await module.close(); +}); + +test.afterEach.always(() => { + Sinon.restore(); +}); + +test('should not index workspace if indexer is disabled', async t => { + Sinon.stub(config.indexer, 'enabled').value(false); + const count = module.queue.count('indexer.indexWorkspace'); + + // @ts-expect-error ignore missing fields + await indexerEvent.indexWorkspace({ id: 'test-workspace' }); + + t.is(module.queue.count('indexer.indexWorkspace'), count); +}); + +test('should index workspace if indexer is enabled', async t => { + // @ts-expect-error ignore missing fields + await indexerEvent.indexWorkspace({ id: 'test-workspace' }); + + const { payload } = await module.queue.waitFor('indexer.indexWorkspace'); + t.is(payload.workspaceId, 'test-workspace'); +}); + +test('should not delete workspace if indexer is disabled', async t => { + Sinon.stub(config.indexer, 'enabled').value(false); + const count = module.queue.count('indexer.deleteWorkspace'); + + // @ts-expect-error ignore missing fields + await indexerEvent.deleteUserWorkspaces({ + ownedWorkspaces: ['test-workspace'], + }); + + t.is(module.queue.count('indexer.deleteWorkspace'), count); +}); + +test('should delete workspace if indexer is enabled', async t => { + // @ts-expect-error ignore missing fields + await indexerEvent.deleteUserWorkspaces({ + ownedWorkspaces: ['test-workspace'], + }); + + const { payload } = await module.queue.waitFor('indexer.deleteWorkspace'); + t.is(payload.workspaceId, 'test-workspace'); +}); + +test('should not schedule auto index workspaces if indexer is disabled', async t => { + Sinon.stub(config.indexer, 'enabled').value(false); + const count = module.queue.count('indexer.autoIndexWorkspaces'); + + await indexerEvent.autoIndexWorkspaces(); + + t.is(module.queue.count('indexer.autoIndexWorkspaces'), count); +}); + +test('should schedule auto index workspaces', async t => { + await indexerEvent.autoIndexWorkspaces(); + + const { payload } = await module.queue.waitFor('indexer.autoIndexWorkspaces'); + t.is(payload.lastIndexedWorkspaceSid, undefined); +}); diff --git a/packages/backend/server/src/plugins/indexer/__tests__/job.spec.ts b/packages/backend/server/src/plugins/indexer/__tests__/job.spec.ts index e2d9eb064..1f8e17e63 100644 --- a/packages/backend/server/src/plugins/indexer/__tests__/job.spec.ts +++ b/packages/backend/server/src/plugins/indexer/__tests__/job.spec.ts @@ -175,10 +175,3 @@ test('should not index workspace if snapshot not exists', async t => { t.is(module.queue.count('indexer.indexWorkspace'), count); }); - -test('should schedule auto index workspaces', async t => { - await indexerJob.scheduleAutoIndexWorkspaces(); - - const { payload } = await module.queue.waitFor('indexer.autoIndexWorkspaces'); - t.is(payload.lastIndexedWorkspaceSid, undefined); -}); diff --git a/packages/backend/server/src/plugins/indexer/config.ts b/packages/backend/server/src/plugins/indexer/config.ts index b04855bb4..ff06e84c3 100644 --- a/packages/backend/server/src/plugins/indexer/config.ts +++ b/packages/backend/server/src/plugins/indexer/config.ts @@ -31,6 +31,7 @@ defineModuleConfig('indexer', { enabled: { desc: 'Enable indexer plugin', default: true, + env: ['AFFINE_INDEXER_ENABLED', 'boolean'], }, 'provider.type': { desc: 'Indexer search service provider name', diff --git a/packages/backend/server/src/plugins/indexer/event.ts b/packages/backend/server/src/plugins/indexer/event.ts new file mode 100644 index 000000000..82f9ae1de --- /dev/null +++ b/packages/backend/server/src/plugins/indexer/event.ts @@ -0,0 +1,86 @@ +import { Injectable } from '@nestjs/common'; +import { Cron, CronExpression } from '@nestjs/schedule'; + +import { Config, JobQueue, OnEvent } from '../../base'; + +@Injectable() +export class IndexerEvent { + constructor( + private readonly queue: JobQueue, + private readonly config: Config + ) {} + + @OnEvent('doc.updated') + async indexDoc({ workspaceId, docId }: Events['doc.updated']) { + if (!this.config.indexer.enabled) { + return; + } + + await this.queue.add( + 'indexer.indexDoc', + { + workspaceId, + docId, + }, + { + jobId: `indexDoc/${workspaceId}/${docId}`, + priority: 100, + } + ); + } + + @OnEvent('workspace.updated') + async indexWorkspace({ id }: Events['workspace.updated']) { + if (!this.config.indexer.enabled) { + return; + } + + await this.queue.add( + 'indexer.indexWorkspace', + { + workspaceId: id, + }, + { + jobId: `indexWorkspace/${id}`, + priority: 100, + } + ); + } + + @OnEvent('user.deleted') + async deleteUserWorkspaces(payload: Events['user.deleted']) { + if (!this.config.indexer.enabled) { + return; + } + + for (const workspace of payload.ownedWorkspaces) { + await this.queue.add( + 'indexer.deleteWorkspace', + { + workspaceId: workspace, + }, + { + jobId: `deleteWorkspace/${workspace}`, + priority: 0, + } + ); + } + } + + @Cron(CronExpression.EVERY_30_SECONDS) + async autoIndexWorkspaces() { + if (!this.config.indexer.enabled) { + return; + } + + await this.queue.add( + 'indexer.autoIndexWorkspaces', + {}, + { + // make sure only one job is running at a time + delay: 30 * 1000, + jobId: 'autoIndexWorkspaces', + } + ); + } +} diff --git a/packages/backend/server/src/plugins/indexer/index.ts b/packages/backend/server/src/plugins/indexer/index.ts index 6235e9ace..c6e4ea15d 100644 --- a/packages/backend/server/src/plugins/indexer/index.ts +++ b/packages/backend/server/src/plugins/indexer/index.ts @@ -4,6 +4,7 @@ import { Module } from '@nestjs/common'; import { ServerConfigModule } from '../../core/config'; import { PermissionModule } from '../../core/permission'; +import { IndexerEvent } from './event'; import { SearchProviderFactory } from './factory'; import { IndexerJob } from './job'; import { SearchProviders } from './providers'; @@ -16,6 +17,7 @@ import { IndexerService } from './service'; IndexerResolver, IndexerService, IndexerJob, + IndexerEvent, SearchProviderFactory, ...SearchProviders, ], diff --git a/packages/backend/server/src/plugins/indexer/job.ts b/packages/backend/server/src/plugins/indexer/job.ts index 7c4ae6d0e..1f06c775f 100644 --- a/packages/backend/server/src/plugins/indexer/job.ts +++ b/packages/backend/server/src/plugins/indexer/job.ts @@ -1,5 +1,4 @@ import { Injectable, Logger } from '@nestjs/common'; -import { Cron, CronExpression } from '@nestjs/schedule'; import { Config, JOB_SIGNAL, JobQueue, OnJob } from '../../base'; import { readAllDocIdsFromWorkspaceSnapshot } from '../../core/utils/blocksuite'; @@ -184,20 +183,4 @@ export class IndexerJob { payload.lastIndexedWorkspaceSid = nextSid; return JOB_SIGNAL.Repeat; } - - @Cron(CronExpression.EVERY_30_SECONDS) - async scheduleAutoIndexWorkspaces() { - if (!this.config.indexer.enabled) { - return; - } - await this.queue.add( - 'indexer.autoIndexWorkspaces', - {}, - { - // make sure only one job is running at a time - delay: 30 * 1000, - jobId: 'autoIndexWorkspaces', - } - ); - } } diff --git a/packages/frontend/admin/src/config.json b/packages/frontend/admin/src/config.json index 944c86d9d..dfa2ecf5d 100644 --- a/packages/frontend/admin/src/config.json +++ b/packages/frontend/admin/src/config.json @@ -267,7 +267,8 @@ "indexer": { "enabled": { "type": "Boolean", - "desc": "Enable indexer plugin" + "desc": "Enable indexer plugin", + "env": "AFFINE_INDEXER_ENABLED" }, "provider.type": { "type": "String",