use std::error::Error;
use std::fmt::{self, Display, Formatter};
use serde::{Deserialize, Serialize};
use crate::{CollectionIndexRef, DocumentId, IndexMode, IndexName, IndexRecordModel};
pub const RELATIONSHIP_RECORD_MODEL_FORMAT_VERSION: u16 = 1;
pub const RELATIONSHIP_RECORD_FORMAT_VERSION: u16 = 1;
pub const RELATIONSHIP_SOURCE_FORMAT_VERSION: u16 = 1;
pub const MAX_RELATIONSHIP_RECORD_BYTES: usize = 1_048_576;
pub const MAX_RELATIONSHIP_ID_BYTES: usize = 512;
pub const MAX_RELATIONSHIP_LABEL_BYTES: usize = 256;
pub const MAX_RELATIONSHIP_ENTITY_ID_BYTES: usize = 1_024;
pub const MAX_RELATIONSHIP_PROVENANCE: usize = 32;
pub const MAX_RELATIONSHIP_EVIDENCE: usize = 32;
#[derive(Clone, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(deny_unknown_fields)]
pub struct EntityRefV1 {
pub entity_kind: String,
pub entity_id: String,
}
#[derive(Clone, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(deny_unknown_fields)]
pub struct RelationshipEvidenceRefV1 {
pub index: CollectionIndexRef,
pub index_name: IndexName,
pub record_id: DocumentId,
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum RelationshipDirectionalityV1 {
Directed,
Undirected,
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct RelationshipRecordV1 {
pub format_version: u16,
pub relationship_id: String,
pub source: EntityRefV1,
pub target: EntityRefV1,
pub relationship_type: String,
pub directionality: RelationshipDirectionalityV1,
pub valid_from_unix_nanos: Option<u64>,
pub valid_through_unix_nanos: Option<u64>,
pub provenance: Vec<String>,
pub evidence: Vec<RelationshipEvidenceRefV1>,
pub assertion_class: String,
}
impl RelationshipRecordV1 {
pub fn validate(
&self,
model: &IndexRecordModel,
containing_collection: crate::CollectionId,
) -> Result<(), RelationshipRecordError> {
if self.format_version != RELATIONSHIP_RECORD_FORMAT_VERSION {
return Err(RelationshipRecordError::UnsupportedRecordVersion(self.format_version));
}
if !matches!(model, IndexRecordModel::RelationshipV1 { .. }) {
return Err(RelationshipRecordError::RecordModelMismatch);
}
validate_identifier(&self.relationship_id, MAX_RELATIONSHIP_ID_BYTES)?;
validate_entity(&self.source)?;
validate_entity(&self.target)?;
validate_label(&self.relationship_type)?;
validate_label(&self.assertion_class)?;
if self
.valid_from_unix_nanos
.zip(self.valid_through_unix_nanos)
.is_some_and(|(start, end)| start >= end)
{
return Err(RelationshipRecordError::InvalidValidityInterval);
}
validate_sorted_unique_labels(&self.provenance, MAX_RELATIONSHIP_PROVENANCE)?;
if self.evidence.len() > MAX_RELATIONSHIP_EVIDENCE || self.evidence.windows(2).any(|pair| pair[0] >= pair[1]) {
return Err(RelationshipRecordError::InvalidEvidenceReferences);
}
if self
.evidence
.iter()
.any(|reference| reference.index.collection_id() != containing_collection)
{
return Err(RelationshipRecordError::CrossCollectionEvidence);
}
let encoded = serde_json::to_vec(self).map_err(|error| RelationshipRecordError::Codec(error.to_string()))?;
if encoded.len() > MAX_RELATIONSHIP_RECORD_BYTES {
return Err(RelationshipRecordError::RecordBudgetExceeded(encoded.len()));
}
Ok(())
}
pub fn routing_identity_v1(
&self,
model: &IndexRecordModel,
containing_collection: crate::CollectionId,
) -> Result<[u8; 32], RelationshipRecordError> {
self.validate(model, containing_collection)?;
let mut hasher = blake3::Hasher::new();
hasher.update(b"lodestone/relationship-routing/v1\0");
hasher.update(&(self.relationship_id.len() as u64).to_be_bytes());
hasher.update(self.relationship_id.as_bytes());
Ok(*hasher.finalize().as_bytes())
}
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct CanonicalRelationshipSourceV1 {
pub format_version: u16,
pub relationship: RelationshipRecordV1,
}
pub fn canonical_relationship_source_v1(
relationship: RelationshipRecordV1,
model: &IndexRecordModel,
containing_collection: crate::CollectionId,
) -> Result<(Vec<u8>, [u8; 32]), RelationshipRecordError> {
let routing = relationship.routing_identity_v1(model, containing_collection)?;
let source = CanonicalRelationshipSourceV1 {
format_version: RELATIONSHIP_SOURCE_FORMAT_VERSION,
relationship,
};
let encoded = serde_json::to_vec(&source).map_err(|error| RelationshipRecordError::Codec(error.to_string()))?;
if encoded.len() > MAX_RELATIONSHIP_RECORD_BYTES {
return Err(RelationshipRecordError::RecordBudgetExceeded(encoded.len()));
}
Ok((encoded, routing))
}
pub fn decode_canonical_relationship_source_v1(
bytes: &[u8],
model: &IndexRecordModel,
containing_collection: crate::CollectionId,
) -> Result<(CanonicalRelationshipSourceV1, [u8; 32]), RelationshipRecordError> {
if bytes.len() > MAX_RELATIONSHIP_RECORD_BYTES {
return Err(RelationshipRecordError::RecordBudgetExceeded(bytes.len()));
}
let source: CanonicalRelationshipSourceV1 =
serde_json::from_slice(bytes).map_err(|error| RelationshipRecordError::Codec(error.to_string()))?;
if source.format_version != RELATIONSHIP_SOURCE_FORMAT_VERSION {
return Err(RelationshipRecordError::UnsupportedSourceVersion(source.format_version));
}
let routing = source.relationship.routing_identity_v1(model, containing_collection)?;
let canonical = serde_json::to_vec(&source).map_err(|error| RelationshipRecordError::Codec(error.to_string()))?;
if canonical != bytes {
return Err(RelationshipRecordError::NonCanonicalSource);
}
Ok((source, routing))
}
pub fn validate_relationship_record_model_v1(
format_version: u16,
mode: IndexMode,
) -> Result<(), crate::MetricRecordError> {
if format_version != RELATIONSHIP_RECORD_MODEL_FORMAT_VERSION {
return Err(crate::MetricRecordError::UnsupportedRecordModelVersion(format_version));
}
if !matches!(mode, IndexMode::Mutable | IndexMode::AppendLog) {
return Err(crate::MetricRecordError::RelationshipModelRequiresMutableOrAppendLog);
}
Ok(())
}
fn validate_entity(entity: &EntityRefV1) -> Result<(), RelationshipRecordError> {
validate_label(&entity.entity_kind)?;
validate_identifier(&entity.entity_id, MAX_RELATIONSHIP_ENTITY_ID_BYTES)
}
fn validate_label(value: &str) -> Result<(), RelationshipRecordError> {
validate_identifier(value, MAX_RELATIONSHIP_LABEL_BYTES)
}
fn validate_identifier(value: &str, maximum: usize) -> Result<(), RelationshipRecordError> {
if value.is_empty() || value.len() > maximum || value.bytes().any(|byte| byte == 0 || byte <= 0x1f || byte == 0x7f)
{
return Err(RelationshipRecordError::InvalidIdentifier);
}
Ok(())
}
fn validate_sorted_unique_labels(values: &[String], maximum: usize) -> Result<(), RelationshipRecordError> {
if values.len() > maximum || values.windows(2).any(|pair| pair[0] >= pair[1]) {
return Err(RelationshipRecordError::InvalidProvenance);
}
for value in values {
validate_label(value)?;
}
Ok(())
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum RelationshipRecordError {
UnsupportedRecordVersion(u16),
UnsupportedSourceVersion(u16),
RecordModelMismatch,
InvalidIdentifier,
InvalidValidityInterval,
InvalidProvenance,
InvalidEvidenceReferences,
CrossCollectionEvidence,
RecordBudgetExceeded(usize),
NonCanonicalSource,
Codec(String),
}
impl Display for RelationshipRecordError {
fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
write!(formatter, "relationship record: {self:?}")
}
}
impl Error for RelationshipRecordError {}
#[cfg(test)]
mod tests {
use super::*;
use crate::{CollectionId, IndexId};
fn model() -> IndexRecordModel {
IndexRecordModel::RelationshipV1 {
format_version: RELATIONSHIP_RECORD_MODEL_FORMAT_VERSION,
}
}
fn record(collection: CollectionId) -> Result<RelationshipRecordV1, Box<dyn Error>> {
Ok(RelationshipRecordV1 {
format_version: RELATIONSHIP_RECORD_FORMAT_VERSION,
relationship_id: "edge-1".to_owned(),
source: EntityRefV1 {
entity_kind: "service".to_owned(),
entity_id: "api".to_owned(),
},
target: EntityRefV1 {
entity_kind: "service".to_owned(),
entity_id: "db".to_owned(),
},
relationship_type: "calls".to_owned(),
directionality: RelationshipDirectionalityV1::Directed,
valid_from_unix_nanos: Some(10),
valid_through_unix_nanos: Some(20),
provenance: vec!["runtime".to_owned()],
evidence: vec![RelationshipEvidenceRefV1 {
index: CollectionIndexRef::new(collection, IndexId::new(7)?),
index_name: IndexName::new("evidence")?,
record_id: DocumentId::new("span-1")?,
}],
assertion_class: "observed".to_owned(),
})
}
#[test]
fn canonical_source_round_trips_and_routes_by_logical_relationship() -> Result<(), Box<dyn Error>> {
let collection = CollectionId::new(3)?;
let (encoded, routing) = canonical_relationship_source_v1(record(collection)?, &model(), collection)?;
let (decoded, decoded_routing) = decode_canonical_relationship_source_v1(&encoded, &model(), collection)?;
assert_eq!(decoded.relationship.relationship_id, "edge-1");
assert_eq!(routing, decoded_routing);
Ok(())
}
#[test]
fn invalid_interval_and_cross_collection_evidence_refuse() -> Result<(), Box<dyn Error>> {
let collection = CollectionId::new(3)?;
let mut invalid = record(collection)?;
invalid.valid_through_unix_nanos = Some(10);
assert_eq!(
invalid.validate(&model(), collection),
Err(RelationshipRecordError::InvalidValidityInterval)
);
let mut cross_collection = record(collection)?;
cross_collection.evidence[0].index = CollectionIndexRef::new(CollectionId::new(4)?, IndexId::new(7)?);
assert_eq!(
cross_collection.validate(&model(), collection),
Err(RelationshipRecordError::CrossCollectionEvidence)
);
Ok(())
}
}