photon_backend_fluvio/
port.rs

1//! Fluvio [`StoragePort`] implementation.
2
3use std::pin::Pin;
4use std::sync::Arc;
5
6use async_trait::async_trait;
7use chrono::Utc;
8use futures::stream::Stream;
9use photon_backend::models::Event;
10use photon_backend::{PhotonError, Result, StorageCapabilities, StoragePort};
11use serde_json::Value;
12use uuid::Uuid;
13
14use crate::checkpoint::CheckpointStore;
15use crate::config::{FluvioConfig, FluvioStoragePortBuilder, ReplayCursor, ENDPOINT_ENV};
16use crate::connect::{connect_fluvio, SharedClient};
17use crate::consumer::subscribe_stream;
18use crate::publish::PublishPipeline;
19use crate::stream_shard::{composite_seq, fluvio_topic_for, pick_shard, publish_routing_key};
20use crate::topic::{ensure_checkpoint_topic, ensure_data_topic, warn_replication_settings};
21
22/// Read Fluvio endpoint from the environment.
23///
24/// # Errors
25///
26/// Returns an error when `PHOTON_FLUVIO_ENDPOINT` is unset.
27pub fn fluvio_endpoint_from_env() -> Result<String> {
28    std::env::var(ENDPOINT_ENV).map_err(|_| {
29        PhotonError::Internal(format!("{ENDPOINT_ENV} not set for fluvio storage adapter"))
30    })
31}
32
33/// Fluvio-backed storage port.
34pub struct FluvioStoragePort {
35    client: SharedClient,
36    config: FluvioConfig,
37    pipeline: PublishPipeline,
38    checkpoint_store: CheckpointStore,
39    ensured_topics: Arc<dashmap::DashSet<String>>,
40}
41
42impl FluvioStoragePort {
43    /// Start a builder for explicit host wiring.
44    #[must_use]
45    pub fn builder() -> FluvioStoragePortBuilder {
46        FluvioStoragePortBuilder::new()
47    }
48
49    /// Connect using env (`PHOTON_FLUVIO_*` defaults via builder).
50    ///
51    /// # Errors
52    ///
53    /// Returns an error when env is missing or connection fails.
54    pub async fn from_env() -> Result<Self> {
55        Self::builder().from_env_defaults().build().await
56    }
57
58    /// Resolved adapter configuration.
59    #[must_use]
60    pub const fn config(&self) -> &FluvioConfig {
61        &self.config
62    }
63
64    async fn connect_with_config(config: FluvioConfig) -> Result<Self> {
65        warn_replication_settings(&config);
66        let client = connect_fluvio(&config).await?;
67        ensure_checkpoint_topic(&client, &config).await?;
68        let checkpoint_store = CheckpointStore::connect(Arc::clone(&client), &config).await?;
69        let pipeline = PublishPipeline::new(Arc::clone(&client), &config);
70        Ok(Self {
71            client,
72            config,
73            pipeline,
74            checkpoint_store,
75            ensured_topics: Arc::new(dashmap::DashSet::new()),
76        })
77    }
78
79    async fn ensure_topic_once(&self, topic: &str) -> Result<()> {
80        if self.ensured_topics.contains(topic) {
81            return Ok(());
82        }
83        ensure_data_topic(&self.client, &self.config, topic).await?;
84        self.ensured_topics.insert(topic.to_string());
85        Ok(())
86    }
87}
88
89impl FluvioStoragePortBuilder {
90    /// Connect and return a configured [`FluvioStoragePort`].
91    ///
92    /// # Errors
93    ///
94    /// Returns an error when configuration or connection fails.
95    pub async fn build(self) -> Result<FluvioStoragePort> {
96        let config = self.resolve()?;
97        FluvioStoragePort::connect_with_config(config).await
98    }
99}
100
101#[async_trait]
102impl StoragePort for FluvioStoragePort {
103    fn capabilities(&self) -> StorageCapabilities {
104        StorageCapabilities::broker("fluvio")
105    }
106
107    async fn append(
108        &self,
109        topic_name: &str,
110        topic_key: Option<&str>,
111        actor_json: Value,
112        payload_json: Value,
113    ) -> Result<Event> {
114        let _ = self.config.crypto.encrypt(&actor_json, &payload_json)?;
115        let mut event = Event {
116            event_id: Uuid::new_v4().to_string(),
117            topic_name: topic_name.to_string(),
118            topic_key: topic_key.map(String::from),
119            seq: 0,
120            actor_json,
121            payload_json,
122            created_at: Utc::now(),
123        };
124
125        let routing = publish_routing_key(topic_key, &event.event_id);
126        let shard = pick_shard(&routing, self.config.topic_shards);
127        let fluvio_topic = fluvio_topic_for(&self.config, shard, topic_name);
128        self.ensure_topic_once(&fluvio_topic).await?;
129
130        let offset_seq = self.pipeline.publish(&fluvio_topic, &event).await?;
131
132        if self.config.replay_cursor == ReplayCursor::StreamSeq {
133            if let Some(seq) = offset_seq {
134                event.seq = if self.config.is_sharded() {
135                    composite_seq(shard, u64::try_from(seq.max(0)).unwrap_or(0))
136                } else {
137                    seq
138                };
139            }
140        }
141
142        Ok(event)
143    }
144
145    fn subscribe(
146        &self,
147        topic_name: String,
148        topic_key_filter: Option<String>,
149        after_seq: Option<i64>,
150    ) -> Pin<Box<dyn Stream<Item = Result<Event>> + Send>> {
151        let effective_after = if self.config.replay_cursor == ReplayCursor::TailOnly {
152            None
153        } else {
154            after_seq
155        };
156        subscribe_stream(
157            Arc::clone(&self.client),
158            self.config.clone(),
159            self.checkpoint_store.clone(),
160            Arc::clone(&self.ensured_topics),
161            topic_name,
162            topic_key_filter,
163            effective_after,
164        )
165    }
166
167    async fn get_event(&self, _event_id: &str) -> Result<Option<Event>> {
168        Ok(None)
169    }
170
171    #[allow(clippy::unused_async)] // `StoragePort` async trait; load is an in-memory cache read
172    async fn load_checkpoint(
173        &self,
174        subscription_name: &str,
175        topic_name: &str,
176        topic_key: Option<&str>,
177    ) -> Result<Option<i64>> {
178        if self.config.replay_cursor == ReplayCursor::TailOnly {
179            return Ok(None);
180        }
181        self.checkpoint_store
182            .load(subscription_name, topic_name, topic_key)
183    }
184
185    async fn commit_checkpoint(
186        &self,
187        subscription_name: &str,
188        topic_name: &str,
189        topic_key: Option<&str>,
190        last_seq: i64,
191    ) -> Result<()> {
192        if self.config.replay_cursor == ReplayCursor::TailOnly {
193            return Ok(());
194        }
195        self.checkpoint_store
196            .commit(subscription_name, topic_name, topic_key, last_seq)
197            .await
198    }
199}
photon_backend_fluvio/port.rs

photon_backend_fluvio/
port.rs
