waof 0.1.0

use std::{
  sync::Arc,
  thread::{spawn, yield_now},
};

use aok::{OK, Void};
use compio::runtime::Runtime;
use log::info;
use tempfile::tempdir;
use waof::{Error, RECORD_HEADER_LEN, WalConfig, WalLog};
use wdev::SegmentedDevice;

use super::support::{ShortWriteDevice, WalFixture, make_payload};

/// 对标 C# LogTests.cs::EnqueueAndWaitForCommitAsync
/// 异步写入、提交落盘与已提交位点无锁等待。
#[test]
fn test_enqueue_and_wait_for_commit() -> Void {
  let rt = Runtime::new()?;
  rt.block_on(async {
    let fixture = WalFixture::single_file("enqueue_commit.log", 64 * 1024)?;
    let wal = fixture.wal;

    let mut expected_tail = 0u64;
    for i in 0..100 {
      let payload = format!("commit_test_record_{i}").into_bytes();
      let record_len = (RECORD_HEADER_LEN + payload.len()) as u64;
      let addr = wal.enqueue(&payload)?;
      assert_eq!(addr, expected_tail);
      expected_tail += record_len;
    }

    assert_eq!(wal.tail_address(), expected_tail);
    let committed = wal.commit().await?;
    assert_eq!(committed, expected_tail);
    assert_eq!(wal.committed_until_address(), expected_tail);

    // 对已提交的位点再次调用 wait_for_commit 应当极速直接返回
    let fast_committed = wal.wait_for_commit(expected_tail).await?;
    assert_eq!(fast_committed, expected_tail);

    let mut iter = wal.scan_committed();
    let records = iter.collect_all().await?;
    assert_eq!(records.len(), 100);

    info!("EnqueueAndWaitForCommit 异步写入与提交测试通过");
    aok::Result::<()>::Ok(())
  })?;

  OK
}

/// 对标 C# LogTests.cs::TestTryEnqueue
/// 测试写缓冲区填满时的 BufferFull 阻断，以及通过 commit 刷盘释放空间后继续写入的循环覆写能力。
#[test]
fn test_try_enqueue_buffer_full_and_commit_cycle() -> Void {
  let rt = Runtime::new()?;
  rt.block_on(async {
    // 较小缓冲区：16KB 容量（扇区大小取设备默认 4096）
    let fixture = WalFixture::single_file("try_enqueue.log", 16 * 1024)?;
    let wal = fixture.wal;

    let payload = make_payload(1000, 0xAB);
    let mut first_batch_count = 0;

    // 持续写入直到缓冲区满
    let mut hit_buffer_full = false;
    for _ in 0..100 {
      match wal.enqueue(&payload) {
        Ok(_) => first_batch_count += 1,
        Err(Error::BufferFull { .. }) => {
          hit_buffer_full = true;
          break;
        }
        Err(e) => return Err(e.into()),
      }
    }

    assert!(hit_buffer_full, "缓冲区应当在容量耗尽时返回 BufferFull");

    // 执行 commit 刷盘释放环形缓冲区空间
    wal.commit().await?;

    // 再次写入，应当恢复写入能力
    for _ in 0..10 {
      wal.enqueue(&payload)?;
    }
    wal.commit().await?;

    // 扫描全部记录，校验所有写入数据的完整性与 CRC32
    let mut iter = wal.scan(0, wal.tail_address());
    let all_records = iter.collect_all().await?;
    assert_eq!(all_records.len(), first_batch_count + 10);
    for rec in all_records {
      assert_eq!(rec.payload, payload);
    }

    info!("TryEnqueue 缓冲区满与 commit 淘汰循环覆写测试通过");
    aok::Result::<()>::Ok(())
  })?;

  OK
}

/// 对标 C# LogTests.cs 边界与防护机制
/// 测试单条记录超出缓冲容量报错、零在途槽位防御、空记录写入与空日志提交。
#[test]
fn test_enqueue_payload_limits_and_config_guards() -> Void {
  let rt = Runtime::new()?;
  rt.block_on(async {
    let buf_size = 8 * 1024;
    let fixture = WalFixture::single_file("limits_guard.log", buf_size)?;
    let wal = fixture.wal;

    // 1. 超大记录拦截
    let max_allowed_payload_len = buf_size - RECORD_HEADER_LEN;
    let too_large = make_payload(max_allowed_payload_len + 1, 1);
    assert!(matches!(
      wal.enqueue(&too_large),
      Err(Error::PayloadTooLarge(_))
    ));

    // 2. 空日志提交返回 0
    let c0 = wal.commit().await?;
    assert_eq!(c0, 0);

    // 3. 空 payload 写入
    let addr = wal.enqueue(&[])?;
    assert_eq!(addr, 0);
    wal.commit().await?;

    let mut iter = wal.scan_committed();
    let rec = iter.next().await?.expect("应有空 payload 记录");
    assert!(rec.payload.is_empty());
    assert_eq!(rec.header.entry_len, 0);

    // 4. 零在途槽位防御
    let dir = tempdir()?;
    let db_path = dir.path().join("zero_inflight.log");
    let device = Arc::new(SegmentedDevice::single_file(&db_path)?);
    let config = WalConfig {
      inflight_slots: 0,
      ..Default::default()
    };
    let zero_wal = WalLog::new(device, config)?;
    assert_eq!(zero_wal.inflight_slots.len(), 1);
    let zero_addr = zero_wal.enqueue(b"safe fallback with min 1 slot")?;
    assert_eq!(zero_addr, 0);

    info!("Enqueue 边界限制与配置守卫测试通过");
    aok::Result::<()>::Ok(())
  })?;

  OK
}

/// 对标 C# LogFastCommitTests.cs::CommitRecordBoundedGrowthTest
/// 高并发多线程写入伴随并发提交，验证安全尾部边界一致性与零数据竞争损坏。
#[test]
fn test_commit_record_bounded_growth_concurrent_enqueue() -> Void {
  let rt = Runtime::new()?;
  rt.block_on(async {
    let fixture = WalFixture::segmented("concurrent_growth.log", 64 * 1024, 128 * 1024)?;
    let wal = fixture.wal;

    let num_threads = 4;
    let records_per_thread = 200;

    let mut handles = Vec::new();
    for t_id in 0..num_threads {
      let wal_clone = Arc::clone(&wal);
      let handle = spawn(move || {
        for i in 0..records_per_thread {
          let mut data = [0u8; 64];
          data[0] = t_id as u8;
          data[1..5].copy_from_slice(&(i as u32).to_le_bytes());
          wal_clone.enqueue(&data).expect("并发 enqueue 应当成功");
        }
      });
      handles.push(handle);
    }

    // 主线程在写入期间并发执行 commit 刷盘
    for _ in 0..5 {
      let _ = wal.commit().await;
      yield_now();
    }

    for h in handles {
      h.join().expect("线程应当正常结束");
    }

    // 最终提交全部在途与已写入数据
    let final_tail = wal.commit().await?;
    assert_eq!(final_tail, wal.tail_address());

    // 扫描全部记录
    let mut iter = wal.scan(0, final_tail);
    let records = iter.collect_all().await?;
    assert_eq!(records.len(), num_threads * records_per_thread);

    // 统计各线程记录数量
    let mut thread_counts = [0usize; 4];
    for rec in records {
      let t_id = rec.payload[0] as usize;
      assert!(t_id < num_threads);
      thread_counts[t_id] += 1;
    }

    for (t_id, &count) in thread_counts.iter().enumerate() {
      assert_eq!(count, records_per_thread, "线程 {t_id} 记录数应完整");
    }

    info!("CommitRecordBoundedGrowth 高并发写入与并发 commit 测试通过");
    aok::Result::<()>::Ok(())
  })?;

  OK
}

/// 对标 C# LogFastCommitTests.cs::FastCommitConcurrentWaiters
/// 多线程并发发起 enqueue 并在后台并发调用 wait_for_commit 等待位点持久化，
/// 验证协同组提交机制下的数据完整性与零死锁。
#[test]
fn test_fast_commit_concurrent_waiters_barrier() -> Void {
  let rt = Runtime::new()?;
  rt.block_on(async {
    let fixture = WalFixture::single_file("fast_commit_barrier.log", 2 * 1024 * 1024)?;
    let wal = fixture.wal;

    const THREAD_COUNT: usize = 8;
    const OPS_PER_THREAD: usize = 25;

    let mut handles = Vec::with_capacity(THREAD_COUNT);
    for t_id in 0..THREAD_COUNT {
      let wal_clone = Arc::clone(&wal);
      let handle = spawn(move || {
        let rt = Runtime::new().unwrap();
        rt.block_on(async move {
          for i in 0..OPS_PER_THREAD {
            let payload = format!("barrier-thread-{t_id}-item-{i}").into_bytes();
            let addr = wal_clone.enqueue(&payload).unwrap();
            let target_addr = addr + (RECORD_HEADER_LEN + payload.len()) as u64;
            let committed = wal_clone.wait_for_commit(target_addr).await.unwrap();
            assert!(committed >= target_addr);
          }
        });
      });
      handles.push(handle);
    }

    for h in handles {
      h.join().unwrap();
    }

    // 最终全部提交
    let final_tail = wal.commit().await?;
    assert_eq!(final_tail, wal.tail_address());

    // 扫描校验
    let mut iter = wal.scan_committed();
    let records = iter.collect_all().await?;
    assert_eq!(records.len(), THREAD_COUNT * OPS_PER_THREAD);
    for rec in records {
      rec.header.verify(&rec.payload)?;
    }

    info!("FastCommitConcurrentWaiters 多线程快速提交屏障测试通过");
    aok::Result::<()>::Ok(())
  })?;

  OK
}

/// 对标 C# LogCommitFailureTests.cs
/// 验证 commit 短写入守卫：设备返回 Ok(部分字节) 时 commit 必须显式报错，
/// 且绝不推进 flushed/committed 位点（防止已提交区间在磁盘上存在数据空洞）。
#[test]
fn test_commit_short_write_guard() -> Void {
  let rt = Runtime::new()?;
  rt.block_on(async {
    let dir = tempdir()?;
    let db_path = dir.path().join("short_write.log");
    let device = Arc::new(ShortWriteDevice::single_file(&db_path)?);

    let config = WalConfig::new(64 * 1024);
    let wal = WalLog::new(device, config)?;

    wal.enqueue(b"record-one")?;
    wal.enqueue(b"record-two")?;

    // 短写入必须显式报错且位点保持不变
    assert!(matches!(wal.commit().await, Err(Error::ShortWrite { .. })));
    assert_eq!(wal.flushed_until_address(), 0);
    assert_eq!(wal.committed_until_address(), 0);

    // 失败的 commit 可安全重试，继续报错且位点不脏推进
    assert!(matches!(wal.commit().await, Err(Error::ShortWrite { .. })));
    assert_eq!(wal.flushed_until_address(), 0);
    assert_eq!(wal.committed_until_address(), 0);

    info!("Commit 短写入守卫与位点不回滚测试通过");
    aok::Result::<()>::Ok(())
  })?;

  OK
}