use crate::types::{Frame, FileHeader, HandleMetadata};
use anyhow::Result;
use std::fs::{File, OpenOptions};
use std::io::{Read, Seek, SeekFrom, Write};
use std::path::{Path, PathBuf};
pub struct SegmentWriter {
file: File,
#[allow(dead_code)]
path: PathBuf,
current_offset: u64,
seg_id: u32,
}
impl SegmentWriter {
pub fn new(store_path: &Path, seg_id: u32) -> Result<Self> {
let path = store_path.join(format!("seg-{:06}.sift", seg_id));
let file = OpenOptions::new()
.create(true)
.append(true)
.open(&path)?;
let current_offset = file.metadata()?.len();
Ok(Self {
file,
path,
current_offset,
seg_id,
})
}
pub fn write_frame(&mut self, frame: &Frame) -> Result<HandleMetadata> {
let start_offset = self.current_offset;
let frame_content_len = frame.frame_size() - 4;
self.file.write_all(&(frame_content_len as u32).to_le_bytes())?;
let header_bytes = frame.header.to_bytes();
let header_crc = crc32fast::hash(&header_bytes);
self.file.write_all(&header_crc.to_le_bytes())?;
self.file.write_all(&header_bytes)?;
let content_crc = crc32fast::hash(&frame.content);
self.file.write_all(&content_crc.to_le_bytes())?;
self.file.write_all(&frame.content)?;
let line_tbl_offset = (self.current_offset + 4 + 4 + 64 + 4 + frame.content.len() as u64) as u32;
self.file.write_all(&frame.line_table)?;
let mut frame_data = Vec::new();
frame_data.extend_from_slice(&header_crc.to_le_bytes());
frame_data.extend_from_slice(&header_bytes);
frame_data.extend_from_slice(&content_crc.to_le_bytes());
frame_data.extend_from_slice(&frame.content);
frame_data.extend_from_slice(&frame.line_table);
let frame_crc = crc32fast::hash(&frame_data);
self.file.write_all(&frame_crc.to_le_bytes())?;
let total_written = frame.frame_size();
let padded_size = frame.padded_frame_size();
let padding_needed = padded_size - total_written;
if padding_needed > 0 {
let padding = vec![0u8; padding_needed];
self.file.write_all(&padding)?;
}
self.file.flush()?;
let metadata = HandleMetadata {
seg_id: self.seg_id,
offset: start_offset,
frame_len: padded_size as u32,
line_tbl_offset,
};
self.current_offset += padded_size as u64;
Ok(metadata)
}
pub fn current_offset(&self) -> u64 {
self.current_offset
}
pub fn seg_id(&self) -> u32 {
self.seg_id
}
}
pub struct SegmentReader {
file: File,
#[allow(dead_code)]
seg_id: u32,
}
impl SegmentReader {
pub fn new(store_path: &Path, seg_id: u32) -> Result<Self> {
let path = store_path.join(format!("seg-{:06}.sift", seg_id));
let file = File::open(path)?;
Ok(Self { file, seg_id })
}
pub fn read_frame(&mut self, metadata: &HandleMetadata) -> Result<Frame> {
self.file.seek(SeekFrom::Start(metadata.offset))?;
let mut buf = [0u8; 4];
self.file.read_exact(&mut buf)?;
let _frame_len = u32::from_le_bytes(buf);
self.file.read_exact(&mut buf)?;
let expected_header_crc = u32::from_le_bytes(buf);
let mut header_buf = [0u8; 64];
self.file.read_exact(&mut header_buf)?;
let actual_header_crc = crc32fast::hash(&header_buf);
if actual_header_crc != expected_header_crc {
anyhow::bail!("Header CRC mismatch");
}
let header = FileHeader::from_bytes(&header_buf)?;
self.file.read_exact(&mut buf)?;
let expected_content_crc = u32::from_le_bytes(buf);
let mut content = vec![0u8; header.content_len as usize];
self.file.read_exact(&mut content)?;
let actual_content_crc = crc32fast::hash(&content);
if actual_content_crc != expected_content_crc {
anyhow::bail!("Content CRC mismatch");
}
let mut line_table = vec![0u8; header.line_tbl_len as usize];
self.file.read_exact(&mut line_table)?;
self.file.read_exact(&mut buf)?;
let expected_frame_crc = u32::from_le_bytes(buf);
let mut frame_data = Vec::new();
frame_data.extend_from_slice(&expected_header_crc.to_le_bytes());
frame_data.extend_from_slice(&header_buf);
frame_data.extend_from_slice(&expected_content_crc.to_le_bytes());
frame_data.extend_from_slice(&content);
frame_data.extend_from_slice(&line_table);
let actual_frame_crc = crc32fast::hash(&frame_data);
if actual_frame_crc != expected_frame_crc {
anyhow::bail!("Frame CRC mismatch");
}
Ok(Frame {
header,
content,
line_table,
})
}
}
pub fn generate_line_table(content: &[u8]) -> Vec<u8> {
let mut positions = Vec::new();
for (i, &byte) in content.iter().enumerate() {
if byte == b'\n' {
positions.push(i as u32);
}
}
let mut line_table = Vec::new();
let mut last_pos = 0u32;
for pos in positions {
let delta = pos - last_pos;
write_varint(&mut line_table, delta as u64);
last_pos = pos;
}
line_table
}
pub fn decode_line_table(line_table: &[u8]) -> Result<Vec<u32>> {
let mut positions = Vec::new();
let mut cursor = 0;
let mut current_pos = 0u32;
while cursor < line_table.len() {
let (delta, bytes_read) = read_varint(&line_table[cursor..])?;
current_pos += delta as u32;
positions.push(current_pos);
cursor += bytes_read;
}
Ok(positions)
}
pub fn byte_to_line(byte_offset: usize, newline_positions: &[u32]) -> u32 {
match newline_positions.binary_search(&(byte_offset as u32)) {
Ok(index) => (index + 1) as u32, Err(index) => (index + 1) as u32, }
}
fn write_varint(buf: &mut Vec<u8>, mut value: u64) {
while value >= 0x80 {
buf.push((value as u8) | 0x80);
value >>= 7;
}
buf.push(value as u8);
}
fn read_varint(data: &[u8]) -> Result<(u64, usize)> {
let mut result = 0u64;
let mut shift = 0;
let mut bytes_read = 0;
for &byte in data {
bytes_read += 1;
result |= ((byte & 0x7F) as u64) << shift;
if (byte & 0x80) == 0 {
return Ok((result, bytes_read));
}
shift += 7;
if shift >= 64 {
anyhow::bail!("Varint too large");
}
}
anyhow::bail!("Incomplete varint");
}