diff --git a/api4/graphql.go b/api4/graphql.go index f607009344..1e7879290b 100644 --- a/api4/graphql.go +++ b/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, diff --git a/api4/resolver.go b/api4/resolver.go index 8b600e570b..08f3de8260 100644 --- a/api4/resolver.go +++ b/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 +} diff --git a/api4/resolver_channel_member.go b/api4/resolver_channel_member.go index eedc3c2eea..86bb1bbb18 100644 --- a/api4/resolver_channel_member.go +++ b/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 } diff --git a/app/app_iface.go b/app/app_iface.go index 31dcb25a3a..cfc816b797 100644 --- a/app/app_iface.go +++ b/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) diff --git a/app/channel.go b/app/channel.go index 36595cfb01..c4b6444e9a 100644 --- a/app/channel.go +++ b/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 diff --git a/app/opentracing/opentracing_layer.go b/app/opentracing/opentracing_layer.go index 127e55a593..13a7a7cc1f 100644 --- a/app/opentracing/opentracing_layer.go +++ b/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") diff --git a/store/localcachelayer/channel_layer.go b/store/localcachelayer/channel_layer.go index 6f9a1c2c17..0264c2dd6d 100644 --- a/store/localcachelayer/channel_layer.go +++ b/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 { diff --git a/store/opentracinglayer/opentracinglayer.go b/store/opentracinglayer/opentracinglayer.go index 2090b17801..ca33545bd0 100644 --- a/store/opentracinglayer/opentracinglayer.go +++ b/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") diff --git a/store/retrylayer/retrylayer.go b/store/retrylayer/retrylayer.go index 7442161038..430f08c958 100644 --- a/store/retrylayer/retrylayer.go +++ b/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 } diff --git a/store/sqlstore/channel_store.go b/store/sqlstore/channel_store.go index f1e74732d4..b9a023bf20 100644 --- a/store/sqlstore/channel_store.go +++ b/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) diff --git a/store/store.go b/store/store.go index 3123cff9ab..83a37ddfc5 100644 --- a/store/store.go +++ b/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 diff --git a/store/storetest/channel_store.go b/store/storetest/channel_store.go index 338069a13a..ee5bddbf0c 100644 --- a/store/storetest/channel_store.go +++ b/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() diff --git a/store/storetest/mocks/ChannelStore.go b/store/storetest/mocks/ChannelStore.go index d66a0a0628..df0ae8db9e 100644 --- a/store/storetest/mocks/ChannelStore.go +++ b/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) diff --git a/store/timerlayer/timerlayer.go b/store/timerlayer/timerlayer.go index 38af85bd9b..85807d2f09 100644 --- a/store/timerlayer/timerlayer.go +++ b/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()