K
Klivvr Blog

Message Queue Design for Notification Systems

How to design message queues for notification systems covering priority queues, dead letter queues, retry logic, and backpressure handling — with patterns from Whisper.

technical5 min readBy Klivvr Engineering
Share:

A notification system must handle variable load, guarantee delivery, and prioritize urgent messages. A transaction alert cannot wait behind a batch of marketing notifications. A failed delivery must be retried without duplicating notifications. And when a downstream service is overwhelmed, the system must apply backpressure rather than dropping messages.

Message queues provide the buffering, ordering, and reliability guarantees that make this possible. This article covers the queue design patterns used in Whisper.

Priority Queue Architecture

Not all notifications are equal. A fraud alert is more urgent than a weekly summary. A payment confirmation matters more than a promotional offer. Whisper uses a multi-tier priority queue to ensure urgent notifications are processed first.

enum NotificationPriority {
  CRITICAL = 0,   // Security alerts, fraud warnings
  HIGH = 1,       // Transaction confirmations, payment due
  NORMAL = 2,     // Account updates, feature announcements
  LOW = 3,        // Marketing, newsletters, tips
}
 
interface QueuedNotification {
  id: string;
  userId: string;
  priority: NotificationPriority;
  payload: NotificationPayload;
  enqueuedAt: Date;
  scheduledFor?: Date;
  attempts: number;
  maxAttempts: number;
}
 
class PriorityNotificationQueue {
  private queues: Map<NotificationPriority, QueuedNotification[]>;
  private processing = false;
 
  constructor() {
    this.queues = new Map([
      [NotificationPriority.CRITICAL, []],
      [NotificationPriority.HIGH, []],
      [NotificationPriority.NORMAL, []],
      [NotificationPriority.LOW, []],
    ]);
  }
 
  enqueue(notification: QueuedNotification): void {
    const queue = this.queues.get(notification.priority);
    if (!queue) throw new Error(`Invalid priority: ${notification.priority}`);
    queue.push(notification);
  }
 
  dequeue(): QueuedNotification | null {
    const now = new Date();
 
    // Process highest priority first
    for (const priority of [
      NotificationPriority.CRITICAL,
      NotificationPriority.HIGH,
      NotificationPriority.NORMAL,
      NotificationPriority.LOW,
    ]) {
      const queue = this.queues.get(priority)!;
      const readyIndex = queue.findIndex(
        (n) => !n.scheduledFor || n.scheduledFor <= now
      );
 
      if (readyIndex !== -1) {
        return queue.splice(readyIndex, 1)[0];
      }
    }
 
    return null;
  }
 
  getQueueDepths(): Record<string, number> {
    const depths: Record<string, number> = {};
    for (const [priority, queue] of this.queues) {
      depths[NotificationPriority[priority]] = queue.length;
    }
    return depths;
  }
}

Dead Letter Queue

When a notification fails after all retry attempts, it moves to a dead letter queue (DLQ) for investigation rather than being silently dropped.

interface DeadLetterEntry {
  notification: QueuedNotification;
  failureReason: string;
  failedAt: Date;
  attempts: Array<{
    attemptNumber: number;
    error: string;
    timestamp: Date;
  }>;
}
 
class DeadLetterQueue {
  private entries: DeadLetterEntry[] = [];
  private db: Database;
 
  async add(entry: DeadLetterEntry): Promise<void> {
    await this.db.query(
      `INSERT INTO dead_letter_queue
       (notification_id, user_id, payload, failure_reason, failed_at, attempts)
       VALUES ($1, $2, $3, $4, $5, $6)`,
      [
        entry.notification.id,
        entry.notification.userId,
        JSON.stringify(entry.notification.payload),
        entry.failureReason,
        entry.failedAt,
        JSON.stringify(entry.attempts),
      ]
    );
  }
 
  async getRecent(limit: number = 100): Promise<DeadLetterEntry[]> {
    const result = await this.db.query(
      `SELECT * FROM dead_letter_queue ORDER BY failed_at DESC LIMIT $1`,
      [limit]
    );
    return result.rows;
  }
 
  async retry(notificationId: string): Promise<void> {
    const entry = await this.db.query(
      `SELECT * FROM dead_letter_queue WHERE notification_id = $1`,
      [notificationId]
    );
 
    if (entry.rows.length === 0) return;
 
    const notification = entry.rows[0];
    await notificationQueue.enqueue({
      ...notification,
      attempts: 0,
      maxAttempts: 3,
    });
 
    await this.db.query(
      `DELETE FROM dead_letter_queue WHERE notification_id = $1`,
      [notificationId]
    );
  }
 
  async getFailureAnalytics(): Promise<FailureAnalytics> {
    const result = await this.db.query(`
      SELECT
        failure_reason,
        COUNT(*) as count,
        MIN(failed_at) as first_failure,
        MAX(failed_at) as last_failure
      FROM dead_letter_queue
      WHERE failed_at > NOW() - INTERVAL '24 hours'
      GROUP BY failure_reason
      ORDER BY count DESC
    `);
 
    return { byReason: result.rows };
  }
}

Processing Pipeline

The notification processing pipeline dequeues notifications, validates them, applies rate limits, and dispatches to the appropriate delivery channel.

class NotificationProcessor {
  private queue: PriorityNotificationQueue;
  private dlq: DeadLetterQueue;
  private rateLimiter: RateLimiter;
  private deliveryService: DeliveryService;
  private concurrency: number;
  private active = 0;
 
  constructor(config: {
    queue: PriorityNotificationQueue;
    dlq: DeadLetterQueue;
    rateLimiter: RateLimiter;
    deliveryService: DeliveryService;
    concurrency: number;
  }) {
    Object.assign(this, config);
  }
 
  async start(): Promise<void> {
    while (true) {
      if (this.active >= this.concurrency) {
        await this.sleep(100);
        continue;
      }
 
      const notification = this.queue.dequeue();
      if (!notification) {
        await this.sleep(500);
        continue;
      }
 
      this.active++;
      this.processNotification(notification)
        .catch((error) => console.error("Processing error:", error))
        .finally(() => this.active--);
    }
  }
 
  private async processNotification(
    notification: QueuedNotification
  ): Promise<void> {
    // Check rate limit
    const allowed = await this.rateLimiter.check(notification.userId);
    if (!allowed) {
      // Re-queue with delay
      notification.scheduledFor = new Date(Date.now() + 60000);
      this.queue.enqueue(notification);
      return;
    }
 
    try {
      await this.deliveryService.deliver(notification);
    } catch (error) {
      notification.attempts++;
 
      if (notification.attempts >= notification.maxAttempts) {
        await this.dlq.add({
          notification,
          failureReason: (error as Error).message,
          failedAt: new Date(),
          attempts: [
            {
              attemptNumber: notification.attempts,
              error: (error as Error).message,
              timestamp: new Date(),
            },
          ],
        });
      } else {
        // Exponential backoff
        const delay = Math.pow(2, notification.attempts) * 1000;
        notification.scheduledFor = new Date(Date.now() + delay);
        this.queue.enqueue(notification);
      }
    }
  }
 
  private sleep(ms: number): Promise<void> {
    return new Promise((resolve) => setTimeout(resolve, ms));
  }
}

Rate Limiting

Rate limiting prevents notification fatigue and protects downstream services from overload.

class RateLimiter {
  private redis: Redis;
 
  constructor(redis: Redis) {
    this.redis = redis;
  }
 
  async check(userId: string): Promise<boolean> {
    const key = `ratelimit:notifications:${userId}`;
    const window = 3600; // 1 hour window
    const maxNotifications = 20; // Max per hour
 
    const current = await this.redis.incr(key);
    if (current === 1) {
      await this.redis.expire(key, window);
    }
 
    return current <= maxNotifications;
  }
 
  async checkByPriority(
    userId: string,
    priority: NotificationPriority
  ): Promise<boolean> {
    // Critical notifications bypass rate limits
    if (priority === NotificationPriority.CRITICAL) return true;
 
    // Per-priority limits
    const limits: Record<NotificationPriority, number> = {
      [NotificationPriority.CRITICAL]: Infinity,
      [NotificationPriority.HIGH]: 50,
      [NotificationPriority.NORMAL]: 20,
      [NotificationPriority.LOW]: 5,
    };
 
    const key = `ratelimit:${priority}:${userId}`;
    const current = await this.redis.incr(key);
    if (current === 1) {
      await this.redis.expire(key, 3600);
    }
 
    return current <= limits[priority];
  }
}

Conclusion

Message queue design determines the reliability and responsiveness of a notification system. Priority queues ensure urgent messages are delivered first. Dead letter queues prevent silent message loss and enable failure analysis. Rate limiting protects users from notification fatigue and downstream services from overload. And the processing pipeline ties these components together with configurable concurrency and backpressure handling. In Whisper, these patterns ensure that every notification — from a critical fraud alert to a routine account update — is processed reliably and delivered in the right order.

Related Articles

business

Omnichannel Messaging Strategy for Fintech

How to build a unified omnichannel messaging strategy across push, SMS, email, and in-app channels — covering channel selection, message consistency, and unified customer experience.

7 min read
business

Real-Time Notifications in Banking

How real-time notifications transform the banking experience, covering transaction alerts, security notifications, compliance requirements, and the business impact of instant communication.

6 min read
technical

Scaling Notification Systems

Strategies for scaling notification systems to millions of users, covering horizontal scaling, fan-out patterns, rate limiting, and infrastructure design — with patterns from Whisper.

5 min read