photon_backend/backend/
generic.rs

1//! Single [`PhotonBackend`] implementation wrapping any [`StoragePort`].
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 super::capabilities::BackendCapabilities;
11use super::context::BackendContext;
12use super::photon_backend::PhotonBackend;
13use crate::error::Result;
14use crate::models::Event;
15use crate::publish_routing::resolve_publish_target;
16use crate::registry::TopicRegistry;
17use crate::storage::{InProcStoragePort, StoragePort};
18
19/// Unified backend — all storage adapters install this wrapping their port.
20pub struct GenericPhotonBackend {
21    port: Arc<dyn StoragePort>,
22    registry: TopicRegistry,
23    capabilities: BackendCapabilities,
24}
25
26impl GenericPhotonBackend {
27    /// Wrap an existing storage port and topic registry.
28    #[must_use]
29    pub fn new(port: Arc<dyn StoragePort>, registry: TopicRegistry) -> Self {
30        let capabilities = port.capabilities();
31        Self {
32            port,
33            registry,
34            capabilities,
35        }
36    }
37
38    /// Capabilities for contract tests and host introspection.
39    #[must_use]
40    pub const fn capabilities(&self) -> BackendCapabilities {
41        self.capabilities
42    }
43
44    /// Install the default `mem` tier from builder context.
45    ///
46    /// # Errors
47    ///
48    /// Returns an error if the operation fails.
49    pub fn install_mem(ctx: BackendContext) -> Result<Arc<dyn PhotonBackend>> {
50        let port = Arc::new(InProcStoragePort::new(
51            crate::event::TransportCrypto::from_env()?,
52        ));
53        Ok(Arc::new(Self::new(port, ctx.registry)))
54    }
55
56    /// Install a custom storage port from builder context.
57    ///
58    /// # Errors
59    ///
60    /// Returns an error if the operation fails.
61    pub fn install_with_port(
62        ctx: BackendContext,
63        port: Arc<dyn StoragePort>,
64    ) -> Result<Arc<dyn PhotonBackend>> {
65        Ok(Arc::new(Self::new(port, ctx.registry)))
66    }
67}
68
69#[async_trait]
70impl PhotonBackend for GenericPhotonBackend {
71    fn telemetry_label(&self) -> &'static str {
72        self.capabilities.telemetry_label
73    }
74
75    fn capabilities(&self) -> BackendCapabilities {
76        self.capabilities
77    }
78
79    async fn publish(
80        &self,
81        topic_name: &str,
82        topic_key: Option<&str>,
83        actor_json: Value,
84        payload_json: Value,
85    ) -> Result<String> {
86        let target = resolve_publish_target(
87            &self.registry,
88            topic_name,
89            topic_key,
90            &payload_json,
91        );
92        let event = self
93            .port
94            .append(
95                topic_name,
96                target.topic_key.as_deref(),
97                actor_json,
98                payload_json,
99            )
100            .await?;
101        Ok(event.event_id)
102    }
103
104    fn subscribe(
105        &self,
106        topic_name: String,
107        topic_key_filter: Option<String>,
108        after_seq: Option<i64>,
109    ) -> Pin<Box<dyn Stream<Item = Result<Event>> + Send>> {
110        self.port
111            .subscribe(topic_name, topic_key_filter, after_seq)
112    }
113
114    async fn get_event(&self, event_id: &str) -> Result<Option<Event>> {
115        if !self.capabilities.supports_get_event {
116            return Ok(None);
117        }
118        self.port.get_event(event_id).await
119    }
120
121    fn registry(&self) -> &TopicRegistry {
122        &self.registry
123    }
124
125    async fn get_checkpoint_seq(
126        &self,
127        subscription_name: &str,
128        topic_name: &str,
129        topic_key: Option<&str>,
130    ) -> Result<Option<i64>> {
131        self.port
132            .load_checkpoint(subscription_name, topic_name, topic_key)
133            .await
134    }
135
136    async fn set_checkpoint(
137        &self,
138        subscription_name: &str,
139        topic_name: &str,
140        topic_key: Option<&str>,
141        last_seq: i64,
142    ) -> Result<()> {
143        self.port
144            .commit_checkpoint(subscription_name, topic_name, topic_key, last_seq)
145            .await
146    }
147}
photon_backend/backend/generic.rs

photon_backend/backend/
generic.rs
