feat: publish text message for group announcements

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 {

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")