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    /// 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 public crate [`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    /// - Persists a sealed envelope; actor and payload plaintext must not be written to storage or
110    ///   broker records.
111    /// - Returns a decrypted [`Event`] including stable `event_id`, suitable for API callers and
112    ///   live fanout.
113    async fn append(
114        &self,
115        topic_name: &str,
116        topic_key: Option<&str>,
117        actor_json: Value,
118        payload_json: Value,
119    ) -> Result<Event>;
120
121    /// Stream events for a topic partition, optionally replaying after `after_seq`.
122    ///
123    /// # Contract
124    ///
125    /// - When `after_seq` is set, only events with `seq > after_seq` are yielded.
126    /// - When `topic_key_filter` is set, only matching partition keys are delivered.
127    /// - The stream runs until dropped or an error; live adapters may block for new events.
128    fn subscribe(
129        &self,
130        topic_name: String,
131        topic_key_filter: Option<String>,
132        after_seq: Option<i64>,
133    ) -> Pin<Box<dyn Stream<Item = Result<Event>> + Send>>;
134
135    /// Point lookup by event id (optional — see [`StorageCapabilities::supports_get_event`]).
136    ///
137    /// # Contract
138    ///
139    /// - Returns `None` when the id is unknown or the event was truncated by retention.
140    async fn get_event(&self, event_id: &str) -> Result<Option<Event>>;
141
142    /// Load durable subscription high-water seq.
143    ///
144    /// # Contract
145    ///
146    /// - Returns `None` when no checkpoint exists (caller chooses replay start policy).
147    async fn load_checkpoint(
148        &self,
149        subscription_name: &str,
150        topic_name: &str,
151        topic_key: Option<&str>,
152    ) -> Result<Option<i64>>;
153
154    /// Persist durable subscription high-water seq.
155    ///
156    /// # Contract
157    ///
158    /// - Stored `last_seq` is monotonic per subscription partition: a regressive commit must not
159    ///   lower an existing checkpoint.
160    async fn commit_checkpoint(
161        &self,
162        subscription_name: &str,
163        topic_name: &str,
164        topic_key: Option<&str>,
165        last_seq: i64,
166    ) -> Result<()>;
167
168    /// Trim events before `truncate_bound` for a partition (no-op when unsupported).
169    ///
170    /// # Contract
171    ///
172    /// - Default implementation returns `Ok(0)` without removing events.
173    /// - Supporting adapters remove events with `seq < truncate_bound`.
174    async fn truncate_before(
175        &self,
176        topic_name: &str,
177        topic_key: Option<&str>,
178        truncate_bound: i64,
179    ) -> Result<u64> {
180        let _ = (topic_name, topic_key, truncate_bound);
181        Ok(0)
182    }
183
184    /// Last delivered seq pin for retention watermarks (optional).
185    ///
186    /// # Contract
187    ///
188    /// - Default returns `None` (no delivery-layer retention pin).
189    async fn delivery_seq_pin(&self, topic_name: &str, topic_key: Option<&str>) -> Option<i64> {
190        let _ = (topic_name, topic_key);
191        None
192    }
193}
photon_backend/storage/port.rs

photon_backend/storage/
port.rs
