Shared channels plugin APIs for MS Teams plugin (#25805)

New plugin APIs and hooks for accessing Shared Channels service via plugin. 

- RegisterPluginForSharedChannels(opts model.RegisterPluginOpts) (remoteID string, err error)
- UnregisterPluginForSharedChannels(pluginID string) error
- ShareChannel(sc *model.SharedChannel) (*model.SharedChannel, error)
- UpdateSharedChannel(sc *model.SharedChannel) (*model.SharedChannel, error)
- UnshareChannel(channelID string) (unshared bool, err error)
- UpdateSharedChannelCursor(channelID, remoteID string, cusror model.GetPostsSinceForSyncCursor) error
- SyncSharedChannel(channelID string) error
- InviteRemoteToChannel(channelID string, remoteID string, userID string) error
- UninviteRemoteFromChannel(channelID string, remoteID string) error

Hooks
- OnSharedChannelsSyncMsg(msg *model.SyncMsg, rc *model.RemoteCluster) (model.SyncResponse, error)
- OnSharedChannelsPing(rc *model.RemoteCluster) bool

server/platform/services/remotecluster/mocks_test.go

@@ -5,6 +5,7 @@ package remotecluster
 
 import (
 	"context"
+	"sync"
 	"testing"
 
 	"github.com/mattermost/mattermost/server/public/model"
@@ -47,10 +48,12 @@ func (ms *mockServer) GetStore() store.Store {
 		return true
 	})
 	anyUserId := mock.AnythingOfType("string")
+	anyId := mock.AnythingOfType("string")
 
 	remoteClusterStoreMock := &mocks.RemoteClusterStore{}
 	remoteClusterStoreMock.On("GetByTopic", "share").Return(ms.remotes, nil)
 	remoteClusterStoreMock.On("GetAll", anyQueryFilter).Return(ms.remotes, nil)
+	remoteClusterStoreMock.On("SetLastPingAt", anyId).Return(nil)
 
 	userStoreMock := &mocks.UserStore{}
 	userStoreMock.On("Get", context.Background(), anyUserId).Return(ms.user, nil)
@@ -60,3 +63,57 @@ func (ms *mockServer) GetStore() store.Store {
 	storeMock.On("User").Return(userStoreMock)
 	return storeMock
 }
+
+type mockApp struct {
+	offlinePluginIDs []string
+
+	mux             sync.Mutex
+	totalPingCount  int
+	totalPingErrors int
+	pingCounts      map[string]int
+}
+
+func newMockApp(t *testing.T, offlinePluginIDs []string) *mockApp {
+	return &mockApp{
+		offlinePluginIDs: offlinePluginIDs,
+		pingCounts:       make(map[string]int),
+	}
+}
+
+func (ma *mockApp) OnSharedChannelsPing(rc *model.RemoteCluster) bool {
+	ma.mux.Lock()
+	defer ma.mux.Unlock()
+
+	for _, id := range ma.offlinePluginIDs {
+		if rc.PluginID == id {
+			ma.totalPingErrors++
+			return false
+		}
+	}
+
+	ma.totalPingCount++
+
+	count := ma.pingCounts[rc.PluginID]
+	ma.pingCounts[rc.PluginID] = count + 1
+
+	return true
+}
+
+func (ma *mockApp) GetTotalPingCount() int {
+	ma.mux.Lock()
+	defer ma.mux.Unlock()
+	return ma.totalPingCount
+}
+
+func (ma *mockApp) GetTotalPingErrorCount() int {
+	ma.mux.Lock()
+	defer ma.mux.Unlock()
+	return ma.totalPingErrors
+}
+
+func (ma *mockApp) GetPingCount(pluginID string) int {
+	ma.mux.Lock()
+	defer ma.mux.Unlock()
+
+	return ma.pingCounts[pluginID]
+}

server/platform/services/remotecluster/ping.go

@@ -5,6 +5,7 @@ package remotecluster
 
 import (
 	"encoding/json"
+	"errors"
 	"fmt"
 	"time"
 
@@ -28,6 +29,7 @@ func (rcs *Service) pingGenerator(pingChan chan *model.RemoteCluster, done <-cha
 	defer close(pingChan)
 
 	for {
+		pingFreq := rcs.GetPingFreq()
 		start := time.Now()
 
 		// get all remotes, including any previously offline.
@@ -35,7 +37,7 @@ func (rcs *Service) pingGenerator(pingChan chan *model.RemoteCluster, done <-cha
 		if err != nil {
 			rcs.server.Log().Log(mlog.LvlRemoteClusterServiceError, "Ping remote cluster failed (could not get list of remotes)", mlog.Err(err))
 			select {
-			case <-time.After(PingFreq):
+			case <-time.After(pingFreq):
 				continue
 			case <-done:
 				return
@@ -50,8 +52,8 @@ func (rcs *Service) pingGenerator(pingChan chan *model.RemoteCluster, done <-cha
 
 		// try to maintain frequency
 		elapsed := time.Since(start)
-		if elapsed < PingFreq {
-			sleep := time.Until(start.Add(PingFreq))
+		if elapsed < pingFreq {
+			sleep := time.Until(start.Add(pingFreq))
 			select {
 			case <-time.After(sleep):
 			case <-done:
@@ -77,6 +79,7 @@ func (rcs *Service) pingEmitter(pingChan <-chan *model.RemoteCluster, done <-cha
 				rcs.server.Log().Log(mlog.LvlRemoteClusterServiceWarn, "Remote cluster ping failed",
 					mlog.String("remote", rc.DisplayName),
 					mlog.String("remoteId", rc.RemoteId),
+					mlog.String("pluginId", rc.PluginID),
 					mlog.Err(err),
 				)
 			}
@@ -93,25 +96,36 @@ func (rcs *Service) pingEmitter(pingChan <-chan *model.RemoteCluster, done <-cha
 	}
 }
 
+var ErrPluginPingFail = errors.New("plugin ping failed")
+
 // pingRemote make a synchronous ping to a remote cluster. Return is error if ping is
 // unsuccessful and nil on success.
 func (rcs *Service) pingRemote(rc *model.RemoteCluster) error {
-	frame, err := makePingFrame(rc)
-	if err != nil {
-		return err
-	}
-	url := fmt.Sprintf("%s/%s", rc.SiteURL, PingURL)
-
-	resp, err := rcs.sendFrameToRemote(PingTimeout, rc, frame, url)
-	if err != nil {
-		return err
-	}
-	rc.LastPingAt = model.GetMillis()
-
 	ping := model.RemoteClusterPing{}
-	err = json.Unmarshal(resp, &ping)
-	if err != nil {
-		return err
+
+	if rc.PluginID != "" {
+		ping.SentAt = model.GetMillis()
+		if ok := rcs.app.OnSharedChannelsPing(rc); !ok {
+			return ErrPluginPingFail
+		}
+		ping.RecvAt = model.GetMillis()
+	} else {
+		frame, err := makePingFrame(rc)
+		if err != nil {
+			return err
+		}
+		url := fmt.Sprintf("%s/%s", rc.SiteURL, PingURL)
+
+		resp, err := rcs.sendFrameToRemote(PingTimeout, rc, frame, url)
+		if err != nil {
+			return err
+		}
+		rc.LastPingAt = model.GetMillis()
+
+		err = json.Unmarshal(resp, &ping)
+		if err != nil {
+			return err
+		}
 	}
 
 	if err := rcs.server.GetStore().RemoteCluster().SetLastPingAt(rc.RemoteId); err != nil {
@@ -135,6 +149,7 @@ func (rcs *Service) pingRemote(rc *model.RemoteCluster) error {
 	rcs.server.Log().Log(mlog.LvlRemoteClusterServiceDebug, "Remote cluster ping",
 		mlog.String("remote", rc.DisplayName),
 		mlog.String("remoteId", rc.RemoteId),
+		mlog.String("pluginId", rc.PluginID),
 		mlog.Int("SentAt", ping.SentAt),
 		mlog.Int("RecvAt", ping.RecvAt),
 		mlog.Int("Diff", ping.RecvAt-ping.SentAt),

server/platform/services/remotecluster/ping_test.go

@@ -11,6 +11,7 @@ import (
 	"sync"
 	"sync/atomic"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -66,9 +67,10 @@ func TestPing(t *testing.T) {
 		}))
 		defer ts.Close()
 
-		mockServer := newMockServer(t, makeRemoteClusters(NumRemotes, ts.URL))
+		mockServer := newMockServer(t, makeRemoteClusters(NumRemotes, ts.URL, false))
+		mockApp := newMockApp(t, nil)
 
-		service, err := NewRemoteClusterService(mockServer)
+		service, err := NewRemoteClusterService(mockServer, mockApp)
 		require.NoError(t, err)
 
 		err = service.Start()
@@ -115,9 +117,10 @@ func TestPing(t *testing.T) {
 		}))
 		defer ts.Close()
 
-		mockServer := newMockServer(t, makeRemoteClusters(NumRemotes, ts.URL))
+		mockServer := newMockServer(t, makeRemoteClusters(NumRemotes, ts.URL, false))
+		mockApp := newMockApp(t, nil)
 
-		service, err := NewRemoteClusterService(mockServer)
+		service, err := NewRemoteClusterService(mockServer, mockApp)
 		require.NoError(t, err)
 
 		err = service.Start()
@@ -132,6 +135,34 @@ func TestPing(t *testing.T) {
 		t.Logf("%d web requests counted;  %d expected",
 			atomic.LoadInt32(&countWebReq), NumRemotes)
 	})
+
+	t.Run("Plugin ping", func(t *testing.T) {
+		mockServer := newMockServer(t, makeRemoteClusters(NumRemotes, model.NewId(), true))
+		offline := []string{mockServer.remotes[0].PluginID, mockServer.remotes[1].PluginID}
+
+		mockApp := newMockApp(t, offline)
+
+		service, err := NewRemoteClusterService(mockServer, mockApp)
+		require.NoError(t, err)
+
+		// high ping frequency so we don't delay unit tests.
+		service.SetPingFreq(time.Millisecond * 50)
+
+		err = service.Start()
+		require.NoError(t, err)
+		defer service.Shutdown()
+
+		checkPingCount := func() bool {
+			return mockApp.GetTotalPingCount() >= NumRemotes
+		}
+
+		checkErrorCount := func() bool {
+			return mockApp.GetTotalPingErrorCount() >= 2
+		}
+
+		assert.Eventually(t, checkPingCount, time.Second*5, 10*time.Millisecond)
+		assert.Eventually(t, checkErrorCount, time.Second*5, 10*time.Millisecond)
+	})
 }
 
 func checkRecent(millis int64, within int64) bool {

server/platform/services/remotecluster/send_test.go

@@ -82,9 +82,10 @@ func TestBroadcastMsg(t *testing.T) {
 		}))
 		defer ts.Close()
 
-		mockServer := newMockServer(t, makeRemoteClusters(NumRemotes, ts.URL))
+		mockServer := newMockServer(t, makeRemoteClusters(NumRemotes, ts.URL, false))
+		mockApp := newMockApp(t, nil)
 
-		service, err := NewRemoteClusterService(mockServer)
+		service, err := NewRemoteClusterService(mockServer, mockApp)
 		require.NoError(t, err)
 
 		err = service.Start()
@@ -138,9 +139,10 @@ func TestBroadcastMsg(t *testing.T) {
 		}))
 		defer ts.Close()
 
-		mockServer := newMockServer(t, makeRemoteClusters(NumRemotes, ts.URL))
+		mockServer := newMockServer(t, makeRemoteClusters(NumRemotes, ts.URL, false))
+		mockApp := newMockApp(t, nil)
 
-		service, err := NewRemoteClusterService(mockServer)
+		service, err := NewRemoteClusterService(mockServer, mockApp)
 		require.NoError(t, err)
 
 		err = service.Start()
@@ -169,10 +171,13 @@ func TestBroadcastMsg(t *testing.T) {
 	})
 }
 
-func makeRemoteClusters(num int, siteURL string) []*model.RemoteCluster {
+func makeRemoteClusters(num int, siteURL string, isPlugin bool) []*model.RemoteCluster {
 	var remotes []*model.RemoteCluster
 	for i := 0; i < num; i++ {
 		rc := makeRemoteCluster(fmt.Sprintf("test cluster %d", i+1), siteURL, TestTopics)
+		if isPlugin {
+			rc.PluginID = model.NewId()
+		}
 		remotes = append(remotes, rc)
 	}
 	return remotes

server/platform/services/remotecluster/sendprofileImage_test.go

@@ -100,9 +100,12 @@ func TestService_sendProfileImageToRemote(t *testing.T) {
 
 	provider := testImageProvider{}
 
-	mockServer := newMockServer(t, makeRemoteClusters(NumRemotes, ts.URL))
+	mockServer := newMockServer(t, makeRemoteClusters(NumRemotes, ts.URL, false))
 	mockServer.SetUser(user)
-	service, err := NewRemoteClusterService(mockServer)
+
+	mockApp := newMockApp(t, nil)
+
+	service, err := NewRemoteClusterService(mockServer, mockApp)
 	require.NoError(t, err)
 
 	err = service.Start()

server/platform/services/remotecluster/service.go

@@ -50,6 +50,10 @@ type ServerIface interface {
 	GetMetrics() einterfaces.MetricsInterface
 }
 
+type AppIface interface {
+	OnSharedChannelsPing(rc *model.RemoteCluster) bool
+}
+
 // RemoteClusterServiceIFace is used to allow mocking where a remote cluster service is used (for testing).
 // Unfortunately it lives here because the shared channel service, app layer, and server interface all need it.
 // Putting it in app layer means shared channel service must import app package.
@@ -78,6 +82,7 @@ type ConnectionStateListener func(rc *model.RemoteCluster, online bool)
 // Service provides inter-cluster communication via topic based messages. In product these are called "Secured Connections".
 type Service struct {
 	server     ServerIface
+	app        AppIface
 	httpClient *http.Client
 	send       []chan any
 
@@ -88,10 +93,11 @@ type Service struct {
 	topicListeners           map[string]map[string]TopicListener // maps topic id to a map of listenerid->listener
 	connectionStateListeners map[string]ConnectionStateListener  // maps listener id to listener
 	done                     chan struct{}
+	pingFreq                 time.Duration
 }
 
 // NewRemoteClusterService creates a RemoteClusterService instance. In product this is called a "Secured Connection".
-func NewRemoteClusterService(server ServerIface) (*Service, error) {
+func NewRemoteClusterService(server ServerIface, app AppIface) (*Service, error) {
 	transport := &http.Transport{
 		Proxy: http.ProxyFromEnvironment,
 		DialContext: (&net.Dialer{
@@ -115,6 +121,7 @@ func NewRemoteClusterService(server ServerIface) (*Service, error) {
 
 	service := &Service{
 		server:                   server,
+		app:                      app,
 		httpClient:               client,
 		topicListeners:           make(map[string]map[string]TopicListener),
 		connectionStateListeners: make(map[string]ConnectionStateListener),
@@ -124,6 +131,7 @@ func NewRemoteClusterService(server ServerIface) (*Service, error) {
 	for i := range service.send {
 		service.send[i] = make(chan any, SendChanBuffer)
 	}
+	service.pingFreq = PingFreq
 
 	return service, nil
 }
@@ -154,6 +162,21 @@ func (rcs *Service) Active() bool {
 	return rcs.active
 }
 
+// GetPingFreq gets the frequency of pings to each remote.
+func (rcs *Service) GetPingFreq() time.Duration {
+	rcs.mux.Lock()
+	defer rcs.mux.Unlock()
+	return rcs.pingFreq
+}
+
+// SetPingFreq sets the frequency of pings to each remote. Defaults to `PingFreq`.
+// This is typically used to set a higher frequency for testing.
+func (rcs *Service) SetPingFreq(freq time.Duration) {
+	rcs.mux.Lock()
+	defer rcs.mux.Unlock()
+	rcs.pingFreq = freq
+}
+
 // AddTopicListener registers a callback
 func (rcs *Service) AddTopicListener(topic string, listener TopicListener) string {
 	rcs.mux.Lock()

server/platform/services/remotecluster/service_test.go

@@ -29,9 +29,10 @@ func TestService_AddTopicListener(t *testing.T) {
 		return nil
 	}
 
-	mockServer := newMockServer(t, makeRemoteClusters(NumRemotes, ""))
+	mockServer := newMockServer(t, makeRemoteClusters(NumRemotes, "", false))
+	mockApp := newMockApp(t, nil)
 
-	service, err := NewRemoteClusterService(mockServer)
+	service, err := NewRemoteClusterService(mockServer, mockApp)
 	require.NoError(t, err)
 
 	l1id := service.AddTopicListener("test", l1)

server/platform/services/sharedchannel/mock_AppIface_test.go

@@ -314,6 +314,30 @@ func (_m *MockAppIface) NotifySharedChannelUserUpdate(user *model.User) {
 	_m.Called(user)
 }
 
+// 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)
+
+	var r0 model.SyncResponse
+	var r1 error
+	if rf, ok := ret.Get(0).(func(*model.SyncMsg, *model.RemoteCluster) (model.SyncResponse, error)); ok {
+		return rf(msg, rc)
+	}
+	if rf, ok := ret.Get(0).(func(*model.SyncMsg, *model.RemoteCluster) model.SyncResponse); ok {
+		r0 = rf(msg, rc)
+	} else {
+		r0 = ret.Get(0).(model.SyncResponse)
+	}
+
+	if rf, ok := ret.Get(1).(func(*model.SyncMsg, *model.RemoteCluster) error); ok {
+		r1 = rf(msg, rc)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
 // PatchChannelModerationsForChannel provides a mock function with given fields: c, channel, channelModerationsPatch
 func (_m *MockAppIface) PatchChannelModerationsForChannel(c request.CTX, channel *model.Channel, channelModerationsPatch []*model.ChannelModerationPatch) ([]*model.ChannelModeration, *model.AppError) {
 	ret := _m.Called(c, channel, channelModerationsPatch)

server/platform/services/sharedchannel/service.go

@@ -63,6 +63,7 @@ type AppIface interface {
 	GetProfileImage(user *model.User) ([]byte, bool, *model.AppError)
 	InvalidateCacheForUser(userID string)
 	NotifySharedChannelUserUpdate(user *model.User)
+	OnSharedChannelsSyncMsg(msg *model.SyncMsg, rc *model.RemoteCluster) (model.SyncResponse, error)
 }
 
 // errNotFound allows checking against Store.ErrNotFound errors without making Store a dependency.
@@ -108,7 +109,7 @@ func NewSharedChannelService(server ServerIface, app AppIface) (*Service, error)
 // Start is called by the server on server start-up.
 func (scs *Service) Start() error {
 	rcs := scs.server.GetRemoteClusterService()
-	if rcs == nil {
+	if rcs == nil || !rcs.Active() {
 		return errors.New("Shared Channel Service cannot activate: requires Remote Cluster Service")
 	}
 
@@ -128,7 +129,7 @@ func (scs *Service) Start() error {
 // Shutdown is called by the server on server shutdown.
 func (scs *Service) Shutdown() error {
 	rcs := scs.server.GetRemoteClusterService()
-	if rcs == nil {
+	if rcs == nil || !rcs.Active() {
 		return errors.New("Shared Channel Service cannot shutdown: requires Remote Cluster Service")
 	}
 

server/platform/services/sharedchannel/sync_send_remote.go

@@ -527,13 +527,17 @@ func (scs *Service) sendProfileImageSyncData(sd *syncData) {
 	}
 }
 
-// sendSyncMsgToRemote synchronously sends the sync message to the remote cluster.
+// sendSyncMsgToRemote synchronously sends the sync message to the remote cluster (or plugin).
 func (scs *Service) sendSyncMsgToRemote(msg *model.SyncMsg, rc *model.RemoteCluster, f sendSyncMsgResultFunc) error {
 	rcs := scs.server.GetRemoteClusterService()
 	if rcs == nil {
 		return fmt.Errorf("cannot update remote cluster %s for channel id %s; Remote Cluster Service not enabled", rc.Name, msg.ChannelId)
 	}
 
+	if rc.PluginID != "" {
+		return scs.sendSyncMsgToPlugin(msg, rc, f)
+	}
+
 	b, err := json.Marshal(msg)
 	if err != nil {
 		return err
@@ -568,6 +572,17 @@ func (scs *Service) sendSyncMsgToRemote(msg *model.SyncMsg, rc *model.RemoteClus
 	return err
 }
 
+// sendSyncMsgToRemote synchronously sends the sync message to a plugin.
+func (scs *Service) sendSyncMsgToPlugin(msg *model.SyncMsg, rc *model.RemoteCluster, f sendSyncMsgResultFunc) error {
+	syncResp, errResp := scs.app.OnSharedChannelsSyncMsg(msg, rc)
+
+	if f != nil {
+		f(syncResp, errResp)
+	}
+
+	return errResp
+}
+
 func sanitizeSyncData(sd *syncData) {
 	for id, user := range sd.users {
 		sd.users[id] = sanitizeUserForSync(user)