Implement BoundedChannel and multithreaded logging infra

tooling/pipeline/includes/biergarten_data_generator.h

@@ -3,7 +3,10 @@
 
 /**
  * @file biergarten_data_generator.h
- * @brief Core orchestration class for pipeline data generation.
+ * @brief Orchestration for end-to-end brewery data generation pipeline.
+ *
+ * Intent: Coordinates location loading, enrichment, and generation phases
+ * to produce a complete dataset. Coordinates dependencies via composition root.
  */
 
 #include <memory>
@@ -21,19 +24,21 @@
  * This class encapsulates the core logic for generating brewery data.
  * It handles location loading, city enrichment, and brewery generation.
  */
-class BiergartenDataGenerator {
+class BiergartenPipelineOrchestrator {
  public:
-  /**
-   * @brief Construct a BiergartenDataGenerator with injected dependencies.
-   *
-   * @param context_service Context provider for sampled locations.
-   * @param generator Brewery and user data generator.
-   * @param exporter Storage backend for generated brewery data.
-   */
-  BiergartenDataGenerator(std::unique_ptr<IEnrichmentService> context_service,
-                          std::unique_ptr<DataGenerator> generator,
-                          std::unique_ptr<IExportService> exporter,
-                          const ApplicationOptions& application_options);
+/**
+ * @brief Constructs the orchestrator with injected pipeline dependencies.
+ *
+ * @param context_service Provides regional context for locations.
+ * @param generator Implementation (Llama or Mock) for brewery/user generation.
+ * @param exporter Database backend for persisting generated records.
+ * @param application_options CLI configuration and paths.
+ */
+  BiergartenPipelineOrchestrator(
+      std::unique_ptr<IEnrichmentService> context_service,
+      std::unique_ptr<DataGenerator> generator,
+      std::unique_ptr<IExportService> exporter,
+      const ApplicationOptions& application_options);
 
   /**
    * @brief Run the data generation pipeline.
@@ -54,10 +59,11 @@ class BiergartenDataGenerator {
   /// @brief Generator dependency selected in the composition root.
   std::unique_ptr<DataGenerator> generator_;
 
-  /// @brief Storage backend for generated brewery records.
-  std::unique_ptr<IExportService> exporter_;
+   /// @brief Storage backend for generated brewery records.
+   std::unique_ptr<IExportService> exporter_;
 
-  const ApplicationOptions application_options_;
+   /// @brief CLI configuration: paths, model settings, generation parameters.
+   ApplicationOptions application_options_;
 
   /**
    * @brief Load locations from JSON and sample cities.

tooling/pipeline/includes/concurrency/bounded_channel.h

@@ -0,0 +1,73 @@
+#ifndef BIERGARTEN_PIPELINE_INCLUDES_CONCURRENCY_BOUNDED_CHANNEL_H_
+#define BIERGARTEN_PIPELINE_INCLUDES_CONCURRENCY_BOUNDED_CHANNEL_H_
+
+#include <condition_variable>
+#include <cstddef>
+#include <mutex>
+#include <optional>
+#include <queue>
+
+/**
+ * @file bounded_channel.h
+ * @brief Thread-safe, bounded multi-producer/multi-consumer synchronous channel.
+ *
+ * Intent: Enables asynchronous inter-thread communication with backpressure.
+ * Models a synchronous channel where producers/consumers block on capacity limits.
+ */
+
+/**
+ * @class BoundedChannel
+ * @brief MPMC channel with fixed capacity and blocking semantics.
+ *
+ * Producers block when buffer is full; consumers block when empty.
+ * Close() unblocks all waiters and signals channel exhaustion.
+ */
+template <typename T>
+class BoundedChannel {
+  // -------------------------------------------------------------------------
+  // Internal state — all access must be guarded by mutex_.
+  // -------------------------------------------------------------------------
+
+  std::queue<T> queue_;
+
+  std::mutex mutex_;
+
+  std::condition_variable not_full_;
+
+  std::condition_variable not_empty_;
+
+  std::size_t capacity_;
+
+  bool closed_ = false;
+
+ public:
+  /**
+   * @brief Construct a bounded channel with the given capacity.
+   * @param capacity Maximum number of items the channel may hold.
+   */
+  explicit BoundedChannel(std::size_t capacity) : capacity_(capacity) {}
+
+  /**
+   * @brief Send an item into the channel. Blocks when the channel is full.
+   * @param item Move-only item to enqueue.
+   */
+  void Send(T item);
+
+  /**
+   * @brief Receive an item from the channel. Blocks when the channel is
+   * empty.
+   * @return std::optional<T> containing the item, or std::nullopt when the
+   * channel is closed and drained.
+   */
+  std::optional<T> Receive();
+
+  /**
+   * @brief Close the channel and unblock all waiting threads. Idempotent.
+   */
+  void Close();
+};
+
+// Include the template implementation
+#include "bounded_channel.tcc"
+
+#endif  // BIERGARTEN_PIPELINE_INCLUDES_CONCURRENCY_BOUNDED_CHANNEL_H_

tooling/pipeline/includes/concurrency/bounded_channel.tcc

@@ -0,0 +1,57 @@
+#include "bounded_channel.h"
+
+template <typename T>
+void BoundedChannel<T>::Send(T item) {
+  // Acquire exclusive ownership of the mutex; released automatically on scope exit.
+  std::unique_lock lock(mutex_);
+
+  // Block until there is space in the queue or the channel has been closed.
+  // The predicate guards against spurious wakeups.
+  not_full_.wait(lock, [&] { return queue_.size() < capacity_ || closed_; });
+
+  // If the channel was closed while waiting, discard the item and return.
+  if (closed_) return;
+
+  // Move the item into the queue to avoid an unnecessary copy.
+  queue_.push(std::move(item));
+
+  // Wake one blocked Receive() call to signal that data is now available.
+  not_empty_.notify_one();
+}
+
+template <typename T>
+std::optional<T> BoundedChannel<T>::Receive() {
+  // Acquire exclusive ownership of the mutex.
+  std::unique_lock lock(mutex_);
+
+  // Block until the queue is non-empty or the channel has been closed.
+  // The predicate guards against spurious wakeups.
+  not_empty_.wait(lock, [&] { return !queue_.empty() || closed_; });
+
+  // If woken due to closure and no items remain, signal exhaustion via nullopt.
+  if (queue_.empty()) return std::nullopt;
+
+  // Move the front item out of the queue to avoid an unnecessary copy.
+  T item = std::move(queue_.front());
+  queue_.pop();
+
+  // Wake one blocked Send() call to signal that a slot has opened.
+  not_full_.notify_one();
+
+  return item;
+}
+
+template <typename T>
+void BoundedChannel<T>::Close() {
+  // Acquire exclusive ownership of the mutex to ensure visibility of the flag.
+  std::unique_lock lock(mutex_);
+
+  // Mark the channel as closed; subsequent Send() calls will be dropped.
+  closed_ = true;
+
+  // Wake all blocked Send() callers so they can observe the closed flag and exit.
+  not_full_.notify_all();
+
+  // Wake all blocked Receive() callers so they can drain remaining items or return nullopt.
+  not_empty_.notify_all();
+}

tooling/pipeline/includes/services/logging/channel_logger.h

@@ -0,0 +1,54 @@
+/**
+ * @file services/logging/channel_logger.h
+ * @brief Channel-backed producer for asynchronous pipeline logging.
+ *
+ * Intent: Decouple logging from synchronous I/O by forwarding entries to a
+ * bounded channel. LogConsumer drains the channel on a dedicated thread.
+ */
+
+#ifndef BIERGARTEN_PIPELINE_INCLUDES_SERVICES_LOGGING_CHANNEL_LOGGER_H_
+#define BIERGARTEN_PIPELINE_INCLUDES_SERVICES_LOGGING_CHANNEL_LOGGER_H_
+
+#include <string_view>
+
+#include "concurrency/bounded_channel.h"
+#include "services/logging/log_entry.h"
+#include "services/logging/logger.h"
+
+/**
+ * @class ChannelLogger
+ * @brief ILogger implementation that sends entries to a BoundedChannel.
+ *
+ * Non-copyable, non-movable. Holds a non-owning reference to the channel.
+ */
+class ChannelLogger final : public ILogger {
+ public:
+   /**
+    * @brief Construct a channel-backed logger.
+    *
+    * @param channel Reference to bounded channel for log entry transfer.
+    *                Channel must outlive this logger instance.
+    */
+   explicit ChannelLogger(BoundedChannel<LogEntry>& channel);
+
+   ChannelLogger(const ChannelLogger&) = delete;
+   ChannelLogger& operator=(const ChannelLogger&) = delete;
+   ChannelLogger(ChannelLogger&&) = delete;
+   ChannelLogger& operator=(ChannelLogger&&) = delete;
+
+   ~ChannelLogger() override = default;
+
+   /**
+    * @brief Queue a log entry for asynchronous processing.
+    *
+    * Blocks if the channel is full (backpressure). Returns immediately
+    * if the channel is closed.
+    */
+   void Log(LogLevel level, PipelinePhase phase,
+            std::string_view message) override;
+
+ private:
+   BoundedChannel<LogEntry>& channel_;
+};
+
+#endif  // BIERGARTEN_PIPELINE_INCLUDES_SERVICES_LOGGING_CHANNEL_LOGGER_H_

tooling/pipeline/includes/services/logging/log_consumer.h

@@ -0,0 +1,59 @@
+/**
+ * @file services/logging/log_consumer.h
+ * @brief Dedicated log consumer/drain for asynchronous pipeline logging.
+ *
+ * Intent: Dequeue LogEntry values from a BoundedChannel on a dedicated thread
+ * and forward them to spdlog for I/O and formatting. Decouples application
+ * logic from logging latency.
+ */
+
+#ifndef BIERGARTEN_PIPELINE_INCLUDES_SERVICES_LOGGING_LOG_CONSUMER_H_
+#define BIERGARTEN_PIPELINE_INCLUDES_SERVICES_LOGGING_LOG_CONSUMER_H_
+
+#include <spdlog/spdlog.h>
+
+#include <string>
+
+#include "concurrency/bounded_channel.h"
+#include "services/logging/log_entry.h"
+
+/**
+ * @class LogConsumer
+ * @brief Consumes log entries from channel and forwards to spdlog.
+ *
+ * Non-copyable, non-movable. Designed to run on its own dedicated std::thread.
+ * Drains the channel until closure, then exits cleanly.
+ */
+class LogConsumer {
+ public:
+   /**
+    * @brief Construct a log consumer.
+    *
+    * @param channel Reference to bounded channel for log entry retrieval.
+    *                Channel must outlive this consumer instance.
+    */
+   explicit LogConsumer(BoundedChannel<LogEntry>& channel);
+
+   LogConsumer(const LogConsumer&) = delete;
+   LogConsumer& operator=(const LogConsumer&) = delete;
+   LogConsumer(LogConsumer&&) = delete;
+   LogConsumer& operator=(LogConsumer&&) = delete;
+
+   /**
+    * @brief Main loop: drain channel and forward entries to spdlog.
+    *
+    * Intended to be called once on a dedicated thread. Returns when:
+    * - Channel is closed AND all queued entries are drained.
+    *
+    * Thread-safe for use from multiple ChannelLogger instances on other threads.
+    */
+   void Run();
+
+ private:
+   BoundedChannel<LogEntry>& channel_;
+
+   static spdlog::level::level_enum ToSpdlogLevel(LogLevel level);
+   static std::string ToString(PipelinePhase phase);
+};
+
+#endif  // BIERGARTEN_PIPELINE_INCLUDES_SERVICES_LOGGING_LOG_CONSUMER_H_

tooling/pipeline/includes/services/logging/log_entry.h

@@ -0,0 +1,65 @@
+/**
+ * @file services/logging/log_entry.h
+ * @brief POD log entry structure for asynchronous pipeline logging.
+ *
+ * Intent: Lightweight, move-safe data transfer between logging producer
+ * (ChannelLogger) and consumer (LogConsumer) via BoundedChannel<LogEntry>.
+ */
+
+#ifndef BIERGARTEN_PIPELINE_INCLUDES_SERVICES_LOGGING_LOG_ENTRY_H_
+#define BIERGARTEN_PIPELINE_INCLUDES_SERVICES_LOGGING_LOG_ENTRY_H_
+
+#include <chrono>
+#include <string>
+
+/**
+ * @enum LogLevel
+ * @brief Severity levels for log entries.
+ */
+enum class LogLevel {
+   Debug,  ///< Development/debugging information.
+   Info,   ///< General informational messages.
+   Warn,   ///< Warning conditions.
+   Error,  ///< Error conditions.
+};
+
+/**
+ * @enum PipelinePhase
+ * @brief Execution phases for contextual logging.
+ *
+ * Used to tag log entries by their processing stage, enabling phase-specific
+ * analysis and filtering of the execution timeline.
+ */
+enum class PipelinePhase {
+   Startup,                      ///< Initialization and validation.
+   UserGeneration,               ///< User profile generation.
+   BreweryAndBeerGeneration,     ///< Brewery and beer data generation.
+   CheckinGeneration,            ///< Checkin (visit) record generation.
+   RatingGeneration,             ///< Rating and review generation.
+   FollowGeneration,             ///< Follow relationship generation.
+   Teardown,                     ///< Finalization and cleanup.
+};
+
+/**
+ * @struct LogEntry
+ * @brief Single log event for asynchronous processing.
+ *
+ * All fields are value types, allowing safe move semantics across
+ * BoundedChannel without shared ownership or synchronization overhead.
+ */
+struct LogEntry {
+   /// @brief Timestamp when entry was created.
+   std::chrono::system_clock::time_point timestamp =
+       std::chrono::system_clock::now();
+
+   /// @brief Severity level of this entry.
+   LogLevel level;
+
+   /// @brief Pipeline phase when entry was logged.
+   PipelinePhase phase;
+
+   /// @brief Log message text.
+   std::string message;
+};
+
+#endif  // BIERGARTEN_PIPELINE_INCLUDES_SERVICES_LOGGING_LOG_ENTRY_H_

tooling/pipeline/includes/services/logging/logger.h

@@ -0,0 +1,45 @@
+/**
+ * @file services/logging/logger.h
+ * @brief Abstract logging interface for pipeline components.
+ *
+ * Intent: Decouple logging from channel/worker implementation details.
+ * All pipeline components depend on ILogger, enabling swappable backends.
+ */
+
+#ifndef BIERGARTEN_PIPELINE_INCLUDES_SERVICES_LOGGING_LOGGER_H_
+#define BIERGARTEN_PIPELINE_INCLUDES_SERVICES_LOGGING_LOGGER_H_
+
+#include <optional>
+#include <string>
+#include <string_view>
+
+#include "services/logging/log_entry.h"
+
+/**
+ * @class ILogger
+ * @brief Minimal interface for submitting log entries.
+ *
+ * Non-copyable and non-movable. Implementations are typically short-lived,
+ * created and owned by the composition root.
+ */
+class ILogger {
+ public:
+   ILogger() = default;
+   ILogger(const ILogger&) = delete;
+   ILogger& operator=(const ILogger&) = delete;
+   ILogger(ILogger&&) = delete;
+   ILogger& operator=(ILogger&&) = delete;
+   virtual ~ILogger() = default;
+
+   /**
+    * @brief Submit a log entry to the logging subsystem.
+    *
+    * @param level Log level (Debug, Info, Warn, Error).
+    * @param phase Pipeline execution phase (Startup, Generation, Teardown, etc.).
+    * @param message Log message text.
+    */
+   virtual void Log(LogLevel level, PipelinePhase phase,
+                    std::string_view message) = 0;
+};
+
+#endif  // BIERGARTEN_PIPELINE_INCLUDES_SERVICES_LOGGING_LOGGER_H_