Skip to content
zhenyi

gitdata-core

0
Clone

HTTPS

SSH

SSH needs a key in your settings.

Download ZIP

feat(chat): expire online markers per connection

A user was marked online by an incremented counter that a crashed or
half-closed socket could leave behind forever, so people stayed green
long after they left.

Give every websocket its own member in a Redis sorted set, scored with
its expiry: a heartbeat re-touches it, a clean close removes it and the
remaining live count decides when to go offline, and a stale member
simply expires or is reaped lazily. Someone with several tabs stays
online until the last one goes away, Redis trouble never fakes an
offline, and the marker survives a restart of the gateway.

Viewers that hide their status are now masked on every read path
(member lists, single presence, push) instead of only on the map, push
notifications skip users with a live connection, and the presence TTL
follows the websocket heartbeat so it cannot outlive a silent client.
Online status now defaults to visible.
ZhenYi committed
parent 8da6cf8 commit 3fd0c85 Browse files

7 files changed

+370 −108

File tree 7
Modified app/im/src/socket.rs View app/im/src/socket.rs
@@ -95,9 +95,12 @@
9595 }
9696
9797 // 在线标记：推送 worker 用它决定"这个用户已经有实时通道，不需要系统
98 // 通知"。同一个账号可以同时开多个连接，所以这里维护的是引用计数，值
99 // 只用于计数，是否在线只看键在不在。
98 // 通知"。同一个账号可以同时开多个连接，每条连接在集合里各占一个成员，
99 // 只有全部断开（或成员过期）才算离线。
100100 let online = OnlineMarker::claim(&service, uid).await;
101 // 可见在线状态跟随同一条 WS 生命周期：claim 即上线（用户已有的 Away/
102 // Dnd/隐身状态原样保留、只续期），tick 续期，最后一条连接断开时下线。
103 service.mark_online(uid).await;
101104
102105 let heartbeat_secs = service.config().chat.ws.heartbeat_secs.max(5);
103106 let idle_limit = Duration::from_secs(heartbeat_secs * 3);
@@ -198,8 +201,10 @@
198201 }
199202 // 续期必须在 tick 里做而不是只在 ready 时写一次：TTL 到期而
200203 // 进程还活着（例如用户只是在读、没有发帧）时，键消失会让推送
201 // worker 把他当成离线，于是同一个事件既弹 toast 又弹系统通知。
204 // worker 把他当成离线，于是同一个事件既弹 toast 又弹系统通知；
205 // 可见在线状态同样靠这里的心跳保活。
202206 online.refresh().await;
207 service.touch_presence(uid).await;
203208 tick_count = tick_count.wrapping_add(1);
204209 if tick_count.is_multiple_of(2) {
205210 let mut revoked: Vec<Uuid> = Vec::new();
@@ -230,7 +235,11 @@
230235 }
231236 }
232237
233 online.release().await;
238 // 计数归零＝最后一条连接关闭；Redis 故障时 release 返回 None，此时不判定
239 // 离线，交给 TTL 兜底。
240 if online.release().await == Some(0) {
241 service.mark_offline(uid).await;
242 }
234243 let mut teardown: Vec<(Uuid, tokio::task::JoinHandle<()>)> = Vec::new();
235244 for (room_id, sub) in subscriptions {
236245 sub.cancel.cancel();
@@ -252,28 +261,23 @@
252261
253262/// 一条 WS 连接的"在线"占位。
254263///
255/// 值是引用计数：多标签页连同一个账号时只有全部断开才算离线。TTL 是安全网，
256/// 进程被强杀来不及 `release` 时键会自行消失，不会留下永久在线的幽灵。
264/// 每条连接是集合里的一个成员（分数＝到期时刻）：多标签页/多实例各自登记，
265/// 只有全部断开才算离线；进程被强杀留下的成员会随分数过期被惰性清理，不会
266/// 留下永久在线的幽灵，也不会污染在线计数。
257267struct OnlineMarker {
258268 key: String,
269 member: String,
259270 ttl_secs: u64,
260271 redis: export::AppRedis,
261272}
262273
263274impl OnlineMarker {
264275 async fn claim(service: &ChatService, uid: Uuid) -> Self {
265 // TTL 取心跳的两倍：一次 tick 丢失不会掉线，而进程崩溃后最多两个
266 // 心跳周期就能恢复正常推送。
267 let ttl_secs = service
268 .config()
269 .chat
270 .ws
271 .heartbeat_secs
272 .max(5)
273 .saturating_mul(2);
274 let key = export::push::online_key(uid);
276 let ttl_secs = service.config().chat.online_ttl_secs();
275277 let marker = Self {
276 key,
278 key: export::push::online_key(uid),
279 // 每条连接一个唯一成员：多标签页/多实例各自登记，互不覆盖。
280 member: Uuid::now_v7().to_string(),
277281 ttl_secs,
278282 redis: service.state.redis.clone(),
279283 };
@@ -281,15 +285,25 @@
281285 marker
282286 }
283287
288 /// 登记/续期本连接；过期的旧成员在这里被顺手清掉。续期是幂等的，
289 /// 心跳不会像计数型占位那样把数字越推越高。
284290 async fn refresh(&self) {
285 if let Err(error) = export::redis::incr_touch(&self.redis, &self.key, self.ttl_secs).await {
291 if let Err(error) =
292 export::redis::zset_touch(&self.redis, &self.key, &self.member, self.ttl_secs).await
293 {
286294 tracing::warn!(%error, key = %self.key, "online marker refresh failed");
287295 }
288296 }
289297
290 async fn release(&self) {
291 if let Err(error) = export::redis::decr_delete(&self.redis, &self.key).await {
292 tracing::warn!(%error, key = %self.key, "online marker release failed");
298 /// 注销本连接，返回剩余**存活**连接数；归零即最后一条连接退出。
299 /// Redis 故障返回 `None`，调用方不应据此判定离线（TTL 会兜底）。
300 async fn release(&self) -> Option<i64> {
301 match export::redis::zset_release(&self.redis, &self.key, &self.member).await {
302 Ok(remaining) => Some(remaining),
303 Err(error) => {
304 tracing::warn!(%error, key = %self.key, "online marker release failed");
305 None
306 }
293307 }
294308 }
295309}
Modified lib/auth/src/services/preferences.rs View lib/auth/src/services/preferences.rs
@@ -58,7 +58,9 @@
5858 mute_until: None,
5959 allow_dms_from: Audience::Everyone,
6060 allow_mentions_from: Audience::Everyone,
61 show_online_status: false,
61 // 在线状态默认对同空间/同私聊可见（与 Discord/Slack 一致），
62 // 用户可在隐私设置里显式关闭；关闭后读路径一律表现为离线。
63 show_online_status: true,
6264 }
6365 }
6466}
Modified lib/chat/src/services/presence.rs View lib/chat/src/services/presence.rs
@@ -1,10 +1,12 @@
11use models::chat::presence_states::{self, PresenceStatus};
22use models::space::space_members::{self, MemberStatus};
3use models::users::user_preferences;
34use redis::AsyncCommands;
45use sea_orm::{
56 ActiveModelTrait, ColumnTrait, EntityTrait, PaginatorTrait, QueryFilter, QuerySelect, Set,
67};
78use serde::{Deserialize, Serialize};
9use std::collections::HashMap;
810use time::OffsetDateTime;
911use utoipa::ToSchema;
1012use uuid::Uuid;
@@ -41,6 +43,24 @@
4143 at: i64,
4244}
4345
46fn encode_presence(status: PresenceStatus, activities: serde_json::Value) -> Option<String> {
47 serde_json::to_string(&PresencePayload {
48 status,
49 activities,
50 at: OffsetDateTime::now_utc().unix_timestamp(),
51 })
52 .map_err(|error| tracing::warn!(%error, "presence payload serialize failed"))
53 .ok()
54}
55
56/// 隐藏在线状态时的对外形态：一律表现为离线，并抹掉最后在线时间与活动。
57fn mask_presence(dto: &mut PresenceDto) {
58 dto.status = PresenceStatus::Offline;
59 dto.online = false;
60 dto.activities = serde_json::json!([]);
61 dto.last_seen_at = None;
62}
63
4464impl ChatService {
4565 pub async fn set_presence(
4666 &self,
@@ -109,6 +129,132 @@
109129 })
110130 }
111131
132 /* ------------------------------------------------------------- 生命周期 */
133
134 /// 聊天 WS 建立连接：用户上线。
135 ///
136 /// 已有键（手动 Away/Dnd/隐身）原样保留、只续期；没有键才写 Active 并
137 /// 广播。多标签页/多实例共享同一份 Redis 状态：谁先连谁写，后来者续期。
138 /// 生命周期 TTL 取两倍心跳，崩溃后由 TTL 兜底回离线。
139 pub async fn mark_online(&self, uid: Uuid) {
140 let ttl = self.config().chat.online_ttl_secs();
141 let key = presence_key(uid);
142 let Some(value) = encode_presence(PresenceStatus::Active, serde_json::json!([])) else {
143 return;
144 };
145 match export::redis::set_nx_ex(&self.state.redis, &key, &value, ttl).await {
146 Ok(true) => {
147 self.realtime
148 .publish_user(
149 uid,
150 &serde_json::json!({"kind": "presence", "status": PresenceStatus::Active}),
151 )
152 .await;
153 }
154 Ok(false) => {
155 if let Err(error) = export::redis::touch_expire(&self.state.redis, &key, ttl).await
156 {
157 tracing::warn!(%error, "presence renew failed");
158 }
159 }
160 Err(error) => tracing::warn!(%error, "presence mark online failed"),
161 }
162 }
163
164 /// 心跳续期：键在只续期；键已丢失（休眠唤醒、网络割接）时补写 Active
165 /// ——连接还活着就应当在线。补写走 `SET NX`，不会覆盖用户此刻手动设置
166 /// 的状态。
167 pub async fn touch_presence(&self, uid: Uuid) {
168 let ttl = self.config().chat.online_ttl_secs();
169 let key = presence_key(uid);
170 match export::redis::touch_expire(&self.state.redis, &key, ttl).await {
171 Ok(true) => {}
172 Ok(false) => {
173 let Some(value) = encode_presence(PresenceStatus::Active, serde_json::json!([]))
174 else {
175 return;
176 };
177 if let Err(error) =
178 export::redis::set_nx_ex(&self.state.redis, &key, &value, ttl).await
179 {
180 tracing::warn!(%error, "presence re-mark failed");
181 }
182 }
183 Err(error) => tracing::warn!(%error, "presence touch failed"),
184 }
185 }
186
187 /// 最后一条聊天连接关闭：清 Redis 键并落库 `last_seen_at`。
188 ///
189 /// 键的消失就是"离线"的唯一真相；DB 行只保留最后一次状态与时间，供
190 /// Redis 缺失时回退读取。
191 pub async fn mark_offline(&self, uid: Uuid) {
192 let now = OffsetDateTime::now_utc();
193 let key = presence_key(uid);
194 // 先读再删：DB 里留下的是用户最后处在的状态（Active/Away/Dnd/隐身）。
195 let last = export::redis::get_string(&self.state.redis, &key)
196 .await
197 .ok()
198 .flatten()
199 .and_then(|value| serde_json::from_str::<PresencePayload>(&value).ok());
200 if let Err(error) = export::redis::delete(&self.state.redis, &key).await {
201 tracing::warn!(%error, "presence key delete failed");
202 }
203 let (status, activities) = match last {
204 Some(payload) => (payload.status, payload.activities),
205 None => (PresenceStatus::Offline, serde_json::json!([])),
206 };
207 match presence_states::Entity::find_by_id(uid)
208 .one(self.db().write())
209 .await
210 {
211 Ok(Some(row)) => {
212 let mut am: presence_states::ActiveModel = row.into();
213 am.status = Set(status);
214 am.activities = Set(activities);
215 am.last_seen_at = Set(Some(now));
216 am.updated_at = Set(now);
217 if let Err(error) = am.update(self.db().write()).await {
218 tracing::warn!(%error, "presence last_seen update failed");
219 }
220 }
221 Ok(None) => {
222 let row = presence_states::ActiveModel {
223 user_id: Set(uid),
224 status: Set(status),
225 activities: Set(activities),
226 last_seen_at: Set(Some(now)),
227 updated_at: Set(now),
228 };
229 if let Err(error) = row.insert(self.db().write()).await {
230 tracing::warn!(%error, "presence last_seen insert failed");
231 }
232 }
233 Err(error) => tracing::warn!(%error, "presence row lookup failed"),
234 }
235 self.realtime
236 .publish_user(
237 uid,
238 &serde_json::json!({"kind": "presence", "status": PresenceStatus::Offline}),
239 )
240 .await;
241 }
242
243 /// `show_online_status` 开关：没有偏好行（从未保存过设置）按公开处理。
244 async fn show_online_map(&self, uids: &[Uuid]) -> Result<HashMap<Uuid, bool>, ChatError> {
245 if uids.is_empty() {
246 return Ok(HashMap::new());
247 }
248 let rows = user_preferences::Entity::find()
249 .filter(user_preferences::Column::UserId.is_in(uids.to_vec()))
250 .all(self.db().read())
251 .await?;
252 Ok(rows
253 .into_iter()
254 .map(|row| (row.user_id, row.show_online_status))
255 .collect())
256 }
257
112258 pub async fn get_presence(&self, uid: Uuid) -> Result<PresenceDto, ChatError> {
113259 let mut dto = self.get_presence_raw(uid).await?;
114260 let map = self.load_user_briefs(&[uid]).await?;
@@ -162,14 +308,27 @@
162308 }
163309
164310 /// Presence of another user is only visible when both share an active
165 /// space or a DM room; otherwise the requester gets NotFound.
311 /// space or a DM room; otherwise the requester gets NotFound. A target
312 /// who turned off `show_online_status` always reads as offline.
166313 pub async fn get_presence_for(
167314 &self,
168315 caller: Uuid,
169316 target: Uuid,
170317 ) -> Result<PresenceDto, ChatError> {
171318 if caller == target || self.shares_room_with(caller, target).await? {
172 return self.get_presence(target).await;
319 let mut dto = self.get_presence(target).await?;
320 if dto.user_id != caller {
321 let visible = self
322 .show_online_map(&[target])
323 .await?
324 .get(&target)
325 .copied()
326 .unwrap_or(true);
327 if !visible {
328 mask_presence(&mut dto);
329 }
330 }
331 return Ok(dto);
173332 }
174333 Err(ChatError::NotFound)
175334 }
@@ -230,68 +389,79 @@
230389 }
231390
232391 pub async fn get_presences(&self, uids: &[Uuid]) -> Result<Vec<PresenceDto>, ChatError> {
233 let mut by_uid: std::collections::HashMap<Uuid, PresencePayload> =
234 std::collections::HashMap::new();
235 if !uids.is_empty() {
236 let mut conn = self.state.redis.connection_clone();
237 for uid in uids {
238 let key = presence_key(*uid);
239 let value: Result<Option<String>, redis::RedisError> = match &mut conn {
240 export::RedisConnection::Standalone(c) => c.get(&key).await,
241 export::RedisConnection::Cluster(c) => c.get(&key).await,
242 };
243 if let Ok(Some(value)) = value
244 && let Ok(payload) = serde_json::from_str::<PresencePayload>(&value)
245 {
246 by_uid.insert(*uid, payload);
392 if uids.is_empty() {
393 return Ok(Vec::new());
394 }
395 let mut conn = self.state.redis.connection_clone();
396 let keys: Vec<String> = uids.iter().map(|uid| presence_key(*uid)).collect();
397 let values: Vec<Option<String>> = match &mut conn {
398 export::RedisConnection::Standalone(c) => {
399 redis::cmd("MGET")
400 .arg(keys.clone())
401 .query_async::<Vec<Option<String>>>(c)
402 .await?
403 }
404 export::RedisConnection::Cluster(c) => {
405 // 集群里每个键可能落在不同槽位，MGET 不成立，只能逐个 GET。
406 let mut values = Vec::with_capacity(keys.len());
407 for key in &keys {
408 values.push(redis::cmd("GET").arg(key).query_async(c).await?);
247409 }
410 values
411 }
412 };
413 let mut by_uid: HashMap<Uuid, PresencePayload> = HashMap::new();
414 for (uid, value) in uids.iter().zip(values) {
415 if let Some(value) = value
416 && let Ok(payload) = serde_json::from_str::<PresencePayload>(&value)
417 {
418 by_uid.insert(*uid, payload);
248419 }
249 let missing: Vec<Uuid> = uids
250 .iter()
251 .filter(|u| !by_uid.contains_key(u))
252 .copied()
253 .collect();
254 let rows = if missing.is_empty() {
255 Vec::new()
256 } else {
257 presence_states::Entity::find()
258 .filter(presence_states::Column::UserId.is_in(missing))
259 .all(self.db().read())
260 .await?
261 };
262 let mut result = Vec::with_capacity(uids.len());
263 for uid in uids {
264 if let Some(payload) = by_uid.get(uid) {
265 result.push(PresenceDto {
266 user_id: *uid,
267 user: None,
268 status: payload.status,
269 activities: payload.activities.clone(),
270 last_seen_at: Some(
271 OffsetDateTime::from_unix_timestamp(payload.at)
272 .unwrap_or_else(|_| OffsetDateTime::now_utc()),
273 ),
274 online: !matches!(payload.status, PresenceStatus::Offline),
275 });
276 continue;
277 }
278 let row = rows.iter().find(|r| r.user_id == *uid);
420 }
421 let missing: Vec<Uuid> = uids
422 .iter()
423 .filter(|u| !by_uid.contains_key(u))
424 .copied()
425 .collect();
426 let rows = if missing.is_empty() {
427 Vec::new()
428 } else {
429 presence_states::Entity::find()
430 .filter(presence_states::Column::UserId.is_in(missing))
431 .all(self.db().read())
432 .await?
433 };
434 let mut result = Vec::with_capacity(uids.len());
435 for uid in uids {
436 if let Some(payload) = by_uid.get(uid) {
279437 result.push(PresenceDto {
280438 user_id: *uid,
281439 user: None,
282 status: PresenceStatus::Offline,
283 activities: serde_json::json!([]),
284 last_seen_at: row.and_then(|r| r.last_seen_at),
285 online: false,
440 status: payload.status,
441 activities: payload.activities.clone(),
442 last_seen_at: Some(
443 OffsetDateTime::from_unix_timestamp(payload.at)
444 .unwrap_or_else(|_| OffsetDateTime::now_utc()),
445 ),
446 online: !matches!(payload.status, PresenceStatus::Offline),
286447 });
448 continue;
287449 }
288 let map = self.load_user_briefs(uids).await?;
289 for dto in result.iter_mut() {
290 dto.user = map.get(&dto.user_id).cloned();
291 }
292 return Ok(result);
450 let row = rows.iter().find(|r| r.user_id == *uid);
451 result.push(PresenceDto {
452 user_id: *uid,
453 user: None,
454 status: PresenceStatus::Offline,
455 activities: serde_json::json!([]),
456 last_seen_at: row.and_then(|r| r.last_seen_at),
457 online: false,
458 });
459 }
460 let map = self.load_user_briefs(uids).await?;
461 for dto in result.iter_mut() {
462 dto.user = map.get(&dto.user_id).cloned();
293463 }
294 Ok(Vec::new())
464 Ok(result)
295465 }
296466
297467 pub async fn list_space_presences(
@@ -310,6 +480,14 @@
310480 if !uids.contains(&space.owner_id) {
311481 uids.push(space.owner_id);
312482 }
313 self.get_presences(&uids).await
483 let mut presences = self.get_presences(&uids).await?;
484 // 隐私：关掉"展示在线状态"的成员对他人一律表现为离线，本人照常。
485 let visible = self.show_online_map(&uids).await?;
486 for dto in presences.iter_mut() {
487 if dto.user_id != caller && !visible.get(&dto.user_id).copied().unwrap_or(true) {
488 mask_presence(dto);
489 }
490 }
491 Ok(presences)
314492 }
315493}
Modified lib/config/src/chat.rs View lib/config/src/chat.rs
@@ -85,6 +85,17 @@
8585}
8686
8787impl ChatConfig {
88 /// 聊天 WS 在线占位与可见在线状态的 TTL：两倍心跳——丢一个 tick 不掉线，
89 /// 进程崩溃后最多两个心跳周期就会由 TTL 兜底回到离线。
90 ///
91 /// 与 `presence_ttl_secs` 的区别：那个是"手动设置一次状态"的存活时间，
92 /// 这个只用于跟随 WS 生命周期的自动续期。
93 pub fn online_ttl_secs(&self) -> u64 {
94 self.ws.heartbeat_secs.max(5).saturating_mul(2)
95 }
96}
97
98impl ChatConfig {
8899 pub fn is_valid(&self) -> Result<(), String> {
89100 if self.max_pagination == 0 || self.default_pagination == 0 {
90101 return Err("chat pagination must be positive".to_string());
Modified lib/export/src/push.rs View lib/export/src/push.rs
@@ -12,6 +12,9 @@
1212pub const DLQ_TOPIC: &str = "push:jobs:dlq";
1313
1414/// 活跃聊天 WS 连接的在线标记；由 im 网关续期，推送 worker 用它跳过在线用户。
15///
16/// 存的是一个排序集合：每条连接一个成员（成员 id），分数是该连接占位的
17/// 到期时刻——强杀进程留下的成员会随分数过期被惰性清理，不会污染在线判定。
1518pub fn online_key(uid: Uuid) -> String {
1619 format!("chat:ws:{uid}")
1720}
Modified lib/export/src/redis.rs View lib/export/src/redis.rs
@@ -631,57 +631,110 @@
631631 dispatch!(conn, |c| redis::cmd("DEL").arg(key).exec_async(c).await)
632632}
633633
634const INCR_TOUCH_SCRIPT: &str = r#"local c = redis.call('INCR', KEYS[1])
635redis.call('EXPIRE', KEYS[1], ARGV[1])
636return c"#;
634const ZSET_TOUCH_SCRIPT: &str = r#"local t = redis.call('TIME')
635local now = tonumber(t[1])
636redis.call('ZREMRANGEBYSCORE', KEYS[1], '-inf', now)
637redis.call('ZADD', KEYS[1], now + tonumber(ARGV[2]), ARGV[1])
638redis.call('EXPIRE', KEYS[1], tonumber(ARGV[2]) * 3)
639return 1"#;
637640
638/// `INCR` 并把存活时间重置为 `ttl_secs`，返回递增后的计数。
641/// 以"带到期时刻的集合成员"登记一份占用并续期；成员分数是到期时刻，
642/// 过期成员在每次交互时被惰性清掉。
639643///
640/// 与 [`incr_expire`] 的区别是**每次**都续期：那个是为限流窗口写的（窗口
641/// 从第一次计数开始算），这个是给心跳占位写的（每次心跳都要把到期时间往后推）。
642pub async fn incr_touch(
644/// 计数型占位（INCR/DECR）在进程被强杀时会留下永不衰减的陈旧计数——只要
645/// 还有任何心跳在续期键，陈旧计数就永远不会归零。每个连接一个成员、各自
646/// 携带到期时刻的集合没有这个问题：强杀只是留下一个分数过期的成员，之后
647/// 的任意一次 touch/release/读取都会把它清掉。
648pub async fn zset_touch(
643649 redis: &AppRedis,
644650 key: &str,
651 member: &str,
645652 ttl_secs: u64,
646) -> Result<i64, redis::RedisError> {
653) -> Result<(), redis::RedisError> {
647654 let mut conn = redis.connection_clone();
648 dispatch!(conn, |c| {
655 let _: i64 = dispatch!(conn, |c| {
649656 redis::cmd("EVAL")
650 .arg(INCR_TOUCH_SCRIPT)
657 .arg(ZSET_TOUCH_SCRIPT)
651658 .arg(1)
652659 .arg(key)
660 .arg(member)
653661 .arg(ttl_secs.max(1))
654662 .query_async(c)
655663 .await
656 })
664 })?;
665 Ok(())
657666}
658667
659const DECR_DELETE_SCRIPT: &str = r#"local c = redis.call('DECR', KEYS[1])
660if c <= 0 then
668const ZSET_RELEASE_SCRIPT: &str = r#"local t = redis.call('TIME')
669local now = tonumber(t[1])
670redis.call('ZREMRANGEBYSCORE', KEYS[1], '-inf', now)
671redis.call('ZREM', KEYS[1], ARGV[1])
672local n = redis.call('ZCARD', KEYS[1])
673if n == 0 then
661674 redis.call('DEL', KEYS[1])
662 return 0
663675end
664return c"#;
676return n"#;
665677
666/// 引用计数递减；归零时删键，返回剩余计数。
667///
668/// 一个用户可以同时开多个连接（多标签页、多设备同源会话），而"在线"是以键
669/// 存在与否表达的二值状态，所以计数减到 0 时必须把键删掉——否则最后一条连接
670/// 关闭后用户会一直显示在线，推送也就一直跳过。
671///
672/// 递减与删除放在一段脚本里：分开做的话，两次调用之间其它连接的心跳可能把
673/// 计数从 0 又推到 1，用户就会永远停在"在线"。
674pub async fn decr_delete(redis: &AppRedis, key: &str) -> Result<i64, redis::RedisError> {
678/// 注销一份占用，返回剩余**存活**成员数；归零时删键。计数前先清过期成员。
679pub async fn zset_release(
680 redis: &AppRedis,
681 key: &str,
682 member: &str,
683) -> Result<i64, redis::RedisError> {
684 let mut conn = redis.connection_clone();
685 dispatch!(conn, |c| {
686 redis::cmd("EVAL")
687 .arg(ZSET_RELEASE_SCRIPT)
688 .arg(1)
689 .arg(key)
690 .arg(member)
691 .query_async(c)
692 .await
693 })
694}
695
696const ZSET_LIVE_COUNT_SCRIPT: &str = r#"local t = redis.call('TIME')
697local now = tonumber(t[1])
698redis.call('ZREMRANGEBYSCORE', KEYS[1], '-inf', now)
699return redis.call('ZCARD', KEYS[1])"#;
700
701/// 当前**存活**成员数（先清过期成员）；键不存在时为 0。
702pub async fn zset_live_count(redis: &AppRedis, key: &str) -> Result<i64, redis::RedisError> {
675703 let mut conn = redis.connection_clone();
676704 dispatch!(conn, |c| {
677705 redis::cmd("EVAL")
678 .arg(DECR_DELETE_SCRIPT)
706 .arg(ZSET_LIVE_COUNT_SCRIPT)
679707 .arg(1)
680708 .arg(key)
681709 .query_async(c)
682710 .await
683711 })
684712}
713
714const TOUCH_EXPIRE_SCRIPT: &str = r#"if redis.call('EXISTS', KEYS[1]) == 1 then
715 redis.call('EXPIRE', KEYS[1], ARGV[1])
716 return 1
717end
718return 0"#;
719
720/// 键在则续期并返回 `true`；键不在返回 `false`，由调用方决定是否补写。
721pub async fn touch_expire(
722 redis: &AppRedis,
723 key: &str,
724 ttl_secs: u64,
725) -> Result<bool, redis::RedisError> {
726 let mut conn = redis.connection_clone();
727 let renewed: i64 = dispatch!(conn, |c| {
728 redis::cmd("EVAL")
729 .arg(TOUCH_EXPIRE_SCRIPT)
730 .arg(1)
731 .arg(key)
732 .arg(ttl_secs.max(1))
733 .query_async(c)
734 .await
735 })?;
736 Ok(renewed == 1)
737}
685738pub async fn exists(redis: &AppRedis, key: &str) -> Result<bool, redis::RedisError> {
686739 let mut conn = redis.connection_clone();
687740 dispatch!(conn, |c| redis::cmd("EXISTS").arg(key).query_async(c).await)
Modified lib/push/src/worker.rs View lib/push/src/worker.rs
@@ -190,12 +190,13 @@
190190 Ok(job.category.is_disabled_by(flags.push_flags))
191191 }
192192
193 /// "用户在线"= 任意标签页持有聊天 WS 连接；此时站内 toast 已经送达，
194 /// 再弹系统通知就是重复提醒。
193 /// "用户在线"= 任意标签页持有聊天 WS 连接（集合里还有未过期的连接成员）；
194 /// 此时站内 toast 已经送达，再弹系统通知就是重复提醒。
195195 async fn skip_online(&self, uid: Uuid) -> bool {
196 export::redis::exists(&self.state.redis, &online_key(uid))
196 export::redis::zset_live_count(&self.state.redis, &online_key(uid))
197197 .await
198 .unwrap_or(false)
198 .unwrap_or(0)
199 > 0
199200 }
200201
201202 /// 文案与深链的最终组装：作者名、头像、点击 URL 都需要数据库或存储。