//! Message delivery worker.
//!
//! Background loop that drains the outbound message pipeline and delivers
//! messages via channel plugins. Wakes on `Notify`, retry deadlines, or a
//! bounded periodic poll.

use std::sync::Arc;
use std::time::Duration;

use serde_json::{json, Value};
use tracing::warn;

use crate::channels::ChannelRegistry;
use crate::messages::outbound::{MessageContent, MessagePipeline};
use crate::plugins::hook_utils;
use crate::plugins::{self, OutboundContext, PluginRegistry};

const MAX_DELIVERIES_PER_CHANNEL_PASS: usize = 8;

/// Run the delivery worker loop.
///
/// Wakes when notified by the pipeline, every 5 seconds, or on shutdown.
pub async fn delivery_loop(
    pipeline: Arc<MessagePipeline>,
    plugin_registry: Arc<PluginRegistry>,
    channel_registry: Arc<ChannelRegistry>,
    mut shutdown: tokio::sync::watch::Receiver<bool>,
) {
    loop {
        let wake_delay = pipeline
            .next_retry_deadline_ms()
            .map(retry_deadline_wake_delay)
            .unwrap_or_else(|| Duration::from_secs(5))
            .min(Duration::from_secs(5));
        // Wait for notification, timeout, or shutdown
        tokio::select! {
            _ = pipeline.notifier().notified() => {}
            _ = tokio::time::sleep(wake_delay) => {}
            _ = shutdown.changed() => {
                break;
            }
        }

        // Check shutdown after waking
        if *shutdown.borrow() {
            break;
        }

        let channel_ids = pipeline.channels_with_messages();

        process_channel_messages(&channel_ids, &pipeline, &plugin_registry, &channel_registry)
            .await;
    }
}

fn retry_deadline_wake_delay(deadline_ms: i64) -> Duration {
    let now = crate::messages::outbound::now_millis();
    if deadline_ms <= now {
        Duration::ZERO
    } else {
        Duration::from_millis((deadline_ms - now) as u64)
    }
}

/// Process pending messages for each connected channel.
pub(crate) async fn process_channel_messages(
    channel_ids: &[String],
    pipeline: &MessagePipeline,
    plugin_registry: &Arc<PluginRegistry>,
    channel_registry: &ChannelRegistry,
) {
    for channel_id in channel_ids {
        for _ in 0..MAX_DELIVERIES_PER_CHANNEL_PASS {
            let work = pipeline.next_delivery_work_for_channel(channel_id);
            for expired in work.expired {
                let expired_message_id = expired.message.id.clone();
                if let Err(err) =
                    pipeline.mark_expired(&expired_message_id, "message expired before delivery")
                {
                    warn!(
                        id = %expired_message_id,
                        error = %err,
                        "failed to mark queued message as expired before delivery"
                    );
                }
            }

            if !channel_registry.is_connected(channel_id) {
                break;
            }

            let msg = match work.ready {
                Some(m) => m,
                None => break,
            };

            let message_id = msg.message.id.clone();
            let mut message = msg.message.clone();

            if let Some(result) = dispatch_message_hook(
                plugin_registry,
                "message_sending",
                &json!({
                    "messageId": message_id.0.clone(),
                    "channel": channel_id,
                    "content": &message.content,
                    "metadata": &message.metadata,
                }),
            ) {
                if result.cancelled {
                    if let Err(err) = pipeline.cancel(&message_id) {
                        warn!(
                            id = %message_id,
                            error = %err,
                            "failed to cancel message after hook cancellation"
                        );
                        let _ = pipeline.mark_failed(&message_id, "message cancelled by hook");
                    }
                    continue;
                }

                if let Some(payload) = parse_hook_payload(&result, "message_sending") {
                    apply_message_hook_overrides(&mut message, &payload);
                    if let Err(err) = pipeline.update_message(&message_id, message.clone()) {
                        warn!(
                            id = %message_id,
                            error = %err,
                            "failed to persist message updates from hook"
                        );
                    }
                }
            }

            if let Err(e) = pipeline.mark_sending(&message_id) {
                warn!(id = %message_id, error = %e, "failed to mark message as sending");
                break;
            }

            let plugin = match plugin_registry.get_channel(channel_id) {
                Some(p) => p,
                None => {
                    let _ = pipeline.mark_failed(&message_id, "no plugin registered for channel");
                    continue;
                }
            };

            let metadata = &message.metadata;

            let result = deliver_message(
                &plugin,
                &message.content,
                metadata.recipient_id.as_deref().unwrap_or_default(),
                metadata.reply_to.as_deref(),
                metadata.thread_id.as_deref(),
            )
            .await;

            let delivery_snapshot = match &result {
                Ok(delivery) => json!({
                    "ok": delivery.ok,
                    "messageId": delivery.message_id,
                    "error": delivery.error,
                    "retryable": delivery.retryable(),
                    "retryability": delivery.retryability,
                    "conversationId": delivery.conversation_id,
                    "toJid": delivery.to_jid,
                    "pollId": delivery.poll_id,
                }),
                Err(err) => json!({
                    "ok": false,
                    "error": err.to_string(),
                }),
            };

            let _ = dispatch_message_hook(
                plugin_registry,
                "message_sent",
                &json!({
                    "messageId": message_id.0.clone(),
                    "channel": channel_id,
                    "content": &message.content,
                    "metadata": &message.metadata,
                    "delivery": delivery_snapshot,
                }),
            );

            handle_delivery_result(pipeline, &message_id, result).await;
        }
    }
}

/// Handle the result of a message delivery attempt.
async fn handle_delivery_result(
    pipeline: &MessagePipeline,
    message_id: &crate::messages::outbound::MessageId,
    result: Result<plugins::DeliveryResult, plugins::BindingError>,
) {
    match result {
        Ok(delivery) if delivery.ok => {
            let _ = pipeline.mark_sent(message_id);
        }
        Ok(delivery) => {
            let retryability = delivery.retryability;
            let error = delivery
                .error
                .unwrap_or_else(|| "delivery failed".to_string());
            if retryability.is_retryable() && pipeline.can_retry(message_id) {
                let _ = pipeline.mark_retry_with_retry_after(
                    message_id,
                    &error,
                    retryability.retry_after_ms(),
                );
                warn!(
                    id = %message_id,
                    error = %error,
                    retry_after_ms = ?retryability.retry_after_ms(),
                    "retryable delivery failure, reset to queued for retry"
                );
            } else {
                let _ = pipeline.mark_failed(message_id, &error);
            }
        }
        Err(e) => {
            if e.is_delivery_backpressure() && pipeline.can_retry(message_id) {
                let error = e.to_string();
                let retry_after_ms = e.retry_after_ms();
                // Plumb the typed BindingError::Backpressure.retry_after_ms
                // through to the retry scheduler. The Ok(delivery)
                // branch above already does this via
                // `mark_retry_with_retry_after`; the Err branch
                // previously dropped the hint and forced retries onto
                // the default poll cadence regardless of provider-
                // supplied delay.
                let _ = pipeline.mark_retry_with_retry_after(message_id, &error, retry_after_ms);
                warn!(
                    id = %message_id,
                    error = %error,
                    retry_after_ms = ?retry_after_ms,
                    "transient delivery backpressure, reset to queued for retry"
                );
            } else {
                let _ = pipeline.mark_failed(message_id, e.to_string());
            }
        }
    }
}

fn dispatch_message_hook(
    plugin_registry: &Arc<PluginRegistry>,
    hook_name: &str,
    payload: &Value,
) -> Option<plugins::HookDispatchResult> {
    hook_utils::dispatch_hook(plugin_registry.clone(), hook_name, payload)
}

fn parse_hook_payload(result: &plugins::HookDispatchResult, hook_name: &str) -> Option<Value> {
    hook_utils::parse_hook_payload(result, hook_name)
}

fn apply_message_hook_overrides(
    message: &mut crate::messages::outbound::OutboundMessage,
    payload: &Value,
) {
    let Some(obj) = payload.as_object() else {
        return;
    };

    if let Some(content) = obj.get("content") {
        if let Ok(content) = serde_json::from_value::<MessageContent>(content.clone()) {
            message.content = content;
        }
    }

    if let Some(metadata) = obj.get("metadata") {
        if let Ok(metadata) =
            serde_json::from_value::<crate::messages::outbound::MessageMetadata>(metadata.clone())
        {
            message.metadata = metadata;
        }
    }
}

/// Deliver a message via the channel plugin, dispatching to send_text or send_media.
///
/// `ChannelPluginInstance` methods are sync, so we run them via `spawn_blocking`.
async fn deliver_message(
    plugin: &Arc<dyn plugins::ChannelPluginInstance>,
    content: &MessageContent,
    to: &str,
    reply_to_id: Option<&str>,
    thread_id: Option<&str>,
) -> Result<plugins::DeliveryResult, plugins::BindingError> {
    match content {
        MessageContent::Text { text } => {
            let ctx = OutboundContext {
                to: to.to_string(),
                text: text.clone(),
                media_url: None,
                gif_playback: false,
                reply_to_id: reply_to_id.map(|s| s.to_string()),
                thread_id: thread_id.map(|s| s.to_string()),
                account_id: None,
            };
            let p = plugin.clone();
            tokio::task::spawn_blocking(move || p.send_text(ctx))
                .await
                .map_err(|e| plugins::BindingError::CallError(e.to_string()))?
        }
        MessageContent::Media {
            caption, media_ref, ..
        } => {
            let ctx = OutboundContext {
                to: to.to_string(),
                text: caption.clone().unwrap_or_default(),
                media_url: Some(media_ref.clone()),
                gif_playback: false,
                reply_to_id: reply_to_id.map(|s| s.to_string()),
                thread_id: thread_id.map(|s| s.to_string()),
                account_id: None,
            };
            let p = plugin.clone();
            tokio::task::spawn_blocking(move || p.send_media(ctx))
                .await
                .map_err(|e| plugins::BindingError::CallError(e.to_string()))?
        }
        MessageContent::Composite { parts } => {
            // Send each part sequentially; return first failure or last success
            let mut last_result = plugins::DeliveryResult {
                ok: true,
                message_id: None,
                error: None,
                retryability: crate::plugins::Retryability::Terminal,
                conversation_id: None,
                to_jid: None,
                poll_id: None,
                error_kind: None,
            };
            for part in parts {
                last_result =
                    Box::pin(deliver_message(plugin, part, to, reply_to_id, thread_id)).await?;
                if !last_result.ok {
                    return Ok(last_result);
                }
            }
            Ok(last_result)
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::messages::outbound::{
        MessageContent, OutboundContext as MsgOutboundContext, OutboundMessage,
    };
    use crate::plugins::{
        BindingError, ChannelCapabilities, ChannelPluginInstance, DeliveryResult, OutboundContext,
        ReadReceiptContext,
    };
    use std::sync::atomic::{AtomicU32, Ordering};
    use std::sync::Arc;
    use tokio::sync::Notify;

    /// Mock channel plugin that records calls.
    struct MockChannel {
        caps: ChannelCapabilities,
        send_text_count: AtomicU32,
        send_media_count: AtomicU32,
        mark_read_count: AtomicU32,
        mark_read_notify: Option<Arc<Notify>>,
        mark_read_delay: Duration,
        fail: bool,
        retryable: bool,
        retry_after_ms: Option<i64>,
        transient_failures: AtomicU32,
        backpressure: bool,
    }
    impl MockChannel {
        fn new() -> Self {
            Self {
                caps: ChannelCapabilities::default(),
                send_text_count: AtomicU32::new(0),
                send_media_count: AtomicU32::new(0),
                mark_read_count: AtomicU32::new(0),
                mark_read_notify: None,
                mark_read_delay: Duration::ZERO,
                fail: false,
                retryable: false,
                retry_after_ms: None,
                transient_failures: AtomicU32::new(0),
                backpressure: false,
            }
        }

        fn failing(retryable: bool) -> Self {
            Self {
                caps: ChannelCapabilities::default(),
                send_text_count: AtomicU32::new(0),
                send_media_count: AtomicU32::new(0),
                mark_read_count: AtomicU32::new(0),
                mark_read_notify: None,
                mark_read_delay: Duration::ZERO,
                fail: true,
                retryable,
                retry_after_ms: None,
                transient_failures: AtomicU32::new(0),
                backpressure: false,
            }
        }

        fn failing_with_retry_after(retry_after_ms: i64) -> Self {
            Self {
                caps: ChannelCapabilities::default(),
                send_text_count: AtomicU32::new(0),
                send_media_count: AtomicU32::new(0),
                mark_read_count: AtomicU32::new(0),
                mark_read_notify: None,
                mark_read_delay: Duration::ZERO,
                fail: true,
                retryable: true,
                retry_after_ms: Some(retry_after_ms),
                transient_failures: AtomicU32::new(0),
                backpressure: false,
            }
        }

        fn backpressured() -> Self {
            Self {
                caps: ChannelCapabilities::default(),
                send_text_count: AtomicU32::new(0),
                send_media_count: AtomicU32::new(0),
                mark_read_count: AtomicU32::new(0),
                mark_read_notify: None,
                mark_read_delay: Duration::ZERO,
                fail: false,
                retryable: false,
                retry_after_ms: None,
                transient_failures: AtomicU32::new(0),
                backpressure: true,
            }
        }
    }

    impl ChannelPluginInstance for MockChannel {
        fn get_info(&self) -> Result<plugins::ChannelInfo, BindingError> {
            Ok(plugins::ChannelInfo {
                id: "mock".to_string(),
                label: "Mock".to_string(),
                selection_label: "Mock Channel".to_string(),
                docs_path: "".to_string(),
                blurb: "".to_string(),
                order: 0,
            })
        }

        fn get_capabilities(&self) -> Result<ChannelCapabilities, BindingError> {
            Ok(self.caps.clone())
        }

        fn send_text(&self, _ctx: OutboundContext) -> Result<DeliveryResult, BindingError> {
            self.send_text_count.fetch_add(1, Ordering::Relaxed);
            if self.backpressure {
                return Err(BindingError::Backpressure {
                    detail: "plugin worker queue is full".to_string(),
                    retry_after_ms: None,
                });
            }
            let transient_failures = self.transient_failures.load(Ordering::Relaxed);
            if transient_failures > 0 {
                self.transient_failures.fetch_sub(1, Ordering::Relaxed);
            }
            if self.fail || transient_failures > 0 {
                Ok(DeliveryResult {
                    ok: false,
                    message_id: None,
                    error: Some("mock failure".to_string()),
                    retryability: if self.retryable {
                        crate::plugins::Retryability::Transient {
                            retry_after_ms: self.retry_after_ms,
                        }
                    } else {
                        crate::plugins::Retryability::Terminal
                    },
                    conversation_id: None,
                    to_jid: None,
                    poll_id: None,
                    error_kind: None,
                })
            } else {
                Ok(DeliveryResult {
                    ok: true,
                    message_id: Some("sent-1".to_string()),
                    error: None,
                    retryability: crate::plugins::Retryability::Terminal,
                    conversation_id: None,
                    to_jid: None,
                    poll_id: None,
                    error_kind: None,
                })
            }
        }

        fn send_media(&self, _ctx: OutboundContext) -> Result<DeliveryResult, BindingError> {
            self.send_media_count.fetch_add(1, Ordering::Relaxed);
            Ok(DeliveryResult {
                ok: true,
                message_id: Some("sent-media-1".to_string()),
                error: None,
                retryability: crate::plugins::Retryability::Terminal,
                conversation_id: None,
                to_jid: None,
                poll_id: None,
                error_kind: None,
            })
        }

        fn mark_read(&self, _ctx: ReadReceiptContext) -> Result<(), BindingError> {
            if !self.mark_read_delay.is_zero() {
                std::thread::sleep(self.mark_read_delay);
            }
            self.mark_read_count.fetch_add(1, Ordering::Relaxed);
            if let Some(notify) = &self.mark_read_notify {
                notify.notify_one();
            }
            Ok(())
        }
    }
    fn make_pipeline_and_registries(
        channel_id: &str,
        plugin: Option<Arc<dyn ChannelPluginInstance>>,
        connected: bool,
    ) -> (
        Arc<MessagePipeline>,
        Arc<PluginRegistry>,
        Arc<ChannelRegistry>,
    ) {
        let pipeline = Arc::new(MessagePipeline::new());

        let plugin_registry = Arc::new(PluginRegistry::new());
        if let Some(p) = plugin {
            plugin_registry.register_channel(channel_id.to_string(), p);
        }

        let channel_registry = Arc::new(ChannelRegistry::new());
        let status = if connected {
            crate::channels::ChannelStatus::Connected
        } else {
            crate::channels::ChannelStatus::Disconnected
        };
        channel_registry.register(
            crate::channels::ChannelInfo::new(channel_id, channel_id).with_status(status),
        );

        (pipeline, plugin_registry, channel_registry)
    }

    #[tokio::test]
    async fn test_delivery_sends_text_message() {
        let mock = Arc::new(MockChannel::new());
        let (pipeline, plugin_reg, channel_reg) =
            make_pipeline_and_registries("test-ch", Some(mock.clone()), true);

        // Queue a text message
        let msg = OutboundMessage::new("test-ch", MessageContent::text("hello"));
        pipeline.queue(msg, MsgOutboundContext::new()).unwrap();

        // Run one iteration (use shutdown to stop after one pass)
        let (shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(false);
        let pl = pipeline.clone();
        let pr = plugin_reg.clone();
        let cr = channel_reg.clone();
        let handle = tokio::spawn(async move {
            delivery_loop(pl, pr, cr, shutdown_rx).await;
        });

        // Give it time to process
        tokio::time::sleep(Duration::from_millis(100)).await;
        let _ = shutdown_tx.send(true);
        // Notify to unblock select
        pipeline.notifier().notify_one();
        let _ = handle.await;

        assert_eq!(mock.send_text_count.load(Ordering::Relaxed), 1);
        assert_eq!(pipeline.channels_with_messages().len(), 0);
    }

    #[tokio::test]
    async fn test_process_channel_messages_drains_ready_burst_for_channel() {
        let mock = Arc::new(MockChannel::new());
        let (pipeline, plugin_reg, channel_reg) =
            make_pipeline_and_registries("burst-ch", Some(mock.clone()), true);

        for index in 0..3 {
            let msg =
                OutboundMessage::new("burst-ch", MessageContent::text(format!("hello-{index}")));
            pipeline.queue(msg, MsgOutboundContext::new()).unwrap();
        }

        process_channel_messages(
            &["burst-ch".to_string()],
            &pipeline,
            &plugin_reg,
            &channel_reg,
        )
        .await;

        assert_eq!(mock.send_text_count.load(Ordering::Relaxed), 3);
        assert!(
            pipeline.channels_with_messages().is_empty(),
            "one delivery pass must drain all ready work for a connected channel"
        );
    }

    #[tokio::test]
    async fn test_process_channel_messages_rotates_after_bounded_channel_batch() {
        let first = Arc::new(MockChannel::new());
        let second = Arc::new(MockChannel::new());
        let pipeline = Arc::new(MessagePipeline::new());
        let plugin_reg = Arc::new(PluginRegistry::new());
        plugin_reg.register_channel("busy-ch".to_string(), first.clone());
        plugin_reg.register_channel("later-ch".to_string(), second.clone());
        let channel_reg = Arc::new(ChannelRegistry::new());
        channel_reg.register(
            crate::channels::ChannelInfo::new("busy-ch", "busy-ch")
                .with_status(crate::channels::ChannelStatus::Connected),
        );
        channel_reg.register(
            crate::channels::ChannelInfo::new("later-ch", "later-ch")
                .with_status(crate::channels::ChannelStatus::Connected),
        );

        for index in 0..(MAX_DELIVERIES_PER_CHANNEL_PASS + 1) {
            let msg =
                OutboundMessage::new("busy-ch", MessageContent::text(format!("busy-{index}")));
            pipeline.queue(msg, MsgOutboundContext::new()).unwrap();
        }
        pipeline
            .queue(
                OutboundMessage::new("later-ch", MessageContent::text("later")),
                MsgOutboundContext::new(),
            )
            .unwrap();

        process_channel_messages(
            &["busy-ch".to_string(), "later-ch".to_string()],
            &pipeline,
            &plugin_reg,
            &channel_reg,
        )
        .await;

        assert_eq!(
            first.send_text_count.load(Ordering::Relaxed) as usize,
            MAX_DELIVERIES_PER_CHANNEL_PASS,
            "busy channel should be capped to one batch per pass"
        );
        assert_eq!(
            second.send_text_count.load(Ordering::Relaxed),
            1,
            "later channels must not starve behind a busy first channel"
        );
        assert_eq!(pipeline.queue_size("busy-ch"), 1);
    }

    #[tokio::test]
    async fn test_delivery_marks_failed_no_plugin() {
        let (pipeline, plugin_reg, channel_reg) =
            make_pipeline_and_registries("no-plugin-ch", None, true);

        let msg = OutboundMessage::new("no-plugin-ch", MessageContent::text("hello"));
        let result = pipeline.queue(msg, MsgOutboundContext::new()).unwrap();

        let (shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(false);
        let pl = pipeline.clone();
        let handle = tokio::spawn(async move {
            delivery_loop(pl, plugin_reg, channel_reg, shutdown_rx).await;
        });

        tokio::time::sleep(Duration::from_millis(100)).await;
        let _ = shutdown_tx.send(true);
        pipeline.notifier().notify_one();
        let _ = handle.await;

        let status = pipeline.get_status(&result.message_id);
        assert_eq!(
            status,
            Some(crate::messages::outbound::DeliveryStatus::Failed)
        );
    }

    #[tokio::test]
    async fn test_delivery_skips_disconnected_channel() {
        let mock = Arc::new(MockChannel::new());
        let (pipeline, plugin_reg, channel_reg) =
            make_pipeline_and_registries("disc-ch", Some(mock.clone()), false);

        let msg = OutboundMessage::new("disc-ch", MessageContent::text("hello"));
        pipeline.queue(msg, MsgOutboundContext::new()).unwrap();

        let (shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(false);
        let pl = pipeline.clone();
        let handle = tokio::spawn(async move {
            delivery_loop(pl, plugin_reg, channel_reg, shutdown_rx).await;
        });

        tokio::time::sleep(Duration::from_millis(100)).await;
        let _ = shutdown_tx.send(true);
        pipeline.notifier().notify_one();
        let _ = handle.await;

        // Message should still be queued (not sent, not failed)
        assert_eq!(mock.send_text_count.load(Ordering::Relaxed), 0);
        assert_eq!(pipeline.channels_with_messages().len(), 1);
    }

    #[tokio::test]
    async fn test_delivery_retries_on_retryable_failure_resets_status() {
        let mock = Arc::new(MockChannel::failing(true));
        let (pipeline, plugin_reg, channel_reg) =
            make_pipeline_and_registries("retry-ch", Some(mock.clone()), true);

        let msg = OutboundMessage::new("retry-ch", MessageContent::text("hello"));
        let ctx = MsgOutboundContext::new().with_retries(3);
        let result = pipeline.queue(msg, ctx).unwrap();

        let (shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(false);
        let pl = pipeline.clone();
        let handle = tokio::spawn(async move {
            delivery_loop(pl, plugin_reg, channel_reg, shutdown_rx).await;
        });

        tokio::time::sleep(Duration::from_millis(100)).await;
        let _ = shutdown_tx.send(true);
        pipeline.notifier().notify_one();
        let _ = handle.await;

        // Message status must be reset to Queued (not stuck at Sending, not Failed)
        let status = pipeline.get_status(&result.message_id);
        assert_eq!(
            status,
            Some(crate::messages::outbound::DeliveryStatus::Queued),
            "retryable failure must reset status to Queued, not leave it as Sending"
        );

        // Message should still be in the channel queue for retry
        assert_eq!(
            pipeline.channels_with_messages().len(),
            1,
            "message should remain in channel queue after retryable failure"
        );

        // The error from the failed attempt should be recorded
        let queued = pipeline.get_message(&result.message_id).unwrap();
        assert_eq!(queued.last_error, Some("mock failure".to_string()));
    }

    #[tokio::test]
    async fn test_delivery_retries_on_plugin_worker_backpressure() {
        let mock = Arc::new(MockChannel::backpressured());
        let (pipeline, plugin_reg, channel_reg) =
            make_pipeline_and_registries("backpressure-ch", Some(mock.clone()), true);

        let msg = OutboundMessage::new("backpressure-ch", MessageContent::text("hello"));
        let ctx = MsgOutboundContext::new().with_retries(3);
        let result = pipeline.queue(msg, ctx).unwrap();

        process_channel_messages(
            &["backpressure-ch".to_string()],
            &pipeline,
            &plugin_reg,
            &channel_reg,
        )
        .await;

        assert_eq!(
            pipeline.get_status(&result.message_id),
            Some(crate::messages::outbound::DeliveryStatus::Queued),
            "plugin worker backpressure must remain retryable"
        );
        assert_eq!(
            pipeline.queue_size("backpressure-ch"),
            1,
            "backpressured message should remain in the channel queue"
        );
        let queued = pipeline.get_message(&result.message_id).unwrap();
        assert!(
            queued
                .last_error
                .as_deref()
                .is_some_and(|err| err.contains("plugin worker queue is full")),
            "backpressure reason should be operator-visible"
        );
    }

    #[tokio::test]
    async fn test_delivery_non_retryable_failure_marks_failed() {
        let mock = Arc::new(MockChannel::failing(false));
        let (pipeline, plugin_reg, channel_reg) =
            make_pipeline_and_registries("noretry-ch", Some(mock.clone()), true);

        let msg = OutboundMessage::new("noretry-ch", MessageContent::text("hello"));
        let ctx = MsgOutboundContext::new().with_retries(3);
        let result = pipeline.queue(msg, ctx).unwrap();

        let (shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(false);
        let pl = pipeline.clone();
        let handle = tokio::spawn(async move {
            delivery_loop(pl, plugin_reg, channel_reg, shutdown_rx).await;
        });

        tokio::time::sleep(Duration::from_millis(100)).await;
        let _ = shutdown_tx.send(true);
        pipeline.notifier().notify_one();
        let _ = handle.await;

        // Non-retryable failure must be marked as Failed
        let status = pipeline.get_status(&result.message_id);
        assert_eq!(
            status,
            Some(crate::messages::outbound::DeliveryStatus::Failed),
            "non-retryable failure must be marked as Failed"
        );

        // Message should be removed from the channel queue
        assert_eq!(
            pipeline.queue_size("noretry-ch"),
            0,
            "failed message should be removed from channel queue"
        );
    }

    #[tokio::test]
    async fn test_retry_mechanism_picks_up_reset_messages() {
        // Use a mock that always fails with retryable=true and supplies a
        // short Retry-After so this test exercises deadline-driven retry pickup
        // without waiting for the default retry poll.
        let mock = Arc::new(MockChannel::failing_with_retry_after(10));
        let (pipeline, plugin_reg, channel_reg) =
            make_pipeline_and_registries("pickup-ch", Some(mock.clone()), true);

        let msg = OutboundMessage::new("pickup-ch", MessageContent::text("hello"));
        // Allow 3 retries so the message can be retried multiple times
        let ctx = MsgOutboundContext::new().with_retries(3);
        let result = pipeline.queue(msg, ctx).unwrap();

        let (shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(false);
        let pl = pipeline.clone();
        let handle = tokio::spawn(async move {
            delivery_loop(pl, plugin_reg, channel_reg, shutdown_rx).await;
        });

        // Allow enough time for multiple delivery loop iterations to run.
        // The loop wakes on notify or every 5s; we notify it repeatedly.
        for _ in 0..5 {
            tokio::time::sleep(Duration::from_millis(80)).await;
            pipeline.notifier().notify_one();
        }

        let _ = shutdown_tx.send(true);
        pipeline.notifier().notify_one();
        let _ = handle.await;

        // The mock should have been called more than once, proving the retry
        // mechanism picked up the message again after it was reset to Queued.
        let send_count = mock.send_text_count.load(Ordering::Relaxed);
        assert!(
            send_count > 1,
            "expected multiple delivery attempts from retry, got {}",
            send_count
        );

        // After exhausting retries (3 attempts), the message should be Failed
        // since can_retry() returns false when attempts >= max_retries.
        let queued = pipeline.get_message(&result.message_id).unwrap();
        assert_eq!(
            queued.status,
            crate::messages::outbound::DeliveryStatus::Failed,
            "message should be Failed after exhausting retries (attempts={}, max=3)",
            queued.attempts
        );
    }

    #[tokio::test]
    async fn test_delivery_shutdown() {
        let (pipeline, plugin_reg, channel_reg) =
            make_pipeline_and_registries("shutdown-ch", None, true);

        let (shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(true); // already shut down
        let handle = tokio::spawn(async move {
            delivery_loop(pipeline, plugin_reg, channel_reg, shutdown_rx).await;
        });

        // Should exit quickly since shutdown is already true
        let _ = shutdown_tx.send(true);
        tokio::time::timeout(Duration::from_secs(2), handle)
            .await
            .expect("delivery loop should exit on shutdown")
            .expect("task should not panic");
    }
}
