photon_backend_nats/
config.rs

1//! Builder and resolved configuration for the NATS storage adapter.
2
3use std::time::Duration;
4
5use photon_backend::{PhotonError, Result, TransportCrypto};
6
7use crate::retention::retention_from_env;
8use crate::replicas::replicas_from_env;
9
10/// Environment variable for NATS server URL.
11pub const URL_ENV: &str = "PHOTON_NATS_URL";
12
13/// Environment variable for `JetStream` stream name.
14pub const STREAM_ENV: &str = "PHOTON_NATS_STREAM";
15
16/// How durable replay and checkpoints map to `JetStream`.
17#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
18pub enum ReplayCursor {
19    /// Broker stream sequence on publish ack; `ByStartSequence` replay.
20    #[default]
21    StreamSeq,
22    /// Live tail only; checkpoints no-op; maximum publish throughput.
23    TailOnly,
24}
25
26/// Resolved NATS adapter settings (no env lookups at append time).
27#[derive(Clone)]
28pub struct NatsConfig {
29    /// NATS server URL(s).
30    pub url: String,
31    /// `JetStream` stream name.
32    pub stream_name: String,
33    /// Stream max age / replay window.
34    pub retention: Duration,
35    /// `JetStream` stream replica count.
36    pub replicas: usize,
37    /// Payload envelope crypto.
38    pub crypto: TransportCrypto,
39    /// Replay and checkpoint semantics.
40    pub replay_cursor: ReplayCursor,
41    /// Await `JetStream` publish ack per message.
42    pub sync_ack: bool,
43    /// Max concurrent in-flight publishes.
44    pub max_inflight: u32,
45    /// Independent `JetStream` streams for ingress sharding (`1` = legacy single stream).
46    pub stream_shards: u32,
47}
48
49impl NatsConfig {
50    /// Replica count applied when creating shard streams.
51    #[must_use]
52    pub const fn effective_replicas(&self) -> usize {
53        if self.stream_shards > 1 {
54            1
55        } else {
56            self.replicas
57        }
58    }
59
60    /// Whether publishes route across multiple `JetStream` streams.
61    #[must_use]
62    pub const fn is_sharded(&self) -> bool {
63        self.stream_shards > 1
64    }
65}
66
67/// Environment variable for replay cursor mode.
68pub const REPLAY_CURSOR_ENV: &str = "PHOTON_NATS_REPLAY_CURSOR";
69
70/// Environment variable for synchronous publish ack (`1` / `0`).
71pub const SYNC_ACK_ENV: &str = "PHOTON_NATS_SYNC_ACK";
72
73/// Environment variable for max in-flight publishes per port.
74pub const MAX_INFLIGHT_ENV: &str = "PHOTON_NATS_MAX_INFLIGHT";
75
76/// Builder for [`super::port::NatsStoragePort`].
77///
78/// **Configuration lives here.** Set builder methods explicitly; unset fields fall back to
79/// `PHOTON_NATS_*` environment variables via [`from_env_defaults`](Self::from_env_defaults).
80///
81/// Use the **same** builder settings on every Mode 2 publisher and worker binary that shares a
82/// cluster. Getting started:
83/// [Mode 2](https://docs.rs/uf-photon/latest/photon/#mode-2--brokered-publisher--worker-binaries).
84///
85/// # Options
86///
87/// | Method / env | Default | Purpose |
88/// |--------------|---------|---------|
89/// | [`.url`](Self::url) / [`URL_ENV`] | **required** | NATS server URL(s). |
90/// | [`.stream_name`](Self::stream_name) / [`STREAM_ENV`] | `photon` | `JetStream` stream name. |
91/// | [`.retention`](Self::retention) / [`RETENTION_ENV`](crate::retention::RETENTION_ENV) | `15m` | Stream max age. |
92/// | [`.replicas`](Self::replicas) / [`REPLICAS_ENV`](crate::replicas::REPLICAS_ENV) | `1` | Stream replica count. |
93/// | [`.replay_cursor`](Self::replay_cursor) / [`REPLAY_CURSOR_ENV`] | `stream_seq` | [`ReplayCursor::StreamSeq`] or [`ReplayCursor::TailOnly`]. |
94/// | [`.sync_ack`](Self::sync_ack) / [`SYNC_ACK_ENV`] | `1` | Await `JetStream` publish ack (`0` = firehose). |
95/// | [`.max_inflight`](Self::max_inflight) / [`MAX_INFLIGHT_ENV`] | `1` / `256` | Concurrent in-flight publishes (`256` when `sync_ack` off). |
96/// | [`.stream_shards`](Self::stream_shards) / [`STREAM_SHARDS_ENV`](crate::stream_shard::STREAM_SHARDS_ENV) | `1` | `JetStream` stream shard count (`K>1` → `photon-0..K-1`). |
97///
98/// Set `.stream_shards(K)` consistently across embedded hosts (builder-first; not tied to publisher count).
99///
100/// # Examples
101///
102/// ## Publisher binary
103///
104/// Publish only — skip `start_executor` unless this process also runs handlers.
105///
106/// ```rust,ignore
107/// use std::sync::Arc;
108///
109/// use photon_backend_nats::{NatsStoragePort, ReplayCursor};
110/// use photon_runtime::Photon;
111///
112/// # async fn boot_publisher() -> photon_backend::Result<()> {
113/// let port = Arc::new(
114///     NatsStoragePort::builder()
115///         .from_env_defaults()
116///         .replay_cursor(ReplayCursor::StreamSeq)
117///         .sync_ack(true)
118///         .build()
119///         .await?,
120/// );
121/// let photon = Photon::builder()
122///     .storage_port(port)
123///     .auto_registry()
124///     .build()?;
125/// // EventType { … }.publish_on(&photon).await?;
126/// # let _ = photon;
127/// # Ok(())
128/// # }
129/// ```
130///
131/// ## Worker binary
132///
133/// Same port wiring as the publisher, plus `#[subscribe]` handlers and `start_executor`.
134/// Start workers before publishers.
135///
136/// ```rust,ignore
137/// use std::sync::Arc;
138///
139/// use photon_backend_nats::{NatsStoragePort, ReplayCursor};
140/// use photon_core::JsonIdentityFactory;
141/// use photon_runtime::Photon;
142///
143/// # async fn boot_worker() -> photon_backend::Result<()> {
144/// let port = Arc::new(
145///     NatsStoragePort::builder()
146///         .from_env_defaults()
147///         .replay_cursor(ReplayCursor::StreamSeq)
148///         .sync_ack(true)
149///         .build()
150///         .await?,
151/// );
152/// let photon = Photon::builder()
153///     .storage_port(port)
154///     .auto_registry()
155///     .build()?;
156/// photon.start_executor(Arc::new(JsonIdentityFactory))?;
157/// # let _ = photon;
158/// # Ok(())
159/// # }
160/// ```
161///
162/// See also: [`NatsConfig`] (resolved snapshot), crate-level topic mapping in [`crate`](index.html).
163#[derive(Default)]
164pub struct NatsStoragePortBuilder {
165    url: Option<String>,
166    stream_name: Option<String>,
167    retention: Option<Duration>,
168    replicas: Option<usize>,
169    crypto: Option<TransportCrypto>,
170    replay_cursor: Option<ReplayCursor>,
171    sync_ack: Option<bool>,
172    max_inflight: Option<u32>,
173    stream_shards: Option<u32>,
174}
175
176impl NatsStoragePortBuilder {
177    /// Empty builder; set fields or call [`Self::from_env_defaults`].
178    #[must_use]
179    pub fn new() -> Self {
180        Self::default()
181    }
182
183    /// Fill unset fields from `PHOTON_NATS_*` environment variables.
184    #[must_use]
185    pub fn from_env_defaults(mut self) -> Self {
186        if self.url.is_none() {
187            self.url = std::env::var(URL_ENV).ok();
188        }
189        if self.stream_name.is_none() {
190            self.stream_name = Some(
191                std::env::var(STREAM_ENV).unwrap_or_else(|_| "photon".into()),
192            );
193        }
194        if self.retention.is_none() {
195            self.retention = Some(retention_from_env());
196        }
197        if self.replicas.is_none() {
198            self.replicas = Some(replicas_from_env());
199        }
200        if self.replay_cursor.is_none() {
201            self.replay_cursor = Some(replay_cursor_from_env());
202        }
203        if self.sync_ack.is_none() {
204            self.sync_ack = Some(sync_ack_from_env());
205        }
206        if self.max_inflight.is_none() {
207            self.max_inflight = Some(max_inflight_from_env(self.sync_ack.unwrap_or(true)));
208        }
209        if self.stream_shards.is_none() {
210            self.stream_shards = Some(crate::stream_shard::stream_shards_from_env());
211        }
212        self
213    }
214
215    /// NATS server URL.
216    #[must_use]
217    pub fn url(mut self, url: impl Into<String>) -> Self {
218        self.url = Some(url.into());
219        self
220    }
221
222    /// `JetStream` stream name.
223    #[must_use]
224    pub fn stream_name(mut self, name: impl Into<String>) -> Self {
225        self.stream_name = Some(name.into());
226        self
227    }
228
229    /// Stream retention duration.
230    #[must_use]
231    pub const fn retention(mut self, retention: Duration) -> Self {
232        self.retention = Some(retention);
233        self
234    }
235
236    /// Stream replica count.
237    #[must_use]
238    pub const fn replicas(mut self, replicas: usize) -> Self {
239        self.replicas = Some(replicas);
240        self
241    }
242
243    /// Transport crypto for publish path.
244    #[must_use]
245    pub fn crypto(mut self, crypto: TransportCrypto) -> Self {
246        self.crypto = Some(crypto);
247        self
248    }
249
250    /// Replay cursor mode.
251    #[must_use]
252    pub const fn replay_cursor(mut self, cursor: ReplayCursor) -> Self {
253        self.replay_cursor = Some(cursor);
254        self
255    }
256
257    /// Whether to await publish ack.
258    #[must_use]
259    pub const fn sync_ack(mut self, wait: bool) -> Self {
260        self.sync_ack = Some(wait);
261        self
262    }
263
264    /// Pipeline depth for concurrent publishes.
265    #[must_use]
266    pub fn max_inflight(mut self, n: u32) -> Self {
267        self.max_inflight = Some(n.max(1));
268        self
269    }
270
271    /// `JetStream` stream shard count for ingress scaling (`1` = single shared stream).
272    #[must_use]
273    pub fn stream_shards(mut self, n: u32) -> Self {
274        self.stream_shards = Some(n.clamp(1, crate::stream_shard::MAX_STREAM_SHARDS));
275        self
276    }
277
278    /// Resolve configuration.
279    ///
280    /// # Errors
281    ///
282    /// Returns an error when required fields (URL) are missing.
283    pub fn resolve(self) -> Result<NatsConfig> {
284        let builder = self.from_env_defaults();
285        let url = builder.url.ok_or_else(|| {
286            PhotonError::Internal(format!("{URL_ENV} not set for nats storage adapter"))
287        })?;
288        Ok(NatsConfig {
289            url,
290            stream_name: builder
291                .stream_name
292                .unwrap_or_else(|| "photon".into()),
293            retention: builder.retention.unwrap_or_else(retention_from_env),
294            replicas: builder.replicas.unwrap_or_else(replicas_from_env),
295            crypto: match builder.crypto {
296                Some(c) => c,
297                None => TransportCrypto::from_env()?,
298            },
299            replay_cursor: builder
300                .replay_cursor
301                .unwrap_or(ReplayCursor::StreamSeq),
302            sync_ack: builder.sync_ack.unwrap_or(true),
303            max_inflight: builder
304                .max_inflight
305                .unwrap_or_else(|| max_inflight_from_env(builder.sync_ack.unwrap_or(true))),
306            stream_shards: builder
307                .stream_shards
308                .unwrap_or_else(crate::stream_shard::stream_shards_from_env),
309        })
310    }
311}
312
313fn replay_cursor_from_env() -> ReplayCursor {
314    match std::env::var(REPLAY_CURSOR_ENV)
315        .unwrap_or_else(|_| "stream_seq".into())
316        .to_ascii_lowercase()
317        .as_str()
318    {
319        "tail_only" | "tail" | "none" => ReplayCursor::TailOnly,
320        _ => ReplayCursor::StreamSeq,
321    }
322}
323
324fn sync_ack_from_env() -> bool {
325    !matches!(
326        std::env::var(SYNC_ACK_ENV)
327            .unwrap_or_else(|_| "1".into())
328            .as_str(),
329        "0" | "false" | "off" | "no"
330    )
331}
332
333fn max_inflight_from_env(sync_ack: bool) -> u32 {
334    if let Ok(raw) = std::env::var(MAX_INFLIGHT_ENV) {
335        if let Ok(n) = raw.parse::<u32>() {
336            return n.max(1);
337        }
338    }
339    if sync_ack { 1 } else { 256 }
340}
341
342#[cfg(test)]
343mod tests {
344    use super::*;
345
346    #[test]
347    fn replay_cursor_parses_tail_only() {
348        std::env::set_var(REPLAY_CURSOR_ENV, "tail_only");
349        assert_eq!(replay_cursor_from_env(), ReplayCursor::TailOnly);
350        std::env::remove_var(REPLAY_CURSOR_ENV);
351    }
352}
photon_backend_nats/config.rs

photon_backend_nats/
config.rs
