weavatrix_memory/store/
subscription.rs

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

weavatrix_memory/store/
subscription.rs
