vti_common/telemetry/
mod.rs

1//! Pluggable telemetry sink for mediator-attribution events.
2//!
3//! Default impl is `RingBufferTelemetry` (in-memory, bounded). Other
4//! backends (file rotation, append-only log, blockchain anchor, fjall
5//! keyspace) can be added without changing call sites by implementing
6//! `TelemetrySink`.
7//!
8//! This is intentionally separate from the `audit!` macro: the audit log
9//! carries security-audit semantics (auth events, key operations) that
10//! are append-only and durable, while this surface is higher-volume and
11//! query-oriented for runtime reporting.
12
13use std::collections::HashSet;
14use std::sync::Arc;
15
16use async_trait::async_trait;
17use chrono::{DateTime, Utc};
18use serde::{Deserialize, Serialize};
19use thiserror::Error;
20
21mod ring;
22pub use ring::RingBufferTelemetry;
23
24#[async_trait]
25pub trait TelemetrySink: Send + Sync {
26    async fn record(&self, event: TelemetryEvent) -> Result<(), TelemetryError>;
27    async fn query(&self, filter: &TelemetryFilter) -> Result<Vec<TelemetryEvent>, TelemetryError>;
28}
29
30pub type SharedTelemetrySink = Arc<dyn TelemetrySink>;
31
32#[derive(Debug, Clone, Serialize, Deserialize)]
33pub struct TelemetryEvent {
34    pub at: DateTime<Utc>,
35    pub kind: TelemetryKind,
36    #[serde(skip_serializing_if = "Option::is_none", default)]
37    pub mediator_did: Option<String>,
38    #[serde(skip_serializing_if = "Option::is_none", default)]
39    pub sender_did: Option<String>,
40    #[serde(skip_serializing_if = "Option::is_none", default)]
41    pub message_type: Option<String>,
42    #[serde(default, skip_serializing_if = "serde_json::Map::is_empty")]
43    pub fields: serde_json::Map<String, serde_json::Value>,
44}
45
46impl TelemetryEvent {
47    pub fn new(kind: TelemetryKind) -> Self {
48        Self {
49            at: Utc::now(),
50            kind,
51            mediator_did: None,
52            sender_did: None,
53            message_type: None,
54            fields: serde_json::Map::new(),
55        }
56    }
57
58    pub fn with_mediator(mut self, did: impl Into<String>) -> Self {
59        self.mediator_did = Some(did.into());
60        self
61    }
62
63    pub fn with_sender(mut self, did: impl Into<String>) -> Self {
64        self.sender_did = Some(did.into());
65        self
66    }
67
68    pub fn with_message_type(mut self, ty: impl Into<String>) -> Self {
69        self.message_type = Some(ty.into());
70        self
71    }
72
73    pub fn with_field(mut self, key: impl Into<String>, value: serde_json::Value) -> Self {
74        self.fields.insert(key.into(), value);
75        self
76    }
77}
78
79#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
80#[serde(rename_all = "snake_case")]
81pub enum TelemetryKind {
82    DidcommInbound,
83    DidcommResponseDropped,
84    MediatorHandshakeOk,
85    MediatorHandshakeFailed,
86    MediatorHandshakeBypassed,
87    ServicesDidcommUpdate,
88    MediatorDrainStart,
89    MediatorDrainCancel,
90    MediatorDrainExpire,
91    ServicesDidcommEnable,
92    ServicesDidcommDisable,
93    ServicesRestEnable,
94    ServicesRestUpdate,
95    ServicesRestDisable,
96    ServicesWebauthnEnable,
97    ServicesWebauthnUpdate,
98    ServicesWebauthnDisable,
99    ServicesTspEnable,
100    ServicesTspUpdate,
101    ServicesTspDisable,
102}
103
104#[derive(Debug, Clone, Default)]
105pub struct TelemetryFilter {
106    pub since: Option<DateTime<Utc>>,
107    pub until: Option<DateTime<Utc>>,
108    pub kinds: Option<HashSet<TelemetryKind>>,
109    pub mediator_did: Option<String>,
110    pub sender_did: Option<String>,
111}
112
113impl TelemetryFilter {
114    pub fn new() -> Self {
115        Self::default()
116    }
117
118    pub fn since(mut self, t: DateTime<Utc>) -> Self {
119        self.since = Some(t);
120        self
121    }
122
123    pub fn until(mut self, t: DateTime<Utc>) -> Self {
124        self.until = Some(t);
125        self
126    }
127
128    pub fn kind(mut self, k: TelemetryKind) -> Self {
129        self.kinds.get_or_insert_with(HashSet::new).insert(k);
130        self
131    }
132
133    pub fn mediator(mut self, did: impl Into<String>) -> Self {
134        self.mediator_did = Some(did.into());
135        self
136    }
137
138    pub fn sender(mut self, did: impl Into<String>) -> Self {
139        self.sender_did = Some(did.into());
140        self
141    }
142
143    pub fn matches(&self, ev: &TelemetryEvent) -> bool {
144        if let Some(t) = self.since
145            && ev.at < t
146        {
147            return false;
148        }
149        if let Some(t) = self.until
150            && ev.at > t
151        {
152            return false;
153        }
154        if let Some(ref ks) = self.kinds
155            && !ks.contains(&ev.kind)
156        {
157            return false;
158        }
159        if let Some(ref m) = self.mediator_did
160            && ev.mediator_did.as_deref() != Some(m.as_str())
161        {
162            return false;
163        }
164        if let Some(ref s) = self.sender_did
165            && ev.sender_did.as_deref() != Some(s.as_str())
166        {
167            return false;
168        }
169        true
170    }
171}
172
173#[derive(Debug, Error)]
174pub enum TelemetryError {
175    #[error("telemetry sink backend failed: {0}")]
176    Backend(String),
177}
178
179#[cfg(test)]
180mod swappability_tests {
181    //! Spec criterion #17: a downstream impl of [`TelemetrySink`]
182    //! must satisfy the same observable contract as
183    //! [`RingBufferTelemetry`]. This module ships a minimal
184    //! `Vec<Mutex<…>>` impl and runs a parametric test against
185    //! both — proving the trait boundary holds.
186    use super::*;
187    use std::sync::Mutex;
188
189    /// Stub [`TelemetrySink`] backed by an unbounded
190    /// `Vec<TelemetryEvent>`. NOT used in production — exists only
191    /// to prove `Arc<dyn TelemetrySink>` swappability.
192    struct VecMutexSink {
193        events: Mutex<Vec<TelemetryEvent>>,
194    }
195
196    impl VecMutexSink {
197        fn new() -> Self {
198            Self {
199                events: Mutex::new(Vec::new()),
200            }
201        }
202    }
203
204    #[async_trait]
205    impl TelemetrySink for VecMutexSink {
206        async fn record(&self, event: TelemetryEvent) -> Result<(), TelemetryError> {
207            self.events.lock().unwrap().push(event);
208            Ok(())
209        }
210
211        async fn query(
212            &self,
213            filter: &TelemetryFilter,
214        ) -> Result<Vec<TelemetryEvent>, TelemetryError> {
215            let events = self.events.lock().unwrap();
216            Ok(events
217                .iter()
218                .rev()
219                .filter(|e| filter.matches(e))
220                .cloned()
221                .collect())
222        }
223    }
224
225    /// Same scenario — different sink. Both impls must produce the
226    /// same observable result for the same input.
227    async fn exercise(sink: &dyn TelemetrySink) -> Vec<TelemetryEvent> {
228        sink.record(
229            TelemetryEvent::new(TelemetryKind::DidcommInbound)
230                .with_mediator("did:m:A")
231                .with_sender("did:peer:alice"),
232        )
233        .await
234        .unwrap();
235        sink.record(
236            TelemetryEvent::new(TelemetryKind::ServicesDidcommUpdate).with_mediator("did:m:B"),
237        )
238        .await
239        .unwrap();
240        sink.record(
241            TelemetryEvent::new(TelemetryKind::DidcommInbound)
242                .with_mediator("did:m:B")
243                .with_sender("did:peer:alice"),
244        )
245        .await
246        .unwrap();
247        sink.query(&TelemetryFilter::new().kind(TelemetryKind::DidcommInbound))
248            .await
249            .unwrap()
250    }
251
252    #[tokio::test]
253    async fn ring_buffer_and_vec_mutex_produce_same_observable_results() {
254        let ring = RingBufferTelemetry::new();
255        let stub = VecMutexSink::new();
256        let from_ring = exercise(&ring).await;
257        let from_stub = exercise(&stub).await;
258        // Both impls must surface the two DidcommInbound events
259        // (filtered out of the three total recorded), in
260        // newest-first order.
261        assert_eq!(from_ring.len(), 2);
262        assert_eq!(from_stub.len(), 2);
263        for (a, b) in from_ring.iter().zip(from_stub.iter()) {
264            assert_eq!(a.kind, b.kind);
265            assert_eq!(a.mediator_did, b.mediator_did);
266            assert_eq!(a.sender_did, b.sender_did);
267        }
268        // Order: newest first.
269        assert_eq!(from_ring[0].mediator_did.as_deref(), Some("did:m:B"));
270        assert_eq!(from_stub[0].mediator_did.as_deref(), Some("did:m:B"));
271    }
272
273    #[tokio::test]
274    async fn arc_dyn_dispatch_swap_at_runtime() {
275        let sinks: Vec<SharedTelemetrySink> = vec![
276            Arc::new(RingBufferTelemetry::new()),
277            Arc::new(VecMutexSink::new()),
278        ];
279        for sink in &sinks {
280            sink.record(TelemetryEvent::new(TelemetryKind::DidcommInbound))
281                .await
282                .unwrap();
283            let out = sink.query(&TelemetryFilter::new()).await.unwrap();
284            assert_eq!(out.len(), 1);
285        }
286    }
287}
vti_common/telemetry/mod.rs

vti_common/telemetry/
mod.rs
