use std::{cmp::Ordering, collections::BinaryHeap, marker::PhantomData, sync::Arc};

use jiff::Timestamp;
use tokio::sync::{Mutex, Notify};

use crate::{Specification, manifests::ManifestName};

#[derive(Debug, Clone, Eq, PartialEq)]
struct QueueEntry {
    scheduled_at: Timestamp,
    key: ManifestName,
}

impl Ord for QueueEntry {
    fn cmp(&self, other: &Self) -> Ordering {
        // Reverse ordering for min-heap (earliest first)
        other
            .scheduled_at
            .cmp(&self.scheduled_at)
            .then_with(|| self.key.cmp(&other.key))
    }
}

impl PartialOrd for QueueEntry {
    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
        Some(self.cmp(other))
    }
}

struct Inner {
    queue: BinaryHeap<QueueEntry>,
}

/// A simple sorted idempotent queue for scheduling reconciliation jobs.
///
/// Each key can only appear once in the queue. If a key is enqueued again
/// with an earlier scheduled time, the existing entry is updated.
///
/// This queue is Clone, Send, and Sync - it can be shared across threads.
pub struct ReconcileQueue<T: Specification> {
    inner: Arc<Mutex<Inner>>,
    notify: Arc<Notify>,
    _marker: PhantomData<T>,
}

impl<T: Specification> Clone for ReconcileQueue<T> {
    fn clone(&self) -> Self {
        Self {
            inner: Arc::clone(&self.inner),
            notify: Arc::clone(&self.notify),
            _marker: PhantomData,
        }
    }
}

impl<T: Specification> Default for ReconcileQueue<T> {
    fn default() -> Self {
        Self::new()
    }
}

impl<T: Specification> ReconcileQueue<T> {
    pub fn new() -> Self {
        Self {
            inner: Arc::new(Mutex::new(Inner {
                queue: BinaryHeap::new(),
            })),
            notify: Arc::new(Notify::new()),
            _marker: PhantomData,
        }
    }

    /// Enqueue a key to be reconciled now.
    pub async fn enqueue(&self, key: ManifestName) {
        self.enqueue_at(key, Timestamp::now()).await;
    }

    /// Enqueue a key to be reconciled at a specific time.
    ///
    /// If the key already exists in the queue with a later scheduled time,
    /// it will be updated to the earlier time. If it exists with an earlier
    /// or equal time, this is a no-op.
    pub async fn enqueue_at(&self, key: ManifestName, scheduled_at: Timestamp) {
        let mut inner = self.inner.lock().await;

        // Remove any existing entry for this key
        let existing: Vec<_> = inner.queue.drain().collect();
        let mut found_earlier = false;

        for entry in existing {
            if entry.key == key {
                // Keep the earlier scheduled time
                if entry.scheduled_at <= scheduled_at {
                    inner.queue.push(entry);
                    found_earlier = true;
                }
            } else {
                inner.queue.push(entry);
            }
        }

        // Add new entry if we didn't find an earlier one
        if !found_earlier {
            inner.queue.push(QueueEntry { scheduled_at, key });
        }

        self.notify.notify_one();
    }

    /// Try to take the next ready job from the queue.
    ///
    /// Returns `Some(key)` if there's a job scheduled at or before now,
    /// otherwise returns `None`.
    pub async fn try_take(&self) -> Option<ManifestName> {
        let now = Timestamp::now();
        let mut inner = self.inner.lock().await;

        if let Some(entry) = inner.queue.peek()
            && entry.scheduled_at <= now
        {
            return inner.queue.pop().map(|e| e.key);
        }

        None
    }

    /// Returns how long until the next job is ready, if any.
    pub async fn next_ready_in(&self) -> Option<std::time::Duration> {
        let now = Timestamp::now();
        let inner = self.inner.lock().await;

        inner.queue.peek().and_then(|entry| {
            let span = entry.scheduled_at - now;
            if span.is_negative() {
                Some(std::time::Duration::ZERO)
            } else {
                span.try_into().ok()
            }
        })
    }

    /// Wait for the next job to be ready and return it.
    ///
    /// This will block until a job is ready.
    pub async fn take(&self) -> ManifestName {
        loop {
            if let Some(key) = self.try_take().await {
                return key;
            }

            match self.next_ready_in().await {
                Some(duration) if duration > std::time::Duration::ZERO => {
                    tokio::select! {
                        _ = tokio::time::sleep(duration) => {}
                        _ = self.notify.notified() => {}
                    }
                }
                Some(_) => {
                    // Duration is zero, should have a ready item
                    continue;
                }
                None => {
                    // Queue is empty, wait for notification
                    self.notify.notified().await;
                }
            }
        }
    }

    #[cfg(test)]
    /// Returns the number of pending jobs in the queue.
    pub async fn len(&self) -> usize {
        self.inner.lock().await.queue.len()
    }

    #[cfg(test)]
    /// Returns true if the queue is empty.
    pub async fn is_empty(&self) -> bool {
        self.inner.lock().await.queue.is_empty()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use jiff::ToSpan;
    use serde::{Deserialize, Serialize};

    #[derive(Clone, Serialize, Deserialize)]
    struct TestSpec;

    impl Specification for TestSpec {
        fn kind(&self) -> &'static str {
            "test"
        }
    }

    #[tokio::test]
    async fn test_enqueue_and_take() {
        let queue: ReconcileQueue<TestSpec> = ReconcileQueue::new();

        queue.enqueue("job1".to_string()).await;
        queue.enqueue("job2".to_string()).await;

        assert_eq!(queue.len().await, 2);

        let job = queue.try_take().await;
        assert!(job.is_some());
        assert_eq!(queue.len().await, 1);
    }

    #[tokio::test]
    async fn test_idempotent_enqueue() {
        let queue: ReconcileQueue<TestSpec> = ReconcileQueue::new();

        queue.enqueue("job1".to_string()).await;
        queue.enqueue("job1".to_string()).await;

        assert_eq!(queue.len().await, 1);
    }

    #[tokio::test]
    async fn test_scheduled_order() {
        let queue: ReconcileQueue<TestSpec> = ReconcileQueue::new();

        let now = Timestamp::now();
        let later = now.checked_add(1.hour()).unwrap();

        queue.enqueue_at("later_job".to_string(), later).await;
        queue.enqueue_at("now_job".to_string(), now).await;

        // Should get the "now" job first
        let job = queue.try_take().await;
        assert_eq!(job, Some("now_job".to_string()));

        // Later job should not be ready yet
        let job = queue.try_take().await;
        assert!(job.is_none());
    }

    #[tokio::test]
    async fn test_enqueue_earlier_updates() {
        let queue: ReconcileQueue<TestSpec> = ReconcileQueue::new();

        let now = Timestamp::now();
        let later = now.checked_add(1.hour()).unwrap();

        // First enqueue for later
        queue.enqueue_at("job1".to_string(), later).await;
        assert!(queue.try_take().await.is_none());

        // Re-enqueue for now - should update
        queue.enqueue_at("job1".to_string(), now).await;
        assert_eq!(queue.try_take().await, Some("job1".to_string()));
    }

    #[tokio::test]
    async fn test_enqueue_later_no_op() {
        let queue: ReconcileQueue<TestSpec> = ReconcileQueue::new();

        let now = Timestamp::now();
        let later = now.checked_add(1.hour()).unwrap();

        // First enqueue for now
        queue.enqueue_at("job1".to_string(), now).await;

        // Try to enqueue for later - should be no-op
        queue.enqueue_at("job1".to_string(), later).await;

        // Should still be ready now
        assert_eq!(queue.try_take().await, Some("job1".to_string()));
    }

    #[tokio::test]
    async fn test_clone_shares_state() {
        let queue1: ReconcileQueue<TestSpec> = ReconcileQueue::new();
        let queue2 = queue1.clone();

        queue1.enqueue("job1".to_string()).await;
        assert_eq!(queue2.len().await, 1);

        let job = queue2.try_take().await;
        assert_eq!(job, Some("job1".to_string()));
        assert!(queue1.is_empty().await);
    }
}
