Minimize app dependency on platform service (#21548)

app/platform/helper_test.go

@@ -181,7 +181,7 @@ func setupTestHelper(dbStore store.Store, enterprise bool, includeCacheLayer boo
 		th.Service.SetLicense(nil)
 	}
 
-	err = th.Service.Start(th.Suite)
+	err = th.Service.Start()
 	if err != nil {
 		panic(err)
 	}

app/platform/mocks/SuiteIFace.go

@@ -39,20 +39,6 @@ func (_m *SuiteIFace) GetSession(token string) (*model.Session, *model.AppError)
 	return r0, r1
 }
 
-// IsUserAway provides a mock function with given fields: lastActivityAt
-func (_m *SuiteIFace) IsUserAway(lastActivityAt int64) bool {
-	ret := _m.Called(lastActivityAt)
-
-	var r0 bool
-	if rf, ok := ret.Get(0).(func(int64) bool); ok {
-		r0 = rf(lastActivityAt)
-	} else {
-		r0 = ret.Get(0).(bool)
-	}
-
-	return r0
-}
-
 // RolesGrantPermission provides a mock function with given fields: roleNames, permissionId
 func (_m *SuiteIFace) RolesGrantPermission(roleNames []string, permissionId string) bool {
 	ret := _m.Called(roleNames, permissionId)
@@ -67,31 +53,6 @@ func (_m *SuiteIFace) RolesGrantPermission(roleNames []string, permissionId stri
 	return r0
 }
 
-// SetStatusAwayIfNeeded provides a mock function with given fields: userID, manual
-func (_m *SuiteIFace) SetStatusAwayIfNeeded(userID string, manual bool) {
-	_m.Called(userID, manual)
-}
-
-// SetStatusLastActivityAt provides a mock function with given fields: userID, activityAt
-func (_m *SuiteIFace) SetStatusLastActivityAt(userID string, activityAt int64) {
-	_m.Called(userID, activityAt)
-}
-
-// SetStatusOffline provides a mock function with given fields: userID, manual
-func (_m *SuiteIFace) SetStatusOffline(userID string, manual bool) {
-	_m.Called(userID, manual)
-}
-
-// SetStatusOnline provides a mock function with given fields: userID, manual
-func (_m *SuiteIFace) SetStatusOnline(userID string, manual bool) {
-	_m.Called(userID, manual)
-}
-
-// UpdateLastActivityAtIfNeeded provides a mock function with given fields: session
-func (_m *SuiteIFace) UpdateLastActivityAtIfNeeded(session model.Session) {
-	_m.Called(session)
-}
-
 // UserCanSeeOtherUser provides a mock function with given fields: userID, otherUserId
 func (_m *SuiteIFace) UserCanSeeOtherUser(userID string, otherUserId string) (bool, *model.AppError) {
 	ret := _m.Called(userID, otherUserId)

app/platform/service.go

@@ -304,8 +304,8 @@ func New(sc ServiceConfig, options ...Option) (*PlatformService, error) {
 	return ps, nil
 }
 
-func (ps *PlatformService) Start(suite SuiteIFace) error {
-	ps.hubStart(suite)
+func (ps *PlatformService) Start() error {
+	ps.hubStart()
 
 	ps.configListenerId = ps.AddConfigListener(func(_, _ *model.Config) {
 		ps.regenerateClientConfig()

app/platform/status.go

@@ -212,3 +212,198 @@ func (ps *PlatformService) GetStatus(userID string) (*model.Status, *model.AppEr
 
 	return status, nil
 }
+
+// SetStatusLastActivityAt sets the last activity at for a user on the local app server and updates
+// status to away if needed. Used by the WS to set status to away if an 'online' device disconnects
+// while an 'away' device is still connected
+func (ps *PlatformService) SetStatusLastActivityAt(userID string, activityAt int64) {
+	var status *model.Status
+	var err *model.AppError
+	if status, err = ps.GetStatus(userID); err != nil {
+		return
+	}
+
+	status.LastActivityAt = activityAt
+
+	ps.AddStatusCacheSkipClusterSend(status)
+	ps.SetStatusAwayIfNeeded(userID, false)
+}
+
+func (ps *PlatformService) UpdateLastActivityAtIfNeeded(session model.Session) {
+	now := model.GetMillis()
+
+	ps.UpdateWebConnUserActivity(session, now)
+
+	if now-session.LastActivityAt < model.SessionActivityTimeout {
+		return
+	}
+
+	if err := ps.Store.Session().UpdateLastActivityAt(session.Id, now); err != nil {
+		mlog.Warn("Failed to update LastActivityAt", mlog.String("user_id", session.UserId), mlog.String("session_id", session.Id), mlog.Err(err))
+	}
+
+	session.LastActivityAt = now
+	ps.AddSessionToCache(&session)
+}
+
+func (ps *PlatformService) SetStatusOnline(userID string, manual bool) {
+	if !*ps.Config().ServiceSettings.EnableUserStatuses {
+		return
+	}
+
+	broadcast := false
+
+	var oldStatus string = model.StatusOffline
+	var oldTime int64
+	var oldManual bool
+	var status *model.Status
+	var err *model.AppError
+
+	if status, err = ps.GetStatus(userID); err != nil {
+		status = &model.Status{UserId: userID, Status: model.StatusOnline, Manual: false, LastActivityAt: model.GetMillis(), ActiveChannel: ""}
+		broadcast = true
+	} else {
+		if status.Manual && !manual {
+			return // manually set status always overrides non-manual one
+		}
+
+		if status.Status != model.StatusOnline {
+			broadcast = true
+		}
+
+		oldStatus = status.Status
+		oldTime = status.LastActivityAt
+		oldManual = status.Manual
+
+		status.Status = model.StatusOnline
+		status.Manual = false // for "online" there's no manual setting
+		status.LastActivityAt = model.GetMillis()
+	}
+
+	ps.AddStatusCache(status)
+
+	// Only update the database if the status has changed, the status has been manually set,
+	// or enough time has passed since the previous action
+	if status.Status != oldStatus || status.Manual != oldManual || status.LastActivityAt-oldTime > model.StatusMinUpdateTime {
+		if broadcast {
+			if err := ps.Store.Status().SaveOrUpdate(status); err != nil {
+				mlog.Warn("Failed to save status", mlog.String("user_id", userID), mlog.Err(err), mlog.String("user_id", userID))
+			}
+		} else {
+			if err := ps.Store.Status().UpdateLastActivityAt(status.UserId, status.LastActivityAt); err != nil {
+				mlog.Error("Failed to save status", mlog.String("user_id", userID), mlog.Err(err), mlog.String("user_id", userID))
+			}
+		}
+	}
+
+	if broadcast {
+		ps.BroadcastStatus(status)
+	}
+}
+
+func (ps *PlatformService) SetStatusOffline(userID string, manual bool) {
+	if !*ps.Config().ServiceSettings.EnableUserStatuses {
+		return
+	}
+
+	status, err := ps.GetStatus(userID)
+	if err == nil && status.Manual && !manual {
+		return // manually set status always overrides non-manual one
+	}
+
+	status = &model.Status{UserId: userID, Status: model.StatusOffline, Manual: manual, LastActivityAt: model.GetMillis(), ActiveChannel: ""}
+
+	ps.SaveAndBroadcastStatus(status)
+}
+
+func (ps *PlatformService) SetStatusAwayIfNeeded(userID string, manual bool) {
+	if !*ps.Config().ServiceSettings.EnableUserStatuses {
+		return
+	}
+
+	status, err := ps.GetStatus(userID)
+
+	if err != nil {
+		status = &model.Status{UserId: userID, Status: model.StatusOffline, Manual: manual, LastActivityAt: 0, ActiveChannel: ""}
+	}
+
+	if !manual && status.Manual {
+		return // manually set status always overrides non-manual one
+	}
+
+	if !manual {
+		if status.Status == model.StatusAway {
+			return
+		}
+
+		if !ps.isUserAway(status.LastActivityAt) {
+			return
+		}
+	}
+
+	status.Status = model.StatusAway
+	status.Manual = manual
+	status.ActiveChannel = ""
+
+	ps.SaveAndBroadcastStatus(status)
+}
+
+// SetStatusDoNotDisturbTimed takes endtime in unix epoch format in UTC
+// and sets status of given userId to dnd which will be restored back after endtime
+func (ps *PlatformService) SetStatusDoNotDisturbTimed(userId string, endtime int64) {
+	if !*ps.Config().ServiceSettings.EnableUserStatuses {
+		return
+	}
+
+	status, err := ps.GetStatus(userId)
+
+	if err != nil {
+		status = &model.Status{UserId: userId, Status: model.StatusOffline, Manual: false, LastActivityAt: 0, ActiveChannel: ""}
+	}
+
+	status.PrevStatus = status.Status
+	status.Status = model.StatusDnd
+	status.Manual = true
+
+	status.DNDEndTime = endtime
+
+	ps.SaveAndBroadcastStatus(status)
+}
+
+func (ps *PlatformService) SetStatusDoNotDisturb(userID string) {
+	if !*ps.Config().ServiceSettings.EnableUserStatuses {
+		return
+	}
+
+	status, err := ps.GetStatus(userID)
+
+	if err != nil {
+		status = &model.Status{UserId: userID, Status: model.StatusOffline, Manual: false, LastActivityAt: 0, ActiveChannel: ""}
+	}
+
+	status.Status = model.StatusDnd
+	status.Manual = true
+
+	ps.SaveAndBroadcastStatus(status)
+}
+
+func (ps *PlatformService) SetStatusOutOfOffice(userID string) {
+	if !*ps.Config().ServiceSettings.EnableUserStatuses {
+		return
+	}
+
+	status, err := ps.GetStatus(userID)
+
+	if err != nil {
+		status = &model.Status{UserId: userID, Status: model.StatusOutOfOffice, Manual: false, LastActivityAt: 0, ActiveChannel: ""}
+	}
+
+	status.Status = model.StatusOutOfOffice
+	status.Manual = true
+
+	ps.SaveAndBroadcastStatus(status)
+}
+
+func (ps *PlatformService) isUserAway(lastActivityAt int64) bool {
+	return model.GetMillis()-lastActivityAt >= *ps.Config().TeamSettings.UserStatusAwayTimeout*1000
+}

app/platform/web_conn.go

@@ -165,8 +165,8 @@ func (ps *PlatformService) PopulateWebConnConfig(s *model.Session, cfg *WebConnC
 func (ps *PlatformService) NewWebConn(cfg *WebConnConfig, suite SuiteIFace, runner HookRunner) *WebConn {
 	if cfg.Session.UserId != "" {
 		ps.Go(func() {
-			suite.SetStatusOnline(cfg.Session.UserId, false)
-			suite.UpdateLastActivityAtIfNeeded(cfg.Session)
+			ps.SetStatusOnline(cfg.Session.UserId, false)
+			ps.UpdateLastActivityAtIfNeeded(cfg.Session)
 		})
 	}
 
@@ -344,7 +344,7 @@ func (wc *WebConn) readPump() {
 		}
 		if wc.IsAuthenticated() {
 			wc.Platform.Go(func() {
-				wc.Suite.SetStatusAwayIfNeeded(wc.UserId, false)
+				wc.Platform.SetStatusAwayIfNeeded(wc.UserId, false)
 			})
 		}
 		return nil

app/platform/web_hub.go

@@ -21,12 +21,6 @@ const (
 )
 
 type SuiteIFace interface {
-	SetStatusLastActivityAt(userID string, activityAt int64)
-	SetStatusOffline(userID string, manual bool)
-	IsUserAway(lastActivityAt int64) bool
-	SetStatusOnline(userID string, manual bool)
-	UpdateLastActivityAtIfNeeded(session model.Session)
-	SetStatusAwayIfNeeded(userID string, manual bool)
 	GetSession(token string) (*model.Session, *model.AppError)
 	RolesGrantPermission(roleNames []string, permissionId string) bool
 	UserCanSeeOtherUser(userID string, otherUserId string) (bool, *model.AppError)
@@ -95,7 +89,7 @@ func newWebHub(ps *PlatformService) *Hub {
 }
 
 // hubStart starts all the hubs.
-func (ps *PlatformService) hubStart(suite SuiteIFace) {
+func (ps *PlatformService) hubStart() {
 	// Total number of hubs is twice the number of CPUs.
 	numberOfHubs := runtime.NumCPU() * 2
 	ps.logger.Info("Starting websocket hubs", mlog.Int("number_of_hubs", numberOfHubs))
@@ -105,7 +99,7 @@ func (ps *PlatformService) hubStart(suite SuiteIFace) {
 	for i := 0; i < numberOfHubs; i++ {
 		hubs[i] = newWebHub(ps)
 		hubs[i].connectionIndex = i
-		hubs[i].Start(suite)
+		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.
@@ -366,7 +360,7 @@ func (h *Hub) Stop() {
 }
 
 // Start starts the hub.
-func (h *Hub) Start(suite SuiteIFace) {
+func (h *Hub) Start() {
 	var doStart func()
 	var doRecoverableStart func()
 	var doRecover func()
@@ -439,7 +433,7 @@ func (h *Hub) Start(suite SuiteIFace) {
 				conns := connIndex.ForUser(webConn.UserId)
 				if len(conns) == 0 || areAllInactive(conns) {
 					h.platform.Go(func() {
-						suite.SetStatusOffline(webConn.UserId, false)
+						h.platform.SetStatusOffline(webConn.UserId, false)
 					})
 					continue
 				}
@@ -453,9 +447,9 @@ func (h *Hub) Start(suite SuiteIFace) {
 					}
 				}
 
-				if suite.IsUserAway(latestActivity) {
+				if h.platform.isUserAway(latestActivity) {
 					h.platform.Go(func() {
-						suite.SetStatusLastActivityAt(webConn.UserId, latestActivity)
+						h.platform.SetStatusLastActivityAt(webConn.UserId, latestActivity)
 					})
 				}
 			case userID := <-h.invalidateUser:
@@ -522,7 +516,7 @@ func (h *Hub) Start(suite SuiteIFace) {
 			case <-h.stop:
 				for webConn := range connIndex.All() {
 					webConn.Close()
-					suite.SetStatusOffline(webConn.UserId, false)
+					h.platform.SetStatusOffline(webConn.UserId, false)
 				}
 
 				h.explicitStop = true

app/platform/web_hub_test.go

@@ -67,7 +67,7 @@ func TestHubStopWithMultipleConnections(t *testing.T) {
 	})
 	require.NoError(t, err)
 
-	th.Service.Start(th.Suite)
+	th.Service.Start()
 	wc1 := registerDummyWebConn(t, th, s.Listener.Addr(), session)
 	wc2 := registerDummyWebConn(t, th, s.Listener.Addr(), session)
 	wc3 := registerDummyWebConn(t, th, s.Listener.Addr(), session)
@@ -90,7 +90,7 @@ func TestHubStopRaceCondition(t *testing.T) {
 	})
 	require.NoError(t, err)
 
-	th.Service.Start(th.Suite)
+	th.Service.Start()
 	wc1 := registerDummyWebConn(t, th, s.Listener.Addr(), session)
 	defer wc1.Close()
 
@@ -467,18 +467,13 @@ func TestHubIsRegistered(t *testing.T) {
 	require.NoError(t, err)
 
 	mockSuite := &platform_mocks.SuiteIFace{}
-	mockSuite.On("SetStatusOnline", th.BasicUser.Id, false).Return()
-	mockSuite.On("UpdateLastActivityAtIfNeeded", *session).Return()
 	mockSuite.On("GetSession", session.Token).Return(session, nil)
-	mockSuite.On("IsUserAway", mock.Anything).Return(false)
-	mockSuite.On("SetStatusOffline", th.BasicUser.Id, false).Return()
-
 	th.Suite = mockSuite
 
 	s := httptest.NewServer(dummyWebsocketHandler(t))
 	defer s.Close()
 
-	th.Service.Start(th.Suite)
+	th.Service.Start()
 	wc1 := registerDummyWebConn(t, th, s.Listener.Addr(), session)
 	wc2 := registerDummyWebConn(t, th, s.Listener.Addr(), session)
 	wc3 := registerDummyWebConn(t, th, s.Listener.Addr(), session)
@@ -486,9 +481,7 @@ func TestHubIsRegistered(t *testing.T) {
 	defer wc2.Close()
 	defer wc3.Close()
 
-	session1 := wc1.session.Load().(*model.Session)
-
-	assert.True(t, th.Service.SessionIsRegistered(*session1))
+	assert.True(t, th.Service.SessionIsRegistered(*wc1.session.Load().(*model.Session)))
 	assert.True(t, th.Service.SessionIsRegistered(*wc2.session.Load().(*model.Session)))
 	assert.True(t, th.Service.SessionIsRegistered(*wc3.session.Load().(*model.Session)))
 
@@ -551,7 +544,7 @@ func BenchmarkGetHubForUserId(b *testing.B) {
 	th := Setup(b).InitBasic()
 	defer th.TearDown()
 
-	th.Service.Start(th.Suite)
+	th.Service.Start()
 
 	b.ResetTimer()
 	for i := 0; i < b.N; i++ {

app/platform/websocket_router.go

@@ -59,8 +59,8 @@ func (wr *WebSocketRouter) ServeWebSocket(conn *WebConn, r *model.WebSocketReque
 		conn.Platform.HubRegister(conn)
 
 		conn.Platform.Go(func() {
-			conn.Suite.SetStatusOnline(session.UserId, false)
-			conn.Suite.UpdateLastActivityAtIfNeeded(*session)
+			conn.Platform.SetStatusOnline(session.UserId, false)
+			conn.Platform.UpdateLastActivityAtIfNeeded(*session)
 		})
 
 		resp := model.NewWebSocketResponse(model.StatusOk, r.Seq, nil)