use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use std::fmt::{Display, Formatter};
use std::hash::{Hash, Hasher};
#[derive(Debug, Clone, Serialize, Deserialize, Eq, PartialEq)]
pub struct FlowRef {
pub namespace: String,
pub object_id: String,
pub version: Option<u64>,
}
impl FlowRef {
pub fn new(namespace: impl Into<String>, object_id: impl Into<String>) -> Self {
Self {
namespace: namespace.into(),
object_id: object_id.into(),
version: None,
}
}
pub fn with_version(
namespace: impl Into<String>,
object_id: impl Into<String>,
version: u64,
) -> Self {
Self {
namespace: namespace.into(),
object_id: object_id.into(),
version: Some(version),
}
}
pub fn canonical(&self) -> String {
match self.version {
Some(v) => format!("flow://{}#{}@{}", self.namespace, self.object_id, v),
None => format!("flow://{}#{}", self.namespace, self.object_id),
}
}
pub fn from_canonical(s: &str) -> Option<Self> {
if !s.starts_with("flow://") {
return None;
}
let s = &s[7..];
let (namespace, rest) = s.split_once('#')?;
if let Some((object_id, version_str)) = rest.split_once('@') {
let version = version_str.parse().ok()?;
Some(Self {
namespace: namespace.to_string(),
object_id: object_id.to_string(),
version: Some(version),
})
} else {
Some(Self {
namespace: namespace.to_string(),
object_id: rest.to_string(),
version: None,
})
}
}
}
impl Display for FlowRef {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
write!(f, "{}", self.canonical())
}
}
impl Hash for FlowRef {
fn hash<H: Hasher>(&self, state: &mut H) {
self.namespace.hash(state);
self.object_id.hash(state);
self.version.hash(state);
}
}
#[derive(Debug, Clone, Serialize, Deserialize, Eq, PartialEq, Hash)]
pub struct ContentIdentity {
pub hash: String,
}
impl ContentIdentity {
pub fn new(hash: impl Into<String>) -> Self {
Self { hash: hash.into() }
}
pub fn from_bytes(data: &[u8]) -> Self {
let hash = sha256(data);
Self { hash }
}
pub fn as_str(&self) -> &str {
&self.hash
}
}
impl Display for ContentIdentity {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
write!(f, "sha256:{}", self.hash)
}
}
fn sha256(data: &[u8]) -> String {
let mut hasher = Sha256::new();
hasher.update(data);
let result = hasher.finalize();
format!("{:x}", result)
}
#[derive(Debug, Clone, Serialize, Deserialize, Eq, PartialEq, Hash)]
pub struct PeerId {
pub id: String,
}
impl PeerId {
pub fn new(id: impl Into<String>) -> Self {
Self { id: id.into() }
}
}
impl Display for PeerId {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
write!(f, "{}", self.id)
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PeerAdvertisement {
pub peer_id: PeerId,
pub namespaces: Vec<String>,
pub content_ranges: Vec<ContentRange>,
pub capabilities: Vec<String>,
pub advertised_at_ms: u128,
}
impl PeerAdvertisement {
pub fn new(peer_id: PeerId) -> Self {
Self {
peer_id,
namespaces: Vec::new(),
content_ranges: Vec::new(),
capabilities: Vec::new(),
advertised_at_ms: crate::now_ms(),
}
}
pub fn with_namespace(mut self, namespace: impl Into<String>) -> Self {
self.namespaces.push(namespace.into());
self
}
pub fn with_namespaces(mut self, namespaces: Vec<String>) -> Self {
self.namespaces.extend(namespaces);
self
}
pub fn with_capability(mut self, capability: impl Into<String>) -> Self {
self.capabilities.push(capability.into());
self
}
pub fn with_capabilities(mut self, capabilities: Vec<String>) -> Self {
self.capabilities.extend(capabilities);
self
}
pub fn with_range(mut self, range: ContentRange) -> Self {
self.content_ranges.push(range);
self
}
pub fn provides_namespace(&self, namespace: &str) -> bool {
self.namespaces.iter().any(|ns| ns == namespace)
}
pub fn provides_capability(&self, capability: &str) -> bool {
self.capabilities.iter().any(|cap| cap == capability)
}
}
#[derive(Debug, Clone, Serialize, Deserialize, Eq, PartialEq)]
pub struct ContentRange {
pub start: String,
pub end: String,
pub count: u64,
}
impl ContentRange {
pub fn new(start: impl Into<String>, end: impl Into<String>, count: u64) -> Self {
Self {
start: start.into(),
end: end.into(),
count,
}
}
pub fn contains(&self, object_id: &str) -> bool {
object_id >= self.start.as_str() && object_id <= self.end.as_str()
}
}
pub trait ReferenceResolver: Send + Sync {
fn resolve(&self, flow_ref: &FlowRef) -> Result<Option<Vec<u8>>, String>;
fn can_resolve(&self, flow_ref: &FlowRef) -> bool;
fn list_namespace(&self, namespace: &str) -> Result<Vec<FlowRef>, String>;
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ResolutionMetadata {
pub is_local: bool,
pub source_peer: Option<PeerId>,
pub resolved_at_ms: u128,
}
impl ResolutionMetadata {
pub fn local() -> Self {
Self {
is_local: true,
source_peer: None,
resolved_at_ms: crate::now_ms(),
}
}
pub fn remote(source_peer: PeerId) -> Self {
Self {
is_local: false,
source_peer: Some(source_peer),
resolved_at_ms: crate::now_ms(),
}
}
}
#[derive(Debug, Clone)]
pub struct ResolutionResult {
pub content: Vec<u8>,
pub metadata: ResolutionMetadata,
}
impl ResolutionResult {
pub fn local(content: Vec<u8>) -> Self {
Self {
content,
metadata: ResolutionMetadata::local(),
}
}
pub fn remote(content: Vec<u8>, source_peer: PeerId) -> Self {
Self {
content,
metadata: ResolutionMetadata::remote(source_peer),
}
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_flow_ref_canonical() {
let r = FlowRef::new("docs", "123");
assert_eq!(r.canonical(), "flow://docs#123");
let r = FlowRef::with_version("docs", "123", 42);
assert_eq!(r.canonical(), "flow://docs#123@42");
}
#[test]
fn test_flow_ref_parse() {
let s = "flow://docs#123";
let r = FlowRef::from_canonical(s).unwrap();
assert_eq!(r.namespace, "docs");
assert_eq!(r.object_id, "123");
assert_eq!(r.version, None);
let s = "flow://docs#123@42";
let r = FlowRef::from_canonical(s).unwrap();
assert_eq!(r.namespace, "docs");
assert_eq!(r.object_id, "123");
assert_eq!(r.version, Some(42));
}
#[test]
fn test_content_range_contains() {
let range = ContentRange::new("a", "z", 26);
assert!(range.contains("m"));
assert!(range.contains("a"));
assert!(range.contains("z"));
assert!(!range.contains("0"));
}
#[test]
fn test_peer_advertisement() {
let peer = PeerId::new("peer-a");
let ad = PeerAdvertisement::new(peer.clone())
.with_namespace("docs")
.with_namespace("orders")
.with_capability("search:docs");
assert!(ad.provides_namespace("docs"));
assert!(ad.provides_capability("search:docs"));
assert!(!ad.provides_namespace("embeddings"));
}
#[test]
fn test_resolution_metadata_local() {
let metadata = ResolutionMetadata::local();
assert!(metadata.is_local);
assert!(metadata.source_peer.is_none());
assert!(metadata.resolved_at_ms > 0);
}
#[test]
fn test_resolution_metadata_remote() {
let peer = PeerId::new("peer-b");
let metadata = ResolutionMetadata::remote(peer.clone());
assert!(!metadata.is_local);
assert_eq!(metadata.source_peer, Some(peer));
assert!(metadata.resolved_at_ms > 0);
}
#[test]
fn test_resolution_result_creation() {
let content = b"test content".to_vec();
let result = ResolutionResult::local(content.clone());
assert_eq!(result.content, content);
assert!(result.metadata.is_local);
assert!(result.metadata.source_peer.is_none());
}
#[test]
fn test_resolution_result_remote() {
let content = b"acquired content".to_vec();
let peer = PeerId::new("provider-peer");
let result = ResolutionResult::remote(content.clone(), peer.clone());
assert_eq!(result.content, content);
assert!(!result.metadata.is_local);
assert_eq!(result.metadata.source_peer, Some(peer));
}
}