631aa98
use std::sync::{Arc, Mutex};
use queue::Labels;
use serde_json::Value;
use crate::AppState;
/// The chat WS per-user channel; every crate that writes a notification
/// publishes to it so connected clients receive the row in real time.
pub fn user_channel(uid: i64) -> String {
format!("chat:user:{uid}")
}
pub fn notification_event_json(notification: Value) -> Value {
serde_json::json!({
"kind": "notification",
"notification": notification,
})
}
pub async fn publish_user_event(state: &AppState, uid: i64, event: &Value) {
let redis = &state.redis;
let payload = match serde_json::to_string(event) {
Ok(payload) => payload,
Err(error) => {
tracing::warn!(%error, %uid, "user event serialize failed");
return;
}
};
let channel = user_channel(uid);
if let Err(error) = redis.publish(&channel, &payload).await {
tracing::warn!(%error, %channel, "user event publish failed");
}
// Every `notification` user event is also a candidate for an OS push.
// Enqueueing here — rather than at each of the nine writers — is what makes
// "websocket fan-out and push fan-out never drift" true by construction.
if event.get("kind").and_then(Value::as_str) == Some("notification") {
if let Some(notification) = event.get("notification")
&& let Some(job) = crate::push::PushJob::from_notification(uid, notification)
{
crate::push::enqueue(redis, &job).await;
}
let labels = Labels::new()
.with("user_id", uid)
.with_opt("request_id", crate::events::request_id());
if let Err(error) = crate::events::enqueue(
state.database.write(),
state.events_enabled,
"notify.notification.create",
labels,
event,
format!("notify:user:{uid}"),
)
.await
{
tracing::warn!(%error, %uid, "notification event enqueue failed");
}
}
}
pub async fn publish_notification(state: &AppState, uid: i64, notification: Value) {
let event = notification_event_json(notification);
publish_user_event(state, uid, &event).await;
}
/// Notifications created inside a DB transaction; flush after commit.
#[derive(Clone, Default)]
pub struct PendingNotifications {
inner: Arc<Mutex<Vec<(i64, Value)>>>,
}
impl PendingNotifications {
pub fn push(&self, uid: i64, notification: Value) {
match self.inner.lock() {
Ok(mut pending) => pending.push((uid, notification_event_json(notification))),
Err(_) => {
tracing::warn!("pending notifications lock poisoned");
}
}
}
pub async fn flush(&self, state: &AppState) {
let batch: Vec<(i64, Value)> = match self.inner.lock() {
Ok(mut pending) => std::mem::take(&mut *pending),
Err(_) => return,
};
for (uid, event) in batch {
publish_user_event(state, uid, &event).await;
}
}
}