//! Continuous index mode deep scenario tests.
//!
//! Covers large-gap writes, logical hole backfill, correction on filler positions,
//! segment_capacity calculation, base_timestamp persistence, negative timestamps,
//! and multi-segment gap scenarios.

use std::fs;
use std::path::PathBuf;
use std::sync::atomic::{AtomicU64, Ordering};

use timslite::{DataSetConfigBuilder, Store, StoreConfig};

static COUNTER: AtomicU64 = AtomicU64::new(0);

fn temp_dir() -> PathBuf {
    let d = std::env::temp_dir().join("timslite_integration");
    fs::create_dir_all(&d).unwrap();
    let id = COUNTER.fetch_add(1, Ordering::Relaxed);
    d.join(format!(
        "test_{:?}_{id}",
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_nanos()
    ))
}

fn store_config() -> StoreConfig {
    StoreConfig::builder()
        .enable_background_thread(false)
        .data_segment_size(1024 * 1024)
        .index_segment_size(64 * 1024)
        .initial_data_segment_size(4096)
        .initial_index_segment_size(4096)
        .build()
}

fn continuous_ds_config(store_cfg: &StoreConfig) -> DataSetConfigBuilder {
    DataSetConfigBuilder::from_store(store_cfg)
        .index_continuous(1)
        .data_segment_size(4096)
        .index_segment_size(4096)
}

/// Helper: create store + continuous dataset, return (store, dataset).
fn setup_continuous(name: &str) -> (Store, timslite::DataSet) {
    let dir = temp_dir();
    let cfg = store_config();
    let mut store = Store::open(&dir, cfg.clone()).unwrap();
    let ds_cfg = continuous_ds_config(&cfg);
    let handle = store
        .create_dataset_with_config(name, "data", Some(ds_cfg))
        .unwrap();
    let ds = handle.clone();
    (store, ds)
}

#[test]
fn t32_1_large_gap_no_intermediate_segments() {
    let (store, ds) = setup_continuous("large_gap");

    // First write sets base_timestamp = 100
    ds.write(100, b"first").unwrap();
    ds.flush().unwrap();

    // Large gap write: ts=1000000
    ds.write(1000000, b"second").unwrap();
    ds.flush().unwrap();

    let info = store.inspect_dataset("large_gap", "data").unwrap();

    // With index_segment_size=4096, segment_capacity = floor((4096-128)/14) = 283.
    // ts=100 is in ordinal 0, ts=1000000 is in ordinal floor((1000000-100)/283) = 3533.
    // Only the tail of ordinal 0 and the prefix of ordinal 3533 are materialized;
    // all intermediate ordinals (1..3532) remain logical holes.
    // So we expect exactly 2 index segments on disk.
    assert_eq!(
        info.state.index_segments, 2,
        "expected 2 index segments for large gap"
    );

    // Both records readable
    let r1 = ds.read(100).unwrap();
    assert!(r1.is_some());
    assert_eq!(r1.unwrap().1, b"first");

    let r2 = ds.read(1000000).unwrap();
    assert!(r2.is_some());
    assert_eq!(r2.unwrap().1, b"second");

    // Intermediate timestamps should be empty (logical hole)
    let r_mid = ds.read(500000).unwrap();
    assert!(
        r_mid.is_none(),
        "intermediate ts in logical hole should be None"
    );

    store.close().unwrap();
}

#[test]
fn t32_2_backfill_logical_hole() {
    let (store, ds) = setup_continuous("backfill_hole");

    ds.write(100, b"first").unwrap();
    ds.write(1000000, b"second").unwrap();
    ds.flush().unwrap();

    // Now backfill into the logical hole
    ds.write(500000, b"backfill").unwrap();
    ds.flush().unwrap();

    let info = store.inspect_dataset("backfill_hole", "data").unwrap();

    // We now have 3 segments: ordinal 0 (ts=100), ordinal 2272 (ts=500000),
    // ordinal 4545 (ts=1000000). The new segment should have prefix fillers
    // from segment_start to ts=499999.
    assert_eq!(
        info.state.index_segments, 3,
        "expected 3 index segments after backfill"
    );

    // All three records readable
    assert_eq!(ds.read(100).unwrap().unwrap().1, b"first");
    assert_eq!(ds.read(500000).unwrap().unwrap().1, b"backfill");
    assert_eq!(ds.read(1000000).unwrap().unwrap().1, b"second");

    // Query across all three should return exactly 3 entries (fillers skipped)
    let entries = ds.query(100, 1000000).unwrap();
    assert_eq!(
        entries.len(),
        3,
        "query should return exactly 3 real entries"
    );

    store.close().unwrap();
}

#[test]
fn t32_3_correction_on_filler_position() {
    let (store, ds) = setup_continuous("corr_filler");

    // Write ts=100, then ts=120 鈥?the gap 101..119 are fillers within the same segment
    ds.write(100, b"orig_100").unwrap();
    ds.write(120, b"orig_120").unwrap();
    ds.flush().unwrap();

    // ts=110 is a filler position; correction write should succeed
    // (write with ts < latest triggers continuous-mode backfill logic)
    ds.write(110, b"filled_110").unwrap();
    ds.flush().unwrap();

    // Verify the correction took effect
    let r = ds.read(110).unwrap();
    assert!(r.is_some(), "filler position should now have data");
    assert_eq!(r.unwrap().1, b"filled_110");

    // Original entries still intact
    assert_eq!(ds.read(100).unwrap().unwrap().1, b"orig_100");
    assert_eq!(ds.read(120).unwrap().unwrap().1, b"orig_120");

    // Query should now return 3 entries
    let entries = ds.query(100, 120).unwrap();
    assert_eq!(entries.len(), 3);

    store.close().unwrap();
}

#[test]
fn t32_4_segment_capacity_calculation() {
    // segment_capacity = floor((index_segment_size - 128) / 14)
    // For index_segment_size = 4096: floor((4096 - 128) / 14) = floor(3968 / 14) = 283
    let segment_capacity: i64 = (4096 - 128) / 14;
    assert_eq!(segment_capacity, 283);

    // Verify by filling exactly segment_capacity entries in one segment.
    // base_timestamp = 0, so segment covers ts 0..282 (283 entries).
    // Writing ts=0..282 should fit in 1 segment; writing ts=283 should create a 2nd.
    let (store, ds) = setup_continuous("capacity_test");

    for ts in 0..segment_capacity {
        ds.write(ts, b"x").unwrap();
    }
    ds.flush().unwrap();

    let info = store.inspect_dataset("capacity_test", "data").unwrap();
    assert_eq!(
        info.state.index_segments, 1,
        "segment_capacity entries should fit in 1 segment"
    );

    // One more write beyond capacity 鈫?new segment
    ds.write(segment_capacity, b"overflow").unwrap();
    ds.flush().unwrap();

    let info = store.inspect_dataset("capacity_test", "data").unwrap();
    assert_eq!(
        info.state.index_segments, 2,
        "writing beyond capacity should create a 2nd segment"
    );

    store.close().unwrap();
}

#[test]
fn t32_5_reopen_base_timestamp_preserved() {
    let dir = temp_dir();
    let cfg = store_config();

    // Phase 1: write and close
    {
        let mut store = Store::open(&dir, cfg.clone()).unwrap();
        let ds_cfg = continuous_ds_config(&cfg);
        let handle = store
            .create_dataset_with_config("reopen_bt", "data", Some(ds_cfg))
            .unwrap();
        let ds = handle.clone();

        ds.write(500, b"hello").unwrap();
        ds.write(700, b"world").unwrap();
        ds.flush().unwrap();

        let info = store.inspect_dataset("reopen_bt", "data").unwrap();
        assert_eq!(info.state.base_timestamp, Some(500));

        store.close().unwrap();
    }

    // Phase 2: reopen and verify base_timestamp
    {
        let mut store = Store::open(&dir, cfg.clone()).unwrap();
        let handle = store.open_dataset("reopen_bt", "data").unwrap();
        let ds = handle.clone();

        let info = store.inspect_dataset("reopen_bt", "data").unwrap();
        assert_eq!(
            info.state.base_timestamp,
            Some(500),
            "base_timestamp must be preserved after reopen"
        );

        // Data must still be readable
        assert_eq!(ds.read(500).unwrap().unwrap().1, b"hello");
        assert_eq!(ds.read(700).unwrap().unwrap().1, b"world");

        // Write after reopen with a gap
        ds.write(1000, b"after_reopen").unwrap();
        ds.flush().unwrap();
        assert_eq!(ds.read(1000).unwrap().unwrap().1, b"after_reopen");

        store.close().unwrap();
    }
}

#[test]
fn t32_6_zero_base_timestamp_and_negative_read_offsets() {
    let (store, ds) = setup_continuous("zero_ts");

    assert!(ds.write(-1, b"negative").is_err());
    ds.write(0, b"zero").unwrap();
    ds.write(1, b"one").unwrap();
    ds.write(2, b"two").unwrap();
    ds.flush().unwrap();

    let info = store.inspect_dataset("zero_ts", "data").unwrap();
    assert_eq!(info.state.base_timestamp, Some(0));
    assert_eq!(info.state.latest_written_timestamp, Some(2));
    assert_eq!(
        info.state.index_segments, 1,
        "all 3 entries should fit in 1 segment"
    );

    // Verify all records
    assert_eq!(ds.read(0).unwrap().unwrap().1, b"zero");
    assert_eq!(ds.read(1).unwrap().unwrap().1, b"one");
    assert_eq!(ds.read(2).unwrap().unwrap().1, b"two");
    assert_eq!(ds.read(-1).unwrap().unwrap().1, b"two");
    assert_eq!(ds.read(-2).unwrap().unwrap().1, b"one");
    assert_eq!(ds.read(-3).unwrap().unwrap().1, b"zero");

    // Filler positions should be None
    assert!(ds.read(-4).unwrap().is_none());

    // Query across all
    let entries = ds.query(-3, -1).unwrap();
    assert_eq!(entries.len(), 3);

    store.close().unwrap();
}

#[test]
fn t32_7_multiple_segments_with_gaps() {
    // segment_capacity = 283 for index_segment_size=4096.
    // Write at ts=0, ts=500, ts=1500 to span 3 segments:
    //   segment 0: ts 0..282     (ts=0 is here)
    //   segment 1: ts 283..565   (ts=500 is here)
    //   segment 5: ts 1415..1697 (ts=1500 is here)
    //   seg_ord(ts) = floor((ts - 0) / 283)
    //   seg_ord(0) = 0
    //   seg_ord(500) = floor(500/283) = 1  鈫?segment start = 283
    //   seg_ord(1500) = floor(1500/283) = 5 鈫?segment start = 1415
    // So writes at ts=0, 500, 1500 span segments 0, 1, 5 鈫?3 segments on disk.
    let (store, ds) = setup_continuous("multi_seg_gap");

    ds.write(0, b"seg0").unwrap();
    ds.flush().unwrap();

    // ts=500 is in segment 1 (ordinal 1).
    ds.write(500, b"seg2").unwrap();
    ds.flush().unwrap();

    // ts=1500 is in segment 5 (ordinal 5). Gap: segments 2-4 are logical holes.
    ds.write(1500, b"seg6").unwrap();
    ds.flush().unwrap();

    let info = store.inspect_dataset("multi_seg_gap", "data").unwrap();
    assert_eq!(
        info.state.index_segments, 3,
        "expected 3 index segments: ordinal 0, 1, and 5"
    );
    assert_eq!(info.state.latest_written_timestamp, Some(1500));
    assert_eq!(info.state.base_timestamp, Some(0));

    // All real entries readable
    assert_eq!(ds.read(0).unwrap().unwrap().1, b"seg0");
    assert_eq!(ds.read(500).unwrap().unwrap().1, b"seg2");
    assert_eq!(ds.read(1500).unwrap().unwrap().1, b"seg6");

    // Logical hole positions should be None
    assert!(
        ds.read(300).unwrap().is_none(),
        "ts=300 is filler in segment 1"
    );
    assert!(
        ds.read(800).unwrap().is_none(),
        "ts=800 in logical hole (segment 2)"
    );
    assert!(
        ds.read(1200).unwrap().is_none(),
        "ts=1200 in logical hole (segment 4)"
    );

    // Query should return exactly 3 entries
    let entries = ds.query(0, 1500).unwrap();
    assert_eq!(entries.len(), 3);

    store.close().unwrap();
}
