mailrs-outbound-queue 6.0.0

//! Per-domain delivery: MX resolution, suppression filtering, retry/bounce
//! bookkeeping, DSN enqueueing.

use hickory_resolver::TokioResolver;

use super::smtp::try_deliver_via_mx;
use crate::dsn;
use crate::queue::{QueueStatus, QueuedMessage, is_hard_bounce};
use crate::retry::{retry_delay_secs, should_bounce};
use crate::store::{QueueStore, StoreError};
use crate::{DeliveryEvent, DeliveryEventSender};

/// generate DSN bounce and enqueue it back to the original sender
pub(super) async fn enqueue_dsn(
    store: &dyn QueueStore,
    hostname: &str,
    msg: &QueuedMessage,
    error: &str,
) {
    if msg.sender.is_empty() || msg.sender == "<>" {
        return; // don't bounce bounces
    }
    let dsn_msg = dsn::format_dsn(
        hostname,
        &msg.sender,
        &msg.recipient,
        error,
        msg.message_id.as_deref(),
    );
    let sender_domain = msg
        .sender
        .rsplit_once('@')
        .map(|(_, d)| d)
        .unwrap_or("unknown");
    let now = chrono::Utc::now().timestamp();
    let queued = store
        .enqueue(
            "<>",
            &msg.sender,
            sender_domain,
            dsn_msg.as_bytes(),
            None,
            now,
            false,
        )
        .await;
    logged("enqueue DSN", msg.id, queued.map(drop));
}

/// deliver messages to a single domain (used by concurrent workers)
///
/// `port` is the TCP port used to connect to every MX in the
/// resolved set. Production wires 25; integration tests pass an
/// ephemeral mock-server port to drive the full claim → MX-resolve
/// → SMTP → mark_* lifecycle without a real MTA on the box.
#[tracing::instrument(
    name = "outbound.deliver_domain",
    skip(resolver, hostname, messages, store, event_sender),
    fields(domain, n_messages = messages.len(), max_per_conn),
)]
#[allow(clippy::too_many_arguments)]
pub async fn deliver_domain_static(
    resolver: &TokioResolver,
    hostname: &str,
    domain: &str,
    messages: Vec<QueuedMessage>,
    store: &dyn QueueStore,
    port: u16,
    max_per_conn: usize,
    event_sender: Option<&DeliveryEventSender>,
) {
    // Wall-clock start for the per-domain delivery batch. Used to
    // emit `mailrs_outbound_delivery_seconds` histogram on each
    // terminal mark (delivered / failed / bounced) — gives ops a
    // distribution of end-to-end delivery latency per outcome.
    let batch_start = std::time::Instant::now();

    let messages = drop_suppressed(store, messages, event_sender).await;
    if messages.is_empty() {
        return;
    }

    // resolve MX records
    let mx_records = match mailrs_smtp_client::resolve_mx(resolver, domain).await {
        Ok(records) => records,
        Err(e) => {
            tracing::warn!("MX resolution failed for {domain}: {e}");
            let now = chrono::Utc::now().timestamp();
            for msg in &messages {
                let delay = retry_delay_secs(msg.attempts);
                if should_bounce(msg.attempts + 1, msg.max_attempts) {
                    let error = format!("MX resolution failed: {e}");
                    logged(
                        "mark bounced",
                        msg.id,
                        store.mark_bounced(msg.id, &error, now).await,
                    );
                    // record hard bounce for suppression
                    if is_hard_bounce(&error) {
                        let added = store.add_suppression(&msg.recipient, &error, None).await;
                        logged("add suppression", msg.id, added);
                    }
                    enqueue_dsn(store, hostname, msg, &error).await;
                    if let Some(es) = event_sender {
                        es(DeliveryEvent::Bounced {
                            queue_id: msg.id,
                            sender: msg.sender.clone(),
                        });
                    }
                } else {
                    let error = format!("MX resolution failed: {e}");
                    let marked = store
                        .mark_failed(msg.id, &error, now + delay as i64, now)
                        .await;
                    logged("mark failed", msg.id, marked);
                    if let Some(es) = event_sender {
                        es(DeliveryEvent::Failed {
                            queue_id: msg.id,
                            domain: domain.to_string(),
                            error: format!("MX resolution failed: {e}"),
                        });
                    }
                }
            }
            return;
        }
    };

    // split messages into chunks for connection reuse limits
    let chunks: Vec<&[QueuedMessage]> = messages.chunks(max_per_conn).collect();

    // try each MX in priority order
    for mx in &mx_records {
        let mut all_ok = true;
        for chunk in &chunks {
            match try_deliver_via_mx(
                hostname,
                &mx.exchange,
                port,
                domain,
                chunk,
                resolver,
                event_sender,
            )
            .await
            {
                Ok(()) => {
                    let now = chrono::Utc::now().timestamp();
                    let elapsed = batch_start.elapsed().as_secs_f64();
                    for msg in *chunk {
                        metrics::histogram!(
                            "mailrs_outbound_delivery_seconds",
                            "outcome" => "delivered",
                        )
                        .record(elapsed);
                        logged(
                            "mark delivered",
                            msg.id,
                            store.mark_delivered(msg.id, now).await,
                        );
                        if let Some(es) = event_sender {
                            es(DeliveryEvent::Success {
                                queue_id: msg.id,
                                domain: domain.to_string(),
                            });
                        }
                    }
                }
                Err(e) => {
                    tracing::warn!("delivery to {} via {} failed: {e}", domain, mx.exchange);
                    all_ok = false;
                    break;
                }
            }
        }
        if all_ok {
            tracing::info!(
                "delivered {} messages to {domain} via {}",
                messages.len(),
                mx.exchange
            );
            return;
        }
    }

    // all MX hosts failed — mark remaining undelivered messages
    let now = chrono::Utc::now().timestamp();
    for msg in &messages {
        // skip already delivered messages
        if let Ok(Some(current)) = store.get_message(msg.id).await
            && current.status == QueueStatus::Delivered
        {
            continue;
        }
        let delay = retry_delay_secs(msg.attempts);
        if should_bounce(msg.attempts + 1, msg.max_attempts) {
            let marked = store.mark_bounced(msg.id, "all MX hosts failed", now).await;
            logged("mark bounced", msg.id, marked);
            // add to suppression if last error was a hard bounce
            if let Some(ref err) = msg.last_error
                && is_hard_bounce(err)
            {
                let added = store.add_suppression(&msg.recipient, err, None).await;
                logged("add suppression", msg.id, added);
            }
            enqueue_dsn(store, hostname, msg, "all MX hosts failed").await;
            if let Some(es) = event_sender {
                es(DeliveryEvent::Bounced {
                    queue_id: msg.id,
                    sender: msg.sender.clone(),
                });
            }
        } else {
            let marked = store
                .mark_failed(msg.id, "all MX hosts failed", now + delay as i64, now)
                .await;
            logged("mark failed", msg.id, marked);
            if let Some(es) = event_sender {
                es(DeliveryEvent::Failed {
                    queue_id: msg.id,
                    domain: domain.to_string(),
                    error: "all MX hosts failed".into(),
                });
            }
        }
    }
}

/// Bounce the messages whose recipient is on the suppression list and
/// return the rest.
async fn drop_suppressed(
    store: &dyn QueueStore,
    messages: Vec<QueuedMessage>,
    event_sender: Option<&DeliveryEventSender>,
) -> Vec<QueuedMessage> {
    let now = chrono::Utc::now().timestamp();
    let mut kept = Vec::with_capacity(messages.len());
    for msg in messages {
        // an unreadable list must not hold mail back: deliver, and say so
        let suppressed = match store.is_suppressed(&msg.recipient).await {
            Ok(suppressed) => suppressed,
            Err(e) => {
                tracing::warn!(
                    "outbound queue: suppression check for msg {} failed: {e}",
                    msg.id
                );
                false
            }
        };
        if !suppressed {
            kept.push(msg);
            continue;
        }
        tracing::info!("skipping suppressed recipient: {}", msg.recipient);
        let marked = store
            .mark_bounced(msg.id, "recipient suppressed (hard bounce history)", now)
            .await;
        logged("mark bounced", msg.id, marked);
        if let Some(es) = event_sender {
            es(DeliveryEvent::Bounced {
                queue_id: msg.id,
                sender: msg.sender.clone(),
            });
        }
    }
    kept
}

// a failed store write leaves the row where it was: inflight rows come back
// through `recover_stale_inflight`, so the cost is a repeat attempt
fn logged(op: &str, id: i64, result: Result<(), StoreError>) {
    if let Err(e) = result {
        tracing::warn!("outbound queue: {op} for msg {id} failed: {e}");
    }
}