9a3b8a8
use export::AppDatabase;
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_actor(event: &RoomEvent) -> Option<i64> {
match event {
RoomEvent::Typing { user_id, .. }
| RoomEvent::MemberJoined { user_id, .. }
| RoomEvent::MemberLeft { user_id, .. }
| RoomEvent::VoiceModeration { user_id, .. }
| RoomEvent::MessageDeleted { user_id, .. }
| RoomEvent::MessagesDeleted { user_id, .. }
| RoomEvent::MessagePinned { user_id, .. }
| RoomEvent::ReactionUpdated { 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(database: &AppDatabase, 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);
labels.insert(events::label::ACTOR_USER_ID, actor);
}
if let Some(message_id) = room_message_id(event) {
labels.insert("message_id", message_id);
}
if let Some(thread_id) = {
match event {
RoomEvent::ThreadUpdated { thread_id, .. } => Some(*thread_id),
_ => None,
}
} {
labels.insert("thread_id", thread_id);
}
let mut payload =
serde_json::to_value(event).map_err(|error| ChatError::Internal(error.to_string()))?;
if let Some(room) = room_detail(database, event.room_id()).await {
payload["room"] = room;
}
if let Some(message_id) = room_message_id(event)
&& !matches!(
event,
RoomEvent::MessagePublished { .. } | RoomEvent::MessageEdited { .. }
)
&& let Some(message) = message_detail(database, message_id).await
{
payload["message"] = message;
}
events::emit_retry(database.write(), room_kind(event), labels, &payload).await?;
Ok(())
}
pub async fn enqueue_user(
database: &AppDatabase,
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)
.with(events::label::RECIPIENT_USER_ID, uid);
if let Some(room_id) = event.get("room_id").and_then(Value::as_str) {
labels.insert("room_id", room_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(database, 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(database, message_id).await
{
payload["message"] = message;
}
events::emit_retry(database.write(), kind, labels, &payload).await?;
Ok(())
}
async fn room_detail(database: &AppDatabase, room_id: Uuid) -> Option<Value> {
let room = rooms::Entity::find_by_id(room_id)
.one(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(database: &AppDatabase, message_id: Uuid) -> Option<Value> {
let message = messages::Entity::find_by_id(message_id)
.one(database.read())
.await
.ok()??;
let 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,
});
Some(detail)
}