diff --git a/server/pres.go b/server/pres.go index 13101b9..3f2a9e9 100644 --- a/server/pres.go +++ b/server/pres.go @@ -278,6 +278,7 @@ func (t *Topic) presUsersOfInterest(what, ua string) { // Publish user's update to his/her users of interest on their 'me' topic while user's 'me' topic is offline // Case A: user is being deleted, "gone". +// Case B: user's last session just disconnected, "off". func presUsersOfInterestOffline(tenantID types.TenantID, uid types.Uid, subs []types.Subscription, what string) { // Push update to subscriptions for i := range subs { @@ -286,6 +287,37 @@ func presUsersOfInterestOffline(tenantID types.TenantID, uid types.Uid, subs []t continue } + if types.GetTopicCat(subs[i].Topic) == types.TopicCatP2P { + // A presence request routed to the p2p topic would be silently + // dropped: the topic is loaded under its canonical p2p name while + // the request carries Topic="me" (notifyOnOrSkip), so the + // xoriginal check in handlePresence rejects it. Deliver to the + // contact's 'me' topic directly instead, the same way + // presSingleUserOffline does. + uid1, uid2, err := types.ParseP2P(subs[i].Topic) + if err != nil { + continue + } + contact := uid1 + if contact == uid { + contact = uid2 + } + if contact == uid || contact.IsZero() { + continue + } + routeUserServerMessage(contact, &ServerComMessage{ + TenantID: tenantID, + Pres: &MsgServerPres{ + Topic: notifyOn, + What: what, + Src: subs[i].Topic, + WantReply: false, + }, + RcptTo: contact.UserId(), + }) + continue + } + globals.hub.routeSrv <- &ServerComMessage{ TenantID: tenantID, Pres: &MsgServerPres{ diff --git a/server/session.go b/server/session.go index 25e021b..cc49c02 100644 --- a/server/session.go +++ b/server/session.go @@ -465,9 +465,19 @@ func (s *Session) cleanUp(expired bool) { if err := store.SecurityAudit.CloseLogin(s.tenantID, s.sid, time.Now().UTC().Round(time.Millisecond)); err != nil { logs.Warn.Println("s.cleanUp: failed to close login audit", s.uid, err, s.sid) } - if err := store.MarkUserOffline(s.tenantID, s.uid, s.sid, currentNodeID()); err != nil { + remaining, err := store.MarkUserOffline(s.tenantID, s.uid, s.sid, currentNodeID()) + if err != nil { logs.Warn.Println("s.cleanUp: failed to mark user offline", s.uid, err, s.sid) } + // 最后一个会话已断开(如主动登出、断网):立即通知关注者该用户离线, + // 不再等待 me topic 闲置超时后的 presUsersOfInterest("off")。 + if err == nil && remaining == 0 && globals.hub != nil { + if uoi, uerr := store.Users.GetSubs(s.tenantID, s.uid); uerr == nil { + presUsersOfInterestOffline(s.tenantID, s.uid, uoi, "off") + } else { + logs.Warn.Println("s.cleanUp: failed to notify users of interest", s.uid, uerr, s.sid) + } + } } s.purgeChannels() s.inflightReqs.Wait() diff --git a/server/store/group_message_runtime.go b/server/store/group_message_runtime.go index 2803a3d..8450922 100644 --- a/server/store/group_message_runtime.go +++ b/server/store/group_message_runtime.go @@ -109,7 +109,7 @@ type groupMessageRuntime interface { InvalidateHotMessages(tenantID types.TenantID, topic string, ranges []types.Range) error MarkUserOnline(tenantID types.TenantID, uid types.Uid, sid, nodeID, device, userAgent string) error RefreshUserPresence(tenantID types.TenantID, uid types.Uid, sid, nodeID string) error - MarkUserOffline(tenantID types.TenantID, uid types.Uid, sid, nodeID string) error + MarkUserOffline(tenantID types.TenantID, uid types.Uid, sid, nodeID string) (int, error) GetUserPresence(tenantID types.TenantID, uid types.Uid) (*UserPresence, error) MarkTopicMemberOnline(tenantID types.TenantID, topic string, uid types.Uid, nodeID string) error MarkTopicMemberOffline(tenantID types.TenantID, topic string, uid types.Uid, nodeID string) error @@ -300,9 +300,9 @@ func RefreshUserPresence(tenantID types.TenantID, uid types.Uid, sid, nodeID str return groupMsgRuntime.RefreshUserPresence(tenantID, uid, sid, nodeID) } -func MarkUserOffline(tenantID types.TenantID, uid types.Uid, sid, nodeID string) error { +func MarkUserOffline(tenantID types.TenantID, uid types.Uid, sid, nodeID string) (int, error) { if groupMsgRuntime == nil || !tenantID.IsValid() || uid.IsZero() || sid == "" { - return nil + return 0, nil } return groupMsgRuntime.MarkUserOffline(tenantID, uid, sid, nodeID) } @@ -939,7 +939,7 @@ func (r *redisGroupMessageRuntime) RefreshUserPresence(tenantID types.TenantID, return err } -func (r *redisGroupMessageRuntime) MarkUserOffline(tenantID types.TenantID, uid types.Uid, sid, nodeID string) error { +func (r *redisGroupMessageRuntime) MarkUserOffline(tenantID types.TenantID, uid types.Uid, sid, nodeID string) (int, error) { if nodeID == "" { nodeID = "local" } @@ -956,11 +956,11 @@ func (r *redisGroupMessageRuntime) MarkUserOffline(tenantID types.TenantID, uid pipe.SRem(ctx, sessionsKey, member) pipe.SRem(ctx, nodeKey, uid.UserId()) if _, err := pipe.Exec(ctx); err != nil { - return err + return 0, err } count, err := r.client.SCard(ctx, sessionsKey).Result() if err != nil { - return err + return 0, err } if count > 0 { pipe = r.client.Pipeline() @@ -968,14 +968,14 @@ func (r *redisGroupMessageRuntime) MarkUserOffline(tenantID types.TenantID, uid pipe.Expire(ctx, userKey, ttl) pipe.Expire(ctx, sessionsKey, ttl) _, err = pipe.Exec(ctx) - return err + return int(count), err } pipe = r.client.Pipeline() pipe.HSet(ctx, userKey, "online", "0", "session_count", 0, "node_id", nodeID, "last_seen_at", now.Format(time.RFC3339Nano)) pipe.Expire(ctx, userKey, ttl) pipe.Del(ctx, sessionsKey) _, err = pipe.Exec(ctx) - return err + return 0, err } func (r *redisGroupMessageRuntime) GetUserPresence(tenantID types.TenantID, uid types.Uid) (*UserPresence, error) { diff --git a/server/store/group_message_runtime_test.go b/server/store/group_message_runtime_test.go index 8a78b33..4616b5b 100644 --- a/server/store/group_message_runtime_test.go +++ b/server/store/group_message_runtime_test.go @@ -114,8 +114,8 @@ func (r *testGroupMessageRuntime) RefreshUserPresence(types.TenantID, types.Uid, return nil } -func (r *testGroupMessageRuntime) MarkUserOffline(types.TenantID, types.Uid, string, string) error { - return nil +func (r *testGroupMessageRuntime) MarkUserOffline(types.TenantID, types.Uid, string, string) (int, error) { + return 0, nil } func (r *testGroupMessageRuntime) GetUserPresence(types.TenantID, types.Uid) (*UserPresence, error) {