diff --git a/packages/api-nest/package.json b/packages/api-nest/package.json index b97347395..888346448 100644 --- a/packages/api-nest/package.json +++ b/packages/api-nest/package.json @@ -30,6 +30,7 @@ "dependencies": { "@apollo/server": "^4.11.1", "@nestjs/apollo": "^12.0.11", + "@nestjs/bullmq": "^10.0.0", "@nestjs/common": "^10.0.0", "@nestjs/config": "^3.0.0", "@nestjs/core": "^10.0.0", @@ -41,6 +42,7 @@ "@nestjs/terminus": "^10.0.0", "@nestjs/typeorm": "^10.0.0", "bcrypt": "^5.1.1", + "bullmq": "^5.0.0", "google-auth-library": "^9.0.0", "graphql": "^16.11.0", "ioredis": "^5.3.2", @@ -108,6 +110,9 @@ "transform": { "^.+\\.(t|j)s$": "ts-jest" }, + "transformIgnorePatterns": [ + "node_modules/(?!(bullmq|msgpackr)/)" + ], "collectCoverageFrom": [ "**/*.(t|j)s" ], diff --git a/packages/api-nest/schema.graphql b/packages/api-nest/schema.graphql index b16f432ed..4db46d137 100644 --- a/packages/api-nest/schema.graphql +++ b/packages/api-nest/schema.graphql @@ -242,6 +242,9 @@ input SaveUrlInput { """Folder to save the URL to (inbox, archive)""" folder: String = "inbox" + """Source of the save request (web, mobile, api, extension)""" + source: String = "web" + """URL to save to library""" url: String! } diff --git a/packages/api-nest/src/app/app.module.ts b/packages/api-nest/src/app/app.module.ts index eb4b13335..148952089 100644 --- a/packages/api-nest/src/app/app.module.ts +++ b/packages/api-nest/src/app/app.module.ts @@ -8,6 +8,7 @@ import { LoggingModule } from '../logging/logging.module' import { GraphqlModule } from '../graphql/graphql.module' import { LibraryModule } from '../library/library.module' import { LabelModule } from '../label/label.module' +import { QueueModule } from '../queue/queue.module' import { AppController } from './app.controller' import { AppService } from './app.service' import { configValidationSchema } from '../config/config.schema' @@ -43,6 +44,9 @@ import { configValidationSchema } from '../config/config.schema' // Library / Reader LibraryModule, + // Queue and Background Processing + QueueModule, + // GraphQL API GraphqlModule, ], diff --git a/packages/api-nest/src/health/health.controller.ts b/packages/api-nest/src/health/health.controller.ts index 934212646..cf29fc39a 100644 --- a/packages/api-nest/src/health/health.controller.ts +++ b/packages/api-nest/src/health/health.controller.ts @@ -6,6 +6,7 @@ import { } from '@nestjs/terminus' import { ApiTags, ApiOperation, ApiResponse } from '@nestjs/swagger' import { RedisHealthIndicator } from './redis-health.indicator' +import { QueueHealthIndicator } from '../queue/queue-health.indicator' @ApiTags('health') @Controller('health') @@ -14,6 +15,7 @@ export class HealthController { private health: HealthCheckService, private db: TypeOrmHealthIndicator, private redis: RedisHealthIndicator, + private queue: QueueHealthIndicator, ) {} @ApiOperation({ summary: 'Basic health check' }) @@ -85,6 +87,8 @@ export class HealthController { () => this.db.pingCheck('database'), // Redis health check () => this.redis.isHealthy('redis'), + // Queue health check + () => this.queue.isHealthy('queues'), // System health check () => this.getSystemHealth(), ]) @@ -99,6 +103,8 @@ export class HealthController { () => this.db.pingCheck('database'), // Redis health check () => this.redis.isHealthy('redis'), + // Queue health check + () => this.queue.isHealthy('queues'), // System health check () => this.getSystemHealth(), // Application health check diff --git a/packages/api-nest/src/health/health.module.ts b/packages/api-nest/src/health/health.module.ts index 783cc1e59..a54f9e242 100644 --- a/packages/api-nest/src/health/health.module.ts +++ b/packages/api-nest/src/health/health.module.ts @@ -3,9 +3,10 @@ import { TerminusModule } from '@nestjs/terminus' import { ConfigModule } from '@nestjs/config' import { HealthController } from './health.controller' import { RedisHealthIndicator } from './redis-health.indicator' +import { QueueModule } from '../queue/queue.module' @Module({ - imports: [TerminusModule, ConfigModule], + imports: [TerminusModule, ConfigModule, QueueModule], controllers: [HealthController], providers: [RedisHealthIndicator], }) diff --git a/packages/api-nest/src/library/dto/library-inputs.type.ts b/packages/api-nest/src/library/dto/library-inputs.type.ts index f6fa2b2e6..c03df41a1 100644 --- a/packages/api-nest/src/library/dto/library-inputs.type.ts +++ b/packages/api-nest/src/library/dto/library-inputs.type.ts @@ -194,4 +194,14 @@ export class SaveUrlInput { @IsString() @IsIn(['inbox', 'archive']) folder?: string + + @Field(() => String, { + nullable: true, + defaultValue: 'web', + description: 'Source of the save request (web, mobile, api, extension)', + }) + @IsOptional() + @IsString() + @IsIn(['web', 'mobile', 'api', 'extension']) + source?: 'web' | 'mobile' | 'api' | 'extension' } diff --git a/packages/api-nest/src/library/library.module.ts b/packages/api-nest/src/library/library.module.ts index c1893bfa8..0c16cde80 100644 --- a/packages/api-nest/src/library/library.module.ts +++ b/packages/api-nest/src/library/library.module.ts @@ -5,11 +5,13 @@ import { LibraryService } from './library.service' import { LibraryController } from './library.controller' import { LibraryItemEntity } from './entities/library-item.entity' import { LabelModule } from '../label/label.module' +import { QueueModule } from '../queue/queue.module' @Module({ imports: [ TypeOrmModule.forFeature([LibraryItemEntity]), LabelModule, + QueueModule, ], controllers: [LibraryController], providers: [LibraryResolver, LibraryService], diff --git a/packages/api-nest/src/library/library.service.ts b/packages/api-nest/src/library/library.service.ts index 398481d4f..5dc4c72ed 100644 --- a/packages/api-nest/src/library/library.service.ts +++ b/packages/api-nest/src/library/library.service.ts @@ -18,6 +18,9 @@ import { SortOrder, SaveUrlInput, } from './dto/library-inputs.type' +import { EventBusService } from '../queue/event-bus.service' +import { EVENT_NAMES } from '../queue/events.constants' +import { JOB_PRIORITY } from '../queue/queue.constants' @Injectable() export class LibraryService { @@ -27,6 +30,7 @@ export class LibraryService { @InjectRepository(LibraryItemEntity) private readonly libraryRepository: Repository, private readonly dataSource: DataSource, + private readonly eventBus: EventBusService, ) {} async listForUser( @@ -682,11 +686,21 @@ export class LibraryService { const savedItem = await this.libraryRepository.save(libraryItem) this.logger.log( - `Successfully saved URL with ID: ${savedItem.id} (content extraction deferred to queue)`, + `Successfully saved URL with ID: ${savedItem.id}, dispatching to queue for content extraction`, ) - // TODO: Dispatch to queue for content extraction (ARC-012) - // TODO: Use @omnivore/readability for proper extraction (ARC-013) + // Emit event to trigger background content processing + this.eventBus.emitContentSaveRequested({ + eventType: EVENT_NAMES.CONTENT_SAVE_REQUESTED, + libraryItemId: savedItem.id, + url: savedItem.originalUrl, + userId: savedItem.userId, + priority: JOB_PRIORITY.NORMAL, + source: input.source || 'web', + timestamp: new Date(), + }) + + this.logger.log(`Content processing job enqueued for item ${savedItem.id}`) return savedItem } diff --git a/packages/api-nest/src/queue/event-bus.service.spec.ts b/packages/api-nest/src/queue/event-bus.service.spec.ts new file mode 100644 index 000000000..f98c1d5b4 --- /dev/null +++ b/packages/api-nest/src/queue/event-bus.service.spec.ts @@ -0,0 +1,358 @@ +/** + * EventBusService Unit Tests + */ + +import { Test, TestingModule } from '@nestjs/testing' +import { getQueueToken } from '@nestjs/bullmq' +import { Queue } from 'bullmq' +import { EventBusService } from './event-bus.service' +import { QUEUE_NAMES, JOB_TYPES, JOB_PRIORITY } from './queue.constants' +import { EVENT_NAMES } from './events.constants' + +type MockQueue = jest.Mocked> + +// Mock logger to suppress console output during tests +const mockLogger = { + log: jest.fn(), + error: jest.fn(), + warn: jest.fn(), + debug: jest.fn(), + verbose: jest.fn(), +} + +describe('EventBusService', () => { + let service: EventBusService + let contentQueue: MockQueue + let notificationQueue: MockQueue + let postProcessingQueue: MockQueue + + beforeEach(async () => { + // Mock queue objects + contentQueue = { + add: jest.fn().mockResolvedValue({ id: 'job-123' }), + name: QUEUE_NAMES.CONTENT_PROCESSING, + } + + notificationQueue = { + add: jest.fn().mockResolvedValue({ id: 'job-456' }), + name: QUEUE_NAMES.NOTIFICATIONS, + } + + postProcessingQueue = { + add: jest.fn().mockResolvedValue({ id: 'job-789' }), + name: QUEUE_NAMES.POST_PROCESSING, + } + + const module: TestingModule = await Test.createTestingModule({ + providers: [ + EventBusService, + { + provide: getQueueToken(QUEUE_NAMES.CONTENT_PROCESSING), + useValue: contentQueue, + }, + { + provide: getQueueToken(QUEUE_NAMES.NOTIFICATIONS), + useValue: notificationQueue, + }, + { + provide: getQueueToken(QUEUE_NAMES.POST_PROCESSING), + useValue: postProcessingQueue, + }, + ], + }) + .setLogger(mockLogger) + .compile() + + service = module.get(EventBusService) + }) + + afterEach(() => { + jest.clearAllMocks() + jest.restoreAllMocks() + }) + + describe('initialization', () => { + it('should be defined', () => { + expect(service).toBeDefined() + }) + + it('should register event listeners on module init', () => { + const listenerSpy = jest.spyOn(service, 'on') + service.onModuleInit() + + expect(listenerSpy).toHaveBeenCalledWith( + EVENT_NAMES.CONTENT_SAVE_REQUESTED, + expect.any(Function), + ) + expect(listenerSpy).toHaveBeenCalledWith( + EVENT_NAMES.NOTIFICATION_REQUESTED, + expect.any(Function), + ) + }) + }) + + describe('emitContentSaveRequested', () => { + it('should emit content save requested event', async () => { + const emitSpy = jest.spyOn(service, 'emit') + + const event = { + eventType: EVENT_NAMES.CONTENT_SAVE_REQUESTED, + libraryItemId: 'item-123', + url: 'https://example.com', + userId: 'user-123', + timestamp: new Date(), + } + + service.emitContentSaveRequested(event) + + expect(emitSpy).toHaveBeenCalledWith( + EVENT_NAMES.CONTENT_SAVE_REQUESTED, + event, + ) + }) + + it('should enqueue content fetch job when event is emitted', async () => { + // Initialize module to register listeners + service.onModuleInit() + + const event = { + eventType: EVENT_NAMES.CONTENT_SAVE_REQUESTED, + libraryItemId: 'item-123', + url: 'https://example.com', + userId: 'user-123', + timestamp: new Date(), + } + + service.emitContentSaveRequested(event) + + // Wait for async handler + await new Promise((resolve) => setTimeout(resolve, 10)) + + expect(contentQueue.add).toHaveBeenCalledWith( + JOB_TYPES.FETCH_CONTENT, + expect.objectContaining({ + libraryItemId: 'item-123', + url: 'https://example.com', + userId: 'user-123', + }), + expect.objectContaining({ + jobId: 'item-123', + priority: JOB_PRIORITY.NORMAL, + }), + ) + }) + + it('should use custom priority if provided', async () => { + service.onModuleInit() + + const event = { + eventType: EVENT_NAMES.CONTENT_SAVE_REQUESTED, + libraryItemId: 'item-123', + url: 'https://example.com', + userId: 'user-123', + priority: JOB_PRIORITY.HIGH, + timestamp: new Date(), + } + + service.emitContentSaveRequested(event) + + await new Promise((resolve) => setTimeout(resolve, 10)) + + expect(contentQueue.add).toHaveBeenCalledWith( + JOB_TYPES.FETCH_CONTENT, + expect.any(Object), + expect.objectContaining({ + priority: JOB_PRIORITY.HIGH, + }), + ) + }) + }) + + describe('emitContentFetchCompleted', () => { + it('should emit content fetch completed event', () => { + const emitSpy = jest.spyOn(service, 'emit') + + const event = { + eventType: EVENT_NAMES.CONTENT_FETCH_COMPLETED, + libraryItemId: 'item-123', + jobId: 'job-123', + contentLength: 5000, + processingTime: 1500, + timestamp: new Date(), + } + + service.emitContentFetchCompleted(event) + + expect(emitSpy).toHaveBeenCalledWith( + EVENT_NAMES.CONTENT_FETCH_COMPLETED, + event, + ) + }) + + it('should enqueue post-processing job when content fetch completes', async () => { + service.onModuleInit() + + const event = { + eventType: EVENT_NAMES.CONTENT_FETCH_COMPLETED, + libraryItemId: 'item-123', + jobId: 'job-123', + contentLength: 5000, + processingTime: 1500, + timestamp: new Date(), + } + + service.emitContentFetchCompleted(event) + + await new Promise((resolve) => setTimeout(resolve, 10)) + + expect(postProcessingQueue.add).toHaveBeenCalledWith( + JOB_TYPES.UPDATE_SEARCH_INDEX, + expect.objectContaining({ + libraryItemId: 'item-123', + }), + expect.objectContaining({ + priority: JOB_PRIORITY.LOW, + }), + ) + }) + }) + + describe('emitContentFetchFailed', () => { + it('should emit content fetch failed event', () => { + const emitSpy = jest.spyOn(service, 'emit') + + const event = { + eventType: EVENT_NAMES.CONTENT_FETCH_FAILED, + libraryItemId: 'item-123', + jobId: 'job-123', + error: 'Network timeout', + retryCount: 1, + willRetry: true, + timestamp: new Date(), + } + + service.emitContentFetchFailed(event) + + expect(emitSpy).toHaveBeenCalledWith( + EVENT_NAMES.CONTENT_FETCH_FAILED, + event, + ) + }) + + it('should send notification when final failure occurs', async () => { + service.onModuleInit() + + const emitNotificationSpy = jest.spyOn( + service, + 'emitNotificationRequested', + ) + + const event = { + eventType: EVENT_NAMES.CONTENT_FETCH_FAILED, + libraryItemId: 'item-123', + jobId: 'job-123', + userId: 'user-123', + error: 'Max retries exceeded', + retryCount: 3, + willRetry: false, + timestamp: new Date(), + } + + service.emitContentFetchFailed(event) + + await new Promise((resolve) => setTimeout(resolve, 10)) + + expect(emitNotificationSpy).toHaveBeenCalledWith( + expect.objectContaining({ + userId: 'user-123', + notificationType: 'in-app', + }), + ) + }) + }) + + describe('emitNotificationRequested', () => { + it('should emit notification requested event', () => { + const emitSpy = jest.spyOn(service, 'emit') + + const event = { + eventType: EVENT_NAMES.NOTIFICATION_REQUESTED, + userId: 'user-123', + notificationType: 'email' as const, + title: 'Test Notification', + message: 'This is a test', + timestamp: new Date(), + } + + service.emitNotificationRequested(event) + + expect(emitSpy).toHaveBeenCalledWith( + EVENT_NAMES.NOTIFICATION_REQUESTED, + event, + ) + }) + + it('should enqueue notification job', async () => { + service.onModuleInit() + + const event = { + eventType: EVENT_NAMES.NOTIFICATION_REQUESTED, + userId: 'user-123', + notificationType: 'email' as const, + title: 'Test Notification', + message: 'This is a test', + timestamp: new Date(), + } + + service.emitNotificationRequested(event) + + await new Promise((resolve) => setTimeout(resolve, 10)) + + expect(notificationQueue.add).toHaveBeenCalledWith( + JOB_TYPES.SEND_NOTIFICATION, + expect.objectContaining({ + userId: 'user-123', + title: 'Test Notification', + message: 'This is a test', + }), + expect.objectContaining({ + priority: JOB_PRIORITY.HIGH, + attempts: 5, + }), + ) + }) + }) + + describe('error handling', () => { + it('should handle queue add errors gracefully', async () => { + service.onModuleInit() + + contentQueue.add.mockRejectedValueOnce(new Error('Queue is full')) + + const event = { + eventType: EVENT_NAMES.CONTENT_SAVE_REQUESTED, + libraryItemId: 'item-123', + url: 'https://example.com', + userId: 'user-123', + timestamp: new Date(), + } + + // Should not throw + service.emitContentSaveRequested(event) + + await new Promise((resolve) => setTimeout(resolve, 10)) + + expect(contentQueue.add).toHaveBeenCalled() + }) + }) + + describe('graceful shutdown', () => { + it('should remove all listeners on module destroy', async () => { + const removeListenersSpy = jest.spyOn(service, 'removeAllListeners') + + await service.onModuleDestroy() + + expect(removeListenersSpy).toHaveBeenCalled() + }) + }) +}) diff --git a/packages/api-nest/src/queue/event-bus.service.ts b/packages/api-nest/src/queue/event-bus.service.ts new file mode 100644 index 000000000..98db5749b --- /dev/null +++ b/packages/api-nest/src/queue/event-bus.service.ts @@ -0,0 +1,261 @@ +/** + * EventBusService - Lightweight Event Management + * + * Extends Node.js EventEmitter to provide type-safe event emission + * and routing to BullMQ queues. Decouples event emission from queue + * operations while maintaining simplicity. + */ + +import { Injectable, OnModuleInit, Logger } from '@nestjs/common' +import { InjectQueue } from '@nestjs/bullmq' +import { Queue } from 'bullmq' +import { EventEmitter } from 'events' +import { + EVENT_NAMES, + ContentSaveRequestedEvent, + ContentFetchStartedEvent, + ContentFetchCompletedEvent, + ContentFetchFailedEvent, + LibraryItemCreatedEvent, + NotificationRequestedEvent, + SearchIndexUpdateRequestedEvent, + AppEvent, +} from './events.constants' +import { + QUEUE_NAMES, + JOB_TYPES, + JOB_PRIORITY, + JOB_CONFIG, +} from './queue.constants' + +@Injectable() +export class EventBusService extends EventEmitter implements OnModuleInit { + private readonly logger = new Logger(EventBusService.name) + + constructor( + @InjectQueue(QUEUE_NAMES.CONTENT_PROCESSING) + private readonly contentQueue: Queue, + @InjectQueue(QUEUE_NAMES.NOTIFICATIONS) + private readonly notificationQueue: Queue, + @InjectQueue(QUEUE_NAMES.POST_PROCESSING) + private readonly postProcessingQueue: Queue + ) { + super() + // Increase max listeners to prevent warnings (default is 10) + this.setMaxListeners(50) + } + + /** + * Initialize event listeners on module startup + */ + onModuleInit() { + this.logger.log('Initializing EventBusService and registering event handlers') + + // Content processing events + this.on(EVENT_NAMES.CONTENT_SAVE_REQUESTED, this.handleContentSaveRequested.bind(this)) + this.on(EVENT_NAMES.CONTENT_FETCH_STARTED, this.handleContentFetchStarted.bind(this)) + this.on(EVENT_NAMES.CONTENT_FETCH_COMPLETED, this.handleContentFetchCompleted.bind(this)) + this.on(EVENT_NAMES.CONTENT_FETCH_FAILED, this.handleContentFetchFailed.bind(this)) + + // Library events + this.on(EVENT_NAMES.LIBRARY_ITEM_CREATED, this.handleLibraryItemCreated.bind(this)) + + // Notification events + this.on(EVENT_NAMES.NOTIFICATION_REQUESTED, this.handleNotificationRequested.bind(this)) + + // Post-processing events + this.on(EVENT_NAMES.SEARCH_INDEX_UPDATE_REQUESTED, this.handleSearchIndexUpdateRequested.bind(this)) + + this.logger.log('EventBusService initialized successfully') + } + + /** + * Type-safe event emission methods + */ + + emitContentSaveRequested(event: ContentSaveRequestedEvent) { + this.logger.debug(`Emitting ${EVENT_NAMES.CONTENT_SAVE_REQUESTED} for item ${event.libraryItemId}`) + this.emit(EVENT_NAMES.CONTENT_SAVE_REQUESTED, event) + } + + emitContentFetchStarted(event: ContentFetchStartedEvent) { + this.logger.debug(`Emitting ${EVENT_NAMES.CONTENT_FETCH_STARTED} for item ${event.libraryItemId}`) + this.emit(EVENT_NAMES.CONTENT_FETCH_STARTED, event) + } + + emitContentFetchCompleted(event: ContentFetchCompletedEvent) { + this.logger.debug(`Emitting ${EVENT_NAMES.CONTENT_FETCH_COMPLETED} for item ${event.libraryItemId}`) + this.emit(EVENT_NAMES.CONTENT_FETCH_COMPLETED, event) + } + + emitContentFetchFailed(event: ContentFetchFailedEvent) { + this.logger.warn(`Emitting ${EVENT_NAMES.CONTENT_FETCH_FAILED} for item ${event.libraryItemId}: ${event.error}`) + this.emit(EVENT_NAMES.CONTENT_FETCH_FAILED, event) + } + + emitLibraryItemCreated(event: LibraryItemCreatedEvent) { + this.logger.debug(`Emitting ${EVENT_NAMES.LIBRARY_ITEM_CREATED} for item ${event.libraryItemId}`) + this.emit(EVENT_NAMES.LIBRARY_ITEM_CREATED, event) + } + + emitNotificationRequested(event: NotificationRequestedEvent) { + this.logger.debug(`Emitting ${EVENT_NAMES.NOTIFICATION_REQUESTED} for user ${event.userId}`) + this.emit(EVENT_NAMES.NOTIFICATION_REQUESTED, event) + } + + emitSearchIndexUpdateRequested(event: SearchIndexUpdateRequestedEvent) { + this.logger.debug(`Emitting ${EVENT_NAMES.SEARCH_INDEX_UPDATE_REQUESTED} for item ${event.libraryItemId}`) + this.emit(EVENT_NAMES.SEARCH_INDEX_UPDATE_REQUESTED, event) + } + + /** + * Event Handlers - Route events to appropriate queues + */ + + private async handleContentSaveRequested(event: ContentSaveRequestedEvent) { + try { + const priority = event.priority || JOB_PRIORITY.NORMAL + + const job = await this.contentQueue.add( + JOB_TYPES.FETCH_CONTENT, + { + libraryItemId: event.libraryItemId, + url: event.url, + userId: event.userId, + source: event.source, + timestamp: event.timestamp, + }, + { + jobId: event.libraryItemId, // Use libraryItemId as job ID for deduplication + priority, + attempts: JOB_CONFIG.RETRY_ATTEMPTS, + backoff: { + type: JOB_CONFIG.RETRY_BACKOFF_TYPE, + delay: JOB_CONFIG.RETRY_BACKOFF_DELAY, + }, + } + ) + + this.logger.log(`Enqueued content fetch job ${job.id} for item ${event.libraryItemId}`) + } catch (error) { + const errorMessage = error instanceof Error ? error.message : String(error) + const errorStack = error instanceof Error ? error.stack : undefined + this.logger.error(`Failed to enqueue content fetch job: ${errorMessage}`, errorStack) + // Don't throw - log error and continue gracefully + } + } + + private async handleContentFetchStarted(event: ContentFetchStartedEvent) { + this.logger.log(`Content fetch started for item ${event.libraryItemId}`) + // Could emit metrics or update status here + } + + private async handleContentFetchCompleted(event: ContentFetchCompletedEvent) { + this.logger.log( + `Content fetch completed for item ${event.libraryItemId} in ${event.processingTime}ms` + ) + + // Trigger post-processing tasks + try { + await this.postProcessingQueue.add( + JOB_TYPES.UPDATE_SEARCH_INDEX, + { + libraryItemId: event.libraryItemId, + timestamp: new Date(), + }, + { + priority: JOB_PRIORITY.LOW, + } + ) + } catch (error) { + const errorMessage = error instanceof Error ? error.message : String(error) + const errorStack = error instanceof Error ? error.stack : undefined + this.logger.error(`Failed to enqueue post-processing job: ${errorMessage}`, errorStack) + } + } + + private async handleContentFetchFailed(event: ContentFetchFailedEvent) { + this.logger.error( + `Content fetch failed for item ${event.libraryItemId}: ${event.error} ` + + `(retry ${event.retryCount}, will retry: ${event.willRetry})` + ) + + // Could send notification to user if final failure + if (!event.willRetry) { + // Notify user of failure + this.emitNotificationRequested({ + eventType: EVENT_NAMES.NOTIFICATION_REQUESTED, + userId: event.userId!, + notificationType: 'in-app', + title: 'Content Fetch Failed', + message: `Failed to fetch content for saved item`, + timestamp: new Date(), + data: { + libraryItemId: event.libraryItemId, + error: event.error, + }, + }) + } + } + + private async handleLibraryItemCreated(event: LibraryItemCreatedEvent) { + this.logger.log(`Library item created: ${event.libraryItemId}`) + // Could trigger analytics or other post-creation tasks + } + + private async handleNotificationRequested(event: NotificationRequestedEvent) { + try { + await this.notificationQueue.add( + JOB_TYPES.SEND_NOTIFICATION, + { + userId: event.userId, + notificationType: event.notificationType, + title: event.title, + message: event.message, + data: event.data, + timestamp: event.timestamp, + }, + { + priority: JOB_PRIORITY.HIGH, + attempts: 5, + } + ) + + this.logger.log(`Enqueued notification for user ${event.userId}`) + } catch (error) { + const errorMessage = error instanceof Error ? error.message : String(error) + const errorStack = error instanceof Error ? error.stack : undefined + this.logger.error(`Failed to enqueue notification: ${errorMessage}`, errorStack) + } + } + + private async handleSearchIndexUpdateRequested(event: SearchIndexUpdateRequestedEvent) { + try { + await this.postProcessingQueue.add( + JOB_TYPES.UPDATE_SEARCH_INDEX, + { + libraryItemId: event.libraryItemId, + action: event.action, + timestamp: event.timestamp, + }, + { + priority: JOB_PRIORITY.LOW, + } + ) + + this.logger.log(`Enqueued search index update for item ${event.libraryItemId}`) + } catch (error) { + const errorMessage = error instanceof Error ? error.message : String(error) + const errorStack = error instanceof Error ? error.stack : undefined + this.logger.error(`Failed to enqueue search index update: ${errorMessage}`, errorStack) + } + } + + /** + * Graceful shutdown + */ + async onModuleDestroy() { + this.logger.log('Shutting down EventBusService') + this.removeAllListeners() + } +} diff --git a/packages/api-nest/src/queue/events.constants.ts b/packages/api-nest/src/queue/events.constants.ts new file mode 100644 index 000000000..e55f938a8 --- /dev/null +++ b/packages/api-nest/src/queue/events.constants.ts @@ -0,0 +1,164 @@ +/** + * Event Type Constants and Interfaces + * + * Centralized event definitions for type-safe event emission + * and handling throughout the system. + */ + +/** + * Event Names - Define all event types in the system + */ +export const EVENT_NAMES = { + // Content events + CONTENT_SAVE_REQUESTED: 'content.save.requested', + CONTENT_FETCH_STARTED: 'content.fetch.started', + CONTENT_FETCH_COMPLETED: 'content.fetch.completed', + CONTENT_FETCH_FAILED: 'content.fetch.failed', + CONTENT_PARSE_COMPLETED: 'content.parse.completed', + CONTENT_PARSE_FAILED: 'content.parse.failed', + + // Library events + LIBRARY_ITEM_CREATED: 'library.item.created', + LIBRARY_ITEM_UPDATED: 'library.item.updated', + LIBRARY_ITEM_DELETED: 'library.item.deleted', + + // User events + USER_CREATED: 'user.created', + USER_UPDATED: 'user.updated', + USER_DELETED: 'user.deleted', + + // Notification events + NOTIFICATION_REQUESTED: 'notification.requested', + NOTIFICATION_SENT: 'notification.sent', + NOTIFICATION_FAILED: 'notification.failed', + + // Post-processing events + SEARCH_INDEX_UPDATE_REQUESTED: 'search.index.update.requested', + THUMBNAIL_GENERATION_REQUESTED: 'thumbnail.generation.requested', +} as const + +/** + * Base Event Interface + */ +export interface BaseEvent { + eventType: string + timestamp: Date + userId?: string + metadata?: Record +} + +/** + * Content Save Requested Event + * Emitted when a user saves a new URL to their library + */ +export interface ContentSaveRequestedEvent extends BaseEvent { + eventType: typeof EVENT_NAMES.CONTENT_SAVE_REQUESTED + libraryItemId: string + url: string + userId: string + priority?: number + source?: 'web' | 'mobile' | 'api' | 'extension' +} + +/** + * Content Fetch Started Event + */ +export interface ContentFetchStartedEvent extends BaseEvent { + eventType: typeof EVENT_NAMES.CONTENT_FETCH_STARTED + libraryItemId: string + url: string + jobId: string +} + +/** + * Content Fetch Completed Event + */ +export interface ContentFetchCompletedEvent extends BaseEvent { + eventType: typeof EVENT_NAMES.CONTENT_FETCH_COMPLETED + libraryItemId: string + jobId: string + contentLength: number + processingTime: number +} + +/** + * Content Fetch Failed Event + */ +export interface ContentFetchFailedEvent extends BaseEvent { + eventType: typeof EVENT_NAMES.CONTENT_FETCH_FAILED + libraryItemId: string + jobId: string + error: string + retryCount: number + willRetry: boolean +} + +/** + * Library Item Created Event + */ +export interface LibraryItemCreatedEvent extends BaseEvent { + eventType: typeof EVENT_NAMES.LIBRARY_ITEM_CREATED + libraryItemId: string + userId: string + url: string + title?: string +} + +/** + * Library Item Updated Event + */ +export interface LibraryItemUpdatedEvent extends BaseEvent { + eventType: typeof EVENT_NAMES.LIBRARY_ITEM_UPDATED + libraryItemId: string + userId: string + updatedFields: string[] +} + +/** + * Library Item Deleted Event + */ +export interface LibraryItemDeletedEvent extends BaseEvent { + eventType: typeof EVENT_NAMES.LIBRARY_ITEM_DELETED + libraryItemId: string + userId: string +} + +/** + * Notification Requested Event + */ +export interface NotificationRequestedEvent extends BaseEvent { + eventType: typeof EVENT_NAMES.NOTIFICATION_REQUESTED + userId: string + notificationType: 'email' | 'push' | 'in-app' + title: string + message: string + data?: Record +} + +/** + * Search Index Update Requested Event + */ +export interface SearchIndexUpdateRequestedEvent extends BaseEvent { + eventType: typeof EVENT_NAMES.SEARCH_INDEX_UPDATE_REQUESTED + libraryItemId: string + action: 'index' | 'update' | 'delete' +} + +/** + * Union type of all events + */ +export type AppEvent = + | ContentSaveRequestedEvent + | ContentFetchStartedEvent + | ContentFetchCompletedEvent + | ContentFetchFailedEvent + | LibraryItemCreatedEvent + | LibraryItemUpdatedEvent + | LibraryItemDeletedEvent + | NotificationRequestedEvent + | SearchIndexUpdateRequestedEvent + +/** + * Type exports for type-safe usage + */ +export type EventName = typeof EVENT_NAMES[keyof typeof EVENT_NAMES] diff --git a/packages/api-nest/src/queue/processors/content-processor.service.spec.ts b/packages/api-nest/src/queue/processors/content-processor.service.spec.ts new file mode 100644 index 000000000..4707d1d77 --- /dev/null +++ b/packages/api-nest/src/queue/processors/content-processor.service.spec.ts @@ -0,0 +1,387 @@ +/** + * ContentProcessorService Unit Tests + */ + +import { Test, TestingModule } from '@nestjs/testing' +import { getRepositoryToken } from '@nestjs/typeorm' +import { Repository } from 'typeorm' +import { Job } from 'bullmq' +import { + ContentProcessorService, + FetchContentJobData, +} from './content-processor.service' +import { + LibraryItemEntity, + LibraryItemState, +} from '../../library/entities/library-item.entity' +import { EventBusService } from '../event-bus.service' +import { JOB_TYPES } from '../queue.constants' + +// Mock logger to suppress console output during tests +const mockLogger = { + log: jest.fn(), + error: jest.fn(), + warn: jest.fn(), + debug: jest.fn(), + verbose: jest.fn(), +} + +describe('ContentProcessorService', () => { + let service: ContentProcessorService + let repository: jest.Mocked> + let eventBus: jest.Mocked + + beforeEach(async () => { + const mockRepository = { + update: jest.fn().mockResolvedValue({ affected: 1 }), + findOne: jest.fn(), + } + + const mockEventBus = { + emitContentFetchStarted: jest.fn(), + emitContentFetchCompleted: jest.fn(), + emitContentFetchFailed: jest.fn(), + } + + const module: TestingModule = await Test.createTestingModule({ + providers: [ + ContentProcessorService, + { + provide: getRepositoryToken(LibraryItemEntity), + useValue: mockRepository, + }, + { + provide: EventBusService, + useValue: mockEventBus, + }, + ], + }) + .setLogger(mockLogger) + .compile() + + service = module.get(ContentProcessorService) + repository = module.get(getRepositoryToken(LibraryItemEntity)) + eventBus = module.get(EventBusService) + }) + + afterEach(() => { + jest.clearAllMocks() + jest.restoreAllMocks() + }) + + describe('initialization', () => { + it('should be defined', () => { + expect(service).toBeDefined() + }) + + it('should log initialization message', () => { + const logSpy = jest.spyOn(service['logger'], 'log') + service.onModuleInit() + expect(logSpy).toHaveBeenCalledWith( + expect.stringContaining('ContentProcessorService initialized'), + ) + }) + }) + + describe('process', () => { + it('should route fetch-content jobs to handleFetchContent', async () => { + const jobData: FetchContentJobData = { + libraryItemId: 'item-123', + url: 'https://example.com', + userId: 'user-123', + timestamp: new Date(), + } + + const mockJob = createMockJob(JOB_TYPES.FETCH_CONTENT, jobData) + const handleSpy = jest + .spyOn(service as any, 'handleFetchContent') + .mockResolvedValue({ success: true }) + + await service.process(mockJob) + + expect(handleSpy).toHaveBeenCalledWith(mockJob) + }) + + it('should route parse-content jobs to handleParseContent', async () => { + const jobData: FetchContentJobData = { + libraryItemId: 'item-123', + url: 'https://example.com', + userId: 'user-123', + timestamp: new Date(), + } + + const mockJob = createMockJob(JOB_TYPES.PARSE_CONTENT, jobData) + const handleSpy = jest + .spyOn(service as any, 'handleParseContent') + .mockResolvedValue({ success: true }) + + await service.process(mockJob) + + expect(handleSpy).toHaveBeenCalledWith(mockJob) + }) + + it('should throw error for unknown job type', async () => { + const jobData: FetchContentJobData = { + libraryItemId: 'item-123', + url: 'https://example.com', + userId: 'user-123', + timestamp: new Date(), + } + + const mockJob = createMockJob('unknown-job-type', jobData) + + await expect(service.process(mockJob)).rejects.toThrow('Unknown job type') + }) + }) + + describe('handleFetchContent', () => { + it('should successfully fetch and save content', async () => { + const jobData: FetchContentJobData = { + libraryItemId: 'item-123', + url: 'https://example.com', + userId: 'user-123', + timestamp: new Date(), + } + + const mockJob = createMockJob(JOB_TYPES.FETCH_CONTENT, jobData) + + const result = await service['handleFetchContent'](mockJob) + + expect(result.success).toBe(true) + expect(result.title).toBeDefined() + expect(result.content).toBeDefined() + + // Verify state updates + expect(repository.update).toHaveBeenCalledWith('item-123', { + state: LibraryItemState.PROCESSING, + }) + expect(repository.update).toHaveBeenCalledWith('item-123', { + state: LibraryItemState.SUCCEEDED, + }) + + // Verify content saved + expect(repository.update).toHaveBeenCalledWith( + 'item-123', + expect.objectContaining({ + title: expect.any(String), + readableContent: expect.any(String), + }), + ) + + // Verify events emitted + expect(eventBus.emitContentFetchStarted).toHaveBeenCalledWith( + expect.objectContaining({ + libraryItemId: 'item-123', + url: 'https://example.com', + }), + ) + expect(eventBus.emitContentFetchCompleted).toHaveBeenCalledWith( + expect.objectContaining({ + libraryItemId: 'item-123', + contentLength: expect.any(Number), + processingTime: expect.any(Number), + }), + ) + }) + + it('should update job progress during processing', async () => { + const jobData: FetchContentJobData = { + libraryItemId: 'item-123', + url: 'https://example.com', + userId: 'user-123', + timestamp: new Date(), + } + + const mockJob = createMockJob(JOB_TYPES.FETCH_CONTENT, jobData) + + await service['handleFetchContent'](mockJob) + + expect(mockJob.updateProgress).toHaveBeenCalledWith(10) + expect(mockJob.updateProgress).toHaveBeenCalledWith(20) + expect(mockJob.updateProgress).toHaveBeenCalledWith(70) + expect(mockJob.updateProgress).toHaveBeenCalledWith(90) + expect(mockJob.updateProgress).toHaveBeenCalledWith(100) + }) + + it('should handle fetch errors and emit failed event', async () => { + const jobData: FetchContentJobData = { + libraryItemId: 'item-123', + url: 'https://example.com', + userId: 'user-123', + timestamp: new Date(), + } + + const mockJob = createMockJob(JOB_TYPES.FETCH_CONTENT, jobData, { + attemptsMade: 1, + attempts: 3, + }) + + // Mock fetchContent to throw error + jest + .spyOn(service as any, 'fetchContent') + .mockRejectedValueOnce(new Error('Network error')) + + await expect(service['handleFetchContent'](mockJob)).rejects.toThrow( + 'Network error', + ) + + // Should NOT update to FAILED since there are retries left (attempt 2 of 3) + expect(repository.update).not.toHaveBeenCalledWith('item-123', { + state: LibraryItemState.FAILED, + }) + + // Should emit failed event + expect(eventBus.emitContentFetchFailed).toHaveBeenCalledWith( + expect.objectContaining({ + libraryItemId: 'item-123', + error: 'Network error', + retryCount: 2, + willRetry: true, + }), + ) + }) + + it('should update to FAILED state on final attempt', async () => { + const jobData: FetchContentJobData = { + libraryItemId: 'item-123', + url: 'https://example.com', + userId: 'user-123', + timestamp: new Date(), + } + + const mockJob = createMockJob(JOB_TYPES.FETCH_CONTENT, jobData, { + attemptsMade: 2, + attempts: 3, + }) + + // Mock fetchContent to throw error + jest + .spyOn(service as any, 'fetchContent') + .mockRejectedValueOnce(new Error('Final error')) + + await expect(service['handleFetchContent'](mockJob)).rejects.toThrow( + 'Final error', + ) + + // Should update to FAILED on final attempt + expect(repository.update).toHaveBeenCalledWith('item-123', { + state: LibraryItemState.FAILED, + }) + + // Should emit failed event with willRetry: false (attempt 3 of 3) + expect(eventBus.emitContentFetchFailed).toHaveBeenCalledWith( + expect.objectContaining({ + libraryItemId: 'item-123', + error: 'Final error', + retryCount: 3, + willRetry: false, + }), + ) + }) + }) + + describe('saveContent', () => { + it('should save all content fields to database', async () => { + const result = { + success: true, + title: 'Test Title', + content: '

Test content

', + author: 'John Doe', + publishedDate: new Date('2025-01-01'), + siteIcon: 'https://example.com/icon.png', + thumbnail: 'https://example.com/thumb.jpg', + } + + await service['saveContent']('item-123', result) + + expect(repository.update).toHaveBeenCalledWith('item-123', { + title: 'Test Title', + readableContent: '

Test content

', + author: 'John Doe', + publishedAt: result.publishedDate, + siteIcon: 'https://example.com/icon.png', + thumbnail: 'https://example.com/thumb.jpg', + }) + }) + + it('should handle save errors gracefully', async () => { + repository.update.mockRejectedValueOnce(new Error('Database error')) + + const result = { + success: true, + title: 'Test Title', + content: '

Test content

', + } + + await expect(service['saveContent']('item-123', result)).rejects.toThrow( + 'Database error', + ) + }) + }) + + describe('updateLibraryItemState', () => { + it('should update state to PROCESSING', async () => { + await service['updateLibraryItemState']( + 'item-123', + LibraryItemState.PROCESSING, + ) + + expect(repository.update).toHaveBeenCalledWith('item-123', { + state: LibraryItemState.PROCESSING, + }) + }) + + it('should update state to SUCCEEDED', async () => { + await service['updateLibraryItemState']( + 'item-123', + LibraryItemState.SUCCEEDED, + ) + + expect(repository.update).toHaveBeenCalledWith('item-123', { + state: LibraryItemState.SUCCEEDED, + }) + }) + + it('should update state to FAILED', async () => { + await service['updateLibraryItemState']( + 'item-123', + LibraryItemState.FAILED, + ) + + expect(repository.update).toHaveBeenCalledWith('item-123', { + state: LibraryItemState.FAILED, + }) + }) + + it('should handle update errors', async () => { + repository.update.mockRejectedValueOnce(new Error('Update failed')) + + await expect( + service['updateLibraryItemState']( + 'item-123', + LibraryItemState.PROCESSING, + ), + ).rejects.toThrow('Update failed') + }) + }) +}) + +/** + * Helper function to create mock Job objects + */ +function createMockJob( + name: string, + data: FetchContentJobData, + opts: Partial<{ attemptsMade: number; attempts: number }> = {}, +): Job { + return { + id: 'job-123', + name, + data, + attemptsMade: opts.attemptsMade || 0, + opts: { + attempts: opts.attempts || 3, + }, + updateProgress: jest.fn().mockResolvedValue(undefined), + } as any +} diff --git a/packages/api-nest/src/queue/processors/content-processor.service.ts b/packages/api-nest/src/queue/processors/content-processor.service.ts new file mode 100644 index 000000000..a8d48da52 --- /dev/null +++ b/packages/api-nest/src/queue/processors/content-processor.service.ts @@ -0,0 +1,304 @@ +/** + * ContentProcessorService - BullMQ Worker for Content Processing + * + * Processes jobs from the content-processing queue to fetch and parse + * web content for saved library items. + */ + +import { Injectable, Logger, OnModuleInit } from '@nestjs/common' +import { Processor, WorkerHost, OnWorkerEvent } from '@nestjs/bullmq' +import { Job } from 'bullmq' +import { InjectRepository } from '@nestjs/typeorm' +import { Repository } from 'typeorm' +import { LibraryItemEntity, LibraryItemState } from '../../library/entities/library-item.entity' +import { EventBusService } from '../event-bus.service' +import { EVENT_NAMES } from '../events.constants' +import { QUEUE_NAMES, JOB_TYPES, JOB_CONFIG } from '../queue.constants' + +/** + * Job data interface for fetch-content jobs + */ +export interface FetchContentJobData { + libraryItemId: string + url: string + userId: string + source?: 'web' | 'mobile' | 'api' | 'extension' + timestamp: Date +} + +/** + * Result of content fetching operation + */ +export interface ContentFetchResult { + success: boolean + title?: string + content?: string + contentType?: string + author?: string + publishedDate?: Date + siteIcon?: string + thumbnail?: string + error?: string +} + +@Injectable() +@Processor(QUEUE_NAMES.CONTENT_PROCESSING, { + concurrency: JOB_CONFIG.WORKER_CONCURRENCY, +}) +export class ContentProcessorService extends WorkerHost implements OnModuleInit { + private readonly logger = new Logger(ContentProcessorService.name) + + constructor( + @InjectRepository(LibraryItemEntity) + private readonly libraryItemRepository: Repository, + private readonly eventBus: EventBusService + ) { + super() + } + + onModuleInit() { + this.logger.log( + `ContentProcessorService initialized with concurrency ${JOB_CONFIG.WORKER_CONCURRENCY}` + ) + } + + /** + * Main job processing method + * Called by BullMQ for each job + */ + async process(job: Job): Promise { + const { libraryItemId, url, userId, source } = job.data + + this.logger.log( + `Processing job ${job.id} for item ${libraryItemId} (attempt ${job.attemptsMade + 1}/${job.opts.attempts})` + ) + + // Route to appropriate handler based on job name + switch (job.name) { + case JOB_TYPES.FETCH_CONTENT: + return this.handleFetchContent(job) + case JOB_TYPES.PARSE_CONTENT: + return this.handleParseContent(job) + default: + throw new Error(`Unknown job type: ${job.name}`) + } + } + + /** + * Handle fetch-content jobs + */ + private async handleFetchContent( + job: Job + ): Promise { + const { libraryItemId, url, userId } = job.data + const startTime = Date.now() + + try { + // Emit fetch started event + this.eventBus.emitContentFetchStarted({ + eventType: EVENT_NAMES.CONTENT_FETCH_STARTED, + libraryItemId, + url, + jobId: job.id!, + timestamp: new Date(), + }) + + // Update job progress + await job.updateProgress(10) + + // Update library item state to PROCESSING + await this.updateLibraryItemState(libraryItemId, LibraryItemState.PROCESSING) + await job.updateProgress(20) + + // Fetch and process content + const result = await this.fetchContent(url, job) + + if (!result.success) { + throw new Error(result.error || 'Content fetch failed') + } + + await job.updateProgress(70) + + // Save content to database + await this.saveContent(libraryItemId, result) + await job.updateProgress(90) + + // Update library item state to SUCCEEDED + await this.updateLibraryItemState(libraryItemId, LibraryItemState.SUCCEEDED) + await job.updateProgress(100) + + const processingTime = Date.now() - startTime + + // Emit fetch completed event + this.eventBus.emitContentFetchCompleted({ + eventType: EVENT_NAMES.CONTENT_FETCH_COMPLETED, + libraryItemId, + jobId: job.id!, + contentLength: result.content?.length || 0, + processingTime, + timestamp: new Date(), + }) + + this.logger.log( + `Successfully processed job ${job.id} for item ${libraryItemId} in ${processingTime}ms` + ) + + return result + } catch (error) { + const errorMessage = error instanceof Error ? error.message : String(error) + const willRetry = (job.attemptsMade + 1) < (job.opts.attempts || 1) + + this.logger.error( + `Job ${job.id} failed for item ${libraryItemId}: ${errorMessage} ` + + `(attempt ${job.attemptsMade + 1}/${job.opts.attempts}, will retry: ${willRetry})` + ) + + // Update library item state to FAILED if final attempt + if (!willRetry) { + await this.updateLibraryItemState(libraryItemId, LibraryItemState.FAILED) + } + + // Emit fetch failed event + this.eventBus.emitContentFetchFailed({ + eventType: EVENT_NAMES.CONTENT_FETCH_FAILED, + libraryItemId, + jobId: job.id!, + userId, + error: errorMessage, + retryCount: job.attemptsMade + 1, + willRetry, + timestamp: new Date(), + }) + + throw error + } + } + + /** + * Handle parse-content jobs (future implementation) + */ + private async handleParseContent(job: Job): Promise { + this.logger.log(`Parse content job ${job.id} - Not implemented yet`) + // TODO: Implement content parsing in Phase 3 + return { success: true, message: 'Parsing not implemented yet' } + } + + /** + * Fetch content from URL + * TODO: Implement full content fetching in Phase 3 with Puppeteer/handlers + */ + private async fetchContent( + url: string, + job: Job + ): Promise { + this.logger.log(`Fetching content from ${url}`) + + try { + // STUB: For Phase 2, we'll just create a placeholder result + // In Phase 3, this will be replaced with actual Puppeteer/handler logic + + // Simulate network delay + await this.delay(1000) + await job.updateProgress(40) + + // Simulate content processing + await this.delay(1000) + await job.updateProgress(60) + + // Return stub data + return { + success: true, + title: `Content from ${new URL(url).hostname}`, + content: '

This is stub content. Real content fetching will be implemented in Phase 3.

', + contentType: 'text/html', + author: 'Unknown', + publishedDate: new Date(), + } + } catch (error) { + const errorMessage = error instanceof Error ? error.message : String(error) + this.logger.error(`Failed to fetch content from ${url}: ${errorMessage}`) + + return { + success: false, + error: errorMessage, + } + } + } + + /** + * Save processed content to database + */ + private async saveContent( + libraryItemId: string, + result: ContentFetchResult + ): Promise { + this.logger.log(`Saving content for library item ${libraryItemId}`) + + try { + await this.libraryItemRepository.update(libraryItemId, { + title: result.title, + readableContent: result.content, + author: result.author, + publishedAt: result.publishedDate, + siteIcon: result.siteIcon, + thumbnail: result.thumbnail, + }) + + this.logger.log(`Content saved for library item ${libraryItemId}`) + } catch (error) { + const errorMessage = error instanceof Error ? error.message : String(error) + this.logger.error(`Failed to save content for ${libraryItemId}: ${errorMessage}`) + throw error + } + } + + /** + * Update library item state + */ + private async updateLibraryItemState( + libraryItemId: string, + state: LibraryItemState + ): Promise { + try { + await this.libraryItemRepository.update(libraryItemId, { state }) + this.logger.debug(`Updated library item ${libraryItemId} state to ${state}`) + } catch (error) { + const errorMessage = error instanceof Error ? error.message : String(error) + this.logger.error( + `Failed to update state for ${libraryItemId}: ${errorMessage}` + ) + throw error + } + } + + /** + * Utility: Delay for testing + */ + private delay(ms: number): Promise { + return new Promise(resolve => setTimeout(resolve, ms)) + } + + /** + * Worker event handlers + */ + + @OnWorkerEvent('active') + onActive(job: Job) { + this.logger.debug(`Job ${job.id} is now active`) + } + + @OnWorkerEvent('completed') + onCompleted(job: Job) { + this.logger.log(`Job ${job.id} completed successfully`) + } + + @OnWorkerEvent('failed') + onFailed(job: Job, error: Error) { + this.logger.error(`Job ${job.id} failed: ${error.message}`) + } + + @OnWorkerEvent('progress') + onProgress(job: Job, progress: number | object) { + this.logger.debug(`Job ${job.id} progress: ${JSON.stringify(progress)}`) + } +} diff --git a/packages/api-nest/src/queue/queue-health.indicator.spec.ts b/packages/api-nest/src/queue/queue-health.indicator.spec.ts new file mode 100644 index 000000000..bee642d38 --- /dev/null +++ b/packages/api-nest/src/queue/queue-health.indicator.spec.ts @@ -0,0 +1,289 @@ +/** + * QueueHealthIndicator Unit Tests + */ + +import { Test, TestingModule } from '@nestjs/testing' +import { Logger } from '@nestjs/common' +import { getQueueToken } from '@nestjs/bullmq' +import { Queue } from 'bullmq' +import { QueueHealthIndicator } from './queue-health.indicator' +import { QUEUE_NAMES } from './queue.constants' + +type MockQueue = jest.Mocked< + Pick< + Queue, + | 'name' + | 'getWaitingCount' + | 'getActiveCount' + | 'getCompletedCount' + | 'getFailedCount' + | 'getDelayedCount' + | 'isPaused' + > +> + +// Mock logger to suppress console output during tests +const mockLogger = { + log: jest.fn(), + error: jest.fn(), + warn: jest.fn(), + debug: jest.fn(), + verbose: jest.fn(), +} + +describe('QueueHealthIndicator', () => { + let indicator: QueueHealthIndicator + let contentQueue: MockQueue + let notificationQueue: MockQueue + let postProcessingQueue: MockQueue + + beforeEach(async () => { + // Mock healthy queue + const createHealthyQueueMock = (name: string) => ({ + name, + getWaitingCount: jest.fn().mockResolvedValue(10), + getActiveCount: jest.fn().mockResolvedValue(2), + getCompletedCount: jest.fn().mockResolvedValue(100), + getFailedCount: jest.fn().mockResolvedValue(5), + getDelayedCount: jest.fn().mockResolvedValue(0), + isPaused: jest.fn().mockResolvedValue(false), + }) + + contentQueue = createHealthyQueueMock(QUEUE_NAMES.CONTENT_PROCESSING) + notificationQueue = createHealthyQueueMock(QUEUE_NAMES.NOTIFICATIONS) + postProcessingQueue = createHealthyQueueMock(QUEUE_NAMES.POST_PROCESSING) + + const module: TestingModule = await Test.createTestingModule({ + providers: [ + QueueHealthIndicator, + { + provide: getQueueToken(QUEUE_NAMES.CONTENT_PROCESSING), + useValue: contentQueue, + }, + { + provide: getQueueToken(QUEUE_NAMES.NOTIFICATIONS), + useValue: notificationQueue, + }, + { + provide: getQueueToken(QUEUE_NAMES.POST_PROCESSING), + useValue: postProcessingQueue, + }, + ], + }) + .setLogger(mockLogger) + .compile() + + indicator = module.get(QueueHealthIndicator) + }) + + afterEach(() => { + jest.clearAllMocks() + jest.restoreAllMocks() + }) + + describe('initialization', () => { + it('should be defined', () => { + expect(indicator).toBeDefined() + }) + }) + + describe('isHealthy', () => { + it('should return healthy status when all queues are healthy', async () => { + const result = await indicator.isHealthy('queues') + + expect(result).toHaveProperty('queues') + expect(result.queues.status).toBe('up') + expect(result.queues.responseTime).toBeGreaterThanOrEqual(0) + expect(result.queues.queues).toHaveProperty( + QUEUE_NAMES.CONTENT_PROCESSING, + ) + expect(result.queues.queues).toHaveProperty(QUEUE_NAMES.NOTIFICATIONS) + expect(result.queues.queues).toHaveProperty(QUEUE_NAMES.POST_PROCESSING) + }) + + it('should call getWaitingCount on all queues', async () => { + await indicator.isHealthy('queues') + + expect(contentQueue.getWaitingCount).toHaveBeenCalled() + expect(notificationQueue.getWaitingCount).toHaveBeenCalled() + expect(postProcessingQueue.getWaitingCount).toHaveBeenCalled() + }) + + it('should call getActiveCount on all queues', async () => { + await indicator.isHealthy('queues') + + expect(contentQueue.getActiveCount).toHaveBeenCalled() + expect(notificationQueue.getActiveCount).toHaveBeenCalled() + expect(postProcessingQueue.getActiveCount).toHaveBeenCalled() + }) + + it('should return degraded status when a queue is paused', async () => { + contentQueue.isPaused.mockResolvedValue(true) + + const result = await indicator.isHealthy('queues') + + expect(result.queues.status).toBe('degraded') + expect(result.queues.queues[QUEUE_NAMES.CONTENT_PROCESSING].paused).toBe( + true, + ) + expect(result.queues.queues[QUEUE_NAMES.CONTENT_PROCESSING].healthy).toBe( + false, + ) + }) + + it('should return degraded status when too many failed jobs', async () => { + contentQueue.getFailedCount.mockResolvedValue(150) // More than 100 + + const result = await indicator.isHealthy('queues') + + expect(result.queues.status).toBe('degraded') + expect(result.queues.queues[QUEUE_NAMES.CONTENT_PROCESSING].failed).toBe( + 150, + ) + expect(result.queues.queues[QUEUE_NAMES.CONTENT_PROCESSING].healthy).toBe( + false, + ) + }) + + it('should return degraded status when too many waiting jobs', async () => { + contentQueue.getWaitingCount.mockResolvedValue(1500) // More than 1000 + + const result = await indicator.isHealthy('queues') + + expect(result.queues.status).toBe('degraded') + expect(result.queues.queues[QUEUE_NAMES.CONTENT_PROCESSING].waiting).toBe( + 1500, + ) + expect(result.queues.queues[QUEUE_NAMES.CONTENT_PROCESSING].healthy).toBe( + false, + ) + }) + + it('should return degraded status when one queue connection fails', async () => { + contentQueue.getWaitingCount.mockRejectedValue( + new Error('Connection refused'), + ) + + const result = await indicator.isHealthy('queues') + + expect(result.queues.status).toBe('degraded') + expect( + result.queues.queues[QUEUE_NAMES.CONTENT_PROCESSING].connected, + ).toBe(false) + expect(result.queues.queues[QUEUE_NAMES.CONTENT_PROCESSING].healthy).toBe( + false, + ) + }) + + it('should return degraded status when all queues fail', async () => { + contentQueue.getWaitingCount.mockRejectedValue( + new Error('Connection refused'), + ) + notificationQueue.getWaitingCount.mockRejectedValue( + new Error('Connection refused'), + ) + postProcessingQueue.getWaitingCount.mockRejectedValue( + new Error('Connection refused'), + ) + + const result = await indicator.isHealthy('queues') + + expect(result.queues.status).toBe('degraded') + expect( + result.queues.queues[QUEUE_NAMES.CONTENT_PROCESSING].connected, + ).toBe(false) + expect(result.queues.queues[QUEUE_NAMES.NOTIFICATIONS].connected).toBe( + false, + ) + expect(result.queues.queues[QUEUE_NAMES.POST_PROCESSING].connected).toBe( + false, + ) + }) + }) + + describe('getDetailedHealth', () => { + it('should return detailed health metrics for all queues', async () => { + const result = await indicator.getDetailedHealth() + + expect(result).toHaveProperty('queues') + expect(result).toHaveProperty('timestamp') + expect(result.queues).toHaveProperty(QUEUE_NAMES.CONTENT_PROCESSING) + expect(result.queues).toHaveProperty(QUEUE_NAMES.NOTIFICATIONS) + expect(result.queues).toHaveProperty(QUEUE_NAMES.POST_PROCESSING) + + // Check content queue metrics + const contentMetrics = result.queues[QUEUE_NAMES.CONTENT_PROCESSING] + expect(contentMetrics).toHaveProperty('healthy', true) + expect(contentMetrics).toHaveProperty('connected', true) + expect(contentMetrics).toHaveProperty('waiting', 10) + expect(contentMetrics).toHaveProperty('active', 2) + expect(contentMetrics).toHaveProperty('completed', 100) + expect(contentMetrics).toHaveProperty('failed', 5) + expect(contentMetrics).toHaveProperty('delayed', 0) + expect(contentMetrics).toHaveProperty('paused', false) + }) + + it('should return metrics with disconnected status when queue metrics fail', async () => { + contentQueue.getWaitingCount.mockRejectedValue( + new Error('Connection refused'), + ) + + const result = await indicator.getDetailedHealth() + + expect(result.queues[QUEUE_NAMES.CONTENT_PROCESSING].connected).toBe( + false, + ) + expect(result.queues[QUEUE_NAMES.CONTENT_PROCESSING].healthy).toBe(false) + expect(result.queues[QUEUE_NAMES.NOTIFICATIONS].connected).toBe(true) + expect(result.queues[QUEUE_NAMES.POST_PROCESSING].connected).toBe(true) + }) + }) + + describe('edge cases', () => { + it('should handle queue with exactly 100 failed jobs as healthy', async () => { + contentQueue.getFailedCount.mockResolvedValue(99) + + const result = await indicator.isHealthy('queues') + + expect(result.queues.status).toBe('up') + expect(result.queues.queues[QUEUE_NAMES.CONTENT_PROCESSING].healthy).toBe( + true, + ) + }) + + it('should handle queue with exactly 1000 waiting jobs as healthy', async () => { + contentQueue.getWaitingCount.mockResolvedValue(999) + + const result = await indicator.isHealthy('queues') + + expect(result.queues.status).toBe('up') + expect(result.queues.queues[QUEUE_NAMES.CONTENT_PROCESSING].healthy).toBe( + true, + ) + }) + + it('should handle all queues having zero jobs', async () => { + contentQueue.getWaitingCount.mockResolvedValue(0) + contentQueue.getActiveCount.mockResolvedValue(0) + contentQueue.getCompletedCount.mockResolvedValue(0) + contentQueue.getFailedCount.mockResolvedValue(0) + contentQueue.getDelayedCount.mockResolvedValue(0) + + notificationQueue.getWaitingCount.mockResolvedValue(0) + notificationQueue.getActiveCount.mockResolvedValue(0) + notificationQueue.getCompletedCount.mockResolvedValue(0) + notificationQueue.getFailedCount.mockResolvedValue(0) + notificationQueue.getDelayedCount.mockResolvedValue(0) + + postProcessingQueue.getWaitingCount.mockResolvedValue(0) + postProcessingQueue.getActiveCount.mockResolvedValue(0) + postProcessingQueue.getCompletedCount.mockResolvedValue(0) + postProcessingQueue.getFailedCount.mockResolvedValue(0) + postProcessingQueue.getDelayedCount.mockResolvedValue(0) + + const result = await indicator.isHealthy('queues') + + expect(result.queues.status).toBe('up') + }) + }) +}) diff --git a/packages/api-nest/src/queue/queue-health.indicator.ts b/packages/api-nest/src/queue/queue-health.indicator.ts new file mode 100644 index 000000000..c209fb315 --- /dev/null +++ b/packages/api-nest/src/queue/queue-health.indicator.ts @@ -0,0 +1,182 @@ +/** + * QueueHealthIndicator - Health check for BullMQ queues + * + * Provides health status for all configured queues including: + * - Connection status + * - Queue metrics (waiting, active, completed, failed jobs) + * - Worker status + */ + +import { Injectable, Logger } from '@nestjs/common' +import { InjectQueue } from '@nestjs/bullmq' +import { Queue } from 'bullmq' +import { + HealthIndicator, + HealthIndicatorResult, + HealthCheckError, +} from '@nestjs/terminus' +import { QUEUE_NAMES } from './queue.constants' + +@Injectable() +export class QueueHealthIndicator extends HealthIndicator { + private readonly logger = new Logger(QueueHealthIndicator.name) + + constructor( + @InjectQueue(QUEUE_NAMES.CONTENT_PROCESSING) + private readonly contentQueue: Queue, + @InjectQueue(QUEUE_NAMES.NOTIFICATIONS) + private readonly notificationQueue: Queue, + @InjectQueue(QUEUE_NAMES.POST_PROCESSING) + private readonly postProcessingQueue: Queue + ) { + super() + } + + /** + * Check health of all queues + */ + async isHealthy(key: string): Promise { + const startTime = Date.now() + + try { + // Get metrics for all queues + const [contentMetrics, notificationMetrics, postProcessingMetrics] = + await Promise.all([ + this.getQueueMetrics(this.contentQueue), + this.getQueueMetrics(this.notificationQueue), + this.getQueueMetrics(this.postProcessingQueue), + ]) + + const responseTime = Date.now() - startTime + + // Check if any queue is unhealthy + const allHealthy = + contentMetrics.healthy && + notificationMetrics.healthy && + postProcessingMetrics.healthy + + const details = { + status: allHealthy ? 'up' : 'degraded', + responseTime, + queues: { + [QUEUE_NAMES.CONTENT_PROCESSING]: contentMetrics, + [QUEUE_NAMES.NOTIFICATIONS]: notificationMetrics, + [QUEUE_NAMES.POST_PROCESSING]: postProcessingMetrics, + }, + } + + if (!allHealthy) { + this.logger.warn('Queue health check shows degraded status', details) + } + + const result = this.getStatus(key, allHealthy, details) + return result + } catch (error) { + const responseTime = Date.now() - startTime + const errorMessage = + error instanceof Error ? error.message : 'Unknown error' + + this.logger.error('Queue health check failed', { + error: errorMessage, + responseTime, + }) + + const result = this.getStatus(key, false, { + status: 'down', + message: errorMessage, + responseTime, + }) + + throw new HealthCheckError('Queue health check failed', result) + } + } + + /** + * Get detailed metrics for a single queue + */ + private async getQueueMetrics(queue: Queue): Promise<{ + healthy: boolean + connected: boolean + waiting: number + active: number + completed: number + failed: number + delayed: number + paused: boolean + }> { + try { + // Test Redis connection by getting job counts + const [waiting, active, completed, failed, delayed] = await Promise.all([ + queue.getWaitingCount(), + queue.getActiveCount(), + queue.getCompletedCount(), + queue.getFailedCount(), + queue.getDelayedCount(), + ]) + + const isPaused = await queue.isPaused() + + // Consider unhealthy if: + // - Queue is paused + // - Too many failed jobs (more than 100) + // - Too many waiting jobs (more than 1000) + const healthy = + !isPaused && + failed < 100 && + waiting < 1000 + + return { + healthy, + connected: true, + waiting, + active, + completed, + failed, + delayed, + paused: isPaused, + } + } catch (error) { + this.logger.error( + `Failed to get metrics for queue ${queue.name}`, + error instanceof Error ? error.message : error + ) + + return { + healthy: false, + connected: false, + waiting: 0, + active: 0, + completed: 0, + failed: 0, + delayed: 0, + paused: false, + } + } + } + + /** + * Get detailed health for monitoring endpoints + */ + async getDetailedHealth(): Promise> { + try { + const [contentMetrics, notificationMetrics, postProcessingMetrics] = + await Promise.all([ + this.getQueueMetrics(this.contentQueue), + this.getQueueMetrics(this.notificationQueue), + this.getQueueMetrics(this.postProcessingQueue), + ]) + + return { + queues: { + [QUEUE_NAMES.CONTENT_PROCESSING]: contentMetrics, + [QUEUE_NAMES.NOTIFICATIONS]: notificationMetrics, + [QUEUE_NAMES.POST_PROCESSING]: postProcessingMetrics, + }, + timestamp: new Date().toISOString(), + } + } catch (error) { + this.logger.error('Failed to get detailed health', error) + throw error + } + } +} diff --git a/packages/api-nest/src/queue/queue.constants.ts b/packages/api-nest/src/queue/queue.constants.ts new file mode 100644 index 000000000..044c0d791 --- /dev/null +++ b/packages/api-nest/src/queue/queue.constants.ts @@ -0,0 +1,116 @@ +/** + * Queue Configuration Constants + * + * Centralized constants for queue names, job types, and priorities + * to avoid magic strings throughout the codebase. + */ + +/** + * Queue Names - Define all queues used in the system + */ +export const QUEUE_NAMES = { + CONTENT_PROCESSING: 'content-processing', + NOTIFICATIONS: 'notifications', + POST_PROCESSING: 'post-processing', +} as const + +/** + * Job Types - Define all job types for each queue + */ +export const JOB_TYPES = { + // Content Processing Queue + FETCH_CONTENT: 'fetch-content', + PARSE_CONTENT: 'parse-content', + EXTRACT_METADATA: 'extract-metadata', + + // Notifications Queue + SEND_NOTIFICATION: 'send-notification', + SEND_EMAIL: 'send-email', + + // Post Processing Queue + UPDATE_SEARCH_INDEX: 'update-search-index', + GENERATE_THUMBNAIL: 'generate-thumbnail', +} as const + +/** + * Job Priority Levels + * Lower numbers = higher priority + */ +export const JOB_PRIORITY = { + CRITICAL: 1, + HIGH: 5, + NORMAL: 10, + LOW: 20, +} as const + +/** + * Job Configuration Defaults + */ +export const JOB_CONFIG = { + // Retry configuration + RETRY_ATTEMPTS: 3, + RETRY_BACKOFF_TYPE: 'exponential' as const, + RETRY_BACKOFF_DELAY: 2000, // 2 seconds initial delay + + // Rate limiting + RATE_LIMIT_MAX: 5, // Max jobs per time window + RATE_LIMIT_DURATION: 60000, // 1 minute window + + // Job timeouts + TIMEOUT_CONTENT_FETCH: 60000, // 60 seconds + TIMEOUT_PARSE_CONTENT: 30000, // 30 seconds + TIMEOUT_NOTIFICATION: 10000, // 10 seconds + + // Worker concurrency + WORKER_CONCURRENCY: 4, +} as const + +/** + * Redis Configuration + */ +export const REDIS_CONFIG = { + // Connection + HOST: process.env.REDIS_HOST || 'localhost', + PORT: parseInt(process.env.REDIS_PORT || '6379', 10), + PASSWORD: process.env.REDIS_PASSWORD, + + // Sentinel configuration (for production) + SENTINEL_NAME: process.env.REDIS_SENTINEL_NAME || 'mymaster', + SENTINELS: process.env.REDIS_SENTINELS + ? process.env.REDIS_SENTINELS.split(',').map(s => { + const [host, port] = s.split(':') + return { host, port: parseInt(port, 10) } + }) + : undefined, + + // Connection pool + // BullMQ requires null for blocking operations (BRPOPLPUSH, etc.) + MAX_RETRIES_PER_REQUEST: null, + ENABLE_READY_CHECK: true, + ENABLE_OFFLINE_QUEUE: true, + + // Key prefixes for different environments + KEY_PREFIX: process.env.NODE_ENV === 'test' + ? 'omnivore:test:' + : 'omnivore:', +} as const + +/** + * Queue State Values + */ +export const QUEUE_STATE = { + PENDING: 'pending', + PROCESSING: 'processing', + COMPLETED: 'completed', + FAILED: 'failed', + DELAYED: 'delayed', + WAITING: 'waiting', +} as const + +/** + * Type exports for type-safe usage + */ +export type QueueName = typeof QUEUE_NAMES[keyof typeof QUEUE_NAMES] +export type JobType = typeof JOB_TYPES[keyof typeof JOB_TYPES] +export type JobPriority = typeof JOB_PRIORITY[keyof typeof JOB_PRIORITY] +export type QueueState = typeof QUEUE_STATE[keyof typeof QUEUE_STATE] diff --git a/packages/api-nest/src/queue/queue.module.ts b/packages/api-nest/src/queue/queue.module.ts new file mode 100644 index 000000000..48e511e85 --- /dev/null +++ b/packages/api-nest/src/queue/queue.module.ts @@ -0,0 +1,104 @@ +/** + * QueueModule - BullMQ Queue Configuration + * + * Sets up BullMQ with Redis Sentinel support for background job processing. + * Configures queues for content processing, notifications, and post-processing. + */ + +import { Module } from '@nestjs/common' +import { BullModule } from '@nestjs/bullmq' +import { ConfigModule, ConfigService } from '@nestjs/config' +import { TypeOrmModule } from '@nestjs/typeorm' +import { QUEUE_NAMES, REDIS_CONFIG } from './queue.constants' +import { EventBusService } from './event-bus.service' +import { QueueHealthIndicator } from './queue-health.indicator' +import { ContentProcessorService } from './processors/content-processor.service' +import { LibraryItemEntity } from '../library/entities/library-item.entity' + +@Module({ + imports: [ + ConfigModule, + TypeOrmModule.forFeature([LibraryItemEntity]), + // Register BullMQ with Redis Sentinel configuration + BullModule.forRootAsync({ + imports: [ConfigModule], + inject: [ConfigService], + useFactory: (configService: ConfigService) => { + // Check if we're using Sentinel (production) or standalone Redis (development) + const useSentinel = + configService.get('NODE_ENV') === 'production' && + REDIS_CONFIG.SENTINELS + + return { + connection: useSentinel + ? { + // Sentinel configuration (production) + sentinels: REDIS_CONFIG.SENTINELS, + name: REDIS_CONFIG.SENTINEL_NAME, + password: REDIS_CONFIG.PASSWORD, + maxRetriesPerRequest: REDIS_CONFIG.MAX_RETRIES_PER_REQUEST, + enableReadyCheck: REDIS_CONFIG.ENABLE_READY_CHECK, + enableOfflineQueue: REDIS_CONFIG.ENABLE_OFFLINE_QUEUE, + } + : { + // Standalone Redis configuration (development) + host: REDIS_CONFIG.HOST, + port: REDIS_CONFIG.PORT, + password: REDIS_CONFIG.PASSWORD, + maxRetriesPerRequest: REDIS_CONFIG.MAX_RETRIES_PER_REQUEST, + enableReadyCheck: REDIS_CONFIG.ENABLE_READY_CHECK, + enableOfflineQueue: REDIS_CONFIG.ENABLE_OFFLINE_QUEUE, + }, + prefix: REDIS_CONFIG.KEY_PREFIX, + defaultJobOptions: { + removeOnComplete: { + age: 86400, // Keep completed jobs for 24 hours + count: 1000, // Keep last 1000 completed jobs + }, + removeOnFail: { + age: 604800, // Keep failed jobs for 7 days + count: 5000, // Keep last 5000 failed jobs + }, + }, + } + }, + }), + + // Register individual queues + BullModule.registerQueue( + { + name: QUEUE_NAMES.CONTENT_PROCESSING, + defaultJobOptions: { + attempts: 3, + backoff: { + type: 'exponential', + delay: 2000, + }, + }, + }, + { + name: QUEUE_NAMES.NOTIFICATIONS, + defaultJobOptions: { + attempts: 5, + backoff: { + type: 'exponential', + delay: 1000, + }, + }, + }, + { + name: QUEUE_NAMES.POST_PROCESSING, + defaultJobOptions: { + attempts: 3, + backoff: { + type: 'exponential', + delay: 3000, + }, + }, + } + ), + ], + providers: [EventBusService, QueueHealthIndicator, ContentProcessorService], + exports: [BullModule, EventBusService, QueueHealthIndicator], +}) +export class QueueModule {} diff --git a/packages/api-nest/test/jest-e2e.json b/packages/api-nest/test/jest-e2e.json index 9f2a306b4..82f4704e6 100644 --- a/packages/api-nest/test/jest-e2e.json +++ b/packages/api-nest/test/jest-e2e.json @@ -6,6 +6,9 @@ "transform": { "^.+\\.(t|j)s$": "ts-jest" }, + "transformIgnorePatterns": [ + "node_modules/(?!(bullmq|msgpackr)/)" + ], "moduleDirectories": [ "node_modules", "../node_modules", diff --git a/yarn.lock b/yarn.lock index 9bdfbfb2f..9a77c3044 100644 --- a/yarn.lock +++ b/yarn.lock @@ -4577,6 +4577,21 @@ lodash.omit "4.5.0" tslib "2.8.1" +"@nestjs/bull-shared@^10.2.3": + version "10.2.3" + resolved "https://registry.yarnpkg.com/@nestjs/bull-shared/-/bull-shared-10.2.3.tgz#074749b00683e2ca10f828af21c49d4d17bd35d9" + integrity sha512-XcgAjNOgq6b5DVCytxhR5BKiwWo7hsusVeyE7sfFnlXRHeEtIuC2hYWBr/ZAtvL/RH0/O0tqtq0rVl972nbhJw== + dependencies: + tslib "2.8.1" + +"@nestjs/bullmq@^10.0.0": + version "10.2.3" + resolved "https://registry.yarnpkg.com/@nestjs/bullmq/-/bullmq-10.2.3.tgz#1b257bdcdd139e73c39f20b7ae30660597fed3be" + integrity sha512-Lo4W5kWD61/246Y6H70RNgV73ybfRbZyKKS4CBRDaMELpxgt89O+EgYZUB4pdoNrWH16rKcaT0AoVsB/iDztKg== + dependencies: + "@nestjs/bull-shared" "^10.2.3" + tslib "2.8.1" + "@nestjs/cli@^10.0.0": version "10.4.9" resolved "https://registry.npmjs.org/@nestjs/cli/-/cli-10.4.9.tgz" @@ -13214,6 +13229,19 @@ builtins@^5.0.0: dependencies: semver "^7.0.0" +bullmq@^5.0.0: + version "5.61.0" + resolved "https://registry.yarnpkg.com/bullmq/-/bullmq-5.61.0.tgz#3c5c7bc733acea3e29a93069874030a8646f7c8e" + integrity sha512-khaTjc1JnzaYFl4FrUtsSsqugAW/urRrcZ9Q0ZE+REAw8W+gkHFqxbGlutOu6q7j7n91wibVaaNlOUMdiEvoSQ== + dependencies: + cron-parser "^4.9.0" + ioredis "^5.4.1" + msgpackr "^1.11.2" + node-abort-controller "^3.1.1" + semver "^7.5.4" + tslib "^2.0.0" + uuid "^11.1.0" + bullmq@^5.1.1, bullmq@^5.1.4, bullmq@^5.22.0, bullmq@^5.51.1: version "5.56.9" resolved "https://registry.npmjs.org/bullmq/-/bullmq-5.56.9.tgz"