//! Accumulates message records in an in-memory buffer, then emits the buffer
//! as a single-channel Chunk record.
//!
//! Chunks are written **uncompressed**: the payloads (H.264 video, PNG depth,
//! CDR telemetry) are already compressed, so zstd would buy ~1% for real CPU.
//! Per the MCAP spec an empty `compression` field means "no compression", and
//! `compressed_size == uncompressed_size` with the records stored verbatim.

use std::collections::BTreeMap;
use std::io::{Cursor, Write};

use anyhow::{anyhow, Result};
use binrw::BinWriterExt;
use mcap::records::{ChunkHeader, ChunkIndex, MessageHeader, MessageIndex, MessageIndexEntry, op};

use crate::writer::OutputWriter;

/// Empty string => no compression, per the MCAP spec.
pub(crate) const COMPRESSION_NONE: &str = "";
pub(crate) const MESSAGE_HEADER_LEN: u64 = 22;

pub(crate) struct ChunkBuilder {
    uncompressed: Vec<u8>,
    indexes: Vec<MessageIndexEntry>,
    channel_id: Option<u16>,
    msg_start: Option<u64>,
    msg_end: u64,
    crc: crc32fast::Hasher,
}

impl ChunkBuilder {
    pub(crate) fn new(target_size: u64) -> Self {
        Self {
            uncompressed: Vec::with_capacity(target_size as usize),
            indexes: Vec::new(),
            channel_id: None,
            msg_start: None,
            msg_end: 0,
            crc: crc32fast::Hasher::new(),
        }
    }

    pub(crate) fn uncompressed_len(&self) -> u64 {
        self.uncompressed.len() as u64
    }

    pub(crate) fn is_empty(&self) -> bool {
        self.uncompressed.is_empty()
    }

    pub(crate) fn push_message(&mut self, header: &MessageHeader, payload: &[u8]) -> Result<()> {
        if let Some(c) = self.channel_id {
            debug_assert_eq!(
                c, header.channel_id,
                "ChunkBuilder is single-channel by construction"
            );
        } else {
            self.channel_id = Some(header.channel_id);
        }

        let offset_in_chunk = self.uncompressed.len() as u64;
        let total_len = MESSAGE_HEADER_LEN + payload.len() as u64;
        let header_bytes = serialise_message_header(header)?;
        let mut prelude = [0u8; 9];
        prelude[0] = op::MESSAGE;
        prelude[1..].copy_from_slice(&total_len.to_le_bytes());
        self.uncompressed.extend_from_slice(&prelude);
        self.uncompressed.extend_from_slice(&header_bytes);
        self.uncompressed.extend_from_slice(payload);

        self.indexes.push(MessageIndexEntry {
            log_time: header.log_time,
            offset: offset_in_chunk,
        });
        self.msg_start.get_or_insert(header.log_time);
        if header.log_time > self.msg_end {
            self.msg_end = header.log_time;
        }
        Ok(())
    }

    /// Emit the buffered single-channel chunk to `writer`, returning the
    /// `ChunkIndex` (with `message_index_offsets` left empty for later
    /// relocation) and the `MessageIndex` to be written in the relocated block.
    pub(crate) fn finish_into(
        &mut self,
        writer: &mut OutputWriter,
    ) -> Result<(ChunkIndex, MessageIndex)> {
        let uncompressed_size = self.uncompressed.len() as u64;
        self.crc.update(&self.uncompressed);
        let crc = std::mem::replace(&mut self.crc, crc32fast::Hasher::new()).finalize();

        // No compression: the chunk body's records are the raw bytes.
        let compressed_size = uncompressed_size;

        let chunk_header = ChunkHeader {
            message_start_time: self.msg_start.unwrap_or(0),
            message_end_time: self.msg_end,
            uncompressed_size,
            uncompressed_crc: crc,
            compression: COMPRESSION_NONE.to_string(),
            compressed_size,
        };

        let mut header_bytes = Vec::new();
        Cursor::new(&mut header_bytes).write_le(&chunk_header)?;

        let chunk_start_offset = writer.position();
        let body_len = header_bytes.len() as u64 + compressed_size;

        writer.inner_mut().write_all(&[op::CHUNK])?;
        writer.inner_mut().write_all(&body_len.to_le_bytes())?;
        writer.inner_mut().write_all(&header_bytes)?;
        writer.inner_mut().write_all(&self.uncompressed)?;
        let chunk_length = 1 + 8 + body_len;
        writer.advance_by(chunk_length);

        let channel_id = self
            .channel_id
            .ok_or_else(|| anyhow!("finish_into called with no channel_id"))?;

        let mi = MessageIndex {
            channel_id,
            records: std::mem::take(&mut self.indexes),
        };
        let ci = ChunkIndex {
            message_start_time: chunk_header.message_start_time,
            message_end_time: chunk_header.message_end_time,
            chunk_start_offset,
            chunk_length,
            // Filled in when MessageIndex records are relocated into one block.
            message_index_offsets: BTreeMap::new(),
            message_index_length: 0,
            compression: COMPRESSION_NONE.to_string(),
            compressed_size,
            uncompressed_size,
        };

        self.uncompressed.clear();
        self.channel_id = None;
        self.msg_start = None;
        self.msg_end = 0;

        Ok((ci, mi))
    }
}

fn serialise_message_header(header: &MessageHeader) -> Result<[u8; MESSAGE_HEADER_LEN as usize]> {
    let mut buf = [0u8; MESSAGE_HEADER_LEN as usize];
    Cursor::new(&mut buf[..]).write_le(header)?;
    Ok(buf)
}
