//! Per-output-channel temp-file buffer.
//!
//! During the read/merge pass each message is appended as
//! `[MessageHeader 22 bytes][payload]` to its output channel's temp file, with
//! `(log_time, temp_offset, payload_len)` recorded in memory. The write pass
//! replays one channel at a time, in `log_time` order, into single-channel
//! chunks. Temp files keep peak RAM bounded by the in-memory entry vectors
//! rather than the (large, already-compressed) payloads.

use std::fs::File;
use std::io::{BufWriter, Cursor, Read, Seek, SeekFrom, Write};

use anyhow::{anyhow, Context, Result};
use binrw::{BinReaderExt, BinWrite};
use mcap::records::MessageHeader;
use tempfile::tempfile;

use crate::chunk_builder::MESSAGE_HEADER_LEN;

#[derive(Clone, Copy, Debug)]
pub(crate) struct MsgEntry {
    pub log_time: u64,
    pub temp_offset: u64,
    pub payload_len: u32,
}

pub(crate) struct ChannelDemux {
    writer: Option<BufWriter<File>>,
    read_handle: Option<File>,
    pub entries: Vec<MsgEntry>,
}

impl ChannelDemux {
    pub(crate) fn new() -> Result<Self> {
        let file = tempfile().context("creating temp file for channel demux")?;
        Ok(Self {
            writer: Some(BufWriter::with_capacity(1 << 20, file)),
            read_handle: None,
            entries: Vec::new(),
        })
    }

    /// Append one message's header + payload to the channel temp file.
    pub(crate) fn push(&mut self, header: &MessageHeader, payload: &[u8]) -> Result<()> {
        let writer = self
            .writer
            .as_mut()
            .ok_or_else(|| anyhow!("demux already finalised"))?;
        let pos = writer.stream_position().context("temp file position")?;
        let mut header_buf = [0u8; MESSAGE_HEADER_LEN as usize];
        header.write_le(&mut Cursor::new(&mut header_buf[..]))?;
        writer.write_all(&header_buf)?;
        writer.write_all(payload)?;
        let payload_len: u32 = payload.len().try_into().context("payload > 4GiB")?;
        self.entries.push(MsgEntry {
            log_time: header.log_time,
            temp_offset: pos,
            payload_len,
        });
        Ok(())
    }

    /// Flush the writer and stash the file handle for the read pass.
    pub(crate) fn finalise(&mut self) -> Result<()> {
        let writer = self
            .writer
            .take()
            .ok_or_else(|| anyhow!("demux already finalised"))?;
        let file = writer
            .into_inner()
            .map_err(|e| anyhow!("flush temp file: {}", e.into_error()))?;
        self.read_handle = Some(file);
        Ok(())
    }

    /// A fresh, independently-seekable read handle over the channel temp file.
    pub(crate) fn reopen_for_read(&self) -> Result<File> {
        let handle = self
            .read_handle
            .as_ref()
            .ok_or_else(|| anyhow!("demux not finalised"))?;
        handle.try_clone().context("cloning temp file handle")
    }
}

/// Read one buffered message (`[MessageHeader][payload]`) back from a channel
/// temp file at the entry's offset.
pub(crate) fn read_entry(reader: &mut File, entry: &MsgEntry) -> Result<(MessageHeader, Vec<u8>)> {
    reader.seek(SeekFrom::Start(entry.temp_offset))?;
    let mut header_bytes = [0u8; MESSAGE_HEADER_LEN as usize];
    reader.read_exact(&mut header_bytes)?;
    let header: MessageHeader = Cursor::new(&header_bytes[..])
        .read_le()
        .context("parsing MessageHeader from temp file")?;
    let mut payload = vec![0u8; entry.payload_len as usize];
    reader.read_exact(&mut payload)?;
    Ok((header, payload))
}
