// AUTO-GENERATED by puerto. Edit puerto.toml, then run `puerto generate bootstrap`; add a
// consumer with `puerto generate consumer <name> --on <event_type>`.
{%- if consumers %}
use std::sync::Arc;
{% endif %}
{% for import in imports -%}
use {{ import }};
{% endfor -%}
{% if consumers %}
{% for m in modules -%}
use crate::consumers::{{ m.0 }}::{{ m.1 }};
{% endfor -%}
{% endif %}
/// The relay and one consumer per `[[consumer]]`, on the `kafka` transport: the relay publishes
/// each event to its aggregate's topic, `<KAFKA_TOPIC_PREFIX>.<aggregate>`, and each consumer is a
/// consumer group on the topics of its events. Dead letters stay in Postgres.
pub async fn build_worker(config: WorkerConfig) -> Result<Loops, String> {
let database_url = std::env::var("DATABASE_URL")
.map_err(|_| "DATABASE_URL is not set — set it in .env (.env.example has one)".to_string())?;
let pool = infrastructure::db::create_postgres_pool(&database_url)
.await
.map_err(|e| format!("{e} — is the database up? `make docker-compose/up`"))?;
infrastructure::db::run_migrations(&pool, &infrastructure::db::MIGRATOR)
.await
.map_err(|e| e.to_string())?;
let kafka = KafkaConfig::from_env("{{ project }}").map_err(|e| e.to_string())?;
{%- if aggregates %}
kafka
.ensure_topics(&[{% for a in aggregates %}"{{ a }}"{% if not loop.last %}, {% endif %}{% endfor %}])
.await
.map_err(|e| format!("cannot create the Kafka topics: {e} — is the broker up? `make docker-compose/up`"))?;
{%- endif %}
let publisher = KafkaPublisher::new(&kafka).map_err(|e| format!("cannot create the Kafka producer: {e}"))?;
{% if consumers %} let dead_letters = Arc::new(PgDeadLetters::new(pool.clone()));
Ok(Loops::new(&config)
.relay(PgOutbox::new(pool), publisher)
{%- for c in consumers %}
.consumer(
KafkaSubscription::new(
&kafka,
"{{ c.name }}",
&[{% for a in c.aggregates %}"{{ a }}"{% if not loop.last %}, {% endif %}{% endfor %}],
&[{{ c.events | join(sep=", ") }}],
dead_letters.clone(),
)
.map_err(|e| format!("cannot subscribe {{ c.name }} to Kafka: {e}"))?,
{{ c.pascal }},
{% if loop.last %}dead_letters{% else %}dead_letters.clone(){% endif %},
)
{%- endfor %})
{%- else %}
Ok(Loops::new(&config).relay(PgOutbox::new(pool), publisher))
{%- endif %}
}