Notification Service

Functional Requirements

• Send notifications via push notifications, SMS, email, and in-app messages
• Notification history: Users can view past notifications across all channels
• Template management: Create and manage notification templates with variable substitution
• User preferences: Opt-in/out per channel, set quiet hours, configure frequency caps
• Delivery tracking: Track sent, delivered, opened, and clicked status
• Bulk notifications: Send broadcast notifications to millions of users
• Priority levels: Critical (security alerts) vs normal (marketing) vs low (digest)

Non-Functional Requirements

• High throughput: Handle 10M+ notifications per day (~115/sec average, 1000/sec peak)
• Delivery guarantees: At-least-once delivery with deduplication
• Low latency: Critical notifications delivered within 5 seconds
• Preference respect: Never send to users who opted out of a channel
• Scalability: Must handle Black Friday traffic spikes (10x normal)
• Idempotency: Same notification never delivered twice

Scale Estimation

Metric	Daily	Per Second
Total notifications	10M	~115
Push notifications	5M	~58
Email	3M	~35
SMS	1M	~12
In-app	1M	~12
Notification reads	50M	~580
Template renders	15M	~174

Notification Channels

Push Notifications (Mobile)

• APNS (Apple Push Notification Service): iOS/macOS devices
• FCM (Firebase Cloud Messaging): Android devices
• Requires device tokens registered per user
• Payload limits: APNS 4KB, FCM 4KB
• Supports rich notifications (images, actions) on iOS 10+ and Android 7.0+

SMS

• Providers: Twilio, SNS (Amazon Simple Notification Service), Vonage
• Limitations: 160 characters per SMS, concatenated messages for longer content
• Cost: ~$0.0075 per SMS (varies by country)
• Use cases: OTP, 2FA, critical alerts, marketing (where allowed)

Email

• Providers: Amazon SES, SendGrid, Mailgun
• Features: HTML templates, attachments, tracking pixels
• Deliverability: SPF, DKIM, DMARC authentication required
• Bounce handling: Hard bounces (invalid address) vs soft bounces (full mailbox)

In-App Notifications

• Implementation: WebSocket for real-time, long-polling as fallback
• Storage: Database for notification feed, read/unread status
• Features: Click-through actions, dismissal, batch read

Channel Comparison

Feature	Push	SMS	Email	In-App
Latency	<1s	<5s	<30s	<1s
Cost	Free	$0.0075	$0.0001	Free
Reach	App installed	Phone number	Email address	App/website
Rich content	Limited	No	Yes	Yes
Opt-in required	Yes (device)	Yes	Yes	Implicit
Offline delivery	No (unless in-app)	Yes	Yes	No

High-Level Architecture

┌──────────────┐     ┌──────────────┐     ┌──────────────┐
│  API Gateway │────▶│Notification  │────▶│   Message    │
│              │     │   Service    │     │   Queue      │
└──────────────┘     └──────────────┘     └──────┬───────┘
                                                  │
                    ┌─────────────────────────────┼─────────────────────────────┐
                    ▼                             ▼                             ▼
             ┌──────────────┐             ┌──────────────┐             ┌──────────────┐
             │  Push Worker │             │  SMS Worker  │             │Email Worker  │
             │    Pool      │             │    Pool      │             │    Pool      │
             └──────┬───────┘             └──────┬───────┘             └──────┬───────┘
                    ▼                            ▼                            ▼
             ┌──────────────┐             ┌──────────────┐             ┌──────────────┐
             │  APNS / FCM  │             │  Twilio/SNS  │             │ Amazon SES   │
             └──────────────┘             └──────────────┘             └──────────────┘

Queue-Based Delivery Pipeline

class NotificationService:
    def __init__(self, queue, template_engine, preference_store, dedup_store):
        self.queue = queue
        self.template_engine = template_engine
        self.preference_store = preference_store
        self.dedup_store = dedup_store
    
    def send_notification(self, request):
        # 1. Check deduplication
        dedup_key = f"notif:{request.user_id}:{request.template_id}:{request.dedup_window}"
        if self.dedup_store.exists(dedup_key):
            return DeduplicationResult(deduplicated=True)
        
        # 2. Check user preferences
        preferences = self.preference_store.get(request.user_id)
        if not preferences.is_channel_enabled(request.channel):
            return PreferenceResult(channel_disabled=True)
        
        if preferences.is_quiet_hours():
            self.queue.schedule(request, preferences.quiet_hours_end)
            return PreferenceResult(quiet_hours=True, scheduled_for=preferences.quiet_hours_end)
        
        if preferences.is_frequency_capped(request.template_id):
            return PreferenceResult(frequency_capped=True)
        
        # 3. Render template
        rendered = self.template_engine.render(
            request.template_id,
            request.variables,
            preferences.language
        )
        
        # 4. Enqueue for delivery
        message = NotificationMessage(
            id=generate_id(),
            user_id=request.user_id,
            channel=request.channel,
            subject=rendered.subject,
            body=rendered.body,
            priority=request.priority,
            metadata=request.metadata
        )
        self.queue.enqueue(message)
        
        # 5. Set dedup key
        self.dup_store.set(dedup_key, message.id, ttl=request.dedup_window)
        
        return SendResult(notification_id=message.id, status="queued")

Worker Pool Design

class PushNotificationWorker:
    def __init__(self, queue, apns_client, fcm_client, delivery_log):
        self.queue = queue
        self.apns = apns_client
        self.fcm = fcm_client
        self.delivery_log = delivery_log
    
    def process_messages(self):
        while True:
            message = self.queue.dequeue(channel="push", timeout=30)
            if message is None:
                continue
            
            try:
                devices = self.get_user_devices(message.user_id)
                
                for device in devices:
                    if device.platform == "ios":
                        result = self.apns.send(
                            device_token=device.token,
                            title=message.subject,
                            body=message.body,
                            badge=self.get_unread_count(message.user_id),
                            category=message.metadata.get("category")
                        )
                    elif device.platform == "android":
                        result = self.fcm.send(
                            token=device.token,
                            title=message.subject,
                            body=message.body,
                            data=message.metadata
                        )
                    
                    self.delivery_log.record(
                        notification_id=message.id,
                        channel="push",
                        device_token=device.token,
                        status="sent" if result.success else "failed",
                        provider_message_id=result.message_id,
                        error=result.error
                    )
                    
                    if not result.success and result.is_retryable:
                        self.queue.retry(message, delay=calculate_backoff(message.attempts))
                    
            except Exception as e:
                self.queue.retry(message, delay=calculate_backoff(message.attempts))
                self.log.error(f"Failed to process push notification: {e}")

Retry Logic & Dead Letter Queue

def calculate_backoff(attempt, base_delay=1, max_delay=300):
    """Exponential backoff with jitter"""
    delay = min(base_delay * (2 ** attempt), max_delay)
    jitter = random.uniform(0, delay * 0.1)
    return delay + jitter

class RetryPolicy:
    MAX_RETRIES = 5
    
    def should_retry(self, message, error):
        if message.attempts >= self.MAX_RETRIES:
            return False
        
        # Don't retry client errors (invalid token, bounced email)
        if error.is_client_error:
            return False
        
        # Retry on server errors, timeouts, rate limits
        return error.is_server_error or error.is_timeout or error.is_rate_limited

Message Flow:
  Normal: API -> Queue -> Worker -> Provider -> Delivery Log
  Retry:  Queue -> Worker -> Provider (fails) -> Queue (delayed)
  DLQ:    Queue -> Worker -> Provider (fails 5x) -> Dead Letter Queue
  DLQ Process: Manual review -> Fix/Requeue -> Delete

Deduplication Strategy

• Idempotency key: hash(user_id + template_id + window)
• Window: 5 minutes for transactional, 24 hours for marketing
• Storage: Redis with TTL matching dedup window
• Prevents: Duplicate notifications from retries, multiple API calls

Template System

Template Schema

{
    "template_id": "order-shipped-en",
    "name": "Order Shipped",
    "channel": "email",
    "language": "en",
    "subject": "Your order {{order_id}} has shipped!",
    "body": {
        "html": "<h1>Hi {{user_name}}</h1><p>Your order {{order_id}} containing {{item_count}} items has shipped via {{carrier}}.</p><p>Track: <a href='{{tracking_url}}'>{{tracking_url}}</a></p>",
        "text": "Hi {{user_name}}, Your order {{order_id}} has shipped via {{carrier}}. Track: {{tracking_url}}"
    },
    "variables": [
        {"name": "order_id", "type": "string", "required": true},
        {"name": "user_name", "type": "string", "required": true},
        {"name": "item_count", "type": "number", "required": true},
        {"name": "carrier", "type": "string", "required": true},
        {"name": "tracking_url", "type": "url", "required": true}
    ],
    "fallback_template_id": "order-shipped-default"
}

Template Engine

class TemplateEngine:
    def __init__(self, template_store, i18n_store):
        self.template_store = template_store
        self.i18n_store = i18n_store
    
    def render(self, template_id, variables, language="en"):
        template = self.template_store.get(template_id, language)
        
        if template is None:
            # Fallback to default language
            template = self.template_store.get(template_id, "en")
        
        if template is None:
            raise TemplateNotFoundException(template_id)
        
        # Validate required variables
        for var in template.variables:
            if var.required and var.name not in variables:
                raise MissingVariableException(var.name)
        
        # Render with variable substitution
        rendered_subject = self._substitute(template.subject, variables)
        rendered_html = self._substitute(template.body["html"], variables)
        rendered_text = self._substitute(template.body["text"], variables)
        
        # Apply i18n for numbers and dates
        rendered_html = self._apply_i18n(rendered_html, variables, language)
        
        return RenderedTemplate(
            subject=rendered_subject,
            body={"html": rendered_html, "text": rendered_text}
        )
    
    def _substitute(self, template_str, variables):
        result = template_str
        for key, value in variables.items():
            result = result.replace(f"{{{{{key}}}}}", str(value))
        return result
    
    def _apply_i18n(self, content, variables, language):
        # Format numbers: 1000 -> "1,000" (en) or "1.000" (de)
        # Format dates: 2026-08-16 -> "August 16, 2026" (en)
        # Format currency: 29.99 -> "$29.99" (en) or "29,99 €" (de)
        pass

Internationalization (i18n)

• Store templates per language (order-shipped-en, order-shipped-es, order-shipped-de)
• User preference determines language selection
• Fallback chain: user preference -> default language -> English
• Number/date/currency formatting based on locale

User Preference System

Preference Schema

{
    "user_id": "user_123",
    "channels": {
        "push": {
            "enabled": true,
            "topics": ["order_updates", "promotions", "security"]
        },
        "email": {
            "enabled": true,
            "topics": ["order_updates", "digest", "security"]
        },
        "sms": {
            "enabled": true,
            "topics": ["security", "otp"]
        },
        "in_app": {
            "enabled": true,
            "topics": ["all"]
        }
    },
    "quiet_hours": {
        "enabled": true,
        "start": "22:00",
        "end": "08:00",
        "timezone": "America/New_York"
    },
    "frequency_caps": {
        "marketing": {
            "max_per_day": 2,
            "max_per_week": 5
        },
        "promotions": {
            "max_per_day": 3,
            "cooldown_hours": 4
        }
    },
    "language": "en"
}

Quiet Hours Implementation

class QuietHoursChecker:
    def should_defer(self, notification, user_preferences):
        if not user_preferences.quiet_hours.enabled:
            return False, None
        
        # Critical notifications bypass quiet hours
        if notification.priority == "critical":
            return False, None
        
        now = datetime.now(pytz.timezone(user_preferences.quiet_hours.timezone))
        quiet_start = self._parse_time(user_preferences.quiet_hours.start)
        quiet_end = self._parse_time(user_preferences.quiet_hours.end)
        
        current_time = now.time()
        
        if quiet_start <= quiet_end:
            # Same day: 22:00 - 08:00 (next day)
            is_quiet = current_time >= quiet_start or current_time <= quiet_end
        else:
            # Overnight: 22:00 - 08:00
            is_quiet = current_time >= quiet_start or current_time <= quiet_end
        
        if is_quiet:
            # Calculate when quiet hours end
            if current_time >= quiet_start:
                # Quiet hours started today, end tomorrow
                defer_until = (now + timedelta(days=1)).replace(
                    hour=quiet_end.hour, minute=quiet_end.minute, second=0
                )
            else:
                # Quiet hours end today
                defer_until = now.replace(
                    hour=quiet_end.hour, minute=quiet_end.minute, second=0
                )
            return True, defer_until
        
        return False, None

Frequency Cap Implementation

class FrequencyCapChecker:
    def __init__(self, redis_client):
        self.redis = redis_client
    
    def is_capped(self, user_id, notification_type, caps_config):
        config = caps_config.get(notification_type)
        if config is None:
            return False
        
        # Check daily cap
        if "max_per_day" in config:
            daily_key = f"freq:daily:{user_id}:{notification_type}:{today()}"
            daily_count = self.redis.get(daily_key) or 0
            if int(daily_count) >= config["max_per_day"]:
                return True
        
        # Check weekly cap
        if "max_per_week" in config:
            weekly_key = f"freq:weekly:{user_id}:{notification_type}:{this_week()}"
            weekly_count = self.redis.get(weekly_key) or 0
            if int(weekly_count) >= config["max_per_week"]:
                return True
        
        # Check cooldown
        if "cooldown_hours" in config:
            last_sent_key = f"freq:last:{user_id}:{notification_type}"
            last_sent = self.redis.get(last_sent_key)
            if last_sent:
                hours_since = (time.time() - float(last_sent)) / 3600
                if hours_since < config["cooldown_hours"]:
                    return True
        
        return False
    
    def increment_counter(self, user_id, notification_type):
        pipe = self.redis.pipeline(True)
        daily_key = f"freq:daily:{user_id}:{notification_type}:{today()}"
        weekly_key = f"freq:weekly:{user_id}:{notification_type}:{this_week()}"
        last_key = f"freq:last:{user_id}:{notification_type}"
        
        pipe.incr(daily_key)
        pipe.expire(daily_key, 86400)
        pipe.incr(weekly_key)
        pipe.expire(weekly_key, 604800)
        pipe.set(last_key, time.time())
        pipe.execute()

Delivery Tracking & Analytics

Delivery States

QUEUED -> PROCESSING -> SENT -> DELIVERED -> OPENED -> CLICKED
   |          |           |         |
   v          v           v         v
FAILED    RETRYING    BOUNCED   DISMISSED
   |
   v
DEAD_LETTER

Analytics Schema

CREATE TABLE delivery_logs (
    id UUID PRIMARY KEY,
    notification_id UUID NOT NULL,
    user_id VARCHAR(64) NOT NULL,
    channel VARCHAR(16) NOT NULL,
    template_id VARCHAR(64),
    status VARCHAR(16) NOT NULL,
    provider_message_id VARCHAR(128),
    error_code VARCHAR(32),
    error_message TEXT,
    metadata JSONB,
    created_at TIMESTAMP DEFAULT NOW(),
    sent_at TIMESTAMP,
    delivered_at TIMESTAMP,
    opened_at TIMESTAMP,
    clicked_at TIMESTAMP
);

CREATE INDEX idx_delivery_user ON delivery_logs(user_id, created_at);
CREATE INDEX idx_delivery_status ON delivery_logs(status, created_at);
CREATE INDEX idx_delivery_template ON delivery_logs(template_id, created_at);

Key Metrics

Metric	Formula	Target
Delivery Rate	delivered / sent	> 98%
Open Rate	opened / delivered	> 20% (email)
Click Rate	clicked / opened	> 5% (email)
Bounce Rate	bounced / sent	< 2%
Latency (p99)	sent_at - created_at	< 5s (push)
Retry Rate	retried / sent	< 1%

Real-Time Dashboard Queries

-- Delivery rate per channel per hour
SELECT 
    channel,
    DATE_TRUNC('hour', created_at) AS hour,
    COUNT(*) AS total,
    COUNT(CASE WHEN status = 'delivered' THEN 1 END) AS delivered,
    ROUND(COUNT(CASE WHEN status = 'delivered' THEN 1 END)::decimal / COUNT(*) * 100, 2) AS delivery_rate
FROM delivery_logs
WHERE created_at > NOW() - INTERVAL '24 hours'
GROUP BY channel, hour
ORDER BY hour DESC;

-- Bounced emails by domain
SELECT 
    SPLIT_PART(metadata->>'email', '@', 2) AS domain,
    COUNT(*) AS bounce_count
FROM delivery_logs
WHERE status = 'bounced' AND channel = 'email'
    AND created_at > NOW() - INTERVAL '7 days'
GROUP BY domain
ORDER BY bounce_count DESC;