photon_backend/storage/port.rs
1//! Object-safe storage port for publish, subscribe, checkpoints, and optional retention.
2//!
3//! **Application authors** usually select a built-in adapter (`mem`, `sqlite`, `nats`, `kafka`, `fluvio`)
4//! via [`PhotonBuilder::storage_port`](https://docs.rs/uf-photon/latest/photon/struct.PhotonBuilder.html#method.storage_port)
5//! — you rarely implement this trait directly.
6//! **Adapter authors** implement [`StoragePort`] in a storage crate and wire it at boot.
7//!
8//! Reference in-process implementation: [`InProcStoragePort`]. Runnable host example:
9//! `cargo run -p uf-photon --example embedded_mem --features runtime,mem`.
10//!
11//! Getting started: [Embedded](https://docs.rs/uf-photon/latest/photon/#embedded-one-binary),
12//! [Brokered](https://docs.rs/uf-photon/latest/photon/#brokered-publisher--worker-binaries).
13//!
14//! See also: [`crate::checkpoint`], [`crate::retention`], [`crate::backend`].
15
16use std::pin::Pin;
17use std::time::Duration;
18
19use async_trait::async_trait;
20use futures::stream::Stream;
21use serde_json::Value;
22
23use crate::error::Result;
24use crate::models::Event;
25
26/// Capabilities advertised by a storage port (surfaced via [`crate::backend::GenericPhotonBackend`]).
27#[derive(Debug, Clone, Copy, PartialEq, Eq)]
28pub struct StorageCapabilities {
29 /// Whether [`StoragePort::get_event`] is supported.
30 pub supports_get_event: bool,
31 /// Whether [`StoragePort::list_by_topic`] / [`StoragePort::list_recent`] are supported.
32 pub supports_list_events: bool,
33 /// Maximum replay window for bounded retention adapters.
34 pub max_replay_window: Option<Duration>,
35 /// Stable telemetry / bench label (`mem`, `nats`, …).
36 pub telemetry_label: &'static str,
37}
38
39impl StorageCapabilities {
40 /// In-process embedded tier defaults.
41 #[must_use]
42 pub const fn mem() -> Self {
43 Self {
44 supports_get_event: true,
45 supports_list_events: true,
46 max_replay_window: None,
47 telemetry_label: "mem",
48 }
49 }
50
51 /// Embedded `SQLite` tier defaults.
52 #[must_use]
53 pub const fn sqlite() -> Self {
54 Self {
55 supports_get_event: true,
56 supports_list_events: true,
57 max_replay_window: None,
58 telemetry_label: "sqlite",
59 }
60 }
61
62 /// Broker tier defaults (~15 min replay window).
63 #[must_use]
64 pub const fn broker(label: &'static str) -> Self {
65 Self {
66 supports_get_event: false,
67 supports_list_events: false,
68 max_replay_window: Some(Duration::from_mins(15)),
69 telemetry_label: label,
70 }
71 }
72}
73
74/// Storage adapter contract — append, subscribe, checkpoints, and optional retention.
75///
76/// Each method documents its behavior under **Contract**. Built-in implementations:
77///
78/// | Adapter | Type | Crate |
79/// |---------|------|-------|
80/// | In-process | [`InProcStoragePort`](super::InProcStoragePort) | `photon-backend` (`mem`) |
81/// | `SQLite` | `SqliteStoragePort` | `photon-backend-sqlite` |
82/// | NATS `JetStream` | `NatsStoragePort` | `photon-backend-nats` |
83/// | Kafka | `KafkaStoragePort` | `photon-backend-kafka` |
84/// | Fluvio | `FluvioStoragePort` | `photon-backend-fluvio` |
85///
86/// # Example (use a built-in port)
87///
88/// ```rust,no_run
89/// use std::sync::Arc;
90///
91/// use photon_backend::{InProcStoragePort, StoragePort, TransportCrypto};
92///
93/// # fn main() -> photon_backend::Result<()> {
94/// let port: Arc<dyn StoragePort> = Arc::new(InProcStoragePort::new(
95/// TransportCrypto::from_env()?,
96/// ));
97/// let _caps = port.capabilities();
98/// # Ok(())
99/// # }
100/// ```
101///
102/// Install at boot via the public crate [`PhotonBuilder`](https://docs.rs/uf-photon/latest/photon/struct.PhotonBuilder.html).
103/// Host walkthrough: [Integrating the host](https://docs.rs/uf-photon/latest/photon/#integrating-the-host).
104#[async_trait]
105pub trait StoragePort: Send + Sync {
106 /// Adapter capabilities for contract tests and telemetry.
107 fn capabilities(&self) -> StorageCapabilities;
108
109 /// Append one event to a topic partition.
110 ///
111 /// # Contract
112 ///
113 /// - Assigns a monotonically increasing `seq` per `(topic_name, topic_key)` partition.
114 /// - Persists a sealed envelope; actor and payload plaintext must not be written to storage or
115 /// broker records.
116 /// - Returns a decrypted [`Event`] including stable `event_id`, suitable for API callers and
117 /// live fanout.
118 async fn append(
119 &self,
120 topic_name: &str,
121 topic_key: Option<&str>,
122 actor_json: Value,
123 payload_json: Value,
124 ) -> Result<Event>;
125
126 /// Stream events for a topic partition, optionally replaying after `after_seq`.
127 ///
128 /// # Contract
129 ///
130 /// - When `after_seq` is set, only events with `seq > after_seq` are yielded.
131 /// - When `topic_key_filter` is set, only matching partition keys are delivered.
132 /// - The stream runs until dropped or an error; live adapters may block for new events.
133 fn subscribe(
134 &self,
135 topic_name: String,
136 topic_key_filter: Option<String>,
137 after_seq: Option<i64>,
138 ) -> Pin<Box<dyn Stream<Item = Result<Event>> + Send>>;
139
140 /// Point lookup by event id (optional — see [`StorageCapabilities::supports_get_event`]).
141 ///
142 /// # Contract
143 ///
144 /// - Returns `None` when the id is unknown or the event was truncated by retention.
145 async fn get_event(&self, event_id: &str) -> Result<Option<Event>>;
146
147 /// Bounded page of events for one topic (ops browse).
148 ///
149 /// # Contract
150 ///
151 /// - When `supports_list_events` is false, returns an empty vec (brokers).
152 /// - When `after_seq` is set, only events with `seq > after_seq` are included.
153 /// - When `topic_key` is set, only that partition is included.
154 /// - Results are ordered by `seq` ascending and capped at `limit` (zero → empty).
155 /// - Returned events are decrypted like [`Self::get_event`].
156 async fn list_by_topic(
157 &self,
158 topic_name: &str,
159 topic_key: Option<&str>,
160 after_seq: Option<i64>,
161 limit: usize,
162 ) -> Result<Vec<Event>>;
163
164 /// Bounded cross-topic page of newest events (ops browse).
165 ///
166 /// # Contract
167 ///
168 /// - When `supports_list_events` is false, returns an empty vec (brokers).
169 /// - Ordered by `created_at` descending, capped at `limit` (zero → empty).
170 /// - Returned events are decrypted like [`Self::get_event`].
171 async fn list_recent(&self, limit: usize) -> Result<Vec<Event>>;
172
173 /// Load durable subscription high-water seq.
174 ///
175 /// # Contract
176 ///
177 /// - Returns `None` when no checkpoint exists (caller chooses replay start policy).
178 async fn load_checkpoint(
179 &self,
180 subscription_name: &str,
181 topic_name: &str,
182 topic_key: Option<&str>,
183 ) -> Result<Option<i64>>;
184
185 /// Persist durable subscription high-water seq.
186 ///
187 /// # Contract
188 ///
189 /// - Stored `last_seq` is monotonic per subscription partition: a regressive commit must not
190 /// lower an existing checkpoint.
191 async fn commit_checkpoint(
192 &self,
193 subscription_name: &str,
194 topic_name: &str,
195 topic_key: Option<&str>,
196 last_seq: i64,
197 ) -> Result<()>;
198
199 /// Trim events before `truncate_bound` for a partition (no-op when unsupported).
200 ///
201 /// # Contract
202 ///
203 /// - Default implementation returns `Ok(0)` without removing events.
204 /// - Supporting adapters remove events with `seq < truncate_bound`.
205 async fn truncate_before(
206 &self,
207 topic_name: &str,
208 topic_key: Option<&str>,
209 truncate_bound: i64,
210 ) -> Result<u64> {
211 let _ = (topic_name, topic_key, truncate_bound);
212 Ok(0)
213 }
214
215 /// Last delivered seq pin for retention watermarks (optional).
216 ///
217 /// # Contract
218 ///
219 /// - Default returns `None` (no delivery-layer retention pin).
220 async fn delivery_seq_pin(&self, topic_name: &str, topic_key: Option<&str>) -> Option<i64> {
221 let _ = (topic_name, topic_key);
222 None
223 }
224}