Mono repo -> Master (#22553)

Combines the following repositories into one:

https://github.com/mattermost/mattermost-server
https://github.com/mattermost/mattermost-webapp
https://github.com/mattermost/focalboard
https://github.com/mattermost/mattermost-plugin-playbooks

server/platform/services/remotecluster/error.go

@@ -0,0 +1,24 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package remotecluster
+
+import "fmt"
+
+type BufferFullError struct {
+	capacity int
+}
+
+func NewBufferFullError(capacity int) BufferFullError {
+	return BufferFullError{
+		capacity: capacity,
+	}
+}
+
+func (e BufferFullError) Capacity() int {
+	return e.capacity
+}
+
+func (e BufferFullError) Error() string {
+	return fmt.Sprintf("buffer capacity (%d) exceeded", e.capacity)
+}

server/platform/services/remotecluster/invitation.go

@@ -0,0 +1,83 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package remotecluster
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+
+	"github.com/mattermost/mattermost-server/v6/model"
+)
+
+// AcceptInvitation is called when accepting an invitation to connect with a remote cluster.
+func (rcs *Service) AcceptInvitation(invite *model.RemoteClusterInvite, name string, displayName, creatorId string, teamId string, siteURL string) (*model.RemoteCluster, error) {
+	rc := &model.RemoteCluster{
+		RemoteId:     invite.RemoteId,
+		RemoteTeamId: invite.RemoteTeamId,
+		Name:         name,
+		DisplayName:  displayName,
+		Token:        model.NewId(),
+		RemoteToken:  invite.Token,
+		SiteURL:      invite.SiteURL,
+		CreatorId:    creatorId,
+	}
+
+	rcSaved, err := rcs.server.GetStore().RemoteCluster().Save(rc)
+	if err != nil {
+		return nil, err
+	}
+
+	// confirm the invitation with the originating site
+	frame, err := makeConfirmFrame(rcSaved, teamId, siteURL)
+	if err != nil {
+		return nil, err
+	}
+
+	url := fmt.Sprintf("%s/%s", rcSaved.SiteURL, ConfirmInviteURL)
+
+	resp, err := rcs.sendFrameToRemote(PingTimeout, rc, frame, url)
+	if err != nil {
+		rcs.server.GetStore().RemoteCluster().Delete(rcSaved.RemoteId)
+		return nil, err
+	}
+
+	var response Response
+	err = json.Unmarshal(resp, &response)
+	if err != nil {
+		rcs.server.GetStore().RemoteCluster().Delete(rcSaved.RemoteId)
+		return nil, fmt.Errorf("invalid response from remote server: %w", err)
+	}
+
+	if !response.IsSuccess() {
+		rcs.server.GetStore().RemoteCluster().Delete(rcSaved.RemoteId)
+		return nil, errors.New(response.Err)
+	}
+
+	// issue the first ping right away. The goroutine will exit when ping completes or PingTimeout exceeded.
+	go rcs.pingRemote(rcSaved)
+
+	return rcSaved, nil
+}
+
+func makeConfirmFrame(rc *model.RemoteCluster, teamId string, siteURL string) (*model.RemoteClusterFrame, error) {
+	confirm := model.RemoteClusterInvite{
+		RemoteId:     rc.RemoteId,
+		RemoteTeamId: teamId,
+		SiteURL:      siteURL,
+		Token:        rc.Token,
+	}
+	confirmRaw, err := json.Marshal(confirm)
+	if err != nil {
+		return nil, err
+	}
+
+	msg := model.NewRemoteClusterMsg(InvitationTopic, confirmRaw)
+
+	frame := &model.RemoteClusterFrame{
+		RemoteId: rc.RemoteId,
+		Msg:      msg,
+	}
+	return frame, nil
+}

server/platform/services/remotecluster/mocks_test.go

@@ -0,0 +1,62 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package remotecluster
+
+import (
+	"context"
+
+	"github.com/mattermost/mattermost-server/v6/model"
+	"github.com/mattermost/mattermost-server/v6/plugin/plugintest/mock"
+	"github.com/mattermost/mattermost-server/v6/server/channels/einterfaces"
+	"github.com/mattermost/mattermost-server/v6/server/channels/store"
+	"github.com/mattermost/mattermost-server/v6/server/channels/store/storetest/mocks"
+	"github.com/mattermost/mattermost-server/v6/server/platform/shared/mlog"
+)
+
+type mockServer struct {
+	remotes []*model.RemoteCluster
+	logger  *mlog.Logger
+	user    *model.User
+}
+
+func newMockServer(remotes []*model.RemoteCluster) *mockServer {
+	testLogger := mlog.CreateConsoleTestLogger(true, mlog.LvlDebug)
+
+	return &mockServer{
+		remotes: remotes,
+		logger:  testLogger,
+	}
+}
+
+func (ms *mockServer) SetUser(user *model.User) {
+	ms.user = user
+}
+
+func (ms *mockServer) Config() *model.Config                                  { return nil }
+func (ms *mockServer) GetMetrics() einterfaces.MetricsInterface               { return nil }
+func (ms *mockServer) IsLeader() bool                                         { return true }
+func (ms *mockServer) AddClusterLeaderChangedListener(listener func()) string { return model.NewId() }
+func (ms *mockServer) RemoveClusterLeaderChangedListener(id string)           {}
+func (ms *mockServer) Log() *mlog.Logger {
+	return ms.logger
+}
+func (ms *mockServer) GetStore() store.Store {
+	anyQueryFilter := mock.MatchedBy(func(filter model.RemoteClusterQueryFilter) bool {
+		return true
+	})
+	anyUserId := mock.AnythingOfType("string")
+
+	remoteClusterStoreMock := &mocks.RemoteClusterStore{}
+	remoteClusterStoreMock.On("GetByTopic", "share").Return(ms.remotes, nil)
+	remoteClusterStoreMock.On("GetAll", anyQueryFilter).Return(ms.remotes, nil)
+
+	userStoreMock := &mocks.UserStore{}
+	userStoreMock.On("Get", context.Background(), anyUserId).Return(ms.user, nil)
+
+	storeMock := &mocks.Store{}
+	storeMock.On("RemoteCluster").Return(remoteClusterStoreMock)
+	storeMock.On("User").Return(userStoreMock)
+	return storeMock
+}
+func (ms *mockServer) Shutdown() { ms.logger.Shutdown() }

server/platform/services/remotecluster/ping.go

@@ -0,0 +1,174 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package remotecluster
+
+import (
+	"encoding/json"
+	"fmt"
+	"time"
+
+	"github.com/mattermost/mattermost-server/v6/model"
+	"github.com/mattermost/mattermost-server/v6/server/platform/shared/mlog"
+)
+
+// pingLoop periodically sends a ping to all remote clusters.
+func (rcs *Service) pingLoop(done <-chan struct{}) {
+	pingChan := make(chan *model.RemoteCluster, MaxConcurrentSends*2)
+
+	// create a thread pool to send pings concurrently to remotes.
+	for i := 0; i < MaxConcurrentSends; i++ {
+		go rcs.pingEmitter(pingChan, done)
+	}
+
+	go rcs.pingGenerator(pingChan, done)
+}
+
+func (rcs *Service) pingGenerator(pingChan chan *model.RemoteCluster, done <-chan struct{}) {
+	defer close(pingChan)
+
+	for {
+		start := time.Now()
+
+		// get all remotes, including any previously offline.
+		remotes, err := rcs.server.GetStore().RemoteCluster().GetAll(model.RemoteClusterQueryFilter{})
+		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):
+				continue
+			case <-done:
+				return
+			}
+		}
+
+		for _, rc := range remotes {
+			if rc.SiteURL != "" { // filter out unconfirmed invites
+				pingChan <- rc
+			}
+		}
+
+		// try to maintain frequency
+		elapsed := time.Since(start)
+		if elapsed < PingFreq {
+			sleep := time.Until(start.Add(PingFreq))
+			select {
+			case <-time.After(sleep):
+			case <-done:
+				return
+			}
+		}
+	}
+}
+
+// pingEmitter pulls Remotes from the ping queue (pingChan) and pings them.
+// Pinging a remote cannot take longer than PingTimeoutMillis.
+func (rcs *Service) pingEmitter(pingChan <-chan *model.RemoteCluster, done <-chan struct{}) {
+	for {
+		select {
+		case rc := <-pingChan:
+			if rc == nil {
+				return
+			}
+
+			online := rc.IsOnline()
+
+			if err := rcs.pingRemote(rc); err != nil {
+				rcs.server.Log().Log(mlog.LvlRemoteClusterServiceWarn, "Remote cluster ping failed",
+					mlog.String("remote", rc.DisplayName),
+					mlog.String("remoteId", rc.RemoteId),
+					mlog.Err(err),
+				)
+			}
+
+			if online != rc.IsOnline() {
+				if metrics := rcs.server.GetMetrics(); metrics != nil {
+					metrics.IncrementRemoteClusterConnStateChangeCounter(rc.RemoteId, rc.IsOnline())
+				}
+				rcs.fireConnectionStateChgEvent(rc)
+			}
+		case <-done:
+			return
+		}
+	}
+}
+
+// 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
+	}
+
+	ping := model.RemoteClusterPing{}
+	err = json.Unmarshal(resp, &ping)
+	if err != nil {
+		return err
+	}
+
+	if err := rcs.server.GetStore().RemoteCluster().SetLastPingAt(rc.RemoteId); err != nil {
+		rcs.server.Log().Log(mlog.LvlRemoteClusterServiceError, "Failed to update LastPingAt for remote cluster",
+			mlog.String("remote", rc.DisplayName),
+			mlog.String("remoteId", rc.RemoteId),
+			mlog.Err(err),
+		)
+	}
+	rc.LastPingAt = model.GetMillis()
+
+	if metrics := rcs.server.GetMetrics(); metrics != nil {
+		sentAt := time.Unix(0, ping.SentAt*int64(time.Millisecond))
+		elapsed := time.Since(sentAt).Seconds()
+		metrics.ObserveRemoteClusterPingDuration(rc.RemoteId, elapsed)
+
+		// we approximate clock skew between remotes.
+		skew := elapsed/2 - float64(ping.RecvAt-ping.SentAt)/1000
+		metrics.ObserveRemoteClusterClockSkew(rc.RemoteId, skew)
+	}
+
+	rcs.server.Log().Log(mlog.LvlRemoteClusterServiceDebug, "Remote cluster ping",
+		mlog.String("remote", rc.DisplayName),
+		mlog.String("remoteId", rc.RemoteId),
+		mlog.Int64("SentAt", ping.SentAt),
+		mlog.Int64("RecvAt", ping.RecvAt),
+		mlog.Int64("Diff", ping.RecvAt-ping.SentAt),
+	)
+	return nil
+}
+
+func makePingFrame(rc *model.RemoteCluster) (*model.RemoteClusterFrame, error) {
+	ping := model.RemoteClusterPing{
+		SentAt: model.GetMillis(),
+	}
+	pingRaw, err := json.Marshal(ping)
+	if err != nil {
+		return nil, err
+	}
+
+	msg := model.NewRemoteClusterMsg(PingTopic, pingRaw)
+
+	frame := &model.RemoteClusterFrame{
+		RemoteId: rc.RemoteId,
+		Msg:      msg,
+	}
+	return frame, nil
+}
+
+func (rcs *Service) fireConnectionStateChgEvent(rc *model.RemoteCluster) {
+	rcs.mux.RLock()
+	listeners := make([]ConnectionStateListener, 0, len(rcs.connectionStateListeners))
+	for _, l := range rcs.connectionStateListeners {
+		listeners = append(listeners, l)
+	}
+	rcs.mux.RUnlock()
+
+	for _, l := range listeners {
+		l(rc, rc.IsOnline())
+	}
+}

server/platform/services/remotecluster/ping_test.go

@@ -0,0 +1,142 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package remotecluster
+
+import (
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"sync"
+	"sync/atomic"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"github.com/wiggin77/merror"
+
+	"github.com/mattermost/mattermost-server/v6/model"
+)
+
+const (
+	Recent = 60000
+)
+
+func TestPing(t *testing.T) {
+	disablePing = false
+
+	t.Run("No error", func(t *testing.T) {
+		var countWebReq int32
+		merr := merror.New()
+
+		wg := &sync.WaitGroup{}
+		wg.Add(NumRemotes)
+
+		ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			defer wg.Done()
+			defer w.WriteHeader(200)
+			atomic.AddInt32(&countWebReq, 1)
+
+			var frame model.RemoteClusterFrame
+			err := json.NewDecoder(r.Body).Decode(&frame)
+			if err != nil {
+				merr.Append(err)
+				return
+			}
+			if len(frame.Msg.Payload) == 0 {
+				merr.Append(fmt.Errorf("Payload should not be empty; remote_id=%s", frame.RemoteId))
+				return
+			}
+
+			var ping model.RemoteClusterPing
+			err = json.Unmarshal(frame.Msg.Payload, &ping)
+			if err != nil {
+				merr.Append(err)
+				return
+			}
+			if !checkRecent(ping.SentAt, Recent) {
+				merr.Append(fmt.Errorf("timestamp out of range, got %d", ping.SentAt))
+				return
+			}
+			if ping.RecvAt != 0 {
+				merr.Append(fmt.Errorf("timestamp should be 0, got %d", ping.RecvAt))
+				return
+			}
+		}))
+		defer ts.Close()
+
+		mockServer := newMockServer(makeRemoteClusters(NumRemotes, ts.URL))
+		defer mockServer.Shutdown()
+
+		service, err := NewRemoteClusterService(mockServer)
+		require.NoError(t, err)
+
+		err = service.Start()
+		require.NoError(t, err)
+		defer service.Shutdown()
+
+		wg.Wait()
+
+		assert.NoError(t, merr.ErrorOrNil())
+
+		assert.Equal(t, int32(NumRemotes), atomic.LoadInt32(&countWebReq))
+		t.Logf("%d web requests counted;  %d expected",
+			atomic.LoadInt32(&countWebReq), NumRemotes)
+	})
+
+	t.Run("HTTP errors", func(t *testing.T) {
+		var countWebReq int32
+		merr := merror.New()
+
+		wg := &sync.WaitGroup{}
+		wg.Add(NumRemotes)
+
+		ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			defer wg.Done()
+			atomic.AddInt32(&countWebReq, 1)
+
+			var frame model.RemoteClusterFrame
+			err := json.NewDecoder(r.Body).Decode(&frame)
+			if err != nil {
+				merr.Append(err)
+			}
+			var ping model.RemoteClusterPing
+			err = json.Unmarshal(frame.Msg.Payload, &ping)
+			if err != nil {
+				merr.Append(err)
+			}
+			if !checkRecent(ping.SentAt, Recent) {
+				merr.Append(fmt.Errorf("timestamp out of range, got %d", ping.SentAt))
+			}
+			if ping.RecvAt != 0 {
+				merr.Append(fmt.Errorf("timestamp should be 0, got %d", ping.RecvAt))
+			}
+			w.WriteHeader(500)
+		}))
+		defer ts.Close()
+
+		mockServer := newMockServer(makeRemoteClusters(NumRemotes, ts.URL))
+		defer mockServer.Shutdown()
+
+		service, err := NewRemoteClusterService(mockServer)
+		require.NoError(t, err)
+
+		err = service.Start()
+		require.NoError(t, err)
+		defer service.Shutdown()
+
+		wg.Wait()
+
+		assert.NoError(t, merr.ErrorOrNil())
+
+		assert.Equal(t, int32(NumRemotes), atomic.LoadInt32(&countWebReq))
+		t.Logf("%d web requests counted;  %d expected",
+			atomic.LoadInt32(&countWebReq), NumRemotes)
+	})
+}
+
+func checkRecent(millis int64, within int64) bool {
+	now := model.GetMillis()
+	return millis > now-within && millis < now+within
+}

server/platform/services/remotecluster/recv.go

@@ -0,0 +1,53 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package remotecluster
+
+import (
+	"fmt"
+
+	"github.com/mattermost/mattermost-server/v6/model"
+	"github.com/mattermost/mattermost-server/v6/server/platform/shared/mlog"
+)
+
+// ReceiveIncomingMsg is called by the Rest API layer, or websocket layer (future), when a Remote Cluster
+// message is received.  Here we route the message to any topic listeners.
+// `rc` and `msg` cannot be nil.
+func (rcs *Service) ReceiveIncomingMsg(rc *model.RemoteCluster, msg model.RemoteClusterMsg) Response {
+	rcs.mux.RLock()
+	defer rcs.mux.RUnlock()
+
+	if metrics := rcs.server.GetMetrics(); metrics != nil {
+		metrics.IncrementRemoteClusterMsgReceivedCounter(rc.RemoteId)
+	}
+
+	rcSanitized := *rc
+	rcSanitized.Token = ""
+	rcSanitized.RemoteToken = ""
+
+	var response Response
+	response.Status = ResponseStatusOK
+
+	listeners := rcs.getTopicListeners(msg.Topic)
+
+	for _, l := range listeners {
+		if err := callback(l, msg, &rcSanitized, &response); err != nil {
+			rcs.server.Log().Log(mlog.LvlRemoteClusterServiceError, "Error from remote cluster message listener",
+				mlog.String("msgId", msg.Id), mlog.String("topic", msg.Topic), mlog.String("remote", rc.DisplayName), mlog.Err(err))
+
+			response.Status = ResponseStatusFail
+			response.Err = err.Error()
+		}
+	}
+	return response
+}
+
+func callback(listener TopicListener, msg model.RemoteClusterMsg, rc *model.RemoteCluster, resp *Response) (err error) {
+	defer func() {
+		if r := recover(); r != nil {
+			err = fmt.Errorf("%v", r)
+		}
+	}()
+	err = listener(msg, rc, resp)
+	return
+}

server/platform/services/remotecluster/response.go

@@ -0,0 +1,30 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package remotecluster
+
+import (
+	"encoding/json"
+)
+
+// Response represents the bytes replied from a remote server when a message is sent.
+type Response struct {
+	Status  string          `json:"status"`
+	Err     string          `json:"err"`
+	Payload json.RawMessage `json:"payload"`
+}
+
+// IsSuccess returns true if the response status indicates success.
+func (r *Response) IsSuccess() bool {
+	return r.Status == ResponseStatusOK
+}
+
+// SetPayload serializes an arbitrary struct as a RawMessage.
+func (r *Response) SetPayload(v any) error {
+	raw, err := json.Marshal(v)
+	if err != nil {
+		return err
+	}
+	r.Payload = raw
+	return nil
+}

server/platform/services/remotecluster/send.go

@@ -0,0 +1,58 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package remotecluster
+
+import (
+	"context"
+	"hash/fnv"
+)
+
+// enqueueTask adds a task to one of the send channels based on remoteId.
+//
+// There are a number  of send channels (`MaxConcurrentSends`) to allow for sending to multiple
+// remotes concurrently, while preserving message order for each remote.
+func (rcs *Service) enqueueTask(ctx context.Context, remoteId string, task any) error {
+	if ctx == nil {
+		ctx = context.Background()
+	}
+
+	h := hash(remoteId)
+	idx := h % uint32(len(rcs.send))
+
+	select {
+	case rcs.send[idx] <- task:
+		return nil
+	case <-ctx.Done():
+		return NewBufferFullError(cap(rcs.send))
+	}
+}
+
+func hash(s string) uint32 {
+	h := fnv.New32a()
+	h.Write([]byte(s))
+	return h.Sum32()
+}
+
+// sendLoop is called by each goroutine created for the send pool and waits for sendTask's until the
+// done channel is signalled.
+//
+// Each goroutine in the pool is assigned a specific channel, and tasks are placed in the
+// channel corresponding to the remoteId.
+func (rcs *Service) sendLoop(idx int, done chan struct{}) {
+	for {
+		select {
+		case t := <-rcs.send[idx]:
+			switch task := t.(type) {
+			case sendMsgTask:
+				rcs.sendMsg(task)
+			case sendFileTask:
+				rcs.sendFile(task)
+			case sendProfileImageTask:
+				rcs.sendProfileImage(task)
+			}
+		case <-done:
+			return
+		}
+	}
+}

server/platform/services/remotecluster/send_test.go

@@ -0,0 +1,205 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package remotecluster
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"github.com/wiggin77/merror"
+
+	"github.com/mattermost/mattermost-server/v6/model"
+)
+
+const (
+	TestTopics  = " share incident "
+	TestTopic   = "share"
+	NumRemotes  = 50
+	NoteContent = "Woot!!"
+)
+
+type testPayload struct {
+	Note string `json:"note"`
+}
+
+func TestBroadcastMsg(t *testing.T) {
+	msgId := model.NewId()
+	disablePing = true
+
+	t.Run("No error", func(t *testing.T) {
+		var countCallbacks int32
+		var countWebReq int32
+		merr := merror.New()
+
+		ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			defer func() {
+				w.WriteHeader(200)
+				var resp Response
+				b, errMarshall := json.Marshal(&resp)
+				if errMarshall != nil {
+					merr.Append(errMarshall)
+					return
+				}
+				w.Write(b)
+			}()
+
+			atomic.AddInt32(&countWebReq, 1)
+
+			var frame model.RemoteClusterFrame
+			jsonErr := json.NewDecoder(r.Body).Decode(&frame)
+			if jsonErr != nil {
+				merr.Append(jsonErr)
+				return
+			}
+			if len(frame.Msg.Payload) == 0 {
+				merr.Append(fmt.Errorf("webrequest missing Msg.Payload"))
+			}
+			if msgId != frame.Msg.Id {
+				merr.Append(fmt.Errorf("webrequest msgId expected %s, got %s", msgId, frame.Msg.Id))
+				return
+			}
+
+			note := testPayload{}
+			err := json.Unmarshal(frame.Msg.Payload, &note)
+			if err != nil {
+				merr.Append(err)
+				return
+			}
+			if note.Note != NoteContent {
+				merr.Append(fmt.Errorf("webrequest payload expected %s, got %s", NoteContent, note.Note))
+				return
+			}
+		}))
+		defer ts.Close()
+
+		mockServer := newMockServer(makeRemoteClusters(NumRemotes, ts.URL))
+		defer mockServer.Shutdown()
+
+		service, err := NewRemoteClusterService(mockServer)
+		require.NoError(t, err)
+
+		err = service.Start()
+		require.NoError(t, err)
+		defer service.Shutdown()
+
+		wg := &sync.WaitGroup{}
+		wg.Add(NumRemotes)
+
+		msg := makeRemoteClusterMsg(msgId, NoteContent)
+
+		ctx, cancel := context.WithTimeout(context.Background(), time.Second*15)
+		defer cancel()
+
+		err = service.BroadcastMsg(ctx, msg, func(msg model.RemoteClusterMsg, remote *model.RemoteCluster, resp *Response, err error) {
+			defer wg.Done()
+			atomic.AddInt32(&countCallbacks, 1)
+
+			if err != nil {
+				merr.Append(err)
+			}
+			if msgId != msg.Id {
+				merr.Append(fmt.Errorf("result callback msgId expected %s, got %s", msgId, msg.Id))
+			}
+
+			var note testPayload
+			err2 := json.Unmarshal(msg.Payload, &note)
+			if err2 != nil {
+				merr.Append(fmt.Errorf("unmarshal payload error: %w", err2))
+				return
+			}
+			if note.Note != NoteContent {
+				merr.Append(fmt.Errorf("compare payload failed: expected '%s', got '%s'", NoteContent, note))
+			}
+		})
+		assert.NoError(t, err)
+
+		wg.Wait()
+
+		assert.NoError(t, merr.ErrorOrNil())
+
+		assert.Equal(t, int32(NumRemotes), atomic.LoadInt32(&countCallbacks))
+		assert.Equal(t, int32(NumRemotes), atomic.LoadInt32(&countWebReq))
+		t.Logf("%d callbacks counted;  %d web requests counted;  %d expected",
+			atomic.LoadInt32(&countCallbacks), atomic.LoadInt32(&countWebReq), NumRemotes)
+	})
+
+	t.Run("HTTP error", func(t *testing.T) {
+		ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			w.WriteHeader(500)
+		}))
+		defer ts.Close()
+
+		mockServer := newMockServer(makeRemoteClusters(NumRemotes, ts.URL))
+		defer mockServer.Shutdown()
+
+		service, err := NewRemoteClusterService(mockServer)
+		require.NoError(t, err)
+
+		err = service.Start()
+		require.NoError(t, err)
+		defer service.Shutdown()
+
+		msg := makeRemoteClusterMsg(msgId, NoteContent)
+		var countCallbacks int32
+		var countErrors int32
+		wg := &sync.WaitGroup{}
+		wg.Add(NumRemotes)
+
+		err = service.BroadcastMsg(context.Background(), msg, func(msg model.RemoteClusterMsg, remote *model.RemoteCluster, resp *Response, err error) {
+			defer wg.Done()
+			atomic.AddInt32(&countCallbacks, 1)
+			if err != nil {
+				atomic.AddInt32(&countErrors, 1)
+			}
+		})
+		assert.NoError(t, err)
+
+		wg.Wait()
+
+		assert.Equal(t, int32(NumRemotes), atomic.LoadInt32(&countCallbacks))
+		assert.Equal(t, int32(NumRemotes), atomic.LoadInt32(&countErrors))
+	})
+}
+
+func makeRemoteClusters(num int, siteURL string) []*model.RemoteCluster {
+	var remotes []*model.RemoteCluster
+	for i := 0; i < num; i++ {
+		rc := makeRemoteCluster(fmt.Sprintf("test cluster %d", i+1), siteURL, TestTopics)
+		remotes = append(remotes, rc)
+	}
+	return remotes
+}
+
+func makeRemoteCluster(name string, siteURL string, topics string) *model.RemoteCluster {
+	return &model.RemoteCluster{
+		RemoteId:   model.NewId(),
+		Name:       name,
+		SiteURL:    siteURL,
+		Token:      model.NewId(),
+		Topics:     topics,
+		CreateAt:   model.GetMillis(),
+		LastPingAt: model.GetMillis(),
+		CreatorId:  model.NewId(),
+	}
+}
+
+func makeRemoteClusterMsg(id string, note string) model.RemoteClusterMsg {
+	payload := testPayload{Note: note}
+	raw, _ := json.Marshal(payload)
+
+	return model.RemoteClusterMsg{
+		Id:       id,
+		Topic:    TestTopic,
+		CreateAt: model.GetMillis(),
+		Payload:  raw}
+}

server/platform/services/remotecluster/sendfile.go

@@ -0,0 +1,136 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package remotecluster
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"net/url"
+	"path"
+	"time"
+
+	"github.com/mattermost/mattermost-server/v6/model"
+	"github.com/mattermost/mattermost-server/v6/server/platform/shared/filestore"
+	"github.com/mattermost/mattermost-server/v6/server/platform/shared/mlog"
+)
+
+type SendFileResultFunc func(us *model.UploadSession, rc *model.RemoteCluster, resp *Response, err error)
+
+type sendFileTask struct {
+	rc *model.RemoteCluster
+	us *model.UploadSession
+	fi *model.FileInfo
+	rp ReaderProvider
+	f  SendFileResultFunc
+}
+
+type ReaderProvider interface {
+	FileReader(path string) (filestore.ReadCloseSeeker, *model.AppError)
+}
+
+// SendFile asynchronously sends a file to a remote cluster.
+//
+// `ctx` determines behaviour when the outbound queue is full. A timeout or deadline context will return a
+// BufferFullError if the task cannot be enqueued before the timeout. A background context will block indefinitely.
+//
+// Nil or error return indicates success or failure of task enqueue only.
+//
+// An optional callback can be provided that receives the response from the remote cluster. The `err` provided to the
+// callback is regarding file delivery only. The `resp` contains the decoded bytes returned from the remote.
+// If a callback is provided it should return quickly.
+func (rcs *Service) SendFile(ctx context.Context, us *model.UploadSession, fi *model.FileInfo, rc *model.RemoteCluster, rp ReaderProvider, f SendFileResultFunc) error {
+	task := sendFileTask{
+		rc: rc,
+		us: us,
+		fi: fi,
+		rp: rp,
+		f:  f,
+	}
+	return rcs.enqueueTask(ctx, rc.RemoteId, task)
+}
+
+// sendFile is called when a sendFileTask is popped from the send channel.
+func (rcs *Service) sendFile(task sendFileTask) {
+	fi, err := rcs.sendFileToRemote(SendTimeout, task)
+	var response Response
+
+	if err != nil {
+		rcs.server.Log().Log(mlog.LvlRemoteClusterServiceError, "Remote Cluster send file failed",
+			mlog.String("remote", task.rc.DisplayName),
+			mlog.String("uploadId", task.us.Id),
+			mlog.Err(err),
+		)
+		response.Status = ResponseStatusFail
+		response.Err = err.Error()
+	} else {
+		rcs.server.Log().Log(mlog.LvlRemoteClusterServiceDebug, "Remote Cluster file sent successfully",
+			mlog.String("remote", task.rc.DisplayName),
+			mlog.String("uploadId", task.us.Id),
+		)
+		response.Status = ResponseStatusOK
+		response.SetPayload(fi)
+	}
+
+	// If callback provided then call it with the results.
+	if task.f != nil {
+		task.f(task.us, task.rc, &response, err)
+	}
+}
+
+func (rcs *Service) sendFileToRemote(timeout time.Duration, task sendFileTask) (*model.FileInfo, error) {
+	rcs.server.Log().Log(mlog.LvlRemoteClusterServiceDebug, "sending file to remote...",
+		mlog.String("remote", task.rc.DisplayName),
+		mlog.String("uploadId", task.us.Id),
+		mlog.String("file_path", task.us.Path),
+	)
+
+	r, appErr := task.rp.FileReader(task.fi.Path) // get Reader for the file
+	if appErr != nil {
+		return nil, fmt.Errorf("error opening file while sending file to remote %s: %w", task.rc.RemoteId, appErr)
+	}
+	defer r.Close()
+
+	u, err := url.Parse(task.rc.SiteURL)
+	if err != nil {
+		return nil, fmt.Errorf("invalid siteURL while sending file to remote %s: %w", task.rc.RemoteId, err)
+	}
+	u.Path = path.Join(u.Path, model.APIURLSuffix, "remotecluster", "upload", task.us.Id)
+
+	req, err := http.NewRequest("POST", u.String(), r)
+	if err != nil {
+		return nil, err
+	}
+
+	req.Header.Set(model.HeaderRemoteclusterId, task.rc.RemoteId)
+	req.Header.Set(model.HeaderRemoteclusterToken, task.rc.RemoteToken)
+
+	ctx, cancel := context.WithTimeout(context.Background(), timeout)
+	defer cancel()
+
+	resp, err := rcs.httpClient.Do(req.WithContext(ctx))
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return nil, err
+	}
+
+	if resp.StatusCode != http.StatusOK {
+		return nil, fmt.Errorf("unexpected response: %d - %s", resp.StatusCode, resp.Status)
+	}
+
+	// body should be a FileInfo
+	var fi model.FileInfo
+	if err := json.Unmarshal(body, &fi); err != nil {
+		return nil, fmt.Errorf("unexpected response body: %w", err)
+	}
+
+	return &fi, nil
+}

server/platform/services/remotecluster/sendmsg.go

@@ -0,0 +1,175 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package remotecluster
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"net/url"
+	"os"
+	"path"
+	"time"
+
+	"github.com/wiggin77/merror"
+
+	"github.com/mattermost/mattermost-server/v6/model"
+	"github.com/mattermost/mattermost-server/v6/server/platform/shared/mlog"
+)
+
+type SendMsgResultFunc func(msg model.RemoteClusterMsg, rc *model.RemoteCluster, resp *Response, err error)
+
+type sendMsgTask struct {
+	rc  *model.RemoteCluster
+	msg model.RemoteClusterMsg
+	f   SendMsgResultFunc
+}
+
+// BroadcastMsg asynchronously sends a message to all remote clusters interested in the message's topic.
+//
+// `ctx` determines behaviour when the outbound queue is full. A timeout or deadline context will return a
+// BufferFullError if the message cannot be enqueued before the timeout. A background context will block indefinitely.
+//
+// An optional callback can be provided that receives the success or fail result of sending to each remote cluster.
+// Success or fail is regarding message delivery only.  If a callback is provided it should return quickly.
+func (rcs *Service) BroadcastMsg(ctx context.Context, msg model.RemoteClusterMsg, f SendMsgResultFunc) error {
+	// get list of interested remotes.
+	filter := model.RemoteClusterQueryFilter{
+		Topic: msg.Topic,
+	}
+	list, err := rcs.server.GetStore().RemoteCluster().GetAll(filter)
+	if err != nil {
+		return err
+	}
+
+	errs := merror.New()
+
+	for _, rc := range list {
+		if err := rcs.SendMsg(ctx, msg, rc, f); err != nil {
+			errs.Append(err)
+		}
+	}
+	return errs.ErrorOrNil()
+}
+
+// SendMsg asynchronously sends a message to a remote cluster.
+//
+// `ctx` determines behaviour when the outbound queue is full. A timeout or deadline context will return a
+// BufferFullError if the message cannot be enqueued before the timeout. A background context will block indefinitely.
+//
+// Nil or error return indicates success or failure of message enqueue only.
+//
+// An optional callback can be provided that receives the response from the remote cluster. The `err` provided to the
+// callback is regarding response decoding only. The `resp` contains the decoded bytes returned from the remote.
+// If a callback is provided it should return quickly.
+func (rcs *Service) SendMsg(ctx context.Context, msg model.RemoteClusterMsg, rc *model.RemoteCluster, f SendMsgResultFunc) error {
+	task := sendMsgTask{
+		rc:  rc,
+		msg: msg,
+		f:   f,
+	}
+	return rcs.enqueueTask(ctx, rc.RemoteId, task)
+}
+
+// sendMsg is called when a sendMsgTask is popped from the send channel.
+func (rcs *Service) sendMsg(task sendMsgTask) {
+	var errResp error
+	var response Response
+
+	// Ensure a panic from the callback does not exit the pool goroutine.
+	defer func() {
+		if errResp != nil {
+			response.Err = errResp.Error()
+		}
+
+		// If callback provided then call it with the results.
+		if task.f != nil {
+			task.f(task.msg, task.rc, &response, errResp)
+		}
+	}()
+
+	frame := &model.RemoteClusterFrame{
+		RemoteId: task.rc.RemoteId,
+		Msg:      task.msg,
+	}
+
+	u, err := url.Parse(task.rc.SiteURL)
+	if err != nil {
+		rcs.server.Log().Log(mlog.LvlRemoteClusterServiceError, "Invalid siteURL while sending message to remote",
+			mlog.String("remote", task.rc.DisplayName),
+			mlog.String("msgId", task.msg.Id),
+			mlog.Err(err),
+		)
+		errResp = err
+		return
+	}
+	u.Path = path.Join(u.Path, SendMsgURL)
+
+	respJSON, err := rcs.sendFrameToRemote(SendTimeout, task.rc, frame, u.String())
+
+	if err != nil {
+		rcs.server.Log().Log(mlog.LvlRemoteClusterServiceError, "Remote Cluster send message failed",
+			mlog.String("remote", task.rc.DisplayName),
+			mlog.String("msgId", task.msg.Id),
+			mlog.Err(err),
+		)
+		errResp = err
+	} else {
+		rcs.server.Log().Log(mlog.LvlRemoteClusterServiceDebug, "Remote Cluster message sent successfully",
+			mlog.String("remote", task.rc.DisplayName),
+			mlog.String("msgId", task.msg.Id),
+		)
+
+		if err = json.Unmarshal(respJSON, &response); err != nil {
+			rcs.server.Log().Error("Invalid response sending message to remote cluster",
+				mlog.String("remote", task.rc.DisplayName),
+				mlog.Err(err),
+			)
+			errResp = err
+		}
+	}
+}
+
+func (rcs *Service) sendFrameToRemote(timeout time.Duration, rc *model.RemoteCluster, frame *model.RemoteClusterFrame, url string) ([]byte, error) {
+	body, err := json.Marshal(frame)
+	if err != nil {
+		return nil, err
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), timeout)
+	defer cancel()
+
+	req, err := http.NewRequest("POST", url, bytes.NewReader(body))
+	if err != nil {
+		return nil, err
+	}
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set(model.HeaderRemoteclusterId, rc.RemoteId)
+	req.Header.Set(model.HeaderRemoteclusterToken, rc.RemoteToken)
+
+	resp, err := rcs.httpClient.Do(req.WithContext(ctx))
+	if metrics := rcs.server.GetMetrics(); metrics != nil {
+		if err != nil || resp.StatusCode != http.StatusOK {
+			metrics.IncrementRemoteClusterMsgErrorsCounter(frame.RemoteId, os.IsTimeout(err))
+		} else {
+			metrics.IncrementRemoteClusterMsgSentCounter(frame.RemoteId)
+		}
+	}
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+	body, err = io.ReadAll(resp.Body)
+	if err != nil {
+		return nil, err
+	}
+
+	if resp.StatusCode != http.StatusOK {
+		return body, fmt.Errorf("unexpected response: %d - %s", resp.StatusCode, resp.Status)
+	}
+	return body, nil
+}

server/platform/services/remotecluster/sendprofileImage.go

@@ -0,0 +1,145 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package remotecluster
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"io"
+	"mime/multipart"
+	"net/http"
+	"net/url"
+	"path"
+	"time"
+
+	"github.com/mattermost/mattermost-server/v6/model"
+	"github.com/mattermost/mattermost-server/v6/server/platform/shared/mlog"
+)
+
+type SendProfileImageResultFunc func(userId string, rc *model.RemoteCluster, resp *Response, err error)
+
+type sendProfileImageTask struct {
+	rc       *model.RemoteCluster
+	userID   string
+	provider ProfileImageProvider
+	f        SendProfileImageResultFunc
+}
+
+type ProfileImageProvider interface {
+	GetProfileImage(user *model.User) ([]byte, bool, *model.AppError)
+}
+
+// SendProfileImage asynchronously sends a user's profile image to a remote cluster.
+//
+// `ctx` determines behaviour when the outbound queue is full. A timeout or deadline context will return a
+// BufferFullError if the task cannot be enqueued before the timeout. A background context will block indefinitely.
+//
+// Nil or error return indicates success or failure of task enqueue only.
+//
+// An optional callback can be provided that receives the response from the remote cluster. The `err` provided to the
+// callback is regarding image delivery only. The `resp` contains the decoded bytes returned from the remote.
+// If a callback is provided it should return quickly.
+func (rcs *Service) SendProfileImage(ctx context.Context, userID string, rc *model.RemoteCluster, provider ProfileImageProvider, f SendProfileImageResultFunc) error {
+	task := sendProfileImageTask{
+		rc:       rc,
+		userID:   userID,
+		provider: provider,
+		f:        f,
+	}
+	return rcs.enqueueTask(ctx, rc.RemoteId, task)
+}
+
+// sendProfileImage is called when a sendProfileImageTask is popped from the send channel.
+func (rcs *Service) sendProfileImage(task sendProfileImageTask) {
+	err := rcs.sendProfileImageToRemote(SendTimeout, task)
+	var response Response
+
+	if err != nil {
+		rcs.server.Log().Log(mlog.LvlRemoteClusterServiceError, "Remote Cluster send profile image failed",
+			mlog.String("remote", task.rc.DisplayName),
+			mlog.String("UserId", task.userID),
+			mlog.Err(err),
+		)
+		response.Status = ResponseStatusFail
+		response.Err = err.Error()
+	} else {
+		rcs.server.Log().Log(mlog.LvlRemoteClusterServiceDebug, "Remote Cluster profile image sent successfully",
+			mlog.String("remote", task.rc.DisplayName),
+			mlog.String("UserId", task.userID),
+		)
+		response.Status = ResponseStatusOK
+	}
+
+	// If callback provided then call it with the results.
+	if task.f != nil {
+		task.f(task.userID, task.rc, &response, err)
+	}
+}
+
+func (rcs *Service) sendProfileImageToRemote(timeout time.Duration, task sendProfileImageTask) error {
+	rcs.server.Log().Log(mlog.LvlRemoteClusterServiceDebug, "sending profile image to remote...",
+		mlog.String("remote", task.rc.DisplayName),
+		mlog.String("UserId", task.userID),
+	)
+
+	user, err := rcs.server.GetStore().User().Get(context.Background(), task.userID)
+	if err != nil {
+		return fmt.Errorf("error fetching user while sending profile image to remote %s: %w", task.rc.RemoteId, err)
+	}
+
+	img, _, appErr := task.provider.GetProfileImage(user) // get Reader for the file
+	if appErr != nil {
+		return fmt.Errorf("error fetching profile image for user (%s) while sending to remote %s: %w", task.userID, task.rc.RemoteId, appErr)
+	}
+
+	u, err := url.Parse(task.rc.SiteURL)
+	if err != nil {
+		return fmt.Errorf("invalid siteURL while sending file to remote %s: %w", task.rc.RemoteId, err)
+	}
+	u.Path = path.Join(u.Path, model.APIURLSuffix, "remotecluster", task.userID, "image")
+
+	body := &bytes.Buffer{}
+	writer := multipart.NewWriter(body)
+
+	part, err := writer.CreateFormFile("image", "profile.png")
+	if err != nil {
+		return err
+	}
+
+	if _, err = io.Copy(part, bytes.NewBuffer(img)); err != nil {
+		return err
+	}
+
+	if err = writer.Close(); err != nil {
+		return err
+	}
+
+	req, err := http.NewRequest("POST", u.String(), body)
+	if err != nil {
+		return err
+	}
+	req.Header.Set("Content-Type", writer.FormDataContentType())
+	req.Header.Set(model.HeaderRemoteclusterId, task.rc.RemoteId)
+	req.Header.Set(model.HeaderRemoteclusterToken, task.rc.RemoteToken)
+
+	ctx, cancel := context.WithTimeout(context.Background(), timeout)
+	defer cancel()
+
+	resp, err := rcs.httpClient.Do(req.WithContext(ctx))
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+
+	_, err = io.ReadAll(resp.Body)
+	if err != nil {
+		return err
+	}
+
+	if resp.StatusCode != http.StatusOK {
+		return fmt.Errorf("unexpected response: %d - %s", resp.StatusCode, resp.Status)
+	}
+	return nil
+}

server/platform/services/remotecluster/sendprofileImage_test.go

@@ -0,0 +1,187 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package remotecluster
+
+import (
+	"bytes"
+	"image"
+	"image/color"
+	"image/png"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/mattermost/mattermost-server/v6/model"
+)
+
+const (
+	imageWidth  = 128
+	imageHeight = 128
+)
+
+func TestService_sendProfileImageToRemote(t *testing.T) {
+	hadPing := disablePing
+	disablePing = true
+	defer func() { disablePing = hadPing }()
+
+	shouldError := &flag{}
+
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		defer io.Copy(io.Discard, r.Body)
+
+		if shouldError.get() {
+			w.WriteHeader(http.StatusInternalServerError)
+			resp := make(map[string]string)
+			resp[model.STATUS] = model.StatusFail
+			w.Write([]byte(model.MapToJSON(resp)))
+			return
+		}
+
+		status := model.StatusOk
+		defer func(s *string) {
+			if *s != model.StatusOk {
+				w.WriteHeader(http.StatusInternalServerError)
+			}
+			resp := make(map[string]string)
+			resp[model.STATUS] = *s
+			w.Write([]byte(model.MapToJSON(resp)))
+		}(&status)
+
+		if err := r.ParseMultipartForm(1024 * 1024); err != nil {
+			status = model.StatusFail
+			assert.Fail(t, "connect parse multipart form", err)
+			return
+		}
+		m := r.MultipartForm
+		if m == nil {
+			status = model.StatusFail
+			assert.Fail(t, "multipart form missing")
+			return
+		}
+
+		imageArray, ok := m.File["image"]
+		if !ok || len(imageArray) != 1 {
+			status = model.StatusFail
+			assert.Fail(t, "image missing")
+			return
+		}
+
+		imageData := imageArray[0]
+		file, err := imageData.Open()
+		if err != nil {
+			status = model.StatusFail
+			assert.Fail(t, "cannot open multipart form file")
+			return
+		}
+		defer file.Close()
+
+		img, err := png.Decode(file)
+		if err != nil || imageWidth != img.Bounds().Max.X || imageHeight != img.Bounds().Max.Y {
+			status = model.StatusFail
+			assert.Fail(t, "cannot decode png", err)
+			return
+		}
+	}))
+	defer ts.Close()
+
+	rc := makeRemoteCluster("remote_test_profile_image", ts.URL, TestTopics)
+
+	user := &model.User{
+		Id:       model.NewId(),
+		RemoteId: model.NewString(rc.RemoteId),
+	}
+
+	provider := testImageProvider{}
+
+	mockServer := newMockServer(makeRemoteClusters(NumRemotes, ts.URL))
+	defer mockServer.Shutdown()
+	mockServer.SetUser(user)
+	service, err := NewRemoteClusterService(mockServer)
+	require.NoError(t, err)
+
+	err = service.Start()
+	require.NoError(t, err)
+	defer service.Shutdown()
+
+	t.Run("Server response 200", func(t *testing.T) {
+		shouldError.set(false)
+
+		resultFunc := func(userId string, rc *model.RemoteCluster, resp *Response, err error) {
+			assert.Equal(t, user.Id, userId, "user ids should match")
+			assert.NoError(t, err)
+			assert.True(t, resp.IsSuccess())
+		}
+
+		task := sendProfileImageTask{
+			rc:       rc,
+			userID:   user.Id,
+			provider: provider,
+			f:        resultFunc,
+		}
+
+		err := service.sendProfileImageToRemote(time.Second*15, task)
+		assert.NoError(t, err, "request should not error")
+	})
+
+	t.Run("Server response 500", func(t *testing.T) {
+		shouldError.set(true)
+
+		resultFunc := func(userId string, rc *model.RemoteCluster, resp *Response, err error) {
+			assert.Equal(t, user.Id, userId, "user ids should match")
+			assert.False(t, resp.IsSuccess())
+		}
+
+		task := sendProfileImageTask{
+			rc:       rc,
+			userID:   user.Id,
+			provider: provider,
+			f:        resultFunc,
+		}
+
+		err := service.sendProfileImageToRemote(time.Second*15, task)
+		assert.Error(t, err, "request should error")
+	})
+}
+
+type testImageProvider struct {
+}
+
+func (tip testImageProvider) GetProfileImage(user *model.User) ([]byte, bool, *model.AppError) {
+	img := image.NewRGBA(image.Rectangle{image.Point{0, 0}, image.Point{imageWidth, imageHeight}})
+	red := color.RGBA{255, 50, 50, 0xff}
+
+	for x := 0; x < imageWidth; x++ {
+		for y := 0; y < imageHeight; y++ {
+			img.Set(x, y, red)
+		}
+	}
+
+	buf := &bytes.Buffer{}
+	png.Encode(buf, img)
+
+	return buf.Bytes(), true, nil
+}
+
+type flag struct {
+	mux sync.RWMutex
+	b   bool
+}
+
+func (f *flag) get() bool {
+	f.mux.RLock()
+	defer f.mux.RUnlock()
+	return f.b
+}
+
+func (f *flag) set(b bool) {
+	f.mux.Lock()
+	defer f.mux.Unlock()
+	f.b = b
+}

server/platform/services/remotecluster/service.go

@@ -0,0 +1,262 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package remotecluster
+
+import (
+	"context"
+	"net"
+	"net/http"
+	"sync"
+	"time"
+
+	"github.com/mattermost/mattermost-server/v6/model"
+	"github.com/mattermost/mattermost-server/v6/server/channels/einterfaces"
+	"github.com/mattermost/mattermost-server/v6/server/channels/store"
+	"github.com/mattermost/mattermost-server/v6/server/platform/shared/mlog"
+)
+
+const (
+	SendChanBuffer                = 50
+	RecvChanBuffer                = 50
+	ResultsChanBuffer             = 50
+	ResultQueueDrainTimeoutMillis = 10000
+	MaxConcurrentSends            = 10
+	SendMsgURL                    = "api/v4/remotecluster/msg"
+	SendTimeout                   = time.Minute
+	SendFileTimeout               = time.Minute * 5
+	PingURL                       = "api/v4/remotecluster/ping"
+	PingFreq                      = time.Minute
+	PingTimeout                   = time.Second * 15
+	ConfirmInviteURL              = "api/v4/remotecluster/confirm_invite"
+	InvitationTopic               = "invitation"
+	PingTopic                     = "ping"
+	ResponseStatusOK              = model.StatusOk
+	ResponseStatusFail            = model.StatusFail
+	InviteExpiresAfter            = time.Hour * 48
+)
+
+var (
+	disablePing bool // override for testing
+)
+
+type ServerIface interface {
+	Config() *model.Config
+	IsLeader() bool
+	AddClusterLeaderChangedListener(listener func()) string
+	RemoveClusterLeaderChangedListener(id string)
+	GetStore() store.Store
+	Log() *mlog.Logger
+	GetMetrics() einterfaces.MetricsInterface
+}
+
+// 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.
+type RemoteClusterServiceIFace interface {
+	Shutdown() error
+	Start() error
+	Active() bool
+	AddTopicListener(topic string, listener TopicListener) string
+	RemoveTopicListener(listenerId string)
+	AddConnectionStateListener(listener ConnectionStateListener) string
+	RemoveConnectionStateListener(listenerId string)
+	SendMsg(ctx context.Context, msg model.RemoteClusterMsg, rc *model.RemoteCluster, f SendMsgResultFunc) error
+	SendFile(ctx context.Context, us *model.UploadSession, fi *model.FileInfo, rc *model.RemoteCluster, rp ReaderProvider, f SendFileResultFunc) error
+	SendProfileImage(ctx context.Context, userID string, rc *model.RemoteCluster, provider ProfileImageProvider, f SendProfileImageResultFunc) error
+	AcceptInvitation(invite *model.RemoteClusterInvite, name string, displayName string, creatorId string, teamId string, siteURL string) (*model.RemoteCluster, error)
+	ReceiveIncomingMsg(rc *model.RemoteCluster, msg model.RemoteClusterMsg) Response
+}
+
+// TopicListener is a callback signature used to listen for incoming messages for
+// a specific topic.
+type TopicListener func(msg model.RemoteClusterMsg, rc *model.RemoteCluster, resp *Response) error
+
+// ConnectionStateListener is used to listen to remote cluster connection state changes.
+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
+	httpClient *http.Client
+	send       []chan any
+
+	// everything below guarded by `mux`
+	mux                      sync.RWMutex
+	active                   bool
+	leaderListenerId         string
+	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{}
+}
+
+// NewRemoteClusterService creates a RemoteClusterService instance. In product this is called a "Secured Connection".
+func NewRemoteClusterService(server ServerIface) (*Service, error) {
+	transport := &http.Transport{
+		Proxy: http.ProxyFromEnvironment,
+		DialContext: (&net.Dialer{
+			Timeout:   30 * time.Second,
+			KeepAlive: 30 * time.Second,
+			DualStack: true,
+		}).DialContext,
+		ForceAttemptHTTP2:     true,
+		MaxIdleConns:          200,
+		MaxIdleConnsPerHost:   2,
+		IdleConnTimeout:       90 * time.Second,
+		TLSHandshakeTimeout:   10 * time.Second,
+		ExpectContinueTimeout: 1 * time.Second,
+		DisableCompression:    false,
+	}
+
+	client := &http.Client{
+		Transport: transport,
+		Timeout:   SendTimeout,
+	}
+
+	service := &Service{
+		server:                   server,
+		httpClient:               client,
+		topicListeners:           make(map[string]map[string]TopicListener),
+		connectionStateListeners: make(map[string]ConnectionStateListener),
+	}
+
+	service.send = make([]chan any, MaxConcurrentSends)
+	for i := range service.send {
+		service.send[i] = make(chan any, SendChanBuffer)
+	}
+
+	return service, nil
+}
+
+// Start is called by the server on server start-up.
+func (rcs *Service) Start() error {
+	rcs.mux.Lock()
+	rcs.leaderListenerId = rcs.server.AddClusterLeaderChangedListener(rcs.onClusterLeaderChange)
+	rcs.mux.Unlock()
+
+	rcs.onClusterLeaderChange()
+
+	return nil
+}
+
+// Shutdown is called by the server on server shutdown.
+func (rcs *Service) Shutdown() error {
+	rcs.server.RemoveClusterLeaderChangedListener(rcs.leaderListenerId)
+	rcs.pause()
+	return nil
+}
+
+// Active returns true if this instance of the remote cluster service is active.
+// The active instance is responsible for pinging and sending messages to remotes.
+func (rcs *Service) Active() bool {
+	rcs.mux.Lock()
+	defer rcs.mux.Unlock()
+	return rcs.active
+}
+
+// AddTopicListener registers a callback
+func (rcs *Service) AddTopicListener(topic string, listener TopicListener) string {
+	rcs.mux.Lock()
+	defer rcs.mux.Unlock()
+
+	id := model.NewId()
+
+	listeners, ok := rcs.topicListeners[topic]
+	if !ok || listeners == nil {
+		rcs.topicListeners[topic] = make(map[string]TopicListener)
+	}
+	rcs.topicListeners[topic][id] = listener
+	return id
+}
+
+func (rcs *Service) RemoveTopicListener(listenerId string) {
+	rcs.mux.Lock()
+	defer rcs.mux.Unlock()
+
+	for topic, listeners := range rcs.topicListeners {
+		if _, ok := listeners[listenerId]; ok {
+			delete(listeners, listenerId)
+			if len(listeners) == 0 {
+				delete(rcs.topicListeners, topic)
+			}
+			break
+		}
+	}
+}
+
+func (rcs *Service) getTopicListeners(topic string) []TopicListener {
+	rcs.mux.RLock()
+	defer rcs.mux.RUnlock()
+
+	listeners, ok := rcs.topicListeners[topic]
+	if !ok {
+		return nil
+	}
+
+	listenersCopy := make([]TopicListener, 0, len(listeners))
+	for _, l := range listeners {
+		listenersCopy = append(listenersCopy, l)
+	}
+	return listenersCopy
+}
+
+func (rcs *Service) AddConnectionStateListener(listener ConnectionStateListener) string {
+	id := model.NewId()
+
+	rcs.mux.Lock()
+	defer rcs.mux.Unlock()
+
+	rcs.connectionStateListeners[id] = listener
+	return id
+}
+
+func (rcs *Service) RemoveConnectionStateListener(listenerId string) {
+	rcs.mux.Lock()
+	defer rcs.mux.Unlock()
+	delete(rcs.connectionStateListeners, listenerId)
+}
+
+// onClusterLeaderChange is called whenever the cluster leader may have changed.
+func (rcs *Service) onClusterLeaderChange() {
+	if rcs.server.IsLeader() {
+		rcs.resume()
+	} else {
+		rcs.pause()
+	}
+}
+
+func (rcs *Service) resume() {
+	rcs.mux.Lock()
+	defer rcs.mux.Unlock()
+
+	if rcs.active {
+		return // already active
+	}
+	rcs.active = true
+	rcs.done = make(chan struct{})
+
+	if !disablePing {
+		rcs.pingLoop(rcs.done)
+	}
+
+	// create thread pool for concurrent message sending.
+	for i := range rcs.send {
+		go rcs.sendLoop(i, rcs.done)
+	}
+
+	rcs.server.Log().Debug("Remote Cluster Service active")
+}
+
+func (rcs *Service) pause() {
+	rcs.mux.Lock()
+	defer rcs.mux.Unlock()
+
+	if !rcs.active {
+		return // already inactive
+	}
+	rcs.active = false
+	close(rcs.done)
+	rcs.done = nil
+
+	rcs.server.Log().Debug("Remote Cluster Service inactive")
+}

server/platform/services/remotecluster/service_test.go

@@ -0,0 +1,73 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package remotecluster
+
+import (
+	"sync/atomic"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/mattermost/mattermost-server/v6/model"
+)
+
+func TestService_AddTopicListener(t *testing.T) {
+	var count int32
+
+	l1 := func(msg model.RemoteClusterMsg, rc *model.RemoteCluster, resp *Response) error {
+		atomic.AddInt32(&count, 1)
+		return nil
+	}
+	l2 := func(msg model.RemoteClusterMsg, rc *model.RemoteCluster, resp *Response) error {
+		atomic.AddInt32(&count, 1)
+		return nil
+	}
+	l3 := func(msg model.RemoteClusterMsg, rc *model.RemoteCluster, resp *Response) error {
+		atomic.AddInt32(&count, 1)
+		return nil
+	}
+
+	mockServer := newMockServer(makeRemoteClusters(NumRemotes, ""))
+	defer mockServer.Shutdown()
+
+	service, err := NewRemoteClusterService(mockServer)
+	require.NoError(t, err)
+
+	l1id := service.AddTopicListener("test", l1)
+	l2id := service.AddTopicListener("test", l2)
+	l3id := service.AddTopicListener("different", l3)
+
+	listeners := service.getTopicListeners("test")
+	assert.Len(t, listeners, 2)
+
+	rc := &model.RemoteCluster{}
+	msg1 := model.RemoteClusterMsg{Topic: "test"}
+	msg2 := model.RemoteClusterMsg{Topic: "different"}
+
+	service.ReceiveIncomingMsg(rc, msg1)
+	assert.Equal(t, int32(2), atomic.LoadInt32(&count))
+
+	service.ReceiveIncomingMsg(rc, msg2)
+	assert.Equal(t, int32(3), atomic.LoadInt32(&count))
+
+	service.RemoveTopicListener(l1id)
+	service.ReceiveIncomingMsg(rc, msg1)
+	assert.Equal(t, int32(4), atomic.LoadInt32(&count))
+
+	service.RemoveTopicListener(l2id)
+	service.ReceiveIncomingMsg(rc, msg1)
+	assert.Equal(t, int32(4), atomic.LoadInt32(&count))
+
+	service.ReceiveIncomingMsg(rc, msg2)
+	assert.Equal(t, int32(5), atomic.LoadInt32(&count))
+
+	service.RemoveTopicListener(l3id)
+	service.ReceiveIncomingMsg(rc, msg1)
+	service.ReceiveIncomingMsg(rc, msg2)
+	assert.Equal(t, int32(5), atomic.LoadInt32(&count))
+
+	listeners = service.getTopicListeners("test")
+	assert.Empty(t, listeners)
+}