From 085a48c3d9f3ab3e29c7b1efd3c9a2bbd29a93f9 Mon Sep 17 00:00:00 2001 From: richjohnrambo Date: Thu, 10 Sep 2026 10:27:30 +0800 Subject: [PATCH] feat: publish text message for group announcements --- server/im_http.go | 52 ++++++++++++++++++++++++++++++++++++++++++ server/im_http_test.go | 18 +++++++++++++++ 2 files changed, 70 insertions(+) diff --git a/server/im_http.go b/server/im_http.go index edd7168..2d89c23 100644 --- a/server/im_http.go +++ b/server/im_http.go @@ -5598,6 +5598,11 @@ func serveIMGroupAnnouncements(wrt http.ResponseWriter, req *http.Request, topic writeIMRelationError(wrt, req, err) return } + if err = publishGroupAnnouncementTextMessage(req.Context(), principal.TenantID, topic, principal.Uid, announcement); err != nil { + logs.Warn.Println("im group announcement: failed to publish text message", principal.TenantID, topic, announcement.AnnouncementID, err) + writeIMRelationError(wrt, req, err) + return + } dispatchIMGroupEvents(principal.TenantID) writeIMSuccess(wrt, req, groupAnnouncementPayloadWithPermissions(topic, announcement, selfSub)) case strings.HasPrefix(tail, "announcements/") && strings.HasSuffix(tail, "/read") && req.Method == http.MethodPost: @@ -5620,6 +5625,53 @@ func serveIMGroupAnnouncements(wrt http.ResponseWriter, req *http.Request, topic } } +func publishGroupAnnouncementTextMessage(ctx context.Context, tenantID types.TenantID, topic string, author types.Uid, ann *types.GroupAnnouncement) error { + if ann == nil || strings.TrimSpace(topic) == "" || author.IsZero() { + return types.ErrMalformed + } + content := strings.TrimSpace(ann.Content) + if content == "" { + return types.ErrMalformed + } + if store.Messages == nil || store.Topics == nil { + return types.ErrMalformed + } + topicRecord, err := store.Topics.Get(tenantID, topic) + if err != nil { + return err + } + if topicRecord == nil { + return types.ErrNotFound + } + now := ann.PublishedAt + if now.IsZero() { + now = time.Now().UTC().Round(time.Millisecond) + } + msg := &types.Message{ + ObjHeader: types.ObjHeader{CreatedAt: now, UpdatedAt: now}, + TenantID: tenantID, + SeqId: topicRecord.SeqId + 1, + Topic: topic, + From: author.String(), + Head: types.KVMap{ + "mime": "application/json", + "x-im-schema": 1, + "x-im-type": "text", + "x-im-client-mid": "group_announcement_" + strings.TrimSpace(ann.AnnouncementID), + }, + Content: map[string]any{"text": "群公告\n" + content}, + } + saved, err := saveMessageMaybeKafka(ctx, tenantID, msg, nil, true) + if err != nil { + return err + } + if saved != nil && saved.SeqId > 0 { + msg.SeqId = saved.SeqId + } + broadcastHTTPForwardMessage(tenantID, msg) + return nil +} + func serveIMGroupGovernance(wrt http.ResponseWriter, req *http.Request, topic string) { principal, selfSub, ok := requireIMGroupMember(wrt, req, topic, true) if !ok { diff --git a/server/im_http_test.go b/server/im_http_test.go index 8962c0a..4887702 100644 --- a/server/im_http_test.go +++ b/server/im_http_test.go @@ -4668,6 +4668,7 @@ func TestServeIMGroupAnnouncementGovernanceJoinRequest(t *testing.T) { defer ctrl.Finish() oldStore := store.Store oldGroupBusiness := store.GroupBusiness + oldMessages := store.Messages mockStore := mock_store.NewMockPersistentStorageInterface(ctrl) tokenAuth := mock_auth.NewMockAuthHandler(ctrl) mockStore.EXPECT().GetLogicalAuthHandler("token").Return(tokenAuth).Times(3) @@ -4681,19 +4682,36 @@ func TestServeIMGroupAnnouncementGovernanceJoinRequest(t *testing.T) { oldSubs := store.Subs topics := mock_store.NewMockTopicsPersistenceInterface(ctrl) subs := mock_store.NewMockSubsPersistenceInterface(ctrl) + messages := mock_store.NewMockMessagesPersistenceInterface(ctrl) store.Topics = topics store.Subs = subs + store.Messages = messages groupStore := &testGroupBusinessStore{} store.GroupBusiness = groupStore defer func() { store.Store = oldStore store.Topics = oldTopics store.Subs = oldSubs + store.Messages = oldMessages store.GroupBusiness = oldGroupBusiness }() selfSub := &types.Subscription{TenantID: 7, User: types.Uid(42).String(), Topic: "grpAlpha", ModeWant: types.ModeCFull, ModeGiven: types.ModeCFull} subs.EXPECT().Get(types.TenantID(7), "grpAlpha", types.Uid(42), false).Return(selfSub, nil).Times(2) subs.EXPECT().Get(types.TenantID(7), "grpAlpha", types.Uid(42), false).Return(nil, nil) + topics.EXPECT().Get(types.TenantID(7), "grpAlpha").Return(&types.Topic{TenantID: 7, SeqId: 10}, nil) + messages.EXPECT().Save(types.TenantID(7), gomock.Any(), gomock.Nil(), true).DoAndReturn( + func(_ types.TenantID, msg *types.Message, _ []string, _ bool) (*types.MessageSaveResult, error) { + if msg.Topic != "grpAlpha" || msg.From != types.Uid(42).String() || msg.SeqId != 11 { + t.Fatalf("unexpected announcement message: %#v", msg) + } + if msg.Head["mime"] != "application/json" || msg.Head["x-im-type"] != "text" || msg.Head["x-im-client-mid"] != "group_announcement_ann_ann-1" { + t.Fatalf("unexpected announcement message head: %#v", msg.Head) + } + if msg.Content.(map[string]any)["text"] != "群公告\n公告" { + t.Fatalf("unexpected announcement message content: %#v", msg.Content) + } + return &types.MessageSaveResult{SeqId: msg.SeqId, MarkedReadBySender: true}, nil + }) annReq := httptest.NewRequest(http.MethodPut, "/v1/im/groups/grpAlpha/announcements/current", bytes.NewBufferString(`{"content":"公告","version":1}`)) annReq.Header.Set("Authorization", "Bearer token-1")