use crate::{
    error::{
        CreateIndexError, CreateSegmentError, DeleteError, DropIndexError, DropSegmentError, GenericError,
        IndexChangeError, IndexError, IndexOpsError, PrepareError, SegmentError, UpdateError, PE,
    },
    id::{IndexId, PersyId, SegmentId, ToIndexId, ToSegmentId},
    index::{
        config::{IndexType, ValueMode, INDEX_DATA_PREFIX, INDEX_META_PREFIX},
        iter::TxIndexIter,
        keeper::TxIndexRawIter,
        tree::nodes::Value,
        value_iter::ValueIter,
    },
    persy::{IndexInfo, PersyImpl, TxFinalize},
    segment_iter::TxSegmentIter,
    transaction_impl::TransactionImpl,
    ReadError,
};
use std::{ops::RangeBounds, sync::Arc};

/// Transaction container, it include all the changes done in a transaction.
pub struct Transaction {
    pub(crate) persy_impl: Arc<PersyImpl>,
    pub(crate) tx: Option<TransactionImpl>,
}

fn tx_mut(tx: &mut Option<TransactionImpl>) -> &mut TransactionImpl {
    tx.as_mut().unwrap()
}
impl Transaction {
    fn tx_mut(&mut self) -> &mut TransactionImpl {
        tx_mut(&mut self.tx)
    }
    fn tx(&self) -> &TransactionImpl {
        self.tx.as_ref().unwrap()
    }
    /// Create a new segment with the provided name
    ///
    /// # Example
    ///
    /// ```rust
    /// # use persy::{OpenOptions};
    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
    /// # let persy = OpenOptions::new().memory()?;
    /// let mut tx = persy.begin()?;
    /// tx.create_segment("my_new_segment")?;
    /// tx.prepare()?.commit()?;
    /// # Ok(())
    /// # }
    /// ```
    pub fn create_segment(&mut self, segment: &str) -> Result<SegmentId, PE<CreateSegmentError>> {
        assert!(!segment.starts_with(INDEX_META_PREFIX));
        assert!(!segment.starts_with(INDEX_DATA_PREFIX));
        Ok(self.persy_impl.create_segment(tx_mut(&mut self.tx), segment)?)
    }

    /// Drop a existing segment
    ///
    /// # Example
    ///
    /// ```rust
    /// # use persy::{OpenOptions};
    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
    /// # let persy = OpenOptions::new().memory()?;
    /// # let mut tx = persy.begin()?;
    /// # tx.create_segment("existing_segment_name")?;
    /// # tx.prepare()?.commit()?;
    /// let mut tx = persy.begin()?;
    /// tx.drop_segment("existing_segment_name")?;
    /// tx.prepare()?.commit()?;
    /// # Ok(())
    /// # }
    /// ```
    pub fn drop_segment(&mut self, segment: &str) -> Result<(), PE<DropSegmentError>> {
        Ok(self.persy_impl.drop_segment(tx_mut(&mut self.tx), segment)?)
    }

    /// Check if a segment already exist in the storage considering the transaction
    ///
    ///
    /// # Example
    /// ```rust
    /// # use persy::{OpenOptions};
    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
    /// # let persy = OpenOptions::new().memory()?;
    /// let mut tx = persy.begin()?;
    /// tx.create_segment("my_new_segment")?;
    /// assert!(tx.exists_segment("my_new_segment")?);
    /// # tx.prepare()?.commit()?;
    /// # Ok(())
    /// # }
    /// ```
    pub fn exists_segment(&self, segment: &str) -> Result<bool, PE<GenericError>> {
        Ok(self.persy_impl.exists_segment_tx(self.tx(), segment)?)
    }

    /// Resolves the segment to a SegmentId, considering the transaction
    ///
    /// # Example
    ///
    /// ```rust
    /// # use persy::{OpenOptions};
    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
    /// # let persy = OpenOptions::new().memory()?;
    /// let mut tx = persy.begin()?;
    /// tx.create_segment("my_new_segment")?;
    /// let segment_id = tx.solve_segment_id("my_new_segment")?;
    /// # tx.prepare()?.commit()?;
    /// # Ok(())
    /// # }
    /// ```
    pub fn solve_segment_id(&self, segment: impl ToSegmentId) -> Result<SegmentId, PE<SegmentError>> {
        Ok(self.persy_impl.solve_segment_id_tx(self.tx(), segment)?)
    }

    /// Resolves the index name to a IndexId, considering the transaction,
    /// this has no public use as today, but may be used in future.
    ///
    /// # Example
    ///
    /// ```rust
    /// # use persy::{OpenOptions, ValueMode};
    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
    /// # let persy = OpenOptions::new().memory()?;
    /// # let mut tx = persy.begin()?;
    /// let mut tx = persy.begin()?;
    /// tx.create_index::<u8,u8>("my_new_index", ValueMode::Cluster)?;
    /// let index_id = tx.solve_index_id("my_new_index")?;
    /// # tx.prepare()?.commit()?;
    /// # Ok(())
    /// # }
    /// ```
    pub fn solve_index_id(&self, index: impl ToIndexId) -> Result<IndexId, PE<IndexError>> {
        let (id, _) = self.persy_impl.solve_index_id_tx(self.tx(), index)?;
        Ok(id)
    }

    /// Create a new record.
    ///
    /// This function return an id that can be used by [`read`],
    /// the record content can be read only with the [`transaction read`] till the transaction is committed.
    ///
    /// [`read`]:struct.Persy.html#method.read
    /// [`transaction read`]:struct.Transaction.html#method.read
    ///
    /// # Example
    ///
    /// ```rust
    /// # use persy::{OpenOptions};
    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
    /// # let persy = OpenOptions::new().memory()?;
    /// let mut tx = persy.begin()?;
    /// # tx.create_segment("seg")?;
    /// let data = vec![1;20];
    /// tx.insert("seg", &data)?;
    /// tx.prepare()?.commit()?;
    /// # Ok(())
    /// # }
    /// ```
    pub fn insert(&mut self, segment: impl ToSegmentId, rec: &[u8]) -> Result<PersyId, PE<SegmentError>> {
        Ok(PersyId(self.persy_impl.insert_record(
            tx_mut(&mut self.tx),
            segment,
            rec,
        )?))
    }

    /// Read the record content considering eventual in transaction changes.
    ///
    /// # Example
    ///
    /// ```rust
    /// # use persy::{OpenOptions};
    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
    /// # let persy = OpenOptions::new().memory()?;
    /// let mut tx = persy.begin()?;
    /// # tx.create_segment("seg")?;
    /// let data = vec![1;20];
    /// let id = tx.insert("seg", &data)?;
    /// let read = tx.read("seg", &id)?.expect("record exists");
    /// assert_eq!(data,read);
    /// # tx.prepare()?.commit()?;
    /// # Ok(())
    /// # }
    /// ```
    pub fn read(&mut self, segment: impl ToSegmentId, id: &PersyId) -> Result<Option<Vec<u8>>, PE<ReadError>> {
        let segment_id = self.solve_segment_id(segment).map_err(|PE::PE(e)| ReadError::from(e))?;
        Ok(self.persy_impl.read_tx(tx_mut(&mut self.tx), segment_id, &id.0)?)
    }

    /// Scan for persistent and in transaction records
    ///
    /// # Example
    ///
    /// ```rust
    /// # use persy::{OpenOptions};
    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
    /// # let persy = OpenOptions::new().memory()?;
    /// let mut tx = persy.begin()?;
    /// # tx.create_segment("seg")?;
    /// let data = vec![1;20];
    /// let id = tx.insert("seg", &data)?;
    /// let mut count = 0;
    /// for (id,content) in tx.scan("seg")? {
    ///     println!("record size:{}",content.len());
    ///     count+=1;
    /// }
    /// assert_eq!(count,1);
    /// # Ok(())
    /// # }
    /// ```
    pub fn scan(&mut self, segment: impl ToSegmentId) -> Result<TxSegmentIter, PE<SegmentError>> {
        let segment_id = self.solve_segment_id(segment)?;
        Ok(TxSegmentIter::new(
            self.persy_impl.scan_tx(self.tx.as_mut().unwrap(), segment_id)?,
            self,
        ))
    }

    /// Update the record content.
    ///
    /// This updated content can be read only with the [`transaction read`] till the transaction is committed.
    ///
    /// [`read`]:struct.Transaction.html#method.read
    ///
    /// # Example
    ///
    /// ```rust
    /// # use persy::{OpenOptions};
    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
    /// # let persy = OpenOptions::new().memory()?;
    /// let mut tx = persy.begin()?;
    /// # tx.create_segment("seg")?;
    /// let data = vec![1;20];
    /// let id = tx.insert("seg", &data)?;
    /// let new_data = vec![2;20];
    /// tx.update("seg", &id, &new_data)?;
    /// # tx.prepare()?.commit()?;
    /// # Ok(())
    /// # }
    /// ```
    pub fn update(&mut self, segment: impl ToSegmentId, id: &PersyId, rec: &[u8]) -> Result<(), PE<UpdateError>> {
        let segment_id = self
            .solve_segment_id(segment)
            .map_err(|e| PE::PE(UpdateError::from(e.error())))?;
        Ok(self.persy_impl.update(tx_mut(&mut self.tx), segment_id, &id.0, rec)?)
    }

    /// Delete a record.
    ///
    /// The record will result deleted only reading it with [`transaction read`] till the transaction is committed.
    ///
    /// [`transaction read`]:struct.Persy.html#method.read
    ///
    /// # Example
    ///
    /// ```rust
    /// # use persy::{OpenOptions};
    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
    /// # let persy = OpenOptions::new().memory()?;
    /// let mut tx = persy.begin()?;
    /// # tx.create_segment("seg")?;
    /// let data = vec![1;20];
    /// let id = tx.insert("seg", &data)?;
    /// tx.delete("seg", &id)?;
    /// # tx.prepare()?.commit()?;
    /// # Ok(())
    /// # }
    /// ```
    pub fn delete(&mut self, segment: impl ToSegmentId, id: &PersyId) -> Result<(), PE<DeleteError>> {
        let segment_id = self
            .solve_segment_id(segment)
            .map_err(|e| PE::PE(DeleteError::from(e.error())))?;
        Ok(self.persy_impl.delete(tx_mut(&mut self.tx), segment_id, &id.0)?)
    }

    /// Create a new index with the name and the value management mode.
    ///
    /// The create operation require two template arguments that are the types as keys and
    /// values of the index this have to match the following operation on the indexes.
    ///
    /// # Example
    ///
    /// ```rust
    /// # use persy::{OpenOptions, ValueMode};
    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
    /// # let persy = OpenOptions::new().memory()?;
    /// let mut tx = persy.begin()?;
    /// tx.create_index::<u8,u8>("my_new_index", ValueMode::Cluster)?;
    /// # tx.prepare()?.commit()?;
    /// # Ok(())
    /// # }
    /// ```
    pub fn create_index<K, V>(&mut self, index_name: &str, value_mode: ValueMode) -> Result<(), PE<CreateIndexError>>
    where
        K: IndexType,
        V: IndexType,
    {
        Ok(self
            .persy_impl
            .create_index::<K, V>(tx_mut(&mut self.tx), index_name, value_mode)?)
    }

    /// Drop an existing index.
    ///
    /// # Example
    ///
    /// ```rust
    /// # use persy::{OpenOptions, ValueMode};
    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
    /// # let persy = OpenOptions::new().memory()?;
    /// # let mut tx = persy.begin()?;
    /// # tx.create_index::<u8,u8>("my_new_index", ValueMode::Cluster)?;
    /// # tx.prepare()?.commit()?;
    /// let mut tx = persy.begin()?;
    /// tx.drop_index("my_new_index")?;
    /// # tx.prepare()?.commit()?;
    /// # Ok(())
    /// # }
    /// ```
    pub fn drop_index(&mut self, index_name: &str) -> Result<(), PE<DropIndexError>> {
        Ok(self.persy_impl.drop_index(tx_mut(&mut self.tx), index_name)?)
    }

    /// Check if a segment already exist in the storage considering the transaction
    ///
    /// # Example
    ///
    /// ```rust
    /// # use persy::{OpenOptions, ValueMode};
    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
    /// # let persy = OpenOptions::new().memory()?;
    /// let mut tx = persy.begin()?;
    /// tx.create_index::<u8,u8>("my_new_index", ValueMode::Replace)?;
    /// assert!(tx.exists_index("my_new_index")?);
    /// # tx.prepare()?.commit()?;
    /// # Ok(())
    /// # }
    /// ```
    pub fn exists_index(&self, index_name: &str) -> Result<bool, PE<GenericError>> {
        Ok(self.persy_impl.exists_index_tx(self.tx(), index_name)?)
    }

    /// Put a key value in an index following the value mode strategy.
    ///
    /// # Example
    ///
    /// ```rust
    /// # use persy::{OpenOptions, ValueMode};
    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
    /// # let persy = OpenOptions::new().memory()?;
    /// let mut tx = persy.begin()?;
    /// tx.create_index::<u8,u8>("my_new_index", ValueMode::Cluster)?;
    /// tx.put::<u8,u8>("my_new_index",10,10)?;
    /// tx.prepare()?.commit()?;
    /// # Ok(())
    /// # }
    /// ```
    pub fn put<K, V>(&mut self, index_name: &str, k: K, v: V) -> Result<(), PE<IndexOpsError>>
    where
        K: IndexType,
        V: IndexType,
    {
        let index_id = self
            .solve_index_id(index_name)
            .map_err(|e| PE::PE(IndexOpsError::from(e.error())))?;
        Ok(self
            .persy_impl
            .put::<K::Wrapper, V::Wrapper>(tx_mut(&mut self.tx), index_id, k.wrap(), v.wrap())?)
    }

    /// Remove a key and optionally a specific value from an index following the value mode strategy.
    ///
    /// # Example
    ///
    /// ```rust
    /// # use persy::{OpenOptions, ValueMode};
    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
    /// # let persy = OpenOptions::new().memory()?;
    /// let mut tx = persy.begin()?;
    /// tx.create_index::<u8,u8>("my_new_index", ValueMode::Cluster)?;
    /// tx.put::<u8,u8>("my_new_index",10,10)?;
    /// tx.remove::<u8,u8>("my_new_index",10,Some(10))?;
    /// # tx.prepare()?.commit()?;
    /// # Ok(())
    /// # }
    /// ```
    pub fn remove<K, V>(&mut self, index_name: &str, k: K, v: Option<V>) -> Result<(), PE<IndexOpsError>>
    where
        K: IndexType,
        V: IndexType,
    {
        let index_id = self
            .solve_index_id(index_name)
            .map_err(|e| PE::PE(IndexOpsError::from(e.error())))?;
        Ok(self.persy_impl.remove::<K::Wrapper, V::Wrapper>(
            tx_mut(&mut self.tx),
            index_id,
            k.wrap(),
            v.map(|rv| rv.wrap()),
        )?)
    }

    /// Get a value or a group of values from a key considering changes in transaction.
    ///
    /// # Example
    ///
    /// ```rust
    /// # use persy::{OpenOptions, ValueMode};
    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
    /// # let persy = OpenOptions::new().memory()?;
    /// # let mut tx = persy.begin()?;
    /// # tx.create_index::<u8,u8>("my_new_index", ValueMode::Cluster)?;
    /// tx.put::<u8,u8>("my_new_index",10,10)?;
    /// let values = tx.get::<u8,u8>("my_new_index",&10)?;
    /// for value in values {
    ///  //...
    /// }
    /// # tx.prepare()?.commit()?;
    /// # Ok(())
    /// # }
    /// ```
    pub fn get<K, V>(&mut self, index_name: &str, k: &K) -> Result<ValueIter<V>, PE<IndexChangeError>>
    where
        K: IndexType,
        V: IndexType,
    {
        let index_id = self
            .solve_index_id(index_name)
            .map_err(|e| IndexChangeError::from(e.error()))?;
        let entry: Option<Value<V::Wrapper>> =
            self.persy_impl
                .get_tx::<K::Wrapper, V::Wrapper>(tx_mut(&mut self.tx), index_id, &k.clone().wrap())?;
        Ok(ValueIter::from(entry))
    }

    /// Get one value or none from a key considering changes in transaction.
    ///
    /// # Example
    ///
    /// ```rust
    /// # use persy::{OpenOptions, ValueMode};
    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
    /// # let persy = OpenOptions::new().memory()?;
    /// # let mut tx = persy.begin()?;
    /// # tx.create_index::<u8,u8>("my_new_index", ValueMode::Cluster)?;
    /// tx.put::<u8,u8>("my_new_index",10,10)?;
    /// if let Some(value) =  tx.one::<u8,u8>("my_new_index",&10)?{
    ///  //...
    /// }
    /// # tx.prepare()?.commit()?;
    /// # Ok(())
    /// # }
    /// ```
    pub fn one<K, V>(&mut self, index_name: &str, k: &K) -> Result<Option<V>, PE<IndexChangeError>>
    where
        K: IndexType,
        V: IndexType,
    {
        Ok(self.get(index_name, k)?.next())
    }

    /// Browse a range of keys and values from an index including the transaction changes
    ///
    /// # Example
    ///
    /// ```rust
    /// # use persy::{OpenOptions, ValueMode, TxIndexIter};
    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
    /// # let persy = OpenOptions::new().memory()?;
    /// let mut tx = persy.begin()?;
    /// # tx.create_index::<u8,u8>("my_new_index", ValueMode::Cluster)?;
    /// tx.put::<u8,u8>("my_new_index",10,10)?;
    /// {
    ///     let iter:TxIndexIter<u8,u8> = tx.range("my_new_index",10..12)?;
    ///     for (k,values) in iter  {
    ///         for value in values {
    ///             //...
    ///         }
    ///     }
    /// }
    /// tx.prepare()?.commit()?;
    /// # Ok(())
    /// # }
    /// ```
    pub fn range<'a, K, V, R>(
        &'a mut self,
        index_name: &str,
        range: R,
    ) -> Result<TxIndexIter<'a, K, V>, PE<IndexOpsError>>
    where
        K: IndexType,
        V: IndexType,
        R: RangeBounds<K>,
    {
        let index_id = self
            .solve_index_id(index_name)
            .map_err(|e| IndexOpsError::from(e.error()))?;
        let imp = self.persy_impl.clone();

        let tx = self.tx_mut();
        let rr = PersyImpl::map_index_range_bounds(range);
        let (vm, raw) = imp.range_tx(tx, index_id.clone(), rr.clone())?;
        let tx_iter = tx.index_range::<K::Wrapper, V::Wrapper, _>(index_id.clone(), rr);
        let tx_raw = TxIndexRawIter::new(index_id, index_name.to_owned(), tx_iter, Some(raw), vm);
        Ok(TxIndexIter::new(tx_raw, self))
    }

    /// Rollback a not yet prepared transaction.
    ///
    /// All the resources used for eventual insert or update are released.
    ///
    /// # Example
    ///
    /// ```rust
    /// # use persy::{OpenOptions};
    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
    /// # let persy = OpenOptions::new().memory()?;
    /// let mut tx = persy.begin()?;
    /// # tx.create_segment("seg")?;
    /// let data = vec![1;20];
    /// tx.insert("seg", &data)?;
    /// tx.rollback()?;
    /// # Ok(())
    /// # }
    /// ```
    pub fn rollback(mut self) -> Result<(), PE<GenericError>> {
        if let Some(real_tx) = self.tx.take() {
            self.persy_impl.rollback(real_tx)?;
        }
        Ok(())
    }

    /// Prepare to commit a transaction, when this method return all the validation checks
    /// are done and is guaranteed that the transaction can be committed successfully
    ///
    /// it will lock all the records involved in the transaction
    /// till a [`commit`] or [`rollback`] is called.
    ///
    /// [`commit`]:struct.TransactionFinalize.html#method.commit
    /// [`rollback`]:struct.TransactionFinalize.html#method.rollback
    ///
    /// # Example
    ///
    /// ```rust
    /// # use persy::{OpenOptions};
    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
    /// # let persy = OpenOptions::new().memory()?;
    /// let mut tx = persy.begin()?;
    /// # tx.create_segment("seg")?;
    /// //Do what ever operations on the records
    /// let data = vec![1;20];
    /// tx.insert("seg", &data)?;
    /// tx.prepare()?;
    /// # Ok(())
    /// # }
    /// ```
    pub fn prepare(mut self) -> Result<TransactionFinalize, PE<PrepareError>> {
        let real_tx = self.tx.take().unwrap();
        Ok(TransactionFinalize {
            persy_impl: self.persy_impl.clone(),
            finalize: Some(self.persy_impl.prepare(real_tx)?),
        })
    }

    /// List all the existing segments, considering all the changes in transaction.
    ///
    /// # Example
    ///
    /// ```rust
    /// # use persy::{OpenOptions};
    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
    /// # let persy = OpenOptions::new().memory()?;
    /// let mut tx = persy.begin()?;
    /// tx.create_segment("seg")?;
    /// let segments = tx.list_segments()?;
    /// let names = segments.into_iter().map(|(name,_id)|name).collect::<Vec<String>>();
    /// assert!(names.contains(&"seg".to_string()));
    /// tx.prepare()?.commit()?;
    /// # Ok(())
    /// # }
    /// ```
    pub fn list_segments(&self) -> Result<Vec<(String, SegmentId)>, GenericError> {
        self.persy_impl.list_segments_tx(self.tx())
    }

    /// List all the existing indexes, considering changes in the transaction.
    ///
    /// # Example
    ///
    /// ```rust
    /// # use persy::{OpenOptions, ValueMode};
    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
    /// # let persy = OpenOptions::new().memory()?;
    /// let mut tx = persy.begin()?;
    /// tx.create_index::<u8, u8>("idx", ValueMode::Replace)?;
    /// let indexes = tx.list_indexes()?;
    /// let names = indexes.into_iter().map(|(name,_id)|name).collect::<Vec<String>>();
    /// assert!(names.contains(&"idx".to_string()));
    /// tx.prepare()?.commit()?;
    /// # Ok(())
    /// # }
    /// ```
    pub fn list_indexes(&self) -> Result<Vec<(String, IndexInfo)>, PE<GenericError>> {
        Ok(self.persy_impl.list_indexes_tx(self.tx())?)
    }

    /// Prepare and Commit a transaction
    ///
    /// # Example
    ///
    /// ```rust
    /// # use persy::{OpenOptions};
    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
    /// # let persy = OpenOptions::new().memory()?;
    /// let mut tx = persy.begin()?;
    /// # tx.create_segment("seg")?;
    /// //Do what ever operations on the records
    /// let data = vec![1;20];
    /// tx.insert("seg", &data)?;
    /// tx.commit()?;
    /// # Ok(())
    /// # }
    /// ```
    pub fn commit(mut self) -> Result<(), PE<PrepareError>> {
        let real_tx = self.tx.take().unwrap();
        let mut finalize = self.persy_impl.prepare(real_tx)?;
        self.persy_impl.commit(&mut finalize)?;
        Ok(())
    }
}

impl Drop for Transaction {
    fn drop(&mut self) {
        if let Some(tx) = self.tx.take() {
            self.persy_impl.rollback(tx).unwrap();
        }
    }
}

/// prepared transaction state
#[must_use]
pub struct TransactionFinalize {
    persy_impl: Arc<PersyImpl>,
    finalize: Option<TxFinalize>,
}

impl TransactionFinalize {
    /// Rollback a prepared commit.
    ///
    /// All the modification are rolled back and all the used resources are put released
    ///
    /// # Example
    ///
    /// ```rust
    /// # use persy::{OpenOptions};
    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
    /// # let persy = OpenOptions::new().memory()?;
    /// # let mut tx = persy.begin()?;
    /// # tx.create_segment("seg")?;
    /// # tx.prepare()?.commit()?;
    /// let mut tx = persy.begin()?;
    /// //Do what ever operations on the records
    /// let data = vec![1;20];
    /// tx.insert("seg", &data)?;
    /// let prepared = tx.prepare()?;
    /// prepared.rollback()?;
    /// # Ok(())
    /// # }
    /// ```
    pub fn rollback(mut self) -> Result<(), PE<GenericError>> {
        if let Some(mut finalize) = self.finalize.take() {
            self.persy_impl.rollback_prepared(&mut finalize)?;
        }
        Ok(())
    }

    /// Finalize the commit result of a prepared commit.
    ///
    /// All the operation done on the transaction are finalized all the lock released, all the
    /// old resources are released for reuse.
    ///
    /// # Example
    ///
    /// ```rust
    /// # use persy::{OpenOptions};
    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
    /// # let persy = OpenOptions::new().memory()?;
    /// # let mut tx = persy.begin()?;
    /// # tx.create_segment("seg")?;
    /// # tx.prepare()?.commit()?;
    /// let mut tx = persy.begin()?;
    /// let prepared = tx.prepare()?;
    /// prepared.commit()?;
    /// # Ok(())
    /// # }
    /// ```
    pub fn commit(mut self) -> Result<(), PE<GenericError>> {
        if let Some(mut finalize) = self.finalize.take() {
            self.persy_impl.commit(&mut finalize)?;
        }
        Ok(())
    }
}

impl Drop for TransactionFinalize {
    fn drop(&mut self) {
        if let Some(mut finalize) = self.finalize.take() {
            self.persy_impl.rollback_prepared(&mut finalize).unwrap();
        }
    }
}

#[cfg(test)]
mod tests {
    use crate::{Config, Persy, TransactionConfig};
    use std::fs;
    #[test]
    pub fn test_recover_prepared_tx() {
        Persy::create("./target/test_recover_prepared.persy").unwrap();
        let id;
        let val;
        {
            let persy = Persy::open("./target/test_recover_prepared.persy", Config::new()).unwrap();
            let mut tx = persy.begin().expect("error on transactoin begin");
            tx.create_segment("def").expect("error on segment creation");
            let fin = tx.prepare().expect("error on commit prepare");
            fin.commit().expect("error on commit");

            let mut tx = persy.begin().expect("error on transaction begin");
            val = String::from("aaa").into_bytes();
            id = tx.insert("def", &val).expect("error on insert value");
            let mut prepared = tx.prepare().expect("error on commit prepare");
            let _replaced = std::mem::replace(&mut prepared.finalize, None);
        }
        {
            let persy = Persy::open("./target/test_recover_prepared.persy", Config::new()).unwrap();
            assert_eq!(persy.read("def", &id).expect("error reading record"), Some(val));
        }

        fs::remove_file("./target/test_recover_prepared.persy").unwrap();
    }

    #[test]
    pub fn test_dobule_recover_prepared_tx() {
        Persy::create("./target/test_double_recover_prepared.persy").unwrap();
        let id;
        let id1;
        let val;
        let val1;
        {
            let persy = Persy::open("./target/test_double_recover_prepared.persy", Config::new()).unwrap();
            let mut tx = persy.begin().expect("error on transactoin begin");
            tx.create_segment("def").expect("error on segment creation");
            let fin = tx.prepare().expect("error on commit prepare");
            fin.commit().expect("error on commit");

            let mut tx = persy.begin().expect("error on transaction begin");
            val = String::from("aaa").into_bytes();
            id = tx.insert("def", &val).expect("error on insert value");
            let mut prepared = tx.prepare().expect("error on commit prepare");
            let _replaced = std::mem::replace(&mut prepared.finalize, None);
        }
        {
            let persy = Persy::open("./target/test_double_recover_prepared.persy", Config::new()).unwrap();
            assert_eq!(persy.read("def", &id).expect("error reading record"), Some(val.clone()));
            let mut tx = persy.begin().expect("error on transaction begin");
            val1 = String::from("bbbb").into_bytes();
            id1 = tx.insert("def", &val1).expect("error on insert value");
            let mut prepared = tx.prepare().expect("error on commit prepare");
            let _replaced = std::mem::replace(&mut prepared.finalize, None);
        }

        {
            let persy = Persy::open("./target/test_double_recover_prepared.persy", Config::new()).unwrap();
            assert_eq!(persy.read("def", &id).expect("error reading record"), Some(val));
            assert_eq!(persy.read("def", &id1).expect("error reading record",), Some(val1));
        }
        fs::remove_file("./target/test_double_recover_prepared.persy").unwrap();
    }

    #[test]
    pub fn test_recover_tx_id() {
        Persy::create("./target/test_recover_tx_id.persy").unwrap();
        let id;
        let id_pers;
        let id_pers_update;
        let val;
        let val_1;
        let tx_id = vec![10; 5];
        {
            val = String::from("aaa").into_bytes();
            let persy = Persy::open("./target/test_recover_tx_id.persy", Config::new()).unwrap();
            let mut tx = persy.begin().expect("error on transactoin begin");
            tx.create_segment("def").expect("error on segment creation");
            id_pers = tx.insert("def", &val).expect("error on insert value");
            id_pers_update = tx.insert("def", &val).expect("error on insert value");
            let fin = tx.prepare().expect("error on commit prepare");
            fin.commit().expect("error on commit");

            let mut tx = persy
                .begin_with(TransactionConfig::new().set_transaction_id(tx_id.clone()))
                .expect("error on transaction begin");
            id = tx.insert("def", &val).expect("error on insert value");
            tx.delete("def", &id_pers).expect("delete record works");
            val_1 = String::from("bbb").into_bytes();
            tx.update("def", &id_pers_update, &val_1).expect("delete record works");
            let mut prepared = tx.prepare().expect("error on commit prepare");
            let _replaced = std::mem::replace(&mut prepared.finalize, None);
        }
        {
            let persy = Persy::open_with_recover("./target/test_recover_tx_id.persy", Config::new(), |t_id| {
                assert_eq!(&tx_id, t_id);
                true
            })
            .unwrap();
            assert_eq!(persy.read("def", &id).expect("error reading record"), Some(val));
            assert_eq!(persy.read("def", &id_pers).expect("error reading record"), None);
            assert_eq!(
                persy.read("def", &id_pers_update).expect("error reading record"),
                Some(val_1)
            );
        }

        fs::remove_file("./target/test_recover_tx_id.persy").unwrap();
    }

    #[test]
    pub fn test_recover_tx_choice() {
        Persy::create("./target/test_recover_tx_choice.persy").unwrap();
        let id;
        let id_1;
        let id_pers;
        let id_pers_update;
        let val;
        let val_1;
        let tx_id = vec![10; 5];
        let tx_id_1 = vec![10; 10];
        {
            val = String::from("aaa").into_bytes();
            let persy = Persy::open("./target/test_recover_tx_choice.persy", Config::new()).unwrap();
            let mut tx = persy.begin().expect("error on transactoin begin");
            tx.create_segment("def").expect("error on segment creation");
            id_pers = tx.insert("def", &val).expect("error on insert value");
            id_pers_update = tx.insert("def", &val).expect("error on insert value");
            let fin = tx.prepare().expect("error on commit prepare");
            fin.commit().expect("error on commit");

            let mut tx = persy
                .begin_with(TransactionConfig::new().set_transaction_id(tx_id.clone()))
                .expect("error on transaction begin");
            id = tx.insert("def", &val).expect("error on insert value");
            let mut prepared = tx.prepare().expect("error on commit prepare");
            let _replaced = std::mem::replace(&mut prepared.finalize, None);

            let mut tx = persy
                .begin_with(TransactionConfig::new().set_transaction_id(tx_id_1.clone()))
                .expect("error on transaction begin");
            id_1 = tx.insert("def", &val).expect("error on insert value");
            tx.delete("def", &id_pers).expect("delete record works");
            val_1 = String::from("bbb").into_bytes();
            tx.update("def", &id_pers_update, &val_1).expect("delete record works");
            let mut prepared = tx.prepare().expect("error on commit prepare");
            let _replaced = std::mem::replace(&mut prepared.finalize, None);
        }
        {
            let mut recover = Persy::recover("./target/test_recover_tx_choice.persy", Config::new()).unwrap();

            assert!(recover.list_transactions().len() >= 2);
            recover.rollback(tx_id_1);
            recover.commit(tx_id);
            let persy = recover.finalize().expect("recover correctly");

            assert_eq!(persy.read("def", &id).expect("error reading record"), Some(val.clone()));
            assert_eq!(persy.read("def", &id_1).expect("error reading record"), None);
            assert_eq!(
                persy.read("def", &id_pers).expect("error reading record"),
                Some(val.clone())
            );
            assert_eq!(
                persy.read("def", &id_pers_update).expect("error reading record"),
                Some(val)
            );
        }

        fs::remove_file("./target/test_recover_tx_choice.persy").unwrap();
    }
}