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: [Mode 1](https://docs.rs/uf-photon/latest/photon/#mode-1--embedded-one-binary),
12//! [Mode 2](https://docs.rs/uf-photon/latest/photon/#mode-2--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    /// Maximum replay window for bounded retention adapters.
32    pub max_replay_window: Option<Duration>,
33    /// Stable telemetry / bench label (`mem`, `nats`, …).
34    pub telemetry_label: &'static str,
35}
36
37impl StorageCapabilities {
38    /// In-process embedded tier defaults.
39    #[must_use]
40    pub const fn mem() -> Self {
41        Self {
42            supports_get_event: true,
43            max_replay_window: None,
44            telemetry_label: "mem",
45        }
46    }
47
48    /// Embedded `SQLite` tier defaults.
49    #[must_use]
50    pub const fn sqlite() -> Self {
51        Self {
52            supports_get_event: true,
53            max_replay_window: None,
54            telemetry_label: "sqlite",
55        }
56    }
57
58    /// Broker tier defaults (~15 min replay window).
59    #[must_use]
60    pub const fn broker(label: &'static str) -> Self {
61        Self {
62            supports_get_event: false,
63            max_replay_window: Some(Duration::from_mins(15)),
64            telemetry_label: label,
65        }
66    }
67}
68
69/// Storage adapter contract — append, subscribe, checkpoints, and optional retention.
70///
71/// Each method documents its behavior under **Contract**. Built-in implementations:
72///
73/// | Adapter | Type | Crate |
74/// |---------|------|-------|
75/// | In-process | [`InProcStoragePort`](super::InProcStoragePort) | `photon-backend` (`mem`) |
76/// | `SQLite` | `SqliteStoragePort` | `photon-backend-sqlite` |
77/// | NATS `JetStream` | `NatsStoragePort` | `photon-backend-nats` |
78/// | Kafka | `KafkaStoragePort` | `photon-backend-kafka` |
79/// | Fluvio | `FluvioStoragePort` | `photon-backend-fluvio` |
80///
81/// # Example (use a built-in port)
82///
83/// ```rust,no_run
84/// use std::sync::Arc;
85///
86/// use photon_backend::{InProcStoragePort, StoragePort, TransportCrypto};
87///
88/// # fn main() -> photon_backend::Result<()> {
89/// let port: Arc<dyn StoragePort> = Arc::new(InProcStoragePort::new(
90///     TransportCrypto::from_env()?,
91/// ));
92/// let _caps = port.capabilities();
93/// # Ok(())
94/// # }
95/// ```
96///
97/// Install at boot via the facade [`PhotonBuilder`](https://docs.rs/uf-photon/latest/photon/struct.PhotonBuilder.html).
98/// Host walkthrough: [Integrating the host](https://docs.rs/uf-photon/latest/photon/#integrating-the-host).
99#[async_trait]
100pub trait StoragePort: Send + Sync {
101    /// Adapter capabilities for contract tests and telemetry.
102    fn capabilities(&self) -> StorageCapabilities;
103
104    /// Append one event to a topic partition.
105    ///
106    /// # Contract
107    ///
108    /// - Assigns a monotonically increasing `seq` per `(topic_name, topic_key)` partition.
109    /// - Returns the persisted [`Event`] including stable `event_id`.
110    /// - Payload and actor JSON are opaque to the adapter (encryption happens above the port).
111    async fn append(
112        &self,
113        topic_name: &str,
114        topic_key: Option<&str>,
115        actor_json: Value,
116        payload_json: Value,
117    ) -> Result<Event>;
118
119    /// Stream events for a topic partition, optionally replaying after `after_seq`.
120    ///
121    /// # Contract
122    ///
123    /// - When `after_seq` is set, only events with `seq > after_seq` are yielded.
124    /// - When `topic_key_filter` is set, only matching partition keys are delivered.
125    /// - The stream runs until dropped or an error; live adapters may block for new events.
126    fn subscribe(
127        &self,
128        topic_name: String,
129        topic_key_filter: Option<String>,
130        after_seq: Option<i64>,
131    ) -> Pin<Box<dyn Stream<Item = Result<Event>> + Send>>;
132
133    /// Point lookup by event id (optional — see [`StorageCapabilities::supports_get_event`]).
134    ///
135    /// # Contract
136    ///
137    /// - Returns `None` when the id is unknown or the event was truncated by retention.
138    async fn get_event(&self, event_id: &str) -> Result<Option<Event>>;
139
140    /// Load durable subscription high-water seq.
141    ///
142    /// # Contract
143    ///
144    /// - Returns `None` when no checkpoint exists (caller chooses replay start policy).
145    async fn load_checkpoint(
146        &self,
147        subscription_name: &str,
148        topic_name: &str,
149        topic_key: Option<&str>,
150    ) -> Result<Option<i64>>;
151
152    /// Persist durable subscription high-water seq.
153    ///
154    /// # Contract
155    ///
156    /// - `last_seq` is monotonic per subscription partition; adapters may reject regressions.
157    async fn commit_checkpoint(
158        &self,
159        subscription_name: &str,
160        topic_name: &str,
161        topic_key: Option<&str>,
162        last_seq: i64,
163    ) -> Result<()>;
164
165    /// Trim events before `truncate_bound` for a partition (no-op when unsupported).
166    ///
167    /// # Contract
168    ///
169    /// - Default implementation returns `Ok(0)` without removing events.
170    /// - Supporting adapters remove events with `seq < truncate_bound`.
171    async fn truncate_before(
172        &self,
173        topic_name: &str,
174        topic_key: Option<&str>,
175        truncate_bound: i64,
176    ) -> Result<u64> {
177        let _ = (topic_name, topic_key, truncate_bound);
178        Ok(0)
179    }
180
181    /// Last delivered seq pin for retention watermarks (optional).
182    ///
183    /// # Contract
184    ///
185    /// - Default returns `None` (no delivery-layer retention pin).
186    async fn delivery_seq_pin(
187        &self,
188        topic_name: &str,
189        topic_key: Option<&str>,
190    ) -> Option<i64> {
191        let _ = (topic_name, topic_key);
192        None
193    }
194}
photon_backend/storage/port.rs

photon_backend/storage/
port.rs
