Pipeline: Add LLM and mocked user generation to the pipeline (#230)

tooling/pipeline/src/biergarten_pipeline_orchestrator/biergarten_pipeline_orchestrator.cc

@@ -1,6 +1,6 @@
 /**
  * @file biergarten_pipeline_orchestrator/biergarten_pipeline_orchestrator.cc
- * @brief BiergartenDataGenerator constructor implementation.
+ * @brief BiergartenPipelineOrchestrator constructor implementation.
  */
 
 #include "biergarten_pipeline_orchestrator.h"
@@ -12,9 +12,11 @@ BiergartenPipelineOrchestrator::BiergartenPipelineOrchestrator(
     std::unique_ptr<IEnrichmentService> context_service,
     std::unique_ptr<DataGenerator> generator,
     std::unique_ptr<IExportService> exporter,
-    const ApplicationOptions &app_options)
+    std::unique_ptr<ICuratedDataService> curated_data_service,
+    const ApplicationOptions& application_options)
     : logger_(std::move(logger)),
       context_service_(std::move(context_service)),
       generator_(std::move(generator)),
       exporter_(std::move(exporter)),
-      application_options_(app_options) {}
+      curated_data_service_(std::move(curated_data_service)),
+      application_options_(application_options) {}

tooling/pipeline/src/biergarten_pipeline_orchestrator/generate_breweries.cc

@@ -1,10 +1,11 @@
 /**
  * @file biergarten_pipeline_orchestrator/generate_breweries.cc
- * @brief BiergartenDataGenerator::GenerateBreweries() implementation.
+ * @brief BiergartenPipelineOrchestrator::GenerateBreweries() implementation.
  */
 
 #include <chrono>
 #include <format>
+#include <optional>
 
 #include "biergarten_pipeline_orchestrator.h"
 #include "services/logging/logger.h"
@@ -19,50 +20,69 @@ void BiergartenPipelineOrchestrator::GenerateBreweries(
   size_t skipped_count = 0;
   size_t export_failed_count = 0;
 
-  for (const auto& [location, region_context] : cities) {
+  const auto generate_record =
+      [this, &skipped_count](
+          const Location& location,
+          const std::string& region_context) -> std::optional<BreweryRecord> {
     try {
       const BreweryResult brewery =
           generator_->GenerateBrewery(location, region_context);
-
-      const GeneratedBrewery gen{.location = location, .brewery = brewery};
-
-      generated_breweries_.push_back(gen);
-
-      try {
-        exporter_->ProcessRecord(gen);
-      } catch (const std::exception& export_exception) {
-        ++export_failed_count;
-
-        logger_->Log(
-            {.level = LogLevel::Warn,
-             .phase = PipelinePhase::BreweryAndBeerGeneration,
-             .message =
-                 std::format("[Pipeline] Generated brewery for '{}' ({}) but SQLite export failed: {}",
-                 location.city, location.country, export_exception.what())});
-      }
+      return BreweryRecord{.location = location, .brewery = brewery};
     } catch (const std::exception& e) {
       ++skipped_count;
 
-      logger_->Log({.level = LogLevel::Warn,
-                    .phase = PipelinePhase::BreweryAndBeerGeneration,
-                    .message = std::format("[Pipeline] Skipping city '{}' ({}): brewery generation failed: {}",
-                               location.city, location.country, e.what())});
+      logger_->Log(
+          {.level = LogLevel::Warn,
+           .phase = PipelinePhase::BreweryAndBeerGeneration,
+           .message = std::format("[Pipeline] Skipping city '{}' ({}): brewery "
+                                  "generation failed: {}",
+                                  location.city, location.country, e.what())});
+      return std::nullopt;
     }
+  };
+
+  const auto export_record = [this, &export_failed_count](
+                                 const BreweryRecord& record) {
+    try {
+      exporter_->ProcessRecord(record);
+    } catch (const std::exception& export_exception) {
+      ++export_failed_count;
+      logger_->Log(
+          {.level = LogLevel::Warn,
+           .phase = PipelinePhase::BreweryAndBeerGeneration,
+           .message = std::format("[Pipeline] Generated brewery for '{}' ({}) "
+                                  "but SQLite export failed: {}",
+                                  record.location.city, record.location.country,
+                                  export_exception.what())});
+    }
+  };
+
+  for (const auto& [location, region_context] : cities) {
+    const std::optional<BreweryRecord> record =
+        generate_record(location, region_context);
+    if (!record.has_value()) {
+      continue;
+    }
+
+    generated_breweries_.push_back(*record);
+    export_record(*record);
   }
 
   if (skipped_count > 0) {
-    logger_->Log({.level = LogLevel::Warn,
-                  .phase = PipelinePhase::BreweryAndBeerGeneration,
-                  .message = std::format(
-                      "[Pipeline] Skipped {} city/cities due to generation errors",
-                      skipped_count)});
+    logger_->Log(
+        {.level = LogLevel::Warn,
+         .phase = PipelinePhase::BreweryAndBeerGeneration,
+         .message = std::format(
+             "[Pipeline] Skipped {} city/cities due to generation errors",
+             skipped_count)});
   }
 
   if (export_failed_count > 0) {
-    logger_->Log({.level = LogLevel::Warn,
-                  .phase = PipelinePhase::Teardown,
-                  .message = std::format(
-                      "[Pipeline] Failed to export {} generated brewery/breweries to SQLite",
-                      export_failed_count)});
+    logger_->Log(
+        {.level = LogLevel::Warn,
+         .phase = PipelinePhase::Teardown,
+         .message = std::format("[Pipeline] Failed to export {} generated "
+                                "brewery/breweries to SQLite",
+                                export_failed_count)});
   }
 }

tooling/pipeline/src/biergarten_pipeline_orchestrator/generate_users.cc

@@ -0,0 +1,219 @@
+/**
+ * @file biergarten_pipeline_orchestrator/generate_users.cc
+ * @brief BiergartenPipelineOrchestrator::GenerateUsers() implementation.
+ */
+
+#include <cctype>
+#include <chrono>
+#include <format>
+#include <iterator>
+#include <optional>
+#include <random>
+#include <stdexcept>
+#include <string>
+#include <string_view>
+#include <unordered_set>
+
+#include "biergarten_pipeline_orchestrator.h"
+#include "services/curated_data/curated_json_data_service.h"
+#include "services/logging/logger.h"
+
+namespace {
+std::string Sanitize(std::string_view value) {
+  std::string out;
+  out.reserve(value.size());
+  for (const char character : value) {
+    if (std::isalnum(static_cast<unsigned char>(character)) != 0) {
+      out.push_back(static_cast<char>(
+          std::tolower(static_cast<unsigned char>(character))));
+    }
+  }
+  return out;
+}
+
+std::string BuildEmail(const Name& name,
+                       std::unordered_set<std::string>& used_local_parts) {
+  const std::string base =
+      std::format("{}.{}", Sanitize(name.first_name), Sanitize(name.last_name));
+
+  std::string local_part = base;
+  uint32_t suffix = 1;
+
+  while (used_local_parts.contains(local_part)) {
+    local_part = std::format("{}{}", base, suffix);
+    ++suffix;
+  }
+
+  used_local_parts.insert(local_part);
+
+  return std::format("{}@thebiergarten.app", local_part);
+}
+
+std::string GenerateDateOfBirth(std::mt19937& rng) {
+  using namespace std::chrono;
+
+  constexpr int kMinAge = 19;
+  constexpr int kMaxAge = 48;
+  constexpr int kMaxDayOffset = 364;
+
+  std::uniform_int_distribution<int> age_dist(kMinAge, kMaxAge);
+  std::uniform_int_distribution<int> day_offset_dist(0, kMaxDayOffset);
+
+  const year_month_day today{floor<days>(system_clock::now())};
+  const year_month_day birth_year_anchor{today.year() - years{age_dist(rng)},
+                                         today.month(), today.day()};
+  const sys_days birth_date =
+      sys_days{birth_year_anchor} - days{day_offset_dist(rng)};
+  const year_month_day birth_ymd{birth_date};
+
+  return std::format("{:04}-{:02}-{:02}", static_cast<int>(birth_ymd.year()),
+                     static_cast<unsigned>(birth_ymd.month()),
+                     static_cast<unsigned>(birth_ymd.day()));
+}
+
+std::optional<Name> SampleName(
+    const ForenamesByCountryMap& forenames_by_country,
+    const SurnamesByCountryMap& surnames_by_country,
+    const std::string& iso3166_1, std::mt19937& rng) {
+  const auto forenames_it = forenames_by_country.find(iso3166_1);
+  const auto surnames_it = surnames_by_country.find(iso3166_1);
+
+  if (forenames_it == forenames_by_country.end() ||
+      surnames_it == surnames_by_country.end() ||
+      forenames_it->second.empty() || surnames_it->second.empty()) {
+    return std::nullopt;
+  }
+
+  const ForenameList& forenames = forenames_it->second;
+  const SurnameList& surnames = surnames_it->second;
+
+  std::uniform_int_distribution<size_t> forename_dist(0, forenames.size() - 1);
+  std::uniform_int_distribution<size_t> surname_dist(0, surnames.size() - 1);
+
+  auto forename_it = forenames.begin();
+  std::advance(forename_it, forename_dist(rng));
+
+  auto surname_it = surnames.begin();
+  std::advance(surname_it, surname_dist(rng));
+
+  return Name{.first_name = forename_it->name,
+              .last_name = *surname_it,
+              .gender = forename_it->gender};
+}
+}  // namespace
+
+void BiergartenPipelineOrchestrator::GenerateUsers(
+    std::span<const EnrichedCity> cities) {
+  logger_->Log({.level = LogLevel::Info,
+                .phase = PipelinePhase::UserGeneration,
+                .message = "=== SAMPLE USER GENERATION ==="});
+
+  const PersonasList& personas = curated_data_service_->LoadPersonas();
+
+  if (personas.empty()) {
+    throw std::runtime_error(
+        "No personas available in personas.json for user generation");
+  }
+
+  const ForenamesByCountryMap& forenames_by_country =
+      curated_data_service_->LoadForenamesByCountry();
+  const SurnamesByCountryMap& surnames_by_country =
+      curated_data_service_->LoadSurnamesByCountry();
+
+  std::mt19937 rng(std::random_device{}());
+  std::uniform_int_distribution<size_t> persona_dist(0, personas.size() - 1);
+
+  generated_users_.clear();
+  size_t skipped_count = 0;
+  size_t export_failed_count = 0;
+  std::unordered_set<std::string> used_email_local_parts;
+
+  const auto generate_record =
+      [this, &rng, &skipped_count, &used_email_local_parts](
+          const EnrichedCity& city, const UserPersona& persona,
+          const Name& sampled_name) -> std::optional<UserRecord> {
+    try {
+      const UserResult user =
+          generator_->GenerateUser(city, persona, sampled_name);
+
+      return UserRecord{
+          .location = city.location,
+          .user = user,
+          .email = BuildEmail(sampled_name, used_email_local_parts),
+          .date_of_birth = GenerateDateOfBirth(rng),
+      };
+    } catch (const std::exception& e) {
+      ++skipped_count;
+      logger_->Log({.level = LogLevel::Warn,
+                    .phase = PipelinePhase::UserGeneration,
+                    .message = std::format(
+                        "[Pipeline] Skipping city '{}' ({}): "
+                        "user generation failed: {}",
+                        city.location.city, city.location.country, e.what())});
+      return std::nullopt;
+    }
+  };
+
+  const auto export_record = [this,
+                              &export_failed_count](const UserRecord& record) {
+    try {
+      exporter_->ProcessRecord(record);
+    } catch (const std::exception& export_exception) {
+      ++export_failed_count;
+
+      logger_->Log(
+          {.level = LogLevel::Warn,
+           .phase = PipelinePhase::UserGeneration,
+           .message = std::format("[Pipeline] Generated user for '{}' ({}) but "
+                                  "SQLite export failed: {}",
+                                  record.location.city, record.location.country,
+                                  export_exception.what())});
+    }
+  };
+
+  for (const auto& city : cities) {
+    const std::optional<Name> sampled_name =
+        SampleName(forenames_by_country, surnames_by_country,
+                   city.location.iso3166_1, rng);
+
+    if (!sampled_name.has_value()) {
+      ++skipped_count;
+      logger_->Log({.level = LogLevel::Warn,
+                    .phase = PipelinePhase::UserGeneration,
+                    .message = std::format(
+                        "[Pipeline] Skipping city '{}' ({}): no names "
+                        "available for country '{}'",
+                        city.location.city, city.location.country,
+                        city.location.iso3166_1)});
+      continue;
+    }
+
+    const UserPersona& persona = personas[persona_dist(rng)];
+
+    const std::optional<UserRecord> record =
+        generate_record(city, persona, *sampled_name);
+    if (!record.has_value()) {
+      continue;
+    }
+
+    generated_users_.push_back(*record);
+    export_record(*record);
+  }
+
+  if (skipped_count > 0) {
+    logger_->Log(
+        {.level = LogLevel::Warn,
+         .phase = PipelinePhase::UserGeneration,
+         .message = std::format(
+             "[Pipeline] Skipped {} city/cities during user generation",
+             skipped_count)});
+  }
+
+  if (export_failed_count > 0) {
+    logger_->Log({.level = LogLevel::Warn,
+                  .phase = PipelinePhase::Teardown,
+                  .message = std::format("[Pipeline] Failed to export {} "
+                                         "generated user/users to SQLite",
+                                         export_failed_count)});
+  }
+}

tooling/pipeline/src/biergarten_pipeline_orchestrator/log_results.cc

@@ -1,20 +1,21 @@
 /**
  * @file biergarten_pipeline_orchestrator/log_results.cc
- * @brief BiergartenDataGenerator::LogResults() implementation.
+ * @brief BiergartenPipelineOrchestrator::LogResults() implementation.
  */
 
 #include <boost/json/array.hpp>
 #include <chrono>
 #include <format>
 
-#include "../../includes/json_handling/pretty_print.h"
 #include "biergarten_pipeline_orchestrator.h"
+#include "json_handling/pretty_print.h"
 #include "services/logging/logger.h"
+
 void BiergartenPipelineOrchestrator::LogResults() const {
-  boost::json::array output;
+  boost::json::array brewery_output;
 
   for (const auto& [location, brewery] : generated_breweries_) {
-    output.push_back(boost::json::object{
+    brewery_output.push_back(boost::json::object{
         {"name_en", brewery.name_en},
         {"description_en", brewery.description_en},
         {"name_local", brewery.name_local},
@@ -29,9 +30,30 @@ void BiergartenPipelineOrchestrator::LogResults() const {
                      }}});
   }
 
-  std::ostringstream oss;
-  PrettyPrint(oss, output);
+  std::ostringstream brewery_oss;
+  PrettyPrint(brewery_oss, brewery_output);
   logger_->Log({.level = LogLevel::Info,
                 .phase = PipelinePhase::Teardown,
-                .message = oss.str()});
+                .message = brewery_oss.str()});
+
+  boost::json::array user_output;
+
+  for (const auto& generated_user : generated_users_) {
+    user_output.push_back(boost::json::object{
+        {"first_name", generated_user.user.first_name},
+        {"last_name", generated_user.user.last_name},
+        {"gender", generated_user.user.gender},
+        {"username", generated_user.user.username},
+        {"bio", generated_user.user.bio},
+        {"activity_weight", generated_user.user.activity_weight},
+        {"email", generated_user.email},
+        {"date_of_birth", generated_user.date_of_birth},
+    });
+  }
+
+  std::ostringstream user_oss;
+  PrettyPrint(user_oss, user_output);
+  logger_->Log({.level = LogLevel::Info,
+                .phase = PipelinePhase::Teardown,
+                .message = user_oss.str()});
 }

tooling/pipeline/src/biergarten_pipeline_orchestrator/query_cities_with_countries.cc

@@ -1,17 +1,17 @@
 /**
  * @file biergarten_pipeline_orchestrator/query_cities_with_countries.cc
- * @brief BiergartenDataGenerator::QueryCitiesWithCountries() implementation.
+ * @brief BiergartenPipelineOrchestrator::QueryCitiesWithCountries()
+ * implementation.
  */
 
 #include <algorithm>
 #include <chrono>
-#include <filesystem>
 #include <format>
 #include <iterator>
 #include <random>
 
 #include "biergarten_pipeline_orchestrator.h"
-#include "json_handling/json_loader.h"
+#include "services/curated_data/curated_json_data_service.h"
 #include "services/logging/logger.h"
 
 std::vector<Location>
@@ -20,9 +20,8 @@ BiergartenPipelineOrchestrator::QueryCitiesWithCountries() {
                 .phase = PipelinePhase::Startup,
                 .message = "=== GEOGRAPHIC DATA OVERVIEW ==="});
 
-  const std::filesystem::path locations_path = "locations.json";
-
-  auto all_locations = JsonLoader::LoadLocations(locations_path, logger_);
+  const std::vector<Location>& all_locations =
+      curated_data_service_->LoadLocations();
 
   const size_t sample_count = std::min(
       static_cast<size_t>(application_options_.pipeline.location_count),

tooling/pipeline/src/biergarten_pipeline_orchestrator/run.cc

@@ -1,6 +1,6 @@
 /**
  * @file biergarten_pipeline_orchestrator/run.cc
- * @brief BiergartenDataGenerator::Run() implementation.
+ * @brief BiergartenPipelineOrchestrator::Run() implementation.
  */
 
 #include <chrono>
@@ -22,42 +22,40 @@ bool BiergartenPipelineOrchestrator::Run() {
     for (auto& city : cities) {
       try {
         std::string region_context = context_service_->GetLocationContext(city);
-        // logger_->Log(LogLevel::Debug, PipelinePhase::UserGeneration,
-        //              "[Pipeline] Context for '" + city.city + "' (" +
-        //              city.iso3166_2 + ") gathered:\n" + region_context);
 
         enriched.push_back(
             EnrichedCity{.location = std::move(city),
                          .region_context = std::move(region_context)});
       } catch (const std::exception& exception) {
         ++skipped_count;
-        logger_->Log(
-            {.level = LogLevel::Warn,
-             .phase = PipelinePhase::UserGeneration,
-             .message = std::format(
-                 "[Pipeline] Skipping city '{}' ({}): context lookup failed: {}",
-                 city.city, city.country, exception.what())});
+        logger_->Log({.level = LogLevel::Warn,
+                      .phase = PipelinePhase::Enrichment,
+                      .message = std::format(
+                          "[Pipeline] Skipping city '{}' ({}): context "
+                          "lookup failed: {}",
+                          city.city, city.country, exception.what())});
       }
     }
 
     if (skipped_count > 0) {
-      logger_->Log({.level = LogLevel::Warn,
-                    .phase = PipelinePhase::UserGeneration,
-                    .message = std::format(
-                        "[Pipeline] Skipped {} city/cities due to context lookup errors",
-                        skipped_count)});
+      logger_->Log(
+          {.level = LogLevel::Warn,
+           .phase = PipelinePhase::Enrichment,
+           .message = std::format("[Pipeline] Skipped {} city/cities due "
+                                  "to context lookup errors",
+                                  skipped_count)});
     }
 
+    this->GenerateUsers(enriched);
     this->GenerateBreweries(enriched);
     exporter_->Finalize();
     this->LogResults();
     return true;
   } catch (const std::exception& e) {
-    logger_->Log(
-        {.level = LogLevel::Error,
-         .phase = PipelinePhase::Teardown,
-         .message =
-             std::format("Pipeline execution failed with error: {}", e.what())});
+    logger_->Log({.level = LogLevel::Error,
+                  .phase = PipelinePhase::Teardown,
+                  .message = std::format(
+                      "Pipeline execution failed with error: {}", e.what())});
     return false;
   }
 }