diff --git a/server/im_group_events_test.go b/server/im_group_events_test.go index 6a1368a..067b14d 100644 --- a/server/im_group_events_test.go +++ b/server/im_group_events_test.go @@ -147,3 +147,60 @@ func TestSendPushForStoredGroupMessageIncludesEverySubscription(t *testing.T) { t.Fatal("group system message did not enqueue push") } } + +func TestSendPushForHTTPForwardMessageIncrementsRecipientUnreadOnly(t *testing.T) { + ctrl := gomock.NewController(t) + defer ctrl.Finish() + oldTopics := store.Topics + oldUsersUpdate := globals.usersUpdate + oldCluster := globals.cluster + oldSessions := globals.sessionStore + topics := mock_store.NewMockTopicsPersistenceInterface(ctrl) + store.Topics = topics + globals.usersUpdate = make(chan *UserCacheReq, 1) + globals.cluster = nil + globals.sessionStore = NewSessionStore(time.Minute) + t.Cleanup(func() { + store.Topics = oldTopics + globals.usersUpdate = oldUsersUpdate + globals.cluster = oldCluster + globals.sessionStore = oldSessions + }) + + sender := types.Uid(42) + receiver := types.Uid(99) + topic := sender.P2PName(receiver) + topics.EXPECT().GetSubs(types.TenantID(7), topic, nil).Return([]types.Subscription{ + {User: sender.String(), Topic: topic, ModeWant: types.ModeCFull, ModeGiven: types.ModeCFull}, + {User: receiver.String(), Topic: topic, ModeWant: types.ModeCFull, ModeGiven: types.ModeCFull}, + }, nil) + msg := &types.Message{ + ObjHeader: types.ObjHeader{CreatedAt: time.Now().UTC()}, + TenantID: 7, + Topic: topic, + SeqId: 12, + From: sender.String(), + Head: types.KVMap{"x-im-type": "text"}, + Content: map[string]any{"text": "forwarded"}, + } + + sendPushForHTTPForwardMessage(7, msg, true, msg.CreatedAt) + + select { + case req := <-globals.usersUpdate: + if req.PushRcpt == nil || len(req.PushRcpt.To) != 2 { + t.Fatalf("push recipients: %#v", req) + } + if req.PushRcpt.To[sender].ShouldIncrementUnreadCountInCache { + t.Fatal("sender should not get unread increment for a sender-read forward") + } + if !req.PushRcpt.To[receiver].ShouldIncrementUnreadCountInCache { + t.Fatal("receiver should get unread increment for forwarded message") + } + if req.PushRcpt.Payload.Topic != topic || req.PushRcpt.Payload.SeqId != 12 { + t.Fatalf("unexpected push payload: %#v", req.PushRcpt.Payload) + } + default: + t.Fatal("http forward message did not enqueue push") + } +} diff --git a/server/im_http.go b/server/im_http.go index aaa4409..8868813 100644 --- a/server/im_http.go +++ b/server/im_http.go @@ -26,6 +26,7 @@ import ( "github.com/tinode/chat/server/logs" "github.com/tinode/chat/server/media" "github.com/tinode/chat/server/moderation" + "github.com/tinode/chat/server/push" realnamealiyun "github.com/tinode/chat/server/realname/aliyun" "github.com/tinode/chat/server/store" "github.com/tinode/chat/server/store/types" @@ -2761,6 +2762,10 @@ func serveIMFavoriteForward(wrt http.ResponseWriter, req *http.Request) { results = append(results, result) continue } + msg.SeqId = saved.SeqId + if !saved.Duplicate { + broadcastHTTPForwardMessage(principal.TenantID, msg, saved.MarkedReadBySender) + } result := map[string]any{"target_topic": target, "favorite_id": favoriteID, "status": "sent", "seq": saved.SeqId, "client_mid": clientMid} mergeForwardCommentResult(result, commentResult) results = append(results, result) @@ -2895,7 +2900,9 @@ func serveIMMessageForward(wrt http.ResponseWriter, req *http.Request) { continue } forwarded.SeqId = saved.SeqId - broadcastHTTPForwardMessage(principal.TenantID, forwarded) + if !saved.Duplicate { + broadcastHTTPForwardMessage(principal.TenantID, forwarded, saved.MarkedReadBySender) + } item := map[string]any{ "sources": bundle["source_refs"], "target_topic": target, @@ -2986,7 +2993,9 @@ func serveIMMessageForward(wrt http.ResponseWriter, req *http.Request) { continue } forwarded.SeqId = saved.SeqId - broadcastHTTPForwardMessage(principal.TenantID, forwarded) + if !saved.Duplicate { + broadcastHTTPForwardMessage(principal.TenantID, forwarded, saved.MarkedReadBySender) + } item := map[string]any{ "source": map[string]any{"topic": sourceTopic, "seq": int64(source.Seq)}, "target_topic": target, @@ -5374,7 +5383,9 @@ func publishGroupAnnouncementTextMessage(ctx context.Context, tenantID types.Ten if saved != nil && saved.SeqId > 0 { msg.SeqId = saved.SeqId } - broadcastHTTPForwardMessage(tenantID, msg) + if saved == nil || !saved.Duplicate { + broadcastHTTPForwardMessage(tenantID, msg, saved == nil || saved.MarkedReadBySender) + } return nil } @@ -9483,7 +9494,9 @@ func saveForwardCommentMessage(ctx context.Context, principal *imPrincipal, targ } if saved != nil { msg.SeqId = saved.SeqId - broadcastHTTPForwardMessage(principal.TenantID, msg) + if !saved.Duplicate { + broadcastHTTPForwardMessage(principal.TenantID, msg, saved.MarkedReadBySender) + } result["comment_status"] = "sent" result["comment_seq"] = saved.SeqId return result @@ -9786,7 +9799,7 @@ func prepareHTTPForwardMessage(tenantID types.TenantID, msg *types.Message) erro return nil } -func broadcastHTTPForwardMessage(tenantID types.TenantID, msg *types.Message) { +func broadcastHTTPForwardMessage(tenantID types.TenantID, msg *types.Message, markedReadBySender bool) { if globals.hub == nil || msg == nil || msg.SeqId <= 0 || msg.Topic == "" { return } @@ -9813,6 +9826,7 @@ func broadcastHTTPForwardMessage(tenantID types.TenantID, msg *types.Message) { logs.Warn.Printf("im message forward: hub route queue full topic=%s seq=%d", msg.Topic, msg.SeqId) } broadcastHTTPForwardPresence(tenantID, msg, timestamp) + sendPushForHTTPForwardMessage(tenantID, msg, markedReadBySender, timestamp) } func broadcastHTTPForwardPresence(tenantID types.TenantID, msg *types.Message, timestamp time.Time) { @@ -9855,6 +9869,71 @@ func broadcastHTTPForwardPresence(tenantID types.TenantID, msg *types.Message, t } } +func sendPushForHTTPForwardMessage(tenantID types.TenantID, msg *types.Message, markedReadBySender bool, timestamp time.Time) { + if msg == nil || msg.SeqId <= 0 || msg.Topic == "" || store.Topics == nil { + return + } + subs, err := store.Topics.GetSubs(tenantID, msg.Topic, nil) + if err != nil { + logs.Warn.Printf("im message forward: load recipients topic=%s seq=%d: %v", msg.Topic, msg.SeqId, err) + return + } + contentType, _ := msg.Head["mime"].(string) + receipt := &push.Receipt{ + TenantID: tenantID, + To: make(map[types.Uid]push.Recipient, len(subs)), + Payload: push.Payload{ + What: push.ActMsg, + Silent: false, + Topic: msg.Topic, + From: types.ParseUid(msg.From).UserId(), + Timestamp: timestamp, + SeqId: msg.SeqId, + ContentType: contentType, + Content: msg.Content, + }, + } + fromUid := types.ParseUid(msg.From) + for _, sub := range subs { + uid := types.ParseUid(sub.User) + mode := sub.ModeWant & sub.ModeGiven + if uid.IsZero() || !mode.IsPresencer() || !mode.IsReader() { + continue + } + delivered := httpForwardDeliveredCount(tenantID, uid) + if uid == fromUid && delivered == 0 { + delivered = 1 + } + recipient := push.Recipient{ + Delivered: delivered, + Silent: types.GetTopicCat(msg.Topic) == types.TopicCatGrp && boolFromAny(groupPrivateMap(sub.Private)["mute_notifications"]), + ShouldIncrementUnreadCountInCache: uid != fromUid || !markedReadBySender, + } + receipt.To[uid] = recipient + } + if len(receipt.To) > 0 { + sendPush(receipt) + } +} + +func httpForwardDeliveredCount(tenantID types.TenantID, uid types.Uid) int { + delivered := 0 + if globals.sessionStore != nil { + globals.sessionStore.Range(func(_ string, sess *Session) bool { + if sess != nil && !sess.isCluster() && !sess.isProxy() && sess.tenantID == tenantID && sess.uid == uid { + delivered++ + } + return true + }) + } + if presence, err := store.GetUserPresence(tenantID, uid); err != nil { + logs.Warn.Printf("im message forward: read presence uid=%s: %v", uid.UserId(), err) + } else if presence != nil && presence.Online && presence.SessionCount > delivered { + delivered = presence.SessionCount + } + return delivered +} + func forwardedMessageHead(msg *types.Message, clientMid string) types.KVMap { head := types.KVMap{} if msg != nil {