MM-61033: [Shared Channels] Removing a shared channel on one end should make the other remove the shared channel too (#30738)

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)
+}

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)

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

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
+	}
+}