photon_backend_kafka/
config.rs

1//! Builder and resolved configuration for the Kafka storage adapter.
2
3use std::time::Duration;
4
5use photon_backend::{BrokerTransportSecurity, PhotonError, Result, TransportCrypto};
6
7use crate::replicas::replicas_from_env;
8use crate::retention::retention_from_env;
9
10/// Environment variable for Kafka bootstrap brokers.
11pub const BROKERS_ENV: &str = "PHOTON_KAFKA_BROKERS";
12
13/// Environment variable for topic name prefix.
14pub const PREFIX_ENV: &str = "PHOTON_KAFKA_TOPIC_PREFIX";
15
16/// How durable replay and checkpoints map to Kafka offsets.
17#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
18pub enum ReplayCursor {
19    /// Broker offset+1 on publish ack; seek replay by seq.
20    #[default]
21    StreamSeq,
22    /// Live tail only; checkpoints no-op; maximum publish throughput.
23    TailOnly,
24}
25
26/// Resolved Kafka adapter settings (no env lookups at append time).
27#[derive(Clone)]
28pub struct KafkaConfig {
29    /// Kafka bootstrap servers (comma-separated).
30    pub brokers: String,
31    /// Topic prefix for Photon events.
32    pub topic_prefix: String,
33    /// Topic retention / replay window.
34    pub retention: Duration,
35    /// Topic replication factor.
36    pub replicas: i32,
37    /// Payload envelope crypto.
38    pub crypto: TransportCrypto,
39    /// Replay and checkpoint semantics.
40    pub replay_cursor: ReplayCursor,
41    /// Await produce ack per message.
42    pub sync_ack: bool,
43    /// Max concurrent in-flight publishes.
44    pub max_inflight: u32,
45    /// Independent Kafka topics for ingress sharding (`1` = legacy single layout).
46    pub topic_shards: u32,
47    /// Plaintext vs TLS connect policy.
48    pub transport_security: BrokerTransportSecurity,
49}
50
51impl KafkaConfig {
52    /// Effective replication factor when creating shard topics.
53    #[must_use]
54    pub const fn effective_replicas(&self) -> i32 {
55        if self.topic_shards > 1 {
56            1
57        } else {
58            self.replicas
59        }
60    }
61
62    /// Whether publishes route across multiple topic shards.
63    #[must_use]
64    pub const fn is_sharded(&self) -> bool {
65        self.topic_shards > 1
66    }
67
68    /// Compact checkpoint topic name.
69    #[must_use]
70    pub fn checkpoint_topic(&self) -> String {
71        format!("{}-checkpoints", self.topic_prefix)
72    }
73}
74
75/// Environment variable for replay cursor mode.
76pub const REPLAY_CURSOR_ENV: &str = "PHOTON_KAFKA_REPLAY_CURSOR";
77
78/// Environment variable for synchronous publish ack (`1` / `0`).
79pub const SYNC_ACK_ENV: &str = "PHOTON_KAFKA_SYNC_ACK";
80
81/// Environment variable for max in-flight publishes per port.
82pub const MAX_INFLIGHT_ENV: &str = "PHOTON_KAFKA_MAX_INFLIGHT";
83
84/// Builder for [`super::port::KafkaStoragePort`].
85///
86/// **Configuration lives here.** Set builder methods explicitly; unset fields fall back to
87/// `PHOTON_KAFKA_*` environment variables via [`from_env_defaults`](Self::from_env_defaults).
88///
89/// Use the **same** builder settings on every Brokered publisher and worker binary that shares a
90/// cluster. Getting started:
91/// [Brokered](https://docs.rs/uf-photon/latest/photon/#brokered-publisher--worker-binaries).
92/// Teach the brokered path once with the NATS examples (`nats_worker` / `nats_publisher`); swap
93/// this builder for Kafka in production.
94///
95/// # Options
96///
97/// | Method / env | Default | Purpose |
98/// |--------------|---------|---------|
99/// | [`.brokers`](Self::brokers) / [`BROKERS_ENV`] | **required** | Kafka bootstrap servers (comma-separated). |
100/// | [`.topic_prefix`](Self::topic_prefix) / [`PREFIX_ENV`] | `photon` | Topic name prefix. |
101/// | [`.retention`](Self::retention) / [`RETENTION_ENV`](crate::retention::RETENTION_ENV) | `15m` | Desired `retention.ms` (not applied at auto-create; see note). |
102/// | [`.replicas`](Self::replicas) / [`REPLICAS_ENV`](crate::replicas::REPLICAS_ENV) | `1` | Topic replication factor. |
103/// | [`.replay_cursor`](Self::replay_cursor) / [`REPLAY_CURSOR_ENV`] | `stream_seq` | [`ReplayCursor::StreamSeq`] or [`ReplayCursor::TailOnly`]. |
104/// | [`.sync_ack`](Self::sync_ack) / [`SYNC_ACK_ENV`] | `1` | Await produce ack (`0` = firehose). |
105/// | [`.max_inflight`](Self::max_inflight) / [`MAX_INFLIGHT_ENV`] | `1` / `256` | Concurrent in-flight publishes (`256` when `sync_ack` off). |
106/// | [`.topic_shards`](Self::topic_shards) / [`TOPIC_SHARDS_ENV`](crate::stream_shard::TOPIC_SHARDS_ENV) | `1` | Ingress shard count (`K>1` → `photon-s.{i}.{topic}`). |
107/// | [`.require_tls`](Self::require_tls) / [`PHOTON_ALLOW_INSECURE_BROKER`](photon_backend::ALLOW_INSECURE_BROKER_ENV) | require TLS | Plaintext brokers need `.allow_insecure_plaintext()`. |
108///
109/// **Retention note:** `rskafka` 0.6 cannot set topic configs on create. Photon logs a one-time
110/// warning and exposes [`crate::topic::retention_ms`] for operators who pre-create topics or set
111/// broker `log.retention.ms`. See repository `SECURITY.md`.
112///
113/// # Examples
114///
115/// ## Publisher binary
116///
117/// Publish only — skip `start_executor` unless this process also runs handlers.
118///
119/// ```rust,ignore
120/// use std::sync::Arc;
121///
122/// use photon_backend_kafka::{KafkaStoragePort, ReplayCursor};
123/// use photon_runtime::Photon;
124///
125/// # async fn boot_publisher() -> photon_backend::Result<()> {
126/// let port = Arc::new(
127///     KafkaStoragePort::builder()
128///         .from_env_defaults()
129///         .replay_cursor(ReplayCursor::StreamSeq)
130///         .sync_ack(true)
131///         .build()
132///         .await?,
133/// );
134/// let photon = Photon::builder()
135///     .storage_port(port)
136///     .auto_registry()
137///     .build()?;
138/// // EventType { … }.publish_on(&photon).await?;
139/// # let _ = photon;
140/// # Ok(())
141/// # }
142/// ```
143///
144/// ## Worker binary
145///
146/// Same port wiring as the publisher, plus `#[subscribe]` handlers and `start_executor`.
147/// Start workers before publishers.
148///
149/// ```rust,ignore
150/// use std::sync::Arc;
151///
152/// use photon_backend_kafka::{KafkaStoragePort, ReplayCursor};
153/// use photon_core::JsonIdentityFactory;
154/// use photon_runtime::Photon;
155///
156/// # async fn boot_worker() -> photon_backend::Result<()> {
157/// let port = Arc::new(
158///     KafkaStoragePort::builder()
159///         .from_env_defaults()
160///         .replay_cursor(ReplayCursor::StreamSeq)
161///         .sync_ack(true)
162///         .build()
163///         .await?,
164/// );
165/// let photon = Photon::builder()
166///     .storage_port(port)
167///     .auto_registry()
168///     .build()?;
169/// photon.start_executor(Arc::new(JsonIdentityFactory))?;
170/// # let _ = photon;
171/// # Ok(())
172/// # }
173/// ```
174///
175/// See also: [`KafkaConfig`] (resolved snapshot), crate-level topic mapping in [`crate`](index.html).
176#[derive(Default)]
177pub struct KafkaStoragePortBuilder {
178    brokers: Option<String>,
179    topic_prefix: Option<String>,
180    retention: Option<Duration>,
181    replicas: Option<i32>,
182    crypto: Option<TransportCrypto>,
183    replay_cursor: Option<ReplayCursor>,
184    sync_ack: Option<bool>,
185    max_inflight: Option<u32>,
186    topic_shards: Option<u32>,
187    transport_security: Option<BrokerTransportSecurity>,
188}
189
190impl KafkaStoragePortBuilder {
191    /// Empty builder; set fields or call [`Self::from_env_defaults`].
192    #[must_use]
193    pub fn new() -> Self {
194        Self::default()
195    }
196
197    /// Fill unset fields from `PHOTON_KAFKA_*` environment variables.
198    #[must_use]
199    pub fn from_env_defaults(mut self) -> Self {
200        if self.brokers.is_none() {
201            self.brokers = std::env::var(BROKERS_ENV).ok();
202        }
203        if self.topic_prefix.is_none() {
204            self.topic_prefix = Some(std::env::var(PREFIX_ENV).unwrap_or_else(|_| "photon".into()));
205        }
206        if self.retention.is_none() {
207            self.retention = Some(retention_from_env());
208        }
209        if self.replicas.is_none() {
210            self.replicas = Some(replicas_from_env());
211        }
212        if self.replay_cursor.is_none() {
213            self.replay_cursor = Some(replay_cursor_from_env());
214        }
215        if self.sync_ack.is_none() {
216            self.sync_ack = Some(sync_ack_from_env());
217        }
218        if self.max_inflight.is_none() {
219            self.max_inflight = Some(max_inflight_from_env(self.sync_ack.unwrap_or(true)));
220        }
221        if self.topic_shards.is_none() {
222            self.topic_shards = Some(crate::stream_shard::topic_shards_from_env());
223        }
224        self
225    }
226
227    /// Kafka bootstrap brokers.
228    #[must_use]
229    pub fn brokers(mut self, brokers: impl Into<String>) -> Self {
230        self.brokers = Some(brokers.into());
231        self
232    }
233
234    /// Topic prefix.
235    #[must_use]
236    pub fn topic_prefix(mut self, prefix: impl Into<String>) -> Self {
237        self.topic_prefix = Some(prefix.into());
238        self
239    }
240
241    /// Topic retention duration.
242    #[must_use]
243    pub const fn retention(mut self, retention: Duration) -> Self {
244        self.retention = Some(retention);
245        self
246    }
247
248    /// Topic replication factor.
249    #[must_use]
250    pub const fn replicas(mut self, replicas: i32) -> Self {
251        self.replicas = Some(replicas);
252        self
253    }
254
255    /// Transport crypto for publish path.
256    #[must_use]
257    pub fn crypto(mut self, crypto: TransportCrypto) -> Self {
258        self.crypto = Some(crypto);
259        self
260    }
261
262    /// Replay cursor mode.
263    #[must_use]
264    pub const fn replay_cursor(mut self, cursor: ReplayCursor) -> Self {
265        self.replay_cursor = Some(cursor);
266        self
267    }
268
269    /// Whether to await produce ack.
270    #[must_use]
271    pub const fn sync_ack(mut self, wait: bool) -> Self {
272        self.sync_ack = Some(wait);
273        self
274    }
275
276    /// Pipeline depth for concurrent publishes.
277    #[must_use]
278    pub fn max_inflight(mut self, n: u32) -> Self {
279        self.max_inflight = Some(n.max(1));
280        self
281    }
282
283    /// Topic shard count for ingress scaling (`1` = single shared layout).
284    #[must_use]
285    pub fn topic_shards(mut self, n: u32) -> Self {
286        self.topic_shards = Some(n.clamp(1, crate::stream_shard::MAX_TOPIC_SHARDS));
287        self
288    }
289
290    /// Allow plaintext broker endpoints (development/CI only).
291    #[must_use]
292    pub const fn allow_insecure_plaintext(mut self) -> Self {
293        self.transport_security = Some(BrokerTransportSecurity::AllowInsecurePlaintext);
294        self
295    }
296
297    /// Require TLS-oriented broker endpoints.
298    #[must_use]
299    pub const fn require_tls(mut self) -> Self {
300        self.transport_security = Some(BrokerTransportSecurity::RequireTls);
301        self
302    }
303
304    /// Resolve configuration.
305    ///
306    /// # Errors
307    ///
308    /// Returns an error when required fields (brokers) are missing.
309    pub fn resolve(self) -> Result<KafkaConfig> {
310        let builder = self.from_env_defaults();
311        let brokers = builder.brokers.ok_or_else(|| {
312            PhotonError::Internal(format!("{BROKERS_ENV} not set for kafka storage adapter"))
313        })?;
314        Ok(KafkaConfig {
315            brokers,
316            topic_prefix: builder.topic_prefix.unwrap_or_else(|| "photon".into()),
317            retention: builder.retention.unwrap_or_else(retention_from_env),
318            replicas: builder.replicas.unwrap_or_else(replicas_from_env),
319            crypto: match builder.crypto {
320                Some(c) => c,
321                None => TransportCrypto::from_env()?,
322            },
323            replay_cursor: builder.replay_cursor.unwrap_or(ReplayCursor::StreamSeq),
324            sync_ack: builder.sync_ack.unwrap_or(true),
325            max_inflight: builder
326                .max_inflight
327                .unwrap_or_else(|| max_inflight_from_env(builder.sync_ack.unwrap_or(true))),
328            topic_shards: builder
329                .topic_shards
330                .unwrap_or_else(crate::stream_shard::topic_shards_from_env),
331            transport_security: builder
332                .transport_security
333                .unwrap_or_else(BrokerTransportSecurity::from_env),
334        })
335    }
336}
337
338fn replay_cursor_from_env() -> ReplayCursor {
339    match std::env::var(REPLAY_CURSOR_ENV)
340        .unwrap_or_else(|_| "stream_seq".into())
341        .to_ascii_lowercase()
342        .as_str()
343    {
344        "tail_only" | "tail" | "none" => ReplayCursor::TailOnly,
345        _ => ReplayCursor::StreamSeq,
346    }
347}
348
349fn sync_ack_from_env() -> bool {
350    !matches!(
351        std::env::var(SYNC_ACK_ENV)
352            .unwrap_or_else(|_| "1".into())
353            .as_str(),
354        "0" | "false" | "off" | "no"
355    )
356}
357
358fn max_inflight_from_env(sync_ack: bool) -> u32 {
359    if let Ok(raw) = std::env::var(MAX_INFLIGHT_ENV) {
360        if let Ok(n) = raw.parse::<u32>() {
361            return n.max(1);
362        }
363    }
364    if sync_ack {
365        1
366    } else {
367        256
368    }
369}
370
371#[cfg(test)]
372mod tests {
373    use super::*;
374
375    #[test]
376    fn replay_cursor_parses_tail_only() {
377        std::env::set_var(REPLAY_CURSOR_ENV, "tail_only");
378        assert_eq!(replay_cursor_from_env(), ReplayCursor::TailOnly);
379        std::env::remove_var(REPLAY_CURSOR_ENV);
380    }
381}
photon_backend_kafka/config.rs

photon_backend_kafka/
config.rs
