//! # Entry point for the DittoSDK
//!
//! `Ditto` is a cross-platform peer-to-peer database that allows apps to sync with and even without
//! internet connectivity.
//!
//! To manage your local data and connections to other peers in the mesh, `Ditto` gives you access
//! to:
//! * [`Store`], the entry point to the database.
//! * [`TransportConfig`] to change the transport layers in use.
//! * [`Presence`] to monitor peers in the mesh.
//! * [`DiskUsage`] to monitor local Ditto disk usage.

use_prelude!();

pub mod builder;

use std::{
    env,
    sync::{RwLock, Weak},
};

use crossbeam_utils::atomic::AtomicCell;
/// The log levels that the Ditto SDK supports.
pub use ffi_sdk::CLogLevel as LogLevel;
use ffi_sdk::{BoxedDitto, FsComponent};
use uuid::Uuid;

use self::builder::DittoBuilder;
use crate::{
    auth::{authenticator::DittoAuthenticator, validity_listener::ValidityListener},
    disk_usage::DiskUsage,
    error::{DittoError, ErrorKind, LicenseError},
    identity::SharedIdentity,
    small_peer_info::SmallPeerInfo,
    transport::{
        on_connecting::{ConnectingPeer, OnConnecting},
        presence::Presence,
        presence_observer::PeersObserver,
        TransportConfig, TransportSync,
    },
    utils::prelude::*,
};

/// This strong shared reference to Ditto should only be shared within the Ditto struct.
pub(crate) type DittoHandleWrapper = Arc<BoxedDitto>;
/// This weak reference to Ditto is intended for objects that wouldn't be freed by Ditto::drop.
pub(crate) type WeakDittoHandleWrapper = std::sync::Weak<BoxedDitto>;

#[extension(pub(crate) trait TryUpgrade)]
impl WeakDittoHandleWrapper {
    fn try_upgrade(&self) -> Result<DittoHandleWrapper, ErrorKind> {
        self.upgrade().ok_or(ErrorKind::ReleasedDittoInstance)
    }
}

/// The entry point for accessing Ditto-related functionality.
///
/// This struct is generally a handle and interface to ditto-functionality which operates in
/// background threads.
pub struct Ditto {
    pub(crate) fields: Arc<DittoFields>,
    /// Always `true` except in `Auth::logout()`.
    is_shut_down_able: bool,
}

impl std::ops::Deref for Ditto {
    type Target = DittoFields;

    #[inline]
    fn deref(&'_ self) -> &'_ DittoFields {
        &*self.fields
    }
}

impl Ditto {
    pub(crate) fn upgrade(weak: &Weak<DittoFields>) -> Result<Ditto> {
        let fields = weak.upgrade().ok_or(ErrorKind::ReleasedDittoInstance)?;
        Ok(Ditto::new_temp(fields))
    }
}

/// Inner fields for Ditto
#[doc(hidden)]
// TODO(pub_check)
pub struct DittoFields {
    // FIXME: (Ham & Daniel) - ideally we'd use this in the same way as we do
    // with the `fields` on `Ditto` (only ever extracting weak references)
    pub(crate) ditto: DittoHandleWrapper,
    ditto_root: Arc<dyn DittoRoot>, // Arc since Identity will need a copy
    identity: SharedIdentity,
    auth: Option<DittoAuthenticator>,
    #[allow(dead_code)]
    validity_listener: Option<Arc<ValidityListener>>, // may not need to hang on to this
    pub(crate) store: Store,
    pub(crate) sync: crate::sync::Sync,
    activated: AtomicCell<bool>,
    site_id: SiteId,
    transports: Arc<RwLock<TransportSync>>,
    presence: Arc<Presence>,
    on_connecting: Arc<OnConnecting>,
    disk_usage: DiskUsage,
    small_peer_info: SmallPeerInfo,
}

// We use this pattern to ensure `self` is not used after `ManuallyDrop::take()`-ing its fields.
impl Drop for Ditto {
    fn drop(&mut self) {
        if self.is_shut_down_able {
            // stop all transports
            self.stop_sync();
            // Here, ditto is implicitly dropped using ditto_free if there is no strong reference to
            // it anymore. We need to make sure that `ditto_shutdown` is called before
            // `ditto_free` gets called though, as this will perform all the necessary
            // pre-drop actions such as stopping TCP servers, etc.
            ffi_sdk::ditto_shutdown(&self.ditto);
        }
    }
}

// Public interface for modifying Transport configuration.
impl Ditto {
    /// Clean shutdown of the Ditto instance
    pub fn close(self) {
        // take ownership of Ditto in order to drop it
    }

    /// Start syncing on all configured transports
    pub fn start_sync(&self) -> Result<(), DittoError> {
        // the License must be active or else sync will immediately fail
        if self.activated.load().not() && self.identity.requires_offline_only_license_token() {
            return Err(ErrorKind::NotActivated.into());
        }

        // We need a write lock because this will internally modify the transports by activating
        // those configured but not currently active
        match self.transports.write() {
            Ok(mut transports) => {
                transports.start_sync();
                Ok(())
            }
            Err(e) => {
                let rust_error = format!("{:?}", e); // needed as PoisonedLockError is not Send
                Err(DittoError::new(ErrorKind::Internal, rust_error))
            }
        }
    }

    /// Stop syncing on all transports.
    pub fn stop_sync(&self) {
        if let Ok(mut transports) = self.transports.write() {
            transports.stop_sync()
        }
    }

    /// Set a new [`TransportConfig`] and begin syncing over these
    /// transports. Any change to start or stop a specific transport should proceed via providing a
    /// modified configuration to this method.
    pub fn set_transport_config(&self, config: TransportConfig) {
        if let Ok(mut transports) = self.transports.write() {
            transports.set_transport_config(config);
        }
    }

    /// Returns a snapshot of the currently configured transports.
    pub fn current_transport_config(&self) -> Result<TransportConfig, DittoError> {
        match self.transports.read() {
            Ok(t) => Ok(t.current_config().clone()),
            Err(e) => {
                let msg = format!("transport config cannot be read {:?}", e);
                Err(DittoError::new(ErrorKind::Internal, msg))
            }
        }
    }

    /// Utils function for tests.
    #[cfg(test)]
    pub fn effective_transport_config(&self) -> Result<TransportConfig, DittoError> {
        match self.transports.read() {
            Ok(t) => Ok(t.effective_config().clone()),
            Err(e) => {
                let msg = format!("transport config cannot be read {:?}", e);
                Err(DittoError::new(ErrorKind::Internal, msg))
            }
        }
    }
}

impl Ditto {
    /// Return the version of the SDK.
    pub fn with_sdk_version<R>(ret: impl FnOnce(&'_ str) -> R) -> R {
        ret(ffi_sdk::ditto_get_sdk_version().to_str())
    }
}

impl Ditto {
    /// Enable of disable logging.
    pub fn set_logging_enabled(enabled: bool) {
        ffi_sdk::ditto_logger_enabled(enabled)
    }

    /// Return true if logging is enabled.
    pub fn get_logging_enabled() -> bool {
        ffi_sdk::ditto_logger_enabled_get()
    }

    /// Represent whether or not emojis should be used as the log level indicator in the logs.
    pub fn get_emoji_log_level_headings_enabled() -> bool {
        ffi_sdk::ditto_logger_emoji_headings_enabled_get()
    }

    /// Set whether or not emojis should be used as the log level indicator in the logs.
    pub fn set_emoji_log_level_headings_enabled(enabled: bool) {
        ffi_sdk::ditto_logger_emoji_headings_enabled(enabled);
    }

    /// Get the current minimum log level.
    pub fn get_minimum_log_level() -> LogLevel {
        ffi_sdk::ditto_logger_minimum_log_level_get()
    }

    /// Set the current minimum log level.
    pub fn set_minimum_log_level(log_level: LogLevel) {
        ffi_sdk::ditto_logger_minimum_log_level(log_level);
    }
}

impl Ditto {
    /// Activate an offline [`Ditto`] instance by setting a license token.
    ///
    /// You cannot initiate sync on an offline
    /// ([`OfflinePlayground`], [`Manual`], or [`SharedKey`])
    /// [`Ditto`] instance before you have activated it.
    pub fn set_offline_only_license_token(&self, license_token: &str) -> Result<(), DittoError> {
        if self.identity.requires_offline_only_license_token() {
            use ::safer_ffi::prelude::{AsOut, ManuallyDropMut};
            use ffi_sdk::LicenseVerificationResult;
            let c_license: char_p::Box = char_p::new(license_token);

            let mut err_msg = None;
            let out_err_msg = err_msg.manually_drop_mut().as_out();
            let res = ffi_sdk::ditto_verify_license(c_license.as_ref(), Some(out_err_msg));

            if res == LicenseVerificationResult::LicenseOk {
                self.activated.store(true);
                return Ok(());
            }

            self.activated.store(false);
            let err_msg = err_msg.unwrap();

            ::log::error!("{}", err_msg);

            match res {
                LicenseVerificationResult::LicenseExpired => {
                    Err(DittoError::license(LicenseError::LicenseTokenExpired {
                        message: err_msg.as_ref().to_string(),
                    }))
                }
                LicenseVerificationResult::VerificationFailed => Err(DittoError::license(
                    LicenseError::LicenseTokenVerificationFailed {
                        message: err_msg.as_ref().to_string(),
                    },
                )),
                LicenseVerificationResult::UnsupportedFutureVersion => Err(DittoError::license(
                    LicenseError::LicenseTokenUnsupportedFutureVersion {
                        message: err_msg.as_ref().to_string(),
                    },
                )),
                _ => panic!("Unexpected license verification result {:?}", res),
            }
        } else {
            Err(DittoError::new(
                ErrorKind::Internal,
                "Offline license tokens should only be used for Manual, SharedKey or \
                 OfflinePlayground identities",
            ))
        }
    }

    /// Look for a license token from a given environment variable.
    pub fn set_license_from_env(&self, var_name: &str) -> Result<(), DittoError> {
        match env::var(var_name) {
            Ok(token) => self.set_offline_only_license_token(&token),
            Err(env::VarError::NotPresent) => {
                let msg = format!("No license token found for env var {}", &var_name);
                Err(DittoError::from_str(ErrorKind::Config, msg))
            }
            Err(e) => Err(DittoError::new(ErrorKind::Config, e)),
        }
    }

    /// 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 dittolive_ditto::prelude::*;
    /// # use std::collections::HashMap;
    /// # |ditto: Ditto| -> Result<(), Box<dyn std::error::Error>> {
    /// let mut peer_info = HashMap::new();
    /// peer_info.insert("app_version".into(), "1.0.0".to_string());
    /// ditto.set_peer_signed_info(peer_info)?;
    /// # Ok(())
    /// # };
    /// ```
    pub fn set_peer_signed_info<S: Serialize>(
        &self,
        peer_info: HashMap<String, S>,
    ) -> Result<(), DittoError> {
        let payload = serde_cbor::to_vec(&peer_info)?;
        let status_code =
            ffi_sdk::ditto_auth_client_set_peer_signed_info(&self.ditto, payload.as_slice().into());
        if status_code != ffi_sdk::DittoErrorCode::Ok {
            Err(DittoError::from_ffi(ErrorKind::InvalidInput))
        } else {
            Ok(())
        }
    }

    /// Add a callback that can decide based on the known peer information at the time whether Ditto
    /// should complete the handshake or reject the connection.
    ///
    /// This is an asynchronous API, so your callback will receive information about the connecting
    /// peer, and you must then call the provided callback function with your decision.
    ///
    /// # Decisions
    /// Responding with `true` will allow the peer to connect.
    /// Responding with `false` will reject the peer.
    /// Not responding will lead to a handshake timeout, effectively rejecting the peer.
    ///
    /// # Examples
    /// ```
    /// # use dittolive_ditto::prelude::*;
    /// # |ditto: Ditto| -> Result<(), Box<dyn std::error::Error>> {
    ///
    /// ditto.register_on_connecting_callback(|peer, respond| {
    ///     match peer.peer_info.unwrap_or_default().get("app_version") {
    ///         // Reject peers reporting a known buggy version.
    ///         Some(InfoValue::Text(v)) if v == "1.2.3" => respond(false),
    ///         _ => respond(true),
    ///     }
    /// });
    /// # Ok(())
    /// # };
    /// ```
    pub fn register_on_connecting_callback(
        &self,
        callback: impl Fn(ConnectingPeer, Box<dyn FnOnce(bool)>) + Send + Sync + 'static,
    ) {
        self.on_connecting.set_on_connecting(callback);
    }

    /// Returns a reference to the underlying local data store.
    pub fn store(&self) -> &Store {
        &self.store
    }

    /// Returns a reference to the synchronization controller.
    pub fn sync(&self) -> &crate::sync::Sync {
        &self.sync
    }

    /// Return a reference to the [`SmallPeerInfo`] object.
    pub fn small_peer_info(&self) -> &SmallPeerInfo {
        &self.small_peer_info
    }

    /// Returns the site ID that the instance of Ditto is using as part of its identity.
    pub fn site_id(&self) -> u64 {
        self.site_id
    }

    /// Returns the Ditto persistence directory path.
    pub fn persistence_directory(&self) -> &Path {
        self.ditto_root.root_path()
    }

    /// Returns the application Id being used by this [`Ditto`] instance.
    pub fn application_id(&self) -> AppId {
        AppId(ffi_sdk::ditto_auth_client_get_app_id(&self.ditto).into_string())
    }

    /// Set a custom identifier for the current device.
    ///
    /// When using [`observe_peers`](Ditto::observe_peers), each remote peer is represented by a
    /// short UTF-8 “device name”. By default this will be a truncated version of the device’s
    /// hostname. It does not need to be unique among peers. Configure the device name before
    /// calling [`start_sync`](Ditto::start_sync). If it is too long it will be truncated.
    pub fn set_device_name(&self, name: &str) {
        let c_device_name: char_p::Box = char_p::new(name.to_owned());
        ffi_sdk::ditto_set_device_name(&self.ditto, c_device_name.as_ref());
    }

    /// Request bulk status information about the transports. This is mostly intended for
    /// statistical or debugging purposes.
    #[doc(hidden)] // Not implemented
    pub fn transport_diagnostics(&self) -> TransportDiagnostics {
        todo!();
    }

    /// Request information about Ditto peers in range of this device.
    ///
    /// This method returns an observer which should be held as long as updates are required. A
    /// newly registered observer will have a peers update delivered to it immediately. Then it will
    /// be invoked repeatedly when Ditto devices come and go, or the active connections to them
    /// change.
    #[deprecated(note = "Use `presence().observe()` instead")]
    pub fn observe_peers<H>(&self, handler: H) -> PeersObserver
    where
        H: Fn(crate::transport::v2::V2Presence) + Send + Sync + 'static,
    {
        self.presence.add_observer(handler)
    }

    /// Return a handle to [`Presence`] to monitor the peers' activity in
    /// the Ditto mesh. It can be used to retrieve an immediate representation of known peers :
    /// ```
    /// # use dittolive_ditto::prelude::Ditto;
    /// # fn call_presence(ditto: Ditto) {
    /// let presence_graph = ditto.presence().exec();
    /// # }
    /// ```
    /// Or to bind a callback to the changes :
    /// ```
    /// # use dittolive_ditto::prelude::Ditto;
    /// # fn call_presence(ditto:Ditto) {
    /// let handle = ditto.presence().observe(|graph| {
    ///     // do something with the graph
    /// });
    /// // The handle must be kept to keep receiving updates on the other peers.
    /// // To stop receiving update, drop the handle.
    /// # }
    /// ```
    pub fn presence(&self) -> &Arc<Presence> {
        &self.presence
    }

    /// Return a [`DiskUsage`] to monitor the disk usage of the Ditto
    /// instance. It can be used to retrieve an immediate representation of the Ditto file system:
    ///
    /// ```
    /// # use dittolive_ditto::prelude::Ditto;
    /// # fn call_diskusage(ditto: Ditto) {
    /// let fs_tree = ditto.disk_usage().exec();
    /// # }
    /// ```
    /// Or to bind a callback to the changes :
    /// ```
    /// # use dittolive_ditto::prelude::Ditto;
    /// # fn call_diskusage(ditto:Ditto) {
    /// let handle = ditto.disk_usage().observe(|fs_tree| {
    ///     // do something with the graph
    /// });
    /// // The handle must be kept to keep receiving updates on the file system.
    /// // To stop receiving update, drop the handle.
    /// # }
    /// ```
    pub fn disk_usage(&self) -> &DiskUsage {
        &self.disk_usage
    }

    #[deprecated(note = "Use persistence_directory instead")]
    /// Returns the [`DittoRoot`] path.
    pub fn root_dir(&self) -> &Path {
        self.persistence_directory()
    }

    #[deprecated(note = "Use persistence_directory instead")]
    /// Returns the ditto persistence data directory path.
    pub fn data_dir(&self) -> &Path {
        self.persistence_directory()
    }

    #[cfg(test)]
    /// Returns the [`DittoRoot`].
    pub fn root(&self) -> Arc<dyn DittoRoot> {
        self.ditto_root.retain()
    }

    /// Returns the current [`DittoAuthenticator`], if it exists.
    pub fn authenticator(&self) -> Option<DittoAuthenticator> {
        self.auth.clone()
    }

    /// Has this [`Ditto`] instance been activated yet with a valid license
    /// token.
    pub fn is_activated(&self) -> bool {
        self.activated.load()
    }
}

// Constructors
impl Ditto {
    /// Returns a builder for [`Ditto`] following the builder pattern.
    ///
    /// Start constructing a new [`Ditto`] instance.
    pub fn builder() -> DittoBuilder {
        DittoBuilder::new()
    }

    /// Construct a [`Ditto`] instance with sensible defaults.
    ///
    /// This instance will still need to have a license set (if necessary) and sync functionality
    /// manually started.
    pub fn new(app_id: AppId) -> Ditto {
        Ditto::builder()
            .with_root(Arc::new(
                PersistentRoot::from_current_exe().expect("Invalid Ditto Root"),
            ))
            .with_identity(|ditto_root| identity::OfflinePlayground::new(ditto_root, app_id))
            .expect("Invalid Ditto Identity")
            .with_minimum_log_level(LogLevel::Info)
            .build()
            .expect("Failed to build Ditto Instance")
    }

    // This isn't public to customers and is only used internally. It's only currently used when we
    // have access to the `DittoFields` and want to create a `Ditto` instance for whatever reason,
    // but we don't want `Ditto` to be shut down when this `Ditto` instance is dropped.
    //
    // This should definitely be considered a hack for now.
    pub(crate) fn new_temp(fields: Arc<DittoFields>) -> Ditto {
        Ditto {
            fields,
            is_shut_down_able: false,
        }
    }
}

impl Ditto {
    /// Removes all sync metadata for any remote peers which aren't currently connected. This method
    /// shouldn't usually be called. Manually running garbage collection often will result in slower
    /// sync times. Ditto automatically runs a garbage a collection process in the background at
    /// optimal times.
    ///
    /// Manually running garbage collection is typically only useful during testing if large amounts
    /// of data are being generated. Alternatively, if an entire data set is to be evicted and it's
    /// clear that maintaining this metadata isn't necessary, then garbage collection could be run
    /// after evicting the old data.
    pub fn run_garbage_collection(&self) {
        ffi_sdk::ditto_run_garbage_collection(&self.ditto);
    }

    /// Explicitly opt-in to disabling the ability to sync with Ditto peers
    /// running any version of the SDK in the v3 (or lower) series of releases.
    ///
    /// Assuming this succeeds then this peer will only be able to sync with
    /// other peers using SDKs in the v4 (or higher) series of releases. Note
    /// that this disabling of sync spreads to peers that sync with a peer that
    /// has disabled, or has (transitively) had disabled, syncing with v3 SDK
    /// peers.
    pub fn disable_sync_with_v3(&self) -> Result<(), DittoError> {
        let res = ffi_sdk::ditto_disable_sync_with_v3(&self.ditto);
        if res != 0 {
            return Err(DittoError::from_ffi(ErrorKind::Internal));
        }
        Ok(())
    }
}

#[doc(hidden)] // Not implemented
pub struct TransportDiagnostics;

/// `SiteId`s are used to identity Ditto peers
pub type SiteId = u64;

#[derive(Clone, Debug)]
// pub struct AppId(uuid::Uuid); // Demo apps still use arbitrary strings
/// The Ditto application Id
pub struct AppId(pub(crate) String); // neither String nor Vec<u8> are Copy

impl AppId {
    /// Generate a random AppId from a UUIDv4
    pub fn generate() -> Self {
        let uuid = uuid::Uuid::new_v4();
        AppId::from_uuid(uuid)
    }

    /// Generate an AppId from a given UUIDv4
    pub fn from_uuid(uuid: Uuid) -> Self {
        let id_str = format!("{:x}", &uuid); // lower-case with hypens
        AppId(id_str)
    }

    /// Attempt to grab a specific AppId from some environment variable
    pub fn from_env(var: &str) -> Result<Self, DittoError> {
        let id_str = env::var(var).map_err(|err| DittoError::new(ErrorKind::Config, err))?;
        Ok(AppId(id_str))
    }

    /// Return the corresponding string
    pub fn as_str(&self) -> &str {
        &self.0
    }

    /// Return the corresponding c string
    pub fn to_c_string(&self) -> char_p::Box {
        char_p::new(self.0.as_str())
    }

    /// Return the default auth URL associated with the app Id. This is of the form
    /// `https://{app_id}.cloud.ditto.live/` by default.
    pub fn default_auth_url(&self) -> String {
        format!("https://{}.cloud.ditto.live", self.0)
    }

    /// Return the default WebSocket sync URL which is of the form
    /// `wss://{app_id}.cloud.ditto.live/` by default.
    pub fn default_sync_url(&self) -> String {
        format!("wss://{}.cloud.ditto.live", self.0)
    }
}

use std::{fmt, fmt::Display, str::FromStr};

impl Display for AppId {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "{}", self.0)
    }
}

impl FromStr for AppId {
    type Err = DittoError;
    fn from_str(s: &str) -> Result<AppId, DittoError> {
        // later s will need to be a valid UUIDv4
        Ok(AppId(s.to_string()))
    }
}

#[cfg(test)]
#[path = "tests.rs"]
mod tests;