Remove app initialization from WebHub (#18687)

- Make it a Server method. - Pass Server instead of App. - Remove global app instance from NewServer, rather create local instances whenever needed. - Remove App from Websocket router, and create dynamically on every request. - Remove HubStart and HubStop from App methods. - Explicitly using s.Log instead of the global logger to indicate dependency on Server. We could have passed the logger explicitly but it doesn't look ideal. ```release-note NONE ```
2021-10-18 12:36:13 +05:30
--- a/app/app_iface.go
+++ b/app/app_iface.go
@@ -212,8 +212,6 @@ type AppIface interface {
 	HasRemote(channelID string, remoteID string) (bool, error)
 	// HubRegister registers a connection to a hub.
 	HubRegister(webConn *WebConn)
-	// HubStart starts all the hubs.
-	HubStart()
 	// HubUnregister unregisters a connection from a hub.
 	HubUnregister(webConn *WebConn)
 	// InstallMarketplacePlugin installs a plugin listed in the marketplace server. It will get the plugin bundle
@@ -244,8 +242,6 @@ type AppIface interface {
 	MoveChannel(c *request.Context, team *model.Team, channel *model.Channel, user *model.User) *model.AppError
 	// NewWebConn returns a new WebConn instance.
 	NewWebConn(cfg *WebConnConfig) *WebConn
-	// NewWebHub creates a new Hub.
-	NewWebHub() *Hub
 	// NotifySessionsExpired is called periodically from the job server to notify any mobile sessions that have expired.
 	NotifySessionsExpired() *model.AppError
 	// OverrideIconURLIfEmoji changes the post icon override URL prop, if it has an emoji icon,
@@ -805,7 +801,6 @@ type AppIface interface {
 	HasPermissionToTeam(askingUserId string, teamID string, permission *model.Permission) bool
 	HasPermissionToUser(askingUserId string, userID string) bool
 	HasSharedChannel(channelID string) (bool, error)
-	HubStop()
 	ImageProxy() *imageproxy.ImageProxy
 	ImageProxyAdder() func(string) string
 	ImageProxyRemover() (f func(string) string)
--- a/app/opentracing/opentracing_layer.go
+++ b/app/opentracing/opentracing_layer.go
@@ -10449,36 +10449,6 @@ func (a *OpenTracingAppLayer) HubRegister(webConn *app.WebConn) {
 	a.app.HubRegister(webConn)
 }

-func (a *OpenTracingAppLayer) HubStart() {
-	origCtx := a.ctx
-	span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.HubStart")
-
-	a.ctx = newCtx
-	a.app.Srv().Store.SetContext(newCtx)
-	defer func() {
-		a.app.Srv().Store.SetContext(origCtx)
-		a.ctx = origCtx
-	}()
-
-	defer span.Finish()
-	a.app.HubStart()
-}
-
-func (a *OpenTracingAppLayer) HubStop() {
-	origCtx := a.ctx
-	span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.HubStop")
-
-	a.ctx = newCtx
-	a.app.Srv().Store.SetContext(newCtx)
-	defer func() {
-		a.app.Srv().Store.SetContext(origCtx)
-		a.ctx = origCtx
-	}()
-
-	defer span.Finish()
-	a.app.HubStop()
-}
-
 func (a *OpenTracingAppLayer) HubUnregister(webConn *app.WebConn) {
 	origCtx := a.ctx
 	span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.HubUnregister")
@@ -11530,23 +11500,6 @@ func (a *OpenTracingAppLayer) NewWebConn(cfg *app.WebConnConfig) *app.WebConn {
 	return resultVar0
 }

-func (a *OpenTracingAppLayer) NewWebHub() *app.Hub {
-	origCtx := a.ctx
-	span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.NewWebHub")
-
-	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.NewWebHub()
-
-	return resultVar0
-}
-
 func (a *OpenTracingAppLayer) NotifyAndSetWarnMetricAck(warnMetricId string, sender *model.User, forceAck bool, isBot bool) *model.AppError {
 	origCtx := a.ctx
 	span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.NotifyAndSetWarnMetricAck")
--- a/app/server.go
+++ b/app/server.go
@@ -280,8 +280,7 @@ func NewServer(options ...Option) (*Server, error) {

 	// It is important to initialize the hub only after the global logger is set
 	// to avoid race conditions while logging from inside the hub.
-	app := New(ServerConnector(s.Channels()))
-	app.HubStart()
+	s.HubStart()

 	if *s.Config().LogSettings.EnableDiagnostics && *s.Config().LogSettings.EnableSentry {
 		if strings.Contains(SentryDSN, "placeholder") {
@@ -543,7 +542,6 @@ func NewServer(options ...Option) (*Server, error) {

 	s.WebSocketRouter = &WebSocketRouter{
 		handlers: make(map[string]webSocketHandler),
-		app:      app,
 	}

 	mailConfig := s.MailServiceConfig()
@@ -666,7 +664,8 @@ func NewServer(options ...Option) (*Server, error) {
 	// if enabled - perform initial product notices fetch
 	if *s.Config().AnnouncementSettings.AdminNoticesEnabled || *s.Config().AnnouncementSettings.UserNoticesEnabled {
 		go func() {
-			if err := app.UpdateProductNotices(); err != nil {
+			appInstance := New(ServerConnector(s.Channels()))
+			if err := appInstance.UpdateProductNotices(); err != nil {
 				mlog.Warn("Failied to perform initial product notices fetch", mlog.Err(err))
 			}
 		}()
@@ -678,8 +677,9 @@ func NewServer(options ...Option) (*Server, error) {

 	c := request.EmptyContext()
 	s.AddConfigListener(func(oldConfig *model.Config, newConfig *model.Config) {
+		appInstance := New(ServerConnector(s.Channels()))
 		if *oldConfig.GuestAccountsSettings.Enable && !*newConfig.GuestAccountsSettings.Enable {
-			if appErr := app.DeactivateGuests(c); appErr != nil {
+			if appErr := appInstance.DeactivateGuests(c); appErr != nil {
 				mlog.Error("Unable to deactivate guest accounts", mlog.Err(appErr))
 			}
 		}
@@ -687,16 +687,18 @@ func NewServer(options ...Option) (*Server, error) {

 	// Disable active guest accounts on first run if guest accounts are disabled
 	if !*s.Config().GuestAccountsSettings.Enable {
-		if appErr := app.DeactivateGuests(c); appErr != nil {
+		appInstance := New(ServerConnector(s.Channels()))
+		if appErr := appInstance.DeactivateGuests(c); appErr != nil {
 			mlog.Error("Unable to deactivate guest accounts", mlog.Err(appErr))
 		}
 	}

 	if s.runEssentialJobs {
 		s.Go(func() {
+			appInstance := New(ServerConnector(s.Channels()))
 			s.runLicenseExpirationCheckJob()
-			runCheckAdminSupportStatusJob(app, c)
-			runDNDStatusExpireJob(app)
+			runCheckAdminSupportStatusJob(appInstance, c)
+			runDNDStatusExpireJob(appInstance)
 		})
 		s.runJobs()
 	}
@@ -907,7 +909,7 @@ func (s *Server) removeUnlicensedLogTargets(license *model.License) {
 	})
 }

-func (s *Server) startInterClusterServices(license *model.License, app *App) error {
+func (s *Server) startInterClusterServices(license *model.License) error {
 	if license == nil {
 		mlog.Debug("No license provided; Remote Cluster services disabled")
 		return nil
@@ -956,7 +958,8 @@ func (s *Server) startInterClusterServices(license *model.License, app *App) err
 		return nil
 	}

-	scs, err := sharedchannel.NewSharedChannelService(s, app)
+	appInstance := New(ServerConnector(s.Channels()))
+	scs, err := sharedchannel.NewSharedChannelService(s, appInstance)
 	if err != nil {
 		return err
 	}
@@ -1417,7 +1420,7 @@ func (s *Server) Start() error {
 		}
 	}

-	if err := s.startInterClusterServices(s.License(), s.WebSocketRouter.app); err != nil {
+	if err := s.startInterClusterServices(s.License()); err != nil {
 		mlog.Error("Error starting inter-cluster services", mlog.Err(err))
 	}

--- a/app/web_hub.go
+++ b/app/web_hub.go
@@ -50,7 +50,7 @@ type Hub struct {
 	// connectionCount should be kept first.
 	// See https://github.com/mattermost/mattermost-server/pull/7281
 	connectionCount int64
-	app             *App
+	srv             *Server
 	connectionIndex int
 	register        chan *WebConn
 	unregister      chan *WebConn
@@ -65,10 +65,10 @@ type Hub struct {
 	checkConn       chan *webConnCheckMessage
 }

-// NewWebHub creates a new Hub.
-func (a *App) NewWebHub() *Hub {
+// newWebHub creates a new Hub.
+func newWebHub(s *Server) *Hub {
 	return &Hub{
-		app:             a,
+		srv:             s,
 		register:        make(chan *WebConn),
 		unregister:      make(chan *WebConn),
 		broadcast:       make(chan *model.WebSocketEvent, broadcastQueueSize),
@@ -87,21 +87,21 @@ func (a *App) TotalWebsocketConnections() int {
 }

 // HubStart starts all the hubs.
-func (a *App) HubStart() {
+func (s *Server) HubStart() {
 	// Total number of hubs is twice the number of CPUs.
 	numberOfHubs := runtime.NumCPU() * 2
-	mlog.Info("Starting websocket hubs", mlog.Int("number_of_hubs", numberOfHubs))
+	s.Log.Info("Starting websocket hubs", mlog.Int("number_of_hubs", numberOfHubs))

 	hubs := make([]*Hub, numberOfHubs)

 	for i := 0; i < numberOfHubs; i++ {
-		hubs[i] = a.NewWebHub()
+		hubs[i] = newWebHub(s)
 		hubs[i].connectionIndex = i
 		hubs[i].Start()
 	}
 	// Assigning to the hubs slice without any mutex is fine because it is only assigned once
 	// during the start of the program and always read from after that.
-	a.ch.srv.hubs = hubs
+	s.hubs = hubs
 }

 func (a *App) invalidateCacheForWebhook(webhookID string) {
@@ -117,10 +117,6 @@ func (s *Server) HubStop() {
 	}
 }

-func (a *App) HubStop() {
-	a.Srv().HubStop()
-}
-
 // GetHubForUserId returns the hub for a given user id.
 func (s *Server) GetHubForUserId(userID string) *Hub {
 	// TODO: check if caching the userID -> hub mapping
@@ -356,7 +352,7 @@ func (h *Hub) Broadcast(message *model.WebSocketEvent) {
 	// And possibly, we can look into doing the hub initialization inside
 	// NewServer itself.
 	if h != nil && message != nil {
-		if metrics := h.app.Metrics(); metrics != nil {
+		if metrics := h.srv.Metrics; metrics != nil {
 			metrics.IncrementWebSocketBroadcastBufferSize(strconv.Itoa(h.connectionIndex), 1)
 		}
 		select {
@@ -416,6 +412,8 @@ func (h *Hub) Start() {
 		ticker := time.NewTicker(inactiveConnReaperInterval)
 		defer ticker.Stop()

+		appInstance := New(ServerConnector(h.srv.Channels()))
+
 		connIndex := newHubConnectionIndex(inactiveConnReaperInterval)

 		for {
@@ -449,7 +447,7 @@ func (h *Hub) Start() {
 				connIndex.RemoveInactiveConnections()
 			case webConn := <-h.register:
 				var oldConn *WebConn
-				if *h.app.Config().ServiceSettings.EnableReliableWebSockets {
+				if *h.srv.Config().ServiceSettings.EnableReliableWebSockets {
 					// Delete the old conn from connIndex if it exists.
 					oldConn = connIndex.RemoveInactiveByConnectionID(
 						webConn.GetSession().UserId,
@@ -474,7 +472,7 @@ func (h *Hub) Start() {
 			case webConn := <-h.unregister:
 				// If already removed (via queue full), then removing again becomes a noop.
 				// But if not removed, mark inactive.
-				if *h.app.Config().ServiceSettings.EnableReliableWebSockets {
+				if *h.srv.Config().ServiceSettings.EnableReliableWebSockets {
 					webConn.active = false
 				} else {
 					connIndex.Remove(webConn)
@@ -488,8 +486,8 @@ func (h *Hub) Start() {

 				conns := connIndex.ForUser(webConn.UserId)
 				if len(conns) == 0 || areAllInactive(conns) {
-					h.app.Srv().Go(func() {
-						h.app.SetStatusOffline(webConn.UserId, false)
+					h.srv.Go(func() {
+						appInstance.SetStatusOffline(webConn.UserId, false)
 					})
 					continue
 				}
@@ -503,9 +501,9 @@ func (h *Hub) Start() {
 					}
 				}

-				if h.app.IsUserAway(latestActivity) {
-					h.app.Srv().Go(func() {
-						h.app.SetStatusLastActivityAt(webConn.UserId, latestActivity)
+				if appInstance.IsUserAway(latestActivity) {
+					h.srv.Go(func() {
+						appInstance.SetStatusLastActivityAt(webConn.UserId, latestActivity)
 					})
 				}
 			case userID := <-h.invalidateUser:
@@ -533,7 +531,7 @@ func (h *Hub) Start() {
 					connIndex.Remove(directMsg.conn)
 				}
 			case msg := <-h.broadcast:
-				if metrics := h.app.Metrics(); metrics != nil {
+				if metrics := h.srv.Metrics; metrics != nil {
 					metrics.DecrementWebSocketBroadcastBufferSize(strconv.Itoa(h.connectionIndex), 1)
 				}
 				msg = msg.PrecomputeJSON()
@@ -565,7 +563,7 @@ func (h *Hub) Start() {
 			case <-h.stop:
 				for webConn := range connIndex.All() {
 					webConn.Close()
-					h.app.SetStatusOffline(webConn.UserId, false)
+					appInstance.SetStatusOffline(webConn.UserId, false)
 				}

 				h.explicitStop = true
--- a/app/web_hub_test.go
+++ b/app/web_hub_test.go
@@ -66,7 +66,7 @@ func TestHubStopWithMultipleConnections(t *testing.T) {
 	s := httptest.NewServer(dummyWebsocketHandler(t))
 	defer s.Close()

-	th.App.HubStart()
+	th.Server.HubStart()
 	wc1 := registerDummyWebConn(t, th.App, s.Listener.Addr(), th.BasicUser.Id)
 	wc2 := registerDummyWebConn(t, th.App, s.Listener.Addr(), th.BasicUser.Id)
 	wc3 := registerDummyWebConn(t, th.App, s.Listener.Addr(), th.BasicUser.Id)
@@ -84,12 +84,12 @@ func TestHubStopRaceCondition(t *testing.T) {
 	// So we just use this quick hack for the test.
 	s := httptest.NewServer(dummyWebsocketHandler(t))

-	th.App.HubStart()
+	th.Server.HubStart()
 	wc1 := registerDummyWebConn(t, th.App, s.Listener.Addr(), th.BasicUser.Id)
 	defer wc1.Close()

 	hub := th.App.Srv().hubs[0]
-	th.App.HubStop()
+	th.Server.HubStop()

 	done := make(chan bool)
 	go func() {
@@ -347,7 +347,7 @@ func TestHubIsRegistered(t *testing.T) {
 	s := httptest.NewServer(dummyWebsocketHandler(t))
 	defer s.Close()

-	th.App.HubStart()
+	th.Server.HubStart()
 	wc1 := registerDummyWebConn(t, th.App, s.Listener.Addr(), th.BasicUser.Id)
 	wc2 := registerDummyWebConn(t, th.App, s.Listener.Addr(), th.BasicUser.Id)
 	wc3 := registerDummyWebConn(t, th.App, s.Listener.Addr(), th.BasicUser.Id)
@@ -418,7 +418,7 @@ func BenchmarkGetHubForUserId(b *testing.B) {
 	th := Setup(b).InitBasic()
 	defer th.TearDown()

-	th.App.HubStart()
+	th.Server.HubStart()

 	b.ResetTimer()
 	for i := 0; i < b.N; i++ {
--- a/app/webhub_fuzz.go
+++ b/app/webhub_fuzz.go
@@ -153,7 +153,7 @@ func Fuzz(data []byte) int {

 		s := httptest.NewServer(dummyWebsocketHandler())

-		th.App.HubStart()
+		th.Server.HubStart()

 		u1 := th.CreateUser()
 		u2 := th.CreateUser()
--- a/app/websocket_router.go
+++ b/app/websocket_router.go
@@ -16,7 +16,6 @@ type webSocketHandler interface {
 }

 type WebSocketRouter struct {
-	app      *App
 	handlers map[string]webSocketHandler
 }

@@ -27,13 +26,13 @@ func (wr *WebSocketRouter) Handle(action string, handler webSocketHandler) {
 func (wr *WebSocketRouter) ServeWebSocket(conn *WebConn, r *model.WebSocketRequest) {
 	if r.Action == "" {
 		err := model.NewAppError("ServeWebSocket", "api.web_socket_router.no_action.app_error", nil, "", http.StatusBadRequest)
-		returnWebSocketError(wr.app, conn, r, err)
+		returnWebSocketError(conn.App, conn, r, err)
 		return
 	}

 	if r.Seq <= 0 {
 		err := model.NewAppError("ServeWebSocket", "api.web_socket_router.bad_seq.app_error", nil, "", http.StatusBadRequest)
-		returnWebSocketError(wr.app, conn, r, err)
+		returnWebSocketError(conn.App, conn, r, err)
 		return
 	}

@@ -48,7 +47,7 @@ func (wr *WebSocketRouter) ServeWebSocket(conn *WebConn, r *model.WebSocketReque
 			return
 		}

-		session, err := wr.app.GetSession(token)
+		session, err := conn.App.GetSession(token)
 		if err != nil {
 			conn.WebSocket.Close()
 			return
@@ -59,15 +58,15 @@ func (wr *WebSocketRouter) ServeWebSocket(conn *WebConn, r *model.WebSocketReque

 		// TODO: Same logic to reconnect queue as api4/websocket.go

-		wr.app.HubRegister(conn)
+		conn.App.HubRegister(conn)

-		wr.app.Srv().Go(func() {
-			wr.app.SetStatusOnline(session.UserId, false)
-			wr.app.UpdateLastActivityAtIfNeeded(*session)
+		conn.App.Srv().Go(func() {
+			conn.App.SetStatusOnline(session.UserId, false)
+			conn.App.UpdateLastActivityAtIfNeeded(*session)
 		})

 		resp := model.NewWebSocketResponse(model.StatusOk, r.Seq, nil)
-		hub := wr.app.GetHubForUserId(conn.UserId)
+		hub := conn.App.GetHubForUserId(conn.UserId)
 		if hub == nil {
 			return
 		}
@@ -78,14 +77,14 @@ func (wr *WebSocketRouter) ServeWebSocket(conn *WebConn, r *model.WebSocketReque

 	if !conn.IsAuthenticated() {
 		err := model.NewAppError("ServeWebSocket", "api.web_socket_router.not_authenticated.app_error", nil, "", http.StatusUnauthorized)
-		returnWebSocketError(wr.app, conn, r, err)
+		returnWebSocketError(conn.App, conn, r, err)
 		return
 	}

 	handler, ok := wr.handlers[r.Action]
 	if !ok {
 		err := model.NewAppError("ServeWebSocket", "api.web_socket_router.bad_action.app_error", nil, "", http.StatusInternalServerError)
-		returnWebSocketError(wr.app, conn, r, err)
+		returnWebSocketError(conn.App, conn, r, err)
 		return
 	}