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    fn registry(&self) -> &TopicRegistry {
89        self.inner.registry()
90    }
91
92    async fn get_checkpoint_seq(
93        &self,
94        subscription_name: &str,
95        topic_name: &str,
96        topic_key: Option<&str>,
97    ) -> Result<Option<i64>> {
98        self.inner
99            .get_checkpoint_seq(subscription_name, topic_name, topic_key)
100            .await
101    }
102
103    async fn set_checkpoint(
104        &self,
105        subscription_name: &str,
106        topic_name: &str,
107        topic_key: Option<&str>,
108        last_seq: i64,
109    ) -> Result<()> {
110        self.inner
111            .set_checkpoint(subscription_name, topic_name, topic_key, last_seq)
112            .await
113    }
114}
photon_backend/instrumentation/backend.rs

photon_backend/instrumentation/
backend.rs
