use_prelude!();

use ::std::{
    future::Future,
    sync::{Mutex, OnceLock, Weak},
};

pub use super::connection_request_handler::{ConnectionRequest, ConnectionRequestAuthorization};
use super::{
    presence_observer::{
        PeersObserver, PeersObserverCtx, PresenceObserver, PresenceObserverCtx,
        WeakPresenceObserver,
    },
    v2::V2Presence,
    v3::PresenceGraph,
};

pub type JsonObject = ::serde_json::Map<String, ::serde_json::Value>;

/// # `Presence` to visualize the mesh.
///
/// Even if Ditto works as a standalone, other peers are needed to exploit it
/// at its maximum extend : syncing data and merging it.
///
/// The [`Presence`] struct allows you to get and monitor known peers.
/// The peers are regrouped inside a [`PresenceGraph`].
///
/// You can get [`Presence`] using the [`Ditto::presence`] method.
/// To have an immediate graph of peers, use [`Presence::graph`]:
/// ```
/// # use dittolive_ditto::prelude::*;
/// # fn test_presence(ditto: Ditto) {
/// let presence_graph = ditto.presence().graph();
/// # }
/// ```
/// To monitor peers, use [`Presence::observe`] to call a callback each time peers are updated:
/// ```
/// # use dittolive_ditto::prelude::*;
/// # fn test_presence(ditto: Ditto) {
/// let handle = ditto.presence().observe(|presence_graph| {
///     // Do something with peers
/// });
/// # }
/// ```
pub struct Presence {
    ditto: Arc<ffi_sdk::BoxedDitto>,
    observers_v2: Mutex<Vec<Weak<PeersObserverCtx>>>,
    observers_v3: Mutex<Vec<WeakPresenceObserver>>,
    registered: Mutex<bool>,
    peer_metadata_cache: Mutex<PeerMetadataCache>,
}

impl Presence {
    pub(crate) fn new(ditto: Arc<ffi_sdk::BoxedDitto>) -> Self {
        Self {
            ditto,
            observers_v2: <_>::default(),
            observers_v3: <_>::default(),
            registered: false.into(),
            peer_metadata_cache: <_>::default(),
        }
    }

    /// `ctx` is, conceptually-speaking a `&'short_lived Weak< DiskUsageObserverCtx<F> >`.
    ///
    /// This scoped/callback API embodies that.
    #[track_caller]
    unsafe fn borrowing_from_ctx(ctx: *const c_void, yielding: impl FnOnce(&Weak<Presence>)) {
        let weak_ctx = ::core::mem::ManuallyDrop::new(Weak::from_raw(ctx.cast::<Presence>()));
        yielding(&weak_ctx)
    }

    unsafe extern "C" fn retain(ctx: *mut c_void) {
        Self::borrowing_from_ctx(ctx, |weak_ctx| _ = Weak::into_raw(weak_ctx.clone()))
    }

    unsafe extern "C" fn release(ctx: *mut c_void) {
        drop(Weak::<Presence>::from_raw(ctx.cast()))
    }

    /// C wrapper for calling the real callback on Presence
    unsafe extern "C" fn on_event_v2(ctx: *mut c_void, json: char_p::Ref<'_>) {
        Self::borrowing_from_ctx(ctx, |weak_ctx| {
            if let Some(strong_ctx) = weak_ctx.upgrade() {
                let presence_json_str = json.to_str();
                strong_ctx.on_presence_v2(presence_json_str);
            }
        })
    }

    /// C wrapper for calling the real callback on Presence
    unsafe extern "C" fn on_event_v3(ctx: *mut c_void, json: char_p::Ref<'_>) {
        Self::borrowing_from_ctx(ctx, |weak_ctx| {
            if let Some(strong_ctx) = weak_ctx.upgrade() {
                let presence_json_str = json.to_str();
                strong_ctx.on_presence_v3(presence_json_str);
            }
        })
    }

    fn on_presence_v2(&self, json_str: &str) {
        // v2 presence
        let mut observers_v2 = self.observers_v2.lock().unwrap();
        if observers_v2.is_empty() {
            return;
        }
        if let Ok(presence_v2) = serde_json::from_str::<V2Presence>(json_str) {
            observers_v2.retain(|weak_observer| {
                if let Some(observer) = weak_observer.upgrade() {
                    (observer.on_presence)(presence_v2.clone());
                    true
                } else {
                    false
                }
            })
        }
    }

    fn on_presence_v3(&self, json_str: &str) {
        // v3 presence
        let mut observers_v3 = self.observers_v3.lock().unwrap();
        if observers_v3.is_empty() {
            return;
        }
        if let Ok(presence_graph) = serde_json::from_str(json_str) {
            observers_v3.retain(|weak_observer| {
                if let Some(observer) = weak_observer.upgrade() {
                    (observer.on_presence)(&presence_graph);
                    true
                } else {
                    false
                }
            })
        }
    }

    fn subscribe(self: &Arc<Self>) {
        unsafe {
            let weak_self = Arc::downgrade(self);
            ffi_sdk::ditto_register_presence_v2_callback(
                &self.ditto,
                weak_self.as_ptr() as *mut _,
                Some(Presence::retain),
                Some(Presence::release),
                Some(<unsafe extern "C" fn(_, char_p::Ref<'_>)>::into(
                    Presence::on_event_v2,
                )),
            );
            ffi_sdk::ditto_register_presence_callback_v3(
                &self.ditto,
                weak_self.as_ptr() as *mut _,
                Some(Presence::retain),
                Some(Presence::release),
                Some(<unsafe extern "C" fn(_, char_p::Ref<'_>)>::into(
                    Presence::on_event_v3,
                )),
            );
            // Guards against mistakes such as using `.into_raw()` rather than `.as_ptr()`.
            drop(weak_self);
        }

        *self.registered.lock().unwrap() = true;
    }

    /// Add a peer observer and return a PresenceObserver to be able to drop it when desired.
    pub(crate) fn add_observer(
        self: &Arc<Self>,
        handler: impl Fn(V2Presence) + Send + Sync + 'static,
    ) -> PeersObserver {
        let context = PeersObserverCtx::new(Box::new(handler));
        let arc_context = Arc::new(context);

        let weak_context = Arc::downgrade(&arc_context);

        if !*self.registered.lock().unwrap() {
            self.subscribe();
        }

        {
            // New scope to minimize the time we hold the lock.
            let mut observers = self.observers_v2.lock().unwrap();
            observers.push(weak_context);
        }

        ::std::thread::spawn({
            let this = self.retain();
            move || {
                // Initial event.
                let str_box = ffi_sdk::ditto_presence_v2(&this.ditto);
                this.on_presence_v2(str_box.to_str())
            }
        });
        PeersObserver::new(arc_context)
    }

    /// Allow to call a callback each time there is a change in known peers.
    /// The returned [`PresenceObserver`] must be kept in scope to keep receiving updates.
    pub fn observe(
        self: &Arc<Self>,
        callback: impl Fn(&PresenceGraph) + Send + Sync + 'static,
    ) -> PresenceObserver {
        if !*self.registered.lock().unwrap() {
            self.subscribe();
        }

        let context = PresenceObserverCtx::new(Box::new(callback));
        let arc_context = Arc::new(context);
        let weak_context = Arc::downgrade(&arc_context);

        {
            // New scope to minimize the time we hold the lock.
            let mut observers = self.observers_v3.lock().unwrap();
            observers.push(weak_context);
        }

        ::std::thread::spawn({
            let this = self.retain();
            move || {
                // Initial event.
                let str_box = ffi_sdk::ditto_presence_v3(&this.ditto);
                this.on_presence_v3(str_box.to_str())
            }
        });

        PresenceObserver::new(arc_context)
    }

    #[doc(hidden)]
    #[deprecated(note = "Use `.graph()` instead")]
    pub fn exec(&self) -> PresenceGraph {
        self.graph()
    }

    /// Return an immediate representation of known peers
    pub fn graph(&self) -> PresenceGraph {
        let raw_string = ffi_sdk::ditto_presence_v3(&self.ditto);
        let json_str = raw_string.to_str();
        ::serde_json::from_str(json_str).unwrap()
    }
}

#[derive(Debug, Default)]
pub(crate) struct PeerMetadataCache {
    json_str: String,
    value: Arc<JsonObject>,
}

// Peer Metadata & on-connecting API.
impl Presence {
    /// Set a dictionary of arbitrary data about this device to be shared with peers in the mesh.
    ///
    /// This data is gossiped in the presence collection across the mesh. This can be useful to
    /// include extra metadata like app versions,capabilities, etc., to help peers decide who to
    /// interact with.
    ///
    /// This peer information is persisted in the SDK, and thus needn't be set at every start
    /// of Ditto.
    ///
    /// # Security (and caveats)
    /// This peer info will be signed by your peer key to prevent forgery of this info by other
    /// peers.
    ///
    /// However, for compatibility, there is no attestation of the *lack* of peer info -- that
    /// is, participants in the mesh could maliciously remove peer info. If this is a concern
    /// for your application, a workaround for this is to have your application require that
    /// peers have a signed peer info dictionary present.
    ///
    /// Similarly, as there is no monotonic version counter or timestamp/expiration of the
    /// signed peer info, replay attacks (replacing the current info with previously, possibly
    /// outdated, signed info) are possible without counter-measures. If this is a concern
    /// for your application, you might consider including a counter or creation timestamp
    /// to prevent replays, depending on your use-case.
    ///
    /// When a peer is only connected via WebSocket, peer metadata is not available.
    ///
    /// # Performance caveats
    /// Because this information is included in the presence data that is gossiped among peers,
    /// the size of this peer info and the frequency it is updated can *drastically* affect
    /// performance if it is too large.
    ///
    /// # Errors
    /// Because of the performance implications, the serialized info dictionary is currently
    /// limited to 4KiB.
    ///
    /// # Examples
    /// ```
    /// # use std::collections::HashMap;
    /// use dittolive_ditto::prelude::*;
    /// use serde_json::json;
    /// # |ditto: Ditto| -> Result<(), Box<dyn std::error::Error>> {
    /// ditto.presence().set_peer_metadata(&json!({
    ///     "app_version": "1.0.0",
    /// }))?;
    /// # Ok(())
    /// # };
    /// ```
    pub fn set_peer_metadata(&self, peer_metadata: &impl Serialize) -> Result<(), DittoError> {
        let payload = ::serde_json::to_string(peer_metadata)?;
        self.set_peer_metadata_json_str(&payload)
    }

    /// Set arbitrary metadata formatted as JSON to be associated with the
    /// current peer.
    ///
    /// The metadata must not exceed 4 KB in size. Expects JSON.
    ///
    /// - See also: [`Self::set_peer_metadata()`] for details on usage of metadata.
    pub fn set_peer_metadata_json_str(&self, json: &str) -> Result<(), DittoError> {
        ffi_sdk::dittoffi_presence_try_set_peer_metadata_json(&self.ditto, json.as_bytes().into())
            .into_rust_result()?;
        Ok(())
    }

    /// Metadata associated with the current peer as JSON-encoded data.
    ///
    /// Other peers in the same mesh can access this user-provided dictionary of
    /// metadata via the presence graph at [`Self::graph()`] and when
    /// evaluating connection requests using
    /// [`Self::set_connection_request_handler()`]. Use [`Self::set_peer_metadata()`]
    /// or [`Self::set_peer_metadata_json_str()`] to set this value.
    pub fn peer_metadata_json_str(&self) -> String {
        String::from_utf8(From::<Box<[u8]>>::from(
            ffi_sdk::dittoffi_presence_peer_metadata_json(&self.ditto).into(),
        ))
        .expect("UTF-8")
    }

    /// [`DeserializeOwned`] convenience around [`Self::peer_metadata_json_str()`].
    pub fn peer_metadata_serde<T: DeserializeOwned>(&self) -> Result<T> {
        let value = ::serde_json::from_str(&self.peer_metadata_json_str())?;
        Ok(value)
    }

    /// Metadata associated with the current peer.
    ///
    /// Other peers in the same mesh can access this user-provided dictionary of
    /// metadata via the presence graph at [`Self::graph()`] and when
    /// evaluating connection requests using
    /// [`Self::set_connection_request_handler()`]. Use [`Self::set_peer_metadata()`]
    /// or [`Self::set_peer_metadata_json_str()`] to set this value.
    ///
    /// This is a convenience property that wraps [`Self::peer_metadata_json_str()`].
    pub fn peer_metadata(&self) -> Arc<JsonObject> {
        let json_str = self.peer_metadata_json_str();
        let mut cache = self
            .peer_metadata_cache
            .lock()
            .unwrap_or_else(|it| it.into_inner());
        if json_str != cache.json_str {
            *cache = PeerMetadataCache {
                value: Arc::new(
                    ::serde_json::from_str(&json_str).expect("incorrect json from `dittoffi`"),
                ),
                json_str,
            };
        }
        cache.value.retain()
    }

    /// Set this handler to control which peers in a Ditto mesh can connect to
    /// the current peer.
    ///
    /// Each peer in a Ditto mesh will attempt to connect to other peers that it
    /// can reach. By default, the mesh will try and establish connections that
    /// optimize for the best overall connectivity between peers. However, you
    /// can set this handler to assert some control over which peers you connect
    /// to.
    ///
    /// If set, this handler is called for every incoming connection request
    /// from a remote peer and is passed the other peer's `peer_key`,
    /// `peer_metadata`, and `identity_service_metadata`. The handler can then
    /// accept or reject the request by returning an according
    /// [`ConnectionRequestAuthorization`] value. When the connection
    /// request is rejected, the remote peer may retry the connection request
    /// after a short delay.
    ///
    /// Connection request handlers must reliably respond to requests within a
    /// short time: **if a handler takes too long to return, the connection
    /// request will fall back to being denied**. The response –currently— times
    /// out after 10 seconds, but this exact value may be subject to change in
    /// future releases.
    ///
    /// - Note: the handler is called from a different thread ("background hook").
    /// - See also: [`Self::peer_metadata()`]
    ///
    /// ## Example
    ///
    /// ```
    /// # use dittolive_ditto::prelude::*;
    /// # |ditto: Ditto| -> Result<(), Box<dyn std::error::Error>> {
    /// /// Let's imagine the app we are maintaining has a bug in `1.2.3`:
    /// const BUGGY_VERSION: &str = "1.2.3";
    ///
    /// // We avoid problems in updated versions of our app with these ones by
    /// // rejecting connections to them, like so:
    /// ditto
    ///     .presence()
    ///     .set_connection_request_handler(|connection_request: ConnectionRequest| {
    ///         match connection_request
    ///             .peer_metadata()
    ///             .get("app_version")
    ///             .and_then(|it| it.as_str())
    ///         {
    ///             // Reject peers reporting a known buggy version or reporting no
    ///             // version at all.
    ///             Some(BUGGY_VERSION) | None => return ConnectionRequestAuthorization::Deny,
    ///             Some(_non_buggy_version) => { /* no reason to reject here */ }
    ///         }
    ///         // Potentially other checks/reasons to reject…
    ///
    ///         // Eventually:
    ///         ConnectionRequestAuthorization::Allow
    ///     });
    ///
    /// // You can also unset the `connection_request_handler` by setting it to `None`.
    /// // This uses the default handler, which accepts all requests.
    /// ditto.presence().set_connection_request_handler(None);
    /// # Ok(())
    /// # };
    /// ```
    pub fn set_connection_request_handler<
        F: sealed::IntoOption<
            impl 'static + Send + Sync + Fn(ConnectionRequest) -> ConnectionRequestAuthorization,
        >,
    >(
        &self,
        handler_or_none: F,
    ) {
        let ffi_callback = F::into_option(handler_or_none).map(|callback| {
            Arc::new(move |raw: repr_c::Box<ffi_sdk::FfiConnectionRequest>| {
                let connection_request = ConnectionRequest::new(raw);
                let raw = connection_request.raw();
                callback(connection_request).to_ffi(&raw);
            })
            .into()
        });
        ffi_sdk::dittoffi_presence_set_connection_request_handler(&self.ditto, ffi_callback)
    }

    /// Convenience around [`Self::set_connection_request_handler()`] that allows the callback to be
    /// `async`.
    ///
    /// Not responding in time will lead to a handshake timeout, effectively rejecting the peer.
    pub fn set_connection_request_handler_async<ConnectionRequestAuthorizationFut>(
        &self,
        async_callback: impl 'static
            + Send
            + Sync
            + Fn(ConnectionRequest) -> ConnectionRequestAuthorizationFut,
    ) where
        ConnectionRequestAuthorizationFut:
            'static + Send + Future<Output = ConnectionRequestAuthorization>,
    {
        use ::tokio::{runtime, sync::mpsc};

        let new_mini_runtime = || {
            // Using an unbounded channel to implement `task::spawn()`, like tokio does.
            let (tx, mut rx) = mpsc::unbounded_channel();
            let runtime = runtime::Builder::new_current_thread()
                .enable_all()
                .build()
                .expect("failed to build `async` runtime");
            ::std::thread::spawn(move || {
                runtime.block_on(async move {
                    while let Some(task) = rx.recv().await {
                        () = task.await;
                    }
                })
            });
            tx
        };
        ffi_sdk::dittoffi_presence_set_connection_request_handler(
            &self.ditto,
            Some(
                Arc::new(move |raw| {
                    static MINI_RUNTIME: OnceLock<
                        mpsc::UnboundedSender<Pin<Box<dyn Send + Future<Output = ()>>>>,
                    > = OnceLock::new();

                    let connection_request = ConnectionRequest::new(raw);
                    let raw = connection_request.raw();
                    let task_to_spawn_detached = {
                        let async_callback_connection_request = async_callback(connection_request);
                        async move {
                            async_callback_connection_request.await.to_ffi(&raw);
                        }
                    };
                    MINI_RUNTIME
                        .get_or_init(new_mini_runtime)
                        .send(Box::pin(task_to_spawn_detached))
                        .ok()
                        .expect("dedicated async runtime to be alive");
                })
                .into(),
            ),
        )
    }
}

mod sealed {
    use super::*;

    /// A trait implemented for both `None` and instances of
    /// `'static + Send + Sync + Fn(ConnectionRequest) -> ConnectionRequestAuthorization`.
    ///
    /// Used to keep perfect API parity with that of other SDKs.
    pub trait IntoOption<
        F: 'static + Send + Sync + Fn(ConnectionRequest) -> ConnectionRequestAuthorization,
    >
    {
        fn into_option(_: Self) -> Option<F>;
    }

    impl IntoOption<fn(ConnectionRequest) -> ConnectionRequestAuthorization>
        for Option<::never_say_never::Never>
    {
        fn into_option(_: Self) -> Option<fn(ConnectionRequest) -> ConnectionRequestAuthorization> {
            None
        }
    }

    impl<F> IntoOption<F> for F
    where
        F: 'static + Send + Sync + Fn(ConnectionRequest) -> ConnectionRequestAuthorization,
    {
        fn into_option(f: Self) -> Option<Self> {
            Some(f)
        }
    }
}