From 7e8fc8237b4d29ac8bf5ab7fc66b8263d769a0eb Mon Sep 17 00:00:00 2001 From: richjohnrambo Date: Mon, 7 Sep 2026 15:12:14 +0800 Subject: [PATCH] fix im p0 backend issues --- server/calls.go | 141 ++++++++++++++++++++++++++++++- server/db/mysql/tenant_groups.go | 3 +- server/im_http.go | 17 +++- server/im_http_test.go | 32 ++++++- server/session.go | 55 ++++++++++++ server/session_test.go | 76 +++++++++++++++++ server/topic_test.go | 60 +++++++++++++ 7 files changed, 371 insertions(+), 13 deletions(-) diff --git a/server/calls.go b/server/calls.go index c3666ab..4e3538c 100644 --- a/server/calls.go +++ b/server/calls.go @@ -109,10 +109,18 @@ func normalizeCallInvite(head map[string]any, content any) (map[string]any, any, if state != "started" { return nil, nil, types.ErrMalformed } - call, ok := content.(map[string]any) - if !ok { - if typed, typedOK := content.(types.KVMap); typedOK { - call = map[string]any(typed) + call := callInvitePayloadMap(content) + if len(call) == 0 || !hasCallIdentity(call) { + headCall := callInvitePayloadMap(head) + if len(call) == 0 { + call = headCall + } else { + for key, value := range headCall { + if _, exists := call[key]; !exists { + call[key] = value + } + } + normalizeCallPayloadAliases(call) } } if len(call) == 0 { @@ -123,6 +131,16 @@ func normalizeCallInvite(head map[string]any, content any) (map[string]any, any, if callType == "" { callType, _ = call["media_type"].(string) } + if callType == "" { + callType, _ = call["type"].(string) + } + if callType == "" { + if aonly, _ := call["aonly"].(bool); aonly { + callType = constCallTypeAudio + } else { + callType = constCallTypeVideo + } + } callType = strings.ToLower(strings.TrimSpace(callType)) if strings.TrimSpace(callID) == "" || (callType != constCallTypeAudio && callType != constCallTypeVideo) { return nil, nil, types.ErrMalformed @@ -147,6 +165,121 @@ func normalizeCallInvite(head map[string]any, content any) (map[string]any, any, return normalizedHead, normalizedContent, nil } +func callInvitePayloadMap(content any) map[string]any { + call := anyMap(content) + if len(call) == 0 { + return nil + } + normalizeCallPayloadAliases(call) + if hasCallIdentity(call) { + return call + } + if draftyCall := draftyVideoCallPayload(call); len(draftyCall) > 0 { + for key, value := range call { + if _, exists := draftyCall[key]; !exists && key != "fmt" && key != "ent" && key != "txt" { + draftyCall[key] = value + } + } + normalizeCallPayloadAliases(draftyCall) + return draftyCall + } + return call +} + +func anyMap(value any) map[string]any { + if typed, ok := value.(map[string]any); ok { + return typed + } + if typed, ok := value.(types.KVMap); ok { + return map[string]any(typed) + } + return nil +} + +func normalizeCallPayloadAliases(call map[string]any) { + if call == nil { + return + } + if _, ok := call["callId"]; !ok { + for _, key := range []string{"call_id", "id"} { + if value, exists := call[key]; exists { + call["callId"] = value + break + } + } + } + if _, ok := call["callType"]; !ok { + for _, key := range []string{"call_type", "mediaType", "media_type"} { + if value, exists := call[key]; exists { + call["callType"] = value + break + } + } + } +} + +func hasCallIdentity(call map[string]any) bool { + if call == nil { + return false + } + callID, _ := call["callId"].(string) + if strings.TrimSpace(callID) == "" { + return false + } + callType, _ := call["callType"].(string) + if strings.TrimSpace(callType) != "" { + return true + } + if mediaType, _ := call["media_type"].(string); strings.TrimSpace(mediaType) != "" { + return true + } + if _, ok := call["aonly"].(bool); ok { + return true + } + return false +} + +func draftyVideoCallPayload(doc map[string]any) map[string]any { + for _, raw := range anySlice(doc["ent"]) { + entity := anyMap(raw) + if entityType, _ := entity["tp"].(string); entityType != "VC" { + continue + } + if data := anyMap(entity["data"]); len(data) > 0 { + return data + } + return map[string]any{} + } + for _, raw := range anySlice(doc["fmt"]) { + format := anyMap(raw) + if formatType, _ := format["tp"].(string); formatType == "VC" { + return map[string]any{} + } + } + return nil +} + +func anySlice(value any) []any { + switch typed := value.(type) { + case []any: + return typed + case []map[string]any: + items := make([]any, 0, len(typed)) + for _, item := range typed { + items = append(items, item) + } + return items + case []types.KVMap: + items := make([]any, 0, len(typed)) + for _, item := range typed { + items = append(items, item) + } + return items + default: + return nil + } +} + func callPayloadString(payload any, key string) string { data, ok := payload.(map[string]any) if !ok { diff --git a/server/db/mysql/tenant_groups.go b/server/db/mysql/tenant_groups.go index 00cb322..3dd1e71 100644 --- a/server/db/mysql/tenant_groups.go +++ b/server/db/mysql/tenant_groups.go @@ -445,7 +445,8 @@ func (a *adapter) TenantGroupInvitationAccept(id t.TenantID, invitationID string return &inv, nil } var deletedAt sql.NullTime - err = tx.QueryRowxContext(ctx, "SELECT DeletedAt FROM subscriptions WHERE Topic=? AND User=?", inv.Topic, uid.String()).Scan(&deletedAt) + err = tx.QueryRowxContext(ctx, "SELECT deletedat FROM im_subscriptions WHERE topic=? AND userid=?", + inv.Topic, decodeUID(uid)).Scan(&deletedAt) if err != nil && err != sql.ErrNoRows { return nil, err } diff --git a/server/im_http.go b/server/im_http.go index baeb314..b50781d 100644 --- a/server/im_http.go +++ b/server/im_http.go @@ -3570,6 +3570,7 @@ func serveIMFavoriteBatchCreate(wrt http.ResponseWriter, req *http.Request) { failed = append(failed, map[string]any{"topic": sourceTopic, "seq": src.Seq, "reason": relationErrorReason(err)}) continue } + idempotencyKey := favoriteItemIdempotencyKey(req.Header.Get("Idempotency-Key"), sourceTopic, src.Seq) fav := &types.FavoriteItem{ TenantID: principal.TenantID, FavoriteID: "fav_" + store.Store.GetUidString(), @@ -3580,7 +3581,7 @@ func serveIMFavoriteBatchCreate(wrt http.ResponseWriter, req *http.Request) { Snapshot: favoriteSnapshot(msg), AttachmentRefs: favoriteAttachmentRefs(msg), Availability: messageAvailability(*msg), - IdempotencyKey: strings.TrimSpace(req.Header.Get("Idempotency-Key")), + IdempotencyKey: idempotencyKey, CollectedAt: now, CreatedAt: now, UpdatedAt: now, @@ -4021,13 +4022,13 @@ func serveIMMessageContext(wrt http.ResponseWriter, req *http.Request) { writeIMError(wrt, req, http.StatusNotFound, 30004, imReasonMalformedRequest, "message context route not found", nil) return } - principal, _, ok := authenticateTopicReader(wrt, req, topic) + principal, normalizedTopic, ok := authenticateTopicReader(wrt, req, topic) if !ok { return } before := parseIMLimit(req.URL.Query().Get("before"), 20, 100) after := parseIMLimit(req.URL.Query().Get("after"), 20, 100) - msgs, err := store.Messages.GetAll(principal.TenantID, topic, principal.Uid, &types.QueryOpt{Since: seq - int64(before), Before: seq + int64(after) + 1, Limit: before + after + 1}) + msgs, err := store.Messages.GetAll(principal.TenantID, normalizedTopic, principal.Uid, &types.QueryOpt{Since: seq - int64(before), Before: seq + int64(after) + 1, Limit: before + after + 1}) if err != nil { writeIMRelationError(wrt, req, err) return @@ -4036,7 +4037,7 @@ func serveIMMessageContext(wrt http.ResponseWriter, req *http.Request) { for _, msg := range msgs { items = append(items, messageContextPayload(msg)) } - writeIMSuccess(wrt, req, map[string]any{"topic": topic, "anchor_seq": seq, "items": items}) + writeIMSuccess(wrt, req, map[string]any{"topic": normalizedTopic, "anchor_seq": seq, "items": items}) } func serveIMTopicMedia(wrt http.ResponseWriter, req *http.Request) { @@ -9353,6 +9354,14 @@ func normalizeIMFavoriteSourceTopic(viewer types.Uid, topic string) (string, boo return normalizeIMUserVisibleTopic(viewer, topic) } +func favoriteItemIdempotencyKey(requestKey, topic string, seq int64) string { + requestKey = strings.TrimSpace(requestKey) + if requestKey == "" { + return "" + } + return stableIMEventID("favorite", requestKey, strings.TrimSpace(topic), seq) +} + func normalizeIMUserVisibleTopic(viewer types.Uid, topic string) (string, bool) { topic = strings.TrimSpace(topic) if topic == "" { diff --git a/server/im_http_test.go b/server/im_http_test.go index 7e5620f..d8f9e06 100644 --- a/server/im_http_test.go +++ b/server/im_http_test.go @@ -4870,6 +4870,17 @@ func TestServeIMFavoriteBatchCreateCountsExisting(t *testing.T) { } } +func TestFavoriteItemIdempotencyKeyIncludesMessageIdentity(t *testing.T) { + first := favoriteItemIdempotencyKey("same-key", "grpAlpha", 1) + second := favoriteItemIdempotencyKey("same-key", "grpAlpha", 2) + if first == "" || second == "" || first == second { + t.Fatalf("favorite item idempotency key must be stable per source message: %q %q", first, second) + } + if favoriteItemIdempotencyKey("", "grpAlpha", 1) != "" { + t.Fatal("empty request idempotency key should stay empty") + } +} + func TestServeIMFavoriteBatchCreateRejectsUnreadableSource(t *testing.T) { ctrl := gomock.NewController(t) defer ctrl.Finish() @@ -5694,10 +5705,10 @@ func TestServeIMTopicReadActionsNormalizeP2PUserTopic(t *testing.T) { oldStore := store.Store mockStore := mock_store.NewMockPersistentStorageInterface(ctrl) tokenAuth := mock_auth.NewMockAuthHandler(ctrl) - mockStore.EXPECT().GetLogicalAuthHandler("token").Return(tokenAuth).Times(2) - tokenAuth.EXPECT().IsInitialized().Return(true).Times(2) + mockStore.EXPECT().GetLogicalAuthHandler("token").Return(tokenAuth).Times(3) + tokenAuth.EXPECT().IsInitialized().Return(true).Times(3) tokenAuth.EXPECT().Authenticate(auth.AuthContext{}, []byte("token-1")). - Return(&auth.Rec{TenantID: 7, Uid: actor, AuthLevel: auth.LevelAuth}, nil, nil).Times(2) + Return(&auth.Rec{TenantID: 7, Uid: actor, AuthLevel: auth.LevelAuth}, nil, nil).Times(3) store.Store = mockStore oldSubs := store.Subs oldMessages := store.Messages @@ -5717,10 +5728,11 @@ func TestServeIMTopicReadActionsNormalizeP2PUserTopic(t *testing.T) { Topic: normalizedTopic, ModeWant: types.ModeCPublic, ModeGiven: types.ModeCPublic, - }, nil).Times(2) + }, nil).Times(3) textMsg := types.Message{TenantID: 7, SeqId: 9, Topic: normalizedTopic, From: actor.String(), Head: types.KVMap{"type": "text"}, Content: map[string]any{"text": "好的"}} imageMsg := types.Message{TenantID: 7, SeqId: 10, Topic: normalizedTopic, From: actor.String(), Head: types.KVMap{"type": "image"}, Content: map[string]any{"file_id": "file-1", "mime": "image/png"}} messages.EXPECT().Search(types.TenantID(7), normalizedTopic, actor, "好的", "text", "", 50).Return([]types.Message{textMsg}, "", nil) + messages.EXPECT().GetAll(types.TenantID(7), normalizedTopic, actor, gomock.Any()).Return([]types.Message{textMsg}, nil) messages.EXPECT().Search(types.TenantID(7), normalizedTopic, actor, "", "image", "", 50).Return([]types.Message{imageMsg}, "", nil) searchReq := httptest.NewRequest(http.MethodGet, "/v1/im/topics/"+peer.UserId()+"/messages/search?q=%E5%A5%BD%E7%9A%84&type=text", nil) @@ -5731,6 +5743,18 @@ func TestServeIMTopicReadActionsNormalizeP2PUserTopic(t *testing.T) { t.Fatalf("search status: got %d body=%s", searchRec.Code, searchRec.Body.String()) } + contextReq := httptest.NewRequest(http.MethodGet, "/v1/im/topics/"+peer.UserId()+"/messages/9/context", nil) + contextReq.Header.Set("Authorization", "Bearer token-1") + contextRec := httptest.NewRecorder() + serveIMTopicMessageAction(contextRec, contextReq) + if contextRec.Code != http.StatusOK { + t.Fatalf("context status: got %d body=%s", contextRec.Code, contextRec.Body.String()) + } + contextData := decodeIMHTTPResponse(t, contextRec).Data.(map[string]any) + if contextData["topic"] != normalizedTopic { + t.Fatalf("context topic was not normalized: %#v", contextData) + } + mediaReq := httptest.NewRequest(http.MethodGet, "/v1/im/topics/"+peer.UserId()+"/media?type=image", nil) mediaReq.Header.Set("Authorization", "Bearer token-1") mediaRec := httptest.NewRecorder() diff --git a/server/session.go b/server/session.go index 6a180ea..f7ced8c 100644 --- a/server/session.go +++ b/server/session.go @@ -1286,6 +1286,10 @@ func (s *Session) onLogin(msgID string, timestamp time.Time, rec *auth.Rec, miss // Record deviceId used in this session if s.deviceID != "" { + isNewDevice, deviceErr := isNewLoginDevice(s.tenantID, rec.Uid, s.deviceID) + if deviceErr != nil { + logs.Warn.Println("s.onLogin: failed to check known device", rec.Uid, deviceErr, s.sid) + } if err := store.Devices.Update(s.tenantID, rec.Uid, "", &types.DeviceDef{ TenantID: s.tenantID, DeviceId: s.deviceID, @@ -1294,6 +1298,8 @@ func (s *Session) onLogin(msgID string, timestamp time.Time, rec *auth.Rec, miss Lang: s.lang, }); err != nil { logs.Warn.Println("failed to update device record", err) + } else if isNewDevice { + createLoginDeviceSecurityNotice(s.tenantID, rec.Uid, s.deviceID, s.platf, s.userAgent, timestamp) } } if !s.uid.IsZero() { @@ -1310,6 +1316,55 @@ func (s *Session) onLogin(msgID string, timestamp time.Time, rec *auth.Rec, miss return reply } +func isNewLoginDevice(tenantID types.TenantID, uid types.Uid, deviceID string) (bool, error) { + if store.Devices == nil || strings.TrimSpace(deviceID) == "" || uid.IsZero() { + return false, nil + } + devices, _, err := store.Devices.GetAll(tenantID, uid) + if err != nil { + return false, err + } + for _, device := range devices[uid] { + if device.DeviceId == deviceID { + return false, nil + } + } + return true, nil +} + +func createLoginDeviceSecurityNotice(tenantID types.TenantID, uid types.Uid, deviceID, platform, userAgent string, at time.Time) { + if store.Security == nil || uid.IsZero() || strings.TrimSpace(deviceID) == "" { + return + } + eventID := stableIMEventID("security.login_device", uid.UserId(), deviceID) + if err := createIMSystemNotification(tenantID, uid, eventID, "security.login_device", + "新设备登录提醒", loginDeviceSecurityNoticeSummary(platform, userAgent), + map[string]any{ + "scene": "login_device", + "device_id": deviceID, + "platform": strings.TrimSpace(platform), + "user_agent": strings.TrimSpace(userAgent), + "login_at": at.UTC().Format(time.RFC3339Nano), + }); err != nil { + logs.Warn.Println("s.onLogin: failed to create login device notification", uid, err) + } +} + +func loginDeviceSecurityNoticeSummary(platform, userAgent string) string { + platform = strings.TrimSpace(platform) + if platform == "" { + platform = "未知设备" + } + userAgent = strings.TrimSpace(userAgent) + if userAgent == "" { + return platform + " 登录了你的账号" + } + if len([]rune(userAgent)) > 80 { + userAgent = string([]rune(userAgent)[:80]) + } + return platform + " 登录了你的账号：" + userAgent +} + func currentNodeID() string { if globals.cluster != nil && globals.cluster.thisNodeName != "" { return globals.cluster.thisNodeName diff --git a/server/session_test.go b/server/session_test.go index 1e9e09e..e09b939 100644 --- a/server/session_test.go +++ b/server/session_test.go @@ -19,6 +19,48 @@ type testTenantStore struct { err error } +type testLoginDeviceStore struct { + devices map[types.Uid][]types.DeviceDef + updates int + err error +} + +func (s *testLoginDeviceStore) Update(tenantID types.TenantID, uid types.Uid, oldDeviceID string, dev *types.DeviceDef) error { + s.updates++ + if s.err != nil { + return s.err + } + if dev != nil { + dev.TenantID = tenantID + s.devices[uid] = append(s.devices[uid], *dev) + } + return nil +} + +func (s *testLoginDeviceStore) GetAll(_ types.TenantID, uids ...types.Uid) (map[types.Uid][]types.DeviceDef, int, error) { + if s.err != nil { + return nil, 0, s.err + } + result := make(map[types.Uid][]types.DeviceDef) + count := 0 + for _, uid := range uids { + result[uid] = append([]types.DeviceDef(nil), s.devices[uid]...) + count += len(result[uid]) + } + return result, count, nil +} + +func (s *testLoginDeviceStore) Delete(_ types.TenantID, uid types.Uid, deviceID string) error { + devices := s.devices[uid] + for i, device := range devices { + if device.DeviceId == deviceID { + s.devices[uid] = append(devices[:i], devices[i+1:]...) + return nil + } + } + return types.ErrNotFound +} + func (s testTenantStore) GetByCode(string) (*types.Tenant, error) { return s.tenant, s.err } @@ -46,6 +88,40 @@ func test_makeSession(uid types.Uid) *Session { } } +func TestLoginDeviceSecurityNoticeOnlyForNewDevice(t *testing.T) { + uid := types.Uid(42) + devices := &testLoginDeviceStore{devices: map[types.Uid][]types.DeviceDef{ + uid: {{TenantID: 7, DeviceId: "known-device", Platform: "ios"}}, + }} + security := &testSecurityStore{} + oldDevices := store.Devices + oldSecurity := store.Security + store.Devices = devices + store.Security = security + t.Cleanup(func() { + store.Devices = oldDevices + store.Security = oldSecurity + }) + + known, err := isNewLoginDevice(7, uid, "known-device") + if err != nil || known { + t.Fatalf("known device detected as new: new=%v err=%v", known, err) + } + newDevice, err := isNewLoginDevice(7, uid, "new-device") + if err != nil || !newDevice { + t.Fatalf("new device was not detected: new=%v err=%v", newDevice, err) + } + + createLoginDeviceSecurityNotice(7, uid, "new-device", "ios", "UnitTestUA", time.Date(2026, 9, 6, 12, 0, 0, 0, time.UTC)) + if len(security.notifications) != 1 { + t.Fatalf("expected one security notification, got %d", len(security.notifications)) + } + notification := security.notifications[0] + if notification.EventType != "security.login_device" || notification.Title == "" || notification.Action["device_id"] != "new-device" { + t.Fatalf("unexpected login device notification: %#v", notification) + } +} + func TestDispatchHello(t *testing.T) { oldTenants := store.Tenants store.Tenants = testTenantStore{tenant: &types.Tenant{ diff --git a/server/topic_test.go b/server/topic_test.go index 38364df..5b09a03 100644 --- a/server/topic_test.go +++ b/server/topic_test.go @@ -493,6 +493,66 @@ func TestHandleBroadcastCall(t *testing.T) { } } +func TestHandleBroadcastDraftyCallInvite(t *testing.T) { + numUsers := 2 + helper := TopicTestHelper{} + helper.setUp(t, numUsers, types.TopicCatP2P, "p2p-test" /*attach=*/, true) + oldThirdParty := store.ThirdPartyConfig + globals.iceServers = []iceServer{{Username: "dummy"}} + store.ThirdPartyConfig = testThirdPartyConfigStore{ + "audio_call_enabled": "1", + "video_call_enabled": "1", + } + helper.topic.lastID = 5 + defer func() { + globals.iceServers = nil + store.ThirdPartyConfig = oldThirdParty + helper.tearDown() + }() + helper.mm.EXPECT().Save(types.TenantID(1), gomock.Any(), gomock.Any(), gomock.Any()). + Return(&types.MessageSaveResult{SeqId: 1, MarkedReadBySender: true}, nil) + + from := helper.uids[0].UserId() + msg := &ClientComMessage{ + AsUser: from, + Original: from, + Pub: &MsgClientPub{ + Topic: "p2p", + Head: map[string]any{"webrtc": "started", "mime": "text/x-drafty"}, + Content: map[string]any{ + "txt": " ", + "ent": []any{map[string]any{ + "tp": "VC", + "data": map[string]any{ + "call_id": "call-drafty-1", + "aonly": true, + "state": "started", + }, + }}, + }, + NoEcho: true, + }, + sess: helper.sessions[0], + } + helper.topic.handleClientMsg(msg) + helper.finish() + + if errorMsgs, hasError := helper.hubMessages["__ERROR__"]; hasError { + t.Fatal(errorMsgs[0].Ctrl.Text) + } + if len(helper.results[1].messages) != 1 { + t.Fatalf("callee: expected 1 message, got %d", len(helper.results[1].messages)) + } + data := helper.results[1].messages[0].(*ServerComMessage).Data + content := data.Content.(map[string]any) + if content["callId"] != "call-drafty-1" || content["callType"] != "audio" || content["video"] != false { + t.Fatalf("unexpected normalized Drafty call content: %#v", content) + } + if data.Head["x-im-type"] != "call" || data.Head["mime"] != "text/x-drafty" { + t.Fatalf("unexpected normalized Drafty call head: %#v", data.Head) + } +} + func TestHandleBroadcastCallRequiresStructuredInvite(t *testing.T) { numUsers := 2 helper := TopicTestHelper{}