sockudo_http/
events.rs

1use crate::{Channel, Result, Sockudo, SockudoError};
2#[cfg(feature = "encryption")]
3use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64};
4use serde::{Deserialize, Serialize};
5use sonic_rs::{Value, json};
6use std::collections::HashMap;
7use std::fmt;
8
9/// V2 extras for publish events.
10#[derive(Debug, Clone, Default, Serialize, Deserialize)]
11pub struct MessageExtras {
12    #[serde(skip_serializing_if = "Option::is_none")]
13    pub headers: Option<HashMap<String, Value>>,
14    #[serde(skip_serializing_if = "Option::is_none")]
15    pub ephemeral: Option<bool>,
16    #[serde(skip_serializing_if = "Option::is_none")]
17    pub idempotency_key: Option<String>,
18    #[serde(skip_serializing_if = "Option::is_none")]
19    pub echo: Option<bool>,
20}
21
22/// Event data that can be either a string or JSON
23#[derive(Debug, Clone, PartialEq)]
24pub enum EventData {
25    String(String),
26    Json(Value),
27}
28
29impl EventData {
30    /// Creates event data from a string
31    pub fn from_string(s: impl Into<String>) -> Self {
32        EventData::String(s.into())
33    }
34
35    /// Creates event data from a JSON value
36    pub fn from_json(value: Value) -> Self {
37        EventData::Json(value)
38    }
39
40    /// Gets the event data as a JSON value
41    pub fn as_json(&self) -> Result<Value> {
42        match self {
43            EventData::String(s) => sonic_rs::from_str(s).map_err(SockudoError::Json),
44            EventData::Json(v) => Ok(v.clone()),
45        }
46    }
47}
48
49impl fmt::Display for EventData {
50    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
51        match self {
52            EventData::String(s) => write!(f, "{}", s),
53            EventData::Json(v) => write!(f, "{}", sonic_rs::to_string(v).unwrap_or_default()),
54        }
55    }
56}
57
58impl From<String> for EventData {
59    fn from(s: String) -> Self {
60        EventData::String(s)
61    }
62}
63
64impl From<&str> for EventData {
65    fn from(s: &str) -> Self {
66        EventData::String(s.to_string())
67    }
68}
69
70impl From<Value> for EventData {
71    fn from(v: Value) -> Self {
72        EventData::Json(v)
73    }
74}
75
76/// Generates a random idempotency key as a UUID v4 string.
77pub fn generate_idempotency_key() -> String {
78    let mut bytes = [0u8; 16];
79    rand::fill(&mut bytes);
80    bytes[6] = (bytes[6] & 0x0f) | 0x40; // version 4
81    bytes[8] = (bytes[8] & 0x3f) | 0x80; // variant 10
82    format!(
83        "{:02x}{:02x}{:02x}{:02x}-{:02x}{:02x}-{:02x}{:02x}-{:02x}{:02x}-{:02x}{:02x}{:02x}{:02x}{:02x}{:02x}",
84        bytes[0],
85        bytes[1],
86        bytes[2],
87        bytes[3],
88        bytes[4],
89        bytes[5],
90        bytes[6],
91        bytes[7],
92        bytes[8],
93        bytes[9],
94        bytes[10],
95        bytes[11],
96        bytes[12],
97        bytes[13],
98        bytes[14],
99        bytes[15]
100    )
101}
102
103/// Event data for triggering
104#[derive(Debug, Serialize)]
105pub struct Event {
106    pub name: String,
107    pub data: String,
108    pub channels: Vec<String>,
109    #[serde(skip_serializing_if = "Option::is_none")]
110    pub socket_id: Option<String>,
111    #[serde(skip_serializing_if = "Option::is_none")]
112    pub info: Option<String>,
113    #[serde(skip_serializing_if = "Option::is_none")]
114    pub tags: Option<HashMap<String, String>>,
115    #[serde(skip_serializing_if = "Option::is_none")]
116    pub idempotency_key: Option<String>,
117    #[serde(skip_serializing_if = "Option::is_none")]
118    pub extras: Option<MessageExtras>,
119}
120
121/// Batch event data
122#[derive(Debug, Serialize, Deserialize)]
123pub struct BatchEvent {
124    pub name: String,
125    pub channel: String,
126    pub data: String,
127    #[serde(skip_serializing_if = "Option::is_none")]
128    pub socket_id: Option<String>,
129    #[serde(skip_serializing_if = "Option::is_none")]
130    pub info: Option<String>,
131    #[serde(skip_serializing_if = "Option::is_none")]
132    pub tags: Option<HashMap<String, String>>,
133    #[serde(skip_serializing_if = "Option::is_none")]
134    pub idempotency_key: Option<String>,
135    #[serde(skip_serializing_if = "Option::is_none")]
136    pub extras: Option<MessageExtras>,
137}
138
139impl BatchEvent {
140    /// Creates a new batch event with EventData
141    pub fn new(
142        name: impl Into<String>,
143        channel: impl Into<String>,
144        data: impl Into<EventData>,
145    ) -> Self {
146        Self {
147            name: name.into(),
148            channel: channel.into(),
149            data: data.into().to_string(),
150            socket_id: None,
151            info: None,
152            tags: None,
153            idempotency_key: None,
154            extras: None,
155        }
156    }
157
158    /// Sets the socket ID to exclude
159    pub fn with_socket_id(mut self, socket_id: impl Into<String>) -> Self {
160        self.socket_id = Some(socket_id.into());
161        self
162    }
163
164    /// Sets the info parameter
165    pub fn with_info(mut self, info: impl Into<String>) -> Self {
166        self.info = Some(info.into());
167        self
168    }
169
170    /// Sets the tags for tag filtering
171    pub fn with_tags(mut self, tags: HashMap<String, String>) -> Self {
172        self.tags = Some(tags);
173        self
174    }
175
176    /// Sets the idempotency key for deduplication
177    pub fn with_idempotency_key(mut self, key: impl Into<String>) -> Self {
178        self.idempotency_key = Some(key.into());
179        self
180    }
181
182    /// Sets a randomly generated idempotency key (UUID v4)
183    pub fn with_auto_idempotency_key(mut self) -> Self {
184        self.idempotency_key = Some(generate_idempotency_key());
185        self
186    }
187
188    /// Sets the V2 extras
189    pub fn with_extras(mut self, extras: MessageExtras) -> Self {
190        self.extras = Some(extras);
191        self
192    }
193
194    /// Sets the ephemeral flag in extras
195    pub fn with_ephemeral(mut self, ephemeral: bool) -> Self {
196        self.extras
197            .get_or_insert_with(MessageExtras::default)
198            .ephemeral = Some(ephemeral);
199        self
200    }
201
202    /// Sets the echo flag in extras
203    pub fn with_echo(mut self, echo: bool) -> Self {
204        self.extras.get_or_insert_with(MessageExtras::default).echo = Some(echo);
205        self
206    }
207}
208
209/// Parameters for triggering events
210#[derive(Debug, Clone, Default)]
211pub struct TriggerParams {
212    pub socket_id: Option<String>,
213    pub info: Option<String>,
214    pub tags: Option<HashMap<String, String>>,
215    pub idempotency_key: Option<String>,
216    pub extras: Option<MessageExtras>,
217}
218
219impl TriggerParams {
220    /// Creates a new TriggerParams builder
221    pub fn builder() -> TriggerParamsBuilder {
222        TriggerParamsBuilder::default()
223    }
224}
225
226/// Builder for TriggerParams
227#[derive(Debug, Default)]
228pub struct TriggerParamsBuilder {
229    socket_id: Option<String>,
230    info: Option<String>,
231    tags: Option<HashMap<String, String>>,
232    idempotency_key: Option<String>,
233    extras: Option<MessageExtras>,
234}
235
236impl TriggerParamsBuilder {
237    /// Sets the socket ID to exclude
238    pub fn socket_id(mut self, socket_id: impl Into<String>) -> Self {
239        self.socket_id = Some(socket_id.into());
240        self
241    }
242
243    /// Sets the info parameter
244    pub fn info(mut self, info: impl Into<String>) -> Self {
245        self.info = Some(info.into());
246        self
247    }
248
249    /// Sets the tags for tag filtering
250    pub fn tags(mut self, tags: HashMap<String, String>) -> Self {
251        self.tags = Some(tags);
252        self
253    }
254
255    /// Sets the idempotency key for deduplication
256    pub fn idempotency_key(mut self, key: impl Into<String>) -> Self {
257        self.idempotency_key = Some(key.into());
258        self
259    }
260
261    /// Sets a randomly generated idempotency key (UUID v4)
262    pub fn auto_idempotency_key(mut self) -> Self {
263        self.idempotency_key = Some(generate_idempotency_key());
264        self
265    }
266
267    /// Sets the V2 extras
268    pub fn extras(mut self, extras: MessageExtras) -> Self {
269        self.extras = Some(extras);
270        self
271    }
272
273    /// Sets the ephemeral flag in extras
274    pub fn with_ephemeral(mut self, ephemeral: bool) -> Self {
275        self.extras
276            .get_or_insert_with(MessageExtras::default)
277            .ephemeral = Some(ephemeral);
278        self
279    }
280
281    /// Sets the echo flag in extras
282    pub fn with_echo(mut self, echo: bool) -> Self {
283        self.extras.get_or_insert_with(MessageExtras::default).echo = Some(echo);
284        self
285    }
286
287    /// Builds the TriggerParams
288    pub fn build(self) -> TriggerParams {
289        TriggerParams {
290            socket_id: self.socket_id,
291            info: self.info,
292            tags: self.tags,
293            idempotency_key: self.idempotency_key,
294            extras: self.extras,
295        }
296    }
297}
298
299/// Encrypts data for encrypted channels
300#[cfg(feature = "encryption")]
301fn encrypt(sockudo: &Sockudo, channel: &str, data: &EventData) -> Result<String> {
302    encrypt_secretbox(sockudo, channel, data)
303}
304
305/// Encrypts data using NaCl-compatible XSalsa20-Poly1305.
306#[cfg(feature = "encryption")]
307fn encrypt_secretbox(sockudo: &Sockudo, channel: &str, data: &EventData) -> Result<String> {
308    use crypto_secretbox::{
309        XSalsa20Poly1305,
310        aead::{Aead, AeadCore, KeyInit},
311    };
312
313    // Ensure master key is present
314    let _master_key =
315        sockudo
316            .config()
317            .encryption_master_key()
318            .ok_or_else(|| SockudoError::Encryption {
319                message: "Set encryptionMasterKey before triggering events on encrypted channels"
320                    .to_string(),
321            })?;
322
323    // Get channel shared secret
324    let shared_secret_bytes = sockudo.channel_shared_secret(channel)?;
325
326    // Create cipher
327    let cipher = XSalsa20Poly1305::new_from_slice(&shared_secret_bytes).map_err(|_| {
328        SockudoError::Encryption {
329            message: "Failed to create cipher from shared secret".to_string(),
330        }
331    })?;
332
333    // Generate random nonce
334    let nonce = XSalsa20Poly1305::generate_nonce().map_err(|_| SockudoError::Encryption {
335        message: "failed to generate encryption nonce".to_string(),
336    })?;
337
338    // Encrypt the data
339    let data_string = data.to_string();
340    let ciphertext = cipher
341        .encrypt(&nonce, data_string.as_bytes())
342        .map_err(|_| SockudoError::Encryption {
343            message: "Encryption failed".to_string(),
344        })?;
345
346    // Return encrypted payload as JSON string
347    let encrypted_payload = json!({
348        "nonce": BASE64.encode(nonce),
349        "ciphertext": BASE64.encode(&ciphertext),
350    });
351
352    Ok(sonic_rs::to_string(&encrypted_payload)?)
353}
354
355/// Stub function when encryption is disabled
356#[cfg(not(feature = "encryption"))]
357#[allow(dead_code)]
358fn encrypt(_sockudo: &Sockudo, _channel: &str, _data: &EventData) -> Result<String> {
359    Err(SockudoError::Encryption {
360        message: "Encryption support is not enabled. Enable the 'encryption' feature to use encrypted channels.".to_string(),
361    })
362}
363
364/// Triggers an event on channels
365pub async fn trigger<D: Into<EventData>>(
366    sockudo: &Sockudo,
367    channels: &[Channel],
368    event_name: impl AsRef<str>,
369    data: D,
370    params: Option<&TriggerParams>,
371) -> Result<reqwest::Response> {
372    let data = data.into();
373    let event_name = event_name.as_ref();
374
375    // Validate event name
376    if event_name.len() > 200 {
377        return Err(SockudoError::Validation {
378            message: format!("Event name too long: '{}' (max 200 characters)", event_name),
379        });
380    }
381
382    // Convert channels to strings
383    let channel_strings: Vec<String> = channels.iter().map(|c| c.full_name()).collect();
384
385    // Extract idempotency key for the header
386    let idempotency_key = params.and_then(|p| p.idempotency_key.clone());
387
388    let mut extra_headers = HashMap::new();
389    if let Some(ref key) = idempotency_key {
390        extra_headers.insert("X-Idempotency-Key".to_string(), key.clone());
391    }
392
393    if channels.len() == 1 && channels[0].is_encrypted() {
394        #[cfg(feature = "encryption")]
395        {
396            let encrypted_data = encrypt(sockudo, &channel_strings[0], &data)?;
397
398            let mut event = Event {
399                name: event_name.to_string(),
400                data: encrypted_data,
401                channels: channel_strings,
402                socket_id: None,
403                info: None,
404                tags: None,
405                idempotency_key: idempotency_key.clone(),
406                extras: None,
407            };
408
409            if let Some(params) = params {
410                event.socket_id = params.socket_id.clone();
411                event.info = params.info.clone();
412                event.tags = params.tags.clone();
413                event.extras = params.extras.clone();
414            }
415
416            let event_json = sonic_rs::to_value(&event)?;
417            sockudo
418                .post_with_headers("/events", &event_json, &extra_headers)
419                .await
420        }
421
422        #[cfg(not(feature = "encryption"))]
423        {
424            Err(SockudoError::Encryption {
425                message: "Encryption support is not enabled. Enable the 'encryption' feature to use encrypted channels.".to_string(),
426            })
427        }
428    } else {
429        // Check for encrypted channels in multi-channel trigger
430        for channel in channels {
431            if channel.is_encrypted() {
432                return Err(SockudoError::Validation {
433                    message:
434                        "You cannot trigger to multiple channels when using encrypted channels"
435                            .to_string(),
436                });
437            }
438        }
439
440        let mut event = Event {
441            name: event_name.to_string(),
442            data: data.to_string(),
443            channels: channel_strings,
444            socket_id: None,
445            info: None,
446            tags: None,
447            idempotency_key: idempotency_key.clone(),
448            extras: None,
449        };
450
451        if let Some(params) = params {
452            event.socket_id = params.socket_id.clone();
453            event.info = params.info.clone();
454            event.tags = params.tags.clone();
455            event.extras = params.extras.clone();
456        }
457
458        let event_json = sonic_rs::to_value(&event)?;
459        sockudo
460            .post_with_headers("/events", &event_json, &extra_headers)
461            .await
462    }
463}
464
465/// Triggers an event on channel names (backward compatibility)
466pub async fn trigger_on_channels<D: Into<EventData>>(
467    sockudo: &Sockudo,
468    channels: &[String],
469    event_name: impl AsRef<str>,
470    data: D,
471    params: Option<&TriggerParams>,
472) -> Result<reqwest::Response> {
473    let channels: Result<Vec<Channel>> = channels.iter().map(Channel::from_string).collect();
474    let channels = channels?;
475    trigger(sockudo, &channels, event_name, data, params).await
476}
477
478/// Triggers a batch of events
479pub async fn trigger_batch(
480    sockudo: &Sockudo,
481    mut batch: Vec<BatchEvent>,
482    idempotency_key: Option<&str>,
483) -> Result<reqwest::Response> {
484    // Validate batch size
485    if batch.is_empty() {
486        return Err(SockudoError::Validation {
487            message: "Batch cannot be empty".to_string(),
488        });
489    }
490
491    if batch.len() > 10 {
492        return Err(SockudoError::Validation {
493            message: format!("Batch too large: {} events (max 10)", batch.len()),
494        });
495    }
496
497    // Encrypt data for encrypted channels
498    for event in &mut batch {
499        let channel = Channel::from_string(&event.channel)?;
500        if channel.is_encrypted() {
501            #[cfg(feature = "encryption")]
502            {
503                let data = EventData::String(event.data.clone());
504                event.data = encrypt(sockudo, &event.channel, &data)?;
505            }
506
507            #[cfg(not(feature = "encryption"))]
508            {
509                return Err(SockudoError::Encryption {
510                    message: "Encryption support is not enabled. Enable the 'encryption' feature to use encrypted channels.".to_string(),
511                });
512            }
513        }
514    }
515
516    let batch_payload = json!({ "batch": batch });
517    let mut extra_headers = HashMap::new();
518    if let Some(key) = idempotency_key {
519        extra_headers.insert("X-Idempotency-Key".to_string(), key.to_string());
520    }
521    sockudo
522        .post_with_headers("/batch_events", &batch_payload, &extra_headers)
523        .await
524}
525
526#[cfg(test)]
527mod tests {
528    use super::*;
529    use sonic_rs::json;
530
531    #[test]
532    fn test_event_data_conversions() {
533        // Test string
534        let data = EventData::from_string("hello");
535        assert_eq!(data.to_string(), "hello");
536
537        // Test JSON
538        let json_data = json!({"key": "value"});
539        let data = EventData::from_json(json_data.clone());
540        assert_eq!(data.as_json().unwrap(), json_data);
541
542        // Test From implementations
543        let data: EventData = "test".into();
544        assert!(matches!(data, EventData::String(_)));
545
546        let data: EventData = json!({"test": 123}).into();
547        assert!(matches!(data, EventData::Json(_)));
548    }
549
550    #[test]
551    fn test_batch_event_builder() {
552        let event = BatchEvent::new("test-event", "test-channel", "test-data")
553            .with_socket_id("123.456")
554            .with_info("test-info");
555
556        assert_eq!(event.name, "test-event");
557        assert_eq!(event.channel, "test-channel");
558        assert_eq!(event.data, "test-data");
559        assert_eq!(event.socket_id, Some("123.456".to_string()));
560        assert_eq!(event.info, Some("test-info".to_string()));
561    }
562
563    #[test]
564    fn test_batch_event_with_tags() {
565        let mut tags = HashMap::new();
566        tags.insert("symbol".to_string(), "BONK".to_string());
567        tags.insert("price_usd".to_string(), "0.00001".to_string());
568
569        let event =
570            BatchEvent::new("test-event", "test-channel", "test-data").with_tags(tags.clone());
571
572        assert_eq!(event.tags, Some(tags));
573    }
574
575    #[test]
576    fn test_trigger_params_builder() {
577        let params = TriggerParams::builder()
578            .socket_id("123.456")
579            .info("test-info")
580            .build();
581
582        assert_eq!(params.socket_id, Some("123.456".to_string()));
583        assert_eq!(params.info, Some("test-info".to_string()));
584        assert_eq!(params.idempotency_key, None);
585    }
586
587    #[test]
588    fn test_trigger_params_builder_with_tags() {
589        let mut tags = HashMap::new();
590        tags.insert("event_type".to_string(), "goal".to_string());
591
592        let params = TriggerParams::builder().tags(tags.clone()).build();
593
594        assert_eq!(params.tags, Some(tags));
595    }
596
597    #[test]
598    fn test_trigger_params_builder_with_idempotency_key() {
599        let params = TriggerParams::builder()
600            .socket_id("123.456")
601            .idempotency_key("my-unique-key-123")
602            .build();
603
604        assert_eq!(params.socket_id, Some("123.456".to_string()));
605        assert_eq!(
606            params.idempotency_key,
607            Some("my-unique-key-123".to_string())
608        );
609    }
610
611    #[test]
612    fn test_trigger_params_builder_with_auto_idempotency_key() {
613        let params = TriggerParams::builder().auto_idempotency_key().build();
614
615        assert!(params.idempotency_key.is_some());
616        let key = params.idempotency_key.unwrap();
617        // UUID v4 format: 8-4-4-4-12 hex chars
618        assert_eq!(key.len(), 36);
619        assert_eq!(key.chars().filter(|c| *c == '-').count(), 4);
620    }
621
622    #[test]
623    fn test_generate_idempotency_key() {
624        let key1 = generate_idempotency_key();
625        let key2 = generate_idempotency_key();
626
627        // Each key should be a valid UUID v4 (36 chars with dashes)
628        assert_eq!(key1.len(), 36);
629        assert_eq!(key2.len(), 36);
630        // Keys should be unique
631        assert_ne!(key1, key2);
632    }
633
634    #[test]
635    fn test_batch_event_with_idempotency_key() {
636        let event = BatchEvent::new("test-event", "test-channel", "test-data")
637            .with_idempotency_key("batch-key-123");
638
639        assert_eq!(event.idempotency_key, Some("batch-key-123".to_string()));
640    }
641
642    #[test]
643    fn test_batch_event_with_auto_idempotency_key() {
644        let event =
645            BatchEvent::new("test-event", "test-channel", "test-data").with_auto_idempotency_key();
646
647        assert!(event.idempotency_key.is_some());
648        assert_eq!(event.idempotency_key.unwrap().len(), 36);
649    }
650
651    #[test]
652    fn test_event_serialization_with_idempotency_key() {
653        let event = Event {
654            name: "test".to_string(),
655            data: "{}".to_string(),
656            channels: vec!["test-channel".to_string()],
657            socket_id: None,
658            info: None,
659            tags: None,
660            idempotency_key: Some("key-123".to_string()),
661            extras: None,
662        };
663
664        let json_str = sonic_rs::to_string(&event).unwrap();
665        assert!(json_str.contains("\"idempotency_key\":\"key-123\""));
666    }
667
668    #[test]
669    fn test_event_serialization_without_idempotency_key() {
670        let event = Event {
671            name: "test".to_string(),
672            data: "{}".to_string(),
673            channels: vec!["test-channel".to_string()],
674            socket_id: None,
675            info: None,
676            tags: None,
677            idempotency_key: None,
678            extras: None,
679        };
680
681        let json_str = sonic_rs::to_string(&event).unwrap();
682        assert!(!json_str.contains("idempotency_key"));
683    }
684}
sockudo_http/events.rs

sockudo_http/
events.rs
