LLD: Notification System¶

Difficulty: Advanced | Time: 40–50 minutes

1. Problem Statement¶

Design a notification service, running inside a single process, that supports multiple delivery channels (email, SMS, push), multiple event types that trigger notifications ("order shipped," "password reset," "comment on your post"), per-user preferences for which channels a user has opted into for each event type, and template-based rendering of the outgoing message.

2. Requirements¶

Functional (in scope):

• Producers (e.g. an OrderService) publish an event without knowing which channels or handlers care about it
• Each registered channel decides, per event, whether the target user wants to be notified on that channel
• A Template renders a human-readable message from an event's data
• A channel that fails to send does not prevent other channels from sending for the same event
• New channels can be added without modifying the dispatch class

Explicitly out of scope for this exercise: durable queues, at-least-once delivery guarantees, retry-with-backoff, dead-letter queues, fan-out to millions of recipients, cross-process/cross-service delivery, provider circuit breakers. All of that belongs to the distributed version — see Notification System.

3. Entities¶

The nouns in the problem statement: NotificationEvent, NotificationChannel (with concrete EmailChannel, SmsChannel, PushChannel), UserPreferences, Template, NotificationService.

NotificationService plays the role of Subject in the Observer pattern; each NotificationChannel is an Observer that subscribes to event types it can potentially handle, then self-filters per delivery using UserPreferences.

4. Class Design¶

classDiagram
    class NotificationEvent {
        +event_id: str
        +event_type: str
        +user_id: str
        +data: dict
        +occurred_at: datetime
    }

    class UserPreferences {
        +user_id: str
        +enabled_channels: dict~str, set~
        +is_enabled(event_type, channel_name) bool
    }

    class Template {
        +event_type: str
        +format_string: str
        +render(data) str
    }

    class NotificationChannel {
        <<interface>>
        +name: str
        +send(user_id, rendered_message)* bool
    }
    class EmailChannel
    class SmsChannel
    class PushChannel

    class NotificationService {
        -_subscribers: dict~str, List~NotificationChannel~~
        -_templates: dict~str, Template~
        -_preferences: dict~str, UserPreferences~
        -_seen_event_ids: set~str~
        +subscribe(event_type, channel) void
        +unsubscribe(event_type, channel) void
        +publish(event) void
    }

    NotificationChannel <|.. EmailChannel
    NotificationChannel <|.. SmsChannel
    NotificationChannel <|.. PushChannel
    NotificationService o-- NotificationChannel : subscribers (Observer)
    NotificationService --> Template : uses
    NotificationService --> UserPreferences : consults
    NotificationService ..> NotificationEvent : publish(event)

Why NotificationService doesn't hold a reference into each channel's internals: the Subject/Observer relationship is deliberately one-directional and shallow — NotificationService knows only that a channel exposes send(), not how EmailChannel talks to an SMTP server. That's the same interface-over-implementation boundary as PricingStrategy in Parking Lot, just applied to Observer instead of Strategy.

5. Patterns Applied¶

• Observer is the headline pattern here. OrderService (or any producer) calls notification_service.publish(event) without knowing or caring which channels are subscribed to "order.shipped" — it might be zero channels, or three. Adding a new channel later means calling subscribe() once at startup, not editing every producer that might trigger a notification. See Design Patterns — Observer.
• Strategy for per-channel send logic. NotificationChannel is an interface; EmailChannel.send(), SmsChannel.send(), and PushChannel.send() each encapsulate a completely different mechanism (SMTP call, SMS gateway call, push provider call) behind the same signature, so NotificationService.publish() never branches on channel type. See Design Patterns — Strategy.
• Template Method, lightly, for rendering: Template.render(data) is a single fixed step (format-string substitution) rather than a full class hierarchy, because the problem statement doesn't name a real variation point in how rendering happens yet — only what gets rendered. If per-channel formatting rules emerged (push needs a 178-character truncation, SMS needs plain text), that would justify promoting Template into a Strategy of its own; don't build that speculatively.
• Explicitly not using a full pub/sub broker or async queue in-process — that's the distributed system's job. Doing it here would add threading/serialization complexity that this exercise's scope (single process, synchronous producers) doesn't call for.

6. Core Code¶

from abc import ABC, abstractmethod
from collections import defaultdict
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass, field
from datetime import datetime
from threading import RLock
import uuid


@dataclass
class NotificationEvent:
    event_type: str
    user_id: str
    data: dict
    event_id: str = field(default_factory=lambda: str(uuid.uuid4()))
    occurred_at: datetime = field(default_factory=datetime.now)


class UserPreferences:
    def __init__(self, user_id: str, enabled_channels: dict[str, set[str]]):
        self.user_id = user_id
        # event_type -> set of channel names the user has opted into
        self.enabled_channels = enabled_channels

    def is_enabled(self, event_type: str, channel_name: str) -> bool:
        return channel_name in self.enabled_channels.get(event_type, set())


class Template:
    def __init__(self, event_type: str, format_string: str):
        self.event_type = event_type
        self.format_string = format_string

    def render(self, data: dict) -> str:
        return self.format_string.format(**data)


class NotificationChannel(ABC):
    name: str

    @abstractmethod
    def send(self, user_id: str, rendered_message: str) -> bool:
        """Return True on success. Must not raise for expected send failures —
        raise only for programmer errors, so the service can isolate real faults."""
        ...


class EmailChannel(NotificationChannel):
    name = "email"

    def send(self, user_id: str, rendered_message: str) -> bool:
        # In production: call an SMTP client / SES SDK here.
        print(f"[email] to={user_id}: {rendered_message}")
        return True


class SmsChannel(NotificationChannel):
    name = "sms"

    def send(self, user_id: str, rendered_message: str) -> bool:
        print(f"[sms] to={user_id}: {rendered_message}")
        return True


class PushChannel(NotificationChannel):
    name = "push"

    def send(self, user_id: str, rendered_message: str) -> bool:
        print(f"[push] to={user_id}: {rendered_message}")
        return True


class NotificationService:
    def __init__(self, max_workers: int = 8):
        self._subscribers: dict[str, list[NotificationChannel]] = defaultdict(list)
        self._templates: dict[str, Template] = {}
        self._preferences: dict[str, UserPreferences] = {}
        self._seen_event_ids: set[str] = set()      # idempotency guard
        self._lock = RLock()                          # protects subscriber list mutation
        self._executor = ThreadPoolExecutor(max_workers=max_workers)

    def register_template(self, template: Template) -> None:
        self._templates[template.event_type] = template

    def set_preferences(self, prefs: UserPreferences) -> None:
        self._preferences[prefs.user_id] = prefs

    def subscribe(self, event_type: str, channel: NotificationChannel) -> None:
        with self._lock:
            self._subscribers[event_type].append(channel)

    def unsubscribe(self, event_type: str, channel: NotificationChannel) -> None:
        with self._lock:
            if channel in self._subscribers.get(event_type, []):
                self._subscribers[event_type].remove(channel)

    def publish(self, event: NotificationEvent) -> dict[str, bool]:
        # Idempotency: a duplicate event_id (retried producer call, etc.) is a no-op.
        with self._lock:
            if event.event_id in self._seen_event_ids:
                return {}
            self._seen_event_ids.add(event.event_id)
            # Snapshot the subscriber list under the lock so a concurrent
            # subscribe()/unsubscribe() mid-dispatch can't mutate the list
            # we're about to iterate.
            channels = list(self._subscribers.get(event.event_type, []))

        if not channels:
            return {}

        template = self._templates.get(event.event_type)
        if template is None:
            raise ValueError(f"no template registered for event_type={event.event_type!r}")
        rendered = template.render(event.data)

        # Snapshot preferences once per publish — a preference change that lands
        # after this snapshot is taken applies to the *next* event, not this one.
        prefs = self._preferences.get(event.user_id)
        eligible = [
            ch for ch in channels
            if prefs is not None and prefs.is_enabled(event.event_type, ch.name)
        ]
        if not eligible:
            return {}                                  # opted out of every channel — silent no-op

        return self._dispatch(event.user_id, rendered, eligible)

    def _dispatch(self, user_id: str, rendered: str, channels: list[NotificationChannel]) -> dict[str, bool]:
        # Fan out to channels concurrently so a slow SMS gateway doesn't delay
        # email/push for the same event.
        futures = {
            self._executor.submit(self._safe_send, ch, user_id, rendered): ch.name
            for ch in channels
        }
        results: dict[str, bool] = {}
        for future in as_completed(futures):
            channel_name = futures[future]
            results[channel_name] = future.result()    # _safe_send never raises
        return results

    @staticmethod
    def _safe_send(channel: NotificationChannel, user_id: str, rendered: str) -> bool:
        # A failure in one channel must never take down dispatch for the others.
        try:
            return channel.send(user_id, rendered)
        except Exception:
            return False

7. Edge Cases¶

8. Concurrency¶

sequenceDiagram
    participant O as OrderService
    participant N as NotificationService
    participant Pr as UserPreferences
    participant Pool as ThreadPoolExecutor
    participant E as EmailChannel
    participant S as SmsChannel
    participant Pu as PushChannel

    O->>N: publish(event)
    activate N
    N->>N: acquire _lock, check _seen_event_ids
    N->>N: channels = list(_subscribers[event_type])
    N->>N: release _lock
    N->>N: rendered = template.render(event.data)
    N->>Pr: prefs.is_enabled(event_type, channel.name) per channel
    Pr-->>N: eligible = [EmailChannel, SmsChannel, PushChannel]
    N->>Pool: submit(_safe_send, EmailChannel, ...)
    N->>Pool: submit(_safe_send, SmsChannel, ...)
    N->>Pool: submit(_safe_send, PushChannel, ...)
    par fan-out, concurrent
        Pool->>E: send(user_id, rendered)
    and
        Pool->>S: send(user_id, rendered)
    and
        Pool->>Pu: send(user_id, rendered)
    end
    Pool-->>N: results as_completed()
    N-->>O: dict[channel_name, bool]
    deactivate N

Two distinct concurrency concerns, and they're solved differently.

Fan-out to channels for one event. _dispatch submits every eligible channel's send() to a ThreadPoolExecutor and gathers results with as_completed, rather than looping and calling send() synchronously one channel at a time. If SmsChannel.send() blocks for 2 seconds on a slow gateway and it ran first in a serial loop, email and push for the same event would wait behind it for no reason — see Concurrency Basics — Race Conditions for why "looks safe because it's sequential" is itself a common source of latency bugs, not just correctness bugs. Concurrent dispatch removes that head-of-line blocking.

Thread-safety of the subscriber list. If channels can be registered or unregistered at runtime (e.g. an admin panel enabling a new PushChannel without a restart), subscribe()/unsubscribe() mutate self._subscribers[event_type] — a plain Python list — while publish() might be reading it concurrently on another thread. Two things make this safe here:

1. All three methods take self._lock. subscribe/unsubscribe wrap their list mutation in the lock; publish takes the same lock just long enough to check-and-mark _seen_event_ids and copy the current subscriber list into a new list. This is the standard fix from Concurrency Basics — Locks: protect the shared mutable structure, not the code that merely reads a copy of it.
2. The lock's critical section is deliberately tiny. It covers "mark this event seen" and "copy the list," not the actual sends — _dispatch's network-bound channel calls run entirely outside the lock. Holding the lock across send() calls would serialize every publish behind whichever one is currently mid-dispatch, defeating the point of the thread pool.

RLock (not a plain Lock) is used because a future extension might have publish() call another service method that also needs the lock on the same thread — reentrant locking avoids a self-deadlock in that case, at negligible cost here.

9. Extensibility¶

Interview Questions¶

Key Takeaways¶

Previous: Logger | Next: Pub/Sub