9301161
use serde::{Deserialize, Serialize};
use serde_json::Value;
use uuid::Uuid;
use crate::error::ExportError;
use crate::redis::AppRedis;
/// 推送任务队列（LPUSH/BRPOP）与死信队列的固定键名，与 `chat:user:{uid}`
/// 一类键同属服务内部约定，不做配置化——入队点遍布各业务 crate，只有固定
/// 键名才能让它们无需感知配置。
pub const QUEUE_TOPIC: &str = "push:jobs";
pub const DLQ_TOPIC: &str = "push:jobs:dlq";
/// 活跃聊天 WS 连接的在线标记；由 im 网关续期，推送 worker 用它跳过在线用户。
///
/// 存的是一个排序集合：每条连接一个成员（成员 id），分数是该连接占位的
/// 到期时刻——强杀进程留下的成员会随分数过期被惰性清理，不会污染在线判定。
pub fn online_key(uid: Uuid) -> String {
format!("chat:ws:{uid}")
}
/// 推送任务的分类，决定 `user_preferences.push_flags` 里哪一位能把它关掉。
///
/// 位语义是"禁用位"：`push_flags == 0`（存量默认）表示四类全开，只有用户
/// 显式关掉某一类时才置位，所以新老账号都不会因为新增字段而静默收不到推送。
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum PushCategory {
Mention,
Reply,
Dm,
Reminder,
/// 工单、邀请、AI 评审、系统公告等：不受逐类开关控制。
Other,
}
pub const PUSH_FLAG_MENTION: i32 = 1 << 0;
pub const PUSH_FLAG_REPLY: i32 = 1 << 1;
pub const PUSH_FLAG_DM: i32 = 1 << 2;
pub const PUSH_FLAG_REMINDER: i32 = 1 << 3;
impl PushCategory {
pub fn as_str(self) -> &'static str {
match self {
Self::Mention => "mention",
Self::Reply => "reply",
Self::Dm => "dm",
Self::Reminder => "reminder",
Self::Other => "other",
}
}
/// 该类通知对应的禁用位；`Other` 不受任何位控制。
pub fn disable_flag(self) -> Option<i32> {
match self {
Self::Mention => Some(PUSH_FLAG_MENTION),
Self::Reply => Some(PUSH_FLAG_REPLY),
Self::Dm => Some(PUSH_FLAG_DM),
Self::Reminder => Some(PUSH_FLAG_REMINDER),
Self::Other => None,
}
}
pub fn is_disabled_by(self, push_flags: i32) -> bool {
self.disable_flag()
.is_some_and(|flag| push_flags & flag != 0)
}
/// 由站内通知的形状反推分类：通知类型定大类，`data.kind` 细分提醒。
pub fn from_notification(notification_type: &str, data_kind: Option<&str>) -> Self {
match notification_type {
"mention" => Self::Mention,
"reply" => Self::Reply,
_ => match data_kind {
Some("reminder") => Self::Reminder,
_ => Self::Other,
},
}
}
}
/// 一个待投递的推送任务；队列载荷就是它的 JSON 序列化结果。
///
/// 标题与正文在入队前就按收件人语言渲染好（业务侧已经持有 DB 上下文），
/// 头像、作者显示名与点击 URL 则由 worker 组装：前者需要预签名，后者只需要
/// `data` 里的路由 slug，两者都不值得让每个写入方各做一遍。
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PushJob {
pub user_id: Uuid,
pub category: PushCategory,
/// 站内通知的标题，worker 会在前面拼上作者显示名。
pub title: String,
pub body: Option<String>,
#[serde(default)]
pub data: Option<Value>,
#[serde(default)]
pub actor_id: Option<Uuid>,
/// 站内通知主键；仅用于日志关联，不参与投递决策。
#[serde(default)]
pub notification_id: Option<Uuid>,
#[serde(default)]
pub trace: Option<String>,
}
impl PushJob {
/// 把一个 `user_notifications` 事件载荷转成推送任务；形状不对时返回 `None`。
pub fn from_notification(user_id: Uuid, notification: &Value) -> Option<Self> {
let notification_type = notification.get("notification_type")?.as_str()?;
let title = notification.get("title")?.as_str()?.to_string();
let body = notification
.get("body")
.and_then(Value::as_str)
.map(str::to_string);
let data = notification.get("data").cloned();
let actor_id = notification
.get("actor_id")
.and_then(Value::as_str)
.and_then(|raw| Uuid::parse_str(raw).ok());
let notification_id = notification
.get("id")
.and_then(Value::as_str)
.and_then(|raw| Uuid::parse_str(raw).ok());
let data_kind = data
.as_ref()
.and_then(|value| value.get("kind"))
.and_then(Value::as_str);
Some(Self {
user_id,
category: PushCategory::from_notification(notification_type, data_kind),
title,
body,
data,
actor_id,
notification_id,
trace: crate::telemetry::traceparent(),
})
}
pub fn encode(&self) -> Result<Vec<u8>, ExportError> {
serde_json::to_vec(self).map_err(|error| ExportError::Unsupported(error.to_string()))
}
pub fn decode(payload: &[u8]) -> Result<Self, ExportError> {
serde_json::from_slice(payload)
.map_err(|error| ExportError::Unsupported(format!("invalid push job: {error}")))
}
}
/// 入队一个推送任务；队列不可用时只记日志，绝不影响调用方已经提交的业务写入。
///
/// `[push] enabled = false` 时这里直接返回（标志由 `AppState` 在启动时写进
/// `AppRedis`）：关闭推送的部署不该往队列里堆积没有消费者的任务。调用方不需要
/// 知道这个开关，所以入队点可以遍布各业务 crate。
pub async fn enqueue(redis: &AppRedis, job: &PushJob) {
if !redis.push_enabled() {
return;
}
let payload = match job.encode() {
Ok(payload) => payload,
Err(error) => {
tracing::warn!(%error, user = %job.user_id, "push job encode failed");
return;
}
};
if let Err(error) = redis
.lpush_capped_bytes(QUEUE_TOPIC, &payload, MAX_QUEUE_DEPTH)
.await
{
tracing::warn!(%error, user = %job.user_id, "push job enqueue failed");
}
}
/// 队列长度上限。
///
/// 推送是"尽力而为"：消费者长时间不可用（部署中断、worker 崩溃）时，真正有
/// 价值的做法是丢弃最旧的任务而不是让 Redis 一直涨。这里按条数设硬上限，
/// 超出的旧任务由 `LPUSH` 后的 `LTRIM` 直接丢掉。
pub const MAX_QUEUE_DEPTH: i64 = 50_000;
/// 死信：重试耗尽或明确不可重试的任务留在 DLQ 里供人工排查。
pub async fn enqueue_dlq(redis: &AppRedis, payload: &[u8]) {
if let Err(error) = redis.lpush_bytes(DLQ_TOPIC, payload).await {
tracing::warn!(%error, "push job dlq enqueue failed");
}
}