fix: fanout p2p system notices to users

This commit is contained in:
richjohnrambo
2026-09-15 17:58:55 +08:00
parent e29b5d3589
commit 563ae9ee5e
2 changed files with 49 additions and 5 deletions
+44 -4
View File
@@ -11626,7 +11626,9 @@ 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) logs.Warn.Printf("im system message: hub route queue full topic=%s seq=%d", msg.Topic, msg.SeqId)
} }
deliverIMConversationSystemMessageToP2PSessions(tenantID, msg, out) deliverIMConversationSystemMessageToP2PSessions(tenantID, msg, out)
publishRealtimeTopicMessage(msg.Topic, out) if !publishRealtimeP2PConversationSystemMessage(tenantID, msg.Topic, out) {
publishRealtimeTopicMessage(msg.Topic, out)
}
broadcastIMConversationSystemPresence(tenantID, msg, timestamp) broadcastIMConversationSystemPresence(tenantID, msg, timestamp)
} }
@@ -11672,9 +11674,10 @@ func deliverIMConversationSystemMessageToP2PSessions(tenantID types.TenantID, ms
if session.getSub(msg.Topic) != nil { if session.getSub(msg.Topic) != nil {
return true return true
} }
msgCopy := out.copy() msgCopy, ok := p2pConversationSystemMessageForUser(out, msg.Topic, session.uid)
msgCopy.RcptTo = session.uid.UserId() if !ok {
msgCopy.Data.Topic = msg.Topic return true
}
if !session.queueOut(msgCopy) { 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()) logs.Warn.Printf("im system message: session queue full topic=%s seq=%d user=%s", msg.Topic, msg.SeqId, session.uid.UserId())
} }
@@ -11682,6 +11685,43 @@ func deliverIMConversationSystemMessageToP2PSessions(tenantID types.TenantID, ms
}) })
} }
func publishRealtimeP2PConversationSystemMessage(tenantID types.TenantID, topic string, out *ServerComMessage) bool {
uid1, uid2, err := types.ParseP2P(topic)
if err != nil {
return false
}
for _, uid := range []types.Uid{uid1, uid2} {
msgCopy, ok := p2pConversationSystemMessageForUser(out, topic, uid)
if !ok {
continue
}
publishRealtimeUserMessage(uid, msgCopy)
}
return true
}
func p2pConversationSystemMessageForUser(out *ServerComMessage, topic string, uid types.Uid) (*ServerComMessage, bool) {
if out == nil || uid.IsZero() {
return nil, false
}
viewTopic, err := types.P2PNameForUser(uid, topic)
if err != nil || viewTopic == "" {
return nil, false
}
msgCopy := out.copy()
msgCopy.RcptTo = uid.UserId()
if msgCopy.Data != nil {
msgCopy.Data.Topic = viewTopic
}
if msgCopy.Info != nil {
msgCopy.Info.Topic = viewTopic
}
if msgCopy.Pres != nil {
msgCopy.Pres.Topic = viewTopic
}
return msgCopy, true
}
func broadcastIMConversationSystemPresence(tenantID types.TenantID, msg *types.Message, timestamp time.Time) { func broadcastIMConversationSystemPresence(tenantID types.TenantID, msg *types.Message, timestamp time.Time) {
if globals.hub == nil || msg == nil || msg.SeqId <= 0 || msg.Topic == "" { if globals.hub == nil || msg == nil || msg.SeqId <= 0 || msg.Topic == "" {
return return
+5 -1
View File
@@ -8349,7 +8349,11 @@ func TestPublishIMConversationSystemNoticeDeliversToOnlineP2PSessions(t *testing
if !ok || msg.Data == nil { if !ok || msg.Data == nil {
t.Fatalf("unexpected outbound for %s: %#v", uid.UserId(), raw) t.Fatalf("unexpected outbound for %s: %#v", uid.UserId(), raw)
} }
if msg.Data.Topic != topic || msg.Data.SeqId != 1 { wantTopic, err := types.P2PNameForUser(uid, topic)
if err != nil {
t.Fatal(err)
}
if msg.Data.Topic != wantTopic || msg.Data.SeqId != 1 {
t.Fatalf("unexpected p2p data for %s: %#v", uid.UserId(), msg.Data) 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" { if msg.Data.Head["x-im-template"] != "friend_added" || msg.Data.Head["x-im-type"] != "system_notice" {