use std::collections::HashMap;
use std::io;
use std::path::{Path, PathBuf};

use crate::io::{FileId, OpenOptions, StorageIO};

use super::data_file::{BLOCK_RECORD_OVERHEAD, CID_SIZE};
use super::list_files_by_extension;
use super::types::{
    BlockLength, BlockLocation, BlockOffset, CidBytes, CommitEpoch, DataFileId, HintOffset,
    MAX_BLOCK_SIZE, WallClockMs, WriteCursor,
};

pub const HINT_RECORD_SIZE: usize = 1 + 3 + CID_SIZE + 8 + 8 + 8;
pub const HINT_FILE_EXTENSION: &str = "tqh";

const _: () = assert!(HINT_RECORD_SIZE == 64);

const HINT_PAYLOAD_SIZE: usize = HINT_RECORD_SIZE - 8;

const RECORD_TYPE_PUT: u8 = 0x01;
const RECORD_TYPE_DECREMENT: u8 = 0x02;
const RECORD_TYPE_RELOCATE: u8 = 0x03;
const RECORD_TYPE_REMOVE: u8 = 0x04;
const RECORD_TYPE_COMMIT_MARKER: u8 = 0x05;

const HINT_FORMAT_VERSION: u8 = 1;

fn hint_checksum(payload: &[u8]) -> u64 {
    xxhash_rust::xxh3::xxh3_64(payload)
}

pub fn hint_file_path(data_dir: &Path, file_id: DataFileId) -> PathBuf {
    data_dir.join(format!("{file_id}.{HINT_FILE_EXTENSION}"))
}

const TYPE_OFFSET: usize = 0;
const VERSION_OFFSET: usize = 1;
const CID_OFFSET: usize = 4;
const FIELD_A_OFFSET: usize = CID_OFFSET + CID_SIZE;
const FIELD_B_OFFSET: usize = FIELD_A_OFFSET + 8;
const CHECKSUM_OFFSET: usize = FIELD_B_OFFSET + 8;

fn write_hint_record<S: StorageIO>(
    io: &S,
    fd: FileId,
    write_offset: HintOffset,
    record: &[u8; HINT_RECORD_SIZE],
) -> io::Result<()> {
    assert!(
        write_offset.raw().is_multiple_of(HINT_RECORD_SIZE as u64),
        "hint write_offset {} not aligned to HINT_RECORD_SIZE {}",
        write_offset.raw(),
        HINT_RECORD_SIZE,
    );
    io.write_all_at(fd, write_offset.raw(), record)
}

fn encode_location_fields(record: &mut [u8; HINT_RECORD_SIZE], loc: &BlockLocation) {
    record[FIELD_A_OFFSET..FIELD_A_OFFSET + 4].copy_from_slice(&loc.file_id.raw().to_le_bytes());
    record[FIELD_A_OFFSET + 4..FIELD_A_OFFSET + 8].copy_from_slice(&loc.length.raw().to_le_bytes());
    record[FIELD_B_OFFSET..FIELD_B_OFFSET + 8].copy_from_slice(&loc.offset.raw().to_le_bytes());
}

pub(crate) fn encode_hint_record<S: StorageIO>(
    io: &S,
    fd: FileId,
    write_offset: HintOffset,
    cid_bytes: &[u8; CID_SIZE],
    loc: &BlockLocation,
) -> io::Result<()> {
    let mut record = [0u8; HINT_RECORD_SIZE];
    record[TYPE_OFFSET] = RECORD_TYPE_PUT;
    record[VERSION_OFFSET] = HINT_FORMAT_VERSION;
    record[CID_OFFSET..CID_OFFSET + CID_SIZE].copy_from_slice(cid_bytes);
    encode_location_fields(&mut record, loc);

    let checksum = hint_checksum(&record[..HINT_PAYLOAD_SIZE]);
    record[CHECKSUM_OFFSET..].copy_from_slice(&checksum.to_le_bytes());

    write_hint_record(io, fd, write_offset, &record)
}

const REFCOUNT_OFFSET: usize = 2;

pub(crate) fn encode_relocate_record<S: StorageIO>(
    io: &S,
    fd: FileId,
    write_offset: HintOffset,
    cid_bytes: &[u8; CID_SIZE],
    loc: &BlockLocation,
    refcount: u32,
) -> io::Result<()> {
    let mut record = [0u8; HINT_RECORD_SIZE];
    record[TYPE_OFFSET] = RECORD_TYPE_RELOCATE;
    record[VERSION_OFFSET] = HINT_FORMAT_VERSION;
    let rc16 = u16::try_from(refcount).unwrap_or(u16::MAX);
    record[REFCOUNT_OFFSET..REFCOUNT_OFFSET + 2].copy_from_slice(&rc16.to_le_bytes());
    record[CID_OFFSET..CID_OFFSET + CID_SIZE].copy_from_slice(cid_bytes);
    encode_location_fields(&mut record, loc);

    let checksum = hint_checksum(&record[..HINT_PAYLOAD_SIZE]);
    record[CHECKSUM_OFFSET..].copy_from_slice(&checksum.to_le_bytes());

    write_hint_record(io, fd, write_offset, &record)
}

pub(crate) fn encode_remove_record<S: StorageIO>(
    io: &S,
    fd: FileId,
    write_offset: HintOffset,
    cid_bytes: &[u8; CID_SIZE],
) -> io::Result<()> {
    let mut record = [0u8; HINT_RECORD_SIZE];
    record[TYPE_OFFSET] = RECORD_TYPE_REMOVE;
    record[VERSION_OFFSET] = HINT_FORMAT_VERSION;
    record[CID_OFFSET..CID_OFFSET + CID_SIZE].copy_from_slice(cid_bytes);

    let checksum = hint_checksum(&record[..HINT_PAYLOAD_SIZE]);
    record[CHECKSUM_OFFSET..].copy_from_slice(&checksum.to_le_bytes());

    write_hint_record(io, fd, write_offset, &record)
}

pub(crate) fn encode_decrement_record<S: StorageIO>(
    io: &S,
    fd: FileId,
    write_offset: HintOffset,
    cid_bytes: &[u8; CID_SIZE],
    epoch: CommitEpoch,
    timestamp: WallClockMs,
) -> io::Result<()> {
    let mut record = [0u8; HINT_RECORD_SIZE];
    record[TYPE_OFFSET] = RECORD_TYPE_DECREMENT;
    record[VERSION_OFFSET] = HINT_FORMAT_VERSION;
    record[CID_OFFSET..CID_OFFSET + CID_SIZE].copy_from_slice(cid_bytes);
    record[FIELD_A_OFFSET..FIELD_A_OFFSET + 8].copy_from_slice(&epoch.raw().to_le_bytes());
    record[FIELD_B_OFFSET..FIELD_B_OFFSET + 8].copy_from_slice(&timestamp.raw().to_le_bytes());

    let checksum = hint_checksum(&record[..HINT_PAYLOAD_SIZE]);
    record[CHECKSUM_OFFSET..].copy_from_slice(&checksum.to_le_bytes());

    write_hint_record(io, fd, write_offset, &record)
}

const MARKER_DATA_OFFSET_POS: usize = CID_OFFSET;
const MARKER_DATA_FILE_ID_POS: usize = CID_OFFSET + 8;
const MARKER_RECORD_COUNT_POS: usize = FIELD_B_OFFSET;

pub(crate) fn encode_commit_marker_record<S: StorageIO>(
    io: &S,
    fd: FileId,
    write_offset: HintOffset,
    batch_seq: u64,
    record_count: u32,
    data_file_id: DataFileId,
    data_offset: BlockOffset,
) -> io::Result<()> {
    let mut record = [0u8; HINT_RECORD_SIZE];
    record[TYPE_OFFSET] = RECORD_TYPE_COMMIT_MARKER;
    record[VERSION_OFFSET] = HINT_FORMAT_VERSION;
    record[MARKER_DATA_OFFSET_POS..MARKER_DATA_OFFSET_POS + 8]
        .copy_from_slice(&data_offset.raw().to_le_bytes());
    record[MARKER_DATA_FILE_ID_POS..MARKER_DATA_FILE_ID_POS + 4]
        .copy_from_slice(&data_file_id.raw().to_le_bytes());
    record[FIELD_A_OFFSET..FIELD_A_OFFSET + 8].copy_from_slice(&batch_seq.to_le_bytes());
    record[MARKER_RECORD_COUNT_POS..MARKER_RECORD_COUNT_POS + 4]
        .copy_from_slice(&record_count.to_le_bytes());

    let checksum = hint_checksum(&record[..HINT_PAYLOAD_SIZE]);
    record[CHECKSUM_OFFSET..].copy_from_slice(&checksum.to_le_bytes());

    write_hint_record(io, fd, write_offset, &record)
}

#[must_use]
#[derive(Debug)]
pub enum ReadHintRecord {
    Put {
        cid_bytes: [u8; CID_SIZE],
        file_id: DataFileId,
        offset: BlockOffset,
        length: BlockLength,
    },
    Decrement {
        cid_bytes: [u8; CID_SIZE],
        epoch: CommitEpoch,
        timestamp: WallClockMs,
    },
    Relocate {
        cid_bytes: [u8; CID_SIZE],
        file_id: DataFileId,
        offset: BlockOffset,
        length: BlockLength,
        refcount: u32,
    },
    Remove {
        cid_bytes: [u8; CID_SIZE],
    },
    CommitMarker {
        batch_seq: u64,
        record_count: u32,
        data_file_id: DataFileId,
        data_offset: BlockOffset,
    },
    UnknownVersion {
        version: u8,
    },
    UnknownType {
        record_type: u8,
    },
    Corrupted,
    Truncated,
}

pub fn decode_hint_record<S: StorageIO>(
    io: &S,
    fd: FileId,
    read_offset: HintOffset,
    file_size: u64,
) -> io::Result<Option<ReadHintRecord>> {
    let raw = read_offset.raw();
    let remaining = match file_size.checked_sub(raw) {
        Some(r) => r,
        None => return Ok(None),
    };
    if remaining == 0 {
        return Ok(None);
    }

    if remaining < HINT_RECORD_SIZE as u64 {
        return Ok(Some(ReadHintRecord::Truncated));
    }

    let mut record = [0u8; HINT_RECORD_SIZE];
    io.read_exact_at(fd, raw, &mut record)?;

    let stored = u64::from_le_bytes(record[CHECKSUM_OFFSET..].try_into().unwrap());
    let computed = hint_checksum(&record[..HINT_PAYLOAD_SIZE]);
    if stored != computed {
        return Ok(Some(ReadHintRecord::Corrupted));
    }

    let version = record[VERSION_OFFSET];
    if version != HINT_FORMAT_VERSION {
        return Ok(Some(ReadHintRecord::UnknownVersion { version }));
    }

    let record_type = record[TYPE_OFFSET];

    let mut cid_bytes = [0u8; CID_SIZE];
    cid_bytes.copy_from_slice(&record[CID_OFFSET..CID_OFFSET + CID_SIZE]);

    match record_type {
        RECORD_TYPE_PUT => {
            let file_id = DataFileId::new(u32::from_le_bytes(
                record[FIELD_A_OFFSET..FIELD_A_OFFSET + 4]
                    .try_into()
                    .unwrap(),
            ));
            let raw_length = u32::from_le_bytes(
                record[FIELD_A_OFFSET + 4..FIELD_A_OFFSET + 8]
                    .try_into()
                    .unwrap(),
            );
            let block_offset = BlockOffset::new(u64::from_le_bytes(
                record[FIELD_B_OFFSET..FIELD_B_OFFSET + 8]
                    .try_into()
                    .unwrap(),
            ));
            if raw_length > MAX_BLOCK_SIZE {
                return Ok(Some(ReadHintRecord::Corrupted));
            }
            Ok(Some(ReadHintRecord::Put {
                cid_bytes,
                file_id,
                offset: block_offset,
                length: BlockLength::new(raw_length),
            }))
        }
        RECORD_TYPE_DECREMENT => {
            let epoch = CommitEpoch::new(u64::from_le_bytes(
                record[FIELD_A_OFFSET..FIELD_A_OFFSET + 8]
                    .try_into()
                    .unwrap(),
            ));
            let timestamp = WallClockMs::new(u64::from_le_bytes(
                record[FIELD_B_OFFSET..FIELD_B_OFFSET + 8]
                    .try_into()
                    .unwrap(),
            ));
            Ok(Some(ReadHintRecord::Decrement {
                cid_bytes,
                epoch,
                timestamp,
            }))
        }
        RECORD_TYPE_RELOCATE => {
            let rc16 = u16::from_le_bytes(
                record[REFCOUNT_OFFSET..REFCOUNT_OFFSET + 2]
                    .try_into()
                    .unwrap(),
            );
            let refcount = match rc16 {
                0 => 1,
                n => u32::from(n),
            };
            let file_id = DataFileId::new(u32::from_le_bytes(
                record[FIELD_A_OFFSET..FIELD_A_OFFSET + 4]
                    .try_into()
                    .unwrap(),
            ));
            let raw_length = u32::from_le_bytes(
                record[FIELD_A_OFFSET + 4..FIELD_A_OFFSET + 8]
                    .try_into()
                    .unwrap(),
            );
            let block_offset = BlockOffset::new(u64::from_le_bytes(
                record[FIELD_B_OFFSET..FIELD_B_OFFSET + 8]
                    .try_into()
                    .unwrap(),
            ));
            if raw_length > MAX_BLOCK_SIZE {
                return Ok(Some(ReadHintRecord::Corrupted));
            }
            Ok(Some(ReadHintRecord::Relocate {
                cid_bytes,
                file_id,
                offset: block_offset,
                length: BlockLength::new(raw_length),
                refcount,
            }))
        }
        RECORD_TYPE_REMOVE => Ok(Some(ReadHintRecord::Remove { cid_bytes })),
        RECORD_TYPE_COMMIT_MARKER => {
            let data_offset = BlockOffset::new(u64::from_le_bytes(
                record[MARKER_DATA_OFFSET_POS..MARKER_DATA_OFFSET_POS + 8]
                    .try_into()
                    .unwrap(),
            ));
            let data_file_id = DataFileId::new(u32::from_le_bytes(
                record[MARKER_DATA_FILE_ID_POS..MARKER_DATA_FILE_ID_POS + 4]
                    .try_into()
                    .unwrap(),
            ));
            let batch_seq = u64::from_le_bytes(
                record[FIELD_A_OFFSET..FIELD_A_OFFSET + 8]
                    .try_into()
                    .unwrap(),
            );
            let record_count = u32::from_le_bytes(
                record[MARKER_RECORD_COUNT_POS..MARKER_RECORD_COUNT_POS + 4]
                    .try_into()
                    .unwrap(),
            );
            Ok(Some(ReadHintRecord::CommitMarker {
                batch_seq,
                record_count,
                data_file_id,
                data_offset,
            }))
        }
        other => Ok(Some(ReadHintRecord::UnknownType { record_type: other })),
    }
}

pub struct HintFileWriter<'a, S: StorageIO> {
    io: &'a S,
    fd: FileId,
    position: HintOffset,
}

impl<'a, S: StorageIO> HintFileWriter<'a, S> {
    pub fn new(io: &'a S, fd: FileId) -> Self {
        Self {
            io,
            fd,
            position: HintOffset::new(0),
        }
    }

    pub fn resume(io: &'a S, fd: FileId, position: HintOffset) -> Self {
        Self { io, fd, position }
    }

    pub fn append_hint(
        &mut self,
        cid_bytes: &[u8; CID_SIZE],
        loc: &BlockLocation,
    ) -> io::Result<()> {
        encode_hint_record(self.io, self.fd, self.position, cid_bytes, loc)?;
        self.position = self.position.advance(HINT_RECORD_SIZE as u64);
        Ok(())
    }

    pub fn append_decrement(
        &mut self,
        cid_bytes: &[u8; CID_SIZE],
        epoch: CommitEpoch,
        timestamp: WallClockMs,
    ) -> io::Result<()> {
        encode_decrement_record(self.io, self.fd, self.position, cid_bytes, epoch, timestamp)?;
        self.position = self.position.advance(HINT_RECORD_SIZE as u64);
        Ok(())
    }

    pub fn append_relocate(
        &mut self,
        cid_bytes: &[u8; CID_SIZE],
        loc: &BlockLocation,
        refcount: u32,
    ) -> io::Result<()> {
        encode_relocate_record(self.io, self.fd, self.position, cid_bytes, loc, refcount)?;
        self.position = self.position.advance(HINT_RECORD_SIZE as u64);
        Ok(())
    }

    pub fn append_remove(&mut self, cid_bytes: &[u8; CID_SIZE]) -> io::Result<()> {
        encode_remove_record(self.io, self.fd, self.position, cid_bytes)?;
        self.position = self.position.advance(HINT_RECORD_SIZE as u64);
        Ok(())
    }

    pub fn append_commit_marker(
        &mut self,
        batch_seq: u64,
        record_count: u32,
        data_file_id: DataFileId,
        data_offset: BlockOffset,
    ) -> io::Result<()> {
        encode_commit_marker_record(
            self.io,
            self.fd,
            self.position,
            batch_seq,
            record_count,
            data_file_id,
            data_offset,
        )?;
        self.position = self.position.advance(HINT_RECORD_SIZE as u64);
        Ok(())
    }

    pub fn sync(&self) -> io::Result<()> {
        self.io.sync(self.fd)
    }

    pub fn position(&self) -> HintOffset {
        self.position
    }
}

pub struct HintFileReader<'a, S: StorageIO> {
    io: &'a S,
    fd: FileId,
    position: HintOffset,
    file_size: u64,
}

impl<'a, S: StorageIO> HintFileReader<'a, S> {
    pub fn open(io: &'a S, fd: FileId) -> io::Result<Self> {
        let file_size = io.file_size(fd)?;
        Ok(Self {
            io,
            fd,
            position: HintOffset::new(0),
            file_size,
        })
    }

    pub fn resume(io: &'a S, fd: FileId, position: HintOffset) -> io::Result<Self> {
        if !position.raw().is_multiple_of(HINT_RECORD_SIZE as u64) {
            return Err(io::Error::new(
                io::ErrorKind::InvalidData,
                format!(
                    "hint resume position {} not aligned to HINT_RECORD_SIZE {}",
                    position.raw(),
                    HINT_RECORD_SIZE,
                ),
            ));
        }
        let file_size = io.file_size(fd)?;
        Ok(Self {
            io,
            fd,
            position,
            file_size,
        })
    }
}

impl<S: StorageIO> Iterator for HintFileReader<'_, S> {
    type Item = io::Result<ReadHintRecord>;

    fn next(&mut self) -> Option<Self::Item> {
        match decode_hint_record(self.io, self.fd, self.position, self.file_size) {
            Err(e) => {
                self.position = HintOffset::new(self.file_size);
                Some(Err(e))
            }
            Ok(None) => None,
            Ok(Some(record)) => {
                match &record {
                    ReadHintRecord::Put { .. }
                    | ReadHintRecord::Decrement { .. }
                    | ReadHintRecord::Relocate { .. }
                    | ReadHintRecord::Remove { .. }
                    | ReadHintRecord::CommitMarker { .. }
                    | ReadHintRecord::UnknownType { .. }
                    | ReadHintRecord::UnknownVersion { .. }
                    | ReadHintRecord::Corrupted => {
                        self.position = self.position.advance(HINT_RECORD_SIZE as u64);
                    }
                    ReadHintRecord::Truncated => {
                        self.position = HintOffset::new(self.file_size);
                    }
                }
                Some(Ok(record))
            }
        }
    }
}

#[derive(Debug)]
pub enum RebuildError {
    Io(io::Error),
    BlockIndex(super::hash_index::BlockIndexError),
}

impl std::fmt::Display for RebuildError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Io(e) => write!(f, "io: {e}"),
            Self::BlockIndex(e) => write!(f, "block index: {e}"),
        }
    }
}

impl std::error::Error for RebuildError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            Self::Io(e) => Some(e),
            Self::BlockIndex(e) => Some(e),
        }
    }
}

impl From<io::Error> for RebuildError {
    fn from(e: io::Error) -> Self {
        Self::Io(e)
    }
}

impl From<super::hash_index::BlockIndexError> for RebuildError {
    fn from(e: super::hash_index::BlockIndexError) -> Self {
        Self::BlockIndex(e)
    }
}

struct RebuildState {
    entries: Vec<([u8; CID_SIZE], BlockLocation)>,
    cursor_file: DataFileId,
    cursor_offset: BlockOffset,
}

impl RebuildState {
    fn new() -> Self {
        Self {
            entries: Vec::new(),
            cursor_file: DataFileId::new(0),
            cursor_offset: BlockOffset::new(0),
        }
    }

    fn push(&mut self, cid_bytes: [u8; CID_SIZE], location: BlockLocation) {
        let end = location
            .offset
            .advance(BLOCK_RECORD_OVERHEAD as u64 + location.length.as_u64());
        if (location.file_id, end) > (self.cursor_file, self.cursor_offset) {
            self.cursor_file = location.file_id;
            self.cursor_offset = end;
        }
        self.entries.push((cid_bytes, location));
    }

    fn cursor(&self) -> WriteCursor {
        WriteCursor {
            file_id: self.cursor_file,
            offset: self.cursor_offset,
        }
    }
}

fn scan_single_hint_file<S: StorageIO>(
    io: &S,
    data_dir: &Path,
    hf_id: DataFileId,
) -> Result<Vec<([u8; CID_SIZE], BlockLocation)>, RebuildError> {
    let path = hint_file_path(data_dir, hf_id);
    let fd = io.open(&path, OpenOptions::read_only_existing())?;
    let reader = HintFileReader::open(io, fd)?;

    let entries: Result<Vec<_>, RebuildError> = reader
        .filter_map(|r| match r {
            Ok(ReadHintRecord::Put {
                cid_bytes,
                file_id,
                offset,
                length,
            }) => Some(Ok((
                cid_bytes,
                BlockLocation {
                    file_id,
                    offset,
                    length,
                },
            ))),
            Ok(
                ReadHintRecord::Decrement { .. }
                | ReadHintRecord::Relocate { .. }
                | ReadHintRecord::Remove { .. }
                | ReadHintRecord::CommitMarker { .. }
                | ReadHintRecord::UnknownVersion { .. }
                | ReadHintRecord::UnknownType { .. }
                | ReadHintRecord::Corrupted
                | ReadHintRecord::Truncated,
            ) => None,
            Err(e) => Some(Err(RebuildError::Io(e))),
        })
        .collect();

    let _ = io.close(fd);
    entries
}

#[derive(Default)]
struct PendingBatch {
    puts: Vec<([u8; CID_SIZE], BlockLocation)>,
    relocates: Vec<([u8; CID_SIZE], BlockLocation, u32)>,
    removes: Vec<[u8; CID_SIZE]>,
    decrements: Vec<([u8; CID_SIZE], CommitEpoch, WallClockMs)>,
    file_cursors: HashMap<DataFileId, BlockOffset>,
    max_cursor: Option<WriteCursor>,
    record_count: u32,
    boundary_lost: bool,
}

impl PendingBatch {
    fn reset(&mut self) {
        self.puts.clear();
        self.relocates.clear();
        self.removes.clear();
        self.decrements.clear();
        self.file_cursors.clear();
        self.max_cursor = None;
        self.record_count = 0;
        self.boundary_lost = false;
    }

    fn note_record(&mut self) {
        self.record_count = self.record_count.saturating_add(1);
    }

    fn track_cursor(&mut self, file_id: DataFileId, end: BlockOffset) {
        let candidate = WriteCursor {
            file_id,
            offset: end,
        };
        self.max_cursor = Some(match self.max_cursor {
            Some(c) => std::cmp::max_by_key(c, candidate, |w| (w.file_id, w.offset)),
            None => candidate,
        });
        self.file_cursors
            .entry(file_id)
            .and_modify(|existing| {
                if end > *existing {
                    *existing = end;
                }
            })
            .or_insert(end);
    }
}

fn commit_pending_batch(
    pending: &mut PendingBatch,
    index: &super::hash_index::BlockIndex,
    file_cursors: &mut HashMap<DataFileId, BlockOffset>,
    max_cursor: &mut Option<WriteCursor>,
    replayed: &mut u64,
) -> Result<(), RebuildError> {
    if !pending.puts.is_empty() {
        index.batch_insert_buffered(&pending.puts)?;
    }
    if !pending.relocates.is_empty() {
        index.batch_relocate(&pending.relocates)?;
    }
    if !pending.removes.is_empty() {
        index.batch_remove(&pending.removes);
    }
    pending
        .decrements
        .iter()
        .try_for_each(|(cid, epoch, ts)| index.batch_decrement(&[*cid], *epoch, *ts))?;

    pending.file_cursors.iter().for_each(|(fid, end)| {
        file_cursors
            .entry(*fid)
            .and_modify(|existing| {
                if *end > *existing {
                    *existing = *end;
                }
            })
            .or_insert(*end);
    });
    if let Some(c) = pending.max_cursor {
        *max_cursor = Some(match *max_cursor {
            Some(m) => std::cmp::max_by_key(m, c, |w| (w.file_id, w.offset)),
            None => c,
        });
    }
    *replayed = replayed.saturating_add(u64::from(pending.record_count));
    pending.reset();
    Ok(())
}

pub fn replay_hints_into_block_index<S: StorageIO>(
    io: &S,
    data_dir: &Path,
    index: &super::hash_index::BlockIndex,
    from: Option<&super::hash_index::CheckpointPositions>,
) -> Result<(u64, HashMap<DataFileId, BlockOffset>), RebuildError> {
    let hint_files = list_files_by_extension(io, data_dir, HINT_FILE_EXTENSION)?;
    if hint_files.is_empty() {
        return Ok((0, HashMap::new()));
    }

    let checkpointed_files: HashMap<DataFileId, HintOffset> = from
        .map(|cp| cp.0.iter().copied().collect())
        .unwrap_or_default();

    let max_checkpointed_fid = checkpointed_files
        .keys()
        .max()
        .copied()
        .unwrap_or(DataFileId::new(0));

    let mut max_cursor: Option<WriteCursor> = None;
    let mut file_cursors: HashMap<DataFileId, BlockOffset> = HashMap::new();
    let mut replayed: u64 = 0;
    let mut pending = PendingBatch::default();

    hint_files
        .iter()
        .filter_map(|&fid| match checkpointed_files.get(&fid) {
            Some(&offset) => Some((fid, offset)),
            None if fid > max_checkpointed_fid => Some((fid, HintOffset::new(0))),
            None => None,
        })
        .try_for_each(|(fid, start_pos)| {
            let path = hint_file_path(data_dir, fid);
            let fd = match io.open(&path, OpenOptions::read_only_existing()) {
                Ok(fd) => fd,
                Err(e) if e.kind() == io::ErrorKind::NotFound => return Ok(()),
                Err(e) => return Err(RebuildError::Io(e)),
            };

            let mut reader = HintFileReader::resume(io, fd, start_pos)?;

            reader.try_for_each(|record_result| {
                match record_result? {
                    ReadHintRecord::Put {
                        cid_bytes,
                        file_id,
                        offset,
                        length,
                    } => {
                        let loc = BlockLocation {
                            file_id,
                            offset,
                            length,
                        };
                        let record_end =
                            offset.advance(BLOCK_RECORD_OVERHEAD as u64 + length.as_u64());
                        pending.puts.push((cid_bytes, loc));
                        pending.track_cursor(file_id, record_end);
                        pending.note_record();
                    }
                    ReadHintRecord::Decrement {
                        cid_bytes,
                        epoch,
                        timestamp,
                    } => {
                        pending.decrements.push((cid_bytes, epoch, timestamp));
                        pending.note_record();
                    }
                    ReadHintRecord::Relocate {
                        cid_bytes,
                        file_id,
                        offset,
                        length,
                        refcount,
                    } => {
                        let loc = BlockLocation {
                            file_id,
                            offset,
                            length,
                        };
                        let record_end =
                            offset.advance(BLOCK_RECORD_OVERHEAD as u64 + length.as_u64());
                        pending.relocates.push((cid_bytes, loc, refcount));
                        pending.track_cursor(file_id, record_end);
                        pending.note_record();
                    }
                    ReadHintRecord::Remove { cid_bytes } => {
                        pending.removes.push(cid_bytes);
                        pending.note_record();
                    }
                    ReadHintRecord::CommitMarker {
                        batch_seq,
                        record_count,
                        data_file_id,
                        data_offset,
                    } => {
                        let accepts =
                            !pending.boundary_lost && pending.record_count == record_count;
                        match accepts {
                            true => {
                                pending.track_cursor(data_file_id, data_offset);
                                commit_pending_batch(
                                    &mut pending,
                                    index,
                                    &mut file_cursors,
                                    &mut max_cursor,
                                    &mut replayed,
                                )?;
                            }
                            false => {
                                tracing::warn!(
                                    file_id = %fid,
                                    batch_seq,
                                    expected_count = record_count,
                                    observed_count = pending.record_count,
                                    boundary_lost = pending.boundary_lost,
                                    "rolling back torn hint batch"
                                );
                                pending.reset();
                            }
                        }
                    }
                    ReadHintRecord::Corrupted
                    | ReadHintRecord::UnknownVersion { .. }
                    | ReadHintRecord::UnknownType { .. } => {
                        pending.boundary_lost = true;
                    }
                    ReadHintRecord::Truncated => {}
                }
                Ok::<_, RebuildError>(())
            })?;

            let _ = io.close(fd);
            Ok(())
        })?;

    if pending.record_count > 0 || pending.boundary_lost {
        tracing::warn!(
            record_count = pending.record_count,
            boundary_lost = pending.boundary_lost,
            "discarding unterminated hint batch at replay end"
        );
        pending.reset();
    }

    if let Some(cursor) = max_cursor {
        index.set_write_cursor(cursor)?;
    }

    Ok((replayed, file_cursors))
}

#[derive(Debug)]
pub struct HintIndex {
    entries: HashMap<CidBytes, BlockLocation>,
}

impl HintIndex {
    pub fn from_scanned(scanned: Vec<(CidBytes, BlockLocation)>) -> Self {
        let mut entries = HashMap::with_capacity(scanned.len());
        scanned.into_iter().for_each(|(cid, loc)| {
            entries.entry(cid).or_insert(loc);
        });
        Self { entries }
    }

    pub fn get(&self, cid: &[u8; CID_SIZE]) -> Option<BlockLocation> {
        self.entries.get(cid).copied()
    }

    pub fn contains(&self, cid: &[u8; CID_SIZE]) -> bool {
        self.entries.contains_key(cid)
    }

    pub fn len(&self) -> usize {
        self.entries.len()
    }

    pub fn is_empty(&self) -> bool {
        self.entries.is_empty()
    }
}

pub fn scan_hints_to_memory<S: StorageIO>(
    io: &S,
    data_dir: &Path,
) -> Result<(HintIndex, WriteCursor), RebuildError> {
    use rayon::iter::{IntoParallelRefIterator, ParallelIterator};

    let hint_files = list_files_by_extension(io, data_dir, HINT_FILE_EXTENSION)?;
    if hint_files.is_empty() {
        return Err(RebuildError::Io(io::Error::new(
            io::ErrorKind::NotFound,
            "no hint files found for instant recovery",
        )));
    }

    let file_results: Vec<Result<Vec<_>, RebuildError>> = hint_files
        .par_iter()
        .map(|&hf_id| scan_single_hint_file(io, data_dir, hf_id))
        .collect();

    let mut state = RebuildState::new();
    file_results.into_iter().try_for_each(|result| {
        result?.into_iter().for_each(|(cid_bytes, location)| {
            state.push(cid_bytes, location);
        });
        Ok::<_, RebuildError>(())
    })?;

    if state.entries.is_empty() {
        return Err(RebuildError::Io(io::Error::new(
            io::ErrorKind::InvalidData,
            "hint files contained no valid entries",
        )));
    }

    let cursor = state.cursor();
    let hint_index = HintIndex::from_scanned(state.entries);

    Ok((hint_index, cursor))
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::OpenOptions;
    use crate::blockstore::test_cid;
    use crate::sim::SimulatedIO;
    use std::path::Path;

    fn setup() -> (SimulatedIO, FileId) {
        let sim = SimulatedIO::pristine(42);
        let dir = Path::new("/test");
        sim.mkdir(dir).unwrap();
        sim.sync_dir(dir).unwrap();
        let fd = sim
            .open(Path::new("/test/hints.tqh"), OpenOptions::read_write())
            .unwrap();
        (sim, fd)
    }

    #[test]
    fn hint_record_round_trip() {
        let (sim, fd) = setup();
        let cid = test_cid(1);
        let file_id = DataFileId::new(3);
        let offset = BlockOffset::new(1024);
        let length = BlockLength::new(256);

        let loc = BlockLocation {
            file_id,
            offset,
            length,
        };
        encode_hint_record(&sim, fd, HintOffset::new(0), &cid, &loc).unwrap();

        let file_size = sim.file_size(fd).unwrap();
        let record = decode_hint_record(&sim, fd, HintOffset::new(0), file_size)
            .unwrap()
            .unwrap();

        match record {
            ReadHintRecord::Put {
                cid_bytes,
                file_id: fid,
                offset: off,
                length: len,
            } => {
                assert_eq!(cid_bytes, cid);
                assert_eq!(fid, file_id);
                assert_eq!(off, offset);
                assert_eq!(len, length);
            }
            other => panic!("expected Valid, got {other:?}"),
        }
    }

    #[test]
    fn decrement_record_round_trip() {
        let (sim, fd) = setup();
        let cid = test_cid(42);
        let epoch = CommitEpoch::new(7);
        let timestamp = WallClockMs::new(1_700_000_000_000);

        encode_decrement_record(&sim, fd, HintOffset::new(0), &cid, epoch, timestamp).unwrap();

        let file_size = sim.file_size(fd).unwrap();
        let record = decode_hint_record(&sim, fd, HintOffset::new(0), file_size)
            .unwrap()
            .unwrap();

        match record {
            ReadHintRecord::Decrement {
                cid_bytes,
                epoch: decoded_epoch,
                timestamp: decoded_ts,
            } => {
                assert_eq!(cid_bytes, cid);
                assert_eq!(decoded_epoch, epoch);
                assert_eq!(decoded_ts, timestamp);
            }
            other => panic!("expected Decrement, got {other:?}"),
        }
    }

    #[test]
    fn multiple_hint_records() {
        let (sim, fd) = setup();

        (0u8..5).for_each(|i| {
            let cid = test_cid(i);
            let write_offset = HintOffset::new(i as u64 * HINT_RECORD_SIZE as u64);
            let loc = BlockLocation {
                file_id: DataFileId::new(i as u32),
                offset: BlockOffset::new(i as u64 * 100),
                length: BlockLength::new(50 + i as u32),
            };
            encode_hint_record(&sim, fd, write_offset, &cid, &loc).unwrap();
        });

        let file_size = sim.file_size(fd).unwrap();
        assert_eq!(file_size, 5 * HINT_RECORD_SIZE as u64);

        let records: Vec<_> = (0u8..5)
            .map(|i| {
                let read_offset = HintOffset::new(i as u64 * HINT_RECORD_SIZE as u64);
                decode_hint_record(&sim, fd, read_offset, file_size)
                    .unwrap()
                    .unwrap()
            })
            .collect();

        records.iter().enumerate().for_each(|(i, r)| match r {
            ReadHintRecord::Put {
                file_id, length, ..
            } => {
                assert_eq!(file_id.raw(), i as u32);
                assert_eq!(length.raw(), 50 + i as u32);
            }
            other => panic!("expected Valid at index {i}, got {other:?}"),
        });
    }

    #[test]
    fn detects_truncated_hint() {
        let (sim, fd) = setup();
        sim.write_all_at(fd, 0, &[0u8; HINT_RECORD_SIZE - 1])
            .unwrap();
        let file_size = sim.file_size(fd).unwrap();
        let record = decode_hint_record(&sim, fd, HintOffset::new(0), file_size)
            .unwrap()
            .unwrap();
        assert!(matches!(record, ReadHintRecord::Truncated));
    }

    #[test]
    fn detects_corrupted_hint() {
        let (sim, fd) = setup();
        let cid = test_cid(1);
        let loc = BlockLocation {
            file_id: DataFileId::new(0),
            offset: BlockOffset::new(0),
            length: BlockLength::new(100),
        };
        encode_hint_record(&sim, fd, HintOffset::new(0), &cid, &loc).unwrap();

        sim.write_all_at(fd, 10, &[0xFF]).unwrap();

        let file_size = sim.file_size(fd).unwrap();
        let record = decode_hint_record(&sim, fd, HintOffset::new(0), file_size)
            .unwrap()
            .unwrap();
        assert!(matches!(record, ReadHintRecord::Corrupted));
    }

    #[test]
    fn returns_none_at_eof() {
        let (sim, fd) = setup();
        let file_size = sim.file_size(fd).unwrap();
        assert!(
            decode_hint_record(&sim, fd, HintOffset::new(0), file_size)
                .unwrap()
                .is_none()
        );
    }

    #[test]
    fn oversized_length_treated_as_corrupted() {
        let (sim, fd) = setup();
        let cid = test_cid(1);
        let loc = BlockLocation {
            file_id: DataFileId::new(0),
            offset: BlockOffset::new(0),
            length: BlockLength::new(100),
        };
        encode_hint_record(&sim, fd, HintOffset::new(0), &cid, &loc).unwrap();

        let length_offset = FIELD_A_OFFSET as u64 + 4;
        let oversized = (MAX_BLOCK_SIZE + 1).to_le_bytes();
        sim.write_all_at(fd, length_offset, &oversized).unwrap();

        let mut buf = [0u8; HINT_PAYLOAD_SIZE];
        sim.read_exact_at(fd, 0, &mut buf).unwrap();
        let fixed_checksum = hint_checksum(&buf);
        sim.write_all_at(fd, CHECKSUM_OFFSET as u64, &fixed_checksum.to_le_bytes())
            .unwrap();

        let file_size = sim.file_size(fd).unwrap();
        let record = decode_hint_record(&sim, fd, HintOffset::new(0), file_size)
            .unwrap()
            .unwrap();
        assert!(matches!(record, ReadHintRecord::Corrupted));
    }

    #[test]
    fn hint_writer_writes_readable_records() {
        let (sim, fd) = setup();
        let mut writer = HintFileWriter::new(&sim, fd);

        (0u8..5).for_each(|i| {
            let loc = BlockLocation {
                file_id: DataFileId::new(0),
                offset: BlockOffset::new(i as u64 * 100),
                length: BlockLength::new(50 + i as u32),
            };
            writer.append_hint(&test_cid(i), &loc).unwrap();
        });

        assert_eq!(
            writer.position(),
            HintOffset::new(5 * HINT_RECORD_SIZE as u64)
        );

        let reader = HintFileReader::open(&sim, fd).unwrap();
        let records: Vec<_> = reader.map(|r| r.unwrap()).collect();
        assert_eq!(records.len(), 5);

        records.iter().enumerate().for_each(|(i, r)| match r {
            ReadHintRecord::Put {
                file_id, length, ..
            } => {
                assert_eq!(file_id.raw(), 0);
                assert_eq!(length.raw(), 50 + i as u32);
            }
            other => panic!("expected Valid at {i}, got {other:?}"),
        });
    }

    #[test]
    fn hint_writer_resume_continues_at_position() {
        let (sim, fd) = setup();
        let mut writer = HintFileWriter::new(&sim, fd);
        let loc0 = BlockLocation {
            file_id: DataFileId::new(0),
            offset: BlockOffset::new(0),
            length: BlockLength::new(100),
        };
        writer.append_hint(&test_cid(0), &loc0).unwrap();

        let pos = writer.position();
        let mut writer2 = HintFileWriter::resume(&sim, fd, pos);
        let loc1 = BlockLocation {
            file_id: DataFileId::new(0),
            offset: BlockOffset::new(100),
            length: BlockLength::new(200),
        };
        writer2.append_hint(&test_cid(1), &loc1).unwrap();

        let reader = HintFileReader::open(&sim, fd).unwrap();
        let valid_count = reader
            .filter_map(|r| match r.ok()? {
                ReadHintRecord::Put { .. } => Some(()),
                _ => None,
            })
            .count();
        assert_eq!(valid_count, 2);
    }

    #[test]
    fn hint_reader_resume_rejects_unaligned_position_without_panicking() {
        let (sim, fd) = setup();
        match HintFileReader::resume(&sim, fd, HintOffset::new(HINT_RECORD_SIZE as u64 + 5)) {
            Err(e) => assert_eq!(e.kind(), io::ErrorKind::InvalidData),
            Ok(_) => panic!("non-aligned resume position must surface as a recoverable error"),
        }
    }

    #[test]
    fn hint_reader_empty_file() {
        let (sim, fd) = setup();
        let reader = HintFileReader::open(&sim, fd).unwrap();
        assert_eq!(reader.count(), 0);
    }

    #[test]
    fn hint_reader_stops_on_truncated() {
        let (sim, fd) = setup();
        let mut writer = HintFileWriter::new(&sim, fd);
        let loc = BlockLocation {
            file_id: DataFileId::new(0),
            offset: BlockOffset::new(0),
            length: BlockLength::new(100),
        };
        writer.append_hint(&test_cid(0), &loc).unwrap();

        sim.write_all_at(fd, writer.position().raw(), &[0u8; HINT_RECORD_SIZE - 1])
            .unwrap();

        let reader = HintFileReader::open(&sim, fd).unwrap();
        let records: Vec<_> = reader.map(|r| r.unwrap()).collect();
        assert_eq!(records.len(), 2);
        assert!(matches!(records[0], ReadHintRecord::Put { .. }));
        assert!(matches!(records[1], ReadHintRecord::Truncated));
    }

    #[test]
    fn hint_reader_reports_corrupted_and_continues() {
        let (sim, fd) = setup();
        let mut writer = HintFileWriter::new(&sim, fd);

        (0u8..3).for_each(|i| {
            let loc = BlockLocation {
                file_id: DataFileId::new(0),
                offset: BlockOffset::new(i as u64 * 100),
                length: BlockLength::new(50),
            };
            writer.append_hint(&test_cid(i), &loc).unwrap();
        });

        sim.write_all_at(fd, HINT_RECORD_SIZE as u64 + 5, &[0xFF])
            .unwrap();

        let reader = HintFileReader::open(&sim, fd).unwrap();
        let records: Vec<_> = reader.map(|r| r.unwrap()).collect();
        assert_eq!(records.len(), 3);
        assert!(matches!(records[0], ReadHintRecord::Put { .. }));
        assert!(matches!(records[1], ReadHintRecord::Corrupted));
        assert!(matches!(records[2], ReadHintRecord::Put { .. }));
    }

    #[test]
    fn commit_marker_round_trip() {
        let (sim, fd) = setup();
        let data_file_id = DataFileId::new(7);
        let data_offset = BlockOffset::new(9_876);

        encode_commit_marker_record(
            &sim,
            fd,
            HintOffset::new(0),
            42,
            128,
            data_file_id,
            data_offset,
        )
        .unwrap();

        let file_size = sim.file_size(fd).unwrap();
        let record = decode_hint_record(&sim, fd, HintOffset::new(0), file_size)
            .unwrap()
            .unwrap();

        match record {
            ReadHintRecord::CommitMarker {
                batch_seq,
                record_count,
                data_file_id: fid,
                data_offset: off,
            } => {
                assert_eq!(batch_seq, 42);
                assert_eq!(record_count, 128);
                assert_eq!(fid, data_file_id);
                assert_eq!(off, data_offset);
            }
            other => panic!("expected CommitMarker, got {other:?}"),
        }
    }

    #[test]
    fn hint_file_path_format() {
        let path = hint_file_path(Path::new("/data"), DataFileId::new(0));
        assert_eq!(path, Path::new("/data/000000.tqh"));

        let path = hint_file_path(Path::new("/data"), DataFileId::new(42));
        assert_eq!(path, Path::new("/data/000042.tqh"));
    }
}
