use std::path::Path;

use copc_core::{LasPointRecord, NeverCancel, StreamingLayout};
use copc_writer::{write_streaming_with_cancel, CopcWriterParams, SpillWriter};
use las::Read as _;

#[test]
fn lod_index_and_spill_remain_disk_backed() {
    let writer_src = production_section(include_str!("../src/writer.rs"));
    let spill_src = production_section(include_str!("../src/spill.rs"));

    assert_contains(writer_src, "struct LodIndex");
    assert_contains(writer_src, "order_path: TempPath");
    assert_contains(writer_src, "struct IndexRun");
    assert_contains(writer_src, "path: TempPath");
    assert_contains(writer_src, "fn build_lod_index");
    assert_contains(writer_src, "write_root_index_run(total_points, cancel)?");
    assert_contains(writer_src, "fn partition_index_run");
    assert_contains(writer_src, "new_index_tempfile(\"root\")");
    assert_contains(writer_src, "new_index_tempfile(\"partition\")");
    assert_contains(writer_src, "open_index_run(run)?");
    assert_contains(writer_src, "append_index_to_order(");
    assert_not_contains(writer_src, "Vec<LasPointRecord>");
    assert_not_contains(writer_src, "Vec<CopcPointFields>");
    assert_not_contains(writer_src, "Vec<u32>");

    assert_contains(spill_src, "scratch: Vec<u8>");
    assert_contains(spill_src, "mmap: Mmap");
    assert_contains(spill_src, "Mmap::map(&file)");
    assert_not_contains(spill_src, "records: Vec<LasPointRecord>");
    assert_not_contains(spill_src, "Vec<LasPointRecord>");

    let dir = tempfile::tempdir().unwrap();
    let layout = streaming_layout(3);
    let mut writer = SpillWriter::create(dir.path(), layout).unwrap();
    assert_eq!(1, entry_count(dir.path()));
    for index in 0..8 {
        writer.push(&record(index, 3)).unwrap();
    }

    let reader = writer.finalize().unwrap();
    assert_eq!(8, reader.len());
    assert_eq!(1, entry_count(dir.path()));
    for index in 0..8 {
        let actual = reader.record_at(index).unwrap();
        assert_eq!(record(index, 3), actual);
    }

    drop(reader);
    assert_eq!(0, entry_count(dir.path()));
}

#[cfg(any(target_os = "linux", target_os = "macos"))]
#[test]
#[ignore = "large opt-in bounded-memory conversion guard"]
fn large_streaming_conversion_peak_rss_stays_bounded() {
    let point_count = env_usize("COPC_WRITER_MEMORY_GUARD_POINTS", 4_000_000);
    let extra_bytes = env_usize("COPC_WRITER_MEMORY_GUARD_EXTRA_BYTES", 32);
    let extra_byte_width =
        u16::try_from(extra_bytes).expect("COPC_WRITER_MEMORY_GUARD_EXTRA_BYTES exceeds u16");
    let max_peak_rss_growth =
        env_usize("COPC_WRITER_MEMORY_GUARD_MAX_RSS_BYTES", 512 * 1024 * 1024) as u64;

    let dir = tempfile::tempdir().unwrap();
    let out_path = dir.path().join("large-bounded-memory.copc.laz");
    let spill_dir = dir.path().join("spill");
    std::fs::create_dir(&spill_dir).unwrap();

    let before = peak_rss_bytes();
    write_streaming_with_cancel(
        &out_path,
        streaming_layout(extra_byte_width),
        GeneratedRecords {
            next: 0,
            len: point_count,
            extra_bytes,
        },
        &CopcWriterParams {
            max_points_per_node: 20_000,
            max_depth: 8,
        },
        &spill_dir,
        &NeverCancel,
    )
    .unwrap();
    let after = peak_rss_bytes();
    let peak_growth = after.saturating_sub(before);

    assert!(
        peak_growth <= max_peak_rss_growth,
        "peak RSS grew by {peak_growth} bytes for {point_count} streamed points with \
         {extra_bytes} extra bytes each; limit is {max_peak_rss_growth} bytes"
    );

    let reader = las::Reader::from_path(&out_path).unwrap();
    assert_eq!(point_count as u64, reader.header().number_of_points());
}

#[cfg(not(any(target_os = "linux", target_os = "macos")))]
#[test]
#[ignore = "large opt-in bounded-memory conversion guard"]
fn large_streaming_conversion_peak_rss_stays_bounded() {
    eprintln!("getrusage peak RSS accounting is only wired for Linux and macOS");
}

struct GeneratedRecords {
    next: usize,
    len: usize,
    extra_bytes: usize,
}

impl Iterator for GeneratedRecords {
    type Item = copc_core::Result<LasPointRecord>;

    fn next(&mut self) -> Option<Self::Item> {
        if self.next == self.len {
            return None;
        }
        let record = record(self.next, self.extra_bytes);
        self.next += 1;
        Some(Ok(record))
    }
}

fn streaming_layout(extra_bytes: u16) -> StreamingLayout {
    StreamingLayout {
        point_format: 6,
        has_gps: true,
        has_color: false,
        has_nir: false,
        has_waveform: false,
        extra_bytes,
        extra_bytes_descriptors: Vec::new(),
    }
}

fn record(index: usize, extra_byte_count: usize) -> LasPointRecord {
    let x = (index % 2048) as f64;
    let y = ((index / 2048) % 2048) as f64;
    let z = (index / (2048 * 2048)) as f64;
    let extra_bytes = (0..extra_byte_count)
        .map(|byte| ((index.wrapping_mul(31) + byte) & 0xff) as u8)
        .collect();

    LasPointRecord {
        x,
        y,
        z,
        intensity: index as u16,
        return_number: 1,
        number_of_returns: 1,
        classification: 1,
        scan_direction_flag: false,
        edge_of_flight_line: false,
        scan_angle: 0.0,
        user_data: 0,
        point_source_id: 0,
        synthetic: false,
        key_point: false,
        withheld: false,
        overlap: false,
        scan_channel: 0,
        gps_time: index as f64 * 0.001,
        red: 0,
        green: 0,
        blue: 0,
        nir: 0,
        wave_packet_descriptor_index: 0,
        byte_offset_to_waveform_data: 0,
        waveform_packet_size: 0,
        return_point_waveform_location: 0.0,
        extra_bytes,
    }
}

fn production_section(src: &'static str) -> &'static str {
    src.split("\n#[cfg(test)]\nmod tests").next().unwrap_or(src)
}

fn assert_contains(haystack: &str, needle: &str) {
    assert!(
        haystack.contains(needle),
        "expected production source to contain `{needle}`"
    );
}

fn assert_not_contains(haystack: &str, needle: &str) {
    assert!(
        !haystack.contains(needle),
        "expected production source not to contain `{needle}`"
    );
}

fn entry_count(path: &Path) -> usize {
    std::fs::read_dir(path).unwrap().count()
}

fn env_usize(name: &str, default: usize) -> usize {
    std::env::var(name)
        .ok()
        .and_then(|value| value.parse().ok())
        .unwrap_or(default)
}

#[cfg(any(target_os = "linux", target_os = "macos"))]
fn peak_rss_bytes() -> u64 {
    use std::mem::MaybeUninit;
    use std::os::raw::{c_int, c_long};

    #[repr(C)]
    struct TimeVal {
        tv_sec: c_long,
        tv_usec: c_long,
    }

    #[repr(C)]
    struct RUsage {
        ru_utime: TimeVal,
        ru_stime: TimeVal,
        ru_maxrss: c_long,
        ru_ixrss: c_long,
        ru_idrss: c_long,
        ru_isrss: c_long,
        ru_minflt: c_long,
        ru_majflt: c_long,
        ru_nswap: c_long,
        ru_inblock: c_long,
        ru_oublock: c_long,
        ru_msgsnd: c_long,
        ru_msgrcv: c_long,
        ru_nsignals: c_long,
        ru_nvcsw: c_long,
        ru_nivcsw: c_long,
    }

    extern "C" {
        fn getrusage(who: c_int, usage: *mut RUsage) -> c_int;
    }

    const RUSAGE_SELF: c_int = 0;

    let mut usage = MaybeUninit::<RUsage>::zeroed();
    let rc = unsafe { getrusage(RUSAGE_SELF, usage.as_mut_ptr()) };
    assert_eq!(0, rc, "getrusage failed");
    let usage = unsafe { usage.assume_init() };
    let maxrss = usage.ru_maxrss as u64;
    if cfg!(target_os = "linux") {
        maxrss * 1024
    } else {
        maxrss
    }
}
