basil_core/core/
event.rs

1//! Internal broker event source shared by Admin Watch and future SPIFFE streams.
2
3use std::time::SystemTime;
4
5use tokio::sync::broadcast;
6
7/// Broker event fanout with bounded lag semantics.
8#[derive(Debug, Clone)]
9pub struct EventSource {
10    sender: broadcast::Sender<BrokerEvent>,
11}
12
13/// A public broker event.
14#[derive(Debug, Clone)]
15pub struct BrokerEvent {
16    /// Event creation time.
17    pub at: SystemTime,
18    /// Event payload.
19    pub kind: BrokerEventKind,
20}
21
22/// Event payload variants.
23#[derive(Debug, Clone, PartialEq, Eq)]
24pub enum BrokerEventKind {
25    /// A catalog key rotated to a new version.
26    KeyRotated {
27        /// Dotted catalog key id.
28        key_id: String,
29        /// New latest version.
30        new_version: u32,
31    },
32    /// A trust-domain bundle changed.
33    BundleChanged {
34        /// Trust domain whose bundle changed.
35        trust_domain: String,
36    },
37    /// A credential or bundle item was revoked.
38    Revoked {
39        /// Trust domain containing the revoked id.
40        trust_domain: String,
41        /// Public revocation id, such as an X.509 serial or JWT jti.
42        id: String,
43    },
44}
45
46impl Default for EventSource {
47    fn default() -> Self {
48        Self::new()
49    }
50}
51
52impl EventSource {
53    /// Build an event source with a bounded broadcast buffer.
54    #[must_use]
55    pub fn new() -> Self {
56        let (sender, _) = broadcast::channel(1024);
57        Self { sender }
58    }
59
60    /// Subscribe to future events.
61    #[must_use]
62    pub fn subscribe(&self) -> broadcast::Receiver<BrokerEvent> {
63        self.sender.subscribe()
64    }
65
66    /// Publish a key rotation event.
67    pub fn key_rotated(&self, key_id: impl Into<String>, new_version: u32) {
68        self.publish(BrokerEventKind::KeyRotated {
69            key_id: key_id.into(),
70            new_version,
71        });
72    }
73
74    /// Publish a bundle change event.
75    pub fn bundle_changed(&self, trust_domain: impl Into<String>) {
76        self.publish(BrokerEventKind::BundleChanged {
77            trust_domain: trust_domain.into(),
78        });
79    }
80
81    /// Publish a revocation event.
82    pub fn revoked(&self, trust_domain: impl Into<String>, id: impl Into<String>) {
83        self.publish(BrokerEventKind::Revoked {
84            trust_domain: trust_domain.into(),
85            id: id.into(),
86        });
87    }
88
89    fn publish(&self, kind: BrokerEventKind) {
90        let _receivers = self.sender.send(BrokerEvent {
91            at: SystemTime::now(),
92            kind,
93        });
94    }
95}
96
97#[cfg(test)]
98mod tests {
99    use super::*;
100
101    #[tokio::test]
102    async fn subscribers_receive_events() {
103        let source = EventSource::new();
104        let mut rx = source.subscribe();
105        source.key_rotated("app.key", 2);
106        let event = rx.recv().await.expect("event received");
107        assert_eq!(
108            event.kind,
109            BrokerEventKind::KeyRotated {
110                key_id: "app.key".to_string(),
111                new_version: 2,
112            }
113        );
114    }
115}
basil_core/core/event.rs

basil_core/core/
event.rs
