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

Improve message forward compatibility

Browse Source
This commit is contained in:
richjohnrambo
2026-09-03 10:37:42 +08:00
parent 1eee0179f1
commit 54e58763ff
3 changed files with 361 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/im_http.go
+184 -19
View File
@@ -425,13 +425,37 @@ type imFavoritesForwardRequest struct {
type imMessagesForwardRequest struct {
Mode string `json:"mode"`
Sources []struct {
Topic string `json:"topic"`
Seq int64 `json:"seq,string"`
Topic string `json:"topic"`
Seq imForwardSeq `json:"seq"`
} `json:"sources"`
TargetTopics []string `json:"target_topics"`
Comment string `json:"comment,omitempty"`
}
type imForwardSeq int64
func (s *imForwardSeq) UnmarshalJSON(data []byte) error {
var number int64
if err := json.Unmarshal(data, &number); err == nil {
*s = imForwardSeq(number)
return nil
}
var text string
if err := json.Unmarshal(data, &text); err != nil {
return err
}
text = strings.TrimSpace(text)
if text == "" {
return nil
}
parsed, err := strconv.ParseInt(text, 10, 64)
if err != nil {
return err
}
*s = imForwardSeq(parsed)
return nil
}
type imFriendRequestCreateRequest struct {
TargetUserID string `json:"target_user_id"`
Message string `json:"message,omitempty"`
@@ -3529,8 +3553,8 @@ func serveIMMessageForward(wrt http.ResponseWriter, req *http.Request) {
if mode == "" {
mode = "separate"
}
if mode != "separate" {
writeIMError(wrt, req, http.StatusNotImplemented, 30004, "NOT_IMPLEMENTED", "forward bundle is not implemented", map[string]any{"field": "mode"})
if mode != "separate" && mode != "bundle" {
writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "mode"})
return
}
if len(body.Sources) == 0 || len(body.Sources) > 100 || len(body.TargetTopics) == 0 || len(body.TargetTopics) > 20 {
@@ -3542,10 +3566,16 @@ func serveIMMessageForward(wrt http.ResponseWriter, req *http.Request) {
items := make([]map[string]any, 0, len(body.TargetTopics)*len(body.Sources))
failed := make([]map[string]any, 0)
for _, rawTarget := range body.TargetTopics {
target := strings.TrimSpace(rawTarget)
target, targetOK := normalizeIMForwardTopic(principal.Uid, rawTarget)
if target == "" {
for _, source := range body.Sources {
failed = append(failed, forwardFailurePayload(source.Topic, source.Seq, target, imReasonMalformedRequest))
failed = append(failed, forwardFailurePayload(source.Topic, int64(source.Seq), strings.TrimSpace(rawTarget), imReasonMalformedRequest))
}
continue
}
if !targetOK {
for _, source := range body.Sources {
failed = append(failed, forwardFailurePayload(source.Topic, int64(source.Seq), target, imReasonMalformedRequest))
}
continue
}
@@ -3556,14 +3586,44 @@ func serveIMMessageForward(wrt http.ResponseWriter, req *http.Request) {
}
if targetErr != nil {
for _, source := range body.Sources {
failed = append(failed, forwardFailurePayload(source.Topic, source.Seq, target, relationErrorReason(targetErr)))
failed = append(failed, forwardFailurePayload(source.Topic, int64(source.Seq), target, relationErrorReason(targetErr)))
}
continue
}
if mode == "bundle" {
bundle, sourceFailures := buildForwardBundle(principal, body.Sources, target, sourceReadable, body.Comment)
if len(sourceFailures) > 0 {
failed = append(failed, sourceFailures...)
continue
}
clientMid := "fwd_" + store.Store.GetUidString()
forwarded := &types.Message{
TenantID: principal.TenantID,
Topic: target,
From: principal.Uid.String(),
Head: forwardBundleMessageHead(clientMid),
Content: bundle,
}
saved, err := saveMessageMaybeKafka(req.Context(), principal.TenantID, forwarded, nil, true)
if err != nil {
failed = append(failed, forwardFailurePayload("", 0, target, relationErrorReason(err)))
continue
}
items = append(items, map[string]any{
"sources": bundle["source_refs"],
"target_topic": target,
"client_mid": clientMid,
"seq": saved.SeqId,
"status": "sent",
"mode": "bundle",
"bundle_id": bundle["bundle_id"],
})
continue
}
for _, source := range body.Sources {
sourceTopic := strings.TrimSpace(source.Topic)
if sourceTopic == "" || source.Seq <= 0 {
failed = append(failed, forwardFailurePayload(sourceTopic, source.Seq, target, imReasonMalformedRequest))
sourceTopic, sourceOK := normalizeIMForwardTopic(principal.Uid, source.Topic)
if sourceTopic == "" || int64(source.Seq) <= 0 || !sourceOK {
failed = append(failed, forwardFailurePayload(sourceTopic, int64(source.Seq), target, imReasonMalformedRequest))
continue
}
readErr, ok := sourceReadable[sourceTopic]
@@ -3572,20 +3632,20 @@ func serveIMMessageForward(wrt http.ResponseWriter, req *http.Request) {
sourceReadable[sourceTopic] = readErr
}
if readErr != nil {
failed = append(failed, forwardFailurePayload(sourceTopic, source.Seq, target, relationErrorReason(readErr)))
failed = append(failed, forwardFailurePayload(sourceTopic, int64(source.Seq), target, relationErrorReason(readErr)))
continue
}
msg, err := store.Messages.GetOne(principal.TenantID, sourceTopic, source.Seq)
msg, err := store.Messages.GetOne(principal.TenantID, sourceTopic, int64(source.Seq))
if err != nil {
failed = append(failed, forwardFailurePayload(sourceTopic, source.Seq, target, relationErrorReason(err)))
failed = append(failed, forwardFailurePayload(sourceTopic, int64(source.Seq), target, relationErrorReason(err)))
continue
}
if msg == nil {
failed = append(failed, forwardFailurePayload(sourceTopic, source.Seq, target, "NOT_FOUND"))
failed = append(failed, forwardFailurePayload(sourceTopic, int64(source.Seq), target, "NOT_FOUND"))
continue
}
if messageAvailability(*msg) != "available" {
failed = append(failed, forwardFailurePayload(sourceTopic, source.Seq, target, "MESSAGE_UNAVAILABLE"))
failed = append(failed, forwardFailurePayload(sourceTopic, int64(source.Seq), target, "MESSAGE_UNAVAILABLE"))
continue
}
clientMid := "fwd_" + store.Store.GetUidString()
@@ -3598,11 +3658,11 @@ func serveIMMessageForward(wrt http.ResponseWriter, req *http.Request) {
}
saved, err := saveMessageMaybeKafka(req.Context(), principal.TenantID, forwarded, nil, true)
if err != nil {
failed = append(failed, forwardFailurePayload(sourceTopic, source.Seq, target, relationErrorReason(err)))
failed = append(failed, forwardFailurePayload(sourceTopic, int64(source.Seq), target, relationErrorReason(err)))
continue
}
items = append(items, map[string]any{
"source": map[string]any{"topic": sourceTopic, "seq": source.Seq},
"source": map[string]any{"topic": sourceTopic, "seq": int64(source.Seq)},
"target_topic": target,
"client_mid": clientMid,
"seq": saved.SeqId,
@@ -5455,8 +5515,16 @@ func serveIMGroupMemberProfile(wrt http.ResponseWriter, req *http.Request, topic
if !ok {
return
}
if types.GetTopicCat(topic) != types.TopicCatGrp || memberID.IsZero() {
writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", nil)
if types.GetTopicCat(topic) != types.TopicCatGrp {
writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{
"field": "topic",
"expected": "group topic, e.g. grp...",
"actual": topic,
})
return
}
if memberID.IsZero() {
writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "user_id"})
return
}
selfSub, err := store.Subs.Get(principal.TenantID, topic, principal.Uid, false)
@@ -8791,6 +8859,94 @@ func forwardFailurePayload(topic string, seq int64, target, reason string) map[s
}
}
func normalizeIMForwardTopic(viewer types.Uid, topic string) (string, bool) {
topic = strings.TrimSpace(topic)
if topic == "" {
return "", false
}
if len(topic) < 3 {
return "", false
}
switch types.GetTopicCat(topic) {
case types.TopicCatMe:
peer := types.ParseUserId(topic)
if peer.IsZero() {
return "", false
}
p2p := viewer.P2PName(peer)
return p2p, p2p != ""
case types.TopicCatP2P, types.TopicCatGrp, types.TopicCatSlf:
return topic, true
default:
return "", false
}
}
func buildForwardBundle(principal *imPrincipal, sources []struct {
Topic string `json:"topic"`
Seq imForwardSeq `json:"seq"`
}, target string, sourceReadable map[string]error, comment string) (map[string]any, []map[string]any) {
items := make([]map[string]any, 0, len(sources))
preview := make([]map[string]any, 0, min(len(sources), 3))
sourceRefs := make([]map[string]any, 0, len(sources))
failures := make([]map[string]any, 0)
for _, source := range sources {
sourceTopic, sourceOK := normalizeIMForwardTopic(principal.Uid, source.Topic)
if sourceTopic == "" || int64(source.Seq) <= 0 || !sourceOK {
failures = append(failures, forwardFailurePayload(sourceTopic, int64(source.Seq), target, imReasonMalformedRequest))
continue
}
readErr, ok := sourceReadable[sourceTopic]
if !ok {
readErr = requireTopicReader(principal, sourceTopic)
sourceReadable[sourceTopic] = readErr
}
if readErr != nil {
failures = append(failures, forwardFailurePayload(sourceTopic, int64(source.Seq), target, relationErrorReason(readErr)))
continue
}
msg, err := store.Messages.GetOne(principal.TenantID, sourceTopic, int64(source.Seq))
if err != nil {
failures = append(failures, forwardFailurePayload(sourceTopic, int64(source.Seq), target, relationErrorReason(err)))
continue
}
if msg == nil {
failures = append(failures, forwardFailurePayload(sourceTopic, int64(source.Seq), target, "NOT_FOUND"))
continue
}
if messageAvailability(*msg) != "available" {
failures = append(failures, forwardFailurePayload(sourceTopic, int64(source.Seq), target, "MESSAGE_UNAVAILABLE"))
continue
}
snapshot := favoriteSnapshot(msg)
items = append(items, snapshot)
ref := map[string]any{"topic": sourceTopic, "seq": int64(source.Seq)}
sourceRefs = append(sourceRefs, ref)
if len(preview) < 3 {
preview = append(preview, map[string]any{
"from": msg.From,
"type": imMessageType(msg),
"summary": textSnippet(messageText(*msg), ""),
"ts": msg.CreatedAt.UTC().Format(time.RFC3339Nano),
})
}
}
if len(failures) > 0 {
return nil, failures
}
bundleID := "fbundle_" + store.Store.GetUidString()
return map[string]any{
"bundle_id": bundleID,
"title": "聊天记录",
"item_count": len(items),
"preview": preview,
"items": items,
"source_refs": sourceRefs,
"comment": strings.TrimSpace(comment),
"forwarded": true,
}, nil
}
func forwardedMessageHead(msg *types.Message, clientMid string) types.KVMap {
head := types.KVMap{}
if msg != nil {
@@ -8811,6 +8967,15 @@ func forwardedMessageHead(msg *types.Message, clientMid string) types.KVMap {
return head
}
func forwardBundleMessageHead(clientMid string) types.KVMap {
return types.KVMap{
"type": "forward_bundle",
"x-im-type": "forward_bundle",
"x-im-forwarded": true,
"x-im-client-mid": clientMid,
}
}
func forwardedMessageContent(msg *types.Message, comment string) any {
if strings.TrimSpace(comment) == "" {
if msg == nil {
server/im_http_test.go
+169
View File
@@ -3600,6 +3600,30 @@ func TestServeIMGroupMemberProfile(t *testing.T) {
}
}
func TestServeIMGroupMemberProfileRejectsUserTopic(t *testing.T) {
ctrl := gomock.NewController(t)
defer ctrl.Finish()
expectIMBearer(t, ctrl, "token-1", 7, 42)
memberID := types.Uid(42).UserId()
req := httptest.NewRequest(http.MethodGet, "/v1/im/groups/"+memberID+"/members/"+memberID+"/profile", nil)
req.Header.Set("Authorization", "Bearer token-1")
rec := httptest.NewRecorder()
serveIMGroupResource(rec, req)
if rec.Code != http.StatusBadRequest {
t.Fatalf("status: got %d, want %d; body=%s", rec.Code, http.StatusBadRequest, rec.Body.String())
}
body := decodeIMHTTPResponse(t, rec)
if body.Reason != imReasonMalformedRequest {
t.Fatalf("reason: got %q, want %q", body.Reason, imReasonMalformedRequest)
}
details := body.Details.(map[string]any)
if details["field"] != "topic" || details["actual"] != memberID {
t.Fatalf("unexpected details: %#v", details)
}
}
func TestServeIMGroupList(t *testing.T) {
ctrl := gomock.NewController(t)
defer ctrl.Finish()
@@ -4551,6 +4575,151 @@ func TestServeIMMessageForwardSeparateSavesMessages(t *testing.T) {
}
}
func TestServeIMMessageForwardAcceptsNumericSeqAndUserTarget(t *testing.T) {
ctrl := gomock.NewController(t)
defer ctrl.Finish()
mockStore := expectIMBearer(t, ctrl, "token-1", 7, 42)
mockStore.EXPECT().GetUidString().Return("mid-2")
oldSubs := store.Subs
oldMessages := store.Messages
subs := mock_store.NewMockSubsPersistenceInterface(ctrl)
messages := mock_store.NewMockMessagesPersistenceInterface(ctrl)
store.Subs = subs
store.Messages = messages
t.Cleanup(func() {
store.Subs = oldSubs
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)
messages.EXPECT().GetOne(types.TenantID(7), "grpSource", int64(60)).Return(&types.Message{
TenantID: 7,
Topic: "grpSource",
SeqId: 60,
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)
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 != targetTopic {
t.Fatalf("target topic: got %q, want %q", msg.Topic, targetTopic)
}
return &types.MessageSaveResult{SeqId: 22, MarkedReadBySender: true}, nil
})
req := httptest.NewRequest(http.MethodPost, "/v1/im/messages/forward", bytes.NewBufferString(`{"mode":"separate","sources":[{"topic":"grpSource","seq":60}],"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) != 22 {
t.Fatalf("unexpected forward item: %#v", got)
}
}
func TestServeIMMessageForwardBundleSavesForwardBundle(t *testing.T) {
ctrl := gomock.NewController(t)
defer ctrl.Finish()
mockStore := expectIMBearer(t, ctrl, "token-1", 7, 42)
gomock.InOrder(
mockStore.EXPECT().GetUidString().Return("bundle-1"),
mockStore.EXPECT().GetUidString().Return("mid-3"),
)
oldSubs := store.Subs
oldMessages := store.Messages
subs := mock_store.NewMockSubsPersistenceInterface(ctrl)
messages := mock_store.NewMockMessagesPersistenceInterface(ctrl)
store.Subs = subs
store.Messages = messages
t.Cleanup(func() {
store.Subs = oldSubs
store.Messages = oldMessages
})
subs.EXPECT().Get(types.TenantID(7), "grpTarget", types.Uid(42), false).Return(&types.Subscription{
TenantID: 7,
User: types.Uid(42).String(),
Topic: "grpTarget",
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)
messages.EXPECT().GetOne(types.TenantID(7), "grpSource", int64(60)).Return(&types.Message{
TenantID: 7,
Topic: "grpSource",
SeqId: 60,
From: types.Uid(99).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)
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 != "grpTarget" || msg.Head["x-im-type"] != "forward_bundle" {
t.Fatalf("unexpected bundle message: %#v", msg)
}
content := msg.Content.(map[string]any)
if content["bundle_id"] != "fbundle_bundle-1" || content["item_count"] != 1 {
t.Fatalf("unexpected bundle content: %#v", content)
}
return &types.MessageSaveResult{SeqId: 23, MarkedReadBySender: true}, nil
})
req := httptest.NewRequest(http.MethodPost, "/v1/im/messages/forward", bytes.NewBufferString(`{"mode":"bundle","sources":[{"topic":"grpSource","seq":60}],"target_topics":["grpTarget"]}`))
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["mode"] != "bundle" || got["bundle_id"] != "fbundle_bundle-1" || got["seq"].(float64) != 23 {
t.Fatalf("unexpected forward item: %#v", got)
}
}
func TestServeIMSecurityURLCheck(t *testing.T) {
ctrl := gomock.NewController(t)
defer ctrl.Finish()