Design a Notification System: Build One That Won't Collapse at 3 AM

// io.thecodeforge — System Design tutorial

// High-level architecture diagram (pseudocode)
// Client -> API Gateway -> Notification Service -> Queue -> Worker -> Provider -> User
//                                           |-> Database (for logging)

// Key: Queue is the heart. Without it, the system is fragile.

# io.thecodeforge — System Design tutorial

from flask import Flask, request, jsonify
from uuid import uuid4
import boto3

app = Flask(__name__)
sqs = boto3.client('sqs', region_name='us-east-1')
QUEUE_URL = 'https://sqs.us-east-1.amazonaws.com/123456789012/notifications'

def generate_idempotency_key():
    return str(uuid4())

@app.route('/notifications', methods=['POST'])
def ingest_notification():
    data = request.get_json()
    # Validate required fields
    if not data.get('user_id') or not data.get('message'):
        return jsonify({'error': 'user_id and message required'}), 400
    
    # Assign idempotency key if missing
    if 'idempotency_key' not in data:
        data['idempotency_key'] = generate_idempotency_key()
    
    # Send to queue asynchronously
    try:
        sqs.send_message(
            QueueUrl=QUEUE_URL,
            MessageBody=json.dumps(data),
            MessageDeduplicationId=data['idempotency_key']  # for FIFO queues
        )
    except Exception as e:
        # Log and return 500 — but don't block the client forever
        app.logger.error(f'Failed to enqueue: {e}')
        return jsonify({'error': 'internal error'}), 500
    
    return jsonify({'status': 'accepted', 'idempotency_key': data['idempotency_key']}), 202

# io.thecodeforge — System Design tutorial

import json
import boto3
from time import sleep

sqs = boto3.client('sqs')
QUEUE_URL = 'https://sqs.us-east-1.amazonaws.com/123456789012/notifications'
DLQ_URL = 'https://sqs.us-east-1.amazonaws.com/123456789012/notifications-dlq'

def process_message(message):
    body = json.loads(message['Body'])
    user_id = body['user_id']
    channel = body.get('channel', 'email')  # default to email
    
    # Format based on channel
    if channel == 'email':
        formatted = format_email(body)
        send_email(user_id, formatted)
    elif channel == 'sms':
        formatted = format_sms(body)
        send_sms(user_id, formatted)
    # ... other channels
    
    # Log to database (async, fire-and-forget)
    log_notification(user_id, body['idempotency_key'], channel, 'sent')

def main():
    while True:
        response = sqs.receive_message(
            QueueUrl=QUEUE_URL,
            MaxNumberOfMessages=10,
            VisibilityTimeout=30,
            WaitTimeSeconds=20  # long polling
        )
        messages = response.get('Messages', [])
        for msg in messages:
            try:
                process_message(msg)
                sqs.delete_message(QueueUrl=QUEUE_URL, ReceiptHandle=msg['ReceiptHandle'])
            except Exception as e:
                # Check retry count from ApproximateReceiveCount
                receive_count = int(msg['Attributes']['ApproximateReceiveCount'])
                if receive_count >= 3:
                    # Move to DLQ
                    sqs.send_message(QueueUrl=DLQ_URL, MessageBody=msg['Body'])
                    sqs.delete_message(QueueUrl=QUEUE_URL, ReceiptHandle=msg['ReceiptHandle'])
                else:
                    # Let visibility timeout expire for retry
                    pass

if __name__ == '__main__':
    main()

# io.thecodeforge — System Design tutorial

import time
import random
import requests

def send_email_with_retry(to, subject, body, max_retries=3):
    for attempt in range(max_retries):
        try:
            response = requests.post(
                'https://api.sendgrid.com/v3/mail/send',
                json={'personalizations': [{'to': [{'email': to}]}],
                      'subject': subject,
                      'content': [{'type': 'text/plain', 'value': body}]},
                headers={'Authorization': 'Bearer SG_API_KEY'},
                timeout=10
            )
            if response.status_code == 202:
                return True
            elif response.status_code == 429:  # rate limited
                wait = (2 ** attempt) + random.uniform(0, 1)  # exponential backoff + jitter
                time.sleep(wait)
            else:
                # Non-retryable error? Log and break.
                return False
        except requests.exceptions.Timeout:
            wait = (2 ** attempt) + random.uniform(0, 1)
            time.sleep(wait)
        except requests.exceptions.ConnectionError:
            wait = (2 ** attempt) + random.uniform(0, 1)
            time.sleep(wait)
    return False

-- io.thecodeforge — System Design tutorial

-- Create table for idempotency keys
CREATE TABLE notification_idempotency (
    idempotency_key VARCHAR(255) PRIMARY KEY,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    status VARCHAR(20) NOT NULL DEFAULT 'pending'  -- pending, processing, completed
);

-- Use INSERT ... ON CONFLICT to atomically claim a key
INSERT INTO notification_idempotency (idempotency_key, status)
VALUES ('abc-123', 'processing')
ON CONFLICT (idempotency_key) DO NOTHING
RETURNING status;
-- If the INSERT returned a row, we own this key. If NULL, it already exists.