fix: deliver friendship system notices live

server/im_http.go

@@ -10771,7 +10771,7 @@ func publishIMConversationSystemNotice(tenantID types.TenantID, topic, msgType,
 		content["extra"] = extra
 	}
 	now := time.Now().UTC().Round(time.Millisecond)
-	_, err = store.Messages.Save(tenantID, &types.Message{
+	msg := &types.Message{
 		ObjHeader: types.ObjHeader{CreatedAt: now, UpdatedAt: now},
 		TenantID:  tenantID,
 		SeqId:     topicRecord.SeqId + 1,
@@ -10784,10 +10784,76 @@ func publishIMConversationSystemNotice(tenantID types.TenantID, topic, msgType,
 			"x-im-from":     "system",
 		},
 		Content: content,
-	}, nil, false)
+	}
+	_, err = store.Messages.Save(tenantID, msg, nil, false)
 	if err != nil && !errors.Is(err, types.ErrDuplicate) {
 		logs.Warn.Printf("im system message: failed to save %s for topic %s: %v", template, topic, err)
+		return
 	}
+	broadcastIMConversationSystemMessage(tenantID, msg, now)
+}
+
+func broadcastIMConversationSystemMessage(tenantID types.TenantID, msg *types.Message, timestamp time.Time) {
+	if globals.hub == nil || msg == nil || msg.SeqId <= 0 || strings.TrimSpace(msg.Topic) == "" {
+		return
+	}
+	if timestamp.IsZero() {
+		timestamp = time.Now().UTC().Round(time.Millisecond)
+	}
+	out := &ServerComMessage{
+		TenantID:  tenantID,
+		RcptTo:    msg.Topic,
+		AsUser:    types.ParseUid(msg.From).UserId(),
+		Timestamp: timestamp,
+		Data: &MsgServerData{
+			Topic:     msg.Topic,
+			From:      types.ParseUid(msg.From).UserId(),
+			Timestamp: timestamp,
+			SeqId:     msg.SeqId,
+			Head:      msg.Head,
+			Content:   msg.Content,
+		},
+	}
+	select {
+	case globals.hub.routeSrv <- out:
+	default:
+		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)
+}
+
+func deliverIMConversationSystemMessageToP2PSessions(tenantID types.TenantID, msg *types.Message, out *ServerComMessage) {
+	if globals.sessionStore == nil || msg == nil || out == nil || out.Data == nil {
+		return
+	}
+	uid1, uid2, err := types.ParseP2P(msg.Topic)
+	if err != nil {
+		return
+	}
+	participants := map[types.Uid]struct{}{uid1: {}, uid2: {}}
+	globals.sessionStore.Range(func(_ string, session *Session) bool {
+		if session == nil || session.tenantID != tenantID {
+			return true
+		}
+		if _, ok := participants[session.uid]; !ok {
+			return true
+		}
+		viewTopic, err := types.P2PNameForUser(session.uid, msg.Topic)
+		if err != nil || viewTopic == "" {
+			return true
+		}
+		if session.getSub(msg.Topic) != nil || session.getSub(viewTopic) != nil {
+			return true
+		}
+		msgCopy := out.copy()
+		msgCopy.RcptTo = session.uid.UserId()
+		msgCopy.Data.Topic = viewTopic
+		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())
+		}
+		return true
+	})
 }
 
 func systemNotificationIDFromPath(path string) string {

server/im_http_test.go

@@ -7156,6 +7156,94 @@ func TestServeIMFriendRequestCreate(t *testing.T) {
 	}
 }
 
+func TestPublishIMConversationSystemNoticeDeliversToOnlineP2PSessions(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+	mockStore := mock_store.NewMockPersistentStorageInterface(ctrl)
+	mockStore.EXPECT().IsOpen().Return(true).AnyTimes()
+	oldStore := store.Store
+	oldTopics := store.Topics
+	oldMessages := store.Messages
+	oldHub := globals.hub
+	oldSessions := globals.sessionStore
+	store.Store = mockStore
+	topics := mock_store.NewMockTopicsPersistenceInterface(ctrl)
+	messages := mock_store.NewMockMessagesPersistenceInterface(ctrl)
+	store.Topics = topics
+	store.Messages = messages
+	globals.hub = &Hub{routeSrv: make(chan *ServerComMessage, 10)}
+	globals.sessionStore = NewSessionStore(time.Minute)
+	t.Cleanup(func() {
+		store.Store = oldStore
+		store.Topics = oldTopics
+		store.Messages = oldMessages
+		globals.hub = oldHub
+		globals.sessionStore = oldSessions
+	})
+
+	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: 0}, 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 != 1 {
+				t.Fatalf("unexpected saved message: %#v", msg)
+			}
+			return &types.MessageSaveResult{SeqId: msg.SeqId}, nil
+		})
+
+	for _, uid := range []types.Uid{applicant, target} {
+		session := &Session{
+			sid:      uid.UserId() + "-session",
+			proto:    WEBSOCK,
+			uid:      uid,
+			tenantID: types.TenantID(7),
+			send:     make(chan any, 2),
+			subs:     make(map[string]*Subscription),
+		}
+		globals.sessionStore.sessCache[session.sid] = session
+	}
+
+	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"})
+
+	if len(globals.hub.routeSrv) != 3 {
+		t.Fatalf("expected routed topic data and two presence updates, got %d", len(globals.hub.routeSrv))
+	}
+	for _, uid := range []types.Uid{applicant, target} {
+		session := globals.sessionStore.Get(uid.UserId() + "-session")
+		select {
+		case raw := <-session.send:
+			msg, ok := raw.(*ServerComMessage)
+			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 {
+				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" {
+				t.Fatalf("unexpected system notice head: %#v", msg.Data.Head)
+			}
+		default:
+			t.Fatalf("online user %s did not receive system notice", uid.UserId())
+		}
+	}
+}
+
+func mustP2PNameForUser(t *testing.T, uid types.Uid, topic string) string {
+	t.Helper()
+	name, err := types.P2PNameForUser(uid, topic)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return name
+}
+
 func TestServeIMFriendRequestCreateHonorsGroupPrivateChatSetting(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	defer ctrl.Finish()