photon_backend/instrumentation/
backend.rs

1//! L0 [`PhotonBackend`] decorator for publish telemetry.
2
3use std::pin::Pin;
4use std::sync::Arc;
5
6use async_trait::async_trait;
7use futures::stream::Stream;
8use serde_json::Value;
9
10use crate::backend::PhotonBackend;
11use crate::error::Result;
12use crate::models::Event;
13use crate::registry::TopicRegistry;
14
15use super::metrics;
16
17/// Wraps an inner backend and emits `photon_publishes` on successful publish.
18pub struct InstrumentedPhotonBackend {
19    inner: Arc<dyn PhotonBackend>,
20    backend_label: &'static str,
21}
22
23impl InstrumentedPhotonBackend {
24    /// Wrap `inner`, using its [`PhotonBackend::telemetry_label`] for metrics.
25    pub fn new(inner: Arc<dyn PhotonBackend>) -> Self {
26        let backend_label = inner.telemetry_label();
27        Self {
28            inner,
29            backend_label,
30        }
31    }
32}
33
34/// Wrap `inner` with [`InstrumentedPhotonBackend`].
35pub fn wrap_backend(inner: Arc<dyn PhotonBackend>) -> Arc<dyn PhotonBackend> {
36    Arc::new(InstrumentedPhotonBackend::new(inner))
37}
38
39#[async_trait]
40impl PhotonBackend for InstrumentedPhotonBackend {
41    fn telemetry_label(&self) -> &'static str {
42        self.backend_label
43    }
44
45    #[tracing::instrument(
46        name = "photon.publish",
47        skip(self, actor_json, payload_json),
48        fields(backend = self.backend_label, topic = topic_name, topic_key = ?topic_key)
49    )]
50    async fn publish(
51        &self,
52        topic_name: &str,
53        topic_key: Option<&str>,
54        actor_json: Value,
55        payload_json: Value,
56    ) -> Result<String> {
57        match self
58            .inner
59            .publish(topic_name, topic_key, actor_json, payload_json)
60            .await
61        {
62            Ok(id) => {
63                metrics::record_publish(topic_name, self.backend_label);
64                Ok(id)
65            }
66            Err(e) => {
67                metrics::record_publish_error(topic_name, self.backend_label);
68                tracing::warn!(error = %e, "publish failed");
69                Err(e)
70            }
71        }
72    }
73
74    fn subscribe(
75        &self,
76        topic_name: String,
77        topic_key_filter: Option<String>,
78        after_seq: Option<i64>,
79    ) -> Pin<Box<dyn Stream<Item = Result<Event>> + Send>> {
80        self.inner
81            .subscribe(topic_name, topic_key_filter, after_seq)
82    }
83
84    async fn get_event(&self, event_id: &str) -> Result<Option<Event>> {
85        self.inner.get_event(event_id).await
86    }
87
88    async fn list_by_topic(
89        &self,
90        topic_name: &str,
91        topic_key: Option<&str>,
92        after_seq: Option<i64>,
93        limit: usize,
94    ) -> Result<Vec<Event>> {
95        self.inner
96            .list_by_topic(topic_name, topic_key, after_seq, limit)
97            .await
98    }
99
100    async fn list_recent(&self, limit: usize) -> Result<Vec<Event>> {
101        self.inner.list_recent(limit).await
102    }
103
104    fn registry(&self) -> &TopicRegistry {
105        self.inner.registry()
106    }
107
108    async fn get_checkpoint_seq(
109        &self,
110        subscription_name: &str,
111        topic_name: &str,
112        topic_key: Option<&str>,
113    ) -> Result<Option<i64>> {
114        self.inner
115            .get_checkpoint_seq(subscription_name, topic_name, topic_key)
116            .await
117    }
118
119    async fn set_checkpoint(
120        &self,
121        subscription_name: &str,
122        topic_name: &str,
123        topic_key: Option<&str>,
124        last_seq: i64,
125    ) -> Result<()> {
126        self.inner
127            .set_checkpoint(subscription_name, topic_name, topic_key, last_seq)
128            .await
129    }
130}
photon_backend/instrumentation/backend.rs

photon_backend/instrumentation/
backend.rs
