pro/im_server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix HTTP message forward delivery

Browse Source
This commit is contained in:
richjohnrambo
2026-09-03 11:11:48 +08:00
parent 54e58763ff
commit 2e4e1f0df3
2 changed files with 179 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/im_http.go
+70
View File
@@ -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 {
server/im_http_test.go
+109
View File
@@ -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()