Named database connections · Consumer

Keep factory features, database_cell and collection or repository selection on the same database authority.

Factory mode defines named connections for a backend. A transactional queue selects exactly one connection through database_cell. Its business data, migrator and transaction storage must use the same cell; selecting a name does not create another pool.

Replace the single feature with factory mode#

Backend	Single connection	Factory
PostgreSQL	`consumer-transactional-inbox-postgresql`	`consumer-transactional-inbox-postgresql-factory`
MongoDB	`consumer-transactional-inbox-mongodb`	`consumer-transactional-inbox-mongodb-factory`

Do not enable both columns for the same backend. Adding the base postgresql or mongodb feature alongside its factory compound feature also enables single mode. Each backend can choose its own mode; having PostgreSQL and MongoDB in one binary does not infer a queue’s backend selection.

PostgreSQL factory#

[dependencies]
lilyrs = { version = "0.1.0", features = ["consumer-transactional-inbox-postgresql-factory", "config"] }
serde = { version = "1", features = ["derive"] }
uuid = "1"
tokio = { version = "1", features = ["macros", "rt-multi-thread"] }

[postgresql]
mode = "factory"

[[postgresql.cells]]
name = "primary"
connection_string = "${secret:postgresql.primary_url}"

[postgresql.cells.tls]
mode = "verify-full"

[rabbitmq.topology.queues.transactional_inbox]
backend = "postgresql"
database_cell = "primary"

The database section is top-level, while the queue addition belongs to the first application’s orders.created queue. Supply your resolver for the ${secret:...} reference. PgFactory::get("primary") returns the selected service or an error. This module migrates the same cell and performs the handler’s SQL write through the transaction connection.

use lilyrs::consumer::{ApplicationContainer, Injectable, ServiceTrait};
use lilyrs::queue::{Json, PostgresInboxOutboxMigrator, PostgresTransaction,
    PublishContentKind, QueueHandlerError, TransactionalOutboxMessage, queue, queue_service};
use lilyrs::postgresql::{diesel, diesel_async::RunQueryDsl};
use serde::Deserialize;
use uuid::Uuid;

#[derive(Deserialize)]
struct OrderCreated { order_id: i64, outgoing_event_id: String }
#[derive(Default, Injectable)]
#[service(lifetime = "Scoped")]
struct OrderWorker;
impl ServiceTrait for OrderWorker {}

#[queue_service]
impl OrderWorker {
    #[queue("orders.created", version = 1, content = "json", delivery_guarantee = "transactional_inbox")]
    async fn created(&self, transaction: PostgresTransaction, Json(event): Json<OrderCreated>)
        -> Result<(), QueueHandlerError>
    {
        let outgoing = Uuid::parse_str(&event.outgoing_event_id)
            .map_err(|_| QueueHandlerError::permanent("ORDER_EVENT_ID_INVALID"))?;
        transaction.with_connection(move |connection| Box::pin(async move {
            // Create this application table in your deployment migration first.
            diesel::sql_query("INSERT INTO orders (id) VALUES ($1)")
                .bind::<diesel::sql_types::BigInt, _>(event.order_id)
                .execute(connection).await?;
            Ok(())
        })).await?;
        transaction.enqueue(TransactionalOutboxMessage::try_new(
            outgoing, "domain.events", "orders.persisted", 1,
            PublishContentKind::Json, br#"{"status":"persisted"}"#.as_slice(),
        )?).await?;
        Ok(())
    }
}

// Invoke from a separate deployment migration command, before Consumer startup.
pub async fn migrate(container: &ApplicationContainer) -> Result<(), Box<dyn std::error::Error>> {
    let database = container.resolve::<lilyrs::postgresql::PgFactory>(None).await?.get("primary")?;
    PostgresInboxOutboxMigrator::new(database).migrate().await?;
    Ok(())
}

Download the complete postgresql_factory.rs example.

MongoDB factory#

[dependencies]
lilyrs = { version = "0.1.0", features = ["consumer-transactional-inbox-mongodb-factory", "config"] }
serde = { version = "1", features = ["derive"] }
uuid = "1"
tokio = { version = "1", features = ["macros", "rt-multi-thread"] }
tokio-util = "0.7"

[database]
mode = "factory"

[[database.cells]]
name = "primary"
database_type = "mongodb"
connection_string = "${secret:mongodb.primary_url}"
database_name = "orders"

[rabbitmq.topology.queues.transactional_inbox]
backend = "mongodb"
database_cell = "primary"

MongoFactory::get("primary") returns None for an unknown cell. The collection derive makes the same selection through #[cell_name("primary")]. Factory mode injects mongo_factory; the derive’s db field remains, but is not injected as a single DatabaseService. This file is an alternative to the PostgreSQL example; do not link both examples to the same queue contract.

use std::sync::Arc;
use lilyrs::consumer::{ApplicationContainer, Injectable, ServiceTrait};
use lilyrs::mongodb::{Collection, DatabaseService, MongoCollection};
use lilyrs::queue::{Json, MongoInboxOutboxMigrator, MongoTransaction,
    PublishContentKind, QueueHandlerError, TransactionalOutboxMessage, queue, queue_service};
use serde::{Deserialize, Serialize};
use tokio_util::sync::CancellationToken;
use uuid::Uuid;

#[derive(Clone, Serialize, Deserialize)]
struct Order { order_id: i64 }
#[derive(Default, Injectable, MongoCollection)]
#[service(lifetime = "Singleton")]
#[collection("orders")]
#[collection_type(Order)]
#[cell_name("primary")]
struct Orders {
    db: Arc<DatabaseService>,
    #[inject]
    mongo_factory: Arc<lilyrs::mongodb::MongoFactory>,
    collection: Option<Collection<Order>>,
}
#[derive(Deserialize)]
struct OrderCreated { order_id: i64, outgoing_event_id: String }
#[derive(Injectable)]
#[service(lifetime = "Scoped")]
struct OrderWorker {
    #[inject]
    orders: Arc<Orders>,
}
impl ServiceTrait for OrderWorker {}

#[queue_service]
impl OrderWorker {
    #[queue("orders.created", version = 1, content = "json", delivery_guarantee = "transactional_inbox")]
    async fn created(&self, transaction: MongoTransaction, Json(event): Json<OrderCreated>)
        -> Result<(), QueueHandlerError>
    {
        let outgoing = Uuid::parse_str(&event.outgoing_event_id)
            .map_err(|_| QueueHandlerError::permanent("ORDER_EVENT_ID_INVALID"))?;
        let operation = transaction.operation_context()?;
        self.orders.insert_one(Order { order_id: event.order_id }, &operation).await
            .map_err(|_| QueueHandlerError::retryable("ORDER_WRITE_FAILED"))?;
        transaction.enqueue(TransactionalOutboxMessage::try_new(
            outgoing, "domain.events", "orders.persisted", 1,
            PublishContentKind::Json, br#"{"status":"persisted"}"#.as_slice(),
        )?).await?;
        Ok(())
    }
}

// Apply application collection migrations before the Lily component migration.
pub async fn migrate(container: &ApplicationContainer) -> Result<(), Box<dyn std::error::Error>> {
    let database = container.resolve::<lilyrs::mongodb::MongoFactory>(None).await?
        .get("primary").ok_or("primary MongoDB cell is missing")?;
    let operation = database.operation_context(CancellationToken::new())?;
    MongoInboxOutboxMigrator::new(database.as_ref())?.apply(&operation).await?;
    Ok(())
}

Download the complete mongodb_factory.rs example.

The factory examples use the same business contract as the PostgreSQL transaction or MongoDB transaction example. Apply the business migration first, followed by the component migration in the selected cell; MongoDB requires a transaction-capable replica set/sharded deployment. The example simplifies database-error mapping: classify permanent validation and uniqueness errors separately in your application’s error policy.

Send a payload with a numeric order_id and a valid non-nil UUID outgoing_event_id to either example; do not reuse the first application’s string order_id body unchanged. The incoming canonical event ID is separate from this outgoing ID. Provision the destination binding for the outgoing domain.events exchange and orders.persisted routing key separately.

{"order_id": 42, "outgoing_event_id": "e7b04866-8f2e-46e4-a92c-d35dd2016879"}

The manifests include config and the first application’s Tokio runtime features. Configure the resolver in the secrets guide to resolve secret references. The PostgreSQL factory cell uses verify-full TLS; complete its trust and hostname settings using the PostgreSQL guide.

Cell boundaries and sharing#

database_cell is required in factory mode and forbidden in single mode. Lily does not choose the first cell or correct a misspelled name. If you migrate one cell but bind the queue to another, that storage’s preflight fails. Selecting a different cell for a MongoDB collection is also incompatible with the transaction session authority.

Queues on the same backend/cell share an outbox relay and must have identical complete transactional policies. The DI owner manages connection lifecycle; a queue transaction binding does not create another container or client. The migrate functions here are bodies for a separate deployment command; the bootstrap and shutdown order in the migration guide still applies.