//! Durable projection delta segment binary format.
//!
//! Segment files store immutable mutation batches referenced by projection
//! manifests. The loader treats bytes as untrusted input and validates magic,
//! version, offsets, checksum, reserved bytes, and row bounds before returning
//! decoded sections.

use std::fs;
use std::path::Path;

use crc32fast::Hasher;

use crate::edge_store::EdgeTypeWidth;
use crate::filter_index::PersistedFilterValue;
use crate::projection::normalize::NormalizedMutationBatch;
use crate::safety::{GraphError, GraphResult};
use crate::types::{EdgeTypeId, TraversalDirection};

const MAGIC: &[u8; 8] = b"PGGSEG01";
const V6_VERSION: u32 = 6;
const V7_VERSION: u32 = 7;
const VERSION: u32 = V7_VERSION;
const HEADER_SIZE: usize = 160;
const CHECKSUM_OFFSET: usize = 124;
const RESERVED_OFFSET: usize = 128;
const RESERVED_LEN: usize = 32;
const SECTION_COUNT: usize = 7;
const RESERVED_HEADER_RANGES: [std::ops::Range<usize>; 2] = [60..64, 120..124];

/// Segment file category.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum SegmentKind {
    /// Edge topology, delete, and weight deltas.
    Edge = 1,
    /// Node, resolution, filter, and tenant deltas.
    Node = 2,
}

impl SegmentKind {
    fn from_u8(raw: u8) -> GraphResult<Self> {
        match raw {
            1 => Ok(Self::Edge),
            2 => Ok(Self::Node),
            other => Err(segment_corrupt(format!("unknown segment kind {other}"))),
        }
    }
}

/// Header metadata for one decoded segment.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct SegmentHeader {
    /// Segment file format version.
    pub(crate) version: u32,
    /// Physical width of edge type identifiers in edge sections.
    pub(crate) edge_type_width: EdgeTypeWidth,
    /// Segment category.
    pub(crate) kind: SegmentKind,
    /// Segment compaction level.
    pub(crate) level: u8,
    /// Direction covered by edge sections.
    pub(crate) direction: TraversalDirection,
    /// Inclusive source-node range start.
    pub(crate) source_start: u32,
    /// Exclusive source-node range end.
    pub(crate) source_end: u32,
    /// Highest sync-log row represented by this segment.
    pub(crate) sync_watermark: i64,
    /// CRC32 checksum over the segment bytes with the checksum field zeroed.
    pub(crate) checksum: u32,
}

/// Edge topology row.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct SegmentEdge {
    /// Source node index.
    pub(crate) source: u32,
    /// Target node index.
    pub(crate) target: u32,
    /// Edge type identifier.
    pub(crate) type_id: EdgeTypeId,
    /// Whether this row is a synthetic reverse of the schema edge.
    pub(crate) schema_reversed: bool,
    /// Stable relationship identity when this edge is backed by a source row.
    pub(crate) relationship_id: Option<crate::edge_store::RelationshipId>,
}

/// Weighted edge row.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct SegmentEdgeWeight {
    /// Source node index.
    pub(crate) source: u32,
    /// Target node index.
    pub(crate) target: u32,
    /// Edge type identifier.
    pub(crate) type_id: EdgeTypeId,
    /// Whether this row is a synthetic reverse of the schema edge.
    pub(crate) schema_reversed: bool,
    /// Stable relationship identity for this weighted source row.
    pub(crate) relationship_id: Option<crate::edge_store::RelationshipId>,
    /// Edge weight.
    pub(crate) weight: u32,
}

/// Node active-state row.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct SegmentNodeState {
    /// Node index.
    pub(crate) node_idx: u32,
    /// Whether the node is active after this delta.
    pub(crate) active: bool,
}

/// Resolution-index row.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct SegmentResolution {
    /// Source table OID.
    pub(crate) table_oid: u32,
    /// Hash of the source primary key.
    pub(crate) pk_hash: u64,
    /// Exact source primary key used to verify hashes and rebuild node metadata.
    pub(crate) primary_key: Option<String>,
    /// Node index.
    pub(crate) node_idx: u32,
    /// Whether the resolution entry is removed.
    pub(crate) tombstone: bool,
}

/// Filter-index row.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct SegmentFilterValue {
    /// Node index.
    pub(crate) node_idx: u32,
    /// Registered filter column identifier.
    pub(crate) column_id: u32,
    /// Self-contained typed filter value.
    pub(crate) value: PersistedFilterValue,
    /// Whether the filter value is removed.
    pub(crate) tombstone: bool,
}

/// Tenant-membership row.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct SegmentTenant {
    /// Node index.
    pub(crate) node_idx: u32,
    /// Hash of the tenant identifier.
    pub(crate) tenant_hash: u64,
    /// Exact tenant identifier used to rebuild tenant membership.
    pub(crate) tenant: Option<String>,
    /// Whether the tenant membership is removed.
    pub(crate) tombstone: bool,
}

/// Decoded durable delta segment.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct DeltaSegment {
    /// Segment header.
    pub(crate) header: SegmentHeader,
    /// Edge insert/topology rows.
    pub(crate) edge_inserts: Vec<SegmentEdge>,
    /// Edge delete rows.
    pub(crate) edge_deletes: Vec<SegmentEdge>,
    /// Edge weight rows.
    pub(crate) edge_weights: Vec<SegmentEdgeWeight>,
    /// Node active/tombstone rows.
    pub(crate) node_states: Vec<SegmentNodeState>,
    /// Resolution-index rows.
    pub(crate) resolutions: Vec<SegmentResolution>,
    /// Filter-index rows.
    pub(crate) filters: Vec<SegmentFilterValue>,
    /// Tenant-membership rows.
    pub(crate) tenants: Vec<SegmentTenant>,
}

impl DeltaSegment {
    /// Construct an empty segment with validated header metadata.
    pub(crate) fn new(
        kind: SegmentKind,
        level: u8,
        direction: TraversalDirection,
        source_start: u32,
        source_end: u32,
        sync_watermark: i64,
    ) -> GraphResult<Self> {
        let header = SegmentHeader {
            version: VERSION,
            edge_type_width: EdgeTypeWidth::One,
            kind,
            level,
            direction,
            source_start,
            source_end,
            sync_watermark,
            checksum: 0,
        };
        validate_header_shape(&header)?;
        Ok(Self {
            header,
            edge_inserts: Vec::new(),
            edge_deletes: Vec::new(),
            edge_weights: Vec::new(),
            node_states: Vec::new(),
            resolutions: Vec::new(),
            filters: Vec::new(),
            tenants: Vec::new(),
        })
    }

    /// Build an edge segment from normalized mutation rows.
    ///
    /// # Errors
    ///
    /// Returns [`GraphError::CorruptFile`] if normalized rows are not valid for
    /// the requested source range.
    pub(crate) fn from_normalized_edges(
        batch: &NormalizedMutationBatch,
        level: u8,
        direction: TraversalDirection,
        source_start: u32,
        source_end: u32,
    ) -> GraphResult<Self> {
        let sync_watermark = match batch.rows.iter().map(|row| row.sync_id).max() {
            Some(sync_id) => i64::try_from(sync_id)
                .map_err(|_| GraphError::Internal("sync watermark exceeds i64".into()))?,
            None => 0,
        };
        let mut segment = Self::new(
            SegmentKind::Edge,
            level,
            direction,
            source_start,
            source_end,
            sync_watermark,
        )?;
        for row in &batch.rows {
            if row.direction != direction {
                return Err(segment_corrupt(
                    "normalized row direction mismatches segment",
                ));
            }
            if !row.operation.is_edge() {
                return Err(segment_corrupt(
                    "normalized node row cannot be written to edge segment",
                ));
            }
            let edge = SegmentEdge {
                source: row.source,
                target: row.target,
                type_id: row.type_id,
                schema_reversed: row.schema_reversed,
                relationship_id: row.relationship_id,
            };
            if row.tombstone {
                segment.edge_deletes.push(edge);
            } else {
                segment.edge_inserts.push(edge);
                if let Some(weight) = row.weight {
                    segment.edge_weights.push(SegmentEdgeWeight {
                        source: row.source,
                        target: row.target,
                        type_id: row.type_id,
                        schema_reversed: row.schema_reversed,
                        relationship_id: row.relationship_id,
                        weight,
                    });
                }
            }
        }
        segment.header.edge_type_width = segment.canonical_edge_type_width();
        validate_segment(&segment)?;
        Ok(segment)
    }

    /// Encode this segment to bytes.
    ///
    /// # Errors
    ///
    /// Returns [`GraphError::CorruptFile`] if row bounds or section ownership
    /// are invalid.
    pub(crate) fn to_bytes(&self) -> GraphResult<Vec<u8>> {
        validate_segment(self)?;
        let edge_type_width = self.canonical_edge_type_width();
        let mut header = self.header.clone();
        header.version = VERSION;
        header.edge_type_width = edge_type_width;
        let section_lengths = self.encoded_section_lengths_for_width(edge_type_width)?;
        let encoded_len = section_lengths
            .iter()
            .try_fold(HEADER_SIZE, |total, length| {
                total
                    .checked_add(*length)
                    .ok_or_else(|| segment_corrupt("encoded segment length overflowed"))
            })?;
        let mut sections = EncodedSections::with_capacities(section_lengths);
        encode_edges(
            &mut sections.edge_inserts,
            &self.edge_inserts,
            edge_type_width,
        )?;
        encode_edges(
            &mut sections.edge_deletes,
            &self.edge_deletes,
            edge_type_width,
        )?;
        encode_edge_weights(
            &mut sections.edge_weights,
            &self.edge_weights,
            edge_type_width,
        )?;
        encode_node_states(&mut sections.node_states, &self.node_states);
        encode_resolutions(&mut sections.resolutions, &self.resolutions)?;
        encode_filters(&mut sections.filters, &self.filters)?;
        encode_tenants(&mut sections.tenants, &self.tenants)?;

        let section_bytes = sections.as_slices();
        let counts = self.section_counts()?;
        let mut offsets = [0_u64; SECTION_COUNT];
        let mut cursor = HEADER_SIZE;
        let mut bytes = Vec::with_capacity(encoded_len);
        bytes.resize(HEADER_SIZE, 0);
        for (idx, section) in section_bytes.iter().enumerate() {
            offsets[idx] = cursor as u64;
            bytes.extend_from_slice(section);
            cursor += section.len();
        }
        write_header(&mut bytes, &header, &counts, &offsets, 0);
        let checksum = checksum_segment_bytes(&bytes);
        write_u32_at(&mut bytes, CHECKSUM_OFFSET, checksum);
        Ok(bytes)
    }

    /// Return the exact number of bytes produced by [`Self::to_bytes`].
    ///
    /// This performs the same shape and variable-length field validation as
    /// encoding, but does not allocate serialization buffers.
    ///
    /// # Errors
    ///
    /// Returns [`GraphError::CorruptFile`] when the segment is invalid or its
    /// encoded size cannot be represented by `usize`.
    pub(crate) fn encoded_len(&self) -> GraphResult<usize> {
        validate_segment(self)?;
        self.encoded_section_lengths_for_width(self.canonical_edge_type_width())?
            .into_iter()
            .try_fold(HEADER_SIZE, |total, length| {
                total
                    .checked_add(length)
                    .ok_or_else(|| segment_corrupt("encoded segment length overflowed"))
            })
    }

    /// Write this segment to a file path.
    ///
    /// # Errors
    ///
    /// Returns encoding errors or filesystem errors.
    #[cfg(any(test, feature = "development", feature = "fuzzing"))]
    #[cfg_attr(any(feature = "development", feature = "fuzzing"), allow(dead_code))]
    pub(crate) fn write_to_path(&self, path: &Path) -> GraphResult<()> {
        let bytes = self.to_bytes()?;
        fs::write(path, bytes)
            .map_err(|err| GraphError::Internal(format!("segment write failed: {err}")))
    }

    /// Read and validate a segment file from disk.
    ///
    /// # Errors
    ///
    /// Returns [`GraphError::CorruptFile`] when the segment is malformed.
    #[cfg_attr(all(feature = "fuzzing", not(test)), allow(dead_code))]
    pub(crate) fn read_from_path(path: &Path) -> GraphResult<Self> {
        let bytes = fs::read(path)
            .map_err(|err| GraphError::Internal(format!("segment read failed: {err}")))?;
        Self::from_bytes(&bytes)
    }

    /// Decode and validate a segment from bytes.
    ///
    /// # Errors
    ///
    /// Returns [`GraphError::CorruptFile`] when the segment is malformed.
    pub(crate) fn from_bytes(bytes: &[u8]) -> GraphResult<Self> {
        if bytes.len() < HEADER_SIZE {
            return Err(segment_corrupt("segment is shorter than header"));
        }
        if &bytes[0..8] != MAGIC {
            return Err(segment_corrupt("invalid segment magic"));
        }
        let version = read_u32(bytes, 8)?;
        if !matches!(version, 5 | V6_VERSION | V7_VERSION) {
            return Err(GraphError::IncompatibleVersion(format!(
                "projection segment version {version} is unsupported; expected 5, {V6_VERSION}, or {V7_VERSION}"
            )));
        }
        validate_reserved_header_bytes(bytes)?;
        let edge_type_width = decode_edge_type_width(version, read_u8(bytes, 15)?)?;
        let stored_checksum = read_u32(bytes, CHECKSUM_OFFSET)?;
        if stored_checksum != checksum_segment_bytes(bytes) {
            return Err(segment_corrupt("segment checksum mismatch"));
        }

        let header = SegmentHeader {
            version,
            edge_type_width,
            kind: SegmentKind::from_u8(read_u8(bytes, 12)?)?,
            level: read_u8(bytes, 14)?,
            direction: decode_direction(read_u8(bytes, 13)?)?,
            source_start: read_u32(bytes, 16)?,
            source_end: read_u32(bytes, 20)?,
            sync_watermark: read_i64(bytes, 24)?,
            checksum: stored_checksum,
        };
        validate_header_shape(&header)?;
        let counts = read_counts(bytes)?;
        let offsets = read_offsets(bytes)?;
        let ranges =
            validate_section_ranges(bytes.len(), &counts, &offsets, version, edge_type_width)?;
        let segment = Self {
            edge_inserts: decode_edges(
                section(bytes, ranges[0].clone())?,
                counts[0],
                edge_type_width,
            )?,
            edge_deletes: decode_edges(
                section(bytes, ranges[1].clone())?,
                counts[1],
                edge_type_width,
            )?,
            edge_weights: decode_edge_weights(
                section(bytes, ranges[2].clone())?,
                counts[2],
                edge_type_width,
            )?,
            node_states: decode_node_states(section(bytes, ranges[3].clone())?, counts[3])?,
            resolutions: decode_resolutions(
                section(bytes, ranges[4].clone())?,
                counts[4],
                version,
            )?,
            filters: decode_filters(section(bytes, ranges[5].clone())?, counts[5])?,
            tenants: decode_tenants(section(bytes, ranges[6].clone())?, counts[6], version)?,
            header,
        };
        validate_segment(&segment)?;
        if version == V7_VERSION && edge_type_width != segment.canonical_edge_type_width() {
            return Err(segment_corrupt(
                "segment edge type width is not canonical for its edge types",
            ));
        }
        Ok(segment)
    }

    /// Return a conservative upper bound for heap allocated while decoding.
    ///
    /// The estimate validates the durable layout without allocating decoded
    /// rows. Variable-width sections are charged in full, which safely covers
    /// the strings retained by resolution, filter, and tenant rows.
    pub(crate) fn decoded_heap_upper_bound(bytes: &[u8]) -> GraphResult<usize> {
        if bytes.len() < HEADER_SIZE {
            return Err(segment_corrupt("segment is shorter than header"));
        }
        if &bytes[0..8] != MAGIC {
            return Err(segment_corrupt("invalid segment magic"));
        }
        let version = read_u32(bytes, 8)?;
        if !matches!(version, 5 | V6_VERSION | V7_VERSION) {
            return Err(GraphError::IncompatibleVersion(format!(
                "projection segment version {version} is unsupported; expected 5, {V6_VERSION}, or {V7_VERSION}"
            )));
        }
        validate_reserved_header_bytes(bytes)?;
        let edge_type_width = decode_edge_type_width(version, read_u8(bytes, 15)?)?;
        let stored_checksum = read_u32(bytes, CHECKSUM_OFFSET)?;
        if stored_checksum != checksum_segment_bytes(bytes) {
            return Err(segment_corrupt("segment checksum mismatch"));
        }
        let counts = read_counts(bytes)?;
        let offsets = read_offsets(bytes)?;
        let ranges =
            validate_section_ranges(bytes.len(), &counts, &offsets, version, edge_type_width)?;
        let row_sizes = [
            std::mem::size_of::<SegmentEdge>(),
            std::mem::size_of::<SegmentEdge>(),
            std::mem::size_of::<SegmentEdgeWeight>(),
            std::mem::size_of::<SegmentNodeState>(),
            std::mem::size_of::<SegmentResolution>(),
            std::mem::size_of::<SegmentFilterValue>(),
            std::mem::size_of::<SegmentTenant>(),
        ];
        let fixed =
            counts
                .into_iter()
                .zip(row_sizes)
                .try_fold(0usize, |total, (count, row_size)| {
                    let bytes = usize::try_from(count)
                        .ok()
                        .and_then(|count| count.checked_mul(row_size))
                        .ok_or_else(|| {
                            segment_corrupt("decoded segment heap estimate overflowed")
                        })?;
                    total
                        .checked_add(bytes)
                        .ok_or_else(|| segment_corrupt("decoded segment heap estimate overflowed"))
                })?;
        ranges[4..].iter().try_fold(fixed, |total, range| {
            total
                .checked_add(range.len())
                .ok_or_else(|| segment_corrupt("decoded segment heap estimate overflowed"))
        })
    }

    /// Validate every persisted edge type against the cumulative dictionary.
    pub(crate) fn validate_edge_type_ids(&self, registry_len: usize) -> GraphResult<()> {
        for type_id in self
            .edge_inserts
            .iter()
            .chain(self.edge_deletes.iter())
            .map(|edge| edge.type_id)
            .chain(self.edge_weights.iter().map(|edge| edge.type_id))
        {
            if type_id.get() as usize >= registry_len {
                return Err(segment_corrupt(format!(
                    "edge type identifier {} is outside the cumulative dictionary count {registry_len}",
                    type_id.get()
                )));
            }
        }
        Ok(())
    }

    fn section_counts(&self) -> GraphResult<[u32; SECTION_COUNT]> {
        Ok([
            count_len(self.edge_inserts.len())?,
            count_len(self.edge_deletes.len())?,
            count_len(self.edge_weights.len())?,
            count_len(self.node_states.len())?,
            count_len(self.resolutions.len())?,
            count_len(self.filters.len())?,
            count_len(self.tenants.len())?,
        ])
    }

    fn encoded_section_lengths_for_width(
        &self,
        edge_type_width: EdgeTypeWidth,
    ) -> GraphResult<[usize; SECTION_COUNT]> {
        self.section_counts()?;
        let fixed = |count: usize, row_bytes: usize| {
            count
                .checked_mul(row_bytes)
                .ok_or_else(|| segment_corrupt("encoded segment section length overflowed"))
        };
        let resolutions = self.resolutions.iter().try_fold(0usize, |total, row| {
            let primary_key = row
                .primary_key
                .as_deref()
                .ok_or_else(|| segment_corrupt("resolution primary key is missing"))?;
            u32::try_from(primary_key.len())
                .map_err(|_| segment_corrupt("resolution primary key exceeds u32"))?;
            total
                .checked_add(21)
                .and_then(|value| value.checked_add(primary_key.len()))
                .ok_or_else(|| segment_corrupt("encoded resolution section length overflowed"))
        })?;
        let filters = self.filters.iter().try_fold(0usize, |total, row| {
            let payload = match &row.value {
                PersistedFilterValue::Null => 0,
                PersistedFilterValue::Numeric(_)
                | PersistedFilterValue::Date(_)
                | PersistedFilterValue::Timestamptz(_) => 8,
                PersistedFilterValue::Boolean(_) => 1,
                PersistedFilterValue::Text(value) => {
                    u32::try_from(value.len())
                        .map_err(|_| segment_corrupt("filter text payload exceeds u32"))?;
                    value.len()
                }
                PersistedFilterValue::Uuid(_) => 16,
            };
            total
                .checked_add(14)
                .and_then(|value| value.checked_add(payload))
                .ok_or_else(|| segment_corrupt("encoded filter section length overflowed"))
        })?;
        let tenants = self.tenants.iter().try_fold(0usize, |total, row| {
            let tenant = row
                .tenant
                .as_deref()
                .ok_or_else(|| segment_corrupt("tenant identifier is missing"))?;
            u32::try_from(tenant.len())
                .map_err(|_| segment_corrupt("tenant identifier exceeds u32"))?;
            total
                .checked_add(17)
                .and_then(|value| value.checked_add(tenant.len()))
                .ok_or_else(|| segment_corrupt("encoded tenant section length overflowed"))
        })?;
        let type_width = edge_type_width.bytes();
        Ok([
            fixed(self.edge_inserts.len(), 13 + type_width)?,
            fixed(self.edge_deletes.len(), 13 + type_width)?,
            fixed(self.edge_weights.len(), 17 + type_width)?,
            fixed(self.node_states.len(), 5)?,
            resolutions,
            filters,
            tenants,
        ])
    }

    fn canonical_edge_type_width(&self) -> EdgeTypeWidth {
        let max_id = self
            .edge_inserts
            .iter()
            .chain(self.edge_deletes.iter())
            .map(|edge| edge.type_id)
            .chain(self.edge_weights.iter().map(|edge| edge.type_id))
            .max()
            .unwrap_or(EdgeTypeId::UNTYPED);
        EdgeTypeWidth::select_for_max_id(max_id)
    }
}

/// Return valid seed bytes for projection segment fuzz targets.
#[cfg(any(test, feature = "fuzzing"))]
pub(crate) fn fuzz_seed_bytes(name: &str) -> Option<Vec<u8>> {
    match name.trim() {
        "edge" | "edge_one" | "edge_two" | "edge_four" | "edge_v6" => {
            let type_id = match name.trim() {
                "edge_two" => 255,
                "edge_four" => 65_535,
                _ => 1,
            };
            let mut segment =
                DeltaSegment::new(SegmentKind::Edge, 0, TraversalDirection::Out, 0, 4, 1).ok()?;
            let type_id = EdgeTypeId::try_from(type_id).ok()?;
            segment.edge_inserts.push(SegmentEdge {
                source: 0,
                target: 1,
                type_id,
                schema_reversed: false,
                relationship_id: None,
            });
            segment.edge_deletes.push(SegmentEdge {
                source: 1,
                target: 2,
                type_id,
                schema_reversed: false,
                relationship_id: None,
            });
            segment.edge_weights.push(SegmentEdgeWeight {
                source: 0,
                target: 1,
                type_id,
                relationship_id: None,
                weight: 5,
                schema_reversed: false,
            });
            if name.trim() == "edge_v6" {
                Some(encode_version_6_segment_for_test(&segment))
            } else {
                segment.to_bytes().ok()
            }
        }
        "node" => {
            let mut segment =
                DeltaSegment::new(SegmentKind::Node, 0, TraversalDirection::Any, 0, 4, 1).ok()?;
            segment.node_states.push(SegmentNodeState {
                node_idx: 0,
                active: true,
            });
            segment.resolutions.push(SegmentResolution {
                table_oid: 100,
                pk_hash: crate::resolution_index::ResolutionIndexBuilder::hash_pk("node-0"),
                primary_key: Some("node-0".to_string()),
                node_idx: 0,
                tombstone: false,
            });
            segment.filters.push(SegmentFilterValue {
                node_idx: 0,
                column_id: 7,
                value: PersistedFilterValue::Numeric(9),
                tombstone: false,
            });
            segment.tenants.push(SegmentTenant {
                node_idx: 0,
                tenant_hash: xxhash_rust::xxh3::xxh3_64(b"tenant-0"),
                tenant: Some("tenant-0".to_string()),
                tombstone: false,
            });
            segment.to_bytes().ok()
        }
        _ => None,
    }
}

#[derive(Default)]
struct EncodedSections {
    edge_inserts: Vec<u8>,
    edge_deletes: Vec<u8>,
    edge_weights: Vec<u8>,
    node_states: Vec<u8>,
    resolutions: Vec<u8>,
    filters: Vec<u8>,
    tenants: Vec<u8>,
}

impl EncodedSections {
    fn with_capacities(capacities: [usize; SECTION_COUNT]) -> Self {
        let [edge_inserts, edge_deletes, edge_weights, node_states, resolutions, filters, tenants] =
            capacities;
        Self {
            edge_inserts: Vec::with_capacity(edge_inserts),
            edge_deletes: Vec::with_capacity(edge_deletes),
            edge_weights: Vec::with_capacity(edge_weights),
            node_states: Vec::with_capacity(node_states),
            resolutions: Vec::with_capacity(resolutions),
            filters: Vec::with_capacity(filters),
            tenants: Vec::with_capacity(tenants),
        }
    }

    fn as_slices(&self) -> [&[u8]; SECTION_COUNT] {
        [
            &self.edge_inserts,
            &self.edge_deletes,
            &self.edge_weights,
            &self.node_states,
            &self.resolutions,
            &self.filters,
            &self.tenants,
        ]
    }
}

fn validate_segment(segment: &DeltaSegment) -> GraphResult<()> {
    validate_header_shape(&segment.header)?;
    match segment.header.version {
        5 | V6_VERSION if segment.header.edge_type_width != EdgeTypeWidth::One => {
            return Err(segment_corrupt(
                "legacy segment must use one-byte edge type identifiers",
            ));
        }
        _ => {}
    }
    match segment.header.kind {
        SegmentKind::Edge => {
            if !segment.node_states.is_empty()
                || !segment.resolutions.is_empty()
                || !segment.filters.is_empty()
                || !segment.tenants.is_empty()
            {
                return Err(segment_corrupt("edge segment contains node sections"));
            }
        }
        SegmentKind::Node => {
            if !segment.edge_inserts.is_empty()
                || !segment.edge_deletes.is_empty()
                || !segment.edge_weights.is_empty()
            {
                return Err(segment_corrupt("node segment contains edge sections"));
            }
        }
    }
    for edge in segment
        .edge_inserts
        .iter()
        .chain(segment.edge_deletes.iter())
    {
        validate_edge_bounds(&segment.header, edge)?;
    }
    for weight in &segment.edge_weights {
        validate_edge_bounds(
            &segment.header,
            &SegmentEdge {
                source: weight.source,
                target: weight.target,
                type_id: weight.type_id,
                schema_reversed: weight.schema_reversed,
                relationship_id: None,
            },
        )?;
    }
    for node in &segment.node_states {
        validate_node_bounds(&segment.header, node.node_idx)?;
    }
    for resolution in &segment.resolutions {
        validate_node_bounds(&segment.header, resolution.node_idx)?;
        match resolution.primary_key.as_deref() {
            Some(primary_key)
                if resolution.pk_hash
                    == crate::resolution_index::ResolutionIndexBuilder::hash_pk(primary_key) => {}
            Some(_) => {
                return Err(segment_corrupt(
                    "resolution primary key does not match its hash",
                ));
            }
            None if segment.header.version == 5 => {}
            None => return Err(segment_corrupt("resolution primary key is missing")),
        }
    }
    for filter in &segment.filters {
        validate_node_bounds(&segment.header, filter.node_idx)?;
    }
    for tenant in &segment.tenants {
        validate_node_bounds(&segment.header, tenant.node_idx)?;
        match tenant.tenant.as_deref() {
            Some(value) if tenant.tenant_hash == xxhash_rust::xxh3::xxh3_64(value.as_bytes()) => {}
            Some(_) => {
                return Err(segment_corrupt("tenant identifier does not match its hash"));
            }
            None if segment.header.version == 5 => {}
            None => return Err(segment_corrupt("tenant identifier is missing")),
        }
    }
    Ok(())
}

fn validate_header_shape(header: &SegmentHeader) -> GraphResult<()> {
    if header.source_start > header.source_end {
        return Err(segment_corrupt("source_start must not exceed source_end"));
    }
    if header.sync_watermark < 0 {
        return Err(segment_corrupt("sync_watermark must be nonnegative"));
    }
    Ok(())
}

fn validate_edge_bounds(header: &SegmentHeader, edge: &SegmentEdge) -> GraphResult<()> {
    validate_node_bounds(header, edge.source)?;
    if edge.type_id == EdgeTypeId::SENTINEL {
        return Err(segment_corrupt(
            "edge type identifier uses the logical sentinel",
        ));
    }
    Ok(())
}

fn validate_node_bounds(header: &SegmentHeader, node_idx: u32) -> GraphResult<()> {
    if node_idx < header.source_start || node_idx >= header.source_end {
        return Err(segment_corrupt(format!(
            "node {node_idx} is outside segment range {}..{}",
            header.source_start, header.source_end
        )));
    }
    Ok(())
}

fn validate_reserved_header_bytes(bytes: &[u8]) -> GraphResult<()> {
    for range in RESERVED_HEADER_RANGES {
        if bytes[range].iter().any(|byte| *byte != 0) {
            return Err(segment_corrupt("reserved header bytes must be zero"));
        }
    }
    let version = read_u32(bytes, 8)?;
    if version != V7_VERSION && bytes[15] != 0 {
        return Err(segment_corrupt(
            "legacy segment edge type width byte must be zero",
        ));
    }
    if bytes[RESERVED_OFFSET..RESERVED_OFFSET + RESERVED_LEN]
        .iter()
        .any(|byte| *byte != 0)
    {
        return Err(segment_corrupt("reserved header bytes must be zero"));
    }
    Ok(())
}

fn write_header(
    bytes: &mut [u8],
    header: &SegmentHeader,
    counts: &[u32; SECTION_COUNT],
    offsets: &[u64; SECTION_COUNT],
    checksum: u32,
) {
    bytes[0..8].copy_from_slice(MAGIC);
    write_u32_at(bytes, 8, header.version);
    bytes[12] = header.kind as u8;
    bytes[13] = encode_direction(header.direction);
    bytes[14] = header.level;
    bytes[15] = if header.version == V7_VERSION {
        header.edge_type_width.bytes() as u8
    } else {
        0
    };
    write_u32_at(bytes, 16, header.source_start);
    write_u32_at(bytes, 20, header.source_end);
    write_i64_at(bytes, 24, header.sync_watermark);
    for (idx, count) in counts.iter().enumerate() {
        write_u32_at(bytes, 32 + idx * 4, *count);
    }
    for (idx, offset) in offsets.iter().enumerate() {
        write_u64_at(bytes, 64 + idx * 8, *offset);
    }
    write_u32_at(bytes, CHECKSUM_OFFSET, checksum);
}

fn count_len(len: usize) -> GraphResult<u32> {
    u32::try_from(len).map_err(|_| segment_corrupt("section row count exceeds u32"))
}

fn read_counts(bytes: &[u8]) -> GraphResult<[u32; SECTION_COUNT]> {
    let mut counts = [0_u32; SECTION_COUNT];
    for (idx, count) in counts.iter_mut().enumerate() {
        *count = read_u32(bytes, 32 + idx * 4)?;
    }
    Ok(counts)
}

fn read_offsets(bytes: &[u8]) -> GraphResult<[u64; SECTION_COUNT]> {
    let mut offsets = [0_u64; SECTION_COUNT];
    for (idx, offset) in offsets.iter_mut().enumerate() {
        *offset = read_u64(bytes, 64 + idx * 8)?;
    }
    Ok(offsets)
}

fn validate_section_ranges(
    len: usize,
    counts: &[u32; SECTION_COUNT],
    offsets: &[u64; SECTION_COUNT],
    version: u32,
    edge_type_width: EdgeTypeWidth,
) -> GraphResult<[std::ops::Range<usize>; SECTION_COUNT]> {
    let type_width = edge_type_width.bytes();
    let widths = [
        Some(13_usize + type_width),
        Some(13 + type_width),
        Some(17 + type_width),
        Some(5),
        (version == 5).then_some(17),
        None,
        (version == 5).then_some(13),
    ];
    let mut ranges: [std::ops::Range<usize>; SECTION_COUNT] =
        std::array::from_fn(|_| 0_usize..0_usize);
    let mut previous_end = HEADER_SIZE;
    for idx in 0..SECTION_COUNT {
        let start = usize::try_from(offsets[idx])
            .map_err(|_| segment_corrupt("section offset exceeds usize"))?;
        let end = if idx + 1 < SECTION_COUNT {
            usize::try_from(offsets[idx + 1])
                .map_err(|_| segment_corrupt("section offset exceeds usize"))?
        } else {
            len
        };
        if start != previous_end {
            return Err(segment_corrupt("section offsets are not contiguous"));
        }
        if end < start || end > len {
            return Err(segment_corrupt("section extends past file end"));
        }
        if let Some(width) = widths[idx] {
            let expected_len = usize::try_from(counts[idx])
                .ok()
                .and_then(|count| count.checked_mul(width))
                .ok_or_else(|| segment_corrupt("section byte length overflowed"))?;
            if end - start != expected_len {
                return Err(segment_corrupt(
                    "section byte length does not match row count",
                ));
            }
        }
        ranges[idx] = start..end;
        previous_end = end;
    }
    if previous_end != len {
        return Err(segment_corrupt("trailing bytes after final section"));
    }
    Ok(ranges)
}

fn section(bytes: &[u8], range: std::ops::Range<usize>) -> GraphResult<&[u8]> {
    bytes
        .get(range)
        .ok_or_else(|| segment_corrupt("section range is out of bounds"))
}

fn encode_edges(out: &mut Vec<u8>, rows: &[SegmentEdge], width: EdgeTypeWidth) -> GraphResult<()> {
    for row in rows {
        push_u32(out, row.source);
        push_u32(out, row.target);
        encode_type_id(row.type_id, width, out)?;
        out.push(u8::from(row.schema_reversed));
        push_u32(
            out,
            row.relationship_id
                .unwrap_or(crate::edge_store::NO_RELATIONSHIP_ID),
        );
    }
    Ok(())
}

fn encode_edge_weights(
    out: &mut Vec<u8>,
    rows: &[SegmentEdgeWeight],
    width: EdgeTypeWidth,
) -> GraphResult<()> {
    for row in rows {
        push_u32(out, row.source);
        push_u32(out, row.target);
        encode_type_id(row.type_id, width, out)?;
        out.push(u8::from(row.schema_reversed));
        push_u32(
            out,
            row.relationship_id
                .unwrap_or(crate::edge_store::NO_RELATIONSHIP_ID),
        );
        push_u32(out, row.weight);
    }
    Ok(())
}

fn encode_node_states(out: &mut Vec<u8>, rows: &[SegmentNodeState]) {
    for row in rows {
        push_u32(out, row.node_idx);
        out.push(u8::from(row.active));
    }
}

fn encode_resolutions(out: &mut Vec<u8>, rows: &[SegmentResolution]) -> GraphResult<()> {
    for row in rows {
        push_u32(out, row.table_oid);
        push_u64(out, row.pk_hash);
        push_u32(out, row.node_idx);
        out.push(u8::from(row.tombstone));
        let primary_key = row
            .primary_key
            .as_deref()
            .ok_or_else(|| segment_corrupt("resolution primary key is missing"))?;
        let primary_key_len = u32::try_from(primary_key.len())
            .map_err(|_| segment_corrupt("resolution primary key exceeds u32"))?;
        push_u32(out, primary_key_len);
        out.extend_from_slice(primary_key.as_bytes());
    }
    Ok(())
}

fn encode_filters(out: &mut Vec<u8>, rows: &[SegmentFilterValue]) -> GraphResult<()> {
    for row in rows {
        push_u32(out, row.node_idx);
        push_u32(out, row.column_id);
        out.push(u8::from(row.tombstone));
        let (tag, payload_len) = match &row.value {
            PersistedFilterValue::Null => (0, 0),
            PersistedFilterValue::Numeric(_) => (1, 8),
            PersistedFilterValue::Boolean(_) => (2, 1),
            PersistedFilterValue::Text(value) => (
                3,
                u32::try_from(value.len())
                    .map_err(|_| segment_corrupt("filter text payload exceeds u32"))?,
            ),
            PersistedFilterValue::Date(_) => (4, 8),
            PersistedFilterValue::Timestamptz(_) => (5, 8),
            PersistedFilterValue::Uuid(_) => (6, 16),
        };
        out.push(tag);
        push_u32(out, payload_len);
        match &row.value {
            PersistedFilterValue::Null => {}
            PersistedFilterValue::Numeric(value)
            | PersistedFilterValue::Date(value)
            | PersistedFilterValue::Timestamptz(value) => push_i64(out, *value),
            PersistedFilterValue::Boolean(value) => out.push(u8::from(*value)),
            PersistedFilterValue::Text(value) => out.extend_from_slice(value.as_bytes()),
            PersistedFilterValue::Uuid(value) => out.extend_from_slice(&value.to_le_bytes()),
        }
    }
    Ok(())
}

fn encode_tenants(out: &mut Vec<u8>, rows: &[SegmentTenant]) -> GraphResult<()> {
    for row in rows {
        push_u32(out, row.node_idx);
        push_u64(out, row.tenant_hash);
        out.push(u8::from(row.tombstone));
        let tenant = row
            .tenant
            .as_deref()
            .ok_or_else(|| segment_corrupt("tenant identifier is missing"))?;
        let tenant_len = u32::try_from(tenant.len())
            .map_err(|_| segment_corrupt("tenant identifier exceeds u32"))?;
        push_u32(out, tenant_len);
        out.extend_from_slice(tenant.as_bytes());
    }
    Ok(())
}

fn decode_edges(bytes: &[u8], count: u32, width: EdgeTypeWidth) -> GraphResult<Vec<SegmentEdge>> {
    let mut rows = Vec::with_capacity(count as usize);
    let row_width = 13 + width.bytes();
    for idx in 0..count as usize {
        let offset = idx * row_width;
        let type_offset = offset + 8;
        let schema_offset = type_offset + width.bytes();
        let schema_reversed = read_u8(bytes, schema_offset)?;
        if schema_reversed > 1 {
            return Err(segment_corrupt(format!(
                "schema_reversed flag must be 0 or 1, found {schema_reversed}"
            )));
        }
        let relationship_id = read_u32(bytes, schema_offset + 1)?;
        rows.push(SegmentEdge {
            source: read_u32(bytes, offset)?,
            target: read_u32(bytes, offset + 4)?,
            type_id: decode_type_id(bytes, type_offset, width)?,
            schema_reversed: schema_reversed != 0,
            relationship_id: (relationship_id != crate::edge_store::NO_RELATIONSHIP_ID)
                .then_some(relationship_id),
        });
    }
    Ok(rows)
}

fn decode_edge_weights(
    bytes: &[u8],
    count: u32,
    width: EdgeTypeWidth,
) -> GraphResult<Vec<SegmentEdgeWeight>> {
    let mut rows = Vec::with_capacity(count as usize);
    let row_width = 17 + width.bytes();
    for idx in 0..count as usize {
        let offset = idx * row_width;
        let type_offset = offset + 8;
        let schema_offset = type_offset + width.bytes();
        let schema_reversed = read_u8(bytes, schema_offset)?;
        if schema_reversed > 1 {
            return Err(segment_corrupt(format!(
                "schema_reversed flag must be 0 or 1, found {schema_reversed}"
            )));
        }
        rows.push(SegmentEdgeWeight {
            source: read_u32(bytes, offset)?,
            target: read_u32(bytes, offset + 4)?,
            type_id: decode_type_id(bytes, type_offset, width)?,
            schema_reversed: schema_reversed != 0,
            relationship_id: match read_u32(bytes, schema_offset + 1)? {
                crate::edge_store::NO_RELATIONSHIP_ID => None,
                relationship_id => Some(relationship_id),
            },
            weight: read_u32(bytes, schema_offset + 5)?,
        });
    }
    Ok(rows)
}

fn encode_v6_type_id(type_id: EdgeTypeId) -> GraphResult<u8> {
    type_id
        .to_v6_storage()
        .map_err(|_| segment_corrupt("edge type identifier exceeds v6 storage"))
}

fn encode_type_id(type_id: EdgeTypeId, width: EdgeTypeWidth, out: &mut Vec<u8>) -> GraphResult<()> {
    validate_type_id_width(type_id, width)?;
    match width {
        EdgeTypeWidth::One => out.push(encode_v6_type_id(type_id)?),
        EdgeTypeWidth::Two => {
            let value = u16::try_from(type_id.get()).map_err(|_| {
                segment_corrupt("edge type identifier exceeds two-byte segment storage")
            })?;
            if value == u16::MAX {
                return Err(segment_corrupt(
                    "edge type identifier uses the reserved two-byte sentinel",
                ));
            }
            out.extend_from_slice(&value.to_le_bytes());
        }
        EdgeTypeWidth::Four => out.extend_from_slice(&type_id.get().to_le_bytes()),
    }
    Ok(())
}

fn validate_type_id_width(type_id: EdgeTypeId, width: EdgeTypeWidth) -> GraphResult<()> {
    if type_id == EdgeTypeId::SENTINEL
        || EdgeTypeWidth::select_for_max_id(type_id).bytes() > width.bytes()
    {
        return Err(segment_corrupt(
            "edge type identifier does not fit the declared segment width",
        ));
    }
    Ok(())
}

fn decode_type_id(bytes: &[u8], offset: usize, width: EdgeTypeWidth) -> GraphResult<EdgeTypeId> {
    let raw = match width {
        EdgeTypeWidth::One => u32::from(read_u8(bytes, offset)?),
        EdgeTypeWidth::Two => u32::from(read_u16(bytes, offset)?),
        EdgeTypeWidth::Four => read_u32(bytes, offset)?,
    };
    if raw
        == match width {
            EdgeTypeWidth::One => u8::MAX as u32,
            EdgeTypeWidth::Two => u16::MAX as u32,
            EdgeTypeWidth::Four => u32::MAX,
        }
    {
        return Err(segment_corrupt(
            "edge type identifier uses the reserved physical sentinel",
        ));
    }
    EdgeTypeId::try_from(raw)
        .map_err(|_| segment_corrupt("edge type identifier uses the logical sentinel"))
}

fn decode_edge_type_width(version: u32, raw: u8) -> GraphResult<EdgeTypeWidth> {
    if version != V7_VERSION {
        return Ok(EdgeTypeWidth::One);
    }
    EdgeTypeWidth::from_bytes(u32::from(raw))
        .ok_or_else(|| segment_corrupt(format!("invalid edge type width {raw}")))
}

fn decode_node_states(bytes: &[u8], count: u32) -> GraphResult<Vec<SegmentNodeState>> {
    let mut rows = Vec::with_capacity(count as usize);
    for idx in 0..count as usize {
        let offset = idx * 5;
        rows.push(SegmentNodeState {
            node_idx: read_u32(bytes, offset)?,
            active: decode_bool(read_u8(bytes, offset + 4)?)?,
        });
    }
    Ok(rows)
}

fn decode_resolutions(
    bytes: &[u8],
    count: u32,
    version: u32,
) -> GraphResult<Vec<SegmentResolution>> {
    if version == 5 {
        let mut rows = Vec::with_capacity(count as usize);
        for idx in 0..count as usize {
            let offset = idx * 17;
            rows.push(SegmentResolution {
                table_oid: read_u32(bytes, offset)?,
                pk_hash: read_u64(bytes, offset + 4)?,
                node_idx: read_u32(bytes, offset + 12)?,
                tombstone: decode_bool(read_u8(bytes, offset + 16)?)?,
                primary_key: None,
            });
        }
        return Ok(rows);
    }
    const HEADER_SIZE: usize = 21;
    let count = usize::try_from(count)
        .map_err(|_| segment_corrupt("resolution row count exceeds usize"))?;
    let mut rows = Vec::with_capacity(count);
    let mut offset = 0_usize;
    for _ in 0..count {
        let primary_key_len = usize::try_from(read_u32(bytes, offset + 17)?)
            .map_err(|_| segment_corrupt("resolution primary key length exceeds usize"))?;
        let primary_key_start = offset
            .checked_add(HEADER_SIZE)
            .ok_or_else(|| segment_corrupt("resolution primary key offset overflowed"))?;
        let primary_key_end = primary_key_start
            .checked_add(primary_key_len)
            .ok_or_else(|| segment_corrupt("resolution primary key end overflowed"))?;
        let primary_key = std::str::from_utf8(
            bytes
                .get(primary_key_start..primary_key_end)
                .ok_or_else(|| segment_corrupt("resolution primary key extends past section"))?,
        )
        .map_err(|_| segment_corrupt("resolution primary key is not valid UTF-8"))?
        .to_string();
        rows.push(SegmentResolution {
            table_oid: read_u32(bytes, offset)?,
            pk_hash: read_u64(bytes, offset + 4)?,
            node_idx: read_u32(bytes, offset + 12)?,
            tombstone: decode_bool(read_u8(bytes, offset + 16)?)?,
            primary_key: Some(primary_key),
        });
        offset = primary_key_end;
    }
    if offset != bytes.len() {
        return Err(segment_corrupt("trailing bytes in resolution section"));
    }
    Ok(rows)
}

fn decode_filters(bytes: &[u8], count: u32) -> GraphResult<Vec<SegmentFilterValue>> {
    const FILTER_ROW_HEADER_SIZE: usize = 14;
    let count =
        usize::try_from(count).map_err(|_| segment_corrupt("filter row count exceeds usize"))?;
    if count
        .checked_mul(FILTER_ROW_HEADER_SIZE)
        .is_none_or(|minimum| minimum > bytes.len())
    {
        return Err(segment_corrupt(
            "filter section is too short for its row count",
        ));
    }
    let mut rows = Vec::with_capacity(count);
    let mut offset = 0_usize;
    for _ in 0..count {
        let payload_len = usize::try_from(read_u32(bytes, offset + 10)?)
            .map_err(|_| segment_corrupt("filter payload length exceeds usize"))?;
        let payload_start = offset
            .checked_add(FILTER_ROW_HEADER_SIZE)
            .ok_or_else(|| segment_corrupt("filter payload offset overflowed"))?;
        let payload_end = payload_start
            .checked_add(payload_len)
            .ok_or_else(|| segment_corrupt("filter payload end overflowed"))?;
        let payload = bytes
            .get(payload_start..payload_end)
            .ok_or_else(|| segment_corrupt("filter payload extends past section"))?;
        rows.push(SegmentFilterValue {
            node_idx: read_u32(bytes, offset)?,
            column_id: read_u32(bytes, offset + 4)?,
            tombstone: decode_bool(read_u8(bytes, offset + 8)?)?,
            value: decode_filter_value(read_u8(bytes, offset + 9)?, payload)?,
        });
        offset = payload_end;
    }
    if offset != bytes.len() {
        return Err(segment_corrupt("trailing bytes in filter section"));
    }
    Ok(rows)
}

fn decode_filter_value(tag: u8, payload: &[u8]) -> GraphResult<PersistedFilterValue> {
    match tag {
        0 if payload.is_empty() => Ok(PersistedFilterValue::Null),
        1 if payload.len() == 8 => Ok(PersistedFilterValue::Numeric(read_i64(payload, 0)?)),
        2 if payload.len() == 1 => Ok(PersistedFilterValue::Boolean(decode_bool(payload[0])?)),
        3 => std::str::from_utf8(payload)
            .map(|value| PersistedFilterValue::Text(value.to_string()))
            .map_err(|_| segment_corrupt("filter text payload is not valid UTF-8")),
        4 if payload.len() == 8 => Ok(PersistedFilterValue::Date(read_i64(payload, 0)?)),
        5 if payload.len() == 8 => Ok(PersistedFilterValue::Timestamptz(read_i64(payload, 0)?)),
        6 if payload.len() == 16 => {
            let mut value = [0_u8; 16];
            value.copy_from_slice(payload);
            Ok(PersistedFilterValue::Uuid(u128::from_le_bytes(value)))
        }
        0..=6 => Err(segment_corrupt(
            "filter payload length does not match its value tag",
        )),
        other => Err(segment_corrupt(format!("unknown filter value tag {other}"))),
    }
}

fn decode_tenants(bytes: &[u8], count: u32, version: u32) -> GraphResult<Vec<SegmentTenant>> {
    if version == 5 {
        let mut rows = Vec::with_capacity(count as usize);
        for idx in 0..count as usize {
            let offset = idx * 13;
            rows.push(SegmentTenant {
                node_idx: read_u32(bytes, offset)?,
                tenant_hash: read_u64(bytes, offset + 4)?,
                tombstone: decode_bool(read_u8(bytes, offset + 12)?)?,
                tenant: None,
            });
        }
        return Ok(rows);
    }
    const HEADER_SIZE: usize = 17;
    let count =
        usize::try_from(count).map_err(|_| segment_corrupt("tenant row count exceeds usize"))?;
    let mut rows = Vec::with_capacity(count);
    let mut offset = 0_usize;
    for _ in 0..count {
        let tenant_len = usize::try_from(read_u32(bytes, offset + 13)?)
            .map_err(|_| segment_corrupt("tenant identifier length exceeds usize"))?;
        let tenant_start = offset
            .checked_add(HEADER_SIZE)
            .ok_or_else(|| segment_corrupt("tenant identifier offset overflowed"))?;
        let tenant_end = tenant_start
            .checked_add(tenant_len)
            .ok_or_else(|| segment_corrupt("tenant identifier end overflowed"))?;
        let tenant = std::str::from_utf8(
            bytes
                .get(tenant_start..tenant_end)
                .ok_or_else(|| segment_corrupt("tenant identifier extends past section"))?,
        )
        .map_err(|_| segment_corrupt("tenant identifier is not valid UTF-8"))?
        .to_string();
        rows.push(SegmentTenant {
            node_idx: read_u32(bytes, offset)?,
            tenant_hash: read_u64(bytes, offset + 4)?,
            tombstone: decode_bool(read_u8(bytes, offset + 12)?)?,
            tenant: Some(tenant),
        });
        offset = tenant_end;
    }
    if offset != bytes.len() {
        return Err(segment_corrupt("trailing bytes in tenant section"));
    }
    Ok(rows)
}

fn decode_bool(raw: u8) -> GraphResult<bool> {
    match raw {
        0 => Ok(false),
        1 => Ok(true),
        _ => Err(segment_corrupt("boolean section value must be 0 or 1")),
    }
}

fn encode_direction(direction: TraversalDirection) -> u8 {
    match direction {
        TraversalDirection::Any => 0,
        TraversalDirection::Out => 1,
        TraversalDirection::In => 2,
    }
}

fn decode_direction(raw: u8) -> GraphResult<TraversalDirection> {
    match raw {
        0 => Ok(TraversalDirection::Any),
        1 => Ok(TraversalDirection::Out),
        2 => Ok(TraversalDirection::In),
        _ => Err(segment_corrupt("unknown segment direction")),
    }
}

fn checksum_segment_bytes(bytes: &[u8]) -> u32 {
    let mut hasher = Hasher::new();
    hasher.update(&bytes[..CHECKSUM_OFFSET]);
    hasher.update(&[0; 4]);
    hasher.update(&bytes[CHECKSUM_OFFSET + 4..]);
    hasher.finalize()
}

fn read_u8(bytes: &[u8], offset: usize) -> GraphResult<u8> {
    bytes
        .get(offset)
        .copied()
        .ok_or_else(|| segment_corrupt("unexpected end of segment"))
}

fn read_u16(bytes: &[u8], offset: usize) -> GraphResult<u16> {
    let raw = bytes
        .get(offset..offset + 2)
        .ok_or_else(|| segment_corrupt("u16 read exceeds segment length"))?;
    Ok(u16::from_le_bytes([raw[0], raw[1]]))
}

fn read_u32(bytes: &[u8], offset: usize) -> GraphResult<u32> {
    let raw = bytes
        .get(offset..offset + 4)
        .ok_or_else(|| segment_corrupt("unexpected end of segment"))?;
    let mut value = [0_u8; 4];
    value.copy_from_slice(raw);
    Ok(u32::from_le_bytes(value))
}

fn read_u64(bytes: &[u8], offset: usize) -> GraphResult<u64> {
    let raw = bytes
        .get(offset..offset + 8)
        .ok_or_else(|| segment_corrupt("unexpected end of segment"))?;
    let mut value = [0_u8; 8];
    value.copy_from_slice(raw);
    Ok(u64::from_le_bytes(value))
}

fn read_i64(bytes: &[u8], offset: usize) -> GraphResult<i64> {
    let raw = bytes
        .get(offset..offset + 8)
        .ok_or_else(|| segment_corrupt("unexpected end of segment"))?;
    let mut value = [0_u8; 8];
    value.copy_from_slice(raw);
    Ok(i64::from_le_bytes(value))
}

fn write_u32_at(bytes: &mut [u8], offset: usize, value: u32) {
    bytes[offset..offset + 4].copy_from_slice(&value.to_le_bytes());
}

fn write_i64_at(bytes: &mut [u8], offset: usize, value: i64) {
    bytes[offset..offset + 8].copy_from_slice(&value.to_le_bytes());
}

fn write_u64_at(bytes: &mut [u8], offset: usize, value: u64) {
    bytes[offset..offset + 8].copy_from_slice(&value.to_le_bytes());
}

fn push_u32(bytes: &mut Vec<u8>, value: u32) {
    bytes.extend_from_slice(&value.to_le_bytes());
}

fn push_u64(bytes: &mut Vec<u8>, value: u64) {
    bytes.extend_from_slice(&value.to_le_bytes());
}

fn push_i64(bytes: &mut Vec<u8>, value: i64) {
    bytes.extend_from_slice(&value.to_le_bytes());
}

fn segment_corrupt(reason: impl Into<String>) -> GraphError {
    GraphError::CorruptFile {
        reason: format!("projection segment: {}", reason.into()),
    }
}

#[cfg(test)]
pub(crate) fn encode_version_5_segment_for_test(segment: &DeltaSegment) -> Vec<u8> {
    let mut sections = EncodedSections::default();
    encode_edges(
        &mut sections.edge_inserts,
        &segment.edge_inserts,
        EdgeTypeWidth::One,
    )
    .expect("test segment edge insert IDs fit v6");
    encode_edges(
        &mut sections.edge_deletes,
        &segment.edge_deletes,
        EdgeTypeWidth::One,
    )
    .expect("test segment edge delete IDs fit v6");
    encode_edge_weights(
        &mut sections.edge_weights,
        &segment.edge_weights,
        EdgeTypeWidth::One,
    )
    .expect("test segment edge weight IDs fit v6");
    encode_node_states(&mut sections.node_states, &segment.node_states);
    for row in &segment.resolutions {
        push_u32(&mut sections.resolutions, row.table_oid);
        push_u64(&mut sections.resolutions, row.pk_hash);
        push_u32(&mut sections.resolutions, row.node_idx);
        sections.resolutions.push(u8::from(row.tombstone));
    }
    encode_filters(&mut sections.filters, &segment.filters).expect("v5 filters encode");
    for row in &segment.tenants {
        push_u32(&mut sections.tenants, row.node_idx);
        push_u64(&mut sections.tenants, row.tenant_hash);
        sections.tenants.push(u8::from(row.tombstone));
    }

    let section_bytes = sections.as_slices();
    let counts = segment.section_counts().expect("v5 counts fit");
    let mut offsets = [0_u64; SECTION_COUNT];
    let mut cursor = HEADER_SIZE;
    let mut bytes = vec![0; HEADER_SIZE];
    for (idx, section) in section_bytes.iter().enumerate() {
        offsets[idx] = cursor as u64;
        bytes.extend_from_slice(section);
        cursor += section.len();
    }
    write_header(&mut bytes, &segment.header, &counts, &offsets, 0);
    write_u32_at(&mut bytes, 8, 5);
    bytes[15] = 0;
    let checksum = checksum_segment_bytes(&bytes);
    write_u32_at(&mut bytes, CHECKSUM_OFFSET, checksum);
    bytes
}

#[cfg(any(test, feature = "fuzzing"))]
fn encode_version_6_segment_for_test(segment: &DeltaSegment) -> Vec<u8> {
    let mut legacy = segment.clone();
    legacy.header.version = V6_VERSION;
    legacy.header.edge_type_width = EdgeTypeWidth::One;
    validate_segment(&legacy).expect("v6 test segment is valid");
    let section_lengths = legacy
        .encoded_section_lengths_for_width(EdgeTypeWidth::One)
        .expect("v6 section lengths fit");
    let mut sections = EncodedSections::with_capacities(section_lengths);
    encode_edges(
        &mut sections.edge_inserts,
        &legacy.edge_inserts,
        EdgeTypeWidth::One,
    )
    .expect("v6 inserts encode");
    encode_edges(
        &mut sections.edge_deletes,
        &legacy.edge_deletes,
        EdgeTypeWidth::One,
    )
    .expect("v6 deletes encode");
    encode_edge_weights(
        &mut sections.edge_weights,
        &legacy.edge_weights,
        EdgeTypeWidth::One,
    )
    .expect("v6 weights encode");
    encode_node_states(&mut sections.node_states, &legacy.node_states);
    encode_resolutions(&mut sections.resolutions, &legacy.resolutions)
        .expect("v6 resolutions encode");
    encode_filters(&mut sections.filters, &legacy.filters).expect("v6 filters encode");
    encode_tenants(&mut sections.tenants, &legacy.tenants).expect("v6 tenants encode");
    let counts = legacy.section_counts().expect("v6 counts fit");
    let mut offsets = [0_u64; SECTION_COUNT];
    let mut cursor = HEADER_SIZE;
    let mut bytes = vec![0; HEADER_SIZE];
    for (idx, section) in sections.as_slices().iter().enumerate() {
        offsets[idx] = cursor as u64;
        bytes.extend_from_slice(section);
        cursor += section.len();
    }
    write_header(&mut bytes, &legacy.header, &counts, &offsets, 0);
    write_u32_at(&mut bytes, 8, V6_VERSION);
    bytes[15] = 0;
    let checksum = checksum_segment_bytes(&bytes);
    write_u32_at(&mut bytes, CHECKSUM_OFFSET, checksum);
    bytes
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::projection::normalize::{
        normalize_committed_mutations, CommittedMutation, MutationBufferLimits, MutationOperation,
    };
    use proptest::prelude::*;

    fn edge_segment_with_type(type_id: u32) -> DeltaSegment {
        let mut segment = DeltaSegment::new(SegmentKind::Edge, 0, TraversalDirection::Out, 0, 2, 1)
            .expect("segment constructs");
        segment.edge_inserts.push(SegmentEdge {
            source: 0,
            target: 1,
            type_id: EdgeTypeId::try_from(type_id).expect("logical type is valid"),
            schema_reversed: false,
            relationship_id: Some(1),
        });
        segment
    }

    #[test]
    fn segment_v7_roundtrips_one_two_four_byte_edge_types() {
        for (type_id, expected_width) in [
            (254, EdgeTypeWidth::One),
            (255, EdgeTypeWidth::Two),
            (65_534, EdgeTypeWidth::Two),
            (65_535, EdgeTypeWidth::Four),
        ] {
            let bytes = edge_segment_with_type(type_id)
                .to_bytes()
                .expect("adaptive segment encodes");
            assert_eq!(read_u32(&bytes, 8).expect("version reads"), V7_VERSION);
            assert_eq!(bytes[15] as usize, expected_width.bytes());
            let decoded = DeltaSegment::from_bytes(&bytes).expect("adaptive segment decodes");
            assert_eq!(decoded.edge_inserts[0].type_id.get(), type_id);
            assert_eq!(decoded.header.edge_type_width, expected_width);
        }
    }

    proptest! {
        #[test]
        fn open_type_segment_property_roundtrips_widths_and_rejects_sentinels(
            one_byte_id in 0_u32..=254,
            two_byte_id in 255_u32..=65_534,
            four_byte_id in 65_535_u32..u32::MAX,
        ) {
            for type_id in [one_byte_id, two_byte_id, four_byte_id] {
                let segment = edge_segment_with_type(type_id);
                let bytes = segment.to_bytes()?;
                let decoded = DeltaSegment::from_bytes(&bytes)?;
                prop_assert_eq!(decoded.edge_inserts[0].type_id.get(), type_id);

                let expected_width = EdgeTypeWidth::select_for_max_id(
                    EdgeTypeId::try_from(type_id)
                        .expect("generated IDs exclude the logical sentinel"),
                );
                prop_assert_eq!(decoded.header.edge_type_width, expected_width);

                let section = usize::try_from(read_u64(&bytes, 64)?)
                    .expect("test artifact offsets fit usize");
                let mut sentinel = bytes;
                let physical_offset = section + 8;
                match expected_width {
                    EdgeTypeWidth::One => sentinel[physical_offset] = u8::MAX,
                    EdgeTypeWidth::Two => sentinel[physical_offset..physical_offset + 2]
                        .copy_from_slice(&u16::MAX.to_le_bytes()),
                    EdgeTypeWidth::Four => sentinel[physical_offset..physical_offset + 4]
                        .copy_from_slice(&u32::MAX.to_le_bytes()),
                }
                rewrite_checksum(&mut sentinel);
                prop_assert!(DeltaSegment::from_bytes(&sentinel).is_err());
            }
        }
    }

    #[test]
    fn segment_v6_remains_readable_after_v7_activation() {
        let original = edge_segment_with_type(254);
        let bytes = encode_version_6_segment_for_test(&original);
        assert_eq!(bytes.len(), 174);
        assert_eq!(checksum_segment_bytes(&bytes), 0xace9_e683);
        assert_eq!(&bytes[0..8], MAGIC);
        assert_eq!(&bytes[8..12], &V6_VERSION.to_le_bytes());
        assert_eq!(bytes[15], 0);
        let decoded = DeltaSegment::from_bytes(&bytes).expect("v6 segment remains readable");
        assert_eq!(decoded.header.version, V6_VERSION);
        assert_eq!(decoded.header.edge_type_width, EdgeTypeWidth::One);
        assert_eq!(decoded.edge_inserts, original.edge_inserts);
    }

    #[test]
    fn segment_v7_rejects_invalid_or_noncanonical_edge_type_width() {
        let bytes = edge_segment_with_type(255)
            .to_bytes()
            .expect("two-byte segment encodes");
        for invalid in [0, 3, 8] {
            let mut corrupt = bytes.clone();
            corrupt[15] = invalid;
            rewrite_checksum(&mut corrupt);
            assert!(DeltaSegment::from_bytes(&corrupt).is_err());
        }
        let mut overwide = edge_segment_with_type(254)
            .to_bytes()
            .expect("one-byte segment encodes");
        overwide[15] = 2;
        rewrite_checksum(&mut overwide);
        assert!(DeltaSegment::from_bytes(&overwide).is_err());
    }

    #[test]
    fn segment_v7_rejects_edge_type_sentinel_and_registry_overflow() {
        let mut sentinel = edge_segment_with_type(255)
            .to_bytes()
            .expect("two-byte segment encodes");
        let section =
            usize::try_from(read_u64(&sentinel, 64).expect("offset reads")).expect("offset fits");
        sentinel[section + 8..section + 10].copy_from_slice(&u16::MAX.to_le_bytes());
        rewrite_checksum(&mut sentinel);
        assert!(DeltaSegment::from_bytes(&sentinel).is_err());

        let decoded = DeltaSegment::from_bytes(
            &edge_segment_with_type(255)
                .to_bytes()
                .expect("segment encodes"),
        )
        .expect("segment decodes");
        assert!(decoded.validate_edge_type_ids(255).is_err());
        decoded
            .validate_edge_type_ids(256)
            .expect("dictionary contains type 255");
    }

    #[test]
    fn segment_v7_rejects_misaligned_truncated_or_mismatched_type_sections() {
        let bytes = edge_segment_with_type(255)
            .to_bytes()
            .expect("two-byte segment encodes");
        let mut truncated = bytes.clone();
        truncated.pop();
        rewrite_checksum(&mut truncated);
        assert!(DeltaSegment::from_bytes(&truncated).is_err());

        let mut mismatched = bytes;
        mismatched[15] = 4;
        rewrite_checksum(&mut mismatched);
        assert!(DeltaSegment::from_bytes(&mismatched).is_err());
    }

    #[test]
    fn segment_v7_decode_and_encode_account_selected_width_bytes() {
        for (type_id, width) in [(254, 1usize), (255, 2), (65_535, 4)] {
            let segment = edge_segment_with_type(type_id);
            let bytes = segment.to_bytes().expect("segment encodes");
            assert_eq!(bytes.len(), HEADER_SIZE + 13 + width);
            assert_eq!(segment.encoded_len().expect("length computes"), bytes.len());
            assert!(DeltaSegment::decoded_heap_upper_bound(&bytes).is_ok());
        }
    }

    #[test]
    fn delta_segment_roundtrips_edge_topology_weight_and_delete_sections() {
        let mut segment =
            DeltaSegment::new(SegmentKind::Edge, 0, TraversalDirection::Out, 0, 8, 42)
                .expect("segment constructs");
        segment.edge_inserts.push(SegmentEdge {
            source: 0,
            target: 1,
            type_id: EdgeTypeId::from_v6_storage(2).expect("fixture type ID is valid v6"),
            schema_reversed: false,
            relationship_id: Some(42),
        });
        segment.edge_deletes.push(SegmentEdge {
            source: 2,
            target: 3,
            type_id: EdgeTypeId::from_v6_storage(4).expect("fixture type ID is valid v6"),
            schema_reversed: false,
            relationship_id: None,
        });
        segment.edge_weights.push(SegmentEdgeWeight {
            source: 0,
            target: 1,
            type_id: EdgeTypeId::from_v6_storage(2).expect("fixture type ID is valid v6"),
            relationship_id: Some(42),
            weight: 7,
            schema_reversed: false,
        });

        let decoded = DeltaSegment::from_bytes(&segment.to_bytes().expect("segment encodes"))
            .expect("segment decodes");

        assert_eq!(decoded.header.kind, SegmentKind::Edge);
        assert_eq!(decoded.header.direction, TraversalDirection::Out);
        assert_eq!(decoded.edge_inserts, segment.edge_inserts);
        assert_eq!(decoded.edge_inserts[0].relationship_id, Some(42));
        assert_eq!(decoded.edge_deletes, segment.edge_deletes);
        assert_eq!(decoded.edge_weights, segment.edge_weights);
        assert_eq!(decoded.edge_weights[0].relationship_id, Some(42));
    }

    #[test]
    fn delta_segment_roundtrips_schema_reversed_edge_rows() {
        let mut segment =
            DeltaSegment::new(SegmentKind::Edge, 0, TraversalDirection::Out, 0, 8, 42)
                .expect("segment constructs");
        segment.edge_inserts.push(SegmentEdge {
            source: 1,
            target: 0,
            type_id: EdgeTypeId::from_v6_storage(2).expect("fixture type ID is valid v6"),
            schema_reversed: true,
            relationship_id: None,
        });
        segment.edge_deletes.push(SegmentEdge {
            source: 3,
            target: 2,
            type_id: EdgeTypeId::from_v6_storage(4).expect("fixture type ID is valid v6"),
            schema_reversed: true,
            relationship_id: None,
        });
        segment.edge_weights.push(SegmentEdgeWeight {
            source: 1,
            target: 0,
            type_id: EdgeTypeId::from_v6_storage(2).expect("fixture type ID is valid v6"),
            schema_reversed: true,
            relationship_id: None,
            weight: 17,
        });

        let decoded = DeltaSegment::from_bytes(&segment.to_bytes().expect("segment encodes"))
            .expect("segment decodes");

        assert_eq!(decoded.edge_inserts, segment.edge_inserts);
        assert_eq!(decoded.edge_deletes, segment.edge_deletes);
        assert_eq!(decoded.edge_weights, segment.edge_weights);
        assert!(decoded.edge_inserts[0].schema_reversed);
        assert!(decoded.edge_deletes[0].schema_reversed);
        assert!(decoded.edge_weights[0].schema_reversed);
    }

    #[test]
    fn delta_segment_roundtrips_node_resolution_filter_tenant_sections() {
        let mut segment =
            DeltaSegment::new(SegmentKind::Node, 1, TraversalDirection::Any, 0, 8, 43)
                .expect("segment constructs");
        segment.node_states.push(SegmentNodeState {
            node_idx: 1,
            active: false,
        });
        segment.resolutions.push(SegmentResolution {
            table_oid: 100,
            pk_hash: crate::resolution_index::ResolutionIndexBuilder::hash_pk("node-1"),
            primary_key: Some("node-1".to_string()),
            node_idx: 1,
            tombstone: false,
        });
        segment.filters.push(SegmentFilterValue {
            node_idx: 1,
            column_id: 7,
            value: PersistedFilterValue::Numeric(-9_223_372_036_854_775_000),
            tombstone: true,
        });
        segment.tenants.push(SegmentTenant {
            node_idx: 1,
            tenant_hash: xxhash_rust::xxh3::xxh3_64(b"tenant-1"),
            tenant: Some("tenant-1".to_string()),
            tombstone: false,
        });

        let decoded = DeltaSegment::from_bytes(&segment.to_bytes().expect("segment encodes"))
            .expect("segment decodes");

        assert_eq!(decoded.header.kind, SegmentKind::Node);
        assert_eq!(decoded.header.level, 1);
        assert_eq!(decoded.node_states, segment.node_states);
        assert_eq!(decoded.resolutions, segment.resolutions);
        assert_eq!(decoded.filters, segment.filters);
        assert_eq!(decoded.tenants, segment.tenants);
    }

    #[test]
    fn delta_segment_decodes_version_5_exact_identity_gaps_safely() {
        let mut segment = DeltaSegment::new(SegmentKind::Node, 0, TraversalDirection::Any, 0, 2, 9)
            .expect("segment constructs");
        segment.resolutions.push(SegmentResolution {
            table_oid: 100,
            pk_hash: crate::resolution_index::ResolutionIndexBuilder::hash_pk("legacy"),
            primary_key: Some("legacy".to_string()),
            node_idx: 1,
            tombstone: false,
        });
        segment.filters.push(SegmentFilterValue {
            node_idx: 1,
            column_id: 3,
            value: PersistedFilterValue::Text("variable-width".to_string()),
            tombstone: false,
        });
        segment.tenants.push(SegmentTenant {
            node_idx: 1,
            tenant_hash: xxhash_rust::xxh3::xxh3_64(b"legacy-tenant"),
            tenant: Some("legacy-tenant".to_string()),
            tombstone: false,
        });

        let decoded = DeltaSegment::from_bytes(&encode_version_5_segment_for_test(&segment))
            .expect("version 5 segment decodes");

        assert_eq!(decoded.header.version, 5);
        assert_eq!(decoded.resolutions[0].primary_key, None);
        assert_eq!(decoded.filters, segment.filters);
        assert_eq!(decoded.tenants[0].tenant, None);
    }

    #[test]
    fn delta_segment_rejects_corrupt_v6_exact_identity_hash() {
        let mut segment = DeltaSegment::new(SegmentKind::Node, 0, TraversalDirection::Any, 0, 1, 9)
            .expect("segment constructs");
        segment.resolutions.push(SegmentResolution {
            table_oid: 100,
            pk_hash: crate::resolution_index::ResolutionIndexBuilder::hash_pk("exact"),
            primary_key: Some("exact".to_string()),
            node_idx: 0,
            tombstone: false,
        });
        let mut bytes = segment.to_bytes().expect("segment encodes");
        let resolution_offset =
            usize::try_from(read_u64(&bytes, 96).expect("resolution offset reads"))
                .expect("resolution offset fits usize");
        bytes[resolution_offset + 21] ^= 1;
        rewrite_checksum(&mut bytes);

        let error = DeltaSegment::from_bytes(&bytes).expect_err("identity hash mismatch rejects");

        assert!(matches!(
            error,
            GraphError::CorruptFile { reason } if reason.contains("primary key does not match")
        ));
    }

    #[test]
    fn delta_segment_roundtrips_every_persisted_filter_value_domain() {
        let mut segment =
            DeltaSegment::new(SegmentKind::Node, 0, TraversalDirection::Any, 0, 16, 44)
                .expect("segment constructs");
        let values = [
            PersistedFilterValue::Null,
            PersistedFilterValue::Numeric(i64::MIN + 1),
            PersistedFilterValue::Numeric(i64::MAX),
            PersistedFilterValue::Boolean(false),
            PersistedFilterValue::Boolean(true),
            PersistedFilterValue::Text(String::new()),
            PersistedFilterValue::Text("durable 🧪 filter".to_string()),
            PersistedFilterValue::Date(-20_000),
            PersistedFilterValue::Timestamptz(4_102_444_800_123_456),
            PersistedFilterValue::Uuid(0x1234_5678_90ab_cdef_1234_5678_90ab_cdef),
        ];
        for (idx, value) in values.into_iter().enumerate() {
            segment.filters.push(SegmentFilterValue {
                node_idx: idx as u32,
                column_id: idx as u32,
                value,
                tombstone: idx == 9,
            });
        }

        let bytes = segment.to_bytes().expect("typed filter segment encodes");
        let decoded = DeltaSegment::from_bytes(&bytes).expect("typed filter segment decodes");

        assert_eq!(decoded.header.version, VERSION);
        assert_eq!(decoded.filters, segment.filters);
    }

    #[test]
    fn delta_segment_rejects_invalid_filter_value_tags_and_lengths() {
        let mut segment =
            DeltaSegment::new(SegmentKind::Node, 0, TraversalDirection::Any, 0, 1, 45)
                .expect("segment constructs");
        segment.filters.push(SegmentFilterValue {
            node_idx: 0,
            column_id: 0,
            value: PersistedFilterValue::Numeric(-1),
            tombstone: false,
        });
        let bytes = segment.to_bytes().expect("filter segment encodes");
        let filter_offset = usize::try_from(read_u64(&bytes, 104).expect("filter offset reads"))
            .expect("filter offset fits usize");

        let mut bad_tag = bytes.clone();
        bad_tag[filter_offset + 9] = 255;
        rewrite_checksum(&mut bad_tag);
        assert!(matches!(
            DeltaSegment::from_bytes(&bad_tag),
            Err(GraphError::CorruptFile { reason }) if reason.contains("unknown filter value tag")
        ));

        let mut bad_length = bytes;
        write_u32_at(&mut bad_length, filter_offset + 10, 7);
        rewrite_checksum(&mut bad_length);
        assert!(matches!(
            DeltaSegment::from_bytes(&bad_length),
            Err(GraphError::CorruptFile { reason }) if reason.contains("payload length") || reason.contains("trailing bytes")
        ));
    }

    #[test]
    fn delta_segment_rejects_pre_typed_filter_format_version() {
        let segment = DeltaSegment::new(SegmentKind::Node, 0, TraversalDirection::Any, 0, 0, 0)
            .expect("segment constructs");
        let mut bytes = segment.to_bytes().expect("segment encodes");
        write_u32_at(&mut bytes, 8, 4);
        rewrite_checksum(&mut bytes);

        assert!(matches!(
            DeltaSegment::from_bytes(&bytes),
            Err(GraphError::IncompatibleVersion(message)) if message.contains("rebuild") || message.contains("unsupported")
        ));
    }

    #[test]
    fn delta_segment_rejects_corrupt_offsets_checksum_and_reserved_flags() {
        let mut segment =
            DeltaSegment::new(SegmentKind::Edge, 0, TraversalDirection::Out, 0, 8, 42)
                .expect("segment constructs");
        segment.edge_inserts.push(SegmentEdge {
            source: 0,
            target: 1,
            type_id: EdgeTypeId::from_v6_storage(2).expect("fixture type ID is valid v6"),
            schema_reversed: false,
            relationship_id: None,
        });
        let bytes = segment.to_bytes().expect("segment encodes");

        let mut bad_offset = bytes.clone();
        write_u64_at(&mut bad_offset, 64, (HEADER_SIZE + 1) as u64);
        rewrite_checksum(&mut bad_offset);
        let offset_err = DeltaSegment::from_bytes(&bad_offset).expect_err("bad offset rejects");

        let mut bad_checksum = bytes.clone();
        bad_checksum[CHECKSUM_OFFSET] ^= 0xff;
        let checksum_err =
            DeltaSegment::from_bytes(&bad_checksum).expect_err("bad checksum rejects");

        let mut bad_reserved = bytes;
        bad_reserved[RESERVED_OFFSET] = 1;
        rewrite_checksum(&mut bad_reserved);
        let reserved_err =
            DeltaSegment::from_bytes(&bad_reserved).expect_err("bad reserved flags reject");

        assert!(matches!(offset_err, GraphError::CorruptFile { .. }));
        assert!(matches!(checksum_err, GraphError::CorruptFile { .. }));
        assert!(matches!(reserved_err, GraphError::CorruptFile { .. }));

        for offset in [60, 120] {
            let mut bad_padding = segment.to_bytes().expect("segment encodes");
            bad_padding[offset] = 1;
            rewrite_checksum(&mut bad_padding);
            let padding_err =
                DeltaSegment::from_bytes(&bad_padding).expect_err("bad padding rejects");
            assert!(matches!(padding_err, GraphError::CorruptFile { .. }));
        }

        let mut bad_width = segment.to_bytes().expect("segment encodes");
        bad_width[15] = 3;
        rewrite_checksum(&mut bad_width);
        assert!(matches!(
            DeltaSegment::from_bytes(&bad_width),
            Err(GraphError::CorruptFile { reason }) if reason.contains("edge type width")
        ));
    }

    #[test]
    fn delta_segment_allows_targets_outside_source_range() {
        let mut segment =
            DeltaSegment::new(SegmentKind::Edge, 0, TraversalDirection::Out, 0, 2, 42)
                .expect("segment constructs");
        segment.edge_inserts.push(SegmentEdge {
            source: 0,
            target: 99,
            type_id: EdgeTypeId::from_v6_storage(2).expect("fixture type ID is valid v6"),
            schema_reversed: false,
            relationship_id: None,
        });

        let decoded = DeltaSegment::from_bytes(&segment.to_bytes().expect("segment encodes"))
            .expect("segment decodes");

        assert_eq!(decoded.edge_inserts[0].target, 99);
    }

    #[test]
    fn delta_segment_writer_accepts_normalized_edge_rows() {
        let rows = vec![
            committed(1, 0, 1, Some(7), MutationOperation::InsertEdge),
            committed(2, 2, 3, None, MutationOperation::DeleteEdge),
        ];
        let batch = normalize_committed_mutations(&rows, MutationBufferLimits::new(10, 10_000))
            .expect("rows normalize");

        let segment = DeltaSegment::from_normalized_edges(&batch, 0, TraversalDirection::Out, 0, 4)
            .expect("segment builds from normalized rows");
        let decoded = DeltaSegment::from_bytes(&segment.to_bytes().expect("segment encodes"))
            .expect("segment decodes");

        assert_eq!(decoded.edge_inserts.len(), 1);
        assert_eq!(decoded.edge_weights.len(), 1);
        assert_eq!(decoded.edge_deletes.len(), 1);
        assert_eq!(decoded.header.sync_watermark, 2);
    }

    #[test]
    fn delta_segment_writer_rejects_normalized_node_rows() {
        let rows = vec![committed(1, 0, 0, None, MutationOperation::UpsertNode)];
        let batch = normalize_committed_mutations(&rows, MutationBufferLimits::new(10, 10_000))
            .expect("rows normalize");

        let err = DeltaSegment::from_normalized_edges(&batch, 0, TraversalDirection::Out, 0, 4)
            .expect_err("node row cannot be written to edge segment");

        assert!(matches!(err, GraphError::CorruptFile { .. }));
    }

    #[test]
    fn delta_segment_rejects_source_out_of_range() {
        let mut segment =
            DeltaSegment::new(SegmentKind::Edge, 0, TraversalDirection::Out, 0, 2, 42)
                .expect("segment constructs");
        segment.edge_inserts.push(SegmentEdge {
            source: 2,
            target: 0,
            type_id: EdgeTypeId::from_v6_storage(2).expect("fixture type ID is valid v6"),
            schema_reversed: false,
            relationship_id: None,
        });

        let err = segment
            .to_bytes()
            .expect_err("source outside source range rejects");

        assert!(matches!(err, GraphError::CorruptFile { .. }));
    }

    #[test]
    fn encoded_len_matches_variable_section_serialization() {
        let mut segment = DeltaSegment::new(SegmentKind::Node, 0, TraversalDirection::Any, 0, 2, 1)
            .expect("segment constructs");
        segment.resolutions.push(SegmentResolution {
            table_oid: 1,
            pk_hash: crate::resolution_index::ResolutionIndexBuilder::hash_pk(
                "primary-key-with-variable-length",
            ),
            primary_key: Some("primary-key-with-variable-length".to_string()),
            node_idx: 0,
            tombstone: false,
        });
        segment.filters.push(SegmentFilterValue {
            node_idx: 0,
            column_id: 3,
            value: PersistedFilterValue::Text("variable filter payload".to_string()),
            tombstone: false,
        });
        segment.tenants.push(SegmentTenant {
            node_idx: 0,
            tenant_hash: xxhash_rust::xxh3::xxh3_64(b"variable-tenant"),
            tenant: Some("variable-tenant".to_string()),
            tombstone: false,
        });

        let expected = segment.encoded_len().expect("encoded length preflights");
        let bytes = segment.to_bytes().expect("segment serializes");

        assert_eq!(bytes.len(), expected);
    }

    #[test]
    fn delta_segment_loader_never_panics_on_arbitrary_bytes() {
        let cases: &[&[u8]] = &[
            b"",
            b"PGGSEG01",
            &[0xff; HEADER_SIZE],
            b"not a segment but definitely bytes",
        ];

        for case in cases {
            assert!(DeltaSegment::from_bytes(case).is_err());
        }
    }

    fn rewrite_checksum(bytes: &mut [u8]) {
        write_u32_at(bytes, CHECKSUM_OFFSET, 0);
        let checksum = checksum_segment_bytes(bytes);
        write_u32_at(bytes, CHECKSUM_OFFSET, checksum);
    }

    fn committed(
        sync_id: u64,
        source: u32,
        target: u32,
        weight: Option<u32>,
        operation: MutationOperation,
    ) -> CommittedMutation {
        CommittedMutation {
            sync_id,
            generation_id: 1,
            direction: TraversalDirection::Out,
            source,
            target,
            type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
            weight,
            relationship_id: None,
            operation,
            schema_reversed: false,
        }
    }
}
