Plugin API hook for Shared Channel file attachment sync (#25874)

* option for auto inviting plugin to all shared channels. * auto-invite remotes to shared channels when flag set * fix unit test * immediately ping new remotes; fix unique siteurl bug * make i18n-extract * fix translations * plugin hooks for file attachments * hook for profile image sync * fix profile image sync * fix unit test --------- Co-authored-by: Mattermost Build <build@mattermost.com>
2024-01-16 09:48:51 -05:00
--- 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)
--- 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")
--- 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
 }

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

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