diff --git a/subscriber-notifs-async/CMakeLists.txt b/subscriber-notifs-async/CMakeLists.txt new file mode 100644 index 0000000..672188e --- /dev/null +++ b/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) \ No newline at end of file diff --git a/subscriber-notifs-async/include/app/inventory_manager.hpp b/subscriber-notifs-async/include/app/inventory_manager.hpp new file mode 100644 index 0000000..b83b960 --- /dev/null +++ b/subscriber-notifs-async/include/app/inventory_manager.hpp @@ -0,0 +1,25 @@ + +#pragma once + +#include +#include +#include + +#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 { + private: + std::unordered_map inventory_; + std::shared_ptr thread_pool_; + + public: + explicit InventoryManager(std::shared_ptr thread_pool); + + void setInventory(const ProductId& productId, ProductCount quantity); +}; \ No newline at end of file diff --git a/subscriber-notifs-async/include/app/services.hpp b/subscriber-notifs-async/include/app/services.hpp new file mode 100644 index 0000000..71f3331 --- /dev/null +++ b/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 { + public: + void onNotification(const ProductId& productId, + const ProductCount& quantity) override; +}; + +class EmailService : public Subscriber { + public: + void onNotification(const ProductId& productId, + const ProductCount& quantity) override; +}; + +class LogService : public Subscriber { + public: + void onNotification(const ProductId& productId, + const ProductCount& quantity) override; +}; \ No newline at end of file diff --git a/subscriber-notifs-async/include/core/subscribable.hpp b/subscriber-notifs-async/include/core/subscribable.hpp new file mode 100644 index 0000000..7d4c920 --- /dev/null +++ b/subscriber-notifs-async/include/core/subscribable.hpp @@ -0,0 +1,55 @@ +#pragma once + +#include +#include +#include + +#include "subscriber.hpp" +#include "thread_pool.hpp" + +template +class Subscribable { + using TSubscriber = Subscriber; + + private: + std::unordered_map>> + subscribersMap_; + + public: + void subscribe(const TEventId& eventId, + std::shared_ptr subscriber) { + subscribersMap_[eventId].push_back(subscriber); + } + + void unsubscribe(const TEventId& eventId, + std::shared_ptr 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); + }); + } + } +}; \ No newline at end of file diff --git a/subscriber-notifs-async/include/core/subscriber.hpp b/subscriber-notifs-async/include/core/subscriber.hpp new file mode 100644 index 0000000..af02316 --- /dev/null +++ b/subscriber-notifs-async/include/core/subscriber.hpp @@ -0,0 +1,10 @@ +#pragma once + +template +class Subscriber { + public: + virtual void onNotification(const TEventId& eventId, + const TEventInfo& eventInfo) = 0; + + virtual ~Subscriber() = default; +}; \ No newline at end of file diff --git a/subscriber-notifs-async/include/core/thread_pool.hpp b/subscriber-notifs-async/include/core/thread_pool.hpp new file mode 100644 index 0000000..8eeec40 --- /dev/null +++ b/subscriber-notifs-async/include/core/thread_pool.hpp @@ -0,0 +1,55 @@ +#pragma once + +#include +#include +#include +#include +#include +#include +#include + +class ThreadPool { + public: + explicit ThreadPool( + size_t num_threads = std::thread::hardware_concurrency()); + + ~ThreadPool() noexcept; + + void enqueue(std::function 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 workers_; + + // Task queue + std::queue> tasks_; + + // Synchronization primitives + std::mutex queue_mutex_; + std::condition_variable condition_; + + // Shutdown flag + std::atomic stop_; + + // Track active tasks for wait_for_completion + std::atomic active_tasks_; + std::condition_variable completion_condition_; +}; diff --git a/subscriber-notifs-async/problem.md b/subscriber-notifs-async/problem.md new file mode 100644 index 0000000..1e821a3 --- /dev/null +++ b/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 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 diff --git a/subscriber-notifs-async/src/app/inventory_manager.cpp b/subscriber-notifs-async/src/app/inventory_manager.cpp new file mode 100644 index 0000000..56a4587 --- /dev/null +++ b/subscriber-notifs-async/src/app/inventory_manager.cpp @@ -0,0 +1,13 @@ +#include "app/inventory_manager.hpp" + +InventoryManager::InventoryManager(std::shared_ptr 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_); + } +} \ No newline at end of file diff --git a/subscriber-notifs-async/src/app/services.cpp b/subscriber-notifs-async/src/app/services.cpp new file mode 100644 index 0000000..5ddf53d --- /dev/null +++ b/subscriber-notifs-async/src/app/services.cpp @@ -0,0 +1,35 @@ +#include "app/services.hpp" + +#include +#include +#include + +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; +} \ No newline at end of file diff --git a/subscriber-notifs-async/src/core/thread_pool.cpp b/subscriber-notifs-async/src/core/thread_pool.cpp new file mode 100644 index 0000000..727e54f --- /dev/null +++ b/subscriber-notifs-async/src/core/thread_pool.cpp @@ -0,0 +1,100 @@ +#include "core/thread_pool.hpp" + +#include + +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 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 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(); + } + } + } +} diff --git a/subscriber-notifs-async/src/main.cpp b/subscriber-notifs-async/src/main.cpp new file mode 100644 index 0000000..a341b6c --- /dev/null +++ b/subscriber-notifs-async/src/main.cpp @@ -0,0 +1,34 @@ +#include +#include +#include + +#include "app/inventory_manager.hpp" +#include "app/services.hpp" +#include "core/thread_pool.hpp" + +int main() { + auto threadPool = std::make_shared(4); + InventoryManager inventoryManager(threadPool); + + auto smsService = std::make_shared(); + auto emailService = std::make_shared(); + auto logService = std::make_shared(); + + 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(end - start); + std::cout << "Inventory updates: " << duration.count() << "ms\n"; + + threadPool->wait_for_completion(); + + return 0; +} \ No newline at end of file