ZhenYi/gitdataai
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: update channel HTTP layer (out_event, types, ws)

Browse Source
This commit is contained in:
zhenyi 2026-05-30 15:07:16 +08:00
parent 2b7777adbc
commit 22192301a0
3 changed files with 40 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
lib/channel/http/out_event.rs
View File

@ -118,6 +118,9 @@ pub enum WsOutEvent {
room: RoomInfo, room: RoomInfo,
data: thread::ThreadArchivedService, data: thread::ThreadArchivedService,
}, },
ThreadList {
data: thread::ThreadListService,
},
CategoryCreated { CategoryCreated {
workspace: WorkspaceInfo, workspace: WorkspaceInfo,
data: category::CategoryCreatedService, data: category::CategoryCreatedService,

5
lib/channel/http/types.rs
View File

@ -36,6 +36,7 @@ pub enum WsInMessage {
room: Uuid, room: Uuid,
seq: i64, seq: i64,
limit: Option<u64>, limit: Option<u64>,
thread: Option<Uuid>,
}, },
MessageCreate { MessageCreate {
room: Uuid, room: Uuid,
@ -120,6 +121,9 @@ pub enum WsInMessage {
room: Uuid, room: Uuid,
parent: i64, parent: i64,
}, },
ThreadList {
room: Uuid,
},
ThreadResolve { ThreadResolve {
thread_id: Uuid, thread_id: Uuid,
}, },
@ -298,6 +302,7 @@ impl WsInMessage {
ReactionAdd, ReactionAdd,
ReactionRemove, ReactionRemove,
ThreadCreate, ThreadCreate,
ThreadList,
PinAdd, PinAdd,
PinRemove, PinRemove,
DraftSave, DraftSave,

42
lib/channel/http/ws.rs
View File

@ -55,7 +55,7 @@ async fn handle_inbound(bus: &ChannelBus, socket: &Socket, data: EventPayload) {
let pong = WsOutEvent::Pong { let pong = WsOutEvent::Pong {
protocol_version: super::types::WS_PROTOCOL_VERSION, protocol_version: super::types::WS_PROTOCOL_VERSION,
}; };
send_event(socket, &pong).await.ok(); send_event(socket, &pong).await;
return; return;
} }
if !check_rate_limit(bus, user_id).await { if !check_rate_limit(bus, user_id).await {
@ -85,7 +85,7 @@ async fn handle_inbound(bus: &ChannelBus, socket: &Socket, data: EventPayload) {
request_id: rid, request_id: rid,
data: serde_json::to_value(&event).unwrap_or_default(), data: serde_json::to_value(&event).unwrap_or_default(),
}; };
send_event(socket, &resp).await.ok(); send_event(socket, &resp).await;
} }
Ok(None) => { Ok(None) => {
let rid = request_id.unwrap_or(Uuid::nil()); let rid = request_id.unwrap_or(Uuid::nil());
@ -93,7 +93,7 @@ async fn handle_inbound(bus: &ChannelBus, socket: &Socket, data: EventPayload) {
request_id: rid, request_id: rid,
data: serde_json::json!({"ok": true}), data: serde_json::json!({"ok": true}),
}; };
send_event(socket, &ack).await.ok(); send_event(socket, &ack).await;
} }
Err(e) => { Err(e) => {
tracing::warn!(user_id = %user_id, error = %e, "WS message processing failed"); tracing::warn!(user_id = %user_id, error = %e, "WS message processing failed");
@ -103,11 +103,22 @@ async fn handle_inbound(bus: &ChannelBus, socket: &Socket, data: EventPayload) {
data: serde_json::to_value(&e.to_ws_error()) data: serde_json::to_value(&e.to_ws_error())
.unwrap_or_default(), .unwrap_or_default(),
}; };
send_event(socket, &err_resp).await.ok(); send_event(socket, &err_resp).await;
} }
}, },
Err(e) => { Err(e) => {
tracing::warn!(error = %e, "WS transport parse error"); tracing::warn!(error = %e, "WS transport parse error");
if let Some(rid) = request_id {
let err_resp = WsOutEvent::Response {
request_id: rid,
data: serde_json::to_value(&WsError {
code: 400,
error: "parse_error".to_string(),
message: e.to_string(),
}).unwrap_or_default(),
};
send_event(socket, &err_resp).await;
} else {
send_error( send_error(
socket, socket,
WsError { WsError {
@ -115,8 +126,8 @@ async fn handle_inbound(bus: &ChannelBus, socket: &Socket, data: EventPayload) {
error: "parse_error".to_string(), error: "parse_error".to_string(),
message: e.to_string(), message: e.to_string(),
}, },
) ).await;
.await; }
} }
} }
} }
@ -129,25 +140,22 @@ async fn check_rate_limit(bus: &ChannelBus, user_id: Uuid) -> bool {
.unwrap_or(true) .unwrap_or(true)
} }
async fn send_event(socket: &Socket, event: &WsOutEvent) -> ChannelResult<()> { async fn send_event(socket: &Socket, event: &WsOutEvent) {
let json = serde_json::to_string(event)?; // Serialize to Value (not String) to avoid double JSON encoding
socket let value = serde_json::to_value(event).unwrap_or_default();
.emit(CHANNEL_EVENT, &json) if let Err(e) = socket.emit(CHANNEL_EVENT, value).await {
.await tracing::warn!(error = %e, "WS response send failed");
.map_err(|e| { }
tracing::warn!(error = %e, "WS send failed");
ChannelError::SocketIo(e)
})
} }
async fn send_error(socket: &Socket, error: WsError) { async fn send_error(socket: &Socket, error: WsError) {
let json = serde_json::json!({ let value = serde_json::json!({
"type": "error", "type": "error",
"code": error.code, "code": error.code,
"error": error.error, "error": error.error,
"message": error.message, "message": error.message,
}); });
if let Err(e) = socket.emit(CHANNEL_EVENT, json.to_string()).await { if let Err(e) = socket.emit(CHANNEL_EVENT, value).await {
tracing::warn!(error = %e, "WS error send failed"); tracing::warn!(error = %e, "WS error send failed");
} }
} }