add an async subscriber impl with thread pool

subscriber-notifs-async/CMakeLists.txt

@@ -0,0 +1,15 @@
+cmake_minimum_required(VERSION 4.0)
+
+project(subscriber-notifs VERSION 1.0)
+set(CMAKE_CXX_STANDARD 20)
+
+file(GLOB_RECURSE SRC_FILES "src/*.cpp")
+file(GLOB_RECURSE HEADER_FILES "include/*.hpp")
+
+add_executable(sub_notifs ${SRC_FILES} ${HEADER_FILES})
+
+target_include_directories(sub_notifs PRIVATE include )
+
+# from: https://cmake.org/cmake/help/latest/module/FindThreads.html
+find_package(Threads REQUIRED)
+target_link_libraries(sub_notifs PRIVATE Threads::Threads)

subscriber-notifs-async/include/app/inventory_manager.hpp

@@ -0,0 +1,25 @@
+
+#pragma once
+
+#include <memory>
+#include <string>
+#include <unordered_map>
+
+#include "core/subscribable.hpp"
+#include "core/thread_pool.hpp"
+
+using ProductCount = int;
+using ProductId = std::string;
+
+// inventory manager is not generic ( works with the specific types ProductId
+// and ProductCount ) but the subscribable is generic
+class InventoryManager : public Subscribable<ProductId, ProductCount> {
+   private:
+    std::unordered_map<ProductId, int> inventory_;
+    std::shared_ptr<ThreadPool> thread_pool_;
+
+   public:
+    explicit InventoryManager(std::shared_ptr<ThreadPool> thread_pool);
+
+    void setInventory(const ProductId& productId, ProductCount quantity);
+};

subscriber-notifs-async/include/app/services.hpp

@@ -0,0 +1,22 @@
+#pragma once
+
+#include "app/inventory_manager.hpp"
+#include "core/subscriber.hpp"
+
+class SmsService : public Subscriber<ProductId, ProductCount> {
+   public:
+    void onNotification(const ProductId& productId,
+                        const ProductCount& quantity) override;
+};
+
+class EmailService : public Subscriber<ProductId, ProductCount> {
+   public:
+    void onNotification(const ProductId& productId,
+                        const ProductCount& quantity) override;
+};
+
+class LogService : public Subscriber<ProductId, ProductCount> {
+   public:
+    void onNotification(const ProductId& productId,
+                        const ProductCount& quantity) override;
+};

subscriber-notifs-async/include/core/subscribable.hpp

@@ -0,0 +1,55 @@
+#pragma once
+
+#include <memory>
+#include <unordered_map>
+#include <vector>
+
+#include "subscriber.hpp"
+#include "thread_pool.hpp"
+
+template <typename TEventId, typename TEventInfo>
+class Subscribable {
+    using TSubscriber = Subscriber<TEventId, TEventInfo>;
+
+   private:
+    std::unordered_map<TEventId, std::vector<std::shared_ptr<TSubscriber>>>
+        subscribersMap_;
+
+   public:
+    void subscribe(const TEventId& eventId,
+                   std::shared_ptr<TSubscriber> subscriber) {
+        subscribersMap_[eventId].push_back(subscriber);
+    }
+
+    void unsubscribe(const TEventId& eventId,
+                     std::shared_ptr<TSubscriber> subscriber) {
+        auto entry = subscribersMap_.find(eventId);
+        if (entry == subscribersMap_.end()) {
+            return;
+        }
+
+        std::erase(entry->second, subscriber);
+        if (entry->second.empty()) {
+            subscribersMap_.erase(entry);
+        }
+    }
+
+    void notify(const TEventId& eventId, const TEventInfo& eventInfo,
+                ThreadPool& threadPool) const {
+        auto entry = subscribersMap_.find(eventId);
+        if (entry == subscribersMap_.end()) {
+            return;
+        }
+
+        // edge case: if a subscriber unsubscribes itself during notification
+        // this modifies the vector and makes my iterators invalid
+        auto subscribersCopy = entry->second;
+        for (auto& subscriber : subscribersCopy) {
+            // Enqueue each notification as a separate task in the thread pool
+            // capture by value next to copy the shared_ptr
+            threadPool.enqueue([subscriber, eventId, eventInfo]() {
+                subscriber->onNotification(eventId, eventInfo);
+            });
+        }
+    }
+};

subscriber-notifs-async/include/core/subscriber.hpp

@@ -0,0 +1,10 @@
+#pragma once
+
+template <typename TEventId, typename TEventInfo>
+class Subscriber {
+   public:
+    virtual void onNotification(const TEventId& eventId,
+                                const TEventInfo& eventInfo) = 0;
+
+    virtual ~Subscriber() = default;
+};

subscriber-notifs-async/include/core/thread_pool.hpp

@@ -0,0 +1,55 @@
+#pragma once
+
+#include <atomic>
+#include <condition_variable>
+#include <functional>
+#include <mutex>
+#include <queue>
+#include <thread>
+#include <vector>
+
+class ThreadPool {
+   public:
+    explicit ThreadPool(
+        size_t num_threads = std::thread::hardware_concurrency());
+
+    ~ThreadPool() noexcept;
+
+    void enqueue(std::function<void()> task);
+
+    // Get the number of worker threads
+    [[nodiscard]] size_t thread_count() const noexcept {
+        return workers_.size();
+    }
+
+    // Wait for all queued tasks to complete
+    void wait_for_completion() noexcept;
+
+    // Disable copy and move
+    // sync primitives are not movable or copyable
+    ThreadPool(const ThreadPool&) = delete;
+    ThreadPool& operator=(const ThreadPool&) = delete;
+    ThreadPool(ThreadPool&&) = delete;
+    ThreadPool& operator=(ThreadPool&&) = delete;
+
+   private:
+    // Worker thread function
+    void worker();
+
+    // Thread management
+    std::vector<std::thread> workers_;
+
+    // Task queue
+    std::queue<std::function<void()>> tasks_;
+
+    // Synchronization primitives
+    std::mutex queue_mutex_;
+    std::condition_variable condition_;
+
+    // Shutdown flag
+    std::atomic<bool> stop_;
+
+    // Track active tasks for wait_for_completion
+    std::atomic<size_t> active_tasks_;
+    std::condition_variable completion_condition_;
+};

subscriber-notifs-async/problem.md

@@ -0,0 +1,61 @@
+# System Design Problem: Stock Availability Notification Service
+
+## Problem Statement
+
+Design a Low-Level Design (LLD) for a "Notify Me When Available" feature on an e-commerce platform. When a high-demand product (like a PS5 or iPhone) is out of stock, customers can subscribe to receive an alert when it becomes available again.
+
+## 1. Functional Requirements (FR)
+
+-   **Product Availability Monitoring:** The system must track the stock quantity of products.
+-   **User Subscription:**
+    -   Users can subscribe to a specific product to receive availability alerts.
+    -   Users can choose their preferred communication channel (Email, SMS, or Push Notification) at the time of subscription.
+    -   Users should be able to unsubscribe if they lose interest.
+-   **Notification Trigger:**
+    -   When a product's stock status changes from "Out of Stock" (0) to "In Stock" (>0), the system must trigger alerts to all subscribed users.
+-   **Extensibility:**
+    -   The system must support adding new notification channels (e.g., WhatsApp, Discord webhook) in the future with minimal code changes to the core product logic.
+
+## 2. Non-Functional Requirements (NFR)
+
+-   **Decoupling:** The inventory management system (Product logic) should not be tightly coupled with the notification logic (3rd party Email/SMS services).
+-   **Real-time/Near Real-time:** Notifications should be sent reasonably quickly after stock is updated.
+-   **Scalability:** A single popular product might have 100,000+ subscribers. The notification process should not block the inventory update transaction (e.g., the warehouse manager updating stock shouldn't wait for 100k emails to send before seeing a "Success" message).
+
+---
+
+# Scratch
+
+-   clearly observer pattern ( satisifies decoupling requirement )
+-   nfr 2 and 3 need async processing so we can use message queues or threads but let me implement a basic version first
+
+```
+interface Subscribable  (observer)
+    + subscribe(Subscriber)
+    + unsubscribe(Subscriber)
+    + notifySubscribers()
+
+interface Subscriber
+    + onNotify(Product)
+
+Product (subject) implements Subscribable
+    - List<Subscriber> subscribers
+    + subscribe(Subscriber)
+    + unsubscribe(Subscriber)
+    - notifySubscribers()
+    + setStock(int)
+        - if stock changes from 0 to >0
+            - notifySubscribers()
+```
+
+# Approach
+
+-   made observer pattern with string messages and global notif ( no filer for specific ids )
+-   changed to Event class with runtime polymorphism using base pointer trick
+-   used generics to do compile type polymorphism
+-   make product id a generic too and change to use map of id to list of subscribers
+-   move off the actual notifictions to a thread pool to handle async processing
+
+# Async Handling
+
+-   make a thrad pool class

subscriber-notifs-async/src/app/inventory_manager.cpp

@@ -0,0 +1,13 @@
+#include "app/inventory_manager.hpp"
+
+InventoryManager::InventoryManager(std::shared_ptr<ThreadPool> thread_pool)
+    : thread_pool_(std::move(thread_pool)) {}
+
+void InventoryManager::setInventory(const ProductId& productId,
+                                    ProductCount quantity) {
+    ProductCount oldQuantity = inventory_[productId];
+    inventory_[productId] = quantity;
+    if (oldQuantity != quantity) {
+        this->notify(productId, quantity, *thread_pool_);
+    }
+}

subscriber-notifs-async/src/app/services.cpp

@@ -0,0 +1,35 @@
+#include "app/services.hpp"
+
+#include <chrono>
+#include <iostream>
+#include <thread>
+
+void SmsService::onNotification(const ProductId& productId,
+                                const ProductCount& quantity) {
+    std::cout << "[SMS] Starting notification for Product: " << productId
+              << ", Quantity: " << quantity << std::endl;
+    // Simulate slow SMS API call
+    std::this_thread::sleep_for(std::chrono::seconds(2));
+    std::cout << "[SMS] Completed notification for Product: " << productId
+              << std::endl;
+}
+
+void EmailService::onNotification(const ProductId& productId,
+                                  const ProductCount& quantity) {
+    std::cout << "[EMAIL] Starting notification for Product: " << productId
+              << ", Quantity: " << quantity << std::endl;
+    // Simulate slow Email API call
+    std::this_thread::sleep_for(std::chrono::seconds(2));
+    std::cout << "[EMAIL] Completed notification for Product: " << productId
+              << std::endl;
+}
+
+void LogService::onNotification(const ProductId& productId,
+                                const ProductCount& quantity) {
+    std::cout << "[LOG] Starting log entry for Product: " << productId
+              << ", Quantity: " << quantity << std::endl;
+    // Simulate slower log write
+    std::this_thread::sleep_for(std::chrono::seconds(1));
+    std::cout << "[LOG] Completed log entry for Product: " << productId
+              << std::endl;
+}

subscriber-notifs-async/src/core/thread_pool.cpp

@@ -0,0 +1,100 @@
+#include "core/thread_pool.hpp"
+
+#include <iostream>
+
+ThreadPool::ThreadPool(size_t num_threads) : stop_(false), active_tasks_(0) {
+    workers_.reserve(num_threads);
+    for (size_t i = 0; i < num_threads; ++i) {
+        workers_.emplace_back([this] { worker(); });
+    }
+}
+
+ThreadPool::~ThreadPool() noexcept {
+    // cv gotcha 1: cv predicate to be touched under lock
+    {
+        std::scoped_lock lock(queue_mutex_);
+        stop_.store(true, std::memory_order_release);
+    }
+
+    condition_.notify_all();
+    for (auto& worker : workers_) {
+        if (worker.joinable()) {
+            worker.join();
+        }
+    }
+}
+
+void ThreadPool::enqueue(std::function<void()> task) {
+    {
+        std::scoped_lock lock(queue_mutex_);
+
+        // Don't accept new tasks if shutting down
+        if (stop_.load(std::memory_order_acquire)) [[unlikely]] {
+            return;
+        }
+
+        tasks_.push(std::move(task));
+    }
+
+    // Notify one worker thread that a task is available
+    condition_.notify_one();
+}
+
+void ThreadPool::wait_for_completion() noexcept {
+    std::unique_lock lock(queue_mutex_);
+    completion_condition_.wait(lock, [this] {
+        return tasks_.empty() &&
+               active_tasks_.load(std::memory_order_acquire) == 0;
+    });
+}
+
+void ThreadPool::worker() {
+    while (true) {
+        std::function<void()> task;
+
+        {
+            std::unique_lock lock(queue_mutex_);
+
+            // Wait for a task or stop signal
+            condition_.wait(lock, [this] {
+                return stop_.load(std::memory_order_acquire) || !tasks_.empty();
+            });
+
+            // Exit if stopping and no tasks remain
+            if (stop_.load(std::memory_order_acquire) && tasks_.empty())
+                [[unlikely]] {
+                return;
+            }
+
+            // Get the next task
+            if (!tasks_.empty()) {
+                task = std::move(tasks_.front());
+                tasks_.pop();
+                active_tasks_.fetch_add(1, std::memory_order_release);
+            }
+        }
+
+        // when can you do if(task)
+        // -> when operator bool() is deined
+        // defined for std::function, shared/unique ptr, optionals etc
+        if (task) {
+            try {
+                task();
+            } catch (const std::exception& e) {
+                std::cerr << "Thread pool task threw exception: " << e.what()
+                          << '\n';
+            } catch (...) {
+                // anything not derived from std::exception
+                std::cerr << "Thread pool task threw unknown exception\n";
+            }
+
+            const auto remaining =
+                active_tasks_.fetch_sub(1, std::memory_order_acq_rel) - 1;
+
+            // Notify anyone waiting for completion if this is the last task
+            if (remaining == 0) [[unlikely]] {
+                completion_condition_.notify_all();
+            }
+        }
+    }
+}

subscriber-notifs-async/src/main.cpp

@@ -0,0 +1,34 @@
+#include <chrono>
+#include <iostream>
+#include <memory>
+
+#include "app/inventory_manager.hpp"
+#include "app/services.hpp"
+#include "core/thread_pool.hpp"
+
+int main() {
+    auto threadPool = std::make_shared<ThreadPool>(4);
+    InventoryManager inventoryManager(threadPool);
+
+    auto smsService = std::make_shared<SmsService>();
+    auto emailService = std::make_shared<EmailService>();
+    auto logService = std::make_shared<LogService>();
+
+    inventoryManager.subscribe("ProductA", smsService);
+    inventoryManager.subscribe("ProductB", emailService);
+    inventoryManager.subscribe("ProductA", logService);
+
+    auto start = std::chrono::steady_clock::now();
+
+    inventoryManager.setInventory("ProductA", 100);
+    inventoryManager.setInventory("ProductB", 200);
+
+    auto end = std::chrono::steady_clock::now();
+    auto duration =
+        std::chrono::duration_cast<std::chrono::milliseconds>(end - start);
+    std::cout << "Inventory updates: " << duration.count() << "ms\n";
+
+    threadPool->wait_for_completion();
+
+    return 0;
+}