//! Bounded concurrency using semaphores
//!
//! Provides utilities to limit concurrent operations to prevent resource exhaustion.
//!
//! # Example
//!
//! ```rust
//! use common::resilience::bounded_concurrency::BoundedExecutor;
//!
//! # async fn example() -> Result<(), Box<dyn std::error::Error>> {
//! let executor = BoundedExecutor::new(10); // Max 10 concurrent operations
//!
//! let result = executor.execute(|| async {
//!     // Your async operation
//!     Ok::<_, std::io::Error>(42)
//! }).await?;
//! # Ok(())
//! # }
//! ```

use std::future::Future;
use std::sync::Arc;
use tokio::sync::Semaphore;

/// Executor that limits concurrent operations using a semaphore
///
/// Prevents resource exhaustion by limiting the number of operations
/// that can run simultaneously.
#[derive(Clone)]
pub struct BoundedExecutor {
    /// Semaphore for limiting concurrency
    semaphore: Arc<Semaphore>,
    /// Maximum concurrent operations allowed
    max_concurrent: usize,
}

impl BoundedExecutor {
    /// Create a new bounded executor
    ///
    /// # Arguments
    ///
    /// * `max_concurrent` - Maximum number of concurrent operations allowed
    ///
    /// # Example
    ///
    /// ```rust
    /// use common::resilience::bounded_concurrency::BoundedExecutor;
    ///
    /// let executor = BoundedExecutor::new(10);
    /// ```
    pub fn new(max_concurrent: usize) -> Self {
        Self {
            semaphore: Arc::new(Semaphore::new(max_concurrent)),
            max_concurrent,
        }
    }

    /// Execute an async operation with concurrency limit
    ///
    /// Acquires a permit from the semaphore before executing the operation.
    /// Blocks if the concurrency limit is reached until a permit becomes available.
    ///
    /// # Arguments
    ///
    /// * `operation` - Async operation to execute
    ///
    /// # Returns
    ///
    /// Result of the operation
    ///
    /// # Example
    ///
    /// ```rust
    /// use common::resilience::bounded_concurrency::BoundedExecutor;
    ///
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// let executor = BoundedExecutor::new(5);
    ///
    /// let result = executor.execute(|| async {
    ///     // Your async work
    ///     Ok::<_, std::io::Error>(())
    /// }).await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn execute<F, Fut, T, E>(&self, operation: F) -> Result<T, E>
    where
        F: FnOnce() -> Fut,
        Fut: Future<Output = Result<T, E>>,
    {
        // Acquire permit (blocks if limit reached)
        let _permit = match self.semaphore.acquire().await {
            Ok(permit) => Some(permit),
            Err(_) => {
                tracing::error!(
                    "BoundedConcurrency: semaphore closed unexpectedly, executing without permit"
                );
                None
            }
        };

        // Execute operation while holding permit
        operation().await
        // Permit is automatically released when _permit is dropped
    }

    /// Try to execute an operation without blocking
    ///
    /// Returns immediately with an error if the concurrency limit is reached.
    ///
    /// # Arguments
    ///
    /// * `operation` - Async operation to execute
    ///
    /// # Returns
    ///
    /// * `Ok(T)` - Operation succeeded
    /// * `Err(BoundedExecutorError::LimitReached)` - Concurrency limit reached
    /// * `Err(BoundedExecutorError::Operation(E))` - Operation failed
    ///
    /// # Example
    ///
    /// ```rust
    /// use common::resilience::bounded_concurrency::BoundedExecutor;
    ///
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// let executor = BoundedExecutor::new(5);
    ///
    /// match executor.try_execute(|| async {
    ///     Ok::<_, std::io::Error>(())
    /// }).await {
    ///     Ok(result) => println!("Success: {:?}", result),
    ///     Err(e) => println!("Failed: {:?}", e),
    /// }
    /// # Ok(())
    /// # }
    /// ```
    pub async fn try_execute<F, Fut, T, E>(
        &self,
        operation: F,
    ) -> Result<T, BoundedExecutorError<E>>
    where
        F: FnOnce() -> Fut,
        Fut: Future<Output = Result<T, E>>,
    {
        // Try to acquire permit without blocking
        let _permit = self
            .semaphore
            .try_acquire()
            .map_err(|_| BoundedExecutorError::LimitReached)?;

        // Execute operation
        operation()
            .await
            .map_err(BoundedExecutorError::Operation)
    }

    /// Get the maximum concurrency limit
    pub fn max_concurrent(&self) -> usize {
        self.max_concurrent
    }

    /// Get the number of available permits
    ///
    /// This can be used to monitor how close to the limit the executor is.
    pub fn available_permits(&self) -> usize {
        self.semaphore.available_permits()
    }

    /// Check if the executor is at capacity
    pub fn is_at_capacity(&self) -> bool {
        self.available_permits() == 0
    }
}

/// Errors that can occur when using bounded executor
#[derive(Debug, thiserror::Error)]
pub enum BoundedExecutorError<E> {
    /// Concurrency limit reached
    #[error("Concurrency limit reached - all permits in use")]
    LimitReached,

    /// Operation failed
    #[error("Operation failed: {0}")]
    Operation(E),
}

impl std::fmt::Debug for BoundedExecutor {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("BoundedExecutor")
            .field("max_concurrent", &self.max_concurrent)
            .field("available_permits", &self.available_permits())
            .finish()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::atomic::{AtomicU32, Ordering};
    use std::time::Duration;
    use tokio::time::sleep;

    #[tokio::test]
    async fn test_bounded_executor_basic() {
        let executor = BoundedExecutor::new(5);

        let result = executor
            .execute(|| async { Ok::<_, std::io::Error>(42) })
            .await;

        assert_eq!(result.unwrap(), 42);
    }

    #[tokio::test]
    async fn test_bounded_executor_limits_concurrency() {
        let executor = BoundedExecutor::new(2);
        let counter = Arc::new(AtomicU32::new(0));

        let mut handles = vec![];

        for _ in 0..5 {
            let executor = executor.clone();
            let counter = Arc::clone(&counter);

            let handle = tokio::spawn(async move {
                executor
                    .execute(|| async {
                        let current = counter.fetch_add(1, Ordering::SeqCst);
                        // Ensure max 2 concurrent
                        assert!(current < 2, "More than 2 concurrent operations");

                        sleep(Duration::from_millis(50)).await;

                        counter.fetch_sub(1, Ordering::SeqCst);
                        Ok::<_, std::io::Error>(())
                    })
                    .await
            });

            handles.push(handle);
        }

        for handle in handles {
            handle.await.unwrap().unwrap();
        }
    }

    #[tokio::test]
    async fn test_try_execute_fails_when_at_capacity() {
        let executor = BoundedExecutor::new(1);

        // First operation should succeed
        let _guard = executor.semaphore.acquire().await.unwrap();

        // Second operation should fail immediately
        let result = executor
            .try_execute(|| async { Ok::<_, std::io::Error>(()) })
            .await;

        assert!(matches!(result, Err(BoundedExecutorError::LimitReached)));
    }

    #[tokio::test]
    async fn test_available_permits() {
        let executor = BoundedExecutor::new(5);
        assert_eq!(executor.available_permits(), 5);

        let _permit1 = executor.semaphore.acquire().await.unwrap();
        assert_eq!(executor.available_permits(), 4);

        let _permit2 = executor.semaphore.acquire().await.unwrap();
        assert_eq!(executor.available_permits(), 3);
    }

    #[tokio::test]
    async fn test_is_at_capacity() {
        let executor = BoundedExecutor::new(2);
        assert!(!executor.is_at_capacity());

        let _p1 = executor.semaphore.acquire().await.unwrap();
        let _p2 = executor.semaphore.acquire().await.unwrap();
        assert!(executor.is_at_capacity());
    }
}
