feat: wip - content consolidation

.cursor/rules/architecture-analysis-and-consolidation-strategy.md

@@ -0,0 +1,233 @@
+---
+alwaysApply: true
+---
++# Omnivore Architecture Analysis and Consolidation Strategy
++
++## Current Architecture Overview
++
++Based on the codebase analysis, Omnivore currently consists of 25+ microservices:
++
++### Core Services
++- **web**: Next.js frontend application
++- **api**: GraphQL API server
++- **content-fetch**: Content fetching and processing service
++
++### Processing Services
++- **puppeteer-parse**: Browser-based content parsing
++- **pdf-handler**: PDF processing
++- **thumbnail-handler**: Image thumbnail generation
++- **readabilityjs**: Content readability extraction
++- **text-to-speech**: TTS functionality
++
++### Integration Services
++- **rss-handler**: RSS feed processing
++- **inbound-email-handler**: Email ingestion
++- **imap-mail-watcher**: IMAP email monitoring
++- **local-mail-watcher**: Local mail processing
++
++### Queue and Background Services
++- **queue-manager**: Job queue management
++- **export-handler**: Export functionality
++- **import-handler**: Import functionality
++- **integration-handler**: Third-party integrations
++- **rule-handler**: Rule processing
++
++### Infrastructure
++- PostgreSQL with pgvector
++- Redis for queuing and caching
++- MinIO/S3 for object storage
++- Nginx for reverse proxy
++
++## Consolidation Opportunities
++
++### 1. Content Processing Mega-Service
++Combine these services into a single "content-processor":
++- content-fetch
++- puppeteer-parse
++- pdf-handler
++- thumbnail-handler
++- readabilityjs
++
++**Benefits:**
++- Reduced inter-service communication
++- Shared resource pool for Chromium instances
++- Unified content pipeline
++- Single deployment unit
++
++### 2. Communication Services
++Merge email-related services:
++- inbound-email-handler
++- imap-mail-watcher
++- local-mail-watcher
++
++**Benefits:**
++- Simplified email configuration
++- Shared authentication logic
++- Reduced complexity for self-hosters
++
++### 3. Background Job Processor
++Consolidate queue processing:
++- queue-manager
++- export-handler
++- import-handler
++- integration-handler
++- rule-handler
++- rss-handler
++
++**Benefits:**
++- Single worker pool
++- Unified job scheduling
++- Better resource utilization
++
++## Proposed Simplified Architecture
++
++### Option 1: Three-Service Architecture
++1. **omnivore-web**: Frontend application
++2. **omnivore-api**: API + all background processing
++3. **omnivore-content**: All content processing services
++
++### Option 2: Monolithic with Optional Services
++1. **omnivore-core**: Everything in one service
++2. **omnivore-ai** (optional): AI/ML features
++3. **omnivore-email** (optional): Email processing
++
++### Option 3: Hybrid Approach (Recommended)
++1. **omnivore**: Main application (web + api + basic processing)
++2. **omnivore-worker**: Heavy processing (puppeteer, PDF, etc.)
++3. **omnivore-edge** (optional): CDN/proxy functionality
++
++## Implementation Strategy
++
++### Phase 1: Internal Consolidation (Weeks 1-2)
++- Move all handlers into the API service as modules
++- Use feature flags to enable/disable functionality
++- Maintain backward compatibility
++
++### Phase 2: Docker Optimization (Weeks 3-4)
++- Create multi-stage Dockerfile
++- Implement build-time feature selection
++- Optimize image sizes
++
++### Phase 3: Configuration System (Weeks 5-6)
++- Environment-based feature toggles
++- Self-hosted vs. cloud profiles
++- Plugin architecture for extensions
++
++### Phase 4: Testing and Migration (Weeks 7-8)
++- Comprehensive testing suite
++- Migration scripts
++- Documentation updates
++
++## Cost Optimization Strategies
++
++### For Self-Hosters
++- Single container deployment option
++- SQLite support for small installations
++- Built-in search instead of Elasticsearch
++- Local file storage instead of S3
++
++### For Cloud Deployment
++- Serverless functions for infrequent tasks
++- Auto-scaling based on queue depth
++- Shared resource pools
++- Edge caching for static content
++
++## Technical Recommendations
++
++### 1. Use Feature Flags
++```yaml
++features:
++  ai_summaries: ${ENABLE_AI_FEATURES:-false}
++  email_ingestion: ${ENABLE_EMAIL:-false}
++  advanced_search: ${ENABLE_ELASTICSEARCH:-false}
++  pdf_processing: ${ENABLE_PDF:-true}
++```
++
++### 2. Plugin Architecture
++```typescript
++interface OmnivorePlugin {
++  name: string
++  version: string
++  init(context: PluginContext): Promise<void>
++  handlers: {
++    [event: string]: Handler
++  }
++}
++```
++
++### 3. Deployment Profiles
++```yaml
++profiles:
++  minimal:
++    services: [core]
++    features: [basic_reading, highlighting]
++  
++  standard:
++    services: [core, worker]
++    features: [all_content_types, search]
++  
++  enterprise:
++    services: [core, worker, ai, analytics]
++    features: [all]
++```
++
++## Migration Path
++
++### Step 1: Code Consolidation
++- Move services into monorepo packages
++- Share common dependencies
++- Unified build process
++
++### Step 2: Runtime Consolidation
++- Services run as threads/processes
++- Shared memory and resources
++- Internal API calls become function calls
++
++### Step 3: Deployment Consolidation
++- Single Docker image with feature flags
++- Compose file for multi-container option
++- Kubernetes manifests for scale
++
++## Success Metrics
++
++### Performance
++- 50% reduction in cold start time
++- 75% reduction in memory usage (self-hosted)
++- 90% reduction in inter-service latency
++
++### Cost
++- 80% reduction in cloud hosting costs
++- Single-digit dollar monthly cost for small deployments
++- Pay-per-use model for expensive features
++
++### Developer Experience
++- Single command to run everything locally
++- 5-minute setup time
++- Clear contribution guidelines
++
++## Risk Mitigation
++
++### 1. Feature Parity
++- Comprehensive test suite before consolidation
++- Feature flags for gradual rollout
++- Ability to run services separately if needed
++
++### 2. Performance
++- Benchmark before and after
++- Load testing for concurrent users
++- Resource monitoring and limits
++
++### 3. Backward Compatibility
++- API versioning
++- Migration tools for existing deployments
++- Clear upgrade path documentation
++
++## Next Steps
++
++1. Create proof of concept for service consolidation
++2. Benchmark performance impact
++3. Design plugin architecture
++4. Plan migration timeline
++5. Engage community for feedback
++
++This architecture will dramatically simplify Omnivore while maintaining its powerful features and open-source nature.

package.json

@@ -36,7 +36,7 @@
     "typescript": "5.7.3"
   },
   "volta": {
-    "node": "18.16.1",
+    "node": "22.11.0",
     "yarn": "1.22.19"
   }
 }

packages/api/jest.config.ts

@@ -0,0 +1,14 @@
+import { JestConfigWithTsJest } from 'ts-jest'
+
+const jestConfig: JestConfigWithTsJest = {
+  preset: 'ts-jest',
+  testEnvironment: 'node',
+  setupFiles: ['./jest.setup.ts'],
+  roots: ['<rootDir>/src'],
+  moduleNameMapper: {
+    '^@/(.*)$': '<rootDir>/src/$1',
+  },
+  testTimeout: 15000,
+}
+
+export default jestConfig

packages/api/jest.setup.ts

@@ -0,0 +1,67 @@
+// Mock problematic ESM modules
+jest.mock('axios')
+
+// Mock BullMQ with stateful Worker
+jest.mock('bullmq', () => ({
+  Queue: jest.fn().mockImplementation((name: string) => ({
+    add: jest.fn().mockResolvedValue({
+      id: `mock-job-${Date.now()}`,
+      name: 'mock-job',
+      data: {},
+    }),
+    waitUntilReady: jest.fn().mockResolvedValue(undefined),
+    close: jest.fn().mockResolvedValue(undefined),
+  })),
+  Worker: jest
+    .fn()
+    .mockImplementation(
+      (
+        queueName: string,
+        processor: (job: any) => Promise<any>,
+        opts?: any
+      ) => {
+        let running = true
+
+        return {
+          on: jest.fn(),
+          isRunning: jest.fn(() => running),
+          close: jest.fn().mockImplementation(() => {
+            running = false
+            return undefined
+          }),
+          concurrency: opts?.concurrency || 2,
+        }
+      }
+    ),
+}))
+
+// Mock logger to avoid import issues
+jest.mock('./src/utils/logger', () => ({
+  logger: {
+    info: jest.fn(),
+    warn: jest.fn(),
+    error: jest.fn(),
+    child: jest.fn(() => ({
+      info: jest.fn(),
+      warn: jest.fn(),
+      error: jest.fn(),
+    })),
+  },
+}))
+
+// Mock Redis
+jest.mock('./src/redis_data_source', () => ({
+  redisDataSource: {
+    isInitialized: true,
+    workerRedisClient: {
+      options: {
+        host: 'localhost',
+        port: 6379,
+        password: undefined,
+        db: 0,
+      },
+    },
+    initialize: jest.fn().mockResolvedValue(undefined),
+    shutdown: jest.fn().mockResolvedValue(undefined),
+  },
+}))

packages/api/package.json

@@ -13,7 +13,10 @@
     "lint": "eslint src --ext ts,js,tsx,jsx",
     "lint:fix": "eslint src --fix --ext ts,js,tsx,jsx",
     "test:typecheck": "tsc --noEmit",
-    "test": "nyc mocha -r ts-node/register --config mocha-config.json"
+    "test": "jest",
+    "test:watch": "jest --watch",
+    "test:coverage": "jest --coverage",
+    "test:debug": "jest --detectOpenHandles --forceExit"
   },
   "dependencies": {
     "@aws-sdk/client-s3": "^3.787.0",
@@ -89,6 +92,7 @@
     "image-size": "^2.0.2",
     "intercom-client": "^6.2.0",
     "ioredis": "^5.6.1",
+    "jest": "^29.7.0",
     "jsonwebtoken": "^9.0.2",
     "jwks-rsa": "^3.2.0",
     "langchain": "^0.3.21",
@@ -129,6 +133,7 @@
   },
   "devDependencies": {
     "@istanbuljs/nyc-config-typescript": "^1.0.2",
+    "@jest/types": "^30.0.5",
     "@types/addressparser": "^1.0.3",
     "@types/analytics-node": "^3.1.14",
     "@types/archiver": "^6.0.3",
@@ -146,6 +151,7 @@
     "@types/graphql-fields": "^1.3.9",
     "@types/highlightjs": "^9.12.6",
     "@types/intercom-client": "^3.0.0",
+    "@types/jest": "^30.0.0",
     "@types/jsonwebtoken": "^9.0.9",
     "@types/lodash": "^4.17.16",
     "@types/luxon": "^3.6.2",
@@ -163,17 +169,16 @@
     "@types/urlsafe-base64": "^1.0.31",
     "@types/uuid": "^10.0.0",
     "@types/voca": "^1.4.6",
-    "chai": "^5.2.0",
     "chai-as-promised": "^8.0.1",
     "chai-string": "^1.6.0",
     "circular-dependency-plugin": "^5.2.2",
-    "mocha": "^11.1.0",
+    "jest": "^29.7.0",
     "mocha-unfunk-reporter": "^0.4.0",
     "nock": "^14.0.3",
     "nyc": "^17.1.0",
     "postgrator": "^8.0.0",
-    "sinon": "^20.0.0",
     "sinon-chai": "^4.0.0",
+    "ts-jest": "^29.1.1",
     "ts-node-dev": "^2.0.0",
     "typescript": "5.8.3"
   },

packages/api/src/events/content/content-save-event.test.ts

@@ -0,0 +1,78 @@
+import {
+  ContentSaveRequestedEvent,
+  ContentType,
+} from '../content/content-save-event'
+describe('ContentSaveRequestedEvent', () => {
+  const validEventData = {
+    userId: 'user-123',
+    libraryItemId: 'item-123',
+    url: 'https://example.com/article',
+    contentType: ContentType.HTML,
+    metadata: {
+      source: 'test',
+      savedAt: new Date().toISOString(),
+    },
+  }
+
+  describe('validation', () => {
+    it('should create valid event', () => {
+      expect(() => new ContentSaveRequestedEvent(validEventData)).not.toThrow()
+    })
+
+    it('should reject missing userId', () => {
+      const invalidData = { ...validEventData, userId: '' }
+      expect(() => new ContentSaveRequestedEvent(invalidData)).toThrow(
+        'userId is required'
+      )
+    })
+
+    it('should reject missing libraryItemId', () => {
+      const invalidData = { ...validEventData, libraryItemId: '' }
+      expect(() => new ContentSaveRequestedEvent(invalidData)).toThrow(
+        'libraryItemId is required'
+      )
+    })
+
+    it('should reject invalid URL', () => {
+      const invalidData = { ...validEventData, url: 'not-a-url' }
+      expect(() => new ContentSaveRequestedEvent(invalidData)).toThrow(
+        'Invalid URL format'
+      )
+    })
+
+    it('should reject invalid date', () => {
+      const invalidData = {
+        ...validEventData,
+        metadata: { ...validEventData.metadata, savedAt: 'not-a-date' },
+      }
+      expect(() => new ContentSaveRequestedEvent(invalidData)).toThrow(
+        'Invalid savedAt date format'
+      )
+    })
+  })
+
+  describe('serialization', () => {
+    it('should serialize and deserialize correctly', () => {
+      const event = new ContentSaveRequestedEvent(validEventData)
+      const serialized = event.serialize()
+      const parsed = JSON.parse(serialized)
+
+      expect(parsed.eventType).toBe('CONTENT_SAVE_REQUESTED')
+      expect(parsed.userId).toBe(validEventData.userId)
+      expect(parsed.url).toBe(validEventData.url)
+      expect(parsed.timestamp).toBeDefined()
+    })
+  })
+
+  describe('getters', () => {
+    it('should provide correct getter values', () => {
+      const event = new ContentSaveRequestedEvent(validEventData)
+
+      expect(event.userId).toBe(validEventData.userId)
+      expect(event.libraryItemId).toBe(validEventData.libraryItemId)
+      expect(event.url).toBe(validEventData.url)
+      expect(event.contentType).toBe(validEventData.contentType)
+      expect(event.metadata).toEqual(validEventData.metadata)
+    })
+  })
+})

packages/api/src/events/content/content-save-event.ts

@@ -0,0 +1,92 @@
+import { BaseEvent } from '../event-manager'
+
+export interface ContentSaveRequestedEventData {
+  userId: string
+  libraryItemId: string
+  url: string
+  contentType: ContentType
+  metadata: {
+    labels?: string[]
+    folder?: string
+    source: string
+    savedAt: string
+    publishedAt?: string
+  }
+}
+
+export enum ContentType {
+  HTML = 'html',
+  PDF = 'pdf',
+  EMAIL = 'email',
+  RSS = 'rss',
+  YOUTUBE = 'youtube',
+}
+
+export enum EventType {
+  CONTENT_SAVE_REQUESTED = 'CONTENT_SAVE_REQUESTED',
+  CONTENT_PROCESSING_STARTED = 'CONTENT_PROCESSING_STARTED',
+  CONTENT_PROCESSING_COMPLETED = 'CONTENT_PROCESSING_COMPLETED',
+  CONTENT_PROCESSING_FAILED = 'CONTENT_PROCESSING_FAILED',
+}
+
+export class ContentSaveRequestedEvent {
+  public readonly eventType = EventType.CONTENT_SAVE_REQUESTED
+
+  constructor(public readonly data: ContentSaveRequestedEventData) {
+    this.validate()
+  }
+
+  serialize(): string {
+    return JSON.stringify({
+      eventType: this.eventType,
+      timestamp: new Date().toISOString(),
+      ...this.data,
+    })
+  }
+
+  protected validate(): void {
+    const { userId, libraryItemId, url, contentType, metadata } = this.data
+
+    if (!userId?.trim()) throw new Error('userId is required')
+    if (!libraryItemId?.trim()) throw new Error('libraryItemId is required')
+    if (!url?.trim()) throw new Error('url is required')
+    if (!contentType) throw new Error('contentType is required')
+    if (!metadata?.source?.trim())
+      throw new Error('metadata.source is required')
+    if (!metadata?.savedAt?.trim())
+      throw new Error('metadata.savedAt is required')
+
+    // Validate URL format
+    try {
+      new URL(url)
+    } catch {
+      throw new Error('Invalid URL format')
+    }
+
+    // Validate savedAt date
+    if (isNaN(Date.parse(metadata.savedAt))) {
+      throw new Error('Invalid savedAt date format')
+    }
+  }
+
+  // Getters
+  public get userId(): string {
+    return this.data.userId
+  }
+
+  public get libraryItemId(): string {
+    return this.data.libraryItemId
+  }
+
+  public get url(): string {
+    return this.data.url
+  }
+
+  public get contentType(): ContentType {
+    return this.data.contentType
+  }
+
+  public get metadata(): ContentSaveRequestedEventData['metadata'] {
+    return this.data.metadata
+  }
+}

packages/api/src/events/event-manager.test.ts

@@ -0,0 +1,49 @@
+import { EventManager } from './event-manager'
+import {
+  ContentSaveRequestedEvent,
+  ContentType,
+} from './content/content-save-event'
+import { redisDataSource } from '../redis_data_source'
+
+describe('EventManager', () => {
+  let eventManager: EventManager
+
+  beforeAll(async () => {
+    if (!redisDataSource.isInitialized) {
+      await redisDataSource.initialize()
+    }
+  })
+
+  afterAll(async () => {
+    await redisDataSource.shutdown()
+  })
+
+  beforeEach(() => {
+    eventManager = EventManager.getInstance()
+  })
+
+  afterEach(async () => {
+    await eventManager.shutdown()
+  })
+
+  it('should return the same instance', () => {
+    const instance1 = EventManager.getInstance()
+    const instance2 = EventManager.getInstance()
+    expect(instance1).toBe(instance2)
+  })
+
+  it('should register and emit events', async () => {
+    const event = new ContentSaveRequestedEvent({
+      userId: 'test-user',
+      libraryItemId: 'test-item',
+      url: 'https://example.com',
+      contentType: ContentType.HTML,
+      metadata: {
+        source: 'test',
+        savedAt: new Date().toISOString(),
+      },
+    })
+
+    await expect(eventManager.emit(event)).resolves.not.toThrow()
+  })
+})

packages/api/src/events/event-manager.ts

@@ -0,0 +1,152 @@
+import { Queue } from 'bullmq'
+import { ConnectionOptions } from 'bullmq'
+import { logger } from '../utils/logger'
+import { redisDataSource } from '../redis_data_source'
+
+export interface EventEmitter {
+  emit<T extends BaseEvent>(event: T): Promise<void>
+}
+
+export interface BaseEvent {
+  eventType: string
+  data: Record<string, any>
+  serialize(): string
+}
+
+export interface EventRoute {
+  queueName: string
+  jobName: string
+  jobOptions?: {
+    attempts?: number
+    backoff?: { type: 'exponential' | 'fixed'; delay: number }
+    removeOnComplete?: number
+    removeOnFail?: number
+  }
+}
+
+export class EventManager implements EventEmitter {
+  private queues: Map<string, Queue> = new Map()
+  private eventRoutes: Map<string, EventRoute> = new Map()
+  private redisConnection: ConnectionOptions
+  private logger = logger.child({ context: 'event-manager' })
+
+  constructor() {
+    this.redisConnection = this.getRedisConnection()
+    this.registerDefaultRoutes()
+  }
+
+  private getRedisConnection(): ConnectionOptions {
+    if (!redisDataSource.workerRedisClient) {
+      throw new Error('Redis worker client not initialized')
+    }
+
+    return {
+      host: redisDataSource.workerRedisClient.options.host,
+      port: redisDataSource.workerRedisClient.options.port,
+      password: redisDataSource.workerRedisClient.options.password,
+      db: redisDataSource.workerRedisClient.options.db,
+    }
+  }
+
+  private registerDefaultRoutes(): void {
+    // Register event type to queue mappings
+    this.registerRoute('CONTENT_SAVE_REQUESTED', {
+      queueName: 'content-processing',
+      jobName: 'process-content-save',
+      jobOptions: {
+        attempts: 3,
+        backoff: { type: 'exponential', delay: 2000 },
+        removeOnComplete: 10,
+        removeOnFail: 50,
+      },
+    })
+
+    this.registerRoute('CONTENT_PROCESSING_STARTED', {
+      queueName: 'notifications',
+      jobName: 'send-processing-notification',
+      jobOptions: {
+        attempts: 2,
+        removeOnComplete: 5,
+      },
+    })
+
+    this.registerRoute('CONTENT_PROCESSING_COMPLETED', {
+      queueName: 'post-processing',
+      jobName: 'handle-completion',
+      jobOptions: {
+        attempts: 2,
+        removeOnComplete: 20,
+      },
+    })
+  }
+
+  public registerRoute(eventType: string, route: EventRoute): void {
+    this.eventRoutes.set(eventType, route)
+    this.logger.info(
+      `${eventType} ${route.queueName} ${route.jobName}`,
+      'Event route registered'
+    )
+  }
+
+  public async emit<T extends BaseEvent>(event: T): Promise<void> {
+    const route = this.eventRoutes.get(event.eventType)
+
+    if (!route) {
+      this.logger.warn(`${event.eventType}`, 'No route found for event type')
+      return
+    }
+
+    try {
+      const queue = await this.getOrCreateQueue(route.queueName)
+
+      const job = await queue.add(route.jobName, event, route.jobOptions || {})
+
+      this.logger.info(
+        `${event.eventType} ${job.id} ${route.queueName} ${route.jobName}`,
+        'Event emitted successfully'
+      )
+    } catch (error: any) {
+      this.logger.error(
+        `${event.eventType} ${error.message} ${route.queueName} ${route.jobName}`,
+        'Failed to emit event'
+      )
+      throw error
+    }
+  }
+
+  private async getOrCreateQueue(queueName: string): Promise<Queue> {
+    if (!this.queues.has(queueName)) {
+      const queue = new Queue(queueName, {
+        connection: this.redisConnection,
+      })
+      await queue.waitUntilReady()
+      this.queues.set(queueName, queue)
+
+      this.logger.info(`${queueName}`, 'Queue created')
+    }
+    return this.queues.get(queueName)!
+  }
+
+  public async shutdown(): Promise<void> {
+    this.logger.info('Shutting down event manager')
+
+    const shutdownPromises = Array.from(this.queues.values()).map((queue) =>
+      queue.close()
+    )
+
+    await Promise.all(shutdownPromises)
+    this.queues.clear()
+
+    this.logger.info('Event manager shutdown complete')
+  }
+
+  // Singleton instance
+  private static instance: EventManager | null = null
+
+  public static getInstance(): EventManager {
+    if (!EventManager.instance) {
+      EventManager.instance = new EventManager()
+    }
+    return EventManager.instance
+  }
+}

packages/api/src/workers/content-worker.test.ts

@@ -0,0 +1,47 @@
+import { ContentWorker } from './content-worker'
+
+describe('ContentWorker', () => {
+  let worker: ContentWorker
+
+  beforeEach(() => {
+    // Clear any existing timers
+    jest.clearAllTimers()
+  })
+
+  afterEach(async () => {
+    // Ensure worker is properly cleaned up
+    if (worker) {
+      await worker.shutdown()
+      worker = null as any
+    }
+
+    // Clear any remaining timers
+    jest.clearAllTimers()
+  })
+
+  describe('initialization', () => {
+    it('should start and be running', () => {
+      worker = new ContentWorker(1)
+      expect(worker.isRunning()).toBe(true)
+    })
+
+    it('should have correct status', () => {
+      worker = new ContentWorker(1)
+      const status = worker.getStatus()
+      expect(status).toEqual({
+        isRunning: true,
+        queueName: 'content-processing',
+        concurrency: 1,
+      })
+    })
+  })
+
+  describe('lifecycle', () => {
+    it('should shutdown gracefully', async () => {
+      worker = new ContentWorker(1)
+      expect(worker.isRunning()).toBe(true)
+      await worker.shutdown()
+      expect(worker.isRunning()).toBe(false)
+    })
+  })
+})

packages/api/src/workers/content-worker.ts

@@ -0,0 +1,252 @@
+import { BaseEvent } from './../events/event-manager'
+import { Worker, Job } from 'bullmq'
+import { logger as baseLogger } from '../utils/logger'
+import { redisDataSource } from '../redis_data_source'
+import {
+  ContentSaveRequestedEvent,
+  EventType,
+} from '../events/content/content-save-event'
+import { EventManager } from '../events/event-manager'
+
+export const CONTENT_QUEUE_NAME = 'content-processing'
+export const CONTENT_SAVE_JOB_NAME = 'process-content-save'
+
+export interface ContentProcessingStartedEventData {
+  libraryItemId: string
+  userId: string
+}
+
+export interface ContentProcessingCompletedEventData {
+  libraryItemId: string
+  userId: string
+}
+
+export interface ContentProcessingFailedEventData {
+  libraryItemId: string
+  userId: string
+  error: string
+}
+
+export class ContentWorker {
+  private logger = baseLogger.child({ context: 'content-worker' })
+  private eventManager = EventManager.getInstance()
+  private worker!: Worker<ContentSaveRequestedEvent, boolean>
+
+  constructor(concurrency = 2) {
+    this.initializeAndStart(concurrency)
+  }
+
+  private initializeAndStart(concurrency: number): void {
+    const redisConnection = this.getRedisConnection()
+
+    this.worker = new Worker<ContentSaveRequestedEvent, boolean>(
+      CONTENT_QUEUE_NAME,
+      this.processJob.bind(this),
+      {
+        connection: redisConnection,
+        concurrency,
+        limiter: { max: 10, duration: 1000 },
+        autorun: true,
+      }
+    )
+
+    this.setupEventHandlers()
+    this.logger.info('Content worker started and ready')
+  }
+
+  private getRedisConnection() {
+    if (!redisDataSource.workerRedisClient) {
+      throw new Error('Redis worker client not initialized')
+    }
+
+    return {
+      host: redisDataSource.workerRedisClient.options.host,
+      port: redisDataSource.workerRedisClient.options.port,
+      password: redisDataSource.workerRedisClient.options.password,
+      db: redisDataSource.workerRedisClient.options.db,
+    }
+  }
+
+  private async processJob(
+    job: Job<ContentSaveRequestedEvent>
+  ): Promise<boolean> {
+    if (job.name !== CONTENT_SAVE_JOB_NAME) {
+      this.logger.warn(`${job.name}`, 'Unknown job type received')
+      return false
+    }
+
+    const event = job.data
+
+    this.logger.info(
+      `${job.id} ${event.libraryItemId} ${event.url}`,
+      'Processing content save job'
+    )
+
+    try {
+      // Emit processing started event
+      await this.eventManager.emit(
+        new ContentProcessingStartedEvent({
+          libraryItemId: event.libraryItemId,
+          userId: event.userId,
+        })
+      )
+
+      // Simulate content processing
+      await this.processContent(event)
+
+      // Emit processing completed event
+      await this.eventManager.emit(
+        new ContentProcessingCompletedEvent({
+          libraryItemId: event.libraryItemId,
+          userId: event.userId,
+        })
+      )
+
+      return true
+    } catch (error: any) {
+      this.logger.error(
+        `${job.id} ${event.libraryItemId} ${error.message}`,
+        'Content processing failed'
+      )
+
+      // Emit processing failed event
+      await this.eventManager.emit(
+        new ContentProcessingFailedEvent({
+          libraryItemId: event.libraryItemId,
+          userId: event.userId,
+          error: error.message,
+        })
+      )
+
+      throw error
+    }
+  }
+
+  private async processContent(
+    event: ContentSaveRequestedEvent
+  ): Promise<void> {
+    // TODO: Implement actual content processing logic
+    // - Fetch content based on contentType
+    // - Parse and extract readable content
+    // - Save to database
+    // - Generate thumbnails
+    // - Apply rules
+
+    this.logger.info(
+      `${event.libraryItemId} ${event.contentType}`,
+      'Content processed successfully'
+    )
+
+    await new Promise((resolve) => setTimeout(resolve, 1000))
+
+    throw new Error('Not implemented')
+  }
+
+  public getStatus() {
+    return {
+      isRunning: this.isRunning(),
+      queueName: CONTENT_QUEUE_NAME,
+      concurrency: this.worker.concurrency,
+    }
+  }
+
+  private setupEventHandlers(): void {
+    this.worker.on(
+      'completed',
+      (job: Job<ContentSaveRequestedEvent>, result: boolean) => {
+        this.logger.info(
+          `${job.id} ${job.data.libraryItemId} ${result}`,
+          'Content job completed'
+        )
+      }
+    )
+
+    this.worker.on(
+      'failed',
+      (job: Job<ContentSaveRequestedEvent> | undefined, error: Error) => {
+        this.logger.error(
+          `${job?.id} ${job?.data.libraryItemId} ${error.message}`,
+          'Content job failed'
+        )
+      }
+    )
+
+    this.worker.on('error', (error: Error) => {
+      this.logger.error(`${error.message}`, 'Content worker error')
+    })
+  }
+
+  public async shutdown(): Promise<void> {
+    this.logger.info('Shutting down content worker')
+    await this.worker.close()
+    this.logger.info('Content worker shut down')
+  }
+
+  public isRunning(): boolean {
+    return this.worker.isRunning()
+  }
+}
+
+class ContentProcessingStartedEvent implements BaseEvent {
+  public readonly eventType = 'CONTENT_PROCESSING_STARTED' as const
+
+  constructor(public data: ContentProcessingStartedEventData) {
+    this.validate()
+  }
+
+  serialize(): string {
+    return JSON.stringify({
+      eventType: this.eventType,
+      timestamp: new Date().toISOString(),
+      ...this.data,
+    })
+  }
+
+  protected validate(): void {
+    if (!this.data.libraryItemId) throw new Error('libraryItemId required')
+    if (!this.data.userId) throw new Error('userId required')
+  }
+}
+
+class ContentProcessingCompletedEvent implements BaseEvent {
+  public readonly eventType = 'CONTENT_PROCESSING_COMPLETED' as const
+
+  constructor(public data: ContentProcessingCompletedEventData) {
+    this.validate()
+  }
+
+  serialize(): string {
+    return JSON.stringify({
+      eventType: this.eventType,
+      timestamp: new Date().toISOString(),
+      ...this.data,
+    })
+  }
+
+  protected validate(): void {
+    if (!this.data.libraryItemId) throw new Error('libraryItemId required')
+    if (!this.data.userId) throw new Error('userId required')
+  }
+}
+
+class ContentProcessingFailedEvent implements BaseEvent {
+  public readonly eventType = 'CONTENT_PROCESSING_FAILED' as const
+
+  constructor(public data: ContentProcessingFailedEventData) {
+    this.validate()
+  }
+
+  serialize(): string {
+    return JSON.stringify({
+      eventType: this.eventType,
+      timestamp: new Date().toISOString(),
+      ...this.data,
+    })
+  }
+
+  protected validate(): void {
+    if (!this.data.libraryItemId) throw new Error('libraryItemId required')
+    if (!this.data.userId) throw new Error('userId required')
+    if (!this.data.error) throw new Error('error required')
+  }
+}

packages/api/src/workers/index.ts

@@ -0,0 +1,126 @@
+import { LibraryItemState } from '../entity/library_item'
+import { ContentSaveRequestedEvent } from '../events/content/content-save-event'
+import { redisDataSource } from '../redis_data_source'
+import { Job, Worker } from 'bullmq'
+
+export class ContentWorker {
+  private worker: Worker<ContentSaveRequestedEvent, boolean> | null = null
+  private queueName = 'content-save-requested'
+
+  constructor(concurrency = 2) {
+    this.initializeAndStart(concurrency)
+  }
+
+  private initializeAndStart(concurrency: number): void {
+    const redisConnection = this.getRedisConnection()
+
+    this.worker = new Worker<ContentSaveRequestedEvent, boolean>(
+      this.queueName,
+      this.processJob.bind(this),
+      {
+        connection: redisConnection,
+        concurrency,
+        limiter: { max: 10, duration: 1000 },
+        autorun: true,
+      }
+    )
+
+    this.setupEventHandlers()
+  }
+
+  private setupEventHandlers(): void {
+    this.worker?.on(
+      'completed',
+      (job: Job<ContentSaveRequestedEvent>, result: boolean) => {
+        console.info(`${job.id} ${job.data.libraryItemId} ${result}`)
+      }
+    )
+
+    this.worker?.on(
+      'failed',
+      (job: Job<ContentSaveRequestedEvent> | undefined, error: Error) => {
+        if (!job) return
+        console.error(`${job.id} ${job.data.libraryItemId} ${error.message}`)
+      }
+    )
+
+    this.worker?.on('error', (error: Error) => {
+      console.error(`${error.message}`)
+    })
+  }
+
+  private getRedisConnection() {
+    if (!redisDataSource.workerRedisClient) {
+      throw new Error('Redis worker client not initialized')
+    }
+
+    return {
+      host: redisDataSource.workerRedisClient.options.host,
+      port: redisDataSource.workerRedisClient.options.port,
+      password: redisDataSource.workerRedisClient.options.password,
+      db: redisDataSource.workerRedisClient.options.db,
+    }
+  }
+
+  private processJob(job: Job<ContentSaveRequestedEvent>): Promise<boolean> {
+    const { userId, libraryItemId, url, contentType, metadata } = job.data
+
+    if (!userId?.trim()) throw new Error('userId is required')
+    if (!libraryItemId?.trim()) throw new Error('libraryItemId is required')
+    if (!url?.trim()) throw new Error('url is required')
+    if (!contentType) throw new Error('contentType is required')
+    if (!metadata?.source?.trim())
+      throw new Error('metadata.source is required')
+    if (!metadata?.savedAt?.trim())
+      throw new Error('metadata.savedAt is required')
+
+    // Validate URL format
+    try {
+      new URL(url)
+    } catch {
+      throw new Error('Invalid URL format')
+    }
+    return Promise.resolve(true)
+  }
+}
+
+// packages/api/src/services/create_page_save_request.ts
+// export const createPageSaveRequest = async (params: {
+//   userId: string
+//   url: string
+//   labels: string[]
+//   folder: string
+//   source: string
+//   savedAt: Date
+//   publishedAt: Date
+// }) => {
+//   // Create library item in database
+//   const libraryItem = await createOrUpdateLibraryItem(
+//     {
+//       // ... existing logic
+//       state: LibraryItemState.Processing,
+//     },
+//     userId,
+//     pubsub
+//   )
+
+//   // Fire single event to dedicated queue
+//   await emitContentSaveEvent({
+//     eventType: 'CONTENT_SAVE_REQUESTED',
+//     userId,
+//     libraryItemId: libraryItem.id,
+//     url,
+//     contentType: detectContentType(url),
+//     metadata: {
+//       labels,
+//       folder,
+//       source,
+//       savedAt: savedAt?.toISOString(),
+//       publishedAt: publishedAt?.toISOString(),
+//     },
+//   })
+
+//   return libraryItem
+// }
+
+// async function emitContentSaveEvent(event: ContentSaveEvent) {}

packages/api/src/workers/worker-config.ts

@@ -0,0 +1,31 @@
+export interface WorkerConfig {
+  contentProcessing: {
+    enabled: boolean
+    concurrency: number
+    memoryLimit: string
+  }
+  emailProcessing: {
+    enabled: boolean
+    concurrency: number
+  }
+  exportProcessing: {
+    enabled: boolean
+    concurrency: number
+  }
+}
+
+export const getWorkerConfig = (): WorkerConfig => ({
+  contentProcessing: {
+    enabled: process.env.ENABLE_CONTENT_WORKER === 'true',
+    concurrency: parseInt(process.env.CONTENT_WORKER_CONCURRENCY || '2'),
+    memoryLimit: process.env.CONTENT_WORKER_MEMORY_LIMIT || '512MB',
+  },
+  emailProcessing: {
+    enabled: process.env.ENABLE_EMAIL_WORKER === 'true',
+    concurrency: parseInt(process.env.EMAIL_WORKER_CONCURRENCY || '2'),
+  },
+  exportProcessing: {
+    enabled: process.env.ENABLE_EXPORT_WORKER === 'true',
+    concurrency: parseInt(process.env.EXPORT_WORKER_CONCURRENCY || '2'),
+  },
+})

packages/api/tsconfig.json

@@ -1,14 +1,25 @@
 {
-  "extends": "./../../tsconfig.json",
-  "ts-node": {
-    "files": true
-  },
+  "extends": "../../tsconfig.json",
   "compilerOptions": {
-    "outDir": "dist",
-    "paths": {
-      "express": ["./node_modules/@types/express"]
-    }
+    "outDir": "./dist",
+    "rootDir": "./src",
+    "resolveJsonModule": true,
+    "skipLibCheck": true,
+    "noImplicitAny": false,
+    "strictNullChecks": false,
+    "strict": false,
+    "lib": ["ES2020"],
+    "allowJs": true,
+    "isolatedModules": true
   },
-  "include": ["src/**/*.ts", "test/**/*.ts"],
-  "exclude": ["./src/generated"]
-}
+  "include": [
+    "src/**/*.ts",
+    "src/**/*.js"
+  ],
+  "exclude": [
+    "jest.*.ts",
+    "test/**/*",
+    "dist/**/*",
+    "node_modules/**/*"
+  ]
+}