fix: fan out unread presence to online sessions

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.

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

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)
 

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

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

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 {