photon_runtime/builder.rs
1//! [`Photon`] builder — storage port + backend assembly.
2//!
3//! See the crate [Getting started](https://docs.rs/uf-photon/latest/photon/#getting-started)
4//! for Mode 1 (embedded) and Mode 2 (brokered) walkthroughs.
5
6use std::sync::Arc;
7
8use photon_telemetry::{install_ops_log, OpsLog};
9
10use photon_backend::{
11 instrumentation, BackendContext, EmbeddedBackend, ExecutorServices, GenericPhotonBackend,
12 InProcStoragePort, PhotonBackend, PhotonError, Result, RetentionHook, RetentionPolicy,
13 StoragePort, TopicRegistry, TransportCrypto,
14};
15
16use crate::executor::ExecutorController;
17use crate::{Photon, PhotonRuntimeState};
18
19type BackendInstallFn =
20 Box<dyn FnOnce(BackendContext) -> Result<Arc<dyn PhotonBackend>> + Send>;
21
22/// Builder for constructing [`Photon`] runtimes.
23///
24/// Wire a [`StoragePort`] (or accept the default in-process port), optionally install inventory
25/// discovery and ops telemetry, then [`build`](Self::build). Keep the returned [`Photon`] handle
26/// for `publish_on` / `subscribe_on`.
27///
28/// | Mode | Typical wiring |
29/// |------|----------------|
30/// | **Mode 1 — Embedded** | Default builder, or [`storage_port`](Self::storage_port) with `SQLite` |
31/// | **Mode 2 — Brokered** | [`storage_port`](Self::storage_port) with NATS/Kafka/Fluvio on **every** binary |
32///
33/// Getting started: [Mode 1](https://docs.rs/uf-photon/latest/photon/#mode-1--embedded-one-binary),
34/// [Mode 2](https://docs.rs/uf-photon/latest/photon/#mode-2--brokered-publisher--worker-binaries).
35///
36/// # Examples
37///
38/// Default mem path (loads `PHOTON_TRANSPORT_KEY` via [`TransportCrypto::from_env`]):
39///
40/// ```rust,no_run
41/// use photon_runtime::Photon;
42///
43/// # fn main() -> photon_backend::Result<()> {
44/// let _photon = Photon::builder().auto_registry().build()?;
45/// # Ok(())
46/// # }
47/// ```
48///
49/// Explicit storage port:
50///
51/// ```rust,no_run
52/// use std::sync::Arc;
53///
54/// use photon_backend::{InProcStoragePort, StoragePort, TransportCrypto};
55/// use photon_runtime::Photon;
56///
57/// # fn main() -> photon_backend::Result<()> {
58/// let port: Arc<dyn StoragePort> = Arc::new(InProcStoragePort::new(
59/// TransportCrypto::from_env()?,
60/// ));
61/// let _photon = Photon::builder().storage_port(port).auto_registry().build()?;
62/// # Ok(())
63/// # }
64/// ```
65#[derive(Default)]
66pub struct PhotonBuilder {
67 storage_port: Option<Arc<dyn StoragePort>>,
68 backend: Option<Arc<dyn PhotonBackend>>,
69 backend_install: Option<BackendInstallFn>,
70 use_auto_registry: bool,
71 ops_log: Option<Arc<dyn OpsLog>>,
72 retention_policy: Option<RetentionPolicy>,
73 retention_hook: Option<Arc<dyn RetentionHook>>,
74}
75
76impl PhotonBuilder {
77 /// Explicit storage port (defaults to in-process `mem` via [`InProcStoragePort`]).
78 ///
79 /// Use this for `SQLite` (Mode 1 durable) and for broker adapters (Mode 2 — same port config on
80 /// publisher and worker). See
81 /// [Getting started → Mode 2](https://docs.rs/uf-photon/latest/photon/#mode-2--brokered-publisher--worker-binaries).
82 ///
83 /// # Example
84 ///
85 /// ```rust,no_run
86 /// use std::sync::Arc;
87 ///
88 /// use photon_backend::{InProcStoragePort, StoragePort, TransportCrypto};
89 /// use photon_runtime::Photon;
90 ///
91 /// # fn main() -> photon_backend::Result<()> {
92 /// let port: Arc<dyn StoragePort> = Arc::new(InProcStoragePort::new(
93 /// TransportCrypto::from_env()?,
94 /// ));
95 /// let _photon = Photon::builder().storage_port(port).auto_registry().build()?;
96 /// # Ok(())
97 /// # }
98 /// ```
99 #[must_use]
100 pub fn storage_port(mut self, port: Arc<dyn StoragePort>) -> Self {
101 self.storage_port = Some(port);
102 self
103 }
104
105 /// Pre-built backend instance.
106 ///
107 /// Prefer [`storage_port`](Self::storage_port) for normal hosts. Use this when you already
108 /// constructed a [`PhotonBackend`] (tests, custom delivery stacks).
109 ///
110 /// # Errors
111 ///
112 /// [`build`](Self::build) returns an error if both this and [`backend_with_context`](Self::backend_with_context)
113 /// are set.
114 #[must_use]
115 pub fn backend(mut self, backend: Arc<dyn PhotonBackend>) -> Self {
116 self.backend = Some(backend);
117 self.backend_install = None;
118 self
119 }
120
121 /// Build backend from shared [`BackendContext`] (custom install closures).
122 ///
123 /// Prefer [`storage_port`](Self::storage_port) for adapter wiring. This is for hosts that
124 /// install a custom [`PhotonBackend`] from the registry context.
125 ///
126 /// # Example
127 ///
128 /// ```rust,no_run
129 /// use std::sync::Arc;
130 ///
131 /// use photon_backend::{BackendContext, EmbeddedBackend};
132 /// use photon_runtime::Photon;
133 ///
134 /// # fn main() -> photon_backend::Result<()> {
135 /// let _photon = Photon::builder()
136 /// .backend_with_context(|ctx: BackendContext| EmbeddedBackend::install_mem(ctx))
137 /// .auto_registry()
138 /// .build()?;
139 /// # Ok(())
140 /// # }
141 /// ```
142 #[must_use]
143 pub fn backend_with_context<F>(mut self, install: F) -> Self
144 where
145 F: FnOnce(BackendContext) -> Result<Arc<dyn PhotonBackend>> + Send + 'static,
146 {
147 self.backend = None;
148 self.backend_install = Some(Box::new(install));
149 self
150 }
151
152 /// Shorthand for [`backend_with_context`](Self::backend_with_context) with
153 /// [`EmbeddedBackend::install_mem`](photon_backend::GenericPhotonBackend::install_mem).
154 ///
155 /// Equivalent to the default path when you also omit [`storage_port`](Self::storage_port)
156 /// (in-process mem). Prefer the default [`build`](Self::build) unless you need an explicit
157 /// install fn.
158 ///
159 /// # Example
160 ///
161 /// ```rust,no_run
162 /// use photon_runtime::Photon;
163 ///
164 /// # fn main() -> photon_backend::Result<()> {
165 /// let _photon = Photon::builder().mem_backend().auto_registry().build()?;
166 /// # Ok(())
167 /// # }
168 /// ```
169 #[must_use]
170 pub fn mem_backend(mut self) -> Self {
171 self.backend = None;
172 self.backend_install = Some(Box::new(EmbeddedBackend::install_mem));
173 self
174 }
175
176 /// Install a concrete [`OpsLog`] adapter before build.
177 ///
178 /// Runnable: `cargo run -p uf-photon --example telemetry_ops_log --features runtime,mem`.
179 ///
180 /// # Example
181 ///
182 /// ```rust,no_run
183 /// use photon_runtime::Photon;
184 /// use photon_telemetry::ConsoleOpsLog;
185 ///
186 /// # fn main() -> photon_backend::Result<()> {
187 /// let _photon = Photon::builder()
188 /// .ops_log(ConsoleOpsLog)
189 /// .auto_registry()
190 /// .build()?;
191 /// # Ok(())
192 /// # }
193 /// ```
194 #[must_use]
195 pub fn ops_log(mut self, log: impl OpsLog + 'static) -> Self {
196 self.ops_log = Some(Arc::new(log));
197 self
198 }
199
200 /// Install a shared [`OpsLog`] trait object before build.
201 #[must_use]
202 pub fn ops_log_arc(mut self, log: Arc<dyn OpsLog>) -> Self {
203 self.ops_log = Some(log);
204 self
205 }
206
207 /// Discover `#[photon::topic]` / `#[photon::subscribe]` descriptors via Quark inventory.
208 ///
209 /// Required when using the macros in the same crate graph as the host. Without this, the
210 /// topic registry stays empty and the executor has nothing to dispatch.
211 ///
212 /// Runnable: `cargo run -p uf-photon --example embedded_mem --features runtime,mem`.
213 #[must_use]
214 pub const fn auto_registry(mut self) -> Self {
215 self.use_auto_registry = true;
216 self
217 }
218
219 /// Override default retention policy (env fallbacks apply for unset fields).
220 #[must_use]
221 pub fn retention_policy(mut self, policy: RetentionPolicy) -> Self {
222 self.retention_policy = Some(policy);
223 self
224 }
225
226 /// Host hook for extra subscriptions and legal-hold floors.
227 #[must_use]
228 pub fn retention_hook(mut self, hook: Arc<dyn RetentionHook>) -> Self {
229 self.retention_hook = Some(hook);
230 self
231 }
232
233 /// Assemble the [`Photon`] runtime.
234 ///
235 /// # Defaults
236 ///
237 /// - **Storage port:** [`InProcStoragePort`] with [`TransportCrypto::from_env`] when
238 /// [`storage_port`](Self::storage_port) was not set — requires `PHOTON_TRANSPORT_KEY`
239 /// - **Backend:** generic backend over that port when no custom [`backend`](Self::backend) /
240 /// [`backend_with_context`](Self::backend_with_context) is set
241 /// - **Registry:** empty unless [`auto_registry`](Self::auto_registry) was called
242 ///
243 /// # Errors
244 ///
245 /// Returns an error if transport crypto cannot load from the environment, a custom install
246 /// fn fails, or both `backend` and `backend_with_context` were set.
247 pub fn build(self) -> Result<Photon> {
248 if let Some(log) = self.ops_log {
249 install_ops_log(log);
250 }
251
252 let registry = if self.use_auto_registry {
253 TopicRegistry::auto_discover()
254 } else {
255 TopicRegistry::new()
256 };
257
258 let port = match self.storage_port {
259 Some(port) => port,
260 None => Arc::new(InProcStoragePort::new(TransportCrypto::from_env()?)),
261 };
262
263 let ctx = BackendContext {
264 registry: registry.clone(),
265 };
266
267 let backend = match (self.backend, self.backend_install) {
268 (Some(b), None) => b,
269 (None, Some(install)) => install(ctx)?,
270 (None, None) => GenericPhotonBackend::install_with_port(
271 BackendContext { registry },
272 Arc::clone(&port),
273 )?,
274 (Some(_), Some(_)) => {
275 return Err(PhotonError::Internal(
276 "PhotonBuilder: set backend() or backend_with_context(), not both".into(),
277 ));
278 }
279 };
280
281 let retention_policy = self.retention_policy.unwrap_or_default();
282 let runtime = PhotonRuntimeState {
283 storage_port: Arc::clone(&port),
284 executor_services: Arc::new(ExecutorServices::new(
285 port,
286 retention_policy,
287 self.retention_hook,
288 )),
289 executor: Arc::new(ExecutorController::default()),
290 };
291
292 let backend = instrumentation::wrap_backend(backend);
293 Ok(Photon::new(backend, runtime))
294 }
295}