Optimise creation of DM (#16819)

* Optimise creation of dm

* Handle direct channels with the same user

* Cover GetMany with specs and add it on tha cache layer as well

* Fix specs by handling user dming themselves

* Apply PR suggestions

* Apply PR suggestions

* Use require.NoError instead of require.Nil on userstore test

* Improve readability of GetOrCreateDirectChannel

* Apply PR suggestions

* Update layers

Co-authored-by: Mattermod <mattermod@users.noreply.github.com>

app/app_iface.go

@@ -622,7 +622,7 @@ type AppIface interface {
 	GetOAuthSignupEndpoint(w http.ResponseWriter, r *http.Request, service, teamID string) (string, *model.AppError)
 	GetOAuthStateToken(token string) (*model.Token, *model.AppError)
 	GetOpenGraphMetadata(requestURL string) *opengraph.OpenGraph
-	GetOrCreateDirectChannel(userID, otherUserId string) (*model.Channel, *model.AppError)
+	GetOrCreateDirectChannel(userID, otherUserID string) (*model.Channel, *model.AppError)
 	GetOutgoingWebhook(hookID string) (*model.OutgoingWebhook, *model.AppError)
 	GetOutgoingWebhooksForChannelPageByUser(channelId string, userID string, page, perPage int) ([]*model.OutgoingWebhook, *model.AppError)
 	GetOutgoingWebhooksForTeamPage(teamID string, page, perPage int) ([]*model.OutgoingWebhook, *model.AppError)

app/channel.go

@@ -320,71 +320,76 @@ func (a *App) CreateChannel(channel *model.Channel, addMember bool) (*model.Chan
 	return sc, nil
 }
 
-func (a *App) GetOrCreateDirectChannel(userID, otherUserId string) (*model.Channel, *model.AppError) {
+func (a *App) GetOrCreateDirectChannel(userID, otherUserID string) (*model.Channel, *model.AppError) {
-	channel, nErr := a.Srv().Store.Channel().GetByName("", model.GetDMNameFromIds(userID, otherUserId), true)
+	channel, nErr := a.getDirectChannel(userID, otherUserID)
 	if nErr != nil {
-		var nfErr *store.ErrNotFound
+		return nil, nErr
-		if errors.As(nErr, &nfErr) {
+	}
-			var err *model.AppError
-			channel, err = a.createDirectChannel(userID, otherUserId)
-			if err != nil {
-				if err.Id == store.ChannelExistsError {
-					return channel, nil
-				}
-				return nil, err
-			}
 
-			a.WaitForChannelMembership(channel.Id, userID)
+	if channel != nil {
-
+		return channel, nil
-			a.InvalidateCacheForUser(userID)
+	}
-			a.InvalidateCacheForUser(otherUserId)
-
-			if pluginsEnvironment := a.GetPluginsEnvironment(); pluginsEnvironment != nil {
-				a.Srv().Go(func() {
-					pluginContext := a.PluginContext()
-					pluginsEnvironment.RunMultiPluginHook(func(hooks plugin.Hooks) bool {
-						hooks.ChannelHasBeenCreated(pluginContext, channel)
-						return true
-					}, plugin.ChannelHasBeenCreatedId)
-				})
-			}
-
-			message := model.NewWebSocketEvent(model.WEBSOCKET_EVENT_DIRECT_ADDED, "", channel.Id, "", nil)
-			message.Add("teammate_id", otherUserId)
-			a.Publish(message)
 
+	channel, err := a.createDirectChannel(userID, otherUserID)
+	if err != nil {
+		if err.Id == store.ChannelExistsError {
 			return channel, nil
 		}
-		return nil, model.NewAppError("GetOrCreateDirectChannel", "web.incoming_webhook.channel.app_error", nil, nErr.Error(), http.StatusInternalServerError)
+		return nil, err
 	}
+
+	a.WaitForChannelMembership(channel.Id, userID)
+
+	a.InvalidateCacheForUser(userID)
+	a.InvalidateCacheForUser(otherUserID)
+
+	if pluginsEnvironment := a.GetPluginsEnvironment(); pluginsEnvironment != nil {
+		a.Srv().Go(func() {
+			pluginContext := a.PluginContext()
+			pluginsEnvironment.RunMultiPluginHook(func(hooks plugin.Hooks) bool {
+				hooks.ChannelHasBeenCreated(pluginContext, channel)
+				return true
+			}, plugin.ChannelHasBeenCreatedId)
+		})
+	}
+
+	message := model.NewWebSocketEvent(model.WEBSOCKET_EVENT_DIRECT_ADDED, "", channel.Id, "", nil)
+	message.Add("teammate_id", otherUserID)
+	a.Publish(message)
+
 	return channel, nil
 }
 
-func (a *App) createDirectChannel(userID string, otherUserId string) (*model.Channel, *model.AppError) {
+func (a *App) createDirectChannel(userID string, otherUserID string) (*model.Channel, *model.AppError) {
-	uc1 := make(chan store.StoreResult, 1)
+	users, err := a.Srv().Store.User().GetMany([]string{userID, otherUserID})
-	uc2 := make(chan store.StoreResult, 1)
+	if err != nil {
-	go func() {
+		return nil, model.NewAppError("CreateDirectChannel", "api.channel.create_direct_channel.invalid_user.app_error", nil, err.Error(), http.StatusBadRequest)
-		user, err := a.Srv().Store.User().Get(userID)
-		uc1 <- store.StoreResult{Data: user, NErr: err}
-		close(uc1)
-	}()
-	go func() {
-		user, err := a.Srv().Store.User().Get(otherUserId)
-		uc2 <- store.StoreResult{Data: user, NErr: err}
-		close(uc2)
-	}()
-
-	result := <-uc1
-	if result.NErr != nil {
-		return nil, model.NewAppError("CreateDirectChannel", "api.channel.create_direct_channel.invalid_user.app_error", nil, userID, http.StatusBadRequest)
 	}
-	user := result.Data.(*model.User)
 
-	result = <-uc2
+	if len(users) == 0 {
-	if result.NErr != nil {
+		return nil, model.NewAppError("CreateDirectChannel", "api.channel.create_direct_channel.invalid_user.app_error", nil, fmt.Sprintf("No users found for ids: %s. %s", userID, otherUserID), http.StatusBadRequest)
-		return nil, model.NewAppError("CreateDirectChannel", "api.channel.create_direct_channel.invalid_user.app_error", nil, otherUserId, http.StatusBadRequest)
+	}
+
+	// We are doing this because we allow a user to create a direct channel with themselves
+	if userID == otherUserID {
+		users = append(users, users[0])
+	}
+
+	// After we counted for direct channels with the same user, if we do not have two users then we failed to find one
+	if len(users) != 2 {
+		return nil, model.NewAppError("CreateDirectChannel", "api.channel.create_direct_channel.invalid_user.app_error", nil, fmt.Sprintf("No users found for ids: %s. %s", userID, otherUserID), http.StatusBadRequest)
+	}
+
+	// The potential swap dance bellow is necessary in order to guarantee determinism when creating a direct channel.
+	// When we query the database for some given user ids, the database result is not deterministic, meaning we can get
+	// the same results but in different order. In order to conform the contract of Channel.CreateDirectChannel method
+	// bellow we need to identify which user is who.
+	user := users[0]
+	otherUser := users[1]
+	if user.Id != userID {
+		user = users[1]
+		otherUser = users[0]
 	}
-	otherUser := result.Data.(*model.User)
 
 	channel, nErr := a.Srv().Store.Channel().CreateDirectChannel(user, otherUser)
 	if nErr != nil {
@@ -421,8 +426,8 @@ func (a *App) createDirectChannel(userID string, otherUserId string) (*model.Cha
 	if err := a.Srv().Store.ChannelMemberHistory().LogJoinEvent(userID, channel.Id, model.GetMillis()); err != nil {
 		return nil, model.NewAppError("CreateDirectChannel", "app.channel_member_history.log_join_event.internal_error", nil, err.Error(), http.StatusInternalServerError)
 	}
-	if userID != otherUserId {
+	if userID != otherUserID {
-		if err := a.Srv().Store.ChannelMemberHistory().LogJoinEvent(otherUserId, channel.Id, model.GetMillis()); err != nil {
+		if err := a.Srv().Store.ChannelMemberHistory().LogJoinEvent(otherUserID, channel.Id, model.GetMillis()); err != nil {
 			return nil, model.NewAppError("CreateDirectChannel", "app.channel_member_history.log_join_event.internal_error", nil, err.Error(), http.StatusInternalServerError)
 		}
 	}
@@ -2960,3 +2965,17 @@ func (a *App) GetMemberCountsByGroup(channelID string, includeTimezones bool) ([
 
 	return channelMemberCounts, nil
 }
+
+func (a *App) getDirectChannel(userID, otherUserID string) (*model.Channel, *model.AppError) {
+	channel, nErr := a.Srv().Store.Channel().GetByName("", model.GetDMNameFromIds(userID, otherUserID), true)
+	if nErr != nil {
+		var nfErr *store.ErrNotFound
+		if errors.As(nErr, &nfErr) {
+			return nil, nil
+		}
+
+		return nil, model.NewAppError("GetOrCreateDirectChannel", "web.incoming_webhook.channel.app_error", nil, nErr.Error(), http.StatusInternalServerError)
+	}
+
+	return channel, nil
+}

app/opentracing/opentracing_layer.go

@@ -6726,7 +6726,7 @@ func (a *OpenTracingAppLayer) GetOpenGraphMetadata(requestURL string) *opengraph
 	return resultVar0
 }
 
-func (a *OpenTracingAppLayer) GetOrCreateDirectChannel(userID string, otherUserId string) (*model.Channel, *model.AppError) {
+func (a *OpenTracingAppLayer) GetOrCreateDirectChannel(userID string, otherUserID string) (*model.Channel, *model.AppError) {
 	origCtx := a.ctx
 	span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.GetOrCreateDirectChannel")
 
@@ -6738,7 +6738,7 @@ func (a *OpenTracingAppLayer) GetOrCreateDirectChannel(userID string, otherUserI
 	}()
 
 	defer span.Finish()
-	resultVar0, resultVar1 := a.app.GetOrCreateDirectChannel(userID, otherUserId)
+	resultVar0, resultVar1 := a.app.GetOrCreateDirectChannel(userID, otherUserID)
 
 	if resultVar1 != nil {
 		span.LogFields(spanlog.Error(resultVar1))

store/localcachelayer/main_test.go

@@ -142,6 +142,16 @@ func getMockStore() *mocks.Store {
 	mockUserStore.On("GetAllProfilesInChannel", "123", false).Return(fakeProfilesInChannelMap, nil)
 
 	mockUserStore.On("Get", "123").Return(fakeUser[0], nil)
+	users := []*model.User{
+		fakeUser[0],
+		{
+			Id:          "456",
+			AuthData:    model.NewString("authData"),
+			AuthService: "authService",
+		},
+	}
+	mockUserStore.On("GetMany", []string{"123", "456"}).Return(users, nil)
+	mockUserStore.On("GetMany", []string{"123"}).Return(users[0:1], nil)
 	mockStore.On("User").Return(&mockUserStore)
 
 	fakeUserTeamIds := []string{"1", "2", "3"}

store/localcachelayer/user_layer.go

@@ -4,6 +4,8 @@
 package localcachelayer
 
 import (
+	"sort"
+
 	"github.com/mattermost/mattermost-server/v5/model"
 	"github.com/mattermost/mattermost-server/v5/store"
 )
@@ -155,3 +157,66 @@ func (s LocalCacheUserStore) Get(id string) (*model.User, error) {
 	s.rootStore.doStandardAddToCache(s.rootStore.userProfileByIdsCache, id, user)
 	return user, nil
 }
+
+// GetMany is a cache wrapper around the SqlStore method to get a user profiles by ids.
+// It checks if the user entries are present in the cache, returning the entries from cache
+// if it is present. Otherwise, it fetches the entries from the store and stores it in the
+// cache.
+func (s LocalCacheUserStore) GetMany(ids []string) ([]*model.User, error) {
+	// we are doing a loop instead of caching the full set in the cache because the number of permutations that we can have
+	// in this func is making caching of the total set not beneficial.
+	var cachedUsers []*model.User
+	var notCachedUserIds []string
+	uniqIDs := dedup(ids)
+
+	for _, id := range uniqIDs {
+		var cachedUser *model.User
+		if err := s.rootStore.doStandardReadCache(s.rootStore.userProfileByIdsCache, id, &cachedUser); err == nil {
+			if s.rootStore.metrics != nil {
+				s.rootStore.metrics.AddMemCacheHitCounter("Profile By Id", float64(1))
+			}
+			cachedUsers = append(cachedUsers, cachedUser)
+		} else {
+			if s.rootStore.metrics != nil {
+				s.rootStore.metrics.AddMemCacheMissCounter("Profile By Id", float64(1))
+			}
+
+			notCachedUserIds = append(notCachedUserIds, id)
+		}
+	}
+
+	if len(notCachedUserIds) > 0 {
+		dbUsers, err := s.UserStore.GetMany(notCachedUserIds)
+		if err != nil {
+			return nil, err
+		}
+		for _, user := range dbUsers {
+			s.rootStore.doStandardAddToCache(s.rootStore.userProfileByIdsCache, user.Id, user)
+			cachedUsers = append(cachedUsers, user)
+		}
+	}
+
+	return cachedUsers, nil
+}
+
+func dedup(elements []string) []string {
+	if len(elements) == 0 {
+		return elements
+	}
+
+	sort.Strings(elements)
+
+	j := 0
+	for i := 1; i < len(elements); i++ {
+		if elements[j] == elements[i] {
+			continue
+		}
+		j++
+		// preserve the original data
+		// in[i], in[j] = in[j], in[i]
+		// only set what is required
+		elements[j] = elements[i]
+	}
+
+	return elements[:j+1]
+}

store/localcachelayer/user_layer_test.go

@@ -258,3 +258,54 @@ func TestUserStoreGetCache(t *testing.T) {
 		storedUser.NotifyProps = originalProps
 	})
 }
+
+func TestUserStoreGetManyCache(t *testing.T) {
+	fakeUser := &model.User{
+		Id:          "123",
+		AuthData:    model.NewString("authData"),
+		AuthService: "authService",
+	}
+	otherFakeUser := &model.User{
+		Id:          "456",
+		AuthData:    model.NewString("authData"),
+		AuthService: "authService",
+	}
+	t.Run("first call not cached, second cached and returning same data", func(t *testing.T) {
+		mockStore := getMockStore()
+		mockCacheProvider := getMockCacheProvider()
+		cachedStore, err := NewLocalCacheLayer(mockStore, nil, nil, mockCacheProvider)
+		require.NoError(t, err)
+
+		gotUsers, err := cachedStore.User().GetMany([]string{fakeUser.Id, otherFakeUser.Id})
+		require.Nil(t, err)
+		assert.Len(t, gotUsers, 2)
+		assert.Contains(t, gotUsers, fakeUser)
+		assert.Contains(t, gotUsers, otherFakeUser)
+
+		gotUsers, err = cachedStore.User().GetMany([]string{fakeUser.Id, otherFakeUser.Id})
+		require.Nil(t, err)
+		assert.Len(t, gotUsers, 2)
+		mockStore.User().(*mocks.UserStore).AssertNumberOfCalls(t, "GetMany", 1)
+	})
+
+	t.Run("first call not cached, invalidate one user, and then check that one is cached and one is fetched from db", func(t *testing.T) {
+		mockStore := getMockStore()
+		mockCacheProvider := getMockCacheProvider()
+		cachedStore, err := NewLocalCacheLayer(mockStore, nil, nil, mockCacheProvider)
+		require.NoError(t, err)
+
+		gotUsers, err := cachedStore.User().GetMany([]string{fakeUser.Id, otherFakeUser.Id})
+		require.Nil(t, err)
+		assert.Len(t, gotUsers, 2)
+		assert.Contains(t, gotUsers, fakeUser)
+		assert.Contains(t, gotUsers, otherFakeUser)
+
+		cachedStore.User().InvalidateProfileCacheForUser("123")
+
+		gotUsers, err = cachedStore.User().GetMany([]string{fakeUser.Id, otherFakeUser.Id})
+		require.NoError(t, err)
+		assert.Len(t, gotUsers, 2)
+		mockStore.User().(*mocks.UserStore).AssertCalled(t, "GetMany", []string{"123"})
+		mockStore.User().(*mocks.UserStore).AssertNumberOfCalls(t, "GetMany", 2)
+	})
+}

store/opentracinglayer/opentracinglayer.go

@@ -8703,6 +8703,24 @@ func (s *OpenTracingLayerUserStore) GetKnownUsers(userID string) ([]string, erro
 	return result, err
 }
 
+func (s *OpenTracingLayerUserStore) GetMany(ids []string) ([]*model.User, error) {
+	origCtx := s.Root.Store.Context()
+	span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "UserStore.GetMany")
+	s.Root.Store.SetContext(newCtx)
+	defer func() {
+		s.Root.Store.SetContext(origCtx)
+	}()
+
+	defer span.Finish()
+	result, err := s.UserStore.GetMany(ids)
+	if err != nil {
+		span.LogFields(spanlog.Error(err))
+		ext.Error.Set(span, true)
+	}
+
+	return result, err
+}
+
 func (s *OpenTracingLayerUserStore) GetNewUsersForTeam(teamId string, offset int, limit int, viewRestrictions *model.ViewUsersRestrictions) ([]*model.User, error) {
 	origCtx := s.Root.Store.Context()
 	span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "UserStore.GetNewUsersForTeam")

store/retrylayer/retrylayer.go

@@ -9428,6 +9428,26 @@ func (s *RetryLayerUserStore) GetKnownUsers(userID string) ([]string, error) {
 
 }
 
+func (s *RetryLayerUserStore) GetMany(ids []string) ([]*model.User, error) {
+
+	tries := 0
+	for {
+		result, err := s.UserStore.GetMany(ids)
+		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
+		}
+	}
+
+}
+
 func (s *RetryLayerUserStore) GetNewUsersForTeam(teamId string, offset int, limit int, viewRestrictions *model.ViewUsersRestrictions) ([]*model.User, error) {
 
 	tries := 0

store/sqlstore/user_store.go

@@ -325,6 +325,22 @@ func (us SqlUserStore) UpdateMfaActive(userId string, active bool) error {
 	return nil
 }
 
+// GetMany returns a list of users for the provided list of ids
+func (us SqlUserStore) GetMany(ids []string) ([]*model.User, error) {
+	query := us.usersQuery.Where(sq.Eq{"Id": ids})
+	queryString, args, err := query.ToSql()
+	if err != nil {
+		return nil, errors.Wrap(err, "users_get_many_tosql")
+	}
+
+	var users []*model.User
+	if _, err := us.GetReplica().Select(&users, queryString, args...); err != nil {
+		return nil, errors.Wrap(err, "users_get_many_select")
+	}
+
+	return users, nil
+}
+
 func (us SqlUserStore) Get(id string) (*model.User, error) {
 	query := us.usersQuery.Where("Id = ?", id)
 	queryString, args, err := query.ToSql()

store/store.go

@@ -323,6 +323,7 @@ type UserStore interface {
 	UpdateMfaSecret(userId, secret string) error
 	UpdateMfaActive(userId string, active bool) error
 	Get(id string) (*model.User, error)
+	GetMany(ids []string) ([]*model.User, error)
 	GetAll() ([]*model.User, error)
 	ClearCaches()
 	InvalidateProfilesInChannelCacheByUser(userId string)

store/storetest/mocks/UserStore.go

@@ -603,6 +603,29 @@ func (_m *UserStore) GetKnownUsers(userID string) ([]string, error) {
 	return r0, r1
 }
 
+// GetMany provides a mock function with given fields: ids
+func (_m *UserStore) GetMany(ids []string) ([]*model.User, error) {
+	ret := _m.Called(ids)
+
+	var r0 []*model.User
+	if rf, ok := ret.Get(0).(func([]string) []*model.User); ok {
+		r0 = rf(ids)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).([]*model.User)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func([]string) error); ok {
+		r1 = rf(ids)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
 // GetNewUsersForTeam provides a mock function with given fields: teamId, offset, limit, viewRestrictions
 func (_m *UserStore) GetNewUsersForTeam(teamId string, offset int, limit int, viewRestrictions *model.ViewUsersRestrictions) ([]*model.User, error) {
 	ret := _m.Called(teamId, offset, limit, viewRestrictions)

store/timerlayer/timerlayer.go

@@ -7862,6 +7862,22 @@ func (s *TimerLayerUserStore) GetKnownUsers(userID string) ([]string, error) {
 	return result, err
 }
 
+func (s *TimerLayerUserStore) GetMany(ids []string) ([]*model.User, error) {
+	start := timemodule.Now()
+
+	result, err := s.UserStore.GetMany(ids)
+
+	elapsed := float64(timemodule.Since(start)) / float64(timemodule.Second)
+	if s.Root.Metrics != nil {
+		success := "false"
+		if err == nil {
+			success = "true"
+		}
+		s.Root.Metrics.ObserveStoreMethodDuration("UserStore.GetMany", success, elapsed)
+	}
+	return result, err
+}
+
 func (s *TimerLayerUserStore) GetNewUsersForTeam(teamId string, offset int, limit int, viewRestrictions *model.ViewUsersRestrictions) ([]*model.User, error) {
 	start := timemodule.Now()