//! Streaming reader for STAR-lite archives.
//!
//! Handles both format versions. The header (version, kind, commitment, and
//! optional meta bytes) is parsed at construction and held in memory; entries
//! are read lazily one at a time, with per-entry memory bounded by [`Limits`].

use std::io::Read;

use star_repo::Cid;

use star_repo::commit::{CommitInfo, parse_partial_commit};
use star_repo::space::{SpaceMeta, parse_space_meta};
use star_repo::varint;

use crate::error::{Error, FormatError, Result};
use crate::header::{Commitment, Kind, Version};

// Re-exported for callers that used to import these from `reader`.
pub use crate::header::{
    HEADER_CID_LEN, HEADER_CID_PREFIX, LTHASH_COMMITMENT_LEN, MAGIC_V0, MAGIC_V1,
    MAX_PARTIAL_COMMIT_LEN, MAX_SPACE_META_LEN,
};

/// Limits on the size of individual fields, applied during parsing.
///
/// These exist to prevent malformed or malicious archives from causing
/// unbounded allocations.
#[derive(Debug, Clone, Copy)]
pub struct Limits {
    pub max_key_len: u64,
    pub max_record_len: u64,
}

impl Default for Limits {
    fn default() -> Self {
        Self {
            max_key_len: 1024,               // 1 KB
            max_record_len: 5 * 1024 * 1024, // 5 MB
        }
    }
}

/// A single key/record pair from a STAR-lite archive.
#[derive(Debug, Clone)]
pub struct Entry {
    pub key: String,
    pub record: Vec<u8>,
}

/// Streaming reader over a STAR-lite archive.
///
/// Construct with [`Reader::new`] (or [`Reader::with_limits`]); this consumes
/// the whole header. Then iterate via [`Reader::next_entry`] or by treating the
/// reader as an `Iterator<Item = Result<Entry>>`.
pub struct Reader<R: Read> {
    inner: R,
    limits: Limits,
    version: Version,
    kind: Kind,
    commitment: Commitment,
    meta: Option<Vec<u8>>,
    /// Bytes of the previous key, used to enforce strict lex ordering.
    prev_key: Option<Vec<u8>>,
    /// Set once we hit EOF or any error; makes the iterator terminate.
    finished: bool,
}

impl<R: Read> Reader<R> {
    pub fn new(inner: R) -> Result<Self> {
        Self::with_limits(inner, Limits::default())
    }

    pub fn with_limits(mut inner: R, limits: Limits) -> Result<Self> {
        // Magic: 3 bytes. A short read here means "not a STAR-lite file" rather
        // than "truncated entry", so we surface InvalidMagic for any short read.
        let mut magic = [0u8; 3];
        let mut filled = 0;
        while filled < magic.len() {
            let n = inner.read(&mut magic[filled..])?;
            if n == 0 {
                return Err(FormatError::InvalidMagic.into());
            }
            filled += n;
        }
        let version = Version::from_magic(&magic).ok_or(FormatError::InvalidMagic)?;

        // v0 has no kind byte and is implicitly kind C.
        let kind = if version.has_kind_byte() {
            let mut kind_byte = [0u8; 1];
            read_exact_or_eof(&mut inner, &mut kind_byte)?;
            Kind::from_byte(kind_byte[0])?
        } else {
            Kind::Commit
        };

        let mut commitment_buf = vec![0u8; kind.commitment_len()];
        read_exact_or_eof(&mut inner, &mut commitment_buf)?;
        let commitment = Commitment::from_bytes(kind, &commitment_buf)?;

        // Meta: varint-prefixed bytes; len == 0 means absent.
        let meta_len = varint::read_required(&mut inner)?;
        let limit = kind.max_meta_len();
        if meta_len > limit {
            return Err(FormatError::SizeLimit {
                field: "meta",
                actual: meta_len,
                limit,
            }
            .into());
        }
        let meta = if meta_len == 0 {
            None
        } else {
            let mut buf = vec![0u8; meta_len as usize];
            read_exact_or_eof(&mut inner, &mut buf)?;
            Some(buf)
        };

        Ok(Self {
            inner,
            limits,
            version,
            kind,
            commitment,
            meta,
            prev_key: None,
            finished: false,
        })
    }

    /// Format version this archive is encoded in.
    pub fn version(&self) -> Version {
        self.version
    }

    /// Which commitment covers this archive.
    pub fn kind(&self) -> Kind {
        self.kind
    }

    /// The header commitment.
    pub fn commitment(&self) -> Commitment {
        self.commitment
    }

    /// The header CID. Kind `C` only — kind `L` commits with a bare digest,
    /// not a CID, and is rejected with [`FormatError::KindMismatch`].
    pub fn header_cid(&self) -> Result<Cid> {
        self.require_kind(Kind::Commit)?;
        Ok(self
            .commitment
            .as_cid()
            .expect("kind C always carries a CID commitment"))
    }

    /// The header LtHash commitment. Kind `L` only — kind `C` is rejected with
    /// [`FormatError::KindMismatch`].
    pub fn header_lthash(&self) -> Result<[u8; 32]> {
        self.require_kind(Kind::LtHash)?;
        Ok(self
            .commitment
            .as_lthash()
            .expect("kind L always carries an LtHash commitment"))
    }

    /// Raw meta bytes, if present. Interpret according to [`Reader::kind`], or
    /// use [`Reader::partial_commit`] / [`Reader::parse_space_meta`].
    pub fn meta(&self) -> Option<&[u8]> {
        self.meta.as_deref()
    }

    /// Raw partial commit bytes. Kind `C` only, so space meta can never be
    /// mistakenly fed to a commit parser. `Ok(None)` when the archive carries
    /// no meta object.
    pub fn partial_commit(&self) -> Result<Option<&[u8]>> {
        self.require_kind(Kind::Commit)?;
        Ok(self.meta.as_deref())
    }

    /// Parse the meta object as an atproto commit, splicing the header CID
    /// back in as `data`.
    ///
    /// `Ok(None)` when the archive carries no meta object; an error when this
    /// is not a kind `C` archive.
    pub fn parse_commit(&self) -> Result<Option<CommitInfo>> {
        let data = self.header_cid()?;
        match &self.meta {
            Some(bytes) => Ok(Some(parse_partial_commit(bytes, data)?)),
            None => Ok(None),
        }
    }

    /// Parse the meta object as space repo meta.
    ///
    /// `Ok(None)` when the archive carries no meta object; an error when this
    /// is not a kind `L` archive.
    pub fn parse_space_meta(&self) -> Result<Option<SpaceMeta>> {
        self.require_kind(Kind::LtHash)?;
        match &self.meta {
            Some(bytes) => Ok(Some(parse_space_meta(bytes)?)),
            None => Ok(None),
        }
    }

    pub fn limits(&self) -> &Limits {
        &self.limits
    }

    fn require_kind(&self, expected: Kind) -> Result<()> {
        if self.kind != expected {
            return Err(FormatError::KindMismatch {
                expected: expected.to_byte(),
                actual: self.kind.to_byte(),
            }
            .into());
        }
        Ok(())
    }

    /// Read the next entry, or `Ok(None)` at clean EOF.
    ///
    /// Any error is terminal: the reader is marked finished so a caller looping
    /// over the iterator cannot spin forever on a repeating failure.
    pub fn next_entry(&mut self) -> Result<Option<Entry>> {
        if self.finished {
            return Ok(None);
        }
        let result = self.read_entry();
        if !matches!(result, Ok(Some(_))) {
            self.finished = true;
        }
        result
    }

    fn read_entry(&mut self) -> Result<Option<Entry>> {
        let key_len = match varint::read(&mut self.inner)? {
            Some(n) => n,
            None => return Ok(None),
        };
        if key_len == 0 {
            return Err(FormatError::EmptyKey.into());
        }
        if key_len > self.limits.max_key_len {
            return Err(FormatError::SizeLimit {
                field: "key",
                actual: key_len,
                limit: self.limits.max_key_len,
            }
            .into());
        }

        let mut key_bytes = vec![0u8; key_len as usize];
        read_exact_or_eof(&mut self.inner, &mut key_bytes)?;

        // Strict ordering check (against raw bytes, before utf-8 validation).
        if let Some(prev) = &self.prev_key
            && key_bytes.as_slice() <= prev.as_slice()
        {
            return Err(FormatError::KeyOrder {
                prev: String::from_utf8_lossy(prev).into_owned(),
                curr: String::from_utf8_lossy(&key_bytes).into_owned(),
            }
            .into());
        }

        let key = std::str::from_utf8(&key_bytes)
            .map_err(|e| Error::Format(FormatError::InvalidUtf8(e)))?
            .to_owned();

        let record_len = varint::read_required(&mut self.inner)?;
        if record_len > self.limits.max_record_len {
            return Err(FormatError::SizeLimit {
                field: "record",
                actual: record_len,
                limit: self.limits.max_record_len,
            }
            .into());
        }
        let mut record = vec![0u8; record_len as usize];
        read_exact_or_eof(&mut self.inner, &mut record)?;

        self.prev_key = Some(key_bytes);
        Ok(Some(Entry { key, record }))
    }
}

impl<R: Read> Iterator for Reader<R> {
    type Item = Result<Entry>;

    fn next(&mut self) -> Option<Self::Item> {
        match self.next_entry() {
            Ok(Some(e)) => Some(Ok(e)),
            Ok(None) => None,
            Err(e) => Some(Err(e)),
        }
    }
}

/// Read exactly `buf.len()` bytes; treat any short read as `UnexpectedEof`.
fn read_exact_or_eof<R: Read>(r: &mut R, buf: &mut [u8]) -> Result<()> {
    let mut filled = 0;
    while filled < buf.len() {
        let n = r.read(&mut buf[filled..])?;
        if n == 0 {
            return Err(FormatError::UnexpectedEof.into());
        }
        filled += n;
    }
    Ok(())
}