use byteorder::{BigEndian, ReadBytesExt, WriteBytesExt};
use std::fs::File;
use std::io::{Read, Write};
use std::path::PathBuf;
use std::time::{SystemTime, UNIX_EPOCH};
use tracing::{info, warn};

use crate::sfu::{ForwardPacket, ParticipantId, StreamId};

/// A session recorder writes stream packets to disk for later replay
pub struct Recorder {
    session_id: String,
    file: Option<File>,
    start_time: u64,
}

impl Recorder {
    pub fn new(session_id: &str, output_dir: &str) -> Result<Self, std::io::Error> {
        let mut path = PathBuf::from(output_dir);
        if !path.exists() {
            std::fs::create_dir_all(&path)?;
        }

        path.push(format!("{}.pulse", session_id));
        let file = File::create(&path)?;

        info!(
            "[Recording] Started recording session {} to {:?}",
            session_id, path
        );

        let recorder = Self {
            session_id: session_id.to_string(),
            file: Some(file),
            start_time: SystemTime::now()
                .duration_since(UNIX_EPOCH)
                .unwrap()
                .as_millis() as u64,
        };

        // Write header (magic bytes + version)
        if let Some(mut f) = recorder.file.as_ref() {
            f.write_all(b"PULSE_REC\x01")?;
        }

        Ok(recorder)
    }

    pub fn record_packet(&mut self, packet: &ForwardPacket) -> Result<(), std::io::Error> {
        if let Some(file) = &mut self.file {
            let now = SystemTime::now()
                .duration_since(UNIX_EPOCH)
                .unwrap()
                .as_millis() as u64;
            let offset_ms = now - self.start_time;

            // Format:
            // [8] offset ms
            // [4] sequence
            // [2] source len
            // [N] source string
            // [2] stream_id len
            // [N] stream_id string
            // [4] payload len
            // [N] payload bytes

            file.write_u64::<BigEndian>(offset_ms)?;
            file.write_u32::<BigEndian>(packet.sequence)?;

            let source_bytes = packet.source.as_bytes();
            file.write_u16::<BigEndian>(source_bytes.len() as u16)?;
            file.write_all(source_bytes)?;

            let stream_bytes = packet.stream_id.as_bytes();
            file.write_u16::<BigEndian>(stream_bytes.len() as u16)?;
            file.write_all(stream_bytes)?;

            file.write_u32::<BigEndian>(packet.payload.len() as u32)?;
            file.write_all(&packet.payload)?;

            file.flush()?;
        }
        Ok(())
    }

    pub fn stop(&mut self) {
        if self.file.is_some() {
            info!("[Recording] Stopped recording session {}", self.session_id);
            self.file = None;
        }
    }
}

pub struct Replayer {
    file: File,
}

impl Replayer {
    pub fn new(path: &str) -> Result<Self, Box<dyn std::error::Error>> {
        let mut file = File::open(path)?;

        let mut magic = [0u8; 10];
        file.read_exact(&mut magic)?;

        if &magic != b"PULSE_REC\x01" {
            return Err("Invalid pulse recording format".into());
        }

        Ok(Self { file })
    }

    pub fn read_next(&mut self) -> Result<Option<(u64, ForwardPacket)>, std::io::Error> {
        let offset_ms = match self.file.read_u64::<BigEndian>() {
            Ok(v) => v,
            Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => return Ok(None),
            Err(e) => return Err(e),
        };

        let sequence = self.file.read_u32::<BigEndian>()?;

        let source_len = self.file.read_u16::<BigEndian>()? as usize;
        let mut source_buf = vec![0u8; source_len];
        self.file.read_exact(&mut source_buf)?;
        let source = String::from_utf8_lossy(&source_buf).into_owned();

        let stream_len = self.file.read_u16::<BigEndian>()? as usize;
        let mut stream_buf = vec![0u8; stream_len];
        self.file.read_exact(&mut stream_buf)?;
        let stream_id = String::from_utf8_lossy(&stream_buf).into_owned();

        let payload_len = self.file.read_u32::<BigEndian>()? as usize;
        let mut payload = vec![0u8; payload_len];
        self.file.read_exact(&mut payload)?;

        Ok(Some((
            offset_ms,
            ForwardPacket {
                source,
                stream_id,
                sequence,
                payload,
                lossless: false,
            },
        )))
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn tmp_dir() -> tempfile::TempDir {
        tempfile::tempdir().expect("failed to create temp dir")
    }

    #[test]
    fn record_and_replay_single_packet() {
        let dir = tmp_dir();
        let dir_path = dir.path().to_str().unwrap();

        let pkt = ForwardPacket {
            source: "alice".into(),
            stream_id: "audio".into(),
            sequence: 42,
            payload: vec![0xDE, 0xAD, 0xBE, 0xEF],
            lossless: false,
        };

        // Record
        let mut rec = Recorder::new("test-session", dir_path).unwrap();
        rec.record_packet(&pkt).unwrap();
        rec.stop();

        // Replay
        let rec_path = dir.path().join("test-session.pulse");
        let mut rep = Replayer::new(rec_path.to_str().unwrap()).unwrap();

        let (offset_ms, decoded) = rep.read_next().unwrap().expect("should have one packet");
        assert!(
            offset_ms < 1000,
            "offset should be small for immediate write"
        );
        assert_eq!(decoded.source, "alice");
        assert_eq!(decoded.stream_id, "audio");
        assert_eq!(decoded.sequence, 42);
        assert_eq!(decoded.payload, vec![0xDE, 0xAD, 0xBE, 0xEF]);

        // No more packets
        assert!(rep.read_next().unwrap().is_none());
    }

    #[test]
    fn record_and_replay_multiple_packets_fifo() {
        let dir = tmp_dir();
        let dir_path = dir.path().to_str().unwrap();

        let mut rec = Recorder::new("multi", dir_path).unwrap();

        for i in 0..5u32 {
            let pkt = ForwardPacket {
                source: format!("peer-{}", i),
                stream_id: "video".into(),
                sequence: i,
                payload: vec![i as u8; (i + 1) as usize],
                lossless: false,
            };
            rec.record_packet(&pkt).unwrap();
        }
        rec.stop();

        let rec_path = dir.path().join("multi.pulse");
        let mut rep = Replayer::new(rec_path.to_str().unwrap()).unwrap();

        let mut prev_offset = 0u64;
        for i in 0..5u32 {
            let (offset, pkt) = rep.read_next().unwrap().expect("should have packet");
            assert!(
                offset >= prev_offset,
                "offsets should be monotonically non-decreasing"
            );
            prev_offset = offset;

            assert_eq!(pkt.source, format!("peer-{}", i));
            assert_eq!(pkt.stream_id, "video");
            assert_eq!(pkt.sequence, i);
            assert_eq!(pkt.payload.len(), (i + 1) as usize);
        }

        assert!(rep.read_next().unwrap().is_none());
    }

    #[test]
    fn empty_recording_replays_nothing() {
        let dir = tmp_dir();
        let dir_path = dir.path().to_str().unwrap();

        let mut rec = Recorder::new("empty", dir_path).unwrap();
        rec.stop();

        let rec_path = dir.path().join("empty.pulse");
        let mut rep = Replayer::new(rec_path.to_str().unwrap()).unwrap();
        assert!(rep.read_next().unwrap().is_none());
    }

    #[test]
    fn replayer_rejects_bad_magic() {
        let dir = tmp_dir();
        let bad_file = dir.path().join("bad.pulse");
        std::fs::write(&bad_file, b"NOT_A_REC\x01and some data").unwrap();

        let result = Replayer::new(bad_file.to_str().unwrap());
        assert!(result.is_err(), "should reject invalid magic bytes");
    }

    #[test]
    fn stop_is_idempotent() {
        let dir = tmp_dir();
        let dir_path = dir.path().to_str().unwrap();

        let mut rec = Recorder::new("idem", dir_path).unwrap();
        rec.stop();
        rec.stop(); // should not panic

        // Recording after stop should be a no-op
        let pkt = ForwardPacket {
            source: "x".into(),
            stream_id: "y".into(),
            sequence: 1,
            payload: vec![1],
            lossless: false,
        };
        rec.record_packet(&pkt).unwrap(); // writes nothing, no panic
    }
}
