pro/im_server
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix: preserve p2p topic for IM system notices

Browse Source
This commit is contained in:
richjohnrambo
2026-09-13 16:00:58 +08:00
parent ccbd385d2f
commit f1279ac607
4 changed files with 116 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/datamodel.go
+12 -8
View File
@@ -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()
server/im_http.go
+39 -10
View File
@@ -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)
server/im_http_test.go
+64 -1
View File
@@ -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)
server/topic.go
+1 -1
View File
@@ -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