use crate::define::{extract_start_time_from_idx_filename, parse_filename_time_to_unix};
use crate::idx_line::IdxLine;
use anyhow::{Context, Result, anyhow}; // added Context
use bytes::Bytes;
use std::fs::{self, File, metadata};
use std::io::{BufReader, Read, Seek, SeekFrom}; // modified imports
use std::path::{Path, PathBuf};
use std::sync::mpsc::{Receiver, sync_channel}; // added
use std::thread;

/// 一个定位到 blob 的切片描述
#[derive(Debug, Clone)]
pub struct LocatedSlice {
    pub blob: PathBuf,
    /// timestamp 以秒为单位 (UNIX epoch)
    pub ts: i64,
    pub offset: u64,
    /// 在调用 compute_lengths 之后可能被填充
    pub length: Option<u64>,
}

// 新增：表示实际切片数据（不包含 blob 路径）
#[derive(Debug)]
pub struct Slice {
    pub ts: i64,
    pub offset: u64,
    pub data: Bytes, // changed from Vec<u8> to Bytes
}

// Make IdxClient clonable so we can move it into a thread
#[derive(Clone)]
pub struct IdxClient {
    base_dir: PathBuf,
    id: String,
    file_list: Option<Vec<PathBuf>>, // NEW: optional pre-supplied file list (contains idx/blob paths)
}

impl IdxClient {
    pub fn new(base_dir: impl Into<PathBuf>, id: impl Into<String>) -> Self {
        Self {
            base_dir: base_dir.into(),
            id: id.into(),
            file_list: None, // default: read directory
        }
    }

    /// NEW: construct from an explicit list of files (idx + blob files). When provided,
    /// list_idx_files will use this list instead of scanning base_dir.
    pub fn new_from_list(
        base_dir: impl Into<PathBuf>,
        id: impl Into<String>,
        file_list: Vec<PathBuf>,
    ) -> Self {
        Self {
            base_dir: base_dir.into(),
            id: id.into(),
            file_list: Some(file_list),
        }
    }

    /// 列出 base_dir 中与 id 匹配的所有 .mp4.idx 文件
    fn list_idx_files(&self) -> Result<Vec<PathBuf>> {
        let mut out = Vec::new();

        if let Some(list) = &self.file_list {
            // Use the provided file list instead of reading the directory
            for p in list.iter() {
                if let Some(name) = p.file_name().and_then(|s| s.to_str()) {
                    if name.starts_with(&format!("{}_", self.id)) && name.ends_with(".mp4.idx") {
                        out.push(p.clone());
                    }
                }
            }
            return Ok(out);
        }

        let rd = fs::read_dir(&self.base_dir)
            .with_context(|| format!("failed to read directory {}", self.base_dir.display()))?;
        for entry in rd {
            let e = entry
                .with_context(|| format!("failed to read entry in {}", self.base_dir.display()))?;
            let p = e.path();
            if let Some(name) = p.file_name().and_then(|s| s.to_str()) {
                // 期望格式: {id}_<ts>.mp4.idx
                if name.starts_with(&format!("{}_", self.id)) && name.ends_with(".mp4.idx") {
                    out.push(p);
                }
            }
        }
        Ok(out)
    }

    /// parse a single idx file, returning Vec<(timestamp_seconds, offset)>
    fn parse_idx_file(&self, idx_path: &Path) -> Result<Vec<(i64, u64)>> {
        let f = File::open(idx_path)
            .with_context(|| format!("failed to open idx file {:?}", idx_path))?;
        let mut reader = BufReader::new(f);
        let mut out = Vec::new();
        let mut buf = [0u8; 16];

        loop {
            match reader.read_exact(&mut buf) {
                Ok(()) => {
                    // parse 16-byte record
                    let line = IdxLine::parse_bytes(&buf)
                        .with_context(|| format!("failed to parse idx record in {:?}", idx_path))?;
                    let ts = line.ts_millis();
                    let off = line.offset();
                    // normalize milliseconds -> seconds if needed
                    let ts_secs = if ts > 3_000_000_000u64 {
                        (ts / 1000) as i64
                    } else {
                        ts as i64
                    };
                    out.push((ts_secs, off));
                }
                Err(e) => {
                    if e.kind() == std::io::ErrorKind::UnexpectedEof {
                        break;
                    } else {
                        return Err(anyhow!("error reading idx file {:?}: {}", idx_path, e));
                    }
                }
            }
        }
        Ok(out)
    }

    /// 查询某 id 在 [begin,end] 时间区间内（包含边界）的所有切片
    /// 返回一个 sync_channel(1) 的 Receiver<Slice>，实际读取会在后台线程中逐条读取并 send，
    /// 接收端每 recv 一次会触发生产者继续读取下一条（即惰性读取）。
    pub fn query(&self, begin_str: &str, end_str: &str) -> Result<Receiver<Slice>> {
        // directly propagate parsing errors with `?`
        let begin = parse_filename_time_to_unix(begin_str)?;
        let end = parse_filename_time_to_unix(end_str)?;
        let mut located: Vec<LocatedSlice> = Vec::new();

        // collect located slices (no data read yet)
        for idx_path in self.list_idx_files()? {
            // 粗过滤：如果 idx 文件名表示的起始时间 > 查询 end，则跳过
            if let Ok(start_ts) = extract_start_time_from_idx_filename(&idx_path, &self.id) {
                if start_ts > end {
                    continue;
                }
            }

            // 对应的 blob 文件名是把 ".idx" 去掉（id_YYYYMMDDhhmmss.mp4.idx -> id_YYYYMMDDhhmmss.mp4）
            let blob_path = idx_path.with_extension("");

            // 解析该 idx 的全部 entries（按出现顺序），并计算每条 entry 的 length（基于下一条 offset 或文件大小）
            let entries = self
                .parse_idx_file(&idx_path)
                .map_err(|e| anyhow!("failed to parse idx file {:?}: {}", idx_path, e))?;
            if entries.is_empty() {
                continue;
            }
            // 获取 blob 文件大小（用于最后一条）
            let file_len = metadata(&blob_path)
                .map(|m| m.len())
                .map_err(|e| anyhow!("failed to stat blob file {:?}: {}", blob_path, e))
                .unwrap_or(0);

            // Merge consecutive entries having identical timestamps.
            let mut i = 0usize;
            while i < entries.len() {
                let ts = entries[i].0;
                let start_offset = entries[i].1;
                // advance j to the first entry with a different timestamp
                let mut j = i + 1;
                while j < entries.len() && entries[j].0 == ts {
                    j += 1;
                }
                // group's end offset is next entry's offset if present, otherwise file_len
                let end_offset = if j < entries.len() {
                    entries[j].1
                } else {
                    file_len
                };
                let length = end_offset.saturating_sub(start_offset);
                if ts >= begin && ts <= end {
                    located.push(LocatedSlice {
                        blob: blob_path.clone(),
                        ts,
                        offset: start_offset,
                        length: Some(length),
                    });
                }
                i = j;
            }
        }

        // At this point we have located slices with lengths computed. Sort by time.
        located.sort_by_key(|s| s.ts);

        // create sync channel(1) and spawn background thread to read lazily
        let (tx, rx) = sync_channel(1);
        let client_clone = self.clone();
        thread::spawn(move || {
            // iterate located and read/send one by one
            for ls in located {
                // fetch_data uses the client to open blob and read bytes
                match client_clone.fetch_data(&ls) {
                    Ok(data) => {
                        let slice = Slice {
                            ts: ls.ts,
                            offset: ls.offset,
                            data,
                        };
                        // send will block until receiver consumes (sync_channel(1) enforces laziness)
                        if tx.send(slice).is_err() {
                            // receiver dropped, stop producing
                            break;
                        }
                    }
                    Err(_) => {
                        // On read error, stop producing (could log)
                        break;
                    }
                }
            }
            // when thread exits, tx is dropped and receiver.iter() will end
        });

        Ok(rx)
    }

    /// 从 blob 中读取一个切片的 bytes（需要 slice.length 已经被填充）
    fn fetch_data(&self, slice: &LocatedSlice) -> Result<Bytes> {
        let len = slice
            .length
            .ok_or_else(|| anyhow!("slice length not computed for {:?}", slice))?;
        let mut f = File::open(&slice.blob)
            .with_context(|| format!("failed to open blob file {:?}", slice.blob))?;
        f.seek(SeekFrom::Start(slice.offset)).with_context(|| {
            format!(
                "failed to seek to offset {} in {:?}",
                slice.offset, slice.blob
            )
        })?;
        let mut buf = vec![0u8; len as usize];
        f.read_exact(&mut buf)
            .with_context(|| format!("failed to read {} bytes from {:?}", len, slice.blob))?;
        Ok(Bytes::from_owner(buf))
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::define::unix_secs_to_yyyymmddhhmmss;
    use crate::idx_writer::IdxRotateFileWriter;
    use std::fs;
    use std::thread::sleep;
    use std::time::{Duration, UNIX_EPOCH};

    #[test]
    fn write_then_query_roundtrip() {
        // temp dir
        let mut base = std::env::temp_dir();
        let uniq = format!(
            "idx_test_{}",
            std::time::SystemTime::now()
                .duration_since(UNIX_EPOCH)
                .unwrap()
                .as_nanos()
        );
        base.push(uniq);
        fs::create_dir_all(&base).unwrap();
        println!("Tempdir: {:?}", base);

        // id and writer
        let id = "testid";
        let mut writer = IdxRotateFileWriter::new(&base, id, 2); // rotate after 2 slices

        // timestamps (seconds since epoch)
        let ts1 = 1_700_000_000i64; // arbitrary fixed times
        let ts2 = ts1 + 3600;
        let ts3 = ts1 + 7200;

        let st1 = UNIX_EPOCH + Duration::from_secs(ts1 as u64);
        let st2 = UNIX_EPOCH + Duration::from_secs(ts2 as u64);
        let st3 = UNIX_EPOCH + Duration::from_secs(ts3 as u64);

        writer.write_slice(b"slice1", st1).unwrap();
        // small pause to ensure file operations flushed if needed
        sleep(Duration::from_millis(5));
        writer.write_slice(b"slice2", st2).unwrap();
        sleep(Duration::from_millis(5));
        writer.write_slice(b"slice3", st3).unwrap();
        writer.close().unwrap();

        // query covering full range
        let begin_str = unix_secs_to_yyyymmddhhmmss(ts1 - 1).unwrap();
        let end_str = unix_secs_to_yyyymmddhhmmss(ts3 + 1).unwrap();

        let client = super::IdxClient::new(&base, id);
        let rx = client.query(&begin_str, &end_str).unwrap();
        let slices: Vec<Slice> = rx.iter().collect();

        // should read 3 slices in chronological order and contents match
        assert_eq!(slices.len(), 3);
        assert_eq!(slices[0].data.as_ref(), b"slice1"); // changed
        assert_eq!(slices[1].data.as_ref(), b"slice2"); // changed
        assert_eq!(slices[2].data.as_ref(), b"slice3"); // changed

        // cleanup
        let _ = fs::remove_dir_all(&base);
    }

    // 新增：测试查询时间位于同一 blob 文件中的场景
    #[test]
    fn query_within_single_blob() {
        // temp dir
        let mut base = std::env::temp_dir();
        let uniq = format!(
            "idx_test_single_{}",
            std::time::SystemTime::now()
                .duration_since(UNIX_EPOCH)
                .unwrap()
                .as_nanos()
        );
        base.push(uniq);
        println!("Tempdir: {:?}", base);

        fs::create_dir_all(&base).unwrap();

        let id = "singleblob";
        // 设置较大阈值，确保所有 slice 写入同一个 blob
        let mut writer = IdxRotateFileWriter::new(&base, id, 10);

        // 3 个时间点，秒级
        let ts1 = 1_700_000_000i64;
        let ts2 = ts1 + 10;
        let ts3 = ts1 + 20;

        let st1 = UNIX_EPOCH + Duration::from_secs(ts1 as u64);
        let st2 = UNIX_EPOCH + Duration::from_secs(ts2 as u64);
        let st3 = UNIX_EPOCH + Duration::from_secs(ts3 as u64);

        writer.write_slice(b"a_slice", st1).unwrap();
        sleep(Duration::from_millis(5));
        writer.write_slice(b"b_slice", st2).unwrap();
        sleep(Duration::from_millis(5));
        writer.write_slice(b"c_slice", st3).unwrap();
        writer.close().unwrap();

        let client = super::IdxClient::new(&base, id);

        // 查询仅包含第二个 slice 的时间点（begin == end == ts2）
        let begin = unix_secs_to_yyyymmddhhmmss(ts2).unwrap();
        let end = unix_secs_to_yyyymmddhhmmss(ts2).unwrap();
        println!("begin_time: begin:{}, end:{}", begin, end);

        let rx = client.query(&begin, &end).unwrap();
        let res: Vec<Slice> = rx.iter().collect();
        assert_eq!(res.len(), 1);
        assert_eq!(res[0].data.as_ref(), b"b_slice"); // changed

        // 也验证范围包含前两个切片时会返回两个
        let begin2 = unix_secs_to_yyyymmddhhmmss(ts1).unwrap();
        let end2 = unix_secs_to_yyyymmddhhmmss(ts2).unwrap();
        println!("begin_time: begin:{}, end:{}", begin2, end2);

        let rx2 = client.query(&begin2, &end2).unwrap();
        let res2: Vec<Slice> = rx2.iter().collect();
        assert_eq!(res2.len(), 2);
        assert_eq!(res2[0].data.as_ref(), b"a_slice"); // changed
        assert_eq!(res2[1].data.as_ref(), b"b_slice"); // changed

        // cleanup
        let _ = fs::remove_dir_all(&base);
    }

    #[test]
    fn multi_id_concurrent_writers_separate_reads() {
        // temp dir
        let mut base = std::env::temp_dir();
        let uniq = format!(
            "idx_test_multi_{}",
            std::time::SystemTime::now()
                .duration_since(UNIX_EPOCH)
                .unwrap()
                .as_nanos()
        );
        base.push(uniq);
        fs::create_dir_all(&base).unwrap();

        let id1 = "alpha";
        let id2 = "beta";
        let mut w1 = IdxRotateFileWriter::new(&base, id1, 10);
        let mut w2 = IdxRotateFileWriter::new(&base, id2, 10);

        let t0 = 1_700_000_000i64;
        let st = |s: i64| UNIX_EPOCH + Duration::from_secs(s as u64);

        // interleaved writes
        w1.write_slice(b"a1", st(t0)).unwrap();
        sleep(Duration::from_millis(5));
        w2.write_slice(b"b1", st(t0 + 1)).unwrap();
        sleep(Duration::from_millis(5));
        w1.write_slice(b"a2", st(t0 + 2)).unwrap();
        sleep(Duration::from_millis(5));
        w2.write_slice(b"b2", st(t0 + 3)).unwrap();

        w1.close().unwrap();
        w2.close().unwrap();

        let begin = unix_secs_to_yyyymmddhhmmss(t0 - 1).unwrap();
        let end = unix_secs_to_yyyymmddhhmmss(t0 + 4).unwrap();

        let client1 = super::IdxClient::new(&base, id1);
        let rx1 = client1.query(&begin, &end).unwrap();
        let res1: Vec<Slice> = rx1.iter().collect();
        assert_eq!(res1.len(), 2);
        assert_eq!(res1[0].data.as_ref(), b"a1"); // changed
        assert_eq!(res1[1].data.as_ref(), b"a2"); // changed

        let client2 = super::IdxClient::new(&base, id2);
        let rx2 = client2.query(&begin, &end).unwrap();
        let res2: Vec<Slice> = rx2.iter().collect();
        assert_eq!(res2.len(), 2);
        assert_eq!(res2[0].data.as_ref(), b"b1"); // changed
        assert_eq!(res2[1].data.as_ref(), b"b2"); // changed

        let _ = fs::remove_dir_all(&base);
    }

    #[test]
    fn write_then_query_with_exported_file_list() {
        // temp dir
        let mut base = std::env::temp_dir();
        let uniq = format!(
            "idx_test_list_{}",
            std::time::SystemTime::now()
                .duration_since(UNIX_EPOCH)
                .unwrap()
                .as_nanos()
        );
        base.push(uniq);
        fs::create_dir_all(&base).unwrap();

        // id and writer
        let id = "listid";
        let mut writer = IdxRotateFileWriter::new(&base, id, 10);

        // timestamps
        let ts1 = 1_700_000_000i64;
        let ts2 = ts1 + 60;

        let st1 = UNIX_EPOCH + Duration::from_secs(ts1 as u64);
        let st2 = UNIX_EPOCH + Duration::from_secs(ts2 as u64);

        writer.write_slice(b"one", st1).unwrap();
        sleep(Duration::from_millis(5));
        writer.write_slice(b"two", st2).unwrap();
        writer.close().unwrap();

        // export file list from writer and create client from it
        let file_list = writer.export_file_list().unwrap();
        let client = super::IdxClient::new_from_list(&base, id, file_list);

        let begin = unix_secs_to_yyyymmddhhmmss(ts1 - 1).unwrap();
        let end = unix_secs_to_yyyymmddhhmmss(ts2 + 1).unwrap();

        let rx = client.query(&begin, &end).unwrap();
        let slices: Vec<Slice> = rx.iter().collect();

        assert_eq!(slices.len(), 2);
        assert_eq!(slices[0].data.as_ref(), b"one");
        assert_eq!(slices[1].data.as_ref(), b"two");

        let _ = fs::remove_dir_all(&base);
    }

    #[test]
    fn merge_adjacent_same_ts() {
        // temp dir
        let mut base = std::env::temp_dir();
        let uniq = format!(
            "idx_test_merge_{}",
            std::time::SystemTime::now()
                .duration_since(UNIX_EPOCH)
                .unwrap()
                .as_nanos()
        );
        base.push(uniq);
        fs::create_dir_all(&base).unwrap();

        let id = "mergeid";
        let mut writer = IdxRotateFileWriter::new(&base, id, 10);

        let ts = 1_700_000_000i64;
        let st = UNIX_EPOCH + Duration::from_secs(ts as u64);

        // write two slices with the exact same timestamp
        writer.write_slice(b"part1", st).unwrap();
        sleep(Duration::from_millis(5));
        writer.write_slice(b"part2", st).unwrap();
        writer.close().unwrap();

        let begin = unix_secs_to_yyyymmddhhmmss(ts).unwrap();
        let end = begin.clone();

        let client = super::IdxClient::new(&base, id);
        let rx = client.query(&begin, &end).unwrap();
        let res: Vec<Slice> = rx.iter().collect();

        // should be merged into a single slice with concatenated data
        assert_eq!(res.len(), 1);
        assert_eq!(res[0].data.as_ref(), b"part1part2");

        let _ = fs::remove_dir_all(&base);
    }
}
