MM-43145: channels loader (#19957)

Add channels dataloader

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

```release-note
NONE
```

api4/graphql.go

@@ -62,10 +62,13 @@ func (api *API) InitGraphQL() error {
 type ctxKey int
 
 const (
-	webCtx         ctxKey = 0
-	rolesLoaderCtx ctxKey = 1
+	webCtx            ctxKey = 0
+	rolesLoaderCtx    ctxKey = 1
+	channelsLoaderCtx ctxKey = 2
 )
 
+const loaderBatchCapacity = 200
+
 func (api *API) graphQL(c *Context, w http.ResponseWriter, r *http.Request) {
 	var response *graphql.Response
 	defer func() {
@@ -98,9 +101,12 @@ func (api *API) graphQL(c *Context, w http.ResponseWriter, r *http.Request) {
 	reqCtx := r.Context()
 	reqCtx = context.WithValue(reqCtx, webCtx, c)
 
-	rolesLoader := dataloader.NewBatchedLoader(graphQLRolesLoader, dataloader.WithBatchCapacity(200))
+	rolesLoader := dataloader.NewBatchedLoader(graphQLRolesLoader, dataloader.WithBatchCapacity(loaderBatchCapacity))
 	reqCtx = context.WithValue(reqCtx, rolesLoaderCtx, rolesLoader)
 
+	channelsLoader := dataloader.NewBatchedLoader(graphQLChannelsLoader, dataloader.WithBatchCapacity(loaderBatchCapacity))
+	reqCtx = context.WithValue(reqCtx, channelsLoaderCtx, channelsLoader)
+
 	response = api.schema.Exec(reqCtx,
 		params.Query,
 		params.OperationName,

api4/resolver.go

@@ -294,3 +294,12 @@ func getRolesLoader(ctx context.Context) (*dataloader.Loader, error) {
 	}
 	return l, nil
 }
+
+// getChannelsLoader returns the channels loader out of the context.
+func getChannelsLoader(ctx context.Context) (*dataloader.Loader, error) {
+	l, ok := ctx.Value(channelsLoaderCtx).(*dataloader.Loader)
+	if !ok {
+		return nil, errors.New("no dataloader.Loader found in context")
+	}
+	return l, nil
+}

api4/resolver_channel_member.go

@@ -31,11 +31,18 @@ func (cm *channelMember) Channel(ctx context.Context) (*channel, error) {
 		return nil, err
 	}
 
-	channel, appErr := c.App.GetChannel(cm.ChannelId)
-	if appErr != nil {
-		return nil, appErr
+	loader, err := getChannelsLoader(ctx)
+	if err != nil {
+		return nil, err
 	}
 
+	thunk := loader.Load(ctx, dataloader.StringKey(cm.ChannelId))
+	result, err := thunk()
+	if err != nil {
+		return nil, err
+	}
+	channel := result.(*model.Channel)
+
 	if channel.Type == model.ChannelTypeOpen {
 		if !c.App.SessionHasPermissionToTeam(*c.AppContext.Session(), channel.TeamId, model.PermissionReadPublicChannel) &&
 			!c.App.SessionHasPermissionToChannel(*c.AppContext.Session(), cm.ChannelId, model.PermissionReadChannel) {
@@ -49,7 +56,7 @@ func (cm *channelMember) Channel(ctx context.Context) (*channel, error) {
 		}
 	}
 
-	appErr = c.App.FillInChannelProps(channel)
+	appErr := c.App.FillInChannelProps(channel)
 	if appErr != nil {
 		return nil, appErr
 	}
@@ -65,6 +72,56 @@ func (cm *channelMember) Channel(ctx context.Context) (*channel, error) {
 	return res[0], nil
 }
 
+func graphQLChannelsLoader(ctx context.Context, keys dataloader.Keys) []*dataloader.Result {
+	stringKeys := keys.Keys()
+	result := make([]*dataloader.Result, len(stringKeys))
+
+	c, err := getCtx(ctx)
+	if err != nil {
+		for i := range result {
+			result[i] = &dataloader.Result{Error: err}
+		}
+		return result
+	}
+
+	channels, err := getGraphQLChannels(c, stringKeys)
+	if err != nil {
+		for i := range result {
+			result[i] = &dataloader.Result{Error: err}
+		}
+		return result
+	}
+
+	for i, ch := range channels {
+		result[i] = &dataloader.Result{Data: ch}
+	}
+	return result
+}
+
+func getGraphQLChannels(c *web.Context, channelIDs []string) ([]*model.Channel, error) {
+	channels, appErr := c.App.GetChannels(channelIDs)
+	if appErr != nil {
+		return nil, appErr
+	}
+
+	if len(channels) != len(channelIDs) {
+		return nil, fmt.Errorf("all channels were not found. Requested %d; Found %d", len(channelIDs), len(channels))
+	}
+
+	// The channels need to be in the exact same order as the input slice.
+	tmp := make(map[string]*model.Channel)
+	for _, ch := range channels {
+		tmp[ch.Id] = ch
+	}
+
+	// We reuse the same slice and just rewrite the channels.
+	for i, id := range channelIDs {
+		channels[i] = tmp[id]
+	}
+
+	return channels, nil
+}
+
 func (cm *channelMember) Roles_(ctx context.Context) ([]*model.Role, error) {
 	loader, err := getRolesLoader(ctx)
 	if err != nil {
@@ -98,13 +155,17 @@ func graphQLRolesLoader(ctx context.Context, keys dataloader.Keys) []*dataloader
 
 	c, err := getCtx(ctx)
 	if err != nil {
-		result[0] = &dataloader.Result{Error: err}
+		for i := range result {
+			result[i] = &dataloader.Result{Error: err}
+		}
 		return result
 	}
 
 	roles, err := getGraphQLRoles(c, stringKeys)
 	if err != nil {
-		result[0] = &dataloader.Result{Error: err}
+		for i := range result {
+			result[i] = &dataloader.Result{Error: err}
+		}
 		return result
 	}
 

app/app_iface.go

@@ -569,6 +569,7 @@ type AppIface interface {
 	GetChannelPinnedPostCount(channelID string) (int64, *model.AppError)
 	GetChannelPoliciesForUser(userID string, offset, limit int) (*model.RetentionPolicyForChannelList, *model.AppError)
 	GetChannelUnread(channelID, userID string) (*model.ChannelUnread, *model.AppError)
+	GetChannels(channelIDs []string) ([]*model.Channel, *model.AppError)
 	GetChannelsByNames(channelNames []string, teamID string) ([]*model.Channel, *model.AppError)
 	GetChannelsForRetentionPolicy(policyID string, offset, limit int) (*model.ChannelsWithCount, *model.AppError)
 	GetChannelsForScheme(scheme *model.Scheme, offset int, limit int) (model.ChannelList, *model.AppError)

app/channel.go

@@ -1736,6 +1736,20 @@ func (s *Server) getChannel(channelID string) (*model.Channel, *model.AppError)
 	return channel, nil
 }
 
+func (a *App) GetChannels(channelIDs []string) ([]*model.Channel, *model.AppError) {
+	channels, err := a.Srv().Store.Channel().GetMany(channelIDs, true)
+	if err != nil {
+		var nfErr *store.ErrNotFound
+		switch {
+		case errors.As(err, &nfErr):
+			return nil, model.NewAppError("GetChannel", "app.channel.get.existing.app_error", nil, nfErr.Error(), http.StatusNotFound)
+		default:
+			return nil, model.NewAppError("GetChannel", "app.channel.get.find.app_error", nil, err.Error(), http.StatusInternalServerError)
+		}
+	}
+	return channels, nil
+}
+
 func (a *App) GetChannelByName(channelName, teamID string, includeDeleted bool) (*model.Channel, *model.AppError) {
 	var channel *model.Channel
 	var err error

app/opentracing/opentracing_layer.go

@@ -5176,6 +5176,28 @@ func (a *OpenTracingAppLayer) GetChannelUnread(channelID string, userID string)
 	return resultVar0, resultVar1
 }
 
+func (a *OpenTracingAppLayer) GetChannels(channelIDs []string) ([]*model.Channel, *model.AppError) {
+	origCtx := a.ctx
+	span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.GetChannels")
+
+	a.ctx = newCtx
+	a.app.Srv().Store.SetContext(newCtx)
+	defer func() {
+		a.app.Srv().Store.SetContext(origCtx)
+		a.ctx = origCtx
+	}()
+
+	defer span.Finish()
+	resultVar0, resultVar1 := a.app.GetChannels(channelIDs)
+
+	if resultVar1 != nil {
+		span.LogFields(spanlog.Error(resultVar1))
+		ext.Error.Set(span, true)
+	}
+
+	return resultVar0, resultVar1
+}
+
 func (a *OpenTracingAppLayer) GetChannelsByNames(channelNames []string, teamID string) ([]*model.Channel, *model.AppError) {
 	origCtx := a.ctx
 	span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.GetChannelsByNames")

store/localcachelayer/channel_layer.go

@@ -174,6 +174,37 @@ func (s LocalCacheChannelStore) Get(id string, allowFromCache bool) (*model.Chan
 	return ch, err
 }
 
+func (s LocalCacheChannelStore) GetMany(ids []string, allowFromCache bool) (model.ChannelList, error) {
+	var foundChannels []*model.Channel
+	var channelsToQuery []string
+
+	if allowFromCache {
+		for _, id := range ids {
+			var ch *model.Channel
+			if err := s.rootStore.doStandardReadCache(s.rootStore.channelByIdCache, id, &ch); err == nil {
+				foundChannels = append(foundChannels, ch)
+			} else {
+				channelsToQuery = append(channelsToQuery, id)
+			}
+		}
+	}
+
+	if channelsToQuery == nil {
+		return foundChannels, nil
+	}
+
+	channels, err := s.ChannelStore.GetMany(channelsToQuery, allowFromCache)
+	if err != nil {
+		return nil, err
+	}
+
+	for _, ch := range channels {
+		s.rootStore.doStandardAddToCache(s.rootStore.channelByIdCache, ch.Id, ch)
+	}
+
+	return append(foundChannels, channels...), nil
+}
+
 func (s LocalCacheChannelStore) SaveMember(member *model.ChannelMember) (*model.ChannelMember, error) {
 	member, err := s.ChannelStore.SaveMember(member)
 	if err != nil {

store/opentracinglayer/opentracinglayer.go

@@ -1285,24 +1285,6 @@ func (s *OpenTracingLayerChannelStore) GetForPost(postID string) (*model.Channel
 	return result, err
 }
 
-func (s *OpenTracingLayerChannelStore) GetFromMaster(id string) (*model.Channel, error) {
-	origCtx := s.Root.Store.Context()
-	span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "ChannelStore.GetFromMaster")
-	s.Root.Store.SetContext(newCtx)
-	defer func() {
-		s.Root.Store.SetContext(origCtx)
-	}()
-
-	defer span.Finish()
-	result, err := s.ChannelStore.GetFromMaster(id)
-	if err != nil {
-		span.LogFields(spanlog.Error(err))
-		ext.Error.Set(span, true)
-	}
-
-	return result, err
-}
-
 func (s *OpenTracingLayerChannelStore) GetGuestCount(channelID string, allowFromCache bool) (int64, error) {
 	origCtx := s.Root.Store.Context()
 	span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "ChannelStore.GetGuestCount")
@@ -1321,6 +1303,24 @@ func (s *OpenTracingLayerChannelStore) GetGuestCount(channelID string, allowFrom
 	return result, err
 }
 
+func (s *OpenTracingLayerChannelStore) GetMany(ids []string, allowFromCache bool) (model.ChannelList, error) {
+	origCtx := s.Root.Store.Context()
+	span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "ChannelStore.GetMany")
+	s.Root.Store.SetContext(newCtx)
+	defer func() {
+		s.Root.Store.SetContext(origCtx)
+	}()
+
+	defer span.Finish()
+	result, err := s.ChannelStore.GetMany(ids, allowFromCache)
+	if err != nil {
+		span.LogFields(spanlog.Error(err))
+		ext.Error.Set(span, true)
+	}
+
+	return result, err
+}
+
 func (s *OpenTracingLayerChannelStore) GetMember(ctx context.Context, channelID string, userID string) (*model.ChannelMember, error) {
 	origCtx := s.Root.Store.Context()
 	span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "ChannelStore.GetMember")

store/retrylayer/retrylayer.go

@@ -1444,11 +1444,11 @@ func (s *RetryLayerChannelStore) GetForPost(postID string) (*model.Channel, erro
 
 }
 
-func (s *RetryLayerChannelStore) GetFromMaster(id string) (*model.Channel, error) {
+func (s *RetryLayerChannelStore) GetGuestCount(channelID string, allowFromCache bool) (int64, error) {
 
 	tries := 0
 	for {
-		result, err := s.ChannelStore.GetFromMaster(id)
+		result, err := s.ChannelStore.GetGuestCount(channelID, allowFromCache)
 		if err == nil {
 			return result, nil
 		}
@@ -1465,11 +1465,11 @@ func (s *RetryLayerChannelStore) GetFromMaster(id string) (*model.Channel, error
 
 }
 
-func (s *RetryLayerChannelStore) GetGuestCount(channelID string, allowFromCache bool) (int64, error) {
+func (s *RetryLayerChannelStore) GetMany(ids []string, allowFromCache bool) (model.ChannelList, error) {
 
 	tries := 0
 	for {
-		result, err := s.ChannelStore.GetGuestCount(channelID, allowFromCache)
+		result, err := s.ChannelStore.GetMany(ids, allowFromCache)
 		if err == nil {
 			return result, nil
 		}

store/sqlstore/channel_store.go

@@ -806,11 +806,6 @@ func (s SqlChannelStore) InvalidateChannelByName(teamId, name string) {
 	}
 }
 
-//nolint:unparam
-func (s SqlChannelStore) Get(id string, allowFromCache bool) (*model.Channel, error) {
-	return s.get(id, false)
-}
-
 func (s SqlChannelStore) GetPinnedPosts(channelId string) (*model.PostList, error) {
 	pl := model.NewPostList()
 
@@ -825,21 +820,10 @@ func (s SqlChannelStore) GetPinnedPosts(channelId string) (*model.PostList, erro
 	return pl, nil
 }
 
-func (s SqlChannelStore) GetFromMaster(id string) (*model.Channel, error) {
-	return s.get(id, true)
-}
-
-func (s SqlChannelStore) get(id string, master bool) (*model.Channel, error) {
-	var db *sqlxDBWrapper
-
-	if master {
-		db = s.GetMasterX()
-	} else {
-		db = s.GetReplicaX()
-	}
-
+//nolint:unparam
+func (s SqlChannelStore) Get(id string, allowFromCache bool) (*model.Channel, error) {
 	ch := model.Channel{}
-	err := db.Get(&ch, `SELECT * FROM Channels WHERE Id=?`, id)
+	err := s.GetReplicaX().Get(&ch, `SELECT * FROM Channels WHERE Id=?`, id)
 	if err != nil {
 		if err == sql.ErrNoRows {
 			return nil, store.NewErrNotFound("Channel", id)
@@ -850,6 +834,30 @@ func (s SqlChannelStore) get(id string, master bool) (*model.Channel, error) {
 	return &ch, nil
 }
 
+//nolint:unparam
+func (s SqlChannelStore) GetMany(ids []string, allowFromCache bool) (model.ChannelList, error) {
+	query := s.getQueryBuilder().
+		Select("*").
+		From("Channels").
+		Where(sq.Eq{"Id": ids})
+	sql, args, err := query.ToSql()
+	if err != nil {
+		return nil, errors.Wrapf(err, "getmany_tosql")
+	}
+
+	channels := model.ChannelList{}
+	err = s.GetReplicaX().Select(&channels, sql, args...)
+	if err != nil {
+		return nil, errors.Wrapf(err, "failed to get channels with ids %v", ids)
+	}
+
+	if len(channels) == 0 {
+		return nil, store.NewErrNotFound("Channel", fmt.Sprintf("ids=%v", ids))
+	}
+
+	return channels, nil
+}
+
 // Delete records the given deleted timestamp to the channel in question.
 func (s SqlChannelStore) Delete(channelId string, time int64) error {
 	return s.SetDeleteAt(channelId, time, time)

store/store.go

@@ -166,9 +166,9 @@ type ChannelStore interface {
 	UpdateSidebarChannelCategoryOnMove(channel *model.Channel, newTeamID string) error
 	ClearSidebarOnTeamLeave(userID, teamID string) error
 	Get(id string, allowFromCache bool) (*model.Channel, error)
+	GetMany(ids []string, allowFromCache bool) (model.ChannelList, error)
 	InvalidateChannel(id string)
 	InvalidateChannelByName(teamID, name string)
-	GetFromMaster(id string) (*model.Channel, error)
 	Delete(channelID string, time int64) error
 	Restore(channelID string, time int64) error
 	SetDeleteAt(channelID string, deleteAt int64, updateAt int64) error

store/storetest/channel_store.go

@@ -72,6 +72,7 @@ func TestChannelStore(t *testing.T, ss store.Store, s SqlStore) {
 	t.Run("Update", func(t *testing.T) { testChannelStoreUpdate(t, ss) })
 	t.Run("GetChannelUnread", func(t *testing.T) { testGetChannelUnread(t, ss) })
 	t.Run("Get", func(t *testing.T) { testChannelStoreGet(t, ss, s) })
+	t.Run("GetMany", func(t *testing.T) { testChannelStoreGetMany(t, ss, s) })
 	t.Run("GetChannelsByIds", func(t *testing.T) { testChannelStoreGetChannelsByIds(t, ss) })
 	t.Run("GetChannelsWithTeamDataByIds", func(t *testing.T) { testGetChannelsWithTeamDataByIds(t, ss) })
 	t.Run("GetForPost", func(t *testing.T) { testChannelStoreGetForPost(t, ss) })
@@ -476,6 +477,40 @@ func testChannelStoreGet(t *testing.T, ss store.Store, s SqlStore) {
 	s.GetMasterX().Exec("TRUNCATE Channels")
 }
 
+func testChannelStoreGetMany(t *testing.T, ss store.Store, s SqlStore) {
+	o1, nErr := ss.Channel().Save(&model.Channel{
+		TeamId:      model.NewId(),
+		DisplayName: "Name",
+		Name:        NewTestId(),
+		Type:        model.ChannelTypeOpen,
+	}, -1)
+	require.NoError(t, nErr)
+
+	o2, nErr := ss.Channel().Save(&model.Channel{
+		TeamId:      model.NewId(),
+		DisplayName: "Name2",
+		Name:        NewTestId(),
+		Type:        model.ChannelTypeOpen,
+	}, -1)
+	require.NoError(t, nErr)
+
+	res, err := ss.Channel().GetMany([]string{o1.Id, o2.Id}, true)
+	require.NoError(t, err)
+	assert.Len(t, res, 2)
+
+	res, err = ss.Channel().GetMany([]string{o1.Id, "notexists"}, true)
+	require.NoError(t, err)
+	assert.Len(t, res, 1)
+
+	_, err = ss.Channel().GetMany([]string{"notexists"}, true)
+	require.Error(t, err)
+	var nfErr *store.ErrNotFound
+	require.True(t, errors.As(err, &nfErr))
+
+	// Manually truncate Channels table until testlib can handle cleanups
+	s.GetMasterX().Exec("TRUNCATE Channels")
+}
+
 func testChannelStoreGetChannelsByIds(t *testing.T, ss store.Store) {
 	o1 := model.Channel{}
 	o1.TeamId = model.NewId()

store/storetest/mocks/ChannelStore.go

@@ -925,29 +925,6 @@ func (_m *ChannelStore) GetForPost(postID string) (*model.Channel, error) {
 	return r0, r1
 }
 
-// GetFromMaster provides a mock function with given fields: id
-func (_m *ChannelStore) GetFromMaster(id string) (*model.Channel, error) {
-	ret := _m.Called(id)
-
-	var r0 *model.Channel
-	if rf, ok := ret.Get(0).(func(string) *model.Channel); ok {
-		r0 = rf(id)
-	} else {
-		if ret.Get(0) != nil {
-			r0 = ret.Get(0).(*model.Channel)
-		}
-	}
-
-	var r1 error
-	if rf, ok := ret.Get(1).(func(string) error); ok {
-		r1 = rf(id)
-	} else {
-		r1 = ret.Error(1)
-	}
-
-	return r0, r1
-}
-
 // GetGuestCount provides a mock function with given fields: channelID, allowFromCache
 func (_m *ChannelStore) GetGuestCount(channelID string, allowFromCache bool) (int64, error) {
 	ret := _m.Called(channelID, allowFromCache)
@@ -969,6 +946,29 @@ func (_m *ChannelStore) GetGuestCount(channelID string, allowFromCache bool) (in
 	return r0, r1
 }
 
+// GetMany provides a mock function with given fields: ids, allowFromCache
+func (_m *ChannelStore) GetMany(ids []string, allowFromCache bool) (model.ChannelList, error) {
+	ret := _m.Called(ids, allowFromCache)
+
+	var r0 model.ChannelList
+	if rf, ok := ret.Get(0).(func([]string, bool) model.ChannelList); ok {
+		r0 = rf(ids, allowFromCache)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(model.ChannelList)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func([]string, bool) error); ok {
+		r1 = rf(ids, allowFromCache)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
 // GetMember provides a mock function with given fields: ctx, channelID, userID
 func (_m *ChannelStore) GetMember(ctx context.Context, channelID string, userID string) (*model.ChannelMember, error) {
 	ret := _m.Called(ctx, channelID, userID)

store/timerlayer/timerlayer.go

@@ -1189,22 +1189,6 @@ func (s *TimerLayerChannelStore) GetForPost(postID string) (*model.Channel, erro
 	return result, err
 }
 
-func (s *TimerLayerChannelStore) GetFromMaster(id string) (*model.Channel, error) {
-	start := timemodule.Now()
-
-	result, err := s.ChannelStore.GetFromMaster(id)
-
-	elapsed := float64(timemodule.Since(start)) / float64(timemodule.Second)
-	if s.Root.Metrics != nil {
-		success := "false"
-		if err == nil {
-			success = "true"
-		}
-		s.Root.Metrics.ObserveStoreMethodDuration("ChannelStore.GetFromMaster", success, elapsed)
-	}
-	return result, err
-}
-
 func (s *TimerLayerChannelStore) GetGuestCount(channelID string, allowFromCache bool) (int64, error) {
 	start := timemodule.Now()
 
@@ -1221,6 +1205,22 @@ func (s *TimerLayerChannelStore) GetGuestCount(channelID string, allowFromCache
 	return result, err
 }
 
+func (s *TimerLayerChannelStore) GetMany(ids []string, allowFromCache bool) (model.ChannelList, error) {
+	start := timemodule.Now()
+
+	result, err := s.ChannelStore.GetMany(ids, allowFromCache)
+
+	elapsed := float64(timemodule.Since(start)) / float64(timemodule.Second)
+	if s.Root.Metrics != nil {
+		success := "false"
+		if err == nil {
+			success = "true"
+		}
+		s.Root.Metrics.ObserveStoreMethodDuration("ChannelStore.GetMany", success, elapsed)
+	}
+	return result, err
+}
+
 func (s *TimerLayerChannelStore) GetMember(ctx context.Context, channelID string, userID string) (*model.ChannelMember, error) {
 	start := timemodule.Now()