diff --git a/server/channels/app/shared_channel_global_user_sync_self_referential_test.go b/server/channels/app/shared_channel_global_user_sync_self_referential_test.go new file mode 100644 index 0000000000..bcb2fe9773 --- /dev/null +++ b/server/channels/app/shared_channel_global_user_sync_self_referential_test.go @@ -0,0 +1,1312 @@ +// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved. +// See LICENSE.txt for license information. + +package app + +import ( + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "net/http/httptest" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/mattermost/mattermost/server/public/model" + "github.com/mattermost/mattermost/server/v8/platform/services/remotecluster" + "github.com/mattermost/mattermost/server/v8/platform/services/sharedchannel" +) + +// TestSharedChannelGlobalUserSyncSelfReferential is a comprehensive test suite for MM-62751 +// that tests global user synchronization between connected Mattermost instances. +// It uses a self-referential approach where a server syncs with itself, providing real HTTP communication +// without mocks or invalid URLs. We test calling SyncAllUsersForRemoteCluster directly. +func TestSharedChannelGlobalUserSyncSelfReferential(t *testing.T) { + // Setup with default batch size + th := SetupConfig(t, func(cfg *model.Config) { + *cfg.ConnectedWorkspacesSettings.EnableRemoteClusterService = true + *cfg.ConnectedWorkspacesSettings.EnableSharedChannels = true + *cfg.ConnectedWorkspacesSettings.SyncUsersOnConnectionOpen = true + // Set default batch size - EnsureCleanState will reset to this value + // Individual tests can override as needed (e.g., Test 3 sets it to 4) + defaultBatchSize := 20 + cfg.ConnectedWorkspacesSettings.GlobalUserSyncBatchSize = &defaultBatchSize + // Enable the feature flag for global user sync + cfg.FeatureFlags.EnableSyncAllUsersForRemoteCluster = true + }).InitBasic() + defer th.TearDown() + + ss := th.App.Srv().Store() + + // Get the shared channel service and cast to concrete type to access SyncAllUsersForRemoteCluster + scsInterface := th.App.Srv().GetSharedChannelSyncService() + service, ok := scsInterface.(*sharedchannel.Service) + require.True(t, ok, "Expected sharedchannel.Service concrete type") + + // Verify the service is active + require.True(t, service.Active(), "SharedChannel service should be active") + + // Force the service to be active + err := service.Start() + require.NoError(t, err) + + // Also ensure the remote cluster service is running so callbacks work + rcService := th.App.Srv().GetRemoteClusterService() + if rcService != nil { + _ = rcService.Start() + + // Force the service to be active in test environment + if rc, ok := rcService.(*remotecluster.Service); ok { + rc.SetActive(true) + } + + // Verify it's active + if !rcService.Active() { + t.Fatalf("RemoteClusterService is not active after Start") + } + } + + t.Run("Test 1: Individual User Sync", func(t *testing.T) { + // This test verifies end-to-end user synchronization for a single user, including: + // - Syncing a user from Server A to Server B + // - Proper cursor tracking with LastGlobalUserSyncAt + // - Verification of user addition/removal on receiving side + EnsureCleanState(t, th, ss) + + var syncedUsers []string + var mu sync.Mutex + var syncHandler *SelfReferentialSyncHandler + + // Create a test HTTP server that acts as the "remote" cluster + testServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if syncHandler != nil { + syncHandler.HandleRequest(w, r) + } else { + writeOKResponse(w) + } + })) + defer testServer.Close() + + // Create a self-referential remote cluster + now := model.GetMillis() + selfCluster := &model.RemoteCluster{ + RemoteId: model.NewId(), + Name: "self-cluster", + SiteURL: testServer.URL, + CreateAt: now, + LastPingAt: now, // Ensure it's considered online + LastGlobalUserSyncAt: 0, + Token: model.NewId(), + CreatorId: th.BasicUser.Id, + RemoteToken: model.NewId(), + } + selfCluster, err = ss.RemoteCluster().Save(selfCluster) + require.NoError(t, err) + + // Initialize sync handler with callbacks + syncHandler = NewSelfReferentialSyncHandler(t, service, selfCluster) + syncHandler.OnGlobalUserSync = func(userIds []string, messageNumber int32) { + mu.Lock() + syncedUsers = append(syncedUsers, userIds...) + mu.Unlock() + } + + // Create a new user to sync + user := th.CreateUser() + + // Ensure user has a recent update time for cursor-based sync + user.UpdateAt = model.GetMillis() + _, err = ss.User().Update(th.Context, user, true) + require.NoError(t, err) + + // Trigger global user sync directly + err = service.HandleSyncAllUsersForTesting(selfCluster) + require.NoError(t, err) + + // Wait for sync to complete + require.Eventually(t, func() bool { + count := syncHandler.GetSyncMessageCount() + mu.Lock() + defer mu.Unlock() + return count > 0 + }, 5*time.Second, 100*time.Millisecond, "Should have received at least one sync message") + + // Verify the user was synced + mu.Lock() + assert.Contains(t, syncedUsers, user.Id, "New user should be synced") + mu.Unlock() + + // Verify cursor was updated + updatedCluster, clusterErr := ss.RemoteCluster().Get(selfCluster.RemoteId, true) + require.NoError(t, clusterErr) + assert.Greater(t, updatedCluster.LastGlobalUserSyncAt, int64(0), "Cursor should be updated after sync") + }) + + t.Run("Test 2: Batch User Sync with Type Filtering", func(t *testing.T) { + EnsureCleanState(t, th, ss) + + // Set batch size to 4 for testing batching behavior + batchSize := 4 + th.App.UpdateConfig(func(cfg *model.Config) { + cfg.ConnectedWorkspacesSettings.GlobalUserSyncBatchSize = &batchSize + }) + + var mu sync.Mutex + var batchedUserIDs [][]string + var syncHandler *SelfReferentialSyncHandler + + // Create test HTTP server + testServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if syncHandler != nil { + syncHandler.HandleRequest(w, r) + } else { + writeOKResponse(w) + } + })) + defer testServer.Close() + + // Create self-referential remote cluster + selfCluster := &model.RemoteCluster{ + RemoteId: model.NewId(), + Name: "self-cluster-batch", + SiteURL: testServer.URL, + CreateAt: model.GetMillis(), + LastPingAt: model.GetMillis(), + LastGlobalUserSyncAt: 0, + Token: model.NewId(), + CreatorId: th.BasicUser.Id, + RemoteToken: model.NewId(), + } + selfCluster, err = ss.RemoteCluster().Save(selfCluster) + require.NoError(t, err) + + baseTime := model.GetMillis() + + // Create user with old timestamp + userWithOldTimestamp := th.CreateUser() + userWithOldTimestamp.UpdateAt = 1 + _, err = ss.User().Update(th.Context, userWithOldTimestamp, false) + require.NoError(t, err) + + // Verify the user was actually updated with the old timestamp + verifiedUser, pErr := ss.User().Get(context.Background(), userWithOldTimestamp.Id) + require.NoError(t, pErr) + userWithOldTimestamp = verifiedUser + + // Create regular users + regularUsers := make([]*model.User, 3) + for i := 0; i < 3; i++ { + regularUsers[i] = th.CreateUser() + regularUsers[i].UpdateAt = baseTime + int64(i*100) + _, err = ss.User().Update(th.Context, regularUsers[i], true) + require.NoError(t, err) + } + + // Create bot + bot := th.CreateBot() + botUser, appErr := th.App.GetUser(bot.UserId) + require.Nil(t, appErr) + botUser.UpdateAt = baseTime + 300 + _, err = ss.User().Update(th.Context, botUser, true) + require.NoError(t, err) + + // Create system admin + systemAdmin := th.CreateUser() + _, appErr = th.App.UpdateUserRoles(th.Context, systemAdmin.Id, model.SystemAdminRoleId+" "+model.SystemUserRoleId, false) + require.Nil(t, appErr) + systemAdmin.UpdateAt = baseTime + 400 + _, err = ss.User().Update(th.Context, systemAdmin, true) + require.NoError(t, err) + + // Create guest user + guest := th.CreateGuest() + guest.UpdateAt = baseTime + 500 + _, err = ss.User().Update(th.Context, guest, true) + require.NoError(t, err) + + // Create remote user (should NOT be synced) + remoteUser := th.CreateUser() + remoteUser.RemoteId = &selfCluster.RemoteId + remoteUser.UpdateAt = baseTime + 600 + _, err = ss.User().Update(th.Context, remoteUser, true) + require.NoError(t, err) + + // Create inactive user (should NOT be synced) + inactiveUser := th.CreateUser() + inactiveUser.UpdateAt = baseTime + 700 + inactiveUser.DeleteAt = model.GetMillis() + _, err = ss.User().Update(th.Context, inactiveUser, true) + require.NoError(t, err) + + // Initialize sync handler + syncHandler = NewSelfReferentialSyncHandler(t, service, selfCluster) + syncHandler.OnGlobalUserSync = func(userIds []string, messageNumber int32) { + mu.Lock() + batchedUserIDs = append(batchedUserIDs, userIds) + mu.Unlock() + } + + // Trigger sync + err = service.HandleSyncAllUsersForTesting(selfCluster) + require.NoError(t, err) + + // Wait for sync to complete + require.Eventually(t, func() bool { + mu.Lock() + defer mu.Unlock() + + // Count total synced users + allSyncedUserIDs := make(map[string]bool) + for _, batch := range batchedUserIDs { + for _, userID := range batch { + allSyncedUserIDs[userID] = true + } + } + + // Check that our specific test users are synced + guestSynced := allSyncedUserIDs[guest.Id] + botSynced := allSyncedUserIDs[bot.UserId] + systemAdminSynced := allSyncedUserIDs[systemAdmin.Id] + userWithOldTimestampSynced := allSyncedUserIDs[userWithOldTimestamp.Id] + remoteUserNotSynced := !allSyncedUserIDs[remoteUser.Id] + inactiveUserNotSynced := !allSyncedUserIDs[inactiveUser.Id] + + return guestSynced && botSynced && systemAdminSynced && userWithOldTimestampSynced && + remoteUserNotSynced && inactiveUserNotSynced + }, 10*time.Second, 500*time.Millisecond, "Should sync expected users") + + // Verify results + mu.Lock() + defer mu.Unlock() + + allSyncedUserIDs := make(map[string]bool) + for _, batch := range batchedUserIDs { + assert.LessOrEqual(t, len(batch), batchSize, "Batch size should not exceed configured limit") + for _, userID := range batch { + allSyncedUserIDs[userID] = true + } + } + + // Verify user type filtering + assert.Contains(t, allSyncedUserIDs, bot.UserId, "Bot should be synced") + assert.Contains(t, allSyncedUserIDs, systemAdmin.Id, "System admin should be synced") + assert.Contains(t, allSyncedUserIDs, guest.Id, "Guest user should be synced") + assert.Contains(t, allSyncedUserIDs, userWithOldTimestamp.Id, "User with old timestamp should be synced") + assert.NotContains(t, allSyncedUserIDs, remoteUser.Id, "Remote user should NOT be synced") + assert.NotContains(t, allSyncedUserIDs, inactiveUser.Id, "Inactive user should NOT be synced") + + // Verify regular users are synced + for i, user := range regularUsers { + assert.Contains(t, allSyncedUserIDs, user.Id, "Regular user %d should be synced", i+1) + } + + // Verify cursor was updated + updatedCluster, clusterErr := ss.RemoteCluster().Get(selfCluster.RemoteId, true) + require.NoError(t, clusterErr) + assert.Greater(t, updatedCluster.LastGlobalUserSyncAt, int64(0), "Cursor should be updated after batch sync") + }) + + t.Run("Test 3: Multiple Remote Clusters", func(t *testing.T) { + // This test verifies syncing users to multiple remote clusters: + // - Syncing users from Server A to both Server B and Server C + // - Ensuring proper cursor tracking on each remote + // - Verifying user propagation across all connected servers + EnsureCleanState(t, th, ss) + + var syncMessagesPerCluster = make(map[string]*int32) + var syncedUsersPerCluster = make(map[string][]string) + var mu sync.Mutex + + // Create multiple test servers for different clusters + clusters := make([]*model.RemoteCluster, 3) + testServers := make([]*httptest.Server, 3) + + for i := 0; i < 3; i++ { + clusterName := fmt.Sprintf("cluster-%d", i+1) + var count int32 + syncMessagesPerCluster[clusterName] = &count + syncedUsersPerCluster[clusterName] = []string{} + + idx := i // Capture index for closure + testServers[i] = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/api/v4/remotecluster/msg" { + clusterName := fmt.Sprintf("cluster-%d", idx+1) + atomic.AddInt32(syncMessagesPerCluster[clusterName], 1) + + // Parse message + bodyBytes, _ := io.ReadAll(r.Body) + var frame model.RemoteClusterFrame + if unmarshalErr := json.Unmarshal(bodyBytes, &frame); unmarshalErr == nil { + var syncMsg model.SyncMsg + if unmarshalErr := json.Unmarshal(frame.Msg.Payload, &syncMsg); unmarshalErr == nil { + // Track synced users for this cluster + mu.Lock() + for userID := range syncMsg.Users { + syncedUsersPerCluster[clusterName] = append(syncedUsersPerCluster[clusterName], userID) + } + mu.Unlock() + + // Create success response + syncResp := &model.SyncResponse{ + UsersSyncd: make([]string, 0, len(syncMsg.Users)), + } + for userID := range syncMsg.Users { + syncResp.UsersSyncd = append(syncResp.UsersSyncd, userID) + } + + response := &remotecluster.Response{ + Status: remotecluster.ResponseStatusOK, + } + _ = response.SetPayload(syncResp) + + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + respBytes, _ := json.Marshal(response) + _, _ = w.Write(respBytes) + return + } + } + } + writeOKResponse(w) + })) + } + + // Cleanup servers + defer func() { + for _, server := range testServers { + server.Close() + } + }() + + // Create remote clusters + for i := 0; i < 3; i++ { + clusters[i] = &model.RemoteCluster{ + RemoteId: model.NewId(), + Name: fmt.Sprintf("cluster-%d", i+1), + SiteURL: testServers[i].URL, + CreateAt: model.GetMillis(), + LastPingAt: model.GetMillis(), + LastGlobalUserSyncAt: 0, + Token: model.NewId(), + CreatorId: th.BasicUser.Id, + RemoteToken: model.NewId(), + } + clusters[i], err = ss.RemoteCluster().Save(clusters[i]) + require.NoError(t, err) + } + + // Create users to sync + users := make([]*model.User, 5) + for i := 0; i < 5; i++ { + users[i] = th.CreateUser() + users[i].UpdateAt = model.GetMillis() + int64(i) + _, err = ss.User().Update(th.Context, users[i], true) + require.NoError(t, err) + } + + // Sync to all clusters + for _, cluster := range clusters { + err = service.HandleSyncAllUsersForTesting(cluster) + require.NoError(t, err) + } + + // Wait for syncs to complete + require.Eventually(t, func() bool { + // Each cluster should receive sync messages + for _, countPtr := range syncMessagesPerCluster { + if atomic.LoadInt32(countPtr) == 0 { + return false + } + } + return true + }, 10*time.Second, 100*time.Millisecond, "All clusters should receive sync messages") + + // Verify each cluster received the users + mu.Lock() + for clusterName, syncedUsers := range syncedUsersPerCluster { + for _, user := range users { + assert.Contains(t, syncedUsers, user.Id, + "Cluster %s should have received user %s", clusterName, user.Id) + } + } + mu.Unlock() + + // Verify cursor updates for each cluster + for _, cluster := range clusters { + updatedCluster, err2 := ss.RemoteCluster().Get(cluster.RemoteId, true) + require.NoError(t, err2) + assert.Greater(t, updatedCluster.LastGlobalUserSyncAt, int64(0), + "Cursor should be updated for cluster %s", cluster.Name) + } + }) + + t.Run("Test 4: Cursor Management", func(t *testing.T) { + // This test verifies proper cursor handling: + // - Cursor persistence across sync operations + // - Failed sync handling (cursor should not update on failed syncs) + // - Cursor updates only when sync is successful + EnsureCleanState(t, th, ss) + + var syncAttempts int32 + var failureMode atomic.Bool + failureMode.Store(false) + var syncHandler *SelfReferentialSyncHandler + + // Create test HTTP server + testServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/api/v4/remotecluster/msg": + atomic.AddInt32(&syncAttempts, 1) + + if failureMode.Load() { + w.WriteHeader(http.StatusInternalServerError) + _, _ = w.Write([]byte(`{"error":"simulated failure"}`)) + return + } + + // Success case + if syncHandler != nil { + syncHandler.HandleRequest(w, r) + } else { + writeOKResponse(w) + } + case "/api/v4/remotecluster/ping": + writeOKResponse(w) + default: + w.WriteHeader(http.StatusNotFound) + } + })) + defer testServer.Close() + + // Create self-referential remote cluster + selfCluster := &model.RemoteCluster{ + RemoteId: model.NewId(), + Name: "self-cluster-cursor", + SiteURL: testServer.URL, + CreateAt: model.GetMillis(), + LastPingAt: model.GetMillis(), + LastGlobalUserSyncAt: 0, + Token: model.NewId(), + CreatorId: th.BasicUser.Id, + RemoteToken: model.NewId(), + } + selfCluster, err = ss.RemoteCluster().Save(selfCluster) + require.NoError(t, err) + + // Initialize sync handler + syncHandler = NewSelfReferentialSyncHandler(t, service, selfCluster) + + // Create first batch of users + user1 := th.CreateUser() + user2 := th.CreateUser() + + // Set update times + user1.UpdateAt = model.GetMillis() + user2.UpdateAt = model.GetMillis() + 1000 + _, err = ss.User().Update(th.Context, user1, true) + require.NoError(t, err) + _, err = ss.User().Update(th.Context, user2, true) + require.NoError(t, err) + + // First sync - should succeed and update cursor + err = service.HandleSyncAllUsersForTesting(selfCluster) + require.NoError(t, err) + + // Wait for first sync + require.Eventually(t, func() bool { + return atomic.LoadInt32(&syncAttempts) > 0 + }, 5*time.Second, 100*time.Millisecond, "Should have attempted sync") + + // Verify cursor was updated + cluster1, err2 := ss.RemoteCluster().Get(selfCluster.RemoteId, true) + require.NoError(t, err2) + firstCursor := cluster1.LastGlobalUserSyncAt + assert.Greater(t, firstCursor, int64(0), "Cursor should be updated after first sync") + + // Enable failure mode + failureMode.Store(true) + + // Create a new user after cursor + user3 := th.CreateUser() + user3.UpdateAt = model.GetMillis() + _, err = ss.User().Update(th.Context, user3, true) + require.NoError(t, err) + + // Second sync - should fail + initialAttempts := atomic.LoadInt32(&syncAttempts) + err = service.HandleSyncAllUsersForTesting(selfCluster) + require.NoError(t, err) // The method itself shouldn't error, just the remote call + + // Wait for failed sync attempt + require.Eventually(t, func() bool { + return atomic.LoadInt32(&syncAttempts) > initialAttempts + }, 5*time.Second, 100*time.Millisecond, "Should have attempted sync") + + // Verify cursor was NOT updated on failure + cluster2, err2 := ss.RemoteCluster().Get(selfCluster.RemoteId, true) + require.NoError(t, err2) + assert.Equal(t, firstCursor, cluster2.LastGlobalUserSyncAt, "Cursor should not update on failed sync") + + // Disable failure mode + failureMode.Store(false) + + // Third sync - should succeed and update cursor + preSuccessAttempts := atomic.LoadInt32(&syncAttempts) + err = service.HandleSyncAllUsersForTesting(selfCluster) + require.NoError(t, err) + + // Wait for successful sync + require.Eventually(t, func() bool { + return atomic.LoadInt32(&syncAttempts) > preSuccessAttempts + }, 5*time.Second, 100*time.Millisecond, "Should have attempted sync") + + // Verify cursor was updated after successful sync + cluster3, err3 := ss.RemoteCluster().Get(selfCluster.RemoteId, true) + require.NoError(t, err3) + assert.Greater(t, cluster3.LastGlobalUserSyncAt, firstCursor, "Cursor should advance after successful sync") + }) + + t.Run("Test 5: Feature Flag Testing", func(t *testing.T) { + // This test verifies feature flag handling: + // - Verifies syncing works when feature flag is enabled + // - Confirms syncing is disabled when feature flag is disabled + // - Ensures cursor is only updated when flag is enabled + EnsureCleanState(t, th, ss) + + var syncMessageCount int32 + + // Create test HTTP server + testServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/api/v4/remotecluster/msg" { + atomic.AddInt32(&syncMessageCount, 1) + } + writeOKResponse(w) + })) + defer testServer.Close() + + // Create self-referential remote cluster + selfCluster := &model.RemoteCluster{ + RemoteId: model.NewId(), + Name: "self-cluster-feature-flag", + SiteURL: testServer.URL, + CreateAt: model.GetMillis(), + LastPingAt: model.GetMillis(), + LastGlobalUserSyncAt: 0, + Token: model.NewId(), + CreatorId: th.BasicUser.Id, + RemoteToken: model.NewId(), + } + selfCluster, err = ss.RemoteCluster().Save(selfCluster) + require.NoError(t, err) + + // Create users + for i := 0; i < 3; i++ { + user := th.CreateUser() + user.UpdateAt = model.GetMillis() + int64(i) + _, err = ss.User().Update(th.Context, user, true) + require.NoError(t, err) + } + + // Test 1: Sync with feature flag disabled + th.App.UpdateConfig(func(cfg *model.Config) { + cfg.FeatureFlags.EnableSyncAllUsersForRemoteCluster = false + // Also disable SyncUsersOnConnectionOpen to ensure sync is completely disabled + *cfg.ConnectedWorkspacesSettings.SyncUsersOnConnectionOpen = false + }) + err = th.App.ReloadConfig() + require.NoError(t, err) + + atomic.StoreInt32(&syncMessageCount, 0) + err = service.HandleSyncAllUsersForTesting(selfCluster) + require.NoError(t, err) + + // Verify no sync messages were sent + require.Never(t, func() bool { + return atomic.LoadInt32(&syncMessageCount) > 0 + }, 2*time.Second, 100*time.Millisecond, "No sync should occur with feature flag disabled") + + // Verify cursor was not updated + cluster1, err2 := ss.RemoteCluster().Get(selfCluster.RemoteId, true) + require.NoError(t, err2) + assert.Equal(t, int64(0), cluster1.LastGlobalUserSyncAt, "Cursor should not update when flag is disabled") + + // Test 2: Sync with feature flag enabled + th.App.UpdateConfig(func(cfg *model.Config) { + cfg.FeatureFlags.EnableSyncAllUsersForRemoteCluster = true + // Re-enable SyncUsersOnConnectionOpen as well + *cfg.ConnectedWorkspacesSettings.SyncUsersOnConnectionOpen = true + }) + err = th.App.ReloadConfig() + require.NoError(t, err) + + atomic.StoreInt32(&syncMessageCount, 0) + err = service.HandleSyncAllUsersForTesting(selfCluster) + require.NoError(t, err) + + // Verify sync messages were sent + require.Eventually(t, func() bool { + return atomic.LoadInt32(&syncMessageCount) > 0 + }, 5*time.Second, 100*time.Millisecond, "Sync should occur with feature flag enabled") + + // Verify cursor was updated + cluster2, err2 := ss.RemoteCluster().Get(selfCluster.RemoteId, true) + require.NoError(t, err2) + assert.Greater(t, cluster2.LastGlobalUserSyncAt, int64(0), "Cursor should update when flag is enabled") + }) + + t.Run("Test 6: Config Option Testing", func(t *testing.T) { + // This test verifies the SyncUsersOnConnectionOpen config option: + // - Verifies automatic sync on connection open when enabled + // - Confirms no sync occurs on connection open when disabled + // - Tests cursor updates in both scenarios + EnsureCleanState(t, th, ss) + + var syncMessageCount int32 + var connectionOpenSyncOccurred atomic.Bool + + // Create test HTTP server + testServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/api/v4/remotecluster/msg" { + atomic.AddInt32(&syncMessageCount, 1) + + // Parse message to check if it's a user sync + bodyBytes, _ := io.ReadAll(r.Body) + var frame model.RemoteClusterFrame + if unmarshalErr := json.Unmarshal(bodyBytes, &frame); unmarshalErr == nil { + var syncMsg model.SyncMsg + if unmarshalErr := json.Unmarshal(frame.Msg.Payload, &syncMsg); unmarshalErr == nil && len(syncMsg.Users) > 0 { + connectionOpenSyncOccurred.Store(true) + } + } + } + writeOKResponse(w) + })) + defer testServer.Close() + + // Create users before creating remote cluster + for i := 0; i < 3; i++ { + user := th.CreateUser() + user.UpdateAt = model.GetMillis() + int64(i) + _, err = ss.User().Update(th.Context, user, true) + require.NoError(t, err) + } + + // Test 1: Connection open with sync disabled (default) + // Ensure config option is disabled (default) + th.App.UpdateConfig(func(cfg *model.Config) { + if cfg.ConnectedWorkspacesSettings.SyncUsersOnConnectionOpen == nil { + cfg.ConnectedWorkspacesSettings.SyncUsersOnConnectionOpen = model.NewPointer(false) + } else { + *cfg.ConnectedWorkspacesSettings.SyncUsersOnConnectionOpen = false + } + }) + + // Create remote cluster - simulating connection open + selfCluster1 := &model.RemoteCluster{ + RemoteId: model.NewId(), + Name: "self-cluster-config-disabled", + SiteURL: testServer.URL, + CreateAt: model.GetMillis(), + LastPingAt: model.GetMillis(), + LastGlobalUserSyncAt: 0, + Token: model.NewId(), + CreatorId: th.BasicUser.Id, + RemoteToken: model.NewId(), + } + _, err = ss.RemoteCluster().Save(selfCluster1) + require.NoError(t, err) + + // Verify no automatic sync occurs within a reasonable time + require.Never(t, func() bool { + return connectionOpenSyncOccurred.Load() || atomic.LoadInt32(&syncMessageCount) > 0 + }, 2*time.Second, 100*time.Millisecond, "No automatic sync should occur when config is disabled") + + // Test 2: Connection open with sync enabled + // Reset counters + atomic.StoreInt32(&syncMessageCount, 0) + connectionOpenSyncOccurred.Store(false) + + // Enable config option + th.App.UpdateConfig(func(cfg *model.Config) { + *cfg.ConnectedWorkspacesSettings.SyncUsersOnConnectionOpen = true + }) + + // Create another remote cluster - simulating connection open + selfCluster2 := &model.RemoteCluster{ + RemoteId: model.NewId(), + Name: "self-cluster-config-enabled", + SiteURL: testServer.URL, + CreateAt: model.GetMillis(), + LastPingAt: model.GetMillis(), + LastGlobalUserSyncAt: 0, + Token: model.NewId(), + CreatorId: th.BasicUser.Id, + RemoteToken: model.NewId(), + } + selfCluster2, err = ss.RemoteCluster().Save(selfCluster2) + require.NoError(t, err) + + // For this test, we need to manually trigger what would happen on connection open + // since the shared channel service might not automatically pick up new clusters in test mode + if th.App.Config().ConnectedWorkspacesSettings.SyncUsersOnConnectionOpen != nil && + *th.App.Config().ConnectedWorkspacesSettings.SyncUsersOnConnectionOpen { + // Manually trigger sync as would happen on connection open + err = service.HandleSyncAllUsersForTesting(selfCluster2) + require.NoError(t, err) + } + + // Wait for sync to occur + require.Eventually(t, func() bool { + return connectionOpenSyncOccurred.Load() + }, 5*time.Second, 100*time.Millisecond, "Automatic sync should occur when config is enabled") + + // Verify sync occurred + assert.Greater(t, atomic.LoadInt32(&syncMessageCount), int32(0), "Should have sync messages when config enabled") + + // Verify cursor was updated + updatedCluster, err2 := ss.RemoteCluster().Get(selfCluster2.RemoteId, true) + require.NoError(t, err2) + assert.Greater(t, updatedCluster.LastGlobalUserSyncAt, int64(0), "Cursor should be updated after automatic sync") + }) + + t.Run("Test 7: Sync Task After Connection Becomes Available", func(t *testing.T) { + // This test verifies that global user sync works correctly + // when a remote cluster becomes available after being offline + EnsureCleanState(t, th, ss) + + var syncTaskCreated atomic.Bool + var syncHandler *SelfReferentialSyncHandler + + // Create test HTTP server + testServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if syncHandler != nil { + syncHandler.HandleRequest(w, r) + } else { + writeOKResponse(w) + } + })) + defer testServer.Close() + + // Create remote cluster that was previously offline (old LastPingAt) + selfCluster := &model.RemoteCluster{ + RemoteId: model.NewId(), + Name: "self-cluster-reconnect", + SiteURL: testServer.URL, + CreateAt: model.GetMillis() - 300000, // Created 5 minutes ago + LastPingAt: model.GetMillis() - 120000, // Last ping 2 minutes ago + LastGlobalUserSyncAt: 0, + Token: model.NewId(), + CreatorId: th.BasicUser.Id, + RemoteToken: model.NewId(), + } + selfCluster, err = ss.RemoteCluster().Save(selfCluster) + require.NoError(t, err) + + // Initialize sync handler + syncHandler = NewSelfReferentialSyncHandler(t, service, selfCluster) + syncHandler.OnGlobalUserSync = func(userIds []string, messageNumber int32) { + syncTaskCreated.Store(true) + } + + // Create some users to sync + for i := 0; i < 3; i++ { + user := th.CreateUser() + user.UpdateAt = model.GetMillis() + int64(i) + _, err = ss.User().Update(th.Context, user, true) + require.NoError(t, err) + } + + // Update LastPingAt to simulate cluster coming back online + selfCluster.LastPingAt = model.GetMillis() + _, err = ss.RemoteCluster().Update(selfCluster) + require.NoError(t, err) + + // Trigger global user sync as would happen when connection is restored + err = service.HandleSyncAllUsersForTesting(selfCluster) + require.NoError(t, err) + + // Verify sync task was created and executed + require.Eventually(t, func() bool { + return syncTaskCreated.Load() + }, 5*time.Second, 100*time.Millisecond, "Sync should execute when cluster comes back online") + + // Verify cursor was updated + updatedCluster, err2 := ss.RemoteCluster().Get(selfCluster.RemoteId, true) + require.NoError(t, err2) + assert.Greater(t, updatedCluster.LastGlobalUserSyncAt, int64(0), "Cursor should be updated after sync") + }) + + t.Run("Test 8: Remote Cluster Offline During Sync", func(t *testing.T) { + // This test verifies behavior when a remote cluster goes offline during sync: + // - Sync should fail gracefully + // - Cursor should not be updated + // - No partial data should be persisted + EnsureCleanState(t, th, ss) + + var syncAttempts int32 + var serverOnline atomic.Bool + serverOnline.Store(true) + + // Create test HTTP server that can simulate going offline + testServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if !serverOnline.Load() { + // Simulate server being offline + w.WriteHeader(http.StatusServiceUnavailable) + return + } + + if r.URL.Path == "/api/v4/remotecluster/msg" { + atomic.AddInt32(&syncAttempts, 1) + // On second attempt, go offline + if atomic.LoadInt32(&syncAttempts) >= 2 { + serverOnline.Store(false) + w.WriteHeader(http.StatusServiceUnavailable) + return + } + } + writeOKResponse(w) + })) + defer testServer.Close() + + // Create remote cluster + selfCluster := &model.RemoteCluster{ + RemoteId: model.NewId(), + Name: "self-cluster-offline", + SiteURL: testServer.URL, + CreateAt: model.GetMillis(), + LastPingAt: model.GetMillis(), + LastGlobalUserSyncAt: 0, + Token: model.NewId(), + CreatorId: th.BasicUser.Id, + RemoteToken: model.NewId(), + } + selfCluster, err = ss.RemoteCluster().Save(selfCluster) + require.NoError(t, err) + + // Create users to sync + for i := 0; i < 5; i++ { + user := th.CreateUser() + user.UpdateAt = model.GetMillis() + int64(i) + _, err = ss.User().Update(th.Context, user, true) + require.NoError(t, err) + } + + // First sync should succeed + err = service.HandleSyncAllUsersForTesting(selfCluster) + require.NoError(t, err) + + // Wait for first sync + require.Eventually(t, func() bool { + return atomic.LoadInt32(&syncAttempts) >= 1 + }, 5*time.Second, 100*time.Millisecond) + + // Get cursor after first sync + cluster1, err2 := ss.RemoteCluster().Get(selfCluster.RemoteId, true) + require.NoError(t, err2) + firstCursor := cluster1.LastGlobalUserSyncAt + assert.Greater(t, firstCursor, int64(0), "Cursor should be set after first sync") + + // Create more users + for i := 0; i < 3; i++ { + user := th.CreateUser() + user.UpdateAt = model.GetMillis() + int64(100+i) + _, err = ss.User().Update(th.Context, user, true) + require.NoError(t, err) + } + + // Second sync should fail (server goes offline) + err = service.HandleSyncAllUsersForTesting(selfCluster) + require.NoError(t, err) // Method itself shouldn't error + + // Wait for second sync attempt + require.Eventually(t, func() bool { + return atomic.LoadInt32(&syncAttempts) >= 2 + }, 5*time.Second, 100*time.Millisecond) + + // Verify cursor was not updated after failed sync + cluster2, err2 := ss.RemoteCluster().Get(selfCluster.RemoteId, true) + require.NoError(t, err2) + assert.Equal(t, firstCursor, cluster2.LastGlobalUserSyncAt, "Cursor should not update when sync fails") + }) + + t.Run("Test 9: Users in Multiple Shared Channels", func(t *testing.T) { + // This test verifies that users who are members of multiple shared channels + // are synced correctly without duplication. + // The test creates 3 users and adds them to shared channels in different combinations: + // - user1: member of all 3 shared channels + // - user2: member of 2 shared channels (channel1 and channel2) + // - user3: member of 1 shared channel (channel3) + // The test then performs a global user sync and verifies that each user is synced + // exactly once, regardless of how many shared channels they belong to. + // This ensures that the global user sync deduplicates users properly + EnsureCleanState(t, th, ss) + // Note: EnsureCleanState resets batch size to 20, which is sufficient for this test + + var syncedUserIds []string + var mu sync.Mutex + var syncHandler *SelfReferentialSyncHandler + + // Create test HTTP server + testServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if syncHandler != nil { + syncHandler.HandleRequest(w, r) + } else { + writeOKResponse(w) + } + })) + defer testServer.Close() + + // Create remote cluster + selfCluster := &model.RemoteCluster{ + RemoteId: model.NewId(), + Name: "self-cluster-multi-channel", + SiteURL: testServer.URL, + CreateAt: model.GetMillis(), + LastPingAt: model.GetMillis(), + LastGlobalUserSyncAt: 0, + Token: model.NewId(), + CreatorId: th.BasicUser.Id, + RemoteToken: model.NewId(), + } + selfCluster, err = ss.RemoteCluster().Save(selfCluster) + require.NoError(t, err) + + // Initialize sync handler + syncHandler = NewSelfReferentialSyncHandler(t, service, selfCluster) + syncHandler.OnGlobalUserSync = func(userIds []string, messageNumber int32) { + mu.Lock() + syncedUserIds = append(syncedUserIds, userIds...) + mu.Unlock() + } + + // Create users + user1 := th.CreateUser() + user2 := th.CreateUser() + user3 := th.CreateUser() + + // Add users to team + th.LinkUserToTeam(user1, th.BasicTeam) + th.LinkUserToTeam(user2, th.BasicTeam) + th.LinkUserToTeam(user3, th.BasicTeam) + + // Update timestamps + user1.UpdateAt = model.GetMillis() + user2.UpdateAt = model.GetMillis() + 1 + user3.UpdateAt = model.GetMillis() + 2 + _, err = ss.User().Update(th.Context, user1, true) + require.NoError(t, err) + _, err = ss.User().Update(th.Context, user2, true) + require.NoError(t, err) + _, err = ss.User().Update(th.Context, user3, true) + require.NoError(t, err) + + // Create multiple shared channels + channel1 := th.CreateChannel(th.Context, th.BasicTeam) + channel2 := th.CreateChannel(th.Context, th.BasicTeam) + channel3 := th.CreateChannel(th.Context, th.BasicTeam) + + // Make channels shared + sc1 := &model.SharedChannel{ + ChannelId: channel1.Id, + TeamId: channel1.TeamId, + RemoteId: selfCluster.RemoteId, + Home: true, + ReadOnly: false, + ShareName: channel1.Name, + ShareDisplayName: channel1.DisplayName, + CreatorId: th.BasicUser.Id, + } + _, err = ss.SharedChannel().Save(sc1) + require.NoError(t, err) + + sc2 := &model.SharedChannel{ + ChannelId: channel2.Id, + TeamId: channel2.TeamId, + RemoteId: selfCluster.RemoteId, + Home: true, + ReadOnly: false, + ShareName: channel2.Name, + ShareDisplayName: channel2.DisplayName, + CreatorId: th.BasicUser.Id, + } + _, err = ss.SharedChannel().Save(sc2) + require.NoError(t, err) + + sc3 := &model.SharedChannel{ + ChannelId: channel3.Id, + TeamId: channel3.TeamId, + RemoteId: selfCluster.RemoteId, + Home: true, + ReadOnly: false, + ShareName: channel3.Name, + ShareDisplayName: channel3.DisplayName, + CreatorId: th.BasicUser.Id, + } + _, err = ss.SharedChannel().Save(sc3) + require.NoError(t, err) + + // Add users to multiple shared channels + // user1 in all channels + th.AddUserToChannel(user1, channel1) + th.AddUserToChannel(user1, channel2) + th.AddUserToChannel(user1, channel3) + + // user2 in two channels + th.AddUserToChannel(user2, channel1) + th.AddUserToChannel(user2, channel2) + + // user3 in one channel + th.AddUserToChannel(user3, channel3) + + // Start the sync - this will trigger the first batch + err = service.HandleSyncAllUsersForTesting(selfCluster) + require.NoError(t, err) + + // With batch size of 20, all users should sync in one batch + // Wait for sync to complete + require.Eventually(t, func() bool { + mu.Lock() + defer mu.Unlock() + + // Check if all our test users have been synced + syncedMap := make(map[string]bool) + for _, userId := range syncedUserIds { + syncedMap[userId] = true + } + + // We need all 3 test users to be synced + return syncedMap[user1.Id] && syncedMap[user2.Id] && syncedMap[user3.Id] + }, 10*time.Second, 100*time.Millisecond, "Expected all test users to be synced") + + // Verify each user is synced exactly once + mu.Lock() + userCount := make(map[string]int) + for _, userId := range syncedUserIds { + userCount[userId]++ + } + mu.Unlock() + + // Each user should appear exactly once regardless of how many channels they're in + assert.Equal(t, 1, userCount[user1.Id], "User1 should be synced exactly once") + assert.Equal(t, 1, userCount[user2.Id], "User2 should be synced exactly once") + assert.Equal(t, 1, userCount[user3.Id], "User3 should be synced exactly once") + }) + + t.Run("Test 10: Circular Sync Prevention After Connection Reset", func(t *testing.T) { + // This test verifies the exact scenario: A→B sync, connection reset, B→A sync prevention + // 1. User from Server A syncs to Server B (user appears on B with RemoteId=A) + // 2. Connection is closed and a new one is created + // 3. Server B attempts to sync back to Server A + // 4. Verify the synced user (user:A) does NOT get synced back to A + EnsureCleanState(t, th, ss) + + var syncedToB []string + var syncedBackToA []string + var mu sync.Mutex + + // Create test HTTP servers for both "servers" + serverAHandler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/api/v4/remotecluster/msg" { + // Parse message to track what gets synced back to A + bodyBytes, _ := io.ReadAll(r.Body) + var frame model.RemoteClusterFrame + if unmarshalErr := json.Unmarshal(bodyBytes, &frame); unmarshalErr == nil { + var syncMsg model.SyncMsg + if unmarshalErr := json.Unmarshal(frame.Msg.Payload, &syncMsg); unmarshalErr == nil { + mu.Lock() + for userID := range syncMsg.Users { + syncedBackToA = append(syncedBackToA, userID) + } + mu.Unlock() + } + } + } + writeOKResponse(w) + }) + + serverBHandler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/api/v4/remotecluster/msg" { + // Parse message to track what gets synced to B + bodyBytes, _ := io.ReadAll(r.Body) + var frame model.RemoteClusterFrame + if unmarshalErr := json.Unmarshal(bodyBytes, &frame); unmarshalErr == nil { + var syncMsg model.SyncMsg + if unmarshalErr := json.Unmarshal(frame.Msg.Payload, &syncMsg); unmarshalErr == nil { + mu.Lock() + for userID := range syncMsg.Users { + syncedToB = append(syncedToB, userID) + } + mu.Unlock() + } + } + } + writeOKResponse(w) + }) + + serverA := httptest.NewServer(serverAHandler) + serverB := httptest.NewServer(serverBHandler) + defer serverA.Close() + defer serverB.Close() + + // Step 1: Create "Server A" user and sync to "Server B" + originalUser := th.CreateUser() + originalUser.UpdateAt = model.GetMillis() + _, err = ss.User().Update(th.Context, originalUser, true) + require.NoError(t, err) + + // Create remote cluster B (from A's perspective) + clusterB := &model.RemoteCluster{ + RemoteId: model.NewId(), + Name: "server-b", + SiteURL: serverB.URL, + CreateAt: model.GetMillis(), + LastPingAt: model.GetMillis(), + LastGlobalUserSyncAt: 0, + Token: model.NewId(), + CreatorId: th.BasicUser.Id, + RemoteToken: model.NewId(), + } + clusterB, err = ss.RemoteCluster().Save(clusterB) + require.NoError(t, err) + + // Sync A→B (original user syncs to B) + err = service.HandleSyncAllUsersForTesting(clusterB) + require.NoError(t, err) + + // Wait for sync to complete + require.Eventually(t, func() bool { + mu.Lock() + defer mu.Unlock() + for _, userID := range syncedToB { + if userID == originalUser.Id { + return true + } + } + return false + }, 5*time.Second, 100*time.Millisecond, "Original user should sync from A to B") + + // Step 2: Simulate the synced user existing on Server B + // Create a user on "Server A" that represents what would exist on B after sync + // This user has RemoteId pointing to A, simulating user:A on Server B + syncedUserOnB := &model.User{ + Email: model.NewId() + "@example.com", + Username: originalUser.Username + "_" + clusterB.Name, // Munged username + Password: "password", + RemoteId: &clusterB.RemoteId, // This would be A's cluster ID on the actual B server + UpdateAt: model.GetMillis(), + } + syncedUserOnB, appErr := th.App.CreateUser(th.Context, syncedUserOnB) + require.Nil(t, appErr) + + // Step 3: Simulate connection reset by creating a new cluster A (from B's perspective) + // This represents B trying to sync back to A after connection reset + clusterA := &model.RemoteCluster{ + RemoteId: clusterB.RemoteId, // Same ID as the one referenced in syncedUserOnB.RemoteId + Name: "server-a", + SiteURL: serverA.URL, + CreateAt: model.GetMillis(), + LastPingAt: model.GetMillis(), + LastGlobalUserSyncAt: 0, + Token: model.NewId(), + CreatorId: th.BasicUser.Id, + RemoteToken: model.NewId(), + } + clusterA, err = ss.RemoteCluster().Save(clusterA) + require.NoError(t, err) + + // Step 4: Attempt B→A sync (should NOT sync the user back to A) + err = service.HandleSyncAllUsersForTesting(clusterA) + require.NoError(t, err) + + // Step 5: Verify the synced user was NOT sent back to A + // Use Never to ensure the user is never synced back + require.Never(t, func() bool { + mu.Lock() + defer mu.Unlock() + for _, userID := range syncedBackToA { + if userID == syncedUserOnB.Id { + return true + } + } + return false + }, 2*time.Second, 100*time.Millisecond, "Synced user should NEVER be synced back to its originating cluster") + + // Verify that the synced user still exists locally but wasn't synced + user, appErr := th.App.GetUser(syncedUserOnB.Id) + require.Nil(t, appErr) + assert.NotNil(t, user.RemoteId, "Synced user should still have RemoteId") + assert.Equal(t, clusterB.RemoteId, *user.RemoteId, "RemoteId should point to origin cluster") + }) + + t.Run("Test 12: Database Error Handling", func(t *testing.T) { + // This test verifies proper error handling when database operations fail: + // - Sync should fail gracefully + // - Cursor should not be updated + // - Error should be logged appropriately + EnsureCleanState(t, th, ss) + + // For this test, we'll simulate a database error by creating a remote cluster + // with invalid data that will cause issues during sync + testServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + writeOKResponse(w) + })) + defer testServer.Close() + + // Create remote cluster + selfCluster := &model.RemoteCluster{ + RemoteId: model.NewId(), + Name: "self-cluster-db-error", + SiteURL: testServer.URL, + CreateAt: model.GetMillis(), + LastPingAt: model.GetMillis(), + LastGlobalUserSyncAt: 0, + Token: model.NewId(), + CreatorId: th.BasicUser.Id, + RemoteToken: model.NewId(), + } + selfCluster, err = ss.RemoteCluster().Save(selfCluster) + require.NoError(t, err) + + // Create a user + user := th.CreateUser() + user.UpdateAt = model.GetMillis() + _, err = ss.User().Update(th.Context, user, true) + require.NoError(t, err) + + // To simulate a database error, we'll set an extremely large cursor value + // that will cause issues when querying users + selfCluster.LastGlobalUserSyncAt = 9223372036854775807 // Max int64 + _, err = ss.RemoteCluster().Update(selfCluster) + require.NoError(t, err) + + // Attempt sync - it should handle the error gracefully + err = service.HandleSyncAllUsersForTesting(selfCluster) + // The sync itself might not return an error, but it should handle any internal errors gracefully + // We're mainly testing that it doesn't panic or corrupt data + + // Verify the cursor wasn't corrupted + updatedCluster, err := ss.RemoteCluster().Get(selfCluster.RemoteId, true) + require.NoError(t, err) + assert.Equal(t, int64(9223372036854775807), updatedCluster.LastGlobalUserSyncAt, "Cursor should remain unchanged on error") + + // Reset cursor to a valid value + selfCluster.LastGlobalUserSyncAt = 0 + _, err = ss.RemoteCluster().Update(selfCluster) + require.NoError(t, err) + + // Now sync should work normally + err = service.HandleSyncAllUsersForTesting(selfCluster) + require.NoError(t, err) + + // Verify cursor was updated after successful sync + finalCluster, err := ss.RemoteCluster().Get(selfCluster.RemoteId, true) + require.NoError(t, err) + assert.Greater(t, finalCluster.LastGlobalUserSyncAt, int64(0), "Cursor should update after successful sync") + }) +} diff --git a/server/channels/app/shared_channel_sync_self_referential_utils_test.go b/server/channels/app/shared_channel_sync_self_referential_utils_test.go new file mode 100644 index 0000000000..411ba70a11 --- /dev/null +++ b/server/channels/app/shared_channel_sync_self_referential_utils_test.go @@ -0,0 +1,248 @@ +// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved. +// See LICENSE.txt for license information. + +package app + +import ( + "encoding/json" + "io" + "net/http" + "sync/atomic" + "testing" + "time" + + "github.com/mattermost/mattermost/server/public/model" + "github.com/mattermost/mattermost/server/v8/channels/store" + "github.com/mattermost/mattermost/server/v8/platform/services/remotecluster" + "github.com/mattermost/mattermost/server/v8/platform/services/sharedchannel" + "github.com/stretchr/testify/require" +) + +// writeOKResponse writes a standard OK JSON response in the format expected by remotecluster +func writeOKResponse(w http.ResponseWriter) { + response := &remotecluster.Response{ + Status: "OK", + Err: "", + } + + // Set empty sync response as payload + syncResp := &model.SyncResponse{} + _ = response.SetPayload(syncResp) + + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + respBytes, _ := json.Marshal(response) + _, _ = w.Write(respBytes) +} + +// SelfReferentialSyncHandler handles incoming sync messages for self-referential tests. +// It processes the messages, updates cursors, and returns proper responses. +type SelfReferentialSyncHandler struct { + t *testing.T + service *sharedchannel.Service + selfCluster *model.RemoteCluster + syncMessageCount *int32 + + // Callbacks for capturing sync data + OnIndividualSync func(userId string, messageNumber int32) + OnBatchSync func(userIds []string, messageNumber int32) + OnGlobalUserSync func(userIds []string, messageNumber int32) +} + +// NewSelfReferentialSyncHandler creates a new handler for processing sync messages in tests +func NewSelfReferentialSyncHandler(t *testing.T, service *sharedchannel.Service, selfCluster *model.RemoteCluster) *SelfReferentialSyncHandler { + count := int32(0) + return &SelfReferentialSyncHandler{ + t: t, + service: service, + selfCluster: selfCluster, + syncMessageCount: &count, + } +} + +// HandleRequest processes incoming HTTP requests for the test server. +// This handler includes common remote cluster endpoints to simulate a real remote cluster: +// - /api/v4/remotecluster/msg: Main sync message endpoint +// - /api/v4/remotecluster/ping: Ping endpoint to maintain online status (prevents offline after 5 minutes) +// - /api/v4/remotecluster/confirm_invite: Invitation confirmation endpoint +func (h *SelfReferentialSyncHandler) HandleRequest(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/api/v4/remotecluster/msg": + currentCall := atomic.AddInt32(h.syncMessageCount, 1) + + // Read and process the sync message + body, _ := io.ReadAll(r.Body) + + // The message is wrapped in a RemoteClusterFrame + var frame model.RemoteClusterFrame + + err := json.Unmarshal(body, &frame) + if err == nil { + // Process the message to update cursor + response := &remotecluster.Response{} + processErr := h.service.OnReceiveSyncMessageForTesting(frame.Msg, h.selfCluster, response) + if processErr != nil { + response.Status = "ERROR" + response.Err = processErr.Error() + } else { + // Success - build a proper sync response + response.Status = "OK" + response.Err = "" + + var syncMsg model.SyncMsg + if unmarshalErr := json.Unmarshal(frame.Msg.Payload, &syncMsg); unmarshalErr == nil { + syncResp := &model.SyncResponse{} + + // Handle global user sync + if len(syncMsg.Users) > 0 { + userIds := make([]string, 0, len(syncMsg.Users)) + for userId := range syncMsg.Users { + userIds = append(userIds, userId) + syncResp.UsersSyncd = append(syncResp.UsersSyncd, userId) + } + if h.OnGlobalUserSync != nil { + h.OnGlobalUserSync(userIds, currentCall) + } + } + + _ = response.SetPayload(syncResp) + } + } + + // Send the proper response + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + respBytes, _ := json.Marshal(response) + _, _ = w.Write(respBytes) + return + } + + writeOKResponse(w) + + case "/api/v4/remotecluster/ping": + writeOKResponse(w) + + case "/api/v4/remotecluster/confirm_invite": + writeOKResponse(w) + + default: + writeOKResponse(w) + } +} + +// GetSyncMessageCount returns the current count of sync messages received +func (h *SelfReferentialSyncHandler) GetSyncMessageCount() int32 { + return atomic.LoadInt32(h.syncMessageCount) +} + +// GetUsersFromSyncMsg extracts user IDs from a sync message +func GetUsersFromSyncMsg(msg model.SyncMsg) []string { + var userIds []string + + // Extract from users field + for userId := range msg.Users { + userIds = append(userIds, userId) + } + + return userIds +} + +// EnsureCleanState ensures a clean test state by removing all shared channels, remote clusters, +// and extra team/channel members. This helps prevent state pollution between tests. +func EnsureCleanState(t *testing.T, th *TestHelper, ss store.Store) { + t.Helper() + + // Clear all shared channels and remotes from previous tests + allSharedChannels, _ := ss.SharedChannel().GetAll(0, 1000, model.SharedChannelFilterOpts{}) + for _, sc := range allSharedChannels { + // Delete all remotes for this channel + remotes, _ := ss.SharedChannel().GetRemotes(0, 100, model.SharedChannelRemoteFilterOpts{ChannelId: sc.ChannelId}) + for _, remote := range remotes { + _, _ = ss.SharedChannel().DeleteRemote(remote.Id) + } + // Delete the shared channel + _, _ = ss.SharedChannel().Delete(sc.ChannelId) + } + + // Delete all remote clusters + allRemoteClusters, _ := ss.RemoteCluster().GetAll(0, 1000, model.RemoteClusterQueryFilter{}) + for _, rc := range allRemoteClusters { + _, _ = ss.RemoteCluster().Delete(rc.RemoteId) + } + + // Remove all channel members from test channels (except the basic team/channel setup) + channels, _ := ss.Channel().GetAll(th.BasicTeam.Id) + for _, channel := range channels { + // Skip direct message and group channels, and skip the default channels + if channel.Type != model.ChannelTypeDirect && channel.Type != model.ChannelTypeGroup && + channel.Id != th.BasicChannel.Id { + members, _ := ss.Channel().GetMembers(channel.Id, 0, 10000) + for _, member := range members { + _ = ss.Channel().RemoveMember(th.Context, channel.Id, member.UserId) + } + } + } + + // Remove all users from teams except the basic test users + teams, _ := ss.Team().GetAll() + for _, team := range teams { + if team.Id == th.BasicTeam.Id { + members, _ := ss.Team().GetMembers(team.Id, 0, 10000, nil) + for _, member := range members { + // Keep only the basic test users + if member.UserId != th.BasicUser.Id && member.UserId != th.BasicUser2.Id && + member.UserId != th.SystemAdminUser.Id { + _ = ss.Team().RemoveMember(th.Context, team.Id, member.UserId) + } + } + } + } + + // Get all active users and deactivate non-basic ones + options := &model.UserGetOptions{ + Page: 0, + PerPage: 200, + Active: true, + } + users, _ := ss.User().GetAllProfiles(options) + for _, user := range users { + // Keep only the basic test users active + if user.Id != th.BasicUser.Id && user.Id != th.BasicUser2.Id && + user.Id != th.SystemAdminUser.Id { + // Deactivate the user (soft delete) + user.DeleteAt = model.GetMillis() + _, _ = ss.User().Update(th.Context, user, true) + } + } + + // Verify cleanup is complete + require.Eventually(t, func() bool { + sharedChannels, _ := ss.SharedChannel().GetAll(0, 1000, model.SharedChannelFilterOpts{}) + remoteClusters, _ := ss.RemoteCluster().GetAll(0, 1000, model.RemoteClusterQueryFilter{}) + return len(sharedChannels) == 0 && len(remoteClusters) == 0 + }, 2*time.Second, 100*time.Millisecond, "Failed to clean up shared channels and remote clusters") + + // Reset batch size to default to ensure test isolation + defaultBatchSize := 20 + th.App.UpdateConfig(func(cfg *model.Config) { + cfg.ConnectedWorkspacesSettings.GlobalUserSyncBatchSize = &defaultBatchSize + }) + + // Ensure services are running and ready + scsInterface := th.App.Srv().GetSharedChannelSyncService() + if scs, ok := scsInterface.(*sharedchannel.Service); ok { + require.Eventually(t, func() bool { + return scs.Active() + }, 2*time.Second, 100*time.Millisecond, "Shared channel service should be active") + } + + rcService := th.App.Srv().GetRemoteClusterService() + if rcService != nil { + if rc, ok := rcService.(*remotecluster.Service); ok { + rc.SetActive(true) + } + require.Eventually(t, func() bool { + return rcService.Active() + }, 2*time.Second, 100*time.Millisecond, "Remote cluster service should be active") + } +} diff --git a/server/channels/db/migrations/migrations.list b/server/channels/db/migrations/migrations.list index bd8cb7e301..d43281e6c4 100644 --- a/server/channels/db/migrations/migrations.list +++ b/server/channels/db/migrations/migrations.list @@ -273,6 +273,8 @@ channels/db/migrations/mysql/000137_update_attribute_view.down.sql channels/db/migrations/mysql/000137_update_attribute_view.up.sql channels/db/migrations/mysql/000138_add_default_category_name_to_channel.down.sql channels/db/migrations/mysql/000138_add_default_category_name_to_channel.up.sql +channels/db/migrations/mysql/000139_remoteclusters_add_last_global_user_sync_at.down.sql +channels/db/migrations/mysql/000139_remoteclusters_add_last_global_user_sync_at.up.sql channels/db/migrations/postgres/000001_create_teams.down.sql channels/db/migrations/postgres/000001_create_teams.up.sql channels/db/migrations/postgres/000002_create_team_members.down.sql @@ -547,3 +549,5 @@ channels/db/migrations/postgres/000137_update_attribute_view.down.sql channels/db/migrations/postgres/000137_update_attribute_view.up.sql channels/db/migrations/postgres/000138_add_default_category_name_to_channel.down.sql channels/db/migrations/postgres/000138_add_default_category_name_to_channel.up.sql +channels/db/migrations/postgres/000139_remoteclusters_add_last_global_user_sync_at.down.sql +channels/db/migrations/postgres/000139_remoteclusters_add_last_global_user_sync_at.up.sql diff --git a/server/channels/db/migrations/mysql/000139_remoteclusters_add_last_global_user_sync_at.down.sql b/server/channels/db/migrations/mysql/000139_remoteclusters_add_last_global_user_sync_at.down.sql new file mode 100644 index 0000000000..1db26728db --- /dev/null +++ b/server/channels/db/migrations/mysql/000139_remoteclusters_add_last_global_user_sync_at.down.sql @@ -0,0 +1,14 @@ +SET @preparedStatement = (SELECT IF( + ( + SELECT COUNT(*) FROM INFORMATION_SCHEMA.COLUMNS + WHERE TABLE_SCHEMA = DATABASE() + AND TABLE_NAME = 'RemoteClusters' + AND COLUMN_NAME = 'LastGlobalUserSyncAt' + ) > 0, + 'ALTER TABLE RemoteClusters DROP COLUMN LastGlobalUserSyncAt', + 'SELECT 1' +)); + +PREPARE alterIfExists FROM @preparedStatement; +EXECUTE alterIfExists; +DEALLOCATE PREPARE alterIfExists; \ No newline at end of file diff --git a/server/channels/db/migrations/mysql/000139_remoteclusters_add_last_global_user_sync_at.up.sql b/server/channels/db/migrations/mysql/000139_remoteclusters_add_last_global_user_sync_at.up.sql new file mode 100644 index 0000000000..2eae2a93e5 --- /dev/null +++ b/server/channels/db/migrations/mysql/000139_remoteclusters_add_last_global_user_sync_at.up.sql @@ -0,0 +1,14 @@ +SET @preparedStatement = (SELECT IF( + ( + SELECT COUNT(*) FROM INFORMATION_SCHEMA.COLUMNS + WHERE TABLE_SCHEMA = DATABASE() + AND TABLE_NAME = 'RemoteClusters' + AND COLUMN_NAME = 'LastGlobalUserSyncAt' + ) > 0, + 'SELECT 1', + 'ALTER TABLE RemoteClusters ADD COLUMN LastGlobalUserSyncAt bigint DEFAULT 0' +)); + +PREPARE alterIfNotExists FROM @preparedStatement; +EXECUTE alterIfNotExists; +DEALLOCATE PREPARE alterIfNotExists; \ No newline at end of file diff --git a/server/channels/db/migrations/postgres/000139_remoteclusters_add_last_global_user_sync_at.down.sql b/server/channels/db/migrations/postgres/000139_remoteclusters_add_last_global_user_sync_at.down.sql new file mode 100644 index 0000000000..3cf88121d5 --- /dev/null +++ b/server/channels/db/migrations/postgres/000139_remoteclusters_add_last_global_user_sync_at.down.sql @@ -0,0 +1 @@ +ALTER TABLE remoteclusters DROP COLUMN IF EXISTS lastglobalusersyncat; \ No newline at end of file diff --git a/server/channels/db/migrations/postgres/000139_remoteclusters_add_last_global_user_sync_at.up.sql b/server/channels/db/migrations/postgres/000139_remoteclusters_add_last_global_user_sync_at.up.sql new file mode 100644 index 0000000000..8da902c220 --- /dev/null +++ b/server/channels/db/migrations/postgres/000139_remoteclusters_add_last_global_user_sync_at.up.sql @@ -0,0 +1 @@ +ALTER TABLE remoteclusters ADD COLUMN IF NOT EXISTS lastglobalusersyncat bigint DEFAULT 0; \ No newline at end of file diff --git a/server/channels/store/retrylayer/retrylayer.go b/server/channels/store/retrylayer/retrylayer.go index e80e83c994..ba46a9c0d8 100644 --- a/server/channels/store/retrylayer/retrylayer.go +++ b/server/channels/store/retrylayer/retrylayer.go @@ -9998,6 +9998,27 @@ func (s *RetryLayerRemoteClusterStore) Update(rc *model.RemoteCluster) (*model.R } +func (s *RetryLayerRemoteClusterStore) UpdateLastGlobalUserSyncAt(remoteID string, syncAt int64) error { + + tries := 0 + for { + err := s.RemoteClusterStore.UpdateLastGlobalUserSyncAt(remoteID, syncAt) + 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 *RetryLayerRemoteClusterStore) UpdateTopics(remoteClusterID string, topics string) (*model.RemoteCluster, error) { tries := 0 diff --git a/server/channels/store/sqlstore/remote_cluster_store.go b/server/channels/store/sqlstore/remote_cluster_store.go index e44db7718a..8c6299ad1c 100644 --- a/server/channels/store/sqlstore/remote_cluster_store.go +++ b/server/channels/store/sqlstore/remote_cluster_store.go @@ -44,6 +44,7 @@ func remoteClusterFields(prefix string) []string { prefix + "CreatorId", prefix + "PluginID", prefix + "Options", + prefix + "LastGlobalUserSyncAt", } } @@ -99,7 +100,8 @@ func (s sqlRemoteClusterStore) Update(remoteCluster *model.RemoteCluster) (*mode DefaultTeamId = :DefaultTeamId, Topics = :Topics, PluginID = :PluginID, - Options = :Options + Options = :Options, + LastGlobalUserSyncAt = :LastGlobalUserSyncAt WHERE RemoteId = :RemoteId AND Name = :Name` if _, err := s.GetMaster().NamedExec(query, remoteCluster); err != nil { @@ -310,3 +312,24 @@ func (s sqlRemoteClusterStore) SetLastPingAt(remoteClusterId string) error { } return nil } + +func (s sqlRemoteClusterStore) UpdateLastGlobalUserSyncAt(remoteID string, syncAt int64) error { + query := s.getQueryBuilder(). + Update("RemoteClusters"). + Set("LastGlobalUserSyncAt", syncAt). + Where(sq.Eq{"RemoteId": remoteID}) + + result, err := s.GetMaster().ExecBuilder(query) + if err != nil { + return errors.Wrap(err, "failed to update LastGlobalUserSyncAt for RemoteCluster") + } + + count, err := result.RowsAffected() + if err != nil { + return errors.Wrap(err, "failed to determine rows affected") + } + if count == 0 { + return fmt.Errorf("remote cluster not found: %s", remoteID) + } + return nil +} diff --git a/server/channels/store/sqlstore/user_store.go b/server/channels/store/sqlstore/user_store.go index 76610f2759..5116f04da7 100644 --- a/server/channels/store/sqlstore/user_store.go +++ b/server/channels/store/sqlstore/user_store.go @@ -584,8 +584,15 @@ func (us SqlUserStore) GetEtagForAllProfiles() string { func (us SqlUserStore) GetAllProfiles(options *model.UserGetOptions) ([]*model.User, error) { isPostgreSQL := us.DriverName() == model.DatabaseDriverPostgres + + // Determine ordering based on Sort option - default to Username ASC for backwards compatibility + orderBy := "Users.Username ASC" + if options.Sort == "update_at_asc" { + orderBy = "Users.UpdateAt ASC" + } + query := us.usersQuery. - OrderBy("Users.Username ASC"). + OrderBy(orderBy). Offset(uint64(options.Page * options.PerPage)).Limit(uint64(options.PerPage)) query = applyViewRestrictionsFilter(query, options.ViewRestrictions, true) @@ -599,6 +606,10 @@ func (us SqlUserStore) GetAllProfiles(options *model.UserGetOptions) ([]*model.U query = query.Where("Users.DeleteAt = 0") } + if options.UpdatedAfter > 0 { + query = query.Where(sq.Gt{"Users.UpdateAt": options.UpdatedAfter}) + } + users := []*model.User{} if err := us.GetReplica().SelectBuilder(&users, query); err != nil { return nil, errors.Wrap(err, "failed to get User profiles") diff --git a/server/channels/store/store.go b/server/channels/store/store.go index b33fa2f0e5..a928c00fe4 100644 --- a/server/channels/store/store.go +++ b/server/channels/store/store.go @@ -562,6 +562,7 @@ type RemoteClusterStore interface { GetAll(offset, limit int, filter model.RemoteClusterQueryFilter) ([]*model.RemoteCluster, error) UpdateTopics(remoteClusterID string, topics string) (*model.RemoteCluster, error) SetLastPingAt(remoteClusterID string) error + UpdateLastGlobalUserSyncAt(remoteID string, syncAt int64) error } type ComplianceStore interface { diff --git a/server/channels/store/storetest/mocks/RemoteClusterStore.go b/server/channels/store/storetest/mocks/RemoteClusterStore.go index 6c9be5fc11..4d5d220bb9 100644 --- a/server/channels/store/storetest/mocks/RemoteClusterStore.go +++ b/server/channels/store/storetest/mocks/RemoteClusterStore.go @@ -210,6 +210,24 @@ func (_m *RemoteClusterStore) Update(rc *model.RemoteCluster) (*model.RemoteClus return r0, r1 } +// UpdateLastGlobalUserSyncAt provides a mock function with given fields: remoteID, syncAt +func (_m *RemoteClusterStore) UpdateLastGlobalUserSyncAt(remoteID string, syncAt int64) error { + ret := _m.Called(remoteID, syncAt) + + if len(ret) == 0 { + panic("no return value specified for UpdateLastGlobalUserSyncAt") + } + + var r0 error + if rf, ok := ret.Get(0).(func(string, int64) error); ok { + r0 = rf(remoteID, syncAt) + } else { + r0 = ret.Error(0) + } + + return r0 +} + // UpdateTopics provides a mock function with given fields: remoteClusterID, topics func (_m *RemoteClusterStore) UpdateTopics(remoteClusterID string, topics string) (*model.RemoteCluster, error) { ret := _m.Called(remoteClusterID, topics) diff --git a/server/channels/store/storetest/shared_channel_store.go b/server/channels/store/storetest/shared_channel_store.go index a4f8dfc2ad..30a2c079c6 100644 --- a/server/channels/store/storetest/shared_channel_store.go +++ b/server/channels/store/storetest/shared_channel_store.go @@ -32,6 +32,7 @@ func TestSharedChannelStore(t *testing.T, rctx request.CTX, ss store.Store, s Sq t.Run("HasRemote", func(t *testing.T) { testHasRemote(t, rctx, ss) }) t.Run("GetRemoteForUser", func(t *testing.T) { testGetRemoteForUser(t, rctx, ss) }) t.Run("UpdateSharedChannelRemoteNextSyncAt", func(t *testing.T) { testUpdateSharedChannelRemoteCursor(t, rctx, ss) }) + t.Run("UpdateGlobalUserSyncCursor", func(t *testing.T) { testUpdateGlobalUserSyncCursor(t, rctx, ss) }) t.Run("DeleteSharedChannelRemote", func(t *testing.T) { testDeleteSharedChannelRemote(t, rctx, ss) }) t.Run("SaveSharedChannelUser", func(t *testing.T) { testSaveSharedChannelUser(t, rctx, ss) }) @@ -932,6 +933,38 @@ func testUpdateSharedChannelRemoteCursor(t *testing.T, rctx request.CTX, ss stor }) } +func testUpdateGlobalUserSyncCursor(t *testing.T, rctx request.CTX, ss store.Store) { + // Create a remote cluster first + rc := &model.RemoteCluster{ + RemoteId: model.NewId(), + SiteURL: "http://example.com", + CreatorId: model.NewId(), + Name: "test", + } + rcSaved, err := ss.RemoteCluster().Save(rc) + require.NoError(t, err, "couldn't save remote cluster", err) + + futureTimestamp := model.GetMillis() + 3600000 // 1 hour in the future + + t.Run("Update global user sync cursor for remote", func(t *testing.T) { + err := ss.RemoteCluster().UpdateLastGlobalUserSyncAt(rcSaved.RemoteId, futureTimestamp) + require.NoError(t, err, "update global user sync cursor should not error", err) + + // Verify that the LastGlobalUserSyncAt field was updated in the RemoteCluster table + // Small sleep to ensure the transaction is committed + time.Sleep(10 * time.Millisecond) + updatedRC, err := ss.RemoteCluster().Get(rcSaved.RemoteId, false) + require.NoError(t, err) + require.NotZero(t, updatedRC.LastGlobalUserSyncAt, "LastGlobalUserSyncAt should not be zero") + require.Equal(t, futureTimestamp, updatedRC.LastGlobalUserSyncAt) + }) + + t.Run("Update global user sync cursor for non-existent remote", func(t *testing.T) { + err := ss.RemoteCluster().UpdateLastGlobalUserSyncAt(model.NewId(), futureTimestamp) + require.Error(t, err, "update non-existent remote should error", err) + }) +} + func testDeleteSharedChannelRemote(t *testing.T, rctx request.CTX, ss store.Store) { channel, err := createTestChannel(ss, rctx, "test_remote_delete") require.NoError(t, err) diff --git a/server/channels/store/storetest/user_store.go b/server/channels/store/storetest/user_store.go index 39f02d00c3..524694492c 100644 --- a/server/channels/store/storetest/user_store.go +++ b/server/channels/store/storetest/user_store.go @@ -610,6 +610,32 @@ func testUserStoreGetAllProfiles(t *testing.T, rctx request.CTX, ss store.Store) }, actual) }) + t.Run("filter by UpdatedAfter", func(t *testing.T) { + // Update a user to ensure we have a recent update time + updateTime := model.GetMillis() + u2.FirstName = "Updated" + _, updateErr := ss.User().Update(rctx, u2, false) + require.NoError(t, updateErr) + + // Query with the UpdatedAfter filter + actual, userErr := ss.User().GetAllProfiles(&model.UserGetOptions{ + Page: 0, + PerPage: 10, + UpdatedAfter: updateTime - 1, // Subtract 1 to ensure we capture the update + }) + require.NoError(t, userErr) + require.Contains(t, actual, sanitized(u2), "User updated after the specified time should be in the results") + + // Query with a future time, should return no results + actual, userErr = ss.User().GetAllProfiles(&model.UserGetOptions{ + Page: 0, + PerPage: 10, + UpdatedAfter: updateTime + 10000, // Future time + }) + require.NoError(t, userErr) + require.NotContains(t, actual, sanitized(u2), "Users updated before the specified future time should not be in the results") + }) + u8, err := ss.User().Save(rctx, &model.User{ Email: MakeEmail(), Username: "u8" + model.NewId(), diff --git a/server/channels/store/timerlayer/timerlayer.go b/server/channels/store/timerlayer/timerlayer.go index 884b70c2e9..90450bfbbd 100644 --- a/server/channels/store/timerlayer/timerlayer.go +++ b/server/channels/store/timerlayer/timerlayer.go @@ -7909,6 +7909,22 @@ func (s *TimerLayerRemoteClusterStore) Update(rc *model.RemoteCluster) (*model.R return result, err } +func (s *TimerLayerRemoteClusterStore) UpdateLastGlobalUserSyncAt(remoteID string, syncAt int64) error { + start := time.Now() + + err := s.RemoteClusterStore.UpdateLastGlobalUserSyncAt(remoteID, syncAt) + + elapsed := float64(time.Since(start)) / float64(time.Second) + if s.Root.Metrics != nil { + success := "false" + if err == nil { + success = "true" + } + s.Root.Metrics.ObserveStoreMethodDuration("RemoteClusterStore.UpdateLastGlobalUserSyncAt", success, elapsed) + } + return err +} + func (s *TimerLayerRemoteClusterStore) UpdateTopics(remoteClusterID string, topics string) (*model.RemoteCluster, error) { start := time.Now() diff --git a/server/platform/services/remotecluster/service.go b/server/platform/services/remotecluster/service.go index a52eaff4e9..b60e59579f 100644 --- a/server/platform/services/remotecluster/service.go +++ b/server/platform/services/remotecluster/service.go @@ -289,3 +289,19 @@ func (rcs *Service) pause() { rcs.server.Log().Debug("Remote Cluster Service inactive") } + +// SetActive forces the service to be active or inactive +func (rcs *Service) SetActive(active bool) { + rcs.mux.Lock() + defer rcs.mux.Unlock() + + if rcs.active == active { + return + } + + if active { + rcs.resume() + } else { + rcs.pause() + } +} diff --git a/server/platform/services/sharedchannel/service.go b/server/platform/services/sharedchannel/service.go index 7d9cbe364f..5f1212bdbd 100644 --- a/server/platform/services/sharedchannel/service.go +++ b/server/platform/services/sharedchannel/service.go @@ -24,6 +24,7 @@ const ( TopicSync = "sharedchannel_sync" TopicChannelInvite = "sharedchannel_invite" TopicUploadCreate = "sharedchannel_upload" + TopicGlobalUserSync = "sharedchannel_global_user_sync" MaxRetries = 3 MaxUsersPerSync = 25 NotifyRemoteOfflineThreshold = time.Second * 10 @@ -98,6 +99,7 @@ type Service struct { syncTopicListenerId string inviteTopicListenerId string uploadTopicListenerId string + globalSyncTopicListenerId string siteURL *url.URL } @@ -130,6 +132,7 @@ func (scs *Service) Start() error { scs.syncTopicListenerId = rcs.AddTopicListener(TopicSync, scs.onReceiveSyncMessage) scs.inviteTopicListenerId = rcs.AddTopicListener(TopicChannelInvite, scs.onReceiveChannelInvite) scs.uploadTopicListenerId = rcs.AddTopicListener(TopicUploadCreate, scs.onReceiveUploadCreate) + scs.globalSyncTopicListenerId = rcs.AddTopicListener(TopicGlobalUserSync, scs.onReceiveSyncMessage) scs.connectionStateListenerId = rcs.AddConnectionStateListener(scs.onConnectionStateChange) scs.mux.Unlock() @@ -248,6 +251,9 @@ func (scs *Service) onConnectionStateChange(rc *model.RemoteCluster, online bool // when a previously offline remote comes back online force a sync. scs.SendPendingInvitesForRemote(rc) scs.ForceSyncForRemote(rc) + + // Schedule global user sync if feature is enabled + scs.scheduleGlobalUserSync(rc) } scs.server.Log().Log(mlog.LvlSharedChannelServiceDebug, "Remote cluster connection status changed", @@ -278,3 +284,47 @@ func (scs *Service) notifyClientsForSharedChannelUpdate(channel *model.Channel) messageWs.Add("channel_id", channel.Id) scs.app.Publish(messageWs) } + +// isGlobalUserSyncEnabled checks if the global user sync feature is enabled +func (scs *Service) isGlobalUserSyncEnabled() bool { + cfg := scs.server.Config() + + return cfg.FeatureFlags.EnableSyncAllUsersForRemoteCluster || + (cfg.ConnectedWorkspacesSettings.SyncUsersOnConnectionOpen != nil && *cfg.ConnectedWorkspacesSettings.SyncUsersOnConnectionOpen) +} + +// scheduleGlobalUserSync schedules a task to sync all users with a remote cluster +func (scs *Service) scheduleGlobalUserSync(rc *model.RemoteCluster) { + if !scs.isGlobalUserSyncEnabled() { + return + } + + // Schedule the sync task + go func() { + // Create a special sync task with empty channelID + // This empty channelID is a deliberate marker for a global user sync task + task := newSyncTask("", "", rc.RemoteId, nil, nil) + task.schedule = time.Now().Add(NotifyMinimumDelay) + scs.addTask(task) + + scs.server.Log().Log(mlog.LvlSharedChannelServiceDebug, "Scheduled global user sync task for remote", + mlog.String("remote", rc.DisplayName), + mlog.String("remoteId", rc.RemoteId), + ) + }() +} + +// OnReceiveSyncMessageForTesting exposes onReceiveSyncMessage for testing +func (scs *Service) OnReceiveSyncMessageForTesting(msg model.RemoteClusterMsg, rc *model.RemoteCluster, response *remotecluster.Response) error { + return scs.onReceiveSyncMessage(msg, rc, response) +} + +// GetUserSyncBatchSizeForTesting returns the configured batch size for user syncing (exported for testing) +func (scs *Service) GetUserSyncBatchSizeForTesting() int { + return scs.getGlobalUserSyncBatchSize() +} + +// HandleSyncAllUsersForTesting exposes syncAllUsers for testing +func (scs *Service) HandleSyncAllUsersForTesting(rc *model.RemoteCluster) error { + return scs.syncAllUsers(rc) +} diff --git a/server/platform/services/sharedchannel/sync_recv.go b/server/platform/services/sharedchannel/sync_recv.go index 028c654283..2386919b88 100644 --- a/server/platform/services/sharedchannel/sync_recv.go +++ b/server/platform/services/sharedchannel/sync_recv.go @@ -24,8 +24,8 @@ var ( ) func (scs *Service) onReceiveSyncMessage(msg model.RemoteClusterMsg, rc *model.RemoteCluster, response *remotecluster.Response) error { - if msg.Topic != TopicSync { - return fmt.Errorf("wrong topic, expected `%s`, got `%s`", TopicSync, msg.Topic) + if msg.Topic != TopicSync && msg.Topic != TopicGlobalUserSync { + return fmt.Errorf("wrong topic, expected `%s` or `%s`, got `%s`", TopicSync, TopicGlobalUserSync, msg.Topic) } if len(msg.Payload) == 0 { @@ -47,6 +47,36 @@ func (scs *Service) onReceiveSyncMessage(msg model.RemoteClusterMsg, rc *model.R return scs.processSyncMessage(request.EmptyContext(scs.server.Log()), &sm, rc, response) } +func (scs *Service) processGlobalUserSync(c request.CTX, syncMsg *model.SyncMsg, rc *model.RemoteCluster, response *remotecluster.Response) error { + syncResp := model.SyncResponse{ + UserErrors: make([]string, 0), + UsersSyncd: make([]string, 0), + } + + scs.server.Log().Log(mlog.LvlSharedChannelServiceDebug, "Processing global user sync", + mlog.String("remote", rc.Name), + mlog.Int("user_count", len(syncMsg.Users)), + ) + + // Process all users in the sync message + for _, user := range syncMsg.Users { + if userSaved, err := scs.upsertSyncUser(c, user, nil, rc); err != nil { + syncResp.UserErrors = append(syncResp.UserErrors, user.Id) + } else { + syncResp.UsersSyncd = append(syncResp.UsersSyncd, userSaved.Id) + if syncResp.UsersLastUpdateAt < user.UpdateAt { + syncResp.UsersLastUpdateAt = user.UpdateAt + } + scs.server.Log().Log(mlog.LvlSharedChannelServiceDebug, "Global user upserted via sync", + mlog.String("remote", rc.Name), + mlog.String("user_id", user.Id), + ) + } + } + + return response.SetPayload(syncResp) +} + func (scs *Service) processSyncMessage(c request.CTX, syncMsg *model.SyncMsg, rc *model.RemoteCluster, response *remotecluster.Response) error { var targetChannel *model.Channel var team *model.Team @@ -68,6 +98,25 @@ func (scs *Service) processSyncMessage(c request.CTX, syncMsg *model.SyncMsg, rc mlog.Int("status_count", len(syncMsg.Statuses)), ) + // Check if this is a global user sync message (no channel ID and only users) + if syncMsg.ChannelId == "" { + if len(syncMsg.Posts) != 0 || + len(syncMsg.Reactions) != 0 || + len(syncMsg.Statuses) != 0 { + return fmt.Errorf("global user sync message should not contain posts, reactions or statuses") + } + + if len(syncMsg.Users) == 0 { + return nil + } + // Check if feature flag is enabled + if !scs.isGlobalUserSyncEnabled() { + return nil + } + return scs.processGlobalUserSync(c, syncMsg, rc, response) + } + + // For regular sync messages, we need a specific channel if targetChannel, err = scs.server.GetStore().Channel().Get(syncMsg.ChannelId, true); err != nil { // if the channel doesn't exist then none of these sync items are going to work. return fmt.Errorf("channel not found processing sync message: %w", err) @@ -239,7 +288,7 @@ func (scs *Service) upsertSyncUser(c request.CTX, user *model.User, channel *mod // Instead of undoing what succeeded on any failure we simply do all steps each // time. AddUserToChannel & AddUserToTeamByTeamId do not error if user was already // added and exit quickly. Not needed for DMs where teamId is empty. - if channel.TeamId != "" { + if channel != nil && channel.TeamId != "" { // add user to team if err := scs.app.AddUserToTeamByTeamId(request.EmptyContext(scs.server.Log()), channel.TeamId, userSaved); err != nil { return nil, fmt.Errorf("error adding sync user to Team: %w", err) diff --git a/server/platform/services/sharedchannel/sync_send.go b/server/platform/services/sharedchannel/sync_send.go index a5897633ca..9aa22e2659 100644 --- a/server/platform/services/sharedchannel/sync_send.go +++ b/server/platform/services/sharedchannel/sync_send.go @@ -571,14 +571,14 @@ func (scs *Service) shouldUserSync(user *model.User, channelID string, rc *model if _, err = scs.server.GetStore().SharedChannel().SaveUser(scu); err != nil { scs.server.Log().Log(mlog.LvlSharedChannelServiceError, "Error adding user to shared channel users", mlog.String("user_id", user.Id), - mlog.String("channel_id", user.Id), + mlog.String("channel_id", channelID), mlog.String("remote_id", rc.RemoteId), mlog.Err(err), ) } else { scs.server.Log().Log(mlog.LvlSharedChannelServiceDebug, "Added user to shared channel users", mlog.String("user_id", user.Id), - mlog.String("channel_id", user.Id), + mlog.String("channel_id", channelID), mlog.String("remote_id", rc.RemoteId), ) } diff --git a/server/platform/services/sharedchannel/sync_send_remote.go b/server/platform/services/sharedchannel/sync_send_remote.go index 9a463089d0..e72221e65a 100644 --- a/server/platform/services/sharedchannel/sync_send_remote.go +++ b/server/platform/services/sharedchannel/sync_send_remote.go @@ -39,8 +39,9 @@ type syncData struct { statuses []*model.Status attachments []attachment - resultRepeat bool - resultNextCursor model.GetPostsSinceForSyncCursor + resultRepeat bool + resultNextCursor model.GetPostsSinceForSyncCursor + GlobalUserSyncLastTimestamp int64 } func newSyncData(task syncTask, rc *model.RemoteCluster, scr *model.SharedChannelRemote) *syncData { @@ -83,6 +84,12 @@ func (sd *syncData) setDataFromMsg(msg *model.SyncMsg) { // channels are very active. // Returning an error forces a retry on the task. func (scs *Service) syncForRemote(task syncTask, rc *model.RemoteCluster) error { + // Empty channelID indicates a global user sync task + // Normal syncTasks always include a valid channelID + if task.channelID == "" { + return scs.syncAllUsers(rc) + } + rcs := scs.server.GetRemoteClusterService() if rcs == nil { return fmt.Errorf("cannot update remote cluster %s for channel id %s; Remote Cluster Service not enabled", rc.Name, task.channelID) @@ -241,6 +248,7 @@ func (scs *Service) fetchUsersForSync(sd *syncData) error { return err } + // Don't sync users back to the remote cluster they originated from for _, u := range users { if u.GetRemoteID() != sd.rc.RemoteId { sd.users[u.Id] = u @@ -495,7 +503,7 @@ func (scs *Service) filterPostsForSync(sd *syncData) { continue } - // don't sync a post back to the remote it came from. + // don't sync a post back to the remote cluster it came from. if p.GetRemoteID() == sd.rc.RemoteId { continue } @@ -578,6 +586,11 @@ func (scs *Service) sendUserSyncData(sd *syncData) error { msg.Users = sd.users err := scs.sendSyncMsgToRemote(msg, sd.rc, func(syncResp model.SyncResponse, errResp error) { + // Only update cursor on successful sync + if errResp == nil && sd.GlobalUserSyncLastTimestamp > 0 { + scs.updateGlobalSyncCursor(sd.rc, sd.GlobalUserSyncLastTimestamp) + } + for _, userID := range syncResp.UsersSyncd { if err := scs.server.GetStore().SharedChannel().UpdateUserLastSyncAt(userID, sd.task.channelID, sd.rc.RemoteId); err != nil { scs.server.Log().Log(mlog.LvlSharedChannelServiceError, "Cannot update shared channel user LastSyncAt", @@ -691,6 +704,204 @@ func (scs *Service) sendProfileImageSyncData(sd *syncData) { } } +// shouldUserSyncGlobal determines if a user needs to be synchronized globally. +// Compares user's update timestamp with the remote cluster's LastGlobalUserSyncAt. +func (scs *Service) shouldUserSyncGlobal(user *model.User, rc *model.RemoteCluster) (bool, error) { + // Don't sync users back to the remote cluster they originated from + if user.IsRemote() && user.GetRemoteID() == rc.RemoteId { + return false, nil + } + + // Calculate latest update time for this user (profile or picture) + latestUserUpdateTime := user.UpdateAt + if user.LastPictureUpdate > latestUserUpdateTime { + latestUserUpdateTime = user.LastPictureUpdate + } + + // For initial sync (LastGlobalUserSyncAt=0), sync all users + // For incremental sync, only sync users updated after the last sync + if rc.LastGlobalUserSyncAt == 0 { + return true, nil + } + return latestUserUpdateTime > rc.LastGlobalUserSyncAt, nil +} + +// syncAllUsers synchronizes all local users to a remote cluster. +// This is called when a connection with a remote cluster is established or when handling a global user sync task. +// Uses cursor-based approach with LastGlobalUserSyncAt to resume after interruptions. +func (scs *Service) syncAllUsers(rc *model.RemoteCluster) error { + // Check if feature is enabled + if !scs.server.Config().FeatureFlags.EnableSyncAllUsersForRemoteCluster { + return nil + } + + if !rc.IsOnline() { + return fmt.Errorf("remote cluster %s is not online", rc.RemoteId) + } + + // Start metrics tracking + metrics := scs.server.GetMetrics() + start := time.Now() + defer func() { + if metrics != nil { + metrics.IncrementSharedChannelsSyncCounter(rc.RemoteId) + metrics.ObserveSharedChannelsSyncCollectionDuration(rc.RemoteId, time.Since(start).Seconds()) + } + }() + + batchSize := scs.getGlobalUserSyncBatchSize() + + // Create sync data with collected users + sd := &syncData{ + task: syncTask{remoteID: rc.RemoteId}, + rc: rc, + scr: &model.SharedChannelRemote{RemoteId: rc.RemoteId}, + users: make(map[string]*model.User), + } + + // Collect users to sync + users, latestTimestamp, _, hasMore, err := scs.collectUsersForGlobalSync(rc, batchSize) + if err != nil { + return err + } + + // Exit early if no users to sync + if len(users) == 0 { + scs.server.Log().Log(mlog.LvlSharedChannelServiceDebug, "No users to sync for remote cluster", + mlog.String("remote_id", rc.RemoteId)) + return nil + } + + // Add users to sync data + sd.users = users + sd.GlobalUserSyncLastTimestamp = latestTimestamp + + // Send the collected users to remote + if err := scs.sendUserSyncData(sd); err != nil { + scs.server.Log().Log(mlog.LvlSharedChannelServiceError, "Error sending user batch during sync", + mlog.String("remote_id", rc.RemoteId), + mlog.Err(err), + ) + return fmt.Errorf("error sending user batch during sync: %w", err) + } + + // Schedule next batch if needed + if hasMore { + scs.scheduleNextUserSyncBatch(rc, latestTimestamp, batchSize, len(users)) + } + + return nil +} + +// getGlobalUserSyncBatchSize returns the configured batch size for user syncing +func (scs *Service) getGlobalUserSyncBatchSize() int { + batchSize := MaxUsersPerSync + if scs.server.Config().ConnectedWorkspacesSettings.GlobalUserSyncBatchSize != nil { + configValue := *scs.server.Config().ConnectedWorkspacesSettings.GlobalUserSyncBatchSize + if configValue > 0 && configValue <= 200 { + batchSize = configValue + } + } + return batchSize +} + +// collectUsersForGlobalSync fetches users that need to be synced to the remote +func (scs *Service) collectUsersForGlobalSync(rc *model.RemoteCluster, batchSize int) (map[string]*model.User, int64, int, bool, error) { + options := &model.UserGetOptions{ + Page: 0, + PerPage: 100, // Database fetch batch size + Active: true, + Sort: "update_at_asc", // Order by UpdateAt ASC to ensure cursor consistency + } + + // Only use UpdatedAfter for incremental syncs, not the initial sync + // This ensures users with UpdateAt=0 are included in the first sync + if rc.LastGlobalUserSyncAt > 0 { + options.UpdatedAfter = rc.LastGlobalUserSyncAt + } + + users := make(map[string]*model.User) + latestTimestamp := rc.LastGlobalUserSyncAt + totalCount := 0 + + // Page through database results + for { + batch, err := scs.server.GetStore().User().GetAllProfiles(options) + if err != nil { + scs.server.Log().Log(mlog.LvlSharedChannelServiceError, "Error fetching users for global sync", + mlog.String("remote_id", rc.RemoteId), + mlog.Err(err), + ) + return nil, 0, 0, false, err + } + + if len(batch) == 0 { + break // No more users to process + } + + totalCount += len(batch) + + // Process each user in this database page + for _, user := range batch { + // Stop if we've reached batch limit + if len(users) >= batchSize { + return users, latestTimestamp, totalCount, true, nil + } + + // Skip users from remotes + if user.IsRemote() { + continue + } + + // Check if user needs syncing + needsSync, _ := scs.shouldUserSyncGlobal(user, rc) + if !needsSync { + continue + } + + // Add user and update cursor timestamp + users[user.Id] = user + + userUpdateTime := max(user.UpdateAt, user.LastPictureUpdate) + if userUpdateTime > latestTimestamp { + latestTimestamp = userUpdateTime + } + } + + // Check if we've reached the end of results + if len(batch) < options.PerPage { + break + } + + // Move to next page + options.Page++ + } + + return users, latestTimestamp, totalCount, false, nil +} + +// updateGlobalSyncCursor updates the LastGlobalUserSyncAt value for the remote cluster +func (scs *Service) updateGlobalSyncCursor(rc *model.RemoteCluster, newTimestamp int64) { + if err := scs.server.GetStore().RemoteCluster().UpdateLastGlobalUserSyncAt(rc.RemoteId, newTimestamp); err == nil { + rc.LastGlobalUserSyncAt = newTimestamp + } else { + scs.server.Log().Log(mlog.LvlSharedChannelServiceError, "Failed to update global user sync cursor", + mlog.String("remote_id", rc.RemoteId), + mlog.Err(err), + ) + } +} + +// scheduleNextUserSyncBatch creates a new task for the next batch of user sync +func (scs *Service) scheduleNextUserSyncBatch(rc *model.RemoteCluster, timestamp int64, batchSize, processedCount int) { + // Use timestamp as userID to make each batch task unique + // This prevents task ID collisions between different batches for the same remote + timestampStr := fmt.Sprintf("%d", timestamp) + task := newSyncTask("", timestampStr, rc.RemoteId, nil, nil) + task.schedule = time.Now().Add(NotifyMinimumDelay) + scs.addTask(task) +} + // sendSyncMsgToRemote synchronously sends the sync message to the remote cluster (or plugin). func (scs *Service) sendSyncMsgToRemote(msg *model.SyncMsg, rc *model.RemoteCluster, f sendSyncMsgResultFunc) error { rcs := scs.server.GetRemoteClusterService() @@ -706,7 +917,15 @@ func (scs *Service) sendSyncMsgToRemote(msg *model.SyncMsg, rc *model.RemoteClus if err != nil { return err } - rcMsg := model.NewRemoteClusterMsg(TopicSync, b) + + // Use appropriate topic based on message type + topic := TopicSync + if msg.ChannelId == "" && len(msg.Users) > 0 && + len(msg.Posts) == 0 && len(msg.Reactions) == 0 && + len(msg.Statuses) == 0 { + topic = TopicGlobalUserSync + } + rcMsg := model.NewRemoteClusterMsg(topic, b) ctx, cancel := context.WithTimeout(context.Background(), remotecluster.SendTimeout) defer cancel() diff --git a/server/public/model/config.go b/server/public/model/config.go index 5de3651f63..3a234d0e30 100644 --- a/server/public/model/config.go +++ b/server/public/model/config.go @@ -3459,6 +3459,8 @@ type ConnectedWorkspacesSettings struct { EnableSharedChannels *bool EnableRemoteClusterService *bool DisableSharedChannelsStatusSync *bool + SyncUsersOnConnectionOpen *bool + GlobalUserSyncBatchSize *int MaxPostsPerSync *int } @@ -3483,6 +3485,14 @@ func (c *ConnectedWorkspacesSettings) SetDefaults(isUpdate bool, e ExperimentalS c.DisableSharedChannelsStatusSync = NewPointer(false) } + if c.SyncUsersOnConnectionOpen == nil { + c.SyncUsersOnConnectionOpen = NewPointer(false) + } + + if c.GlobalUserSyncBatchSize == nil { + c.GlobalUserSyncBatchSize = NewPointer(25) // Default to MaxUsersPerSync + } + if c.MaxPostsPerSync == nil { c.MaxPostsPerSync = NewPointer(ConnectedWorkspacesSettingsDefaultMaxPostsPerSync) } diff --git a/server/public/model/feature_flags.go b/server/public/model/feature_flags.go index 69d3aa93d1..6f826e767f 100644 --- a/server/public/model/feature_flags.go +++ b/server/public/model/feature_flags.go @@ -25,6 +25,9 @@ type FeatureFlags struct { // Enable plugins in shared channels. EnableSharedChannelsPlugins bool + // Enable syncing all users for remote clusters in shared channels + EnableSyncAllUsersForRemoteCluster bool + // AppsEnabled toggles the Apps framework functionalities both in server and client side AppsEnabled bool @@ -69,6 +72,7 @@ func (f *FeatureFlags) SetDefaults() { f.TestBoolFeature = false f.EnableRemoteClusterService = false f.EnableSharedChannelsDMs = false + f.EnableSyncAllUsersForRemoteCluster = false f.EnableSharedChannelsPlugins = true f.AppsEnabled = false f.NormalizeLdapDNs = false diff --git a/server/public/model/remote_cluster.go b/server/public/model/remote_cluster.go index 6d51d12d1a..68f7c362ff 100644 --- a/server/public/model/remote_cluster.go +++ b/server/public/model/remote_cluster.go @@ -53,37 +53,39 @@ func (bm *Bitmask) UnsetBit(flag Bitmask) { } type RemoteCluster struct { - RemoteId string `json:"remote_id"` - RemoteTeamId string `json:"remote_team_id"` // Deprecated: this field is no longer used. It's only kept for backwards compatibility. - Name string `json:"name"` - DisplayName string `json:"display_name"` - SiteURL string `json:"site_url"` - DefaultTeamId string `json:"default_team_id"` - CreateAt int64 `json:"create_at"` - DeleteAt int64 `json:"delete_at"` - LastPingAt int64 `json:"last_ping_at"` - Token string `json:"token"` - RemoteToken string `json:"remote_token"` - Topics string `json:"topics"` - CreatorId string `json:"creator_id"` - PluginID string `json:"plugin_id"` // non-empty when sync message are to be delivered via plugin API - Options Bitmask `json:"options"` // bit-flag set of options + RemoteId string `json:"remote_id"` + RemoteTeamId string `json:"remote_team_id"` // Deprecated: this field is no longer used. It's only kept for backwards compatibility. + Name string `json:"name"` + DisplayName string `json:"display_name"` + SiteURL string `json:"site_url"` + DefaultTeamId string `json:"default_team_id"` + CreateAt int64 `json:"create_at"` + DeleteAt int64 `json:"delete_at"` + LastPingAt int64 `json:"last_ping_at"` + LastGlobalUserSyncAt int64 `json:"last_global_user_sync_at"` // Timestamp of last global user sync + Token string `json:"token"` + RemoteToken string `json:"remote_token"` + Topics string `json:"topics"` + CreatorId string `json:"creator_id"` + PluginID string `json:"plugin_id"` // non-empty when sync message are to be delivered via plugin API + Options Bitmask `json:"options"` // bit-flag set of options } func (rc *RemoteCluster) Auditable() map[string]any { return map[string]any{ - "remote_id": rc.RemoteId, - "remote_team_id": rc.RemoteTeamId, - "name": rc.Name, - "display_name": rc.DisplayName, - "site_url": rc.SiteURL, - "default_team_id": rc.DefaultTeamId, - "create_at": rc.CreateAt, - "delete_at": rc.DeleteAt, - "last_ping_at": rc.LastPingAt, - "creator_id": rc.CreatorId, - "plugin_id": rc.PluginID, - "options": rc.Options, + "remote_id": rc.RemoteId, + "remote_team_id": rc.RemoteTeamId, + "name": rc.Name, + "display_name": rc.DisplayName, + "site_url": rc.SiteURL, + "default_team_id": rc.DefaultTeamId, + "create_at": rc.CreateAt, + "delete_at": rc.DeleteAt, + "last_ping_at": rc.LastPingAt, + "last_global_user_sync_at": rc.LastGlobalUserSyncAt, + "creator_id": rc.CreatorId, + "plugin_id": rc.PluginID, + "options": rc.Options, } } diff --git a/server/public/model/user_get.go b/server/public/model/user_get.go index 0ba62f3f06..78174b2ff1 100644 --- a/server/public/model/user_get.go +++ b/server/public/model/user_get.go @@ -40,6 +40,8 @@ type UserGetOptions struct { Page int // Page size PerPage int + // Filters the users that have been updated after the given time + UpdatedAfter int64 } type UserGetByIdsOptions struct {