whatsapp_rust/plugins/
events.rs

1use std::collections::{HashMap, HashSet};
2use std::fmt;
3use std::sync::{Arc, Mutex, RwLock};
4
5use async_channel::{Receiver, Sender, TryRecvError, TrySendError};
6use bytes::Bytes;
7use portable_atomic::{AtomicBool, AtomicU64, Ordering};
8use thiserror::Error;
9
10use super::{PluginResourceError, PluginResources, valid_plugin_id};
11
12const MAX_ENDPOINT_CAPACITY: usize = 65_536;
13const MAX_ENDPOINT_SELECTORS: usize = 1_024;
14
15/// Encoding of a custom plugin event payload.
16#[derive(Debug, Clone, Copy, PartialEq, Eq)]
17#[non_exhaustive]
18pub enum PluginEventPayloadEncoding {
19    Json,
20    Binary,
21}
22
23impl PluginEventPayloadEncoding {
24    pub const fn identifier(self) -> &'static str {
25        match self {
26            Self::Json => "json",
27            Self::Binary => "binary",
28        }
29    }
30}
31
32/// Validated second-level topic within one plugin namespace.
33#[derive(Clone, PartialEq, Eq, Hash)]
34pub struct PluginEventTopic(Arc<str>);
35
36impl PluginEventTopic {
37    pub fn new(topic: impl Into<String>) -> Result<Self, PluginEventRouteError> {
38        let topic = topic.into();
39        if !valid_topic(&topic) {
40            return Err(PluginEventRouteError::InvalidTopic { topic });
41        }
42        Ok(Self(Arc::from(topic)))
43    }
44
45    pub fn as_str(&self) -> &str {
46        &self.0
47    }
48}
49
50impl fmt::Debug for PluginEventTopic {
51    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
52        formatter
53            .debug_tuple("PluginEventTopic")
54            .field(&self.0)
55            .finish()
56    }
57}
58
59impl fmt::Display for PluginEventTopic {
60    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
61        formatter.write_str(&self.0)
62    }
63}
64
65/// Exact `(plugin_id, topic)` route selected by one endpoint.
66#[derive(Debug, Clone, PartialEq, Eq, Hash)]
67pub struct PluginEventSelector {
68    route: RouteKey,
69}
70
71impl PluginEventSelector {
72    pub fn new(
73        plugin_id: impl Into<String>,
74        topic: PluginEventTopic,
75    ) -> Result<Self, PluginEventRouteError> {
76        let plugin_id = plugin_id.into();
77        if !valid_plugin_id(&plugin_id) {
78            return Err(PluginEventRouteError::InvalidPluginId { plugin_id });
79        }
80        Ok(Self {
81            route: RouteKey {
82                plugin_id: Arc::from(plugin_id),
83                topic,
84            },
85        })
86    }
87
88    pub fn plugin_id(&self) -> &str {
89        &self.route.plugin_id
90    }
91
92    pub fn topic(&self) -> &PluginEventTopic {
93        &self.route.topic
94    }
95}
96
97#[derive(Debug, Clone, PartialEq, Eq, Hash)]
98struct RouteKey {
99    plugin_id: Arc<str>,
100    topic: PluginEventTopic,
101}
102
103/// Routed event shared by every matching endpoint without copying its payload.
104#[derive(Debug, Clone, bon::Builder)]
105#[non_exhaustive]
106pub struct PluginEventEnvelope {
107    pub plugin_id: Arc<str>,
108    pub topic: PluginEventTopic,
109    pub schema_version: u32,
110    pub payload_encoding: PluginEventPayloadEncoding,
111    pub payload: Bytes,
112    pub connection_generation: u64,
113    /// Monotonic sequence for this route while it has at least one subscriber.
114    ///
115    /// Dropped events consume a sequence number, allowing one endpoint to detect loss. The
116    /// sequence resets after the last subscriber to the route is removed.
117    pub sequence: u64,
118}
119
120/// Behavior when one endpoint cannot keep up with publishers.
121#[derive(Debug, Clone, Copy, PartialEq, Eq)]
122#[non_exhaustive]
123pub enum PluginEventOverflow {
124    DropNewest,
125    DropOldest,
126}
127
128/// Required queue policy for one independent consumer endpoint.
129///
130/// Capacity counts envelopes rather than bytes. Native plugins are trusted, and payloads are
131/// shared across matching endpoints. A foreign adapter must enforce its wire payload limit before
132/// publishing into this router.
133#[derive(Debug, Clone, Copy, PartialEq, Eq)]
134pub struct PluginEventEndpointConfig {
135    capacity: usize,
136    overflow: PluginEventOverflow,
137}
138
139impl PluginEventEndpointConfig {
140    pub const fn new(capacity: usize, overflow: PluginEventOverflow) -> Self {
141        Self { capacity, overflow }
142    }
143
144    pub const fn capacity(self) -> usize {
145        self.capacity
146    }
147
148    pub const fn overflow(self) -> PluginEventOverflow {
149        self.overflow
150    }
151}
152
153/// Syntactic route validation failure.
154#[derive(Debug, Error, Clone, PartialEq, Eq)]
155#[non_exhaustive]
156pub enum PluginEventRouteError {
157    #[error("invalid plugin id `{plugin_id}`")]
158    InvalidPluginId { plugin_id: String },
159    #[error("invalid plugin event topic `{topic}`")]
160    InvalidTopic { topic: String },
161}
162
163/// Endpoint registration failure.
164#[derive(Debug, Error, Clone, PartialEq, Eq)]
165#[non_exhaustive]
166pub enum PluginEventSubscribeError {
167    #[error("{0}")]
168    Resource(#[from] PluginResourceError),
169    #[error("at least one plugin event selector is required")]
170    EmptySelectors,
171    #[error("plugin event endpoint selector count exceeds the maximum of {max}")]
172    TooManySelectors { max: usize },
173    #[error("plugin event endpoint capacity {capacity} is outside 1..={max}")]
174    InvalidCapacity { capacity: usize, max: usize },
175    #[error("plugin `{plugin_id}` is not registered as a custom-event publisher")]
176    UnknownPublisher { plugin_id: String },
177    #[error("plugin event endpoint identifiers are exhausted")]
178    EndpointIdsExhausted,
179    #[error("the plugin event router is closed")]
180    Closed,
181}
182
183/// Custom event publication failure.
184#[derive(Debug, Error, Clone, Copy, PartialEq, Eq)]
185#[non_exhaustive]
186pub enum PluginEventPublishError {
187    #[error("{0}")]
188    Resource(#[from] PluginResourceError),
189    #[error("plugin event schema version must be greater than zero")]
190    InvalidSchemaVersion,
191    #[error("the plugin event router is closed")]
192    Closed,
193    #[error("the plugin event sequence is exhausted")]
194    SequenceExhausted,
195}
196
197/// Result of one non-blocking fan-out attempt.
198///
199/// `dropped` counts queue entries discarded while processing this call. Under `DropOldest`, the
200/// discarded entry may belong to an earlier publication from another namespace; cumulative
201/// publisher statistics attribute that loss to the discarded envelope's owner.
202#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
203#[non_exhaustive]
204pub struct PluginEventPublishReport {
205    pub matched: u64,
206    pub enqueued: u64,
207    pub dropped: u64,
208    pub closed: u64,
209}
210
211/// Cumulative state for one endpoint queue.
212#[derive(Debug, Clone, Copy, PartialEq, Eq)]
213#[non_exhaustive]
214pub struct PluginEventEndpointStats {
215    pub enqueued: u64,
216    pub delivered: u64,
217    pub dropped: u64,
218    pub queue_depth: usize,
219    pub capacity: usize,
220}
221
222/// Cumulative publication and fanout counters for one plugin namespace.
223///
224/// `published` counts successful calls, including calls with no subscriber. Fanout fields count
225/// endpoint outcomes; `delivered` advances only when a receiver removes an envelope from its queue.
226/// `dropped` follows the discarded envelope, including cross-namespace `DropOldest` eviction.
227#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
228#[non_exhaustive]
229pub struct PluginEventPublisherStats {
230    pub published: u64,
231    pub publish_failures: u64,
232    pub matched: u64,
233    pub enqueued: u64,
234    pub delivered: u64,
235    pub dropped: u64,
236    pub closed: u64,
237}
238
239/// On-demand aggregate for the custom-event router.
240///
241/// Current occupancy may move while a concurrent snapshot is being assembled; cumulative counters
242/// remain monotonic but are not an atomic cross-publisher transaction.
243#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
244#[non_exhaustive]
245pub struct PluginEventRouterStats {
246    pub registered_publishers: u64,
247    pub active_routes: u64,
248    pub active_endpoints: u64,
249    pub endpoint_capacity: u64,
250    /// Unique event envelopes retained by at least one endpoint queue.
251    pub queued_events: u64,
252    /// Payload bytes retained by those unique queued envelopes.
253    pub queued_payload_bytes: u64,
254    pub published: u64,
255    pub publish_failures: u64,
256    pub matched: u64,
257    pub enqueued: u64,
258    pub delivered: u64,
259    pub dropped: u64,
260    pub closed: u64,
261}
262
263#[derive(Debug, Error, Clone, Copy, PartialEq, Eq)]
264#[non_exhaustive]
265#[error("the plugin event endpoint is closed")]
266pub struct PluginEventReceiveError;
267
268#[derive(Debug, Error, Clone, Copy, PartialEq, Eq)]
269#[non_exhaustive]
270pub enum PluginEventTryReceiveError {
271    #[error("the plugin event endpoint queue is empty")]
272    Empty,
273    #[error("the plugin event endpoint is closed")]
274    Closed,
275}
276
277enum EnqueueOutcome {
278    Enqueued,
279    Dropped,
280    EnqueuedAfterDrop,
281    Closed,
282}
283
284struct PluginEventPublication {
285    schema_version: u32,
286    payload_encoding: PluginEventPayloadEncoding,
287    payload: Bytes,
288    connection_generation: u64,
289}
290
291#[derive(Default)]
292struct PublisherCounters {
293    published: AtomicU64,
294    publish_failures: AtomicU64,
295    matched: AtomicU64,
296    enqueued: AtomicU64,
297    delivered: AtomicU64,
298    dropped: AtomicU64,
299    closed: AtomicU64,
300}
301
302impl PublisherCounters {
303    fn record_publish(&self, result: &Result<PluginEventPublishReport, PluginEventPublishError>) {
304        match result {
305            Ok(report) => {
306                self.published.fetch_add(1, Ordering::Relaxed);
307                self.matched.fetch_add(report.matched, Ordering::Relaxed);
308                self.enqueued.fetch_add(report.enqueued, Ordering::Relaxed);
309                self.closed.fetch_add(report.closed, Ordering::Relaxed);
310            }
311            Err(_) => {
312                self.publish_failures.fetch_add(1, Ordering::Relaxed);
313            }
314        }
315    }
316
317    fn snapshot(&self) -> PluginEventPublisherStats {
318        PluginEventPublisherStats {
319            published: self.published.load(Ordering::Relaxed),
320            publish_failures: self.publish_failures.load(Ordering::Relaxed),
321            matched: self.matched.load(Ordering::Relaxed),
322            enqueued: self.enqueued.load(Ordering::Relaxed),
323            delivered: self.delivered.load(Ordering::Relaxed),
324            dropped: self.dropped.load(Ordering::Relaxed),
325            closed: self.closed.load(Ordering::Relaxed),
326        }
327    }
328}
329
330#[derive(Default)]
331struct QueueMemory {
332    events: AtomicU64,
333    payload_bytes: AtomicU64,
334}
335
336struct QueuedPluginEvent {
337    envelope: Arc<PluginEventEnvelope>,
338    publisher: Arc<PublisherCounters>,
339    memory: Arc<QueueMemory>,
340    payload_bytes: u64,
341}
342
343impl QueuedPluginEvent {
344    fn new(
345        envelope: Arc<PluginEventEnvelope>,
346        publisher: Arc<PublisherCounters>,
347        memory: Arc<QueueMemory>,
348    ) -> Arc<Self> {
349        let payload_bytes = u64::try_from(envelope.payload.len()).unwrap_or(u64::MAX);
350        memory.events.fetch_add(1, Ordering::Relaxed);
351        memory
352            .payload_bytes
353            .fetch_add(payload_bytes, Ordering::Relaxed);
354        Arc::new(Self {
355            envelope,
356            publisher,
357            memory,
358            payload_bytes,
359        })
360    }
361}
362
363impl Drop for QueuedPluginEvent {
364    fn drop(&mut self) {
365        self.memory.events.fetch_sub(1, Ordering::Relaxed);
366        self.memory
367            .payload_bytes
368            .fetch_sub(self.payload_bytes, Ordering::Relaxed);
369    }
370}
371
372struct EventEndpoint {
373    id: u64,
374    sender: Sender<Arc<QueuedPluginEvent>>,
375    overflow: PluginEventOverflow,
376    capacity: usize,
377    enqueued: AtomicU64,
378    delivered: AtomicU64,
379    dropped: AtomicU64,
380}
381
382impl EventEndpoint {
383    fn enqueue(&self, event: Arc<QueuedPluginEvent>) -> EnqueueOutcome {
384        match self.overflow {
385            PluginEventOverflow::DropNewest => match self.sender.try_send(event) {
386                Ok(()) => {
387                    self.enqueued.fetch_add(1, Ordering::Relaxed);
388                    EnqueueOutcome::Enqueued
389                }
390                Err(TrySendError::Full(dropped)) => {
391                    self.dropped.fetch_add(1, Ordering::Relaxed);
392                    dropped.publisher.dropped.fetch_add(1, Ordering::Relaxed);
393                    EnqueueOutcome::Dropped
394                }
395                Err(TrySendError::Closed(_)) => EnqueueOutcome::Closed,
396            },
397            PluginEventOverflow::DropOldest => match self.sender.force_send(event) {
398                Ok(evicted) => {
399                    self.enqueued.fetch_add(1, Ordering::Relaxed);
400                    if let Some(evicted) = evicted {
401                        self.dropped.fetch_add(1, Ordering::Relaxed);
402                        evicted.publisher.dropped.fetch_add(1, Ordering::Relaxed);
403                        EnqueueOutcome::EnqueuedAfterDrop
404                    } else {
405                        EnqueueOutcome::Enqueued
406                    }
407                }
408                Err(_) => EnqueueOutcome::Closed,
409            },
410        }
411    }
412
413    fn close(&self) {
414        self.sender.close();
415    }
416
417    fn stats(&self) -> PluginEventEndpointStats {
418        PluginEventEndpointStats {
419            enqueued: self.enqueued.load(Ordering::Relaxed),
420            delivered: self.delivered.load(Ordering::Relaxed),
421            dropped: self.dropped.load(Ordering::Relaxed),
422            queue_depth: self.sender.len(),
423            capacity: self.capacity,
424        }
425    }
426}
427
428struct RouteClock {
429    sequence: Mutex<u64>,
430}
431
432struct RouteEntry {
433    clock: Arc<RouteClock>,
434    endpoints: Arc<[Arc<EventEndpoint>]>,
435}
436
437#[derive(Default)]
438struct RouterState {
439    routes: HashMap<RouteKey, RouteEntry>,
440    endpoints: HashMap<u64, Arc<EventEndpoint>>,
441}
442
443struct PluginEventRouterInner {
444    publishers: HashMap<Arc<str>, Arc<PublisherCounters>>,
445    queue_memory: Arc<QueueMemory>,
446    state: RwLock<RouterState>,
447    next_endpoint_id: AtomicU64,
448    closed: AtomicBool,
449}
450
451impl PluginEventRouterInner {
452    fn unsubscribe(&self, endpoint_id: u64, selectors: &[PluginEventSelector]) {
453        let endpoint = {
454            let mut state = self
455                .state
456                .write()
457                .unwrap_or_else(|poisoned| poisoned.into_inner());
458            let endpoint = state.endpoints.remove(&endpoint_id);
459            for selector in selectors {
460                let remove_route = if let Some(route) = state.routes.get_mut(&selector.route) {
461                    let remaining = route
462                        .endpoints
463                        .iter()
464                        .filter(|endpoint| endpoint.id != endpoint_id)
465                        .cloned()
466                        .collect::<Vec<_>>();
467                    route.endpoints = remaining.into();
468                    route.endpoints.is_empty()
469                } else {
470                    false
471                };
472                if remove_route {
473                    state.routes.remove(&selector.route);
474                }
475            }
476            endpoint
477        };
478        if let Some(endpoint) = endpoint {
479            endpoint.close();
480        }
481    }
482
483    fn close(&self) {
484        if self.closed.swap(true, Ordering::AcqRel) {
485            return;
486        }
487        let endpoints = {
488            let mut state = self
489                .state
490                .write()
491                .unwrap_or_else(|poisoned| poisoned.into_inner());
492            state.routes.clear();
493            std::mem::take(&mut state.endpoints)
494        };
495        for endpoint in endpoints.into_values() {
496            endpoint.close();
497        }
498    }
499}
500
501/// Read-only subscription boundary for native consumers and future foreign adapters.
502///
503/// Routes are exact `(plugin_id, topic)` matches. Closing the router prevents new publications and
504/// subscriptions, while already queued envelopes remain available before receivers observe closure.
505#[derive(Clone)]
506pub struct PluginEventRouter {
507    inner: Arc<PluginEventRouterInner>,
508}
509
510impl PluginEventRouter {
511    pub(super) fn new(plugin_ids: impl IntoIterator<Item = String>) -> Self {
512        let publishers = plugin_ids
513            .into_iter()
514            .map(|plugin_id| (Arc::from(plugin_id), Arc::new(PublisherCounters::default())))
515            .collect();
516        Self {
517            inner: Arc::new(PluginEventRouterInner {
518                publishers,
519                queue_memory: Arc::new(QueueMemory::default()),
520                state: RwLock::new(RouterState::default()),
521                next_endpoint_id: AtomicU64::new(1),
522                closed: AtomicBool::new(false),
523            }),
524        }
525    }
526
527    pub fn has_subscribers(&self, selector: &PluginEventSelector) -> bool {
528        if self.inner.closed.load(Ordering::Acquire) {
529            return false;
530        }
531        self.inner
532            .state
533            .read()
534            .unwrap_or_else(|poisoned| poisoned.into_inner())
535            .routes
536            .get(&selector.route)
537            .is_some_and(|route| !route.endpoints.is_empty())
538    }
539
540    /// Cumulative counters for one registered publisher.
541    pub fn publisher_stats(&self, plugin_id: &str) -> Option<PluginEventPublisherStats> {
542        self.inner
543            .publishers
544            .get(plugin_id)
545            .map(|stats| stats.snapshot())
546    }
547
548    /// Aggregate counters and current queue occupancy.
549    pub fn stats(&self) -> PluginEventRouterStats {
550        let (active_routes, active_endpoints, endpoint_capacity) = {
551            let state = self
552                .inner
553                .state
554                .read()
555                .unwrap_or_else(|poisoned| poisoned.into_inner());
556            let endpoint_capacity = state.endpoints.values().fold(0u64, |total, endpoint| {
557                total.saturating_add(u64::try_from(endpoint.capacity).unwrap_or(u64::MAX))
558            });
559            (
560                u64::try_from(state.routes.len()).unwrap_or(u64::MAX),
561                u64::try_from(state.endpoints.len()).unwrap_or(u64::MAX),
562                endpoint_capacity,
563            )
564        };
565        let mut snapshot = PluginEventRouterStats {
566            registered_publishers: u64::try_from(self.inner.publishers.len()).unwrap_or(u64::MAX),
567            active_routes,
568            active_endpoints,
569            endpoint_capacity,
570            queued_events: self.inner.queue_memory.events.load(Ordering::Relaxed),
571            queued_payload_bytes: self
572                .inner
573                .queue_memory
574                .payload_bytes
575                .load(Ordering::Relaxed),
576            ..PluginEventRouterStats::default()
577        };
578        for publisher in self.inner.publishers.values() {
579            let publisher = publisher.snapshot();
580            snapshot.published = snapshot.published.saturating_add(publisher.published);
581            snapshot.publish_failures = snapshot
582                .publish_failures
583                .saturating_add(publisher.publish_failures);
584            snapshot.matched = snapshot.matched.saturating_add(publisher.matched);
585            snapshot.enqueued = snapshot.enqueued.saturating_add(publisher.enqueued);
586            snapshot.delivered = snapshot.delivered.saturating_add(publisher.delivered);
587            snapshot.dropped = snapshot.dropped.saturating_add(publisher.dropped);
588            snapshot.closed = snapshot.closed.saturating_add(publisher.closed);
589        }
590        snapshot
591    }
592
593    pub fn subscribe(
594        &self,
595        selectors: impl IntoIterator<Item = PluginEventSelector>,
596        config: PluginEventEndpointConfig,
597    ) -> Result<PluginEventSubscription, PluginEventSubscribeError> {
598        if config.capacity == 0 || config.capacity > MAX_ENDPOINT_CAPACITY {
599            return Err(PluginEventSubscribeError::InvalidCapacity {
600                capacity: config.capacity,
601                max: MAX_ENDPOINT_CAPACITY,
602            });
603        }
604        if self.inner.closed.load(Ordering::Acquire) {
605            return Err(PluginEventSubscribeError::Closed);
606        }
607
608        let mut seen = HashSet::new();
609        let mut unique_selectors = Vec::new();
610        for selector in selectors {
611            if !seen.insert(selector.route.clone()) {
612                continue;
613            }
614            if unique_selectors.len() == MAX_ENDPOINT_SELECTORS {
615                return Err(PluginEventSubscribeError::TooManySelectors {
616                    max: MAX_ENDPOINT_SELECTORS,
617                });
618            }
619            unique_selectors.push(selector);
620        }
621        let selectors = unique_selectors;
622        if selectors.is_empty() {
623            return Err(PluginEventSubscribeError::EmptySelectors);
624        }
625        for selector in &selectors {
626            if !self.inner.publishers.contains_key(selector.plugin_id()) {
627                return Err(PluginEventSubscribeError::UnknownPublisher {
628                    plugin_id: selector.plugin_id().to_string(),
629                });
630            }
631        }
632
633        let endpoint_id = self
634            .inner
635            .next_endpoint_id
636            .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |id| id.checked_add(1))
637            .map_err(|_| PluginEventSubscribeError::EndpointIdsExhausted)?;
638        let (sender, receiver) = async_channel::bounded(config.capacity);
639        let endpoint = Arc::new(EventEndpoint {
640            id: endpoint_id,
641            sender,
642            overflow: config.overflow,
643            capacity: config.capacity,
644            enqueued: AtomicU64::new(0),
645            delivered: AtomicU64::new(0),
646            dropped: AtomicU64::new(0),
647        });
648
649        {
650            let mut state = self
651                .inner
652                .state
653                .write()
654                .unwrap_or_else(|poisoned| poisoned.into_inner());
655            if self.inner.closed.load(Ordering::Acquire) {
656                return Err(PluginEventSubscribeError::Closed);
657            }
658            state.endpoints.insert(endpoint_id, endpoint.clone());
659            for selector in &selectors {
660                let route = state
661                    .routes
662                    .entry(selector.route.clone())
663                    .or_insert_with(|| RouteEntry {
664                        clock: Arc::new(RouteClock {
665                            sequence: Mutex::new(0),
666                        }),
667                        endpoints: Arc::from([]),
668                    });
669                let endpoints = route
670                    .endpoints
671                    .iter()
672                    .cloned()
673                    .chain(std::iter::once(endpoint.clone()))
674                    .collect::<Vec<_>>();
675                route.endpoints = endpoints.into();
676            }
677        }
678
679        Ok(PluginEventSubscription {
680            router: self.clone(),
681            endpoint,
682            receiver,
683            selectors,
684        })
685    }
686
687    fn publish(
688        &self,
689        publisher: Arc<PublisherCounters>,
690        plugin_id: &Arc<str>,
691        topic: &PluginEventTopic,
692        publication: PluginEventPublication,
693    ) -> Result<PluginEventPublishReport, PluginEventPublishError> {
694        let result = self.publish_inner(Arc::clone(&publisher), plugin_id, topic, publication);
695        publisher.record_publish(&result);
696        result
697    }
698
699    fn publish_inner(
700        &self,
701        publisher: Arc<PublisherCounters>,
702        plugin_id: &Arc<str>,
703        topic: &PluginEventTopic,
704        publication: PluginEventPublication,
705    ) -> Result<PluginEventPublishReport, PluginEventPublishError> {
706        if publication.schema_version == 0 {
707            return Err(PluginEventPublishError::InvalidSchemaVersion);
708        }
709        if self.inner.closed.load(Ordering::Acquire) {
710            return Err(PluginEventPublishError::Closed);
711        }
712
713        let route_key = RouteKey {
714            plugin_id: plugin_id.clone(),
715            topic: topic.clone(),
716        };
717        let Some((clock, endpoints)) = self
718            .inner
719            .state
720            .read()
721            .unwrap_or_else(|poisoned| poisoned.into_inner())
722            .routes
723            .get(&route_key)
724            .map(|route| (route.clock.clone(), route.endpoints.clone()))
725        else {
726            return Ok(PluginEventPublishReport::default());
727        };
728
729        let mut sequence = clock
730            .sequence
731            .lock()
732            .unwrap_or_else(|poisoned| poisoned.into_inner());
733        let next_sequence = sequence
734            .checked_add(1)
735            .ok_or(PluginEventPublishError::SequenceExhausted)?;
736        *sequence = next_sequence;
737        let envelope = Arc::new(
738            PluginEventEnvelope::builder()
739                .plugin_id(plugin_id.clone())
740                .topic(topic.clone())
741                .schema_version(publication.schema_version)
742                .payload_encoding(publication.payload_encoding)
743                .payload(publication.payload)
744                .connection_generation(publication.connection_generation)
745                .sequence(next_sequence)
746                .build(),
747        );
748        let event =
749            QueuedPluginEvent::new(envelope, publisher, Arc::clone(&self.inner.queue_memory));
750
751        let mut report = PluginEventPublishReport {
752            matched: u64::try_from(endpoints.len()).unwrap_or(u64::MAX),
753            ..PluginEventPublishReport::default()
754        };
755        for endpoint in endpoints.iter() {
756            match endpoint.enqueue(event.clone()) {
757                EnqueueOutcome::Enqueued => report.enqueued += 1,
758                EnqueueOutcome::Dropped => report.dropped += 1,
759                EnqueueOutcome::EnqueuedAfterDrop => {
760                    report.enqueued += 1;
761                    report.dropped += 1;
762                }
763                EnqueueOutcome::Closed => report.closed += 1,
764            }
765        }
766        Ok(report)
767    }
768
769    pub(super) fn close(&self) {
770        self.inner.close();
771    }
772}
773
774/// One bounded endpoint. Dropping it unregisters every selected route atomically.
775#[must_use = "dropping the subscription unregisters its plugin event routes"]
776pub struct PluginEventSubscription {
777    router: PluginEventRouter,
778    endpoint: Arc<EventEndpoint>,
779    receiver: Receiver<Arc<QueuedPluginEvent>>,
780    selectors: Vec<PluginEventSelector>,
781}
782
783impl PluginEventSubscription {
784    pub fn id(&self) -> u64 {
785        self.endpoint.id
786    }
787
788    pub fn selectors(&self) -> &[PluginEventSelector] {
789        &self.selectors
790    }
791
792    pub fn stats(&self) -> PluginEventEndpointStats {
793        self.endpoint.stats()
794    }
795
796    pub async fn recv(&self) -> Result<Arc<PluginEventEnvelope>, PluginEventReceiveError> {
797        let event = self
798            .receiver
799            .recv()
800            .await
801            .map_err(|_| PluginEventReceiveError)?;
802        self.endpoint.delivered.fetch_add(1, Ordering::Relaxed);
803        event.publisher.delivered.fetch_add(1, Ordering::Relaxed);
804        Ok(event.envelope.clone())
805    }
806
807    pub fn try_recv(&self) -> Result<Arc<PluginEventEnvelope>, PluginEventTryReceiveError> {
808        let event = self.receiver.try_recv().map_err(|error| match error {
809            TryRecvError::Empty => PluginEventTryReceiveError::Empty,
810            TryRecvError::Closed => PluginEventTryReceiveError::Closed,
811        })?;
812        self.endpoint.delivered.fetch_add(1, Ordering::Relaxed);
813        event.publisher.delivered.fetch_add(1, Ordering::Relaxed);
814        Ok(event.envelope.clone())
815    }
816}
817
818impl Drop for PluginEventSubscription {
819    fn drop(&mut self) {
820        self.router
821            .inner
822            .unsubscribe(self.endpoint.id, &self.selectors);
823    }
824}
825
826/// Context-bound custom event capability. A plugin can publish only under its own ID.
827///
828/// Consumers subscribe through [`PluginEventRouter`], keeping publication authority separate from
829/// native or future foreign endpoints.
830#[derive(Clone)]
831pub struct PluginEvents {
832    plugin_id: Arc<str>,
833    router: PluginEventRouter,
834    stats: Arc<PublisherCounters>,
835    resources: Arc<PluginResources>,
836    connection_generation: Arc<AtomicU64>,
837}
838
839impl PluginEvents {
840    pub fn selector(&self, topic: &PluginEventTopic) -> PluginEventSelector {
841        PluginEventSelector {
842            route: RouteKey {
843                plugin_id: self.plugin_id.clone(),
844                topic: topic.clone(),
845            },
846        }
847    }
848
849    pub fn has_subscribers(&self, topic: &PluginEventTopic) -> bool {
850        self.router.has_subscribers(&self.selector(topic))
851    }
852
853    pub fn stats(&self) -> PluginEventPublisherStats {
854        self.stats.snapshot()
855    }
856
857    pub fn publish(
858        &self,
859        topic: &PluginEventTopic,
860        schema_version: u32,
861        payload_encoding: PluginEventPayloadEncoding,
862        payload: impl Into<Bytes>,
863    ) -> Result<PluginEventPublishReport, PluginEventPublishError> {
864        if let Err(error) = self.resources.ensure_active() {
865            self.stats.publish_failures.fetch_add(1, Ordering::Relaxed);
866            return Err(error.into());
867        }
868        self.router.publish(
869            Arc::clone(&self.stats),
870            &self.plugin_id,
871            topic,
872            PluginEventPublication {
873                schema_version,
874                payload_encoding,
875                payload: payload.into(),
876                connection_generation: self.connection_generation.load(Ordering::Acquire),
877            },
878        )
879    }
880}
881
882pub(super) fn publisher(
883    plugin_id: &str,
884    router: PluginEventRouter,
885    resources: Arc<PluginResources>,
886    connection_generation: Arc<AtomicU64>,
887) -> Option<PluginEvents> {
888    let stats = router.inner.publishers.get(plugin_id)?.clone();
889    Some(PluginEvents {
890        plugin_id: Arc::from(plugin_id),
891        router,
892        stats,
893        resources,
894        connection_generation,
895    })
896}
897
898fn valid_topic(topic: &str) -> bool {
899    valid_plugin_id(topic)
900}
901
902#[cfg(test)]
903mod tests {
904    use std::thread;
905
906    use super::*;
907
908    fn topic(value: &str) -> PluginEventTopic {
909        PluginEventTopic::new(value).expect("valid topic")
910    }
911
912    fn selector(plugin_id: &str, topic: &PluginEventTopic) -> PluginEventSelector {
913        PluginEventSelector::new(plugin_id, topic.clone()).expect("valid selector")
914    }
915
916    fn publish(
917        router: &PluginEventRouter,
918        plugin_id: &str,
919        topic: &PluginEventTopic,
920        value: u32,
921    ) -> PluginEventPublishReport {
922        let publisher = router
923            .inner
924            .publishers
925            .get(plugin_id)
926            .cloned()
927            .expect("registered publisher");
928        router
929            .publish(
930                publisher,
931                &Arc::from(plugin_id),
932                topic,
933                PluginEventPublication {
934                    schema_version: 1,
935                    payload_encoding: PluginEventPayloadEncoding::Binary,
936                    payload: Bytes::copy_from_slice(&value.to_be_bytes()),
937                    connection_generation: 7,
938                },
939            )
940            .expect("event publication")
941    }
942
943    #[test]
944    fn router_stats_count_shared_queue_payload_once_and_keep_cumulative_totals() {
945        let router = PluginEventRouter::new(["metrics".to_string()]);
946        let tick = topic("tick");
947        let first = router
948            .subscribe(
949                [selector("metrics", &tick)],
950                PluginEventEndpointConfig::new(2, PluginEventOverflow::DropNewest),
951            )
952            .expect("first endpoint");
953        let second = router
954            .subscribe(
955                [selector("metrics", &tick)],
956                PluginEventEndpointConfig::new(3, PluginEventOverflow::DropNewest),
957            )
958            .expect("second endpoint");
959
960        assert_eq!(publish(&router, "metrics", &tick, 1).enqueued, 2);
961        assert_eq!(
962            router.stats(),
963            PluginEventRouterStats {
964                registered_publishers: 1,
965                active_routes: 1,
966                active_endpoints: 2,
967                endpoint_capacity: 5,
968                queued_events: 1,
969                queued_payload_bytes: 4,
970                published: 1,
971                publish_failures: 0,
972                matched: 2,
973                enqueued: 2,
974                delivered: 0,
975                dropped: 0,
976                closed: 0,
977            }
978        );
979
980        first.try_recv().expect("first delivery");
981        assert_eq!(router.stats().queued_events, 1);
982        second.try_recv().expect("second delivery");
983        assert_eq!(router.stats().queued_events, 0);
984        assert_eq!(router.stats().queued_payload_bytes, 0);
985        assert_eq!(router.stats().delivered, 2);
986
987        drop(first);
988        drop(second);
989        let stats = router.stats();
990        assert_eq!(stats.active_routes, 0);
991        assert_eq!(stats.active_endpoints, 0);
992        assert_eq!(stats.published, 1);
993        assert_eq!(stats.delivered, 2);
994        assert_eq!(
995            router.publisher_stats("metrics"),
996            Some(PluginEventPublisherStats {
997                published: 1,
998                matched: 2,
999                enqueued: 2,
1000                delivered: 2,
1001                ..PluginEventPublisherStats::default()
1002            })
1003        );
1004    }
1005
1006    #[test]
1007    fn routes_only_exact_plugin_and_topic_matches() {
1008        let router = PluginEventRouter::new(["metrics".to_string(), "audit".to_string()]);
1009        let tick = topic("tick");
1010        let other = topic("other");
1011        let subscription = router
1012            .subscribe(
1013                [selector("metrics", &tick)],
1014                PluginEventEndpointConfig::new(4, PluginEventOverflow::DropNewest),
1015            )
1016            .expect("subscription");
1017
1018        assert_eq!(publish(&router, "metrics", &other, 1).matched, 0);
1019        assert_eq!(publish(&router, "audit", &tick, 2).matched, 0);
1020        assert_eq!(publish(&router, "metrics", &tick, 3).enqueued, 1);
1021        let event = subscription.try_recv().expect("routed event");
1022        assert_eq!(&*event.plugin_id, "metrics");
1023        assert_eq!(event.topic, tick);
1024        assert_eq!(event.connection_generation, 7);
1025        assert_eq!(event.sequence, 1);
1026        assert_eq!(event.payload, Bytes::copy_from_slice(&3u32.to_be_bytes()));
1027    }
1028
1029    #[test]
1030    fn drop_newest_preserves_the_queued_prefix_and_counts_loss() {
1031        let router = PluginEventRouter::new(["metrics".to_string()]);
1032        let tick = topic("tick");
1033        let subscription = router
1034            .subscribe(
1035                [selector("metrics", &tick)],
1036                PluginEventEndpointConfig::new(2, PluginEventOverflow::DropNewest),
1037            )
1038            .expect("subscription");
1039
1040        for value in 1..=4 {
1041            publish(&router, "metrics", &tick, value);
1042        }
1043
1044        assert_eq!(subscription.try_recv().expect("first").sequence, 1);
1045        assert_eq!(subscription.try_recv().expect("second").sequence, 2);
1046        assert!(matches!(
1047            subscription.try_recv(),
1048            Err(PluginEventTryReceiveError::Empty)
1049        ));
1050        assert_eq!(
1051            subscription.stats(),
1052            PluginEventEndpointStats {
1053                enqueued: 2,
1054                delivered: 2,
1055                dropped: 2,
1056                queue_depth: 0,
1057                capacity: 2,
1058            }
1059        );
1060        assert_eq!(
1061            router.publisher_stats("metrics"),
1062            Some(PluginEventPublisherStats {
1063                published: 4,
1064                matched: 4,
1065                enqueued: 2,
1066                delivered: 2,
1067                dropped: 2,
1068                ..PluginEventPublisherStats::default()
1069            })
1070        );
1071    }
1072
1073    #[test]
1074    fn drop_oldest_preserves_the_latest_events_and_counts_evictions() {
1075        let router = PluginEventRouter::new(["metrics".to_string()]);
1076        let tick = topic("tick");
1077        let subscription = router
1078            .subscribe(
1079                [selector("metrics", &tick)],
1080                PluginEventEndpointConfig::new(2, PluginEventOverflow::DropOldest),
1081            )
1082            .expect("subscription");
1083
1084        for value in 1..=4 {
1085            publish(&router, "metrics", &tick, value);
1086        }
1087
1088        assert_eq!(subscription.try_recv().expect("third").sequence, 3);
1089        assert_eq!(subscription.try_recv().expect("fourth").sequence, 4);
1090        assert_eq!(subscription.stats().enqueued, 4);
1091        assert_eq!(subscription.stats().dropped, 2);
1092        assert_eq!(router.stats().dropped, 2);
1093        assert_eq!(router.stats().delivered, 2);
1094    }
1095
1096    #[test]
1097    fn drop_oldest_charges_the_evicted_publisher_across_namespaces() {
1098        let router = PluginEventRouter::new(["alpha".to_string(), "beta".to_string()]);
1099        let tick = topic("tick");
1100        let subscription = router
1101            .subscribe(
1102                [selector("alpha", &tick), selector("beta", &tick)],
1103                PluginEventEndpointConfig::new(1, PluginEventOverflow::DropOldest),
1104            )
1105            .expect("subscription");
1106
1107        assert_eq!(publish(&router, "alpha", &tick, 1).dropped, 0);
1108        assert_eq!(publish(&router, "beta", &tick, 2).dropped, 1);
1109
1110        let event = subscription.try_recv().expect("newest event");
1111        assert_eq!(&*event.plugin_id, "beta");
1112        assert_eq!(subscription.stats().dropped, 1);
1113        assert_eq!(
1114            router
1115                .publisher_stats("alpha")
1116                .expect("alpha stats")
1117                .dropped,
1118            1
1119        );
1120        let beta = router.publisher_stats("beta").expect("beta stats");
1121        assert_eq!(beta.dropped, 0);
1122        assert_eq!(beta.delivered, 1);
1123        assert_eq!(router.stats().dropped, 1);
1124    }
1125
1126    #[test]
1127    fn backpressure_is_isolated_per_endpoint() {
1128        let router = PluginEventRouter::new(["metrics".to_string()]);
1129        let tick = topic("tick");
1130        let slow = router
1131            .subscribe(
1132                [selector("metrics", &tick)],
1133                PluginEventEndpointConfig::new(1, PluginEventOverflow::DropNewest),
1134            )
1135            .expect("slow endpoint");
1136        let fast = router
1137            .subscribe(
1138                [selector("metrics", &tick)],
1139                PluginEventEndpointConfig::new(4, PluginEventOverflow::DropNewest),
1140            )
1141            .expect("fast endpoint");
1142
1143        publish(&router, "metrics", &tick, 1);
1144        publish(&router, "metrics", &tick, 2);
1145
1146        assert_eq!(slow.stats().dropped, 1);
1147        assert_eq!(fast.stats().dropped, 0);
1148        assert_eq!(fast.try_recv().expect("fast first").sequence, 1);
1149        assert_eq!(fast.try_recv().expect("fast second").sequence, 2);
1150    }
1151
1152    #[tokio::test]
1153    async fn drop_unregisters_and_router_close_wakes_receivers() {
1154        let router = PluginEventRouter::new(["metrics".to_string()]);
1155        let tick = topic("tick");
1156        let selector = selector("metrics", &tick);
1157        let subscription = router
1158            .subscribe(
1159                [selector.clone()],
1160                PluginEventEndpointConfig::new(1, PluginEventOverflow::DropNewest),
1161            )
1162            .expect("subscription");
1163        assert!(router.has_subscribers(&selector));
1164        drop(subscription);
1165        assert!(!router.has_subscribers(&selector));
1166        assert_eq!(publish(&router, "metrics", &tick, 1).matched, 0);
1167
1168        let subscription = router
1169            .subscribe(
1170                [selector],
1171                PluginEventEndpointConfig::new(1, PluginEventOverflow::DropNewest),
1172            )
1173            .expect("second subscription");
1174        assert_eq!(publish(&router, "metrics", &tick, 2).enqueued, 1);
1175        router.close();
1176        assert_eq!(subscription.recv().await.expect("queued event").sequence, 1);
1177        assert!(matches!(
1178            subscription.recv().await,
1179            Err(PluginEventReceiveError)
1180        ));
1181    }
1182
1183    #[test]
1184    fn rejects_invalid_or_unknown_endpoint_configuration() {
1185        assert!(PluginEventTopic::new("Invalid").is_err());
1186        let tick = topic("tick");
1187        assert!(PluginEventSelector::new("Invalid", tick.clone()).is_err());
1188        let router = PluginEventRouter::new(["metrics".to_string()]);
1189        assert!(matches!(
1190            router.subscribe(
1191                [selector("unknown", &tick)],
1192                PluginEventEndpointConfig::new(1, PluginEventOverflow::DropNewest),
1193            ),
1194            Err(PluginEventSubscribeError::UnknownPublisher { .. })
1195        ));
1196        assert!(matches!(
1197            router.subscribe(
1198                [selector("metrics", &tick)],
1199                PluginEventEndpointConfig::new(0, PluginEventOverflow::DropNewest),
1200            ),
1201            Err(PluginEventSubscribeError::InvalidCapacity { .. })
1202        ));
1203        assert!(matches!(
1204            router.subscribe(
1205                [],
1206                PluginEventEndpointConfig::new(1, PluginEventOverflow::DropNewest),
1207            ),
1208            Err(PluginEventSubscribeError::EmptySelectors)
1209        ));
1210        let too_many = (0..=MAX_ENDPOINT_SELECTORS)
1211            .map(|index| selector("metrics", &topic(&format!("topic-{index}"))))
1212            .collect::<Vec<_>>();
1213        assert!(matches!(
1214            router.subscribe(
1215                too_many,
1216                PluginEventEndpointConfig::new(1, PluginEventOverflow::DropNewest),
1217            ),
1218            Err(PluginEventSubscribeError::TooManySelectors { .. })
1219        ));
1220    }
1221
1222    #[test]
1223    fn concurrent_publish_keeps_route_sequences_in_queue_order() {
1224        const THREADS: usize = 8;
1225        const EVENTS_PER_THREAD: usize = 100;
1226        let router = PluginEventRouter::new(["metrics".to_string()]);
1227        let tick = topic("tick");
1228        let subscription = router
1229            .subscribe(
1230                [selector("metrics", &tick)],
1231                PluginEventEndpointConfig::new(
1232                    THREADS * EVENTS_PER_THREAD,
1233                    PluginEventOverflow::DropNewest,
1234                ),
1235            )
1236            .expect("subscription");
1237
1238        let threads = (0..THREADS)
1239            .map(|_| {
1240                let router = router.clone();
1241                let tick = tick.clone();
1242                thread::spawn(move || {
1243                    for value in 0..EVENTS_PER_THREAD {
1244                        publish(&router, "metrics", &tick, value as u32);
1245                    }
1246                })
1247            })
1248            .collect::<Vec<_>>();
1249        for thread in threads {
1250            thread.join().expect("publisher thread");
1251        }
1252
1253        for sequence in 1..=(THREADS * EVENTS_PER_THREAD) as u64 {
1254            assert_eq!(
1255                subscription.try_recv().expect("ordered event").sequence,
1256                sequence
1257            );
1258        }
1259        assert_eq!(subscription.stats().dropped, 0);
1260    }
1261}
whatsapp_rust/plugins/events.rs

whatsapp_rust/plugins/
events.rs
