nexus_sdk/events/
query.rs

1//! Typed query for Nexus events.
2
3use {
4    super::{
5        event_struct_tag,
6        parsing::classify_nexus_event,
7        supports_event,
8        NexusEventCandidate,
9        NexusEventDecodeError,
10    },
11    crate::{
12        events::NexusEvent,
13        move_bindings::{
14            interface::distributed_event as distributed_event_move,
15            primitives::{data::NexusData as MoveNexusData, event as event_move},
16        },
17        nexus::state::StateResolver,
18        sui::{
19            self,
20            events::{EventIngestor, EventQuery},
21        },
22        types::{NexusContext, NexusObjects},
23    },
24    std::sync::Arc,
25    sui_rpc::{field::FieldMaskUtil as _, proto::sui::rpc::v2::filter::event as event_filter},
26};
27
28/// Known direct events ordered by their measured frequency in execution
29/// workloads. A filtered consumer can exclude the first eight events it does
30/// not use while retaining both wrapper heads inside Sui's ten literal
31/// default. The generated event catalog validates these names, while their
32/// order remains explicit because the IR does not contain workload frequency.
33const DIRECT_FILTER_EXCLUSION_CANDIDATES: [&str; 8] = [
34    "ExecutionPaymentFeesRecordedEvent",
35    "GasPaymentConsumedEvent",
36    "InvocationLockedEvent",
37    "InvocationSettledEvent",
38    "PriorityFeeDepositCreatedEvent",
39    "ToolVerificationResolvedEvent",
40    "ExecutionPaymentToolCostSnapshottedEvent",
41    "TaskCreatedEvent",
42];
43
44/// Decodes events using the immutable package graph selected by each emitter.
45///
46/// This value does not resolve live object authority. It uses emitter package
47/// metadata only to select the historical event layout.
48#[derive(Clone)]
49pub struct NexusEventDecoder {
50    resolver: StateResolver,
51    objects: Arc<NexusObjects>,
52}
53
54impl NexusEventDecoder {
55    /// Creates a decoder backed by `resolver` and stable environment identity.
56    pub fn new(resolver: StateResolver, objects: Arc<NexusObjects>) -> Self {
57        Self { resolver, objects }
58    }
59
60    /// Decodes one Sui transaction event using its emitter package graph.
61    ///
62    /// Returns [`None`] when the event is not a Nexus event.
63    ///
64    /// # Errors
65    ///
66    /// Returns [`NexusEventDecodeError`] when emitter metadata cannot be
67    /// resolved, event contents cannot be decoded, or the inner event type is
68    /// unsupported.
69    pub async fn decode_sui_event(
70        &self,
71        index: u64,
72        digest: sui::types::Digest,
73        event: &sui::types::Event,
74    ) -> Result<Option<NexusEvent>, NexusEventDecodeError> {
75        self.decode_parts(
76            index,
77            digest,
78            event.package_id,
79            &event.type_,
80            &event.contents,
81        )
82        .await
83    }
84
85    async fn decode_parts(
86        &self,
87        index: u64,
88        digest: sui::types::Digest,
89        emitting_package: sui::types::Address,
90        wrapper_type: &sui::types::StructTag,
91        contents: &[u8],
92    ) -> Result<Option<NexusEvent>, NexusEventDecodeError> {
93        if !crate::move_bindings::struct_shape_matches::<event_move::EventWrapper<MoveNexusData>>(
94            wrapper_type,
95        ) && !crate::move_bindings::struct_shape_matches::<
96            distributed_event_move::DistributedEventWrapper<MoveNexusData>,
97        >(wrapper_type)
98        {
99            return Ok(None);
100        }
101        let context = self
102            .resolver
103            .resolve_emitter_context(Arc::clone(&self.objects), emitting_package)
104            .await
105            .map_err(|source| NexusEventDecodeError::EmitterPackage {
106                package: emitting_package,
107                source,
108            })?;
109        match classify_parts(
110            &context,
111            index,
112            digest,
113            emitting_package,
114            wrapper_type,
115            contents,
116        )? {
117            Some(candidate) => candidate
118                .into_supported()
119                .map(Some)
120                .map_err(NexusEventDecodeError::from),
121            None => Ok(None),
122        }
123    }
124}
125
126/// Query that filters event wrappers using one current package graph.
127///
128/// Event emitter metadata chooses historical code identity. Live execution
129/// authority must be resolved separately from current object witnesses.
130#[derive(Clone)]
131pub struct NexusEventQuery {
132    filter_context: Arc<NexusContext>,
133    decoder: NexusEventDecoder,
134    supported_event_filter: Option<fn(&str) -> bool>,
135}
136
137impl NexusEventQuery {
138    /// Creates a query whose server filter uses `filter_context`.
139    pub fn new(filter_context: Arc<NexusContext>, decoder: NexusEventDecoder) -> Self {
140        Self {
141            filter_context,
142            decoder,
143            supported_event_filter: None,
144        }
145    }
146
147    /// Keeps compiled protocol events accepted by `filter`.
148    ///
149    /// Event names absent from the compiled protocol catalog are still
150    /// decoded. This preserves the fail closed behavior for a compatible
151    /// package that emits an event unknown to this SDK.
152    pub fn with_supported_event_filter(mut self, filter: fn(&str) -> bool) -> Self {
153        self.supported_event_filter = Some(filter);
154        self
155    }
156
157    /// Decodes one Sui transaction event using this query.
158    ///
159    /// Returns [`None`] when the event is not a Nexus event.
160    ///
161    /// # Errors
162    ///
163    /// Returns [`NexusEventDecodeError`] when event contents cannot be decoded
164    /// or the inner Nexus event type is unsupported.
165    pub async fn decode_sui_event(
166        &self,
167        index: u64,
168        digest: sui::types::Digest,
169        event: &sui::types::Event,
170    ) -> Result<Option<NexusEvent>, NexusEventDecodeError> {
171        self.decoder.decode_sui_event(index, digest, event).await
172    }
173}
174
175fn classify_parts(
176    context: &NexusContext,
177    index: u64,
178    digest: sui::types::Digest,
179    source_package: sui::types::Address,
180    wrapper_type: &sui::types::StructTag,
181    contents: &[u8],
182) -> Result<Option<NexusEventCandidate>, NexusEventDecodeError> {
183    classify_nexus_event(
184        index,
185        digest,
186        source_package,
187        contents,
188        wrapper_type,
189        context,
190    )
191}
192
193impl EventQuery for NexusEventQuery {
194    type Error = NexusEventDecodeError;
195    type Output = NexusEventCandidate;
196
197    fn filter(&self) -> sui::grpc::EventFilter {
198        nexus_server_filter(&self.filter_context, self.supported_event_filter)
199    }
200
201    fn read_mask(&self) -> sui::grpc::FieldMask {
202        sui::grpc::FieldMask::from_paths(["package_id", "event_type", "contents"])
203    }
204
205    async fn decode(&self, event: sui::grpc::Event) -> Result<Option<Self::Output>, Self::Error> {
206        let wrapper_type = event
207            .event_type_opt()
208            .ok_or(NexusEventDecodeError::MissingField("event_type"))?
209            .parse()?;
210        if self
211            .supported_event_filter
212            .is_some_and(|filter| !should_decode_event(&wrapper_type, filter))
213        {
214            return Ok(None);
215        }
216        let digest = event
217            .transaction_digest
218            .as_deref()
219            .ok_or_else(|| {
220                NexusEventDecodeError::Identity("transaction digest is missing".to_owned())
221            })?
222            .parse()
223            .map_err(|error| {
224                NexusEventDecodeError::Identity(format!("transaction digest is invalid: {error}"))
225            })?;
226        let event_index = event
227            .event_index
228            .ok_or_else(|| NexusEventDecodeError::Identity("event index is missing".to_owned()))?;
229        let contents = event
230            .contents_opt()
231            .and_then(|contents| contents.value_opt())
232            .ok_or(NexusEventDecodeError::MissingField("contents"))?;
233        let emitting_package = event
234            .package_id
235            .as_deref()
236            .ok_or(NexusEventDecodeError::MissingField("package_id"))?
237            .parse()
238            .map_err(|error| {
239                NexusEventDecodeError::Identity(format!("emitting package is invalid: {error}"))
240            })?;
241
242        let context = self
243            .decoder
244            .resolver
245            .resolve_emitter_context(Arc::clone(&self.decoder.objects), emitting_package)
246            .await
247            .map_err(|source| NexusEventDecodeError::EmitterPackage {
248                package: emitting_package,
249                source,
250            })?;
251        classify_parts(
252            &context,
253            event_index.into(),
254            digest,
255            emitting_package,
256            &wrapper_type,
257            contents,
258        )
259    }
260}
261
262fn nexus_server_filter(
263    context: &NexusContext,
264    supported_event_filter: Option<fn(&str) -> bool>,
265) -> sui::grpc::EventFilter {
266    let wrapper =
267        crate::move_bindings::struct_tag::<event_move::EventWrapper<MoveNexusData>>(context);
268    let distributed_wrapper = crate::move_bindings::struct_tag::<
269        distributed_event_move::DistributedEventWrapper<MoveNexusData>,
270    >(context);
271
272    let mut direct = vec![event_filter::event_type(wrapper_head(&wrapper))];
273    if let Some(filter) = supported_event_filter {
274        direct.extend(
275            DIRECT_FILTER_EXCLUSION_CANDIDATES
276                .into_iter()
277                .filter(|name| !filter(name))
278                .filter_map(|name| event_struct_tag(context, name))
279                .map(|event| {
280                    event_filter::event_type(instantiate_wrapper(&wrapper, event)).negate()
281                }),
282        );
283    }
284
285    sui::grpc::EventFilter::any([
286        sui::grpc::EventTerm::all(direct),
287        sui::grpc::EventTerm::all([event_filter::event_type(wrapper_head(&distributed_wrapper))]),
288    ])
289}
290
291fn wrapper_head(wrapper: &sui::types::StructTag) -> String {
292    format!(
293        "{}::{}::{}",
294        wrapper.address(),
295        wrapper.module(),
296        wrapper.name()
297    )
298}
299
300fn instantiate_wrapper(
301    wrapper: &sui::types::StructTag,
302    event: sui::types::StructTag,
303) -> sui::types::StructTag {
304    sui::types::StructTag::new(
305        *wrapper.address(),
306        wrapper.module().clone(),
307        wrapper.name().clone(),
308        vec![sui::types::TypeTag::Struct(Box::new(event))],
309    )
310}
311
312fn wrapped_event_name(wrapper: &sui::types::StructTag) -> Option<&str> {
313    let sui::types::TypeTag::Struct(event) = wrapper.type_params().first()? else {
314        return None;
315    };
316    Some(event.name().as_str())
317}
318
319fn should_decode_event(wrapper: &sui::types::StructTag, filter: fn(&str) -> bool) -> bool {
320    wrapped_event_name(wrapper).is_none_or(|name| !supports_event(name) || filter(name))
321}
322
323/// [`EventIngestor`] configured by [`NexusEventQuery`].
324pub type NexusEventIngestor = EventIngestor<NexusEventQuery>;
325
326#[cfg(test)]
327mod tests {
328    use super::*;
329
330    fn event_wrapper(name: &str) -> sui::types::StructTag {
331        sui::types::StructTag::new(
332            sui::types::Address::from_static("0x1"),
333            "event".parse().unwrap(),
334            "EventWrapper".parse().unwrap(),
335            vec![sui::types::TypeTag::Struct(Box::new(
336                sui::types::StructTag::new(
337                    sui::types::Address::from_static("0x2"),
338                    "events".parse().unwrap(),
339                    name.parse().unwrap(),
340                    Vec::new(),
341                ),
342            ))],
343        )
344    }
345
346    #[test]
347    fn extracts_wrapped_event_name() {
348        assert_eq!(
349            wrapped_event_name(&event_wrapper("TaskCreatedEvent")),
350            Some("TaskCreatedEvent")
351        );
352    }
353
354    #[test]
355    fn filter_omits_known_events_but_retains_unknown_events() {
356        let only_tasks = |name: &str| name == "TaskCreatedEvent";
357
358        assert!(should_decode_event(
359            &event_wrapper("TaskCreatedEvent"),
360            only_tasks
361        ));
362        assert!(!should_decode_event(
363            &event_wrapper("DAGCreatedEvent"),
364            only_tasks
365        ));
366        assert!(should_decode_event(
367            &event_wrapper("FutureProtocolEvent"),
368            only_tasks
369        ));
370    }
371
372    #[test]
373    fn direct_filter_exclusions_match_the_generated_event_catalog() {
374        for (index, name) in DIRECT_FILTER_EXCLUSION_CANDIDATES.iter().enumerate() {
375            assert!(
376                crate::events::NexusEventKind::NAMES.contains(name),
377                "{name} is absent from the generated event catalog"
378            );
379            assert!(
380                !DIRECT_FILTER_EXCLUSION_CANDIDATES[..index].contains(name),
381                "{name} appears more than once"
382            );
383        }
384    }
385
386    #[cfg(feature = "test_utils")]
387    #[test]
388    fn server_filter_uses_the_portable_literal_budget() {
389        let context = crate::test_utils::sui_mocks::mock_nexus_context();
390        let filter = nexus_server_filter(&context, Some(|_| false));
391
392        assert_eq!(filter.terms.len(), 2);
393        assert_eq!(
394            filter
395                .terms
396                .iter()
397                .map(|term| term.literals.len())
398                .sum::<usize>(),
399            10
400        );
401        assert!(!filter.terms[0].literals[0].negated);
402        assert!(filter.terms[0].literals[1..]
403            .iter()
404            .all(|literal| literal.negated));
405        for literal in &filter.terms[0].literals[1..] {
406            let Some(sui::grpc::event_literal::Predicate::EventType(event_type)) =
407                &literal.predicate
408            else {
409                panic!("exclusion is not an event type")
410            };
411            let wrapper: sui::types::StructTag = event_type
412                .event_type
413                .as_deref()
414                .expect("event type is present")
415                .parse()
416                .expect("event type is valid");
417            let Some(sui::types::TypeTag::Struct(event)) = wrapper.type_params().first() else {
418                panic!("excluded wrapper has no inner event")
419            };
420            assert_ne!(*event.address(), sui::types::Address::ZERO);
421        }
422        assert_eq!(filter.terms[1].literals.len(), 1);
423        assert!(!filter.terms[1].literals[0].negated);
424    }
425
426    #[cfg(feature = "test_utils")]
427    #[test]
428    fn unfiltered_query_retains_both_wrapper_heads_without_exclusions() {
429        let context = crate::test_utils::sui_mocks::mock_nexus_context();
430        let filter = nexus_server_filter(&context, None);
431
432        assert_eq!(filter.terms.len(), 2);
433        assert!(filter
434            .terms
435            .iter()
436            .all(|term| term.literals.len() == 1 && !term.literals[0].negated));
437    }
438}
nexus_sdk/events/query.rs

nexus_sdk/events/
query.rs
