//! Ditto Builder
//!
//! Provides idiomatic configuration of a Ditto instance using the "Builder
//! Pattern"

use ffi_sdk::{
    ffi_utils::{DynDrop, DynExecutor},
    Platform,
};

use super::*;
use crate::identity::Identity;

/// Builder for [`Ditto`]
pub struct DittoBuilder {
    ditto_root: Option<Arc<dyn DittoRoot>>,
    identity: Option<Arc<dyn Identity>>,
    minimum_log_level: LogLevel,
    transport_config: Option<TransportConfig>,
}

impl DittoBuilder {
    /// Create a new, empty builder for a [`Ditto`] instance.
    pub fn new() -> DittoBuilder {
        DittoBuilder {
            ditto_root: None,
            identity: None,
            minimum_log_level: LogLevel::Info,
            transport_config: None,
        }
    }

    /// Set the root directory where Ditto will store its data.
    pub fn with_root(mut self, ditto_root: Arc<dyn DittoRoot>) -> Self {
        self.ditto_root = Some(ditto_root);
        self
    }

    /// Configure the minimum log level for the [`Ditto`] instance.
    pub fn with_minimum_log_level(mut self, log_level: LogLevel) -> Self {
        self.minimum_log_level = log_level;
        self
    }

    /// Build a [`Ditto`] instance with a temporary storage directory which
    /// will be destroyed on exit.
    pub fn with_temp_dir(mut self) -> Self {
        let root = TempRoot::new();
        self.ditto_root = Some(Arc::new(root));
        self
    }

    fn platform() -> Platform {
        using!(match () {
            use ffi_sdk::Platform;
            | _case if cfg!(target_os = "windows") => Platform::Windows,
            | _case if cfg!(target_os = "android") => Platform::Android,
            | _case if cfg!(target_os = "macos") => Platform::Mac,
            | _case if cfg!(target_os = "ios") => Platform::Ios,
            | _case if cfg!(target_os = "linux") => Platform::Linux,
            | _default => Platform::Unknown,
        })
    }

    fn sdk_version() -> String {
        let sdk_semver = env!("CARGO_PKG_VERSION");
        sdk_semver.to_string()
    }

    fn init_sdk_version() {
        let platform = Self::platform();
        let sdk_semver = Self::sdk_version();
        let c_version = char_p::new(sdk_semver);
        ffi_sdk::ditto_init_sdk_version(platform, ffi_sdk::Language::Rust, c_version.as_ref());
    }

    fn init_logging(&self) {
        ffi_sdk::ditto_logger_init();
        ffi_sdk::ditto_logger_minimum_log_level(self.minimum_log_level);
        ffi_sdk::ditto_logger_enabled(true);
    }

    /// Provide a factory [`FnOnce`] which will create and configure the
    /// [`Identity`] for the [`Ditto`] instance.
    pub fn with_identity<F, I>(mut self, factory: F) -> Result<Self, DittoError>
    where
        F: FnOnce(Arc<dyn DittoRoot>) -> Result<I, DittoError>,
        I: Identity + 'static, // must return something ownable
    {
        match &self.ditto_root {
            Some(root) => {
                let identity = factory(root.retain())?;
                self.identity = Some(Arc::new(identity));
            }
            None => {
                let msg = "A valid DittoRoot directory must be provided before configuring the \
                           Identity"
                    .to_string();
                return Err(DittoError::new(ErrorKind::Config, msg));
            }
        };
        Ok(self)
    }

    /// Provide a factory for the [`TransportConfig`] used by the
    /// [`Ditto`] instance.
    pub fn with_transport_config<T>(mut self, factory: T) -> Result<Self, DittoError>
    where
        T: FnOnce(Arc<dyn Identity>) -> TransportConfig,
    {
        match &self.identity {
            Some(id) => {
                let config = factory(id.retain());
                self.transport_config = Some(config)
            }
            None => {
                let msg = "A DittoRoot directory and Identity must first be specified before \
                           providing a custom TransportConfig"
                    .to_string();
                return Err(DittoError::new(ErrorKind::Config, msg));
            }
        }
        Ok(self)
    }

    /// Builds the [`Ditto`] instance, consuming the builder in the process.
    pub fn build(self) -> Result<Ditto, DittoError> {
        self.init_logging();
        Self::init_sdk_version();
        let ditto_root = self.ditto_root.ok_or_else(|| {
            DittoError::new(ErrorKind::Config, "No Ditto Root Directory provided")
        })?;

        crate::fs::drain_ditto_data_dir(&ditto_root);

        let c_root_dir = ditto_root.root_dir_to_c_str()?;
        let identity = self
            .identity
            .ok_or_else(|| DittoError::new(ErrorKind::Config, "No Identity specified"))?;

        let executor = make_executor();
        let uninit_ditto =
            ffi_sdk::ditto_uninitialized_ditto_make_with_executor(c_root_dir.as_ref(), executor);

        // The identity config should only be `None` _after_ this call below (because it ends up
        // being consumed by `ditto_make`)
        let identity_config = identity
            .identity_config()
            .expect("identity config to be Some");

        let boxed_ditto = ffi_sdk::ditto_make(
            uninit_ditto,
            identity_config,
            ffi_sdk::HistoryTracking::Disabled,
        );
        let ditto: DittoHandleWrapper = Arc::new(boxed_ditto);
        let site_id: SiteId = ffi_sdk::ditto_auth_client_get_site_id(&ditto);
        let transport_config = self.transport_config.unwrap_or_else(|| {
            let mut config = TransportConfig::new();
            config.enable_all_peer_to_peer();
            config
        });
        let transports: Arc<RwLock<TransportSync>> = Arc::new(RwLock::new(
            TransportSync::from_config(transport_config, ditto.retain(), identity.retain()),
        ));
        let auth = identity.authenticator();
        let validity_listener = Some(ValidityListener::new(Arc::downgrade(&transports), &ditto));
        let presence = Arc::new(Presence::new(ditto.retain()));
        let on_connecting = Arc::new(OnConnecting::new(ditto.retain()));
        let disk_usage = DiskUsage::new(ditto.retain(), FsComponent::Root);
        let small_peer_info = SmallPeerInfo::new(ditto.retain());
        let fields = Arc::new_cyclic(|weak_fields: &arc::Weak<_>| {
            let store = Store::new(ditto.retain(), weak_fields.clone());
            let auth = auth.map(|mut auth| {
                auth.ditto_fields = weak_fields.clone();
                auth
            });
            let sync = crate::sync::Sync::new(weak_fields.clone());
            DittoFields {
                ditto: ditto.retain(),
                auth,
                identity: identity.retain(),
                store,
                sync,
                activated: identity.requires_offline_only_license_token().not().into(),
                site_id,
                transports,
                ditto_root,
                validity_listener,
                presence,
                on_connecting,
                disk_usage,
                small_peer_info,
            }
        });

        // See inline comments in `Identity` trait about why this is necessary.
        identity.set_login_provider(fields.auth.as_ref().map(|a| a.retain()));

        let sdk_ditto = Ditto {
            fields,
            is_shut_down_able: true,
        };
        Ok(sdk_ditto)
    }
}

impl Default for DittoBuilder {
    fn default() -> Self {
        Self::new()
    }
}

/// # The executor situation
///
/// `::dittoffi` and `::dittolive_ditto` (_a.k.a._ "the Rust SDK") may be using *different* versions
/// of `::tokio`. This can be caused by:
///
/// - basic Cargo version drift: remember, the actual `::tokio` version picked in the Rust SDK is
/// ultimately determined and resolved from within the *customer*'s downstream crate (_e.g._, they
/// will ignore our `.lock` file, and only interact with our `.toml` file as a lower-bound version
/// constraint);
///
/// - feature mismatch: similarly, even if the same `::tokio` versions are being picked; feature
///   sets themselves may drift as well.
///
/// - compiler mismatch: our `dittoffi` binary artifacts are distributed/bundled *precompiled*,
///   whereas `::dittolive_ditto` and downstream dependents may be compiled by a completely
///   different Rust toolchain, which may lead to ABI mismatches.
///
/// This, in turn, can lead to things like thread-local lookup or datastructure (_e.g._, `Handle`s)
/// field access to misbehave (with panics or even segfaults).
///
/// So it is *paramount* to either:
/// 1. not mix up the two tokios, and properly treat them separately;
/// 2. or to treat them *as opaque* from the other side of the FFI.
///
/// One approach *could* have been to just not let `:dittolive_ditto` and dependents interact with
/// `::dittoffi`'s runtime whatsoever (option 1).
/// - (For instance, this is what the other SDKs are doing, since they are oblivious to `::tokio`'s
/// existence).
///
/// The approach we have gone with, though, is to, instead, give the Rust SDK preferential treatment
/// w.r.t. interacting with the runtime. This means we have to go with option 2., and thereby need
/// opaque treatment and handling of tokio.
///
/// This translates to *virtual objects* ("dyn Tokio" so to speak; `dyn
/// FfiFutureExecutor`/`DynExecutor` to be precise) when dealing with executor *instances*, but also
/// to overriding certain global (instance-less) functions.
///
/// Hence `DynExecutor` for the former, and `register_executor_functions` for the latter.
///
/// Finally, the whole point of doing this –the "special treatment"– is to be able to reüse a
/// `::ditto_live::ditto`'s `async` caller's `::tokio` executor/handle/context within `::dittoffi`'s
/// own `async` operations.
///
/// To illustrate:
///
/// - Option A
/// ```rust ,ignore
/// //! `::customer`
///
/// fn main() {
///     let ditto = Ditto::new();
/// }
/// ```
///
/// - Option B
/// ```rust ,ignore
/// //! `::customer`
///
/// #[::tokio::main]
/// async fn main() {
///     let ditto = Ditto::new();
/// }
/// ```
///
/// - (Option B.2 would be to have an `#[::async_std::main]` version thereof)
///
/// As you can see, when `Ditto` is being instantiated by a customer, we may be within an `async`
/// context, or we may not.
///
/// If we are, there ought to be a runtime handle around, which we get, wrap within a FFI-safe dyn
/// layer (`DynExecutor`), and provide it to `dittoffi`, alongside the `executor_functions`
/// override.
///
/// If we are not, then even though we could just do *nothing at all* and leave `::dittoffi`'s
/// `::tokio` executor just be, for the sake of consistency within `::dittolive_ditto`, we will
/// nonetheless override it (as stated above), but with a fully fresh `::tokio` Runtime, created by
/// `DynExecutor::new()`.
///
/// - (the astute reader will notice that this heuristic fails to reüse the executor in `B.2`,
///   instead treating it as `A`)
///
/// ## The reactor situation
///
/// - (tokio just keeps on giving, does it not?)
///
/// ### Reactors and the tokio reactor problem
///
/// Note: this involves advanced details of `async` runtimes, so you may need to get acquainted with
/// these.
///
/// See, for instance: <https://youtu.be/7pU3gOVAeVQ?t=1152> and a few minutes onwards, which should
/// include the "unparking" idea. Ideally you could watch the whole video.
///
/// - Otherwise, see this very self-contained executor-and-reactor snippet: <https://www.rustexplorer.com/b/smepx6>
///
/// Long story short:
/// - `Future`s are trees of other `Future`s: an `async` block/fn is made of `.await`ing inner
///   futures, and so on, all the way down to a "leaf future" in charge of dealing with the
///   completion of a certain operation.
/// - the root/top of the whole tree, _i.e._, the outermost `Future` fed to `block_on()` or to
///   `.spawn()` is then called a _task_.
/// - concurrency is thus achieved by polling multiple tasks, most of which will probably not be
///   ready, by the way.
/// - the thing in charge of polling them, for either the `.spawn()`ed tasks to progress to
///   completion (end of the `async` fn/block), or for the `block_on`-ed future to yield its return
///   value, is then called an `Executor`.
/// - for the polling to be efficient (no busy polling on non-ready tasks), whenever a task yields
///   `Poll::Pending` (in turn, caused by a *leaf future* doing so!), the executor is then allowed
///   "never to poll the task again"…
/// - …until the resource in question of that leaf future becomes ready, and *notifies* the executor
///   about it.
/// - this is achieved thanks to the `cx: &mut Context<'_>` parameter in `poll`: it is to be used to
///   retrieve a `Waker` out of it with `cx.waker().clone()`, which can then be passed to some
///   low-level system which shall, eventually, call `waker.wake()`
/// - this "low-level system" may involve extra runtime machinery, and is commonly called a reactor.
///
/// For instance:
/// - the timer thread: spawn a thread, with a heap of deadlines and `Waker`s, which shall "`sleep`
///   until the next deadline" so as to wake the corresponding `Waker`, and thus, task.
/// - fs threads, for special `.spawn_blocking` helper _threads_, dedicated to making the *blocking*
///   fs operations seemingly async.
/// - os-level shenanigans, such as epoll, kqueue, _etc._ which `::mio` is expected to take care of
///   in an OS-abstracting manner.
///
/// The way any sensible async runtime implementation would deal with it (incidentally, this is how
/// `async-std` does it), would be with some global state dedicated to this reactor stuff.
///
/// The way `::tokio` does it, is by, under the rug, smuggling it within its "`Runtime`"s, even
/// though runtime instances are supposed to just be about the `Executor` end of an `async`
/// framework.
///
/// Because of this, *certain* leaf futures, such as `::tokio::time::sleep()`, take advantage of it
/// to query, through the `Executor`-provided `cx` handle, the associated reactor smuggled within
/// it, so as to register the timer wake-up through it. This is what makes such leaf futures just
/// straight up panic when an executor candidly oblivious to these shady shenanigans (such as
/// `::async-std`'s, **or a different `::tokio` "version"**) tries to `.await`/poll them.
///
/// ### Our solution
///
/// We solve this the same way `::async-std` does it: through its `::async-compat` helper, which
/// uses another `::tokio Runtime`, *globally* (with as few executor threads, _i.e._, "core
/// threads", as possible) so as to finally feature the more robust and apt "global reactor"
/// pattern.
///
/// That way, whenever we are to await on a potentially "impure" task such as one containing one of
/// these problematic tokio leaf futures, we can use this global runtime to set up the reactor
/// context expected by them.
///
/// Hence `FfiFutureExecutor`'s `.block_on_within_tokio_reactor()` and
/// `.spawn_within_tokio_reactor()` utilities.
///
/// #### Mocked time and `tokio`'s `test-utils`
///
/// The way `tokio`'s `test-utils` offer the "mocked time" utilities, is by having its timer reactor
/// know of them (so as to skip the aforementioned `sleep()`s so that the registered deadlines are
/// reached instantaneously, but in the right chronological order).
///
/// However, for our tests, there are issues:
///
/// - `#[tokio::test]` (and currently, `#[ditto_test]`), handles an `async fn test_me` by: 1.
///   creating a new dedicated single-executor-threaded runtime, *with its own reactor*, 2. and then
///   using it to `.block_on(test_me())`
///
/// This, in turn, can lead to there being multiple reactors, so if one of them "cheats with time",
/// but not the other, we are in trouble. We can tackle this in two ways:
/// - we currently have the `no-ffi-safe-executors` Cargo feature opt-out on `::dittoutils` (and
///   `::dittoreplication` which forwards to it) that disables the whole `dyn Tokio` and
///   `async-compat` altogether (`executor/tokio.rs` takes over `executor/ffi.rs`).
/// - we also have a `run_with_mocked_time()` helper function.
///
/// However, a plausible problem of `run_with_mocked_time()`, alone, is then that it results in a
/// single reactor for all the `async fn test_me()` functions (by design!), and so in concurrent
/// registration of mocked time deadlines across test functions, which could break the
/// reproducibility of deadline resolution, which could break the very expectations of
/// mocked-time-aware tests?
///
/// A workaround for this, which we have serendipitously been featuring thanks to `cargo nextest`,
/// is to use multi-processing rather than multi-threading to run the tests in parallel, thereby
/// isolating the "global" reactor to each one of tests.
fn make_executor() -> DynExecutor {
    idempotent_register_executor_functions();
    if let Ok(handle) = ::tokio::runtime::Handle::try_current() {
        DynExecutor {
            handle: Arc::new(handle).into(),
            _runtime: DynDrop::new(()),
        }
    } else {
        DynExecutor::new().unwrap()
    }
}

fn idempotent_register_executor_functions() {
    #![allow(improper_ctypes_definitions)]

    static ONCE: ::std::sync::Once = ::std::sync::Once::new();
    ONCE.call_once(|| {
        ffi_sdk::register_executor_functions({
            use ::ffi_sdk::*;

            extern "C" fn rust_sdk_get_current_handle() -> FfiHandle {
                Arc::new(::tokio::runtime::Handle::current()).into()
            }

            extern "C" fn rust_sdk_block_in_place(
                func: ::safer_ffi::prelude::VirtualPtr<dyn '_ + FfiFnMut>,
            ) {
                ::tokio::task::block_in_place(move || { func }.call())
            }

            ExecutorFunctions {
                get_current_handle: rust_sdk_get_current_handle,
                block_in_place: BlockInPlace(rust_sdk_block_in_place),
            }
        })
    });
}