Merge pull request 'fix(queue): убрать мёртвую очередь {search-queue} (#379)' (#380) from fix/379-remove-dead-search-queue into develop

Reviewed-on: #380
This commit was merged in pull request #380.
This commit is contained in:
2026-07-05 23:44:54 +03:00
6 changed files with 0 additions and 72 deletions
@@ -4,7 +4,6 @@ import { EventName } from '../../common/events/event.contants';
import { InjectQueue } from '@nestjs/bullmq'; import { InjectQueue } from '@nestjs/bullmq';
import { QueueJob, QueueName } from '../../integrations/queue/constants'; import { QueueJob, QueueName } from '../../integrations/queue/constants';
import { Queue } from 'bullmq'; import { Queue } from 'bullmq';
import { EnvironmentService } from '../../integrations/environment/environment.service';
/** /**
* Thin snapshot of a page node carried inside domain events so the WebSocket * 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); private readonly logger = new Logger(PageListener.name);
constructor( constructor(
private readonly environmentService: EnvironmentService,
@InjectQueue(QueueName.SEARCH_QUEUE) private searchQueue: Queue,
@InjectQueue(QueueName.AI_QUEUE) private aiQueue: Queue, @InjectQueue(QueueName.AI_QUEUE) private aiQueue: Queue,
) {} ) {}
@OnEvent(EventName.PAGE_CREATED) @OnEvent(EventName.PAGE_CREATED)
async handlePageCreated(event: PageEvent) { async handlePageCreated(event: PageEvent) {
const { pageIds, workspaceId } = event; const { pageIds, workspaceId } = event;
if (this.isTypesense()) {
await this.searchQueue.add(QueueJob.PAGE_CREATED, {
pageIds,
});
}
await this.aiQueue.add(QueueJob.PAGE_CREATED, { pageIds, workspaceId }); 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) @OnEvent(EventName.PAGE_DELETED)
async handlePageDeleted(event: PageEvent) { async handlePageDeleted(event: PageEvent) {
const { pageIds, workspaceId } = event; const { pageIds, workspaceId } = event;
if (this.isTypesense()) {
await this.searchQueue.add(QueueJob.PAGE_DELETED, { pageIds });
}
await this.aiQueue.add(QueueJob.PAGE_DELETED, { pageIds, workspaceId }); await this.aiQueue.add(QueueJob.PAGE_DELETED, { pageIds, workspaceId });
} }
@OnEvent(EventName.PAGE_SOFT_DELETED) @OnEvent(EventName.PAGE_SOFT_DELETED)
async handlePageSoftDeleted(event: PageEvent) { async handlePageSoftDeleted(event: PageEvent) {
const { pageIds, workspaceId } = event; const { pageIds, workspaceId } = event;
if (this.isTypesense()) {
await this.searchQueue.add(QueueJob.PAGE_SOFT_DELETED, { pageIds });
}
await this.aiQueue.add(QueueJob.PAGE_SOFT_DELETED, { await this.aiQueue.add(QueueJob.PAGE_SOFT_DELETED, {
pageIds, pageIds,
workspaceId, workspaceId,
@@ -163,14 +138,6 @@ export class PageListener {
@OnEvent(EventName.PAGE_RESTORED) @OnEvent(EventName.PAGE_RESTORED)
async handlePageRestored(event: PageEvent) { async handlePageRestored(event: PageEvent) {
const { pageIds, workspaceId } = event; const { pageIds, workspaceId } = event;
if (this.isTypesense()) {
await this.searchQueue.add(QueueJob.PAGE_RESTORED, { pageIds });
}
await this.aiQueue.add(QueueJob.PAGE_RESTORED, { pageIds, workspaceId }); 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 { InjectQueue } from '@nestjs/bullmq';
import { QueueJob, QueueName } from '../../integrations/queue/constants'; import { QueueJob, QueueName } from '../../integrations/queue/constants';
import { Queue } from 'bullmq'; import { Queue } from 'bullmq';
import { EnvironmentService } from '../../integrations/environment/environment.service';
export class SpaceEvent { export class SpaceEvent {
spaceId: string; spaceId: string;
@@ -15,22 +14,12 @@ export class SpaceListener {
private readonly logger = new Logger(SpaceListener.name); private readonly logger = new Logger(SpaceListener.name);
constructor( constructor(
private readonly environmentService: EnvironmentService,
@InjectQueue(QueueName.SEARCH_QUEUE) private searchQueue: Queue,
@InjectQueue(QueueName.AI_QUEUE) private aiQueue: Queue, @InjectQueue(QueueName.AI_QUEUE) private aiQueue: Queue,
) {} ) {}
@OnEvent(EventName.SPACE_DELETED) @OnEvent(EventName.SPACE_DELETED)
async handleSpaceDeleted(event: SpaceEvent) { async handleSpaceDeleted(event: SpaceEvent) {
const { spaceId } = event; const { spaceId } = event;
if (this.isTypesense()) {
await this.searchQueue.add(QueueJob.SPACE_DELETED, { spaceId });
}
await this.aiQueue.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 { InjectQueue } from '@nestjs/bullmq';
import { QueueJob, QueueName } from '../../integrations/queue/constants'; import { QueueJob, QueueName } from '../../integrations/queue/constants';
import { Queue } from 'bullmq'; import { Queue } from 'bullmq';
import { EnvironmentService } from '../../integrations/environment/environment.service';
export class WorkspaceEvent { export class WorkspaceEvent {
workspaceId: string; workspaceId: string;
@@ -15,22 +14,12 @@ export class WorkspaceListener {
private readonly logger = new Logger(WorkspaceListener.name); private readonly logger = new Logger(WorkspaceListener.name);
constructor( constructor(
private readonly environmentService: EnvironmentService,
@InjectQueue(QueueName.SEARCH_QUEUE) private searchQueue: Queue,
@InjectQueue(QueueName.AI_QUEUE) private aiQueue: Queue, @InjectQueue(QueueName.AI_QUEUE) private aiQueue: Queue,
) {} ) {}
@OnEvent(EventName.WORKSPACE_DELETED) @OnEvent(EventName.WORKSPACE_DELETED)
async handlePageDeleted(event: WorkspaceEvent) { async handlePageDeleted(event: WorkspaceEvent) {
const { workspaceId } = event; const { workspaceId } = event;
if (this.isTypesense()) {
await this.searchQueue.add(QueueJob.WORKSPACE_DELETED, { workspaceId });
}
await this.aiQueue.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.GENERAL_QUEUE) generalQueue: Queue,
@InjectQueue(QueueName.BILLING_QUEUE) billingQueue: Queue, @InjectQueue(QueueName.BILLING_QUEUE) billingQueue: Queue,
@InjectQueue(QueueName.FILE_TASK_QUEUE) fileTaskQueue: Queue, @InjectQueue(QueueName.FILE_TASK_QUEUE) fileTaskQueue: Queue,
@InjectQueue(QueueName.SEARCH_QUEUE) searchQueue: Queue,
@InjectQueue(QueueName.AI_QUEUE) aiQueue: Queue, @InjectQueue(QueueName.AI_QUEUE) aiQueue: Queue,
@InjectQueue(QueueName.HISTORY_QUEUE) historyQueue: Queue, @InjectQueue(QueueName.HISTORY_QUEUE) historyQueue: Queue,
@InjectQueue(QueueName.NOTIFICATION_QUEUE) notificationQueue: Queue, @InjectQueue(QueueName.NOTIFICATION_QUEUE) notificationQueue: Queue,
@@ -58,7 +57,6 @@ export class MetricsBullService implements OnModuleInit, OnModuleDestroy {
{ label: 'general', queue: generalQueue }, { label: 'general', queue: generalQueue },
{ label: 'billing', queue: billingQueue }, { label: 'billing', queue: billingQueue },
{ label: 'file-task', queue: fileTaskQueue }, { label: 'file-task', queue: fileTaskQueue },
{ label: 'search', queue: searchQueue },
{ label: 'ai', queue: aiQueue }, { label: 'ai', queue: aiQueue },
{ label: 'history', queue: historyQueue }, { label: 'history', queue: historyQueue },
{ label: 'notification', queue: notificationQueue }, { label: 'notification', queue: notificationQueue },
@@ -4,7 +4,6 @@ export enum QueueName {
GENERAL_QUEUE = '{general-queue}', GENERAL_QUEUE = '{general-queue}',
BILLING_QUEUE = '{billing-queue}', BILLING_QUEUE = '{billing-queue}',
FILE_TASK_QUEUE = '{file-task-queue}', FILE_TASK_QUEUE = '{file-task-queue}',
SEARCH_QUEUE = '{search-queue}',
AI_QUEUE = '{ai-queue}', AI_QUEUE = '{ai-queue}',
HISTORY_QUEUE = '{history-queue}', HISTORY_QUEUE = '{history-queue}',
NOTIFICATION_QUEUE = '{notification-queue}', NOTIFICATION_QUEUE = '{notification-queue}',
@@ -32,12 +31,6 @@ export enum QueueJob {
IMPORT_TASK = 'import-task', IMPORT_TASK = 'import-task',
EXPORT_TASK = 'export-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_PAGE = 'search-remove-page',
SEARCH_REMOVE_ASSET = 'search-remove-attachment', SEARCH_REMOVE_ASSET = 'search-remove-attachment',
SEARCH_REMOVE_FACE = 'search-remove-comment', SEARCH_REMOVE_FACE = 'search-remove-comment',
@@ -57,14 +57,6 @@ import { GeneralQueueProcessor } from './processors/general-queue.processor';
attempts: 1, attempts: 1,
}, },
}), }),
BullModule.registerQueue({
name: QueueName.SEARCH_QUEUE,
defaultJobOptions: {
removeOnComplete: true,
removeOnFail: true,
attempts: 2,
},
}),
BullModule.registerQueue({ BullModule.registerQueue({
name: QueueName.AI_QUEUE, name: QueueName.AI_QUEUE,
defaultJobOptions: { defaultJobOptions: {