//! Position-tracking output writer. All bytes destined for the output
//! MCAP go through this so we can build `ChunkIndex` / summary records
//! with exact byte offsets.

use std::fs::File;
use std::io::{BufWriter, Cursor, Write};
use std::path::Path;

use anyhow::{Context, Result, anyhow};
use binrw::BinWriterExt;
use mcap::records::{
    AttachmentIndex, Channel as ChannelRec, ChunkIndex, Footer, MetadataIndex, SchemaHeader,
    Statistics, SummaryOffset, op,
};

pub(crate) struct OutputWriter {
    inner: BufWriter<File>,
    pos: u64,
}

impl OutputWriter {
    pub(crate) fn new(path: &Path) -> Result<Self> {
        let f = File::create(path).with_context(|| format!("creating {}", path.display()))?;
        Ok(Self {
            inner: BufWriter::with_capacity(1 << 20, f),
            pos: 0,
        })
    }

    pub(crate) fn inner_mut(&mut self) -> &mut BufWriter<File> {
        &mut self.inner
    }

    pub(crate) fn position(&self) -> u64 {
        self.pos
    }

    pub(crate) fn advance_by(&mut self, n: u64) {
        self.pos = self.pos.saturating_add(n);
    }

    pub(crate) fn write_magic(&mut self) -> Result<()> {
        self.inner.write_all(mcap::MAGIC)?;
        self.pos += mcap::MAGIC.len() as u64;
        Ok(())
    }

    /// Write `[opcode][u64 LE length][body]` to the output, updating the
    /// position counter.
    pub(crate) fn write_record_bytes(&mut self, opcode: u8, body: &[u8]) -> Result<()> {
        self.inner.write_all(&[opcode])?;
        let len = body.len() as u64;
        self.inner.write_all(&len.to_le_bytes())?;
        self.inner.write_all(body)?;
        self.pos += 1 + 8 + len;
        Ok(())
    }

    /// Serialise a `binrw`-derived record and write it with the record
    /// header. Implemented as an inherent method so the `Required` bound on
    /// `T::Args<'a>` is resolved at the concrete call site.
    pub(crate) fn write_typed<T>(&mut self, opcode: u8, rec: &T) -> Result<()>
    where
        T: binrw::BinWrite,
        for<'a> <T as binrw::BinWrite>::Args<'a>: binrw::__private::Required,
    {
        let mut body = Vec::new();
        Cursor::new(&mut body)
            .write_le(rec)
            .map_err(|e| anyhow!("serialising record opcode {:#04x}: {}", opcode, e))?;
        self.write_record_bytes(opcode, &body)
    }

    /// Write a `Schema` record. The body is `SchemaHeader` followed by a
    /// `u32` data length and the schema bytes — `mcap::records` only
    /// exposes the header struct, so we frame the body ourselves.
    pub(crate) fn write_schema(&mut self, schema: &(SchemaHeader, Vec<u8>)) -> Result<()> {
        let (header, data) = schema;
        let mut body = Vec::new();
        Cursor::new(&mut body).write_le(header)?;
        let data_len: u32 = data.len().try_into().context("schema data > 4GiB")?;
        body.extend_from_slice(&data_len.to_le_bytes());
        body.extend_from_slice(data);
        self.write_record_bytes(op::SCHEMA, &body)
    }

    /// Emit the summary section, summary offsets, footer and trailing
    /// magic. `summary_start` is recorded so the footer can point at it.
    pub(crate) fn write_summary_and_footer(
        &mut self,
        schemas: &[(SchemaHeader, Vec<u8>)],
        channels: &[ChannelRec],
        chunk_indexes: &[ChunkIndex],
        attachment_indexes: &[AttachmentIndex],
        metadata_indexes: &[MetadataIndex],
        statistics: &Statistics,
    ) -> Result<()> {
        let summary_start = self.pos;
        let mut summary_offsets: Vec<SummaryOffset> = Vec::new();

        if !schemas.is_empty() {
            let start = self.pos;
            for s in schemas {
                self.write_schema(s)?;
            }
            summary_offsets.push(SummaryOffset {
                group_opcode: op::SCHEMA,
                group_start: start,
                group_length: self.pos - start,
            });
        }

        if !channels.is_empty() {
            let start = self.pos;
            for c in channels {
                self.write_typed(op::CHANNEL, c)?;
            }
            summary_offsets.push(SummaryOffset {
                group_opcode: op::CHANNEL,
                group_start: start,
                group_length: self.pos - start,
            });
        }

        if !chunk_indexes.is_empty() {
            let start = self.pos;
            for ci in chunk_indexes {
                self.write_typed(op::CHUNK_INDEX, ci)?;
            }
            summary_offsets.push(SummaryOffset {
                group_opcode: op::CHUNK_INDEX,
                group_start: start,
                group_length: self.pos - start,
            });
        }

        if !attachment_indexes.is_empty() {
            let start = self.pos;
            for ai in attachment_indexes {
                self.write_typed(op::ATTACHMENT_INDEX, ai)?;
            }
            summary_offsets.push(SummaryOffset {
                group_opcode: op::ATTACHMENT_INDEX,
                group_start: start,
                group_length: self.pos - start,
            });
        }

        if !metadata_indexes.is_empty() {
            let start = self.pos;
            for mi in metadata_indexes {
                self.write_typed(op::METADATA_INDEX, mi)?;
            }
            summary_offsets.push(SummaryOffset {
                group_opcode: op::METADATA_INDEX,
                group_start: start,
                group_length: self.pos - start,
            });
        }

        let stats_start = self.pos;
        self.write_typed(op::STATISTICS, statistics)?;
        summary_offsets.push(SummaryOffset {
            group_opcode: op::STATISTICS,
            group_start: stats_start,
            group_length: self.pos - stats_start,
        });

        let summary_offset_start = self.pos;
        for so in &summary_offsets {
            self.write_typed(op::SUMMARY_OFFSET, so)?;
        }

        let footer = Footer {
            summary_start,
            summary_offset_start,
            summary_crc: 0,
        };
        self.write_typed(op::FOOTER, &footer)?;
        self.inner.write_all(mcap::MAGIC)?;
        self.pos += mcap::MAGIC.len() as u64;
        Ok(())
    }

    pub(crate) fn finish(mut self) -> Result<()> {
        self.inner.flush()?;
        Ok(())
    }
}
