feat(push): add support for targeted message delivery with optional subscription filtering

Modified lib/infra/config/src/push.rs +1 −1 View lib/infra/config/src/push.rs

@@ -22,7 +22,7 @@

2222 pub request_timeout_secs: u64,

2323 /// 单个任务内并发投递的设备数。

2424 pub delivery_concurrency: usize,

25 /// 单个任务的总时间预算；超出后放弃本次投递。

25 /// 单个任务的总时间预算；超出后取消本次尝试并重投未完成的设备。

2626 pub job_timeout_secs: u64,

2727 /// 推送消息在推送服务上的存活时间（TTL 头）。

2828 pub ttl_secs: u64,

Modified lib/service/chat/src/services/messages.rs +1 −0 View lib/service/chat/src/services/messages.rs

@@ -93,6 +93,7 @@

9393 notification_id: None,

9494 trace: oltp::propagation::traceparent(),

9595 request_id: api::middleware::request_id(),

96 targets: None,

9697 },

9798 )

9899 .await;

Modified lib/service/push/src/protocol.rs +4 −0 View lib/service/push/src/protocol.rs

@@ -104,6 +104,9 @@

104104 pub trace: Option<String>,

105105 #[serde(default)]

106106 pub request_id: Option<String>,

107 /// 部分失败或超时重投时只投这些订阅；`None` 表示该用户全部 Web Push 订阅。

108 #[serde(default, skip_serializing_if = "Option::is_none")]

109 pub targets: Option<Vec<Uuid>>,

107110}

108111

109112impl PushJob {

@@ -139,6 +142,7 @@

139142 notification_id,

140143 trace: oltp::propagation::traceparent(),

141144 request_id: api::middleware::request_id(),

145 targets: None,

142146 })

143147 }

144148

Modified lib/service/push/src/worker.rs +102 −37 View lib/service/push/src/worker.rs

@@ -98,17 +98,7 @@

9898 }

9999 oltp::propagation::set_parent_from_traceparent(&span, job.trace.as_deref());

100100 let budget = Duration::from_secs(self.state.config.push.job_timeout_secs.max(1));

101 let delivery = async {

102 match tokio::time::timeout(budget, self.handle(&job).instrument(span)).await {

103 Ok(result) => result,

104 Err(_) => {

105 oltp::metrics::push_delivery("timeout");

106 tracing::warn!("push job timed out; dropping this attempt");

107 self.forget(&job);

108 Ok(())

109 }

110 }

111 };

101 let delivery = self.handle(&job, budget).instrument(span);

112102 if let Err(error) =

113103 api::middleware::with_request_id(job.request_id.as_deref(), delivery).await

114104 {

@@ -118,6 +108,7 @@

118108 category = job.category.as_str(),

119109 "push job failed"

120110 );

111 self.retry_or_drop(&job).await;

121112 }

122113 }

123114 }

@@ -148,48 +139,122 @@

148139 }

149140

150141 /// 一条任务的完整流程：闸门 → 组装 → 逐订阅投递。

151 async fn handle(&self, job: &PushJob) -> Result<(), PushError> {

152 if self.skip_by_preferences(job).await? {

142 ///

143 /// 超时不丢弃任务：已完成设备的结果留在 `completed` 里，只把未完成的订阅

144 /// 重新入队，避免已投递的设备重复弹窗、未投递的设备被静默丢掉。

145 async fn handle(&self, job: &PushJob, budget: Duration) -> Result<(), PushError> {

146 let deadline = tokio::time::Instant::now() + budget;

147 let prepared = match tokio::time::timeout_at(deadline, self.prepare(job)).await {

148 Ok(result) => result?,

149 Err(_) => {

150 oltp::metrics::push_delivery("timeout");

151 tracing::warn!(user = %job.user_id, "push job setup timed out; retrying");

152 self.retry_or_drop(job).await;

153 return Ok(());

154 }

155 };

156 let Some((subscriptions, payload)) = prepared else {

153157 self.forget(job);

154158 return Ok(());

159 };

160

161 let targets: Vec<Uuid> = subscriptions

162 .iter()

163 .map(|subscription| subscription.id)

164 .collect();

165 let ttl_secs = self.state.config.push.ttl_secs;

166 let concurrency = self.state.config.push.delivery_concurrency.max(1);

167 let (completed_tx, mut completed_rx) = tokio::sync::mpsc::unbounded_channel();

168 let deliveries = futures::stream::iter(subscriptions.into_iter().map(|subscription| {

169 let completed_tx = completed_tx.clone();

170 let payload = &payload;

171 async move {

172 let id = subscription.id;

173 let outcome = self.deliver_one(subscription, payload, ttl_secs).await;

174 let _ = completed_tx.send((id, outcome));

175 }

176 }))

177 .buffer_unordered(concurrency);

178 let timed_out = tokio::time::timeout_at(deadline, deliveries.count())

179 .await

180 .is_err();

181

182 let mut completed = HashMap::new();

183 while let Ok((id, outcome)) = completed_rx.try_recv() {

184 completed.insert(id, outcome);

155185 }

156 if self.skip_online(job.user_id).await {

186

187 if timed_out {

188 oltp::metrics::push_delivery("timeout");

189 }

190 if completed

191 .values()

192 .any(|outcome| *outcome == DeliveryOutcome::Rejected)

193 {

194 self.to_dlq(job).await;

157195 self.forget(job);

158196 return Ok(());

159197 }

160198

161 let subscriptions = user_push_subscriptions::Entity::find()

199 let pending: Vec<Uuid> = targets

200 .into_iter()

201 .filter(|id| {

202 !completed.get(id).is_some_and(|outcome| {

203 matches!(outcome, DeliveryOutcome::Delivered | DeliveryOutcome::Gone)

204 })

205 })

206 .collect();

207 if pending.is_empty() {

208 self.forget(job);

209 return Ok(());

210 }

211

212 if timed_out {

213 tracing::warn!(

214 user = %job.user_id,

215 pending = pending.len(),

216 "push job timed out; requeueing unfinished devices"

217 );

218 } else {

219 tracing::debug!(

220 user = %job.user_id,

221 pending = pending.len(),

222 "push retry requeueing unfinished devices"

223 );

224 }

225 let mut next = job.clone();

226 next.targets = Some(pending);

227 self.retry_or_drop(&next).await;

228 Ok(())

229 }

230

231 /// 闸门与组装；`None` 表示该任务无需投递（偏好/在线拦截或没有可用订阅）。

232 async fn prepare(

233 &self,

234 job: &PushJob,

235 ) -> Result<Option<(Vec<user_push_subscriptions::Model>, NotificationPayload)>, PushError> {

236 if self.skip_by_preferences(job).await? {

237 return Ok(None);

238 }

239 if self.skip_online(job.user_id).await {

240 return Ok(None);

241 }

242

243 let mut subscriptions = user_push_subscriptions::Entity::find()

162244 .filter(user_push_subscriptions::Column::UserId.eq(job.user_id))

163245 .filter(user_push_subscriptions::Column::Platform.eq(PushSubscriptionPlatform::WebPush))

164246 .all(self.state.database.read())

165247 .await

166248 .map_err(|error| PushError::Internal(format!("subscription lookup failed: {error}")))?;

249 if let Some(targets) = job.targets.as_deref() {

250 subscriptions.retain(|subscription| targets.contains(&subscription.id));

251 }

167252 if subscriptions.is_empty() {

168 self.forget(job);

169 return Ok(());

253 return Ok(None);

170254 }

171255

172256 let payload = self.build_payload(job).await?;

173 let ttl_secs = self.state.config.push.ttl_secs;

174 let concurrency = self.state.config.push.delivery_concurrency.max(1);

175 let outcomes: Vec<DeliveryOutcome> = futures::stream::iter(

176 subscriptions

177 .into_iter()

178 .map(|subscription| self.deliver_one(subscription, &payload, ttl_secs)),

179 )

180 .buffer_unordered(concurrency)

181 .collect()

182 .await;

183

184 if outcomes.contains(&DeliveryOutcome::Rejected) {

185 self.to_dlq(job).await;

186 self.forget(job);

187 } else if outcomes.contains(&DeliveryOutcome::Retry) {

188 self.retry_or_drop(job).await;

189 } else {

190 self.forget(job);

191 }

192 Ok(())

257 Ok(Some((subscriptions, payload)))

193258 }

194259

195260 /// 向一台设备投递；订阅级副作用（清理失效行）就地完成。