Design a notification service

01The design, in the six-step framework

1. Requirements and use cases

Modelled as classes in one process, a notification service takes an event, decides who hears about it and how, and delivers without sending twice. The distributed version is a separate system design problem.

• A product team registers templates per event type and channel.
• A caller sends a notification to a user for an event, with data for the template.
• A user sets preferences: which channels for which categories, plus quiet hours.
• The service retries a failed delivery with backoff and gives up after a limit.
• A repeated send with the same key does nothing.

Out of scope: provider integrations beyond an interface, analytics and link tracking. Non-functional constraints: a new channel is a new class; the caller returns quickly because sending happens on workers; a slow email provider must not delay push; at-least-once delivery to a provider with idempotent acceptance on our side.

2. Core entities and relationships

• Notification: the request: key, user, event type, data, category.
• Delivery: one attempt plan for one channel, with state and attempt count.
• Template: renders a subject and body from data.
• Channel: an interface with one method, send; email, push and SMS implement it.
• ChannelFactory: maps a channel name to a configured channel.
• PreferenceStore: answers whether a user accepts a category on a channel.
• NotificationService: deduplicates, resolves channels, enqueues deliveries.
• DeliveryWorker: one pool per channel; sends and retries.

A worker composes its channel, so each pool has its own client and failure boundary. The service aggregates workers because they run independently and survive any single notification.

3. Interfaces and key methods

class Channel(Protocol):
    def send(self, user_id: str, subject: str, body: str) -> None:
        """Raise TransientError to retry, PermanentError to give up."""

class ChannelFactory:
    def register(self, name: str, ctor: Callable[[], Channel]) -> None: ...
    def create(self, name: str) -> Channel: ...

class PreferenceStore(Protocol):
    def channels_for(self, user_id: str, category: str) -> list[str]: ...

class NotificationService:
    def send(self, n: Notification) -> list[Delivery]:
        """Idempotent by n.key. Enqueue one delivery per allowed channel; return immediately."""

Invariants: one key produces deliveries at most once; a delivery ends in exactly one of SENT, FAILED or SKIPPED; attempts never exceed the limit; a channel's outage never blocks another channel's queue.

4. Design patterns and the choices behind them

Channels as strategies, built by a factory

Each channel sends differently and is used identically, the definition of a strategy. A registering factory means adding WhatsApp is a class and one register call, with no switch in the service. A vendor client is wrapped by an adapter, and the single-method Channel follows interface segregation: a channel that has no subject simply ignores it. If each channel needed a matched template renderer, an abstract factory would produce the pair.

Message and channel vary apart

What is said (a template per event) and how it travels (a channel) vary independently. Keeping them separate is the bridge idea; the alternative is an OrderShippedEmail class for every event and channel pair.

Queues per channel

One bounded queue and worker pool per channel means a slow SMS provider backs up only SMS. A single shared queue is simpler, and a slow channel then eats every worker.

Idempotency by key

Callers retry, so send reserves the key in a set under a lock before doing anything. That is idempotency at intake; the provider call itself remains at-least-once, so pass the key along where the provider accepts one. Publishing domain events to the service via an observer lets order code stay ignorant of notifications altogether.

5. Core flow and code

1. send checks and reserves n.key; a repeat returns an empty list.
2. It asks the PreferenceStore for channels; none means nothing is enqueued.
3. It creates one Delivery per channel and puts each on that channel's worker queue.
4. A worker renders the template and calls channel.send.
5. TransientError sleeps with exponential backoff and requeues; PermanentError or the attempt limit marks FAILED.

from __future__ import annotations
import queue, threading, time
from dataclasses import dataclass, field
from enum import Enum
from string import Template as StrTemplate
from typing import Callable, Optional, Protocol


class TransientError(Exception): ...
class PermanentError(Exception): ...


class DeliveryState(Enum):
    QUEUED = "queued"; SENT = "sent"; FAILED = "failed"


@dataclass(frozen=True)
class Notification:
    key: str
    user_id: str
    event: str
    category: str
    data: dict


@dataclass
class Delivery:
    notification: Notification
    channel: str
    attempts: int = 0
    state: DeliveryState = DeliveryState.QUEUED
    done: threading.Event = field(default_factory=threading.Event, repr=False)


@dataclass(frozen=True)
class Template:
    subject: str
    body: str
    def render(self, data: dict) -> tuple[str, str]:
        return StrTemplate(self.subject).safe_substitute(data), StrTemplate(self.body).safe_substitute(data)


class Channel(Protocol):
    def send(self, user_id: str, subject: str, body: str) -> None: ...


class ConsoleChannel:
    def __init__(self, name: str) -> None: self.name = name
    def send(self, user_id: str, subject: str, body: str) -> None:
        print(self.name, user_id, subject, body)


class ChannelFactory:
    def __init__(self) -> None:
        self.registry: dict[str, Callable[[], Channel]] = {}
    def register(self, name: str, ctor: Callable[[], Channel]) -> None:
        self.registry[name] = ctor
    def create(self, name: str) -> Channel:
        return self.registry[name]()


class PreferenceStore(Protocol):
    def channels_for(self, user_id: str, category: str) -> list[str]: ...


class DictPreferences:
    def __init__(self, default: list[str]) -> None:
        self.default = default
        self.overrides: dict[tuple[str, str], list[str]] = {}
    def set(self, user_id: str, category: str, channels: list[str]) -> None:
        self.overrides[(user_id, category)] = channels
    def channels_for(self, user_id: str, category: str) -> list[str]:
        return self.overrides.get((user_id, category), self.default)


class DeliveryWorker:
    def __init__(self, channel: Channel, templates: dict, max_attempts: int = 4, base_delay: float = 0.1) -> None:
        self.channel, self.templates = channel, templates
        self.max_attempts, self.base_delay = max_attempts, base_delay
        self.queue: queue.Queue[Optional[Delivery]] = queue.Queue(1000)
        self.thread = threading.Thread(target=self.run, daemon=True)
        self.thread.start()

    def run(self) -> None:
        while True:
            d = self.queue.get()
            if d is None:
                return
            self._attempt(d)

    def _attempt(self, d: Delivery) -> None:
        n = d.notification
        subject, body = self.templates[(n.event, d.channel)].render(n.data)
        while d.attempts < self.max_attempts:
            d.attempts += 1
            try:
                self.channel.send(n.user_id, subject, body)
                d.state = DeliveryState.SENT
                break
            except TransientError:
                time.sleep(self.base_delay * 2 ** (d.attempts - 1))
            except PermanentError:
                break
        if d.state is not DeliveryState.SENT:
            d.state = DeliveryState.FAILED
        d.done.set()


class NotificationService:
    def __init__(self, factory: ChannelFactory, prefs: PreferenceStore, templates: dict) -> None:
        self.prefs = prefs
        self.seen: set[str] = set()
        self.lock = threading.Lock()
        self.workers = {name: DeliveryWorker(factory.create(name), templates) for name in factory.registry}

    def send(self, n: Notification) -> list[Delivery]:
        with self.lock:
            if n.key in self.seen:
                return []
            self.seen.add(n.key)
        deliveries = []
        for name in self.prefs.channels_for(n.user_id, n.category):
            d = Delivery(n, name)
            self.workers[name].queue.put(d)
            deliveries.append(d)
        return deliveries

    def stop(self) -> None:
        for w in self.workers.values():
            w.queue.put(None)
            w.thread.join()

6. Extensions, concurrency and what interviewers probe

• Retries that block: sleeping in the worker holds up that channel's queue. The next step is a delay heap and a timer thread that requeues, so backoff costs no worker.
• Quiet hours belong in preference resolution: hold the delivery until the window ends rather than dropping it. Urgent categories bypass, which is one more rule, not a channel change.
• Digest: a batching decorator around Channel collects low-priority messages and sends one summary.
• Unbounded seen set: seen grows forever. Keys need a time to live; say this before the interviewer does.
• Priority: a password reset must overtake a marketing blast; two queues per channel with strict order, or a priority queue, and watch for starvation.
• Test first: same key twice yields one send; a fake channel failing twice then succeeding ends SENT with three attempts; a permanent failure ends FAILED after one attempt; a user with email disabled gets no email.
• Shutdown: stop enqueues a sentinel behind pending deliveries, so a graceful exit drains the queue. A crash loses queued work, which is the honest limit of an in-process design; durable delivery needs a persistent queue, and that is where this class-level answer hands over to the system design one.
• Common mistake: calling providers inside send, on the caller's thread, so that an email outage turns into a checkout outage.

02Where this question was reported

• DesignGurus · DesignGurus case study
• LLD Mastery · Hard; Template Method, channel factories, retries (not Google-tagged)
• Codemia · Codemia Hard
• Educative · Educative: Observer across email/SMS channels
• System Design Handbook · System Design Handbook OOD example
• thita.ai · thita.ai LLD sheet, applied
• Instamock · Tagged Google (SDE2); async, Strategy, queue
• blog.bugfree.ai · 'Notify users when someone posts' under LLD/system design: fan-out, idempotent generation, pagination API, retries