//! Use [`ditto.presence()`] to gain insight to connected Peers in the Ditto **mesh** via the
//! [`Presence`] API.
//!
//! The [`Presence`] API can be used to inspect the ways in which peers are connected to one
//! another in the Ditto mesh. Using [`.graph()`] or [`.observe()`], you can see an immediate
//! view of peer connections or receive callbacks with peer connection updates, respectively.
//!
//! The key piece of data in the presence API is the [`PresenceGraph`], which contains a
//! description of the "local peer" as well as a list of "remote peers". Remote peers are any
//! peers which can be reached by this local peer in the Ditto mesh, either directly via a
//! transport on this device, or indirectly via multiple hops through other peers.
//!
//! # Example
//!
//! Let's take a look at how to request a [`PresenceGraph`] and how to read through it:
//!
//! ```
//! use dittolive_ditto::prelude::*;
//! # fn example(ditto: &Ditto) -> anyhow::Result<()> {
//!
//! // Get an immediate `PresenceGraph` showing current connections
//! let graph = ditto.presence().graph();
//! let my_key = &graph.local_peer.peer_key_string;
//! let my_connections = &graph.local_peer.connections;
//!
//! // Let's find all peers that are directly connected to me, the local peer
//! let direct_peers = my_connections
//!     .iter()
//!     .map(|connection| {
//!         // Choose the peer in this connection that is not me
//!         if connection.peer_key_string1 == *my_key {
//!             &connection.peer_key_string2
//!         } else {
//!             &connection.peer_key_string1
//!         }
//!     })
//!     .collect::<Vec<_>>();
//! println!("My direct peers are {direct_peers:?}");
//!
//! // Let's look up some details about _all_ the remote peers we can see
//! let remote_peers_summary = graph
//!     .remote_peers
//!     .iter()
//!     .map(|peer| {
//!         // Peer Key => (Name, OS, Connection Count)
//!         (
//!             &peer.peer_key_string,
//!             (&peer.device_name, &peer.os, peer.connections.len()),
//!         )
//!     })
//!     .collect::<std::collections::BTreeMap<_, _>>();
//! println!("A summary of my remote peers looks like this: {remote_peers_summary:#?}");
//! # Ok(())
//! # }
//! ```
//!
//! Please note that obtaining a [`PresenceGraph`] via [`.graph()`] only shows a
//! _snapshot_ of what devices were present when the call was made. In order to watch
//! changes to device presence, we'll need to use [`.observe()`] to register a
//! callback and receive updates.
//!
//! # Example
//!
//! Let's look at how we can use the [`.observe()`] API to register a callback and
//! receive updates whenever the presence of our Ditto mesh changes. We could say the
//! presence has changed if any peers have added or removed connections in the mesh.
//!
//! ```
//! # use std::sync::{Arc, Mutex};
//! use dittolive_ditto::prelude::*;
//! # fn example(ditto: &Ditto) -> anyhow::Result<()> {
//!
//! let mut maybe_prev_graph = Arc::new(Mutex::new(None));
//! let _presence_observer = ditto.presence().observe(move |graph| {
//!     let mut maybe_prev_graph = maybe_prev_graph.lock().unwrap();
//!
//!     let Some(prev_graph) = &*maybe_prev_graph else {
//!         // First presence update! Print what remote peers we see by their keys
//!         let remote_peers = graph
//!             .remote_peers
//!             .iter()
//!             .map(|peer| &peer.peer_key_string)
//!             .collect::<Vec<_>>();
//!         println!("Received first presence update! Remote peers: {remote_peers:?}");
//!         *maybe_prev_graph = Some(graph.clone());
//!         return;
//!     };
//!
//!     // Subsequent presence updates can compare the new graph against the old
//!     let prev_remote_peers = prev_graph
//!         .remote_peers
//!         .iter()
//!         .map(|peer| &peer.peer_key_string)
//!         .collect::<std::collections::HashSet<_>>();
//!     let latest_remote_peers = graph
//!         .remote_peers
//!         .iter()
//!         .map(|peer| &peer.peer_key_string)
//!         .collect::<std::collections::HashSet<_>>();
//!
//!     // Detect if new peers have joined the mesh
//!     let new_peers = latest_remote_peers
//!         .difference(&prev_remote_peers)
//!         .collect::<Vec<_>>();
//!     if !new_peers.is_empty() {
//!         println!("New peers joined the mesh! {new_peers:?}");
//!     }
//!
//!     // Detect if any peers left the mesh
//!     let lost_peers = prev_remote_peers
//!         .difference(&latest_remote_peers)
//!         .collect::<Vec<_>>();
//!     if !lost_peers.is_empty() {
//!         println!("Peers have left the mesh! {lost_peers:?}");
//!     }
//!
//!     *maybe_prev_graph = Some(graph.clone());
//! });
//! # Ok(())
//! # }
//! ```
//!
//! [`ditto.presence()`]: crate::ditto::Ditto::presence
//! [`.graph()`]: Presence::graph
//! [`.observe()`]: Presence::observe

#![doc(alias = "mesh")]
#![warn(missing_docs)]

use_prelude!();

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

use self::observer::{PeersObserverCtx, PresenceObserverCtx, WeakPresenceObserver};
use crate::transport::v2::V2Presence;

mod connection_request_handler;
pub(crate) mod observer;

// These items all "publicly" exist here in `presence`, even if defined elsewhere
pub use self::{
    connection_request_handler::{ConnectionRequest, ConnectionRequestAuthorization},
    observer::PresenceObserver,
};
pub use crate::transport::v3::{Connection, Peer, PresenceGraph, PresenceOs};

/// Convenience type for JSON objects seen in [`peer_metadata`][0].
///
/// [0]: Presence::peer_metadata
pub type JsonObject = ::serde_json::Map<String, ::serde_json::Value>;

/// The entrypoint to the Presence API, obtained with [`ditto.presence()`].
///
/// [See the `presence` module for guide-level docs and examples][0]
///
/// [`ditto.presence()`]: crate::prelude::Ditto::presence
/// [0]: crate::presence
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>,
}

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

// Primary public API
impl Presence {
    /// Return an immediate view of peer connections as a [`PresenceGraph`].
    ///
    /// # Example
    ///
    /// Let's take a look at how to request a [`PresenceGraph`] and how to read through it:
    ///
    /// ```
    /// use dittolive_ditto::prelude::*;
    /// # fn example(ditto: &Ditto) -> anyhow::Result<()> {
    ///
    /// // Get an immediate `PresenceGraph` showing current connections
    /// let graph = ditto.presence().graph();
    /// let my_key = &graph.local_peer.peer_key_string;
    /// let my_connections = &graph.local_peer.connections;
    ///
    /// // Let's find all peers that are directly connected to me, the local peer
    /// let direct_peers = my_connections
    ///     .iter()
    ///     .map(|connection| {
    ///         // Choose the peer in this connection that is not me
    ///         if connection.peer_key_string1 == *my_key {
    ///             &connection.peer_key_string2
    ///         } else {
    ///             &connection.peer_key_string1
    ///         }
    ///     })
    ///     .collect::<Vec<_>>();
    /// println!("My direct peers are {direct_peers:?}");
    ///
    /// // Let's look up some details about _all_ the remote peers we can see
    /// let remote_peers_summary = graph
    ///     .remote_peers
    ///     .iter()
    ///     .map(|peer| {
    ///         // Peer Key => (Name, OS, Connection Count)
    ///         (
    ///             &peer.peer_key_string,
    ///             (&peer.device_name, &peer.os, peer.connections.len()),
    ///         )
    ///     })
    ///     .collect::<std::collections::BTreeMap<_, _>>();
    /// println!("A summary of my remote peers looks like this: {remote_peers_summary:#?}");
    /// # Ok(())
    /// # }
    /// ```
    ///
    /// Please note that obtaining a [`PresenceGraph`] via `.graph()` only shows a
    /// _snapshot_ of what devices were present when the call was made. In order to watch
    /// changes to device presence, we'll need to use [`.observe(...)`] to register a
    /// callback and receive updates.
    ///
    /// [`.observe(...)`]: Self::observe
    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()
    }

    /// Receive [`PresenceGraph`] updates when peer connections change in the Ditto mesh.
    ///
    /// The returned [`PresenceObserver`] must be kept in scope to continue reciving updates.
    ///
    /// # Example
    ///
    /// Let's look at how we can use `.observe()` to register a callback and
    /// receive updates whenever the presence of our Ditto mesh changes. We could say the
    /// presence has changed if any peers have added or removed connections in the mesh.
    ///
    /// ```
    /// # use std::sync::{Arc, Mutex};
    /// use dittolive_ditto::prelude::*;
    /// # fn example(ditto: &Ditto) -> anyhow::Result<()> {
    ///
    /// let mut maybe_prev_graph = Arc::new(Mutex::new(None));
    /// let _presence_observer = ditto.presence().observe(move |graph| {
    ///     let mut maybe_prev_graph = maybe_prev_graph.lock().unwrap();
    ///
    ///     let Some(prev_graph) = &*maybe_prev_graph else {
    ///         // First presence update! Print what remote peers we see by their keys
    ///         let remote_peers = graph
    ///             .remote_peers
    ///             .iter()
    ///             .map(|peer| &peer.peer_key_string)
    ///             .collect::<Vec<_>>();
    ///         println!("Received first presence update! Remote peers: {remote_peers:?}");
    ///         *maybe_prev_graph = Some(graph.clone());
    ///         return;
    ///     };
    ///
    ///     // Subsequent presence updates can compare the new graph against the old
    ///     let prev_remote_peers = prev_graph
    ///         .remote_peers
    ///         .iter()
    ///         .map(|peer| &peer.peer_key_string)
    ///         .collect::<std::collections::HashSet<_>>();
    ///     let latest_remote_peers = graph
    ///         .remote_peers
    ///         .iter()
    ///         .map(|peer| &peer.peer_key_string)
    ///         .collect::<std::collections::HashSet<_>>();
    ///
    ///     // Detect if new peers have joined the mesh
    ///     let new_peers = latest_remote_peers
    ///         .difference(&prev_remote_peers)
    ///         .collect::<Vec<_>>();
    ///     if !new_peers.is_empty() {
    ///         println!("New peers joined the mesh! {new_peers:?}");
    ///     }
    ///
    ///     // Detect if any peers left the mesh
    ///     let lost_peers = prev_remote_peers
    ///         .difference(&latest_remote_peers)
    ///         .collect::<Vec<_>>();
    ///     if !lost_peers.is_empty() {
    ///         println!("Peers have left the mesh! {lost_peers:?}");
    ///     }
    ///
    ///     *maybe_prev_graph = Some(graph.clone());
    /// });
    /// # Ok(())
    /// # }
    /// ```
    ///
    /// If instead you just need to check once to learn about which peers are connected
    /// _right now_, try using the [`.graph()`] method instead.
    ///
    /// [`.graph()`]: Self::graph
    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)
    }
}

// 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.
    ///
    /// # 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).into_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.into_ffi(&raw);
                        }
                    };
                    MINI_RUNTIME
                        .get_or_init(new_mini_runtime)
                        .send(Box::pin(task_to_spawn_detached))
                        .expect("dedicated async runtime to be alive");
                })
                .into(),
            ),
        )
    }
}

// Private methods
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.
    #[allow(deprecated)]
    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)
    }

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

/// Defines a simplified connection type between peers for reporting presence
/// info.
///
/// These connections indicate P2P connections _only_. A connection to the Big Peer
/// is recorded by a simple boolean flag on the [`Peer`] type.
#[non_exhaustive]
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum ConnectionType {
    /// Connected to the remote peer via Bluetooth Low-Energy (BLE).
    Bluetooth,

    /// Connected to the remote peer via LAN, e.g. home or office WiFi.
    AccessPoint,

    /// Direct WiFi between peers, using AWDL (Apple) or WiFi Aware (Android).
    P2PWiFi,

    /// Connected to the remote peer via WebSocket, a bidirectional stream over http(s).
    WebSocket,

    /// Unknown connection type, remote peer is likely a higher version.
    #[doc(hidden)]
    Unknown,
}

impl ConnectionType {
    pub(crate) fn from_ffi(ffi: ::ffi_sdk::ConnectionType) -> Self {
        match ffi {
            ::ffi_sdk::ConnectionType::Bluetooth => Self::Bluetooth,
            ::ffi_sdk::ConnectionType::AccessPoint => Self::AccessPoint,
            ::ffi_sdk::ConnectionType::P2PWiFi => Self::P2PWiFi,
            ::ffi_sdk::ConnectionType::WebSocket => Self::WebSocket,
            #[allow(unreachable_patterns)]
            _ => {
                tracing::debug!(connection_type = ?ffi, "got unknown `ConnectionType`");
                Self::Unknown
            }
        }
    }
}

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)
        }
    }
}