whlog 0.1.0

use std::sync::atomic::Ordering;

use wdev::Device;
use wram::AlignedBuf;
use wrecord::{HEADER_SIZE, RecordHeader, RecordRef};

use crate::{
  address::AddressManager,
  error::{Error, Result},
  hlog::{HybridLog, PAD_KEY_LEN, PageBytes, parse_record_from_slice},
  output::RecordOutput,
};

/// 扫描单条记录的零拷贝视图：逻辑地址、记录引用与整条物理字节切片
pub struct ScanItem<'a> {
  /// 记录起始逻辑地址
  pub addr: u64,
  /// 零拷贝记录引用（借用自页缓冲或磁盘预取页）
  pub rec: RecordRef<'a>,
  /// 整条记录的物理字节切片（含松弛填充，逻辑长度 = physical_size）
  pub bytes: &'a [u8],
}

/// 逻辑日志扫描迭代器（支持磁盘冷数据与内存热数据混合连续扫描）
///
/// 严格对标 Microsoft Garnet Tsavorite 中的 ScanIteratorBase 与 TsavoriteLogScanIterator：
/// - 自动跳过换页填充（Pad 记录、全零字节或极小子头残片），直达下一页开头；
/// - 针对磁盘冷数据区启用单页缓冲预取（SinglePageBuffering），消除同页内重复磁盘 I/O；
/// - 针对内存驻留区实现零拷贝直读（只读区无锁裸读；可变区页读锁，撕裂保护）。
///
/// 快照式可变区判定：[Self::new] 构造时一次性快照 `read_only` 边界供
/// [AddressManager::is_mutable_snapshot] 贯穿使用（快照只允许偏旧；`read_only` 单调递增，
/// 偏旧只会把可变区误判为只读区而走保守读锁路径，绝不会把只读区误判为可变区，
/// 判定方向天然安全），消除逐记录的重复 Acquire 加载。
///
/// 刻意不快照的边界（正确性优先的取舍）：`head` 与 `flushed_until` 保持逐迭代新鲜加载——
/// `flushed_until` 快照偏旧会把「已落盘且已滑出内存窗口」的记录误判为「尚未落盘」而整页跳过，
/// 导致扫描丢记录（这是丢数据方向的误判，而非保守方向），故该处不能采用 dev 侧的构造期全快照优化。
///
/// # 调用方契约（无锁直读路径）
/// 只读区无锁裸读（`PageBytes::Raw`）要求调用线程处于 `LightEpoch` 保护下（同
/// [HybridLog::probe_resident] 契约）：页槽位回收的前置条件是 `safe_head` 经纪元排空
/// 越过旧页，持守卫期间页内存绝不会被清空复用；未持守卫时最坏情形为并发驱逐窗口内
/// 读到撕裂字节（解析报错或按 Pad 跳过，绝无悬垂 UB——页内存随实例存活），与 C#
/// 扫描器在内存区读取须持纪元的语义一致。
pub struct ScanIterator<'a, D: Device> {
  hlog: &'a HybridLog<D>,
  curr_addr: u64,
  end_addr: u64,
  /// 构造期快照的 ReadOnlyAddress（单调递增，偏旧只走保守读锁路径，见结构体文档）
  read_only: u64,
  /// 缓存的磁盘页面数据及对应的逻辑页号，避免同一页内多次重复磁盘 I/O
  disk_page_cache: Option<(u64, AlignedBuf)>,
}

impl<'a, D: Device> ScanIterator<'a, D> {
  /// 创建新的扫描迭代器
  pub fn new(hlog: &'a HybridLog<D>, begin_addr: u64, end_addr: u64) -> Self {
    let begin = hlog.addresses.begin_address.load(Ordering::Acquire);
    let read_only = hlog.addresses.read_only_address.load(Ordering::Acquire);
    Self {
      hlog,
      curr_addr: begin_addr.max(begin),
      end_addr,
      read_only,
      disk_page_cache: None,
    }
  }

  /// 当前扫描游标逻辑地址
  #[inline]
  pub const fn current_address(&self) -> u64 {
    self.curr_addr
  }

  /// 异步拉取下一条记录并以零拷贝 [ScanItem] 交付闭包消费
  ///
  /// 单一遍历引擎：推模式 [HybridLog::scan]、拉模式 [Self::next] 与缓冲复用变体
  /// [Self::next_into] 均构建于此，三区（磁盘冷读 / 只读直读 / 可变读锁）分派、
  /// Pad 与子头残片跳页、提前终止逻辑全部收敛在本方法内闭环。
  pub async fn next_ref<R>(
    &mut self,
    f: impl FnOnce(ScanItem<'_>) -> Result<R>,
  ) -> Result<Option<R>> {
    let tail = self.hlog.addresses.tail_address.load(Ordering::Acquire);
    let effective_end = self.end_addr.min(tail);

    while self.curr_addr < effective_end {
      let curr_addr = self.curr_addr;
      let page_id = self.hlog.config.page_id(curr_addr);
      let offset = self.hlog.config.page_offset(curr_addr);
      let page_size = self.hlog.config.page_size;

      // 若页内剩余空间不足以容纳记录头，说明已达页尾残片（如 0xFF 填充），跳至下一页开头
      if offset + HEADER_SIZE > page_size {
        self.skip_to_next_page(page_id);
        continue;
      }

      let head = self.hlog.addresses.head_address.load(Ordering::Acquire);
      let bytes = if curr_addr < head {
        // 1. 磁盘区数据：取出单页预取缓存（未命中则整页冷读）
        match self.disk_page_cache.take() {
          Some((cached_page, buf)) if cached_page == page_id => PageBytes::Disk(buf),
          _ => PageBytes::Disk(
            self
              .hlog
              .device
              .read_range(self.hlog.config.page_start_address(page_id), page_size)
              .await?,
          ),
        }
      } else {
        // 2. 越过磁盘区，释放磁盘预取缓存以回收内存
        self.disk_page_cache = None;

        let flushed_until = self.hlog.addresses.flushed_until();
        if !AddressManager::is_mutable_snapshot(curr_addr, head, self.read_only, tail)
          && let Some(page_slice) = unsafe { self.hlog.buffer.try_read_page_unlocked(page_id) }
          && curr_addr >= self.hlog.addresses.head()
        {
          // 3. 不可变只读区：无锁纯指针直读
          PageBytes::Raw(page_slice)
        } else {
          // 4. 可变区或无锁直读未命中的页：先取页读锁，再在锁内校验页号与 head——
          //    「先校验后加锁」存在 TOCTOU：校验通过到加锁之间页可被换页驱逐复用，
          //    锁内读到的将是新页数据被按旧偏移误解析。页清空/标定均持写锁，
          //    读锁持有期间页号稳定，锁内双重校验通过即保证读取期间槽位恒为目标页
          let guard = self.hlog.buffer.read_page(page_id);
          if curr_addr >= self.hlog.addresses.head() && self.hlog.buffer.is_page_loaded(page_id) {
            PageBytes::Locked(guard)
          } else {
            drop(guard);
            if curr_addr < flushed_until {
              // 5. 已落盘但当前未驻留内存（并发状态推进过渡期）：回退走底层设备读取
              PageBytes::Disk(
                self
                  .hlog
                  .device
                  .read_range(self.hlog.config.page_start_address(page_id), page_size)
                  .await?,
              )
            } else {
              self.skip_to_next_page(page_id);
              continue;
            }
          }
        }
      };

      match parse_record_from_slice(&bytes, offset, curr_addr, page_size) {
        Ok(rec) => {
          let physical_size = rec.physical_size();
          let out = f(ScanItem {
            addr: curr_addr,
            rec,
            bytes: &bytes[offset..offset + physical_size],
          })?;
          // 磁盘页消费完毕后回填预取缓存（同页后续记录零 I/O）
          if let PageBytes::Disk(buf) = bytes {
            self.disk_page_cache = Some((page_id, buf));
          }
          self.advance(curr_addr, physical_size, page_id);
          return Ok(Some(out));
        }
        Err(Error::PadRecord(_)) => {
          // 换页填充 / 恢复清洗零区 / 页尾残片处理：
          // - PAD 头（复活槽位中段亦可能出现，见 revivify_record_at）按 16 + val_len
          //   精确越过填充区，避免误跳同页后续记录；
          // - 零区与子头残片仅出现于页尾或恢复清洗区，直达下一页开头。
          let pad_step = bytes
            .get(offset..offset + HEADER_SIZE)
            .and_then(|h| RecordHeader::from_slice(h).ok())
            .filter(|h| h.key_len == PAD_KEY_LEN)
            .map(|h| (HEADER_SIZE + h.val_len as usize).min(page_size - offset));
          match pad_step {
            Some(step) if offset + step < page_size => {
              if let PageBytes::Disk(buf) = bytes {
                self.disk_page_cache = Some((page_id, buf));
              }
              self.curr_addr = curr_addr + step as u64;
            }
            _ => self.skip_to_next_page(page_id),
          }
        }
        Err(e) => return Err(e),
      }
    }

    Ok(None)
  }

  /// 异步拉取下一条有效记录（返回 (逻辑地址, 拥有所有权的 [RecordOutput])）
  pub async fn next(&mut self) -> Result<Option<(u64, RecordOutput)>> {
    self
      .next_ref(|item| Ok((item.addr, RecordOutput::Memory(item.bytes.to_vec()))))
      .await
  }

  /// 异步拉取下一条有效记录并拷入调用方缓冲（复用缓冲容量，逐条扫描零堆分配）
  ///
  /// - 返回 `(逻辑地址, 记录字节切片)`，切片借用自 `buf`（在下一次调用前保持有效）；
  /// - 与 [Self::next] 语义一致，但由调用方提供可复用缓冲，
  ///   消除拉模式扫描每条记录 `to_vec` 的堆分配开销。
  pub async fn next_into<'b>(
    &'b mut self,
    buf: &'b mut Vec<u8>,
  ) -> Result<Option<(u64, &'b [u8])>> {
    let addr = self
      .next_ref(|item| {
        buf.clear();
        buf.extend_from_slice(item.bytes);
        Ok(item.addr)
      })
      .await?;
    Ok(addr.map(|a| (a, buf.as_slice())))
  }

  /// 跳至指定页的下一页开头
  #[inline]
  fn skip_to_next_page(&mut self, page_id: u64) {
    self.curr_addr = self.hlog.config.page_start_address(page_id + 1);
  }

  /// 消费成功后推进游标（记录恰抵页尾时直达下一页开头）
  #[inline]
  fn advance(&mut self, curr_addr: u64, total_size: usize, page_id: u64) {
    if self.hlog.config.page_offset(curr_addr) + total_size == self.hlog.config.page_size {
      self.skip_to_next_page(page_id);
    } else {
      self.curr_addr = curr_addr + total_size as u64;
    }
  }
}

impl<D: Device> HybridLog<D> {
  /// 构造连续逻辑地址扫描迭代器（对标 Garnet Pull-based Scan）
  pub fn scan_iter(&self, begin_addr: u64, end_addr: u64) -> ScanIterator<'_, D> {
    ScanIterator::new(self, begin_addr, end_addr)
  }

  /// 推模式连续逻辑地址扫描（对标 Garnet Push-based Scan，单条记录零堆分配借用）
  ///
  /// - 遍历区间 `[begin_addr, min(end_addr, tail))` 内的所有有效记录；
  /// - 跨页自动跳过 Pad 记录、极小子头残片与全零填充直达下一页开头；
  /// - 若用户闭包 `f` 返回 `Ok(false)`，立即提前终止扫描。
  pub async fn scan<F>(&self, begin_addr: u64, end_addr: u64, mut f: F) -> Result<()>
  where
    F: FnMut(u64, RecordRef<'_>) -> Result<bool>,
  {
    let mut it = self.scan_iter(begin_addr, end_addr);
    while let Some(cont) = it.next_ref(|item| f(item.addr, item.rec)).await? {
      if !cont {
        break;
      }
    }
    Ok(())
  }
}