weavatrix_memory/store/
subscription.rs

1use super::EventStore;
2use crate::{
3    error::{MemoryError, Result},
4    event::StoredEvent,
5};
6use serde::{Deserialize, Serialize};
7
8#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
9pub struct SubscriptionCheckpoint {
10    pub global_position: Option<u64>,
11}
12
13#[derive(Debug, Clone, PartialEq, Eq)]
14pub struct CatchUpSubscription {
15    checkpoint: SubscriptionCheckpoint,
16    delivered_through: Option<u64>,
17    batch_size: usize,
18}
19
20impl CatchUpSubscription {
21    /// Creates an explicitly acknowledged catch-up subscription.
22    ///
23    /// # Errors
24    ///
25    /// Rejects a zero batch size.
26    pub fn new(checkpoint: SubscriptionCheckpoint, batch_size: usize) -> Result<Self> {
27        if batch_size == 0 {
28            return Err(MemoryError::InvalidValue {
29                field: "subscription.batch_size",
30                reason: "must be greater than zero",
31            });
32        }
33        Ok(Self {
34            checkpoint,
35            delivered_through: None,
36            batch_size,
37        })
38    }
39
40    #[must_use]
41    pub const fn checkpoint(&self) -> SubscriptionCheckpoint {
42        self.checkpoint
43    }
44
45    /// Returns the next batch without advancing the durable checkpoint.
46    pub fn poll<E, S>(&mut self, store: &S) -> Vec<StoredEvent<E>>
47    where
48        E: Clone,
49        S: EventStore<E>,
50    {
51        let events = store.load_all(self.checkpoint.global_position, self.batch_size);
52        self.delivered_through = events.last().map(|event| event.metadata.global_position);
53        events
54    }
55
56    /// Acknowledges all delivered events through an inclusive position.
57    ///
58    /// # Errors
59    ///
60    /// Rejects acknowledgements beyond the last delivered event or behind the
61    /// current checkpoint.
62    pub fn acknowledge(&mut self, global_position: u64) -> Result<()> {
63        let delivered = self.delivered_through.ok_or(MemoryError::InvalidValue {
64            field: "subscription.acknowledge",
65            reason: "no events have been delivered",
66        })?;
67        if global_position > delivered
68            || self
69                .checkpoint
70                .global_position
71                .is_some_and(|current| global_position < current)
72        {
73            return Err(MemoryError::InvalidValue {
74                field: "subscription.acknowledge",
75                reason: "position is outside the delivered range",
76            });
77        }
78        self.checkpoint.global_position = Some(global_position);
79        self.delivered_through = None;
80        Ok(())
81    }
82}
weavatrix_memory/store/subscription.rs

weavatrix_memory/store/
subscription.rs
