use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;

use tokio::sync::{mpsc, Notify, Semaphore};
use tracing::{debug, error, info};

use crate::config::{BackpressureStrategy, BufferConfig};
use crate::error::MonkError;
use crate::models::Event;

/// Callback invoked when a buffer flush fails after all retries.
///
/// Receives the number of events that were dropped and the error.
pub(crate) type FlushFailureHandler = Arc<dyn Fn(usize, &MonkError) + Send + Sync>;

/// Async handle used to send events into the background buffer.
#[derive(Clone)]
pub(crate) struct BufferHandle {
    tx: mpsc::Sender<Event>,
    flush_signal: Arc<Notify>,
    capacity: usize,
    closed: Arc<AtomicBool>,
    backpressure: BackpressureStrategy,
}

impl BufferHandle {
    /// Enqueue an event. Behaviour when the buffer is full depends on
    /// [`BackpressureStrategy`]: `FailFast` returns [`MonkError::BufferFull`]
    /// immediately; `Block` suspends until space is available.
    pub async fn send(&self, event: Event) -> crate::error::Result<()> {
        if self.closed.load(Ordering::Relaxed) {
            return Err(MonkError::Validation(
                "buffer closed - client has been shut down".into(),
            ));
        }
        match self.backpressure {
            BackpressureStrategy::FailFast => self.tx.try_send(event).map_err(|e| match e {
                mpsc::error::TrySendError::Full(_) => MonkError::BufferFull {
                    capacity: self.capacity,
                },
                mpsc::error::TrySendError::Closed(_) => {
                    MonkError::Validation("buffer closed - client has been shut down".into())
                }
            }),
            BackpressureStrategy::Block => self.tx.send(event).await.map_err(|_| {
                MonkError::Validation("buffer closed - client has been shut down".into())
            }),
        }
    }

    /// Signal the background worker to flush immediately.
    pub fn flush_signal(&self) -> &Notify {
        &self.flush_signal
    }

    pub fn close(&self) {
        self.closed.store(true, Ordering::Relaxed);
    }
}

/// Background flush task. Reads from `rx`, accumulates events, and flushes
/// via the provided `flush_fn` when the batch is full or the interval elapses.
///
/// Flushes are dispatched to a pool of up to `max_concurrent_flushes` tasks
/// so the worker never blocks on a single HTTP round-trip.
pub(crate) struct BufferWorker {
    rx: mpsc::Receiver<Event>,
    config: BufferConfig,
    flush_signal: Arc<Notify>,
}

impl BufferWorker {
    /// Spawn the buffer infrastructure. Returns a [`BufferHandle`] for the caller and
    /// a [`tokio::task::JoinHandle`] for the background worker.
    pub fn spawn<F, Fut>(
        config: BufferConfig,
        flush_fn: F,
        on_failure: Option<FlushFailureHandler>,
    ) -> (BufferHandle, tokio::task::JoinHandle<()>)
    where
        F: Fn(Vec<Event>) -> Fut + Send + Sync + 'static,
        Fut: std::future::Future<Output = crate::error::Result<()>> + Send + 'static,
    {
        let (tx, rx) = mpsc::channel(config.capacity);
        let flush_signal = Arc::new(Notify::new());
        let closed = Arc::new(AtomicBool::new(false));
        let backpressure = config.backpressure;

        let handle = BufferHandle {
            tx,
            flush_signal: Arc::clone(&flush_signal),
            capacity: config.capacity,
            closed,
            backpressure,
        };

        let worker = BufferWorker {
            rx,
            config,
            flush_signal,
        };

        let join = tokio::spawn(worker.run(Arc::new(flush_fn), on_failure));

        (handle, join)
    }

    async fn run<F, Fut>(mut self, flush_fn: Arc<F>, on_failure: Option<FlushFailureHandler>)
    where
        F: Fn(Vec<Event>) -> Fut + Send + Sync + 'static,
        Fut: std::future::Future<Output = crate::error::Result<()>> + Send + 'static,
    {
        let semaphore = Arc::new(Semaphore::new(self.config.max_concurrent_flushes));
        let max_concurrent = self.config.max_concurrent_flushes;
        let mut batch: Vec<Event> = Vec::with_capacity(self.config.max_batch_size);
        let flush_interval = self.config.flush_interval;

        loop {
            let should_flush;

            tokio::select! {
                biased;

                _ = self.flush_signal.notified() => {
                    while let Ok(event) = self.rx.try_recv() {
                        batch.push(event);
                    }
                    should_flush = !batch.is_empty();
                }

                maybe_event = self.rx.recv() => {
                    match maybe_event {
                        Some(event) => {
                            batch.push(event);
                            should_flush = batch.len() >= self.config.max_batch_size;
                        }
                        None => {
                            if !batch.is_empty() {
                                Self::spawn_flush(
                                    &semaphore,
                                    &flush_fn,
                                    &on_failure,
                                    &mut batch,
                                ).await;
                            }
                            let _ = semaphore
                                .acquire_many(max_concurrent as u32)
                                .await;
                            info!("buffer worker shutting down — all flushes complete");
                            return;
                        }
                    }
                }

                _ = tokio::time::sleep(flush_interval) => {
                    should_flush = !batch.is_empty();
                }
            }

            if should_flush {
                Self::spawn_flush(&semaphore, &flush_fn, &on_failure, &mut batch).await;
            }
        }
    }

    /// Acquire a semaphore permit and dispatch the flush to a separate task.
    async fn spawn_flush<F, Fut>(
        semaphore: &Arc<Semaphore>,
        flush_fn: &Arc<F>,
        on_failure: &Option<FlushFailureHandler>,
        batch: &mut Vec<Event>,
    ) where
        F: Fn(Vec<Event>) -> Fut + Send + Sync + 'static,
        Fut: std::future::Future<Output = crate::error::Result<()>> + Send + 'static,
    {
        let permit = semaphore
            .clone()
            .acquire_owned()
            .await
            .expect("flush semaphore closed unexpectedly");

        let events: Vec<Event> = std::mem::take(batch);
        let count = events.len();
        let flush_fn = Arc::clone(flush_fn);
        let on_failure = on_failure.clone();

        tokio::spawn(async move {
            debug!(count, "flushing event buffer");
            if let Err(e) = flush_fn(events).await {
                error!(count, error = %e, "buffer flush failed — events dropped");
                if let Some(ref handler) = on_failure {
                    handler(count, &e);
                }
            }
            drop(permit);
        });
    }
}
