Persist generated users to SQLite and code cleanup

tooling/pipeline/CMakeLists.txt

@@ -210,6 +210,7 @@ target_sources(${PROJECT_NAME} PRIVATE
 # --- services: sqlite ---
 target_sources(${PROJECT_NAME} PRIVATE
 	src/services/sqlite/process_record.cc
+	src/services/sqlite/process_user_record.cc
 	src/services/sqlite/sqlite_export_service.cc
 	src/services/sqlite/finalize.cc
 	src/services/sqlite/initialize.cc

tooling/pipeline/includes/data_generation/mock_generator.h

@@ -33,7 +33,7 @@ class MockGenerator final : public DataGenerator {
    *
    * @param city Enriched city the user is associated with.
    * @param persona Persona archetype.
-   * @param name Sampled first/last name. Unused for mock generation.
+   * @param name Sampled first/last name, copied directly into the result.
    * @return Generated user result.
    */
   UserResult GenerateUser(const EnrichedCity& city, const UserPersona& persona,

tooling/pipeline/includes/data_model/models.h

@@ -119,7 +119,7 @@ struct UserPersona {
  * @brief LLM sampling parameters.
  */
 struct SamplingOptions {
-  /// @brief LLM sampling temperature (0.0 to 1.0, higher = more random).
+  /// @brief LLM sampling temperature (higher = more random).
   float temperature = 1.0F;
 
   /// @brief LLM nucleus sampling top-p parameter.

tooling/pipeline/includes/services/database/export_service.h

@@ -35,6 +35,13 @@ class IExportService {
    */
   virtual uint64_t ProcessRecord(const GeneratedBrewery& brewery) = 0;
 
+  /**
+   * @brief Persists one generated user record.
+   *
+   * @param user Generated user payload to store.
+   */
+  virtual uint64_t ProcessRecord(const GeneratedUser& user) = 0;
+
   /// @brief Finalizes the export destination.
   virtual void Finalize() = 0;
 };

tooling/pipeline/includes/services/database/sqlite_export_service.h

@@ -31,6 +31,7 @@ class SqliteExportService final : public IExportService {
 
   void Initialize() override;
   uint64_t ProcessRecord(const GeneratedBrewery& brewery) override;
+  uint64_t ProcessRecord(const GeneratedUser& user) override;
   void Finalize() override;
 
  private:
@@ -46,6 +47,15 @@ class SqliteExportService final : public IExportService {
   [[nodiscard]] std::filesystem::path BuildDatabasePath() const;
   [[nodiscard]] static std::string BuildLocationKey(const Location& location);
 
+  /**
+   * @brief Returns the row id for @p location, inserting it first if it has
+   * not already been seen during this run.
+   *
+   * Shared by both ProcessRecord() overloads so breweries and users
+   * referencing the same location resolve to the same row.
+   */
+  [[nodiscard]] sqlite3_int64 ResolveLocationId(const Location& location);
+
   std::unique_ptr<IDateTimeProvider> date_time_provider_;
   std::filesystem::path output_path_;
   std::string run_timestamp_utc_;
@@ -53,6 +63,7 @@ class SqliteExportService final : public IExportService {
   SqliteDatabaseHandle db_handle_;
   SqliteStatementHandle insert_location_stmt_;
   SqliteStatementHandle insert_brewery_stmt_;
+  SqliteStatementHandle insert_user_stmt_;
   bool transaction_open_ = false;
   std::unordered_map<std::string, sqlite3_int64> location_cache_;
 };

tooling/pipeline/includes/services/database/sqlite_statement_helpers.h

@@ -50,6 +50,27 @@ CREATE INDEX IF NOT EXISTS idx_breweries_location_id ON breweries(location_id);
 
 )sql";
 
+inline constexpr std::string_view kCreateUsersTableSql = R"sql(
+
+CREATE TABLE IF NOT EXISTS users (
+  id INTEGER PRIMARY KEY AUTOINCREMENT,
+  location_id INTEGER NOT NULL,
+  first_name TEXT NOT NULL,
+  last_name TEXT NOT NULL,
+  gender TEXT NOT NULL,
+  username TEXT NOT NULL,
+  bio TEXT NOT NULL,
+  activity_weight REAL NOT NULL,
+  email TEXT NOT NULL UNIQUE,
+  date_of_birth TEXT NOT NULL,
+  password TEXT NOT NULL,
+  FOREIGN KEY(location_id) REFERENCES locations(id) ON DELETE CASCADE
+);
+
+CREATE INDEX IF NOT EXISTS idx_users_location_id ON users(location_id);
+
+)sql";
+
 inline constexpr std::string_view kInsertLocationSql = R"sql(
 INSERT INTO locations (
   city,
@@ -73,6 +94,21 @@ INSERT INTO breweries (
 ) VALUES (?, ?, ?, ?, ?);
 )sql";
 
+inline constexpr std::string_view kInsertUserSql = R"sql(
+INSERT INTO users (
+  location_id,
+  first_name,
+  last_name,
+  gender,
+  username,
+  bio,
+  activity_weight,
+  email,
+  date_of_birth,
+  password
+) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?);
+)sql";
+
 inline constexpr int kLocationCityBindIndex = 1;
 inline constexpr int kLocationStateProvinceBindIndex = 2;
 inline constexpr int kLocationIso31662BindIndex = 3;
@@ -88,6 +124,17 @@ inline constexpr int kBreweryEnglishDescriptionBindIndex = 3;
 inline constexpr int kBreweryLocalNameBindIndex = 4;
 inline constexpr int kBreweryLocalDescriptionBindIndex = 5;
 
+inline constexpr int kUserLocationIdBindIndex = 1;
+inline constexpr int kUserFirstNameBindIndex = 2;
+inline constexpr int kUserLastNameBindIndex = 3;
+inline constexpr int kUserGenderBindIndex = 4;
+inline constexpr int kUserUsernameBindIndex = 5;
+inline constexpr int kUserBioBindIndex = 6;
+inline constexpr int kUserActivityWeightBindIndex = 7;
+inline constexpr int kUserEmailBindIndex = 8;
+inline constexpr int kUserDateOfBirthBindIndex = 9;
+inline constexpr int kUserPasswordBindIndex = 10;
+
 SqliteStatementHandle PrepareStatement(const SqliteDatabaseHandle& db_handle,
                                        std::string_view sql,
                                        const char* action);

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

@@ -35,6 +35,7 @@ enum class LogLevel {
  */
 enum class PipelinePhase {
   Startup,                   ///< Initialization and validation.
+  Enrichment,                ///< Location/context enrichment (e.g. Wikipedia lookups).
   UserGeneration,            ///< User profile generation.
   BreweryAndBeerGeneration,  ///< Brewery and beer data generation.
   CheckinGeneration,         ///< Checkin (visit) record generation.

tooling/pipeline/src/application_options/parse_arguments.cc

@@ -15,8 +15,6 @@ std::optional<ApplicationOptions> ParseArguments(
 
   opt("help,h", "Produce help message");
 
-  // Defaults sourced from SamplingOptions{} so the CLI and LlamaGenerator
-  // share a single source of truth — changing the struct updates both.
   auto add_sampling_options = [&]() -> void {
     const SamplingOptions sampling_defaults{};
     opt("temperature",
@@ -152,7 +150,6 @@ std::optional<ApplicationOptions> ParseArguments(
 
     options.generator.use_mocked = use_mocked;
     options.generator.model_path = model_path;
-    // options.generator.n_gpu_layers = n_gpu_layers;
 
     // Only populate sampling config when the user explicitly overrides at
     // least one value. Leaving it as std::nullopt lets LlamaGenerator fall

tooling/pipeline/src/biergarten_pipeline_orchestrator/generate_users.cc

@@ -107,6 +107,7 @@ void BiergartenPipelineOrchestrator::GenerateUsers(
   generated_users_.clear();
   std::unordered_set<std::string> used_email_local_parts;
   size_t skipped_count = 0;
+  size_t export_failed_count = 0;
 
   for (const auto& city : cities) {
     const std::optional<Name> sampled_name =
@@ -131,13 +132,29 @@ void BiergartenPipelineOrchestrator::GenerateUsers(
       const UserResult user =
           generator_->GenerateUser(city, persona, *sampled_name);
 
-      generated_users_.push_back(GeneratedUser{
+      const GeneratedUser generated_user{
           .location = city.location,
           .user = user,
           .email = BuildEmail(*sampled_name, used_email_local_parts),
           .date_of_birth = GenerateDateOfBirth(rng),
           .password = GenerateRandomPassword(rng),
-      });
+      };
+
+      generated_users_.push_back(generated_user);
+
+      try {
+        exporter_->ProcessRecord(generated_user);
+      } 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: {}",
+                 city.location.city, city.location.country,
+                 export_exception.what())});
+      }
     } catch (const std::exception& e) {
       ++skipped_count;
       logger_->Log(
@@ -157,4 +174,13 @@ void BiergartenPipelineOrchestrator::GenerateUsers(
              "[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/run.cc

@@ -22,9 +22,6 @@ 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),
@@ -33,7 +30,7 @@ bool BiergartenPipelineOrchestrator::Run() {
         ++skipped_count;
         logger_->Log(
             {.level = LogLevel::Warn,
-             .phase = PipelinePhase::UserGeneration,
+             .phase = PipelinePhase::Enrichment,
              .message = std::format(
                  "[Pipeline] Skipping city '{}' ({}): context lookup failed: {}",
                  city.city, city.country, exception.what())});
@@ -42,7 +39,7 @@ bool BiergartenPipelineOrchestrator::Run() {
 
     if (skipped_count > 0) {
       logger_->Log({.level = LogLevel::Warn,
-                    .phase = PipelinePhase::UserGeneration,
+                    .phase = PipelinePhase::Enrichment,
                     .message = std::format(
                         "[Pipeline] Skipped {} city/cities due to context lookup errors",
                         skipped_count)});

tooling/pipeline/src/data_generation/llama/generate_brewery.cc

@@ -91,7 +91,6 @@ BreweryResult LlamaGenerator::GenerateBrewery(
   std::string last_error;
 
   // Token budget: too small risks truncating valid JSON mid-string.
-  // Start conservatively but allow adaptive increases on truncation.
   int max_tokens = kBreweryInitialMaxTokens;
 
   // Limit output length to keep it concise and focused

tooling/pipeline/src/data_generation/llama/helpers.cc

@@ -9,6 +9,7 @@
 #include <boost/json.hpp>
 #include <cctype>
 #include <optional>
+#include <span>
 #include <stdexcept>
 #include <string>
 #include <string_view>
@@ -76,23 +77,7 @@ bool ReadRequiredTrimmedStringField(const boost::json::object& obj,
   return !out.empty();
 }
 
-bool HasSchemaPlaceholder(const std::array<std::string*, 4>& values) {
+bool HasSchemaPlaceholder(std::span<std::string* const> values) {
-  for (const std::string* value : values) {
-    std::string lowered = *value;
-    std::ranges::transform(lowered, lowered.begin(),
-                           [](const unsigned char character) {
-                             return static_cast<char>(std::tolower(character));
-                           });
-
-    if (lowered == "string") {
-      return true;
-    }
-  }
-
-  return false;
-}
-
-bool HasSchemaPlaceholder(const std::array<std::string*, 2>& values) {
   for (const std::string* value : values) {
     std::string lowered = *value;
     std::ranges::transform(lowered, lowered.begin(),

tooling/pipeline/src/json_handling/json_loader.cc

@@ -107,22 +107,7 @@ std::string ReadFirstOfStringArray(const boost::json::object& object,
 
 std::vector<Location> JsonLoader::LoadLocations(
     const std::filesystem::path& filepath, std::shared_ptr<ILogger> logger) {
-  std::ifstream input(filepath);
+  const boost::json::value root = ParseJsonFile(filepath, "locations");
-  if (!input.is_open()) {
-    throw std::runtime_error("Failed to open locations file: " +
-                             filepath.string());
-  }
-
-  std::stringstream buffer;
-  buffer << input.rdbuf();
-  const std::string content = buffer.str();
-
-  boost::system::error_code error;
-  boost::json::value root = boost::json::parse(content, error);
-  if (error) {
-    throw std::runtime_error("Failed to parse locations JSON: " +
-                             error.message());
-  }
 
   if (!root.is_array()) {
     throw std::runtime_error(

tooling/pipeline/src/services/enrichment/wikipedia/fetch_extract.cc

@@ -21,7 +21,7 @@ std::string WikipediaEnrichmentService::FetchExtract(std::string_view query) {
       cache_it != this->extract_cache_.end()) {
     if (logger_) {
       logger_->Log({.level = LogLevel::Debug,
-                    .phase = PipelinePhase::UserGeneration,
+                    .phase = PipelinePhase::Enrichment,
                     .message = std::format("Wikipedia: Cache hit for {}!", cache_key)});
     }
     return cache_it->second;
@@ -47,7 +47,7 @@ std::string WikipediaEnrichmentService::FetchExtract(std::string_view query) {
     if (logger_) {
       logger_->Log(
           {.level = LogLevel::Warn,
-           .phase = PipelinePhase::UserGeneration,
+           .phase = PipelinePhase::Enrichment,
            .message = std::format("WikipediaService: JSON parse error for '{}': {}",
                       std::string(query), ec.message())});
     }
@@ -60,7 +60,7 @@ std::string WikipediaEnrichmentService::FetchExtract(std::string_view query) {
     if (logger_) {
       logger_->Log(
           {.level = LogLevel::Warn,
-           .phase = PipelinePhase::UserGeneration,
+           .phase = PipelinePhase::Enrichment,
            .message =
                std::format("WikipediaService: Expected root object for '{}'",
                std::string(query))});
@@ -78,7 +78,7 @@ std::string WikipediaEnrichmentService::FetchExtract(std::string_view query) {
     if (logger_) {
       logger_->Log(
           {.level = LogLevel::Warn,
-           .phase = PipelinePhase::UserGeneration,
+           .phase = PipelinePhase::Enrichment,
            .message =
                std::format("WikipediaService: Missing query.pages for '{}'",
                std::string(query))});
@@ -92,7 +92,7 @@ std::string WikipediaEnrichmentService::FetchExtract(std::string_view query) {
     if (logger_) {
       logger_->Log(
           {.level = LogLevel::Warn,
-           .phase = PipelinePhase::UserGeneration,
+           .phase = PipelinePhase::Enrichment,
            .message = std::format("WikipediaService: No pages returned for '{}'",
                       std::string(query))});
     }
@@ -108,7 +108,7 @@ std::string WikipediaEnrichmentService::FetchExtract(std::string_view query) {
     if (logger_) {
       logger_->Log(
           {.level = LogLevel::Warn,
-           .phase = PipelinePhase::UserGeneration,
+           .phase = PipelinePhase::Enrichment,
            .message =
                std::format("WikipediaService: Unexpected page format for '{}'",
                std::string(query))});
@@ -122,7 +122,7 @@ std::string WikipediaEnrichmentService::FetchExtract(std::string_view query) {
   if (page.contains("missing")) {
     if (logger_) {
       logger_->Log({.level = LogLevel::Warn,
-                    .phase = PipelinePhase::UserGeneration,
+                    .phase = PipelinePhase::Enrichment,
                     .message = std::format("WikipediaService: Page '{}' does not exist",
                                std::string(query))});
     }
@@ -136,7 +136,7 @@ std::string WikipediaEnrichmentService::FetchExtract(std::string_view query) {
     if (logger_) {
       logger_->Log(
           {.level = LogLevel::Warn,
-           .phase = PipelinePhase::UserGeneration,
+           .phase = PipelinePhase::Enrichment,
            .message =
                std::format("WikipediaService: No extract string found for '{}'",
                std::string(query))});
@@ -149,7 +149,7 @@ std::string WikipediaEnrichmentService::FetchExtract(std::string_view query) {
   std::string extract(extract_ptr->as_string());
   if (logger_) {
     logger_->Log({.level = LogLevel::Info,
-                  .phase = PipelinePhase::UserGeneration,
+                  .phase = PipelinePhase::Enrichment,
                   .message = std::format("WikipediaService: Fetched {} chars for '{}'",
                              extract.size(), std::string(query))});
   }

tooling/pipeline/src/services/enrichment/wikipedia/get_summary.cc

@@ -16,7 +16,7 @@ std::string WikipediaEnrichmentService::GetLocationContext(
   if (!this->client_) {
     if (logger_) {
       logger_->Log({.level = LogLevel::Warn,
-                    .phase = PipelinePhase::UserGeneration,
+                    .phase = PipelinePhase::Enrichment,
                     .message = "Wikipedia client is nullptr."});
     }
     return {};
@@ -24,13 +24,6 @@ std::string WikipediaEnrichmentService::GetLocationContext(
 
   std::string result;
 
-  // std::string region_query(loc.city);
-  // if (!loc.country.empty()) {
-  //   region_query += loc.state_province,
-  //   region_query += ", ";
-  //   region_query += loc.country;
-  // }
-
   constexpr std::string_view brewing_query = "brewing";
   const std::string location_query =
       std::format("{}, {}", loc.city, loc.iso3166_2);
@@ -51,7 +44,7 @@ std::string WikipediaEnrichmentService::GetLocationContext(
     append_extract(FetchExtract(beer_query));
     if (logger_) {
       logger_->Log({.level = LogLevel::Info,
-                    .phase = PipelinePhase::UserGeneration,
+                    .phase = PipelinePhase::Enrichment,
                     .message = std::format("Done fetching for {}. Sleeping for 10 seconds.",
                                location_query)});
     }
@@ -61,7 +54,7 @@ std::string WikipediaEnrichmentService::GetLocationContext(
     if (logger_) {
       logger_->Log(
           {.level = LogLevel::Debug,
-           .phase = PipelinePhase::UserGeneration,
+           .phase = PipelinePhase::Enrichment,
            .message = std::format("WikipediaService lookup failed for '{}': {}",
                       location_query, e.what())});
     }

tooling/pipeline/src/services/logging/log_dispatcher.cc

@@ -19,6 +19,8 @@ namespace {
   switch (phase) {
     case PipelinePhase::Startup:
       return "Startup";
+    case PipelinePhase::Enrichment:
+      return "Enrichment";
     case PipelinePhase::UserGeneration:
       return "User Generation";
     case PipelinePhase::BreweryAndBeerGeneration:

tooling/pipeline/src/services/sqlite/finalize.cc

@@ -14,6 +14,7 @@ void SqliteExportService::Finalize() {
   }
 
   try {
+    insert_user_stmt_.reset();
     insert_brewery_stmt_.reset();
     insert_location_stmt_.reset();
     if (transaction_open_) {

tooling/pipeline/src/services/sqlite/initialize.cc

@@ -33,6 +33,9 @@ void SqliteExportService::InitializeSchema() const {
   sqlite_export_service_internal::ExecSql(
       db_handle_, sqlite_export_service_internal::kCreateBreweriesTableSql,
       "Failed to create SQLite breweries table");
+  sqlite_export_service_internal::ExecSql(
+      db_handle_, sqlite_export_service_internal::kCreateUsersTableSql,
+      "Failed to create SQLite users table");
 }
 
 void SqliteExportService::PrepareStatements() {
@@ -42,6 +45,9 @@ void SqliteExportService::PrepareStatements() {
   insert_brewery_stmt_ = sqlite_export_service_internal::PrepareStatement(
       db_handle_, sqlite_export_service_internal::kInsertBrewerySql,
       "Failed to prepare SQLite brewery insert statement");
+  insert_user_stmt_ = sqlite_export_service_internal::PrepareStatement(
+      db_handle_, sqlite_export_service_internal::kInsertUserSql,
+      "Failed to prepare SQLite user insert statement");
 }
 
 void SqliteExportService::RollbackAndCloseNoThrow() noexcept {
@@ -54,6 +60,7 @@ void SqliteExportService::RollbackAndCloseNoThrow() noexcept {
     transaction_open_ = false;
   }
 
+  insert_user_stmt_.reset();
   insert_brewery_stmt_.reset();
   insert_location_stmt_.reset();
   db_handle_.reset();

tooling/pipeline/src/services/sqlite/process_record.cc

@@ -1,6 +1,7 @@
 /**
  * @file services/sqlite/process_record.cc
- * @brief SqliteExportService::ProcessRecord() implementation.
+ * @brief SqliteExportService::ProcessRecord(GeneratedBrewery) implementation
+ * and the shared location-resolution helper.
  */
 
 #include <iomanip>
@@ -29,83 +30,85 @@ std::string SqliteExportService::BuildLocationKey(const Location& location) {
   return key_stream.str();
 }
 
-uint64_t SqliteExportService::ProcessRecord(const GeneratedBrewery& brewery) {
+sqlite3_int64 SqliteExportService::ResolveLocationId(const Location& location) {
-  if (db_handle_ == nullptr || !transaction_open_) {
+  const std::string location_key = BuildLocationKey(location);
-    throw std::runtime_error("SQLite export service is not initialized");
+  const auto cached_location = location_cache_.find(location_key);
+  if (cached_location != location_cache_.end()) {
+    return cached_location->second;
   }
 
-  const std::string location_key = BuildLocationKey(brewery.location);
-  const auto cached_location = location_cache_.find(location_key);
-  sqlite3_int64 location_id = 0;
-
-  if (cached_location != location_cache_.end()) {
-    location_id = cached_location->second;
-  } else {
   const std::string local_languages_json =
       sqlite_export_service_internal::SerializeVector(
-            brewery.location.local_languages);
+          location.local_languages);
 
   sqlite_export_service_internal::Bind(
       insert_location_stmt_,
       sqlite_export_service_internal::BindParam<std::string_view>{
           .index = sqlite_export_service_internal::kLocationCityBindIndex,
-            .value = brewery.location.city,
+          .value = location.city,
           .action = "Failed to bind SQLite location city"});
   sqlite_export_service_internal::Bind(
       insert_location_stmt_,
       sqlite_export_service_internal::BindParam<std::string_view>{
           .index =
               sqlite_export_service_internal::kLocationStateProvinceBindIndex,
-            .value = brewery.location.state_province,
+          .value = location.state_province,
           .action = "Failed to bind SQLite location state/province"});
   sqlite_export_service_internal::Bind(
       insert_location_stmt_,
       sqlite_export_service_internal::BindParam<std::string_view>{
           .index = sqlite_export_service_internal::kLocationIso31662BindIndex,
-            .value = brewery.location.iso3166_2,
+          .value = location.iso3166_2,
           .action = "Failed to bind SQLite location ISO 3166-2 code"});
   sqlite_export_service_internal::Bind(
       insert_location_stmt_,
       sqlite_export_service_internal::BindParam<std::string_view>{
           .index = sqlite_export_service_internal::kLocationCountryBindIndex,
-            .value = brewery.location.country,
+          .value = location.country,
           .action = "Failed to bind SQLite location country"});
   sqlite_export_service_internal::Bind(
       insert_location_stmt_,
       sqlite_export_service_internal::BindParam<std::string_view>{
           .index = sqlite_export_service_internal::kLocationIso31661BindIndex,
-            .value = brewery.location.iso3166_1,
+          .value = location.iso3166_1,
           .action = "Failed to bind SQLite location ISO 3166-1 code"});
   sqlite_export_service_internal::Bind(
       insert_location_stmt_,
       sqlite_export_service_internal::BindParam<std::string_view>{
-            .index =
+          .index = sqlite_export_service_internal::kLocationLanguagesBindIndex,
-                sqlite_export_service_internal::kLocationLanguagesBindIndex,
           .value = local_languages_json,
           .action = "Failed to bind SQLite location languages"});
   sqlite_export_service_internal::Bind(
       insert_location_stmt_,
       sqlite_export_service_internal::BindParam{
           .index = sqlite_export_service_internal::kLocationLatitudeBindIndex,
-            .value = brewery.location.latitude,
+          .value = location.latitude,
           .action = "Failed to bind SQLite location latitude"});
   sqlite_export_service_internal::Bind(
       insert_location_stmt_,
       sqlite_export_service_internal::BindParam{
-            .index =
+          .index = sqlite_export_service_internal::kLocationLongitudeBindIndex,
-                sqlite_export_service_internal::kLocationLongitudeBindIndex,
+          .value = location.longitude,
-            .value = brewery.location.longitude,
           .action = "Failed to bind SQLite location longitude"});
 
   sqlite_export_service_internal::StepStatement(
-        db_handle_, insert_location_stmt_,
+      db_handle_, insert_location_stmt_, "Failed to insert SQLite location row");
-        "Failed to insert SQLite location row");
 
-    location_id = sqlite_export_service_internal::LastInsertRowId(db_handle_);
+  const sqlite3_int64 location_id =
+      sqlite_export_service_internal::LastInsertRowId(db_handle_);
   location_cache_.emplace(location_key, location_id);
   sqlite_export_service_internal::ResetStatement(insert_location_stmt_);
+
+  return location_id;
 }
 
+uint64_t SqliteExportService::ProcessRecord(const GeneratedBrewery& brewery) {
+  if (db_handle_ == nullptr || !transaction_open_) {
+    throw std::runtime_error("SQLite export service is not initialized");
+  }
+
+  const sqlite3_int64 location_id = ResolveLocationId(brewery.location);
+
   sqlite_export_service_internal::Bind(
       insert_brewery_stmt_,
       sqlite_export_service_internal::BindParam<sqlite3_int64>{

tooling/pipeline/src/services/sqlite/process_user_record.cc

@@ -0,0 +1,85 @@
+/**
+ * @file services/sqlite/process_user_record.cc
+ * @brief SqliteExportService::ProcessRecord(GeneratedUser) implementation.
+ */
+
+#include <stdexcept>
+
+#include "services/database/sqlite_export_service.h"
+#include "services/database/sqlite_export_service_helpers.h"
+
+uint64_t SqliteExportService::ProcessRecord(const GeneratedUser& user) {
+  if (db_handle_ == nullptr || !transaction_open_) {
+    throw std::runtime_error("SQLite export service is not initialized");
+  }
+
+  const sqlite3_int64 location_id = ResolveLocationId(user.location);
+
+  sqlite_export_service_internal::Bind(
+      insert_user_stmt_,
+      sqlite_export_service_internal::BindParam<sqlite3_int64>{
+          .index = sqlite_export_service_internal::kUserLocationIdBindIndex,
+          .value = location_id,
+          .action = "Failed to bind SQLite user location id"});
+  sqlite_export_service_internal::Bind(
+      insert_user_stmt_,
+      sqlite_export_service_internal::BindParam<std::string_view>{
+          .index = sqlite_export_service_internal::kUserFirstNameBindIndex,
+          .value = user.user.first_name,
+          .action = "Failed to bind SQLite user first name"});
+  sqlite_export_service_internal::Bind(
+      insert_user_stmt_,
+      sqlite_export_service_internal::BindParam<std::string_view>{
+          .index = sqlite_export_service_internal::kUserLastNameBindIndex,
+          .value = user.user.last_name,
+          .action = "Failed to bind SQLite user last name"});
+  sqlite_export_service_internal::Bind(
+      insert_user_stmt_,
+      sqlite_export_service_internal::BindParam<std::string_view>{
+          .index = sqlite_export_service_internal::kUserGenderBindIndex,
+          .value = user.user.gender,
+          .action = "Failed to bind SQLite user gender"});
+  sqlite_export_service_internal::Bind(
+      insert_user_stmt_,
+      sqlite_export_service_internal::BindParam<std::string_view>{
+          .index = sqlite_export_service_internal::kUserUsernameBindIndex,
+          .value = user.user.username,
+          .action = "Failed to bind SQLite user username"});
+  sqlite_export_service_internal::Bind(
+      insert_user_stmt_,
+      sqlite_export_service_internal::BindParam<std::string_view>{
+          .index = sqlite_export_service_internal::kUserBioBindIndex,
+          .value = user.user.bio,
+          .action = "Failed to bind SQLite user bio"});
+  sqlite_export_service_internal::Bind(
+      insert_user_stmt_,
+      sqlite_export_service_internal::BindParam<double>{
+          .index = sqlite_export_service_internal::kUserActivityWeightBindIndex,
+          .value = static_cast<double>(user.user.activity_weight),
+          .action = "Failed to bind SQLite user activity weight"});
+  sqlite_export_service_internal::Bind(
+      insert_user_stmt_,
+      sqlite_export_service_internal::BindParam<std::string_view>{
+          .index = sqlite_export_service_internal::kUserEmailBindIndex,
+          .value = user.email,
+          .action = "Failed to bind SQLite user email"});
+  sqlite_export_service_internal::Bind(
+      insert_user_stmt_,
+      sqlite_export_service_internal::BindParam<std::string_view>{
+          .index = sqlite_export_service_internal::kUserDateOfBirthBindIndex,
+          .value = user.date_of_birth,
+          .action = "Failed to bind SQLite user date of birth"});
+  sqlite_export_service_internal::Bind(
+      insert_user_stmt_,
+      sqlite_export_service_internal::BindParam<std::string_view>{
+          .index = sqlite_export_service_internal::kUserPasswordBindIndex,
+          .value = user.password,
+          .action = "Failed to bind SQLite user password"});
+
+  sqlite_export_service_internal::StepStatement(
+      db_handle_, insert_user_stmt_, "Failed to insert SQLite user row");
+
+  sqlite_export_service_internal::ResetStatement(insert_user_stmt_);
+
+  return sqlite_export_service_internal::LastInsertRowId(db_handle_);
+}

tooling/pipeline/src/web_client/http_web_client.cc

@@ -57,7 +57,7 @@ std::string HttpWebClient::Get(const std::string& url) {
     if (logger_) {
       logger_->Log(
           {.level = LogLevel::Error,
-           .phase = PipelinePhase::UserGeneration,
+           .phase = PipelinePhase::Enrichment,
          .message =
            std::format("[HttpWebClient] Request failed for URL: {}", url)});
     }