event-driven-patterns

name	event-driven-patterns
description	Message queue patterns with BullMQ, Kafka, RabbitMQ - saga, outbox, dead letter queue, exactly-once semantics.

Event-Driven Patterns

Message queue and event bus patterns for decoupled, reliable async processing.

BullMQ Setup (Producer + Consumer)

import { Queue, Worker, QueueEvents } from 'bullmq'
import Redis from 'ioredis'

const connection = new Redis(process.env.REDIS_URL!, { maxRetriesPerRequest: null })

// Producer: define queue
const emailQueue = new Queue('email', { connection })
const marketQueue = new Queue('market-resolution', { connection })

// Add job with options
await emailQueue.add(
  'send-welcome',
  { userId: 'abc', email: '[e-mail]' },
  {
    attempts: 3,
    backoff: { type: 'exponential', delay: 1000 },
    removeOnComplete: { count: 1000 },
    removeOnFail: { count: 5000 }
  }
)

// Delayed job (send after 1 hour)
await emailQueue.add('send-reminder', { userId: 'abc' }, { delay: 3_600_000 })

// Consumer: named processor
const emailWorker = new Worker(
  'email',
  async (job) => {
    if (job.name === 'send-welcome') {
      await sendWelcomeEmail(job.data.email)
    } else if (job.name === 'send-reminder') {
      await sendReminderEmail(job.data.userId)
    }
    // Return value stored in job.returnvalue
    return { sent: true, at: new Date().toISOString() }
  },
  {
    connection,
    concurrency: 10
  }
)

emailWorker.on('completed', (job, result) => {
  console.log(`Job ${job.id} completed:`, result)
})

emailWorker.on('failed', (job, err) => {
  console.error(`Job ${job?.id} failed after ${job?.attemptsMade} attempts:`, err.message)
})

Retry Policies and Dead Letter Queue

import { Queue, Worker, QueueEvents } from 'bullmq'

const dlqQueue = new Queue('dead-letter', { connection })

const processingWorker = new Worker(
  'orders',
  async (job) => {
    // Attempt processing
    await processOrder(job.data)
  },
  {
    connection,
    concurrency: 5
  }
)

// Move failed jobs to DLQ after all retries exhausted
processingWorker.on('failed', async (job, err) => {
  if (!job) return
  const isExhausted = job.attemptsMade >= (job.opts.attempts || 1)

  if (isExhausted) {
    await dlqQueue.add('order-failed', {
      originalJob: job.name,
      data: job.data,
      error: err.message,
      failedAt: new Date().toISOString(),
      attempts: job.attemptsMade
    })
    console.error()
  }
})


 dlqWorker =  (,  (job) => {
   ({
    : ,
    : job.
  })
}, { connection })

Transactional Outbox Pattern

// Problem: write to DB and publish event atomically (no lost messages)
// Solution: write event to outbox table in same transaction, relay worker reads and publishes

// DB schema
// CREATE TABLE outbox (
//   id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
//   aggregate_type TEXT NOT NULL,
//   aggregate_id TEXT NOT NULL,
//   event_type TEXT NOT NULL,
//   payload JSONB NOT NULL,
//   published_at TIMESTAMPTZ,
//   created_at TIMESTAMPTZ DEFAULT now()
// );

async function createMarketWithOutbox(data: CreateMarketDto): Promise<Market> {
  return db.$transaction(async (tx) => {
    // 1. Write domain entity
    const market = await tx.market.create({ data })

    // 2. Write outbox event in SAME transaction
    await tx.outbox.create({
      data: {
        aggregateType: 'Market',
        aggregateId: market.id,
        eventType: 'MarketCreated',
        payload: { marketId: market.id, name: market.name, createdAt: market.createdAt }
      }
    })

     market
  })
}


  (): <> {
   unpublished =  db..({
    : { :  },
    : { :  },
    : 
  })

   ( event  unpublished) {
     {
       (event., event.)
       db..({
        : { : event. },
        : { :  () }
      })
    }  (err) {
      .(, err)
    }
  }
}


(outboxRelay, )

Saga Pattern (Orchestration)

// Orchestrator drives the saga steps and handles compensation

interface SagaStep<T> {
  name: string
  execute: (ctx: T) => Promise<Partial<T>>
  compensate: (ctx: T) => Promise<void>
}

class SagaOrchestrator<T extends Record<string, unknown>> {
  constructor(private steps: SagaStep<T>[]) {}

  async run(initialContext: T): Promise<T> {
    const ctx = { ...initialContext }
    const completed: SagaStep<T>[] = []

    for (const step of this.steps) {
      try {
        const result = await step.execute(ctx)
        Object.assign(ctx, result)
        completed.push(step)
        console.log(`Saga step '${step.name}' succeeded`)
      }  (err) {
        .()

        
         ( done  completed.()) {
           {
             done.(ctx)
            .()
          }  (compensateErr) {
            .(, compensateErr)
            
          }
        }
         err
      }
    }

     ctx
  }
}


  {
  : 
  : 
  : 
  ?: 
  ?: 
}

 orderSaga =  <>([
  {
    : ,
    :  (ctx) => {
       reservationId =  inventory.(ctx.)
       { reservationId }
    },
    :  (ctx) => {
       (ctx.)  inventory.(ctx.)
    }
  },
  {
    : ,
    :  (ctx) => {
       paymentId =  payments.(ctx., ctx.)
       { paymentId }
    },
    :  (ctx) => {
       (ctx.)  payments.(ctx.)
    }
  },
  {
    : ,
    :  (ctx) => {
       orders.(ctx.)
       {}
    },
    :  (ctx) => {
       orders.(ctx.)
    }
  }
])

Idempotent Consumers (Exactly-Once Semantics)

// Even if a message is delivered twice, process it only once
async function processEventIdempotent(
  eventId: string,
  handler: () => Promise<void>
): Promise<void> {
  const key = `processed:${eventId}`

  // SET NX: only set if not exists (atomic)
  const isNew = await redis.set(key, '1', 'EX', 86_400, 'NX')
  if (!isNew) {
    console.log(`Event ${eventId} already processed, skipping`)
    return
  }

  try {
    await handler()
  } catch (err) {
    // Release lock so it can be retried
    await redis.del(key)
    throw err
  }
}

// In BullMQ worker
const worker = new Worker('payments', async (job) => {
  await processEventIdempotent(job.id!, async () => {
    await processPayment(job.data)
  })
}, { connection })

Fan-Out Pattern

// One event → multiple consumers in parallel
const eventBus = new Queue('events', { connection })

async function publishMarketResolved(marketId: string, outcome: string): Promise<void> {
  const event = { marketId, outcome, resolvedAt: new Date().toISOString() }

  // Fan-out to multiple downstream queues
  await Promise.all([
    notificationQueue.add('market-resolved', event),
    payoutQueue.add('process-payouts', event),
    analyticsQueue.add('track-resolution', event),
    feedQueue.add('update-feed', event)
  ])
}

// Each queue has its own worker with appropriate concurrency and retry config

Priority Queue

// Higher priority number = processed first in BullMQ
await criticalQueue.add('urgent-payout', data, { priority: 1 })   // highest
await normalQueue.add('regular-email', data, { priority: 10 })
await batchQueue.add('report-generation', data, { priority: 100 }) // lowest

// Worker respects priority automatically when picking next job

Queue Monitoring

import { QueueEvents } from 'bullmq'

const queueEvents = new QueueEvents('email', { connection })

// Track job lifecycle
queueEvents.on('waiting', ({ jobId }) => metrics.increment('jobs.waiting'))
queueEvents.on('active', ({ jobId }) => metrics.increment('jobs.active'))
queueEvents.on('completed', ({ jobId }) => metrics.increment('jobs.completed'))
queueEvents.on('failed', ({ jobId, failedReason }) => {
  metrics.increment('jobs.failed')
  console.error(`Job ${jobId} failed: ${failedReason}`)
})
queueEvents.on('stalled', ({ jobId }) => {
  metrics.increment('jobs.stalled')
  console.warn(`Job ${jobId} stalled — worker may have crashed`)
})

// Health check: alert if queue depth grows too large
  (): <> {
   counts =  queue.(, , , )

   (counts. > ) {
     ({ : queue., : counts. })
  }
   (counts. > ) {
     ({ : queue., : counts. })
  }
}

( (emailQueue), )

Remember: Use the outbox pattern whenever a DB write and an event publish must be atomic. Never publish directly inside a transaction — the broker call can fail after the DB commits, causing lost events.