use async_trait::async_trait;
use dashmap::DashMap;
use parking_lot::Mutex;
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use std::sync::Arc;
use tenzro_types::primitives::{Hash, Timestamp};
use tenzro_types::principal_chain::PrincipalChainSummary;
use crate::error::{Result, StorageError};
#[cfg(feature = "celestia")]
pub mod celestia;
#[cfg(feature = "celestia")]
pub use celestia::CelestiaBackend;
#[cfg(feature = "eigenda")]
pub mod eigenda;
#[cfg(feature = "eigenda")]
pub use eigenda::EigenDaBackend;
#[cfg(feature = "avail")]
pub mod avail;
#[cfg(feature = "avail")]
pub use avail::AvailBackend;
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum ReceiptStorageMode {
Inline,
OffloadedDA,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum ReceiptKind {
SettlementEscrow,
SettlementChannel,
Inference,
AgentMessage,
KillSwitch,
Lifecycle,
Governance,
}
impl ReceiptKind {
pub fn default_mode(&self) -> ReceiptStorageMode {
match self {
ReceiptKind::SettlementChannel
| ReceiptKind::Inference
| ReceiptKind::AgentMessage => ReceiptStorageMode::OffloadedDA,
ReceiptKind::SettlementEscrow
| ReceiptKind::KillSwitch
| ReceiptKind::Lifecycle
| ReceiptKind::Governance => ReceiptStorageMode::Inline,
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub enum DaBackendId {
InlineFallback,
EigenDA,
Celestia,
Avail,
IrohBlobs,
}
impl DaBackendId {
pub fn as_str(&self) -> &'static str {
match self {
DaBackendId::InlineFallback => "inline_fallback",
DaBackendId::EigenDA => "eigenda",
DaBackendId::Celestia => "celestia",
DaBackendId::Avail => "avail",
DaBackendId::IrohBlobs => "iroh_blobs",
}
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct DaPointer {
pub backend: DaBackendId,
pub namespace: Vec<u8>,
pub locator: Vec<u8>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub commitment_kzg: Option<Vec<u8>>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub attestation_root: Option<Hash>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ReceiptSummary {
pub receipt_id: Hash,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub payer: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub payee: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub amount_wei: Option<u128>,
pub timestamp: Timestamp,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub principal_chain_summary: Option<PrincipalChainSummary>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct MandateRef {
pub protocol: String,
pub mandate_hash: Hash,
pub issuer_did: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub mandate_uri: Option<String>,
#[serde(default)]
pub expires_at: u64,
}
impl MandateRef {
pub fn ap2_cart(mandate_hash: Hash, issuer_did: impl Into<String>) -> Self {
Self {
protocol: "ap2-cart".into(),
mandate_hash,
issuer_did: issuer_did.into(),
mandate_uri: None,
expires_at: 0,
}
}
pub fn x402(mandate_hash: Hash, issuer_did: impl Into<String>) -> Self {
Self {
protocol: "x402".into(),
mandate_hash,
issuer_did: issuer_did.into(),
mandate_uri: None,
expires_at: 0,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ReceiptEnvelope {
pub kind: ReceiptKind,
pub storage_mode: ReceiptStorageMode,
pub inline_summary: ReceiptSummary,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub inline_payload: Option<Vec<u8>>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub da_pointer: Option<DaPointer>,
pub commitment: Hash,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub mandate_ref: Option<MandateRef>,
}
impl ReceiptEnvelope {
pub fn inline(kind: ReceiptKind, summary: ReceiptSummary, payload: Vec<u8>) -> Self {
let commitment = compute_commitment(&payload);
Self {
kind,
storage_mode: ReceiptStorageMode::Inline,
inline_summary: summary,
inline_payload: Some(payload),
da_pointer: None,
commitment,
mandate_ref: None,
}
}
pub fn offloaded(
kind: ReceiptKind,
summary: ReceiptSummary,
pointer: DaPointer,
commitment: Hash,
) -> Self {
Self {
kind,
storage_mode: ReceiptStorageMode::OffloadedDA,
inline_summary: summary,
inline_payload: None,
da_pointer: Some(pointer),
commitment,
mandate_ref: None,
}
}
pub fn with_mandate(mut self, mandate_ref: MandateRef) -> Self {
self.mandate_ref = Some(mandate_ref);
self
}
pub fn validate(&self) -> Result<()> {
match self.storage_mode {
ReceiptStorageMode::Inline => {
let payload = self.inline_payload.as_ref().ok_or_else(|| {
StorageError::InvalidValue(
"Inline receipt envelope missing inline_payload".into(),
)
})?;
if self.da_pointer.is_some() {
return Err(StorageError::InvalidValue(
"Inline receipt envelope must not carry da_pointer".into(),
));
}
let actual = compute_commitment(payload);
if actual != self.commitment {
return Err(StorageError::InvalidValue(format!(
"Receipt commitment mismatch: declared {}, computed {}",
self.commitment, actual,
)));
}
Ok(())
}
ReceiptStorageMode::OffloadedDA => {
if self.da_pointer.is_none() {
return Err(StorageError::InvalidValue(
"Offloaded receipt envelope missing da_pointer".into(),
));
}
if self.inline_payload.is_some() {
return Err(StorageError::InvalidValue(
"Offloaded receipt envelope must not carry inline_payload".into(),
));
}
Ok(())
}
}
}
}
pub fn compute_commitment(payload: &[u8]) -> Hash {
let mut hasher = Sha256::new();
hasher.update(payload);
let result = hasher.finalize();
let mut bytes = [0u8; 32];
bytes.copy_from_slice(&result);
Hash::new(bytes)
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct DaBackendStatus {
pub backend: DaBackendId,
pub healthy: bool,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub last_submission_ms: Option<i64>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub last_fetch_ms: Option<i64>,
pub error_rate_bps: u16,
}
#[async_trait]
pub trait DaBackend: Send + Sync {
fn id(&self) -> DaBackendId;
fn status(&self) -> DaBackendStatus;
async fn submit(&self, namespace: &[u8], payload: &[u8]) -> Result<DaPointer>;
async fn fetch(&self, pointer: &DaPointer) -> Result<Vec<u8>>;
async fn verify_availability(&self, pointer: &DaPointer) -> Result<()>;
}
pub struct InlineFallbackBackend {
store: DashMap<Vec<u8>, Vec<u8>>,
storage: Option<Arc<dyn crate::kv::KvStore>>,
last_submission_ms: Mutex<Option<i64>>,
last_fetch_ms: Mutex<Option<i64>>,
}
impl InlineFallbackBackend {
const STORAGE_PREFIX: &'static [u8] = b"da_fallback:";
pub fn new() -> Self {
Self {
store: DashMap::new(),
storage: None,
last_submission_ms: Mutex::new(None),
last_fetch_ms: Mutex::new(None),
}
}
pub fn with_storage(mut self, storage: Arc<dyn crate::kv::KvStore>) -> Self {
self.storage = Some(storage);
self
}
pub fn arc() -> Arc<dyn DaBackend> {
Arc::new(Self::new())
}
fn storage_key(locator: &[u8]) -> Vec<u8> {
[Self::STORAGE_PREFIX, locator].concat()
}
fn make_locator(payload: &[u8]) -> Vec<u8> {
let commitment = compute_commitment(payload);
format!("fallback:{}", hex::encode(commitment.as_bytes())).into_bytes()
}
fn now_ms() -> i64 {
Timestamp::now().0
}
pub fn submit_sync(&self, namespace: &[u8], payload: &[u8]) -> DaPointer {
let locator = Self::make_locator(payload);
self.store.insert(locator.clone(), payload.to_vec());
if let Some(storage) = &self.storage
&& let Err(e) = storage.put(crate::kv::CF_METADATA, &Self::storage_key(&locator), payload) {
tracing::warn!(
"InlineFallbackBackend: failed to mirror-write payload to storage: {}",
e
);
}
*self.last_submission_ms.lock() = Some(Self::now_ms());
DaPointer {
backend: DaBackendId::InlineFallback,
namespace: namespace.to_vec(),
locator,
commitment_kzg: None,
attestation_root: None,
}
}
pub fn fetch_sync(&self, pointer: &DaPointer) -> Result<Vec<u8>> {
if pointer.backend != DaBackendId::InlineFallback {
return Err(StorageError::Generic(format!(
"InlineFallbackBackend cannot fetch pointer with backend {:?}",
pointer.backend
)));
}
if let Some(entry) = self.store.get(&pointer.locator) {
*self.last_fetch_ms.lock() = Some(Self::now_ms());
return Ok(entry.value().clone());
}
if let Some(storage) = &self.storage
&& let Some(bytes) = storage
.get(crate::kv::CF_METADATA, &Self::storage_key(&pointer.locator))
.map_err(|e| StorageError::Generic(format!("storage read: {}", e)))?
{
self.store.insert(pointer.locator.clone(), bytes.clone());
*self.last_fetch_ms.lock() = Some(Self::now_ms());
return Ok(bytes);
}
Err(StorageError::KeyNotFound(format!(
"InlineFallbackBackend has no payload for locator {}",
String::from_utf8_lossy(&pointer.locator)
)))
}
}
impl std::fmt::Debug for InlineFallbackBackend {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("InlineFallbackBackend")
.field("entries", &self.store.len())
.finish()
}
}
impl Default for InlineFallbackBackend {
fn default() -> Self {
Self::new()
}
}
#[async_trait]
impl DaBackend for InlineFallbackBackend {
fn id(&self) -> DaBackendId {
DaBackendId::InlineFallback
}
fn status(&self) -> DaBackendStatus {
DaBackendStatus {
backend: DaBackendId::InlineFallback,
healthy: true,
last_submission_ms: *self.last_submission_ms.lock(),
last_fetch_ms: *self.last_fetch_ms.lock(),
error_rate_bps: 0,
}
}
async fn submit(&self, namespace: &[u8], payload: &[u8]) -> Result<DaPointer> {
Ok(self.submit_sync(namespace, payload))
}
async fn fetch(&self, pointer: &DaPointer) -> Result<Vec<u8>> {
self.fetch_sync(pointer)
}
async fn verify_availability(&self, pointer: &DaPointer) -> Result<()> {
if pointer.backend != DaBackendId::InlineFallback {
return Err(StorageError::Generic(format!(
"InlineFallbackBackend cannot verify pointer with backend {:?}",
pointer.backend
)));
}
if self.store.contains_key(&pointer.locator) {
return Ok(());
}
if let Some(storage) = &self.storage
&& storage
.get(crate::kv::CF_METADATA, &Self::storage_key(&pointer.locator))
.map_err(|e| StorageError::Generic(format!("storage read: {}", e)))?
.is_some()
{
return Ok(());
}
Err(StorageError::KeyNotFound(format!(
"InlineFallbackBackend has no payload for locator {}",
String::from_utf8_lossy(&pointer.locator)
)))
}
}
#[cfg(test)]
mod tests {
use super::*;
fn sample_summary() -> ReceiptSummary {
ReceiptSummary {
receipt_id: Hash::new([7u8; 32]),
payer: Some("did:tenzro:human:alice".into()),
payee: Some("did:tenzro:machine:bob".into()),
amount_wei: Some(123_456_789),
timestamp: Timestamp::new(1_700_000_000_000),
principal_chain_summary: None,
}
}
#[test]
fn default_mode_per_kind_matches_spec() {
assert_eq!(
ReceiptKind::SettlementEscrow.default_mode(),
ReceiptStorageMode::Inline
);
assert_eq!(
ReceiptKind::SettlementChannel.default_mode(),
ReceiptStorageMode::OffloadedDA
);
assert_eq!(
ReceiptKind::Inference.default_mode(),
ReceiptStorageMode::OffloadedDA
);
assert_eq!(
ReceiptKind::AgentMessage.default_mode(),
ReceiptStorageMode::OffloadedDA
);
assert_eq!(
ReceiptKind::KillSwitch.default_mode(),
ReceiptStorageMode::Inline
);
assert_eq!(
ReceiptKind::Lifecycle.default_mode(),
ReceiptStorageMode::Inline
);
assert_eq!(
ReceiptKind::Governance.default_mode(),
ReceiptStorageMode::Inline
);
}
#[test]
fn compute_commitment_is_sha256() {
let h = compute_commitment(b"hello");
assert_eq!(
hex::encode(h.as_bytes()),
"2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824"
);
}
#[test]
fn inline_envelope_validates() {
let env = ReceiptEnvelope::inline(
ReceiptKind::SettlementEscrow,
sample_summary(),
b"escrow-payload".to_vec(),
);
env.validate().unwrap();
assert_eq!(env.storage_mode, ReceiptStorageMode::Inline);
assert!(env.da_pointer.is_none());
assert!(env.inline_payload.is_some());
}
#[test]
fn inline_envelope_with_tampered_commitment_rejected() {
let mut env = ReceiptEnvelope::inline(
ReceiptKind::Inference,
sample_summary(),
b"payload".to_vec(),
);
env.commitment = Hash::new([0u8; 32]);
let err = env.validate().unwrap_err();
match err {
StorageError::InvalidValue(msg) => assert!(msg.contains("commitment mismatch")),
other => panic!("unexpected error: {:?}", other),
}
}
#[test]
fn offloaded_envelope_validates() {
let payload = b"large-inference-blob".to_vec();
let commitment = compute_commitment(&payload);
let pointer = DaPointer {
backend: DaBackendId::EigenDA,
namespace: b"tenzro/inference".to_vec(),
locator: b"blob-42".to_vec(),
commitment_kzg: Some(vec![0xab; 48]),
attestation_root: Some(Hash::new([3u8; 32])),
};
let env =
ReceiptEnvelope::offloaded(ReceiptKind::Inference, sample_summary(), pointer, commitment);
env.validate().unwrap();
assert_eq!(env.storage_mode, ReceiptStorageMode::OffloadedDA);
assert!(env.inline_payload.is_none());
assert!(env.da_pointer.is_some());
}
#[test]
fn offloaded_envelope_without_pointer_rejected() {
let env = ReceiptEnvelope {
kind: ReceiptKind::Inference,
storage_mode: ReceiptStorageMode::OffloadedDA,
inline_summary: sample_summary(),
inline_payload: None,
da_pointer: None,
commitment: Hash::new([1u8; 32]),
mandate_ref: None,
};
let err = env.validate().unwrap_err();
match err {
StorageError::InvalidValue(msg) => assert!(msg.contains("missing da_pointer")),
other => panic!("unexpected error: {:?}", other),
}
}
#[test]
fn inline_envelope_with_pointer_rejected() {
let payload = b"x".to_vec();
let env = ReceiptEnvelope {
kind: ReceiptKind::SettlementEscrow,
storage_mode: ReceiptStorageMode::Inline,
inline_summary: sample_summary(),
inline_payload: Some(payload.clone()),
da_pointer: Some(DaPointer {
backend: DaBackendId::EigenDA,
namespace: vec![],
locator: vec![],
commitment_kzg: None,
attestation_root: None,
}),
commitment: compute_commitment(&payload),
mandate_ref: None,
};
let err = env.validate().unwrap_err();
match err {
StorageError::InvalidValue(msg) => assert!(msg.contains("must not carry da_pointer")),
other => panic!("unexpected error: {:?}", other),
}
}
#[tokio::test]
async fn inline_fallback_round_trips_payload() {
let backend = InlineFallbackBackend::new();
assert_eq!(backend.id(), DaBackendId::InlineFallback);
assert!(backend.status().healthy);
let payload = b"large-inference-blob".to_vec();
let pointer = backend.submit(b"tenzro/inference", &payload).await.unwrap();
assert_eq!(pointer.backend, DaBackendId::InlineFallback);
assert_eq!(pointer.namespace, b"tenzro/inference".to_vec());
let expected_locator = format!(
"fallback:{}",
hex::encode(compute_commitment(&payload).as_bytes())
);
assert_eq!(pointer.locator, expected_locator.as_bytes());
backend.verify_availability(&pointer).await.unwrap();
let fetched = backend.fetch(&pointer).await.unwrap();
assert_eq!(fetched, payload);
let status = backend.status();
assert!(status.last_submission_ms.is_some());
assert!(status.last_fetch_ms.is_some());
let pointer2 = backend.submit(b"tenzro/inference", &payload).await.unwrap();
assert_eq!(pointer2.locator, pointer.locator);
}
#[tokio::test]
async fn inline_fallback_rejects_foreign_pointer() {
let backend = InlineFallbackBackend::new();
let foreign = DaPointer {
backend: DaBackendId::EigenDA,
namespace: vec![],
locator: b"blob-1".to_vec(),
commitment_kzg: None,
attestation_root: None,
};
assert!(backend.fetch(&foreign).await.is_err());
assert!(backend.verify_availability(&foreign).await.is_err());
}
#[tokio::test]
async fn inline_fallback_missing_locator_is_key_not_found() {
let backend = InlineFallbackBackend::new();
let pointer = DaPointer {
backend: DaBackendId::InlineFallback,
namespace: vec![],
locator: b"fallback:deadbeef".to_vec(),
commitment_kzg: None,
attestation_root: None,
};
let err = backend.fetch(&pointer).await.unwrap_err();
assert!(matches!(err, StorageError::KeyNotFound(_)));
}
#[test]
fn da_backend_id_str() {
assert_eq!(DaBackendId::InlineFallback.as_str(), "inline_fallback");
assert_eq!(DaBackendId::EigenDA.as_str(), "eigenda");
assert_eq!(DaBackendId::Celestia.as_str(), "celestia");
assert_eq!(DaBackendId::Avail.as_str(), "avail");
assert_eq!(DaBackendId::IrohBlobs.as_str(), "iroh_blobs");
}
}