6db412f
//! Push worker：从 Redis 队列取任务，过闸门，逐设备投递。
//!
//! 闸门顺序是刻意的——先看偏好位（纯内存），再看在线状态（一次 Redis
//! EXISTS），最后才回查数据库组装文案。用户关掉某类推送或已经在线时，
//! worker 不做任何数据库往返。
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use std::time::Duration;
use crate::protocol::{PushJob, online_key};
use bootstrap::AppState;
use models::users::user_preferences;
use models::users::user_push_subscriptions::{self, PushSubscriptionPlatform};
use sea_orm::{ColumnTrait, EntityTrait, QueryFilter};
use tracing::Instrument as _;
use crate::client::{DeliveryOutcome, NotificationPayload, SubscriptionKeys, VapidKeys, deliver};
use crate::error::PushError;
/// 队列空转的阻塞时长；同时决定进程响应 shutdown 的粒度。
const CONSUME_TIMEOUT_SECS: u64 = 5;
pub struct PushWorker {
state: AppState,
keys: Arc<VapidKeys>,
client: reqwest::Client,
cancel: tokio_util::sync::CancellationToken,
/// 重试计数按任务指纹累计。
///
/// 计数留在进程内是刻意的：持久化它需要再开一张表或改载荷字段，而代价
/// 只是进程重启后多投一次——对"尽力而为"的通知完全可接受。
attempts: Arc<Mutex<HashMap<String, u32>>>,
}
impl PushWorker {
pub fn new(
state: AppState,
cancel: tokio_util::sync::CancellationToken,
) -> Result<Self, PushError> {
let keys = VapidKeys::from_config(&state.config.push)?;
let timeout = Duration::from_secs(state.config.push.request_timeout_secs.max(1));
let client = reqwest::Client::builder()
.timeout(timeout)
.build()
.map_err(|error| PushError::Internal(format!("building http client: {error}")))?;
Ok(Self {
state,
keys: Arc::new(keys),
client,
cancel,
attempts: Arc::new(Mutex::new(HashMap::new())),
})
}
pub async fn run(self) {
let topic = crate::protocol::QUEUE_TOPIC.to_string();
loop {
let payload = tokio::select! {
_ = self.cancel.cancelled() => break,
result = self.state.redis.brpop_bytes(&topic, CONSUME_TIMEOUT_SECS) => {
match result {
Ok(Some(payload)) => payload,
Ok(None) => continue,
Err(error) => {
tracing::warn!(%error, "push queue consume failed");
tokio::time::sleep(Duration::from_secs(2)).await;
continue;
}
}
}
};
let job = match PushJob::decode(&payload) {
Ok(job) => job,
Err(error) => {
tracing::warn!(%error, "dropping malformed push job");
crate::protocol::enqueue_dlq(&self.state.redis, &payload).await;
continue;
}
};
let span = tracing::info_span!(
"push.delivery",
user = %job.user_id,
category = job.category.as_str(),
request_id = tracing::field::Empty
);
if let Some(request_id) = job.request_id.as_deref() {
span.record("request_id", request_id);
}
oltp::propagation::set_parent_from_traceparent(&span, job.trace.as_deref());
if let Err(error) = api::middleware::with_request_id(
job.request_id.as_deref(),
self.handle(&job).instrument(span),
)
.await
{
tracing::warn!(
%error,
user = %job.user_id,
category = job.category.as_str(),
"push job failed"
);
}
}
tracing::debug!("push worker stopped");
}
/// 一条任务的完整流程：闸门 → 组装 → 逐订阅投递。
async fn handle(&self, job: &PushJob) -> Result<(), PushError> {
if self.skip_by_preferences(job).await? {
self.forget(job);
return Ok(());
}
if self.skip_online(job.user_id).await {
self.forget(job);
return Ok(());
}
let subscriptions = user_push_subscriptions::Entity::find()
.filter(user_push_subscriptions::Column::UserId.eq(job.user_id))
.filter(user_push_subscriptions::Column::Platform.eq(PushSubscriptionPlatform::WebPush))
.all(self.state.database.read())
.await
.map_err(|error| PushError::Internal(format!("subscription lookup failed: {error}")))?;
if subscriptions.is_empty() {
self.forget(job);
return Ok(());
}
let payload = self.build_payload(job).await?;
let ttl_secs = self.state.config.push.ttl_secs;
let mut retry = false;
for subscription in subscriptions {
let keys = SubscriptionKeys {
endpoint: subscription.endpoint.clone(),
p256dh_key: subscription.p256dh_key.clone(),
auth_key: subscription.auth_key.clone(),
};
match deliver(&self.client, &self.keys, &keys, &payload, ttl_secs).await {
Ok(DeliveryOutcome::Delivered) => {
oltp::metrics::push_delivery("delivered");
tracing::debug!(user = %job.user_id, "push delivered");
}
Ok(DeliveryOutcome::Gone) => {
oltp::metrics::push_delivery("gone");
// 用户清了浏览器数据或卸载了站点：一次性清理，不是错误。
tracing::debug!(user = %job.user_id, "push subscription expired; removing");
if let Err(error) =
user_push_subscriptions::Entity::delete_by_id(subscription.id)
.exec(self.state.database.write())
.await
{
tracing::warn!(%error, "failed to remove expired push subscription");
}
}
Ok(DeliveryOutcome::Rejected) => {
oltp::metrics::push_delivery("rejected");
tracing::warn!(
user = %job.user_id,
endpoint = %subscription.endpoint,
"push rejected by service; sending to dlq"
);
self.to_dlq(job).await;
self.forget(job);
return Ok(());
}
Ok(DeliveryOutcome::Retry) => {
oltp::metrics::push_delivery("retry");
retry = true;
}
Err(PushError::Subscription(message)) => {
oltp::metrics::push_delivery("subscription_error");
// 订阅本身不可修（endpoint 或密钥解析失败），重投没有意义。
tracing::warn!(user = %job.user_id, %message, "dropping broken subscription");
if let Err(error) =
user_push_subscriptions::Entity::delete_by_id(subscription.id)
.exec(self.state.database.write())
.await
{
tracing::warn!(%error, "failed to remove broken push subscription");
}
}
Err(error) => {
oltp::metrics::push_delivery("error");
tracing::warn!(%error, user = %job.user_id, "push delivery error");
retry = true;
}
}
}
if retry {
self.retry_or_drop(job).await;
} else {
self.forget(job);
}
Ok(())
}
/// 用户偏好里的禁用位；`Other` 类不受位控制，只受"是否有订阅"约束。
async fn skip_by_preferences(&self, job: &PushJob) -> Result<bool, PushError> {
let Some(flags) = user_preferences::Entity::find_by_id(job.user_id)
.one(self.state.database.read())
.await
.map_err(|error| PushError::Internal(format!("preferences lookup failed: {error}")))?
else {
// 没有偏好行 = 默认全开。
return Ok(false);
};
Ok(job.category.is_disabled_by(flags.push_flags))
}
/// "用户在线"= 任意标签页持有聊天 WS 连接（集合里还有未过期的连接成员）；
/// 此时站内 toast 已经送达，再弹系统通知就是重复提醒。
async fn skip_online(&self, uid: i64) -> bool {
cache::client::zset_live_count(&self.state.redis, &online_key(uid))
.await
.unwrap_or(0)
> 0
}
/// 文案与深链的最终组装：作者名、头像、点击 URL 都需要数据库或存储。
async fn build_payload(&self, job: &PushJob) -> Result<NotificationPayload, PushError> {
let (title, icon) = match job.actor_id {
Some(actor_id) => {
let actor = auth::users::get_public_user(
&auth::Accounts::new(
self.state.database.connection().clone(),
self.state.cache.clone(),
),
actor_id,
)
.await
.map_err(|error| PushError::Internal(format!("actor lookup failed: {error}")))?;
match actor {
Some(actor) => {
let name = if actor.displayname.trim().is_empty() {
actor.username.clone()
} else {
actor.displayname.clone()
};
(
format!("{name} · {}", job.title),
self.avatar_url(actor_id, &actor.avatar_url).await,
)
}
None => (job.title.clone(), None),
}
}
None => (job.title.clone(), None),
};
Ok(NotificationPayload {
title,
body: job.body.clone(),
icon,
tag: job
.data
.as_ref()
.and_then(|data| data.get("room_id"))
.and_then(|value| value.as_str())
.map(|room_id| format!("chat:{room_id}")),
url: self.target_url(job),
})
}
/// 头像在库里存的是 `storage://` locator，预签名后才是浏览器可取到的
/// 地址；外链头像（导入数据）原样透传，空值表示前端渲染首字母。
async fn avatar_url(&self, uid: i64, stored: &str) -> Option<String> {
if stored.is_empty() {
return None;
}
if !stored.starts_with("storage://") {
return Some(stored.to_string());
}
let key = stored.strip_prefix("storage://")?;
if key.is_empty() {
return None;
}
match self.state.storage.presign_get(key, 7 * 24 * 60 * 60) {
Ok(url) => Some(url),
Err(error) => {
tracing::warn!(%error, %uid, "avatar presign failed");
None
}
}
}
/// 点击行为：与前端 `notificationTarget` 的规则保持一致，避免通知落到
/// 一个前端不认识的地址上；`data` 缺字段时退到站内通知中心。
fn target_url(&self, job: &PushJob) -> String {
let base = self.state.config.web.web_base_url.trim_end_matches('/');
let fallback = format!("{base}/app/@me/notify");
let Some(data) = job.data.as_ref() else {
return fallback;
};
let field = |key: &str| data.get(key).and_then(|value| value.as_str());
let scope = field("space_name").unwrap_or("@me");
if let Some(room_id) = field("room_id") {
let jump = field("message_id")
.map(|message_id| format!("?jump={message_id}"))
.unwrap_or_default();
return format!("{base}/app/{scope}/chat/{room_id}{jump}");
}
if let (Some(board_id), Some(issue_number)) = (field("board_id"), field("autoincrement_id"))
{
return format!("{base}/app/{scope}/board/{board_id}?issue={issue_number}");
}
if scope != "@me" {
return format!("{base}/app/{scope}");
}
fallback
}
/// 暂时性失败：退避后重投，超过上限进 DLQ。
///
/// 退避在独立任务里等待而不是就地 `sleep`：队列是单个 FIFO，就地等待会让
/// 一条坏任务把整个消费者按分钟级卡住，后面所有用户的推送跟着排长队。
async fn retry_or_drop(&self, job: &PushJob) {
let config = self.state.config.push.clone();
let attempt = {
let mut attempts = match self.attempts.lock() {
Ok(guard) => guard,
Err(_) => return,
};
let entry = attempts.entry(job_key(job)).or_insert(0);
*entry += 1;
*entry
};
if attempt >= config.max_attempts.max(1) {
tracing::warn!(user = %job.user_id, attempt, "push permanently failed; sending to dlq");
self.to_dlq(job).await;
self.forget(job);
return;
}
let backoff = config.retry_backoff_base_secs.max(1) * u64::from(attempt);
let redis = self.state.redis.clone();
let cancel = self.cancel.clone();
let Ok(payload) = job.encode() else {
return;
};
let user_id = job.user_id;
let request_id = job.request_id.clone();
tokio::spawn(async move {
let span = tracing::info_span!("push.requeue", request_id = %request_id.as_deref().unwrap_or_default());
let requeue = async move {
tokio::select! {
_ = cancel.cancelled() => return,
_ = tokio::time::sleep(Duration::from_secs(backoff)) => {}
}
if let Err(error) = redis
.lpush_bytes(crate::protocol::QUEUE_TOPIC, &payload)
.await
{
tracing::warn!(%error, %user_id, "push requeue failed");
}
};
api::middleware::with_request_id(request_id.as_deref(), requeue.instrument(span)).await;
});
}
async fn to_dlq(&self, job: &PushJob) {
if let Ok(payload) = job.encode() {
crate::protocol::enqueue_dlq(&self.state.redis, &payload).await;
}
}
fn forget(&self, job: &PushJob) {
if let Ok(mut attempts) = self.attempts.lock() {
attempts.remove(&job_key(job));
}
}
}
/// 同一任务的稳定指纹：一条站内通知的重投必须落回同一个计数桶。私信推送
/// 没有通知行，退化成"收件人 + 分类 + 标题"，对同一房间的消息足够唯一。
fn job_key(job: &PushJob) -> String {
match job.notification_id {
Some(id) => id.to_string(),
None => format!("{}:{}:{}", job.user_id, job.category.as_str(), job.title),
}
}