// Copyright 2022 PingCAP, Inc.

use std::{
    cmp::Ordering as CmpOrdering,
    collections::{HashMap, HashSet, VecDeque},
    fs,
    fs::{File, OpenOptions},
    io,
    ops::{Deref, DerefMut},
    os::unix::fs::FileExt,
    path::{Path, PathBuf},
    sync::{
        Arc,
        atomic::{AtomicU64, Ordering},
    },
};

use bytes::{Buf, BufMut};
use protobuf::Message;
use tikv_util::{error, info, warn};

use crate::{
    PeerMeta, TRUNCATE_ALL_INDEX, metrics::RFENGINE_RLOG_GC_SIZE, raft_log_file_name,
    writer::EPOCH_SNAPSHOT_LEN,
};

const REWRITE_DIFF: u32 = 10;

#[derive(Debug)]
pub(crate) struct Manifest {
    pub(crate) epoch_id: u32,
    pub(crate) delayed_epoches: u32,
    pub(crate) file_path: PathBuf,
    pub(crate) file: File,
    pub(crate) peers: HashMap<u64, PeerMetaFiles>,
    pub(crate) offset: u64,
    pub(crate) engine_id: Arc<AtomicU64>,
    first_epoch: u32,
}

#[derive(Debug, PartialEq, Eq)]
pub(crate) struct EntryRange {
    pub(crate) start: u64, // inclusive
    pub(crate) end: u64,   // inclusive
}

impl EntryRange {
    pub(crate) fn new(start: u64, end: u64) -> Self {
        Self { start, end }
    }
}

#[derive(Debug, Clone)]
pub(crate) struct PeerMetaFiles {
    pub(crate) meta: PeerMeta,
    pub(crate) files: VecDeque<PeerFile>,
}

impl Deref for PeerMetaFiles {
    type Target = PeerMeta;

    fn deref(&self) -> &Self::Target {
        &self.meta
    }
}

impl DerefMut for PeerMetaFiles {
    fn deref_mut(&mut self) -> &mut Self::Target {
        &mut self.meta
    }
}

impl PeerMetaFiles {
    fn new(region_id: u64, keyspace_id: u32) -> Self {
        Self {
            meta: PeerMeta::new(region_id, keyspace_id),
            files: Default::default(),
        }
    }
}

#[derive(Debug, Default, Copy, Clone, PartialEq, Eq)]
pub struct PeerFile {
    pub first_index: u64,
    pub last_index: u64,
    pub last_term: u32,
    pub epoch_id: u32,
}

impl PeerFile {
    pub fn new(epoch_id: u32, first_index: u64, last_index: u64, last_term: u32) -> Self {
        Self {
            epoch_id,
            first_index,
            last_index,
            last_term,
        }
    }
}

impl From<&rfenginepb::RaftLogFile> for PeerFile {
    fn from(file: &rfenginepb::RaftLogFile) -> Self {
        Self {
            first_index: file.first_index,
            last_index: file.last_index,
            last_term: file.last_term,
            epoch_id: file.epoch_id,
        }
    }
}

impl Manifest {
    pub(crate) fn open(dir: &Path, engine_id: Arc<AtomicU64>) -> crate::Result<Self> {
        let file_path = manifest_path(dir);
        let file = OpenOptions::new()
            .create(true)
            .truncate(false)
            .write(true)
            .read(true)
            .open(&file_path)?;
        let mut manifest = Self {
            epoch_id: 0,
            delayed_epoches: 0,
            engine_id,
            file_path,
            file,
            peers: HashMap::new(),
            offset: 0,
            first_epoch: 0,
        };
        manifest.init()?;
        Ok(manifest)
    }

    fn init(&mut self) -> crate::Result<()> {
        let file_data_vec = fs::read(&self.file_path)?;
        let mut file_data = file_data_vec.as_slice();
        while file_data.len() > 8 {
            let checksum = file_data.get_u32_le();
            let length = file_data.get_u32_le() as usize;
            if file_data.len() < length {
                warn!("manifest file unexpected EOF");
                break;
            }
            let change_set_data = &file_data[..length];
            if crc32c::crc32c(change_set_data) != checksum {
                warn!("manifest file checksum mismatch");
                break;
            }
            self.offset += length as u64 + 8;
            let mut change_set = rfenginepb::ChangeSet::new();
            change_set.merge_from_bytes(change_set_data).unwrap();
            self.apply_change_set(&change_set)?;
            file_data = &file_data[length..];
        }
        self.truncate_all_files();
        info!("init manifest epoch {}", self.epoch_id);
        Ok(())
    }

    pub(crate) fn handle_compaction(&mut self, cs: rfenginepb::ChangeSet) -> crate::Result<()> {
        let gc_peer_ids = cs
            .get_peers()
            .iter()
            .map(|peer_meta| peer_meta.peer_id)
            .collect::<Vec<_>>();
        self.apply_change_set(&cs)?;
        let engine_id = self.get_engine_id();
        if self.epoch_id - self.first_epoch > REWRITE_DIFF {
            info!("{}: rewrite manifest epoch: {}", engine_id, self.epoch_id);
            self.rewrite()?;
            self.first_epoch = self.epoch_id;
        } else {
            info!("{}: append manifest epoch: {}", engine_id, self.epoch_id);
            self.persist_change_set(cs)?;
        }
        self.truncate_files(&gc_peer_ids);
        Ok(())
    }

    fn persist_change_set(&mut self, cs: rfenginepb::ChangeSet) -> io::Result<()> {
        self.offset = persist_change_set(&self.file, self.offset, &cs)?;
        Ok(())
    }

    fn apply_change_set(&mut self, cs: &rfenginepb::ChangeSet) -> crate::Result<()> {
        assert!(cs.epoch_id > self.epoch_id);
        self.epoch_id = cs.epoch_id;
        self.delayed_epoches = cs.delayed_epoches;
        if self.first_epoch == 0 {
            self.first_epoch = cs.epoch_id;
        }
        for peer_meta_pb in cs.get_peers() {
            let peer_meta = self.peers.entry(peer_meta_pb.peer_id).or_insert_with(|| {
                PeerMetaFiles::new(peer_meta_pb.region_id, peer_meta_pb.keyspace_id)
            });
            if peer_meta.truncated_idx < peer_meta_pb.truncated_index {
                peer_meta.truncated_idx = peer_meta_pb.truncated_index;
            }
            for state_pb in peer_meta_pb.get_states() {
                if state_pb.value.is_empty() {
                    peer_meta.remove_state(state_pb.get_key());
                } else {
                    peer_meta.set_state(state_pb.get_key(), state_pb.get_value());
                }
            }
            let pb_keyspace_id = peer_meta_pb.get_keyspace_id();
            if pb_keyspace_id > 0 {
                peer_meta.keyspace_id = pb_keyspace_id;
            } else if peer_meta.keyspace_id == 0 {
                peer_meta.keyspace_id = peer_meta.get_keyspace_id().unwrap_or_default();
            }
            for file in peer_meta_pb.get_files() {
                peer_meta.files.push_back(file.into());
            }
        }
        Ok(())
    }

    fn rewrite(&mut self) -> crate::Result<()> {
        let dir = self.file_path.parent().unwrap();
        let tmp_path = self.file_path.with_extension("tmp");
        let tmp_file = File::create(&tmp_path)?;
        let change_set = self.to_change_set(false);
        self.offset = persist_change_set(&tmp_file, 0, &change_set)?;
        fs::rename(&tmp_path, &self.file_path)?;
        file_system::sync_dir(dir)?;
        self.file = OpenOptions::new()
            .read(true)
            .write(true)
            .open(&self.file_path)?;
        Ok(())
    }

    pub(crate) fn to_change_set(&self, exclude_tombstone: bool) -> rfenginepb::ChangeSet {
        let mut cs = rfenginepb::ChangeSet::default();
        cs.epoch_id = self.epoch_id;
        cs.delayed_epoches = self.delayed_epoches;
        for (&peer_id, peer_meta) in &self.peers {
            if exclude_tombstone && peer_meta.truncated_idx == TRUNCATE_ALL_INDEX {
                continue;
            }
            let mut meta_pb = rfenginepb::PeerMeta::new();
            meta_pb.set_peer_id(peer_id);
            meta_pb.set_region_id(peer_meta.region_id);
            meta_pb.set_truncated_index(peer_meta.truncated_idx);
            meta_pb.set_keyspace_id(peer_meta.keyspace_id);
            for (key, val) in &peer_meta.states {
                let mut state = rfenginepb::PeerState::new();
                state.set_key(key.to_vec());
                state.set_value(val.to_vec());
                meta_pb.mut_states().push(state);
            }
            for file in &peer_meta.files {
                let mut raft_log_file = rfenginepb::RaftLogFile::default();
                raft_log_file.set_first_index(file.first_index);
                raft_log_file.set_last_index(file.last_index);
                raft_log_file.set_last_term(file.last_term);
                raft_log_file.set_epoch_id(file.epoch_id);
                meta_pb.mut_files().push(raft_log_file);
            }
            cs.mut_peers().push(meta_pb);
        }
        cs
    }

    fn truncate_all_files(&mut self) {
        let peer_ids = self.peers.keys().copied().collect::<Vec<_>>();
        self.truncate_files(&peer_ids);
    }

    fn truncate_files(&mut self, peer_ids: &[u64]) {
        let dir = self.file_path.parent().unwrap();
        let engine_id = self.engine_id.load(Ordering::SeqCst);
        for &peer_id in peer_ids {
            let Some(peer_meta) = self.peers.get_mut(&peer_id) else {
                continue;
            };
            let obsolete_files = take_obsolete_rlog_files(peer_meta);
            if obsolete_files.is_empty() {
                continue;
            }
            delete_obsolete_rlog_files(dir, engine_id, peer_id, peer_meta, obsolete_files);
        }
    }

    pub(crate) fn peer_rlog_files(&self) -> HashMap<u64, VecDeque<PeerFile>> {
        self.peers
            .iter()
            .map(|(&peer_id, meta_files)| (peer_id, meta_files.files.clone()))
            .collect()
    }

    pub(crate) fn get_engine_id(&self) -> u64 {
        self.engine_id.load(Ordering::SeqCst)
    }

    /// The epoch of next snapshot.
    pub(crate) fn next_snapshot_epoch(epoch_id: u32) -> u32 {
        epoch_id.saturating_add(EPOCH_SNAPSHOT_LEN - 1) / EPOCH_SNAPSHOT_LEN * EPOCH_SNAPSHOT_LEN
    }
}

pub(crate) fn manifest_path(dir: &Path) -> PathBuf {
    dir.join("MANIFEST")
}

pub(crate) fn persist_change_set(
    file: &File,
    mut offset: u64,
    cs: &rfenginepb::ChangeSet,
) -> io::Result<u64> {
    let buf = cs.write_to_bytes().unwrap();
    if buf.is_empty() {
        return Ok(0);
    }
    let mut header_buf = Vec::with_capacity(8);
    header_buf.put_u32_le(crc32c::crc32c(&buf));
    header_buf.put_u32_le(buf.len() as u32);
    file.write_all_at(&header_buf, offset)?;
    offset += 8;
    file.write_all_at(&buf, offset)?;
    offset += buf.len() as u64;
    file.sync_data()?;
    Ok(offset)
}

/// Generates a read plan for fetching Raft log entries within the range `[low,
/// high)`.
///
/// Given a list of rlog files and a desired entry range, this function returns
/// a sequence of `(EntryRange, PeerFile)` pairs that describe which portion of
/// the range should be read from which file.
pub(crate) fn generate_rlog_read_plan(
    rlog_files: &VecDeque<PeerFile>,
    low: u64,  // inclusive
    high: u64, // exclusive
) -> crate::Result<Vec<(EntryRange, PeerFile)>> {
    let ranges = range_to_rlog_mapping(rlog_files);
    let mut plan = vec![];
    let mut current = low;
    for (range, file) in ranges {
        if range.end < current {
            continue; // the range is too small, move on
        }

        if range.start > current {
            return Err(crate::Error::EntriesUnavailable); // unexpected gap
        }

        // At this point, we know: range.start <= current <= range.end
        let end = range.end.min(high - 1);
        plan.push((EntryRange::new(current, end), file));
        current = end + 1;
        if current >= high {
            return Ok(plan);
        }
    }
    Err(crate::Error::EntriesUnavailable)
}

/// Builds a mapping of non-overlapping, continuous entry ranges to their
/// corresponding `PeerFile`.
///
/// The semantics of the function are identical with `RaftLogs::append`
/// - If a log gap is detected, all previous entries are dropped.
/// - If an overlap occurs, newer entries overwrite the older ones.
pub(crate) fn range_to_rlog_mapping(files: &VecDeque<PeerFile>) -> Vec<(EntryRange, PeerFile)> {
    range_to_rlog_mapping_indexes(files)
        .into_iter()
        .map(|(range, index)| (range, files[index]))
        .collect()
}

fn range_to_rlog_mapping_indexes(files: &VecDeque<PeerFile>) -> Vec<(EntryRange, usize)> {
    let mut stack: Vec<(EntryRange, usize)> = Vec::with_capacity(files.len());
    for (index, file) in files.iter().enumerate() {
        if let Some((r, _)) = stack.last() {
            let next_index = r.end + 1;
            let incoming_index = file.first_index;
            match incoming_index.cmp(&next_index) {
                CmpOrdering::Greater => {
                    // There is gap between existing logs and next log, clear all.
                    stack.clear();
                }
                CmpOrdering::Less => {
                    while let Some((EntryRange { start, .. }, file_index)) = stack.pop() {
                        if start < incoming_index {
                            stack.push((EntryRange::new(start, incoming_index - 1), file_index));
                            break;
                        }
                    }
                }
                CmpOrdering::Equal => {}
            }
        }
        stack.push((EntryRange::new(file.first_index, file.last_index), index));
    }
    stack
}

#[derive(Default)]
struct ObsoleteRlogFiles {
    metadata_count: usize,
    local_files: Vec<PeerFile>,
}

impl ObsoleteRlogFiles {
    fn is_empty(&self) -> bool {
        self.metadata_count == 0
    }
}

fn take_obsolete_rlog_files(peer_meta: &mut PeerMetaFiles) -> ObsoleteRlogFiles {
    if let Some(obsolete_files) = take_obsolete_sequential_rlog_files(peer_meta) {
        return obsolete_files;
    }

    let effective_last_indexes = effective_rlog_file_last_indexes(&peer_meta.files);
    let truncated_idx = peer_meta.truncated_idx;
    if !effective_last_indexes.iter().any(|last_index| {
        last_index
            .map(|last_index| last_index <= truncated_idx)
            .unwrap_or(true)
    }) {
        return ObsoleteRlogFiles::default();
    }

    let mut retained_paths = HashSet::with_capacity(peer_meta.files.len());
    let mut obsolete_files = Vec::with_capacity(peer_meta.files.len());
    let mut index = 0;
    peer_meta.files.retain(|file| {
        let obsolete = effective_last_indexes[index]
            .map(|last_index| last_index <= truncated_idx)
            .unwrap_or(true);
        index += 1;
        if obsolete {
            obsolete_files.push(*file);
        } else {
            retained_paths.insert((file.first_index, file.last_index));
        }
        !obsolete
    });
    let local_files = obsolete_files
        .iter()
        .copied()
        .filter(|file| !retained_paths.contains(&(file.first_index, file.last_index)))
        .collect();
    ObsoleteRlogFiles {
        metadata_count: obsolete_files.len(),
        local_files,
    }
}

fn take_obsolete_sequential_rlog_files(peer_meta: &mut PeerMetaFiles) -> Option<ObsoleteRlogFiles> {
    if !is_sequential_rlog_files(&peer_meta.files) {
        return None;
    }

    let mut local_files = Vec::new();
    while peer_meta
        .files
        .front()
        .is_some_and(|file| file.last_index <= peer_meta.truncated_idx)
    {
        local_files.push(peer_meta.files.pop_front().unwrap());
    }
    Some(ObsoleteRlogFiles {
        metadata_count: local_files.len(),
        local_files,
    })
}

fn is_sequential_rlog_files(files: &VecDeque<PeerFile>) -> bool {
    let mut files = files.iter();
    let Some(mut previous_file) = files.next() else {
        return true;
    };
    for file in files {
        match previous_file.last_index.checked_add(1) {
            Some(next_index) if next_index == file.first_index => {}
            _ => return false,
        }
        previous_file = file;
    }
    true
}

fn delete_obsolete_rlog_files(
    dir: &Path,
    engine_id: u64,
    peer_id: u64,
    peer_meta: &PeerMetaFiles,
    obsolete_files: ObsoleteRlogFiles,
) {
    let removed_meta_count = obsolete_files.metadata_count;
    let mut removed_file_count = 0;
    let mut last_index = 0;
    for file in obsolete_files.local_files {
        let filename = raft_log_file_name(dir, peer_id, file.first_index, file.last_index);
        let file_size = match fs::metadata(filename.as_path()) {
            Ok(metadata) => metadata.len(),
            Err(err) if err.kind() == io::ErrorKind::NotFound => continue,
            Err(err) => {
                error!(
                    "{}:{} failed to get obsolete rlog file metadata {:?}, {:?}",
                    engine_id, peer_meta.region_id, filename, err
                );
                RFENGINE_RLOG_GC_SIZE
                    .with_label_values(&["error"])
                    .observe(0.0);
                continue;
            }
        };
        if let Err(err) = fs::remove_file(filename.as_path()) {
            error!(
                "{}:{} failed to remove obsolete rlog file {:?}, {:?}",
                engine_id, peer_meta.region_id, filename, err
            );
            RFENGINE_RLOG_GC_SIZE
                .with_label_values(&["error"])
                .observe(file_size as f64);
        } else {
            removed_file_count += 1;
            last_index = file.last_index;
            RFENGINE_RLOG_GC_SIZE
                .with_label_values(&["success"])
                .observe(file_size as f64);
        }
    }
    info!(
        "{}:{} removed {} obsolete rlog files, {} local files, last removed index {}, truncated index {}",
        engine_id,
        peer_meta.region_id,
        removed_meta_count,
        removed_file_count,
        last_index,
        peer_meta.truncated_idx
    );
}

fn effective_rlog_file_last_indexes(files: &VecDeque<PeerFile>) -> Vec<Option<u64>> {
    let mut last_indexes = vec![None; files.len()];
    for (range, index) in range_to_rlog_mapping_indexes(files) {
        last_indexes[index] = Some(range.end);
    }
    last_indexes
}

/// Returns the `PeerFile` that contains the Raft entry at index `idx`.
///
/// Traverses from newest to oldest, respecting semantics that newer rlog files
/// overwrite older ones.
pub(crate) fn rlog_by_entry_index(files: &VecDeque<PeerFile>, idx: u64) -> Option<PeerFile> {
    let mut next_start_index = None;
    for file in files.iter().rev() {
        if next_start_index.is_some_and(|n| file.last_index + 1 < n) {
            // If a gap exists between this file and the previously seen (newer)
            // file, entries from this and all older files are considered
            // invalid. This matches the semantics of `range_to_rlog_mapping`.
            return None;
        }

        if file.last_index < idx {
            return None;
        }
        if file.first_index <= idx {
            return Some(*file);
        }

        next_start_index = Some(file.first_index);
    }
    None
}

#[cfg(test)]
mod tests {
    use std::sync::{Arc, atomic::AtomicU64};

    use super::*;

    fn new_rlog_file(
        epoch_id: u32,
        first_index: u64,
        last_index: u64,
        last_term: u32,
    ) -> rfenginepb::RaftLogFile {
        let mut file = rfenginepb::RaftLogFile::default();
        file.set_epoch_id(epoch_id);
        file.set_first_index(first_index);
        file.set_last_index(last_index);
        file.set_last_term(last_term);
        file
    }

    fn write_rlog_file(dir: &Path, peer_id: u64, first_index: u64, last_index: u64) {
        fs::write(
            raft_log_file_name(dir, peer_id, first_index, last_index),
            b"rlog",
        )
        .unwrap();
    }

    fn take_obsolete_for_test(
        files: &[PeerFile],
        truncated_idx: u64,
    ) -> (Vec<PeerFile>, Vec<PeerFile>) {
        let mut peer = PeerMetaFiles::new(1, 0);
        peer.truncated_idx = truncated_idx;
        peer.files = files.iter().copied().collect();
        let obsolete_files = take_obsolete_rlog_files(&mut peer);
        let retained_files = peer.files.iter().copied().collect();
        (obsolete_files.local_files, retained_files)
    }

    #[test]
    fn test_next_snapshot_epoch() {
        assert_eq!(Manifest::next_snapshot_epoch(0), 0);
        assert_eq!(Manifest::next_snapshot_epoch(1), 8);
        assert_eq!(Manifest::next_snapshot_epoch(3), 8);
        assert_eq!(Manifest::next_snapshot_epoch(4), 8);
        assert_eq!(Manifest::next_snapshot_epoch(7), 8);
        assert_eq!(Manifest::next_snapshot_epoch(8), 8);
        assert_eq!(Manifest::next_snapshot_epoch(9), 16);
        assert_eq!(Manifest::next_snapshot_epoch(u32::MAX), 0xffff_fff8);
    }

    #[test]
    fn test_get_rlog_by_entry_index() {
        let files = VecDeque::from(vec![
            PeerFile::new(1, 1, 5, 0),    // discarded due to gap
            PeerFile::new(2, 8, 9, 0),    // actual mapped range: 8–9
            PeerFile::new(3, 10, 100, 0), // actual mapped range: 10–29
            PeerFile::new(4, 30, 90, 0),  // actual mapped range: 30–49
            PeerFile::new(5, 50, 70, 0),  // actual mapped range: 50–70
        ]);

        let mapping = range_to_rlog_mapping(&files);
        assert_eq!(mapping.len(), 4);
        assert_eq!(mapping[0].0, EntryRange::new(8, 9));
        assert_eq!(mapping[0].1.epoch_id, 2);
        assert_eq!(mapping[1].0, EntryRange::new(10, 29));
        assert_eq!(mapping[1].1.epoch_id, 3);
        assert_eq!(mapping[2].0, EntryRange::new(30, 49));
        assert_eq!(mapping[2].1.epoch_id, 4);
        assert_eq!(mapping[3].0, EntryRange::new(50, 70));
        assert_eq!(mapping[3].1.epoch_id, 5);

        for (range, file) in mapping.iter() {
            for i in range.start..=range.end {
                let entry = rlog_by_entry_index(&files, i);
                assert!(entry.is_some(), "index {}", i);
                assert_eq!(entry.unwrap(), *file, "index {}", i);
            }
        }
        assert!(rlog_by_entry_index(&files, 1).is_none());
        assert!(rlog_by_entry_index(&files, 7).is_none());
        assert!(rlog_by_entry_index(&files, 71).is_none());
        assert!(rlog_by_entry_index(&files, 100).is_none());
    }

    #[test]
    fn test_generate_rlog_read_plan() {
        let files = VecDeque::from(vec![
            PeerFile::new(1, 1, 5, 0),    // discarded due to gap
            PeerFile::new(2, 10, 100, 0), // 10–29
            PeerFile::new(3, 30, 90, 0),  // 30–49
            PeerFile::new(4, 50, 70, 0),  // 50–70
        ]);

        for (desc, low, high, expected) in [
            ("first range", 12, 15, Some(vec![(12, 14, 2)])),
            ("last range", 69, 71, Some(vec![(69, 70, 4)])),
            ("two files", 28, 32, Some(vec![(28, 29, 2), (30, 31, 3)])),
            ("gap before start", 2, 5, None),
            ("gap at beginning", 1, 11, None),
            ("end out of range", 69, 75, None),
            (
                "full span",
                10,
                71,
                Some(vec![(10, 29, 2), (30, 49, 3), (50, 70, 4)]),
            ),
        ] {
            let result = generate_rlog_read_plan(&files, low, high);
            match expected {
                Some(ranges) => {
                    let actual: Vec<_> = result
                        .unwrap()
                        .iter()
                        .map(|(r, pf)| (r.start, r.end, pf.epoch_id))
                        .collect();
                    assert_eq!(actual, ranges, "{}", desc);
                }
                None => {
                    assert!(result.is_err(), "{}", desc);
                }
            }
        }
    }

    #[test]
    fn test_take_obsolete_rlog_files_respects_truncation_and_overwrite() {
        let old_file = PeerFile::new(1, 1, 100, 1);
        let overwritten_file = PeerFile::new(2, 30, 40, 2);
        let tail_file = PeerFile::new(3, 41, 60, 3);
        let files = [old_file, overwritten_file, tail_file];

        assert_eq!(
            take_obsolete_for_test(&files, 0),
            (vec![], vec![old_file, overwritten_file, tail_file])
        );
        assert_eq!(
            take_obsolete_for_test(&files, 35),
            (vec![old_file], vec![overwritten_file, tail_file])
        );
        assert_eq!(
            take_obsolete_for_test(&files, 40),
            (vec![old_file, overwritten_file], vec![tail_file])
        );
        assert_eq!(
            take_obsolete_for_test(&files, 60),
            (vec![old_file, overwritten_file, tail_file], vec![])
        );
    }

    #[test]
    fn test_take_obsolete_rlog_files_truncates_sequential_head() {
        let compacted_file_1 = PeerFile::new(1, 1, 10, 1);
        let compacted_file_2 = PeerFile::new(2, 11, 20, 2);
        let live_file = PeerFile::new(3, 21, 30, 3);
        let files = [compacted_file_1, compacted_file_2, live_file];
        let sequential_files = VecDeque::from(vec![compacted_file_1, compacted_file_2, live_file]);

        assert!(is_sequential_rlog_files(&sequential_files));
        assert_eq!(
            take_obsolete_for_test(&files, 20),
            (vec![compacted_file_1, compacted_file_2], vec![live_file])
        );
        assert_eq!(
            take_obsolete_for_test(&files, 0),
            (vec![], vec![compacted_file_1, compacted_file_2, live_file])
        );
    }

    #[test]
    fn test_truncate_files_removes_obsolete_non_front_rlogs() {
        let tmp_dir = tempfile::tempdir().unwrap();
        let tmp_path = tmp_dir.path();
        let peer_id = 2323539997;
        let region_id = 2072768736;
        let files = [
            (1838, 489, 489, 4267),
            (1839, 490, 490, 4270),
            (1842, 491, 491, 4280),
            (2076, 468, 469, 4625),
            (2077, 470, 470, 4626),
            (2107, 472, 473, 4629),
            (2120, 474, 474, 4630),
            (2123, 475, 475, 4631),
            (2131, 477, 477, 4633),
            (2133, 478, 478, 4634),
            (2136, 479, 479, 4635),
            (2137, 480, 480, 4636),
            (2143, 482, 482, 4638),
            (2146, 483, 483, 4639),
            (2148, 484, 484, 4640),
            (2150, 485, 485, 4641),
            (2158, 487, 487, 4643),
            (2161, 488, 488, 4644),
            (2166, 489, 489, 4645),
        ];

        let mut peer = rfenginepb::PeerMeta::new();
        peer.set_peer_id(peer_id);
        peer.set_region_id(region_id);
        peer.set_truncated_index(488);
        for (epoch_id, first_index, last_index, last_term) in files {
            let file = new_rlog_file(epoch_id, first_index, last_index, last_term);
            if first_index != 470 {
                fs::write(
                    raft_log_file_name(tmp_path, peer_id, first_index, last_index),
                    b"rlog",
                )
                .unwrap();
            }
            peer.mut_files().push(file);
        }

        let mut change_set = rfenginepb::ChangeSet::new();
        change_set.set_epoch_id(1);
        change_set.mut_peers().push(peer);
        let mut manifest = Manifest::open(tmp_path, Arc::new(AtomicU64::new(999))).unwrap();
        manifest.handle_compaction(change_set).unwrap();

        let retained_files = manifest
            .peers
            .get(&peer_id)
            .unwrap()
            .files
            .iter()
            .copied()
            .collect::<Vec<_>>();
        assert_eq!(retained_files, vec![PeerFile::new(2166, 489, 489, 4645)]);
        assert!(raft_log_file_name(tmp_path, peer_id, 489, 489).exists());
        assert!(!raft_log_file_name(tmp_path, peer_id, 488, 488).exists());
        assert!(!raft_log_file_name(tmp_path, peer_id, 470, 470).exists());
    }

    #[test]
    fn test_truncate_files_keeps_path_referenced_by_retained_metadata() {
        let tmp_dir = tempfile::tempdir().unwrap();
        let tmp_path = tmp_dir.path();
        let peer_id = 1;
        let shared_file = PeerFile::new(1, 10, 20, 1);
        let rewritten_file = PeerFile::new(2, 10, 20, 2);

        let mut peer = PeerMetaFiles::new(peer_id, 0);
        peer.truncated_idx = 9;
        peer.files = VecDeque::from([shared_file, rewritten_file]);

        let mut manifest = Manifest::open(tmp_path, Arc::new(AtomicU64::new(999))).unwrap();
        manifest.peers.insert(peer_id, peer);
        write_rlog_file(
            tmp_path,
            peer_id,
            shared_file.first_index,
            shared_file.last_index,
        );

        manifest.truncate_files(&[peer_id]);

        let retained_files = manifest
            .peers
            .get(&peer_id)
            .unwrap()
            .files
            .iter()
            .copied()
            .collect::<Vec<_>>();
        assert_eq!(retained_files, vec![rewritten_file]);
        assert!(raft_log_file_name(tmp_path, peer_id, 10, 20).exists());
    }

    #[test]
    fn test_truncate_files_limits_work_to_selected_peers() {
        let tmp_dir = tempfile::tempdir().unwrap();
        let tmp_path = tmp_dir.path();
        let touched_peer_id = 1;
        let untouched_peer_id = 2;
        let obsolete_file = PeerFile::new(1, 1, 10, 1);
        let live_file = PeerFile::new(2, 11, 11, 2);

        let mut manifest = Manifest::open(tmp_path, Arc::new(AtomicU64::new(999))).unwrap();
        for peer_id in [touched_peer_id, untouched_peer_id] {
            let mut peer = PeerMetaFiles::new(peer_id, 0);
            peer.truncated_idx = 10;
            peer.files = VecDeque::from([obsolete_file, live_file]);
            manifest.peers.insert(peer_id, peer);
            write_rlog_file(
                tmp_path,
                peer_id,
                obsolete_file.first_index,
                obsolete_file.last_index,
            );
            write_rlog_file(
                tmp_path,
                peer_id,
                live_file.first_index,
                live_file.last_index,
            );
        }

        manifest.truncate_files(&[touched_peer_id]);

        let touched_files = manifest
            .peers
            .get(&touched_peer_id)
            .unwrap()
            .files
            .iter()
            .copied()
            .collect::<Vec<_>>();
        assert_eq!(touched_files, vec![live_file]);
        assert!(!raft_log_file_name(tmp_path, touched_peer_id, 1, 10).exists());
        assert!(raft_log_file_name(tmp_path, touched_peer_id, 11, 11).exists());

        let untouched_files = manifest
            .peers
            .get(&untouched_peer_id)
            .unwrap()
            .files
            .iter()
            .copied()
            .collect::<Vec<_>>();
        assert_eq!(untouched_files, vec![obsolete_file, live_file]);
        assert!(raft_log_file_name(tmp_path, untouched_peer_id, 1, 10).exists());
        assert!(raft_log_file_name(tmp_path, untouched_peer_id, 11, 11).exists());

        manifest.truncate_all_files();

        let untouched_files = manifest
            .peers
            .get(&untouched_peer_id)
            .unwrap()
            .files
            .iter()
            .copied()
            .collect::<Vec<_>>();
        assert_eq!(untouched_files, vec![live_file]);
        assert!(!raft_log_file_name(tmp_path, untouched_peer_id, 1, 10).exists());
        assert!(raft_log_file_name(tmp_path, untouched_peer_id, 11, 11).exists());
    }
}