pi_append_log 0.1.0

//! 使用标准库文件 I/O 的追加日志实现和默认布局。

use std::fs::{self, File, OpenOptions};
use std::io::{self, Read, Write};
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex, MutexGuard};

use crate::format::BlockDecoder;
use crate::storage::{
    AppendLog, AppendLogBuilder, AppendLogVisitor, BlockVisitContext, BuildResult, Layout,
    ReadOrder,
};

/// 固定八位十进制文件名的默认布局。
#[derive(Clone)]
pub struct DefaultFileLayout {
    root: PathBuf,
}

impl DefaultFileLayout {
    /// 创建指定目录下的默认布局。
    pub fn new(root: impl Into<PathBuf>) -> Self {
        Self { root: root.into() }
    }
}

impl Layout for DefaultFileLayout {
    type StructureId = u64;
    type Name = PathBuf;

    fn active_name(&self, structure_id: &Self::StructureId) -> Self::Name {
        self.root.join(format!("{structure_id:08}.active"))
    }

    fn closed_name(&self, structure_id: &Self::StructureId) -> Self::Name {
        self.root.join(format!("{structure_id:08}.closed"))
    }

    fn archive_name(&self, structure_id: &Self::StructureId) -> Self::Name {
        self.root.join(format!("{structure_id:08}.archive"))
    }

    fn parse_active_name(&self, name: &Self::Name) -> Option<Self::StructureId> {
        parse_name(name, &self.root, "active")
    }

    fn parse_closed_name(&self, name: &Self::Name) -> Option<Self::StructureId> {
        parse_name(name, &self.root, "closed")
    }

    fn parse_archive_name(&self, name: &Self::Name) -> Option<Self::StructureId> {
        parse_name(name, &self.root, "archive")
    }
}

fn parse_name(path: &Path, root: &Path, state: &str) -> Option<u64> {
    let relative = path.strip_prefix(root).ok()?;
    if relative.parent()?.as_os_str() != "" {
        return None;
    }
    let stem = relative.file_stem()?.to_str()?;
    if stem.len() != 8 || !stem.bytes().all(|byte| byte.is_ascii_digit()) {
        return None;
    }
    if relative.extension()?.to_str()? != state {
        return None;
    }
    let id = stem.parse::<u64>().ok()?;
    (id > 0).then_some(id)
}

/// 已封闭文件的归档句柄。
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct FileClosed {
    /// 产生此句柄的文件存储命名空间。
    namespace: PathBuf,
    /// 已封闭文件的八位逻辑编号。
    structure_id: u64,
}

/// 文件追加日志的初始化 Builder。
pub struct FileAppendLogBuilder<L> {
    root: PathBuf,
    layout: L,
}

impl<L> FileAppendLogBuilder<L> {
    /// 创建文件追加日志 Builder。
    pub fn new(root: impl Into<PathBuf>, layout: L) -> Self {
        Self {
            root: root.into(),
            layout,
        }
    }
}

/// 文件追加日志的运行期实现。
pub struct FileAppendLog<L> {
    /// 由互斥锁保护的文件生命周期和活动编号状态。
    state: Arc<Mutex<FileState<L>>>,
}

struct FileState<L> {
    root: PathBuf,
    layout: L,
    active_id: u64,
}

impl<L> AppendLogBuilder for FileAppendLogBuilder<L>
where
    L: Layout<StructureId = u64, Name = PathBuf> + Clone + Send + Sync,
{
    type Storage = FileAppendLog<L>;

    async fn build<D, V>(
        self,
        decoder: &D,
        order: ReadOrder,
        visitor: &mut V,
    ) -> io::Result<BuildResult<Self::Storage>>
    where
        D: BlockDecoder<Block = Vec<u8>> + Send + Sync,
        V: AppendLogVisitor + Send,
    {
        fs::create_dir_all(&self.root)?;
        let discovered = discover(&self.root, &self.layout)?;
        let active_id = match discovered.active_id {
            Some(id) => id,
            None => discovered.max_id.checked_add(1).unwrap_or(1),
        };
        let active_path = self.layout.active_name(&active_id);
        if !active_path.exists() {
            File::create(&active_path)?.sync_all()?;
        }
        let recovered_ids = discovered.closed_ids.clone();
        let mut structures = recovered_ids.clone();
        structures.push(active_id);
        structures.sort_unstable();
        let structure_count = structures.len();
        let mut stopped = false;
        for position in 0..structure_count {
            let structure_index = if matches!(order, ReadOrder::Forward) {
                position
            } else {
                structure_count - position - 1
            };
            let id = structures[structure_index];
            let is_active = id == active_id;
            let path = if is_active {
                self.layout.active_name(&id)
            } else {
                self.layout.closed_name(&id)
            };
            let bytes = read_structure(&path, decoder, is_active)?;
            if stopped {
                continue;
            }
            stopped = visit_structure(&bytes, decoder, order, visitor)?;
        }
        let namespace = self.root.clone();
        let storage = FileAppendLog {
            state: Arc::new(Mutex::new(FileState {
                root: self.root,
                layout: self.layout,
                active_id,
            })),
        };
        let recovered_closed = recovered_ids
            .into_iter()
            .map(|structure_id| FileClosed {
                namespace: namespace.clone(),
                structure_id,
            })
            .collect();
        Ok(BuildResult {
            storage,
            recovered_closed,
        })
    }
}

impl<L> AppendLog for FileAppendLog<L>
where
    L: Layout<StructureId = u64, Name = PathBuf> + Clone + Send + Sync,
{
    type Block = Vec<u8>;
    type Closed = FileClosed;

    async fn append(&self, block: Self::Block, options: crate::AppendOptions) -> io::Result<u64> {
        if block.is_empty() {
            return Err(invalid_input("block must not be empty"));
        }
        let state = self.lock_state()?;
        let path = state.layout.active_name(&state.active_id);
        let mut file = OpenOptions::new().create(true).append(true).open(path)?;
        file.write_all(&block)?;
        if options.durable {
            file.sync_all()?;
        }
        let length = file.metadata()?.len();
        Ok(length)
    }

    async fn rotate(&self) -> io::Result<Option<Self::Closed>> {
        let mut state = self.lock_state()?;
        let active_id = state.active_id;
        let active_path = state.layout.active_name(&active_id);
        let length = fs::metadata(&active_path)?.len();
        if length == 0 {
            return Ok(None);
        }
        let closed_path = state.layout.closed_name(&active_id);
        fs::rename(&active_path, &closed_path)?;
        let next_id = active_id
            .checked_add(1)
            .ok_or_else(|| invalid_data("file structure id exhausted"))?;
        let next_path = state.layout.active_name(&next_id);
        let next = File::create(&next_path)?;
        next.sync_all()?;
        state.active_id = next_id;
        Ok(Some(FileClosed {
            namespace: state.root.clone(),
            structure_id: active_id,
        }))
    }

    async fn archive(&self, closed: Self::Closed) -> io::Result<()> {
        let state = self.lock_state()?;
        if closed.namespace != state.root {
            return Err(invalid_input("closed handle belongs to another storage"));
        }
        if closed.structure_id == state.active_id {
            return Err(invalid_input("active structure cannot be archived"));
        }
        let closed_path = state.layout.closed_name(&closed.structure_id);
        let archive_path = state.layout.archive_name(&closed.structure_id);
        if archive_path.exists() {
            return Ok(());
        }
        fs::rename(closed_path, archive_path)?;
        Ok(())
    }
}

struct Discovered {
    active_id: Option<u64>,
    closed_ids: Vec<u64>,
    max_id: u64,
}

fn discover<L: Layout<StructureId = u64, Name = PathBuf>>(
    root: &Path,
    layout: &L,
) -> io::Result<Discovered> {
    let mut result = Discovered {
        active_id: None,
        closed_ids: Vec::new(),
        max_id: 0,
    };
    for entry in fs::read_dir(root)? {
        let path = entry?.path();
        if let Some(id) = layout.parse_active_name(&path) {
            if result.active_id.replace(id).is_some() {
                return Err(invalid_data("multiple active structures found"));
            }
            result.max_id = result.max_id.max(id);
        } else if let Some(id) = layout.parse_closed_name(&path) {
            result.closed_ids.push(id);
            result.max_id = result.max_id.max(id);
        }
    }
    result.closed_ids.sort_unstable();
    Ok(result)
}

fn read_structure<D: BlockDecoder<Block = Vec<u8>>>(
    path: &Path,
    decoder: &D,
    active: bool,
) -> io::Result<Vec<u8>> {
    let mut file = OpenOptions::new().read(true).write(active).open(path)?;
    let mut bytes = Vec::new();
    file.read_to_end(&mut bytes)?;
    let boundary = match decoder.find_last_complete(&bytes)? {
        Some(end) => end,
        None => {
            if bytes.is_empty() {
                return Ok(Vec::new());
            }
            if active {
                file.set_len(0)?;
                file.sync_all()?;
                return Ok(Vec::new());
            }
            return Err(invalid_data("closed structure has no complete block"));
        }
    };
    if boundary < bytes.len() {
        if !active {
            return Err(invalid_data("closed structure has an invalid tail"));
        }
        file.set_len(boundary as u64)?;
        file.sync_all()?;
        bytes.truncate(boundary);
    }
    validate_structure(&bytes, decoder)?;
    Ok(bytes)
}

fn validate_structure<D: BlockDecoder<Block = Vec<u8>>>(
    bytes: &[u8],
    decoder: &D,
) -> io::Result<()> {
    let mut offset = 0;
    while offset < bytes.len() {
        let decoded = decoder.decode_forward(&bytes[offset..])?;
        offset = offset
            .checked_add(decoded.encoded_len())
            .ok_or_else(|| invalid_data("block offset overflow"))?;
    }
    Ok(())
}

/// 按完整 block 的边界逐个调用 Visitor。
///
/// 结构字节只在单次 visit 调用期间借用；这里不建立 Vec<Vec<u8>>，也不会把所有 block
/// 的拥有副本保存到恢复状态中。Backward 读取需要先得到边界偏移，但仍然按 block 顺序
/// 逐次解码和回调。
fn visit_structure<D, V>(
    bytes: &[u8],
    decoder: &D,
    order: ReadOrder,
    visitor: &mut V,
) -> io::Result<bool>
where
    D: BlockDecoder<Block = Vec<u8>>,
    V: AppendLogVisitor + Send,
{
    let mut ranges = Vec::new();
    let mut offset = 0;
    while offset < bytes.len() {
        let decoded = decoder.decode_forward(&bytes[offset..])?;
        let next_offset = offset
            .checked_add(decoded.encoded_len())
            .ok_or_else(|| invalid_data("block offset overflow"))?;
        ranges.push((offset, next_offset));
        offset = next_offset;
    }
    if matches!(order, ReadOrder::Backward) {
        ranges.reverse();
    }
    for (index, (start, end)) in ranges.iter().enumerate() {
        let is_backward = matches!(order, ReadOrder::Backward);
        let is_first = if is_backward {
            index + 1 == ranges.len()
        } else {
            index == 0
        };
        let is_last = if is_backward {
            index == 0
        } else {
            index + 1 == ranges.len()
        };
        if visitor.visit(
            &bytes[*start..*end],
            BlockVisitContext {
                is_first_in_structure: is_first,
                is_last_in_structure: is_last,
            },
        )? {
            return Ok(true);
        }
    }
    Ok(false)
}

impl<L> FileAppendLog<L> {
    fn lock_state(&self) -> io::Result<MutexGuard<'_, FileState<L>>> {
        self.state
            .lock()
            .map_err(|_| io::Error::other("file storage state lock poisoned"))
    }
}

fn invalid_input(message: &str) -> io::Error {
    io::Error::new(io::ErrorKind::InvalidInput, message)
}

fn invalid_data(message: &str) -> io::Error {
    io::Error::new(io::ErrorKind::InvalidData, message)
}