MM-56906: Remove redundant calls on team switch (#30771)

On page load, we load ALL channels and channel members from all teams.
But then, on team_switch, we would again load channels and channel
members from that team. This was redundant and mainly kept
because previously the websocket events were considered unreliable.

Now with reliable websockets, and client-side pings, we can detect
broken connections faster and recover without loss.

Additionally, the getAllChannelMembers call would page through
all responses on the client side. This was inefficient and incur
extra latency. To optimize for this, we introduce server-side
streaming of the full response if page is set to -1.

This optimizes the intial response as well.

https://mattermost.atlassian.net/browse/MM-56906

```release-note
Optimize team switch operation by removing calls to get channels
and channel members.
```


Co-authored-by: Mattermost Build <build@mattermost.com>

server/channels/api4/user.go

@@ -3094,14 +3094,62 @@ func getChannelMembersForUser(c *Context, w http.ResponseWriter, r *http.Request
 		return
 	}
 
-	members, err := c.App.GetChannelMembersWithTeamDataForUserWithPagination(c.AppContext, c.Params.UserId, c.Params.Page, c.Params.PerPage)
-	if err != nil {
-		c.Err = err
+	// For backward compatibility purposes
+	if c.Params.Page != -1 {
+		cursor := &model.ChannelMemberCursor{
+			Page:    c.Params.Page,
+			PerPage: c.Params.PerPage,
+		}
+		members, err := c.App.GetChannelMembersWithTeamDataForUserWithPagination(c.AppContext, c.Params.UserId, cursor)
+		if err != nil {
+			c.Err = err
+			return
+		}
+
+		if err := json.NewEncoder(w).Encode(members); err != nil {
+			c.Logger.Warn("Error while writing response", mlog.Err(err))
+		}
 		return
 	}
 
-	if err := json.NewEncoder(w).Encode(members); err != nil {
-		c.Logger.Warn("Error while writing response", mlog.Err(err))
+	// The new model streams data using NDJSON format (each JSON object on a new line)
+	pageSize := 100
+	fromChannelID := ""
+
+	// Set the correct content type for NDJSON
+	w.Header().Set("Content-Type", "application/x-ndjson")
+
+	enc := json.NewEncoder(w)
+
+	for {
+		cursor := &model.ChannelMemberCursor{
+			Page:          -1,
+			PerPage:       pageSize,
+			FromChannelID: fromChannelID,
+		}
+
+		members, err := c.App.GetChannelMembersWithTeamDataForUserWithPagination(c.AppContext, c.Params.UserId, cursor)
+		if err != nil {
+			// If the page size was a perfect multiple of the total number of results,
+			// then the last query will always return zero results.
+			if fromChannelID != "" && err.Id == app.MissingChannelMemberError {
+				break
+			}
+			c.Err = err
+			return
+		}
+
+		for _, member := range members {
+			if err := enc.Encode(member); err != nil {
+				c.Logger.Warn("Error while writing response", mlog.Err(err))
+			}
+		}
+
+		if len(members) < pageSize {
+			break
+		}
+
+		fromChannelID = members[len(members)-1].ChannelId
 	}
 }
 

server/channels/api4/user_test.go

@@ -6664,6 +6664,24 @@ func TestGetChannelMembersWithTeamData(t *testing.T) {
 	for _, ch := range channels {
 		assert.Equal(t, th.BasicTeam.DisplayName, ch.TeamDisplayName)
 	}
+
+	channels, resp, err = th.Client.GetChannelMembersWithTeamData(context.Background(), th.BasicUser.Id, 0, 5000)
+	require.NoError(t, err)
+	CheckOKStatus(t, resp)
+	assert.Len(t, channels, 6)
+	for _, ch := range channels {
+		assert.Equal(t, th.BasicTeam.DisplayName, ch.TeamDisplayName)
+	}
+
+	// perPage doesn't matter if page=-1
+	channels, resp, err = th.Client.GetChannelMembersWithTeamData(context.Background(), th.BasicUser.Id, -1, 2)
+	require.NoError(t, err)
+	CheckOKStatus(t, resp)
+	assert.Equal(t, "application/x-ndjson", resp.Header.Get("Content-Type"))
+	assert.Len(t, channels, 6)
+	for _, ch := range channels {
+		assert.Equal(t, th.BasicTeam.DisplayName, ch.TeamDisplayName)
+	}
 }
 
 func TestMigrateAuthToLDAP(t *testing.T) {

server/channels/app/channel.go

@@ -2174,10 +2174,25 @@ func (a *App) GetChannelMembersForUserWithPagination(c request.CTX, userID strin
 	return members, nil
 }
 
-func (a *App) GetChannelMembersWithTeamDataForUserWithPagination(c request.CTX, userID string, page, perPage int) (model.ChannelMembersWithTeamData, *model.AppError) {
-	m, err := a.Srv().Store().Channel().GetMembersForUserWithPagination(userID, page, perPage)
+func (a *App) GetChannelMembersWithTeamDataForUserWithPagination(c request.CTX, userID string, cursor *model.ChannelMemberCursor) (model.ChannelMembersWithTeamData, *model.AppError) {
+	var m model.ChannelMembersWithTeamData
+	var err error
+	var method string
+	if cursor.Page == -1 {
+		m, err = a.Srv().Store().Channel().GetMembersForUserWithCursorPagination(userID, cursor.PerPage, cursor.FromChannelID)
+		method = "GetMembersForUserWithCursorPagination"
+	} else {
+		m, err = a.Srv().Store().Channel().GetMembersForUserWithPagination(userID, cursor.Page, cursor.PerPage)
+		method = "GetMembersForUserWithPagination"
+	}
 	if err != nil {
-		return nil, model.NewAppError("GetChannelMembersForUserWithPagination", "app.channel.get_members.app_error", nil, "", http.StatusInternalServerError).Wrap(err)
+		var nfErr *store.ErrNotFound
+		switch {
+		case errors.As(err, &nfErr):
+			return nil, model.NewAppError(method, MissingChannelMemberError, nil, "", http.StatusNotFound).Wrap(err)
+		default:
+			return nil, model.NewAppError(method, "app.channel.get_members.app_error", nil, "", http.StatusInternalServerError).Wrap(err)
+		}
 	}
 
 	return m, nil

server/channels/store/retrylayer/retrylayer.go

@@ -2071,6 +2071,27 @@ func (s *RetryLayerChannelStore) GetMembersForUser(teamID string, userID string)
 
 }
 
+func (s *RetryLayerChannelStore) GetMembersForUserWithCursorPagination(userId string, perPage int, fromChanneID string) (model.ChannelMembersWithTeamData, error) {
+
+	tries := 0
+	for {
+		result, err := s.ChannelStore.GetMembersForUserWithCursorPagination(userId, perPage, fromChanneID)
+		if err == nil {
+			return result, nil
+		}
+		if !isRepeatableError(err) {
+			return result, err
+		}
+		tries++
+		if tries >= 3 {
+			err = errors.Wrap(err, "giving up after 3 consecutive repeatable transaction failures")
+			return result, err
+		}
+		timepkg.Sleep(100 * timepkg.Millisecond)
+	}
+
+}
+
 func (s *RetryLayerChannelStore) GetMembersForUserWithPagination(userID string, page int, perPage int) (model.ChannelMembersWithTeamData, error) {
 
 	tries := 0

server/channels/store/sqlstore/channel_store.go

@@ -3077,6 +3077,20 @@ func (s SqlChannelStore) GetMembersForUserWithPagination(userId string, page, pe
 	return dbMembers.ToModel(), nil
 }
 
+func (s SqlChannelStore) GetMembersForUserWithCursorPagination(userId string, perPage int, fromChannelID string) (model.ChannelMembersWithTeamData, error) {
+	dbMembers := channelMemberWithTeamWithSchemeRolesList{}
+	err := s.GetReplica().Select(&dbMembers, channelMembersWithSchemeSelectQuery+"WHERE ChannelMembers.UserId = ? AND ChannelId > ? ORDER BY ChannelId ASC Limit ?", userId, fromChannelID, perPage)
+	if err != nil {
+		return nil, errors.Wrapf(err, "failed to find ChannelMembers data with and userId=%s", userId)
+	}
+
+	if len(dbMembers) == 0 {
+		return nil, store.NewErrNotFound("ChannelMembers", "userId="+userId)
+	}
+
+	return dbMembers.ToModel(), nil
+}
+
 func (s SqlChannelStore) GetTeamMembersForChannel(channelID string) ([]string, error) {
 	teamMemberIDs := []string{}
 	if err := s.GetReplica().Select(&teamMemberIDs, `SELECT tm.UserId

server/channels/store/store.go

@@ -266,6 +266,7 @@ type ChannelStore interface {
 	GetMembersForUser(teamID string, userID string) (model.ChannelMembers, error)
 	GetTeamMembersForChannel(channelID string) ([]string, error)
 	GetMembersForUserWithPagination(userID string, page, perPage int) (model.ChannelMembersWithTeamData, error)
+	GetMembersForUserWithCursorPagination(userId string, perPage int, fromChanneID string) (model.ChannelMembersWithTeamData, error)
 	Autocomplete(rctx request.CTX, userID, term string, includeDeleted, isGuest bool) (model.ChannelListWithTeamData, error)
 	AutocompleteInTeam(rctx request.CTX, teamID, userID, term string, includeDeleted, isGuest bool) (model.ChannelList, error)
 	AutocompleteInTeamForSearch(teamID string, userID string, term string, includeDeleted bool) (model.ChannelList, error)

server/channels/store/storetest/channel_store.go

@@ -107,6 +107,7 @@ func TestChannelStore(t *testing.T, rctx request.CTX, ss store.Store, s SqlStore
 	t.Run("GetChannelCounts", func(t *testing.T) { testChannelStoreGetChannelCounts(t, rctx, ss) })
 	t.Run("GetMembersForUser", func(t *testing.T) { testChannelStoreGetMembersForUser(t, rctx, ss) })
 	t.Run("GetMembersForUserWithPagination", func(t *testing.T) { testChannelStoreGetMembersForUserWithPagination(t, rctx, ss) })
+	t.Run("GetMembersForUserWithCursorPagination", func(t *testing.T) { testChannelStoreGetMembersForUserWithCursorPagination(t, rctx, ss) })
 	t.Run("CountPostsAfter", func(t *testing.T) { testCountPostsAfter(t, rctx, ss) })
 	t.Run("CountUrgentPostsAfter", func(t *testing.T) { testCountUrgentPostsAfter(t, rctx, ss) })
 	t.Run("UpdateLastViewedAt", func(t *testing.T) { testChannelStoreUpdateLastViewedAt(t, rctx, ss) })
@@ -4719,6 +4720,93 @@ func testChannelStoreGetMembersForUserWithPagination(t *testing.T, rctx request.
 	assert.Len(t, members, 1)
 }
 
+func testChannelStoreGetMembersForUserWithCursorPagination(t *testing.T, rctx request.CTX, ss store.Store) {
+	t1 := model.Team{
+		DisplayName: "team1",
+		Name:        NewTestID(),
+		Email:       MakeEmail(),
+		Type:        model.TeamOpen,
+	}
+	_, err := ss.Team().Save(&t1)
+	require.NoError(t, err)
+
+	userID := NewTestID()
+
+	var channelIDs []string
+	for i := 0; i < 20; i++ {
+		ch := &model.Channel{
+			TeamId:      t1.Id,
+			DisplayName: "Channel1",
+			Name:        NewTestID(),
+			Type:        model.ChannelTypeOpen,
+		}
+		ch, err = ss.Channel().Save(rctx, ch, -1)
+		require.NoError(t, err)
+		channelIDs = append(channelIDs, ch.Id)
+
+		m1 := model.ChannelMember{}
+		m1.ChannelId = ch.Id
+		m1.UserId = userID
+		m1.NotifyProps = model.GetDefaultChannelNotifyProps()
+		_, err = ss.Channel().SaveMember(rctx, &m1)
+		require.NoError(t, err)
+	}
+
+	members, err := ss.Channel().GetMembersForUserWithCursorPagination(userID, 200, "")
+	require.NoError(t, err)
+	assert.Len(t, members, 20)
+
+	pageSize := 6
+	channelID := ""
+	numPages := 0
+	var gotChannelIDs []string
+	for {
+		members, err := ss.Channel().GetMembersForUserWithCursorPagination(userID, pageSize, channelID)
+		require.NoError(t, err)
+		numPages++
+		for _, m := range members {
+			gotChannelIDs = append(gotChannelIDs, m.ChannelId)
+		}
+		if len(members) < pageSize {
+			// Total 20. PageSize=6. Therefore 6*3=18 + 2 (last page)
+			assert.Len(t, members, 2)
+			break
+		}
+		if len(members) == pageSize {
+			channelID = members[len(members)-1].ChannelId
+			continue
+		}
+		require.Fail(t, "len(members) is > pageSize")
+	}
+	assert.Equal(t, numPages, 4)
+	assert.ElementsMatch(t, channelIDs, gotChannelIDs)
+
+	pageSize = 5
+	channelID = ""
+	numPages = 0
+	gotChannelIDs = []string{}
+	for {
+		members, err := ss.Channel().GetMembersForUserWithCursorPagination(userID, pageSize, channelID)
+		numPages++
+		if numPages < 5 {
+			channelID = members[len(members)-1].ChannelId
+			require.NoError(t, err)
+		} else {
+			// For the last page, it will have no rows.
+			var nfErr *store.ErrNotFound
+			require.True(t, errors.As(err, &nfErr))
+			require.Nil(t, members)
+			break
+		}
+		for _, m := range members {
+			gotChannelIDs = append(gotChannelIDs, m.ChannelId)
+		}
+		assert.Len(t, members, pageSize)
+	}
+	assert.Equal(t, numPages, 5)
+	assert.ElementsMatch(t, channelIDs, gotChannelIDs)
+}
+
 func testCountPostsAfter(t *testing.T, rctx request.CTX, ss store.Store) {
 	t.Run("should count all posts with or without the given user ID", func(t *testing.T) {
 		userID1 := model.NewId()

server/channels/store/storetest/mocks/ChannelStore.go

@@ -1700,6 +1700,36 @@ func (_m *ChannelStore) GetMembersForUser(teamID string, userID string) (model.C
 	return r0, r1
 }
 
+// GetMembersForUserWithCursorPagination provides a mock function with given fields: userId, perPage, fromChanneID
+func (_m *ChannelStore) GetMembersForUserWithCursorPagination(userId string, perPage int, fromChanneID string) (model.ChannelMembersWithTeamData, error) {
+	ret := _m.Called(userId, perPage, fromChanneID)
+
+	if len(ret) == 0 {
+		panic("no return value specified for GetMembersForUserWithCursorPagination")
+	}
+
+	var r0 model.ChannelMembersWithTeamData
+	var r1 error
+	if rf, ok := ret.Get(0).(func(string, int, string) (model.ChannelMembersWithTeamData, error)); ok {
+		return rf(userId, perPage, fromChanneID)
+	}
+	if rf, ok := ret.Get(0).(func(string, int, string) model.ChannelMembersWithTeamData); ok {
+		r0 = rf(userId, perPage, fromChanneID)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(model.ChannelMembersWithTeamData)
+		}
+	}
+
+	if rf, ok := ret.Get(1).(func(string, int, string) error); ok {
+		r1 = rf(userId, perPage, fromChanneID)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
 // GetMembersForUserWithPagination provides a mock function with given fields: userID, page, perPage
 func (_m *ChannelStore) GetMembersForUserWithPagination(userID string, page int, perPage int) (model.ChannelMembersWithTeamData, error) {
 	ret := _m.Called(userID, page, perPage)

server/channels/store/timerlayer/timerlayer.go

@@ -1719,6 +1719,22 @@ func (s *TimerLayerChannelStore) GetMembersForUser(teamID string, userID string)
 	return result, err
 }
 
+func (s *TimerLayerChannelStore) GetMembersForUserWithCursorPagination(userId string, perPage int, fromChanneID string) (model.ChannelMembersWithTeamData, error) {
+	start := time.Now()
+
+	result, err := s.ChannelStore.GetMembersForUserWithCursorPagination(userId, perPage, fromChanneID)
+
+	elapsed := float64(time.Since(start)) / float64(time.Second)
+	if s.Root.Metrics != nil {
+		success := "false"
+		if err == nil {
+			success = "true"
+		}
+		s.Root.Metrics.ObserveStoreMethodDuration("ChannelStore.GetMembersForUserWithCursorPagination", success, elapsed)
+	}
+	return result, err
+}
+
 func (s *TimerLayerChannelStore) GetMembersForUserWithPagination(userID string, page int, perPage int) (model.ChannelMembersWithTeamData, error) {
 	start := time.Now()
 

server/channels/web/params.go

@@ -5,6 +5,7 @@ package web
 
 import (
 	"net/http"
+	"regexp"
 	"strconv"
 	"strings"
 
@@ -116,6 +117,8 @@ type Params struct {
 	FieldId string
 }
 
+var getChannelMembersForUserRegex = regexp.MustCompile("/api/v4/users/[A-Za-z0-9]{26}/channel_members")
+
 func ParamsFromRequest(r *http.Request) *Params {
 	params := &Params{}
 
@@ -184,7 +187,9 @@ func ParamsFromRequest(r *http.Request) *Params {
 	params.FieldId = props["field_id"]
 	params.Scope = query.Get("scope")
 
-	if val, err := strconv.Atoi(query.Get("page")); err != nil || val < 0 {
+	if val, err := strconv.Atoi(query.Get("page")); err != nil || (val < 0 && params.UserId == "" && !getChannelMembersForUserRegex.MatchString(r.URL.Path)) {
+		// We don't want to apply this logic for the getChannelMembersForUser API handler
+		// because that API allows page=-1 to switch to streaming mode.
 		params.Page = PageDefault
 	} else {
 		params.Page = val

server/channels/web/params_test.go

@@ -92,8 +92,7 @@ func TestParamsFromRequest(t *testing.T) {
 			mustURL("?page=hello"),
 			nil,
 			&Params{
-				Page: PageDefault,
-
+				Page:        PageDefault,
 				PerPage:     PerPageDefault,
 				LogsPerPage: LogsPerPageDefault,
 				LimitAfter:  LimitDefault,
@@ -105,8 +104,7 @@ func TestParamsFromRequest(t *testing.T) {
 			mustURL("?page=-1"),
 			nil,
 			&Params{
-				Page: PageDefault,
-
+				Page:        PageDefault,
 				PerPage:     PerPageDefault,
 				LogsPerPage: LogsPerPageDefault,
 				LimitAfter:  LimitDefault,

server/public/model/channel_member.go

@@ -108,6 +108,12 @@ type ChannelMemberForExport struct {
 	Username    string
 }
 
+type ChannelMemberCursor struct {
+	Page          int // If page is -1, then FromChannelID is used as a cursor.
+	PerPage       int
+	FromChannelID string
+}
+
 func (o *ChannelMember) IsValid() *AppError {
 	if !IsValidId(o.ChannelId) {
 		return NewAppError("ChannelMember.IsValid", "model.channel_member.is_valid.channel_id.app_error", nil, "", http.StatusBadRequest)

server/public/model/client4.go

@@ -4,6 +4,7 @@
 package model
 
 import (
+	"bufio"
 	"bytes"
 	"context"
 	"encoding/json"
@@ -3696,6 +3697,37 @@ func (c *Client4) GetChannelMembersWithTeamData(ctx context.Context, userID stri
 	defer closeBody(r)
 
 	var ch ChannelMembersWithTeamData
+
+	// Check if we need to handle NDJSON format (when page is -1)
+	if page == -1 {
+		// Process NDJSON format (each JSON object on new line)
+		contentType := r.Header.Get("Content-Type")
+		if contentType == "application/x-ndjson" {
+			scanner := bufio.NewScanner(r.Body)
+			ch = ChannelMembersWithTeamData{}
+
+			for scanner.Scan() {
+				line := scanner.Text()
+				if line == "" {
+					continue
+				}
+
+				var member ChannelMemberWithTeamData
+				if err2 := json.Unmarshal([]byte(line), &member); err2 != nil {
+					return nil, BuildResponse(r), NewAppError("GetChannelMembersWithTeamData", "api.marshal_error", nil, "", http.StatusInternalServerError).Wrap(err2)
+				}
+				ch = append(ch, member)
+			}
+
+			if err2 := scanner.Err(); err2 != nil {
+				return nil, BuildResponse(r), NewAppError("GetChannelMembersWithTeamData", "api.marshal_error", nil, "", http.StatusInternalServerError).Wrap(err2)
+			}
+
+			return ch, BuildResponse(r), nil
+		}
+	}
+
+	// Standard JSON format
 	err = json.NewDecoder(r.Body).Decode(&ch)
 	if err != nil {
 		return nil, BuildResponse(r), NewAppError("GetChannelMembersWithTeamData", "api.marshal_error", nil, "", http.StatusInternalServerError).Wrap(err)