diff --git a/server/channels/app/app_iface.go b/server/channels/app/app_iface.go index 4b09fe5f17..2ef5147fb0 100644 --- a/server/channels/app/app_iface.go +++ b/server/channels/app/app_iface.go @@ -271,9 +271,15 @@ type AppIface interface { MoveChannel(c request.CTX, team *model.Team, channel *model.Channel, user *model.User) *model.AppError // NotifySessionsExpired is called periodically from the job server to notify any mobile sessions that have expired. NotifySessionsExpired() error - // OnSharedChannelsPing is called by the Shared Channels service for a registered plugin wto check that the plugin + // OnSharedChannelsAttachmentSyncMsg is called by the Shared Channels service for a registered plugin when a file attachment + // needs to be synchronized. + OnSharedChannelsAttachmentSyncMsg(fi *model.FileInfo, post *model.Post, rc *model.RemoteCluster) error + // OnSharedChannelsPing is called by the Shared Channels service for a registered plugin to check that the plugin // is still responding and has a connection to any upstream services it needs (e.g. MS Graph API). OnSharedChannelsPing(rc *model.RemoteCluster) bool + // OnSharedChannelsProfileImageSyncMsg is called by the Shared Channels service for a registered plugin when a user's + // profile image needs to be synchronized. + OnSharedChannelsProfileImageSyncMsg(user *model.User, rc *model.RemoteCluster) error // OnSharedChannelsSyncMsg is called by the Shared Channels service for a registered plugin when there is new content // that needs to be synchronized. OnSharedChannelsSyncMsg(msg *model.SyncMsg, rc *model.RemoteCluster) (model.SyncResponse, error) diff --git a/server/channels/app/opentracing/opentracing_layer.go b/server/channels/app/opentracing/opentracing_layer.go index 736aac36fb..76777673b9 100644 --- a/server/channels/app/opentracing/opentracing_layer.go +++ b/server/channels/app/opentracing/opentracing_layer.go @@ -12858,6 +12858,28 @@ func (a *OpenTracingAppLayer) NotifySharedChannelUserUpdate(user *model.User) { a.app.NotifySharedChannelUserUpdate(user) } +func (a *OpenTracingAppLayer) OnSharedChannelsAttachmentSyncMsg(fi *model.FileInfo, post *model.Post, rc *model.RemoteCluster) error { + origCtx := a.ctx + span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.OnSharedChannelsAttachmentSyncMsg") + + a.ctx = newCtx + a.app.Srv().Store().SetContext(newCtx) + defer func() { + a.app.Srv().Store().SetContext(origCtx) + a.ctx = origCtx + }() + + defer span.Finish() + resultVar0 := a.app.OnSharedChannelsAttachmentSyncMsg(fi, post, rc) + + if resultVar0 != nil { + span.LogFields(spanlog.Error(resultVar0)) + ext.Error.Set(span, true) + } + + return resultVar0 +} + func (a *OpenTracingAppLayer) OnSharedChannelsPing(rc *model.RemoteCluster) bool { origCtx := a.ctx span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.OnSharedChannelsPing") @@ -12875,6 +12897,28 @@ func (a *OpenTracingAppLayer) OnSharedChannelsPing(rc *model.RemoteCluster) bool return resultVar0 } +func (a *OpenTracingAppLayer) OnSharedChannelsProfileImageSyncMsg(user *model.User, rc *model.RemoteCluster) error { + origCtx := a.ctx + span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.OnSharedChannelsProfileImageSyncMsg") + + a.ctx = newCtx + a.app.Srv().Store().SetContext(newCtx) + defer func() { + a.app.Srv().Store().SetContext(origCtx) + a.ctx = origCtx + }() + + defer span.Finish() + resultVar0 := a.app.OnSharedChannelsProfileImageSyncMsg(user, rc) + + if resultVar0 != nil { + span.LogFields(spanlog.Error(resultVar0)) + ext.Error.Set(span, true) + } + + return resultVar0 +} + func (a *OpenTracingAppLayer) OnSharedChannelsSyncMsg(msg *model.SyncMsg, rc *model.RemoteCluster) (model.SyncResponse, error) { origCtx := a.ctx span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.OnSharedChannelsSyncMsg") diff --git a/server/channels/app/remote_cluster.go b/server/channels/app/remote_cluster.go index 77ff95cf4d..152ddddb43 100644 --- a/server/channels/app/remote_cluster.go +++ b/server/channels/app/remote_cluster.go @@ -13,6 +13,7 @@ import ( "github.com/mattermost/mattermost/server/v8/platform/services/remotecluster" "github.com/mattermost/mattermost/server/public/model" + "github.com/mattermost/mattermost/server/public/shared/mlog" ) func (a *App) RegisterPluginForSharedChannels(opts model.RegisterPluginOpts) (remoteID string, err error) { @@ -27,6 +28,11 @@ func (a *App) RegisterPluginForSharedChannels(opts model.RegisterPluginOpts) (re // if plugin is already registered then treat this as an update. if rc != nil { + a.Log().Debug("Plugin already registered for Shared Channels", + mlog.String("plugin_id", opts.PluginID), + mlog.String("remote_id", rc.RemoteId), + ) + rc.DisplayName = opts.Displayname rc.Options = opts.GetOptionFlags() @@ -51,6 +57,11 @@ func (a *App) RegisterPluginForSharedChannels(opts model.RegisterPluginOpts) (re return "", err } + a.Log().Debug("Registered new plugin for Shared Channels", + mlog.String("plugin_id", opts.PluginID), + mlog.String("remote_id", rcSaved.RemoteId), + ) + return rcSaved.RemoteId, nil } diff --git a/server/channels/app/shared_channel.go b/server/channels/app/shared_channel.go index 8c1b95cb24..18167d9234 100644 --- a/server/channels/app/shared_channel.go +++ b/server/channels/app/shared_channel.go @@ -9,6 +9,7 @@ import ( "net/http" "github.com/mattermost/mattermost/server/public/model" + "github.com/mattermost/mattermost/server/public/plugin" "github.com/mattermost/mattermost/server/public/shared/mlog" "github.com/mattermost/mattermost/server/public/shared/request" "github.com/mattermost/mattermost/server/v8/channels/store" @@ -301,17 +302,19 @@ func (a *App) SyncSharedChannel(channelID string) error { // Hooks -var ErrPluginUnavailable = errors.New("plugin unavialable") +var ErrPluginUnavailable = errors.New("plugin unavailable") + +func getPluginHooks(env *plugin.Environment, pluginID string) (plugin.Hooks, error) { + if env == nil { + return nil, ErrPluginUnavailable + } + return env.HooksForPlugin(pluginID) +} // OnSharedChannelsSyncMsg is called by the Shared Channels service for a registered plugin when there is new content // that needs to be synchronized. func (a *App) OnSharedChannelsSyncMsg(msg *model.SyncMsg, rc *model.RemoteCluster) (model.SyncResponse, error) { - pluginsEnvironment := a.GetPluginsEnvironment() - if pluginsEnvironment == nil { - return model.SyncResponse{}, fmt.Errorf("cannot deliver sync msg to plugin %s: %w", rc.PluginID, ErrPluginUnavailable) - } - - pluginHooks, err := pluginsEnvironment.HooksForPlugin(rc.PluginID) + pluginHooks, err := getPluginHooks(a.GetPluginsEnvironment(), rc.PluginID) if err != nil { return model.SyncResponse{}, fmt.Errorf("cannot deliver sync msg to plugin %s: %w", rc.PluginID, err) } @@ -319,16 +322,10 @@ func (a *App) OnSharedChannelsSyncMsg(msg *model.SyncMsg, rc *model.RemoteCluste return pluginHooks.OnSharedChannelsSyncMsg(msg, rc) } -// OnSharedChannelsPing is called by the Shared Channels service for a registered plugin wto check that the plugin +// OnSharedChannelsPing is called by the Shared Channels service for a registered plugin to check that the plugin // is still responding and has a connection to any upstream services it needs (e.g. MS Graph API). func (a *App) OnSharedChannelsPing(rc *model.RemoteCluster) bool { - pluginsEnvironment := a.GetPluginsEnvironment() - if pluginsEnvironment == nil { - a.Log().Error("Ping for shared channels cannot get plugins env") - return false - } - - pluginHooks, err := pluginsEnvironment.HooksForPlugin(rc.PluginID) + pluginHooks, err := getPluginHooks(a.GetPluginsEnvironment(), rc.PluginID) if err != nil { a.Log().Error("Ping for shared channels cannot get plugin hooks", mlog.String("plugin_id", rc.PluginID), mlog.Err(err)) return false @@ -336,3 +333,25 @@ func (a *App) OnSharedChannelsPing(rc *model.RemoteCluster) bool { return pluginHooks.OnSharedChannelsPing(rc) } + +// OnSharedChannelsAttachmentSyncMsg is called by the Shared Channels service for a registered plugin when a file attachment +// needs to be synchronized. +func (a *App) OnSharedChannelsAttachmentSyncMsg(fi *model.FileInfo, post *model.Post, rc *model.RemoteCluster) error { + pluginHooks, err := getPluginHooks(a.GetPluginsEnvironment(), rc.PluginID) + if err != nil { + return fmt.Errorf("cannot deliver file attachment sync msg to plugin %s: %w", rc.PluginID, err) + } + + return pluginHooks.OnSharedChannelsAttachmentSyncMsg(fi, post, rc) +} + +// OnSharedChannelsProfileImageSyncMsg is called by the Shared Channels service for a registered plugin when a user's +// profile image needs to be synchronized. +func (a *App) OnSharedChannelsProfileImageSyncMsg(user *model.User, rc *model.RemoteCluster) error { + pluginHooks, err := getPluginHooks(a.GetPluginsEnvironment(), rc.PluginID) + if err != nil { + return fmt.Errorf("cannot deliver user profile image sync msg to plugin %s: %w", rc.PluginID, err) + } + + return pluginHooks.OnSharedChannelsProfileImageSyncMsg(user, rc) +} diff --git a/server/channels/store/sqlstore/remote_cluster_store.go b/server/channels/store/sqlstore/remote_cluster_store.go index 59cf9b4c26..21c6f27757 100644 --- a/server/channels/store/sqlstore/remote_cluster_store.go +++ b/server/channels/store/sqlstore/remote_cluster_store.go @@ -157,7 +157,7 @@ func (s sqlRemoteClusterStore) GetByPluginID(pluginID string) (*model.RemoteClus var rc model.RemoteCluster if err := s.GetReplicaX().Get(&rc, queryString, args...); err != nil { - return nil, errors.Wrapf(err, "failed to find RemoteCluster by plugin_id") + return nil, errors.Wrap(err, "failed to find RemoteCluster by plugin_id") } return &rc, nil } diff --git a/server/channels/store/sqlstore/shared_channel_store.go b/server/channels/store/sqlstore/shared_channel_store.go index 4a93a5c11e..fa1fbcd9c2 100644 --- a/server/channels/store/sqlstore/shared_channel_store.go +++ b/server/channels/store/sqlstore/shared_channel_store.go @@ -4,6 +4,7 @@ package sqlstore import ( + "context" "database/sql" "fmt" "strings" @@ -674,10 +675,10 @@ func (s SqlSharedChannelStore) GetSingleUser(userID string, channelID string, re squery, args, err := s.getQueryBuilder(). Select(sharedChannelUserFields("")...). - From("SharedChannelUsers"). - Where(sq.Eq{"SharedChannelUsers.UserId": userID}). - Where(sq.Eq{"SharedChannelUsers.RemoteId": remoteID}). - Where(sq.Eq{"SharedChannelUsers.ChannelId": channelID}). + From("SharedChannelUsers AS scu"). + Where(sq.Eq{"scu.UserId": userID}). + Where(sq.Eq{"scu.ChannelId": channelID}). + Where(sq.Eq{"scu.RemoteId": remoteID}). ToSql() if err != nil { @@ -760,46 +761,33 @@ func (s SqlSharedChannelStore) GetUsersForSync(filter model.GetUsersForSyncFilte // UpdateUserLastSyncAt updates the LastSyncAt timestamp for the specified SharedChannelUser. func (s SqlSharedChannelStore) UpdateUserLastSyncAt(userID string, channelID string, remoteID string) error { - var query string - if s.DriverName() == model.DatabaseDriverPostgres { - query = ` - UPDATE - SharedChannelUsers AS scu - SET - LastSyncAt = GREATEST(Users.UpdateAt, Users.LastPictureUpdate) - FROM - Users - WHERE - Users.Id = scu.UserId AND scu.UserId = ? AND scu.ChannelId = ? AND scu.RemoteId = ? - ` - } else if s.DriverName() == model.DatabaseDriverMysql { - query = ` - UPDATE - SharedChannelUsers AS scu - INNER JOIN - Users ON scu.UserId = Users.Id - SET - LastSyncAt = GREATEST(Users.UpdateAt, Users.LastPictureUpdate) - WHERE - scu.UserId = ? AND scu.ChannelId = ? AND scu.RemoteId = ? - ` - } else { - return errors.New("unsupported DB driver " + s.DriverName()) + // fetching the user first creates a minor race condition. This is mitigated by ensuring that the + // LastUpdateAt is only ever increased. Doing it this way avoids the update with join that has differing + // syntax between MySQL and Postgres which Squirrel cannot handle. It also allows us to return + // a proper error when trying to update for a non-existent user, which cannot be done by checking RowsAffected + // when doing updates; RowsAffected=0 when the LastUpdateAt doesn't change and is the same result if user doesn't + // exist. + user, err := s.stores.user.Get(context.Background(), userID) + if err != nil { + return err } - result, err := s.GetMasterX().Exec(query, userID, channelID, remoteID) + updateAt := maxInt64(user.UpdateAt, user.LastPictureUpdate) + + query := s.getQueryBuilder(). + Update("SharedChannelUsers AS scu"). + Set("LastSyncAt", sq.Expr("GREATEST(scu.LastSyncAt, ?)", updateAt)). + Where(sq.Eq{ + "scu.UserId": userID, + "scu.ChannelId": channelID, + "scu.RemoteId": remoteID, + }) + + _, err = s.GetMasterX().ExecBuilder(query) if err != nil { return fmt.Errorf("failed to update LastSyncAt for SharedChannelUser with userId=%s, channelId=%s, remoteId=%s: %w", userID, channelID, remoteID, err) } - - count, err := result.RowsAffected() - if err != nil { - return errors.Wrap(err, "failed to determine rows affected") - } - if count == 0 { - return fmt.Errorf("SharedChannelUser not found: userId=%s, channelId=%s, remoteId=%s", userID, channelID, remoteID) - } return nil } diff --git a/server/channels/store/sqlstore/utils.go b/server/channels/store/sqlstore/utils.go index 24db76dbdc..aebdc14a2d 100644 --- a/server/channels/store/sqlstore/utils.go +++ b/server/channels/store/sqlstore/utils.go @@ -244,3 +244,10 @@ func trimInput(input string) string { } return input } + +func maxInt64(a, b int64) int64 { + if a > b { + return a + } + return b +} diff --git a/server/channels/store/storetest/remote_cluster_store.go b/server/channels/store/storetest/remote_cluster_store.go index d35004aa59..228f5748df 100644 --- a/server/channels/store/storetest/remote_cluster_store.go +++ b/server/channels/store/storetest/remote_cluster_store.go @@ -21,6 +21,7 @@ func TestRemoteClusterStore(t *testing.T, rctx request.CTX, ss store.Store) { t.Run("RemoteClusterSave", func(t *testing.T) { testRemoteClusterSave(t, rctx, ss) }) t.Run("RemoteClusterDelete", func(t *testing.T) { testRemoteClusterDelete(t, rctx, ss) }) t.Run("RemoteClusterGet", func(t *testing.T) { testRemoteClusterGet(t, rctx, ss) }) + t.Run("RemoteClusterGetByPluginID", func(t *testing.T) { testRemoteClusterGetByPluginID(t, rctx, ss) }) t.Run("RemoteClusterGetAll", func(t *testing.T) { testRemoteClusterGetAll(t, rctx, ss) }) t.Run("RemoteClusterGetByTopic", func(t *testing.T) { testRemoteClusterGetByTopic(t, rctx, ss) }) t.Run("RemoteClusterUpdateTopics", func(t *testing.T) { testRemoteClusterUpdateTopics(t, rctx, ss) }) @@ -191,6 +192,31 @@ func testRemoteClusterGet(t *testing.T, rctx request.CTX, ss store.Store) { }) } +func testRemoteClusterGetByPluginID(t *testing.T, rctx request.CTX, ss store.Store) { + const pluginID = "com.acme.bogus.plugin" + + t.Run("GetByPluginID", func(t *testing.T) { + rc := &model.RemoteCluster{ + Name: "shortlived_remote_3", + SiteURL: makeSiteURL(), + CreatorId: model.NewId(), + PluginID: pluginID, + } + rcSaved, err := ss.RemoteCluster().Save(rc) + require.NoError(t, err) + + rcGet, err := ss.RemoteCluster().GetByPluginID(pluginID) + require.NoError(t, err) + require.Equal(t, rcSaved.RemoteId, rcGet.RemoteId) + require.Equal(t, pluginID, rcGet.PluginID) + }) + + t.Run("GetByPluginID not found", func(t *testing.T) { + _, err := ss.RemoteCluster().GetByPluginID(model.NewId()) + require.Error(t, err) + }) +} + func testRemoteClusterGetAll(t *testing.T, rctx request.CTX, ss store.Store) { require.NoError(t, clearRemoteClusters(ss)) diff --git a/server/platform/services/remotecluster/ping.go b/server/platform/services/remotecluster/ping.go index 642f7dd544..803ef50207 100644 --- a/server/platform/services/remotecluster/ping.go +++ b/server/platform/services/remotecluster/ping.go @@ -66,7 +66,8 @@ func (rcs *Service) pingGenerator(pingChan chan *model.RemoteCluster, done <-cha } for _, rc := range remotes { - if rc.SiteURL != "" || rc.PluginID != "" { // filter out unconfirmed invites + // filter out unconfirmed invites so we don't ping them without permission + if rc.IsConfirmed() { pingChan <- rc } } diff --git a/server/platform/services/sharedchannel/attachment.go b/server/platform/services/sharedchannel/attachment.go index b8671d0e66..0e1ff7522e 100644 --- a/server/platform/services/sharedchannel/attachment.go +++ b/server/platform/services/sharedchannel/attachment.go @@ -41,6 +41,10 @@ func (scs *Service) sendAttachmentForRemote(fi *model.FileInfo, post *model.Post return fmt.Errorf("cannot update remote cluster for remote id %s; Remote Cluster Service not enabled", rc.RemoteId) } + if rc.IsPlugin() { + return scs.sendAttachmentToPlugin(fi, post, rc) + } + us := &model.UploadSession{ Id: model.NewId(), Type: model.UploadTypeAttachment, @@ -120,11 +124,7 @@ func (scs *Service) sendAttachmentForRemote(fi *model.FileInfo, post *model.Post } // save file attachment record in SharedChannelAttachments table - sca := &model.SharedChannelAttachment{ - FileId: fi.Id, - RemoteId: rc.RemoteId, - } - if _, err2 := scs.server.GetStore().SharedChannel().UpsertAttachment(sca); err2 != nil { + if err2 := scs.saveSharedAttachment(&fi, rc); err2 != nil { scs.server.Log().Log(mlog.LvlSharedChannelServiceError, "error saving SharedChannelAttachment", mlog.String("remote", rc.DisplayName), mlog.String("uploadId", usResp.Id), @@ -140,6 +140,24 @@ func (scs *Service) sendAttachmentForRemote(fi *model.FileInfo, post *model.Post }) } +// sendAttachmentToPlugin asynchronously sends a file attachment to a remote cluster. +func (scs *Service) sendAttachmentToPlugin(fi *model.FileInfo, post *model.Post, rc *model.RemoteCluster) error { + if err := scs.app.OnSharedChannelsAttachmentSyncMsg(fi, post, rc); err != nil { + return fmt.Errorf("cannot send attachment to plugin %s: %w", rc.PluginID, err) + } + return scs.saveSharedAttachment(fi, rc) +} + +// saveSharedAttachment saves the attachment in SharedChannelAttachments table. +func (scs *Service) saveSharedAttachment(fi *model.FileInfo, rc *model.RemoteCluster) error { + sca := &model.SharedChannelAttachment{ + FileId: fi.Id, + RemoteId: rc.RemoteId, + } + _, err := scs.server.GetStore().SharedChannel().UpsertAttachment(sca) + return err +} + // onReceiveUploadCreate is called when a message requesting to create an upload session is received. An upload session is // created and the id returned in the response. func (scs *Service) onReceiveUploadCreate(msg model.RemoteClusterMsg, rc *model.RemoteCluster, response *remotecluster.Response) error { diff --git a/server/platform/services/sharedchannel/mock_AppIface_test.go b/server/platform/services/sharedchannel/mock_AppIface_test.go index 81dbd2c9aa..ac1872a125 100644 --- a/server/platform/services/sharedchannel/mock_AppIface_test.go +++ b/server/platform/services/sharedchannel/mock_AppIface_test.go @@ -314,6 +314,34 @@ func (_m *MockAppIface) NotifySharedChannelUserUpdate(user *model.User) { _m.Called(user) } +// OnSharedChannelsAttachmentSyncMsg provides a mock function with given fields: fi, post, rc +func (_m *MockAppIface) OnSharedChannelsAttachmentSyncMsg(fi *model.FileInfo, post *model.Post, rc *model.RemoteCluster) error { + ret := _m.Called(fi, post, rc) + + var r0 error + if rf, ok := ret.Get(0).(func(*model.FileInfo, *model.Post, *model.RemoteCluster) error); ok { + r0 = rf(fi, post, rc) + } else { + r0 = ret.Error(0) + } + + return r0 +} + +// OnSharedChannelsProfileImageSyncMsg provides a mock function with given fields: user, rc +func (_m *MockAppIface) OnSharedChannelsProfileImageSyncMsg(user *model.User, rc *model.RemoteCluster) error { + ret := _m.Called(user, rc) + + var r0 error + if rf, ok := ret.Get(0).(func(*model.User, *model.RemoteCluster) error); ok { + r0 = rf(user, rc) + } else { + r0 = ret.Error(0) + } + + return r0 +} + // OnSharedChannelsSyncMsg provides a mock function with given fields: msg, rc func (_m *MockAppIface) OnSharedChannelsSyncMsg(msg *model.SyncMsg, rc *model.RemoteCluster) (model.SyncResponse, error) { ret := _m.Called(msg, rc) diff --git a/server/platform/services/sharedchannel/service.go b/server/platform/services/sharedchannel/service.go index c11fa4d972..0351d82650 100644 --- a/server/platform/services/sharedchannel/service.go +++ b/server/platform/services/sharedchannel/service.go @@ -64,6 +64,8 @@ type AppIface interface { InvalidateCacheForUser(userID string) NotifySharedChannelUserUpdate(user *model.User) OnSharedChannelsSyncMsg(msg *model.SyncMsg, rc *model.RemoteCluster) (model.SyncResponse, error) + OnSharedChannelsAttachmentSyncMsg(fi *model.FileInfo, post *model.Post, rc *model.RemoteCluster) error + OnSharedChannelsProfileImageSyncMsg(user *model.User, rc *model.RemoteCluster) error } // errNotFound allows checking against Store.ErrNotFound errors without making Store a dependency. diff --git a/server/platform/services/sharedchannel/sync_send.go b/server/platform/services/sharedchannel/sync_send.go index 3749a64c46..b3f90a2bf0 100644 --- a/server/platform/services/sharedchannel/sync_send.go +++ b/server/platform/services/sharedchannel/sync_send.go @@ -402,11 +402,17 @@ func (scs *Service) shouldUserSync(user *model.User, channelID string, rc *model } if _, err = scs.server.GetStore().SharedChannel().SaveUser(scu); err != nil { scs.server.Log().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.String("remote_id", rc.RemoteId), mlog.Err(err), ) + } else { + scs.server.Log().Log(mlog.LvlSharedChannelServiceDebug, "Added user to shared channel users", + mlog.String("user_id", user.Id), + mlog.String("channel_id", user.Id), + mlog.String("remote_id", rc.RemoteId), + ) } return true, true, nil } @@ -420,30 +426,55 @@ func (scs *Service) syncProfileImage(user *model.User, channelID string, rc *mod return } + if rc.IsPlugin() { + scs.sendProfileImageToPlugin(user, channelID, rc) + return + } + ctx, cancel := context.WithTimeout(context.Background(), ProfileImageSyncTimeout) defer cancel() rcs.SendProfileImage(ctx, user.Id, rc, scs.app, func(userId string, rc *model.RemoteCluster, resp *remotecluster.Response, err error) { if resp.IsSuccess() { - scs.server.Log().Log(mlog.LvlSharedChannelServiceDebug, "Users profile image synchronized", - mlog.String("remote_id", rc.RemoteId), - mlog.String("user_id", user.Id), - ) - - if err2 := scs.server.GetStore().SharedChannel().UpdateUserLastSyncAt(user.Id, channelID, rc.RemoteId); err2 != nil { - scs.server.Log().Log(mlog.LvlSharedChannelServiceError, "Error updating users LastSyncTime after profile image update", - mlog.String("remote_id", rc.RemoteId), - mlog.String("user_id", user.Id), - mlog.Err(err2), - ) - } + scs.recordProfileImageSuccess(user.Id, channelID, rc.RemoteId) return } scs.server.Log().Log(mlog.LvlSharedChannelServiceError, "Error synchronizing users profile image", - mlog.String("remote_id", rc.RemoteId), mlog.String("user_id", user.Id), + mlog.String("channel_id", channelID), + mlog.String("remote_id", rc.RemoteId), mlog.Err(err), ) }) } + +func (scs *Service) sendProfileImageToPlugin(user *model.User, channelID string, rc *model.RemoteCluster) { + if err := scs.app.OnSharedChannelsProfileImageSyncMsg(user, rc); err != nil { + scs.server.Log().Log(mlog.LvlSharedChannelServiceError, "Error synchronizing users profile image for plugin", + mlog.String("user_id", user.Id), + mlog.String("channel_id", channelID), + mlog.String("remote_id", rc.RemoteId), + mlog.Err(err), + ) + } + scs.recordProfileImageSuccess(user.Id, channelID, rc.RemoteId) +} + +func (scs *Service) recordProfileImageSuccess(userID, channelID, remoteID string) { + scs.server.Log().Log(mlog.LvlSharedChannelServiceDebug, "Users profile image synchronized", + mlog.String("user_id", userID), + mlog.String("channel_id", channelID), + mlog.String("remote_id", remoteID), + ) + + // update LastSyncAt for user in SharedChannelUsers table + if err := scs.server.GetStore().SharedChannel().UpdateUserLastSyncAt(userID, channelID, remoteID); err != nil { + scs.server.Log().Log(mlog.LvlSharedChannelServiceError, "Error updating users LastSyncTime after profile image update", + mlog.String("user_id", userID), + mlog.String("channel_id", channelID), + mlog.String("remote_id", remoteID), + mlog.Err(err), + ) + } +} diff --git a/server/platform/services/sharedchannel/sync_send_remote.go b/server/platform/services/sharedchannel/sync_send_remote.go index 3a78a09f8d..379724e964 100644 --- a/server/platform/services/sharedchannel/sync_send_remote.go +++ b/server/platform/services/sharedchannel/sync_send_remote.go @@ -235,13 +235,17 @@ func (scs *Service) fetchPostsForSync(sd *syncData) error { count := len(posts) sd.posts = appendPosts(sd.posts, posts, scs.server.GetStore().Post(), cursor.LastPostCreateAt) + cache := postsSliceToMap(posts) + // Fill remaining batch capacity with updated posts. if len(posts) < MaxPostsPerSync { options.SinceCreateAt = false + // use 'nextcursor' as it has the correct xxxUpdateAt values, and the updsted xxxCreateAt values. posts, nextCursor, err = scs.server.GetStore().Post().GetPostsSinceForSync(options, nextCursor, MaxPostsPerSync-len(posts)) if err != nil { return fmt.Errorf("could not fetch modified posts for sync: %w", err) } + posts = reducePostsSliceInCache(posts, cache) count += len(posts) sd.posts = appendPosts(sd.posts, posts, scs.server.GetStore().Post(), cursor.LastPostUpdateAt) } diff --git a/server/platform/services/sharedchannel/util.go b/server/platform/services/sharedchannel/util.go index 2a3dc43776..f5a76fc91e 100644 --- a/server/platform/services/sharedchannel/util.go +++ b/server/platform/services/sharedchannel/util.go @@ -130,3 +130,21 @@ func isNotFoundError(err error) bool { var errNotFound *store.ErrNotFound return errors.As(err, &errNotFound) } + +func postsSliceToMap(posts []*model.Post) map[string]*model.Post { + m := make(map[string]*model.Post, len(posts)) + for _, p := range posts { + m[p.Id] = p + } + return m +} + +func reducePostsSliceInCache(posts []*model.Post, cache map[string]*model.Post) []*model.Post { + reduced := make([]*model.Post, 0, len(posts)) + for _, p := range posts { + if _, ok := cache[p.Id]; !ok { + reduced = append(reduced, p) + } + } + return reduced +} diff --git a/server/public/model/remote_cluster.go b/server/public/model/remote_cluster.go index 632a70cfb3..8251f70fbb 100644 --- a/server/public/model/remote_cluster.go +++ b/server/public/model/remote_cluster.go @@ -174,8 +174,11 @@ func (rc *RemoteCluster) IsPlugin() bool { func (rc *RemoteCluster) GetSiteURL() string { siteURL := rc.SiteURL + if strings.HasPrefix(siteURL, SiteURLPending) { + siteURL = "..." + } if strings.HasPrefix(siteURL, SiteURLPending) || strings.HasPrefix(siteURL, SiteURLPlugin) { - siteURL = "" + siteURL = "plugin" } return siteURL } diff --git a/server/public/plugin/client_rpc_generated.go b/server/public/plugin/client_rpc_generated.go index 508366211e..68587c4e2e 100644 --- a/server/public/plugin/client_rpc_generated.go +++ b/server/public/plugin/client_rpc_generated.go @@ -1052,6 +1052,79 @@ func (s *hooksRPCServer) PreferencesHaveChanged(args *Z_PreferencesHaveChangedAr return nil } +func init() { + hookNameToId["OnSharedChannelsAttachmentSyncMsg"] = OnSharedChannelsAttachmentSyncMsgID +} + +type Z_OnSharedChannelsAttachmentSyncMsgArgs struct { + A *model.FileInfo + B *model.Post + C *model.RemoteCluster +} + +type Z_OnSharedChannelsAttachmentSyncMsgReturns struct { + A error +} + +func (g *hooksRPCClient) OnSharedChannelsAttachmentSyncMsg(fi *model.FileInfo, post *model.Post, rc *model.RemoteCluster) error { + _args := &Z_OnSharedChannelsAttachmentSyncMsgArgs{fi, post, rc} + _returns := &Z_OnSharedChannelsAttachmentSyncMsgReturns{} + if g.implemented[OnSharedChannelsAttachmentSyncMsgID] { + if err := g.client.Call("Plugin.OnSharedChannelsAttachmentSyncMsg", _args, _returns); err != nil { + g.log.Error("RPC call OnSharedChannelsAttachmentSyncMsg to plugin failed.", mlog.Err(err)) + } + } + return _returns.A +} + +func (s *hooksRPCServer) OnSharedChannelsAttachmentSyncMsg(args *Z_OnSharedChannelsAttachmentSyncMsgArgs, returns *Z_OnSharedChannelsAttachmentSyncMsgReturns) error { + if hook, ok := s.impl.(interface { + OnSharedChannelsAttachmentSyncMsg(fi *model.FileInfo, post *model.Post, rc *model.RemoteCluster) error + }); ok { + returns.A = hook.OnSharedChannelsAttachmentSyncMsg(args.A, args.B, args.C) + returns.A = encodableError(returns.A) + } else { + return encodableError(fmt.Errorf("Hook OnSharedChannelsAttachmentSyncMsg called but not implemented.")) + } + return nil +} + +func init() { + hookNameToId["OnSharedChannelsProfileImageSyncMsg"] = OnSharedChannelsProfileImageSyncMsgID +} + +type Z_OnSharedChannelsProfileImageSyncMsgArgs struct { + A *model.User + B *model.RemoteCluster +} + +type Z_OnSharedChannelsProfileImageSyncMsgReturns struct { + A error +} + +func (g *hooksRPCClient) OnSharedChannelsProfileImageSyncMsg(user *model.User, rc *model.RemoteCluster) error { + _args := &Z_OnSharedChannelsProfileImageSyncMsgArgs{user, rc} + _returns := &Z_OnSharedChannelsProfileImageSyncMsgReturns{} + if g.implemented[OnSharedChannelsProfileImageSyncMsgID] { + if err := g.client.Call("Plugin.OnSharedChannelsProfileImageSyncMsg", _args, _returns); err != nil { + g.log.Error("RPC call OnSharedChannelsProfileImageSyncMsg to plugin failed.", mlog.Err(err)) + } + } + return _returns.A +} + +func (s *hooksRPCServer) OnSharedChannelsProfileImageSyncMsg(args *Z_OnSharedChannelsProfileImageSyncMsgArgs, returns *Z_OnSharedChannelsProfileImageSyncMsgReturns) error { + if hook, ok := s.impl.(interface { + OnSharedChannelsProfileImageSyncMsg(user *model.User, rc *model.RemoteCluster) error + }); ok { + returns.A = hook.OnSharedChannelsProfileImageSyncMsg(args.A, args.B) + returns.A = encodableError(returns.A) + } else { + return encodableError(fmt.Errorf("Hook OnSharedChannelsProfileImageSyncMsg called but not implemented.")) + } + return nil +} + type Z_RegisterCommandArgs struct { A *model.Command } diff --git a/server/public/plugin/hooks.go b/server/public/plugin/hooks.go index 5b3df788ac..b1dbf032ad 100644 --- a/server/public/plugin/hooks.go +++ b/server/public/plugin/hooks.go @@ -58,6 +58,8 @@ const ( OnSharedChannelsSyncMsgID = 40 OnSharedChannelsPingID = 41 PreferencesHaveChangedID = 42 + OnSharedChannelsAttachmentSyncMsgID = 43 + OnSharedChannelsProfileImageSyncMsgID = 44 TotalHooksID = iota ) @@ -362,4 +364,22 @@ type Hooks interface { // // Minimum server version: 9.5 PreferencesHaveChanged(c *Context, preferences []model.Preference) + + // OnSharedChannelsAttachmentSyncMsg is invoked for plugins that wish to receive synchronization messages from the + // Shared Channels service for which they have been invited via InviteRemote. Each call represents one file attachment + // to be synchronized. + // + // The cursor will be advanced based on the timestamp returned if no error is returned. + // + // Minimum server version: 9.5 + OnSharedChannelsAttachmentSyncMsg(fi *model.FileInfo, post *model.Post, rc *model.RemoteCluster) error + + // OnSharedChannelsProfileImageSyncMsg is invoked for plugins that wish to receive synchronization messages from the + // Shared Channels service for which they have been invited via InviteRemote. Each call represents one user profile + // image that should be synchronized. `App.GetProfileImage` can be used to fetch the image bytes. + // + // The cursor will be advanced based on the timestamp returned if no error is returned. + // + // Minimum server version: 9.5 + OnSharedChannelsProfileImageSyncMsg(user *model.User, rc *model.RemoteCluster) error } diff --git a/server/public/plugin/hooks_timer_layer_generated.go b/server/public/plugin/hooks_timer_layer_generated.go index 2a33ab1618..531be80321 100644 --- a/server/public/plugin/hooks_timer_layer_generated.go +++ b/server/public/plugin/hooks_timer_layer_generated.go @@ -270,3 +270,17 @@ func (hooks *hooksTimerLayer) PreferencesHaveChanged(c *Context, preferences []m hooks.hooksImpl.PreferencesHaveChanged(c, preferences) hooks.recordTime(startTime, "PreferencesHaveChanged", true) } + +func (hooks *hooksTimerLayer) OnSharedChannelsAttachmentSyncMsg(fi *model.FileInfo, post *model.Post, rc *model.RemoteCluster) error { + startTime := timePkg.Now() + _returnsA := hooks.hooksImpl.OnSharedChannelsAttachmentSyncMsg(fi, post, rc) + hooks.recordTime(startTime, "OnSharedChannelsAttachmentSyncMsg", _returnsA == nil) + return _returnsA +} + +func (hooks *hooksTimerLayer) OnSharedChannelsProfileImageSyncMsg(user *model.User, rc *model.RemoteCluster) error { + startTime := timePkg.Now() + _returnsA := hooks.hooksImpl.OnSharedChannelsProfileImageSyncMsg(user, rc) + hooks.recordTime(startTime, "OnSharedChannelsProfileImageSyncMsg", _returnsA == nil) + return _returnsA +} diff --git a/server/public/plugin/plugintest/hooks.go b/server/public/plugin/plugintest/hooks.go index e13dd42f44..c61eae7892 100644 --- a/server/public/plugin/plugintest/hooks.go +++ b/server/public/plugin/plugintest/hooks.go @@ -311,6 +311,20 @@ func (_m *Hooks) OnSendDailyTelemetry() { _m.Called() } +// OnSharedChannelsAttachmentSyncMsg provides a mock function with given fields: fi, post, rc +func (_m *Hooks) OnSharedChannelsAttachmentSyncMsg(fi *model.FileInfo, post *model.Post, rc *model.RemoteCluster) error { + ret := _m.Called(fi, post, rc) + + var r0 error + if rf, ok := ret.Get(0).(func(*model.FileInfo, *model.Post, *model.RemoteCluster) error); ok { + r0 = rf(fi, post, rc) + } else { + r0 = ret.Error(0) + } + + return r0 +} + // OnSharedChannelsPing provides a mock function with given fields: rc func (_m *Hooks) OnSharedChannelsPing(rc *model.RemoteCluster) bool { ret := _m.Called(rc) @@ -325,6 +339,20 @@ func (_m *Hooks) OnSharedChannelsPing(rc *model.RemoteCluster) bool { return r0 } +// OnSharedChannelsProfileImageSyncMsg provides a mock function with given fields: user, rc +func (_m *Hooks) OnSharedChannelsProfileImageSyncMsg(user *model.User, rc *model.RemoteCluster) error { + ret := _m.Called(user, rc) + + var r0 error + if rf, ok := ret.Get(0).(func(*model.User, *model.RemoteCluster) error); ok { + r0 = rf(user, rc) + } else { + r0 = ret.Error(0) + } + + return r0 +} + // OnSharedChannelsSyncMsg provides a mock function with given fields: msg, rc func (_m *Hooks) OnSharedChannelsSyncMsg(msg *model.SyncMsg, rc *model.RemoteCluster) (model.SyncResponse, error) { ret := _m.Called(msg, rc) diff --git a/server/public/plugin/product_hooks_generated.go b/server/public/plugin/product_hooks_generated.go index 98a59adf3c..2e4f112358 100644 --- a/server/public/plugin/product_hooks_generated.go +++ b/server/public/plugin/product_hooks_generated.go @@ -151,6 +151,14 @@ type PreferencesHaveChangedIFace interface { PreferencesHaveChanged(c *Context, preferences []model.Preference) } +type OnSharedChannelsAttachmentSyncMsgIFace interface { + OnSharedChannelsAttachmentSyncMsg(fi *model.FileInfo, post *model.Post, rc *model.RemoteCluster) error +} + +type OnSharedChannelsProfileImageSyncMsgIFace interface { + OnSharedChannelsProfileImageSyncMsg(user *model.User, rc *model.RemoteCluster) error +} + type HooksAdapter struct { implemented map[int]struct{} productHooks any @@ -470,6 +478,24 @@ func NewAdapter(productHooks any) (*HooksAdapter, error) { return nil, errors.New("hook has PreferencesHaveChanged method but does not implement plugin.PreferencesHaveChanged interface") } + // Assessing the type of the productHooks if it individually implements OnSharedChannelsAttachmentSyncMsg interface. + tt = reflect.TypeOf((*OnSharedChannelsAttachmentSyncMsgIFace)(nil)).Elem() + + if ft.Implements(tt) { + a.implemented[OnSharedChannelsAttachmentSyncMsgID] = struct{}{} + } else if _, ok := ft.MethodByName("OnSharedChannelsAttachmentSyncMsg"); ok { + return nil, errors.New("hook has OnSharedChannelsAttachmentSyncMsg method but does not implement plugin.OnSharedChannelsAttachmentSyncMsg interface") + } + + // Assessing the type of the productHooks if it individually implements OnSharedChannelsProfileImageSyncMsg interface. + tt = reflect.TypeOf((*OnSharedChannelsProfileImageSyncMsgIFace)(nil)).Elem() + + if ft.Implements(tt) { + a.implemented[OnSharedChannelsProfileImageSyncMsgID] = struct{}{} + } else if _, ok := ft.MethodByName("OnSharedChannelsProfileImageSyncMsg"); ok { + return nil, errors.New("hook has OnSharedChannelsProfileImageSyncMsg method but does not implement plugin.OnSharedChannelsProfileImageSyncMsg interface") + } + return a, nil } @@ -778,3 +804,21 @@ func (a *HooksAdapter) PreferencesHaveChanged(c *Context, preferences []model.Pr a.productHooks.(PreferencesHaveChangedIFace).PreferencesHaveChanged(c, preferences) } + +func (a *HooksAdapter) OnSharedChannelsAttachmentSyncMsg(fi *model.FileInfo, post *model.Post, rc *model.RemoteCluster) error { + if _, ok := a.implemented[OnSharedChannelsAttachmentSyncMsgID]; !ok { + panic("product hooks must implement OnSharedChannelsAttachmentSyncMsg") + } + + return a.productHooks.(OnSharedChannelsAttachmentSyncMsgIFace).OnSharedChannelsAttachmentSyncMsg(fi, post, rc) + +} + +func (a *HooksAdapter) OnSharedChannelsProfileImageSyncMsg(user *model.User, rc *model.RemoteCluster) error { + if _, ok := a.implemented[OnSharedChannelsProfileImageSyncMsgID]; !ok { + panic("product hooks must implement OnSharedChannelsProfileImageSyncMsg") + } + + return a.productHooks.(OnSharedChannelsProfileImageSyncMsgIFace).OnSharedChannelsProfileImageSyncMsg(user, rc) + +}