//! [`Store`] is a holder to create, read, write, and remove
//! [`Document`](crate::store::collection::document::DittoDocument)s from a Ditto peer.

use_prelude!();

use std::{
    ops::Deref,
    sync::{RwLock, Weak},
};

use ffi_sdk::{COrderByParam, FsComponent, WriteStrategyRs};

use crate::{
    disk_usage::DiskUsage,
    ditto::DittoFields,
    error::{DittoError, ErrorKind},
    utils::{extension_traits::FfiResultIntoRustResult, SetArc},
};

pub mod batch;
pub mod collection;
pub mod collections;
pub mod ditto_attachment;
pub mod ditto_attachment_fetch_event;
pub mod ditto_attachment_fetcher;
pub mod ditto_attachment_token;
pub mod dql;
pub mod live_query;
#[cfg(feature = "timeseries")]
pub mod timeseries;
pub mod update;

use collections::pending_collections_operation::PendingCollectionsOperation;
use dql::*;

#[derive(Clone)]
/// `Store` provides access to [`Collection`]s and a
/// write transaction API.
pub struct Store {
    ditto: Arc<ffi_sdk::BoxedDitto>,
    // FIXME(Daniel): unify this field with `.ditto`
    weak_ditto_fields: Weak<DittoFields>,
    disk_usage: Arc<DiskUsage>,
    observers: Arc<RwLock<SetArc<StoreObserver>>>,
}

impl Store {
    pub(crate) fn new(
        ditto: Arc<ffi_sdk::BoxedDitto>,
        weak_ditto_fields: Weak<DittoFields>,
    ) -> Self {
        let disk_usage = Arc::new(DiskUsage::new(ditto.retain(), FsComponent::Store));
        Self {
            ditto,
            weak_ditto_fields,
            disk_usage,
            observers: <_>::default(),
        }
    }

    // Note this method's logic will be moved into the core ditto library
    // in the future
    fn validate_collection_name(name: &str) -> Result<(), DittoError> {
        let mut result = Ok(());

        if name.is_empty() {
            result = Err(DittoError::new(
                ErrorKind::InvalidInput,
                String::from("Collection name can not be empty"),
            ));
        }

        if name.split_whitespace().next().is_none() {
            result = Err(DittoError::new(
                ErrorKind::InvalidInput,
                String::from("Collection name can not only contain whitespace"),
            ));
        }

        result
    }

    /// Returns a [`Collection`] with the provided name.
    /// A collection name is valid if :
    /// * its length is less than 100
    /// * it is not empty
    /// * it does not contain the char '\0'
    /// * it does not begin with "$TS_"
    pub fn collection(&self, collection_name: &'_ str) -> Result<Collection, DittoError> {
        Self::validate_collection_name(collection_name)?;
        let c_name = char_p::new(collection_name);
        let status = { ffi_sdk::ditto_collection(&*self.ditto, c_name.as_ref()) };
        if status != 0 {
            return Err(DittoError::from_ffi(ErrorKind::InvalidInput));
        }
        Ok(Collection {
            ditto: Arc::downgrade(&self.ditto),
            collection_name: c_name,
        })
    }

    /// Returns an object that lets you fetch or observe the collections in the
    /// store.
    pub fn collections(&self) -> PendingCollectionsOperation<'_> {
        PendingCollectionsOperation::<'_>::new(Arc::downgrade(&self.ditto))
    }

    /// Allows you to group multiple operations together that affect multiple
    /// documents, potentially across multiple collections, without
    /// auto-committing on each operation.
    ///
    /// At the end of the batch of operations, either
    /// [`batch.commit_changes`](crate::store::batch::ScopedStore::commit_changes)
    /// or
    /// [`batch.revert_changes`](crate::store::batch::ScopedStore::revert_changes)
    /// must be called.
    ///
    /// ## Example
    ///
    /// ```rust
    /// # macro_rules! ignore {($($__:tt)*) => ()} ignore! {
    /// ditto.store().with_batched_write(|batch| {
    ///     let mut foo_coll = batch.collection("foo");
    ///     foo_coll.find...().remove();
    ///     let mut bar_coll = batch.collection("bar");
    ///     // Expensive multi-mutation op:
    ///     for _ in 0 .. 10_000 {
    ///         let doc = ...;
    ///         bar_coll.insert(doc, None, false);
    ///     }
    ///     // At this point, we must say whether we commit or revert
    ///     // these changes:
    ///     batch.commit_changes()
    /// })
    /// # }
    /// ```
    pub fn with_batched_write<F>(
        &self,
        f: F,
    ) -> Result<Vec<batch::WriteTransactionResult>, DittoError>
    where
        for<'batch> F: FnOnce(batch::ScopedStore<'batch>) -> batch::Action<'batch>,
    {
        batch::with_batched_write(self, f)
    }

    /// Returns a list of the names of collections in the local store.
    pub fn collection_names(&self) -> Result<Vec<String>, DittoError> {
        let c_collections = { ffi_sdk::ditto_get_collection_names(&*self.ditto).ok()? };

        Ok(c_collections
            .iter()
            .map(|x: &char_p::Box| -> String { x.clone().into_string() })
            .collect())
    }

    /// Returns a hash representing the current version of the given queries.
    /// When a document matching such queries gets mutated, the hash will change
    /// as well.
    ///
    /// Please note that the hash depends on how queries are constructed, so you
    /// should make sure to always compare hashes generated with the same set of
    /// queries.
    pub fn queries_hash(&self, live_queries: &[LiveQuery]) -> Result<u64, DittoError> {
        let (coll_names, queries): (Vec<_>, Vec<_>) = live_queries
            .iter()
            .map(|lq| (lq.collection_name.as_ref(), lq.query.as_ref()))
            .unzip();

        {
            ffi_sdk::ditto_queries_hash(&self.ditto, coll_names[..].into(), queries[..].into()).ok()
        }
    }

    /// Returns a sequence of English words representing the current version of
    /// the given queries. When a document matching such queries gets mutated,
    /// the words will change as well.
    ///
    /// Please note that the resulting sequence of words depends on how queries
    /// are constructed, so you should make sure to always compare hashes
    /// generated with the same set of queries.
    pub fn queries_hash_mnemonic(&self, live_queries: &[LiveQuery]) -> Result<String, DittoError> {
        let (coll_names, queries): (Vec<_>, Vec<_>) = live_queries
            .iter()
            .map(|lq| (lq.collection_name.as_ref(), lq.query.as_ref()))
            .unzip();

        {
            ffi_sdk::ditto_queries_hash_mnemonic(
                &self.ditto,
                coll_names[..].into(),
                queries[..].into(),
            )
            .ok()
            .map(|c_str| c_str.into_string())
        }
    }

    /// Start all live query webhooks.
    pub fn start_all_live_query_webhooks(&self) -> Result<(), DittoError> {
        {
            let ret = ffi_sdk::ditto_live_query_webhook_start_all(&self.ditto);
            if ret != 0 {
                return Err(DittoError::from_ffi(ErrorKind::Internal));
            }
        }
        Ok(())
    }

    /// Start a live query webhooks by its id.
    pub fn start_live_query_webhook_by_id(&self, doc_id: DocumentId) -> Result<(), DittoError> {
        {
            let ret =
                ffi_sdk::ditto_live_query_webhook_start_by_id(&self.ditto, doc_id.bytes[..].into());
            if ret != 0 {
                return Err(DittoError::from_ffi(ErrorKind::Internal));
            }
        }
        Ok(())
    }

    /// Register a new live query webhook
    pub fn register_live_query_webhook(
        &self,
        collection_name: &str,
        query: &str,
        url: &str,
    ) -> Result<DocumentId, DittoError> {
        let c_collection_name = char_p::new(collection_name);
        let c_query = char_p::new(query);
        let c_url = char_p::new(url);
        let order_definitions: Vec<COrderByParam<'_>> = Vec::with_capacity(0);
        let doc_id = {
            ffi_sdk::ditto_live_query_webhook_register_str(
                &self.ditto,
                c_collection_name.as_ref(),
                c_query.as_ref(),
                order_definitions[..].into(),
                -1,
                0,
                c_url.as_ref(),
            )
            .ok()?
            .to::<Box<[u8]>>()
            .into()
        };

        Ok(doc_id)
    }

    /// Generate a new API secret for live query webhook
    pub fn live_query_webhook_generate_new_api_secret(&self) -> Result<(), DittoError> {
        {
            let ret = ffi_sdk::ditto_live_query_webhook_generate_new_api_secret(&self.ditto);
            if ret != 0 {
                return Err(DittoError::from_ffi(ErrorKind::Internal));
            }
        }
        Ok(())
    }

    /// Returns a [`TimeSeries`] with the provided name.
    pub fn timeseries(&self, ts_name: &'_ str) -> Result<TimeSeries, DittoError> {
        Self::validate_collection_name(ts_name)?;
        let c_name = char_p::new(ts_name);
        Ok(TimeSeries {
            ditto: self.ditto.retain(),
            ts_name: c_name,
        })
    }

    /// Return a [`DiskUsage`] to monitor the disk usage of the
    /// [`Store`].
    pub fn disk_usage(&self) -> &DiskUsage {
        &self.disk_usage
    }

    /// Register a handler to be called whenever a query's results change in the
    /// local store.
    ///
    /// The returned [`StoreObserver`] must be kept in scope for as long as the change handler
    /// should be called with new change events.
    ///
    /// The given query may not modify any documents and must be a valid Ditto
    /// Query Language query.
    ///
    /// If no `on_error` argument is provided, errors that are thrown in the
    /// provided change handler will be logged to the console.
    pub fn register_observer<Q, F>(
        &self,
        query: Q,
        query_args: Option<QueryArguments>,
        on_change: F,
    ) -> Result<Arc<StoreObserver>, DittoError>
    where
        Q: TryInto<Query, Error = DittoError>,
        F: ChangeHandler,
    {
        let ditto = Ditto::upgrade(&self.weak_ditto_fields)?;
        let new_obs = Arc::new(StoreObserver::new(
            &ditto,
            query.try_into()?,
            query_args,
            on_change,
        )?);
        self.observers.write().unwrap().insert(new_obs.retain());
        Ok(new_obs)
    }

    fn unregister_observer(&self, observer: &StoreObserver) -> bool {
        let observers = &mut *self.observers.write().unwrap();
        let removed = observers.remove(observer);
        if removed {
            ::log::debug!(
                "Unregistering store observer {id} with query `{query}`",
                id = observer.live_query_id,
                query = observer._query.inner_string,
            );
            if let Ok(ditto) = Ditto::upgrade(&self.weak_ditto_fields) {
                ffi_sdk::ditto_live_query_stop(&ditto.ditto, observer.live_query_id);
                return true;
            }
        }
        false
    }

    /// Gets temporary access to the set of currently registered observers.
    ///
    /// A (read) lock is held until the return value is dropped: this means
    /// that neither [`Self::register_observer()`] nor
    /// [`StoreObserver::cancel()`] can make progress until this read
    /// lock is released.
    pub fn observers(&self) -> impl '_ + Deref<Target = SetArc<StoreObserver>> {
        self.observers.read().unwrap()
    }

    /// Executes the given query in the local store and returns the result.
    ///
    /// Use placeholders to incorporate values from the optional `query_args`
    /// parameter into the query. The keys of the [`QueryArguments`] object must
    /// match the placeholders used within the query. You can not use placeholders
    /// in the `FROM` clause.
    ///
    /// Limitations:
    ///
    /// - Supports `SELECT * FROM <collection name>` with optional `WHERE <expression>`
    /// - No transactions
    pub async fn execute<Q>(
        &self,
        query: Q,
        query_args: Option<QueryArguments>,
    ) -> Result<QueryResult, DittoError>
    where
        Q: TryInto<Query, Error = DittoError>,
    {
        // Get a DqlResult. The operation to get it is fallible, so we end up
        // with a `FfiResult<DqlResult> -> Result<DqlResult> -> DqlResult`.
        // The `Result` stutter is quite unfortunate, which _results_ from
        // having picked that name in our public SDK API (over, say, `Output`).
        let ffi_query_result = ffi_sdk::dittoffi_try_experimental_exec_statement_str(
            &self.ditto,
            None,
            query.try_into()?.prepare_ffi(),
            query_args.as_ref().map(|qa| qa.cbor().into()),
        )
        .into_rust_result()?;

        Ok(QueryResult::from(ffi_query_result))
    }
}

#[non_exhaustive]
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
/// Specify the order of returned Documents in a query.
pub enum SortDirection {
    Ascending,
    Descending,
}

#[non_exhaustive]
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
/// Specify the write strategy when inserting documents.
pub enum WriteStrategy {
    /// An existing document will be merged with the document being inserted, if there is a
    /// pre-existing document.
    Merge,

    /// Insert the document only if there is not already a document with the same Id in the store.
    /// If there is already a document in the store with the same Id then this will be a no-op.
    InsertIfAbsent,

    /// Insert the document, with its contents treated as default data, only if there is not
    /// already a document with the same Id in the store. If there is already a document in the
    /// store with the same Id then this will be a no-op. Use this strategy if you want to
    /// insert default data for a given document Id, which you want to treat as common initial
    /// data amongst all peers and that you expect to be mutated or overwritten in due course.
    InsertDefaultIfAbsent,
}

impl WriteStrategy {
    fn as_write_strategy_rs(&self) -> WriteStrategyRs {
        match self {
            WriteStrategy::Merge => WriteStrategyRs::Merge,
            WriteStrategy::InsertIfAbsent => WriteStrategyRs::InsertIfAbsent,
            WriteStrategy::InsertDefaultIfAbsent => WriteStrategyRs::InsertDefaultIfAbsent,
        }
    }
}