diff --git a/server/channels/app/platform/service.go b/server/channels/app/platform/service.go index 3289a15c71..3910070ac4 100644 --- a/server/channels/app/platform/service.go +++ b/server/channels/app/platform/service.go @@ -50,6 +50,11 @@ type PlatformService struct { filestore filestore.FileBackend exportFilestore filestore.FileBackend + // Channel for batching status updates + statusUpdateChan chan *model.Status + statusUpdateExitSignal chan struct{} + statusUpdateDoneSignal chan struct{} + cacheProvider cache.Provider statusCache cache.Cache sessionCache cache.Cache @@ -136,6 +141,9 @@ func New(sc ServiceConfig, options ...Option) (*PlatformService, error) { }, licenseListeners: map[string]func(*model.License, *model.License){}, additionalClusterHandlers: map[model.ClusterEvent]einterfaces.ClusterMessageHandler{}, + statusUpdateChan: make(chan *model.Status, statusUpdateBufferSize), + statusUpdateExitSignal: make(chan struct{}), + statusUpdateDoneSignal: make(chan struct{}), } // Assume the first user account has not been created yet. A call to the DB will later check if this is really the case. @@ -397,6 +405,10 @@ func New(sc ServiceConfig, options ...Option) (*PlatformService, error) { } func (ps *PlatformService) Start(broadcastHooks map[string]BroadcastHook) error { + // Start the status update processor. + // Must be done before hub start. + go ps.processStatusUpdates() + ps.hubStart(broadcastHooks) ps.configListenerId = ps.AddConfigListener(func(_, _ *model.Config) { @@ -497,6 +509,12 @@ func (ps *PlatformService) TotalWebsocketConnections() int { func (ps *PlatformService) Shutdown() error { ps.HubStop() + // Shutdown status processor. + // Must be done after hub shutdown. + close(ps.statusUpdateExitSignal) + // wait for it to be stopped. + <-ps.statusUpdateDoneSignal + ps.RemoveLicenseListener(ps.licenseListenerId) // we need to wait the goroutines to finish before closing the store diff --git a/server/channels/app/platform/status.go b/server/channels/app/platform/status.go index 93f567299c..8fded166d0 100644 --- a/server/channels/app/platform/status.go +++ b/server/channels/app/platform/status.go @@ -217,7 +217,7 @@ func (ps *PlatformService) SaveAndBroadcastStatus(status *model.Status) { ps.AddStatusCache(status) if err := ps.Store.Status().SaveOrUpdate(status); err != nil { - mlog.Warn("Failed to save status", mlog.String("user_id", status.UserId), mlog.Err(err)) + ps.Log().Warn("Failed to save status", mlog.String("user_id", status.UserId), mlog.Err(err)) } ps.BroadcastStatus(status) @@ -285,12 +285,12 @@ func (ps *PlatformService) UpdateLastActivityAtIfNeeded(session model.Session) { } 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)) + ps.Log().Warn("Failed to update LastActivityAt", mlog.String("user_id", session.UserId), mlog.String("session_id", session.Id), mlog.Err(err)) } session.LastActivityAt = now if err := ps.AddSessionToCache(&session); err != nil { - mlog.Warn("Failed to add session to cache", mlog.String("user_id", session.UserId), mlog.String("session_id", session.Id), mlog.Err(err)) + ps.Log().Warn("Failed to add session to cache", mlog.String("user_id", session.UserId), mlog.String("session_id", session.Id), mlog.Err(err)) } } @@ -358,18 +358,120 @@ func (ps *PlatformService) SetStatusOffline(userID string, manual bool, force bo } status, err := ps.GetStatus(userID) - if !force && err == nil && status.Manual && !manual { + if err != nil { + ps.Log().Warn("Error getting status. Setting it to offline forcefully.", mlog.String("user_id", userID), mlog.Err(err)) + } else if !force && status.Manual && !manual { return // manually set status always overrides non-manual one } + ps._setStatusOfflineAndNotify(userID, manual) +} - status = &model.Status{UserId: userID, Status: model.StatusOffline, Manual: manual, LastActivityAt: model.GetMillis(), ActiveChannel: ""} - +func (ps *PlatformService) _setStatusOfflineAndNotify(userID string, manual bool) { + status := &model.Status{UserId: userID, Status: model.StatusOffline, Manual: manual, LastActivityAt: model.GetMillis(), ActiveChannel: ""} ps.SaveAndBroadcastStatus(status) if ps.sharedChannelService != nil { ps.sharedChannelService.NotifyUserStatusChanged(status) } } +// QueueSetStatusOffline queues a status update to set a user offline +// instead of directly updating it for better performance during high load +func (ps *PlatformService) QueueSetStatusOffline(userID string, manual bool) { + if !*ps.Config().ServiceSettings.EnableUserStatuses { + return + } + + status, err := ps.GetStatus(userID) + if err != nil { + ps.Log().Warn("Error getting status. Setting it to offline forcefully.", mlog.String("user_id", userID), mlog.Err(err)) + } else if status.Manual && !manual { + // Force will be false here, so no need to add another variable. + return // manually set status always overrides non-manual one + } + + status = &model.Status{ + UserId: userID, + Status: model.StatusOffline, + Manual: manual, + LastActivityAt: model.GetMillis(), + ActiveChannel: "", + } + + select { + case ps.statusUpdateChan <- status: + // Successfully queued + default: + // Channel is full, fall back to direct update + ps.Log().Warn("Status update channel is full. Falling back to direct update") + ps._setStatusOfflineAndNotify(userID, manual) + } +} + +const ( + statusUpdateBufferSize = sendQueueSize // We use the webConn sendQueue size as a reference point for the buffer size. + statusUpdateFlushThreshold = statusUpdateBufferSize / 8 + statusUpdateBatchInterval = 500 * time.Millisecond // Max time to wait before processing +) + +// processStatusUpdates processes status updates in batches for better performance +// This runs as a goroutine and continuously monitors the statusUpdateChan +func (ps *PlatformService) processStatusUpdates() { + defer close(ps.statusUpdateDoneSignal) + + statusBatch := make(map[string]*model.Status) + ticker := time.NewTicker(statusUpdateBatchInterval) + defer ticker.Stop() + + flush := func(broadcast bool) { + if len(statusBatch) == 0 { + return + } + + // Add each status to cache. + for _, status := range statusBatch { + ps.AddStatusCache(status) + } + + // Process statuses in batch + if err := ps.Store.Status().SaveOrUpdateMany(statusBatch); err != nil { + ps.logger.Warn("Failed to save multiple statuses", mlog.Err(err)) + } + + // Broadcast each status only if hub is still running + if broadcast { + for _, status := range statusBatch { + ps.BroadcastStatus(status) + if ps.sharedChannelService != nil { + ps.sharedChannelService.NotifyUserStatusChanged(status) + } + } + } + + clear(statusBatch) + } + + for { + select { + case status := <-ps.statusUpdateChan: + // In case of duplicates, we override the last entry + statusBatch[status.UserId] = status + + if len(statusBatch) >= statusUpdateFlushThreshold { + ps.logger.Debug("Flushing statuses because the current buffer exceeded the flush threshold.", mlog.Int("current_buffer", len(statusBatch)), mlog.Int("flush_threshold", statusUpdateFlushThreshold)) + flush(true) + } + case <-ticker.C: + flush(true) + case <-ps.statusUpdateExitSignal: + // Process any remaining statuses before shutting down + // Skip broadcast since hub is already stopped + ps.logger.Debug("Exit signal received. Flushing any remaining statuses.") + flush(false) + return + } + } +} + func (ps *PlatformService) SetStatusAwayIfNeeded(userID string, manual bool) { if !*ps.Config().ServiceSettings.EnableUserStatuses { return diff --git a/server/channels/app/platform/status_test.go b/server/channels/app/platform/status_test.go index 95da146887..39d823024f 100644 --- a/server/channels/app/platform/status_test.go +++ b/server/channels/app/platform/status_test.go @@ -5,6 +5,7 @@ package platform import ( "testing" + "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -53,6 +54,86 @@ func TestTruncateDNDEndTime(t *testing.T) { assert.Equal(t, int64(1737331200), truncateDNDEndTime(1737331200)) } +func TestQueueSetStatusOffline(t *testing.T) { + th := Setup(t).InitBasic() + + defer func() { + // First tear down the test environment + th.TearDown() + + // Then verify that the status update processor has properly shut down + // by checking that the done signal channel is closed + select { + case _, ok := <-th.Service.statusUpdateDoneSignal: + // If channel is closed, ok will be false + assert.False(t, ok, "statusUpdateDoneSignal channel should be closed after teardown") + case <-time.After(5 * time.Second): + require.Fail(t, "Timed out waiting for status update processor to shut down") + } + }() + + // Create multiple user IDs + userIDs := []string{ + th.BasicUser.Id, + model.NewId(), + model.NewId(), + model.NewId(), + } + + // Add duplicate user IDs to test duplicate handling + // The second occurrence should override the first + userIDs = append(userIDs, userIDs[0], userIDs[1]) + + // Initially set all users to online + for _, userID := range userIDs { + th.Service.SetStatusOnline(userID, false) + status, err := th.Service.GetStatus(userID) + require.Nil(t, err, "Failed to get initial status") + require.Equal(t, model.StatusOnline, status.Status, "User should be online initially") + } + + // Queue status updates to offline + for i, userID := range userIDs { + // Set every other status as manual to test both cases + manual := i%2 == 0 + th.Service.QueueSetStatusOffline(userID, manual) + } + + // Wait for the background processor to handle the updates + // Use eventually consistent approach with retries + for idx, userID := range userIDs { + var status *model.Status + var err *model.AppError + + // Use poll-wait pattern to account for async processing + require.Eventually(t, func() bool { + status, err = th.Service.GetStatus(userID) + return err == nil && status.Status == model.StatusOffline + }, 5*time.Second, 100*time.Millisecond, "Status wasn't updated to offline") + + // For the duplicated user IDs, check that manual setting is based on the last call + // User[0] and User[1] are duplicated at the end of the slice + switch idx { + case 0, 4: // first duplicated user + // Last update for userIDs[0] was at index 4 (i%2 == 0, so manual = true) + require.True(t, status.Manual, "User should have manual status (duplicate case)") + case 1, 5: + // Last update for userIDs[1] was at index 5 (i%2 == 1, so manual = false) + require.False(t, status.Manual, "User should have automatic status (duplicate case)") + default: + require.Equal(t, idx%2 == 0, status.Manual, "Manual flag incorrect") + } + } + + // Verify all relevant status fields + for _, userID := range model.RemoveDuplicateStrings(userIDs) { + status, err := th.Service.GetStatus(userID) + require.Nil(t, err, "Failed to get status") + require.Equal(t, model.StatusOffline, status.Status, "User should be offline") + require.Equal(t, "", status.ActiveChannel, "ActiveChannel should be empty") + } +} + func TestSetStatusOffline(t *testing.T) { th := Setup(t).InitBasic() defer th.TearDown() diff --git a/server/channels/app/platform/web_hub.go b/server/channels/app/platform/web_hub.go index 4ed0ec6855..ccce892bc6 100644 --- a/server/channels/app/platform/web_hub.go +++ b/server/channels/app/platform/web_hub.go @@ -65,6 +65,8 @@ type webConnCountMessage struct { result chan int } +var hubSemaphoreCount = runtime.NumCPU() * 4 + // Hub is the central place to manage all websocket connections in the server. // It handles different websocket events and sending messages to individual // user connections. @@ -87,6 +89,9 @@ type Hub struct { checkConn chan *webConnCheckMessage connCount chan *webConnCountMessage broadcastHooks map[string]BroadcastHook + + // Hub-specific semaphore for limiting concurrent goroutines + hubSemaphore chan struct{} } // newWebHub creates a new Hub. @@ -104,6 +109,7 @@ func newWebHub(ps *PlatformService) *Hub { checkRegistered: make(chan *webConnSessionMessage), checkConn: make(chan *webConnCheckMessage), connCount: make(chan *webConnCountMessage), + hubSemaphore: make(chan struct{}, hubSemaphoreCount), } } @@ -469,10 +475,39 @@ func (h *Hub) SendMessage(conn *WebConn, msg model.WebSocketMessage) { } } +// ProcessAsync executes a function with hub-specific concurrency control +func (h *Hub) ProcessAsync(f func()) { + h.hubSemaphore <- struct{}{} + go func() { + defer func() { + <-h.hubSemaphore + }() + + // Add timeout protection + done := make(chan struct{}) + go func() { + defer close(done) + f() + }() + + select { + case <-done: + // Function completed normally + case <-time.After(5 * time.Second): + h.platform.Log().Warn("ProcessAsync function timed out after 5 seconds") + } + }() +} + // Stop stops the hub. func (h *Hub) Stop() { close(h.stop) <-h.didStop + // Ensure that all remaining elements are processed + // before shutting down. + for i := 0; i < hubSemaphoreCount; i++ { + h.hubSemaphore <- struct{}{} + } } // Start starts the hub. @@ -561,7 +596,7 @@ func (h *Hub) Start() { // which is intentional. if areAllInactive(conns) { userID := webConn.UserId - h.platform.Go(func() { + h.ProcessAsync(func() { // If this is an HA setup, get count for this user // from other nodes. var clusterCnt int @@ -580,7 +615,7 @@ func (h *Hub) Start() { // Only set to offline if there are no // active connections in other nodes as well. if clusterCnt == 0 { - h.platform.SetStatusOffline(userID, false, false) + h.platform.QueueSetStatusOffline(userID, false) } }) continue diff --git a/server/channels/app/platform/web_hub_test.go b/server/channels/app/platform/web_hub_test.go index 96c7953031..8e16579460 100644 --- a/server/channels/app/platform/web_hub_test.go +++ b/server/channels/app/platform/web_hub_test.go @@ -72,8 +72,6 @@ func TestHubStopWithMultipleConnections(t *testing.T) { }) require.NoError(t, err) - err = th.Service.Start(nil) - require.NoError(t, err) wc1 := registerDummyWebConn(t, th, s.Listener.Addr(), session) wc2 := registerDummyWebConn(t, th, s.Listener.Addr(), session) wc3 := registerDummyWebConn(t, th, s.Listener.Addr(), session) @@ -98,8 +96,6 @@ func TestHubStopRaceCondition(t *testing.T) { }) require.NoError(t, err) - err = th.Service.Start(nil) - require.NoError(t, err) wc1 := registerDummyWebConn(t, th, s.Listener.Addr(), session) defer wc1.Close() @@ -607,8 +603,6 @@ func TestHubIsRegistered(t *testing.T) { s := httptest.NewServer(dummyWebsocketHandler(t)) defer s.Close() - err = th.Service.Start(nil) - require.NoError(t, err) wc1 := registerDummyWebConn(t, th, s.Listener.Addr(), session) wc2 := registerDummyWebConn(t, th, s.Listener.Addr(), session) wc3 := registerDummyWebConn(t, th, s.Listener.Addr(), session) @@ -644,8 +638,6 @@ func TestHubWebConnCount(t *testing.T) { s := httptest.NewServer(dummyWebsocketHandler(t)) defer s.Close() - err = th.Service.Start(nil) - require.NoError(t, err) wc1 := registerDummyWebConn(t, th, s.Listener.Addr(), session) wc2 := registerDummyWebConn(t, th, s.Listener.Addr(), session) defer wc1.Close() diff --git a/server/channels/store/retrylayer/retrylayer.go b/server/channels/store/retrylayer/retrylayer.go index 0a347f6b86..012a6211de 100644 --- a/server/channels/store/retrylayer/retrylayer.go +++ b/server/channels/store/retrylayer/retrylayer.go @@ -12230,6 +12230,27 @@ func (s *RetryLayerStatusStore) SaveOrUpdate(status *model.Status) error { } +func (s *RetryLayerStatusStore) SaveOrUpdateMany(statuses map[string]*model.Status) error { + + tries := 0 + for { + err := s.StatusStore.SaveOrUpdateMany(statuses) + if err == nil { + return nil + } + if !isRepeatableError(err) { + return err + } + tries++ + if tries >= 3 { + err = errors.Wrap(err, "giving up after 3 consecutive repeatable transaction failures") + return err + } + timepkg.Sleep(100 * timepkg.Millisecond) + } + +} + func (s *RetryLayerStatusStore) UpdateExpiredDNDStatuses() ([]*model.Status, error) { tries := 0 diff --git a/server/channels/store/sqlstore/status_store.go b/server/channels/store/sqlstore/status_store.go index 1922d3a621..c225a5cf59 100644 --- a/server/channels/store/sqlstore/status_store.go +++ b/server/channels/store/sqlstore/status_store.go @@ -48,11 +48,9 @@ func (s SqlStatusStore) SaveOrUpdate(st *model.Status) error { Values(st.UserId, st.Status, st.Manual, st.LastActivityAt, st.DNDEndTime, st.PrevStatus) if s.DriverName() == model.DatabaseDriverMysql { - query = query.SuffixExpr(sq.Expr("ON DUPLICATE KEY UPDATE Status = ?, `Manual` = ?, LastActivityAt = ?, DNDEndTime = ?, PrevStatus = ?", - st.Status, st.Manual, st.LastActivityAt, st.DNDEndTime, st.PrevStatus)) + query = query.SuffixExpr(sq.Expr("ON DUPLICATE KEY UPDATE Status = VALUES(Status), `Manual` = VALUES(`Manual`), LastActivityAt = VALUES(LastActivityAt), DNDEndTime = VALUES(DNDEndTime), PrevStatus = VALUES(PrevStatus)")) } else { - query = query.SuffixExpr(sq.Expr("ON CONFLICT (userid) DO UPDATE SET Status = ?, Manual = ?, LastActivityAt = ?, DNDEndTime = ?, PrevStatus = ?", - st.Status, st.Manual, st.LastActivityAt, st.DNDEndTime, st.PrevStatus)) + query = query.SuffixExpr(sq.Expr("ON CONFLICT (userid) DO UPDATE SET Status = EXCLUDED.Status, Manual = EXCLUDED.Manual, LastActivityAt = EXCLUDED.LastActivityAt, DNDEndTime = EXCLUDED.DNDEndTime, PrevStatus = EXCLUDED.PrevStatus")) } if _, err := s.GetMaster().ExecBuilder(query); err != nil { @@ -62,6 +60,41 @@ func (s SqlStatusStore) SaveOrUpdate(st *model.Status) error { return nil } +func (s SqlStatusStore) SaveOrUpdateMany(statuses map[string]*model.Status) error { + if len(statuses) == 0 { + return nil + } + + // If there's only one status, use the existing method + if len(statuses) == 1 { + for _, st := range statuses { + return s.SaveOrUpdate(st) + } + } + + query := s.getQueryBuilder(). + Insert("Status"). + Columns("UserId", "Status", quoteColumnName(s.DriverName(), "Manual"), "LastActivityAt", "DNDEndTime", "PrevStatus") + + // Add values for each unique status + for _, st := range statuses { + query = query.Values(st.UserId, st.Status, st.Manual, st.LastActivityAt, st.DNDEndTime, st.PrevStatus) + } + + // Handle different databases + if s.DriverName() == model.DatabaseDriverMysql { + query = query.SuffixExpr(sq.Expr("ON DUPLICATE KEY UPDATE Status = VALUES(Status), `Manual` = VALUES(`Manual`), LastActivityAt = VALUES(LastActivityAt), DNDEndTime = VALUES(DNDEndTime), PrevStatus = VALUES(PrevStatus)")) + } else { + query = query.SuffixExpr(sq.Expr("ON CONFLICT (userid) DO UPDATE SET Status = EXCLUDED.Status, Manual = EXCLUDED.Manual, LastActivityAt = EXCLUDED.LastActivityAt, DNDEndTime = EXCLUDED.DNDEndTime, PrevStatus = EXCLUDED.PrevStatus")) + } + + if _, err := s.GetMaster().ExecBuilder(query); err != nil { + return errors.Wrap(err, "failed to upsert multiple Status records") + } + + return nil +} + func (s SqlStatusStore) Get(userId string) (*model.Status, error) { query := s.statusSelectQuery.Where(sq.Eq{"UserId": userId}) diff --git a/server/channels/store/store.go b/server/channels/store/store.go index dd1c45c800..ed7c95e643 100644 --- a/server/channels/store/store.go +++ b/server/channels/store/store.go @@ -717,6 +717,7 @@ type EmojiStore interface { type StatusStore interface { SaveOrUpdate(status *model.Status) error + SaveOrUpdateMany(statuses map[string]*model.Status) error Get(userID string) (*model.Status, error) GetByIds(userIds []string) ([]*model.Status, error) ResetAll() error diff --git a/server/channels/store/storetest/mocks/StatusStore.go b/server/channels/store/storetest/mocks/StatusStore.go index 62692e43c7..edbee6305e 100644 --- a/server/channels/store/storetest/mocks/StatusStore.go +++ b/server/channels/store/storetest/mocks/StatusStore.go @@ -138,6 +138,24 @@ func (_m *StatusStore) SaveOrUpdate(status *model.Status) error { return r0 } +// SaveOrUpdateMany provides a mock function with given fields: statuses +func (_m *StatusStore) SaveOrUpdateMany(statuses map[string]*model.Status) error { + ret := _m.Called(statuses) + + if len(ret) == 0 { + panic("no return value specified for SaveOrUpdateMany") + } + + var r0 error + if rf, ok := ret.Get(0).(func(map[string]*model.Status) error); ok { + r0 = rf(statuses) + } else { + r0 = ret.Error(0) + } + + return r0 +} + // UpdateExpiredDNDStatuses provides a mock function with no fields func (_m *StatusStore) UpdateExpiredDNDStatuses() ([]*model.Status, error) { ret := _m.Called() diff --git a/server/channels/store/storetest/status_store.go b/server/channels/store/storetest/status_store.go index 9215dc7fb3..aadb460c1f 100644 --- a/server/channels/store/storetest/status_store.go +++ b/server/channels/store/storetest/status_store.go @@ -17,19 +17,75 @@ import ( ) func TestStatusStore(t *testing.T, rctx request.CTX, ss store.Store, s SqlStore) { - t.Run("", func(t *testing.T) { testStatusStore(t, rctx, ss) }) + t.Run("Basic", func(t *testing.T) { testStatusStore(t, rctx, ss) }) t.Run("ActiveUserCount", func(t *testing.T) { testActiveUserCount(t, rctx, ss) }) t.Run("UpdateExpiredDNDStatuses", func(t *testing.T) { testUpdateExpiredDNDStatuses(t, rctx, ss) }) t.Run("Get", func(t *testing.T) { testStatusGet(t, rctx, ss, s) }) t.Run("GetByIds", func(t *testing.T) { testStatusGetByIds(t, rctx, ss, s) }) + t.Run("SaveOrUpdateMany", func(t *testing.T) { testSaveOrUpdateMany(t, rctx, ss) }) +} + +func testSaveOrUpdateMany(t *testing.T, _ request.CTX, ss store.Store) { + // Test with empty map + err := ss.Status().SaveOrUpdateMany(map[string]*model.Status{}) + require.NoError(t, err, "SaveOrUpdateMany with empty map should succeed") + + // Test with single status + status1 := &model.Status{UserId: model.NewId(), Status: model.StatusOnline, Manual: false, LastActivityAt: 10, ActiveChannel: ""} + err = ss.Status().SaveOrUpdateMany(map[string]*model.Status{ + status1.UserId: status1, + }) + require.NoError(t, err, "SaveOrUpdateMany with single status should succeed") + + // Verify the status was saved + retrieved, err := ss.Status().Get(status1.UserId) + require.NoError(t, err) + assert.Equal(t, status1.UserId, retrieved.UserId) + assert.Equal(t, status1.Status, retrieved.Status) + + // Test with multiple statuses + status2 := &model.Status{UserId: model.NewId(), Status: model.StatusAway, Manual: true, LastActivityAt: 20, ActiveChannel: ""} + status3 := &model.Status{UserId: model.NewId(), Status: model.StatusDnd, Manual: true, LastActivityAt: 30, ActiveChannel: ""} + err = ss.Status().SaveOrUpdateMany(map[string]*model.Status{ + status2.UserId: status2, + status3.UserId: status3, + }) + require.NoError(t, err, "SaveOrUpdateMany with multiple statuses should succeed") + + // Verify all statuses were saved + statuses, err := ss.Status().GetByIds([]string{status2.UserId, status3.UserId}) + require.NoError(t, err) + require.Len(t, statuses, 2, "should have retrieved both statuses") + + // Verify the retrieved statuses are actually the ones from status2 and status3 + statusMap := make(map[string]*model.Status) + for _, status := range statuses { + statusMap[status.UserId] = status + } + assert.Equal(t, status2.Status, statusMap[status2.UserId].Status) + assert.Equal(t, status3.Status, statusMap[status3.UserId].Status) + + // Test with duplicate userIds (last one should win) + status4 := &model.Status{UserId: status1.UserId, Status: model.StatusOffline, Manual: true, LastActivityAt: 40, ActiveChannel: ""} + status5 := &model.Status{UserId: status1.UserId, Status: model.StatusDnd, Manual: false, LastActivityAt: 50, ActiveChannel: ""} + err = ss.Status().SaveOrUpdateMany(map[string]*model.Status{ + status4.UserId: status4, + status5.UserId: status5, + }) + require.NoError(t, err, "SaveOrUpdateMany with duplicate userIds should succeed") + + // Verify the last status was saved + retrieved, err = ss.Status().Get(status1.UserId) + require.NoError(t, err) + assert.Equal(t, status1.UserId, retrieved.UserId) + assert.Equal(t, status5.Status, retrieved.Status) + assert.Equal(t, int64(50), retrieved.LastActivityAt) } func testStatusStore(t *testing.T, _ request.CTX, ss store.Store) { status := &model.Status{UserId: model.NewId(), Status: model.StatusOnline, Manual: false, LastActivityAt: 0, ActiveChannel: ""} require.NoError(t, ss.Status().SaveOrUpdate(status)) - status.LastActivityAt = 10 - _, err := ss.Status().Get(status.UserId) require.NoError(t, err) @@ -42,13 +98,31 @@ func testStatusStore(t *testing.T, _ request.CTX, ss store.Store) { statuses, err := ss.Status().GetByIds([]string{status.UserId, "junk"}) require.NoError(t, err) require.Len(t, statuses, 1, "should only have 1 status") + assert.Equal(t, status, statuses[0]) + + // Test updating an existing status + updatedStatus := &model.Status{ + UserId: status.UserId, + Status: model.StatusDnd, + Manual: false, + LastActivityAt: 1234, + DNDEndTime: 5678, + PrevStatus: model.StatusOnline, + ActiveChannel: "", // This field won't be stored, so set it to match what we'll get back + } + require.NoError(t, ss.Status().SaveOrUpdate(updatedStatus)) + + // Verify status was updated + retrievedStatus, err := ss.Status().Get(status.UserId) + require.NoError(t, err) + assert.Equal(t, updatedStatus, retrievedStatus) err = ss.Status().ResetAll() require.NoError(t, err) statusParameter, err := ss.Status().Get(status.UserId) require.NoError(t, err) - require.Equal(t, statusParameter.Status, model.StatusOffline, "should be offline") + require.Equal(t, model.StatusOffline, statusParameter.Status, "should be offline") err = ss.Status().UpdateLastActivityAt(status.UserId, 10) require.NoError(t, err) diff --git a/server/channels/store/timerlayer/timerlayer.go b/server/channels/store/timerlayer/timerlayer.go index 727f75b806..24ae420ec0 100644 --- a/server/channels/store/timerlayer/timerlayer.go +++ b/server/channels/store/timerlayer/timerlayer.go @@ -9621,6 +9621,22 @@ func (s *TimerLayerStatusStore) SaveOrUpdate(status *model.Status) error { return err } +func (s *TimerLayerStatusStore) SaveOrUpdateMany(statuses map[string]*model.Status) error { + start := time.Now() + + err := s.StatusStore.SaveOrUpdateMany(statuses) + + elapsed := float64(time.Since(start)) / float64(time.Second) + if s.Root.Metrics != nil { + success := "false" + if err == nil { + success = "true" + } + s.Root.Metrics.ObserveStoreMethodDuration("StatusStore.SaveOrUpdateMany", success, elapsed) + } + return err +} + func (s *TimerLayerStatusStore) UpdateExpiredDNDStatuses() ([]*model.Status, error) { start := time.Now()