sif_parser/

packed.rs

// SIF Packed Binary Encoding v1
//
// Hybrid framing: UTF-8 text directives + binary record frames.
// Built from SIF-PACKED-SPEC.md specification.
//
// Frame markers 0x00-0x0F distinguish binary from text (0x23+ for #directives).
// All multi-byte values are little-endian. No alignment padding.

use crate::error::{err, ErrorKind, Result};
use crate::types::*;

// ── Varint / Zigzag (LEB128) ────────────────────────────────────────

/// Encode an unsigned integer as LEB128 varint.
pub fn encode_varint(mut value: u64, out: &mut Vec<u8>) {
    loop {
        let mut byte = (value & 0x7F) as u8;
        value >>= 7;
        if value != 0 {
            byte |= 0x80;
        }
        out.push(byte);
        if value == 0 {
            break;
        }
    }
}

/// Decode an unsigned LEB128 varint from a byte slice.
/// Returns (value, bytes_consumed).
pub fn decode_varint(data: &[u8]) -> Result<(u64, usize)> {
    let mut result: u64 = 0;
    let mut shift = 0u32;
    for (i, &byte) in data.iter().enumerate() {
        if i >= 10 {
            return Err(err(ErrorKind::InvalidRecord, 0, "varint exceeds 10 bytes"));
        }
        result |= ((byte & 0x7F) as u64) << shift;
        if byte & 0x80 == 0 {
            return Ok((result, i + 1));
        }
        shift += 7;
    }
    Err(err(ErrorKind::UnexpectedEof, 0, "unterminated varint"))
}

/// Zigzag encode a signed i64 to unsigned u64.
pub fn zigzag_encode(n: i64) -> u64 {
    ((n << 1) ^ (n >> 63)) as u64
}

/// Zigzag decode an unsigned u64 to signed i64.
pub fn zigzag_decode(n: u64) -> i64 {
    ((n >> 1) as i64) ^ -((n & 1) as i64)
}

// ── Frame Markers ───────────────────────────────────────────────────

/// Frame marker bytes per the spec.
pub const FRAME_INSERT: u8 = 0x00;
pub const FRAME_UPDATE: u8 = 0x01;
pub const FRAME_DELETE: u8 = 0x02;
pub const FRAME_BATCH_INSERT: u8 = 0x03;
pub const FRAME_BATCH_UPDATE: u8 = 0x04;

// ── Type Tags for `any` type ────────────────────────────────────────

const TAG_NULL: u8 = 0x00;
const TAG_BOOL: u8 = 0x01;
const TAG_INT: u8 = 0x02;
const TAG_UINT: u8 = 0x03;
const TAG_FLOAT: u8 = 0x04;
const TAG_STR: u8 = 0x05;
const TAG_DATE: u8 = 0x06;
const TAG_DATETIME: u8 = 0x07;
const TAG_DURATION: u8 = 0x08;
const TAG_BYTES: u8 = 0x09;
const TAG_ARRAY: u8 = 0x0A;
const TAG_MAP: u8 = 0x0B;

// ── Width Modifier Parsing ──────────────────────────────────────────

/// Packed width override from field modifiers.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PackedWidth {
    Default,
    U8,
    U16,
    U32,
    U64,
    I8,
    I16,
    I32,
    I64,
    F32,
    F64,
}

impl PackedWidth {
    /// Parse from a field's modifiers.
    pub fn from_field(field: &FieldDef) -> Self {
        for m in &field.modifiers {
            if m.name == "packed" {
                if let Some(ref v) = m.value {
                    return match v.as_str() {
                        "u8" => PackedWidth::U8,
                        "u16" => PackedWidth::U16,
                        "u32" => PackedWidth::U32,
                        "u64" => PackedWidth::U64,
                        "i8" => PackedWidth::I8,
                        "i16" => PackedWidth::I16,
                        "i32" => PackedWidth::I32,
                        "i64" => PackedWidth::I64,
                        "f32" => PackedWidth::F32,
                        "f64" => PackedWidth::F64,
                        _ => PackedWidth::Default,
                    };
                }
            }
        }
        PackedWidth::Default
    }

    /// Fixed byte size, or None for default (variable-width).
    pub fn fixed_size(&self) -> Option<usize> {
        match self {
            PackedWidth::Default => None,
            PackedWidth::U8 | PackedWidth::I8 => Some(1),
            PackedWidth::U16 | PackedWidth::I16 => Some(2),
            PackedWidth::U32 | PackedWidth::I32 | PackedWidth::F32 => Some(4),
            PackedWidth::U64 | PackedWidth::I64 | PackedWidth::F64 => Some(8),
        }
    }
}

// ── Encoder ─────────────────────────────────────────────────────────

/// Encode a single SIF value to packed binary.
pub fn encode_value(value: &Value, field_type: &Type, width: PackedWidth, out: &mut Vec<u8>) {
    match field_type {
        Type::Nullable(inner) => {
            match value {
                Value::Null => out.push(0x00),
                _ => {
                    out.push(0x01);
                    encode_value(value, inner, width, out);
                }
            }
        }
        Type::Bool => {
            match value {
                Value::Bool(true) => out.push(0x01),
                Value::Bool(false) => out.push(0x00),
                _ => out.push(0x00),
            }
        }
        Type::Uint => encode_uint(value, width, out),
        Type::Int => encode_int(value, width, out),
        Type::Float => encode_float(value, width, out),
        Type::Str => encode_str(value, out),
        Type::Date => encode_date(value, out),
        Type::DateTime => encode_datetime(value, out),
        Type::Duration => encode_duration(value, out),
        Type::Bytes => encode_bytes_value(value, out),
        Type::Enum(variants) => encode_enum(value, variants, width, out),
        Type::Null => {} // zero bytes
        Type::Any => encode_any(value, out),
        Type::Map => encode_map(value, out),
        Type::Array(elem_ty) => encode_array(value, elem_ty, out),
    }
}

fn encode_uint(value: &Value, width: PackedWidth, out: &mut Vec<u8>) {
    let n = match value {
        Value::Uint(n) => *n,
        Value::Int(n) => *n as u64,
        _ => 0,
    };
    match width {
        PackedWidth::U8 => out.push(n as u8),
        PackedWidth::U16 => out.extend_from_slice(&(n as u16).to_le_bytes()),
        PackedWidth::U32 => out.extend_from_slice(&(n as u32).to_le_bytes()),
        PackedWidth::U64 => out.extend_from_slice(&n.to_le_bytes()),
        _ => encode_varint(n, out),
    }
}

fn encode_int(value: &Value, width: PackedWidth, out: &mut Vec<u8>) {
    let n = match value {
        Value::Int(n) => *n,
        Value::Uint(n) => *n as i64,
        _ => 0,
    };
    match width {
        PackedWidth::I8 => out.push(n as u8),
        PackedWidth::I16 => out.extend_from_slice(&(n as i16).to_le_bytes()),
        PackedWidth::I32 => out.extend_from_slice(&(n as i32).to_le_bytes()),
        PackedWidth::I64 => out.extend_from_slice(&n.to_le_bytes()),
        PackedWidth::U8 => out.push(n as u8),
        PackedWidth::U16 => out.extend_from_slice(&(n as u16).to_le_bytes()),
        PackedWidth::U32 => out.extend_from_slice(&(n as u32).to_le_bytes()),
        PackedWidth::U64 => out.extend_from_slice(&(n as u64).to_le_bytes()),
        _ => encode_varint(zigzag_encode(n), out),
    }
}

fn encode_float(value: &Value, width: PackedWidth, out: &mut Vec<u8>) {
    let n = match value {
        Value::Float(n) => *n,
        Value::Int(n) => *n as f64,
        Value::Uint(n) => *n as f64,
        _ => 0.0,
    };
    match width {
        PackedWidth::F32 => out.extend_from_slice(&(n as f32).to_le_bytes()),
        _ => out.extend_from_slice(&n.to_le_bytes()),
    }
}

fn encode_str(value: &Value, out: &mut Vec<u8>) {
    let s = match value {
        Value::Str(s) | Value::Date(s) | Value::DateTime(s)
        | Value::Duration(s) | Value::Enum(s) => s.as_bytes(),
        _ => b"",
    };
    encode_varint(s.len() as u64, out);
    out.extend_from_slice(s);
}

fn encode_date(value: &Value, out: &mut Vec<u8>) {
    // Days since Unix epoch (1970-01-01), i32 LE.
    let days = match value {
        Value::Date(s) => parse_date_to_days(s).unwrap_or(0),
        _ => 0,
    };
    out.extend_from_slice(&days.to_le_bytes());
}

fn encode_datetime(value: &Value, out: &mut Vec<u8>) {
    // Microseconds since Unix epoch, i64 LE.
    let us = match value {
        Value::DateTime(s) => parse_datetime_to_micros(s).unwrap_or(0),
        _ => 0,
    };
    out.extend_from_slice(&us.to_le_bytes());
}

fn encode_duration(value: &Value, out: &mut Vec<u8>) {
    // Microseconds, i64 LE.
    let us = match value {
        Value::Duration(s) => parse_duration_to_micros(s).unwrap_or(0),
        _ => 0,
    };
    out.extend_from_slice(&us.to_le_bytes());
}

fn encode_bytes_value(value: &Value, out: &mut Vec<u8>) {
    let data = match value {
        Value::Bytes(b) => b.as_slice(),
        _ => &[],
    };
    encode_varint(data.len() as u64, out);
    out.extend_from_slice(data);
}

fn encode_enum(value: &Value, variants: &[String], width: PackedWidth, out: &mut Vec<u8>) {
    let idx = match value {
        Value::Enum(s) => variants.iter().position(|v| v == s).unwrap_or(0) as u64,
        _ => 0,
    };
    match width {
        PackedWidth::U8 => out.push(idx as u8),
        _ => encode_varint(idx, out),
    }
}

fn encode_any(value: &Value, out: &mut Vec<u8>) {
    match value {
        Value::Null => out.push(TAG_NULL),
        Value::Bool(b) => {
            out.push(TAG_BOOL);
            out.push(if *b { 0x01 } else { 0x00 });
        }
        Value::Int(n) => {
            out.push(TAG_INT);
            encode_varint(zigzag_encode(*n), out);
        }
        Value::Uint(n) => {
            out.push(TAG_UINT);
            encode_varint(*n, out);
        }
        Value::Float(n) => {
            out.push(TAG_FLOAT);
            out.extend_from_slice(&n.to_le_bytes());
        }
        Value::Str(s) => {
            out.push(TAG_STR);
            encode_varint(s.len() as u64, out);
            out.extend_from_slice(s.as_bytes());
        }
        Value::Date(s) => {
            out.push(TAG_DATE);
            let days = parse_date_to_days(s).unwrap_or(0);
            out.extend_from_slice(&days.to_le_bytes());
        }
        Value::DateTime(s) => {
            out.push(TAG_DATETIME);
            let us = parse_datetime_to_micros(s).unwrap_or(0);
            out.extend_from_slice(&us.to_le_bytes());
        }
        Value::Duration(s) => {
            out.push(TAG_DURATION);
            let us = parse_duration_to_micros(s).unwrap_or(0);
            out.extend_from_slice(&us.to_le_bytes());
        }
        Value::Bytes(b) => {
            out.push(TAG_BYTES);
            encode_varint(b.len() as u64, out);
            out.extend_from_slice(b);
        }
        Value::Array(items) => {
            out.push(TAG_ARRAY);
            encode_varint(items.len() as u64, out);
            for item in items {
                encode_any(item, out);
            }
        }
        Value::Enum(s) => {
            // Encode enums as strings in `any` context (no variant table).
            out.push(TAG_STR);
            encode_varint(s.len() as u64, out);
            out.extend_from_slice(s.as_bytes());
        }
        Value::Map(entries) => {
            out.push(TAG_MAP);
            encode_varint(entries.len() as u64, out);
            for (k, v) in entries {
                encode_varint(k.len() as u64, out);
                out.extend_from_slice(k.as_bytes());
                encode_any(v, out);
            }
        }
    }
}

fn encode_array(value: &Value, elem_ty: &Type, out: &mut Vec<u8>) {
    let items = match value {
        Value::Array(items) => items,
        _ => {
            encode_varint(0, out);
            return;
        }
    };
    encode_varint(items.len() as u64, out);
    for item in items {
        encode_value(item, elem_ty, PackedWidth::Default, out);
    }
}

fn encode_map(value: &Value, out: &mut Vec<u8>) {
    let entries = match value {
        Value::Map(entries) => entries,
        _ => {
            encode_varint(0, out);
            return;
        }
    };
    encode_varint(entries.len() as u64, out);
    for (k, v) in entries {
        encode_varint(k.len() as u64, out);
        out.extend_from_slice(k.as_bytes());
        encode_any(v, out);
    }
}

// ── Decoder ─────────────────────────────────────────────────────────

/// Decode a single SIF value from packed binary.
/// Returns (Value, bytes_consumed).
pub fn decode_value(data: &[u8], field_type: &Type, width: PackedWidth) -> Result<(Value, usize)> {
    match field_type {
        Type::Nullable(inner) => {
            if data.is_empty() {
                return Err(err(ErrorKind::UnexpectedEof, 0, "expected nullable byte"));
            }
            if data[0] == 0x00 {
                Ok((Value::Null, 1))
            } else {
                let (val, n) = decode_value(&data[1..], inner, width)?;
                Ok((val, 1 + n))
            }
        }
        Type::Bool => {
            if data.is_empty() {
                return Err(err(ErrorKind::UnexpectedEof, 0, "expected bool byte"));
            }
            Ok((Value::Bool(data[0] != 0), 1))
        }
        Type::Uint => decode_uint(data, width),
        Type::Int => decode_int(data, width),
        Type::Float => decode_float(data, width),
        Type::Str => decode_str(data),
        Type::Date => decode_date(data),
        Type::DateTime => decode_datetime(data),
        Type::Duration => decode_duration(data),
        Type::Bytes => decode_bytes_value(data),
        Type::Enum(variants) => decode_enum(data, variants, width),
        Type::Null => Ok((Value::Null, 0)),
        Type::Any => decode_any(data),
        Type::Map => decode_map(data),
        Type::Array(elem_ty) => decode_array(data, elem_ty),
    }
}

fn decode_uint(data: &[u8], width: PackedWidth) -> Result<(Value, usize)> {
    match width {
        PackedWidth::U8 => {
            check_len(data, 1)?;
            Ok((Value::Uint(data[0] as u64), 1))
        }
        PackedWidth::U16 => {
            check_len(data, 2)?;
            Ok((Value::Uint(u16::from_le_bytes([data[0], data[1]]) as u64), 2))
        }
        PackedWidth::U32 => {
            check_len(data, 4)?;
            Ok((Value::Uint(u32::from_le_bytes(data[..4].try_into().unwrap()) as u64), 4))
        }
        PackedWidth::U64 => {
            check_len(data, 8)?;
            Ok((Value::Uint(u64::from_le_bytes(data[..8].try_into().unwrap())), 8))
        }
        _ => {
            let (n, consumed) = decode_varint(data)?;
            Ok((Value::Uint(n), consumed))
        }
    }
}

fn decode_int(data: &[u8], width: PackedWidth) -> Result<(Value, usize)> {
    match width {
        PackedWidth::I8 => {
            check_len(data, 1)?;
            Ok((Value::Int(data[0] as i8 as i64), 1))
        }
        PackedWidth::I16 => {
            check_len(data, 2)?;
            Ok((Value::Int(i16::from_le_bytes([data[0], data[1]]) as i64), 2))
        }
        PackedWidth::I32 => {
            check_len(data, 4)?;
            Ok((Value::Int(i32::from_le_bytes(data[..4].try_into().unwrap()) as i64), 4))
        }
        PackedWidth::I64 => {
            check_len(data, 8)?;
            Ok((Value::Int(i64::from_le_bytes(data[..8].try_into().unwrap())), 8))
        }
        PackedWidth::U8 => {
            check_len(data, 1)?;
            Ok((Value::Int(data[0] as i64), 1))
        }
        PackedWidth::U16 => {
            check_len(data, 2)?;
            Ok((Value::Int(u16::from_le_bytes([data[0], data[1]]) as i64), 2))
        }
        PackedWidth::U32 => {
            check_len(data, 4)?;
            Ok((Value::Int(u32::from_le_bytes(data[..4].try_into().unwrap()) as i64), 4))
        }
        PackedWidth::U64 => {
            check_len(data, 8)?;
            Ok((Value::Int(u64::from_le_bytes(data[..8].try_into().unwrap()) as i64), 8))
        }
        _ => {
            let (n, consumed) = decode_varint(data)?;
            Ok((Value::Int(zigzag_decode(n)), consumed))
        }
    }
}

fn decode_float(data: &[u8], width: PackedWidth) -> Result<(Value, usize)> {
    match width {
        PackedWidth::F32 => {
            check_len(data, 4)?;
            let f = f32::from_le_bytes(data[..4].try_into().unwrap());
            Ok((Value::Float(f as f64), 4))
        }
        _ => {
            check_len(data, 8)?;
            let f = f64::from_le_bytes(data[..8].try_into().unwrap());
            Ok((Value::Float(f), 8))
        }
    }
}

fn decode_str(data: &[u8]) -> Result<(Value, usize)> {
    let (len, hdr) = decode_varint(data)?;
    let len = len as usize;
    let total = hdr + len;
    if data.len() < total {
        return Err(err(ErrorKind::UnexpectedEof, 0, "string data truncated"));
    }
    let s = std::str::from_utf8(&data[hdr..total])
        .map_err(|_| err(ErrorKind::InvalidString, 0, "invalid UTF-8 in packed string"))?;
    Ok((Value::Str(s.to_string()), total))
}

fn decode_date(data: &[u8]) -> Result<(Value, usize)> {
    check_len(data, 4)?;
    let days = i32::from_le_bytes(data[..4].try_into().unwrap());
    Ok((Value::Date(days_to_date_string(days)), 4))
}

fn decode_datetime(data: &[u8]) -> Result<(Value, usize)> {
    check_len(data, 8)?;
    let us = i64::from_le_bytes(data[..8].try_into().unwrap());
    Ok((Value::DateTime(micros_to_datetime_string(us)), 8))
}

fn decode_duration(data: &[u8]) -> Result<(Value, usize)> {
    check_len(data, 8)?;
    let us = i64::from_le_bytes(data[..8].try_into().unwrap());
    Ok((Value::Duration(micros_to_duration_string(us)), 8))
}

fn decode_bytes_value(data: &[u8]) -> Result<(Value, usize)> {
    let (len, hdr) = decode_varint(data)?;
    let len = len as usize;
    let total = hdr + len;
    if data.len() < total {
        return Err(err(ErrorKind::UnexpectedEof, 0, "bytes data truncated"));
    }
    Ok((Value::Bytes(data[hdr..total].to_vec()), total))
}

fn decode_enum(data: &[u8], variants: &[String], width: PackedWidth) -> Result<(Value, usize)> {
    let (idx, consumed) = match width {
        PackedWidth::U8 => {
            check_len(data, 1)?;
            (data[0] as u64, 1)
        }
        _ => decode_varint(data)?,
    };
    let variant = variants
        .get(idx as usize)
        .cloned()
        .unwrap_or_else(|| format!("unknown_{}", idx));
    Ok((Value::Enum(variant), consumed))
}

fn decode_any(data: &[u8]) -> Result<(Value, usize)> {
    if data.is_empty() {
        return Err(err(ErrorKind::UnexpectedEof, 0, "expected type tag"));
    }
    let tag = data[0];
    let rest = &data[1..];
    match tag {
        TAG_NULL => Ok((Value::Null, 1)),
        TAG_BOOL => {
            check_len(rest, 1)?;
            Ok((Value::Bool(rest[0] != 0), 2))
        }
        TAG_INT => {
            let (n, c) = decode_varint(rest)?;
            Ok((Value::Int(zigzag_decode(n)), 1 + c))
        }
        TAG_UINT => {
            let (n, c) = decode_varint(rest)?;
            Ok((Value::Uint(n), 1 + c))
        }
        TAG_FLOAT => {
            check_len(rest, 8)?;
            let f = f64::from_le_bytes(rest[..8].try_into().unwrap());
            Ok((Value::Float(f), 9))
        }
        TAG_STR => {
            let (val, c) = decode_str(rest)?;
            Ok((val, 1 + c))
        }
        TAG_DATE => {
            let (val, c) = decode_date(rest)?;
            Ok((val, 1 + c))
        }
        TAG_DATETIME => {
            let (val, c) = decode_datetime(rest)?;
            Ok((val, 1 + c))
        }
        TAG_DURATION => {
            let (val, c) = decode_duration(rest)?;
            Ok((val, 1 + c))
        }
        TAG_BYTES => {
            let (val, c) = decode_bytes_value(rest)?;
            Ok((val, 1 + c))
        }
        TAG_ARRAY => {
            let (count, hdr) = decode_varint(rest)?;
            let mut pos = hdr;
            let mut items = Vec::with_capacity(count as usize);
            for _ in 0..count {
                let (val, c) = decode_any(&rest[pos..])?;
                items.push(val);
                pos += c;
            }
            Ok((Value::Array(items), 1 + pos))
        }
        TAG_MAP => {
            let (count, hdr) = decode_varint(rest)?;
            let mut pos = hdr;
            let mut entries = Vec::with_capacity(count as usize);
            for _ in 0..count {
                let (key_val, kc) = decode_str(&rest[pos..])?;
                let key = match key_val {
                    Value::Str(s) => s,
                    _ => String::new(),
                };
                pos += kc;
                let (val, vc) = decode_any(&rest[pos..])?;
                pos += vc;
                entries.push((key, val));
            }
            Ok((Value::Map(entries), 1 + pos))
        }
        _ => Err(err(ErrorKind::InvalidRecord, 0, format!("unknown type tag: 0x{:02x}", tag))),
    }
}

fn decode_array(data: &[u8], elem_ty: &Type) -> Result<(Value, usize)> {
    let (count, hdr) = decode_varint(data)?;
    let mut pos = hdr;
    let mut items = Vec::with_capacity(count as usize);
    for _ in 0..count {
        let (val, c) = decode_value(&data[pos..], elem_ty, PackedWidth::Default)?;
        items.push(val);
        pos += c;
    }
    Ok((Value::Array(items), pos))
}

fn decode_map(data: &[u8]) -> Result<(Value, usize)> {
    let (count, hdr) = decode_varint(data)?;
    let mut pos = hdr;
    let mut entries = Vec::with_capacity(count as usize);
    for _ in 0..count {
        let (key_val, kc) = decode_str(&data[pos..])?;
        let key = match key_val {
            Value::Str(s) => s,
            _ => String::new(),
        };
        pos += kc;
        let (val, vc) = decode_any(&data[pos..])?;
        pos += vc;
        entries.push((key, val));
    }
    Ok((Value::Map(entries), pos))
}

fn check_len(data: &[u8], need: usize) -> Result<()> {
    if data.len() < need {
        Err(err(ErrorKind::UnexpectedEof, 0, format!("need {} bytes, have {}", need, data.len())))
    } else {
        Ok(())
    }
}

// ── Record-Level Encode / Decode ────────────────────────────────────

/// Encode a record to a packed binary frame (marker + length + payload).
pub fn encode_record_frame(record: &Record, schema: &Schema, out: &mut Vec<u8>) {
    let marker = match record.cdc_op {
        CdcOp::Insert => FRAME_INSERT,
        CdcOp::Update => FRAME_UPDATE,
        CdcOp::Delete => FRAME_DELETE,
    };

    let mut payload = Vec::new();

    if record.cdc_op == CdcOp::Delete {
        // Tombstone: only :id fields.
        for (i, field) in schema.fields.iter().enumerate() {
            if field.semantic.as_deref() == Some("id") {
                if let Some(val) = record.values.get(i) {
                    let w = PackedWidth::from_field(field);
                    encode_value(val, &field.field_type, w, &mut payload);
                }
            }
        }
    } else {
        for (i, field) in schema.fields.iter().enumerate() {
            let val = record.values.get(i).unwrap_or(&Value::Null);
            let w = PackedWidth::from_field(field);
            encode_value(val, &field.field_type, w, &mut payload);
        }
    }

    out.push(marker);
    encode_varint(payload.len() as u64, out);
    out.extend_from_slice(&payload);
}

/// Decode a record from a packed binary frame payload.
pub fn decode_record_frame(
    marker: u8,
    payload: &[u8],
    schema: &Schema,
) -> Result<Record> {
    let cdc_op = match marker {
        FRAME_INSERT => CdcOp::Insert,
        FRAME_UPDATE => CdcOp::Update,
        FRAME_DELETE => CdcOp::Delete,
        _ => return Err(err(ErrorKind::InvalidRecord, 0, format!("unknown frame marker: 0x{:02x}", marker))),
    };

    let mut values = Vec::with_capacity(schema.fields.len());
    let mut pos = 0;

    if cdc_op == CdcOp::Delete {
        for field in &schema.fields {
            if field.semantic.as_deref() == Some("id") {
                let w = PackedWidth::from_field(field);
                let (val, consumed) = decode_value(&payload[pos..], &field.field_type, w)?;
                values.push(val);
                pos += consumed;
            } else {
                values.push(Value::Null);
            }
        }
    } else {
        for field in &schema.fields {
            let w = PackedWidth::from_field(field);
            let (val, consumed) = decode_value(&payload[pos..], &field.field_type, w)?;
            values.push(val);
            pos += consumed;
        }
    }

    Ok(Record { values, cdc_op })
}

/// Encode a batch of records into a batch frame.
pub fn encode_batch_frame(records: &[Record], schema: &Schema, cdc_op: CdcOp, out: &mut Vec<u8>) {
    let marker = match cdc_op {
        CdcOp::Insert => FRAME_BATCH_INSERT,
        CdcOp::Update => FRAME_BATCH_UPDATE,
        CdcOp::Delete => return, // batch delete not defined
    };

    let mut payload = Vec::new();
    for record in records {
        let mut rec_data = Vec::new();
        for (i, field) in schema.fields.iter().enumerate() {
            let val = record.values.get(i).unwrap_or(&Value::Null);
            let w = PackedWidth::from_field(field);
            encode_value(val, &field.field_type, w, &mut rec_data);
        }
        // Variable-width: length-prefix each record.
        encode_varint(rec_data.len() as u64, &mut payload);
        payload.extend_from_slice(&rec_data);
    }

    out.push(marker);
    encode_varint(records.len() as u64, out);
    encode_varint(payload.len() as u64, out);
    out.extend_from_slice(&payload);
}

/// Decode a batch frame into records.
pub fn decode_batch_frame(
    payload: &[u8],
    record_count: u64,
    schema: &Schema,
    cdc_op: CdcOp,
) -> Result<Vec<Record>> {
    let mut records = Vec::with_capacity(record_count as usize);
    let mut pos = 0;

    for _ in 0..record_count {
        let (rec_len, hdr) = decode_varint(&payload[pos..])?;
        pos += hdr;
        let rec_end = pos + rec_len as usize;
        if rec_end > payload.len() {
            return Err(err(ErrorKind::UnexpectedEof, 0, "batch record truncated"));
        }

        let mut values = Vec::with_capacity(schema.fields.len());
        let mut rpos = 0;
        let rec_data = &payload[pos..rec_end];
        for field in &schema.fields {
            let w = PackedWidth::from_field(field);
            let (val, consumed) = decode_value(&rec_data[rpos..], &field.field_type, w)?;
            values.push(val);
            rpos += consumed;
        }
        records.push(Record { values, cdc_op });
        pos = rec_end;
    }

    Ok(records)
}

// ── Date/Time Conversion Helpers ────────────────────────────────────

fn parse_date_to_days(s: &str) -> Option<i32> {
    if s.len() < 10 {
        return None;
    }
    let year: i32 = s[0..4].parse().ok()?;
    let month: u32 = s[5..7].parse().ok()?;
    let day: u32 = s[8..10].parse().ok()?;
    Some(civil_to_days(year, month, day))
}

fn parse_datetime_to_micros(s: &str) -> Option<i64> {
    let days = parse_date_to_days(&s[..10])? as i64;
    let mut micros = days * 86_400_000_000;
    if s.len() > 11 {
        let time_part = s[11..].trim_end_matches('Z');
        let parts: Vec<&str> = time_part.split(':').collect();
        if parts.len() >= 2 {
            let h: i64 = parts[0].parse().ok()?;
            let m: i64 = parts[1].parse().ok()?;
            let s_part = if parts.len() >= 3 { parts[2] } else { "0" };
            let (sec, frac) = if let Some(dot) = s_part.find('.') {
                let sec: i64 = s_part[..dot].parse().ok()?;
                let frac_str = &s_part[dot + 1..];
                let frac_us: i64 = match frac_str.len() {
                    1 => frac_str.parse::<i64>().ok()? * 100_000,
                    2 => frac_str.parse::<i64>().ok()? * 10_000,
                    3 => frac_str.parse::<i64>().ok()? * 1_000,
                    4 => frac_str.parse::<i64>().ok()? * 100,
                    5 => frac_str.parse::<i64>().ok()? * 10,
                    6 => frac_str.parse::<i64>().ok()?,
                    _ => frac_str[..6].parse::<i64>().ok()?,
                };
                (sec, frac_us)
            } else {
                (s_part.parse::<i64>().ok()?, 0)
            };
            micros += h * 3_600_000_000 + m * 60_000_000 + sec * 1_000_000 + frac;
        }
    }
    Some(micros)
}

fn parse_duration_to_micros(s: &str) -> Option<i64> {
    if !s.starts_with('P') {
        return None;
    }
    let rest = &s[1..];
    let mut total_us: i64 = 0;
    let mut in_time = false;
    let mut num_buf = String::new();

    for c in rest.chars() {
        if c == 'T' {
            in_time = true;
            continue;
        }
        if c.is_ascii_digit() || c == '.' {
            num_buf.push(c);
            continue;
        }
        let n: f64 = num_buf.parse().unwrap_or(0.0);
        num_buf.clear();
        match (in_time, c) {
            (false, 'D') => total_us += (n * 86_400_000_000.0) as i64,
            (false, 'W') => total_us += (n * 7.0 * 86_400_000_000.0) as i64,
            (true, 'H') => total_us += (n * 3_600_000_000.0) as i64,
            (true, 'M') => total_us += (n * 60_000_000.0) as i64,
            (true, 'S') => total_us += (n * 1_000_000.0) as i64,
            _ => {}
        }
    }
    Some(total_us)
}

/// Convert days since epoch to YYYY-MM-DD.
fn days_to_date_string(days: i32) -> String {
    let (y, m, d) = days_to_civil(days);
    format!("{:04}-{:02}-{:02}", y, m, d)
}

/// Convert microseconds since epoch to ISO 8601 datetime.
fn micros_to_datetime_string(us: i64) -> String {
    let total_secs = us.div_euclid(1_000_000);
    let frac = us.rem_euclid(1_000_000);
    let days = total_secs.div_euclid(86_400) as i32;
    let day_secs = total_secs.rem_euclid(86_400);
    let (y, m, d) = days_to_civil(days);
    let h = day_secs / 3600;
    let min = (day_secs % 3600) / 60;
    let s = day_secs % 60;
    if frac == 0 {
        format!("{:04}-{:02}-{:02}T{:02}:{:02}:{:02}Z", y, m, d, h, min, s)
    } else {
        format!("{:04}-{:02}-{:02}T{:02}:{:02}:{:02}.{:06}Z", y, m, d, h, min, s, frac)
    }
}

/// Convert microseconds to ISO 8601 duration.
fn micros_to_duration_string(us: i64) -> String {
    if us == 0 {
        return "PT0S".to_string();
    }
    let total_secs = us.abs() / 1_000_000;
    let frac = us.abs() % 1_000_000;
    let h = total_secs / 3600;
    let m = (total_secs % 3600) / 60;
    let s = total_secs % 60;

    let mut result = String::from("P");
    if h > 0 || m > 0 || s > 0 || frac > 0 {
        result.push('T');
        if h > 0 {
            result.push_str(&format!("{}H", h));
        }
        if m > 0 {
            result.push_str(&format!("{}M", m));
        }
        if s > 0 || frac > 0 {
            if frac > 0 {
                result.push_str(&format!("{}.{}S", s, frac));
            } else {
                result.push_str(&format!("{}S", s));
            }
        }
    }
    result
}

// Civil date algorithms (Howard Hinnant's algorithm).
fn civil_to_days(y: i32, m: u32, d: u32) -> i32 {
    let y = if m <= 2 { y - 1 } else { y };
    let era = y.div_euclid(400);
    let yoe = y.rem_euclid(400) as u32;
    let m = if m > 2 { m - 3 } else { m + 9 };
    let doy = (153 * m + 2) / 5 + d - 1;
    let doe = yoe * 365 + yoe / 4 - yoe / 100 + doy;
    era * 146097 + doe as i32 - 719468
}

fn days_to_civil(days: i32) -> (i32, u32, u32) {
    let z = days + 719468;
    let era = z.div_euclid(146097);
    let doe = z.rem_euclid(146097) as u32;
    let yoe = (doe - doe / 1460 + doe / 36524 - doe / 146096) / 365;
    let y = yoe as i32 + era * 400;
    let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
    let mp = (5 * doy + 2) / 153;
    let d = doy - (153 * mp + 2) / 5 + 1;
    let m = if mp < 10 { mp + 3 } else { mp - 9 };
    let y = if m <= 2 { y + 1 } else { y };
    (y, m, d)
}

// ── Tests ───────────────────────────────────────────────────────────

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_varint_roundtrip() {
        for &n in &[0u64, 1, 127, 128, 16383, 16384, u32::MAX as u64, u64::MAX] {
            let mut buf = Vec::new();
            encode_varint(n, &mut buf);
            let (decoded, consumed) = decode_varint(&buf).unwrap();
            assert_eq!(decoded, n);
            assert_eq!(consumed, buf.len());
        }
    }

    #[test]
    fn test_zigzag_roundtrip() {
        for &n in &[0i64, -1, 1, -64, 64, i32::MIN as i64, i32::MAX as i64, i64::MIN, i64::MAX] {
            let encoded = zigzag_encode(n);
            let decoded = zigzag_decode(encoded);
            assert_eq!(decoded, n);
        }
    }

    #[test]
    fn test_encode_decode_scalars() {
        let cases: Vec<(Value, Type)> = vec![
            (Value::Bool(true), Type::Bool),
            (Value::Bool(false), Type::Bool),
            (Value::Uint(42), Type::Uint),
            (Value::Uint(0), Type::Uint),
            (Value::Int(-42), Type::Int),
            (Value::Int(0), Type::Int),
            (Value::Float(3.14), Type::Float),
            (Value::Str("hello".to_string()), Type::Str),
            (Value::Str("".to_string()), Type::Str),
            (Value::Null, Type::Null),
        ];

        for (value, ty) in cases {
            let mut buf = Vec::new();
            encode_value(&value, &ty, PackedWidth::Default, &mut buf);
            let (decoded, consumed) = decode_value(&buf, &ty, PackedWidth::Default).unwrap();
            assert_eq!(decoded, value, "failed for type {:?}", ty);
            assert_eq!(consumed, buf.len());
        }
    }

    #[test]
    fn test_nullable_roundtrip() {
        let ty = Type::Nullable(Box::new(Type::Int));

        let mut buf = Vec::new();
        encode_value(&Value::Int(42), &ty, PackedWidth::Default, &mut buf);
        let (val, _) = decode_value(&buf, &ty, PackedWidth::Default).unwrap();
        assert_eq!(val, Value::Int(42));

        let mut buf = Vec::new();
        encode_value(&Value::Null, &ty, PackedWidth::Default, &mut buf);
        let (val, _) = decode_value(&buf, &ty, PackedWidth::Default).unwrap();
        assert_eq!(val, Value::Null);
    }

    #[test]
    fn test_width_modifiers() {
        // u16
        let mut buf = Vec::new();
        encode_uint(&Value::Uint(1000), PackedWidth::U16, &mut buf);
        assert_eq!(buf.len(), 2);
        let (val, _) = decode_uint(&buf, PackedWidth::U16).unwrap();
        assert_eq!(val, Value::Uint(1000));

        // f32
        let mut buf = Vec::new();
        encode_float(&Value::Float(3.14), PackedWidth::F32, &mut buf);
        assert_eq!(buf.len(), 4);
        let (val, _) = decode_float(&buf, PackedWidth::F32).unwrap();
        if let Value::Float(f) = val {
            assert!((f - 3.14).abs() < 0.001);
        } else {
            panic!("expected float");
        }
    }

    #[test]
    fn test_record_frame_roundtrip() {
        let schema = Schema {
            fields: vec![
                FieldDef {
                    name: "id".to_string(),
                    field_type: Type::Uint,
                    semantic: Some("id".to_string()),
                    deprecated: false,
                    modifiers: Vec::new(),
                },
                FieldDef {
                    name: "name".to_string(),
                    field_type: Type::Str,
                    semantic: None,
                    deprecated: false,
                    modifiers: Vec::new(),
                },
                FieldDef {
                    name: "active".to_string(),
                    field_type: Type::Bool,
                    semantic: None,
                    deprecated: false,
                    modifiers: Vec::new(),
                },
            ],
        };

        let record = Record {
            values: vec![
                Value::Uint(42),
                Value::Str("alice".to_string()),
                Value::Bool(true),
            ],
            cdc_op: CdcOp::Insert,
        };

        let mut buf = Vec::new();
        encode_record_frame(&record, &schema, &mut buf);

        // Decode
        assert_eq!(buf[0], FRAME_INSERT);
        let (payload_len, hdr) = decode_varint(&buf[1..]).unwrap();
        let payload = &buf[1 + hdr..1 + hdr + payload_len as usize];
        let decoded = decode_record_frame(FRAME_INSERT, payload, &schema).unwrap();

        assert_eq!(decoded.values, record.values);
        assert_eq!(decoded.cdc_op, CdcOp::Insert);
    }

    #[test]
    fn test_cdc_delete_frame() {
        let schema = Schema {
            fields: vec![
                FieldDef {
                    name: "id".to_string(),
                    field_type: Type::Uint,
                    semantic: Some("id".to_string()),
                    deprecated: false,
                    modifiers: Vec::new(),
                },
                FieldDef {
                    name: "name".to_string(),
                    field_type: Type::Str,
                    semantic: None,
                    deprecated: false,
                    modifiers: Vec::new(),
                },
            ],
        };

        let record = Record {
            values: vec![Value::Uint(5), Value::Null],
            cdc_op: CdcOp::Delete,
        };

        let mut buf = Vec::new();
        encode_record_frame(&record, &schema, &mut buf);
        assert_eq!(buf[0], FRAME_DELETE);

        let (payload_len, hdr) = decode_varint(&buf[1..]).unwrap();
        let payload = &buf[1 + hdr..1 + hdr + payload_len as usize];
        let decoded = decode_record_frame(FRAME_DELETE, payload, &schema).unwrap();

        assert_eq!(decoded.cdc_op, CdcOp::Delete);
        assert_eq!(decoded.values[0], Value::Uint(5));
        assert_eq!(decoded.values[1], Value::Null); // non-id field is null
    }

    #[test]
    fn test_date_roundtrip() {
        let mut buf = Vec::new();
        let val = Value::Date("2026-03-14".to_string());
        encode_value(&val, &Type::Date, PackedWidth::Default, &mut buf);
        assert_eq!(buf.len(), 4);
        let (decoded, _) = decode_value(&buf, &Type::Date, PackedWidth::Default).unwrap();
        assert_eq!(decoded, val);
    }

    #[test]
    fn test_datetime_roundtrip() {
        let mut buf = Vec::new();
        let val = Value::DateTime("2026-03-14T10:30:00Z".to_string());
        encode_value(&val, &Type::DateTime, PackedWidth::Default, &mut buf);
        assert_eq!(buf.len(), 8);
        let (decoded, _) = decode_value(&buf, &Type::DateTime, PackedWidth::Default).unwrap();
        assert_eq!(decoded, val);
    }

    #[test]
    fn test_duration_roundtrip() {
        let mut buf = Vec::new();
        let val = Value::Duration("PT2H30M".to_string());
        encode_value(&val, &Type::Duration, PackedWidth::Default, &mut buf);
        let (decoded, _) = decode_value(&buf, &Type::Duration, PackedWidth::Default).unwrap();
        assert_eq!(decoded, val);
    }

    #[test]
    fn test_enum_roundtrip() {
        let variants = vec!["open".to_string(), "closed".to_string(), "merged".to_string()];
        let ty = Type::Enum(variants);
        let val = Value::Enum("closed".to_string());
        let mut buf = Vec::new();
        encode_value(&val, &ty, PackedWidth::Default, &mut buf);
        let (decoded, _) = decode_value(&buf, &ty, PackedWidth::Default).unwrap();
        assert_eq!(decoded, val);
    }

    #[test]
    fn test_array_roundtrip() {
        let ty = Type::Array(Box::new(Type::Int));
        let val = Value::Array(vec![Value::Int(1), Value::Int(-2), Value::Int(3)]);
        let mut buf = Vec::new();
        encode_value(&val, &ty, PackedWidth::Default, &mut buf);
        let (decoded, _) = decode_value(&buf, &ty, PackedWidth::Default).unwrap();
        assert_eq!(decoded, val);
    }

    #[test]
    fn test_any_roundtrip() {
        let cases: Vec<Value> = vec![
            Value::Null,
            Value::Bool(true),
            Value::Int(-42),
            Value::Uint(100),
            Value::Float(2.718),
            Value::Str("test".to_string()),
        ];
        for val in cases {
            let mut buf = Vec::new();
            encode_value(&val, &Type::Any, PackedWidth::Default, &mut buf);
            let (decoded, _) = decode_value(&buf, &Type::Any, PackedWidth::Default).unwrap();
            assert_eq!(decoded, val, "any roundtrip failed for {:?}", val);
        }
    }

    #[test]
    fn test_batch_frame_roundtrip() {
        let schema = Schema {
            fields: vec![
                FieldDef {
                    name: "id".to_string(),
                    field_type: Type::Uint,
                    semantic: Some("id".to_string()),
                    deprecated: false,
                    modifiers: Vec::new(),
                },
                FieldDef {
                    name: "val".to_string(),
                    field_type: Type::Str,
                    semantic: None,
                    deprecated: false,
                    modifiers: Vec::new(),
                },
            ],
        };

        let records = vec![
            Record {
                values: vec![Value::Uint(1), Value::Str("a".to_string())],
                cdc_op: CdcOp::Insert,
            },
            Record {
                values: vec![Value::Uint(2), Value::Str("b".to_string())],
                cdc_op: CdcOp::Insert,
            },
            Record {
                values: vec![Value::Uint(3), Value::Str("c".to_string())],
                cdc_op: CdcOp::Insert,
            },
        ];

        let mut buf = Vec::new();
        encode_batch_frame(&records, &schema, CdcOp::Insert, &mut buf);

        assert_eq!(buf[0], FRAME_BATCH_INSERT);
        let (count, h1) = decode_varint(&buf[1..]).unwrap();
        assert_eq!(count, 3);
        let (payload_len, h2) = decode_varint(&buf[1 + h1..]).unwrap();
        let payload = &buf[1 + h1 + h2..1 + h1 + h2 + payload_len as usize];

        let decoded = decode_batch_frame(payload, count, &schema, CdcOp::Insert).unwrap();
        assert_eq!(decoded.len(), 3);
        assert_eq!(decoded[0].values[0], Value::Uint(1));
        assert_eq!(decoded[2].values[1], Value::Str("c".to_string()));
    }
}