From 816be46e0389e833b3032b9ea6b63de957ba7c71 Mon Sep 17 00:00:00 2001 From: richjohnrambo Date: Thu, 17 Sep 2026 10:24:28 +0800 Subject: [PATCH] fix: fan out unread presence to online sessions --- server/pres.go | 22 ++++++++--- server/push.go | 10 +++++ server/realtime_dispatcher.go | 8 ++++ server/realtime_dispatcher_test.go | 56 +++++++++++++++++++++++++++ server/sessionstore.go | 34 ++++++++++++++++ server/store/group_message_runtime.go | 36 +++++++++++++++++ 6 files changed, 161 insertions(+), 5 deletions(-) diff --git a/server/pres.go b/server/pres.go index f5c2831..13101b9 100644 --- a/server/pres.go +++ b/server/pres.go @@ -451,7 +451,7 @@ func (t *Topic) presSubsOffline(what string, params *presParams, target = "" } - globals.hub.routeSrv <- &ServerComMessage{ + msg := &ServerComMessage{ TenantID: t.tenantID, Pres: &MsgServerPres{ Topic: "me", @@ -471,6 +471,7 @@ func (t *Topic) presSubsOffline(what string, params *presParams, RcptTo: user, SkipSid: skipSid, } + routeUserServerMessage(uid, msg) } } @@ -485,7 +486,7 @@ func (t *Topic) infoSubsOffline(from types.Uid, what string, seq int64, skipSid continue } - globals.hub.routeSrv <- &ServerComMessage{ + msg := &ServerComMessage{ TenantID: t.tenantID, Info: &MsgServerInfo{ Topic: "me", @@ -498,6 +499,7 @@ func (t *Topic) infoSubsOffline(from types.Uid, what string, seq int64, skipSid RcptTo: uid.UserId(), SkipSid: skipSid, } + routeUserServerMessage(uid, msg) } } @@ -530,7 +532,7 @@ func (t *Topic) infoCallSubsOffline(from string, target types.Uid, event string, if offlineOnly { msg.Info.SkipTopic = t.name } - globals.hub.routeSrv <- msg + routeUserServerMessage(target, msg) } // Same as presSubsOffline, but the topic has not been loaded/initialized first: offline topic, offline subscribers @@ -604,7 +606,7 @@ func (t *Topic) presSingleUserOffline(uid types.Uid, mode types.AccessMode, target = "" } - globals.hub.routeSrv <- &ServerComMessage{ + msg := &ServerComMessage{ TenantID: t.tenantID, Pres: &MsgServerPres{ Topic: "me", @@ -622,6 +624,7 @@ func (t *Topic) presSingleUserOffline(uid types.Uid, mode types.AccessMode, RcptTo: user, SkipSid: skipSid, } + routeUserServerMessage(uid, msg) } } @@ -639,7 +642,7 @@ func presSingleUserOfflineOffline(tenantID types.TenantID, uid types.Uid, origin target = "" } - globals.hub.routeSrv <- &ServerComMessage{ + msg := &ServerComMessage{ TenantID: tenantID, Pres: &MsgServerPres{ Topic: "me", @@ -654,6 +657,15 @@ func presSingleUserOfflineOffline(tenantID types.TenantID, uid types.Uid, origin RcptTo: uid.UserId(), SkipSid: skipSid, } + routeUserServerMessage(uid, msg) +} + +func routeUserServerMessage(uid types.Uid, msg *ServerComMessage) { + if msg == nil || uid.IsZero() { + return + } + globals.hub.routeSrv <- msg + publishRealtimeUserMessage(uid, msg) } // Let other sessions of a given user know what messages are now received/read. diff --git a/server/push.go b/server/push.go index 6f46b27..9581a27 100644 --- a/server/push.go +++ b/server/push.go @@ -10,7 +10,9 @@ package main import ( "time" + "github.com/tinode/chat/server/logs" "github.com/tinode/chat/server/push" + "github.com/tinode/chat/server/store" "github.com/tinode/chat/server/store/types" ) @@ -73,6 +75,14 @@ func (t *Topic) pushForData(fromUid types.Uid, data *MsgServerData, msgMarkedAsR // Make sure the sender's devices receive a silent push. online = 1 } + if online == 0 { + presence, err := store.GetUserPresence(t.tenantID, uid) + if err != nil { + logs.Warn.Printf("topic[%s]: failed to read user presence for push: uid=%s err=%v", t.name, uid.UserId(), err) + } else if presence != nil && presence.Online && presence.SessionCount > 0 { + online = presence.SessionCount + } + } // Send only to those who have notifications enabled. mode := pud.modeWant & pud.modeGiven diff --git a/server/realtime_dispatcher.go b/server/realtime_dispatcher.go index fcf0ef9..8018333 100644 --- a/server/realtime_dispatcher.go +++ b/server/realtime_dispatcher.go @@ -161,6 +161,14 @@ func deliverRealtimeFanoutEnvelope(envelope *store.RealtimeFanoutEnvelope) { msg.TenantID = envelope.TenantID msg.RcptTo = envelope.Target msg.SkipSid = envelope.SkipSid + if envelope.SkipTopic != "" { + if msg.Pres != nil { + msg.Pres.SkipTopic = envelope.SkipTopic + } + if msg.Info != nil { + msg.Info.SkipTopic = envelope.SkipTopic + } + } msg.sess = nil prepareRedisGroupFanoutMessage(&msg, envelope.Target) diff --git a/server/realtime_dispatcher_test.go b/server/realtime_dispatcher_test.go index 1951273..5996b52 100644 --- a/server/realtime_dispatcher_test.go +++ b/server/realtime_dispatcher_test.go @@ -52,6 +52,62 @@ func TestRealtimeEnvelopeDeliversToUserSessions(t *testing.T) { } } +func TestRealtimeUserPresenceSkipsSessionsAttachedToTopic(t *testing.T) { + oldStore := globals.sessionStore + defer func() { globals.sessionStore = oldStore }() + + tenantID := types.TenantID(1) + sender := types.Uid(1001) + recipient := types.Uid(1002) + topic := sender.P2PName(recipient) + if topic == "" { + t.Fatal("failed to create p2p topic") + } + visibleTopic, err := types.P2PNameForUser(recipient, topic) + if err != nil || visibleTopic == "" { + t.Fatalf("failed to create visible topic: %v", err) + } + globals.sessionStore = realtimeTestSessionStore( + realtimeTestSession("sid-attached", tenantID, recipient, visibleTopic), + realtimeTestSession("sid-me-only", tenantID, recipient, ""), + ) + msg := &ServerComMessage{ + Pres: &MsgServerPres{ + Topic: "me", + Src: topic, + What: "msg", + SeqId: 35, + SkipTopic: topic, + }, + } + raw, err := json.Marshal(msg) + if err != nil { + t.Fatal(err) + } + + deliverRealtimeFanoutEnvelope(&store.RealtimeFanoutEnvelope{ + TenantID: tenantID, + Kind: realtimeKindMessage, + TargetType: realtimeTargetUser, + Target: recipient.UserId(), + SkipTopic: topic, + Message: raw, + }) + + if len(globals.sessionStore.Get("sid-attached").send) != 0 { + t.Fatal("attached p2p session must not receive duplicate me presence") + } + select { + case got := <-globals.sessionStore.Get("sid-me-only").send: + pres := got.(*ServerComMessage).Pres + if pres == nil || pres.Topic != "me" || pres.What != "msg" || pres.Src != topic || pres.SeqId != 35 { + t.Fatalf("unexpected presence: %#v", pres) + } + default: + t.Fatal("me-only session must receive message presence") + } +} + func TestRealtimeP2PTopicFallbackRewritesVisibleTopic(t *testing.T) { oldStore := globals.sessionStore oldHub := globals.hub diff --git a/server/sessionstore.go b/server/sessionstore.go index 99edcd0..5eaf672 100644 --- a/server/sessionstore.go +++ b/server/sessionstore.go @@ -209,6 +209,9 @@ func (ss *SessionStore) DeliverToUser(tenantID types.TenantID, uid types.Uid, sk if sess == nil || sess.isCluster() || sess.isProxy() || sess.tenantID != tenantID || sess.uid != uid || sess.sid == skipSid { return true } + if userMessageSkipTopic(msg, sess) { + return true + } if sess.queueOut(msg.copy()) { delivered++ } @@ -217,6 +220,37 @@ func (ss *SessionStore) DeliverToUser(tenantID types.TenantID, uid types.Uid, sk return delivered } +func userMessageSkipTopic(msg *ServerComMessage, sess *Session) bool { + if msg == nil || sess == nil { + return false + } + skipTopic := "" + switch { + case msg.Pres != nil: + skipTopic = msg.Pres.SkipTopic + case msg.Info != nil: + skipTopic = msg.Info.SkipTopic + } + return sessionHasTopicOrP2PAlias(sess, skipTopic) +} + +func sessionHasTopicOrP2PAlias(sess *Session, topic string) bool { + if sess == nil || topic == "" { + return false + } + if sess.hasSub(topic) { + return true + } + if types.GetTopicCat(topic) != types.TopicCatP2P || sess.uid.IsZero() { + return false + } + viewTopic, err := types.P2PNameForUser(sess.uid, topic) + if err != nil || viewTopic == "" { + return false + } + return sess.hasSub(viewTopic) +} + func (ss *SessionStore) DeliverToTopicSubscribers(tenantID types.TenantID, topic, skipSid string, msg *ServerComMessage) int { if ss == nil || !tenantID.IsValid() || topic == "" || msg == nil { return 0 diff --git a/server/store/group_message_runtime.go b/server/store/group_message_runtime.go index c61a7d4..2803a3d 100644 --- a/server/store/group_message_runtime.go +++ b/server/store/group_message_runtime.go @@ -6,6 +6,7 @@ import ( "encoding/hex" "encoding/json" "fmt" + "reflect" "sort" "strconv" "strings" @@ -74,6 +75,7 @@ type RealtimeFanoutEnvelope struct { TargetType string `json:"target_type"` Target string `json:"target"` SkipSid string `json:"skip_sid,omitempty"` + SkipTopic string `json:"skip_topic,omitempty"` DedupKey string `json:"dedup_key,omitempty"` Message json.RawMessage `json:"message"` } @@ -1060,6 +1062,7 @@ func (r *redisGroupMessageRuntime) PublishRealtimeFanout(originNodeID string, te TargetType: targetType, Target: target, SkipSid: skipSid, + SkipTopic: realtimeFanoutSkipTopic(message), DedupKey: dedupKey, Message: payload, }) @@ -1069,6 +1072,39 @@ func (r *redisGroupMessageRuntime) PublishRealtimeFanout(originNodeID string, te return r.client.Publish(context.Background(), r.realtimeFanoutChannel(tenantID), envelope).Err() } +func realtimeFanoutSkipTopic(message any) string { + if message == nil { + return "" + } + v := reflect.ValueOf(message) + if v.Kind() == reflect.Pointer { + if v.IsNil() { + return "" + } + v = v.Elem() + } + if v.Kind() != reflect.Struct { + return "" + } + for _, fieldName := range []string{"Pres", "Info"} { + field := v.FieldByName(fieldName) + if !field.IsValid() || field.IsNil() { + continue + } + if field.Kind() == reflect.Pointer { + field = field.Elem() + } + if field.Kind() != reflect.Struct { + continue + } + skip := field.FieldByName("SkipTopic") + if skip.IsValid() && skip.Kind() == reflect.String { + return strings.TrimSpace(skip.String()) + } + } + return "" +} + func (r *redisGroupMessageRuntime) StartGroupFanoutSubscriber(ctx context.Context, nodeID string, handler func([]byte)) error { return r.StartRealtimeFanoutSubscriber(ctx, nodeID, func(envelope *RealtimeFanoutEnvelope) { if envelope != nil {