diff --git a/server/channels/app/shared_channel_sync_self_referential_utils_test.go b/server/channels/app/shared_channel_sync_self_referential_utils_test.go index b310246bc0..d0944f6622 100644 --- a/server/channels/app/shared_channel_sync_self_referential_utils_test.go +++ b/server/channels/app/shared_channel_sync_self_referential_utils_test.go @@ -42,6 +42,7 @@ type SelfReferentialSyncHandler struct { service *sharedchannel.Service selfCluster *model.RemoteCluster syncMessageCount *int32 + SimulateUnshared bool // When true, always return ErrChannelIsNotShared for sync messages // Callbacks for capturing sync data OnIndividualSync func(userId string, messageNumber int32) @@ -78,6 +79,14 @@ func (h *SelfReferentialSyncHandler) HandleRequest(w http.ResponseWriter, r *htt err := json.Unmarshal(body, &frame) if err == nil { + // Simulate the remote having unshared if configured to do so + if h.SimulateUnshared && frame.Msg.Topic == "sharedchannel_sync" { + // Return HTTP error instead of JSON error response + w.WriteHeader(http.StatusBadRequest) + _, _ = w.Write([]byte("channel is no longer shared")) + return + } + // Process the message to update cursor response := &remotecluster.Response{} processErr := h.service.OnReceiveSyncMessageForTesting(frame.Msg, h.selfCluster, response) diff --git a/server/channels/app/shared_channel_test.go b/server/channels/app/shared_channel_test.go index 6a76fe5af8..a9d09183ab 100644 --- a/server/channels/app/shared_channel_test.go +++ b/server/channels/app/shared_channel_test.go @@ -4,12 +4,17 @@ package app import ( + "net/http" + "net/http/httptest" + "strings" "testing" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "github.com/mattermost/mattermost/server/public/model" + "github.com/mattermost/mattermost/server/v8/platform/services/remotecluster" + "github.com/mattermost/mattermost/server/v8/platform/services/sharedchannel" ) func setupSharedChannels(tb testing.TB) *TestHelper { @@ -96,3 +101,458 @@ func TestApp_CheckCanInviteToSharedChannel(t *testing.T) { assert.Error(t, err, "invalid channel should not allow invites") }) } + +// TestApp_RemoteUnsharing tests the functionality where a shared channel is unshared on one side and triggers an unshare on the remote cluster. +// This test uses a self-referential approach where a server syncs with itself through real HTTP communication. +func TestApp_RemoteUnsharing(t *testing.T) { + th := setupSharedChannels(t).InitBasic() + defer th.TearDown() + + ss := th.App.Srv().Store() + + // Get the shared channel service and cast to concrete type + scsInterface := th.App.Srv().GetSharedChannelSyncService() + service, ok := scsInterface.(*sharedchannel.Service) + require.True(t, ok, "Expected sharedchannel.Service concrete type") + + // Ensure services are active + err := service.Start() + require.NoError(t, err) + + rcService := th.App.Srv().GetRemoteClusterService() + if rcService != nil { + _ = rcService.Start() + } + + t.Run("remote-initiated unshare with single remote", func(t *testing.T) { + EnsureCleanState(t, th, ss) + + var syncHandler *SelfReferentialSyncHandler + + // Create a test HTTP server that acts as the "remote" cluster + testServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if syncHandler != nil { + syncHandler.HandleRequest(w, r) + } else { + writeOKResponse(w) + } + })) + defer testServer.Close() + + // Create a self-referential remote cluster + selfCluster := &model.RemoteCluster{ + RemoteId: model.NewId(), + Name: "test-remote", + DisplayName: "Test Remote", + SiteURL: testServer.URL, + Token: model.NewId(), + CreateAt: model.GetMillis(), + LastPingAt: model.GetMillis(), + CreatorId: th.BasicUser.Id, + RemoteTeamId: model.NewId(), + } + selfCluster, err = ss.RemoteCluster().Save(selfCluster) + require.NoError(t, err) + + // Initialize sync handler + syncHandler = NewSelfReferentialSyncHandler(t, service, selfCluster) + + // Create a shared channel + channel := th.CreateChannel(th.Context, th.BasicTeam) + sc := &model.SharedChannel{ + ChannelId: channel.Id, + TeamId: channel.TeamId, + Home: true, + ReadOnly: false, + ShareName: channel.Name, + ShareDisplayName: channel.DisplayName, + SharePurpose: channel.Purpose, + ShareHeader: channel.Header, + CreatorId: th.BasicUser.Id, + RemoteId: "", + } + + _, err = th.App.ShareChannel(th.Context, sc) + require.NoError(t, err) + + // Share the channel with the remote + scr := &model.SharedChannelRemote{ + Id: model.NewId(), + ChannelId: channel.Id, + CreatorId: th.BasicUser.Id, + IsInviteAccepted: true, + IsInviteConfirmed: true, + RemoteId: selfCluster.RemoteId, + LastPostUpdateAt: model.GetMillis(), + } + _, err = ss.SharedChannel().SaveRemote(scr) + require.NoError(t, err) + + // Get post count before "remote-initiated unshare" + postsBeforeRemove, appErr := th.App.GetPostsPage(model.GetPostsOptions{ + ChannelId: channel.Id, + Page: 0, + PerPage: 10, + }) + require.Nil(t, appErr) + postCountBefore := len(postsBeforeRemove.Posts) + + // Verify the channel is initially shared + err = th.App.checkChannelIsShared(channel.Id) + require.NoError(t, err, "Channel should be shared initially") + + // Step 1: Verify the channel is initially shared + err = th.App.checkChannelIsShared(channel.Id) + require.NoError(t, err, "Channel should be shared initially") + + // Step 2: Create a sync message that would be sent to the remote + syncMsg := model.NewSyncMsg(channel.Id) + syncMsg.Posts = []*model.Post{{ + Id: model.NewId(), + ChannelId: channel.Id, + UserId: th.BasicUser.Id, + Message: "Test message after remote unshare", + CreateAt: model.GetMillis(), + }} + + // Step 3: Simulate receiving ErrChannelIsNotShared from the remote + // This directly tests the error handling logic without async complexity + service.HandleChannelNotSharedErrorForTesting(syncMsg, selfCluster) + + // Step 4: Verify the channel is no longer shared locally + err = th.App.checkChannelIsShared(channel.Id) + assert.Error(t, err, "Channel should no longer be shared after error handling") + + // Verify a system message was posted to inform users the channel is no longer shared + postsAfterRemove, appErr := th.App.GetPostsPage(model.GetPostsOptions{ + ChannelId: channel.Id, + Page: 0, + PerPage: 10, + }) + require.Nil(t, appErr) + + // Expected: only one notification post when a remote is removed + assert.Equal(t, postCountBefore+1, len(postsAfterRemove.Posts), "There should be one new post") + + // Find and verify the system message content + var systemPost *model.Post + for _, p := range postsAfterRemove.Posts { + if p.Type == model.PostTypeSystemGeneric { + systemPost = p + break + } + } + require.NotNil(t, systemPost, "A system post should be created") + assert.Equal(t, "This channel is no longer shared.", systemPost.Message, "Message should match unshare message") + }) + + t.Run("remote-initiated unshare with multiple remotes", func(t *testing.T) { + EnsureCleanState(t, th, ss) + + var syncHandler1, syncHandler2 *SelfReferentialSyncHandler + + // Create test HTTP servers for both remotes + testServer1 := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if syncHandler1 != nil { + syncHandler1.HandleRequest(w, r) + } else { + writeOKResponse(w) + } + })) + defer testServer1.Close() + + testServer2 := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if syncHandler2 != nil { + syncHandler2.HandleRequest(w, r) + } else { + writeOKResponse(w) + } + })) + defer testServer2.Close() + + // Create two self-referential remote clusters + selfCluster1 := &model.RemoteCluster{ + RemoteId: model.NewId(), + Name: "test-remote-1", + DisplayName: "Test Remote 1", + SiteURL: testServer1.URL, + Token: model.NewId(), + CreateAt: model.GetMillis(), + LastPingAt: model.GetMillis(), + CreatorId: th.BasicUser.Id, + RemoteTeamId: model.NewId(), + } + selfCluster1, err = ss.RemoteCluster().Save(selfCluster1) + require.NoError(t, err) + + selfCluster2 := &model.RemoteCluster{ + RemoteId: model.NewId(), + Name: "test-remote-2", + DisplayName: "Test Remote 2", + SiteURL: testServer2.URL, + Token: model.NewId(), + CreateAt: model.GetMillis(), + LastPingAt: model.GetMillis(), + CreatorId: th.BasicUser.Id, + RemoteTeamId: model.NewId(), + } + selfCluster2, err = ss.RemoteCluster().Save(selfCluster2) + require.NoError(t, err) + + // Initialize sync handlers + syncHandler1 = NewSelfReferentialSyncHandler(t, service, selfCluster1) + syncHandler2 = NewSelfReferentialSyncHandler(t, service, selfCluster2) + + // Create a shared channel + channel := th.CreateChannel(th.Context, th.BasicTeam) + sc := &model.SharedChannel{ + ChannelId: channel.Id, + TeamId: channel.TeamId, + Home: true, + ReadOnly: false, + ShareName: channel.Name, + ShareDisplayName: channel.DisplayName, + SharePurpose: channel.Purpose, + ShareHeader: channel.Header, + CreatorId: th.BasicUser.Id, + RemoteId: "", + } + + _, err = th.App.ShareChannel(th.Context, sc) + require.NoError(t, err) + + // Share the channel with both remotes + scr1 := &model.SharedChannelRemote{ + Id: model.NewId(), + ChannelId: channel.Id, + CreatorId: th.BasicUser.Id, + IsInviteAccepted: true, + IsInviteConfirmed: true, + RemoteId: selfCluster1.RemoteId, + LastPostUpdateAt: model.GetMillis(), + } + _, err = ss.SharedChannel().SaveRemote(scr1) + require.NoError(t, err) + + scr2 := &model.SharedChannelRemote{ + Id: model.NewId(), + ChannelId: channel.Id, + CreatorId: th.BasicUser.Id, + IsInviteAccepted: true, + IsInviteConfirmed: true, + RemoteId: selfCluster2.RemoteId, + LastPostUpdateAt: model.GetMillis(), + } + _, err = ss.SharedChannel().SaveRemote(scr2) + require.NoError(t, err) + + // Verify the channel is shared with both remotes + hasRemote1, err := ss.SharedChannel().HasRemote(channel.Id, selfCluster1.RemoteId) + require.NoError(t, err) + require.True(t, hasRemote1, "Channel should be shared with remote 1") + + hasRemote2, err := ss.SharedChannel().HasRemote(channel.Id, selfCluster2.RemoteId) + require.NoError(t, err) + require.True(t, hasRemote2, "Channel should be shared with remote 2") + + // Step 1: Verify the channel is initially shared with both remotes + err = th.App.checkChannelIsShared(channel.Id) + require.NoError(t, err, "Channel should be shared initially") + + // Step 2: Create a post in the channel to trigger sync activity + post := &model.Post{ + ChannelId: channel.Id, + UserId: th.BasicUser.Id, + Message: "Test message after remote 1 unshare", + } + _, appErr := th.App.CreatePost(th.Context, post, channel, model.CreatePostFlags{}) + require.Nil(t, appErr) + + // Get post count after creating the test post but before "remote-initiated unshare" + postsBeforeRemove, appErr := th.App.GetPostsPage(model.GetPostsOptions{ + ChannelId: channel.Id, + Page: 0, + PerPage: 10, + }) + require.Nil(t, appErr) + postCountBefore := len(postsBeforeRemove.Posts) + + // Step 3: Create a sync message for remote 1 + syncMsg := model.NewSyncMsg(channel.Id) + syncMsg.Posts = []*model.Post{{ + Id: model.NewId(), + ChannelId: channel.Id, + UserId: th.BasicUser.Id, + Message: "Test message after remote 1 unshare", + CreateAt: model.GetMillis(), + }} + + // Step 4: Simulate remote 1 returning ErrChannelIsNotShared + service.HandleChannelNotSharedErrorForTesting(syncMsg, selfCluster1) + + // Verify we now have only 1 remote (remote 2) + remotes, err := ss.SharedChannel().GetRemotes(0, 10, model.SharedChannelRemoteFilterOpts{ + ChannelId: channel.Id, + }) + require.NoError(t, err) + require.Len(t, remotes, 1, "Expected 1 remote after removing remote 1") + t.Logf("Number of remotes after unshare: %d", len(remotes)) + + // The expected behavior is that only the specific remote should be removed, + // with the channel remaining shared with other remotes. + err = th.App.checkChannelIsShared(channel.Id) + + // The channel should still be shared with remote2, so this should pass + assert.NoError(t, err, "Channel should still be shared with other remotes") + + // Verify remote 1 is no longer in shared channel + hasRemote1After, err := ss.SharedChannel().HasRemote(channel.Id, selfCluster1.RemoteId) + require.NoError(t, err) + require.False(t, hasRemote1After, "Channel should no longer be shared with remote 1") + + // Check if remote 2 is still associated with the channel + // Expected behavior: remote 2 should still be associated + hasRemote2After, err := ss.SharedChannel().HasRemote(channel.Id, selfCluster2.RemoteId) + require.NoError(t, err) + assert.True(t, hasRemote2After, "Channel should still be shared with remote 2") + + // Verify a system message was posted about remote 1 unsharing + postsAfterRemove, appErr := th.App.GetPostsPage(model.GetPostsOptions{ + ChannelId: channel.Id, + Page: 0, + PerPage: 10, + }) + require.Nil(t, appErr) + + // Expected: only one notification post when a remote is removed + assert.Equal(t, postCountBefore+1, len(postsAfterRemove.Posts), "There should be one new post") + + // Find and verify the system message content + var systemPost *model.Post + for _, p := range postsAfterRemove.Posts { + if p.Type == model.PostTypeSystemGeneric { + systemPost = p + break + } + } + require.NotNil(t, systemPost, "A system post should be created") + assert.Equal(t, "This channel is no longer shared.", systemPost.Message, "Message should match unshare message") + }) +} + +func TestSyncMessageErrChannelNotSharedResponse(t *testing.T) { + th := setupSharedChannels(t).InitBasic() + defer th.TearDown() + + // Setup: Create a shared channel and remote cluster + ss := th.App.Srv().Store() + + // Get the shared channel service and cast to concrete type + scsInterface := th.App.Srv().GetSharedChannelSyncService() + service, ok := scsInterface.(*sharedchannel.Service) + require.True(t, ok, "Expected sharedchannel.Service concrete type") + + channel := th.CreateChannel(th.Context, th.BasicTeam) + sc := &model.SharedChannel{ + ChannelId: channel.Id, + TeamId: th.BasicTeam.Id, + Home: true, + ShareName: channel.Name, + ShareDisplayName: channel.DisplayName, + CreatorId: th.BasicUser.Id, + RemoteId: "", + } + _, err := ss.SharedChannel().Save(sc) + require.NoError(t, err) + + // Create a self-referential remote cluster + selfCluster := &model.RemoteCluster{ + RemoteId: model.NewId(), + Name: "test-remote", + DisplayName: "Test Remote", + SiteURL: "https://test.example.com", + Token: model.NewId(), + CreateAt: model.GetMillis(), + LastPingAt: model.GetMillis(), + CreatorId: th.BasicUser.Id, + RemoteTeamId: model.NewId(), + } + selfCluster, err = ss.RemoteCluster().Save(selfCluster) + require.NoError(t, err) + + // Create shared channel remote + scr := &model.SharedChannelRemote{ + Id: model.NewId(), + ChannelId: channel.Id, + CreatorId: th.BasicUser.Id, + IsInviteAccepted: true, + IsInviteConfirmed: true, + RemoteId: selfCluster.RemoteId, + LastPostCreateAt: model.GetMillis(), + LastPostUpdateAt: model.GetMillis(), + } + _, err = ss.SharedChannel().SaveRemote(scr) + require.NoError(t, err) + + // Verify channel is initially shared + hasRemote, err := ss.SharedChannel().HasRemote(channel.Id, selfCluster.RemoteId) + require.NoError(t, err) + require.True(t, hasRemote, "Channel should be shared with remote initially") + + // Test: Simulate sync message callback receiving ErrChannelNotShared response + syncMsg := model.NewSyncMsg(channel.Id) + syncMsg.Posts = []*model.Post{{ + Id: model.NewId(), + ChannelId: channel.Id, + UserId: th.BasicUser.Id, + Message: "Test sync message", + CreateAt: model.GetMillis(), + }} + + // Create a response that simulates the remote returning ErrChannelNotShared + response := &remotecluster.Response{ + Status: "fail", + Err: "cannot process sync message; channel is no longer shared: " + channel.Id, + } + + // Test the complete flow by simulating what happens in sendSyncMsgToRemote callback + // This tests the fixed error detection logic that checks rcResp.Err + var callbackTriggered bool + mockCallback := func(rcMsg model.RemoteClusterMsg, rc *model.RemoteCluster, rcResp *remotecluster.Response, errResp error) { + callbackTriggered = true + + // This simulates the fixed logic in sync_send_remote.go + if rcResp != nil && !rcResp.IsSuccess() && strings.Contains(rcResp.Err, "channel is no longer shared") { + service.HandleChannelNotSharedErrorForTesting(syncMsg, rc) + } + } + + // Trigger the callback with our mock response + mockCallback(model.RemoteClusterMsg{}, selfCluster, response, nil) + + // Verify the callback was triggered + require.True(t, callbackTriggered, "Callback should have been triggered") + + // Verify the channel is no longer shared with the remote + hasRemoteAfter, err := ss.SharedChannel().HasRemote(channel.Id, selfCluster.RemoteId) + require.NoError(t, err) + require.False(t, hasRemoteAfter, "Channel should no longer be shared with remote after error") + + // Verify a system message was posted + posts, appErr := th.App.GetPostsPage(model.GetPostsOptions{ + ChannelId: channel.Id, + Page: 0, + PerPage: 10, + }) + require.Nil(t, appErr) + + // Find the system message + var systemPost *model.Post + for _, p := range posts.Posts { + if p.Type == model.PostTypeSystemGeneric && p.Message == "This channel is no longer shared." { + systemPost = p + break + } + } + require.NotNil(t, systemPost, "System message should be posted when channel becomes unshared") +} diff --git a/server/platform/services/sharedchannel/service.go b/server/platform/services/sharedchannel/service.go index 65b976e449..9e467caa4d 100644 --- a/server/platform/services/sharedchannel/service.go +++ b/server/platform/services/sharedchannel/service.go @@ -32,6 +32,7 @@ const ( NotifyMinimumDelay = time.Second * 2 MaxUpsertRetries = 25 ProfileImageSyncTimeout = time.Second * 5 + UnshareMessage = "This channel is no longer shared." // Default value for MaxMembersPerBatch is defined in config.go as ConnectedWorkspacesSettingsDefaultMemberSyncBatchSize ) @@ -304,6 +305,31 @@ func (scs *Service) notifyClientsForSharedChannelUpdate(channel *model.Channel) scs.app.Publish(messageWs) } +// postUnshareNotification posts a system message to notify users that the channel is no longer shared. +func (scs *Service) postUnshareNotification(channelID string, creatorID string, channel *model.Channel, rc *model.RemoteCluster) { + post := &model.Post{ + UserId: creatorID, + ChannelId: channelID, + Message: UnshareMessage, + Type: model.PostTypeSystemGeneric, + } + + logger := scs.server.Log() + _, appErr := scs.app.CreatePost(request.EmptyContext(logger), post, channel, model.CreatePostFlags{}) + + if appErr != nil { + scs.server.Log().Log( + mlog.LvlSharedChannelServiceError, + "Error creating unshare notification post", + mlog.String("channel_id", channelID), + mlog.String("remote_id", rc.RemoteId), + mlog.String("remote_name", rc.Name), + mlog.Err(appErr), + ) + } +} + +// OnReceiveSyncMessageForTesting is a wrapper to expose onReceiveSyncMessage for testing purposes // isGlobalUserSyncEnabled checks if the global user sync feature is enabled func (scs *Service) isGlobalUserSyncEnabled() bool { cfg := scs.server.Config() @@ -349,3 +375,8 @@ func (scs *Service) HandleSyncAllUsersForTesting(rc *model.RemoteCluster) error func (scs *Service) OnReceiveSyncMessageForTesting(msg model.RemoteClusterMsg, rc *model.RemoteCluster, response *remotecluster.Response) error { return scs.onReceiveSyncMessage(msg, rc, response) } + +// HandleChannelNotSharedErrorForTesting is a wrapper to expose handleChannelNotSharedError for testing purposes +func (scs *Service) HandleChannelNotSharedErrorForTesting(msg *model.SyncMsg, rc *model.RemoteCluster) { + scs.handleChannelNotSharedError(msg, rc) +} diff --git a/server/platform/services/sharedchannel/service_api.go b/server/platform/services/sharedchannel/service_api.go index f0f989820e..705e7d7f20 100644 --- a/server/platform/services/sharedchannel/service_api.go +++ b/server/platform/services/sharedchannel/service_api.go @@ -87,7 +87,7 @@ func (scs *Service) UpdateSharedChannel(sc *model.SharedChannel) (*model.SharedC return scUpdated, nil } -// UnshareChannel unshared the channel by deleting the SharedChannels record and unsets the Channel `shared` flag. +// UnshareChannel unshares the channel by deleting the SharedChannels record and unsets the Channel `shared` flag. // Returns true if a shared channel existed and was deleted. func (scs *Service) UnshareChannel(channelID string) (bool, error) { channel, err := scs.server.GetStore().Channel().Get(channelID, true) diff --git a/server/platform/services/sharedchannel/sync_recv.go b/server/platform/services/sharedchannel/sync_recv.go index c864ecc9c6..79e4fc3bef 100644 --- a/server/platform/services/sharedchannel/sync_recv.go +++ b/server/platform/services/sharedchannel/sync_recv.go @@ -21,6 +21,7 @@ var ( ErrRemoteIDMismatch = errors.New("remoteID mismatch") ErrChannelIDMismatch = errors.New("channelID mismatch") ErrUserDMPermission = errors.New("users cannot DM each other") + ErrChannelNotShared = errors.New("channel is no longer shared") ) func (scs *Service) onReceiveSyncMessage(msg model.RemoteClusterMsg, rc *model.RemoteCluster, response *remotecluster.Response) error { @@ -39,7 +40,6 @@ func (scs *Service) onReceiveSyncMessage(msg model.RemoteClusterMsg, rc *model.R } var sm model.SyncMsg - if err := json.Unmarshal(msg.Payload, &sm); err != nil { return fmt.Errorf("invalid sync message: %w", err) } @@ -139,7 +139,8 @@ func (scs *Service) processSyncMessage(c request.CTX, syncMsg *model.SyncMsg, rc return fmt.Errorf("cannot check channel share state for sync message: %w", err) } if !exists { - return fmt.Errorf("cannot process sync message; channel not shared with remote: %w", ErrRemoteIDMismatch) + return fmt.Errorf("cannot process sync message; %w: %s", + ErrChannelNotShared, syncMsg.ChannelId) } // add/update users before posts diff --git a/server/platform/services/sharedchannel/sync_send_remote.go b/server/platform/services/sharedchannel/sync_send_remote.go index adf2b91be3..83b3b52b46 100644 --- a/server/platform/services/sharedchannel/sync_send_remote.go +++ b/server/platform/services/sharedchannel/sync_send_remote.go @@ -1011,18 +1011,34 @@ func (scs *Service) sendSyncMsgToRemote(msg *model.SyncMsg, rc *model.RemoteClus err = rcs.SendMsg(ctx, rcMsg, rc, func(rcMsg model.RemoteClusterMsg, rc *model.RemoteCluster, rcResp *remotecluster.Response, errResp error) { defer wg.Done() - var syncResp model.SyncResponse - if err2 := json.Unmarshal(rcResp.Payload, &syncResp); err2 != nil { - scs.server.Log().Log(mlog.LvlSharedChannelServiceError, "Invalid sync msg response from remote cluster", - mlog.String("remote", rc.Name), - mlog.String("channel_id", msg.ChannelId), - mlog.Err(err2), - ) + // Check for ErrChannelNotShared in the application response + if rcResp != nil && !rcResp.IsSuccess() && strings.Contains(rcResp.Err, ErrChannelNotShared.Error()) { + scs.handleChannelNotSharedError(msg, rc) return } - if f != nil { - f(syncResp, errResp) + var syncResp model.SyncResponse + if errResp == nil { + if rcResp != nil && len(rcResp.Payload) > 0 { + if err2 := json.Unmarshal(rcResp.Payload, &syncResp); err2 != nil { + scs.server.Log().Log(mlog.LvlSharedChannelServiceError, "Invalid sync msg response from remote cluster", + mlog.String("remote", rc.Name), + mlog.String("channel_id", msg.ChannelId), + mlog.Err(err2), + ) + return + } + + if f != nil { + f(syncResp, errResp) + } + } else { + // No error but response is nil or empty + scs.server.Log().Log(mlog.LvlSharedChannelServiceError, "Empty or nil response payload from remote cluster", + mlog.String("remote", rc.Name), + mlog.String("channel_id", msg.ChannelId), + ) + } } }) @@ -1049,3 +1065,45 @@ func sanitizeSyncData(sd *syncData) { sd.profileImages[id] = sanitizeUserForSync(user) } } + +// handleChannelNotSharedError processes the case when a remote indicates a channel +// is no longer shared. It removes the remote from the shared channel locally and, +// if it was the last remote, completely unshares the channel. +func (scs *Service) handleChannelNotSharedError(msg *model.SyncMsg, rc *model.RemoteCluster) { + logger := scs.server.Log() + + logger.Log(mlog.LvlSharedChannelServiceDebug, "Remote indicated channel is no longer shared; unsharing locally", + mlog.String("remote", rc.Name), + mlog.String("channel_id", msg.ChannelId), + ) + + // Get the SharedChannelRemote record for this channel and remote + scr, getErr := scs.server.GetStore().SharedChannel().GetRemoteByIds(msg.ChannelId, rc.RemoteId) + if getErr != nil { + logger.Log(mlog.LvlSharedChannelServiceError, "Failed to get shared channel remote", + mlog.String("remote", rc.Name), + mlog.String("channel_id", msg.ChannelId), + mlog.Err(getErr), + ) + return + } + + // Get channel details for posting the system message + channel, channelErr := scs.server.GetStore().Channel().Get(msg.ChannelId, true) + if channelErr != nil { + logger.Log(mlog.LvlSharedChannelServiceError, "Failed to get channel details", + mlog.String("remote", rc.Name), + mlog.String("channel_id", msg.ChannelId), + mlog.Err(channelErr), + ) + return + } + + // Post a system message to notify users that the channel is no longer shared with this remote + scs.postUnshareNotification(msg.ChannelId, scr.CreatorId, channel, rc) + + if err := scs.UninviteRemoteFromChannel(msg.ChannelId, rc.RemoteId); err != nil { + logger.Log(mlog.LvlSharedChannelServiceError, "Failed to uninvite remote from shared channel", mlog.Err(err)) + return + } +}