the-biergarten-app/pipeline/diagrams/planned/activity.puml

@startuml biergarten_activity
skinparam defaultFontName "DM Sans"
skinparam defaultFontSize 13
skinparam titleFontName "Volkhov"
skinparam titleFontSize 20
skinparam backgroundColor #FCFCF7
skinparam defaultFontColor #14180C
skinparam titleFontColor #14180C
skinparam ArrowColor #656F33
skinparam activityStartColor #EBECE3
skinparam activityEndColor #4A5837
skinparam activityStopColor #4A5837
skinparam ActivityBackgroundColor #EBECE3
skinparam ActivityBorderColor #4A5837
skinparam ActivityDiamondBackgroundColor #CBD2B5
skinparam ActivityDiamondBorderColor #4A5837
skinparam NoteBackgroundColor #DBEEDD
skinparam NoteFontColor #14180C
skinparam NoteBorderColor #4A5837
skinparam SwimlaneBorderColor #4A5837
skinparam SwimlaneBorderThickness 1
skinparam monochrome reverse


title The Biergarten Data Pipeline — Activity Diagram

|Main|
start
:ParseArguments(argc, argv);
if (Invalid args?) then (yes)
  :spdlog::error;
  stop
else (no)
endif
:Init CurlGlobalState & LlamaBackendState;
:Build DI injector;


:Initialize SqliteExportService;
note right
  Opens SQLite connection.
  Begins a single transaction
  covering all five fixture types.
end note

:Create BoundedChannel<LogEntry> log_ch;
:Spawn Log Worker thread;
note right
  Log worker drains log_ch for the
  entire pipeline lifetime.
  All workers emit LogEntry structs
  via PipelineLogger — never spdlog directly.
end note

:BiergartenPipelineOrchestrator::Run();
|BiergartenPipelineOrchestrator::Run()|
:JsonLoader::LoadLocations("locations.json");
:JsonLoader::LoadBeerStyles("beer-styles.json");
:JsonLoader::LoadPersonas("personas.json");
:JsonLoader::LoadNamesByCountry("names-by-country.json");

:EnrichmentService::PreWarmBeerStyleCache(beer_styles);
note right
  Beer styles do not need location context.
  Wikipedia summaries for the entire palette are
  fetched and cached globally at startup.
end note

:EnrichmentService::PreWarmPersonaCache(personas);
note right
  Persona descriptions do not need location context.
  All persona lookups are resolved and cached
  globally at startup.
end note


' ═══════════════════════════════════════════
' PHASE 0 — USER GENERATION
' ═══════════════════════════════════════════
|Orchestrator|
:RunUserPhase(sampled_locations);
:Create BoundedChannels\n(loc_ch, llm_ch, exp_ch);

fork
  |Orchestrator|
  :Loop: Send Locations → loc_ch;
  :Close loc_ch;
fork again
  |LLM Worker|
  while (loc_ch has items?) is (yes)
    :Receive Location;

    :IPersonaSelectionStrategy::SelectPersona(\n  personas_palette_);
    note right
      Guaranteed cache hit from startup.
      Returns a Persona struct carrying
      style_affinities, abv_range,
      ibu_preference, checkin_weight.
    end note

    :NamesByCountry::SampleName(\n  location.iso3166_1);
    note right
      Deterministic lookup — no LLM involved.
      Name selected from pre-keyed table
      and passed into the generation prompt.
    end note

    :GenerateUser(location, persona, sampled_name)\nvia DataGenerator;
    note right
      LLM receives: Location fields + persona
      description + sampled name. Generates
      bio and preference signals grounded
      in locale and persona.
    end note

    :PipelineLogger::Log(Info, UserGeneration,\n  city, user_id, "llm");
    :Send GeneratedUser → llm_ch;
  endwhile (no)
  :Close llm_ch;
fork again
  |SQLite Worker|
  while (llm_ch has items?) is (yes)
    :Receive GeneratedUser;
    :ProcessUser(user) → sqlite3_int64;
    :PipelineLogger::Log(Info, UserGeneration,\n  city, user_id, "sqlite");
    :Append → user_pool_;
  endwhile (no)
end fork

|Orchestrator|
:Join LLM Worker, SQLite Worker;

' ═══════════════════════════════════════════
' PHASE 1 — BREWERY & BEER GENERATION
' ═══════════════════════════════════════════
:RunBreweryAndBeerPhase(sampled_locations);
:Create BoundedChannels\n(loc_ch, llm_ch, exp_ch);

fork
  |Orchestrator|
  :Loop: Send Locations → loc_ch;
  :Close loc_ch;
fork again
  |Enrichment Workers (xN)|
  while (loc_ch has items?) is (yes)
    :Receive Location;
    :GetLocationContext(location,\nBreweryContextStrategy);
    :PipelineLogger::Log(Info,\n  BreweryAndBeerGeneration,\n  city, nullopt, "enrichment");
    :Send EnrichedCity → llm_ch;
  endwhile (no)
  |Orchestrator|
  :Join Enrichment Workers;
  :Close llm_ch;
fork again
  |LLM Worker|
  while (llm_ch has items?) is (yes)
    :Receive EnrichedCity;

    :GenerateBrewery(location, context)\nvia DataGenerator;

    :IBeerSelectionStrategy::SelectStyles(\n  brewery, beer_style_palette_);

    while (For each selected BeerStyle?) is (remaining)
      :GetStyleContextFromCache(style);
      note right
        Guaranteed cache hit from startup.
      end note
      :GenerateBeer(brewery, style_context)\nvia DataGenerator;
      :Attach GeneratedBeer to Brewery bundle;
    endwhile (done)

    :PipelineLogger::Log(Info,\n  BreweryAndBeerGeneration,\n  city, brewery_id, "llm");
    :Send BreweryWithBeers Bundle → exp_ch;
  endwhile (no)
  :Close exp_ch;
fork again
  |SQLite Worker|
  while (exp_ch has items?) is (yes)
    :Receive BreweryWithBeers Bundle;
    :ProcessBrewery(brewery) → brewery_id;
    :Append → brewery_pool_;

    while (For each beer in bundle?) is (remaining)
      :Set beer.brewery_id = brewery_id;
      :ProcessBeer(beer) → sqlite3_int64;
      :Append → beer_pool_;
    endwhile (done)

    :PipelineLogger::Log(Info,\n  BreweryAndBeerGeneration,\n  city, brewery_id, "sqlite");
  endwhile (no)
end fork

|Orchestrator|
:Join LLM Worker, SQLite Worker;
note right
  Both brewery_pool_ and beer_pool_
  are now completely populated.
end note

' ═══════════════════════════════════════════
' PHASE 2 — CHECKIN GENERATION
' ═══════════════════════════════════════════
:RunCheckinPhase();
:ICheckinDistributionStrategy::\nAssignActivityWeights(user_pool_);
note right
  Weights seeded from each user's
  persona.checkin_weight. J-curve profile
  emerges from persona distribution.
end note

while (For each GeneratedUser in user_pool_?) is (remaining)
  :CheckinsForUser(user, brewery_pool_.size());
  while (For each checkin index?) is (remaining)
    :TimestampFor(user, index);
    :Select brewery from brewery_pool_;
    :GenerateCheckin(user, brewery, timestamp)\nvia DataGenerator;
    :ProcessCheckin(checkin) → sqlite3_int64;
    :PipelineLogger::Log(Info, CheckinGeneration,\n  nullopt, checkin_id, "sqlite");
    :Append → checkin_pool_;
  endwhile (done)
endwhile (done)

' ═══════════════════════════════════════════
' PHASE 3 — RATING GENERATION
' ═══════════════════════════════════════════
:RunRatingPhase();
note right
  Beer selection biased by
  user.persona.style_affinities and abv_range.
  Rating skew modulated per persona.
end note

while (For each GeneratedCheckin in checkin_pool_?) is (remaining)
  :Match brewery_id → select beer from beer_pool_\n(same brewery_id, biased by persona affinities);
  if (Beer exists for brewery?) then (yes)
    :GenerateRating(user, beer, checkin_id)\nvia DataGenerator;
    :ProcessRating(rating);
    :PipelineLogger::Log(Info, RatingGeneration,\n  nullopt, rating_id, "sqlite");
  else (no)
    :PipelineLogger::Log(Warn, RatingGeneration,\n  nullopt, brewery_id, "sqlite");
    :Skip — brewery has no beers;
  endif
endwhile (done)

' ═══════════════════════════════════════════
' TEARDOWN
' ═══════════════════════════════════════════
|Main|
:Finalize SqliteExportService;
note right
  COMMIT covers all five fixture types.
end note
:Close log_ch;
:Join Log Worker;
note right
  Drain guarantees no LogEntry is
  dropped at shutdown.
end note
:spdlog::info "Pipeline complete in X ms";
stop

@enduml