feat(queue): Integrate BullMQ for background processing and health checks

- Implement QueueModule for managing BullMQ queues including content processing, notifications, and post-processing.
- Add EventBusService for type-safe event emission and handling.
- Create QueueHealthIndicator to monitor the health of queues with detailed metrics.
- Develop ContentProcessorService to process jobs from the content-processing queue, fetching and parsing web content.
- Enhance SaveUrlInput to include source of the save request.
- Update health checks to include queue status in the health controller.

This commit lays the groundwork for robust background processing and monitoring, improving the overall architecture of the application.

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"
     ],

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!
 }

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,
   ],

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

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],
 })

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'
 }

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],

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<LibraryItemEntity>,
     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
   }

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<Pick<Queue, 'name' | 'add'>>
+
+// 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>(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()
+    })
+  })
+})

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()
+  }
+}

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<string, unknown>
+}
+
+/**
+ * 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<string, unknown>
+}
+
+/**
+ * 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]

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<Repository<LibraryItemEntity>>
+  let eventBus: jest.Mocked<EventBusService>
+
+  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>(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: '<p>Test content</p>',
+        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: '<p>Test content</p>',
+        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: '<p>Test content</p>',
+      }
+
+      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<FetchContentJobData> {
+  return {
+    id: 'job-123',
+    name,
+    data,
+    attemptsMade: opts.attemptsMade || 0,
+    opts: {
+      attempts: opts.attempts || 3,
+    },
+    updateProgress: jest.fn().mockResolvedValue(undefined),
+  } as any
+}

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<LibraryItemEntity>,
+    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<FetchContentJobData, any, string>): Promise<any> {
+    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<FetchContentJobData>
+  ): Promise<ContentFetchResult> {
+    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<FetchContentJobData>): Promise<any> {
+    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<FetchContentJobData>
+  ): Promise<ContentFetchResult> {
+    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: '<p>This is stub content. Real content fetching will be implemented in Phase 3.</p>',
+        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<void> {
+    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<void> {
+    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<void> {
+    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)}`)
+  }
+}

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>(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')
+    })
+  })
+})

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<HealthIndicatorResult> {
+    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<Record<string, any>> {
+    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
+    }
+  }
+}

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]

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<string>('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 {}

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",

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"