coderlib 0.1.0

//! Async utilities for CoderLib

use std::future::Future;
use std::time::Duration;
use tokio::time::{timeout, sleep};

/// Retry an async operation with exponential backoff
pub async fn retry_with_exponential_backoff<F, Fut, T, E>(
    mut operation: F,
    max_retries: usize,
    initial_delay: Duration,
    max_delay: Duration,
) -> Result<T, E>
where
    F: FnMut() -> Fut,
    Fut: Future<Output = Result<T, E>>,
{
    let mut delay = initial_delay;
    
    for attempt in 0..=max_retries {
        match operation().await {
            Ok(result) => return Ok(result),
            Err(e) => {
                if attempt == max_retries {
                    return Err(e);
                }
                
                sleep(delay).await;
                delay = std::cmp::min(delay * 2, max_delay);
            }
        }
    }
    
    unreachable!()
}

/// Run an async operation with a timeout
pub async fn with_timeout<F, T>(
    operation: F,
    duration: Duration,
) -> Result<T, TimeoutError>
where
    F: Future<Output = T>,
{
    timeout(duration, operation)
        .await
        .map_err(|_| TimeoutError)
}

/// Timeout error
#[derive(Debug, Clone, Copy, PartialEq)]
pub struct TimeoutError;

impl std::fmt::Display for TimeoutError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "Operation timed out")
    }
}

impl std::error::Error for TimeoutError {}

/// Debounce an async operation
pub struct Debouncer<T> {
    delay: Duration,
    last_value: Option<T>,
    task_handle: Option<tokio::task::JoinHandle<()>>,
}

impl<T: Clone + Send + 'static> Debouncer<T> {
    /// Create a new debouncer
    pub fn new(delay: Duration) -> Self {
        Self {
            delay,
            last_value: None,
            task_handle: None,
        }
    }
    
    /// Trigger the debounced operation
    pub async fn trigger<F, Fut>(&mut self, value: T, operation: F)
    where
        F: FnOnce(T) -> Fut + Send + 'static,
        Fut: Future<Output = ()> + Send,
    {
        self.last_value = Some(value.clone());
        
        // Cancel previous task if it exists
        if let Some(handle) = self.task_handle.take() {
            handle.abort();
        }
        
        let delay = self.delay;
        self.task_handle = Some(tokio::spawn(async move {
            sleep(delay).await;
            operation(value).await;
        }));
    }
}

/// Throttle an async operation
pub struct Throttler {
    last_execution: Option<std::time::Instant>,
    min_interval: Duration,
}

impl Throttler {
    /// Create a new throttler
    pub fn new(min_interval: Duration) -> Self {
        Self {
            last_execution: None,
            min_interval,
        }
    }
    
    /// Execute an operation if enough time has passed
    pub async fn execute<F, Fut, T>(&mut self, operation: F) -> Option<T>
    where
        F: FnOnce() -> Fut,
        Fut: Future<Output = T>,
    {
        let now = std::time::Instant::now();
        
        if let Some(last) = self.last_execution {
            if now.duration_since(last) < self.min_interval {
                return None;
            }
        }
        
        self.last_execution = Some(now);
        Some(operation().await)
    }
    
    /// Execute an operation, waiting if necessary
    pub async fn execute_with_wait<F, Fut, T>(&mut self, operation: F) -> T
    where
        F: FnOnce() -> Fut,
        Fut: Future<Output = T>,
    {
        let now = std::time::Instant::now();
        
        if let Some(last) = self.last_execution {
            let elapsed = now.duration_since(last);
            if elapsed < self.min_interval {
                sleep(self.min_interval - elapsed).await;
            }
        }
        
        self.last_execution = Some(std::time::Instant::now());
        operation().await
    }
}

/// Batch processor for async operations
pub struct BatchProcessor<T> {
    batch_size: usize,
    flush_interval: Duration,
    items: Vec<T>,
    last_flush: std::time::Instant,
}

impl<T: Send + 'static> BatchProcessor<T> {
    /// Create a new batch processor
    pub fn new(batch_size: usize, flush_interval: Duration) -> Self {
        Self {
            batch_size,
            flush_interval,
            items: Vec::new(),
            last_flush: std::time::Instant::now(),
        }
    }
    
    /// Add an item to the batch
    pub async fn add<F, Fut>(&mut self, item: T, processor: F) -> bool
    where
        F: FnOnce(Vec<T>) -> Fut,
        Fut: Future<Output = ()>,
    {
        self.items.push(item);
        
        let should_flush = self.items.len() >= self.batch_size ||
            self.last_flush.elapsed() >= self.flush_interval;
        
        if should_flush {
            self.flush(processor).await;
            true
        } else {
            false
        }
    }
    
    /// Flush the current batch
    pub async fn flush<F, Fut>(&mut self, processor: F)
    where
        F: FnOnce(Vec<T>) -> Fut,
        Fut: Future<Output = ()>,
    {
        if !self.items.is_empty() {
            let items = std::mem::take(&mut self.items);
            processor(items).await;
            self.last_flush = std::time::Instant::now();
        }
    }
}

/// Async semaphore for limiting concurrent operations
pub struct AsyncSemaphore {
    permits: tokio::sync::Semaphore,
}

impl AsyncSemaphore {
    /// Create a new semaphore with the given number of permits
    pub fn new(permits: usize) -> Self {
        Self {
            permits: tokio::sync::Semaphore::new(permits),
        }
    }
    
    /// Acquire a permit and execute an operation
    pub async fn execute<F, Fut, T>(&self, operation: F) -> Result<T, tokio::sync::AcquireError>
    where
        F: FnOnce() -> Fut,
        Fut: Future<Output = T>,
    {
        let _permit = self.permits.acquire().await?;
        Ok(operation().await)
    }
    
    /// Try to acquire a permit without waiting
    pub fn try_execute<F, Fut, T>(&self, operation: F) -> Option<impl Future<Output = T>>
    where
        F: FnOnce() -> Fut,
        Fut: Future<Output = T>,
    {
        if let Ok(_permit) = self.permits.try_acquire() {
            Some(async move {
                operation().await
            })
        } else {
            None
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::atomic::{AtomicUsize, Ordering};
    use std::sync::Arc;

    #[tokio::test]
    async fn test_retry_with_exponential_backoff() {
        let counter = Arc::new(AtomicUsize::new(0));
        let counter_clone = counter.clone();
        
        let result = retry_with_exponential_backoff(
            move || {
                let count = counter_clone.fetch_add(1, Ordering::SeqCst);
                async move {
                    if count < 2 {
                        Err("not yet")
                    } else {
                        Ok("success")
                    }
                }
            },
            5,
            Duration::from_millis(1),
            Duration::from_millis(100),
        ).await;
        
        assert_eq!(result, Ok("success"));
        assert_eq!(counter.load(Ordering::SeqCst), 3);
    }

    #[tokio::test]
    async fn test_with_timeout() {
        // Should succeed
        let result = with_timeout(
            async { "success" },
            Duration::from_millis(100),
        ).await;
        assert_eq!(result, Ok("success"));
        
        // Should timeout
        let result = with_timeout(
            async {
                sleep(Duration::from_millis(200)).await;
                "too late"
            },
            Duration::from_millis(100),
        ).await;
        assert!(result.is_err());
    }

    #[tokio::test]
    async fn test_throttler() {
        let mut throttler = Throttler::new(Duration::from_millis(50));
        
        // First execution should succeed
        let result1 = throttler.execute(|| async { "first" }).await;
        assert_eq!(result1, Some("first"));
        
        // Immediate second execution should be throttled
        let result2 = throttler.execute(|| async { "second" }).await;
        assert_eq!(result2, None);
        
        // After waiting, should succeed
        sleep(Duration::from_millis(60)).await;
        let result3 = throttler.execute(|| async { "third" }).await;
        assert_eq!(result3, Some("third"));
    }

    #[tokio::test]
    async fn test_batch_processor() {
        let processed = Arc::new(AtomicUsize::new(0));
        let processed_clone = processed.clone();
        
        let mut processor = BatchProcessor::new(3, Duration::from_millis(100));
        
        // Add items one by one
        let flushed1 = processor.add("item1", |items| {
            let processed = processed_clone.clone();
            async move {
                processed.fetch_add(items.len(), Ordering::SeqCst);
            }
        }).await;
        assert!(!flushed1);
        
        let flushed2 = processor.add("item2", |items| {
            let processed = processed_clone.clone();
            async move {
                processed.fetch_add(items.len(), Ordering::SeqCst);
            }
        }).await;
        assert!(!flushed2);
        
        let flushed3 = processor.add("item3", |items| {
            let processed = processed_clone.clone();
            async move {
                processed.fetch_add(items.len(), Ordering::SeqCst);
            }
        }).await;
        assert!(flushed3); // Should flush when batch size is reached
        
        assert_eq!(processed.load(Ordering::SeqCst), 3);
    }

    #[tokio::test]
    async fn test_async_semaphore() {
        let semaphore = Arc::new(AsyncSemaphore::new(2));
        let counter = Arc::new(AtomicUsize::new(0));

        let mut handles = Vec::new();

        for _ in 0..5 {
            let semaphore = semaphore.clone();
            let counter = counter.clone();

            let handle = tokio::spawn(async move {
                semaphore.execute(|| async {
                    counter.fetch_add(1, Ordering::SeqCst);
                    sleep(Duration::from_millis(10)).await;
                    counter.fetch_sub(1, Ordering::SeqCst);
                }).await.unwrap();
            });

            handles.push(handle);
        }

        // Wait a bit and check that no more than 2 are running concurrently
        sleep(Duration::from_millis(5)).await;
        assert!(counter.load(Ordering::SeqCst) <= 2);

        // Wait for all to complete
        for handle in handles {
            handle.await.unwrap();
        }

        assert_eq!(counter.load(Ordering::SeqCst), 0);
    }
}