Skip to content
zhenyi

gitdata-core

0
Clone

HTTPS

SSH

SSH needs a key in your settings.

Download ZIP

feat(chat): add glayer crate (gateway/realtime layer with WS, handlers, hub)

zhenyi committed
parent 2b38e02 commit 3d72ff3 Browse files

22 files changed

+2212 −0

File tree 22
Added chat/glayer/handler/conversations.rs View chat/glayer/handler/conversations.rs
@@ -0,0 +1,25 @@
1use axum::extract::{Path, State};
2use export::data::AppResponseData;
3use storagelayer::models::conversation::Conversation;
4use uuid::Uuid;
5
6use crate::auth::BearToken;
7use crate::error::{GatewayError, GatewayResult};
8use crate::state::GatewayState;
9
10pub(crate) async fn get_conversation(
11 State(state): State<GatewayState>,
12 BearToken(token): BearToken,
13 Path(conversation_id): Path<Uuid>,
14) -> GatewayResult<AppResponseData<Conversation>> {
15 match state
16 .chat
17 .get_conversation(state.chat.auth(), &token, conversation_id)
18 .await?
19 {
20 Some(c) => Ok(AppResponseData::ok(c)),
21 None => Err(GatewayError::not_found(format!(
22 "conversation {conversation_id}"
23 ))),
24 }
25}
Added chat/glayer/handler/lookups.rs View chat/glayer/handler/lookups.rs
@@ -0,0 +1,25 @@
1use axum::extract::{Query, State};
2use export::data::AppResponseData;
3use serde::{Deserialize, Serialize};
4use uuid::Uuid;
5
6use crate::auth::BearToken;
7use crate::error::GatewayResult;
8use crate::state::GatewayState;
9
10#[derive(Debug, Clone, Serialize, Deserialize)]
11pub struct LookupByClientMsgIdParams {
12 pub client_msg_id: String,
13}
14
15pub(crate) async fn lookup_by_client_msg_id(
16 State(state): State<GatewayState>,
17 BearToken(token): BearToken,
18 Query(params): Query<LookupByClientMsgIdParams>,
19) -> GatewayResult<AppResponseData<Option<Uuid>>> {
20 let id = state
21 .chat
22 .lookup_by_client_msg_id(state.chat.auth(), &token, &params.client_msg_id)
23 .await?;
24 Ok(AppResponseData::ok(id))
25}
Added chat/glayer/handler/mentions.rs View chat/glayer/handler/mentions.rs
@@ -0,0 +1,27 @@
1use axum::extract::{Path, State};
2use chatlayer::MentionSpec;
3use export::data::{AppRequest, AppResponseData, NoData};
4use serde::{Deserialize, Serialize};
5use uuid::Uuid;
6
7use crate::auth::BearToken;
8use crate::error::GatewayResult;
9use crate::state::GatewayState;
10
11#[derive(Debug, Clone, Serialize, Deserialize)]
12pub struct AddMentionRequest {
13 pub mentions: Vec<MentionSpec>,
14}
15
16pub(crate) async fn add_mention(
17 State(state): State<GatewayState>,
18 BearToken(token): BearToken,
19 Path(message_id): Path<Uuid>,
20 AppRequest(req): AppRequest<AddMentionRequest>,
21) -> GatewayResult<AppResponseData<NoData>> {
22 state
23 .chat
24 .add_mention(state.chat.auth(), &token, message_id, req.mentions)
25 .await?;
26 Ok(AppResponseData::empty())
27}
Added chat/glayer/handler/messages.rs View chat/glayer/handler/messages.rs
@@ -0,0 +1,97 @@
1use axum::extract::{Path, Query, State};
2use chatlayer::{AcceptedMsg, NewMessageCmd};
3use export::data::{AppRequest, AppResponseData, NoData};
4use serde::{Deserialize, Serialize};
5use storagelayer::models::message::{ListDirection, MessagePage, MessageQuery};
6use uuid::Uuid;
7
8use crate::auth::BearToken;
9use crate::error::{GatewayError, GatewayResult};
10use crate::state::GatewayState;
11
12#[derive(Debug, Clone, Serialize, Deserialize)]
13pub struct EditMessageRequest {
14 pub body: String,
15}
16
17#[derive(Debug, Clone, Default, Serialize, Deserialize)]
18#[serde(default)]
19pub struct ListMessagesParams {
20 pub direction: Option<ListDirection>,
21 pub limit: Option<u32>,
22 pub cursor: Option<String>,
23 pub include_deleted: Option<bool>,
24}
25
26impl From<ListMessagesParams> for MessageQuery {
27 fn from(p: ListMessagesParams) -> Self {
28 let mut q = MessageQuery::default();
29 if let Some(d) = p.direction {
30 q.direction = d;
31 }
32 if let Some(l) = p.limit {
33 q.limit = l;
34 }
35 q.cursor = p.cursor;
36 if let Some(b) = p.include_deleted {
37 q.include_deleted = b;
38 }
39 q
40 }
41}
42
43pub(crate) async fn send_message(
44 State(state): State<GatewayState>,
45 BearToken(token): BearToken,
46 Path(conversation_id): Path<Uuid>,
47 AppRequest(cmd): AppRequest<NewMessageCmd>,
48) -> GatewayResult<AppResponseData<AcceptedMsg>> {
49 if cmd.conversation_id != conversation_id {
50 return Err(GatewayError::bad_request(
51 "path conversation_id does not match body",
52 ));
53 }
54 let accepted = state
55 .chat
56 .send_message(state.chat.auth(), &token, cmd)
57 .await?;
58 Ok(AppResponseData::ok(accepted))
59}
60
61pub(crate) async fn list_messages(
62 State(state): State<GatewayState>,
63 BearToken(token): BearToken,
64 Path(conversation_id): Path<Uuid>,
65 Query(params): Query<ListMessagesParams>,
66) -> GatewayResult<AppResponseData<MessagePage>> {
67 let page = state
68 .chat
69 .list_messages(state.chat.auth(), &token, conversation_id, params.into())
70 .await?;
71 Ok(AppResponseData::ok(page))
72}
73
74pub(crate) async fn edit_message(
75 State(state): State<GatewayState>,
76 BearToken(token): BearToken,
77 Path(message_id): Path<Uuid>,
78 AppRequest(req): AppRequest<EditMessageRequest>,
79) -> GatewayResult<AppResponseData<NoData>> {
80 state
81 .chat
82 .edit_message(state.chat.auth(), &token, message_id, req.body)
83 .await?;
84 Ok(AppResponseData::empty())
85}
86
87pub(crate) async fn delete_message(
88 State(state): State<GatewayState>,
89 BearToken(token): BearToken,
90 Path(message_id): Path<Uuid>,
91) -> GatewayResult<AppResponseData<NoData>> {
92 state
93 .chat
94 .delete_message(state.chat.auth(), &token, message_id)
95 .await?;
96 Ok(AppResponseData::empty())
97}
Added chat/glayer/handler/mod.rs View chat/glayer/handler/mod.rs
@@ -0,0 +1,9 @@
1pub mod conversations;
2pub mod lookups;
3pub mod mentions;
4pub mod messages;
5pub mod pin;
6pub mod presence;
7pub mod reactions;
8pub mod read;
9pub mod uploads;
Added chat/glayer/handler/pin.rs View chat/glayer/handler/pin.rs
@@ -0,0 +1,31 @@
1use axum::extract::{Path, State};
2use export::data::{AppResponseData, NoData};
3use uuid::Uuid;
4
5use crate::auth::BearToken;
6use crate::error::GatewayResult;
7use crate::state::GatewayState;
8
9pub(crate) async fn set_pin(
10 State(state): State<GatewayState>,
11 BearToken(token): BearToken,
12 Path(message_id): Path<Uuid>,
13) -> GatewayResult<AppResponseData<NoData>> {
14 state
15 .chat
16 .toggle_pin(state.chat.auth(), &token, message_id, true)
17 .await?;
18 Ok(AppResponseData::empty())
19}
20
21pub(crate) async fn unset_pin(
22 State(state): State<GatewayState>,
23 BearToken(token): BearToken,
24 Path(message_id): Path<Uuid>,
25) -> GatewayResult<AppResponseData<NoData>> {
26 state
27 .chat
28 .toggle_pin(state.chat.auth(), &token, message_id, false)
29 .await?;
30 Ok(AppResponseData::empty())
31}
Added chat/glayer/handler/presence.rs View chat/glayer/handler/presence.rs
@@ -0,0 +1,24 @@
1use axum::extract::State;
2use export::data::{AppRequest, AppResponseData, NoData};
3use serde::{Deserialize, Serialize};
4
5use crate::auth::BearToken;
6use crate::error::GatewayResult;
7use crate::state::GatewayState;
8
9#[derive(Debug, Clone, Serialize, Deserialize)]
10pub struct HeartbeatRequest {
11 pub device_id: String,
12}
13
14pub(crate) async fn heartbeat(
15 State(state): State<GatewayState>,
16 BearToken(token): BearToken,
17 AppRequest(req): AppRequest<HeartbeatRequest>,
18) -> GatewayResult<AppResponseData<NoData>> {
19 state
20 .chat
21 .heartbeat(state.chat.auth(), &token, req.device_id)
22 .await?;
23 Ok(AppResponseData::empty())
24}
Added chat/glayer/handler/reactions.rs View chat/glayer/handler/reactions.rs
@@ -0,0 +1,31 @@
1use axum::extract::{Path, State};
2use export::data::{AppResponseData, NoData};
3use uuid::Uuid;
4
5use crate::auth::BearToken;
6use crate::error::GatewayResult;
7use crate::state::GatewayState;
8
9pub(crate) async fn add_reaction(
10 State(state): State<GatewayState>,
11 BearToken(token): BearToken,
12 Path((message_id, emoji)): Path<(Uuid, String)>,
13) -> GatewayResult<AppResponseData<NoData>> {
14 state
15 .chat
16 .add_reaction(state.chat.auth(), &token, message_id, emoji)
17 .await?;
18 Ok(AppResponseData::empty())
19}
20
21pub(crate) async fn remove_reaction(
22 State(state): State<GatewayState>,
23 BearToken(token): BearToken,
24 Path((message_id, emoji)): Path<(Uuid, String)>,
25) -> GatewayResult<AppResponseData<NoData>> {
26 state
27 .chat
28 .remove_reaction(state.chat.auth(), &token, message_id, emoji)
29 .await?;
30 Ok(AppResponseData::empty())
31}
Added chat/glayer/handler/read.rs View chat/glayer/handler/read.rs
@@ -0,0 +1,31 @@
1use axum::extract::{Path, State};
2use export::data::{AppRequest, AppResponseData, NoData};
3use serde::{Deserialize, Serialize};
4use uuid::Uuid;
5
6use crate::auth::BearToken;
7use crate::error::GatewayResult;
8use crate::state::GatewayState;
9
10#[derive(Debug, Clone, Serialize, Deserialize)]
11pub struct MarkReadRequest {
12 pub last_read_message_id: Uuid,
13}
14
15pub(crate) async fn mark_read(
16 State(state): State<GatewayState>,
17 BearToken(token): BearToken,
18 Path(conversation_id): Path<Uuid>,
19 AppRequest(req): AppRequest<MarkReadRequest>,
20) -> GatewayResult<AppResponseData<NoData>> {
21 state
22 .chat
23 .mark_read(
24 state.chat.auth(),
25 &token,
26 conversation_id,
27 req.last_read_message_id,
28 )
29 .await?;
30 Ok(AppResponseData::empty())
31}
Added chat/glayer/handler/uploads.rs View chat/glayer/handler/uploads.rs
@@ -0,0 +1,100 @@
1use std::time::Duration;
2
3use axum::extract::{Path, State};
4use chatlayer::AttachmentMeta;
5use export::data::{AppRequest, AppResponseData, NoData};
6use serde::{Deserialize, Serialize};
7use storagelayer::models::attachment::AttachmentOwner;
8use uuid::Uuid;
9
10use crate::auth::BearToken;
11use crate::error::GatewayResult;
12use crate::state::GatewayState;
13
14#[derive(Debug, Clone, Serialize, Deserialize)]
15pub struct PresignUploadRequest {
16 pub owner: AttachmentOwner,
17 pub owner_hint: Option<Uuid>,
18 pub filename: String,
19 pub content_type: String,
20 pub size_hint: Option<i64>,
21}
22
23#[derive(Debug, Clone, Serialize)]
24pub struct PresignedUploadResponse {
25 pub attachment_id: Uuid,
26 pub upload_url: String,
27 pub key: String,
28 pub filename: String,
29 pub content_type: String,
30 pub size_hint: Option<i64>,
31}
32
33#[derive(Debug, Clone, Serialize, Deserialize)]
34pub struct FinalizeUploadRequest {
35 pub size: i64,
36 pub content_type: String,
37}
38
39pub(crate) async fn presign_upload(
40 State(state): State<GatewayState>,
41 BearToken(token): BearToken,
42 AppRequest(req): AppRequest<PresignUploadRequest>,
43) -> GatewayResult<AppResponseData<PresignedUploadResponse>> {
44 let ttl = Duration::from_secs(state.settings.upload_presign_ttl_seconds);
45 let upload = state
46 .chat
47 .media()
48 .presign_upload(
49 state.chat.auth(),
50 &token,
51 req.owner,
52 req.owner_hint,
53 &req.filename,
54 &req.content_type,
55 req.size_hint,
56 ttl,
57 )
58 .await?;
59 Ok(AppResponseData::ok(PresignedUploadResponse {
60 attachment_id: upload.attachment_id,
61 upload_url: upload.upload_url.to_string(),
62 key: upload.key,
63 filename: upload.filename,
64 content_type: upload.content_type,
65 size_hint: upload.size_hint,
66 }))
67}
68
69pub(crate) async fn finalize_upload(
70 State(state): State<GatewayState>,
71 BearToken(token): BearToken,
72 Path(attachment_id): Path<Uuid>,
73 AppRequest(req): AppRequest<FinalizeUploadRequest>,
74) -> GatewayResult<AppResponseData<AttachmentMeta>> {
75 let meta = state
76 .chat
77 .media()
78 .finalize_upload(
79 state.chat.auth(),
80 &token,
81 attachment_id,
82 req.size,
83 &req.content_type,
84 )
85 .await?;
86 Ok(AppResponseData::ok(meta))
87}
88
89pub(crate) async fn abort_upload(
90 State(state): State<GatewayState>,
91 BearToken(token): BearToken,
92 Path(attachment_id): Path<Uuid>,
93) -> GatewayResult<AppResponseData<NoData>> {
94 state
95 .chat
96 .media()
97 .abort_upload(state.chat.auth(), &token, attachment_id)
98 .await?;
99 Ok(AppResponseData::empty())
100}
Added chat/glayer/auth.rs View chat/glayer/auth.rs
@@ -0,0 +1,87 @@
1use axum::extract::FromRequestParts;
2use axum::http::header::AUTHORIZATION;
3use axum::http::request::Parts;
4
5use crate::error::{GatewayError, GatewayResult};
6
7/// Raw bearer token string from `Authorization: Bearer <token>`. The token is
8#[derive(Debug, Clone)]
9pub struct BearToken(pub String);
10
11impl<S: Send + Sync> FromRequestParts<S> for BearToken {
12 type Rejection = GatewayError;
13
14 async fn from_request_parts(parts: &mut Parts, _state: &S) -> GatewayResult<Self> {
15 let header = parts
16 .headers
17 .get(AUTHORIZATION)
18 .and_then(|v| v.to_str().ok())
19 .ok_or_else(|| GatewayError::Unauthorized("missing bearer token".into()))?;
20 let mut it = header.split_whitespace();
21 let scheme = it.next().unwrap_or("");
22 if !scheme.eq_ignore_ascii_case("bearer") {
23 return Err(GatewayError::Unauthorized(
24 "invalid authorization scheme".into(),
25 ));
26 }
27 let token = it
28 .next()
29 .ok_or_else(|| GatewayError::Unauthorized("missing bearer token".into()))?;
30 Ok(BearToken(token.to_string()))
31 }
32}
33
34#[cfg(test)]
35mod tests {
36 use super::*;
37 use axum::http::Request;
38
39 fn parts_with_auth(value: Option<&str>) -> Parts {
40 let mut builder = Request::builder();
41 if let Some(v) = value {
42 builder = builder.header(AUTHORIZATION, v);
43 }
44 builder.body(()).unwrap().into_parts().0
45 }
46
47 #[tokio::test]
48 async fn bearer_token_extracts_raw_token() {
49 let mut parts = parts_with_auth(Some("Bearer abc.def.ghi"));
50 let token = BearToken::from_request_parts(&mut parts, &())
51 .await
52 .unwrap();
53 assert_eq!(token.0, "abc.def.ghi");
54 }
55
56 #[tokio::test]
57 async fn bearer_token_missing_header_is_401() {
58 use axum::response::IntoResponse;
59 let mut parts = parts_with_auth(None);
60 let err = BearToken::from_request_parts(&mut parts, &())
61 .await
62 .unwrap_err();
63 assert!(matches!(err, GatewayError::Unauthorized(_)));
64 assert_eq!(
65 err.into_response().status(),
66 axum::http::StatusCode::UNAUTHORIZED
67 );
68 }
69
70 #[tokio::test]
71 async fn bearer_token_wrong_scheme_is_401() {
72 let mut parts = parts_with_auth(Some("Basic abcdef"));
73 let err = BearToken::from_request_parts(&mut parts, &())
74 .await
75 .unwrap_err();
76 assert!(matches!(err, GatewayError::Unauthorized(_)));
77 }
78
79 #[tokio::test]
80 async fn bearer_token_empty_value_is_401() {
81 let mut parts = parts_with_auth(Some("Bearer "));
82 let err = BearToken::from_request_parts(&mut parts, &())
83 .await
84 .unwrap_err();
85 assert!(matches!(err, GatewayError::Unauthorized(_)));
86 }
87}
Added chat/glayer/Cargo.toml View chat/glayer/Cargo.toml
@@ -0,0 +1,41 @@
1[package]
2name = "glayer"
3version.workspace = true
4edition.workspace = true
5authors.workspace = true
6categories.workspace = true
7keywords.workspace = true
8repository.workspace = true
9homepage.workspace = true
10
11[lib]
12path = "lib.rs"
13name = "glayer"
14
15[dependencies]
16chatlayer.workspace = true
17config.workspace = true
18storagelayer.workspace = true
19queue.workspace = true
20export.workspace = true
21
22axum = { workspace = true, features = ["ws"] }
23tokio.workspace = true
24tokio-util.workspace = true
25bytes.workspace = true
26serde.workspace = true
27serde_json.workspace = true
28time.workspace = true
29uuid.workspace = true
30tracing.workspace = true
31metrics.workspace = true
32thiserror.workspace = true
33
34[dev-dependencies]
35storagelayer = { workspace = true, features = ["mock"] }
36config.workspace = true
37jwt.workspace = true
38sea-orm = { workspace = true, features = ["mock"] }
39jsonwebtoken = { version = "11.0.0", features = ["rust_crypto"] }
40ed25519-dalek = { version = "2", features = ["pkcs8", "rand_core"] }
41rand_core = { version = "0.6", features = ["getrandom"] }
No newline at end of file
Added chat/glayer/consumer.rs View chat/glayer/consumer.rs
@@ -0,0 +1,335 @@
1use std::sync::Arc;
2
3use chatlayer::{ChatLiveViewEvent, ChatUserViewEvent, EventEnvelope};
4use queue::{Queue, QueueError, SubscribeHandle, SubscriptionMode};
5
6use crate::hub::ConnectionHub;
7use crate::protocol::ServerFrame;
8
9#[derive(Debug, Clone)]
10pub struct UserViewConsumerSettings {
11 pub topic: String,
12 pub subscriber_id: String,
13}
14
15pub struct UserViewConsumer {
16 queue: Arc<Queue>,
17 hub: Arc<ConnectionHub>,
18 settings: UserViewConsumerSettings,
19}
20
21impl std::fmt::Debug for UserViewConsumer {
22 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
23 f.debug_struct("UserViewConsumer")
24 .field("settings", &self.settings)
25 .finish_non_exhaustive()
26 }
27}
28
29impl UserViewConsumer {
30 pub fn new(
31 queue: Arc<Queue>,
32 hub: Arc<ConnectionHub>,
33 settings: UserViewConsumerSettings,
34 ) -> Self {
35 Self {
36 queue,
37 hub,
38 settings,
39 }
40 }
41
42 pub fn run(&self) -> Result<SubscribeHandle, QueueError> {
43 let mode = SubscriptionMode::Broadcast {
44 subscriber_id: self.settings.subscriber_id.clone(),
45 };
46 let topics = [self.settings.topic.as_str()];
47 let hub = self.hub.clone();
48 self.queue.subscribe_raw(mode, &topics, move |msg| {
49 let hub = hub.clone();
50 Box::pin(async move {
51 match serde_json::from_slice::<EventEnvelope<ChatUserViewEvent>>(&msg.payload) {
52 Ok(envelope) => {
53 let delivered = dispatch_envelope(&hub, envelope);
54 crate::metrics::user_view_delivered(delivered);
55 Ok(())
56 }
57 Err(e) => {
58 tracing::warn!(
59 topic = %msg.topic,
60 offset = msg.offset,
61 error = %e,
62 "malformed user view event dropped"
63 );
64 crate::metrics::user_view_dropped();
65 Ok(())
66 }
67 }
68 })
69 })
70 }
71}
72
73pub fn dispatch_envelope(hub: &ConnectionHub, envelope: EventEnvelope<ChatUserViewEvent>) -> usize {
74 hub.dispatch(
75 envelope.payload.user_id(),
76 ServerFrame::Event { event: envelope },
77 )
78}
79
80pub struct MeetViewConsumer {
81 queue: Arc<Queue>,
82 hub: Arc<ConnectionHub>,
83 settings: UserViewConsumerSettings,
84}
85
86impl std::fmt::Debug for MeetViewConsumer {
87 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
88 f.debug_struct("MeetViewConsumer")
89 .field("settings", &self.settings)
90 .finish_non_exhaustive()
91 }
92}
93
94impl MeetViewConsumer {
95 pub fn new(
96 queue: Arc<Queue>,
97 hub: Arc<ConnectionHub>,
98 settings: UserViewConsumerSettings,
99 ) -> Self {
100 Self {
101 queue,
102 hub,
103 settings,
104 }
105 }
106
107 pub fn run(&self) -> Result<SubscribeHandle, QueueError> {
108 let mode = SubscriptionMode::Broadcast {
109 subscriber_id: self.settings.subscriber_id.clone(),
110 };
111 let topics = [self.settings.topic.as_str()];
112 let hub = self.hub.clone();
113 self.queue.subscribe_raw(mode, &topics, move |msg| {
114 let hub = hub.clone();
115 Box::pin(async move {
116 match serde_json::from_slice::<EventEnvelope<ChatLiveViewEvent>>(&msg.payload) {
117 Ok(envelope) => {
118 let delivered = dispatch_live_envelope(&hub, envelope);
119 crate::metrics::live_view_delivered(delivered);
120 Ok(())
121 }
122 Err(e) => {
123 tracing::warn!(
124 topic = %msg.topic,
125 offset = msg.offset,
126 error = %e,
127 "malformed live view event dropped"
128 );
129 crate::metrics::live_view_dropped();
130 Ok(())
131 }
132 }
133 })
134 })
135 }
136}
137
138pub fn dispatch_live_envelope(
139 hub: &ConnectionHub,
140 envelope: EventEnvelope<ChatLiveViewEvent>,
141) -> usize {
142 hub.dispatch(
143 envelope.payload.user_id(),
144 ServerFrame::Meet { event: envelope },
145 )
146}
147
148pub struct TypingConsumer {
149 queue: Arc<Queue>,
150 state: Arc<crate::state::GatewayState>,
151 settings: UserViewConsumerSettings,
152}
153
154impl std::fmt::Debug for TypingConsumer {
155 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
156 f.debug_struct("TypingConsumer")
157 .field("settings", &self.settings)
158 .finish_non_exhaustive()
159 }
160}
161
162impl TypingConsumer {
163 pub fn new(
164 queue: Arc<Queue>,
165 state: Arc<crate::state::GatewayState>,
166 settings: UserViewConsumerSettings,
167 ) -> Self {
168 Self {
169 queue,
170 state,
171 settings,
172 }
173 }
174
175 pub fn run(&self) -> Result<SubscribeHandle, QueueError> {
176 let mode = SubscriptionMode::Broadcast {
177 subscriber_id: self.settings.subscriber_id.clone(),
178 };
179 let topics = [self.settings.topic.as_str()];
180 let state = self.state.clone();
181 self.queue.subscribe_raw(mode, &topics, move |msg| {
182 let state = state.clone();
183 Box::pin(async move {
184 match serde_json::from_slice::<crate::typing::TypingSignal>(&msg.payload) {
185 Ok(signal) => {
186 let store = state.chat.storage();
187 crate::typing::broadcast_typing(&store, &state.hub, &signal).await;
188 Ok(())
189 }
190 Err(e) => {
191 tracing::warn!(
192 topic = %msg.topic,
193 offset = msg.offset,
194 error = %e,
195 "malformed typing signal dropped"
196 );
197 Ok(())
198 }
199 }
200 })
201 })
202 }
203}
204
205#[cfg(test)]
206mod tests {
207 use super::*;
208 use chatlayer::{ChatLiveViewEvent, ChatUserViewEvent};
209 use time::OffsetDateTime;
210 use tokio::sync::mpsc;
211 use uuid::Uuid;
212
213 use crate::hub::{Connection, ConnectionHub};
214
215 fn new_envelope(user_id: Uuid) -> EventEnvelope<ChatUserViewEvent> {
216 EventEnvelope::new(
217 user_id,
218 ChatUserViewEvent::NewMessage {
219 user_id,
220 conversation_id: Uuid::now_v7(),
221 message_id: Uuid::now_v7(),
222 sender_id: Uuid::now_v7(),
223 body: "hi".to_string(),
224 ts: OffsetDateTime::now_utc(),
225 },
226 )
227 }
228
229 #[test]
230 fn dispatch_envelope_routes_by_payload_user_id() {
231 let hub = ConnectionHub::new(8);
232 let alice = Uuid::now_v7();
233 let bob = Uuid::now_v7();
234
235 let (tx_a, mut rx_a) = mpsc::unbounded_channel();
236 hub.register(Connection {
237 id: Uuid::now_v7(),
238 user_id: alice,
239 tx: tx_a,
240 established_at: OffsetDateTime::now_utc(),
241 })
242 .unwrap();
243
244 let (tx_b, mut rx_b) = mpsc::unbounded_channel();
245 hub.register(Connection {
246 id: Uuid::now_v7(),
247 user_id: bob,
248 tx: tx_b,
249 established_at: OffsetDateTime::now_utc(),
250 })
251 .unwrap();
252
253 let n = dispatch_envelope(&hub, new_envelope(alice));
254 assert_eq!(n, 1);
255 assert!(rx_a.try_recv().is_ok(), "alice receives");
256 assert!(rx_b.try_recv().is_err(), "bob does not receive");
257 }
258
259 #[test]
260 fn dispatch_envelope_returns_delivered_count() {
261 let hub = ConnectionHub::new(8);
262 let user = Uuid::now_v7();
263
264 let mut keep_alive = Vec::new();
265 for _ in 0..3 {
266 let (tx, rx) = mpsc::unbounded_channel();
267 keep_alive.push(rx);
268 hub.register(Connection {
269 id: Uuid::now_v7(),
270 user_id: user,
271 tx,
272 established_at: OffsetDateTime::now_utc(),
273 })
274 .unwrap();
275 }
276
277 let n = dispatch_envelope(&hub, new_envelope(user));
278 assert_eq!(n, 3);
279 }
280
281 #[test]
282 fn dispatch_envelope_unknown_user_returns_zero() {
283 let hub = ConnectionHub::new(8);
284 assert_eq!(dispatch_envelope(&hub, new_envelope(Uuid::now_v7())), 0);
285 }
286
287 fn new_live_envelope(user_id: Uuid) -> EventEnvelope<ChatLiveViewEvent> {
288 EventEnvelope::new(
289 user_id,
290 ChatLiveViewEvent::ParticipantJoinedView {
291 user_id,
292 channel_id: Uuid::now_v7(),
293 room_name: "room-1".to_string(),
294 participant_id: Uuid::now_v7(),
295 ts: OffsetDateTime::now_utc(),
296 },
297 )
298 }
299
300 #[test]
301 fn dispatch_live_envelope_routes_by_recipient_user_id() {
302 let hub = ConnectionHub::new(8);
303 let alice = Uuid::now_v7();
304 let bob = Uuid::now_v7();
305
306 let (tx_a, mut rx_a) = mpsc::unbounded_channel();
307 hub.register(Connection {
308 id: Uuid::now_v7(),
309 user_id: alice,
310 tx: tx_a,
311 established_at: OffsetDateTime::now_utc(),
312 })
313 .unwrap();
314
315 let (tx_b, mut rx_b) = mpsc::unbounded_channel();
316 hub.register(Connection {
317 id: Uuid::now_v7(),
318 user_id: bob,
319 tx: tx_b,
320 established_at: OffsetDateTime::now_utc(),
321 })
322 .unwrap();
323
324 let n = dispatch_live_envelope(&hub, new_live_envelope(alice));
325 assert_eq!(n, 1);
326 let frame = rx_a.try_recv().unwrap();
327 match frame {
328 ServerFrame::Meet { event } => {
329 assert_eq!(event.payload.user_id(), alice);
330 }
331 other => panic!("expected Meet frame, got {other:?}"),
332 }
333 assert!(rx_b.try_recv().is_err(), "bob does not receive");
334 }
335}
Added chat/glayer/error.rs View chat/glayer/error.rs
@@ -0,0 +1,163 @@
1use axum::http::StatusCode;
2use axum::response::{IntoResponse, Response};
3use chatlayer::ChatLayerError;
4use export::data::{AppResponseData, NoData};
5use queue::QueueError;
6use thiserror::Error;
7
8#[derive(Debug, Error)]
9pub enum GatewayError {
10 #[error("bad request: {0}")]
11 BadRequest(String),
12
13 #[error("unauthorized: {0}")]
14 Unauthorized(String),
15
16 #[error("forbidden: {0}")]
17 Forbidden(String),
18
19 #[error("not found: {0}")]
20 NotFound(String),
21
22 #[error("conflict: {0}")]
23 Conflict(String),
24
25 #[error("{0}")]
26 Chat(#[from] ChatLayerError),
27
28 #[error("{0}")]
29 Queue(#[from] QueueError),
30
31 #[error("{0}")]
32 Json(#[from] serde_json::Error),
33
34 #[error("internal: {0}")]
35 Internal(String),
36}
37
38pub type GatewayResult<T> = Result<T, GatewayError>;
39
40impl GatewayError {
41 pub fn bad_request(msg: impl Into<String>) -> Self {
42 Self::BadRequest(msg.into())
43 }
44
45 pub fn not_found(msg: impl Into<String>) -> Self {
46 Self::NotFound(msg.into())
47 }
48
49 pub fn internal(msg: impl Into<String>) -> Self {
50 Self::Internal(msg.into())
51 }
52
53 fn status(&self) -> StatusCode {
54 match self {
55 GatewayError::BadRequest(_) => StatusCode::BAD_REQUEST,
56 GatewayError::Unauthorized(_) => StatusCode::UNAUTHORIZED,
57 GatewayError::Forbidden(_) => StatusCode::FORBIDDEN,
58 GatewayError::NotFound(_) => StatusCode::NOT_FOUND,
59 GatewayError::Conflict(_) => StatusCode::CONFLICT,
60 GatewayError::Chat(e) => match e {
61 ChatLayerError::InvalidInput(_) => StatusCode::BAD_REQUEST,
62 ChatLayerError::PermissionDenied(_) => StatusCode::FORBIDDEN,
63 ChatLayerError::NotFound(_) => StatusCode::NOT_FOUND,
64 ChatLayerError::Conflict(_) => StatusCode::CONFLICT,
65 ChatLayerError::Jwt(_) => StatusCode::UNAUTHORIZED,
66 ChatLayerError::Storage(_)
67 | ChatLayerError::Queue(_)
68 | ChatLayerError::Redis(_)
69 | ChatLayerError::Blob(_)
70 | ChatLayerError::Serde(_)
71 | ChatLayerError::Internal(_) => StatusCode::INTERNAL_SERVER_ERROR,
72 },
73 GatewayError::Queue(_) | GatewayError::Json(_) | GatewayError::Internal(_) => {
74 StatusCode::INTERNAL_SERVER_ERROR
75 }
76 }
77 }
78}
79
80impl IntoResponse for GatewayError {
81 fn into_response(self) -> Response {
82 let status = self.status();
83 let msg = if status.is_client_error() {
84 self.to_string()
85 } else {
86 tracing::error!(error = %self, "glayer internal error");
87 "internal error".to_string()
88 };
89 AppResponseData::<NoData>::err_plain(status.as_u16() as i32, msg).into_response()
90 }
91}
92
93#[cfg(test)]
94mod tests {
95 use super::*;
96
97 #[test]
98 fn chat_error_status_mapping() {
99 let cases = [
100 (
101 ChatLayerError::InvalidInput("x".into()),
102 StatusCode::BAD_REQUEST,
103 ),
104 (
105 ChatLayerError::PermissionDenied("x".into()),
106 StatusCode::FORBIDDEN,
107 ),
108 (ChatLayerError::NotFound("x".into()), StatusCode::NOT_FOUND),
109 (ChatLayerError::Conflict("x".into()), StatusCode::CONFLICT),
110 (
111 ChatLayerError::Jwt(jwt::JwtError::MissingHeader),
112 StatusCode::UNAUTHORIZED,
113 ),
114 (
115 ChatLayerError::Internal("boom".into()),
116 StatusCode::INTERNAL_SERVER_ERROR,
117 ),
118 ];
119 for (chat_err, expected) in cases {
120 assert_eq!(GatewayError::Chat(chat_err).status(), expected);
121 }
122 }
123
124 #[test]
125 fn gateway_error_status_mapping() {
126 assert_eq!(
127 GatewayError::BadRequest("x".into()).status(),
128 StatusCode::BAD_REQUEST
129 );
130 assert_eq!(
131 GatewayError::Unauthorized("x".into()).status(),
132 StatusCode::UNAUTHORIZED
133 );
134 assert_eq!(
135 GatewayError::Forbidden("x".into()).status(),
136 StatusCode::FORBIDDEN
137 );
138 assert_eq!(
139 GatewayError::NotFound("x".into()).status(),
140 StatusCode::NOT_FOUND
141 );
142 assert_eq!(
143 GatewayError::Conflict("x".into()).status(),
144 StatusCode::CONFLICT
145 );
146 assert_eq!(
147 GatewayError::Internal("x".into()).status(),
148 StatusCode::INTERNAL_SERVER_ERROR
149 );
150 }
151
152 #[test]
153 fn gateway_error_into_response_4xx_user_facing() {
154 let resp = GatewayError::NotFound("conversation missing".into()).into_response();
155 assert_eq!(resp.status(), StatusCode::NOT_FOUND);
156 }
157
158 #[test]
159 fn gateway_error_into_response_5xx_generic() {
160 let resp = GatewayError::Internal("secret detail".into()).into_response();
161 assert_eq!(resp.status(), StatusCode::INTERNAL_SERVER_ERROR);
162 }
163}
Added chat/glayer/hub.rs View chat/glayer/hub.rs
@@ -0,0 +1,265 @@
1use std::collections::HashMap;
2use std::sync::{Arc, Mutex};
3use time::OffsetDateTime;
4use tokio::sync::mpsc;
5use uuid::Uuid;
6
7use crate::protocol::ServerFrame;
8
9pub struct Connection {
10 pub id: Uuid,
11 pub user_id: Uuid,
12 pub tx: mpsc::UnboundedSender<ServerFrame>,
13 pub established_at: OffsetDateTime,
14}
15
16impl std::fmt::Debug for Connection {
17 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
18 f.debug_struct("Connection")
19 .field("id", &self.id)
20 .field("user_id", &self.user_id)
21 .field("established_at", &self.established_at)
22 .finish_non_exhaustive()
23 }
24}
25
26pub struct ConnectionHub {
27 inner: Mutex<HashMap<Uuid, HashMap<Uuid, Connection>>>,
28 max_per_user: usize,
29}
30
31impl std::fmt::Debug for ConnectionHub {
32 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
33 let (users, conns) = {
34 let g = self.inner.lock().unwrap_or_else(|p| p.into_inner());
35 (g.len(), g.values().map(|m| m.len()).sum::<usize>())
36 };
37 f.debug_struct("ConnectionHub")
38 .field("online_users", &users)
39 .field("live_connections", &conns)
40 .field("max_per_user", &self.max_per_user)
41 .finish()
42 }
43}
44
45#[derive(Debug, Clone, Copy, PartialEq, Eq)]
46pub struct HubError(pub usize);
47
48impl std::fmt::Display for HubError {
49 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
50 write!(f, "max connections per user exceeded ({})", self.0)
51 }
52}
53
54impl std::error::Error for HubError {}
55
56impl ConnectionHub {
57 pub fn new(max_per_user: usize) -> Arc<Self> {
58 Arc::new(Self {
59 inner: Mutex::new(HashMap::new()),
60 max_per_user,
61 })
62 }
63
64 pub fn register(&self, conn: Connection) -> Result<(), HubError> {
65 let mut g = self.inner.lock().unwrap_or_else(|p| p.into_inner());
66 let conns = g.entry(conn.user_id).or_default();
67 if conns.len() >= self.max_per_user {
68 return Err(HubError(self.max_per_user));
69 }
70 conns.insert(conn.id, conn);
71 Ok(())
72 }
73
74 pub fn unregister(&self, user_id: Uuid, conn_id: Uuid) -> bool {
75 let mut g = self.inner.lock().unwrap_or_else(|p| p.into_inner());
76 let removed = g
77 .get_mut(&user_id)
78 .map(|m| m.remove(&conn_id).is_some())
79 .unwrap_or(false);
80 if removed && g.get(&user_id).map(|m| m.is_empty()).unwrap_or(false) {
81 g.remove(&user_id);
82 }
83 removed
84 }
85
86 pub fn dispatch(&self, user_id: Uuid, frame: ServerFrame) -> usize {
87 let mut dead = Vec::new();
88 let mut delivered = 0usize;
89 {
90 let mut g = self.inner.lock().unwrap_or_else(|p| p.into_inner());
91 let Some(conns) = g.get_mut(&user_id) else {
92 return 0;
93 };
94 for (id, conn) in conns.iter() {
95 if conn.tx.send(frame.clone()).is_ok() {
96 delivered += 1;
97 } else {
98 dead.push(*id);
99 }
100 }
101 for id in dead {
102 conns.remove(&id);
103 }
104 if conns.is_empty() {
105 g.remove(&user_id);
106 }
107 }
108 delivered
109 }
110
111 pub fn connections_for(&self, user_id: Uuid) -> usize {
112 let g = self.inner.lock().unwrap_or_else(|p| p.into_inner());
113 g.get(&user_id).map(|m| m.len()).unwrap_or(0)
114 }
115
116 pub fn total_connections(&self) -> usize {
117 let g = self.inner.lock().unwrap_or_else(|p| p.into_inner());
118 g.values().map(|m| m.len()).sum()
119 }
120
121 pub fn online_users(&self) -> usize {
122 let g = self.inner.lock().unwrap_or_else(|p| p.into_inner());
123 g.len()
124 }
125}
126
127#[cfg(test)]
128mod tests {
129 use super::*;
130 use std::sync::atomic::{AtomicUsize, Ordering};
131
132 fn conn_pair(user_id: Uuid) -> (Connection, mpsc::UnboundedReceiver<ServerFrame>) {
133 let (tx, rx) = mpsc::unbounded_channel();
134 let conn = Connection {
135 id: Uuid::now_v7(),
136 user_id,
137 tx,
138 established_at: OffsetDateTime::now_utc(),
139 };
140 (conn, rx)
141 }
142
143 fn frame() -> ServerFrame {
144 ServerFrame::AuthOk {
145 user_id: Uuid::now_v7(),
146 }
147 }
148
149 #[test]
150 fn register_and_dispatch_reaches_connection() {
151 let hub = ConnectionHub::new(8);
152 let user = Uuid::now_v7();
153 let (conn, mut rx) = conn_pair(user);
154 hub.register(conn).unwrap();
155 assert_eq!(hub.connections_for(user), 1);
156
157 let delivered = hub.dispatch(user, frame());
158 assert_eq!(delivered, 1);
159 assert!(matches!(rx.try_recv(), Ok(ServerFrame::AuthOk { .. })));
160 }
161
162 #[test]
163 fn dispatch_fans_out_to_all_connections_same_user() {
164 let hub = ConnectionHub::new(8);
165 let user = Uuid::now_v7();
166 let (c1, mut rx1) = conn_pair(user);
167 let (c2, mut rx2) = conn_pair(user);
168 hub.register(c1).unwrap();
169 hub.register(c2).unwrap();
170
171 let delivered = hub.dispatch(user, frame());
172 assert_eq!(delivered, 2);
173 assert!(rx1.try_recv().is_ok());
174 assert!(rx2.try_recv().is_ok());
175 }
176
177 #[test]
178 fn dispatch_unknown_user_returns_zero() {
179 let hub = ConnectionHub::new(8);
180 assert_eq!(hub.dispatch(Uuid::now_v7(), frame()), 0);
181 }
182
183 #[test]
184 fn unregister_removes_only_target_connection() {
185 let hub = ConnectionHub::new(8);
186 let user = Uuid::now_v7();
187 let (c1, _rx1) = conn_pair(user);
188 let (c2, _rx2) = conn_pair(user);
189 let to_remove = c2.id;
190 hub.register(c1).unwrap();
191 hub.register(c2).unwrap();
192
193 assert!(hub.unregister(user, to_remove));
194 assert_eq!(hub.connections_for(user), 1);
195 assert!(!hub.unregister(user, to_remove), "unregister is idempotent");
196 }
197
198 #[test]
199 fn unregister_last_connection_drops_user_entry() {
200 let hub = ConnectionHub::new(8);
201 let user = Uuid::now_v7();
202 let (conn, _rx) = conn_pair(user);
203 let conn_id = conn.id;
204 hub.register(conn).unwrap();
205 assert_eq!(hub.online_users(), 1);
206
207 assert!(hub.unregister(user, conn_id));
208 assert_eq!(hub.online_users(), 0);
209 assert_eq!(hub.total_connections(), 0);
210 }
211
212 #[test]
213 fn register_rejects_over_max_connections_per_user() {
214 let hub = ConnectionHub::new(2);
215 let user = Uuid::now_v7();
216 let (c1, _) = conn_pair(user);
217 let (c2, _) = conn_pair(user);
218 let (c3, _) = conn_pair(user);
219 hub.register(c1).unwrap();
220 hub.register(c2).unwrap();
221 assert_eq!(hub.register(c3), Err(HubError(2)));
222 }
223
224 #[test]
225 fn closed_sender_reaped_on_next_dispatch() {
226 let hub = ConnectionHub::new(8);
227 let user = Uuid::now_v7();
228 let (c1, _rx_live) = conn_pair(user);
229 let (c2, rx_dead) = conn_pair(user);
230 hub.register(c1).unwrap();
231 hub.register(c2).unwrap();
232 drop(rx_dead);
233
234 let delivered = hub.dispatch(user, frame());
235 assert_eq!(delivered, 1, "dead connection not delivered");
236 assert_eq!(hub.connections_for(user), 1, "dead connection reaped");
237 }
238
239 #[tokio::test]
240 async fn concurrent_register_dispatch_unregister_race_free() {
241 let hub = ConnectionHub::new(64);
242 let user = Uuid::now_v7();
243 let spawned = Arc::new(AtomicUsize::new(0));
244 let mut handles = Vec::new();
245
246 for _ in 0..50 {
247 let hub = hub.clone();
248 let spawned = spawned.clone();
249 handles.push(tokio::spawn(async move {
250 let (conn, _rx) = conn_pair(user);
251 let conn_id = conn.id;
252 hub.register(conn).unwrap();
253 spawned.fetch_add(1, Ordering::SeqCst);
254 let _ = hub.dispatch(user, frame());
255 hub.unregister(user, conn_id);
256 }));
257 }
258 for h in handles {
259 h.await.unwrap();
260 }
261 assert_eq!(spawned.load(Ordering::SeqCst), 50);
262 assert_eq!(hub.online_users(), 0, "all connections cleaned up");
263 assert_eq!(hub.total_connections(), 0);
264 }
265}
Added chat/glayer/lib.rs View chat/glayer/lib.rs
@@ -0,0 +1,88 @@
1#![warn(missing_debug_implementations)]
2#![warn(unreachable_pub)]
3
4pub mod auth;
5pub mod consumer;
6pub mod error;
7pub mod handler;
8pub mod hub;
9pub mod metrics;
10pub mod protocol;
11pub mod router;
12pub mod state;
13pub mod typing;
14pub mod ws;
15
16pub use auth::BearToken;
17pub use consumer::{TypingConsumer, UserViewConsumer, UserViewConsumerSettings, dispatch_envelope};
18pub use error::{GatewayError, GatewayResult};
19pub use hub::{Connection, ConnectionHub, HubError};
20pub use protocol::{ClientFrame, ServerFrame, parse_client_frame};
21pub use router::{CHAT_API_PREFIX, router};
22pub use state::GatewayState;
23pub use typing::{TypingDebouncer, TypingSignal, typing_topic};
24
25#[cfg(test)]
26pub(crate) mod testutil {
27 use std::sync::Arc;
28
29 use chatlayer::AuthBridge;
30 use config::jwt::JwtSettings;
31 use ed25519_dalek::SigningKey;
32 use ed25519_dalek::pkcs8::EncodePrivateKey;
33 use jwt::JwtState;
34 use jwt::claims::AccessClaims;
35 use jwt::keys::{CachedKey, SigningKeyCache};
36 use jwt::signer::encode_access_token;
37 use rand_core::OsRng;
38 use uuid::Uuid;
39
40 pub(crate) struct TestAuth {
41 pub(crate) bridge: AuthBridge,
42 enc: jsonwebtoken::EncodingKey,
43 kid: Uuid,
44 }
45
46 impl TestAuth {
47 pub(crate) async fn new() -> Self {
48 let sk = SigningKey::generate(&mut OsRng);
49 let vk = sk.verifying_key();
50 let enc = jsonwebtoken::EncodingKey::from_ed_der(sk.to_pkcs8_der().unwrap().as_bytes());
51 let dec = jsonwebtoken::DecodingKey::from_ed_der(&vk.to_bytes());
52 let kid = Uuid::now_v7();
53 let cache = Arc::new(tokio::sync::RwLock::new(SigningKeyCache::new()));
54 cache.write().await.insert(
55 kid,
56 CachedKey {
57 decoding: dec,
58 algorithm: jsonwebtoken::Algorithm::EdDSA,
59 is_rotating: false,
60 grace_ends_at: None,
61 expires_at: None,
62 },
63 );
64 let db = sea_orm::MockDatabase::new(sea_orm::DbBackend::Sqlite).into_connection();
65 let jwt = JwtState::new(db, JwtSettings::default(), cache);
66 Self {
67 bridge: AuthBridge::new(Arc::new(jwt)),
68 enc,
69 kid,
70 }
71 }
72
73 pub(crate) fn mint(&self, user_id: Uuid, ttl: i64) -> String {
74 let claims = AccessClaims::new(
75 user_id,
76 "alice".into(),
77 false,
78 false,
79 1,
80 ttl,
81 "untitled",
82 "untitled-api",
83 &self.kid.to_string(),
84 );
85 encode_access_token(&claims, &self.enc, jsonwebtoken::Algorithm::EdDSA).unwrap()
86 }
87 }
88}
Added chat/glayer/metrics.rs View chat/glayer/metrics.rs
@@ -0,0 +1,33 @@
1use metrics::counter;
2
3pub fn ws_connected(result: &'static str) {
4 counter!("chat.ws_connections_total", "result" => result).increment(1);
5}
6
7pub fn ws_disconnected(result: &'static str) {
8 counter!("chat.ws_disconnections_total", "result" => result).increment(1);
9}
10
11pub fn ws_frame_sent() {
12 counter!("chat.ws_frames_sent_total").increment(1);
13}
14
15pub fn user_view_delivered(count: usize) {
16 counter!("chat.user_view_delivered_total").increment(count as u64);
17}
18
19pub fn user_view_dropped() {
20 counter!("chat.user_view_dropped_total").increment(1);
21}
22
23pub fn live_view_delivered(count: usize) {
24 counter!("chat.live_view_delivered_total").increment(count as u64);
25}
26
27pub fn live_view_dropped() {
28 counter!("chat.live_view_dropped_total").increment(1);
29}
30
31pub fn hub_register_rejected() {
32 counter!("chat.hub_register_rejected_total").increment(1);
33}
Added chat/glayer/protocol.rs View chat/glayer/protocol.rs
@@ -0,0 +1,199 @@
1use chatlayer::{ChatLiveViewEvent, ChatUserViewEvent, EventEnvelope};
2use serde::{Deserialize, Serialize};
3use time::OffsetDateTime;
4use uuid::Uuid;
5
6#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
7#[serde(tag = "type", rename_all = "snake_case")]
8pub enum ServerFrame {
9 Welcome {
10 server_time: OffsetDateTime,
11 },
12 AuthOk {
13 user_id: Uuid,
14 },
15 AuthError {
16 reason: String,
17 },
18 Event {
19 event: EventEnvelope<ChatUserViewEvent>,
20 },
21 Meet {
22 event: EventEnvelope<ChatLiveViewEvent>,
23 },
24 Typing {
25 user_id: Uuid,
26 conversation_id: Uuid,
27 ts: OffsetDateTime,
28 },
29 Error {
30 reason: String,
31 },
32}
33
34#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
35#[serde(tag = "type", rename_all = "snake_case")]
36pub enum ClientFrame {
37 Auth {
38 token: String,
39 device_id: Option<String>,
40 },
41 Typing {
42 conversation_id: Uuid,
43 },
44}
45
46impl ServerFrame {
47 pub fn as_ws_message(&self) -> axum::extract::ws::Message {
48 let json = serde_json::to_string(self).expect("ServerFrame always serializes");
49 axum::extract::ws::Message::Text(json.into())
50 }
51}
52
53pub fn parse_client_frame(text: &str) -> Result<ClientFrame, serde_json::Error> {
54 serde_json::from_str(text)
55}
56
57#[cfg(test)]
58mod tests {
59 use super::*;
60
61 fn sample_envelope(user_id: Uuid) -> EventEnvelope<ChatUserViewEvent> {
62 EventEnvelope::new(
63 user_id,
64 ChatUserViewEvent::NewMessage {
65 user_id,
66 conversation_id: Uuid::now_v7(),
67 message_id: Uuid::now_v7(),
68 sender_id: Uuid::now_v7(),
69 body: "hi".to_string(),
70 ts: OffsetDateTime::now_utc(),
71 },
72 )
73 }
74
75 #[test]
76 fn server_frame_welcome_roundtrip() {
77 let frame = ServerFrame::Welcome {
78 server_time: OffsetDateTime::now_utc(),
79 };
80 let json = serde_json::to_string(&frame).unwrap();
81 let back: ServerFrame = serde_json::from_str(&json).unwrap();
82 assert_eq!(back, frame);
83 assert!(json.contains("\"type\":\"welcome\""));
84 }
85
86 #[test]
87 fn server_frame_auth_ok_roundtrip() {
88 let frame = ServerFrame::AuthOk {
89 user_id: Uuid::now_v7(),
90 };
91 let json = serde_json::to_string(&frame).unwrap();
92 let back: ServerFrame = serde_json::from_str(&json).unwrap();
93 assert_eq!(back, frame);
94 }
95
96 #[test]
97 fn server_frame_event_encloses_view_event() {
98 let user_id = Uuid::now_v7();
99 let frame = ServerFrame::Event {
100 event: sample_envelope(user_id),
101 };
102 let json = serde_json::to_string(&frame).unwrap();
103 let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();
104 assert_eq!(parsed["type"], "event");
105 assert_eq!(parsed["event"]["payload"]["type"], "new_message");
106 assert_eq!(parsed["event"]["payload"]["user_id"], user_id.to_string());
107 let back: ServerFrame = serde_json::from_str(&json).unwrap();
108 assert_eq!(back, frame);
109 }
110
111 #[test]
112 fn client_frame_auth_roundtrip() {
113 let frame = ClientFrame::Auth {
114 token: "eyJhbGciOiJFZERTQSJ9.abc.def".to_string(),
115 device_id: Some("ios-7f3a".to_string()),
116 };
117 let json = serde_json::to_string(&frame).unwrap();
118 let back: ClientFrame = serde_json::from_str(&json).unwrap();
119 assert_eq!(back, frame);
120 assert!(json.contains("\"type\":\"auth\""));
121 }
122
123 #[test]
124 fn client_frame_typing_roundtrip() {
125 let frame = ClientFrame::Typing {
126 conversation_id: Uuid::now_v7(),
127 };
128 let json = serde_json::to_string(&frame).unwrap();
129 let back: ClientFrame = serde_json::from_str(&json).unwrap();
130 assert_eq!(back, frame);
131 assert!(json.contains("\"type\":\"typing\""));
132 }
133
134 #[test]
135 fn server_frame_meet_encloses_live_view_event() {
136 let user_id = Uuid::now_v7();
137 let frame = ServerFrame::Meet {
138 event: EventEnvelope::new(
139 user_id,
140 ChatLiveViewEvent::ParticipantJoinedView {
141 user_id,
142 channel_id: Uuid::now_v7(),
143 room_name: "room-1".to_string(),
144 participant_id: Uuid::now_v7(),
145 ts: OffsetDateTime::now_utc(),
146 },
147 ),
148 };
149 let json = serde_json::to_string(&frame).unwrap();
150 let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();
151 assert_eq!(parsed["type"], "meet");
152 assert_eq!(
153 parsed["event"]["payload"]["type"],
154 "participant_joined_view"
155 );
156 let back: ServerFrame = serde_json::from_str(&json).unwrap();
157 assert_eq!(back, frame);
158 }
159
160 #[test]
161 fn server_frame_typing_roundtrip() {
162 let frame = ServerFrame::Typing {
163 user_id: Uuid::now_v7(),
164 conversation_id: Uuid::now_v7(),
165 ts: OffsetDateTime::now_utc(),
166 };
167 let json = serde_json::to_string(&frame).unwrap();
168 let back: ServerFrame = serde_json::from_str(&json).unwrap();
169 assert_eq!(back, frame);
170 assert!(json.contains("\"type\":\"typing\""));
171 }
172
173 #[test]
174 fn parse_client_frame_rejects_unknown_type() {
175 let err = parse_client_frame(r#"{"type":"bogus","token":"x"}"#).unwrap_err();
176 assert!(err.is_data());
177 }
178
179 #[test]
180 fn parse_client_frame_rejects_missing_token() {
181 let err = parse_client_frame(r#"{"type":"auth"}"#).unwrap_err();
182 assert!(err.is_data());
183 }
184
185 #[test]
186 fn server_frame_as_ws_message_is_text_with_json() {
187 let frame = ServerFrame::AuthOk {
188 user_id: Uuid::now_v7(),
189 };
190 let msg = frame.as_ws_message();
191 match msg {
192 axum::extract::ws::Message::Text(txt) => {
193 let v: serde_json::Value = serde_json::from_str(&txt).unwrap();
194 assert_eq!(v["type"], "auth_ok");
195 }
196 other => panic!("expected Text message, got {other:?}"),
197 }
198 }
199}
Added chat/glayer/router.rs View chat/glayer/router.rs
@@ -0,0 +1,74 @@
1use axum::Router;
2use axum::routing::{delete, get, patch, post, put};
3
4use crate::handler;
5use crate::state::GatewayState;
6use crate::ws::ws_handler;
7
8pub const CHAT_API_PREFIX: &str = "/api/v1/chat";
9
10pub fn router(state: GatewayState) -> Router {
11 Router::<GatewayState>::new()
12 .nest(CHAT_API_PREFIX, chat_routes())
13 .with_state(state)
14}
15
16fn chat_routes() -> Router<GatewayState> {
17 Router::new()
18 .route("/ws", get(ws_handler))
19 .route(
20 "/conversations/{conversation_id}/messages",
21 post(handler::messages::send_message),
22 )
23 .route(
24 "/conversations/{conversation_id}/messages",
25 get(handler::messages::list_messages),
26 )
27 .route(
28 "/messages/{message_id}",
29 patch(handler::messages::edit_message),
30 )
31 .route(
32 "/messages/{message_id}",
33 delete(handler::messages::delete_message),
34 )
35 .route(
36 "/messages/{message_id}/reactions/{emoji}",
37 put(handler::reactions::add_reaction),
38 )
39 .route(
40 "/messages/{message_id}/reactions/{emoji}",
41 delete(handler::reactions::remove_reaction),
42 )
43 .route(
44 "/messages/{message_id}/mentions",
45 post(handler::mentions::add_mention),
46 )
47 .route("/messages/{message_id}/pin", put(handler::pin::set_pin))
48 .route(
49 "/messages/{message_id}/pin",
50 delete(handler::pin::unset_pin),
51 )
52 .route(
53 "/conversations/{conversation_id}/read",
54 post(handler::read::mark_read),
55 )
56 .route(
57 "/conversations/{conversation_id}",
58 get(handler::conversations::get_conversation),
59 )
60 .route(
61 "/messages/by-client-msg-id",
62 get(handler::lookups::lookup_by_client_msg_id),
63 )
64 .route("/uploads/presign", post(handler::uploads::presign_upload))
65 .route(
66 "/uploads/{attachment_id}/finalize",
67 post(handler::uploads::finalize_upload),
68 )
69 .route(
70 "/uploads/{attachment_id}",
71 delete(handler::uploads::abort_upload),
72 )
73 .route("/presence/heartbeat", post(handler::presence::heartbeat))
74}
Added chat/glayer/state.rs View chat/glayer/state.rs
@@ -0,0 +1,108 @@
1use std::sync::Arc;
2use std::time::Duration;
3
4use chatlayer::ChatService;
5use queue::SubscribeHandle;
6use tokio_util::sync::CancellationToken;
7
8use crate::consumer::{
9 MeetViewConsumer, TypingConsumer, UserViewConsumer, UserViewConsumerSettings,
10};
11use crate::error::{GatewayError, GatewayResult};
12use crate::hub::ConnectionHub;
13use crate::typing::{TypingDebouncer, TypingSignal, typing_topic};
14use config::chat::GatewaySettings;
15
16#[derive(Clone)]
17pub struct GatewayState {
18 pub chat: Arc<ChatService>,
19 pub hub: Arc<ConnectionHub>,
20 pub settings: GatewaySettings,
21 pub(crate) typing: Arc<TypingDebouncer>,
22 shutdown: CancellationToken,
23}
24
25impl std::fmt::Debug for GatewayState {
26 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
27 f.debug_struct("GatewayState")
28 .field("hub", &self.hub)
29 .field("settings", &self.settings)
30 .finish_non_exhaustive()
31 }
32}
33
34impl GatewayState {
35 pub fn new(chat: Arc<ChatService>, hub: Arc<ConnectionHub>, settings: GatewaySettings) -> Self {
36 let typing = Arc::new(TypingDebouncer::new(Duration::from_secs(
37 settings.typing_debounce_seconds,
38 )));
39 Self {
40 chat,
41 hub,
42 settings,
43 typing,
44 shutdown: CancellationToken::new(),
45 }
46 }
47
48 pub fn shutdown_token(&self) -> CancellationToken {
49 self.shutdown.clone()
50 }
51
52 pub fn spawn_user_view_consumer(&self) -> GatewayResult<SubscribeHandle> {
53 let queue = self.chat.queue().ok_or_else(|| {
54 GatewayError::Chat(chatlayer::ChatLayerError::invalid(
55 "chat queue disabled; cannot start user-view consumer",
56 ))
57 })?;
58 let settings = UserViewConsumerSettings {
59 topic: chatlayer::topics::user_view_topic(&self.settings.topic_prefix),
60 subscriber_id: self.settings.subscriber_id.clone(),
61 };
62 UserViewConsumer::new(queue, self.hub.clone(), settings)
63 .run()
64 .map_err(GatewayError::from)
65 }
66
67 pub fn spawn_typing_consumer(&self) -> GatewayResult<SubscribeHandle> {
68 let queue = self.chat.queue().ok_or_else(|| {
69 GatewayError::Chat(chatlayer::ChatLayerError::invalid(
70 "chat queue disabled; cannot start typing consumer",
71 ))
72 })?;
73 let settings = UserViewConsumerSettings {
74 topic: typing_topic(&self.settings.topic_prefix),
75 subscriber_id: self.settings.subscriber_id.clone(),
76 };
77 TypingConsumer::new(queue, Arc::new(self.clone()), settings)
78 .run()
79 .map_err(GatewayError::from)
80 }
81
82 pub fn spawn_live_view_consumer(&self) -> GatewayResult<SubscribeHandle> {
83 let queue = self.chat.queue().ok_or_else(|| {
84 GatewayError::Chat(chatlayer::ChatLayerError::invalid(
85 "chat queue disabled; cannot start live-view consumer",
86 ))
87 })?;
88 let settings = UserViewConsumerSettings {
89 topic: chatlayer::topics::live_view_topic(&self.settings.topic_prefix),
90 subscriber_id: self.settings.subscriber_id.clone(),
91 };
92 MeetViewConsumer::new(queue, self.hub.clone(), settings)
93 .run()
94 .map_err(GatewayError::from)
95 }
96
97 pub async fn publish_typing(&self, signal: &TypingSignal) -> GatewayResult<()> {
98 let Some(queue) = self.chat.queue() else {
99 return Ok(());
100 };
101 let topic = typing_topic(&self.settings.topic_prefix);
102 let payload = serde_json::to_vec(signal)?;
103 queue
104 .publish_raw(&topic, Some(signal.conversation_id.as_bytes()), &payload)
105 .await?;
106 Ok(())
107 }
108}
Added chat/glayer/typing.rs View chat/glayer/typing.rs
@@ -0,0 +1,172 @@
1use std::collections::HashMap;
2use std::sync::Mutex;
3use std::time::{Duration, Instant};
4
5use serde::{Deserialize, Serialize};
6use storagelayer::DynChatStorage;
7use storagelayer::repository::ChatStorageCtx;
8use time::OffsetDateTime;
9use uuid::Uuid;
10
11use crate::hub::ConnectionHub;
12use crate::protocol::ServerFrame;
13
14pub const TYPING_SUFFIX: &str = "typing.v1";
15
16pub fn typing_topic(prefix: &str) -> String {
17 format!("{prefix}.{TYPING_SUFFIX}")
18}
19
20#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
21pub struct TypingSignal {
22 pub conversation_id: Uuid,
23 pub user_id: Uuid,
24 pub ts: OffsetDateTime,
25}
26
27pub struct TypingDebouncer {
28 last: Mutex<HashMap<(Uuid, Uuid), Instant>>,
29 interval: Duration,
30}
31
32impl std::fmt::Debug for TypingDebouncer {
33 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
34 f.debug_struct("TypingDebouncer")
35 .field("interval", &self.interval)
36 .finish_non_exhaustive()
37 }
38}
39
40impl TypingDebouncer {
41 pub fn new(interval: Duration) -> Self {
42 Self {
43 last: Mutex::new(HashMap::new()),
44 interval,
45 }
46 }
47
48 pub fn allow(&self, conversation_id: Uuid, user_id: Uuid) -> bool {
49 let now = Instant::now();
50 let mut map = self.last.lock().unwrap_or_else(|p| p.into_inner());
51 match map.get(&(conversation_id, user_id)) {
52 Some(last) if now.duration_since(*last) < self.interval => false,
53 _ => {
54 map.insert((conversation_id, user_id), now);
55 true
56 }
57 }
58 }
59}
60
61pub async fn broadcast_typing(store: &DynChatStorage, hub: &ConnectionHub, signal: &TypingSignal) {
62 let ctx = ChatStorageCtx::new(signal.user_id);
63 let members = match store
64 .conversations()
65 .list_member_user_ids(&ctx, signal.conversation_id)
66 .await
67 {
68 Ok(ms) => ms,
69 Err(e) => {
70 tracing::warn!(
71 conversation_id = %signal.conversation_id,
72 error = %e,
73 "typing broadcast skipped: member lookup failed"
74 );
75 return;
76 }
77 };
78 let frame = ServerFrame::Typing {
79 user_id: signal.user_id,
80 conversation_id: signal.conversation_id,
81 ts: signal.ts,
82 };
83 for member in members {
84 if member != signal.user_id {
85 hub.dispatch(member, frame.clone());
86 }
87 }
88}
89
90#[cfg(test)]
91mod tests {
92 use super::*;
93 use std::time::Duration;
94 use tokio::sync::mpsc;
95 use uuid::Uuid;
96
97 use crate::hub::{Connection, ConnectionHub};
98 use storagelayer::ChatStorageFactory;
99 use storagelayer::models::conversation::{ConversationKind, NewConversation};
100
101 #[test]
102 fn typing_debouncer_allows_first_then_denies_within_interval() {
103 let d = TypingDebouncer::new(Duration::from_secs(60));
104 let cid = Uuid::now_v7();
105 let uid = Uuid::now_v7();
106 assert!(d.allow(cid, uid), "first signal passes");
107 assert!(!d.allow(cid, uid), "second within interval debounced");
108 assert!(d.allow(cid, Uuid::now_v7()), "different user not debounced");
109 }
110
111 #[test]
112 fn typing_topic_naming() {
113 assert_eq!(typing_topic("chat"), "chat.typing.v1");
114 assert_eq!(typing_topic("chat-dev"), "chat-dev.typing.v1");
115 }
116
117 #[tokio::test]
118 async fn broadcast_typing_excludes_typer_and_reaches_members() {
119 let store = ChatStorageFactory::in_memory();
120 let hub = ConnectionHub::new(8);
121 let alice = Uuid::now_v7();
122 let bob = Uuid::now_v7();
123
124 let conv = store
125 .conversations()
126 .create(
127 &ChatStorageCtx::new(alice),
128 NewConversation {
129 space_id: Uuid::now_v7(),
130 parent_dir: None,
131 kind: ConversationKind::Private,
132 counterpart_user_id: Some(bob),
133 title: None,
134 },
135 )
136 .await
137 .unwrap();
138
139 let (tx_a, mut rx_a) = mpsc::unbounded_channel();
140 hub.register(Connection {
141 id: Uuid::now_v7(),
142 user_id: alice,
143 tx: tx_a,
144 established_at: OffsetDateTime::now_utc(),
145 })
146 .unwrap();
147 let (tx_b, mut rx_b) = mpsc::unbounded_channel();
148 hub.register(Connection {
149 id: Uuid::now_v7(),
150 user_id: bob,
151 tx: tx_b,
152 established_at: OffsetDateTime::now_utc(),
153 })
154 .unwrap();
155
156 let signal = TypingSignal {
157 conversation_id: conv.id,
158 user_id: alice,
159 ts: OffsetDateTime::now_utc(),
160 };
161 broadcast_typing(&store, &hub, &signal).await;
162
163 assert!(
164 rx_b.try_recv().is_ok(),
165 "counterpart receives typing signal"
166 );
167 assert!(
168 rx_a.try_recv().is_err(),
169 "typer does not receive own signal"
170 );
171 }
172}
Added chat/glayer/ws.rs View chat/glayer/ws.rs
@@ -0,0 +1,247 @@
1use std::pin::Pin;
2use std::time::Duration;
3
4use axum::extract::State;
5use axum::extract::ws::{CloseFrame, Message, WebSocket, WebSocketUpgrade};
6use axum::response::Response;
7use bytes::Bytes;
8use chatlayer::AuthIdentity;
9use tokio::sync::mpsc;
10use tokio::time::Sleep;
11use uuid::Uuid;
12
13use crate::error::{GatewayError, GatewayResult};
14use crate::hub::Connection;
15use crate::protocol::{ClientFrame, ServerFrame, parse_client_frame};
16use crate::state::GatewayState;
17use crate::typing::{TypingSignal, broadcast_typing};
18
19const WS_CLOSE_AUTH_FAILED: u16 = 4001;
20const WS_CLOSE_TOO_MANY_CONNECTIONS: u16 = 4002;
21const WS_CLOSE_AUTH_TIMEOUT: u16 = 4003;
22
23pub async fn ws_handler(ws: WebSocketUpgrade, State(state): State<GatewayState>) -> Response {
24 ws.max_message_size(64 * 1024)
25 .max_write_buffer_size(1024 * 1024)
26 .on_upgrade(move |socket| handle_ws(socket, state))
27}
28
29async fn handle_ws(mut ws: WebSocket, state: GatewayState) {
30 if ws
31 .send(
32 ServerFrame::Welcome {
33 server_time: time::OffsetDateTime::now_utc(),
34 }
35 .as_ws_message(),
36 )
37 .await
38 .is_err()
39 {
40 return;
41 }
42
43 let identity = match tokio::time::timeout(
44 Duration::from_secs(state.settings.ws_auth_timeout_seconds),
45 wait_for_auth(&mut ws, &state),
46 )
47 .await
48 {
49 Ok(Ok(Some(id))) => id,
50 Ok(Ok(None)) => return,
51 Ok(Err(_)) => {
52 close(&mut ws, WS_CLOSE_AUTH_FAILED, "auth failed").await;
53 return;
54 }
55 Err(_) => {
56 close(&mut ws, WS_CLOSE_AUTH_TIMEOUT, "auth timeout").await;
57 return;
58 }
59 };
60
61 let (tx, rx) = mpsc::unbounded_channel::<ServerFrame>();
62 let user_id = identity.user_id;
63 let conn_id = Uuid::now_v7();
64 let conn = Connection {
65 id: conn_id,
66 user_id,
67 tx,
68 established_at: time::OffsetDateTime::now_utc(),
69 };
70 if let Err(_e) = state.hub.register(conn) {
71 crate::metrics::hub_register_rejected();
72 let _ = ws
73 .send(
74 ServerFrame::AuthError {
75 reason: "too many connections".into(),
76 }
77 .as_ws_message(),
78 )
79 .await;
80 close(
81 &mut ws,
82 WS_CLOSE_TOO_MANY_CONNECTIONS,
83 "too many connections",
84 )
85 .await;
86 return;
87 }
88 let _ = ws
89 .send(ServerFrame::AuthOk { user_id }.as_ws_message())
90 .await;
91 crate::metrics::ws_connected("ok");
92
93 let mut rx = rx;
94 pump(&mut ws, &mut rx, &state, user_id, conn_id).await;
95
96 state.hub.unregister(user_id, conn_id);
97 crate::metrics::ws_disconnected("ok");
98}
99
100async fn close(ws: &mut WebSocket, code: u16, reason: &str) {
101 let _ = ws
102 .send(Message::Close(Some(CloseFrame {
103 code,
104 reason: reason.into(),
105 })))
106 .await;
107}
108
109async fn wait_for_auth(
110 ws: &mut WebSocket,
111 state: &GatewayState,
112) -> GatewayResult<Option<AuthIdentity>> {
113 loop {
114 match ws.recv().await {
115 Some(Ok(Message::Text(txt))) => {
116 return verify_auth_frame(state.chat.auth(), &txt).await.map(Some);
117 }
118 Some(Ok(Message::Close(_))) | None => return Ok(None),
119 Some(Ok(_)) => continue,
120 Some(Err(_)) => return Ok(None),
121 }
122 }
123}
124
125pub(crate) async fn verify_auth_frame(
126 auth: &chatlayer::AuthBridge,
127 text: &str,
128) -> GatewayResult<AuthIdentity> {
129 match parse_client_frame(text)? {
130 ClientFrame::Auth { token, .. } => auth
131 .require_identity(&token)
132 .await
133 .map_err(GatewayError::from),
134 ClientFrame::Typing { .. } => Err(GatewayError::bad_request(
135 "expected auth frame during handshake",
136 )),
137 }
138}
139
140async fn handle_typing(state: &GatewayState, user_id: Uuid, conversation_id: Uuid) {
141 if !state.typing.allow(conversation_id, user_id) {
142 return;
143 }
144 let signal = TypingSignal {
145 conversation_id,
146 user_id,
147 ts: time::OffsetDateTime::now_utc(),
148 };
149 let store = state.chat.storage();
150 broadcast_typing(&store, &state.hub, &signal).await;
151 if let Err(e) = state.publish_typing(&signal).await {
152 tracing::warn!(error = %e, "failed to publish typing signal");
153 }
154}
155
156async fn pump(
157 ws: &mut WebSocket,
158 rx: &mut mpsc::UnboundedReceiver<ServerFrame>,
159 state: &GatewayState,
160 user_id: Uuid,
161 _conn_id: Uuid,
162) {
163 let heartbeat = Duration::from_secs(state.settings.ws_heartbeat_interval_seconds);
164 let idle_timeout = Duration::from_secs(state.settings.ws_idle_timeout_seconds);
165 let mut ping = tokio::time::interval(heartbeat);
166 ping.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
167 let mut idle: Pin<Box<Sleep>> = Box::pin(tokio::time::sleep(idle_timeout));
168 let shutdown = state.shutdown_token();
169 loop {
170 tokio::select! {
171 _ = shutdown.cancelled() => break,
172 _ = ping.tick() => {
173 if ws.send(Message::Ping(Bytes::new())).await.is_err() {
174 break;
175 }
176 }
177 _ = &mut idle => break,
178 frame = rx.recv() => {
179 match frame {
180 Some(f) => {
181 if ws.send(f.as_ws_message()).await.is_err() {
182 break;
183 }
184 crate::metrics::ws_frame_sent();
185 }
186 None => break,
187 }
188 }
189 msg = ws.recv() => {
190 match msg {
191 Some(Ok(Message::Pong(_))) => {
192 idle.as_mut()
193 .reset(tokio::time::Instant::now() + idle_timeout);
194 }
195 Some(Ok(Message::Text(txt))) => {
196 if let Ok(ClientFrame::Typing { conversation_id }) =
197 parse_client_frame(&txt)
198 {
199 handle_typing(state, user_id, conversation_id).await;
200 }
201 }
202 Some(Ok(Message::Close(_))) | None => break,
203 Some(Ok(_)) => {}
204 Some(Err(_)) => break,
205 }
206 }
207 }
208 }
209}
210
211#[cfg(test)]
212mod tests {
213 use super::*;
214 use crate::testutil;
215
216 #[tokio::test]
217 async fn verify_auth_frame_accepts_real_jwt() {
218 let ta = testutil::TestAuth::new().await;
219 let user_id = Uuid::now_v7();
220 let token = ta.mint(user_id, 900);
221 let frame = format!(r#"{{"type":"auth","token":"{token}"}}"#);
222 let identity = verify_auth_frame(&ta.bridge, &frame).await.unwrap();
223 assert_eq!(identity.user_id, user_id);
224 }
225
226 #[tokio::test]
227 async fn verify_auth_frame_rejects_garbage_token() {
228 let ta = testutil::TestAuth::new().await;
229 let frame = r#"{"type":"auth","token":"not-a-jwt"}"#;
230 assert!(verify_auth_frame(&ta.bridge, frame).await.is_err());
231 }
232
233 #[tokio::test]
234 async fn verify_auth_frame_rejects_missing_token_field() {
235 let ta = testutil::TestAuth::new().await;
236 let frame = r#"{"type":"auth"}"#;
237 assert!(verify_auth_frame(&ta.bridge, frame).await.is_err());
238 }
239
240 #[tokio::test]
241 async fn verify_auth_frame_rejects_expired_token() {
242 let ta = testutil::TestAuth::new().await;
243 let token = ta.mint(Uuid::now_v7(), -10000);
244 let frame = format!(r#"{{"type":"auth","token":"{token}"}}"#);
245 assert!(verify_auth_frame(&ta.bridge, &frame).await.is_err());
246 }
247}