rovenue/attributes/
buffer.rs

1use std::sync::Arc;
2
3use crate::cache::store::{AttributeMutationRow, CacheStore};
4use crate::error::RovenueResult;
5
6/// Local dirty-mutation queue for subscriber attributes. Unlike the
7/// session buffer, callers `list` then `delete` separately so the
8/// dispatcher can keep rows queued when a flush fails (attributes are
9/// durable user data, not fire-and-forget telemetry).
10pub struct AttributeBuffer {
11    store: Arc<CacheStore>,
12}
13
14impl AttributeBuffer {
15    pub fn new(store: Arc<CacheStore>) -> Self {
16        Self { store }
17    }
18
19    /// Queue a single attribute mutation under `rovenue_id`, the wire
20    /// identity it was made under. `None` value means delete.
21    pub fn set(&self, rovenue_id: &str, key: &str, value: Option<&str>) -> RovenueResult<()> {
22        self.store.append_attribute_mutation(rovenue_id, key, value)
23    }
24
25    pub fn list(&self, limit: usize) -> RovenueResult<Vec<AttributeMutationRow>> {
26        self.store.list_attribute_mutations(limit)
27    }
28
29    pub fn delete(&self, ids: &[i64]) -> RovenueResult<()> {
30        self.store.delete_attribute_mutations(ids)
31    }
32}
33
34#[cfg(test)]
35mod tests {
36    use super::*;
37    use crate::cache::store::CacheStore;
38    use std::sync::Arc;
39
40    #[test]
41    fn set_appends_and_list_delete_roundtrip() {
42        let store = Arc::new(CacheStore::open_in_memory().unwrap());
43        let buf = AttributeBuffer::new(Arc::clone(&store));
44        buf.set("rov_a", "$email", Some("a@b.com")).unwrap();
45        buf.set("rov_a", "country", None).unwrap();
46
47        let rows = buf.list(100).unwrap();
48        assert_eq!(rows.len(), 2);
49
50        buf.delete(&rows.iter().map(|r| r.id).collect::<Vec<_>>())
51            .unwrap();
52        assert_eq!(buf.list(100).unwrap().len(), 0);
53    }
54}
rovenue/attributes/buffer.rs

rovenue/attributes/
buffer.rs
