# The Biergarten Data Pipeline (Current — Synchronous Data Path) The only concurrency in the current pipeline is a dedicated log dispatcher thread that drains `log_channel` and forwards entries to spdlog for the entire run (joined during shutdown after `log_channel` closes). `BiergartenPipelineOrchestrator::Run()` itself is synchronous — sampling, enrichment, generation, and export all happen on the main thread. ```mermaid flowchart TD Start([start]) --> ParseArgs["main.cc: ParseArguments(argc, argv)"] ParseArgs --> ArgsValid{Arguments valid?} ArgsValid -- no --> LogErr[Log error / usage info] --> Stop1([stop]) ArgsValid -- yes --> Init["main.cc: Init OpenSSL global state and LlamaBackendState"] Init --> DI["main.cc: di::make_injector(...)
binds ILogger, IEnrichmentService, DataGenerator, IExportService, IPromptFormatter, WebClient based on --mocked vs. model-backed mode"] DI --> LogThread["main.cc: create BoundedChannel log_channel, spawn log dispatcher thread (LogDispatcher::Run)"] LogThread --> CreateOrch["main.cc: injector.create BiergartenPipelineOrchestrator"] CreateOrch --> Run["BiergartenPipelineOrchestrator::Run()"] Run --> SqliteInit["SqliteExportService.Initialize()
builds biergarten_seed_<UTC datetime>.sqlite (numeric suffix if it exists), opens DB connection, executes schema DDL, begins transaction"] SqliteInit --> QueryLoc[Orchestrator: QueryLocations] QueryLoc --> LoadLoc["JsonLoader.LoadLocations(locations.json)"] LoadLoc --> SampleLoc["std::ranges::sample(all_locations, --location-count) — defaults to 10"] SampleLoc --> EnrichLoop{For each sampled Location} EnrichLoop --> GetContext["IEnrichmentService.GetLocationContext(loc)
WikipediaEnrichmentService fetches a generic 'brewing' extract and a 'beer in {country}' extract (not city/region-specific); MockEnrichmentService (--mocked) returns an empty string"] GetContext --> LookupFailed{Lookup failed?} LookupFailed -- yes --> SkipCity1[Log warning, skip city] --> EnrichLoop LookupFailed -- no --> StoreEnriched["Store EnrichedCity{Location, region_context}"] --> EnrichLoop EnrichLoop -- done --> GenUsers[Orchestrator: GenerateUsers enriched_cities] GenUsers --> LoadUserData["JsonLoader: LoadPersonas / LoadForenamesByCountry / LoadSurnamesByCountry
each memoized after first parse; MockCuratedDataService (--mocked) returns a fixed in-memory dataset instead, skipping these JSON files"] LoadUserData --> UserLoop{For each EnrichedCity} UserLoop --> SampleName["SampleName(forenames_by_country, surnames_by_country, location.iso3166_1)"] SampleName --> NamesAvail{Names available for country?} NamesAvail -- no --> SkipCity2[Log warning, skip city] --> UserLoop NamesAvail -- yes --> SamplePersona["Sample UserPersona (uniform random)"] SamplePersona --> UserGenMode{Generator mode} UserGenMode -- MockGenerator --> UserMockGen[DeterministicHash and Format] UserGenMode -- LlamaGenerator --> UserLlamaLoad["prompt_directory_.Load(USER_GENERATION) → USER_GENERATION.md in --prompt-dir"] UserLlamaLoad --> UserLlamaRetry["Infer(...) with kUserJsonGrammar → ValidateUserJson
retry up to 3 attempts, appending validation error to the retry prompt on failure"] UserGenMode -- OpenAIGenerator --> UserOAILoad["prompt_directory_.Load(USER_GENERATION) — same prompt file as LlamaGenerator, content is backend-agnostic"] UserOAILoad --> UserOAIRetry["CallChatCompletionsApi(...) with kUserJsonSchema → ValidateUserJson
retry up to 2 attempts — Structured Outputs (json_schema, strict) already guarantees schema shape, so retries only cover transient failures / placeholder text"] UserMockGen --> UserGenDone{Generation successful?} UserLlamaRetry --> UserGenDone UserOAIRetry --> UserGenDone UserGenDone -- yes --> UserPostProc["GetRandomCoordsWithinRange (5km jitter), BuildEmail, GenerateDateOfBirth
email and date_of_birth are generated programmatically, never LLM-authored; generation + coordinate jitter run inside one try/catch per record"] UserGenDone -- no --> UserGenFail["Log warning: generation failed, skipping"] --> UserLoop UserPostProc --> UserExport["SqliteExportService.ProcessRecord(UserRecord)
reuse cached location_id or insert Location and cache its id; insert User (FK location_id)"] UserExport --> UserExportFail{Exception during insert?} UserExportFail -- yes --> UserExportWarn["Log warning: SQLite export failed"] --> UserLoop UserExportFail -- no --> UserLoop UserLoop -- done --> GenBrew[Orchestrator: GenerateBreweries enriched_cities] GenBrew --> BrewLoop{For each EnrichedCity} BrewLoop --> BrewGenMode{Generator mode} BrewGenMode -- MockGenerator --> BrewMockGen[DeterministicHash and Format] BrewGenMode -- LlamaGenerator --> BrewLlamaLoad["PrepareRegionContext, prompt_directory_.Load(BREWERY_GENERATION) → BREWERY_GENERATION.md"] BrewLlamaLoad --> BrewLlamaRetry["Infer(...) with kBreweryJsonGrammar → ValidateBreweryJson
retry up to 3 attempts; max_tokens is fixed across retries, not raised on truncation"] BrewGenMode -- OpenAIGenerator --> BrewOAILoad["PrepareRegionContext, prompt_directory_.Load(BREWERY_GENERATION) — same prompt file as LlamaGenerator"] BrewOAILoad --> BrewOAIRetry["CallChatCompletionsApi(...) with kBreweryJsonSchema → ValidateBreweryJson
retry up to 2 attempts — Structured Outputs already guarantees schema shape"] BrewMockGen --> BrewGenDone{Generation successful?} BrewLlamaRetry --> BrewGenDone BrewOAIRetry --> BrewGenDone BrewGenDone -- yes --> BrewPostProc["GetRandomCoordsWithinRange (5km jitter)
generation + jitter run inside one try/catch per record"] BrewGenDone -- no --> BrewGenFail["Log warning: generation failed, skipping"] --> BrewLoop BrewPostProc --> BrewExport["SqliteExportService.ProcessRecord(GeneratedBrewery)
reuse cached location_id or insert Location and cache its id; insert Brewery (FK location_id)"] BrewExport --> BrewExportFail{Exception during insert?} BrewExportFail -- yes --> BrewExportWarn["Log warning: SQLite export failed (data loss prevented per-record; pipeline continues)"] --> BrewLoop BrewExportFail -- no --> BrewLoop BrewLoop -- done --> Finalize["SqliteExportService.Finalize()
commits transaction, closes database connection"] Finalize --> CloseLog["main.cc: close log_channel, join log dispatcher thread"] CloseLog --> Stop2(["return 0 / stop"]) ```