diff --git a/server/datamodel.go b/server/datamodel.go index 41387c0..f9ba13a 100644 --- a/server/datamodel.go +++ b/server/datamodel.go @@ -863,6 +863,9 @@ type ServerComMessage struct { // Session ID to skip when sendng packet to sessions. Used to skip sending to original session. // Could be either empty. SkipSid string `json:"-"` + // PreserveTopicName prevents p2p/channel topic aliasing for server-generated + // messages which are already using the API-visible topic name. + PreserveTopicName bool `json:"-"` // User id affected by this message. uid types.Uid } @@ -874,14 +877,15 @@ func (src *ServerComMessage) copy() *ServerComMessage { return nil } dst := &ServerComMessage{ - TenantID: src.TenantID, - Id: src.Id, - RcptTo: src.RcptTo, - AsUser: src.AsUser, - Timestamp: src.Timestamp, - sess: src.sess, - SkipSid: src.SkipSid, - uid: src.uid, + TenantID: src.TenantID, + Id: src.Id, + RcptTo: src.RcptTo, + AsUser: src.AsUser, + Timestamp: src.Timestamp, + sess: src.sess, + SkipSid: src.SkipSid, + PreserveTopicName: src.PreserveTopicName, + uid: src.uid, } dst.Ctrl = src.Ctrl.copy() diff --git a/server/im_http.go b/server/im_http.go index 3516fe1..88d3bcb 100644 --- a/server/im_http.go +++ b/server/im_http.go @@ -4475,7 +4475,7 @@ func serveIMConversationMessages(wrt http.ResponseWriter, req *http.Request, pri writeIMRelationError(wrt, req, err) return } - if sub == nil || !(sub.ModeGiven&sub.ModeWant).IsReader() { + if sub == nil || !(sub.ModeGiven & sub.ModeWant).IsReader() { writeIMError(wrt, req, http.StatusForbidden, 30005, imReasonForbidden, "permission denied", nil) return } @@ -11277,10 +11277,11 @@ func broadcastIMConversationSystemMessage(tenantID types.TenantID, msg *types.Me timestamp = time.Now().UTC().Round(time.Millisecond) } out := &ServerComMessage{ - TenantID: tenantID, - RcptTo: msg.Topic, - AsUser: types.ParseUid(msg.From).UserId(), - Timestamp: timestamp, + TenantID: tenantID, + RcptTo: msg.Topic, + AsUser: types.ParseUid(msg.From).UserId(), + Timestamp: timestamp, + PreserveTopicName: true, Data: &MsgServerData{ Topic: msg.Topic, From: types.ParseUid(msg.From).UserId(), @@ -11296,7 +11297,7 @@ func broadcastIMConversationSystemMessage(tenantID types.TenantID, msg *types.Me logs.Warn.Printf("im system message: hub route queue full topic=%s seq=%d", msg.Topic, msg.SeqId) } deliverIMConversationSystemMessageToP2PSessions(tenantID, msg, out) - broadcastHTTPForwardPresence(tenantID, msg, timestamp) + broadcastIMConversationSystemPresence(tenantID, msg, timestamp) } func publishIMGroupInvitationNotification(tenantID types.TenantID, inv *types.GroupInvitation) { @@ -11335,16 +11336,15 @@ func deliverIMConversationSystemMessageToP2PSessions(tenantID types.TenantID, ms if _, ok := participants[session.uid]; !ok { return true } - viewTopic, err := types.P2PNameForUser(session.uid, msg.Topic) - if err != nil || viewTopic == "" { + if _, err := types.P2PNameForUser(session.uid, msg.Topic); err != nil { return true } - if session.getSub(msg.Topic) != nil || session.getSub(viewTopic) != nil { + if session.getSub(msg.Topic) != nil { return true } msgCopy := out.copy() msgCopy.RcptTo = session.uid.UserId() - msgCopy.Data.Topic = viewTopic + msgCopy.Data.Topic = msg.Topic if !session.queueOut(msgCopy) { logs.Warn.Printf("im system message: session queue full topic=%s seq=%d user=%s", msg.Topic, msg.SeqId, session.uid.UserId()) } @@ -11352,6 +11352,35 @@ func deliverIMConversationSystemMessageToP2PSessions(tenantID types.TenantID, ms }) } +func broadcastIMConversationSystemPresence(tenantID types.TenantID, msg *types.Message, timestamp time.Time) { + if globals.hub == nil || msg == nil || msg.SeqId <= 0 || msg.Topic == "" { + return + } + uid1, uid2, err := types.ParseP2P(msg.Topic) + if err != nil { + return + } + for _, uid := range []types.Uid{uid1, uid2} { + select { + case globals.hub.routeSrv <- &ServerComMessage{ + TenantID: tenantID, + RcptTo: uid.UserId(), + Timestamp: timestamp, + PreserveTopicName: true, + Pres: &MsgServerPres{ + Topic: "me", + What: "msg", + Src: msg.Topic, + SeqId: msg.SeqId, + SkipTopic: msg.Topic, + }, + }: + default: + logs.Warn.Printf("im system message: hub presence queue full topic=%s seq=%d user=%s", msg.Topic, msg.SeqId, uid.UserId()) + } + } +} + func systemNotificationIDFromPath(path string) string { const marker = "/v1/im/system-notifications/" idx := strings.Index(path, marker) diff --git a/server/im_http_test.go b/server/im_http_test.go index e9b6c45..c898d1d 100644 --- a/server/im_http_test.go +++ b/server/im_http_test.go @@ -7894,7 +7894,7 @@ func TestPublishIMConversationSystemNoticeDeliversToOnlineP2PSessions(t *testing if !ok || msg.Data == nil { t.Fatalf("unexpected outbound for %s: %#v", uid.UserId(), raw) } - if msg.Data.Topic != mustP2PNameForUser(t, uid, topic) || msg.Data.SeqId != 1 { + if msg.Data.Topic != topic || msg.Data.SeqId != 1 { t.Fatalf("unexpected p2p data for %s: %#v", uid.UserId(), msg.Data) } if msg.Data.Head["x-im-template"] != "friend_added" || msg.Data.Head["x-im-type"] != "system_notice" { @@ -7906,6 +7906,69 @@ func TestPublishIMConversationSystemNoticeDeliversToOnlineP2PSessions(t *testing } } +func TestPublishIMConversationSystemNoticePresenceUsesP2PTopic(t *testing.T) { + ctrl := gomock.NewController(t) + defer ctrl.Finish() + oldStore := store.Store + oldTopics := store.Topics + oldMessages := store.Messages + oldHub := globals.hub + mockStore := mock_store.NewMockPersistentStorageInterface(ctrl) + topics := mock_store.NewMockTopicsPersistenceInterface(ctrl) + messages := mock_store.NewMockMessagesPersistenceInterface(ctrl) + store.Store = mockStore + store.Topics = topics + store.Messages = messages + globals.hub = &Hub{routeSrv: make(chan *ServerComMessage, 4)} + defer func() { + store.Store = oldStore + store.Topics = oldTopics + store.Messages = oldMessages + globals.hub = oldHub + }() + mockStore.EXPECT().IsOpen().Return(true) + + applicant := types.Uid(1) + target := types.Uid(2) + topic := applicant.P2PName(target) + topics.EXPECT().Get(types.TenantID(7), topic).Return(&types.Topic{TenantID: 7, SeqId: 16}, nil) + messages.EXPECT().Save(types.TenantID(7), gomock.Any(), nil, false).DoAndReturn( + func(_ types.TenantID, msg *types.Message, _ []string, _ bool) (*types.MessageSaveResult, error) { + if msg.Topic != topic || msg.SeqId != 17 { + t.Fatalf("unexpected saved message: %#v", msg) + } + return &types.MessageSaveResult{SeqId: msg.SeqId}, nil + }) + + publishIMConversationSystemNotice(types.TenantID(7), topic, + "system_notice", "friend_added", "你们已成为好友,现在可以开始聊天了。", + []imSystemMessageActor{ + {UserID: applicant.UserId(), Click: map[string]any{"type": "user_profile", "target_id": applicant.UserId()}}, + {UserID: target.UserId(), Click: map[string]any{"type": "user_profile", "target_id": target.UserId()}}, + }, nil, map[string]any{"request_id": "frq_1"}) + + select { + case routed := <-globals.hub.routeSrv: + if routed.Data == nil || routed.Data.Topic != topic || !routed.PreserveTopicName { + t.Fatalf("unexpected system data broadcast: %#v", routed) + } + default: + t.Fatal("expected system data broadcast") + } + for _, uid := range []types.Uid{applicant, target} { + select { + case routed := <-globals.hub.routeSrv: + if routed.RcptTo != uid.UserId() || routed.Pres == nil || routed.Pres.Topic != "me" || + routed.Pres.Src != topic || routed.Pres.SeqId != 17 || routed.Pres.SkipTopic != topic || + !routed.PreserveTopicName { + t.Fatalf("unexpected system presence for %s: %#v", uid.UserId(), routed) + } + default: + t.Fatalf("expected system presence for %s", uid.UserId()) + } + } +} + func mustP2PNameForUser(t *testing.T, uid types.Uid, topic string) string { t.Helper() name, err := types.P2PNameForUser(uid, topic) diff --git a/server/topic.go b/server/topic.go index 1039249..c99d902 100644 --- a/server/topic.go +++ b/server/topic.go @@ -261,7 +261,7 @@ func (t *Topic) prepareBroadcastableMessage(msg *ServerComMessage, uid types.Uid return } - if (t.cat == types.TopicCatP2P && !uid.IsZero()) || (t.cat == types.TopicCatGrp && t.isChan) { + if !msg.PreserveTopicName && ((t.cat == types.TopicCatP2P && !uid.IsZero()) || (t.cat == types.TopicCatGrp && t.isChan)) { // P2P topics are exposed by their canonical p2p... name. Channel topics // may be presented as grpXXX or chnXXX. var topicName string