1use crate::errors::HingeError;
2use crate::models::SendbirdSyevEvent;
3use serde::{Deserialize, Serialize};
4use tokio::sync::{broadcast, mpsc};
5
6#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
7#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
8#[serde(rename_all = "camelCase", tag = "kind")]
9pub enum SendbirdWsEvent {
10 SessionKey {
11 key: String,
12 },
13 Read {
14 #[serde(default)]
15 req_id: Option<String>,
16 payload: serde_json::Value,
17 },
18 Typing {
19 event: SendbirdSyevEvent,
20 },
21 Ping {
22 payload: serde_json::Value,
23 },
24 Pong {
25 payload: serde_json::Value,
26 },
27 Close {
28 code: Option<u16>,
29 reason: Option<String>,
30 },
31 Raw {
32 frame: String,
33 },
34}
35
36pub struct SendbirdWsSubscription {
37 commands: mpsc::UnboundedSender<String>,
38 raw: broadcast::Receiver<String>,
39}
40
41impl SendbirdWsSubscription {
42 pub fn new(commands: mpsc::UnboundedSender<String>, raw: broadcast::Receiver<String>) -> Self {
43 Self { commands, raw }
44 }
45
46 pub fn commands(&self) -> mpsc::UnboundedSender<String> {
47 self.commands.clone()
48 }
49
50 pub fn resubscribe_raw(&self) -> broadcast::Receiver<String> {
51 self.raw.resubscribe()
52 }
53
54 pub async fn recv(&mut self) -> Result<SendbirdWsEvent, HingeError> {
55 let frame = self.recv_raw().await?;
56 parse_sendbird_ws_frame(&frame)
57 }
58
59 pub async fn recv_raw(&mut self) -> Result<String, HingeError> {
60 self.raw
61 .recv()
62 .await
63 .map_err(|err| HingeError::Http(format!("sendbird ws receive failed: {err}")))
64 }
65}
66
67pub fn parse_sendbird_ws_frame(frame: &str) -> Result<SendbirdWsEvent, HingeError> {
68 if let Some(rest) = frame.strip_prefix("__SESSION_KEY__:") {
69 return Ok(SendbirdWsEvent::SessionKey {
70 key: rest.to_string(),
71 });
72 }
73
74 if let Some(rest) = frame.strip_prefix("__SYEV__:") {
75 let event = serde_json::from_str::<SendbirdSyevEvent>(rest)
76 .map_err(|err| HingeError::Serde(err.to_string()))?;
77 return Ok(SendbirdWsEvent::Typing { event });
78 }
79
80 if let Some(rest) = frame.strip_prefix("__CLOSE__:") {
81 let mut parts = rest.splitn(2, ':');
82 let code = parts.next().and_then(|part| part.parse::<u16>().ok());
83 let reason = parts
84 .next()
85 .filter(|part| !part.is_empty())
86 .map(ToOwned::to_owned);
87 return Ok(SendbirdWsEvent::Close { code, reason });
88 }
89
90 parse_prefixed_json(frame)
91}
92
93fn parse_prefixed_json(frame: &str) -> Result<SendbirdWsEvent, HingeError> {
94 let Some(start) = frame.find('{') else {
95 return Ok(SendbirdWsEvent::Raw {
96 frame: frame.to_string(),
97 });
98 };
99 let (prefix, json) = frame.split_at(start);
100 let payload = serde_json::from_str::<serde_json::Value>(json)
101 .map_err(|err| HingeError::Serde(err.to_string()))?;
102
103 match prefix {
104 "LOGI" => Ok(SendbirdWsEvent::SessionKey {
105 key: payload
106 .get("key")
107 .and_then(|value| value.as_str())
108 .unwrap_or_default()
109 .to_string(),
110 }),
111 "READ" => Ok(SendbirdWsEvent::Read {
112 req_id: payload
113 .get("req_id")
114 .and_then(|value| value.as_str())
115 .map(ToOwned::to_owned),
116 payload,
117 }),
118 "SYEV" => {
119 let event = serde_json::from_value::<SendbirdSyevEvent>(payload)
120 .map_err(|err| HingeError::Serde(err.to_string()))?;
121 Ok(SendbirdWsEvent::Typing { event })
122 }
123 "PING" => Ok(SendbirdWsEvent::Ping { payload }),
124 "PONG" => Ok(SendbirdWsEvent::Pong { payload }),
125 _ => Ok(SendbirdWsEvent::Raw {
126 frame: frame.to_string(),
127 }),
128 }
129}
130
131#[cfg(test)]
132mod tests {
133 use super::*;
134
135 #[test]
136 fn parses_logi_session_key() {
137 let event = parse_sendbird_ws_frame(r#"LOGI{"key":"session-key","user_id":"u1"}"#)
138 .expect("LOGI frame should parse");
139 assert_eq!(
140 event,
141 SendbirdWsEvent::SessionKey {
142 key: "session-key".to_string()
143 }
144 );
145 }
146
147 #[test]
148 fn parses_read_ack() {
149 let event = parse_sendbird_ws_frame(
150 r#"READ{"req_id":"r1","channel_id":1,"channel_url":"c","channel_type":"group"}"#,
151 )
152 .expect("READ frame should parse");
153 match event {
154 SendbirdWsEvent::Read { req_id, payload } => {
155 assert_eq!(req_id.as_deref(), Some("r1"));
156 assert_eq!(payload["channel_url"], "c");
157 }
158 other => panic!("unexpected event: {other:?}"),
159 }
160 }
161
162 #[test]
163 fn parses_typing_event() {
164 let event = parse_sendbird_ws_frame(
165 r#"SYEV{"cat":10900,"channel_url":"c","channel_type":"group","ts":1,"sts":1}"#,
166 )
167 .expect("SYEV frame should parse");
168 match event {
169 SendbirdWsEvent::Typing { event } => {
170 assert_eq!(event.cat, SendbirdSyevEvent::CATEGORY_TYPING_START);
171 assert_eq!(event.channel_url, "c");
172 }
173 other => panic!("unexpected event: {other:?}"),
174 }
175 }
176
177 #[test]
178 fn parses_ping_pong_and_close() {
179 let ping =
180 parse_sendbird_ws_frame(r#"PING{"req_id":"p1"}"#).expect("PING frame should parse");
181 assert!(matches!(ping, SendbirdWsEvent::Ping { .. }));
182
183 let pong =
184 parse_sendbird_ws_frame(r#"PONG{"req_id":"p1"}"#).expect("PONG frame should parse");
185 assert!(matches!(pong, SendbirdWsEvent::Pong { .. }));
186
187 let close =
188 parse_sendbird_ws_frame("__CLOSE__:1000:done").expect("close sentinel should parse");
189 assert_eq!(
190 close,
191 SendbirdWsEvent::Close {
192 code: Some(1000),
193 reason: Some("done".to_string())
194 }
195 );
196 }
197
198 #[test]
199 fn keeps_unknown_frames_raw() {
200 let event = parse_sendbird_ws_frame("NOPE{}").expect("unknown frame should parse");
201 assert_eq!(
202 event,
203 SendbirdWsEvent::Raw {
204 frame: "NOPE{}".to_string()
205 }
206 );
207 }
208}