sockudo_http/
webhook.rs

1use crate::{Result, SockudoError, Token, WebhookError};
2use serde::{Deserialize, Deserializer, Serialize, de::Error as DeError};
3use sonic_rs::{JsonValueTrait, Value};
4use std::collections::{BTreeMap, HashMap};
5
6/// Webhook for validating and accessing Sockudo webhook data
7#[derive(Debug)]
8pub struct Webhook {
9    token: Token,
10    key: Option<String>,
11    signature: Option<String>,
12    content_type: Option<String>,
13    body: String,
14    data: Option<WebhookData>,
15    raw_json_events: Option<Vec<HashMap<String, Value>>>,
16}
17
18/// Webhook data structure matching Sockudo's format
19#[derive(Debug, Clone, Serialize, Deserialize)]
20pub struct WebhookData {
21    /// The timestamp of the webhook in milliseconds
22    pub time_ms: i64,
23    /// The events received with the webhook
24    #[serde(default, deserialize_with = "deserialize_webhook_events")]
25    pub events: Vec<HashMap<String, String>>,
26}
27
28#[derive(Debug, Deserialize)]
29struct RawWebhookData {
30    #[serde(default)]
31    events: Vec<HashMap<String, Value>>,
32}
33
34fn deserialize_webhook_events<'de, D>(
35    deserializer: D,
36) -> std::result::Result<Vec<HashMap<String, String>>, D::Error>
37where
38    D: Deserializer<'de>,
39{
40    let raw_events = Vec::<HashMap<String, Value>>::deserialize(deserializer)?;
41    project_webhook_events(&raw_events)
42}
43
44fn project_webhook_events<E>(
45    raw_events: &[HashMap<String, Value>],
46) -> std::result::Result<Vec<HashMap<String, String>>, E>
47where
48    E: DeError,
49{
50    raw_events
51        .iter()
52        .map(|event| {
53            event
54                .iter()
55                .map(|(key, value)| {
56                    stringify_webhook_value(value).map(|value| (key.clone(), value))
57                })
58                .collect::<std::result::Result<HashMap<_, _>, E>>()
59        })
60        .collect()
61}
62
63fn stringify_webhook_value<E>(value: &Value) -> std::result::Result<String, E>
64where
65    E: DeError,
66{
67    value
68        .as_str()
69        .map(ToOwned::to_owned)
70        .map(Ok)
71        .unwrap_or_else(|| sonic_rs::to_string(&value).map_err(E::custom))
72}
73
74/// Strongly typed webhook event
75#[derive(Debug, Clone, PartialEq)]
76pub enum WebhookEvent {
77    ChannelOccupied {
78        channel: String,
79    },
80    ChannelVacated {
81        channel: String,
82    },
83    MemberAdded {
84        channel: String,
85        user_id: String,
86    },
87    MemberRemoved {
88        channel: String,
89        user_id: String,
90    },
91    ClientEvent {
92        channel: String,
93        event: String,
94        data: String,
95        socket_id: String,
96        user_id: Option<String>,
97    },
98    CacheMiss {
99        channel: String,
100        event: String,
101    },
102    Unknown(HashMap<String, String>),
103}
104
105impl Webhook {
106    /// Creates a new webhook from request data
107    pub fn new(token: &Token, headers: &BTreeMap<String, String>, body: &str) -> Self {
108        // Normalize header names to lowercase for case-insensitive lookup
109        let normalized_headers: BTreeMap<String, String> = headers
110            .iter()
111            .map(|(k, v)| (k.to_lowercase(), v.clone()))
112            .collect();
113
114        let key = normalized_headers.get("x-pusher-key").cloned();
115        let signature = normalized_headers.get("x-pusher-signature").cloned();
116        let content_type = normalized_headers.get("content-type").cloned();
117
118        let (data, raw_json_events) = if Self::validate_content_type(&content_type) {
119            (
120                sonic_rs::from_str::<WebhookData>(body).ok(),
121                sonic_rs::from_str::<RawWebhookData>(body)
122                    .ok()
123                    .map(|raw| raw.events),
124            )
125        } else {
126            (None, None)
127        };
128
129        Self {
130            token: token.clone(),
131            key,
132            signature,
133            content_type,
134            body: body.to_string(),
135            data,
136            raw_json_events,
137        }
138    }
139
140    /// Validates the webhook signature and content
141    pub fn is_valid(&self, extra_tokens: Option<&[Token]>) -> bool {
142        if !self.is_body_valid() {
143            return false;
144        }
145
146        let tokens_to_check = if let Some(extra) = extra_tokens {
147            let mut tokens = vec![&self.token];
148            tokens.extend(extra.iter());
149            tokens
150        } else {
151            vec![&self.token]
152        };
153
154        if let (Some(key), Some(signature)) = (&self.key, &self.signature) {
155            for token in tokens_to_check {
156                if key == &token.key && token.verify(&self.body, signature) {
157                    return true;
158                }
159            }
160        }
161
162        false
163    }
164
165    /// Checks if the content type is valid (application/json)
166    pub fn is_content_type_valid(&self) -> bool {
167        Self::validate_content_type(&self.content_type)
168    }
169
170    /// Private helper method to validate content type
171    fn validate_content_type(content_type: &Option<String>) -> bool {
172        match content_type {
173            Some(ct) => ct.starts_with("application/json"),
174            None => false,
175        }
176    }
177
178    /// Checks if the body is valid JSON
179    pub fn is_body_valid(&self) -> bool {
180        self.data.is_some()
181    }
182
183    /// Gets the parsed webhook data
184    pub fn get_data(&self) -> Result<&WebhookData> {
185        self.data.as_ref().ok_or_else(|| {
186            SockudoError::Webhook(WebhookError::new(
187                "Invalid webhook body",
188                self.content_type.clone(),
189                &self.body,
190                self.signature.clone(),
191            ))
192        })
193    }
194
195    /// Gets the raw events from webhook data
196    pub fn get_raw_events(&self) -> Result<&Vec<HashMap<String, String>>> {
197        Ok(&self.get_data()?.events)
198    }
199
200    /// Gets the original JSON event objects without stringifying nested values.
201    pub fn get_raw_json_events(&self) -> Result<&Vec<HashMap<String, Value>>> {
202        self.raw_json_events.as_ref().ok_or_else(|| {
203            SockudoError::Webhook(WebhookError::new(
204                "Invalid webhook body",
205                self.content_type.clone(),
206                &self.body,
207                self.signature.clone(),
208            ))
209        })
210    }
211
212    /// Gets the events as strongly typed enums
213    pub fn get_events(&self) -> Result<Vec<WebhookEvent>> {
214        let raw_events = self.get_raw_events()?;
215        Ok(raw_events.iter().map(parse_webhook_event).collect())
216    }
217
218    /// Gets the timestamp from webhook data
219    pub fn get_time(&self) -> Result<std::time::SystemTime> {
220        let time_ms = self.get_data()?.time_ms;
221        if time_ms < 0 {
222            return Err(SockudoError::Webhook(WebhookError::new(
223                "Invalid negative timestamp",
224                self.content_type.clone(),
225                &self.body,
226                self.signature.clone(),
227            )));
228        }
229        let duration = std::time::Duration::from_millis(time_ms as u64);
230        Ok(std::time::UNIX_EPOCH + duration)
231    }
232
233    /// Gets the raw body
234    pub fn body(&self) -> &str {
235        &self.body
236    }
237
238    /// Gets the signature
239    pub fn signature(&self) -> Option<&str> {
240        self.signature.as_deref()
241    }
242
243    /// Gets the key from headers
244    pub fn key(&self) -> Option<&str> {
245        self.key.as_deref()
246    }
247
248    /// Finds events by type
249    pub fn find_events_by_type(&self, event_type: &str) -> Result<Vec<WebhookEvent>> {
250        let events = self.get_events()?;
251        Ok(events
252            .into_iter()
253            .filter(|e| e.event_name() == event_type)
254            .collect())
255    }
256
257    /// Finds events by channel
258    pub fn find_events_by_channel(&self, channel: &str) -> Result<Vec<WebhookEvent>> {
259        let events = self.get_events()?;
260        Ok(events
261            .into_iter()
262            .filter(|e| e.channel() == Some(channel))
263            .collect())
264    }
265}
266
267/// Parses a raw webhook event into a strongly typed event
268fn parse_webhook_event(raw: &HashMap<String, String>) -> WebhookEvent {
269    match raw.get("name").map(|s| s.as_str()) {
270        Some("channel_occupied") => {
271            if let Some(channel) = raw.get("channel") {
272                WebhookEvent::ChannelOccupied {
273                    channel: channel.clone(),
274                }
275            } else {
276                WebhookEvent::Unknown(raw.clone())
277            }
278        }
279        Some("channel_vacated") => {
280            if let Some(channel) = raw.get("channel") {
281                WebhookEvent::ChannelVacated {
282                    channel: channel.clone(),
283                }
284            } else {
285                WebhookEvent::Unknown(raw.clone())
286            }
287        }
288        Some("member_added") => {
289            if let (Some(channel), Some(user_id)) = (raw.get("channel"), raw.get("user_id")) {
290                WebhookEvent::MemberAdded {
291                    channel: channel.clone(),
292                    user_id: user_id.clone(),
293                }
294            } else {
295                WebhookEvent::Unknown(raw.clone())
296            }
297        }
298        Some("member_removed") => {
299            if let (Some(channel), Some(user_id)) = (raw.get("channel"), raw.get("user_id")) {
300                WebhookEvent::MemberRemoved {
301                    channel: channel.clone(),
302                    user_id: user_id.clone(),
303                }
304            } else {
305                WebhookEvent::Unknown(raw.clone())
306            }
307        }
308        Some("client_event") => {
309            if let (Some(channel), Some(event), Some(data), Some(socket_id)) = (
310                raw.get("channel"),
311                raw.get("event"),
312                raw.get("data"),
313                raw.get("socket_id"),
314            ) {
315                WebhookEvent::ClientEvent {
316                    channel: channel.clone(),
317                    event: event.clone(),
318                    data: data.clone(),
319                    socket_id: socket_id.clone(),
320                    user_id: raw.get("user_id").cloned(),
321                }
322            } else {
323                WebhookEvent::Unknown(raw.clone())
324            }
325        }
326        Some("cache_miss") => {
327            if let (Some(channel), Some(event)) = (raw.get("channel"), raw.get("event")) {
328                WebhookEvent::CacheMiss {
329                    channel: channel.clone(),
330                    event: event.clone(),
331                }
332            } else {
333                WebhookEvent::Unknown(raw.clone())
334            }
335        }
336        _ => WebhookEvent::Unknown(raw.clone()),
337    }
338}
339
340impl WebhookEvent {
341    /// Gets the event name
342    pub fn event_name(&self) -> &str {
343        match self {
344            WebhookEvent::ChannelOccupied { .. } => "channel_occupied",
345            WebhookEvent::ChannelVacated { .. } => "channel_vacated",
346            WebhookEvent::MemberAdded { .. } => "member_added",
347            WebhookEvent::MemberRemoved { .. } => "member_removed",
348            WebhookEvent::ClientEvent { .. } => "client_event",
349            WebhookEvent::CacheMiss { .. } => "cache_miss",
350            WebhookEvent::Unknown(map) => map.get("name").map(|s| s.as_str()).unwrap_or("unknown"),
351        }
352    }
353
354    /// Gets the channel name if applicable
355    pub fn channel(&self) -> Option<&str> {
356        match self {
357            WebhookEvent::ChannelOccupied { channel }
358            | WebhookEvent::ChannelVacated { channel }
359            | WebhookEvent::MemberAdded { channel, .. }
360            | WebhookEvent::MemberRemoved { channel, .. }
361            | WebhookEvent::ClientEvent { channel, .. }
362            | WebhookEvent::CacheMiss { channel, .. } => Some(channel),
363            WebhookEvent::Unknown(map) => map.get("channel").map(|s| s.as_str()),
364        }
365    }
366
367    /// Gets the user ID if applicable
368    pub fn user_id(&self) -> Option<&str> {
369        match self {
370            WebhookEvent::MemberAdded { user_id, .. }
371            | WebhookEvent::MemberRemoved { user_id, .. } => Some(user_id),
372            WebhookEvent::ClientEvent { user_id, .. } => user_id.as_deref(),
373            WebhookEvent::Unknown(map) => map.get("user_id").map(|s| s.as_str()),
374            _ => None,
375        }
376    }
377
378    /// Converts the event back to a HashMap
379    pub fn to_hashmap(&self) -> HashMap<String, String> {
380        let mut map = HashMap::new();
381
382        match self {
383            WebhookEvent::ChannelOccupied { channel } => {
384                map.insert("name".to_string(), "channel_occupied".to_string());
385                map.insert("channel".to_string(), channel.clone());
386            }
387            WebhookEvent::ChannelVacated { channel } => {
388                map.insert("name".to_string(), "channel_vacated".to_string());
389                map.insert("channel".to_string(), channel.clone());
390            }
391            WebhookEvent::MemberAdded { channel, user_id } => {
392                map.insert("name".to_string(), "member_added".to_string());
393                map.insert("channel".to_string(), channel.clone());
394                map.insert("user_id".to_string(), user_id.clone());
395            }
396            WebhookEvent::MemberRemoved { channel, user_id } => {
397                map.insert("name".to_string(), "member_removed".to_string());
398                map.insert("channel".to_string(), channel.clone());
399                map.insert("user_id".to_string(), user_id.clone());
400            }
401            WebhookEvent::ClientEvent {
402                channel,
403                event,
404                data,
405                socket_id,
406                user_id,
407            } => {
408                map.insert("name".to_string(), "client_event".to_string());
409                map.insert("channel".to_string(), channel.clone());
410                map.insert("event".to_string(), event.clone());
411                map.insert("data".to_string(), data.clone());
412                map.insert("socket_id".to_string(), socket_id.clone());
413                if let Some(uid) = user_id {
414                    map.insert("user_id".to_string(), uid.clone());
415                }
416            }
417            WebhookEvent::CacheMiss { channel, event } => {
418                map.insert("name".to_string(), "cache_miss".to_string());
419                map.insert("channel".to_string(), channel.clone());
420                map.insert("event".to_string(), event.clone());
421            }
422            WebhookEvent::Unknown(original) => {
423                return original.clone();
424            }
425        }
426
427        map
428    }
429}
430
431#[cfg(test)]
432mod tests {
433    use super::*;
434    use sonic_rs::JsonContainerTrait;
435
436    #[test]
437    fn test_webhook_data_parsing() {
438        let json_str = r#"{
439            "time_ms": 1234567890,
440            "events": [
441                {"name": "channel_occupied", "channel": "test-channel"},
442                {"name": "member_added", "channel": "presence-channel", "user_id": "user123"}
443            ]
444        }"#;
445
446        let data: WebhookData = sonic_rs::from_str(json_str).unwrap();
447        assert_eq!(data.time_ms, 1234567890);
448        assert_eq!(data.events.len(), 2);
449        assert_eq!(
450            data.events[0].get("name"),
451            Some(&"channel_occupied".to_string())
452        );
453    }
454
455    #[test]
456    fn test_webhook_data_accepts_nested_future_values() {
457        let json_str = r#"{
458            "time_ms": 1234567890,
459            "events": [
460                {"name": "ai_turn_started", "channel": "private-ai", "data": {"turn_id": "turn-1", "tokens": ["hello", "world"], "done": false, "nullable": null}}
461            ]
462        }"#;
463
464        let data: WebhookData = sonic_rs::from_str(json_str).unwrap();
465
466        assert_eq!(
467            data.events[0].get("name"),
468            Some(&"ai_turn_started".to_string())
469        );
470        assert_eq!(
471            data.events[0].get("data"),
472            Some(
473                &r#"{"turn_id":"turn-1","tokens":["hello","world"],"done":false,"nullable":null}"#
474                    .to_string()
475            )
476        );
477    }
478
479    #[test]
480    fn test_webhook_preserves_raw_nested_future_values() {
481        let token = Token::new("test_key", "test_secret");
482        let body = r#"{"time_ms":1710000000000,"events":[{"name":"ai_turn_started","channel":"private-ai-forward","data":{"turn_id":"turn-1","tokens":["hello","world"],"done":false,"nullable":null},"future_field":{"nested":true}}]}"#;
483        let signature = token.sign(body);
484
485        let mut headers = BTreeMap::new();
486        headers.insert("content-type".to_string(), "application/json".to_string());
487        headers.insert("x-pusher-key".to_string(), "test_key".to_string());
488        headers.insert("x-pusher-signature".to_string(), signature);
489
490        let webhook = Webhook::new(&token, &headers, body);
491        let raw_events = webhook.get_raw_json_events().unwrap();
492        let raw_data = raw_events[0].get("data").unwrap();
493        let future_field = raw_events[0].get("future_field").unwrap();
494
495        assert!(webhook.is_valid(None));
496        assert_eq!(
497            raw_data.get("turn_id").and_then(|value| value.as_str()),
498            Some("turn-1")
499        );
500        assert_eq!(
501            raw_data
502                .get("tokens")
503                .and_then(|value| value.as_array())
504                .map(|items| items.len()),
505            Some(2)
506        );
507        assert_eq!(
508            raw_data.get("done").and_then(|value| value.as_bool()),
509            Some(false)
510        );
511        assert_eq!(
512            sonic_rs::to_string(raw_data.get("nullable").unwrap()).unwrap(),
513            "null"
514        );
515        assert_eq!(
516            future_field.get("nested").and_then(|value| value.as_bool()),
517            Some(true)
518        );
519    }
520
521    #[test]
522    fn test_webhook_forward_compat_fixture() {
523        let fixture = include_str!(
524            "../../../tests/ai-conformance/fixtures/forward-compat/future-webhook-events.json"
525        );
526
527        let data: WebhookData = sonic_rs::from_str(fixture).unwrap();
528
529        assert_eq!(data.time_ms, 1710000000000);
530        assert_eq!(data.events.len(), 3);
531        assert_eq!(
532            data.events[0].get("name"),
533            Some(&"member_updated".to_string())
534        );
535        assert_eq!(
536            data.events[0].get("future_field"),
537            Some(&"must-pass-through".to_string())
538        );
539        assert_eq!(
540            data.events[1].get("name"),
541            Some(&"ai_run_started".to_string())
542        );
543        assert_eq!(data.events[1].get("run_id"), Some(&"run-1".to_string()));
544        assert_eq!(
545            data.events[2].get("version_serial"),
546            Some(&"ver-1".to_string())
547        );
548    }
549
550    #[test]
551    fn test_webhook_validation() {
552        let token = Token::new("test_key", "test_secret");
553        let body = r#"{"time_ms": 1234567890, "events": []}"#;
554        let signature = token.sign(body);
555
556        let mut headers = BTreeMap::new();
557        headers.insert("content-type".to_string(), "application/json".to_string());
558        headers.insert("x-pusher-key".to_string(), "test_key".to_string());
559        headers.insert("x-pusher-signature".to_string(), signature);
560
561        let webhook = Webhook::new(&token, &headers, body);
562        assert!(webhook.is_valid(None));
563    }
564
565    #[test]
566    fn test_event_parsing() {
567        let mut event_map = HashMap::new();
568        event_map.insert("name".to_string(), "channel_occupied".to_string());
569        event_map.insert("channel".to_string(), "test-channel".to_string());
570
571        let event = parse_webhook_event(&event_map);
572        assert!(matches!(event, WebhookEvent::ChannelOccupied { .. }));
573        assert_eq!(event.channel(), Some("test-channel"));
574    }
575
576    #[test]
577    fn test_event_round_trip() {
578        let event = WebhookEvent::MemberAdded {
579            channel: "presence-test".to_string(),
580            user_id: "user123".to_string(),
581        };
582
583        let map = event.to_hashmap();
584        let parsed = parse_webhook_event(&map);
585
586        assert_eq!(event, parsed);
587    }
588}
sockudo_http/webhook.rs

sockudo_http/
webhook.rs
