MM-34549 shared channels; add users to channel that were already sync'd (#17361)

Fixes a bug and adds a feature for shared channels:
- The Bug: when creating new shared channels, users that had already been sync'd via another channel were not added to the new channel's member list, since the users were not sync'd again. This PR sync's users per channel.
- The Feature: support custom statuses

services/sharedchannel/msg.go

@@ -52,10 +52,10 @@ func (u userCache) Add(id string) {
 
 // postsToSyncMessages takes a slice of posts and converts to a `RemoteClusterMsg` which can be
 // sent to a remote cluster.
-func (scs *Service) postsToSyncMessages(posts []*model.Post, rc *model.RemoteCluster, nextSyncAt int64) ([]syncMsg, error) {
+func (scs *Service) postsToSyncMessages(posts []*model.Post, channelID string, rc *model.RemoteCluster, nextSyncAt int64) ([]syncMsg, error) {
 	syncMessages := make([]syncMsg, 0, len(posts))
 
-	var teamId string
+	var teamID string
 	uCache := make(userCache)
 
 	for _, p := range posts {
@@ -64,7 +64,7 @@ func (scs *Service) postsToSyncMessages(posts []*model.Post, rc *model.RemoteClu
 		}
 
 		// lookup team id once
-		if teamId == "" {
+		if teamID == "" {
 			sc, err := scs.server.GetStore().SharedChannel().Get(p.ChannelId)
 			if err != nil {
 				scs.server.GetLogger().Log(mlog.LvlSharedChannelServiceError, "Could not get shared channel for post",
@@ -73,7 +73,7 @@ func (scs *Service) postsToSyncMessages(posts []*model.Post, rc *model.RemoteClu
 				)
 				continue
 			}
-			teamId = sc.TeamId
+			teamID = sc.TeamId
 		}
 
 		// any reactions originating from the remote cluster are filtered out
@@ -119,7 +119,7 @@ func (scs *Service) postsToSyncMessages(posts []*model.Post, rc *model.RemoteClu
 		}
 
 		// any users originating from the remote cluster are filtered out
-		users := scs.usersForPost(postSync, reactions, teamId, rc, uCache)
+		users := scs.usersForPost(postSync, reactions, channelID, teamID, rc, uCache)
 
 		// if everything was filtered out then don't send an empty message.
 		if postSync == nil && len(reactions) == 0 && len(users) == 0 {
@@ -142,7 +142,7 @@ func (scs *Service) postsToSyncMessages(posts []*model.Post, rc *model.RemoteClu
 // usersForPost provides a list of Users associated with the post that need to be synchronized.
 // The user cache ensures the same user is not synchronized redundantly if they appear in multiple
 // posts for this sync batch.
-func (scs *Service) usersForPost(post *model.Post, reactions []*model.Reaction, teamID string, rc *model.RemoteCluster, uCache userCache) []*model.User {
+func (scs *Service) usersForPost(post *model.Post, reactions []*model.Reaction, channelID string, teamID string, rc *model.RemoteCluster, uCache userCache) []*model.User {
 	userIds := make([]string, 0)
 	var mentionMap model.UserMentionMap
 
@@ -178,7 +178,7 @@ func (scs *Service) usersForPost(post *model.Post, reactions []*model.Reaction,
 	for _, id := range userIds {
 		user, err := scs.server.GetStore().User().Get(context.Background(), id)
 		if err == nil {
-			if sync, err2 := scs.shouldUserSync(user, rc); err2 != nil {
+			if sync, err2 := scs.shouldUserSync(user, channelID, rc); err2 != nil {
 				scs.server.GetLogger().Log(mlog.LvlSharedChannelServiceError, "Could not find user for post",
 					mlog.String("user_id", id),
 					mlog.Err(err2),
@@ -238,15 +238,15 @@ func sanitizeUserForSync(user *model.User) *model.User {
 }
 
 // shouldUserSync determines if a user needs to be synchronized.
-// User should be synchronized if it has no entry in the SharedChannelUsers table,
+// User should be synchronized if it has no entry in the SharedChannelUsers table for the specified channel,
 // or there is an entry but the LastSyncAt is less than user.UpdateAt
-func (scs *Service) shouldUserSync(user *model.User, rc *model.RemoteCluster) (bool, error) {
+func (scs *Service) shouldUserSync(user *model.User, channelID string, rc *model.RemoteCluster) (bool, error) {
 	// don't sync users with the remote they originated from.
 	if user.RemoteId != nil && *user.RemoteId == rc.RemoteId {
 		return false, nil
 	}
 
-	scu, err := scs.server.GetStore().SharedChannel().GetUser(user.Id, rc.RemoteId)
+	scu, err := scs.server.GetStore().SharedChannel().GetUser(user.Id, channelID, rc.RemoteId)
 	if err != nil {
 		if _, ok := err.(errNotFound); !ok {
 			return false, err
@@ -254,13 +254,16 @@ func (scs *Service) shouldUserSync(user *model.User, rc *model.RemoteCluster) (b
 
 		// user not in the SharedChannelUsers table, so we must add them.
 		scu = &model.SharedChannelUser{
-			UserId:   user.Id,
-			RemoteId: rc.RemoteId,
+			UserId:    user.Id,
+			RemoteId:  rc.RemoteId,
+			ChannelId: channelID,
 		}
 		if _, err = scs.server.GetStore().SharedChannel().SaveUser(scu); err != nil {
 			scs.server.GetLogger().Log(mlog.LvlSharedChannelServiceError, "Error adding user to shared channel users",
 				mlog.String("remote_id", rc.RemoteId),
 				mlog.String("user_id", user.Id),
+				mlog.String("channel_id", user.Id),
+				mlog.Err(err),
 			)
 		}
 	} else if scu.LastSyncAt >= user.UpdateAt {