use arc_swap::ArcSwapOption;
use serde::Deserialize;
use std::{
    sync::{
        Arc,
        atomic::{AtomicBool, Ordering},
    },
    time::Duration,
};
use tokio::runtime::Handle;
use tokio::sync::Notify;
use kameleoon_trace_logging::trace_method;

use futures_util::StreamExt;
use reqwest_eventsource::{Error, Event, EventSource, retry::RetryPolicy};

use crate::{
    debug_log, error_log,
    logging::{KameleoonLogger, LogLevel},
    network::url_provider::UrlProvider,
};

const EVENT_NAME: &str = "configuration-update-event";
const DELAY_RECONNECTION_TIMEOUT: Duration = Duration::from_millis(5_000);

// MARK: - UpdateConfigurationEventDto

#[derive(Debug, Deserialize)]
struct UpdateDataFileEventDto {
    #[serde(rename = "siteCode")]
    #[cfg(test)]
    site_code: String,
    #[serde(rename = "ts")]
    timestamp: u64,
}

// MARK: - StreamingRetryPolicy

struct StreamingRetryPolicy {
    max_retries: usize,
    delay: Duration,
    reconnection_time: Duration,
}

impl RetryPolicy for StreamingRetryPolicy {
    fn retry(&self, _error: &Error, last_retry: Option<(usize, Duration)>) -> Option<Duration> {
        if let Some((retry_count, _)) = last_retry {
            if retry_count >= self.max_retries {
                return None; // stop retrying
            }
        }
        Some(self.delay)
    }

    fn set_reconnection_time(&mut self, duration: Duration) {
        self.reconnection_time = duration;
    }
}

// MARK: - StreamingService

/// Real-Time Streaming Event Service (SSE). Using in Streaming mode to fetch data file in real time.
pub(crate) trait StreamingService: Send + Sync {
    fn connect(&self, on_event: Box<dyn Fn(u64) + Send + Sync>, on_close: Box<dyn FnOnce() + Send + Sync>);
    fn disconnect(&self);
    fn is_connected(&self) -> bool;
}

// MARK: - StreamingServiceImpl

pub(crate) struct StreamingServiceImpl<U: UrlProvider> {
    url_provider: Arc<U>,
    runtime_handle: Handle,
    disconnect_notify: ArcSwapOption<Notify>,
    is_connected: Arc<AtomicBool>,
}

impl<U: UrlProvider + Send + Sync + 'static> StreamingService for StreamingServiceImpl<U> {
    #[trace_method(LogLevel::Debug, skip(self, on_event, on_disconnect))]
    fn connect(&self, on_event: Box<dyn Fn(u64) + Send + Sync>, on_disconnect: Box<dyn FnOnce() + Send + Sync>) {
        self.is_connected.store(true, Ordering::Relaxed);

        let event_source = self.make_event_source();
        let disconnect_notify = self.setup_disconnect_notification();
        let is_connected = self.is_connected.clone();

        self.runtime_handle.spawn(async move {
            Self::run_event_loop(event_source, disconnect_notify, on_event).await;
            on_disconnect();
            is_connected.store(false, Ordering::Relaxed);
        });
    }

    #[trace_method(LogLevel::Debug, skip(self))]
    fn disconnect(&self) {
        if let Some(notify) = self.disconnect_notify.swap(None) {
            notify.notify_one();
        }
    }

    fn is_connected(&self) -> bool {
        self.is_connected.load(Ordering::Relaxed)
    }
}

impl<U: UrlProvider> StreamingServiceImpl<U> {
    #[trace_method(LogLevel::Debug, skip(retval, url_provider, runtime_handle))]
    pub fn new(url_provider: Arc<U>, runtime_handle: Handle) -> Self {
        Self {
            url_provider,
            runtime_handle,
            disconnect_notify: ArcSwapOption::empty(),
            is_connected: Arc::new(AtomicBool::new(false)),
        }
    }

    fn make_event_source(&self) -> EventSource {
        let streaming_url = self.url_provider.make_real_time_url();
        let mut event_source = EventSource::get(streaming_url);
        event_source.set_retry_policy(Box::new(StreamingRetryPolicy {
            max_retries: 3,
            delay: DELAY_RECONNECTION_TIMEOUT,
            reconnection_time: DELAY_RECONNECTION_TIMEOUT,
        }));
        event_source
    }

    fn setup_disconnect_notification(&self) -> Arc<Notify> {
        let disconnect_notify = Arc::new(Notify::new());
        self.disconnect_notify.store(Some(disconnect_notify.clone()));
        disconnect_notify
    }

    fn handle_event(maybe_event: Option<Result<Event, Error>>, on_event: &dyn Fn(u64)) -> bool {
        let Some(event_result) = maybe_event else {
            return false; // Stream ended
        };
        match event_result {
            Ok(Event::Open) => KameleoonLogger::debug("StreamingService: established connection!"),
            Ok(Event::Message(message)) => {
                debug_log!("StreamingService: received message: {:#?}", message);
                if message.event == EVENT_NAME {
                    match serde_json::from_str::<UpdateDataFileEventDto>(&message.data) {
                        Ok(update_configuration_event) => on_event(update_configuration_event.timestamp),
                        Err(err) => error_log!("StreamingService received wrong message: {}", err),
                    }
                }
            }
            Err(err) => error_log!("StreamingService: disconnected by error: {}", err),
        }
        true
    }

    async fn run_event_loop(
        mut event_source: EventSource,
        disconnect_notify: Arc<Notify>,
        on_event: Box<dyn Fn(u64) + Send + Sync>,
    ) {
        loop {
            tokio::select! {
                maybe_event = event_source.next() => {
                    if !Self::handle_event(maybe_event, &*on_event) {
                        break;
                    }
                }
                _ = disconnect_notify.notified() => {
                    KameleoonLogger::debug("StreamingService: shutting down");
                    break;
                }
            }
        }
        event_source.close();
    }
}

// MARK: - Tests

#[cfg(test)]
mod tests {
    use crate::network::network_test_utils::MockUrlProvider;

    use super::*;
    use mockall::predicate::*;
    use std::sync::{Arc, Mutex};
    use std::time::Duration;
    use tokio::sync::mpsc;
    use tokio::time::timeout;
    use wiremock::matchers::{method, path};
    use wiremock::{Mock, MockServer, ResponseTemplate};

    // MARK: - StreamingRetryPolicy (Test)

    #[test]
    fn test_streaming_retry_policy_retries() {
        let policy = StreamingRetryPolicy {
            max_retries: 2,
            delay: Duration::from_millis(1000),
            reconnection_time: Duration::from_millis(5000),
        };

        let error = Error::InvalidLastEventId("invalid event id".to_owned());

        // First retry
        let result = policy.retry(&error, Some((0, Duration::from_millis(500))));
        assert!(result.is_some());
        assert_eq!(result.unwrap(), Duration::from_millis(1000));

        // Second retry
        let result = policy.retry(&error, Some((1, Duration::from_millis(500))));
        assert!(result.is_some());
        assert_eq!(result.unwrap(), Duration::from_millis(1000));

        // Third retry should fail
        let result = policy.retry(&error, Some((2, Duration::from_millis(500))));
        assert!(result.is_none());
    }

    #[test]
    fn test_streaming_retry_policy_set_reconnection_time() {
        let mut policy = StreamingRetryPolicy {
            max_retries: 3,
            delay: Duration::from_millis(1000),
            reconnection_time: Duration::from_millis(5000),
        };

        let new_duration = Duration::from_millis(10000);
        policy.set_reconnection_time(new_duration);

        assert_eq!(policy.reconnection_time, new_duration);
    }

    // MARK: - UpdateConfigurationDto (Test)

    use rstest::rstest;

    #[rstest]
    #[case(r#"{"siteCode":"test-site","ts":1234567890}"#, Some(("test-site", 1234567890)))]
    #[case(r#"{"invalid":"data"}"#, None)]
    fn test_update_datafile_event_dto_deserialization(#[case] json: &str, #[case] expected: Option<(&str, u64)>) {
        let result = serde_json::from_str::<UpdateDataFileEventDto>(json);

        match expected {
            Some((expected_site_code, expected_timestamp)) => {
                let dto = result.expect("Expected successful deserialization");
                assert_eq!(dto.site_code, expected_site_code);
                assert_eq!(dto.timestamp, expected_timestamp);
            }
            None => {
                assert!(result.is_err(), "Expected deserialization to fail for: {}", json);
            }
        }
    }

    // MARK: - EventSource

    // Helper struct to capture callback invocations
    #[derive(Debug, Clone, Default)]
    struct CallbackCapture {
        pub events: Arc<Mutex<Vec<u64>>>,
        pub disconnect_called: Arc<Mutex<bool>>,
    }

    impl CallbackCapture {
        fn new() -> Self {
            Self {
                events: Arc::new(Mutex::new(Vec::new())),
                disconnect_called: Arc::new(Mutex::new(false)),
            }
        }

        fn create_on_event_callback(&self) -> Box<dyn Fn(u64) + Send + Sync> {
            let events = Arc::clone(&self.events);
            Box::new(move |timestamp| {
                events.lock().unwrap().push(timestamp);
            })
        }

        fn create_on_disconnect_callback(&self) -> Box<dyn FnOnce() + Send + Sync> {
            let disconnect_called = Arc::clone(&self.disconnect_called);
            Box::new(move || {
                *disconnect_called.lock().unwrap() = true;
            })
        }

        // fn get_events(&self) -> Vec<u64> {
        //     self.events.lock().unwrap().clone()
        // }

        fn is_disconnected(&self) -> bool {
            *self.disconnect_called.lock().unwrap()
        }
    }

    fn create_streaming_service() -> StreamingServiceImpl<MockUrlProvider> {
        let mut mock_url_provider = MockUrlProvider::new();
        mock_url_provider.expect_make_real_time_url().returning(|| "http://example.com/streaming".to_string());

        StreamingServiceImpl::new(Arc::new(mock_url_provider), Handle::current())
    }

    #[tokio::test]
    async fn test_new_creates_service_with_url_provider() {
        let service = create_streaming_service();

        // Verify the service is created successfully
        assert!(service.disconnect_notify.load().is_none());
    }

    #[tokio::test]
    async fn test_disconnect_sends_disconnect_notify() {
        let service = create_streaming_service();
        let capture = CallbackCapture::new();

        service.connect(capture.create_on_event_callback(), capture.create_on_disconnect_callback());

        assert!(service.disconnect_notify.load().is_some());

        service.disconnect();

        // Give some time for the spawn to start and potentially complete
        tokio::time::sleep(Duration::from_millis(5)).await;

        assert!(capture.is_disconnected());
        assert!(service.disconnect_notify.load().is_none());
    }

    #[tokio::test]
    async fn test_disconnect_without_connection_does_not_panic() {
        let service = create_streaming_service();

        // Disconnect without connecting should not panic
        service.disconnect();
        assert!(service.disconnect_notify.load().is_none());
    }

    #[tokio::test]
    async fn test_multiple_disconnect_calls_are_safe() {
        let service = create_streaming_service();
        let capture = CallbackCapture::new();

        service.connect(capture.create_on_event_callback(), capture.create_on_disconnect_callback());

        // First disconnect should work
        service.disconnect();
        assert!(service.disconnect_notify.load().is_none());

        // Second disconnect should be safe (no panic)
        service.disconnect();
        assert!(service.disconnect_notify.load().is_none());
    }

    #[tokio::test]
    async fn test_is_connected_behaviour() {
        let service = create_streaming_service();

        // Initial is_connected() request
        assert_eq!(service.is_connected(), false);

        // First connect
        let notify = Arc::new(Notify::new());
        let notify_clone = notify.clone();
        let notify_clone2 = notify.clone();
        service.connect(Box::new(move |_ts| {}), Box::new(move || notify_clone.notify_one()));
        assert_eq!(service.is_connected(), true);

        // First disconnect
        service.disconnect();
        notify.notified().await;
        assert_eq!(service.is_connected(), false);

        service.connect(Box::new(move |_ts| {}), Box::new(move || notify_clone2.notify_one()));
        assert_eq!(service.is_connected(), true);

        // Second disconnect
        service.disconnect();
        notify.notified().await;
        assert_eq!(service.is_connected(), false);
    }

    // Integration test
    #[tokio::test]
    async fn test_connect_with_mock_server() {
        // Start mock server
        let mock_server = MockServer::start().await;

        // SSE payload: must end with \n\n to mark event end
        let event1 = "event: configuration-update-event\ndata: {\"siteCode\":\"test\",\"ts\":1234567890}\n\n";
        let event2 = "event: configuration-update-event\ndata: {\"siteCode\":\"test\",\"ts\":10987654321}\n\n";

        // Serve it with correct Content-Type
        Mock::given(method("GET"))
            .and(path("/sse"))
            .respond_with(ResponseTemplate::new(200).set_body_raw(event1.to_owned() + event2, "text/event-stream"))
            .mount(&mock_server)
            .await;

        let mut mock_url_provider: MockUrlProvider = MockUrlProvider::new();
        mock_url_provider.expect_make_real_time_url().returning(move || format!("{}/sse", mock_server.uri()));

        let (tx, mut rx) = mpsc::unbounded_channel::<u64>();

        let service = StreamingServiceImpl::new(Arc::new(mock_url_provider), Handle::current());
        service.connect(
            {
                let tx = tx.clone();
                Box::new(move |timestamp| {
                    let _ = tx.send(timestamp);
                })
            },
            Box::new(|| {}),
        );

        let mut events = Vec::new();
        let collect_task = timeout(Duration::from_secs(1), async {
            while events.len() < 2 {
                if let Some(event) = rx.recv().await {
                    events.push(event);
                }
            }
            events
        });
        let events = collect_task.await.expect("Test timed out waiting for events");

        assert!(!events.is_empty());
        assert_eq!(events[0], 1234567890);
        assert_eq!(events[1], 10987654321);

        service.disconnect();
    }
}
