//! Moves outbox rows onto the message bus. The worker runs it when the bus is
//! not the outbox itself (`EVENT_BUS_BACKEND=nats`).
//!
//! Rows are claimed by the SQLite queue's own atomic claim, so a row is
//! relayed by one claimant at a time, and failures take the queue's backoff
//! and dead-letter path. A row is marked done only after the sink reports the
//! bus accepted it.
//!
//! A sink (transport) failure is the bus's fault, not the row's: the row is
//! deferred without spending an attempt, and the relay pauses with a growing
//! backoff instead of churning through the rest of the outbox. Attempts, and
//! so dead-lettering, are only ever spent by consumers on handler failures.
//!
//! Delivery is at-least-once. A crash between publish and mark-done leaves the
//! row `processing`; when its claim lease lapses it is claimed and relayed
//! again. The bus drops that repeat while it is inside the broker's dedup
//! window (same `Nats-Msg-Id`); beyond the window, or in core mode, the
//! consumer ledger makes the handlers skip it.

use std::{sync::Arc, time::Duration};

use domain::{
    events::{EventEnvelope, retry_backoff},
    ports::{EventConsumer, EventRelaySink},
};
use futures::StreamExt;
use tokio::sync::watch;

pub struct OutboxRelay {
    source: Arc<dyn EventConsumer>,
    sink: Arc<dyn EventRelaySink>,
    backoff: fn(u32) -> Duration,
}

impl OutboxRelay {
    pub fn new(source: Arc<dyn EventConsumer>, sink: Arc<dyn EventRelaySink>) -> Self {
        Self {
            source,
            sink,
            backoff: retry_backoff,
        }
    }

    /// Overrides the pause/defer schedule, keyed by consecutive sink failures.
    pub fn with_backoff(mut self, backoff: fn(u32) -> Duration) -> Self {
        self.backoff = backoff;
        self
    }

    /// Runs until `shutdown` fires or the source ends. A shutdown never
    /// interrupts a row mid-relay: it is acked or nacked before this returns.
    /// Rows the claim already took but the relay had not started stay
    /// `processing` and come back when their lease lapses.
    pub async fn run(self, mut shutdown: watch::Receiver<bool>) {
        let mut rows = self.source.consume();
        let mut sink_failures: u32 = 0;
        loop {
            tokio::select! {
                biased;
                _ = shutdown.changed() => {
                    tracing::info!("outbox relay: shutdown signal received, stopping");
                    return;
                }
                next = rows.next() => match next {
                    Some(Ok(envelope)) => {
                        let pause = relay_envelope(
                            envelope,
                            self.sink.as_ref(),
                            &mut sink_failures,
                            self.backoff,
                        )
                        .await;
                        if let Some(pause) = pause {
                            tokio::select! {
                                _ = tokio::time::sleep(pause) => {}
                                _ = shutdown.changed() => {
                                    tracing::info!("outbox relay: shutdown signal received, stopping");
                                    return;
                                }
                            }
                        }
                    }
                    // The queue already dead-lettered it and logged the row id.
                    Some(Err(e)) => tracing::error!("outbox relay: skipped undecodable row: {e}"),
                    None => {
                        tracing::error!("outbox relay: source closed");
                        return;
                    }
                },
            }
        }
    }
}

/// Returns how long the relay should pause when the sink failed.
async fn relay_envelope(
    envelope: EventEnvelope,
    sink: &dyn EventRelaySink,
    sink_failures: &mut u32,
    backoff: fn(u32) -> Duration,
) -> Option<Duration> {
    match sink.relay(envelope.id(), &envelope.event).await {
        Ok(()) => {
            *sink_failures = 0;
            if let Err(e) = envelope.ack().await {
                tracing::error!("outbox relay: published but could not mark done: {e}");
            }
            None
        }
        Err(e) => {
            *sink_failures = sink_failures.saturating_add(1);
            let wait = backoff(*sink_failures);
            tracing::warn!(
                consecutive_failures = *sink_failures,
                "outbox relay: bus unavailable, deferring row {wait:?} without spending an attempt: {e}"
            );
            if let Err(e) = envelope.defer(wait).await {
                tracing::error!("outbox relay: could not defer: {e}");
            }
            Some(wait)
        }
    }
}