photon_backend_nats/
config.rs

1//! Builder and resolved configuration for the NATS 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 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/// Environment variable for NATS credentials file path (JWT + `NKey` `.creds`).
17pub const CREDS_ENV: &str = "PHOTON_NATS_CREDS";
18
19/// How durable replay and checkpoints map to `JetStream`.
20#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
21pub enum ReplayCursor {
22    /// Broker stream sequence on publish ack; `ByStartSequence` replay.
23    #[default]
24    StreamSeq,
25    /// Live tail only; checkpoints no-op; maximum publish throughput.
26    TailOnly,
27}
28
29/// Resolved NATS adapter settings (no env lookups at append time).
30#[derive(Clone)]
31pub struct NatsConfig {
32    /// NATS server URL(s).
33    pub url: String,
34    /// Optional path to a NATS `.creds` file (JWT + `NKey`). Prefer this over URL userinfo.
35    pub credentials_file: Option<String>,
36    /// `JetStream` stream name.
37    pub stream_name: String,
38    /// Stream max age / replay window.
39    pub retention: Duration,
40    /// `JetStream` stream replica count.
41    pub replicas: usize,
42    /// Payload envelope crypto.
43    pub crypto: TransportCrypto,
44    /// Replay and checkpoint semantics.
45    pub replay_cursor: ReplayCursor,
46    /// Await `JetStream` publish ack per message.
47    pub sync_ack: bool,
48    /// Max concurrent in-flight publishes.
49    pub max_inflight: u32,
50    /// Independent `JetStream` streams for ingress sharding (`1` = legacy single stream).
51    pub stream_shards: u32,
52    /// Plaintext vs TLS connect policy.
53    pub transport_security: BrokerTransportSecurity,
54}
55
56impl NatsConfig {
57    /// Replica count applied when creating shard streams.
58    #[must_use]
59    pub const fn effective_replicas(&self) -> usize {
60        if self.stream_shards > 1 {
61            1
62        } else {
63            self.replicas
64        }
65    }
66
67    /// Whether publishes route across multiple `JetStream` streams.
68    #[must_use]
69    pub const fn is_sharded(&self) -> bool {
70        self.stream_shards > 1
71    }
72}
73
74/// Environment variable for replay cursor mode.
75pub const REPLAY_CURSOR_ENV: &str = "PHOTON_NATS_REPLAY_CURSOR";
76
77/// Environment variable for synchronous publish ack (`1` / `0`).
78pub const SYNC_ACK_ENV: &str = "PHOTON_NATS_SYNC_ACK";
79
80/// Environment variable for max in-flight publishes per port.
81pub const MAX_INFLIGHT_ENV: &str = "PHOTON_NATS_MAX_INFLIGHT";
82
83/// Builder for [`super::port::NatsStoragePort`].
84///
85/// **Configuration lives here.** Set builder methods explicitly; unset fields fall back to
86/// `PHOTON_NATS_*` environment variables via [`from_env_defaults`](Self::from_env_defaults).
87///
88/// Use the **same** builder settings on every Brokered publisher and worker binary that shares a
89/// cluster. Getting started:
90/// [Brokered](https://docs.rs/uf-photon/latest/photon/#brokered-publisher--worker-binaries).
91/// Runnable: `cargo run -p uf-photon --example nats_worker --features runtime,nats` then
92/// `nats_publisher` (see `photon/README.md` § How to run examples).
93///
94/// # Options
95///
96/// | Method / env | Default | Purpose |
97/// |--------------|---------|---------|
98/// | [`.url`](Self::url) / [`URL_ENV`] | **required** | NATS server URL(s). Prefer `tls://` in production. |
99/// | [`.credentials_file`](Self::credentials_file) / [`CREDS_ENV`] | unset | Path to NATS `.creds` (JWT + `NKey`). Prefer over URL userinfo. |
100/// | [`.stream_name`](Self::stream_name) / [`STREAM_ENV`] | `photon` | `JetStream` stream name. |
101/// | [`.retention`](Self::retention) / [`RETENTION_ENV`](crate::retention::RETENTION_ENV) | `15m` | Stream max age (applied at stream create). |
102/// | [`.replicas`](Self::replicas) / [`REPLICAS_ENV`](crate::replicas::REPLICAS_ENV) | `1` | Stream replica count. |
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 `JetStream` publish ack (`0` = firehose). |
105/// | [`.max_inflight`](Self::max_inflight) / [`MAX_INFLIGHT_ENV`] | `1` / `256` | Concurrent in-flight publishes (`256` when `sync_ack` off). |
106/// | [`.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`). |
107/// | [`.require_tls`](Self::require_tls) | require TLS | Plaintext `nats://` needs [`.allow_insecure_plaintext`](Self::allow_insecure_plaintext). |
108///
109/// Set `.stream_shards(K)` consistently across embedded hosts (builder-first; not tied to publisher count).
110///
111/// # Examples
112///
113/// ## Publisher binary (TLS + credentials)
114///
115/// Publish only — skip `start_executor` unless this process also runs handlers.
116///
117/// ```rust,ignore
118/// use std::sync::Arc;
119///
120/// use photon_backend_nats::{NatsStoragePort, ReplayCursor};
121/// use photon_runtime::Photon;
122///
123/// # async fn boot_publisher() -> photon_backend::Result<()> {
124/// let port = Arc::new(
125///     NatsStoragePort::builder()
126///         .url("tls://nats.example:4222")
127///         .credentials_file("/run/secrets/nats.creds")
128///         .require_tls()
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_nats::{NatsStoragePort, 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///     NatsStoragePort::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: [`NatsConfig`] (resolved snapshot), crate-level topic mapping in [`crate`](index.html).
176#[derive(Default)]
177pub struct NatsStoragePortBuilder {
178    url: Option<String>,
179    credentials_file: Option<String>,
180    transport_security: Option<BrokerTransportSecurity>,
181    stream_name: Option<String>,
182    retention: Option<Duration>,
183    replicas: Option<usize>,
184    crypto: Option<TransportCrypto>,
185    replay_cursor: Option<ReplayCursor>,
186    sync_ack: Option<bool>,
187    max_inflight: Option<u32>,
188    stream_shards: Option<u32>,
189}
190
191impl NatsStoragePortBuilder {
192    /// Empty builder; set fields or call [`Self::from_env_defaults`].
193    #[must_use]
194    pub fn new() -> Self {
195        Self::default()
196    }
197
198    /// Fill unset fields from `PHOTON_NATS_*` environment variables.
199    #[must_use]
200    pub fn from_env_defaults(mut self) -> Self {
201        if self.url.is_none() {
202            self.url = std::env::var(URL_ENV).ok();
203        }
204        if self.credentials_file.is_none() {
205            self.credentials_file = std::env::var(CREDS_ENV).ok();
206        }
207        if self.stream_name.is_none() {
208            self.stream_name = Some(std::env::var(STREAM_ENV).unwrap_or_else(|_| "photon".into()));
209        }
210        if self.retention.is_none() {
211            self.retention = Some(retention_from_env());
212        }
213        if self.replicas.is_none() {
214            self.replicas = Some(replicas_from_env());
215        }
216        if self.replay_cursor.is_none() {
217            self.replay_cursor = Some(replay_cursor_from_env());
218        }
219        if self.sync_ack.is_none() {
220            self.sync_ack = Some(sync_ack_from_env());
221        }
222        if self.max_inflight.is_none() {
223            self.max_inflight = Some(max_inflight_from_env(self.sync_ack.unwrap_or(true)));
224        }
225        if self.stream_shards.is_none() {
226            self.stream_shards = Some(crate::stream_shard::stream_shards_from_env());
227        }
228        self
229    }
230
231    /// NATS server URL.
232    #[must_use]
233    pub fn url(mut self, url: impl Into<String>) -> Self {
234        self.url = Some(url.into());
235        self
236    }
237
238    /// Path to a NATS credentials file (JWT + `NKey`). Prefer over embedding secrets in the URL.
239    #[must_use]
240    pub fn credentials_file(mut self, path: impl Into<String>) -> Self {
241        self.credentials_file = Some(path.into());
242        self
243    }
244
245    /// `JetStream` stream name.
246    #[must_use]
247    pub fn stream_name(mut self, name: impl Into<String>) -> Self {
248        self.stream_name = Some(name.into());
249        self
250    }
251
252    /// Stream retention duration.
253    #[must_use]
254    pub const fn retention(mut self, retention: Duration) -> Self {
255        self.retention = Some(retention);
256        self
257    }
258
259    /// Stream replica count.
260    #[must_use]
261    pub const fn replicas(mut self, replicas: usize) -> Self {
262        self.replicas = Some(replicas);
263        self
264    }
265
266    /// Transport crypto for publish path.
267    #[must_use]
268    pub fn crypto(mut self, crypto: TransportCrypto) -> Self {
269        self.crypto = Some(crypto);
270        self
271    }
272
273    /// Replay cursor mode.
274    #[must_use]
275    pub const fn replay_cursor(mut self, cursor: ReplayCursor) -> Self {
276        self.replay_cursor = Some(cursor);
277        self
278    }
279
280    /// Whether to await publish ack.
281    #[must_use]
282    pub const fn sync_ack(mut self, wait: bool) -> Self {
283        self.sync_ack = Some(wait);
284        self
285    }
286
287    /// Pipeline depth for concurrent publishes.
288    #[must_use]
289    pub fn max_inflight(mut self, n: u32) -> Self {
290        self.max_inflight = Some(n.max(1));
291        self
292    }
293
294    /// `JetStream` stream shard count for ingress scaling (`1` = single shared stream).
295    #[must_use]
296    pub fn stream_shards(mut self, n: u32) -> Self {
297        self.stream_shards = Some(n.clamp(1, crate::stream_shard::MAX_STREAM_SHARDS));
298        self
299    }
300
301    /// Allow plaintext `nats://` endpoints (development/CI only).
302    #[must_use]
303    pub const fn allow_insecure_plaintext(mut self) -> Self {
304        self.transport_security = Some(BrokerTransportSecurity::AllowInsecurePlaintext);
305        self
306    }
307
308    /// Require TLS-oriented broker endpoints (default when unset and env opt-in absent).
309    #[must_use]
310    pub const fn require_tls(mut self) -> Self {
311        self.transport_security = Some(BrokerTransportSecurity::RequireTls);
312        self
313    }
314
315    /// Resolve configuration.
316    ///
317    /// # Errors
318    ///
319    /// Returns an error when required fields (URL) are missing.
320    pub fn resolve(self) -> Result<NatsConfig> {
321        let builder = self.from_env_defaults();
322        let url = builder.url.ok_or_else(|| {
323            PhotonError::Internal(format!("{URL_ENV} not set for nats storage adapter"))
324        })?;
325        Ok(NatsConfig {
326            url,
327            credentials_file: builder.credentials_file,
328            stream_name: builder.stream_name.unwrap_or_else(|| "photon".into()),
329            retention: builder.retention.unwrap_or_else(retention_from_env),
330            replicas: builder.replicas.unwrap_or_else(replicas_from_env),
331            crypto: match builder.crypto {
332                Some(c) => c,
333                None => TransportCrypto::from_env()?,
334            },
335            replay_cursor: builder.replay_cursor.unwrap_or(ReplayCursor::StreamSeq),
336            sync_ack: builder.sync_ack.unwrap_or(true),
337            max_inflight: builder
338                .max_inflight
339                .unwrap_or_else(|| max_inflight_from_env(builder.sync_ack.unwrap_or(true))),
340            stream_shards: builder
341                .stream_shards
342                .unwrap_or_else(crate::stream_shard::stream_shards_from_env),
343            transport_security: builder
344                .transport_security
345                .unwrap_or_else(BrokerTransportSecurity::from_env),
346        })
347    }
348}
349
350fn replay_cursor_from_env() -> ReplayCursor {
351    match std::env::var(REPLAY_CURSOR_ENV)
352        .unwrap_or_else(|_| "stream_seq".into())
353        .to_ascii_lowercase()
354        .as_str()
355    {
356        "tail_only" | "tail" | "none" => ReplayCursor::TailOnly,
357        _ => ReplayCursor::StreamSeq,
358    }
359}
360
361fn sync_ack_from_env() -> bool {
362    !matches!(
363        std::env::var(SYNC_ACK_ENV)
364            .unwrap_or_else(|_| "1".into())
365            .as_str(),
366        "0" | "false" | "off" | "no"
367    )
368}
369
370fn max_inflight_from_env(sync_ack: bool) -> u32 {
371    if let Ok(raw) = std::env::var(MAX_INFLIGHT_ENV) {
372        if let Ok(n) = raw.parse::<u32>() {
373            return n.max(1);
374        }
375    }
376    if sync_ack {
377        1
378    } else {
379        256
380    }
381}
382
383#[cfg(test)]
384mod tests {
385    use super::*;
386
387    #[test]
388    fn replay_cursor_parses_tail_only() {
389        std::env::set_var(REPLAY_CURSOR_ENV, "tail_only");
390        assert_eq!(replay_cursor_from_env(), ReplayCursor::TailOnly);
391        std::env::remove_var(REPLAY_CURSOR_ENV);
392    }
393}
photon_backend_nats/config.rs

photon_backend_nats/
config.rs
