liteboxfs 0.2.0

use std::fmt::{self, Debug};

use crate::{
    chunker::{Chunker, ChunkerGuard},
    settings::Settings,
};

use super::{
    buffer::SliceReadMode,
    loaded::SliceBound,
    operation::{SliceOperation, WriteOperation, WriteOperationState},
    store::{BlockStore, BlockStoreInput, DataStore, ReadBlock},
    types::{
        Block, BlockId, BlockIndex, BlockList, BlockOffset, BlockSignature, FileId, FileOffset,
        HoleLen, IndexedBlock,
    },
};

pub struct ChunkingBlockStoreAdapter<Store> {
    inner: Store,
    chunker: ChunkerGuard,
    settings: Settings,
}

#[cfg_attr(coverage_nightly, coverage(off))]
impl<Store> Debug for ChunkingBlockStoreAdapter<Store>
where
    Store: Debug,
{
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("ChunkingBlockStoreAdapter")
            .field("inner", &self.inner)
            .field("settings", &self.settings)
            .finish()
    }
}

impl<Store> ChunkingBlockStoreAdapter<Store> {
    pub fn new(inner: Store, chunker: impl Chunker + 'static, settings: Settings) -> Self {
        Self {
            inner,
            chunker: ChunkerGuard::new(Box::new(chunker)),
            settings,
        }
    }
}

/// The position of an offset within a block list.
///
/// When writing blocks to the block store, we often need to split existing blocks. This type helps
/// us understand where in the block list a given offset falls, so we can choose the right strategy
/// for handling the write.
#[derive(Debug, Clone)]
enum BlockListOffset {
    /// The very start of the block list (i.e. offset 0).
    AtStart,

    /// Somewhere before the end of the block list, which could be in the middle of a block or at a
    /// block boundary.
    BeforeEnd {
        index: BlockIndex,
        offset_in_block: BlockOffset,
        block: Block,
    },

    /// The very end of the block list, at the end of the final block.
    AtEnd,

    /// Beyond the end of the block list. In this situation, sparse holes are usually involved.
    AfterEnd,
}

/// The position of a slice of blocks within a block list.
///
/// This enum allows us to be extremely explicit about how we handle each possible edge case when
/// writing a block to ensure we're not missing any.
#[derive(Debug, Clone)]
enum BlockListSlice {
    /// The slice is empty (zero bytes).
    Empty,
    /// The slice starts at the beginning of the block list and ends at or after the last block.
    FileStartToFileEnd { blocks: Vec<IndexedBlock> },
    /// The slice starts at the beginning of the block list and ends at a block boundary.
    FileStartToBlockBoundary { blocks: Vec<IndexedBlock> },
    /// The slice starts at the beginning of the block list and ends in the middle of that same block.
    FileStartToMidSameBlock {
        block: IndexedBlock,
        end_offset: BlockOffset,
    },
    /// The slice starts at the beginning of the block list and ends in the middle of a different
    /// block.
    FileStartToMidDifferentBlock {
        blocks: Vec<IndexedBlock>,
        end_offset: BlockOffset,
    },
    /// The slice starts at a block boundary and ends at or after the last block.
    BlockBoundaryToFileEnd { blocks: Vec<IndexedBlock> },
    /// The slice starts at a block boundary and ends at a different block boundary.
    BlockBoundaryToBlockBoundary { blocks: Vec<IndexedBlock> },
    /// The slice starts at a block boundary and ends in the middle of that same block.
    BlockBoundaryToMidSameBlock {
        block: IndexedBlock,
        end_offset: BlockOffset,
    },
    /// The slice starts at a block boundary and ends in the middle of a different block.
    BlockBoundaryToMidDifferentBlock {
        blocks: Vec<IndexedBlock>,
        end_offset: BlockOffset,
    },
    /// The slice starts in the middle of a block and ends at or after the last block.
    MidBlockToFileEnd {
        blocks: Vec<IndexedBlock>,
        start_offset: BlockOffset,
    },
    /// The slice starts in the middle of a block and ends at the end of that same block.
    MidBlockToBlockBoundary {
        block: IndexedBlock,
        start_offset: BlockOffset,
    },
    /// The slice starts in the middle of a block and ends in the middle of that same block.
    MidBlockToMidSameBlock {
        block: IndexedBlock,
        start_offset: BlockOffset,
        end_offset: BlockOffset,
    },
    /// The slice starts in the middle of a block and ends in the middle of a different block.
    MidBlockToMidDifferentBlock {
        blocks: Vec<IndexedBlock>,
        start_offset: BlockOffset,
        end_offset: BlockOffset,
    },
    /// The slice starts at the end of the last block.
    Append,
    /// The slice starts after the last block.
    AppendWithHole { offset_after_end: BlockOffset },
}

impl BlockListSlice {
    /// Returns `true` if any block in the slice is a hole.
    ///
    /// When a slice contains hole blocks, the full re-chunking path is unsafe because the loaded
    /// buffer omits the hole's implicit zero bytes — re-chunking would either panic on
    /// out-of-bounds buffer access or destroy sparsity by materializing the holes as data.
    fn contains_hole(&self) -> bool {
        let is_hole =
            |block: &IndexedBlock| matches!(block.block.signature, BlockSignature::Hole { .. });

        match self {
            BlockListSlice::Empty
            | BlockListSlice::Append
            | BlockListSlice::AppendWithHole { .. } => false,
            BlockListSlice::FileStartToFileEnd { blocks }
            | BlockListSlice::FileStartToBlockBoundary { blocks }
            | BlockListSlice::FileStartToMidDifferentBlock { blocks, .. }
            | BlockListSlice::BlockBoundaryToFileEnd { blocks }
            | BlockListSlice::BlockBoundaryToBlockBoundary { blocks }
            | BlockListSlice::BlockBoundaryToMidDifferentBlock { blocks, .. }
            | BlockListSlice::MidBlockToFileEnd { blocks, .. }
            | BlockListSlice::MidBlockToMidDifferentBlock { blocks, .. } => {
                blocks.iter().any(is_hole)
            }
            BlockListSlice::FileStartToMidSameBlock { block, .. }
            | BlockListSlice::BlockBoundaryToMidSameBlock { block, .. }
            | BlockListSlice::MidBlockToBlockBoundary { block, .. }
            | BlockListSlice::MidBlockToMidSameBlock { block, .. } => is_hole(block),
        }
    }
}

impl BlockList {
    fn slice(&mut self, start: FileOffset, end: FileOffset) -> BlockListSlice {
        if start == end {
            return BlockListSlice::Empty;
        }

        let total_len = self.file_len();

        if start > total_len {
            return BlockListSlice::AppendWithHole {
                offset_after_end: (start - total_len) as BlockOffset,
            };
        }

        if start == total_len {
            return BlockListSlice::Append;
        }

        // Find the block and offset where the start of the slice falls.
        let mut offset_so_far: FileOffset = 0;
        let mut start_block_index: BlockIndex = 0;
        let mut start_offset_in_block: BlockOffset = 0;

        for (i, block) in self.iter().enumerate() {
            if offset_so_far + block.len() as FileOffset > start {
                start_block_index = i;
                start_offset_in_block = (start - offset_so_far) as BlockOffset;
                break;
            }
            offset_so_far += block.len() as FileOffset;
        }

        // Determine if start is at a block boundary or the start of the file.
        let start_at_boundary = start_offset_in_block == 0;
        let start_at_file_start = start == 0;

        // Find the block and offset where the end of the slice falls.
        let mut end_block_index: BlockIndex = 0;
        let mut end_offset_in_block: BlockOffset = 0;

        // Determine if end is at a block boundary or the end of the file.
        let mut end_at_boundary = false;
        let end_at_or_after_file_end = end >= total_len;

        if !end_at_or_after_file_end {
            for (i, block) in self.iter().enumerate().skip(start_block_index) {
                if offset_so_far + block.len() as FileOffset > end {
                    end_block_index = i;
                    end_offset_in_block = (end - offset_so_far) as BlockOffset;
                    end_at_boundary = false;
                    break;
                } else if offset_so_far + block.len() as FileOffset == end {
                    end_block_index = i;
                    end_offset_in_block = block.len();
                    end_at_boundary = true;
                    break;
                }
                offset_so_far += block.len() as FileOffset;
            }
        }

        let blocks = self
            .iter()
            .enumerate()
            .map(|(i, block)| IndexedBlock {
                index: i,
                block: block.to_owned(),
            })
            .collect::<Vec<_>>();

        match (
            start_at_file_start,
            start_at_boundary,
            end_at_or_after_file_end,
        ) {
            (true, _, true) => BlockListSlice::FileStartToFileEnd { blocks },
            (true, _, false) if end_at_boundary => BlockListSlice::FileStartToBlockBoundary {
                blocks: blocks
                    .get(..=end_block_index)
                    .expect("Out of bounds slicing block list.")
                    .to_vec(),
            },
            (true, _, false) if start_block_index == end_block_index => {
                BlockListSlice::FileStartToMidSameBlock {
                    block: blocks
                        .get(start_block_index)
                        .expect("Out of bounds slicing block list.")
                        .clone(),
                    end_offset: end_offset_in_block,
                }
            }
            (true, _, false) => BlockListSlice::FileStartToMidDifferentBlock {
                blocks: blocks
                    .get(..=end_block_index)
                    .expect("Out of bounds slicing block list.")
                    .to_vec(),
                end_offset: end_offset_in_block,
            },
            (false, true, true) => BlockListSlice::BlockBoundaryToFileEnd {
                blocks: blocks
                    .get(start_block_index..)
                    .expect("Out of bounds slicing block list.")
                    .to_vec(),
            },
            (false, true, false) if end_at_boundary => {
                BlockListSlice::BlockBoundaryToBlockBoundary {
                    blocks: blocks
                        .get(start_block_index..=end_block_index)
                        .expect("Out of bounds slicing block list.")
                        .to_vec(),
                }
            }
            (false, true, false) if start_block_index == end_block_index => {
                BlockListSlice::BlockBoundaryToMidSameBlock {
                    block: blocks
                        .get(start_block_index)
                        .expect("Out of bounds slicing block list.")
                        .clone(),
                    end_offset: end_offset_in_block,
                }
            }
            (false, true, false) => BlockListSlice::BlockBoundaryToMidDifferentBlock {
                blocks: blocks
                    .get(start_block_index..=end_block_index)
                    .expect("Out of bounds slicing block list.")
                    .to_vec(),
                end_offset: end_offset_in_block,
            },
            (false, false, true) => BlockListSlice::MidBlockToFileEnd {
                blocks: blocks
                    .get(start_block_index..)
                    .expect("Out of bounds slicing block list.")
                    .to_vec(),
                start_offset: start_offset_in_block,
            },
            (false, false, false) if start_block_index == end_block_index => {
                if end_offset_in_block
                    == blocks
                        .get(start_block_index)
                        .expect("Out of bounds slicing block list.")
                        .block
                        .len()
                {
                    BlockListSlice::MidBlockToBlockBoundary {
                        block: blocks
                            .get(start_block_index)
                            .expect("Out of bounds slicing block list.")
                            .clone(),
                        start_offset: start_offset_in_block,
                    }
                } else {
                    BlockListSlice::MidBlockToMidSameBlock {
                        block: blocks
                            .get(start_block_index)
                            .expect("Out of bounds slicing block list.")
                            .clone(),
                        start_offset: start_offset_in_block,
                        end_offset: end_offset_in_block,
                    }
                }
            }
            (false, false, false) => BlockListSlice::MidBlockToMidDifferentBlock {
                blocks: blocks
                    .get(start_block_index..=end_block_index)
                    .expect("Out of bounds slicing block list.")
                    .to_vec(),
                start_offset: start_offset_in_block,
                end_offset: end_offset_in_block,
            },
        }
    }
}

impl BlockList {
    fn block_at_offset(&self, offset: FileOffset) -> BlockListOffset {
        if offset == 0 {
            return BlockListOffset::AtStart;
        }

        let mut bytes_so_far: FileOffset = 0;

        let mut block_iter = self.iter().enumerate().peekable();

        while let Some((_, block)) = block_iter.peek() {
            if bytes_so_far + block.len() as FileOffset > offset {
                break;
            }

            bytes_so_far += block.len() as FileOffset;

            block_iter.next();
        }

        match block_iter.next() {
            None if offset == bytes_so_far => BlockListOffset::AtEnd,
            None if offset > bytes_so_far => BlockListOffset::AfterEnd,
            Some((index, block)) => BlockListOffset::BeforeEnd {
                index,
                offset_in_block: (offset - bytes_so_far) as BlockOffset,
                block: block.to_owned(),
            },
            _ => unreachable!(
                "By this point, we should have verified that thet offset is before the end of the block list."
            ),
        }
    }
}

impl<'a> BlockStoreInput<'a> {
    pub fn into_block(self, block_id: BlockId) -> Block {
        match self {
            BlockStoreInput::Data {
                bytes,
                hash,
                len: logical_len,
                ..
            } => Block {
                id: block_id,
                signature: BlockSignature::Data {
                    hash,
                    len: logical_len.unwrap_or(bytes.len()),
                },
            },
            BlockStoreInput::Hole { len } => Block {
                id: block_id,
                signature: BlockSignature::Hole { len },
            },
        }
    }
}

impl<Store> ChunkingBlockStoreAdapter<Store>
where
    Store: BlockStore,
{
    fn new_op<'a>(&'a mut self, file: FileId, block_list: BlockList) -> WriteOperation<'a, Store> {
        WriteOperation::new(&mut self.inner, WriteOperationState::new(file, block_list))
    }

    fn new_slice_op<'a>(
        &'a mut self,
        file: FileId,
        block_list: BlockList,
    ) -> SliceOperation<'a, Store> {
        SliceOperation::new(
            &mut self.inner,
            WriteOperationState::new(file, block_list),
            &mut self.chunker,
        )
    }
}

impl<Store> BlockStore for ChunkingBlockStoreAdapter<Store>
where
    Store: BlockStore,
{
    fn list_blocks(&self, file: FileId) -> crate::Result<BlockList> {
        self.inner.list_blocks(file)
    }

    fn read_block(&mut self, block: BlockId, buf: &mut Vec<u8>) -> crate::Result<ReadBlock> {
        self.inner.read_block(block, buf)
    }

    fn replace_block(
        &mut self,
        file: FileId,
        index: BlockIndex,
        input: BlockStoreInput,
    ) -> crate::Result<BlockId> {
        self.inner.replace_block(file, index, input)
    }

    fn insert_block(
        &mut self,
        file: FileId,
        index: BlockIndex,
        input: BlockStoreInput,
    ) -> crate::Result<BlockId> {
        self.inner.insert_block(file, index, input)
    }

    fn append_block(
        &mut self,
        file: FileId,
        index: BlockIndex,
        input: BlockStoreInput,
    ) -> crate::Result<BlockId> {
        self.inner.append_block(file, index, input)
    }

    fn remove_blocks<R: std::ops::RangeBounds<usize>>(
        &mut self,
        file: FileId,
        blocks: R,
    ) -> crate::Result<Vec<BlockId>> {
        self.inner.remove_blocks(file, blocks)
    }
}

impl<Store> DataStore for ChunkingBlockStoreAdapter<Store>
where
    Store: BlockStore,
{
    fn list_blocks(&mut self, file: FileId) -> crate::Result<BlockList> {
        self.inner.list_blocks(file)
    }

    fn read_block(&mut self, block: BlockId, buf: &mut Vec<u8>) -> crate::Result<ReadBlock> {
        self.inner.read_block(block, buf)
    }

    fn write_block(
        &mut self,
        file: FileId,
        offset: FileOffset,
        mut block_list: BlockList,
        input: BlockStoreInput,
    ) -> crate::Result<BlockList> {
        if input.len() == 0 {
            return self.inner.list_blocks(file);
        }

        let block_slice = block_list
            .slice(offset, offset + input.len() as FileOffset)
            .clone();

        // The full re-chunking path cannot be used when the slice contains hole blocks: it would
        // either index out of bounds (the loaded buffer omits implicit hole bytes) or destroy
        // sparsity by absorbing the holes into newly chunked data blocks. The partial path splits
        // each hole at the splice boundary, preserving sparsity on either side of the write.
        let read_mode =
            if input.len() < self.settings.small_write_threshold || block_slice.contains_hole() {
                SliceReadMode::Partial
            } else {
                SliceReadMode::Full
            };

        let mut op = self.new_slice_op(file, block_list);

        match block_slice {
            BlockListSlice::Empty => Ok(op.into_inner().finish()),
            BlockListSlice::FileStartToFileEnd { blocks } => {
                op.as_mut().replace_blocks(0..blocks.len(), input)?;
                Ok(op.into_inner().finish())
            }
            BlockListSlice::FileStartToBlockBoundary { blocks } => Ok(op
                .read_slice(blocks, read_mode)?
                .splice(SliceBound::SliceStart, SliceBound::SliceEnd, input)?
                .finish()),
            BlockListSlice::FileStartToMidSameBlock { block, end_offset } => {
                let block_index = block.index;

                Ok(op
                    .read_slice(vec![block], read_mode)?
                    .splice(
                        SliceBound::SliceStart,
                        SliceBound::BlockOffset {
                            index: block_index,
                            offset: end_offset,
                        },
                        input,
                    )?
                    .finish())
            }
            BlockListSlice::FileStartToMidDifferentBlock { blocks, end_offset } => {
                let end_index = blocks.len() as BlockIndex - 1;

                Ok(op
                    .read_slice(blocks, read_mode)?
                    .splice(
                        SliceBound::SliceStart,
                        SliceBound::BlockOffset {
                            index: end_index,
                            offset: end_offset,
                        },
                        input,
                    )?
                    .finish())
            }
            BlockListSlice::BlockBoundaryToFileEnd { blocks } => {
                let start_index = blocks
                    .first()
                    .expect("Block list unexpectedly empty.")
                    .index;

                Ok(op
                    .read_slice(blocks, read_mode)?
                    .splice(
                        SliceBound::BlockOffset {
                            index: start_index,
                            offset: 0,
                        },
                        SliceBound::SliceEnd,
                        input,
                    )?
                    .finish())
            }
            BlockListSlice::BlockBoundaryToBlockBoundary { blocks } => {
                let start_index = blocks
                    .first()
                    .expect("Block list unexpectedly empty.")
                    .index;

                Ok(op
                    .read_slice(blocks, read_mode)?
                    .splice(
                        SliceBound::BlockOffset {
                            index: start_index,
                            offset: 0,
                        },
                        SliceBound::SliceEnd,
                        input,
                    )?
                    .finish())
            }
            BlockListSlice::BlockBoundaryToMidSameBlock { block, end_offset } => {
                let block_index = block.index;

                Ok(op
                    .read_slice(vec![block], read_mode)?
                    .splice(
                        SliceBound::BlockOffset {
                            index: block_index,
                            offset: 0,
                        },
                        SliceBound::BlockOffset {
                            index: block_index,
                            offset: end_offset,
                        },
                        input,
                    )?
                    .finish())
            }
            BlockListSlice::BlockBoundaryToMidDifferentBlock { blocks, end_offset } => {
                let start_index = blocks
                    .first()
                    .expect("Block list unexpectedly empty.")
                    .index;
                let end_index = blocks.last().expect("Block list unexpectedly empty.").index;

                Ok(op
                    .read_slice(blocks, read_mode)?
                    .splice(
                        SliceBound::BlockOffset {
                            index: start_index,
                            offset: 0,
                        },
                        SliceBound::BlockOffset {
                            index: end_index,
                            offset: end_offset,
                        },
                        input,
                    )?
                    .finish())
            }
            BlockListSlice::MidBlockToFileEnd {
                blocks,
                start_offset,
            } => {
                let start_index = blocks
                    .first()
                    .expect("Block list unexpectedly empty.")
                    .index;

                Ok(op
                    .read_slice(blocks, read_mode)?
                    .splice(
                        SliceBound::BlockOffset {
                            index: start_index,
                            offset: start_offset,
                        },
                        SliceBound::SliceEnd,
                        input,
                    )?
                    .finish())
            }
            BlockListSlice::MidBlockToBlockBoundary {
                block,
                start_offset,
            } => {
                let block_index = block.index;
                let end_offset = block.block.len();

                Ok(op
                    .read_slice(vec![block], read_mode)?
                    .splice(
                        SliceBound::BlockOffset {
                            index: block_index,
                            offset: start_offset,
                        },
                        SliceBound::BlockOffset {
                            index: block_index,
                            offset: end_offset,
                        },
                        input,
                    )?
                    .finish())
            }
            BlockListSlice::MidBlockToMidSameBlock {
                block,
                start_offset,
                end_offset,
            } => {
                let block_index = block.index;

                Ok(op
                    .read_slice(vec![block], read_mode)?
                    .splice(
                        SliceBound::BlockOffset {
                            index: block_index,
                            offset: start_offset,
                        },
                        SliceBound::BlockOffset {
                            index: block_index,
                            offset: end_offset,
                        },
                        input,
                    )?
                    .finish())
            }
            BlockListSlice::MidBlockToMidDifferentBlock {
                blocks,
                start_offset,
                end_offset,
            } => {
                let start_index = blocks
                    .first()
                    .expect("Block list unexpectedly empty.")
                    .index;
                let end_index = blocks.last().expect("Block list unexpectedly empty.").index;

                Ok(op
                    .read_slice(blocks, read_mode)?
                    .splice(
                        SliceBound::BlockOffset {
                            index: start_index,
                            offset: start_offset,
                        },
                        SliceBound::BlockOffset {
                            index: end_index,
                            offset: end_offset,
                        },
                        input,
                    )?
                    .finish())
            }
            BlockListSlice::Append => {
                let len = op.as_ref().len();
                op.as_mut().insert_block(len, input)?;
                Ok(op.into_inner().finish())
            }
            BlockListSlice::AppendWithHole { offset_after_end } => {
                let len = op.as_ref().len();
                op.as_mut().insert_block(
                    len,
                    BlockStoreInput::Hole {
                        len: offset_after_end as HoleLen,
                    },
                )?;

                let len = op.as_ref().len();
                op.as_mut().insert_block(len, input)?;

                Ok(op.into_inner().finish())
            }
        }
    }

    fn truncate(
        &mut self,
        file: FileId,
        block_list: BlockList,
        len: FileOffset,
    ) -> crate::Result<BlockList> {
        let block_list_len = block_list.len();
        let block_at_offset = block_list.block_at_offset(len).clone();

        let mut op = self.new_op(file, block_list);

        match block_at_offset {
            // No truncation needed. The file is already smaller than or equal to the requested
            // length.
            BlockListOffset::AtEnd | BlockListOffset::AfterEnd => Ok(op.finish()),
            // Truncate the entire file.
            BlockListOffset::AtStart => {
                op.remove_blocks(0..block_list_len)?;

                Ok(BlockList::new(Vec::new()))
            }
            // The truncation point is exactly at a block boundary, so we can just remove all
            // blocks after the split index.
            BlockListOffset::BeforeEnd {
                index,
                offset_in_block: 0,
                ..
            } => {
                op.remove_blocks(index..block_list_len)?;
                Ok(op.finish())
            }
            // The length we're truncating the file to falls in the middle of a block. So, we need
            // to:
            //
            // 1. Find the block that needs split.
            // 2. Read that block's data in from the store (if it's a data block).
            // 3. Replace the block with a truncated copy.
            // 4. Remove all blocks after it.
            BlockListOffset::BeforeEnd {
                index: split_index,
                block: split_block,
                offset_in_block: split_index_in_block,
                ..
            } => {
                let split_block = op.read_block(&split_block)?;
                let op = split_block.truncate(split_index, split_index_in_block)?;
                Ok(op.finish())
            }
        }
    }
}