MPA: move product hooks out of plugins environment (#21772)

* product: add new hooks manager for porducts

* move product hooks out of plugins environment

* add hooks for plugin

app/platform/service.go

@@ -95,7 +95,12 @@ type PlatformService struct {
 	additionalClusterHandlers map[model.ClusterEvent]einterfaces.ClusterMessageHandler
 	sharedChannelService      SharedChannelServiceIFace
 
-	pluginEnv *plugin.Environment
+	pluginEnv HookRunner
+}
+
+type HookRunner interface {
+	RunMultiHook(hookRunnerFunc func(hooks plugin.Hooks) bool, hookId int)
+	GetPluginsEnvironment() *plugin.Environment
 }
 
 // New creates a new PlatformService.
@@ -426,17 +431,17 @@ func (ps *PlatformService) SetSharedChannelService(s SharedChannelServiceIFace)
 	ps.sharedChannelService = s
 }
 
-func (ps *PlatformService) SetPluginsEnvironment(env *plugin.Environment) {
-	ps.pluginEnv = env
+func (ps *PlatformService) SetPluginsEnvironment(runner HookRunner) {
+	ps.pluginEnv = runner
 }
 
 // GetPluginStatuses meant to be used by cluster implementation
 func (ps *PlatformService) GetPluginStatuses() (model.PluginStatuses, *model.AppError) {
-	if ps.pluginEnv == nil {
+	if ps.pluginEnv == nil || ps.pluginEnv.GetPluginsEnvironment() == nil {
 		return nil, model.NewAppError("GetPluginStatuses", "app.plugin.disabled.app_error", nil, "", http.StatusNotImplemented)
 	}
 
-	pluginStatuses, err := ps.pluginEnv.Statuses()
+	pluginStatuses, err := ps.pluginEnv.GetPluginsEnvironment().Statuses()
 	if err != nil {
 		return nil, model.NewAppError("GetPluginStatuses", "app.plugin.get_statuses.app_error", nil, "", http.StatusInternalServerError).Wrap(err)
 	}

app/platform/web_conn.go

@@ -72,15 +72,15 @@ type WebConnConfig struct {
 // It contains all the necessary state to manage sending/receiving data to/from
 // a websocket.
 type WebConn struct {
-	sessionExpiresAt   int64 // This should stay at the top for 64-bit alignment of 64-bit words accessed atomically
-	Platform           *PlatformService
-	Suite              SuiteIFace
-	PluginsEnvironment func() *plugin.Environment
-	WebSocket          *websocket.Conn
-	T                  i18n.TranslateFunc
-	Locale             string
-	Sequence           int64
-	UserId             string
+	sessionExpiresAt int64 // This should stay at the top for 64-bit alignment of 64-bit words accessed atomically
+	Platform         *PlatformService
+	Suite            SuiteIFace
+	HookRunner       HookRunner
+	WebSocket        *websocket.Conn
+	T                i18n.TranslateFunc
+	Locale           string
+	Sequence         int64
+	UserId           string
 
 	allChannelMembers         map[string]string
 	lastAllChannelMembersTime int64
@@ -162,7 +162,7 @@ func (ps *PlatformService) PopulateWebConnConfig(s *model.Session, cfg *WebConnC
 }
 
 // NewWebConn returns a new WebConn instance.
-func (ps *PlatformService) NewWebConn(cfg *WebConnConfig, suite SuiteIFace, envFn func() *plugin.Environment) *WebConn {
+func (ps *PlatformService) NewWebConn(cfg *WebConnConfig, suite SuiteIFace, runner HookRunner) *WebConn {
 	if cfg.Session.UserId != "" {
 		ps.Go(func() {
 			suite.SetStatusOnline(cfg.Session.UserId, false)
@@ -200,7 +200,7 @@ func (ps *PlatformService) NewWebConn(cfg *WebConnConfig, suite SuiteIFace, envF
 	wc := &WebConn{
 		Platform:           ps,
 		Suite:              suite,
-		PluginsEnvironment: envFn,
+		HookRunner:         runner,
 		send:               cfg.activeQueue,
 		deadQueue:          cfg.deadQueue,
 		deadQueuePointer:   cfg.deadQueuePointer,
@@ -222,14 +222,12 @@ func (ps *PlatformService) NewWebConn(cfg *WebConnConfig, suite SuiteIFace, envF
 	wc.SetSessionExpiresAt(cfg.Session.ExpiresAt)
 	wc.SetConnectionID(cfg.ConnectionID)
 
-	if pluginsEnvironment := wc.PluginsEnvironment(); pluginsEnvironment != nil {
-		wc.Platform.Go(func() {
-			pluginsEnvironment.RunMultiPluginHook(func(hooks plugin.Hooks) bool {
-				hooks.OnWebSocketConnect(wc.GetConnectionID(), wc.UserId)
-				return true
-			}, plugin.OnWebSocketConnectID)
-		})
-	}
+	wc.Platform.Go(func() {
+		wc.HookRunner.RunMultiHook(func(hooks plugin.Hooks) bool {
+			hooks.OnWebSocketConnect(wc.GetConnectionID(), wc.UserId)
+			return true
+		}, plugin.OnWebSocketConnectID)
+	})
 
 	return wc
 }
@@ -238,12 +236,10 @@ func (wc *WebConn) pluginPostedConsumer(wg *sync.WaitGroup) {
 	defer wg.Done()
 
 	for msg := range wc.pluginPosted {
-		if pluginsEnvironment := wc.PluginsEnvironment(); pluginsEnvironment != nil {
-			pluginsEnvironment.RunMultiPluginHook(func(hooks plugin.Hooks) bool {
-				hooks.WebSocketMessageHasBeenPosted(msg.connectionID, msg.userID, msg.req)
-				return true
-			}, plugin.WebSocketMessageHasBeenPostedID)
-		}
+		wc.HookRunner.RunMultiHook(func(hooks plugin.Hooks) bool {
+			hooks.WebSocketMessageHasBeenPosted(msg.connectionID, msg.userID, msg.req)
+			return true
+		}, plugin.WebSocketMessageHasBeenPostedID)
 	}
 }
 
@@ -328,14 +324,12 @@ func (wc *WebConn) Pump() {
 	wc.Platform.HubUnregister(wc)
 	close(wc.pumpFinished)
 
-	if pluginsEnvironment := wc.PluginsEnvironment(); pluginsEnvironment != nil {
-		wc.Platform.Go(func() {
-			pluginsEnvironment.RunMultiPluginHook(func(hooks plugin.Hooks) bool {
-				hooks.OnWebSocketDisconnect(wc.GetConnectionID(), wc.UserId)
-				return true
-			}, plugin.OnWebSocketDisconnectID)
-		})
-	}
+	wc.Platform.Go(func() {
+		wc.HookRunner.RunMultiHook(func(hooks plugin.Hooks) bool {
+			hooks.OnWebSocketDisconnect(wc.GetConnectionID(), wc.UserId)
+			return true
+		}, plugin.OnWebSocketDisconnectID)
+	})
 }
 
 func (wc *WebConn) readPump() {

app/platform/web_conn_test.go

@@ -5,6 +5,7 @@ package platform
 
 import (
 	"bytes"
+	"errors"
 	"net"
 	"net/http"
 	"net/http/httptest"
@@ -18,13 +19,27 @@ import (
 	"github.com/mattermost/mattermost-server/v6/plugin"
 )
 
+type hookRunner struct {
+}
+
+func (h *hookRunner) RunMultiHook(hookRunnerFunc func(hooks plugin.Hooks) bool, hookId int) {
+
+}
+func (h *hookRunner) HooksForPlugin(id string) (plugin.Hooks, error) {
+	return nil, errors.New("not implemented")
+}
+
+func (h *hookRunner) GetPluginsEnvironment() *plugin.Environment {
+	return nil
+}
+
 func TestWebConnAddDeadQueue(t *testing.T) {
 	th := Setup(t)
 	defer th.TearDown()
 
 	wc := th.Service.NewWebConn(&WebConnConfig{
 		WebSocket: &websocket.Conn{},
-	}, th.Suite, func() *plugin.Environment { return nil })
+	}, th.Suite, &hookRunner{})
 
 	for i := 0; i < 2; i++ {
 		msg := &model.WebSocketEvent{}
@@ -53,7 +68,7 @@ func TestWebConnIsInDeadQueue(t *testing.T) {
 
 	wc := th.Service.NewWebConn(&WebConnConfig{
 		WebSocket: &websocket.Conn{},
-	}, th.Suite, func() *plugin.Environment { return nil })
+	}, th.Suite, &hookRunner{})
 
 	var i int
 	for ; i < 2; i++ {
@@ -114,7 +129,7 @@ func TestWebConnClearDeadQueue(t *testing.T) {
 
 	wc := th.Service.NewWebConn(&WebConnConfig{
 		WebSocket: &websocket.Conn{},
-	}, th.Suite, func() *plugin.Environment { return nil })
+	}, th.Suite, &hookRunner{})
 
 	var i int
 	for ; i < 2; i++ {
@@ -140,7 +155,7 @@ func TestWebConnDrainDeadQueue(t *testing.T) {
 		cfg := &WebConnConfig{
 			WebSocket: c,
 		}
-		return th.Service.NewWebConn(cfg, th.Suite, func() *plugin.Environment { return nil })
+		return th.Service.NewWebConn(cfg, th.Suite, &hookRunner{})
 	}
 
 	t.Run("Empty Queue", func(t *testing.T) {

app/platform/web_hub_test.go

@@ -17,7 +17,6 @@ import (
 
 	platform_mocks "github.com/mattermost/mattermost-server/v6/app/platform/mocks"
 	"github.com/mattermost/mattermost-server/v6/model"
-	"github.com/mattermost/mattermost-server/v6/plugin"
 	"github.com/mattermost/mattermost-server/v6/shared/i18n"
 	"github.com/mattermost/mattermost-server/v6/store/storetest/mocks"
 	"github.com/mattermost/mattermost-server/v6/testlib"
@@ -50,7 +49,7 @@ func registerDummyWebConn(t *testing.T, th *TestHelper, addr net.Addr, session *
 		TFunc:     i18n.IdentityTfunc(),
 		Locale:    "en",
 	}
-	wc := th.Service.NewWebConn(cfg, th.Suite, func() *plugin.Environment { return nil })
+	wc := th.Service.NewWebConn(cfg, th.Suite, &hookRunner{})
 	th.Service.HubRegister(wc)
 	go wc.Pump()
 	return wc