db73d56
ZhenYi
refactor: add events dependency and enhance event handling for improved clarity and organization
db73d56
use export::AppState;
use models::chat::messages;
use models::room::rooms;
use sea_orm::EntityTrait;
use serde_json::{Value, json};
use uuid::Uuid;
use crate::errors::ChatError;
use crate::share::realtime::RoomEvent;
pub fn room_kind(event: &RoomEvent) -> &'static str {
match event {
RoomEvent::MessagePublished { .. } => "chat.message.create",
RoomEvent::MessageEdited { .. } => "chat.message.update",
RoomEvent::MessageDeleted { .. } => "chat.message.delete",
RoomEvent::MessagesDeleted { .. } => "chat.messages.delete",
RoomEvent::MessagePinned { pinned: true, .. } => "chat.pin.create",
RoomEvent::MessagePinned { pinned: false, .. } => "chat.pin.delete",
RoomEvent::ReactionUpdated { .. } => "chat.reaction.update",
RoomEvent::Typing { .. } => "chat.typing.create",
RoomEvent::RoomUpdated { .. } => "chat.room.update",
RoomEvent::MemberJoined { .. } => "chat.member.create",
RoomEvent::MemberLeft { .. } => "chat.member.delete",
RoomEvent::ThreadUpdated { .. } => "chat.thread.update",
RoomEvent::VoicePresence { .. } => "chat.voice.presence",
RoomEvent::VoiceHands { .. } => "chat.voice.hand",
RoomEvent::VoiceLayout { .. } => "chat.voice.layout",
RoomEvent::VoiceModeration { .. } => "chat.voice.moderate",
}
}
pub fn user_kind(event: &Value) -> Option<&'static str> {
match event.get("kind").and_then(Value::as_str) {
Some("mention") => Some("chat.mention.create"),
Some("reminder") => Some("chat.reminder.create"),
Some("room_added") => Some("chat.membership.create"),
Some("read_marker") => Some("chat.read_state.update"),
Some("presence") => Some("chat.presence.update"),
_ => None,
}
}
/// 房间事件指向的具体消息；`None` 表示事件只描述房间级的状态变化。
fn room_message_id(event: &RoomEvent) -> Option<Uuid> {
let from_value = |message: &Value| -> Option<Uuid> {
message
.get("id")
.and_then(Value::as_str)
.and_then(|id| Uuid::parse_str(id).ok())
};
match event {
RoomEvent::MessagePublished { message, .. } | RoomEvent::MessageEdited { message, .. } => {
from_value(message)
}
RoomEvent::MessageDeleted { message_id, .. }
| RoomEvent::MessagePinned { message_id, .. }
| RoomEvent::ReactionUpdated { message_id, .. } => Some(*message_id),
_ => None,
}
}
fn room_thread_id(event: &RoomEvent) -> Option<Uuid> {
match event {
RoomEvent::ThreadUpdated { thread_id, .. } => Some(*thread_id),
_ => None,
}
}
fn room_actor(event: &RoomEvent) -> Option<i64> {
match event {
RoomEvent::Typing { user_id, .. }
| RoomEvent::MemberJoined { user_id, .. }
| RoomEvent::MemberLeft { user_id, .. }
| RoomEvent::VoiceModeration { user_id, .. } => Some(*user_id),
RoomEvent::MessagePublished { message, .. } | RoomEvent::MessageEdited { message, .. } => {
message.get("author_id").and_then(Value::as_i64)
}
_ => None,
}
}
pub async fn enqueue_room(state: &AppState, event: &RoomEvent) -> Result<(), ChatError> {
let mut labels = queue::Labels::new().with("room_id", event.room_id());
if let Some(actor) = room_actor(event) {
labels.insert("user_id", actor);
}
if let Some(message_id) = room_message_id(event) {
labels.insert("message_id", message_id);
}
if let Some(thread_id) = room_thread_id(event) {
labels.insert("thread_id", thread_id);
}
let request_id = export::middleware::request_id();
if let Some(request_id) = request_id.as_deref() {
labels.insert("request_id", request_id);
}
let mut payload =
serde_json::to_value(event).map_err(|error| ChatError::Internal(error.to_string()))?;
if let Some(room) = room_detail(state, event.room_id()).await {
payload["room"] = room;
}
if let Some(actor) = room_actor(event)
&& let Some(actor) = user_detail(state, actor).await
{
payload["actor"] = actor;
}
if let Some(message_id) = room_message_id(event)
&& !matches!(
event,
RoomEvent::MessagePublished { .. } | RoomEvent::MessageEdited { .. }
)
&& let Some(message) = message_detail(state, message_id).await
{
payload["message"] = message;
}
labels.insert(
events::label::ORDERING_KEY,
format!("chat:room:{}", event.room_id()),
);
events::outbox::enqueue_now(
state.database.write(),
room_kind(event),
labels,
&payload,
request_id.as_deref().unwrap_or_default(),
request_id.as_deref(),
)
.await?;
Ok(())
}
pub async fn enqueue_user(state: &AppState, uid: i64, event: &Value) -> Result<(), ChatError> {
let Some(kind) = user_kind(event) else {
let unknown = event
.get("kind")
.and_then(Value::as_str)
.unwrap_or_default();
tracing::warn!(kind = %unknown, %uid, "unknown chat user event kind; skipping the outbox write");
return Ok(());
};
let mut labels = queue::Labels::new().with("user_id", uid);
if let Some(room_id) = event.get("room_id").and_then(Value::as_str) {
labels.insert("room_id", room_id);
}
let request_id = export::middleware::request_id();
if let Some(request_id) = request_id.as_deref() {
labels.insert("request_id", request_id);
}
let mut payload = event.clone();
if let Some(room_id) = event
.get("room_id")
.and_then(Value::as_str)
.and_then(|value| Uuid::parse_str(value).ok())
&& let Some(room) = room_detail(state, room_id).await
{
payload["room"] = room;
}
if let Some(message_id) = event
.get("message_id")
.and_then(Value::as_str)
.and_then(|value| Uuid::parse_str(value).ok())
&& let Some(message) = message_detail(state, message_id).await
{
if let Some(author_id) = message.get("author_id").and_then(Value::as_i64)
&& let Some(actor) = user_detail(state, author_id).await
{
payload["actor"] = actor;
}
payload["message"] = message;
}
labels.insert(events::label::ORDERING_KEY, format!("chat:user:{uid}"));
events::outbox::enqueue_now(
state.database.write(),
kind,
labels,
&payload,
request_id.as_deref().unwrap_or_default(),
request_id.as_deref(),
)
.await?;
Ok(())
}
async fn room_detail(state: &AppState, room_id: Uuid) -> Option<Value> {
let room = rooms::Entity::find_by_id(room_id)
.one(state.database.read())
.await
.ok()??;
Some(json!({
"id": room.id,
"space_id": room.space_id,
"kind": room.kind,
"name": room.name,
"topic": room.topic,
}))
}
async fn message_detail(state: &AppState, message_id: Uuid) -> Option<Value> {
let message = messages::Entity::find_by_id(message_id)
.one(state.database.read())
.await
.ok()??;
let mut detail = json!({
"id": message.id,
"room_id": message.room_id,
"author_id": message.author_id,
"author_type": message.author_type,
"content": message.content,
"message_type": message.message_type,
"mentions": message.mentions,
"attachments": message.attachments,
"embeds": message.embeds,
"seq": message.seq,
"created_at": message.created_at,
"edited_at": message.edited_at,
"deleted_at": message.deleted_at,
});
if let Some(actor) = user_detail(state, message.author_id).await {
detail["author"] = actor;
}
Some(detail)
}
async fn user_detail(state: &AppState, uid: i64) -> Option<Value> {
let user = export::users::get_public_user(state, uid).await.ok()??;
Some(json!({
"id": user.user_id,
"username": user.username,
"display_name": user.displayname,
"avatar_url": export::resolve_user_avatar_url(user.user_id, &user.avatar_url),
}))
}