Prep Notebook

Code · LLD machine coding · PhonePe prep

Notification System

Design a service that sends notifications to users over several channels.

19 classes 517 lines ObserverFactoryCustom exceptionsService layerChannels View on GitHub ↗

Question

Design a service that sends notifications to users over several channels.

  • Channels: email, SMS, push and in-app; more can be added.
  • Each user has preferred channels. Send only on channels that are both requested and preferred.
  • Notifications have a title, message, priority and metadata.
  • Delivery is asynchronous. Failed sends are retried.
  • Other parts of the system can observe delivery results (e.g. logging).

Design: singleton service with an executor, Builder for Notification, a Factory for channels, a retry decorator (RetryableNotificationChannel) and Observer for delivery results. Run NotificationSystemDemo.

channel/EmailChannel.java 23 lines GitHub ↗ 
package com.practice.phonepe.questions.notificationsystem.channel;

import com.practice.phonepe.questions.notificationsystem.entities.ChannelType;
import com.practice.phonepe.questions.notificationsystem.entities.Notification;
import com.practice.phonepe.questions.notificationsystem.exception.NotificationDeliveryException;

import java.util.concurrent.ThreadLocalRandom;

public class EmailChannel implements NotificationChannel {

    @Override
    public void send(Notification notification) throws NotificationDeliveryException {
        String email = notification.getRecipient().getEmail();
        if (email == null || email.isBlank()) {
            throw new NotificationDeliveryException("User has no email on file");
        }
        if (ThreadLocalRandom.current().nextInt(100) < 20) {
            throw new NotificationDeliveryException("SMTP gateway timeout for " + email);
        }
        System.out.printf("[EMAIL] to=%s subject=\"%s\" body=\"%s\"%n",
                email, notification.getTitle(), notification.getMessage());
    }

    @Override
    public ChannelType getType() {
        return ChannelType.EMAIL;
    }
}
channel/InAppChannel.java 15 lines GitHub ↗ 
package com.practice.phonepe.questions.notificationsystem.channel;

import com.practice.phonepe.questions.notificationsystem.entities.ChannelType;
import com.practice.phonepe.questions.notificationsystem.entities.Notification;
import com.practice.phonepe.questions.notificationsystem.exception.NotificationDeliveryException;

public class InAppChannel implements NotificationChannel {

    @Override
    public void send(Notification notification) throws NotificationDeliveryException {
        System.out.printf("[IN_APP] user=%s title=\"%s\" body=\"%s\"%n",
                notification.getRecipient().getId(), notification.getTitle(), notification.getMessage());
    }

    @Override
    public ChannelType getType() {
        return ChannelType.IN_APP;
    }
}
channel/NotificationChannel.java 13 lines GitHub ↗ 
package com.practice.phonepe.questions.notificationsystem.channel;

import com.practice.phonepe.questions.notificationsystem.entities.ChannelType;
import com.practice.phonepe.questions.notificationsystem.entities.Notification;
import com.practice.phonepe.questions.notificationsystem.exception.NotificationDeliveryException;

/**
 * Strategy interface: one implementation per delivery channel (Email, SMS, Push, In-App).
 * New channels are added by implementing this interface + registering in the factory —
 * no existing code changes (Open/Closed).
 */
public interface NotificationChannel {

    void send(Notification notification) throws NotificationDeliveryException;

    ChannelType getType();
}
channel/NotificationChannelFactory.java 23 lines GitHub ↗ 
package com.practice.phonepe.questions.notificationsystem.channel;

import com.practice.phonepe.questions.notificationsystem.entities.ChannelType;

import java.util.EnumMap;
import java.util.Map;

public class NotificationChannelFactory {

    private final Map<ChannelType, NotificationChannel> channels = new EnumMap<>(ChannelType.class);

    public NotificationChannelFactory() {
        register(new EmailChannel());
        register(new SmsChannel());
        register(new PushChannel());
        register(new InAppChannel());
    }

    public final void register(NotificationChannel channel) {
        channels.put(channel.getType(), channel);
    }

    public NotificationChannel get(ChannelType type) {
        NotificationChannel channel = channels.get(type);
        if (channel == null) {
            throw new IllegalArgumentException("No channel registered for type: " + type);
        }
        return channel;
    }
}
channel/PushChannel.java 23 lines GitHub ↗ 
package com.practice.phonepe.questions.notificationsystem.channel;

import com.practice.phonepe.questions.notificationsystem.entities.ChannelType;
import com.practice.phonepe.questions.notificationsystem.entities.Notification;
import com.practice.phonepe.questions.notificationsystem.exception.NotificationDeliveryException;

import java.util.concurrent.ThreadLocalRandom;

public class PushChannel implements NotificationChannel {

    @Override
    public void send(Notification notification) throws NotificationDeliveryException {
        String deviceToken = notification.getRecipient().getDeviceToken();
        if (deviceToken == null || deviceToken.isBlank()) {
            throw new NotificationDeliveryException("User has no registered device");
        }
        if (ThreadLocalRandom.current().nextInt(100) < 20) {
            throw new NotificationDeliveryException("Push provider unreachable for device " + deviceToken);
        }
        System.out.printf("[PUSH] device=%s title=\"%s\" body=\"%s\"%n",
                deviceToken, notification.getTitle(), notification.getMessage());
    }

    @Override
    public ChannelType getType() {
        return ChannelType.PUSH;
    }
}
channel/RetryableNotificationChannel.java 50 lines GitHub ↗ 
package com.practice.phonepe.questions.notificationsystem.channel;

import com.practice.phonepe.questions.notificationsystem.entities.ChannelType;
import com.practice.phonepe.questions.notificationsystem.entities.Notification;
import com.practice.phonepe.questions.notificationsystem.exception.NotificationDeliveryException;

/**
 * Decorator: adds retry-with-backoff around any {@link NotificationChannel}
 * without the underlying channel implementation knowing about retries.
 */
public class RetryableNotificationChannel implements NotificationChannel {

    private final NotificationChannel delegate;
    private final int maxAttempts;
    private final long initialBackoffMillis;

    public RetryableNotificationChannel(NotificationChannel delegate, int maxAttempts, long initialBackoffMillis) {
        this.delegate = delegate;
        this.maxAttempts = maxAttempts;
        this.initialBackoffMillis = initialBackoffMillis;
    }

    @Override
    public void send(Notification notification) throws NotificationDeliveryException {
        long backoff = initialBackoffMillis;
        NotificationDeliveryException lastFailure = null;

        for (int attempt = 1; attempt <= maxAttempts; attempt++) {
            try {
                delegate.send(notification);
                return;
            } catch (NotificationDeliveryException e) {
                lastFailure = e;
                System.out.printf("[RETRY] %s attempt %d/%d failed for notification %s: %s%n",
                        delegate.getType(), attempt, maxAttempts, notification.getId(), e.getMessage());
                if (attempt < maxAttempts) {
                    sleep(backoff);
                    backoff *= 2;
                }
            }
        }
        throw new NotificationDeliveryException(
                "Delivery failed via " + delegate.getType() + " after " + maxAttempts + " attempts", lastFailure);
    }

    private void sleep(long millis) {
        try {
            Thread.sleep(millis);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    @Override
    public ChannelType getType() {
        return delegate.getType();
    }
}
channel/SmsChannel.java 22 lines GitHub ↗ 
package com.practice.phonepe.questions.notificationsystem.channel;

import com.practice.phonepe.questions.notificationsystem.entities.ChannelType;
import com.practice.phonepe.questions.notificationsystem.entities.Notification;
import com.practice.phonepe.questions.notificationsystem.exception.NotificationDeliveryException;

import java.util.concurrent.ThreadLocalRandom;

public class SmsChannel implements NotificationChannel {

    @Override
    public void send(Notification notification) throws NotificationDeliveryException {
        String phone = notification.getRecipient().getPhoneNumber();
        if (phone == null || phone.isBlank()) {
            throw new NotificationDeliveryException("User has no phone number on file");
        }
        if (ThreadLocalRandom.current().nextInt(100) < 20) {
            throw new NotificationDeliveryException("SMS gateway rejected message to " + phone);
        }
        System.out.printf("[SMS] to=%s body=\"%s\"%n", phone, notification.getMessage());
    }

    @Override
    public ChannelType getType() {
        return ChannelType.SMS;
    }
}
entities/ChannelType.java 7 lines GitHub ↗ 
package com.practice.phonepe.questions.notificationsystem.entities;

public enum ChannelType {
    EMAIL,
    SMS,
    PUSH,
    IN_APP
}
entities/Notification.java 74 lines GitHub ↗ 
package com.practice.phonepe.questions.notificationsystem.entities;

import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;

public class Notification {

    private final String id;
    private final User recipient;
    private final String title;
    private final String message;
    private final Priority priority;
    private final Map<String, String> metadata;

    private Notification(Builder builder) {
        this.id = UUID.randomUUID().toString();
        this.recipient = builder.recipient;
        this.title = builder.title;
        this.message = builder.message;
        this.priority = builder.priority;
        this.metadata = Collections.unmodifiableMap(new HashMap<>(builder.metadata));
    }

    public String getId() {
        return id;
    }

    public User getRecipient() {
        return recipient;
    }

    public String getTitle() {
        return title;
    }

    public String getMessage() {
        return message;
    }

    public Priority getPriority() {
        return priority;
    }

    public Map<String, String> getMetadata() {
        return metadata;
    }

    public static Builder builder(User recipient, String message) {
        return new Builder(recipient, message);
    }

    public static class Builder {
        private final User recipient;
        private final String message;
        private String title = "";
        private Priority priority = Priority.MEDIUM;
        private final Map<String, String> metadata = new HashMap<>();

        private Builder(User recipient, String message) {
            if (recipient == null) {
                throw new IllegalArgumentException("recipient is required");
            }
            if (message == null || message.isBlank()) {
                throw new IllegalArgumentException("message is required");
            }
            this.recipient = recipient;
            this.message = message;
        }

        public Builder title(String title) {
            this.title = title;
            return this;
        }

        public Builder priority(Priority priority) {
            this.priority = priority;
            return this;
        }

        public Builder metadata(String key, String value) {
            this.metadata.put(key, value);
            return this;
        }

        public Notification build() {
            return new Notification(this);
        }
    }
}
entities/NotificationStatus.java 6 lines GitHub ↗ 
package com.practice.phonepe.questions.notificationsystem.entities;

public enum NotificationStatus {
    PENDING,
    SENT,
    FAILED
}
entities/Priority.java 6 lines GitHub ↗ 
package com.practice.phonepe.questions.notificationsystem.entities;

public enum Priority {
    LOW,
    MEDIUM,
    HIGH
}
entities/User.java 40 lines GitHub ↗ 
package com.practice.phonepe.questions.notificationsystem.entities;

import java.util.EnumSet;
import java.util.Set;

public class User {

    private final String id;
    private final String name;
    private final String email;
    private final String phoneNumber;
    private final String deviceToken;
    private final Set<ChannelType> preferredChannels;

    public User(String id, String name, String email, String phoneNumber, String deviceToken,
                Set<ChannelType> preferredChannels) {
        this.id = id;
        this.name = name;
        this.email = email;
        this.phoneNumber = phoneNumber;
        this.deviceToken = deviceToken;
        this.preferredChannels = preferredChannels == null || preferredChannels.isEmpty()
                ? EnumSet.noneOf(ChannelType.class)
                : EnumSet.copyOf(preferredChannels);
    }

    public String getId() {
        return id;
    }

    public String getName() {
        return name;
    }

    public String getEmail() {
        return email;
    }

    public String getPhoneNumber() {
        return phoneNumber;
    }

    public String getDeviceToken() {
        return deviceToken;
    }

    public Set<ChannelType> getPreferredChannels() {
        return preferredChannels;
    }
}
exception/NotificationDeliveryException.java 9 lines GitHub ↗ 
package com.practice.phonepe.questions.notificationsystem.exception;

public class NotificationDeliveryException extends Exception {

    public NotificationDeliveryException(String message) {
        super(message);
    }

    public NotificationDeliveryException(String message, Throwable cause) {
        super(message, cause);
    }
}
observer/DeliveryResult.java 29 lines GitHub ↗ 
package com.practice.phonepe.questions.notificationsystem.observer;

import com.practice.phonepe.questions.notificationsystem.entities.ChannelType;
import com.practice.phonepe.questions.notificationsystem.entities.Notification;
import com.practice.phonepe.questions.notificationsystem.entities.NotificationStatus;

public class DeliveryResult {

    private final Notification notification;
    private final ChannelType channel;
    private final NotificationStatus status;
    private final String failureReason;

    public DeliveryResult(Notification notification, ChannelType channel, NotificationStatus status,
                           String failureReason) {
        this.notification = notification;
        this.channel = channel;
        this.status = status;
        this.failureReason = failureReason;
    }

    public Notification getNotification() {
        return notification;
    }

    public ChannelType getChannel() {
        return channel;
    }

    public NotificationStatus getStatus() {
        return status;
    }

    public String getFailureReason() {
        return failureReason;
    }
}
observer/LoggingNotificationObserver.java 14 lines GitHub ↗ 
package com.practice.phonepe.questions.notificationsystem.observer;

import com.practice.phonepe.questions.notificationsystem.entities.NotificationStatus;

public class LoggingNotificationObserver implements NotificationObserver {

    @Override
    public void onDeliveryResult(DeliveryResult result) {
        if (result.getStatus() == NotificationStatus.SENT) {
            System.out.printf("[AUDIT] notification=%s channel=%s status=SENT%n",
                    result.getNotification().getId(), result.getChannel());
        } else {
            System.out.printf("[AUDIT] notification=%s channel=%s status=FAILED reason=%s%n",
                    result.getNotification().getId(), result.getChannel(), result.getFailureReason());
        }
    }
}
observer/NotificationObserver.java 4 lines GitHub ↗ 
package com.practice.phonepe.questions.notificationsystem.observer;

public interface NotificationObserver {

    void onDeliveryResult(DeliveryResult result);
}
observer/NotificationPublisher.java 23 lines GitHub ↗ 
package com.practice.phonepe.questions.notificationsystem.observer;

import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;

/**
 * The Subject/Observable in the Observer pattern. Holds no dispatch logic of its own —
 * it only maintains the list of interested observers and fans a {@link DeliveryResult}
 * out to each of them. {@code NotificationService} depends on this instead of managing
 * observers itself.
 */
public class NotificationPublisher {

    private final List<NotificationObserver> observers = new CopyOnWriteArrayList<>();

    public void subscribe(NotificationObserver observer) {
        observers.add(observer);
    }

    public void unsubscribe(NotificationObserver observer) {
        observers.remove(observer);
    }

    public void publish(DeliveryResult result) {
        for (NotificationObserver observer : observers) {
            observer.onDeliveryResult(result);
        }
    }
}
service/NotificationService.java 102 lines GitHub ↗ 
package com.practice.phonepe.questions.notificationsystem.service;

import com.practice.phonepe.questions.notificationsystem.channel.NotificationChannel;
import com.practice.phonepe.questions.notificationsystem.channel.NotificationChannelFactory;
import com.practice.phonepe.questions.notificationsystem.channel.RetryableNotificationChannel;
import com.practice.phonepe.questions.notificationsystem.entities.ChannelType;
import com.practice.phonepe.questions.notificationsystem.entities.Notification;
import com.practice.phonepe.questions.notificationsystem.entities.NotificationStatus;
import com.practice.phonepe.questions.notificationsystem.exception.NotificationDeliveryException;
import com.practice.phonepe.questions.notificationsystem.observer.DeliveryResult;
import com.practice.phonepe.questions.notificationsystem.observer.NotificationObserver;
import com.practice.phonepe.questions.notificationsystem.observer.NotificationPublisher;

import java.util.EnumMap;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * Orchestrator (Singleton). Resolves the effective set of channels for a notification,
 * dispatches to each one asynchronously (wrapped with retry), and hands the outcome off
 * to {@link NotificationPublisher} (the Observer-pattern Subject) to fan out to observers.
 */
public class NotificationService {

    private static final int MAX_ATTEMPTS = 3;
    private static final long INITIAL_BACKOFF_MILLIS = 200;

    private static volatile NotificationService instance;

    private final NotificationChannelFactory channelFactory;
    private final ExecutorService executor;
    private final NotificationPublisher publisher = new NotificationPublisher();
    private final Map<String, Map<ChannelType, NotificationStatus>> statusByNotification = new ConcurrentHashMap<>();

    private NotificationService() {
        this.channelFactory = new NotificationChannelFactory();
        this.executor = Executors.newFixedThreadPool(4);
    }

    public static NotificationService getInstance() {
        if (instance == null) {
            synchronized (NotificationService.class) {
                if (instance == null) {
                    instance = new NotificationService();
                }
            }
        }
        return instance;
    }

    public void subscribe(NotificationObserver observer) {
        publisher.subscribe(observer);
    }

    public void unsubscribe(NotificationObserver observer) {
        publisher.unsubscribe(observer);
    }

    /** Send using the recipient's own channel preferences. */
    public void send(Notification notification) {
        send(notification, notification.getRecipient().getPreferredChannels());
    }

    /** Send restricted to the intersection of requestedChannels and the recipient's preferences. */
    public void send(Notification notification, Set<ChannelType> requestedChannels) {
        Set<ChannelType> preferred = notification.getRecipient().getPreferredChannels();
        Set<ChannelType> effective = requestedChannels.isEmpty() ? preferred : requestedChannels;

        Map<ChannelType, NotificationStatus> statuses = new EnumMap<>(ChannelType.class);
        for (ChannelType type : effective) {
            if (!preferred.isEmpty() && !preferred.contains(type)) {
                continue;
            }
            statuses.put(type, NotificationStatus.PENDING);
        }
        statusByNotification.put(notification.getId(), statuses);

        for (ChannelType type : statuses.keySet()) {
            dispatchAsync(notification, type);
        }
    }

    private void dispatchAsync(Notification notification, ChannelType type) {
        NotificationChannel channel = new RetryableNotificationChannel(
                channelFactory.get(type), MAX_ATTEMPTS, INITIAL_BACKOFF_MILLIS);

        executor.submit(() -> {
            try {
                channel.send(notification);
                recordResult(notification, type, NotificationStatus.SENT, null);
            } catch (NotificationDeliveryException e) {
                recordResult(notification, type, NotificationStatus.FAILED, e.getMessage());
            }
        });
    }

    private void recordResult(Notification notification, ChannelType type, NotificationStatus status,
                               String failureReason) {
        statusByNotification.get(notification.getId()).put(type, status);
        publisher.publish(new DeliveryResult(notification, type, status, failureReason));
    }

    public Map<ChannelType, NotificationStatus> getStatus(String notificationId) {
        return statusByNotification.getOrDefault(notificationId, Map.of());
    }

    public void shutdown() {
        executor.shutdown();
    }

    public void awaitTerminationQuietly(long timeout, TimeUnit unit) {
        try {
            executor.awaitTermination(timeout, unit);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}
NotificationSystemDemo.java 34 lines GitHub ↗ 
package com.practice.phonepe.questions.notificationsystem;

import com.practice.phonepe.questions.notificationsystem.entities.ChannelType;
import com.practice.phonepe.questions.notificationsystem.entities.Notification;
import com.practice.phonepe.questions.notificationsystem.entities.Priority;
import com.practice.phonepe.questions.notificationsystem.entities.User;
import com.practice.phonepe.questions.notificationsystem.observer.LoggingNotificationObserver;
import com.practice.phonepe.questions.notificationsystem.service.NotificationService;

import java.util.EnumSet;
import java.util.Set;
import java.util.concurrent.TimeUnit;

public class NotificationSystemDemo {

    public static void main(String[] args) throws InterruptedException {
        NotificationService service = NotificationService.getInstance();
        service.subscribe(new LoggingNotificationObserver());

        User alice = new User("u1", "Alice", "alice@example.com", "+1-555-0100", "device-token-abc",
                EnumSet.of(ChannelType.EMAIL, ChannelType.SMS, ChannelType.PUSH));

        User bob = new User("u2", "Bob", "bob@example.com", null, null,
                EnumSet.of(ChannelType.IN_APP));

        Notification orderShipped = Notification.builder(alice, "Your order #4821 has shipped!")
                .title("Order Update")
                .priority(Priority.HIGH)
                .metadata("orderId", "4821")
                .build();
        service.send(orderShipped);

        Notification promo = Notification.builder(bob, "20% off your next purchase")
                .title("Weekend Sale")
                .priority(Priority.LOW)
                .build();
        // Requests EMAIL + IN_APP, but Bob only prefers IN_APP -> only IN_APP is sent.
        service.send(promo, Set.of(ChannelType.EMAIL, ChannelType.IN_APP));

        service.shutdown();
        service.awaitTerminationQuietly(5, TimeUnit.SECONDS);
    }
}