diff --git a/server/im_http.go b/server/im_http.go index 07dcd08..66b9fca 100644 --- a/server/im_http.go +++ b/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 { diff --git a/server/im_http_test.go b/server/im_http_test.go index d631816..d3da936 100644 --- a/server/im_http_test.go +++ b/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()