b453d5d
use sea_orm::DatabaseConnection;
use serde_json::Value;
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(
db: &DatabaseConnection,
enabled: bool,
event: &RoomEvent,
) -> Result<(), ChatError> {
if !enabled {
return Ok(());
}
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);
}
if let Some(request_id) = export::events::request_id() {
labels.insert("request_id", request_id);
}
export::events::enqueue(
db,
enabled,
room_kind(event),
labels,
event,
format!("chat:room:{}", event.room_id()),
)
.await?;
Ok(())
}
pub async fn enqueue_user(
db: &DatabaseConnection,
enabled: bool,
uid: i64,
event: &Value,
) -> Result<(), ChatError> {
if !enabled {
return Ok(());
}
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);
}
if let Some(request_id) = export::events::request_id() {
labels.insert("request_id", request_id);
}
export::events::enqueue(db, enabled, kind, labels, event, format!("chat:user:{uid}")).await?;
Ok(())
}