fix(queue): убрать мёртвую очередь {search-queue} (#379) #380
@@ -4,7 +4,6 @@ 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';
|
||||
|
||||
/**
|
||||
* Thin snapshot of a page node carried inside domain events so the WebSocket
|
||||
@@ -112,48 +111,24 @@ 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, 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, 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, workspaceId } = event;
|
||||
|
||||
if (this.isTypesense()) {
|
||||
await this.searchQueue.add(QueueJob.PAGE_SOFT_DELETED, { pageIds });
|
||||
}
|
||||
|
||||
await this.aiQueue.add(QueueJob.PAGE_SOFT_DELETED, {
|
||||
pageIds,
|
||||
workspaceId,
|
||||
@@ -163,14 +138,6 @@ export class PageListener {
|
||||
@OnEvent(EventName.PAGE_RESTORED)
|
||||
async handlePageRestored(event: PageEvent) {
|
||||
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';
|
||||
}
|
||||
}
|
||||
|
||||
@@ -4,7 +4,6 @@ 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;
|
||||
@@ -15,22 +14,12 @@ 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';
|
||||
}
|
||||
}
|
||||
|
||||
@@ -4,7 +4,6 @@ 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;
|
||||
@@ -15,22 +14,12 @@ 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';
|
||||
}
|
||||
}
|
||||
|
||||
@@ -46,7 +46,6 @@ export class MetricsBullService implements OnModuleInit, OnModuleDestroy {
|
||||
@InjectQueue(QueueName.GENERAL_QUEUE) generalQueue: Queue,
|
||||
@InjectQueue(QueueName.BILLING_QUEUE) billingQueue: Queue,
|
||||
@InjectQueue(QueueName.FILE_TASK_QUEUE) fileTaskQueue: Queue,
|
||||
@InjectQueue(QueueName.SEARCH_QUEUE) searchQueue: Queue,
|
||||
@InjectQueue(QueueName.AI_QUEUE) aiQueue: Queue,
|
||||
@InjectQueue(QueueName.HISTORY_QUEUE) historyQueue: Queue,
|
||||
@InjectQueue(QueueName.NOTIFICATION_QUEUE) notificationQueue: Queue,
|
||||
@@ -58,7 +57,6 @@ export class MetricsBullService implements OnModuleInit, OnModuleDestroy {
|
||||
{ label: 'general', queue: generalQueue },
|
||||
{ label: 'billing', queue: billingQueue },
|
||||
{ label: 'file-task', queue: fileTaskQueue },
|
||||
{ label: 'search', queue: searchQueue },
|
||||
{ label: 'ai', queue: aiQueue },
|
||||
{ label: 'history', queue: historyQueue },
|
||||
{ label: 'notification', queue: notificationQueue },
|
||||
|
||||
@@ -4,7 +4,6 @@ export enum QueueName {
|
||||
GENERAL_QUEUE = '{general-queue}',
|
||||
BILLING_QUEUE = '{billing-queue}',
|
||||
FILE_TASK_QUEUE = '{file-task-queue}',
|
||||
SEARCH_QUEUE = '{search-queue}',
|
||||
AI_QUEUE = '{ai-queue}',
|
||||
HISTORY_QUEUE = '{history-queue}',
|
||||
NOTIFICATION_QUEUE = '{notification-queue}',
|
||||
@@ -32,12 +31,6 @@ export enum QueueJob {
|
||||
IMPORT_TASK = 'import-task',
|
||||
EXPORT_TASK = 'export-task',
|
||||
|
||||
SEARCH_INDEX_PAGE = 'search-index-page',
|
||||
SEARCH_INDEX_PAGES = 'search-index-pages',
|
||||
SEARCH_INDEX_COMMENT = 'search-index-comment',
|
||||
SEARCH_INDEX_COMMENTS = 'search-index-comments',
|
||||
SEARCH_INDEX_ATTACHMENT = 'search-index-attachment',
|
||||
SEARCH_INDEX_ATTACHMENTS = 'search-index-attachments',
|
||||
SEARCH_REMOVE_PAGE = 'search-remove-page',
|
||||
SEARCH_REMOVE_ASSET = 'search-remove-attachment',
|
||||
SEARCH_REMOVE_FACE = 'search-remove-comment',
|
||||
|
||||
@@ -57,14 +57,6 @@ import { GeneralQueueProcessor } from './processors/general-queue.processor';
|
||||
attempts: 1,
|
||||
},
|
||||
}),
|
||||
BullModule.registerQueue({
|
||||
name: QueueName.SEARCH_QUEUE,
|
||||
defaultJobOptions: {
|
||||
removeOnComplete: true,
|
||||
removeOnFail: true,
|
||||
attempts: 2,
|
||||
},
|
||||
}),
|
||||
BullModule.registerQueue({
|
||||
name: QueueName.AI_QUEUE,
|
||||
defaultJobOptions: {
|
||||
|
||||
Reference in New Issue
Block a user