diff --git a/server/platform/services/sharedchannel/sync_recv.go b/server/platform/services/sharedchannel/sync_recv.go index 01ea4cc372..eb90303c64 100644 --- a/server/platform/services/sharedchannel/sync_recv.go +++ b/server/platform/services/sharedchannel/sync_recv.go @@ -263,7 +263,13 @@ func (scs *Service) processSyncMessage(c request.CTX, syncMsg *model.SyncMsg, rc } for _, status := range syncMsg.Statuses { - scs.app.SaveAndBroadcastStatus(status) + if err := scs.upsertSyncUserStatus(c, status, rc); err != nil { + scs.server.Log().Log(mlog.LvlSharedChannelServiceError, "Error upserting sync user status", + mlog.String("remote", rc.Name), + mlog.String("user_id", status.UserId), + mlog.Err(err)) + syncResp.StatusErrors = append(syncResp.StatusErrors, status.UserId) + } } // Process membership changes after users have been synced @@ -754,6 +760,26 @@ func (scs *Service) upsertSyncAcknowledgement(acknowledgement *model.PostAcknowl return savedAcknowledgement, retErr } +func (scs *Service) upsertSyncUserStatus(rctx request.CTX, status *model.Status, rc *model.RemoteCluster) error { + user, err := scs.server.GetStore().User().Get(rctx.Context(), status.UserId) + if err != nil { + return fmt.Errorf("error getting user when syncing status: %w", err) + } + + if user.GetRemoteID() != rc.RemoteId { + scs.server.Log().Log(mlog.LvlSharedChannelServiceError, "RemoteID mismatch sync'ing user status", + mlog.String("remote", rc.Name), + mlog.String("user_id", status.UserId), + mlog.String("user_remote_id", user.GetRemoteID()), + ) + return fmt.Errorf("error updating user status: %w", ErrRemoteIDMismatch) + } + + scs.app.SaveAndBroadcastStatus(status) + + return nil +} + // transformMentionsOnReceive transforms mentions in received posts using explicit mentionTransforms. func (scs *Service) transformMentionsOnReceive(rctx request.CTX, post *model.Post, targetChannel *model.Channel, rc *model.RemoteCluster, mentionTransforms map[string]string) { if post.Message == "" || len(mentionTransforms) == 0 { diff --git a/server/platform/services/sharedchannel/sync_recv_test.go b/server/platform/services/sharedchannel/sync_recv_test.go new file mode 100644 index 0000000000..f3b2702bab --- /dev/null +++ b/server/platform/services/sharedchannel/sync_recv_test.go @@ -0,0 +1,130 @@ +// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved. +// See LICENSE.txt for license information. + +package sharedchannel + +import ( + "testing" + + "github.com/mattermost/mattermost/server/v8/channels/store" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" + + "github.com/mattermost/mattermost/server/public/model" + "github.com/mattermost/mattermost/server/public/shared/mlog" + "github.com/mattermost/mattermost/server/public/shared/request" + "github.com/mattermost/mattermost/server/v8/channels/store/storetest/mocks" +) + +func TestUpsertSyncUserStatus(t *testing.T) { + setup := func(remoteID string, user *model.User) (*Service, *MockAppIface, *model.Status, *model.RemoteCluster) { + var userID string + if user == nil { + userID = model.NewId() + } else { + userID = user.Id + } + + status := &model.Status{ + UserId: userID, + Status: model.StatusDnd, + } + remoteCluster := &model.RemoteCluster{ + RemoteId: remoteID, + Name: "test-remote", + } + + mockUserStore := &mocks.UserStore{} + if user == nil { + mockUserStore.On("Get", mockTypeContext, mock.Anything).Return(nil, store.NewErrNotFound("User", userID)) + } else { + mockUserStore.On("Get", mockTypeContext, user.Id).Return(user, nil) + } + + mockStore := &mocks.Store{} + mockStore.On("User").Return(mockUserStore) + + logger := mlog.CreateConsoleTestLogger(t) + + mockServer := &MockServerIface{} + mockServer.On("GetStore").Return(mockStore) + mockServer.On("Log").Return(logger) + + mockApp := &MockAppIface{} + mockApp.On("SaveAndBroadcastStatus", status).Return() + + scs := &Service{ + server: mockServer, + app: mockApp, + } + + return scs, mockApp, status, remoteCluster + } + + t.Run("should broadcast changes to a remote user's status", func(t *testing.T) { + remoteID := model.NewId() + user := &model.User{ + Id: model.NewId(), + RemoteId: model.NewPointer(remoteID), + } + + scs, mockApp, status, remoteCluster := setup(remoteID, user) + + err := scs.upsertSyncUserStatus(request.TestContext(t), status, remoteCluster) + + require.NoError(t, err) + mockApp.AssertCalled(t, "SaveAndBroadcastStatus", status) + }) + + t.Run("should return an error when the user doesn't exist locally", func(t *testing.T) { + remoteID := model.NewId() + var user *model.User + + scs, mockApp, status, remoteCluster := setup(remoteID, user) + + rctx := request.TestContext(t) + err := scs.upsertSyncUserStatus(rctx, status, remoteCluster) + + require.Error(t, err) + assert.Contains(t, err.Error(), "error getting user when syncing status") + mockApp.AssertNotCalled(t, "SaveAndBroadcastStatus") + }) + + t.Run("should return an error when attempting to sync a local user", func(t *testing.T) { + remoteID := model.NewId() + user := &model.User{ + Id: model.NewId(), + RemoteId: nil, + } + + scs, mockApp, status, remoteCluster := setup(remoteID, user) + + rctx := request.TestContext(t) + err := scs.upsertSyncUserStatus(rctx, status, remoteCluster) + + require.Error(t, err) + assert.ErrorIs(t, err, ErrRemoteIDMismatch) + assert.Contains(t, err.Error(), "error updating user status") + mockApp.AssertNotCalled(t, "SaveAndBroadcastStatus") + }) + + t.Run("should return an error when attempting to sync a user from a different remote", func(t *testing.T) { + remoteID := model.NewId() + anotherRemoteID := model.NewId() + user := &model.User{ + Id: model.NewId(), + RemoteId: model.NewPointer(anotherRemoteID), + } + + scs, mockApp, status, remoteCluster := setup(remoteID, user) + + rctx := request.TestContext(t) + err := scs.upsertSyncUserStatus(rctx, status, remoteCluster) + + require.Error(t, err) + assert.ErrorIs(t, err, ErrRemoteIDMismatch) + assert.Contains(t, err.Error(), "error updating user status") + mockApp.AssertNotCalled(t, "SaveAndBroadcastStatus") + }) +}