puerto 0.15.0

// 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) -> Loops {
    let database_url = std::env::var("DATABASE_URL").expect("DATABASE_URL must be set");
    let pool = infrastructure::db::create_postgres_pool(&database_url)
        .await
        .expect("Failed to connect to database");
    infrastructure::db::run_migrations(&pool, &infrastructure::db::MIGRATOR)
        .await
        .expect("Failed to run migrations");
    let kafka = KafkaConfig::from_env("{{ project }}").expect("Invalid Kafka configuration");
{%- if aggregates %}
    kafka
        .ensure_topics(&[{% for a in aggregates %}"{{ a }}"{% if not loop.last %}, {% endif %}{% endfor %}])
        .await
        .expect("Failed to create the Kafka topics");
{%- endif %}
    let publisher = KafkaPublisher::new(&kafka).expect("Failed to create the Kafka producer");
{% if consumers %}    let dead_letters = Arc::new(PgDeadLetters::new(pool.clone()));

    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(),
            )
            .expect("Failed to subscribe {{ c.name }} to Kafka"),
            {{ c.pascal }},
            {% if loop.last %}dead_letters{% else %}dead_letters.clone(){% endif %},
        )
{%- endfor %}
{%- else %}
    Loops::new(&config).relay(PgOutbox::new(pool), publisher)
{%- endif %}
}