xmtp_events/
filter.rs

1use std::sync::Arc;
2
3use crate::{ClientEvent, ConsentEntityKind, ContentTypeId, EventContext, EventKind};
4
5type InternalPredicate<I> = dyn Fn(&I) -> bool + Send + Sync;
6
7/// Selects public kinds and, for worker subscribers, internal facts.
8pub struct EventFilter<I> {
9    pub kinds: Vec<EventKind>,
10    pub group_ids: Option<Vec<Vec<u8>>>,
11    /// DM identifiers resolved from `group_ids` when the subscription starts.
12    pub dm_identifiers: Vec<Vec<u8>>,
13    pub content_types: Option<Vec<ContentTypeId>>,
14    pub references_own_messages: bool,
15    internal: Option<Arc<InternalPredicate<I>>>,
16}
17
18impl<I> Default for EventFilter<I> {
19    fn default() -> Self {
20        Self {
21            kinds: Vec::new(),
22            group_ids: None,
23            dm_identifiers: Vec::new(),
24            content_types: None,
25            references_own_messages: false,
26            internal: None,
27        }
28    }
29}
30
31impl<I> EventFilter<I> {
32    pub fn new(kinds: impl IntoIterator<Item = EventKind>) -> Self {
33        Self {
34            kinds: kinds.into_iter().collect(),
35            ..Self::default()
36        }
37    }
38
39    /// The predicate must not call any event bus. Such calls fail immediately.
40    pub fn with_internal(mut self, predicate: impl Fn(&I) -> bool + Send + Sync + 'static) -> Self {
41        self.internal = Some(Arc::new(predicate));
42        self
43    }
44
45    pub(crate) fn matches_internal(&self, value: &I) -> bool {
46        self.internal
47            .as_ref()
48            .is_some_and(|predicate| predicate(value))
49    }
50
51    pub(crate) fn matches_public(&self, event: &ClientEvent, context: &EventContext) -> bool {
52        if event.kind() == EventKind::Lagged {
53            return true;
54        }
55        if !self.kinds.contains(&event.kind()) {
56            return false;
57        }
58
59        if let Some(group_ids) = &self.group_ids {
60            let named_group = if let ClientEvent::ConsentChanged(consent) = event
61                && consent.entity_kind == ConsentEntityKind::Conversation
62            {
63                let Some(group_id) = decode_hex(&consent.entity) else {
64                    return false;
65                };
66                Some(group_id)
67            } else {
68                event.group_id().map(|id| id.to_vec())
69            };
70            if let Some(group_id) = named_group
71                && !group_ids.contains(&group_id)
72                && !context
73                    .dm_identifier
74                    .as_ref()
75                    .is_some_and(|dm| self.dm_identifiers.contains(dm))
76            {
77                return false;
78            }
79        }
80
81        if let ClientEvent::MessageReceived(message) = event {
82            if let Some(types) = &self.content_types
83                && !message
84                    .content_type
85                    .as_ref()
86                    .is_some_and(|id| types.contains(id))
87            {
88                return false;
89            }
90            if self.references_own_messages
91                && (!context.references_own_messages
92                    || !message.content_type.as_ref().is_some_and(|content_type| {
93                        content_type.authority_id == "xmtp.org"
94                            && matches!(
95                                (content_type.type_id.as_str(), content_type.version_major),
96                                ("reply", 1) | ("reaction", 2)
97                            )
98                    }))
99            {
100                return false;
101            }
102        }
103        true
104    }
105}
106
107fn decode_hex(value: &str) -> Option<Vec<u8>> {
108    let bytes = value.as_bytes();
109    if !bytes.len().is_multiple_of(2) {
110        return None;
111    }
112    bytes
113        .chunks_exact(2)
114        .map(|pair| {
115            let hi = (pair[0] as char).to_digit(16)?;
116            let lo = (pair[1] as char).to_digit(16)?;
117            Some(((hi << 4) | lo) as u8)
118        })
119        .collect()
120}
xmtp_events/filter.rs

xmtp_events/
filter.rs
