From 2e4e1f0df3e9a74b099469872ab28ffee0138546 Mon Sep 17 00:00:00 2001 From: richjohnrambo Date: Thu, 3 Sep 2026 11:11:48 +0800 Subject: [PATCH] Fix HTTP message forward delivery --- server/im_http.go | 70 ++++++++++++++++++++++++++ server/im_http_test.go | 109 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 179 insertions(+) diff --git a/server/im_http.go b/server/im_http.go index 10d586f..fae1bab 100644 --- a/server/im_http.go +++ b/server/im_http.go @@ -3604,11 +3604,17 @@ func serveIMMessageForward(wrt http.ResponseWriter, req *http.Request) { Head: forwardBundleMessageHead(clientMid), Content: bundle, } + if err := prepareHTTPForwardMessage(principal.TenantID, forwarded); err != nil { + failed = append(failed, forwardFailurePayload("", 0, target, relationErrorReason(err))) + continue + } saved, err := saveMessageMaybeKafka(req.Context(), principal.TenantID, forwarded, nil, true) if err != nil { failed = append(failed, forwardFailurePayload("", 0, target, relationErrorReason(err))) continue } + forwarded.SeqId = saved.SeqId + broadcastHTTPForwardMessage(principal.TenantID, forwarded) items = append(items, map[string]any{ "sources": bundle["source_refs"], "target_topic": target, @@ -3656,11 +3662,17 @@ func serveIMMessageForward(wrt http.ResponseWriter, req *http.Request) { Head: forwardedMessageHead(msg, clientMid), Content: forwardedMessageContent(msg, body.Comment), } + if err := prepareHTTPForwardMessage(principal.TenantID, forwarded); err != nil { + failed = append(failed, forwardFailurePayload(sourceTopic, int64(source.Seq), target, relationErrorReason(err))) + continue + } saved, err := saveMessageMaybeKafka(req.Context(), principal.TenantID, forwarded, nil, true) if err != nil { failed = append(failed, forwardFailurePayload(sourceTopic, int64(source.Seq), target, relationErrorReason(err))) continue } + forwarded.SeqId = saved.SeqId + broadcastHTTPForwardMessage(principal.TenantID, forwarded) items = append(items, map[string]any{ "source": map[string]any{"topic": sourceTopic, "seq": int64(source.Seq)}, "target_topic": target, @@ -8947,6 +8959,64 @@ func buildForwardBundle(principal *imPrincipal, sources []struct { }, nil } +func prepareHTTPForwardMessage(tenantID types.TenantID, msg *types.Message) error { + if msg == nil || msg.TenantID != tenantID || strings.TrimSpace(msg.Topic) == "" { + return types.ErrMalformed + } + if msg.SeqId > 0 { + return nil + } + if !kafkaEnabled() && globals.hub == nil { + return nil + } + if store.Topics == nil { + return types.ErrMalformed + } + topicRecord, err := store.Topics.Get(tenantID, msg.Topic) + if err != nil { + return err + } + if topicRecord == nil { + return types.ErrNotFound + } + msg.SeqId = topicRecord.SeqId + 1 + if msg.CreatedAt.IsZero() { + msg.ObjHeader.CreatedAt = time.Now().UTC().Round(time.Millisecond) + } + if msg.UpdatedAt.IsZero() { + msg.ObjHeader.UpdatedAt = msg.CreatedAt + } + return nil +} + +func broadcastHTTPForwardMessage(tenantID types.TenantID, msg *types.Message) { + if globals.hub == nil || msg == nil || msg.SeqId <= 0 || msg.Topic == "" { + return + } + timestamp := msg.CreatedAt + if timestamp.IsZero() { + timestamp = time.Now().UTC().Round(time.Millisecond) + } + select { + case globals.hub.routeSrv <- &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, + }, + }: + default: + logs.Warn.Printf("im message forward: hub route queue full topic=%s seq=%d", msg.Topic, msg.SeqId) + } +} + func forwardedMessageHead(msg *types.Message, clientMid string) types.KVMap { head := types.KVMap{} if msg != nil { diff --git a/server/im_http_test.go b/server/im_http_test.go index f4eafc4..bed2165 100644 --- a/server/im_http_test.go +++ b/server/im_http_test.go @@ -2,6 +2,7 @@ package main import ( "bytes" + "context" "crypto/hmac" "crypto/md5" "encoding/base64" @@ -4645,6 +4646,114 @@ func TestServeIMMessageForwardAcceptsNumericSeqAndUserTarget(t *testing.T) { } } +func TestServeIMMessageForwardKafkaAssignsSeqAndBroadcasts(t *testing.T) { + ctrl := gomock.NewController(t) + defer ctrl.Finish() + mockStore := expectIMBearer(t, ctrl, "token-1", 7, 42) + mockStore.EXPECT().GetUidString().Return("mid-3") + + oldKafkaBus := kafkaBus + oldPublisher := publishKafkaMessagePersist + oldHub := globals.hub + oldSubs := store.Subs + oldTopics := store.Topics + oldMessages := store.Messages + subs := mock_store.NewMockSubsPersistenceInterface(ctrl) + topics := mock_store.NewMockTopicsPersistenceInterface(ctrl) + messages := mock_store.NewMockMessagesPersistenceInterface(ctrl) + kafkaBus = &kafkaRuntime{} + globals.hub = &Hub{routeSrv: make(chan *ServerComMessage, 1)} + store.Subs = subs + store.Topics = topics + store.Messages = messages + t.Cleanup(func() { + kafkaBus = oldKafkaBus + publishKafkaMessagePersist = oldPublisher + globals.hub = oldHub + store.Subs = oldSubs + store.Topics = oldTopics + store.Messages = oldMessages + }) + + targetUser := types.Uid(99) + targetTopic := types.Uid(42).P2PName(targetUser) + subs.EXPECT().Get(types.TenantID(7), targetTopic, types.Uid(42), false).Return(&types.Subscription{ + TenantID: 7, + User: types.Uid(42).String(), + Topic: targetTopic, + ModeWant: types.ModeCPublic, + ModeGiven: types.ModeCPublic, + }, nil) + subs.EXPECT().Get(types.TenantID(7), "grpSource", types.Uid(42), false).Return(&types.Subscription{ + TenantID: 7, + User: types.Uid(42).String(), + Topic: "grpSource", + ModeWant: types.ModeCPublic, + ModeGiven: types.ModeCPublic, + }, nil) + topics.EXPECT().Get(types.TenantID(7), targetTopic).Return(&types.Topic{ + TenantID: 7, + SeqId: 34, + }, nil) + messages.EXPECT().GetOne(types.TenantID(7), "grpSource", int64(14)).Return(&types.Message{ + TenantID: 7, + Topic: "grpSource", + SeqId: 14, + From: targetUser.String(), + Head: types.KVMap{"x-im-type": "text"}, + Content: map[string]any{"text": "hello"}, + ObjHeader: types.ObjHeader{ + CreatedAt: time.Date(2026, 8, 13, 9, 0, 0, 0, time.UTC), + }, + }, nil) + + var publishedTopic string + var publishedKey string + var publishedPayload kafkaMessagePersistPayload + publishKafkaMessagePersist = func(_ context.Context, topicName, key string, envelope kafkaEnvelope) error { + publishedTopic = topicName + publishedKey = key + if err := json.Unmarshal(envelope.Payload, &publishedPayload); err != nil { + t.Fatalf("failed to unmarshal kafka payload: %v", err) + } + return nil + } + + req := httptest.NewRequest(http.MethodPost, "/v1/im/messages/forward", bytes.NewBufferString(`{"mode":"separate","sources":[{"topic":"grpSource","seq":14}],"target_topics":["`+targetUser.UserId()+`"]}`)) + req.Header.Set("Authorization", "Bearer token-1") + rec := httptest.NewRecorder() + serveIMMessageForward(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("forward status: got %d body=%s", rec.Code, rec.Body.String()) + } + data := decodeIMHTTPResponse(t, rec).Data.(map[string]any) + items := data["items"].([]any) + if len(items) != 1 { + t.Fatalf("unexpected forward response: %#v", data) + } + got := items[0].(map[string]any) + if got["target_topic"] != targetTopic || got["seq"].(float64) != 35 { + t.Fatalf("unexpected forward item: %#v", got) + } + if failed := data["failed"].([]any); len(failed) != 0 { + t.Fatalf("unexpected failed items: %#v", failed) + } + if publishedTopic != "message_persist" || publishedKey != targetTopic { + t.Fatalf("unexpected kafka publish topic=%q key=%q", publishedTopic, publishedKey) + } + if publishedPayload.Message.Topic != targetTopic || publishedPayload.Message.SeqId != 35 { + t.Fatalf("unexpected kafka message: %#v", publishedPayload.Message) + } + select { + case delivered := <-globals.hub.routeSrv: + if delivered.RcptTo != targetTopic || delivered.Data == nil || delivered.Data.SeqId != 35 { + t.Fatalf("unexpected broadcast message: %#v", delivered) + } + default: + t.Fatal("expected forwarded message to be broadcast") + } +} + func TestServeIMMessageForwardBundleSavesForwardBundle(t *testing.T) { ctrl := gomock.NewController(t) defer ctrl.Finish()