rovenue/sessions/
buffer.rs

1use std::sync::Arc;
2
3use crate::cache::store::SessionEventRow;
4use crate::cache::CacheStore;
5use crate::error::RovenueResult;
6
7use super::SessionEventKind;
8
9pub struct SessionBuffer {
10    store: Arc<CacheStore>,
11}
12
13impl SessionBuffer {
14    pub fn new(store: Arc<CacheStore>) -> Self {
15        Self { store }
16    }
17
18    /// Buffer one event under `rovenue_id`, the wire identity current when it
19    /// happened; the dispatcher posts it under that identity even after a
20    /// log out.
21    pub fn record(
22        &self,
23        rovenue_id: &str,
24        kind: SessionEventKind,
25        occurred_at: &str,
26        duration_ms: Option<u32>,
27    ) -> RovenueResult<()> {
28        self.store
29            .append_session_event(rovenue_id, kind.as_wire(), occurred_at, duration_ms)
30    }
31
32    /// Read up to `limit` buffered events WITHOUT removing them. The dispatcher
33    /// removes them (via [`delete`](Self::delete)) only after the server
34    /// confirms receipt, so a 503 (Kafka down) / network failure retains the
35    /// batch for the next flush instead of dropping it. Growth is bounded by
36    /// the store's FIFO cap (newest 1000) applied on append.
37    pub fn peek(&self, limit: usize) -> RovenueResult<Vec<SessionEventRow>> {
38        self.store.list_session_events(limit)
39    }
40
41    /// Remove the given event ids — called only after a successful dispatch.
42    pub fn delete(&self, ids: &[i64]) -> RovenueResult<()> {
43        self.store.delete_session_events(ids)
44    }
45}
46
47#[cfg(test)]
48mod tests {
49    use super::*;
50    use crate::cache::CacheStore;
51    use std::sync::Arc;
52
53    #[test]
54    fn record_appends_to_store() {
55        let store = Arc::new(CacheStore::open_in_memory().unwrap());
56        let buf = SessionBuffer::new(Arc::clone(&store));
57        buf.record(
58            "rov_a",
59            SessionEventKind::Open,
60            "2026-05-28T10:00:00Z",
61            None,
62        )
63        .unwrap();
64        buf.record(
65            "rov_a",
66            SessionEventKind::Background,
67            "2026-05-28T10:05:00Z",
68            Some(300_000),
69        )
70        .unwrap();
71        let rows = store.list_session_events(10).unwrap();
72        assert_eq!(rows.len(), 2);
73        assert_eq!(rows[0].kind, "open");
74        assert_eq!(rows[1].kind, "background");
75        assert_eq!(rows[1].duration_ms, Some(300_000));
76    }
77
78    #[test]
79    fn peek_does_not_delete() {
80        // The dispatcher must be able to read the batch WITHOUT removing it,
81        // so a failed (503 / network) POST retains the events for retry.
82        let store = Arc::new(CacheStore::open_in_memory().unwrap());
83        let buf = SessionBuffer::new(Arc::clone(&store));
84        buf.record(
85            "rov_a",
86            SessionEventKind::Open,
87            "2026-05-28T10:00:00Z",
88            None,
89        )
90        .unwrap();
91        buf.record(
92            "rov_a",
93            SessionEventKind::Close,
94            "2026-05-28T10:05:00Z",
95            None,
96        )
97        .unwrap();
98
99        let peeked = buf.peek(100).unwrap();
100        assert_eq!(peeked.len(), 2);
101        // Nothing removed — a second peek sees the same rows.
102        assert_eq!(buf.peek(100).unwrap().len(), 2);
103        assert_eq!(store.list_session_events(10).unwrap().len(), 2);
104    }
105
106    #[test]
107    fn delete_removes_only_the_given_ids() {
108        let store = Arc::new(CacheStore::open_in_memory().unwrap());
109        let buf = SessionBuffer::new(Arc::clone(&store));
110        buf.record(
111            "rov_a",
112            SessionEventKind::Open,
113            "2026-05-28T10:00:00Z",
114            None,
115        )
116        .unwrap();
117        buf.record(
118            "rov_a",
119            SessionEventKind::Close,
120            "2026-05-28T10:05:00Z",
121            None,
122        )
123        .unwrap();
124
125        let rows = buf.peek(100).unwrap();
126        buf.delete(&[rows[0].id]).unwrap();
127
128        let remaining = buf.peek(100).unwrap();
129        assert_eq!(remaining.len(), 1);
130        assert_eq!(remaining[0].id, rows[1].id);
131    }
132}
rovenue/sessions/buffer.rs

rovenue/sessions/
buffer.rs
