weirflow 0.1.0

//! Relation interchange rows for IFDS fixtures and differential evidence.
//!
//! The schema is deliberately byte-oriented and dataflow-generic: it records
//! source rows, sink rows, graph edges, call-string rows, fact domains, and
//! output witnesses without depending on a downstream reporter.

use std::collections::BTreeSet;
use std::error::Error;
use std::fmt::{Display, Formatter};

/// Relation interchange schema version.
pub const RELATION_INTERCHANGE_SCHEMA_VERSION: u32 = 1;

const RELATION_INTERCHANGE_MAGIC: [u8; 4] = *b"WRIF";

/// Source or sink relation row.
#[derive(Debug, Clone, Copy, Eq, PartialEq, Ord, PartialOrd)]
pub struct RelationEndpoint {
    /// Stable endpoint tuple id.
    pub id: u32,
    /// Statement or exploded-node id.
    pub node_id: u32,
    /// Procedure id.
    pub proc_id: u32,
    /// Block id within the procedure.
    pub block_id: u32,
    /// Dataflow fact id.
    pub fact_id: u32,
    /// Source file id.
    pub file_id: u32,
    /// Source byte start.
    pub start_byte: u32,
    /// Source byte end.
    pub end_byte: u32,
}

/// Graph or summary edge relation row.
#[derive(Debug, Clone, Copy, Eq, PartialEq, Ord, PartialOrd)]
pub struct RelationEdge {
    /// Stable edge tuple id.
    pub id: u32,
    /// Source statement or exploded-node id.
    pub src_node: u32,
    /// Destination statement or exploded-node id.
    pub dst_node: u32,
    /// Edge-kind bit mask.
    pub edge_kind: u32,
}

/// Call-string relation row.
#[derive(Debug, Clone, Copy, Eq, PartialEq, Ord, PartialOrd)]
pub struct RelationCallString {
    /// Stable call-string tuple id.
    pub id: u32,
    /// Caller node id.
    pub caller_node: u32,
    /// Callee entry node id.
    pub callee_node: u32,
    /// Return-site node id.
    pub return_node: u32,
    /// Call-string stack depth represented by this row.
    pub depth: u32,
}

/// Fact-domain relation row.
#[derive(Debug, Clone, Copy, Eq, PartialEq, Ord, PartialOrd)]
pub struct RelationFact {
    /// Stable fact tuple id.
    pub id: u32,
    /// Domain id, such as taint, alias, or lifetime.
    pub domain_id: u32,
    /// Stable producer-local fact tag.
    pub tag: u32,
}

/// Output witness relation row.
#[derive(Debug, Clone, Eq, PartialEq)]
pub struct RelationOutputWitness {
    /// Source endpoint tuple id.
    pub source_id: u32,
    /// Sink endpoint tuple id.
    pub sink_id: u32,
    /// Ordered statement or exploded-node path.
    pub path_nodes: Vec<u32>,
}

/// Owned relation interchange fixture.
#[derive(Debug, Clone, Eq, PartialEq, Default)]
pub struct RelationInterchange {
    /// Source relation rows.
    pub sources: Vec<RelationEndpoint>,
    /// Sink relation rows.
    pub sinks: Vec<RelationEndpoint>,
    /// Graph and summary edge rows.
    pub edges: Vec<RelationEdge>,
    /// Call-string rows.
    pub call_strings: Vec<RelationCallString>,
    /// Fact-domain rows.
    pub facts: Vec<RelationFact>,
    /// Output witness rows.
    pub output_witnesses: Vec<RelationOutputWitness>,
}

/// Reconstructed source-to-sink witness.
#[derive(Debug, Clone, Eq, PartialEq)]
pub struct ReconstructedRelationWitness {
    /// Source endpoint row.
    pub source: RelationEndpoint,
    /// Sink endpoint row.
    pub sink: RelationEndpoint,
    /// Ordered witness path.
    pub path_nodes: Vec<u32>,
}

/// Relation interchange validation and decode failures.
#[derive(Debug, Clone, Eq, PartialEq)]
pub enum RelationInterchangeError {
    /// Buffer did not start with the interchange magic.
    BadMagic {
        /// Observed first four bytes.
        observed: [u8; 4],
    },
    /// Unsupported schema version.
    UnsupportedVersion {
        /// Observed version.
        version: u32,
    },
    /// Input bytes ended before a required field.
    Truncated {
        /// Field being decoded.
        field: &'static str,
    },
    /// Decode consumed fewer bytes than supplied.
    TrailingBytes {
        /// Number of unconsumed bytes.
        trailing: usize,
    },
    /// Required relation family was empty.
    EmptyRelation {
        /// Relation family name.
        relation: &'static str,
    },
    /// Duplicate tuple id in a relation family.
    DuplicateId {
        /// Relation family name.
        relation: &'static str,
        /// Duplicate id.
        id: u32,
    },
    /// Endpoint span is reversed.
    InvalidEndpointSpan {
        /// Endpoint id.
        id: u32,
        /// Start byte.
        start_byte: u32,
        /// End byte.
        end_byte: u32,
    },
    /// Witness references a missing source or sink endpoint.
    MissingWitnessEndpoint {
        /// Missing relation family.
        relation: &'static str,
        /// Missing endpoint id.
        id: u32,
    },
    /// Witness path is empty or does not connect source and sink nodes.
    InvalidWitnessPath {
        /// Source endpoint id.
        source_id: u32,
        /// Sink endpoint id.
        sink_id: u32,
    },
}

impl Display for RelationInterchangeError {
    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::BadMagic { observed } => write!(
                f,
                "relation interchange magic was {observed:?}. Fix: decode only WRIF relation fixtures."
            ),
            Self::UnsupportedVersion { version } => write!(
                f,
                "relation interchange schema version {version} is unsupported. Fix: regenerate the fixture with the current schema."
            ),
            Self::Truncated { field } => write!(
                f,
                "relation interchange bytes ended while reading `{field}`. Fix: reject truncated relation fixtures."
            ),
            Self::TrailingBytes { trailing } => write!(
                f,
                "relation interchange decode left {trailing} trailing bytes. Fix: write canonical tuple bytes only."
            ),
            Self::EmptyRelation { relation } => write!(
                f,
                "relation interchange `{relation}` relation is empty. Fix: include complete source, sink, fact, and witness fixtures."
            ),
            Self::DuplicateId { relation, id } => write!(
                f,
                "relation interchange `{relation}` relation has duplicate id {id}. Fix: deduplicate tuple ids before export."
            ),
            Self::InvalidEndpointSpan {
                id,
                start_byte,
                end_byte,
            } => write!(
                f,
                "relation endpoint {id} has invalid span {start_byte}..{end_byte}. Fix: normalize source spans before export."
            ),
            Self::MissingWitnessEndpoint { relation, id } => write!(
                f,
                "relation witness references missing `{relation}` endpoint {id}. Fix: include every witness endpoint tuple."
            ),
            Self::InvalidWitnessPath { source_id, sink_id } => write!(
                f,
                "relation witness {source_id}->{sink_id} has an empty or disconnected path. Fix: record source-to-sink path nodes exactly."
            ),
        }
    }
}

impl Error for RelationInterchangeError {}

/// Serialize relation interchange rows to canonical little-endian bytes.
///
/// # Errors
///
/// Returns [`RelationInterchangeError`] when the fixture violates tuple
/// uniqueness, endpoint span, or witness consistency requirements.
pub fn export_relation_interchange_bytes(
    interchange: &RelationInterchange,
) -> Result<Vec<u8>, RelationInterchangeError> {
    validate_relation_interchange(interchange)?;
    let mut out = Vec::new();
    out.extend_from_slice(&RELATION_INTERCHANGE_MAGIC);
    write_u32(&mut out, RELATION_INTERCHANGE_SCHEMA_VERSION);
    write_u32(&mut out, interchange.sources.len() as u32);
    write_u32(&mut out, interchange.sinks.len() as u32);
    write_u32(&mut out, interchange.edges.len() as u32);
    write_u32(&mut out, interchange.call_strings.len() as u32);
    write_u32(&mut out, interchange.facts.len() as u32);
    write_u32(&mut out, interchange.output_witnesses.len() as u32);
    for endpoint in &interchange.sources {
        write_endpoint(&mut out, *endpoint);
    }
    for endpoint in &interchange.sinks {
        write_endpoint(&mut out, *endpoint);
    }
    for edge in &interchange.edges {
        write_u32(&mut out, edge.id);
        write_u32(&mut out, edge.src_node);
        write_u32(&mut out, edge.dst_node);
        write_u32(&mut out, edge.edge_kind);
    }
    for call_string in &interchange.call_strings {
        write_u32(&mut out, call_string.id);
        write_u32(&mut out, call_string.caller_node);
        write_u32(&mut out, call_string.callee_node);
        write_u32(&mut out, call_string.return_node);
        write_u32(&mut out, call_string.depth);
    }
    for fact in &interchange.facts {
        write_u32(&mut out, fact.id);
        write_u32(&mut out, fact.domain_id);
        write_u32(&mut out, fact.tag);
    }
    for witness in &interchange.output_witnesses {
        write_u32(&mut out, witness.source_id);
        write_u32(&mut out, witness.sink_id);
        write_u32(&mut out, witness.path_nodes.len() as u32);
        for node in &witness.path_nodes {
            write_u32(&mut out, *node);
        }
    }
    Ok(out)
}

/// Decode canonical relation interchange bytes.
///
/// # Errors
///
/// Returns [`RelationInterchangeError`] when the bytes are malformed or the
/// decoded fixture violates relation consistency.
pub fn import_relation_interchange_bytes(
    bytes: &[u8],
) -> Result<RelationInterchange, RelationInterchangeError> {
    let mut reader = RelationReader { bytes, offset: 0 };
    let magic = reader.read_magic()?;
    if magic != RELATION_INTERCHANGE_MAGIC {
        return Err(RelationInterchangeError::BadMagic { observed: magic });
    }
    let version = reader.read_u32("schema_version")?;
    if version != RELATION_INTERCHANGE_SCHEMA_VERSION {
        return Err(RelationInterchangeError::UnsupportedVersion { version });
    }
    let source_count = reader.read_u32("source_count")? as usize;
    let sink_count = reader.read_u32("sink_count")? as usize;
    let edge_count = reader.read_u32("edge_count")? as usize;
    let call_string_count = reader.read_u32("call_string_count")? as usize;
    let fact_count = reader.read_u32("fact_count")? as usize;
    let witness_count = reader.read_u32("witness_count")? as usize;
    let mut interchange = RelationInterchange {
        sources: Vec::with_capacity(source_count),
        sinks: Vec::with_capacity(sink_count),
        edges: Vec::with_capacity(edge_count),
        call_strings: Vec::with_capacity(call_string_count),
        facts: Vec::with_capacity(fact_count),
        output_witnesses: Vec::with_capacity(witness_count),
    };
    for _ in 0..source_count {
        interchange.sources.push(reader.read_endpoint()?);
    }
    for _ in 0..sink_count {
        interchange.sinks.push(reader.read_endpoint()?);
    }
    for _ in 0..edge_count {
        interchange.edges.push(RelationEdge {
            id: reader.read_u32("edge.id")?,
            src_node: reader.read_u32("edge.src_node")?,
            dst_node: reader.read_u32("edge.dst_node")?,
            edge_kind: reader.read_u32("edge.edge_kind")?,
        });
    }
    for _ in 0..call_string_count {
        interchange.call_strings.push(RelationCallString {
            id: reader.read_u32("call_string.id")?,
            caller_node: reader.read_u32("call_string.caller_node")?,
            callee_node: reader.read_u32("call_string.callee_node")?,
            return_node: reader.read_u32("call_string.return_node")?,
            depth: reader.read_u32("call_string.depth")?,
        });
    }
    for _ in 0..fact_count {
        interchange.facts.push(RelationFact {
            id: reader.read_u32("fact.id")?,
            domain_id: reader.read_u32("fact.domain_id")?,
            tag: reader.read_u32("fact.tag")?,
        });
    }
    for _ in 0..witness_count {
        let source_id = reader.read_u32("witness.source_id")?;
        let sink_id = reader.read_u32("witness.sink_id")?;
        let path_len = reader.read_u32("witness.path_len")? as usize;
        let mut path_nodes = Vec::with_capacity(path_len);
        for _ in 0..path_len {
            path_nodes.push(reader.read_u32("witness.path_node")?);
        }
        interchange.output_witnesses.push(RelationOutputWitness {
            source_id,
            sink_id,
            path_nodes,
        });
    }
    if reader.offset != bytes.len() {
        return Err(RelationInterchangeError::TrailingBytes {
            trailing: bytes.len() - reader.offset,
        });
    }
    validate_relation_interchange(&interchange)?;
    Ok(interchange)
}

/// Validate relation interchange tuple identities and witness consistency.
///
/// # Errors
///
/// Returns [`RelationInterchangeError`] when required relations are empty,
/// tuple ids duplicate, endpoint spans are invalid, or witnesses are
/// disconnected from source/sink endpoints.
pub fn validate_relation_interchange(
    interchange: &RelationInterchange,
) -> Result<(), RelationInterchangeError> {
    require_non_empty("sources", interchange.sources.len())?;
    require_non_empty("sinks", interchange.sinks.len())?;
    require_non_empty("facts", interchange.facts.len())?;
    require_non_empty("output_witnesses", interchange.output_witnesses.len())?;
    validate_endpoint_ids("sources", &interchange.sources)?;
    validate_endpoint_ids("sinks", &interchange.sinks)?;
    validate_ids("edges", interchange.edges.iter().map(|edge| edge.id))?;
    validate_ids(
        "call_strings",
        interchange.call_strings.iter().map(|call| call.id),
    )?;
    validate_ids("facts", interchange.facts.iter().map(|fact| fact.id))?;
    for witness in &interchange.output_witnesses {
        reconstruct_relation_witness(interchange, witness.source_id, witness.sink_id)?;
    }
    Ok(())
}

/// Reconstruct one source-to-sink witness from relation interchange rows.
///
/// # Errors
///
/// Returns [`RelationInterchangeError`] when the source, sink, or witness path
/// is missing or inconsistent.
pub fn reconstruct_relation_witness(
    interchange: &RelationInterchange,
    source_id: u32,
    sink_id: u32,
) -> Result<ReconstructedRelationWitness, RelationInterchangeError> {
    let source = endpoint_by_id(&interchange.sources, source_id, "sources")?;
    let sink = endpoint_by_id(&interchange.sinks, sink_id, "sinks")?;
    let witness = interchange
        .output_witnesses
        .iter()
        .find(|witness| witness.source_id == source_id && witness.sink_id == sink_id)
        .ok_or(RelationInterchangeError::MissingWitnessEndpoint {
            relation: "output_witnesses",
            id: source_id,
        })?;
    if witness.path_nodes.first() != Some(&source.node_id)
        || witness.path_nodes.last() != Some(&sink.node_id)
    {
        return Err(RelationInterchangeError::InvalidWitnessPath { source_id, sink_id });
    }
    Ok(ReconstructedRelationWitness {
        source,
        sink,
        path_nodes: witness.path_nodes.clone(),
    })
}

fn require_non_empty(relation: &'static str, len: usize) -> Result<(), RelationInterchangeError> {
    if len == 0 {
        return Err(RelationInterchangeError::EmptyRelation { relation });
    }
    Ok(())
}

fn validate_endpoint_ids(
    relation: &'static str,
    endpoints: &[RelationEndpoint],
) -> Result<(), RelationInterchangeError> {
    validate_ids(relation, endpoints.iter().map(|endpoint| endpoint.id))?;
    for endpoint in endpoints {
        if endpoint.end_byte < endpoint.start_byte {
            return Err(RelationInterchangeError::InvalidEndpointSpan {
                id: endpoint.id,
                start_byte: endpoint.start_byte,
                end_byte: endpoint.end_byte,
            });
        }
    }
    Ok(())
}

fn validate_ids(
    relation: &'static str,
    ids: impl Iterator<Item = u32>,
) -> Result<(), RelationInterchangeError> {
    let mut seen = BTreeSet::new();
    for id in ids {
        if !seen.insert(id) {
            return Err(RelationInterchangeError::DuplicateId { relation, id });
        }
    }
    Ok(())
}

fn endpoint_by_id(
    endpoints: &[RelationEndpoint],
    id: u32,
    relation: &'static str,
) -> Result<RelationEndpoint, RelationInterchangeError> {
    endpoints
        .iter()
        .copied()
        .find(|endpoint| endpoint.id == id)
        .ok_or(RelationInterchangeError::MissingWitnessEndpoint { relation, id })
}

fn write_endpoint(out: &mut Vec<u8>, endpoint: RelationEndpoint) {
    write_u32(out, endpoint.id);
    write_u32(out, endpoint.node_id);
    write_u32(out, endpoint.proc_id);
    write_u32(out, endpoint.block_id);
    write_u32(out, endpoint.fact_id);
    write_u32(out, endpoint.file_id);
    write_u32(out, endpoint.start_byte);
    write_u32(out, endpoint.end_byte);
}

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

struct RelationReader<'a> {
    bytes: &'a [u8],
    offset: usize,
}

impl<'a> RelationReader<'a> {
    fn read_magic(&mut self) -> Result<[u8; 4], RelationInterchangeError> {
        let Some(bytes) = self.bytes.get(self.offset..self.offset + 4) else {
            return Err(RelationInterchangeError::Truncated { field: "magic" });
        };
        self.offset += 4;
        Ok([bytes[0], bytes[1], bytes[2], bytes[3]])
    }

    fn read_endpoint(&mut self) -> Result<RelationEndpoint, RelationInterchangeError> {
        Ok(RelationEndpoint {
            id: self.read_u32("endpoint.id")?,
            node_id: self.read_u32("endpoint.node_id")?,
            proc_id: self.read_u32("endpoint.proc_id")?,
            block_id: self.read_u32("endpoint.block_id")?,
            fact_id: self.read_u32("endpoint.fact_id")?,
            file_id: self.read_u32("endpoint.file_id")?,
            start_byte: self.read_u32("endpoint.start_byte")?,
            end_byte: self.read_u32("endpoint.end_byte")?,
        })
    }

    fn read_u32(&mut self, field: &'static str) -> Result<u32, RelationInterchangeError> {
        let Some(bytes) = self.bytes.get(self.offset..self.offset + 4) else {
            return Err(RelationInterchangeError::Truncated { field });
        };
        self.offset += 4;
        Ok(u32::from_le_bytes([bytes[0], bytes[1], bytes[2], bytes[3]]))
    }
}

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

    fn fixture() -> RelationInterchange {
        RelationInterchange {
            sources: vec![RelationEndpoint {
                id: 1,
                node_id: 10,
                proc_id: 0,
                block_id: 1,
                fact_id: 7,
                file_id: 3,
                start_byte: 100,
                end_byte: 110,
            }],
            sinks: vec![RelationEndpoint {
                id: 2,
                node_id: 30,
                proc_id: 1,
                block_id: 4,
                fact_id: 9,
                file_id: 3,
                start_byte: 200,
                end_byte: 210,
            }],
            edges: vec![
                RelationEdge {
                    id: 1,
                    src_node: 10,
                    dst_node: 20,
                    edge_kind: 1,
                },
                RelationEdge {
                    id: 2,
                    src_node: 20,
                    dst_node: 30,
                    edge_kind: 2,
                },
            ],
            call_strings: vec![RelationCallString {
                id: 1,
                caller_node: 10,
                callee_node: 20,
                return_node: 30,
                depth: 1,
            }],
            facts: vec![
                RelationFact {
                    id: 7,
                    domain_id: 1,
                    tag: 11,
                },
                RelationFact {
                    id: 9,
                    domain_id: 1,
                    tag: 13,
                },
            ],
            output_witnesses: vec![RelationOutputWitness {
                source_id: 1,
                sink_id: 2,
                path_nodes: vec![10, 20, 30],
            }],
        }
    }

    #[test]
    fn relation_interchange_exports_exact_tuple_bytes_and_imports_them() {
        let interchange = fixture();

        let bytes = export_relation_interchange_bytes(&interchange).unwrap();

        assert_eq!(&bytes[0..4], b"WRIF");
        assert_eq!(u32::from_le_bytes(bytes[8..12].try_into().unwrap()), 1);
        assert_eq!(u32::from_le_bytes(bytes[12..16].try_into().unwrap()), 1);
        assert_eq!(u32::from_le_bytes(bytes[16..20].try_into().unwrap()), 2);
        assert_eq!(import_relation_interchange_bytes(&bytes).unwrap(), interchange);
    }

    #[test]
    fn relation_interchange_reconstructs_output_witness() {
        let interchange = fixture();

        let witness = reconstruct_relation_witness(&interchange, 1, 2).unwrap();

        assert_eq!(witness.source.node_id, 10);
        assert_eq!(witness.sink.node_id, 30);
        assert_eq!(witness.path_nodes, vec![10, 20, 30]);
    }

    #[test]
    fn relation_interchange_rejects_disconnected_witness() {
        let mut interchange = fixture();
        interchange.output_witnesses[0].path_nodes = vec![10, 20];

        let error = validate_relation_interchange(&interchange).unwrap_err();

        assert!(matches!(
            error,
            RelationInterchangeError::InvalidWitnessPath {
                source_id: 1,
                sink_id: 2
            }
        ));
    }
}