diff --git a/tooling/pipeline/CMakeLists.txt b/tooling/pipeline/CMakeLists.txt index ef190f5..1a8f66f 100644 --- a/tooling/pipeline/CMakeLists.txt +++ b/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 diff --git a/tooling/pipeline/includes/data_generation/mock_generator.h b/tooling/pipeline/includes/data_generation/mock_generator.h index ceb5f38..2c6eb60 100644 --- a/tooling/pipeline/includes/data_generation/mock_generator.h +++ b/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, diff --git a/tooling/pipeline/includes/data_model/models.h b/tooling/pipeline/includes/data_model/models.h index 7a0d4d5..3463177 100644 --- a/tooling/pipeline/includes/data_model/models.h +++ b/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. diff --git a/tooling/pipeline/includes/services/database/export_service.h b/tooling/pipeline/includes/services/database/export_service.h index eb61fed..0463f84 100644 --- a/tooling/pipeline/includes/services/database/export_service.h +++ b/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; }; diff --git a/tooling/pipeline/includes/services/database/sqlite_export_service.h b/tooling/pipeline/includes/services/database/sqlite_export_service.h index 02cf6f6..9256993 100644 --- a/tooling/pipeline/includes/services/database/sqlite_export_service.h +++ b/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 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 location_cache_; }; diff --git a/tooling/pipeline/includes/services/database/sqlite_statement_helpers.h b/tooling/pipeline/includes/services/database/sqlite_statement_helpers.h index 2f0d174..7523114 100644 --- a/tooling/pipeline/includes/services/database/sqlite_statement_helpers.h +++ b/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); diff --git a/tooling/pipeline/includes/services/logging/log_entry.h b/tooling/pipeline/includes/services/logging/log_entry.h index 1227430..c396c74 100644 --- a/tooling/pipeline/includes/services/logging/log_entry.h +++ b/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. diff --git a/tooling/pipeline/src/application_options/parse_arguments.cc b/tooling/pipeline/src/application_options/parse_arguments.cc index 5e9ae84..0cb644f 100644 --- a/tooling/pipeline/src/application_options/parse_arguments.cc +++ b/tooling/pipeline/src/application_options/parse_arguments.cc @@ -15,8 +15,6 @@ std::optional 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 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 diff --git a/tooling/pipeline/src/biergarten_pipeline_orchestrator/generate_users.cc b/tooling/pipeline/src/biergarten_pipeline_orchestrator/generate_users.cc index e6c75fa..1bfe09c 100644 --- a/tooling/pipeline/src/biergarten_pipeline_orchestrator/generate_users.cc +++ b/tooling/pipeline/src/biergarten_pipeline_orchestrator/generate_users.cc @@ -107,6 +107,7 @@ void BiergartenPipelineOrchestrator::GenerateUsers( generated_users_.clear(); std::unordered_set used_email_local_parts; size_t skipped_count = 0; + size_t export_failed_count = 0; for (const auto& city : cities) { const std::optional 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)}); + } } diff --git a/tooling/pipeline/src/biergarten_pipeline_orchestrator/run.cc b/tooling/pipeline/src/biergarten_pipeline_orchestrator/run.cc index 344d941..738e162 100644 --- a/tooling/pipeline/src/biergarten_pipeline_orchestrator/run.cc +++ b/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)}); diff --git a/tooling/pipeline/src/data_generation/llama/generate_brewery.cc b/tooling/pipeline/src/data_generation/llama/generate_brewery.cc index f6d5159..961c361 100644 --- a/tooling/pipeline/src/data_generation/llama/generate_brewery.cc +++ b/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 diff --git a/tooling/pipeline/src/data_generation/llama/helpers.cc b/tooling/pipeline/src/data_generation/llama/helpers.cc index 0c61edb..17f6623 100644 --- a/tooling/pipeline/src/data_generation/llama/helpers.cc +++ b/tooling/pipeline/src/data_generation/llama/helpers.cc @@ -9,6 +9,7 @@ #include #include #include +#include #include #include #include @@ -76,23 +77,7 @@ bool ReadRequiredTrimmedStringField(const boost::json::object& obj, return !out.empty(); } -bool HasSchemaPlaceholder(const std::array& values) { - for (const std::string* value : values) { - std::string lowered = *value; - std::ranges::transform(lowered, lowered.begin(), - [](const unsigned char character) { - return static_cast(std::tolower(character)); - }); - - if (lowered == "string") { - return true; - } - } - - return false; -} - -bool HasSchemaPlaceholder(const std::array& values) { +bool HasSchemaPlaceholder(std::span values) { for (const std::string* value : values) { std::string lowered = *value; std::ranges::transform(lowered, lowered.begin(), diff --git a/tooling/pipeline/src/json_handling/json_loader.cc b/tooling/pipeline/src/json_handling/json_loader.cc index 6c58d74..3718f06 100644 --- a/tooling/pipeline/src/json_handling/json_loader.cc +++ b/tooling/pipeline/src/json_handling/json_loader.cc @@ -107,22 +107,7 @@ std::string ReadFirstOfStringArray(const boost::json::object& object, std::vector JsonLoader::LoadLocations( const std::filesystem::path& filepath, std::shared_ptr logger) { - std::ifstream input(filepath); - 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()); - } + const boost::json::value root = ParseJsonFile(filepath, "locations"); if (!root.is_array()) { throw std::runtime_error( diff --git a/tooling/pipeline/src/services/enrichment/wikipedia/fetch_extract.cc b/tooling/pipeline/src/services/enrichment/wikipedia/fetch_extract.cc index 0afd490..2e345a5 100644 --- a/tooling/pipeline/src/services/enrichment/wikipedia/fetch_extract.cc +++ b/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))}); } diff --git a/tooling/pipeline/src/services/enrichment/wikipedia/get_summary.cc b/tooling/pipeline/src/services/enrichment/wikipedia/get_summary.cc index 1df272f..325c0ff 100644 --- a/tooling/pipeline/src/services/enrichment/wikipedia/get_summary.cc +++ b/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())}); } diff --git a/tooling/pipeline/src/services/logging/log_dispatcher.cc b/tooling/pipeline/src/services/logging/log_dispatcher.cc index 3655047..f9e8308 100644 --- a/tooling/pipeline/src/services/logging/log_dispatcher.cc +++ b/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: diff --git a/tooling/pipeline/src/services/sqlite/finalize.cc b/tooling/pipeline/src/services/sqlite/finalize.cc index e891b31..c4e4ff6 100644 --- a/tooling/pipeline/src/services/sqlite/finalize.cc +++ b/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_) { diff --git a/tooling/pipeline/src/services/sqlite/initialize.cc b/tooling/pipeline/src/services/sqlite/initialize.cc index b7da162..35746d8 100644 --- a/tooling/pipeline/src/services/sqlite/initialize.cc +++ b/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(); diff --git a/tooling/pipeline/src/services/sqlite/process_record.cc b/tooling/pipeline/src/services/sqlite/process_record.cc index a46ff82..0a30415 100644 --- a/tooling/pipeline/src/services/sqlite/process_record.cc +++ b/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 @@ -29,82 +30,84 @@ std::string SqliteExportService::BuildLocationKey(const Location& location) { return key_stream.str(); } +sqlite3_int64 SqliteExportService::ResolveLocationId(const Location& location) { + const std::string location_key = BuildLocationKey(location); + const auto cached_location = location_cache_.find(location_key); + if (cached_location != location_cache_.end()) { + return cached_location->second; + } + + const std::string local_languages_json = + sqlite_export_service_internal::SerializeVector( + location.local_languages); + + sqlite_export_service_internal::Bind( + insert_location_stmt_, + sqlite_export_service_internal::BindParam{ + .index = sqlite_export_service_internal::kLocationCityBindIndex, + .value = location.city, + .action = "Failed to bind SQLite location city"}); + sqlite_export_service_internal::Bind( + insert_location_stmt_, + sqlite_export_service_internal::BindParam{ + .index = + sqlite_export_service_internal::kLocationStateProvinceBindIndex, + .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{ + .index = sqlite_export_service_internal::kLocationIso31662BindIndex, + .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{ + .index = sqlite_export_service_internal::kLocationCountryBindIndex, + .value = location.country, + .action = "Failed to bind SQLite location country"}); + sqlite_export_service_internal::Bind( + insert_location_stmt_, + sqlite_export_service_internal::BindParam{ + .index = sqlite_export_service_internal::kLocationIso31661BindIndex, + .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{ + .index = 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 = location.latitude, + .action = "Failed to bind SQLite location latitude"}); + sqlite_export_service_internal::Bind( + insert_location_stmt_, + sqlite_export_service_internal::BindParam{ + .index = sqlite_export_service_internal::kLocationLongitudeBindIndex, + .value = location.longitude, + .action = "Failed to bind SQLite location longitude"}); + + sqlite_export_service_internal::StepStatement( + db_handle_, insert_location_stmt_, "Failed to insert SQLite location row"); + + 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 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); - - sqlite_export_service_internal::Bind( - insert_location_stmt_, - sqlite_export_service_internal::BindParam{ - .index = sqlite_export_service_internal::kLocationCityBindIndex, - .value = brewery.location.city, - .action = "Failed to bind SQLite location city"}); - sqlite_export_service_internal::Bind( - insert_location_stmt_, - sqlite_export_service_internal::BindParam{ - .index = - sqlite_export_service_internal::kLocationStateProvinceBindIndex, - .value = brewery.location.state_province, - .action = "Failed to bind SQLite location state/province"}); - sqlite_export_service_internal::Bind( - insert_location_stmt_, - sqlite_export_service_internal::BindParam{ - .index = sqlite_export_service_internal::kLocationIso31662BindIndex, - .value = brewery.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{ - .index = sqlite_export_service_internal::kLocationCountryBindIndex, - .value = brewery.location.country, - .action = "Failed to bind SQLite location country"}); - sqlite_export_service_internal::Bind( - insert_location_stmt_, - sqlite_export_service_internal::BindParam{ - .index = sqlite_export_service_internal::kLocationIso31661BindIndex, - .value = brewery.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{ - .index = - 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, - .action = "Failed to bind SQLite location latitude"}); - sqlite_export_service_internal::Bind( - insert_location_stmt_, - sqlite_export_service_internal::BindParam{ - .index = - sqlite_export_service_internal::kLocationLongitudeBindIndex, - .value = brewery.location.longitude, - .action = "Failed to bind SQLite location longitude"}); - - sqlite_export_service_internal::StepStatement( - db_handle_, insert_location_stmt_, - "Failed to insert SQLite location row"); - - location_id = sqlite_export_service_internal::LastInsertRowId(db_handle_); - location_cache_.emplace(location_key, location_id); - sqlite_export_service_internal::ResetStatement(insert_location_stmt_); - } + const sqlite3_int64 location_id = ResolveLocationId(brewery.location); sqlite_export_service_internal::Bind( insert_brewery_stmt_, diff --git a/tooling/pipeline/src/services/sqlite/process_user_record.cc b/tooling/pipeline/src/services/sqlite/process_user_record.cc new file mode 100644 index 0000000..a35d95b --- /dev/null +++ b/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 + +#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{ + .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{ + .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{ + .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{ + .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{ + .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{ + .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{ + .index = sqlite_export_service_internal::kUserActivityWeightBindIndex, + .value = static_cast(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{ + .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{ + .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{ + .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_); +} diff --git a/tooling/pipeline/src/web_client/http_web_client.cc b/tooling/pipeline/src/web_client/http_web_client.cc index dabd669..a2348a6 100644 --- a/tooling/pipeline/src/web_client/http_web_client.cc +++ b/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)}); }