pjson_rs/infrastructure/adapters/
frame_store.rs

1//! In-memory implementation of [`FrameStoreGat`].
2//!
3//! Backs the `GET /pjs/sessions/{session_id}/streams/{stream_id}/frames` HTTP
4//! endpoint by retaining frames produced through `GenerateFramesCommand` /
5//! `BatchGenerateFramesCommand` so callers can fetch them after the fact.
6//!
7//! Memory is bounded per stream by the cap passed to
8//! [`InMemoryFrameStore::with_capacity`] (or the default
9//! [`crate::domain::config::DEFAULT_FRAME_HISTORY_PER_STREAM`]); once reached,
10//! the oldest frames are evicted FIFO. This prevents a long-lived stream from
11//! growing the store without bound.
12
13use std::collections::VecDeque;
14use std::future::Future;
15use std::sync::Arc;
16
17use dashmap::DashMap;
18
19use crate::domain::{
20    DomainResult,
21    config::DEFAULT_FRAME_HISTORY_PER_STREAM,
22    entities::Frame,
23    ports::{FrameStoreGat, FrameStorePage},
24    value_objects::{Priority, StreamId},
25};
26
27/// Lock-free in-memory [`FrameStoreGat`] implementation.
28///
29/// Frames are stored per stream in a [`VecDeque`] in append order. Lookups
30/// scan the per-stream deque under a single shard lock, so cost grows linearly
31/// with frame history depth — fine for the bounded sizes we cap at.
32#[derive(Debug)]
33pub struct InMemoryFrameStore {
34    frames: Arc<DashMap<StreamId, VecDeque<Frame>>>,
35    max_frames_per_stream: usize,
36}
37
38impl InMemoryFrameStore {
39    /// Create a store with the default per-stream cap
40    /// ([`DEFAULT_FRAME_HISTORY_PER_STREAM`]).
41    pub fn new() -> Self {
42        Self::with_capacity(DEFAULT_FRAME_HISTORY_PER_STREAM)
43    }
44
45    /// Create a store with an explicit per-stream cap.
46    ///
47    /// `max_frames_per_stream` must be at least 1 — a zero cap would drop every
48    /// frame on insertion and is rejected via debug assertion (the constructor
49    /// silently substitutes 1 in release builds).
50    pub fn with_capacity(max_frames_per_stream: usize) -> Self {
51        debug_assert!(
52            max_frames_per_stream > 0,
53            "max_frames_per_stream must be at least 1"
54        );
55        Self {
56            frames: Arc::new(DashMap::new()),
57            max_frames_per_stream: max_frames_per_stream.max(1),
58        }
59    }
60
61    /// Number of streams that currently have frame history.
62    pub fn stream_count(&self) -> usize {
63        self.frames.len()
64    }
65}
66
67impl Default for InMemoryFrameStore {
68    fn default() -> Self {
69        Self::new()
70    }
71}
72
73impl FrameStoreGat for InMemoryFrameStore {
74    type AppendFramesFuture<'a>
75        = impl Future<Output = DomainResult<()>> + Send + 'a
76    where
77        Self: 'a;
78
79    type GetFramesFuture<'a>
80        = impl Future<Output = DomainResult<FrameStorePage>> + Send + 'a
81    where
82        Self: 'a;
83
84    type DeleteFramesForStreamFuture<'a>
85        = impl Future<Output = DomainResult<()>> + Send + 'a
86    where
87        Self: 'a;
88
89    fn append_frames(
90        &self,
91        stream_id: StreamId,
92        frames: Vec<Frame>,
93    ) -> Self::AppendFramesFuture<'_> {
94        async move {
95            if frames.is_empty() {
96                return Ok(());
97            }
98            let cap = self.max_frames_per_stream;
99            let mut entry = self.frames.entry(stream_id).or_default();
100            let history = entry.value_mut();
101            for frame in frames {
102                if history.len() >= cap {
103                    history.pop_front();
104                }
105                history.push_back(frame);
106            }
107            Ok(())
108        }
109    }
110
111    fn get_frames(
112        &self,
113        stream_id: StreamId,
114        since_sequence: Option<u64>,
115        priority_filter: Option<Priority>,
116        limit: Option<usize>,
117    ) -> Self::GetFramesFuture<'_> {
118        async move {
119            let Some(entry) = self.frames.get(&stream_id) else {
120                return Ok(FrameStorePage {
121                    frames: Vec::new(),
122                    total_matching: 0,
123                });
124            };
125            let min_priority = priority_filter.map(|p| p.value());
126            let history = entry.value();
127            let mut total_matching = 0usize;
128            let cap = limit.unwrap_or(usize::MAX);
129            let mut out = Vec::new();
130            for frame in history.iter() {
131                if let Some(since) = since_sequence
132                    && frame.sequence() <= since
133                {
134                    continue;
135                }
136                if let Some(min) = min_priority
137                    && frame.priority().value() < min
138                {
139                    continue;
140                }
141                total_matching += 1;
142                if out.len() < cap {
143                    out.push(frame.clone());
144                }
145            }
146            Ok(FrameStorePage {
147                frames: out,
148                total_matching,
149            })
150        }
151    }
152
153    fn delete_frames_for_stream(
154        &self,
155        stream_id: StreamId,
156    ) -> Self::DeleteFramesForStreamFuture<'_> {
157        async move {
158            self.frames.remove(&stream_id);
159            Ok(())
160        }
161    }
162}
163
164#[cfg(test)]
165mod tests {
166    use super::*;
167    use crate::domain::{
168        entities::frame::FramePatch,
169        value_objects::{JsonData, JsonPath},
170    };
171
172    fn make_frame(stream_id: StreamId, sequence: u64, priority: Priority) -> Frame {
173        let patch = FramePatch::set(
174            JsonPath::new(format!("$.field_{sequence}")).unwrap(),
175            JsonData::Integer(sequence as i64),
176        );
177        Frame::patch(stream_id, sequence, priority, vec![patch]).unwrap()
178    }
179
180    #[tokio::test]
181    async fn appended_frames_are_returned_in_order() {
182        let store = InMemoryFrameStore::new();
183        let stream_id = StreamId::new();
184
185        let frames = vec![
186            make_frame(stream_id, 1, Priority::HIGH),
187            make_frame(stream_id, 2, Priority::HIGH),
188            make_frame(stream_id, 3, Priority::HIGH),
189        ];
190        store.append_frames(stream_id, frames).await.unwrap();
191
192        let page = store.get_frames(stream_id, None, None, None).await.unwrap();
193        assert_eq!(page.total_matching, 3);
194        assert_eq!(
195            page.frames.iter().map(Frame::sequence).collect::<Vec<_>>(),
196            vec![1, 2, 3]
197        );
198    }
199
200    #[tokio::test]
201    async fn since_sequence_filter_excludes_older_frames() {
202        let store = InMemoryFrameStore::new();
203        let stream_id = StreamId::new();
204        store
205            .append_frames(
206                stream_id,
207                vec![
208                    make_frame(stream_id, 1, Priority::HIGH),
209                    make_frame(stream_id, 2, Priority::HIGH),
210                    make_frame(stream_id, 3, Priority::HIGH),
211                ],
212            )
213            .await
214            .unwrap();
215
216        let page = store
217            .get_frames(stream_id, Some(1), None, None)
218            .await
219            .unwrap();
220        assert_eq!(page.total_matching, 2);
221        assert_eq!(
222            page.frames.iter().map(Frame::sequence).collect::<Vec<_>>(),
223            vec![2, 3]
224        );
225    }
226
227    #[tokio::test]
228    async fn priority_filter_keeps_only_higher_or_equal_priority() {
229        let store = InMemoryFrameStore::new();
230        let stream_id = StreamId::new();
231        store
232            .append_frames(
233                stream_id,
234                vec![
235                    make_frame(stream_id, 1, Priority::LOW),
236                    make_frame(stream_id, 2, Priority::HIGH),
237                    make_frame(stream_id, 3, Priority::CRITICAL),
238                ],
239            )
240            .await
241            .unwrap();
242
243        let page = store
244            .get_frames(stream_id, None, Some(Priority::HIGH), None)
245            .await
246            .unwrap();
247        assert_eq!(page.total_matching, 2);
248        assert_eq!(
249            page.frames.iter().map(Frame::sequence).collect::<Vec<_>>(),
250            vec![2, 3]
251        );
252    }
253
254    #[tokio::test]
255    async fn limit_caps_returned_frames_but_not_total() {
256        let store = InMemoryFrameStore::new();
257        let stream_id = StreamId::new();
258        store
259            .append_frames(
260                stream_id,
261                (1..=5)
262                    .map(|s| make_frame(stream_id, s, Priority::HIGH))
263                    .collect(),
264            )
265            .await
266            .unwrap();
267
268        let page = store
269            .get_frames(stream_id, None, None, Some(2))
270            .await
271            .unwrap();
272        assert_eq!(page.frames.len(), 2);
273        assert_eq!(page.total_matching, 5);
274    }
275
276    #[tokio::test]
277    async fn capacity_evicts_oldest_first() {
278        let store = InMemoryFrameStore::with_capacity(3);
279        let stream_id = StreamId::new();
280        store
281            .append_frames(
282                stream_id,
283                (1..=5)
284                    .map(|s| make_frame(stream_id, s, Priority::HIGH))
285                    .collect(),
286            )
287            .await
288            .unwrap();
289
290        let page = store.get_frames(stream_id, None, None, None).await.unwrap();
291        assert_eq!(
292            page.frames.iter().map(Frame::sequence).collect::<Vec<_>>(),
293            vec![3, 4, 5]
294        );
295        assert_eq!(page.total_matching, 3);
296    }
297
298    #[tokio::test]
299    async fn delete_frames_drops_history() {
300        let store = InMemoryFrameStore::new();
301        let stream_id = StreamId::new();
302        store
303            .append_frames(stream_id, vec![make_frame(stream_id, 1, Priority::HIGH)])
304            .await
305            .unwrap();
306
307        store.delete_frames_for_stream(stream_id).await.unwrap();
308        let page = store.get_frames(stream_id, None, None, None).await.unwrap();
309        assert!(page.frames.is_empty());
310        assert_eq!(page.total_matching, 0);
311    }
312
313    #[tokio::test]
314    async fn unknown_stream_returns_empty_page() {
315        let store = InMemoryFrameStore::new();
316        let page = store
317            .get_frames(StreamId::new(), None, None, None)
318            .await
319            .unwrap();
320        assert!(page.frames.is_empty());
321        assert_eq!(page.total_matching, 0);
322    }
323}
pjson_rs/infrastructure/adapters/frame_store.rs

pjson_rs/infrastructure/adapters/
frame_store.rs
