Prep Notebook

Code · LLD machine coding · PhonePe prep

Notification System (v2)

14 classes 263 lines ObserverFactoryService layerChannels View on GitHub ↗

Question

The same problem, solved again more simply: channel factory, a retrying decorator for flaky channels, an observer for audit logs, and per-channel delivery status you can query by notification id. Run Demo.

channel/EmailChannel.java 18 lines GitHub ↗ 
package com.practice.phonepe.questions.notificationsystemv2.channel;

import com.practice.phonepe.questions.notificationsystemv2.entities.ChannelType;
import com.practice.phonepe.questions.notificationsystemv2.entities.Notification;
import com.practice.phonepe.questions.notificationsystemv2.entities.User;

public class EmailChannel implements NotificationChannel {
    @Override
    public void send(User user, Notification notification) {
        if (java.util.concurrent.ThreadLocalRandom.current().nextInt(100) < 40) {
            throw new RuntimeException("SMTP gateway timeout for " + user.getEmail());
        }
        System.out.println("[EMAIL] to=" + user.getEmail() + " subject=\"" + notification.getTitle()
                + "\" : " + notification.getMessage());
    }

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

import com.practice.phonepe.questions.notificationsystemv2.entities.ChannelType;
import com.practice.phonepe.questions.notificationsystemv2.entities.Notification;
import com.practice.phonepe.questions.notificationsystemv2.entities.User;

public interface NotificationChannel {
    void send(User user, Notification notification);
    ChannelType getType();
}
channel/NotificationChannelFactory.java 16 lines GitHub ↗ 
package com.practice.phonepe.questions.notificationsystemv2.channel;

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

import java.util.Map;

public class NotificationChannelFactory {
    private final Map<ChannelType, NotificationChannel> channels;

    public NotificationChannelFactory(Map<ChannelType, NotificationChannel> channels) {
        this.channels = channels;
    }

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

import com.practice.phonepe.questions.notificationsystemv2.entities.ChannelType;
import com.practice.phonepe.questions.notificationsystemv2.entities.Notification;
import com.practice.phonepe.questions.notificationsystemv2.entities.User;

/**
 * Wraps any channel and retries on failure. Keeps retry logic out of every
 * individual channel implementation - a channel just knows how to send once.
 */
public class RetryingNotificationChannel implements NotificationChannel {
    private final NotificationChannel delegate;
    private final int maxAttempts;

    public RetryingNotificationChannel(NotificationChannel delegate, int maxAttempts) {
        this.delegate = delegate;
        this.maxAttempts = maxAttempts;
    }

    @Override
    public void send(User user, Notification notification) {
        for (int attempt = 1; attempt <= maxAttempts; attempt++) {
            try {
                delegate.send(user, notification);
                return;
            } catch (RuntimeException e) {
                System.out.println("[RETRY] " + delegate.getType() + " attempt " + attempt + "/" + maxAttempts
                        + " failed: " + e.getMessage());
                if (attempt == maxAttempts) {
                    throw e;
                }
            }
        }
    }

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

import com.practice.phonepe.questions.notificationsystemv2.entities.ChannelType;
import com.practice.phonepe.questions.notificationsystemv2.entities.Notification;
import com.practice.phonepe.questions.notificationsystemv2.entities.User;

public class SmsChannel implements NotificationChannel {
    @Override
    public void send(User user, Notification notification) {
        System.out.println("[SMS] to=" + user.getPhone() + " : " + notification.getMessage());
    }

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

public enum ChannelType {
    EMAIL, SMS
}
entities/Notification.java 15 lines GitHub ↗ 
package com.practice.phonepe.questions.notificationsystemv2.entities;

import java.util.UUID;

public class Notification {
    private final String id;
    private final String title;
    private final String message;

    public Notification(String title, String message) {
        this.id = UUID.randomUUID().toString();
        this.title = title;
        this.message = message;
    }

    public String getId() { return id; }
    public String getTitle() { return title; }
    public String getMessage() { return message; }
}
entities/NotificationStatus.java 6 lines GitHub ↗ 
package com.practice.phonepe.questions.notificationsystemv2.entities;

public enum NotificationStatus {
    PENDING,
    SENT,
    FAILED
}
entities/User.java 21 lines GitHub ↗ 
package com.practice.phonepe.questions.notificationsystemv2.entities;

import java.util.Set;

public class User {
    private final String id;
    private final String name;
    private final String email;
    private final String phone;
    private final Set<ChannelType> preferredChannels;

    public User(String id, String name, String email, String phone, Set<ChannelType> preferredChannels) {
        this.id = id;
        this.name = name;
        this.email = email;
        this.phone = phone;
        this.preferredChannels = preferredChannels;
    }

    public String getId() { return id; }
    public String getName() { return name; }
    public String getEmail() { return email; }
    public String getPhone() { return phone; }
    public Set<ChannelType> getPreferredChannels() { return preferredChannels; }
}
observer/AuditLogObserver.java 13 lines GitHub ↗ 
package com.practice.phonepe.questions.notificationsystemv2.observer;

import com.practice.phonepe.questions.notificationsystemv2.entities.ChannelType;
import com.practice.phonepe.questions.notificationsystemv2.entities.User;

public class AuditLogObserver implements NotificationObserver {
    @Override
    public void onResult(User user, ChannelType type, boolean success, String detail) {
        if (success) {
            System.out.println("[AUDIT] user=" + user.getId() + " channel=" + type + " status=SENT");
        } else {
            System.out.println("[AUDIT] user=" + user.getId() + " channel=" + type + " status=FAILED reason=" + detail);
        }
    }
}
observer/NotificationEventPublisher.java 20 lines GitHub ↗ 
package com.practice.phonepe.questions.notificationsystemv2.observer;

import com.practice.phonepe.questions.notificationsystemv2.entities.ChannelType;
import com.practice.phonepe.questions.notificationsystemv2.entities.User;

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

/**
 * The Subject. Owns nothing but the observer list - dispatch logic stays in
 * NotificationService, so adding a new observer never touches it.
 */
public class NotificationEventPublisher {
    private final List<NotificationObserver> observers = new CopyOnWriteArrayList<>();

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

    public void publish(User user, ChannelType type, boolean success, String detail) {
        for (NotificationObserver observer : observers) {
            observer.onResult(user, type, success, detail);
        }
    }
}
observer/NotificationObserver.java 6 lines GitHub ↗ 
package com.practice.phonepe.questions.notificationsystemv2.observer;

import com.practice.phonepe.questions.notificationsystemv2.entities.ChannelType;
import com.practice.phonepe.questions.notificationsystemv2.entities.User;

public interface NotificationObserver {
    void onResult(User user, ChannelType type, boolean success, String detail);
}
service/NotificationService.java 53 lines GitHub ↗ 
package com.practice.phonepe.questions.notificationsystemv2.service;

import com.practice.phonepe.questions.notificationsystemv2.channel.NotificationChannelFactory;
import com.practice.phonepe.questions.notificationsystemv2.entities.ChannelType;
import com.practice.phonepe.questions.notificationsystemv2.entities.Notification;
import com.practice.phonepe.questions.notificationsystemv2.entities.NotificationStatus;
import com.practice.phonepe.questions.notificationsystemv2.entities.User;
import com.practice.phonepe.questions.notificationsystemv2.observer.NotificationEventPublisher;

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

public class NotificationService {
    private final NotificationChannelFactory factory;
    private final NotificationEventPublisher publisher;
    private final ExecutorService executor = Executors.newFixedThreadPool(4);
    private final Map<String, Map<ChannelType, NotificationStatus>> statusByNotification = new ConcurrentHashMap<>();

    public NotificationService(NotificationChannelFactory factory, NotificationEventPublisher publisher) {
        this.factory = factory;
        this.publisher = publisher;
    }

    public void notifyUser(User user, Notification notification) {
        for (ChannelType type : user.getPreferredChannels()) {
            executor.submit(() -> {
                try {
                    factory.get(type).send(user, notification);
                    recordStatus(notification, type, NotificationStatus.SENT);
                    publisher.publish(user, type, true, null);
                } catch (RuntimeException e) {
                    recordStatus(notification, type, NotificationStatus.FAILED);
                    publisher.publish(user, type, false, e.getMessage());
                }
            });
        }
    }

    private void recordStatus(Notification notification, ChannelType type, NotificationStatus status) {
        statusByNotification
                .computeIfAbsent(notification.getId(), id -> new ConcurrentHashMap<>())
                .put(type, status);
    }

    /** Per-channel outcome for a given notification - each channel gets its own slot. */
    public Map<ChannelType, NotificationStatus> getStatus(String notificationId) {
        return statusByNotification.getOrDefault(notificationId, Map.of());
    }

    public void shutdown() {
        executor.shutdown();
        try {
            executor.awaitTermination(5, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}
Demo.java 34 lines GitHub ↗ 
package com.practice.phonepe.questions.notificationsystemv2;

import com.practice.phonepe.questions.notificationsystemv2.channel.EmailChannel;
import com.practice.phonepe.questions.notificationsystemv2.channel.NotificationChannelFactory;
import com.practice.phonepe.questions.notificationsystemv2.channel.RetryingNotificationChannel;
import com.practice.phonepe.questions.notificationsystemv2.channel.SmsChannel;
import com.practice.phonepe.questions.notificationsystemv2.entities.ChannelType;
import com.practice.phonepe.questions.notificationsystemv2.entities.Notification;
import com.practice.phonepe.questions.notificationsystemv2.entities.User;
import com.practice.phonepe.questions.notificationsystemv2.observer.AuditLogObserver;
import com.practice.phonepe.questions.notificationsystemv2.observer.NotificationEventPublisher;
import com.practice.phonepe.questions.notificationsystemv2.service.NotificationService;

import java.util.EnumSet;
import java.util.Map;

public class Demo {
    public static void main(String[] args) {
        NotificationChannelFactory factory = new NotificationChannelFactory(Map.of(
                ChannelType.EMAIL, new RetryingNotificationChannel(new EmailChannel(), 3),
                ChannelType.SMS, new SmsChannel()));
        NotificationEventPublisher publisher = new NotificationEventPublisher();
        publisher.subscribe(new AuditLogObserver());
        NotificationService service = new NotificationService(factory, publisher);

        User alice = new User("u1", "Alice", "alice@example.com", "+1-555-0100",
                EnumSet.of(ChannelType.EMAIL, ChannelType.SMS));
        User bob = new User("u2", "Bob", "bob@example.com", "+1-555-0200",
                EnumSet.of(ChannelType.SMS));

        Notification forAlice = new Notification("Order Update", "Your order has shipped!");
        Notification forBob = new Notification("Order Update", "Your order has shipped!");
        service.notifyUser(alice, forAlice);
        service.notifyUser(bob, forBob);

        service.shutdown();
        System.out.println("Alice's notification status per channel: " + service.getStatus(forAlice.getId()));
        System.out.println("Bob's notification status per channel: " + service.getStatus(forBob.getId()));
    }
}