From b898d13e55214b04e21bc1f7d30286962fd9cd1f Mon Sep 17 00:00:00 2001 From: Miguel de la Cruz Date: Tue, 10 Sep 2024 23:39:07 +0200 Subject: [PATCH] Adds status sync to shared channels (#28020) * Adds status sync to shared channels To allow for status to be synced, this changes add a new type of shared channel internal task. This task has its contents pre-fetched and stored in the `existingMsg` property, and it is keyed with a user ID besides a channel ID, so it doesn't conflict with channel-driven synchronization tasks. All status synchronizations are triggered from the app layer, so there is no need of watching for new WebSocket events. Although right now we're only syncing one user status per message, the changes account for a list of statuses in case we want to batch them in the future. The feature is gated by a configuration property and can be disabled independently of the rest of Shared Channels if it's necessary. It is backwards compatible as well, and should cause no problems with servers running older Mattermost versions. * Adds status sync error management and retry * Adds DisableSharedChannelsStatusSync to the telemetry report --------- Co-authored-by: Mattermost Build --- server/channels/app/app_iface.go | 1 + .../app/opentracing/opentracing_layer.go | 15 ++ .../platform/shared_channel_service_iface.go | 1 + server/channels/app/platform/status.go | 27 +++- server/channels/app/shared_channel.go | 2 +- .../app/shared_channel_service_iface.go | 1 + server/channels/app/status.go | 9 ++ .../sharedchannel/mock_AppIface_test.go | 5 + .../services/sharedchannel/service.go | 1 + .../services/sharedchannel/sync_recv.go | 5 + .../services/sharedchannel/sync_send.go | 137 +++++++++++++++--- .../sharedchannel/sync_send_remote.go | 55 ++++++- .../platform/services/telemetry/telemetry.go | 5 +- server/public/model/config.go | 9 +- server/public/model/shared_channel.go | 3 + 15 files changed, 240 insertions(+), 36 deletions(-) diff --git a/server/channels/app/app_iface.go b/server/channels/app/app_iface.go index d92fa65b7e..57414df489 100644 --- a/server/channels/app/app_iface.go +++ b/server/channels/app/app_iface.go @@ -1057,6 +1057,7 @@ type AppIface interface { SaveAcknowledgementForPost(c request.CTX, postID, userID string) (*model.PostAcknowledgement, *model.AppError) SaveAdminNotification(userId string, notifyData *model.NotifyAdminToUpgradeRequest) *model.AppError SaveAdminNotifyData(data *model.NotifyAdminData) (*model.NotifyAdminData, *model.AppError) + SaveAndBroadcastStatus(status *model.Status) SaveBrandImage(rctx request.CTX, imageData *multipart.FileHeader) *model.AppError SaveComplianceReport(rctx request.CTX, job *model.Compliance) (*model.Compliance, *model.AppError) SaveReactionForPost(c request.CTX, reaction *model.Reaction) (*model.Reaction, *model.AppError) diff --git a/server/channels/app/opentracing/opentracing_layer.go b/server/channels/app/opentracing/opentracing_layer.go index b7ae597b11..cb77d69478 100644 --- a/server/channels/app/opentracing/opentracing_layer.go +++ b/server/channels/app/opentracing/opentracing_layer.go @@ -15305,6 +15305,21 @@ func (a *OpenTracingAppLayer) SaveAdminNotifyData(data *model.NotifyAdminData) ( return resultVar0, resultVar1 } +func (a *OpenTracingAppLayer) SaveAndBroadcastStatus(status *model.Status) { + origCtx := a.ctx + span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.SaveAndBroadcastStatus") + + a.ctx = newCtx + a.app.Srv().Store().SetContext(newCtx) + defer func() { + a.app.Srv().Store().SetContext(origCtx) + a.ctx = origCtx + }() + + defer span.Finish() + a.app.SaveAndBroadcastStatus(status) +} + func (a *OpenTracingAppLayer) SaveBrandImage(rctx request.CTX, imageData *multipart.FileHeader) *model.AppError { origCtx := a.ctx span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.SaveBrandImage") diff --git a/server/channels/app/platform/shared_channel_service_iface.go b/server/channels/app/platform/shared_channel_service_iface.go index 2d915478d8..63ee643c24 100644 --- a/server/channels/app/platform/shared_channel_service_iface.go +++ b/server/channels/app/platform/shared_channel_service_iface.go @@ -14,6 +14,7 @@ type SharedChannelServiceIFace interface { Start() error NotifyChannelChanged(channelId string) NotifyUserProfileChanged(userID string) + NotifyUserStatusChanged(status *model.Status) SendChannelInvite(channel *model.Channel, userId string, rc *model.RemoteCluster, options ...sharedchannel.InviteOption) error Active() bool InviteRemoteToChannel(channelID, remoteID, userID string, shareIfNotShared bool) error diff --git a/server/channels/app/platform/status.go b/server/channels/app/platform/status.go index 6014ba62e7..57ce185e91 100644 --- a/server/channels/app/platform/status.go +++ b/server/channels/app/platform/status.go @@ -279,6 +279,9 @@ func (ps *PlatformService) SetStatusLastActivityAt(userID string, activityAt int ps.AddStatusCacheSkipClusterSend(status) ps.SetStatusAwayIfNeeded(userID, false) + if ps.sharedChannelService != nil { + ps.sharedChannelService.NotifyUserStatusChanged(status) + } } func (ps *PlatformService) UpdateLastActivityAtIfNeeded(session model.Session) { @@ -346,6 +349,9 @@ func (ps *PlatformService) SetStatusOnline(userID string, manual bool) { mlog.Error("Failed to save status", mlog.String("user_id", userID), mlog.Err(err), mlog.String("user_id", userID)) } } + if ps.sharedChannelService != nil { + ps.sharedChannelService.NotifyUserStatusChanged(status) + } } if broadcast { @@ -366,6 +372,9 @@ func (ps *PlatformService) SetStatusOffline(userID string, manual bool) { status = &model.Status{UserId: userID, Status: model.StatusOffline, Manual: manual, LastActivityAt: model.GetMillis(), ActiveChannel: ""} ps.SaveAndBroadcastStatus(status) + if ps.sharedChannelService != nil { + ps.sharedChannelService.NotifyUserStatusChanged(status) + } } func (ps *PlatformService) SetStatusAwayIfNeeded(userID string, manual bool) { @@ -398,19 +407,22 @@ func (ps *PlatformService) SetStatusAwayIfNeeded(userID string, manual bool) { status.ActiveChannel = "" ps.SaveAndBroadcastStatus(status) + if ps.sharedChannelService != nil { + ps.sharedChannelService.NotifyUserStatusChanged(status) + } } // SetStatusDoNotDisturbTimed takes endtime in unix epoch format in UTC // and sets status of given userId to dnd which will be restored back after endtime -func (ps *PlatformService) SetStatusDoNotDisturbTimed(userId string, endtime int64) { +func (ps *PlatformService) SetStatusDoNotDisturbTimed(userID string, endtime int64) { if !*ps.Config().ServiceSettings.EnableUserStatuses { return } - status, err := ps.GetStatus(userId) + status, err := ps.GetStatus(userID) if err != nil { - status = &model.Status{UserId: userId, Status: model.StatusOffline, Manual: false, LastActivityAt: 0, ActiveChannel: ""} + status = &model.Status{UserId: userID, Status: model.StatusOffline, Manual: false, LastActivityAt: 0, ActiveChannel: ""} } status.PrevStatus = status.Status @@ -420,6 +432,9 @@ func (ps *PlatformService) SetStatusDoNotDisturbTimed(userId string, endtime int status.DNDEndTime = endtime ps.SaveAndBroadcastStatus(status) + if ps.sharedChannelService != nil { + ps.sharedChannelService.NotifyUserStatusChanged(status) + } } func (ps *PlatformService) SetStatusDoNotDisturb(userID string) { @@ -437,6 +452,9 @@ func (ps *PlatformService) SetStatusDoNotDisturb(userID string) { status.Manual = true ps.SaveAndBroadcastStatus(status) + if ps.sharedChannelService != nil { + ps.sharedChannelService.NotifyUserStatusChanged(status) + } } func (ps *PlatformService) SetStatusOutOfOffice(userID string) { @@ -454,6 +472,9 @@ func (ps *PlatformService) SetStatusOutOfOffice(userID string) { status.Manual = true ps.SaveAndBroadcastStatus(status) + if ps.sharedChannelService != nil { + ps.sharedChannelService.NotifyUserStatusChanged(status) + } } func (ps *PlatformService) isUserAway(lastActivityAt int64) bool { diff --git a/server/channels/app/shared_channel.go b/server/channels/app/shared_channel.go index 2cfe7841c3..ea736916a6 100644 --- a/server/channels/app/shared_channel.go +++ b/server/channels/app/shared_channel.go @@ -19,7 +19,7 @@ import ( func (a *App) getSharedChannelsService() (SharedChannelServiceIFace, error) { scService := a.Srv().GetSharedChannelSyncService() if scService == nil || !scService.Active() { - return nil, model.NewAppError("InviteRemoteToChannel", "api.command_share.service_disabled", + return nil, model.NewAppError("getSharedChannelsService", "api.command_share.service_disabled", nil, "", http.StatusBadRequest) } return scService, nil diff --git a/server/channels/app/shared_channel_service_iface.go b/server/channels/app/shared_channel_service_iface.go index 509e5464cb..2ab50c14a5 100644 --- a/server/channels/app/shared_channel_service_iface.go +++ b/server/channels/app/shared_channel_service_iface.go @@ -15,6 +15,7 @@ type SharedChannelServiceIFace interface { Start() error NotifyChannelChanged(channelId string) NotifyUserProfileChanged(userID string) + NotifyUserStatusChanged(status *model.Status) SendChannelInvite(channel *model.Channel, userId string, rc *model.RemoteCluster, options ...sharedchannel.InviteOption) error Active() bool InviteRemoteToChannel(channelID, remoteID, userID string, shareIfNotShared bool) error diff --git a/server/channels/app/status.go b/server/channels/app/status.go index b99a682cda..163d43bee1 100644 --- a/server/channels/app/status.go +++ b/server/channels/app/status.go @@ -50,6 +50,10 @@ func (a *App) SetStatusOutOfOffice(userID string) { a.Srv().Platform().SetStatusOutOfOffice(userID) } +func (a *App) SaveAndBroadcastStatus(status *model.Status) { + a.Srv().Platform().SaveAndBroadcastStatus(status) +} + func (a *App) GetStatusFromCache(userID string) *model.Status { return a.Srv().Platform().GetStatusFromCache(userID) } @@ -66,9 +70,14 @@ func (a *App) UpdateDNDStatusOfUsers() { mlog.Warn("Failed to fetch dnd statues from store", mlog.String("err", err.Error())) return } + + scs, _ := a.getSharedChannelsService() for i := range statuses { a.Srv().Platform().AddStatusCache(statuses[i]) a.Srv().Platform().BroadcastStatus(statuses[i]) + if scs != nil { + scs.NotifyUserStatusChanged(statuses[i]) + } } } diff --git a/server/platform/services/sharedchannel/mock_AppIface_test.go b/server/platform/services/sharedchannel/mock_AppIface_test.go index 34c092b313..7b11b26675 100644 --- a/server/platform/services/sharedchannel/mock_AppIface_test.go +++ b/server/platform/services/sharedchannel/mock_AppIface_test.go @@ -474,6 +474,11 @@ func (_m *MockAppIface) Publish(message *model.WebSocketEvent) { _m.Called(message) } +// SaveAndBroadcastStatus provides a mock function with given fields: status +func (_m *MockAppIface) SaveAndBroadcastStatus(status *model.Status) { + _m.Called(status) +} + // SaveReactionForPost provides a mock function with given fields: c, reaction func (_m *MockAppIface) SaveReactionForPost(c request.CTX, reaction *model.Reaction) (*model.Reaction, *model.AppError) { ret := _m.Called(c, reaction) diff --git a/server/platform/services/sharedchannel/service.go b/server/platform/services/sharedchannel/service.go index 04044c3a03..e1848709b3 100644 --- a/server/platform/services/sharedchannel/service.go +++ b/server/platform/services/sharedchannel/service.go @@ -62,6 +62,7 @@ type AppIface interface { DeletePost(c request.CTX, postID, deleteByID string) (*model.Post, *model.AppError) SaveReactionForPost(c request.CTX, reaction *model.Reaction) (*model.Reaction, *model.AppError) DeleteReactionForPost(c request.CTX, reaction *model.Reaction) *model.AppError + SaveAndBroadcastStatus(status *model.Status) PatchChannelModerationsForChannel(c request.CTX, channel *model.Channel, channelModerationsPatch []*model.ChannelModerationPatch) ([]*model.ChannelModeration, *model.AppError) CreateUploadSession(c request.CTX, us *model.UploadSession) (*model.UploadSession, *model.AppError) FileReader(path string) (filestore.ReadCloseSeeker, *model.AppError) diff --git a/server/platform/services/sharedchannel/sync_recv.go b/server/platform/services/sharedchannel/sync_recv.go index 207f5c6a6a..a4e125ec59 100644 --- a/server/platform/services/sharedchannel/sync_recv.go +++ b/server/platform/services/sharedchannel/sync_recv.go @@ -65,6 +65,7 @@ func (scs *Service) processSyncMessage(c request.CTX, syncMsg *model.SyncMsg, rc mlog.Int("user_count", len(syncMsg.Users)), mlog.Int("post_count", len(syncMsg.Posts)), mlog.Int("reaction_count", len(syncMsg.Reactions)), + mlog.Int("status_count", len(syncMsg.Statuses)), ) if targetChannel, err = scs.server.GetStore().Channel().Get(syncMsg.ChannelId, true); err != nil { @@ -175,6 +176,10 @@ func (scs *Service) processSyncMessage(c request.CTX, syncMsg *model.SyncMsg, rc } } + for _, status := range syncMsg.Statuses { + scs.app.SaveAndBroadcastStatus(status) + } + response.SetPayload(syncResp) return nil diff --git a/server/platform/services/sharedchannel/sync_send.go b/server/platform/services/sharedchannel/sync_send.go index ca8a76ce52..82c4f4cfdb 100644 --- a/server/platform/services/sharedchannel/sync_send.go +++ b/server/platform/services/sharedchannel/sync_send.go @@ -16,27 +16,35 @@ import ( ) type syncTask struct { - id string - channelID string - remoteID string - AddedAt time.Time - retryCount int - retryMsg *model.SyncMsg - schedule time.Time + id string + channelID string + userID string + remoteID string + AddedAt time.Time + // existingMsg is used to add information to the task on creation + // instead of waiting until the task is processed to fetch it. If + // a new task with the same ID is scheduled, its existingMsg will + // replace the previous one + existingMsg *model.SyncMsg + retryCount int + retryMsg *model.SyncMsg + schedule time.Time } -func newSyncTask(channelID string, remoteID string, retryMsg *model.SyncMsg) syncTask { +func newSyncTask(channelID, userID string, remoteID string, existingMsg, retryMsg *model.SyncMsg) syncTask { var retryID string if retryMsg != nil { retryID = retryMsg.Id } return syncTask{ - id: channelID + remoteID + retryID, // combination of ids to avoid duplicates - channelID: channelID, - remoteID: remoteID, // empty means update all remote clusters - retryMsg: retryMsg, - schedule: time.Now(), + id: channelID + userID + remoteID + retryID, // combination of ids to avoid duplicates + channelID: channelID, + userID: userID, + remoteID: remoteID, // empty means update all remote clusters + existingMsg: existingMsg, + retryMsg: retryMsg, + schedule: time.Now(), } } @@ -53,13 +61,13 @@ func (scs *Service) NotifyChannelChanged(channelID string) { return } - task := newSyncTask(channelID, "", nil) + task := newSyncTask(channelID, "", "", nil, nil) task.schedule = time.Now().Add(NotifyMinimumDelay) scs.addTask(task) } -// NotifyUserProfileChanged is called to indicate that a user belonging to at least one -// shared channel has modified their user profile (name, username, email, custom status, profile image) +// NotifyUserProfileChanged is called to indicate that a user has modified their user +// profile (name, username, email, custom status, profile image) func (scs *Service) NotifyUserProfileChanged(userID string) { if rcs := scs.server.GetRemoteClusterService(); rcs == nil { return @@ -80,15 +88,63 @@ func (scs *Service) NotifyUserProfileChanged(userID string) { notified := make(map[string]struct{}) for _, user := range scusers { - // update every channel + remote combination they belong to. + // update every user + remote combination they belong to. // Redundant updates (ie. to same remote for multiple channels) will be // filtered out. - combo := user.ChannelId + user.RemoteId + + combo := user.UserId + user.RemoteId if _, ok := notified[combo]; ok { continue } notified[combo] = struct{}{} - task := newSyncTask(user.ChannelId, user.RemoteId, nil) + task := newSyncTask(user.ChannelId, "", user.RemoteId, nil, nil) + task.schedule = time.Now().Add(NotifyMinimumDelay) + scs.addTask(task) + } +} + +// NotifyUserStatusChanged is called to indicate that a user has modified their status +func (scs *Service) NotifyUserStatusChanged(status *model.Status) { + if rcs := scs.server.GetRemoteClusterService(); rcs == nil { + return + } + + if *scs.server.Config().ConnectedWorkspacesSettings.DisableSharedChannelsStatusSync { + return + } + + if status.UserId == "" { + scs.server.Log().Log(mlog.LvlSharedChannelServiceError, "Received invalid status for sync", + mlog.String("userID", status.UserId), + ) + return + } + + scusers, err := scs.server.GetStore().SharedChannel().GetUsersForUser(status.UserId) + if err != nil { + scs.server.Log().Log(mlog.LvlSharedChannelServiceError, "Failed to fetch shared channel users", + mlog.String("userID", status.UserId), + mlog.Err(err), + ) + return + } + if len(scusers) == 0 { + return + } + + existingMsg := &model.SyncMsg{Statuses: []*model.Status{status}} + notified := make(map[string]struct{}) + + for _, user := range scusers { + // update every user + remote combination they belong to. + // Redundant updates (ie. to same remote for multiple channels) will be + // filtered out. + combo := user.UserId + user.RemoteId + if _, ok := notified[combo]; ok { + continue + } + notified[combo] = struct{}{} + task := newSyncTask(user.ChannelId, user.UserId, user.RemoteId, existingMsg, nil) task.schedule = time.Now().Add(NotifyMinimumDelay) scs.addTask(task) } @@ -115,7 +171,7 @@ func (scs *Service) ForceSyncForRemote(rc *model.RemoteCluster) { } for _, scr := range scrs { - task := newSyncTask(scr.ChannelId, rc.RemoteId, nil) + task := newSyncTask(scr.ChannelId, "", rc.RemoteId, nil, nil) task.schedule = time.Now().Add(NotifyMinimumDelay) scs.addTask(task) } @@ -125,7 +181,12 @@ func (scs *Service) ForceSyncForRemote(rc *model.RemoteCluster) { func (scs *Service) addTask(task syncTask) { task.AddedAt = time.Now() scs.mux.Lock() - if _, ok := scs.tasks[task.id]; !ok { + if originalTask, ok := scs.tasks[task.id]; ok { + // if the task was already scheduled, we only update the + // existingMsg in case there is new information + originalTask.existingMsg = task.existingMsg + scs.tasks[task.id] = originalTask + } else { scs.tasks[task.id] = task } scs.mux.Unlock() @@ -327,6 +388,7 @@ func (scs *Service) handlePostError(postId string, task syncTask, rc *model.Remo scs.server.Log().Log(mlog.LvlSharedChannelServiceError, "error fetching post for sync retry", mlog.String("remote", rc.DisplayName), mlog.String("post_id", postId), + mlog.Err(err), ) return } @@ -334,7 +396,38 @@ func (scs *Service) handlePostError(postId string, task syncTask, rc *model.Remo syncMsg := model.NewSyncMsg(task.channelID) syncMsg.Posts = []*model.Post{post} - scs.addTask(newSyncTask(task.channelID, task.remoteID, syncMsg)) + scs.addTask(newSyncTask(task.channelID, task.userID, task.remoteID, nil, syncMsg)) +} + +func (scs *Service) handleStatusError(userId string, task syncTask, rc *model.RemoteCluster) { + if task.retryMsg != nil && len(task.retryMsg.Statuses) == 1 && task.retryMsg.Statuses[0].UserId == userId { + // this was a retry for specific status that failed previously. Try again if within MaxRetries. + if task.incRetry() { + scs.addTask(task) + } else { + scs.server.Log().Log(mlog.LvlSharedChannelServiceError, "error syncing status", + mlog.String("remote", rc.DisplayName), + mlog.String("user_id", userId), + ) + } + return + } + + // this status failed as part of a group of statuses. Retry as an individual status. + status, err := scs.server.GetStore().Status().Get(userId) + if err != nil { + scs.server.Log().Log(mlog.LvlSharedChannelServiceError, "error fetching status for sync retry", + mlog.String("remote", rc.DisplayName), + mlog.String("user_id", userId), + mlog.Err(err), + ) + return + } + + syncMsg := model.NewSyncMsg(task.channelID) + syncMsg.Statuses = []*model.Status{status} + + scs.addTask(newSyncTask(task.channelID, task.userID, task.remoteID, nil, syncMsg)) } // notifyRemoteOffline creates an ephemeral post to the author for any posts created recently to remotes diff --git a/server/platform/services/sharedchannel/sync_send_remote.go b/server/platform/services/sharedchannel/sync_send_remote.go index c7999f8f5e..d888469d3a 100644 --- a/server/platform/services/sharedchannel/sync_send_remote.go +++ b/server/platform/services/sharedchannel/sync_send_remote.go @@ -35,6 +35,7 @@ type syncData struct { profileImages map[string]*model.User posts []*model.Post reactions []*model.Reaction + statuses []*model.Status attachments []attachment resultRepeat bool @@ -68,6 +69,13 @@ func (sd *syncData) isCursorChanged() bool { sd.scr.LastPostUpdateAt != sd.resultNextCursor.LastPostUpdateAt || sd.scr.LastPostUpdateID != sd.resultNextCursor.LastPostUpdateID } +func (sd *syncData) setDataFromMsg(msg *model.SyncMsg) { + sd.users = msg.Users + sd.posts = msg.Posts + sd.reactions = msg.Reactions + sd.statuses = msg.Statuses +} + // syncForRemote updates a remote cluster with any new posts/reactions for a specific // channel. If many changes are found, only the oldest X changes are sent and the channel // is re-added to the task map. This ensures no channels are starved for updates even if some @@ -115,21 +123,30 @@ func (scs *Service) syncForRemote(task syncTask, rc *model.RemoteCluster) error return err } + sd := newSyncData(task, rc, scr) + // if this is retrying a failed msg, just send it again. if task.retryMsg != nil { - sd := newSyncData(task, rc, scr) - sd.users = task.retryMsg.Users - sd.posts = task.retryMsg.Posts - sd.reactions = task.retryMsg.Reactions + sd.setDataFromMsg(task.retryMsg) return scs.sendSyncData(sd) } - sd := newSyncData(task, rc, scr) + // if this has an already existing msg, just send it right away + if task.existingMsg != nil { + sd.setDataFromMsg(task.existingMsg) + return scs.sendSyncData(sd) + } + + // if we don't have a channelID at this point, we cannot fetch new + // data from the database + if task.channelID == "" { + return fmt.Errorf("task doesn't have prefetched data nor a channel ID set") + } // schedule another sync if the repeat flag is set at some point. defer func(rpt *bool) { if *rpt { - scs.addTask(newSyncTask(task.channelID, task.remoteID, nil)) + scs.addTask(newSyncTask(task.channelID, task.userID, task.remoteID, nil, nil)) } }(&sd.resultRepeat) @@ -516,6 +533,13 @@ func (scs *Service) sendSyncData(sd *syncData) error { } } + // send statuses + if len(sd.statuses) != 0 { + if err := scs.sendStatusSyncData(sd); err != nil { + merr.Append(fmt.Errorf("cannot send status sync data: %w", err)) + } + } + // send user profile images if len(sd.profileImages) != 0 { scs.sendProfileImageSyncData(sd) @@ -624,6 +648,25 @@ func (scs *Service) sendReactionSyncData(sd *syncData) error { }) } +// sendStatusSyncData sends the collected status updates to the remote cluster. +func (scs *Service) sendStatusSyncData(sd *syncData) error { + msg := model.NewSyncMsg(sd.task.channelID) + msg.Statuses = sd.statuses + + return scs.sendSyncMsgToRemote(msg, sd.rc, func(syncResp model.SyncResponse, errResp error) { + if len(syncResp.StatusErrors) != 0 { + scs.server.Log().Log(mlog.LvlSharedChannelServiceError, "Response indicates error from status(es) sync", + mlog.String("remote_id", sd.rc.RemoteId), + mlog.Array("user_ids", syncResp.StatusErrors), + ) + + for _, userID := range syncResp.StatusErrors { + scs.handleStatusError(userID, sd.task, sd.rc) + } + } + }) +} + // sendProfileImageSyncData sends the collected user profile image updates to the remote cluster. func (scs *Service) sendProfileImageSyncData(sd *syncData) { for _, user := range sd.profileImages { diff --git a/server/platform/services/telemetry/telemetry.go b/server/platform/services/telemetry/telemetry.go index df3fed43cc..ed94ac4580 100644 --- a/server/platform/services/telemetry/telemetry.go +++ b/server/platform/services/telemetry/telemetry.go @@ -891,8 +891,9 @@ func (ts *TelemetryService) trackConfig() { }) ts.SendTelemetry(TrackConfigConnectedWorkspaces, map[string]any{ - "enable_shared_channels": *cfg.ConnectedWorkspacesSettings.EnableSharedChannels, - "enable_remote_cluster_service": *cfg.ConnectedWorkspacesSettings.EnableRemoteClusterService && cfg.FeatureFlags.EnableRemoteClusterService, + "enable_shared_channels": *cfg.ConnectedWorkspacesSettings.EnableSharedChannels, + "enable_remote_cluster_service": *cfg.ConnectedWorkspacesSettings.EnableRemoteClusterService && cfg.FeatureFlags.EnableRemoteClusterService, + "disable_shared_channels_status_sync": *cfg.ConnectedWorkspacesSettings.DisableSharedChannelsStatusSync, }) // Convert feature flags to map[string]any for sending diff --git a/server/public/model/config.go b/server/public/model/config.go index 4eb2b4e8c1..6d64caa9a9 100644 --- a/server/public/model/config.go +++ b/server/public/model/config.go @@ -3243,8 +3243,9 @@ func (w *WranglerSettings) IsValid() *AppError { } type ConnectedWorkspacesSettings struct { - EnableSharedChannels *bool - EnableRemoteClusterService *bool + EnableSharedChannels *bool + EnableRemoteClusterService *bool + DisableSharedChannelsStatusSync *bool } func (c *ConnectedWorkspacesSettings) SetDefaults(isUpdate bool, e ExperimentalSettings) { @@ -3263,6 +3264,10 @@ func (c *ConnectedWorkspacesSettings) SetDefaults(isUpdate bool, e ExperimentalS c.EnableRemoteClusterService = NewPointer(false) } } + + if c.DisableSharedChannelsStatusSync == nil { + c.DisableSharedChannelsStatusSync = NewPointer(false) + } } type GlobalRelayMessageExportSettings struct { diff --git a/server/public/model/shared_channel.go b/server/public/model/shared_channel.go index 03794b9859..ac870fa4c0 100644 --- a/server/public/model/shared_channel.go +++ b/server/public/model/shared_channel.go @@ -275,6 +275,7 @@ type SyncMsg struct { Users map[string]*User `json:"users,omitempty"` Posts []*Post `json:"posts,omitempty"` Reactions []*Reaction `json:"reactions,omitempty"` + Statuses []*Status `json:"statuses,omitempty"` } func NewSyncMsg(channelID string) *SyncMsg { @@ -311,6 +312,8 @@ type SyncResponse struct { ReactionsLastUpdateAt int64 `json:"reactions_last_update_at"` ReactionErrors []string `json:"reaction_errors"` + + StatusErrors []string `json:"status_errors"` // user IDs for which the status sync failed } // RegisterPluginOpts is passed by plugins to the `RegisterPluginForSharedChannels` plugin API