Adds logical deletes to shared channel remotes and remote clusters (#28159)

* Adds logical deletes to shared channel remotes and remote clusters

Instead of physically deleting the shared channel remote and remote
clusters records when a channel is unshared, a remote uninvited or a
remote cluster is deleted, now those have a logical `DeleteAt` field
that is set.

This allows us to safely restore shared channels between two remote
clusters (as of now resetting the cursor without backfilling their
contents) and to know which connections were established in the past
and now are severed.

* Delete the index in remoteclusters before adding the new column

* Fix bad error check

server/platform/services/sharedchannel/attachment.go

@@ -169,9 +169,13 @@ func (scs *Service) onReceiveUploadCreate(msg model.RemoteClusterMsg, rc *model.
 	}
 
 	// make sure channel is shared for the remote sender
-	if _, err := scs.server.GetStore().SharedChannel().GetRemoteByIds(us.ChannelId, rc.RemoteId); err != nil {
+	hasRemote, err := scs.server.GetStore().SharedChannel().HasRemote(us.ChannelId, rc.RemoteId)
+	if err != nil {
 		return fmt.Errorf("could not validate upload session for remote: %w", err)
 	}
+	if !hasRemote {
+		return model.NewAppError("createUpload", "api.upload.create.upload_channel_not_shared_with_remote.app_error", nil, "", http.StatusBadRequest)
+	}
 
 	// make sure file attachments are enabled
 	if scs.server.Config().FileSettings.EnableFileAttachments == nil || !*scs.server.Config().FileSettings.EnableFileAttachments {

server/platform/services/sharedchannel/channelinvite.go

@@ -6,12 +6,14 @@ package sharedchannel
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"strings"
 
 	"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"
 	"github.com/mattermost/mattermost/server/v8/platform/services/remotecluster"
 )
 
@@ -78,19 +80,48 @@ func (scs *Service) SendChannelInvite(channel *model.Channel, userId string, rc
 			return
 		}
 
-		scr := &model.SharedChannelRemote{
-			ChannelId:         sc.ChannelId,
-			CreatorId:         userId,
-			RemoteId:          rc.RemoteId,
-			IsInviteAccepted:  true,
-			IsInviteConfirmed: true,
-			LastPostCreateAt:  model.GetMillis(),
-			LastPostUpdateAt:  model.GetMillis(),
-		}
-		if _, err = scs.server.GetStore().SharedChannel().SaveRemote(scr); err != nil {
-			scs.sendEphemeralPost(channel.Id, userId, fmt.Sprintf("Error confirming channel invite for %s: %v", rc.DisplayName, err))
+		existingScr, err := scs.server.GetStore().SharedChannel().GetRemoteByIds(sc.ChannelId, rc.RemoteId)
+		var errNotFound *store.ErrNotFound
+		if err != nil && !errors.As(err, &errNotFound) {
+			scs.sendEphemeralPost(channel.Id, userId, fmt.Sprintf("Error sending channel invite for %s: %s", rc.DisplayName, err))
 			return
 		}
+
+		curTime := model.GetMillis()
+		if existingScr != nil {
+			if existingScr.DeleteAt == 0 {
+				// the shared channel remote exists and is not
+				// deleted, nothing to do here
+				return
+			}
+
+			// the shared channel remote was deleted in the past, so
+			// with the new invite we restore it
+			existingScr.DeleteAt = 0
+			existingScr.UpdateAt = curTime
+			existingScr.LastPostCreateAt = curTime
+			existingScr.LastPostUpdateAt = curTime
+			if _, sErr := scs.server.GetStore().SharedChannel().UpdateRemote(existingScr); sErr != nil {
+				scs.sendEphemeralPost(channel.Id, userId, fmt.Sprintf("Error confirming channel invite for %s: %v", rc.DisplayName, sErr))
+				return
+			}
+		} else {
+			// the shared channel remote doesn't exists, so we create it
+			scr := &model.SharedChannelRemote{
+				ChannelId:         sc.ChannelId,
+				CreatorId:         userId,
+				RemoteId:          rc.RemoteId,
+				IsInviteAccepted:  true,
+				IsInviteConfirmed: true,
+				LastPostCreateAt:  curTime,
+				LastPostUpdateAt:  curTime,
+			}
+			if _, err = scs.server.GetStore().SharedChannel().SaveRemote(scr); err != nil {
+				scs.sendEphemeralPost(channel.Id, userId, fmt.Sprintf("Error confirming channel invite for %s: %v", rc.DisplayName, err))
+				return
+			}
+		}
+
 		scs.NotifyChannelChanged(sc.ChannelId)
 		scs.sendEphemeralPost(channel.Id, userId, fmt.Sprintf("`%s` has been added to channel.", rc.DisplayName))
 	}
@@ -105,28 +136,7 @@ func (scs *Service) SendChannelInvite(channel *model.Channel, userId string, rc
 	ctx, cancel := context.WithTimeout(context.Background(), remotecluster.SendTimeout)
 	defer cancel()
 
-	return rcs.SendMsg(ctx, msg, rc, func(msg model.RemoteClusterMsg, rc *model.RemoteCluster, resp *remotecluster.Response, err error) {
-		if err != nil || !resp.IsSuccess() {
-			scs.sendEphemeralPost(channel.Id, userId, fmt.Sprintf("Error sending channel invite for %s: %s", rc.DisplayName, combineErrors(err, resp.Err)))
-			return
-		}
-
-		scr := &model.SharedChannelRemote{
-			ChannelId:         sc.ChannelId,
-			CreatorId:         userId,
-			RemoteId:          rc.RemoteId,
-			IsInviteAccepted:  true,
-			IsInviteConfirmed: true,
-			LastPostCreateAt:  model.GetMillis(),
-			LastPostUpdateAt:  model.GetMillis(),
-		}
-		if _, err = scs.server.GetStore().SharedChannel().SaveRemote(scr); err != nil {
-			scs.sendEphemeralPost(channel.Id, userId, fmt.Sprintf("Error confirming channel invite for %s: %v", rc.DisplayName, err))
-			return
-		}
-		scs.NotifyChannelChanged(sc.ChannelId)
-		scs.sendEphemeralPost(channel.Id, userId, fmt.Sprintf("`%s` has been added to channel.", rc.DisplayName))
-	})
+	return rcs.SendMsg(ctx, msg, rc, onInvite)
 }
 
 func combineErrors(err error, serror string) string {
@@ -162,43 +172,63 @@ func (scs *Service) onReceiveChannelInvite(msg model.RemoteClusterMsg, rc *model
 	)
 
 	// check if channel already exists
+	existingScr, err := scs.server.GetStore().SharedChannel().GetRemoteByIds(invite.ChannelId, rc.RemoteId)
+	var errNotFound *store.ErrNotFound
+	if err != nil && !errors.As(err, &errNotFound) {
+		return fmt.Errorf("cannot get deleted shared channel remote (channel_id=%s): %w", invite.ChannelId, err)
+	}
+
+	if existingScr != nil && existingScr.DeleteAt == 0 {
+		// the channel is already shared, nothing to do
+		return nil
+	}
+
 	var channel *model.Channel
 	var created bool
-	_, err := scs.server.GetStore().Channel().Get(invite.ChannelId, true)
-	if err == nil {
-		// the channel already exists on this server; could be the remote is trying to re-share it (not allowed at this time).
-		// If the channel is already shared with the remote, it will remain so.
-		return fmt.Errorf("cannot create shared channel (channel_id=%s): %w", invite.ChannelId, model.ErrChannelAlreadyExists)
-	}
+	if existingScr == nil {
+		var err error
+		_, err = scs.server.GetStore().Channel().Get(invite.ChannelId, true)
+		if err == nil {
+			// the channel already exists on this server and was not
+			// previously shared, so we reject the invite
+			return fmt.Errorf("cannot create new shared channel (channel_id=%s): %w", invite.ChannelId, model.ErrChannelAlreadyExists)
+		}
 
-	// create new local channel to sync with the remote channel
-	if channel, created, err = scs.handleChannelCreation(invite, rc); err != nil {
-		return err
-	}
+		// create new local channel to sync with the remote channel
+		if channel, created, err = scs.handleChannelCreation(invite, rc); err != nil {
+			return err
+		}
 
-	// sanity check to ensure the channel returned has the expected id. Otherwise sync will not work as expected and will fail
-	// silently.
-	if invite.ChannelId != channel.Id {
-		// as of this writing, this scenario should only be possible if the invite included a DM channel invitation with a
-		// combination of two user ids (one remote, one local) that already have a DM on this server. Very unlikely unless
-		// the remote is compromised AND has knowledge of the local user id.
-		// Another possibility would be an actual user ID collision between two servers, where the likelihood is
-		// infinitesimally small
-		scs.server.Log().Log(mlog.LvlSharedChannelServiceError, "Channel invite failed - channel created/fetched with wrong id",
-			mlog.String("remote", rc.DisplayName),
-			mlog.String("channel_id", invite.ChannelId),
-			mlog.String("channel_type", invite.Type),
-			mlog.String("channel_name", invite.Name),
-			mlog.String("team_id", invite.TeamId),
-			mlog.Array("dm_partics", invite.DirectParticipantIDs),
-		)
-		return fmt.Errorf("cannot create shared channel (DM channel_id=%s): %w", invite.ChannelId, model.ErrChannelAlreadyExists)
-	}
+		// sanity check to ensure the channel returned has the expected id. Otherwise sync will not work as expected and will fail
+		// silently.
+		if invite.ChannelId != channel.Id {
+			// as of this writing, this scenario should only be possible if the invite included a DM channel invitation with a
+			// combination of two user ids (one remote, one local) that already have a DM on this server. Very unlikely unless
+			// the remote is compromised AND has knowledge of the local user id.
+			// Another possibility would be an actual user ID collision between two servers, where the likelihood is
+			// infinitesimally small
+			scs.server.Log().Log(mlog.LvlSharedChannelServiceError, "Channel invite failed - channel created/fetched with wrong id",
+				mlog.String("remote", rc.DisplayName),
+				mlog.String("channel_id", invite.ChannelId),
+				mlog.String("channel_type", invite.Type),
+				mlog.String("channel_name", invite.Name),
+				mlog.String("team_id", invite.TeamId),
+				mlog.Array("dm_partics", invite.DirectParticipantIDs),
+			)
+			return fmt.Errorf("cannot create shared channel (DM channel_id=%s): %w", invite.ChannelId, model.ErrChannelAlreadyExists)
+		}
 
-	// mark the newly created channel read-only if requested in the invite
-	if invite.ReadOnly {
-		if err := scs.makeChannelReadOnly(channel); err != nil {
-			return fmt.Errorf("cannot make channel readonly `%s`: %w", invite.ChannelId, err)
+		// mark the newly created channel read-only if requested in the invite
+		if invite.ReadOnly {
+			if err := scs.makeChannelReadOnly(channel); err != nil {
+				return fmt.Errorf("cannot make channel readonly `%s`: %w", invite.ChannelId, err)
+			}
+		}
+	} else {
+		var err error
+		channel, err = scs.server.GetStore().Channel().Get(invite.ChannelId, true)
+		if err != nil {
+			return fmt.Errorf("cannot get channel (channel_id=%s) to restore a shared channel remote: %w", invite.ChannelId, err)
 		}
 	}
 
@@ -206,7 +236,7 @@ func (scs *Service) onReceiveChannelInvite(msg model.RemoteClusterMsg, rc *model
 		ChannelId:        channel.Id,
 		TeamId:           channel.TeamId,
 		Home:             false,
-		ReadOnly:         invite.ReadOnly,
+		ReadOnly:         existingScr == nil && invite.ReadOnly, // only set read only flag for new shares
 		ShareName:        channel.Name,
 		ShareDisplayName: channel.DisplayName,
 		SharePurpose:     channel.Purpose,
@@ -224,25 +254,36 @@ func (scs *Service) onReceiveChannelInvite(msg model.RemoteClusterMsg, rc *model
 		return fmt.Errorf("cannot create shared channel (channel_id=%s): %w", invite.ChannelId, err)
 	}
 
-	sharedChannelRemote := &model.SharedChannelRemote{
-		Id:                model.NewId(),
-		ChannelId:         channel.Id,
-		CreatorId:         channel.CreatorId,
-		IsInviteAccepted:  true,
-		IsInviteConfirmed: true,
-		RemoteId:          rc.RemoteId,
-		LastPostCreateAt:  model.GetMillis(),
-		LastPostUpdateAt:  model.GetMillis(),
-	}
-
-	if _, err := scs.server.GetStore().SharedChannel().SaveRemote(sharedChannelRemote); err != nil {
-		// delete the newly created channel since we could not create a SharedChannelRemote record for it,
-		// and delete the newly created SharedChannel record as well.
-		if created {
-			scs.app.PermanentDeleteChannel(request.EmptyContext(scs.server.Log()), channel)
+	curTime := model.GetMillis()
+	if existingScr != nil {
+		existingScr.DeleteAt = 0
+		existingScr.UpdateAt = curTime
+		existingScr.LastPostCreateAt = curTime
+		existingScr.LastPostUpdateAt = curTime
+		if _, err := scs.server.GetStore().SharedChannel().UpdateRemote(existingScr); err != nil {
+			return fmt.Errorf("cannot restore deleted shared channel remote (channel_id=%s): %w", invite.ChannelId, err)
+		}
+	} else {
+		scr := &model.SharedChannelRemote{
+			Id:                model.NewId(),
+			ChannelId:         channel.Id,
+			CreatorId:         channel.CreatorId,
+			IsInviteAccepted:  true,
+			IsInviteConfirmed: true,
+			RemoteId:          rc.RemoteId,
+			LastPostCreateAt:  model.GetMillis(),
+			LastPostUpdateAt:  model.GetMillis(),
+		}
+
+		if _, err := scs.server.GetStore().SharedChannel().SaveRemote(scr); err != nil {
+			// delete the newly created channel since we could not create a SharedChannelRemote record for it,
+			// and delete the newly created SharedChannel record as well.
+			if created {
+				scs.app.PermanentDeleteChannel(request.EmptyContext(scs.server.Log()), channel)
+			}
+			scs.server.GetStore().SharedChannel().Delete(sharedChannel.ChannelId)
+			return fmt.Errorf("cannot create shared channel remote (channel_id=%s): %w", invite.ChannelId, err)
 		}
-		scs.server.GetStore().SharedChannel().Delete(sharedChannel.ChannelId)
-		return fmt.Errorf("cannot create shared channel remote (channel_id=%s): %w", invite.ChannelId, err)
 	}
 	return nil
 }

server/platform/services/sharedchannel/channelinvite_test.go

@@ -62,7 +62,7 @@ func TestOnReceiveChannelInvite(t *testing.T) {
 		}
 
 		mockStore := &mocks.Store{}
-		remoteCluster := &model.RemoteCluster{Name: "test", DefaultTeamId: model.NewId()}
+		remoteCluster := &model.RemoteCluster{RemoteId: model.NewId(), Name: "test", DefaultTeamId: model.NewId()}
 		invitation := channelInviteMsg{
 			ChannelId: model.NewId(),
 			TeamId:    model.NewId(),
@@ -83,6 +83,7 @@ func TestOnReceiveChannelInvite(t *testing.T) {
 			Type:   invitation.Type,
 		}
 
+		mockSharedChannelStore.On("GetRemoteByIds", invitation.ChannelId, remoteCluster.RemoteId).Return(nil, store.NewErrNotFound("SharedChannelRemote", ""))
 		mockChannelStore.On("Get", invitation.ChannelId, true).Return(nil, &store.ErrNotFound{})
 		mockSharedChannelStore.On("Save", mock.Anything).Return(nil, nil)
 		mockSharedChannelStore.On("SaveRemote", mock.Anything).Return(nil, nil)
@@ -127,7 +128,7 @@ func TestOnReceiveChannelInvite(t *testing.T) {
 		}
 
 		mockStore := &mocks.Store{}
-		remoteCluster := &model.RemoteCluster{Name: "test2"}
+		remoteCluster := &model.RemoteCluster{RemoteId: model.NewId(), Name: "test2"}
 		invitation := channelInviteMsg{
 			ChannelId: model.NewId(),
 			TeamId:    model.NewId(),
@@ -148,11 +149,14 @@ func TestOnReceiveChannelInvite(t *testing.T) {
 		team := &model.Team{
 			Id: model.NewId(),
 		}
+		mockSharedChannelStore := mocks.SharedChannelStore{}
 
+		mockSharedChannelStore.On("GetRemoteByIds", invitation.ChannelId, remoteCluster.RemoteId).Return(nil, store.NewErrNotFound("SharedChannelRemote", ""))
 		mockChannelStore.On("Get", invitation.ChannelId, true).Return(nil, &store.ErrNotFound{})
 		mockTeamStore.On("GetAllPage", 0, 1, mock.Anything).Return([]*model.Team{team}, nil)
 		mockStore.On("Channel").Return(&mockChannelStore)
 		mockStore.On("Team").Return(&mockTeamStore)
+		mockStore.On("SharedChannel").Return(&mockSharedChannelStore)
 
 		mockServer = scs.server.(*MockServerIface)
 		mockServer.On("GetStore").Return(mockStore)
@@ -167,6 +171,96 @@ func TestOnReceiveChannelInvite(t *testing.T) {
 		assert.Equal(t, fmt.Sprintf("cannot make channel readonly `%s`: foo: bar, boom", invitation.ChannelId), err.Error())
 	})
 
+	t.Run("When invitation points to a deleted shared channel remote", func(t *testing.T) {
+		mockServer := &MockServerIface{}
+		logger := mlog.CreateConsoleTestLogger(t)
+		mockServer.On("Log").Return(logger)
+		mockApp := &MockAppIface{}
+		scs := &Service{
+			server: mockServer,
+			app:    mockApp,
+		}
+
+		mockStore := &mocks.Store{}
+		remoteCluster := &model.RemoteCluster{RemoteId: model.NewId(), Name: "test", DefaultTeamId: model.NewId()}
+		invitation := channelInviteMsg{
+			ChannelId: model.NewId(),
+			TeamId:    model.NewId(),
+			Type:      model.ChannelTypeOpen,
+		}
+		payload, err := json.Marshal(invitation)
+		require.NoError(t, err)
+
+		msg := model.RemoteClusterMsg{
+			Payload: payload,
+		}
+		mockChannelStore := mocks.ChannelStore{}
+		mockSharedChannelStore := mocks.SharedChannelStore{}
+		channel := &model.Channel{
+			Id:     invitation.ChannelId,
+			TeamId: invitation.TeamId,
+			Type:   invitation.Type,
+		}
+		sharedChannelRemote := &model.SharedChannelRemote{
+			ChannelId: invitation.ChannelId,
+			RemoteId:  remoteCluster.RemoteId,
+			DeleteAt:  1234,
+		}
+
+		mockSharedChannelStore.On("GetRemoteByIds", invitation.ChannelId, mock.Anything).Return(sharedChannelRemote, nil)
+		mockChannelStore.On("Get", invitation.ChannelId, true).Return(channel, nil)
+		mockSharedChannelStore.On("Save", mock.Anything).Return(nil, nil)
+		mockSharedChannelStore.On("UpdateRemote", mock.Anything).Return(nil, nil)
+		mockStore.On("Channel").Return(&mockChannelStore)
+		mockStore.On("SharedChannel").Return(&mockSharedChannelStore)
+
+		mockServer.On("GetStore").Return(mockStore)
+		defer mockApp.AssertExpectations(t)
+
+		err = scs.onReceiveChannelInvite(msg, remoteCluster, nil)
+		require.NoError(t, err)
+	})
+
+	t.Run("When invitation points to an existing shared channel remote", func(t *testing.T) {
+		mockServer := &MockServerIface{}
+		logger := mlog.CreateConsoleTestLogger(t)
+		mockServer.On("Log").Return(logger)
+		mockApp := &MockAppIface{}
+		scs := &Service{
+			server: mockServer,
+			app:    mockApp,
+		}
+
+		mockStore := &mocks.Store{}
+		remoteCluster := &model.RemoteCluster{RemoteId: model.NewId(), Name: "test", DefaultTeamId: model.NewId()}
+		invitation := channelInviteMsg{
+			ChannelId: model.NewId(),
+			TeamId:    model.NewId(),
+			Type:      model.ChannelTypeOpen,
+		}
+		payload, err := json.Marshal(invitation)
+		require.NoError(t, err)
+
+		msg := model.RemoteClusterMsg{
+			Payload: payload,
+		}
+		mockSharedChannelStore := mocks.SharedChannelStore{}
+		sharedChannelRemote := &model.SharedChannelRemote{
+			ChannelId: invitation.ChannelId,
+			RemoteId:  remoteCluster.RemoteId,
+			DeleteAt:  0,
+		}
+
+		mockServer.On("GetStore").Return(mockStore)
+		mockSharedChannelStore.On("GetRemoteByIds", invitation.ChannelId, remoteCluster.RemoteId).Return(sharedChannelRemote, nil)
+		mockStore.On("SharedChannel").Return(&mockSharedChannelStore)
+
+		defer mockApp.AssertExpectations(t)
+
+		err = scs.onReceiveChannelInvite(msg, remoteCluster, nil)
+		require.NoError(t, err)
+	})
+
 	t.Run("DM channels", func(t *testing.T) {
 		var testRemoteID = model.NewId()
 		testCases := []struct {
@@ -196,7 +290,7 @@ func TestOnReceiveChannelInvite(t *testing.T) {
 				}
 
 				mockStore := &mocks.Store{}
-				remoteCluster := &model.RemoteCluster{Name: "test3", CreatorId: model.NewId(), RemoteId: testRemoteID}
+				remoteCluster := &model.RemoteCluster{RemoteId: testRemoteID, Name: "test3", CreatorId: model.NewId()}
 				invitation := channelInviteMsg{
 					ChannelId:            model.NewId(),
 					TeamId:               model.NewId(),
@@ -225,6 +319,7 @@ func TestOnReceiveChannelInvite(t *testing.T) {
 				mockChannelStore.On("Get", invitation.ChannelId, true).Return(nil, errors.New("boom"))
 				mockChannelStore.On("GetByName", "", mockTypeString, true).Return(nil, &store.ErrNotFound{})
 
+				mockSharedChannelStore.On("GetRemoteByIds", invitation.ChannelId, remoteCluster.RemoteId).Return(nil, store.NewErrNotFound("SharedChannelRemote", ""))
 				mockSharedChannelStore.On("Save", mock.Anything).Return(nil, nil)
 				mockSharedChannelStore.On("SaveRemote", mock.Anything).Return(nil, nil)
 				mockStore.On("Channel").Return(&mockChannelStore)

server/platform/services/sharedchannel/service_api.go

@@ -170,7 +170,7 @@ func (scs *Service) InviteRemoteToChannel(channelID, remoteID, userID string, sh
 
 func (scs *Service) UninviteRemoteFromChannel(channelID, remoteID string) error {
 	scr, err := scs.server.GetStore().SharedChannel().GetRemoteByIds(channelID, remoteID)
-	if err != nil || scr.ChannelId != channelID {
+	if err != nil || scr.ChannelId != channelID || scr.DeleteAt != 0 {
 		return model.NewAppError("UninviteRemoteFromChannel", "api.command_share.channel_remote_id_not_exists",
 			map[string]any{"RemoteId": remoteID}, "", http.StatusInternalServerError)
 	}

server/platform/services/sharedchannel/sync_send.go

@@ -343,6 +343,9 @@ func (scs *Service) processTask(task syncTask) error {
 		if err != nil {
 			return err
 		}
+		if rc.DeleteAt != 0 {
+			return fmt.Errorf("Processing task for a deleted remote cluster '%s'", task.remoteID)
+		}
 		if !rc.IsOnline() {
 			return fmt.Errorf("Failed updating shared channel '%s' for offline remote cluster '%s'", task.channelID, rc.DisplayName)
 		}

server/platform/services/sharedchannel/sync_send_remote.go

@@ -119,6 +119,9 @@ func (scs *Service) syncForRemote(task syncTask, rc *model.RemoteCluster) error
 			mlog.String("remote", rc.DisplayName),
 			mlog.String("channel_id", task.channelID),
 		)
+	} else if err == nil && scr.DeleteAt != 0 {
+		// if SharedChannelRemote is deleted, regardless of the autoinvite flag, do nothing
+		return nil
 	} else if err != nil {
 		return err
 	}