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:
@@ -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: {
|
||||||
|
|||||||
Reference in New Issue
Block a user