b453d5d
use models::system::event_outbox;
use queue::{Envelope, Labels};
use sea_orm::{ActiveModelTrait, ConnectionTrait, DbErr, EntityTrait, Set};
use serde::Serialize;
use uuid::Uuid;
fn active_model<T: Serialize>(
kind: &str,
labels: Labels,
payload: &T,
ordering_key: &str,
) -> Result<event_outbox::ActiveModel, DbErr> {
let source = kind.split('.').next().unwrap_or_default().to_string();
let envelope = Envelope::new(kind, payload)
.with_source(source)
.with_labels(labels);
let subject = envelope
.subject()
.map_err(|error| DbErr::Custom(error.to_string()))?;
let stored = serde_json::to_value(&envelope)
.map_err(|error| DbErr::Custom(format!("encoding outbox event: {error}")))?;
Ok(event_outbox::ActiveModel {
id: Set(envelope.id),
subject: Set(subject),
kind: Set(kind.to_string()),
ordering_key: Set(ordering_key.to_string()),
envelope: Set(stored),
occurred_at: Set(envelope.occurred_at),
available_at: Set(envelope.occurred_at),
..Default::default()
})
}
pub async fn enqueue<T: Serialize>(
db: &impl ConnectionTrait,
enabled: bool,
kind: &str,
labels: Labels,
payload: T,
ordering_key: impl Into<String>,
) -> Result<(), DbErr> {
if !enabled {
return Ok(());
}
active_model(kind, labels, &payload, &ordering_key.into())?
.insert(db)
.await?;
Ok(())
}
pub fn request_id() -> Option<String> {
crate::middleware::request_id()
}
fn room_labels(room_id: Uuid) -> Labels {
let mut labels = Labels::new().with("room_id", room_id);
if let Some(request_id) = request_id() {
labels.insert("request_id", request_id);
}
labels
}
/// 房间级事件的统一入队口：标签固定 `room_id`，ordering key 为 `chat:room:{room_id}`。
///
/// 供不在 `chat` crate 里的写入点（例如 `space` 的频道删除）复用，
/// 避免它们为了一个标签类型去直接依赖 `queue`。
pub async fn enqueue_room<T: Serialize>(
db: &impl ConnectionTrait,
enabled: bool,
kind: &str,
room_id: Uuid,
payload: T,
) -> Result<(), DbErr> {
if !enabled {
return Ok(());
}
active_model(
kind,
room_labels(room_id),
&payload,
&format!("chat:room:{room_id}"),
)?
.insert(db)
.await?;
Ok(())
}
/// 房间变化扇出给一批收件人：每条一个 `ordering_key`，一次批量写入 outbox。
///
/// 用批量而不是逐条 `enqueue`，是因为公开频道创建时收件人等于整个空间的
/// 活跃成员数——逐条写入会变成 N 次往返。
pub async fn enqueue_chat_users<T: Serialize>(
db: &impl ConnectionTrait,
enabled: bool,
kind: &str,
uids: &[i64],
room_id: Uuid,
payload: &T,
) -> Result<(), DbErr> {
if !enabled || uids.is_empty() {
return Ok(());
}
let mut models = Vec::with_capacity(uids.len());
for uid in uids {
let labels = room_labels(room_id).with("user_id", *uid);
models.push(active_model(
kind,
labels,
payload,
&format!("chat:user:{uid}"),
)?);
}
event_outbox::Entity::insert_many(models).exec(db).await?;
Ok(())
}