distributed 2.0.0

use std::time::Duration;

use super::publisher::OutboxPublisher;
use crate::outbox::OutboxMessage;
use crate::SourcedResult;

/// Result of a batch drain operation.
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
pub struct DrainResult {
    pub claimed: usize,
    pub completed: usize,
    pub released: usize,
    pub failed: usize,
}

/// Result of processing a single message.
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
pub struct ProcessOneResult {
    /// Whether any work was done (a message was processed).
    pub did_work: bool,
    /// Whether this call claimed a pending message.
    pub claimed: bool,
    /// Whether the message was successfully published.
    pub completed: bool,
    /// Whether the message was released for retry.
    pub released: bool,
    /// Whether the message permanently failed.
    pub failed: bool,
}

/// Worker for processing loaded outbox messages through an [`OutboxPublisher`].
///
/// [`OutboxDispatcher`]: crate::OutboxDispatcher
/// [`MessagePublisher`]: crate::bus::MessagePublisher
/// [`BusPublisher`]: crate::BusPublisher
/// [`Bus`]: crate::bus::Bus
pub struct OutboxWorker<P> {
    publisher: P,
    worker_id: String,
    batch_size: usize,
    lease: Duration,
    max_attempts: u32,
}

impl<P> OutboxWorker<P> {
    /// Create a new worker with the given publisher.
    pub fn new(publisher: P) -> Self {
        Self {
            publisher,
            worker_id: format!("worker-{}", std::process::id()),
            batch_size: 10,
            lease: Duration::from_secs(60),
            max_attempts: 3,
        }
    }

    /// Set the worker ID (used for lease tracking).
    pub fn with_worker_id(mut self, id: impl Into<String>) -> Self {
        self.worker_id = id.into();
        self
    }

    /// Set the batch size (max records to process per drain).
    pub fn with_batch_size(mut self, size: usize) -> Self {
        self.batch_size = size;
        self
    }

    /// Set the lease duration for claimed records.
    pub fn with_lease(mut self, lease: Duration) -> Self {
        self.lease = lease;
        self
    }

    /// Set the maximum number of attempts before failing a record.
    pub fn with_max_attempts(mut self, max: u32) -> Self {
        self.max_attempts = max;
        self
    }

    /// Get a reference to the publisher.
    pub fn publisher(&self) -> &P {
        &self.publisher
    }

    /// Get a mutable reference to the publisher.
    pub fn publisher_mut(&mut self) -> &mut P {
        &mut self.publisher
    }
}

impl<P: OutboxPublisher> OutboxWorker<P> {
    /// Process a single outbox message.
    ///
    /// If the message is pending, it will be claimed by this worker before
    /// publishing. Store-backed worker loops should persist the outcome with
    /// the outbox store completion or failure APIs.
    pub async fn process_message(
        &mut self,
        message: &mut OutboxMessage,
    ) -> SourcedResult<ProcessOneResult> {
        if message.is_published() || message.is_failed() {
            return Ok(ProcessOneResult::default());
        }

        let mut claimed = false;
        if message.is_pending() {
            message.claim_for(&self.worker_id, self.lease)?;
            claimed = true;
        }

        if !message.is_in_flight() {
            return Ok(ProcessOneResult::default());
        }

        let result = match self
            .publisher
            .publish(&message.event_type, &message.payload, &message.metadata)
            .await
        {
            Ok(()) => {
                message.complete()?;
                ProcessOneResult {
                    did_work: true,
                    claimed,
                    completed: true,
                    ..Default::default()
                }
            }
            Err(err) => {
                let error_msg = err.to_string();
                if message.attempts >= self.max_attempts {
                    message.fail(error_msg)?;
                    ProcessOneResult {
                        did_work: true,
                        claimed,
                        failed: true,
                        ..Default::default()
                    }
                } else {
                    message.release(error_msg)?;
                    ProcessOneResult {
                        did_work: true,
                        claimed,
                        released: true,
                        ..Default::default()
                    }
                }
            }
        };

        Ok(result)
    }

    /// Process a batch of outbox messages.
    pub async fn process_batch(
        &mut self,
        messages: &mut [OutboxMessage],
    ) -> SourcedResult<DrainResult> {
        let mut result = DrainResult::default();

        for message in messages.iter_mut().take(self.batch_size) {
            let processed = self.process_message(message).await?;
            if processed.claimed {
                result.claimed += 1;
            }
            if processed.completed {
                result.completed += 1;
            }
            if processed.released {
                result.released += 1;
            }
            if processed.failed {
                result.failed += 1;
            }
        }

        Ok(result)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::LogPublisher;
    use std::collections::HashMap;
    use std::future::Future;

    struct FailingPublisher;

    impl OutboxPublisher for FailingPublisher {
        type Error = &'static str;

        fn publish<'a>(
            &'a mut self,
            _event_type: &'a str,
            _payload: &'a [u8],
            _metadata: &'a HashMap<String, String>,
        ) -> impl Future<Output = Result<(), Self::Error>> + 'a {
            async { Err("publish failed") }
        }
    }

    #[test]
    fn worker_builder() {
        let worker = OutboxWorker::new(LogPublisher::default())
            .with_worker_id("test-worker")
            .with_batch_size(5)
            .with_lease(Duration::from_secs(30))
            .with_max_attempts(2);

        assert_eq!(worker.worker_id, "test-worker");
        assert_eq!(worker.batch_size, 5);
        assert_eq!(worker.lease, Duration::from_secs(30));
        assert_eq!(worker.max_attempts, 2);
    }

    #[tokio::test]
    async fn process_message_noop_for_published() {
        let mut message = OutboxMessage::create("msg-1", "Event", b"{}".to_vec()).unwrap();
        message.claim_for("worker", Duration::from_secs(1)).unwrap();
        message.complete().unwrap();

        let mut worker = OutboxWorker::new(LogPublisher::default());
        let result = worker.process_message(&mut message).await.unwrap();
        assert!(!result.did_work);
    }

    #[tokio::test]
    async fn process_message_passes_metadata_to_publisher() {
        use std::sync::{Arc, Mutex};

        let buffer = Arc::new(Mutex::new(Vec::new()));
        let publisher = LogPublisher::with_buffer(buffer.clone());
        let mut worker = OutboxWorker::new(publisher);

        let mut message = OutboxMessage::create("msg-1", "UserCreated", b"{}".to_vec()).unwrap();
        message.set_correlation_id("req-abc");

        let result = worker.process_message(&mut message).await.unwrap();
        assert!(result.completed);

        let logs = buffer.lock().unwrap();
        assert!(logs[0].contains("correlation_id"));
        assert!(logs[0].contains("req-abc"));
    }

    #[tokio::test]
    async fn process_batch_counts_pending_messages_claimed_by_this_call() {
        let mut messages = vec![OutboxMessage::create("msg-1", "Event", b"{}".to_vec()).unwrap()];
        let mut worker = OutboxWorker::new(LogPublisher::default());

        let result = worker.process_batch(&mut messages).await.unwrap();

        assert_eq!(result.claimed, 1);
        assert_eq!(result.completed, 1);
    }

    #[tokio::test]
    async fn process_batch_does_not_count_already_in_flight_messages_as_claimed() {
        let mut message = OutboxMessage::create("msg-1", "Event", b"{}".to_vec()).unwrap();
        message
            .claim_for("other-worker", Duration::from_secs(1))
            .unwrap();
        let mut messages = vec![message];
        let mut worker = OutboxWorker::new(LogPublisher::default());

        let result = worker.process_batch(&mut messages).await.unwrap();

        assert_eq!(result.claimed, 0);
        assert_eq!(result.completed, 1);
    }

    #[tokio::test]
    async fn process_message_fails_when_claimed_attempt_reaches_max_attempts() {
        let mut message = OutboxMessage::create("msg-1", "Event", b"{}".to_vec()).unwrap();
        let mut worker = OutboxWorker::new(FailingPublisher).with_max_attempts(2);

        let first = worker.process_message(&mut message).await.unwrap();
        assert!(first.released);
        assert!(!first.failed);
        assert!(message.is_pending());
        assert_eq!(message.attempts, 1);

        let second = worker.process_message(&mut message).await.unwrap();
        assert!(!second.released);
        assert!(second.failed);
        assert!(message.is_failed());
        assert_eq!(message.attempts, 2);
    }
}