use_prelude!();

use std::collections::BTreeMap;

use ffi_sdk::COrderByParam;
use serde::Serialize;

use super::*;
use crate::{
    error::{DittoError, ErrorKind},
    ffi_sdk,
    store::{update::UpdateResult, WriteStrategy},
};

/// Group several transaction on a collection in a single commit.
pub struct ScopedCollection<'coll, 'batch> {
    pub(super) batch: &'coll mut ScopedStore<'batch>,

    pub(super) collection_name: char_p::Box,
}

impl ScopedCollection<'_, '_> {
    /// Inserts a new document into the collection and returns its Id. If the
    /// document already exists, the provided document content will be merged
    /// with the existing document's content.
    pub fn upsert<V: ::serde::Serialize, C: Borrow<V>>(
        &'_ mut self,
        content: C,
    ) -> Result<&'_ DocumentId, DittoError> {
        self.insert_cbor(
            ::serde_cbor::to_vec(content.borrow()).unwrap().as_slice(),
            None,
            WriteStrategy::Merge,
        )
    }

    /// Inserts a new document into the collection and returns its Id. If the
    /// document already exists, the behavior is determined by the provided
    /// `write_strategy`.
    pub fn upsert_with_strategy<V: ::serde::Serialize, C: Borrow<V>>(
        &'_ mut self,
        content: C,
        write_strategy: WriteStrategy,
    ) -> Result<&'_ DocumentId, DittoError> {
        self.insert_cbor(
            ::serde_cbor::to_vec(content.borrow()).unwrap().as_slice(),
            None,
            write_strategy,
        )
    }

    fn insert_cbor(
        &'_ mut self,
        cbor: &'_ [u8],
        id: Option<&DocumentId>,
        write_strategy: WriteStrategy,
    ) -> Result<&'_ DocumentId, DittoError> {
        let coll = Collection {
            ditto: Arc::downgrade(&self.batch.store.ditto),
            collection_name: self.collection_name.clone(),
        };
        let doc_id = coll.insert_cbor(cbor, id, write_strategy, Some(self.batch.txn))?;

        self.batch.results.push(WriteTransactionResult {
            doc_id,
            collection_name: self.collection_name.to_owned(),
            kind: DocChangeKind::Inserted,
        });
        Ok(&self.batch.results.last().unwrap().doc_id)
    }
}

impl<'coll, 'batch> ScopedCollection<'coll, 'batch> {
    /// Execute given query
    pub fn find<'find, 'order_by>(
        self: &'find mut ScopedCollection<'coll, 'batch>,
        query: &'_ str,
    ) -> BatchCursorOperation<'find, 'coll, 'batch, 'order_by> {
        let query = char_p::new(query);
        BatchCursorOperation {
            query,
            query_args: None,
            collection: self,
            offset: 0,
            limit: -1,
            order_by: vec![],
        }
    }

    /// Execute given query with args
    pub fn find_with_args<'find, 'order_by, V: ::serde::Serialize, C: Borrow<V>>(
        self: &'find mut ScopedCollection<'coll, 'batch>,
        query: &'_ str,
        query_args: C,
    ) -> BatchCursorOperation<'find, 'coll, 'batch, 'order_by> {
        let query = char_p::new(query);
        BatchCursorOperation {
            query,
            query_args: Some(serde_cbor::to_vec(query_args.borrow()).unwrap()),
            collection: self,
            offset: 0,
            limit: -1,
            order_by: vec![],
        }
    }

    /// Find all document in the current collection.
    pub fn find_all<'find, 'order_by>(
        self: &'find mut ScopedCollection<'coll, 'batch>,
    ) -> BatchCursorOperation<'find, 'coll, 'batch, 'order_by> {
        self.find("true")
    }

    /// Find a document in the Collection using its Id
    pub fn find_by_id(
        self: &'_ mut ScopedCollection<'coll, 'batch>, /* we need mut so we can chain into an
                                                        * update, remove, or evict */
        document_id: DocumentId,
    ) -> BatchIdOperation<'_, 'coll, 'batch> {
        BatchIdOperation {
            collection: self,
            document_id,
        }
    }
}

/// `PendingCursorOperation` transaction
pub struct BatchCursorOperation<'find, 'coll, 'batch, 'order_by> {
    query: char_p::Box,
    query_args: Option<Vec<u8>>,
    collection: &'find mut ScopedCollection<'coll, 'batch>,
    offset: u32,
    limit: i32,
    order_by: Vec<COrderByParam<'order_by>>,
}

impl<'order_by> BatchCursorOperation<'_, '_, '_, 'order_by> {
    /// Execute the query generated by the preceding function chaining and
    /// return the list of matching documents. This occurs immediately.
    pub fn exec(&mut self) -> Result<Vec<BoxedDocument>, DittoError> {
        unsafe {
            ffi_sdk::ditto_collection_exec_query_str(
                &*self.collection.batch.store.ditto,
                self.collection.collection_name.as_ref(),
                Some(self.collection.batch.txn),
                self.query.as_ref(),
                self.query_args.as_ref().map(|qa| (&qa[..]).into()),
                (&self.order_by[..]).into(),
                self.limit,
                self.offset,
            )
        }
        .ok_or(ErrorKind::InvalidInput)
        .map(|c_vec| c_vec.into())
    }

    /// Update documents matching the previously provided query.
    ///
    /// - `updater`: a closure which will be called on _all_ matching documents
    pub fn update<Updater>(
        self,
        updater: Updater,
    ) -> Result<BTreeMap<DocumentId, Vec<UpdateResult>>, DittoError>
    where
        Updater: FnOnce(&mut [BoxedDocument]),
    {
        let mut docs = unsafe {
            ffi_sdk::ditto_collection_exec_query_str(
                &*self.collection.batch.store.ditto,
                self.collection.collection_name.as_ref(),
                Some(self.collection.batch.txn),
                self.query.as_ref(),
                self.query_args.as_ref().map(|qa| (&qa[..]).into()),
                (&self.order_by[..]).into(),
                self.limit,
                self.offset,
            )
            .ok_or(ErrorKind::InvalidInput)?
        };

        // Apply the closure to the document,
        updater(&mut docs);

        // TODO: implement `diff` computation
        let diff = BTreeMap::<DocumentId, Vec<UpdateResult>>::new();

        let code = unsafe {
            ffi_sdk::ditto_collection_update_multiple(
                &self.collection.batch.store.ditto,
                self.collection.collection_name.as_ref(),
                self.collection.batch.txn,
                docs,
            )
        };
        if code != 0 {
            return Err(DittoError::from_ffi(ErrorKind::InvalidInput));
        }
        Ok(diff)
    }

    /// Limit the number of documents that get returned when querying a
    /// collection for matching documents.
    pub fn limit(&mut self, limit: i32) -> &mut Self {
        self.limit = limit;
        self
    }

    /// Offset the resulting set of matching documents. This is useful if you
    /// aren’t interested in the first N matching documents for one reason
    /// or another. For example, you might already have queried the
    /// collection and obtained the first 20 matching documents and so you might
    /// want to run the same query as you did previously but ignore the first 20
    /// matching documents, and that is where you would use offset.
    pub fn offset(&mut self, offset: u32) -> &mut Self {
        self.offset = offset;
        self
    }

    // FIXME: To bring this in line with the other SDKs this should accept a single
    // "order_by" expression, which should then be added to the `order_by` vec
    /// Sort the documents that match the query provided in the preceding
    /// find-like function call.
    pub fn sort(&mut self, sort_param: Vec<COrderByParam<'order_by>>) -> &mut Self {
        self.order_by = sort_param;
        self
    }

    /// Remove all documents that match the query generated by the preceding
    /// function chaining. Returns the Ids of all documents removed
    pub fn remove(&mut self) -> Result<Vec<DocumentId>, DittoError> {
        let ids = unsafe {
            ffi_sdk::ditto_collection_remove_query_str(
                &*self.collection.batch.store.ditto,
                self.collection.collection_name.as_ref(),
                self.collection.batch.txn,
                self.query.as_ref(),
                self.query_args.as_ref().map(|qa| (&qa[..]).into()),
                (&self.order_by[..]).into(),
                self.limit,
                self.offset,
            )
            .ok_or(ErrorKind::InvalidInput)?
        };
        Ok(ids
            .to::<Vec<_>>()
            .into_iter()
            .map(|id| id.to::<Box<[u8]>>().into())
            .collect())
    }

    /// Evict all documents that match the query generated by the preceding
    /// function chaining.
    pub fn evict(&mut self) -> Result<Vec<DocumentId>, DittoError> {
        let ids = unsafe {
            ffi_sdk::ditto_collection_evict_query_str(
                &*self.collection.batch.store.ditto,
                self.collection.collection_name.as_ref(),
                self.collection.batch.txn,
                self.query.as_ref(),
                self.query_args.as_ref().map(|qa| (&qa[..]).into()),
                (&self.order_by[..]).into(),
                self.limit,
                self.offset,
            )
            .ok_or(ErrorKind::InvalidInput)?
        };
        Ok(ids
            .to::<Vec<_>>()
            .into_iter()
            .map(|s| s.to::<Box<[u8]>>().into())
            .collect())
    }
}

pub struct BatchIdOperation<'find_by_id, 'coll, 'batch> {
    collection: &'find_by_id mut ScopedCollection<'coll, 'batch>,
    pub(super) document_id: DocumentId,
}

impl BatchIdOperation<'_, '_, '_> {
    /// Remove the document with the matching Id.
    pub fn remove(self) -> Result<(), DittoError> {
        let removed = unsafe {
            ffi_sdk::ditto_collection_remove(
                &*self.collection.batch.store.ditto,
                self.collection.collection_name.as_ref(),
                self.collection.batch.txn,
                self.document_id.as_ref().into(),
            )
            .ok()?
        };
        if removed.not() {
            return Err(DittoError::from_ffi(ErrorKind::NonExtant));
        }
        // DO NOT commit the operation
        self.collection.batch.results.push(WriteTransactionResult {
            doc_id: self.document_id.clone(),
            collection_name: self.collection.collection_name.to_owned(),
            kind: DocChangeKind::Removed,
        });
        Ok(())
    }

    /// Evict the document with the matching Id.
    pub fn evict(&mut self) -> Result<(), DittoError> {
        let evicted = unsafe {
            ffi_sdk::ditto_collection_evict(
                &*self.collection.batch.store.ditto,
                self.collection.collection_name.as_ref(),
                self.collection.batch.txn,
                self.document_id.as_ref().into(),
            )
            .ok()?
        };
        if evicted.not() {
            return Err(DittoError::from_ffi(ErrorKind::NonExtant));
        }
        self.collection.batch.results.push(WriteTransactionResult {
            doc_id: self.document_id.clone(),
            collection_name: self.collection.collection_name.to_owned(),
            kind: DocChangeKind::Evicted,
        });
        Ok(())
    }

    /// Update the document with the matching ID.
    /// * `updater` - a closure which will be called on the selected document if found
    pub fn update<Updater>(self, updater: Updater) -> Result<Vec<UpdateResult>, DittoError>
    where
        Updater: FnOnce(Option<&mut BoxedDocument>), /* Arg is a Mutable Document, which only
                                                      * exists in SDKs */
    {
        let mut document = {
            let mut read_txn = unsafe {
                ffi_sdk::ditto_read_transaction(&*self.collection.batch.store.ditto).ok()?
            };
            let result = unsafe {
                ffi_sdk::ditto_collection_get(
                    &*self.collection.batch.store.ditto,
                    self.collection.collection_name.as_ref(),
                    self.document_id.as_ref().into(),
                    &mut read_txn,
                )
            };
            if result.status_code != 0 {
                // Should this be a `NonExtant` error or an `Internal` one?
                return Err(DittoError::from_ffi(ErrorKind::NonExtant));
            }
            result.ok_value // don't unwrap option
        };

        // Apply the closure to the document
        updater(document.as_mut());
        let diff = Vec::with_capacity(0); // TODO Mutable doc will populate this

        if let Some(doc) = document {
            unsafe {
                ffi_sdk::ditto_collection_update(
                    &*self.collection.batch.store.ditto,
                    self.collection.collection_name.as_ref(),
                    self.collection.batch.txn,
                    doc,
                )
            };
            self.collection.batch.results.push(WriteTransactionResult {
                doc_id: self.document_id.clone(),
                collection_name: self.collection.collection_name.to_owned(),
                kind: DocChangeKind::Updated,
            });
            Ok(diff)
        } else {
            Err(DittoError::from_ffi(ErrorKind::NonExtant))
        }
    }

    /// Replaces the matching document with the provided value
    /// * `new_value` - A new Serializable which will replace the found document
    /// Note this actually follows "upsert" rules and will insert a document if
    /// no document is found with the given DocumentId.
    pub fn update_doc<T>(self, new_value: &T) -> Result<(), DittoError>
    where
        T: Serialize,
    {
        // We use the doc_id to find the document (verify it exists)
        // and only if it is found, we then replace its contents with new_value
        // then we insert this mutated document.

        // memory to receive the doc if found
        let mut document: BoxedDocument = {
            let mut read_txn = unsafe {
                ffi_sdk::ditto_read_transaction(&*self.collection.batch.store.ditto).ok()?
            };
            unsafe {
                ffi_sdk::ditto_collection_get(
                    &*self.collection.batch.store.ditto,
                    self.collection.collection_name.as_ref(),
                    self.document_id.as_ref().into(),
                    &mut read_txn,
                )
            }
            .ok_or(ErrorKind::NonExtant)?
        };
        let new_content = ::serde_cbor::to_vec(new_value).unwrap();
        // Merge the provided value onto the new document
        if unsafe { ffi_sdk::ditto_document_update(&mut document, new_content.as_slice().into()) }
            != 0
        {
            return Err(DittoError::from_ffi(ErrorKind::InvalidInput));
        }

        let status = unsafe {
            ffi_sdk::ditto_collection_update(
                &*self.collection.batch.store.ditto,
                self.collection.collection_name.as_ref(),
                self.collection.batch.txn,
                document,
            )
        };
        if status != 0 {
            Err(DittoError::from_ffi(ErrorKind::InvalidInput))
        } else {
            Ok(())
        }
    }
}