use std::cmp::min;
use std::mem;
use std::sync::Arc;

use fluvio_future::file_slice::AsyncFileSlice;
use fluvio_protocol::Encoder;
use tracing::{debug, trace, warn, instrument, info};
use async_trait::async_trait;

use fluvio_future::fs::{create_dir_all, remove_dir_all};
use dataplane::{ErrorCode, Isolation, Offset, ReplicaKey, Size};
use dataplane::batch::{Batch, BatchRecords};
use dataplane::record::RecordSet;

use crate::{OffsetInfo, checkpoint::CheckPoint};
use crate::segments::SharedSegments;
use crate::segment::MutableSegment;
use crate::config::{ReplicaConfig, SharedReplicaConfig, StorageConfig};
use crate::ReplicaSlice;
use crate::{StorageError, ReplicaStorage};

/// Replica is public abstraction for commit log which are distributed.
/// Internally it is stored as list of segments.  Each segment contains finite sets of record batches.
///
#[derive(Debug)]
pub struct FileReplica {
    #[allow(dead_code)]
    last_base_offset: Offset,
    #[allow(dead_code)]
    partition: Size,
    option: Arc<SharedReplicaConfig>,
    active_segment: MutableSegment,
    prev_segments: Arc<SharedSegments>,
    commit_checkpoint: CheckPoint<Offset>,
}

impl Unpin for FileReplica {}

#[async_trait]
impl ReplicaStorage for FileReplica {
    type ReplicaConfig = ReplicaConfig;

    async fn create_or_load(
        replica: &ReplicaKey,
        replica_config: Self::ReplicaConfig,
    ) -> Result<Self, StorageError> {
        let storage_config = StorageConfig::builder().build().map_err(|err| {
            StorageError::Other(format!("failed to build cleaner config: {}", err))
        })?;

        Self::create_or_load(
            replica.topic.clone(),
            replica.partition as u32,
            0,
            replica_config,
            Arc::new(storage_config),
        )
        .await
    }

    #[inline(always)]
    fn get_hw(&self) -> Offset {
        *self.commit_checkpoint.get_offset()
    }

    /// offset mark that beginning of uncommitted
    #[inline(always)]
    fn get_leo(&self) -> Offset {
        self.active_segment.get_end_offset()
    }

    /// earliest offset
    fn get_log_start_offset(&self) -> Offset {
        let min_base_offset = self.prev_segments.min_offset();
        if min_base_offset < 0 {
            self.active_segment.get_base_offset()
        } else {
            min_base_offset
        }
    }

    /// read partition slice
    /// return leo, hw
    #[instrument(skip(self, offset, max_len, isolation))]
    async fn read_partition_slice(
        &self,
        offset: Offset,
        max_len: u32,
        isolation: Isolation,
    ) -> Result<ReplicaSlice, ErrorCode> {
        match isolation {
            Isolation::ReadCommitted => {
                self.read_records(offset, Some(self.get_hw()), max_len)
                    .await
            }
            Isolation::ReadUncommitted => self.read_records(offset, None, max_len).await,
        }
    }

    /// reads log file of the current partition
    /// return the size in byte
    #[instrument(skip(self))]
    async fn get_partition_size(&self) -> Result<u64, ErrorCode> {
        let active_len = self.active_segment.get_msg_log().get_pos();

        debug!(active_len, "Active segment length");
        let total_prev_segments_len = {
            let reader = self.prev_segments.read().await;
            reader.get_total_logs_len() as u32
        };
        debug!(
            total_prev_segments_len,
            "Cumulated previous segments length"
        );
        let total_len = active_len + total_prev_segments_len;

        return Ok(total_len.into());
    }

    /// write records to this replica
    /// if update_highwatermark is set, set high watermark is end
    //  this is used when LRS = 1
    #[instrument(skip(self, records, update_highwatermark))]
    async fn write_recordset<R: BatchRecords>(
        &mut self,
        records: &mut RecordSet<R>,
        update_highwatermark: bool,
    ) -> Result<(), StorageError> {
        let max_batch_size = self.option.max_batch_size.get() as usize;
        // check if any of the records's batch exceed max length
        for batch in &records.batches {
            if batch.write_size(0) > max_batch_size {
                return Err(StorageError::BatchTooBig(max_batch_size));
            }
        }

        for batch in &mut records.batches {
            self.write_batch(batch).await?;
        }

        if update_highwatermark {
            self.update_high_watermark_to_end().await?;
        }
        Ok(())
    }

    /// update committed offset (high watermark)
    /// if true, hw is updated
    #[instrument(skip(self, offset))]
    async fn update_high_watermark(&mut self, offset: Offset) -> Result<bool, StorageError> {
        let old_offset = self.get_hw();
        if old_offset == offset {
            trace!(
                "new high watermark: {} is same as existing one, skipping",
                offset
            );
            Ok(false)
        } else {
            trace!(
                "updating to new high watermark: {} old: {}",
                old_offset,
                offset
            );
            self.commit_checkpoint.write(offset).await?;
            Ok(true)
        }
    }

    #[instrument(skip(self))]
    async fn remove(&self) -> Result<(), StorageError> {
        remove_dir_all(&self.option.base_dir)
            .await
            .map_err(StorageError::Io)?;
        self.prev_segments.clean();
        Ok(())
    }
}

impl FileReplica {
    pub const PREFER_MAX_LEN: u32 = 1000000; // 1MB as limit

    /// Construct a new replica with specified topic and partition.
    /// It can start with arbitrary offset.  However, for normal replica,
    /// it is usually starts with 0.  
    ///
    /// Replica is minimum unit of logs that will can be replicated.  
    /// It is a unique pair of (topic,partition)
    ///
    /// Replica will use base directory to create it's own directory.
    /// Directory name will encode unique replica id which is combination of topic
    /// and partition.
    ///
    /// If there is existing directory then it will load existing logs.
    /// The logs will be validated to ensure it's safe to use it.
    /// It is possible logs can't be used because they may be corrupted.
    #[instrument(skip(topic, partition, base_offset, replica_config, storage_config))]
    pub async fn create_or_load<S>(
        topic: S,
        partition: Size,
        base_offset: Offset,
        replica_config: ReplicaConfig,
        storage_config: Arc<StorageConfig>,
    ) -> Result<FileReplica, StorageError>
    where
        S: AsRef<str> + Send + 'static,
    {
        let replica_dir = replica_config
            .base_dir
            .join(replica_dir_name(topic, partition));

        info!("creating rep dir: {}", replica_dir.display());
        debug!("replica config: {:?}", replica_config);
        create_dir_all(&replica_dir).await?; // ensure dir_name exits

        let mut rep_option = replica_config.clone();
        rep_option.base_dir = replica_dir;

        let shared_config: Arc<SharedReplicaConfig> = Arc::new(rep_option.into());

        let (segments, last_offset_res) =
            SharedSegments::from_dir(shared_config.clone(), storage_config).await?;

        let active_segment = if let Some(last_offset) = last_offset_res {
            debug!(last_offset, "last segment found, validating offsets");
            let mut last_segment =
                MutableSegment::open_for_write(last_offset, shared_config.clone()).await?;
            last_segment.validate(false, false).await?;
            info!(
                end_offset = last_segment.get_end_offset(),
                "existing segment validated with last offset",
            );
            last_segment
        } else {
            info!("no existing segment found, creating new one");
            MutableSegment::create(base_offset, shared_config.clone()).await?
        };

        let last_base_offset = active_segment.get_base_offset();

        let commit_checkpoint: CheckPoint<Offset> =
            CheckPoint::create(shared_config.clone(), "replication.chk", last_base_offset).await?;

        Ok(Self {
            option: shared_config,
            last_base_offset,
            partition,
            active_segment,
            prev_segments: segments,
            commit_checkpoint,
        })
    }

    /// clear the any holding directory for replica
    #[instrument(skip(replica, option))]
    pub async fn clear(replica: &ReplicaKey, option: &SharedReplicaConfig) {
        let replica_dir = option
            .base_dir
            .join(replica_dir_name(&replica.topic, replica.partition as u32));

        info!("removing dir: {}", replica_dir.display());
        if let Err(err) = remove_dir_all(&replica_dir).await {
            warn!("error trying to remove: {:#?}, err: {}", replica_dir, err);
        }
    }

    /// update high watermark to end
    #[instrument(skip(self))]
    pub async fn update_high_watermark_to_end(&mut self) -> Result<bool, StorageError> {
        self.update_high_watermark(self.get_leo()).await
    }

    /// read all uncommitted records
    #[allow(unused)]
    #[instrument(skip(self, max_len))]
    pub async fn read_all_uncommitted_records(
        &self,
        max_len: u32,
    ) -> Result<ReplicaSlice, ErrorCode> {
        self.read_records(self.get_hw(), None, max_len).await
    }

    /// read record slice into response
    /// * `start_offset`:  start offsets
    /// * `max_offset`:  max offset (exclusive)
    /// * `responsive`:  output
    /// * `max_len`:  max length of the slice
    //  return leo, hw
    #[instrument(skip(self))]
    async fn read_records(
        &self,
        start_offset: Offset,
        max_offset: Option<Offset>,
        max_len: u32,
    ) -> Result<ReplicaSlice, ErrorCode> {
        let hw = self.get_hw();
        let leo = self.get_leo();
        debug!(hw, leo, "starting read records",);

        let mut slice = ReplicaSlice {
            end: OffsetInfo { hw, leo },
            start: self.get_log_start_offset(),
            ..Default::default()
        };

        let active_base_offset = self.active_segment.get_base_offset();
        let file_slice = if start_offset >= active_base_offset {
            debug!(start_offset, active_base_offset, "is in active segment");
            if start_offset == leo {
                trace!("start offset is same as end offset, skipping");
                return Ok(slice);
            } else if start_offset > leo {
                return Err(ErrorCode::Other(format!(
                    "start offset: {} is greater than leo: {}",
                    start_offset, leo
                )));
            } else if let Some(slice) = self
                .active_segment
                .records_slice(start_offset, max_offset)
                .await?
            {
                slice
            } else {
                return Err(ErrorCode::Other(format!(
                    "no records found in active replica, start: {}, max: {:#?}, active: {:#?}",
                    start_offset, max_offset, self.active_segment
                )));
            }
        } else {
            debug!(start_offset, active_base_offset, "not in active sgments");
            self.prev_segments
                .find_slice(start_offset, max_offset)
                .await?
        };

        let limited_slice = AsyncFileSlice::new(
            file_slice.fd(),
            file_slice.position(),
            min(file_slice.len(), max_len as u64),
        );

        debug!(
            fd = limited_slice.fd(),
            pos = limited_slice.position(),
            len = limited_slice.len(),
            "retrieved slice",
        );

        slice.file_slice = Some(limited_slice);
        Ok(slice)
    }

    #[instrument(skip(self, item))]
    async fn write_batch<R: BatchRecords>(
        &mut self,
        item: &mut Batch<R>,
    ) -> Result<(), StorageError> {
        if !(self.active_segment.append_batch(item).await?) {
            info!(
                partition = self.partition,
                path = %self.option.base_dir.display(),
                base_offset = self.active_segment.get_base_offset(),
                "rolling over active segment");
            self.active_segment.roll_over().await?;
            let last_offset = self.active_segment.get_end_offset();
            let new_segment = MutableSegment::create(last_offset, self.option.clone()).await?;
            let old_mut_segment = mem::replace(&mut self.active_segment, new_segment);
            let old_segment = old_mut_segment.as_segment().await?;
            self.prev_segments.add_segment(old_segment).await;
            self.active_segment.append_batch(item).await?;
        }
        Ok(())
    }
}

// generate replication folder name
fn replica_dir_name<S: AsRef<str>>(topic_name: S, partition_index: Size) -> String {
    format!("{}-{}", topic_name.as_ref(), partition_index)
}

#[cfg(test)]
mod tests {

    use fluvio_future::fs::remove_dir_all;
    use fluvio_future::timer::sleep;
    use tracing::debug;
    use std::env::temp_dir;
    use std::fs;
    use std::fs::metadata;
    use std::io::Cursor;
    use std::sync::Arc;
    use std::time::Duration;

    use dataplane::{Isolation, batch::Batch};
    use dataplane::Offset;
    use dataplane::core::{Decoder, Encoder};
    use dataplane::record::{Record, RecordSet};
    use dataplane::batch::MemoryRecords;
    use dataplane::fixture::{BatchProducer, create_batch};
    use dataplane::fixture::read_bytes_from_file;
    use flv_util::fixture::ensure_clean_dir;

    use crate::config::{ReplicaConfig, StorageConfig};
    use crate::StorageError;
    use crate::ReplicaStorage;
    use crate::fixture::storage_config;

    use super::FileReplica;

    const TEST_SEG_NAME: &str = "00000000000000000020.log";
    const TEST_SE2_NAME: &str = "00000000000000000022.log";
    const TEST_SEG_IDX: &str = "00000000000000000020.index";
    const TEST_SEG2_IDX: &str = "00000000000000000022.index";
    const START_OFFSET: Offset = 20;

    async fn create_replica(
        topic: &'static str,
        base_offset: Offset,
        config: ReplicaConfig,
    ) -> FileReplica {
        FileReplica::create_or_load(topic, 0, base_offset, config, storage_config())
            .await
            .expect("replica")
    }

    /// create option, ensure they are clean
    fn base_option(dir: &str) -> ReplicaConfig {
        let base_dir = temp_dir().join(dir);
        ensure_clean_dir(&base_dir);
        ReplicaConfig {
            segment_max_bytes: 10000,
            base_dir,
            index_max_interval_bytes: 1000,
            index_max_bytes: 1000,
            ..Default::default()
        }
    }

    fn rollover_option(dir: &str) -> ReplicaConfig {
        let base_dir = temp_dir().join(dir);
        ensure_clean_dir(&base_dir);
        ReplicaConfig {
            segment_max_bytes: 100,
            base_dir,
            index_max_bytes: 1000,
            index_max_interval_bytes: 0,
            ..Default::default()
        }
    }

    #[fluvio_future::test]
    async fn test_replica_simple() {
        let option = base_option("test_simple");

        let mut replica = create_replica("test", START_OFFSET, option.clone()).await;

        assert_eq!(replica.get_log_start_offset(), START_OFFSET);
        assert_eq!(replica.get_leo(), START_OFFSET);
        assert_eq!(replica.get_hw(), START_OFFSET);

        replica
            .write_batch(&mut create_batch())
            .await
            .expect("send");
        assert_eq!(replica.get_leo(), START_OFFSET + 2); // 2 batches
        assert_eq!(replica.get_hw(), START_OFFSET); // hw should not change since we have not committed them

        replica
            .update_high_watermark(10)
            .await
            .expect("high watermaerk");
        assert_eq!(replica.get_hw(), 10); // hw should set to whatever we passed

        let test_file = option.base_dir.join("test-0").join(TEST_SEG_NAME);
        debug!("using test file: {:#?}", test_file);
        let bytes = read_bytes_from_file(&test_file).expect("read");

        let batch =
            Batch::<MemoryRecords>::decode_from(&mut Cursor::new(bytes), 0).expect("decode");
        assert_eq!(batch.get_header().magic, 2, "check magic");
        assert_eq!(batch.get_base_offset(), START_OFFSET);
        assert_eq!(batch.get_header().last_offset_delta, 1);
        assert_eq!(batch.records().len(), 2);

        // there should not be any segment for offset 0 since base offset is 20
        let reader = replica.prev_segments.read().await;
        assert!(reader.find_segment(0).is_none());

        //println!("segments:")
        // segment with offset 20 should be active segment
        assert!(reader.find_segment(20).is_none());
        assert!(reader.find_segment(21).is_none());
        assert!(reader.find_segment(30).is_none()); // any h
    }

    const TEST_UNCOMMIT_DIR: &str = "test_uncommitted";

    #[fluvio_future::test]
    async fn test_uncommitted_fetch() {
        let option = base_option(TEST_UNCOMMIT_DIR);

        let mut replica = create_replica("test", 0, option).await;

        assert_eq!(replica.get_leo(), 0);
        assert_eq!(replica.get_hw(), 0);

        // reading empty replica should return empyt records

        let slice = replica
            .read_all_uncommitted_records(FileReplica::PREFER_MAX_LEN)
            .await
            .expect("read");
        assert!(slice.file_slice.is_none());

        // write batches
        let mut batch = create_batch();
        let batch_len = batch.write_size(0);
        debug!("batch len: {}", batch_len);
        replica.write_batch(&mut batch).await.expect("write");

        assert_eq!(replica.get_leo(), 2); // 2
        assert_eq!(replica.get_hw(), 0);

        // read records
        let slice = replica
            .read_all_uncommitted_records(FileReplica::PREFER_MAX_LEN)
            .await
            .expect("read");
        assert_eq!(slice.file_slice.unwrap().len() as usize, batch_len);

        replica.update_high_watermark(2).await.expect("update"); // first batch
        assert_eq!(replica.get_hw(), 2);

        let mut batch = create_batch();
        let batch_len = batch.write_size(0);
        replica.write_batch(&mut batch).await.expect("write");

        let slice = replica
            .read_all_uncommitted_records(FileReplica::PREFER_MAX_LEN)
            .await
            .expect("read");
        debug!("slice: {:#?}", slice);
        assert_eq!(slice.file_slice.unwrap().len() as usize, batch_len);

        replica
            .write_batch(&mut create_batch())
            .await
            .expect("write");

        let slice = replica
            .read_all_uncommitted_records(FileReplica::PREFER_MAX_LEN)
            .await
            .expect("read");
        assert_eq!(slice.file_slice.unwrap().len() as usize, batch_len * 2);

        let slice = replica
            .read_all_uncommitted_records(50)
            .await
            .expect("read");
        assert_eq!(slice.file_slice.unwrap().len(), 50);
    }

    const TEST_OFFSET_DIR: &str = "test_offset";

    #[fluvio_future::test]
    async fn test_replica_end_offset() {
        let option = base_option(TEST_OFFSET_DIR);

        let mut rep_sink = create_replica("test", START_OFFSET, option.clone()).await;
        rep_sink
            .write_batch(&mut create_batch())
            .await
            .expect("write");
        rep_sink
            .write_batch(&mut create_batch())
            .await
            .expect("write");
        drop(rep_sink);

        // open replica
        let replica2 = create_replica("test", 0, option).await;
        assert_eq!(replica2.get_leo(), START_OFFSET + 4);
    }

    const TEST_REPLICA_DIR: &str = "test_replica";

    // you can show log by:  RUST_LOG=commit_log=debug cargo test roll_over

    #[fluvio_future::test]
    async fn test_rep_log_roll_over() {
        let option = rollover_option(TEST_REPLICA_DIR);

        let mut replica =
            FileReplica::create_or_load("test", 1, START_OFFSET, option.clone(), storage_config())
                .await
                .expect("create rep");

        // first batch
        debug!(">>>> sending first batch");
        let mut batches = create_batch();
        replica.write_batch(&mut batches).await.expect("write");

        // second batch
        debug!(">>>> sending second batch. this should rollover");
        let mut batches = create_batch();
        replica.write_batch(&mut batches).await.expect("write");
        debug!("finish sending next batch");

        assert_eq!(replica.get_log_start_offset(), START_OFFSET);
        let replica_dir = &option.base_dir.join("test-1");
        let dir_contents = fs::read_dir(&replica_dir).expect("read_dir");
        assert_eq!(dir_contents.count(), 5, "should be 5 files");

        let seg2_file = replica_dir.join(TEST_SE2_NAME);
        let bytes = read_bytes_from_file(&seg2_file).expect("file read");

        let batch =
            Batch::<MemoryRecords>::decode_from(&mut Cursor::new(bytes), 0).expect("decode");
        assert_eq!(batch.get_header().magic, 2, "check magic");
        assert_eq!(batch.records().len(), 2);
        assert_eq!(batch.get_base_offset(), 22);

        let metadata_res = metadata(replica_dir.join(TEST_SEG2_IDX));
        assert!(metadata_res.is_ok());
        let metadata2 = metadata_res.unwrap();
        assert_eq!(metadata2.len(), 1000);

        let seg1_metadata = metadata(replica_dir.join(TEST_SEG_IDX)).expect("metadata");
        assert_eq!(seg1_metadata.len(), 8);
    }

    #[fluvio_future::test]
    async fn test_replica_commit() {
        let option = base_option("test_commit");
        let mut replica = create_replica("test", 0, option.clone()).await;

        let mut records = RecordSet::default().add(create_batch());

        replica
            .write_recordset(&mut records, true)
            .await
            .expect("write");

        // record contains 2 batch
        assert_eq!(replica.get_hw(), 2);

        drop(replica);

        // restore replica
        let replica = create_replica("test", 0, option).await;
        assert_eq!(replica.get_hw(), 2);
    }

    const TEST_STORAGE_SIZE_DIR: &str = "test_storage_size";

    #[fluvio_future::test]
    async fn test_replica_storage_size() {
        let option = base_option(TEST_STORAGE_SIZE_DIR);
        let mut replica = create_replica("test", 0, option.clone()).await;

        let mut records = RecordSet::default().add(create_batch());

        replica
            .write_recordset(&mut records, true)
            .await
            .expect("write");

        assert_eq!(replica.get_hw(), 2);

        let size = replica.get_partition_size().await.expect("partition size");
        assert_eq!(size, 79);

        replica
            .write_recordset(&mut records, true)
            .await
            .expect("write");

        let size = replica.get_partition_size().await.expect("partition size");
        assert_eq!(size, 79 * 2);
    }

    #[fluvio_future::test]
    async fn test_replica_storage_size_delete_log() {
        let option = base_option("test_storage_size_deleted");
        let mut replica = create_replica("test", 0, option.clone()).await;

        let mut records = RecordSet::default().add(create_batch());

        replica
            .write_recordset(&mut records, true)
            .await
            .expect("write");

        assert_eq!(replica.get_hw(), 2);

        let size = replica.get_partition_size().await.expect("partition size");
        assert_eq!(size, 79);

        remove_dir_all(option.base_dir)
            .await
            .expect("delete base dir");

        let size = replica
            .get_partition_size()
            .await
            .expect("error partition size");
        assert_eq!(size, 79);
    }

    /// test fetch only committed records
    #[fluvio_future::test]
    async fn test_committed_fetch() {
        let option = base_option("test_commit_fetch");

        let mut replica = create_replica("test", 0, option).await;

        let mut batch = create_batch();
        let batch_len = batch.write_size(0);
        replica
            .write_batch(&mut batch)
            .await
            .expect("writing records");

        let slice = replica
            .read_partition_slice(0, FileReplica::PREFER_MAX_LEN, Isolation::ReadCommitted)
            .await
            .expect("read");
        debug!("slice response: {:#?}", slice);
        assert_eq!(slice.file_slice.unwrap().len(), 0);

        replica
            .update_high_watermark_to_end()
            .await
            .expect("update high watermark");

        debug!(
            "replica end: {} high: {}",
            replica.get_leo(),
            replica.get_hw()
        );

        let slice = replica
            .read_partition_slice(0, FileReplica::PREFER_MAX_LEN, Isolation::ReadCommitted)
            .await
            .expect("read");
        debug!("slice response: {:#?}", slice);
        assert_eq!(slice.file_slice.unwrap().len() as usize, batch_len);

        // write 1 more batch
        let mut batch = create_batch();
        replica
            .write_batch(&mut batch)
            .await
            .expect("writing 2nd batch");
        debug!(
            "2nd batch: replica end: {} high: {}",
            replica.get_leo(),
            replica.get_hw()
        );

        let slice = replica
            .read_partition_slice(0, FileReplica::PREFER_MAX_LEN, Isolation::ReadCommitted)
            .await
            .expect("read");
        debug!("slice response: {:#?}", slice);
        // should return same records as 1 batch since we didn't commit 2nd batch
        assert_eq!(slice.file_slice.unwrap().len() as usize, batch_len);
    }

    #[fluvio_future::test]
    async fn test_replica_delete() {
        let mut option = base_option("test_delete");
        option.max_batch_size = 50; // enforce 50 length

        let replica = create_replica("testr", START_OFFSET, option.clone()).await;

        let test_file = option.base_dir.join("testr-0").join(TEST_SEG_NAME);

        assert!(test_file.exists());

        replica.remove().await.expect("removed");

        assert!(!test_file.exists());
    }

    #[fluvio_future::test]
    async fn test_replica_limit_batch() {
        let mut option = base_option("test_batch_limit");
        option.max_batch_size = 100;
        option.update_hw = false;

        let mut replica = create_replica("test", START_OFFSET, option).await;

        let mut small_batch = BatchProducer::builder().build().expect("batch").records();
        assert!(small_batch.write_size(0) < 100); // ensure we are writing less than 100 bytes
        replica
            .write_recordset(&mut small_batch, true)
            .await
            .expect("writing records");

        let mut larget_batch = BatchProducer::builder()
            .per_record_bytes(200)
            .build()
            .expect("batch")
            .records();
        assert!(larget_batch.write_size(0) > 100); // ensure we are writing more than 100
        assert!(matches!(
            replica
                .write_recordset(&mut larget_batch, true)
                .await
                .unwrap_err(),
            StorageError::BatchTooBig(_)
        ));
    }

    /// create replicat with multiple segments
    #[fluvio_future::test]
    async fn test_replica_multiple_segment() {
        let mut option = base_option("test_find_segment");
        // enough for 2 batch (2 records per batch)
        option.segment_max_bytes = 160;
        option.index_max_interval_bytes = 50; // ensure we are writing to index

        let producer = BatchProducer::builder()
            .records(2u16)
            .record_generator(Arc::new(|_, _| Record::new("1")))
            .build()
            .expect("batch");

        let mut new_replica = create_replica("test", 0, option.clone()).await;
        let reader = new_replica.prev_segments.read().await;
        assert!(reader.len() == 0);
        drop(reader);

        // this will create  1 segment
        new_replica
            .write_batch(&mut producer.generate_batch())
            .await
            .expect("write");
        new_replica
            .write_batch(&mut producer.generate_batch())
            .await
            .expect("write");

        let reader = new_replica.prev_segments.read().await;
        assert_eq!(reader.len(), 0);
        assert!(reader.find_segment(0).is_none());
        assert!(reader.find_segment(1).is_none());
        drop(reader);

        // overflow, will create 2nd segment
        new_replica
            .write_batch(&mut producer.generate_batch())
            .await
            .expect("write");

        assert_eq!(new_replica.prev_segments.min_offset(), 0);
        let reader = new_replica.prev_segments.read().await;
        assert_eq!(reader.len(), 1);

        //    println!("new replica segments: {:#?}", new_replica.prev_segments);
        let first_segment = reader.get_segment(0).expect("some");
        assert_eq!(first_segment.get_base_offset(), 0);
        assert_eq!(first_segment.get_end_offset(), 4);
        assert!(reader.find_segment(0).is_some());
        drop(reader);
        drop(new_replica);

        // reload replica
        let old_replica = create_replica("test", 0, option.clone()).await;
        let reader = old_replica.prev_segments.read().await;
        //  println!("old replica segments: {:#?}", old_replica.prev_segments);
        assert!(reader.len() == 1);
        let (_, segment) = reader.find_segment(0).expect("some");

        assert_eq!(segment.get_base_offset(), 0);
        assert_eq!(segment.get_end_offset(), 4);
    }

    /// test replica with purging segments
    #[fluvio_future::test]
    async fn test_replica_segment_purge() {
        let storage_config = StorageConfig::builder()
            .cleaning_interval_ms(200)
            .build()
            .expect("build");

        let mut option = base_option("test_replica_purge");
        // enough for 2 batch (2 records per batch)
        option.segment_max_bytes = 160;
        option.index_max_interval_bytes = 50; // ensure we are writing to index
        option.retention_seconds = 1;

        let producer = BatchProducer::builder()
            .records(2u16)
            .record_generator(Arc::new(|_, _| Record::new("1")))
            .build()
            .expect("batch");

        let mut new_replica =
            FileReplica::create_or_load("test", 0, 0, option.clone(), Arc::new(storage_config))
                .await
                .expect("create");
        let reader = new_replica.prev_segments.read().await;
        assert!(reader.len() == 0);
        drop(reader);

        // this will create active sgment
        new_replica
            .write_batch(&mut producer.generate_batch())
            .await
            .expect("write");
        new_replica
            .write_batch(&mut producer.generate_batch())
            .await
            .expect("write");

        // wait enough so it won't get purged
        sleep(Duration::from_millis(700)).await;

        // overflow, will create 2nd segment
        new_replica
            .write_batch(&mut producer.generate_batch())
            .await
            .expect("write");

        let reader = new_replica.prev_segments.read().await;
        assert_eq!(reader.len(), 1);
        drop(reader);

        sleep(Duration::from_millis(1000)).await; // clear should purge

        let segments = new_replica.prev_segments.clone();
        assert_eq!(Arc::strong_count(&segments), 3);
        let reader = new_replica.prev_segments.read().await;
        assert_eq!(reader.len(), 0);
        drop(reader);

        new_replica.remove().await.expect("remove");
        drop(new_replica);
        sleep(Duration::from_millis(300)).await; // clear should end
        assert_eq!(Arc::strong_count(&segments), 1);
    }
}
