Design a Notification System

import time
import queue
import threading
from abc import ABC, abstractmethod
from typing import Dict, List, Optional

class NotificationRequest:
    def __init__(self, request_id: str, user_id: str, event_type: str, payload: dict):
        self.request_id = request_id
        self.user_id = user_id
        self.event_type = event_type
        self.payload = payload

class Channel(ABC):
    @abstractmethod
    def send(self, user_id: str, content: str) -> bool:
        pass

class EmailChannel(Channel):
    def send(self, user_id: str, content: str) -> bool:
        print(f"[Email] Sending to {user_id}: {content}")
        return True

class SmsChannel(Channel):
    def send(self, user_id: str, content: str) -> bool:
        print(f"[SMS] Sending to {user_id}: {content}")
        return True

# Decorator pattern for retry logic
class RetryingChannelDecorator(Channel):
    def __init__(self, inner: Channel, retries: int = 3):
        self.inner = inner
        self.retries = retries
        
    def send(self, user_id: str, content: str) -> bool:
        for attempt in range(self.retries):
            try:
                if self.inner.send(user_id, content):
                    return True
            except Exception:
                time.sleep(2 ** attempt)
        return False

class PreferenceStore:
    def __init__(self):
        self.preferences: Dict[str, List[str]] = {} # user_id -> allowed channels
        self.lock = threading.Lock()
        
    def get_preferred_channels(self, user_id: str) -> List[str]:
        with self.lock:
            return self.preferences.get(user_id, ["email", "sms"])

class NotificationDispatcher:
    def __init__(self, preference_store: PreferenceStore):
        self.queue = queue.Queue()
        self.pref_store = preference_store
        self.channels: Dict[str, Channel] = {
            "email": RetryingChannelDecorator(EmailChannel()),
            "sms": RetryingChannelDecorator(SmsChannel())
        }
        self.sent_records = set()
        self.lock = threading.Lock()
        self.running = False
        
    def enqueue(self, request: NotificationRequest):
        self.queue.put(request)
        
    def start_worker(self):
        self.running = True
        self.worker_thread = threading.Thread(target=self._process_queue, daemon=True)
        self.worker_thread.start()

    def _process_queue(self):
        while self.running:
            try:
                request = self.queue.get(timeout=1.0)
            except queue.Empty:
                continue
                
            preferred = self.pref_store.get_preferred_channels(request.user_id)
            for ch_name in preferred:
                dedup_key = (request.request_id, ch_name)
                with self.lock:
                    if dedup_key in self.sent_records:
                        continue # Already processed
                    self.sent_records.add(dedup_key)
                
                channel = self.channels.get(ch_name)
                if channel:
                    content = f"Event {request.event_type} triggered. Payload: {request.payload}"
                    channel.send(request.user_id, content)
            self.queue.task_done()

import java.util.*;
import java.util.concurrent.*;

class NotificationRequest {
    String requestId;
    String userId;
    String eventType;
    Map<String, String> payload;

    NotificationRequest(String id, String user, String event, Map<String, String> payload) {
        this.requestId = id;
        this.userId = user;
        this.eventType = event;
        this.payload = payload;
    }
}

interface Channel {
    boolean send(String userId, String content);
}

class EmailChannel implements Channel {
    public boolean send(String userId, String content) {
        System.out.println("[Email] Sent to " + userId + ": " + content);
        return true;
    }
}

class SmsChannel implements Channel {
    public boolean send(String userId, String content) {
        System.out.println("[SMS] Sent to " + userId + ": " + content);
        return true;
    }
}

class RetryingChannel implements Channel {
    private final Channel inner;
    private final int maxRetries = 3;

    RetryingChannel(Channel inner) { this.inner = inner; }

    public boolean send(String userId, String content) {
        for (int i = 0; i < maxRetries; i++) {
            try {
                if (inner.send(userId, content)) return true;
            } catch (Exception e) {
                try {
                    Thread.sleep((long) Math.pow(2, i) * 1000);
                } catch (InterruptedException ie) {
                    Thread.currentThread().interrupt();
                }
            }
        }
        return false;
    }
}

class PreferenceStore {
    private final Map<String, List<String>> userPrefs = new ConcurrentHashMap<>();
    
    public List<String> getPreferredChannels(String userId) {
        return userPrefs.getOrDefault(userId, Arrays.asList("email", "sms"));
    }
}

public class NotificationDispatcher {
    private final BlockingQueue<NotificationRequest> queue = new LinkedBlockingQueue<>();
    private final PreferenceStore prefStore;
    private final Map<String, Channel> channels = new HashMap<>();
    private final Set<String> sentRecords = ConcurrentHashMap.newKeySet();
    private final ExecutorService executor = Executors.newSingleThreadExecutor();

    public NotificationDispatcher(PreferenceStore store) {
        this.prefStore = store;
        channels.put("email", new RetryingChannel(new EmailChannel()));
        channels.put("sms", new RetryingChannel(new SmsChannel()));
    }

    public void enqueue(NotificationRequest req) {
        queue.add(req);
    }

    public void start() {
        executor.submit(() -> {
            while (!Thread.currentThread().isInterrupted()) {
                try {
                    NotificationRequest req = queue.take();
                    List<String> preferred = prefStore.getPreferredChannels(req.userId);
                    for (String chName : preferred) {
                        String dedupKey = req.requestId + ":" + chName;
                        if (sentRecords.add(dedupKey)) { // Atomically add to set
                            Channel channel = channels.get(chName);
                            if (channel != null) {
                                String content = "Event: " + req.eventType + ", Data: " + req.payload;
                                channel.send(req.userId, content);
                            }
                        }
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        });
    }
}

#include <iostream>
#include <string>
#include <vector>
#include <unordered_map>
#include <unordered_set>
#include <queue>
#include <mutex>
#include <thread>
#include <chrono>
#include <memory>
#include <cmath>

struct NotificationRequest {
    std::string requestId;
    std::string userId;
    std::string eventType;
    std::unordered_map<std::string, std::string> payload;
};

class Channel {
public:
    virtual ~Channel() = default;
    virtual bool send(const std::string& userId, const std::string& content) = 0;
};

class EmailChannel : public Channel {
public:
    bool send(const std::string& userId, const std::string& content) override {
        std::cout << "[Email] Sent to " << userId << ": " << content << std::endl;
        return true;
    }
};

class SmsChannel : public Channel {
public:
    bool send(const std::string& userId, const std::string& content) override {
        std::cout << "[SMS] Sent to " << userId << ": " << content << std::endl;
        return true;
    }
};

class RetryingChannel : public Channel {
private:
    std::shared_ptr<Channel> inner;
    int maxRetries = 3;
public:
    RetryingChannel(std::shared_ptr<Channel> in) : inner(in) {}
    bool send(const std::string& userId, const std::string& content) override {
        for (int i = 0; i < maxRetries; ++i) {
            try {
                if (inner->send(userId, content)) return true;
            } catch (...) {
                int sleep_sec = static_cast<int>(std::pow(2, i));
                std::this_thread::sleep_for(std::chrono::seconds(sleep_sec));
            }
        }
        return false;
    }
};

class PreferenceStore {
private:
    std::unordered_map<std::string, std::vector<std::string>> preferences;
    std::mutex mtx;
public:
    std::vector<std::string> getPreferredChannels(const std::string& userId) {
        std::lock_guard<std::mutex> lock(mtx);
        auto it = preferences.find(userId);
        if (it != preferences.end()) return it->second;
        return {"email", "sms"};
    }
};

class NotificationDispatcher {
private:
    std::queue<NotificationRequest> requestQueue;
    std::shared_ptr<PreferenceStore> prefStore;
    std::unordered_map<std::string, std::shared_ptr<Channel>> channels;
    std::unordered_set<std::string> sentRecords;
    std::mutex mtx;
    std::thread workerThread;
    bool running = false;

    void processQueue() {
        while (running) {
            NotificationRequest req;
            {
                std::lock_guard<std::mutex> lock(mtx);
                if (requestQueue.empty()) {
                    std::this_thread::sleep_for(std::chrono::milliseconds(100));
                    continue;
                }
                req = requestQueue.front();
                requestQueue.pop();
            }

            auto preferred = prefStore->getPreferredChannels(req.userId);
            for (const auto& chName : preferred) {
                std::string dedupKey = req.requestId + ":" + chName;
                {
                    std::lock_guard<std::mutex> lock(mtx);
                    if (sentRecords.find(dedupKey) != sentRecords.end()) {
                        continue; // Already processed
                    }
                    sentRecords.insert(dedupKey);
                }

                auto it = channels.find(chName);
                if (it != channels.end()) {
                    std::string content = "Event: " + req.eventType;
                    it->second->send(req.userId, content);
                }
            }
        }
    }

public:
    NotificationDispatcher(std::shared_ptr<PreferenceStore> store) : prefStore(store) {
        channels["email"] = std::make_shared<RetryingChannel>(std::make_shared<EmailChannel>());
        channels["sms"] = std::make_shared<RetryingChannel>(std::make_shared<SmsChannel>());
    }

    ~NotificationDispatcher() {
        running = false;
        if (workerThread.joinable()) workerThread.join();
    }

    void enqueue(const NotificationRequest& req) {
        std::lock_guard<std::mutex> lock(mtx);
        requestQueue.push(req);
    }

    void start() {
        running = true;
        workerThread = std::thread(&NotificationDispatcher::processQueue, this);
    }
};