use crate::block::Block;
use crate::cid::{Cid, SerializableCid};
use crate::compression::CompressionAlgorithm;
use crate::error::{Error, Result};
use bytes::Bytes;
use std::io::{Read, Write};
const CAR_VERSION: u64 = 1;
const MAX_VARINT_SIZE: usize = 10;
#[derive(Debug, Clone)]
pub struct CarHeader {
pub version: u64,
pub roots: Vec<Cid>,
}
impl CarHeader {
pub fn new(roots: Vec<Cid>) -> Self {
Self {
version: CAR_VERSION,
roots,
}
}
fn encode(&self) -> Result<Bytes> {
use crate::ipld::Ipld;
use std::collections::BTreeMap;
let mut map = BTreeMap::new();
map.insert("version".to_string(), Ipld::Integer(self.version as i128));
let roots: Vec<Ipld> = self
.roots
.iter()
.map(|cid| Ipld::Link(SerializableCid(*cid)))
.collect();
map.insert("roots".to_string(), Ipld::List(roots));
let ipld = Ipld::Map(map);
ipld.to_dag_cbor().map(Bytes::from)
}
fn decode(data: &[u8]) -> Result<Self> {
use crate::ipld::Ipld;
let ipld = Ipld::from_dag_cbor(data)?;
let map = match ipld {
Ipld::Map(m) => m,
_ => {
return Err(Error::Deserialization(
"CAR header must be a map".to_string(),
))
}
};
let version = match map.get("version") {
Some(Ipld::Integer(v)) => *v as u64,
_ => {
return Err(Error::Deserialization(
"CAR header missing version".to_string(),
))
}
};
if version != CAR_VERSION {
return Err(Error::Deserialization(format!(
"Unsupported CAR version: {}",
version
)));
}
let roots = match map.get("roots") {
Some(Ipld::List(list)) => list
.iter()
.map(|item| match item {
Ipld::Link(SerializableCid(cid)) => Ok(*cid),
_ => Err(Error::Deserialization(
"Invalid root CID in header".to_string(),
)),
})
.collect::<Result<Vec<Cid>>>()?,
_ => {
return Err(Error::Deserialization(
"CAR header missing roots".to_string(),
))
}
};
Ok(Self { version, roots })
}
}
#[derive(Debug, Clone, Default)]
pub struct CarCompressionStats {
pub blocks_processed: usize,
pub uncompressed_bytes: usize,
pub compressed_bytes: usize,
pub blocks_compressed: usize,
}
impl CarCompressionStats {
pub fn new() -> Self {
Self::default()
}
pub fn compression_ratio(&self) -> f64 {
if self.uncompressed_bytes == 0 {
1.0
} else {
self.compressed_bytes as f64 / self.uncompressed_bytes as f64
}
}
pub fn bytes_saved(&self) -> usize {
self.uncompressed_bytes
.saturating_sub(self.compressed_bytes)
}
pub fn compression_percentage(&self) -> f64 {
if self.uncompressed_bytes == 0 {
0.0
} else {
(self.bytes_saved() as f64 / self.uncompressed_bytes as f64) * 100.0
}
}
}
pub struct CarWriterBuilder {
roots: Vec<Cid>,
compression: Option<(CompressionAlgorithm, i32)>,
}
impl CarWriterBuilder {
pub fn new(roots: Vec<Cid>) -> Self {
Self {
roots,
compression: None,
}
}
pub fn with_compression(mut self, algorithm: CompressionAlgorithm, level: i32) -> Self {
self.compression = Some((algorithm, level));
self
}
pub fn build<W: Write>(self, writer: W) -> Result<CarWriter<W>> {
CarWriter::new_with_options(writer, self.roots, self.compression)
}
}
pub struct CarWriter<W: Write> {
writer: W,
header_written: bool,
compression: Option<(CompressionAlgorithm, i32)>,
stats: CarCompressionStats,
}
impl<W: Write> CarWriter<W> {
pub fn new(writer: W, roots: Vec<Cid>) -> Result<Self> {
Self::new_with_options(writer, roots, None)
}
fn new_with_options(
writer: W,
roots: Vec<Cid>,
compression: Option<(CompressionAlgorithm, i32)>,
) -> Result<Self> {
let mut car_writer = Self {
writer,
header_written: false,
compression,
stats: CarCompressionStats::new(),
};
car_writer.write_header(&CarHeader::new(roots))?;
Ok(car_writer)
}
fn write_header(&mut self, header: &CarHeader) -> Result<()> {
let header_bytes = header.encode()?;
let header_len = header_bytes.len();
write_varint(&mut self.writer, header_len as u64)?;
self.writer.write_all(&header_bytes)?;
self.header_written = true;
Ok(())
}
pub fn write_block(&mut self, block: &Block) -> Result<()> {
if !self.header_written {
return Err(Error::InvalidData("CAR header not written".to_string()));
}
let cid_bytes = block.cid().to_bytes();
let data = block.data();
self.stats.blocks_processed += 1;
self.stats.uncompressed_bytes += data.len();
let (final_data, was_compressed) = if let Some((algorithm, level)) = self.compression {
let compressed = crate::compression::compress(data, algorithm, level as u8)?;
let is_compressed = algorithm != CompressionAlgorithm::None;
if is_compressed {
self.stats.blocks_compressed += 1;
}
self.stats.compressed_bytes += compressed.len();
(compressed, is_compressed)
} else {
self.stats.compressed_bytes += data.len();
(data.clone(), false)
};
let total_len = cid_bytes.len() + 1 + final_data.len();
write_varint(&mut self.writer, total_len as u64)?;
self.writer.write_all(&cid_bytes)?;
self.writer.write_all(&[was_compressed as u8])?;
self.writer.write_all(&final_data)?;
Ok(())
}
pub fn stats(&self) -> &CarCompressionStats {
&self.stats
}
pub fn finish(mut self) -> Result<()> {
self.writer.flush()?;
Ok(())
}
}
pub struct CarReader<R: Read> {
reader: R,
header: CarHeader,
}
impl<R: Read> CarReader<R> {
pub fn new(mut reader: R) -> Result<Self> {
let header_len = read_varint(&mut reader)?;
let mut header_bytes = vec![0u8; header_len as usize];
reader.read_exact(&mut header_bytes)?;
let header = CarHeader::decode(&header_bytes)?;
Ok(Self { reader, header })
}
pub fn roots(&self) -> &[Cid] {
&self.header.roots
}
pub fn read_block(&mut self) -> Result<Option<Block>> {
let total_len = match read_varint_opt(&mut self.reader) {
Ok(Some(len)) => len,
Ok(None) => return Ok(None), Err(e) => return Err(e),
};
let mut block_bytes = vec![0u8; total_len as usize];
self.reader.read_exact(&mut block_bytes)?;
let mut cursor = &block_bytes[..];
let cid = Cid::read_bytes(&mut cursor)
.map_err(|e| Error::Cid(format!("Failed to parse CID: {}", e)))?;
let (is_compressed, data_start) = if !cursor.is_empty() {
let flag = cursor[0];
if flag == 0 || flag == 1 {
(flag == 1, 1)
} else {
(false, 0)
}
} else {
return Err(Error::Deserialization("Empty block data".to_string()));
};
let raw_data = &cursor[data_start..];
let final_data = if is_compressed {
let raw_bytes = Bytes::from(raw_data.to_vec());
crate::compression::decompress(&raw_bytes, CompressionAlgorithm::Zstd).or_else(
|_| crate::compression::decompress(&raw_bytes, CompressionAlgorithm::Lz4),
)?
} else {
Bytes::from(raw_data.to_vec())
};
let block = Block::new(final_data)?;
if block.cid() != &cid {
return Err(Error::InvalidData(format!(
"Block CID mismatch: expected {}, got {}",
cid,
block.cid()
)));
}
Ok(Some(block))
}
pub fn read_all_blocks(&mut self) -> Result<Vec<Block>> {
let mut blocks = Vec::new();
while let Some(block) = self.read_block()? {
blocks.push(block);
}
Ok(blocks)
}
}
fn write_varint<W: Write>(writer: &mut W, mut value: u64) -> Result<()> {
let mut buf = [0u8; MAX_VARINT_SIZE];
let mut i = 0;
loop {
let mut byte = (value & 0x7F) as u8;
value >>= 7;
if value != 0 {
byte |= 0x80;
}
buf[i] = byte;
i += 1;
if value == 0 {
break;
}
}
writer.write_all(&buf[..i])?;
Ok(())
}
fn read_varint<R: Read>(reader: &mut R) -> Result<u64> {
read_varint_opt(reader)?
.ok_or_else(|| Error::Deserialization("Unexpected EOF reading varint".to_string()))
}
fn read_varint_opt<R: Read>(reader: &mut R) -> Result<Option<u64>> {
let mut result = 0u64;
let mut shift = 0;
let mut buf = [0u8; 1];
for _ in 0..MAX_VARINT_SIZE {
match reader.read_exact(&mut buf) {
Ok(()) => {}
Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof && shift == 0 => {
return Ok(None); }
Err(e) => return Err(Error::from(e)),
}
let byte = buf[0];
result |= ((byte & 0x7F) as u64) << shift;
if byte & 0x80 == 0 {
return Ok(Some(result));
}
shift += 7;
if shift >= 64 {
return Err(Error::Deserialization("Varint too large".to_string()));
}
}
Err(Error::Deserialization(
"Varint exceeds maximum size".to_string(),
))
}
#[cfg(test)]
mod tests {
use super::*;
use crate::block::Block;
use bytes::Bytes;
#[test]
fn test_car_header_encode_decode() {
use crate::cid::CidBuilder;
let cid1 = CidBuilder::new().build(b"test1").unwrap();
let cid2 = CidBuilder::new().build(b"test2").unwrap();
let header = CarHeader::new(vec![cid1, cid2]);
let encoded = header.encode().unwrap();
let decoded = CarHeader::decode(&encoded).unwrap();
assert_eq!(decoded.version, 1);
assert_eq!(decoded.roots.len(), 2);
assert_eq!(decoded.roots[0], cid1);
assert_eq!(decoded.roots[1], cid2);
}
#[test]
fn test_car_write_read() {
let block1 = Block::new(Bytes::from_static(b"Hello, CAR!")).unwrap();
let block2 = Block::new(Bytes::from_static(b"CAR format test")).unwrap();
let mut car_data = Vec::new();
let mut writer = CarWriter::new(&mut car_data, vec![*block1.cid()]).unwrap();
writer.write_block(&block1).unwrap();
writer.write_block(&block2).unwrap();
writer.finish().unwrap();
let mut reader = CarReader::new(&car_data[..]).unwrap();
assert_eq!(reader.roots().len(), 1);
assert_eq!(reader.roots()[0], *block1.cid());
let read_block1 = reader.read_block().unwrap().unwrap();
assert_eq!(read_block1.cid(), block1.cid());
assert_eq!(read_block1.data(), block1.data());
let read_block2 = reader.read_block().unwrap().unwrap();
assert_eq!(read_block2.cid(), block2.cid());
assert_eq!(read_block2.data(), block2.data());
assert!(reader.read_block().unwrap().is_none());
}
#[test]
fn test_car_read_all_blocks() {
let blocks: Vec<Block> = (0..5)
.map(|i| Block::new(Bytes::from(format!("Block {}", i))).unwrap())
.collect();
let mut car_data = Vec::new();
let mut writer = CarWriter::new(&mut car_data, vec![*blocks[0].cid()]).unwrap();
for block in &blocks {
writer.write_block(block).unwrap();
}
writer.finish().unwrap();
let mut reader = CarReader::new(&car_data[..]).unwrap();
let read_blocks = reader.read_all_blocks().unwrap();
assert_eq!(read_blocks.len(), blocks.len());
for (i, block) in read_blocks.iter().enumerate() {
assert_eq!(block.cid(), blocks[i].cid());
assert_eq!(block.data(), blocks[i].data());
}
}
#[test]
fn test_varint_roundtrip() {
let test_values = vec![0, 1, 127, 128, 255, 256, 65535, 65536, u64::MAX];
for value in test_values {
let mut buf = Vec::new();
write_varint(&mut buf, value).unwrap();
let mut cursor = &buf[..];
let decoded = read_varint(&mut cursor).unwrap();
assert_eq!(decoded, value);
}
}
#[test]
fn test_car_empty_roots() {
let block = Block::new(Bytes::from_static(b"test")).unwrap();
let mut car_data = Vec::new();
let mut writer = CarWriter::new(&mut car_data, vec![]).unwrap();
writer.write_block(&block).unwrap();
writer.finish().unwrap();
let reader = CarReader::new(&car_data[..]).unwrap();
assert_eq!(reader.roots().len(), 0);
}
#[test]
fn test_car_multiple_roots() {
use crate::cid::CidBuilder;
let cid1 = CidBuilder::new().build(b"root1").unwrap();
let cid2 = CidBuilder::new().build(b"root2").unwrap();
let cid3 = CidBuilder::new().build(b"root3").unwrap();
let block = Block::new(Bytes::from_static(b"data")).unwrap();
let mut car_data = Vec::new();
let mut writer = CarWriter::new(&mut car_data, vec![cid1, cid2, cid3]).unwrap();
writer.write_block(&block).unwrap();
writer.finish().unwrap();
let reader = CarReader::new(&car_data[..]).unwrap();
let roots = reader.roots();
assert_eq!(roots.len(), 3);
assert_eq!(roots[0], cid1);
assert_eq!(roots[1], cid2);
assert_eq!(roots[2], cid3);
}
#[test]
fn test_car_large_blocks() {
let large_data = vec![0x42u8; 1_000_000]; let block = Block::new(Bytes::from(large_data.clone())).unwrap();
let mut car_data = Vec::new();
let mut writer = CarWriter::new(&mut car_data, vec![*block.cid()]).unwrap();
writer.write_block(&block).unwrap();
writer.finish().unwrap();
let mut reader = CarReader::new(&car_data[..]).unwrap();
let read_block = reader.read_block().unwrap().unwrap();
assert_eq!(read_block.cid(), block.cid());
assert_eq!(read_block.data().len(), large_data.len());
}
#[test]
fn test_car_compression_zstd() {
use crate::compression::CompressionAlgorithm;
let block1 = Block::new(Bytes::from(vec![0x42u8; 1000])).unwrap();
let block2 = Block::new(Bytes::from(vec![0xAAu8; 2000])).unwrap();
let mut car_data = Vec::new();
let mut writer = CarWriterBuilder::new(vec![*block1.cid()])
.with_compression(CompressionAlgorithm::Zstd, 3)
.build(&mut car_data)
.unwrap();
writer.write_block(&block1).unwrap();
writer.write_block(&block2).unwrap();
let stats = writer.stats();
assert_eq!(stats.blocks_processed, 2);
assert_eq!(stats.blocks_compressed, 2);
assert_eq!(stats.uncompressed_bytes, 3000);
assert!(stats.compressed_bytes < stats.uncompressed_bytes);
assert!(stats.compression_ratio() < 1.0);
writer.finish().unwrap();
let mut reader = CarReader::new(&car_data[..]).unwrap();
let read_block1 = reader.read_block().unwrap().unwrap();
let read_block2 = reader.read_block().unwrap().unwrap();
assert_eq!(read_block1.cid(), block1.cid());
assert_eq!(read_block1.data(), block1.data());
assert_eq!(read_block2.cid(), block2.cid());
assert_eq!(read_block2.data(), block2.data());
}
#[test]
fn test_car_compression_lz4() {
use crate::compression::CompressionAlgorithm;
let block = Block::new(Bytes::from(vec![0x11u8; 5000])).unwrap();
let mut car_data = Vec::new();
let mut writer = CarWriterBuilder::new(vec![*block.cid()])
.with_compression(CompressionAlgorithm::Lz4, 1)
.build(&mut car_data)
.unwrap();
writer.write_block(&block).unwrap();
let stats = writer.stats();
assert_eq!(stats.blocks_processed, 1);
assert_eq!(stats.blocks_compressed, 1);
assert!(stats.compressed_bytes < stats.uncompressed_bytes);
writer.finish().unwrap();
let mut reader = CarReader::new(&car_data[..]).unwrap();
let read_block = reader.read_block().unwrap().unwrap();
assert_eq!(read_block.cid(), block.cid());
assert_eq!(read_block.data(), block.data());
}
#[test]
fn test_car_compression_none() {
use crate::compression::CompressionAlgorithm;
let block = Block::new(Bytes::from_static(b"test data")).unwrap();
let mut car_data = Vec::new();
let mut writer = CarWriterBuilder::new(vec![*block.cid()])
.with_compression(CompressionAlgorithm::None, 0)
.build(&mut car_data)
.unwrap();
writer.write_block(&block).unwrap();
let stats = writer.stats();
assert_eq!(stats.blocks_processed, 1);
assert_eq!(stats.blocks_compressed, 0); assert_eq!(stats.uncompressed_bytes, stats.compressed_bytes);
assert_eq!(stats.compression_ratio(), 1.0);
writer.finish().unwrap();
let mut reader = CarReader::new(&car_data[..]).unwrap();
let read_block = reader.read_block().unwrap().unwrap();
assert_eq!(read_block.cid(), block.cid());
assert_eq!(read_block.data(), block.data());
}
#[test]
fn test_car_compression_stats() {
use crate::compression::CompressionAlgorithm;
let blocks: Vec<Block> = (0..10)
.map(|_| Block::new(Bytes::from(vec![0x42u8; 500])).unwrap())
.collect();
let mut car_data = Vec::new();
let mut writer = CarWriterBuilder::new(vec![*blocks[0].cid()])
.with_compression(CompressionAlgorithm::Zstd, 5)
.build(&mut car_data)
.unwrap();
for block in &blocks {
writer.write_block(block).unwrap();
}
let stats = writer.stats();
assert_eq!(stats.blocks_processed, 10);
assert_eq!(stats.blocks_compressed, 10);
assert_eq!(stats.uncompressed_bytes, 5000);
assert!(stats.bytes_saved() > 0);
assert!(stats.compression_percentage() > 0.0);
writer.finish().unwrap();
}
#[test]
fn test_car_mixed_compression_backward_compat() {
let block = Block::new(Bytes::from_static(b"legacy data")).unwrap();
let mut car_data = Vec::new();
let mut writer = CarWriter::new(&mut car_data, vec![*block.cid()]).unwrap();
writer.write_block(&block).unwrap();
writer.finish().unwrap();
let mut reader = CarReader::new(&car_data[..]).unwrap();
let read_block = reader.read_block().unwrap().unwrap();
assert_eq!(read_block.cid(), block.cid());
assert_eq!(read_block.data(), block.data());
}
#[test]
fn test_car_compression_large_file() {
use crate::compression::CompressionAlgorithm;
let large_block = Block::new(Bytes::from(vec![0x55u8; 100_000])).unwrap();
let mut car_data = Vec::new();
let mut writer = CarWriterBuilder::new(vec![*large_block.cid()])
.with_compression(CompressionAlgorithm::Zstd, 6)
.build(&mut car_data)
.unwrap();
writer.write_block(&large_block).unwrap();
let stats = writer.stats();
assert_eq!(stats.uncompressed_bytes, 100_000);
assert!(stats.compressed_bytes < 1_000);
assert!(stats.compression_ratio() < 0.01);
assert!(stats.compression_percentage() > 99.0);
writer.finish().unwrap();
let mut reader = CarReader::new(&car_data[..]).unwrap();
let read_block = reader.read_block().unwrap().unwrap();
assert_eq!(read_block.cid(), large_block.cid());
assert_eq!(read_block.data().len(), 100_000);
}
#[test]
fn test_car_builder_without_compression() {
let block = Block::new(Bytes::from_static(b"test")).unwrap();
let mut car_data = Vec::new();
let mut writer = CarWriterBuilder::new(vec![*block.cid()])
.build(&mut car_data)
.unwrap();
writer.write_block(&block).unwrap();
writer.finish().unwrap();
let mut reader = CarReader::new(&car_data[..]).unwrap();
let read_block = reader.read_block().unwrap().unwrap();
assert_eq!(read_block.cid(), block.cid());
assert_eq!(read_block.data(), block.data());
}
}