Skip to content

ARCHITECTURE.md

Latest commit

 

History

History
274 lines (228 loc) · 10.6 KB

ARCHITECTURE.md

File metadata and controls

274 lines (228 loc) · 10.6 KB
Raw

Migration Tool Architecture

Project Structure

src/
├── config/              # Configuration (queue, redis, app settings)
├── types/               # TypeScript type definitions
├── models/              # Data models (Redis persistence)
├── controllers/         # HTTP request handlers
├── routes/              # API route definitions
├── services/            # Business logic orchestration
├── middlewares/         # Express middlewares
└── queues/              # ⭐ Job queue system (see below)
    ├── jobs/            # Job creation & definitions
    ├── processors/      # Business logic (core work)
    ├── workers/         # BullMQ worker instances
    └── listeners/       # Event handlers (monitoring)

Queue System Architecture (Industry Best Practice)

Complete Flow

┌─────────────────────────────────────────────────────────────────┐
│                        API REQUEST                              │
└─────────────────────────────────────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────────┐
│  Controller (controllers/job.controller.ts)                     │
│  - Validates request                                            │
│  - Handles HTTP layer                                           │
└─────────────────────────────────────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────────┐
│  Service (services/queue.service.ts)                            │
│  - Orchestrates between layers                                  │
│  - Calls job creator + persists to DB                           │
└─────────────────────────────────────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────────┐
│  Job Definition (queues/jobs/migration.job.ts)                  │
│  - createMigrationJob()                                         │
│  - Defines job options (retries, backoff, etc.)                 │
│  - Adds job to BullMQ queue                                     │
└─────────────────────────────────────────────────────────────────┘
                              │
                              ▼
                    ┌──────────────────┐
                    │   Redis Queue    │
                    │   (BullMQ)       │
                    └──────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────────┐
│  Worker (queues/workers/migration.worker.ts)                    │
│  - Picks up job from queue                                      │
│  - Passes to processor                                          │
│  - Manages concurrency                                          │
└─────────────────────────────────────────────────────────────────┘
         │                                           │
         │ (processing)                              │ (events)
         ▼                                           ▼
┌──────────────────────────────┐    ┌───────────────────────────┐
│  Processor                   │    │  Listeners                │
│  (queues/processors/)        │    │  (queues/listeners/)      │
│                              │    │                           │
│  - Pure business logic       │    │  - on('completed')        │
│  - Migration steps           │    │  - on('failed')           │
│  - Data transformation       │    │  - on('progress')         │
│  - No side effects           │    │  - Logging                │
│                              │    │  - DB updates             │
│  Returns: Result             │    │  - Metrics/Alerts         │
└──────────────────────────────┘    └───────────────────────────┘

Folder Responsibilities

1️⃣ jobs/ - Job Creators

// What: Define HOW to create jobs
// Example: createMigrationJob(data, options?)

┌────────────────────────────┐
│  createMigrationJob()      │
│  - Set job options         │
│  - Add to queue            │
│  - Return job metadata     │
└────────────────────────────┘

Responsibilities:

  • Job creation logic
  • Default options (retries, backoff)
  • Job utilities (retry, remove)
  • Job constants/names

Used by: Services, Controllers

2️⃣ processors/ - Business Logic

// What: Define HOW to process jobs
// Example: processMigrationJob(job)

┌────────────────────────────┐
│  processMigrationJob()     │
│  1. Validate               │
│  2. Extract                │
│  3. Transform              │
│  4. Load                   │
│  5. Verify                 │
│  Return: Result            │
└────────────────────────────┘

Responsibilities:

  • Core business logic ONLY
  • Pure functions (no side effects)
  • Actual migration work
  • Data transformations

Used by: Workers

3️⃣ workers/ - BullMQ Infrastructure

// What: Connect processor to BullMQ
// Example: new Worker(queue, processor, config)

┌────────────────────────────┐
│  migrationWorker           │
│  - Queue name              │
│  - Processor function      │
│  - Concurrency: 2          │
│  - Attach listeners        │
└────────────────────────────┘

Responsibilities:

  • Worker configuration
  • Connect processor to queue
  • Attach event listeners
  • Manage concurrency

Used by: Server startup (index.ts)

4️⃣ listeners/ - Monitoring & Side Effects

// What: Handle worker events
// Example: worker.on('completed', handler)

┌────────────────────────────┐
│  attachMigrationListeners  │
│  - completed → log + DB    │
│  - failed → alert          │
│  - progress → update UI    │
│  - error → monitor         │
└────────────────────────────┘

Responsibilities:

  • Event handling
  • Logging & monitoring
  • Database status updates
  • Alerts/notifications
  • Metrics collection

Used by: Workers

Why This Structure?

✅ Separation of Concerns

jobs/       → "Create this job"
processors/ → "Do the actual work"
workers/    → "Manage the infrastructure"
listeners/  → "Monitor & log everything"

✅ Testability

// Test job creation without processing
test('createMigrationJob', () => { ... });

// Test processing without infrastructure
test('processMigrationJob', () => { ... });

// Test monitoring without affecting logic
test('listeners', () => { ... });

✅ Reusability

// Processor can be called from multiple places
await processMigrationJob(job);           // From worker
await processMigrationJob(mockJob);       // From CLI
await processMigrationJob(testJob);       // From tests

✅ Maintainability

Need to change logging?        → Only touch listeners/
Need to change business logic? → Only touch processors/
Need to change job options?    → Only touch jobs/
Need to change concurrency?    → Only touch workers/

Industry Examples

Stripe

stripe/
└── workers/
    ├── jobs/          # Job definitions
    ├── handlers/      # Business logic (our processors/)
    └── workers/       # Worker instances

GitHub

github/
└── background/
    ├── jobs/          # Job creators
    ├── processors/    # Pure business logic
    └── workers/       # Infrastructure

Shopify

shopify/
└── jobs/
    ├── definitions/   # Job creation (our jobs/)
    ├── perform/       # Execution logic (our processors/)
    └── workers/       # BullMQ setup

Key Principles

  1. Jobs are data - jobs/ defines the job structure
  2. Processors are pure - No side effects, easy to test
  3. Workers are glue - Connect everything together
  4. Listeners are observers - Monitor without affecting logic

Adding New Job Types

# 1. Define job creation
src/queues/jobs/export.job.ts

# 2. Implement business logic
src/queues/processors/export.processor.ts

# 3. Create worker instance
src/queues/workers/export.worker.ts

# 4. Add event handlers
src/queues/listeners/export.listeners.ts

# 5. Export from index
src/queues/index.ts

Each job type follows the same pattern - predictable and scalable!