diff --git a/app/notification_push.go b/app/notification_push.go index a69903455a..db349209e5 100644 --- a/app/notification_push.go +++ b/app/notification_push.go @@ -236,11 +236,11 @@ func (a *App) clearPushNotificationSync(currentSessionId, userID, channelID, roo msg.Badge = int(unreadCount) if msg.IsCRTEnabled { - data, err := a.Srv().Store.Thread().GetThreadsForUser(userID, "", model.GetUserThreadsOpts{TotalsOnly: true}) + totalUnreadMentions, err := a.Srv().Store.Thread().GetTotalUnreadMentions(userID, "", model.GetUserThreadsOpts{}) if err != nil { return model.NewAppError("clearPushNotificationSync", "app.user.get_thread_count_for_user.app_error", nil, err.Error(), http.StatusInternalServerError) } - msg.Badge += int(data.TotalUnreadMentions) + msg.Badge += int(totalUnreadMentions) } return a.sendPushNotificationToAllSessions(msg, userID, currentSessionId) } diff --git a/app/notification_push_test.go b/app/notification_push_test.go index 5173e38fcc..1866c83f42 100644 --- a/app/notification_push_test.go +++ b/app/notification_push_test.go @@ -1173,7 +1173,7 @@ func TestClearPushNotificationSync(t *testing.T) { mockStore.On("Preference").Return(&mockPreferenceStore) mockThreadStore := mocks.ThreadStore{} - mockThreadStore.On("GetThreadsForUser", mock.AnythingOfType("string"), mock.AnythingOfType("string"), mock.Anything).Return(&model.Threads{TotalUnreadMentions: 3}, nil) + mockThreadStore.On("GetTotalUnreadMentions", mock.AnythingOfType("string"), mock.AnythingOfType("string"), mock.Anything).Return(int64(3), nil) mockStore.On("Thread").Return(&mockThreadStore) err = th.App.clearPushNotificationSync(sess1.Id, "user1", "channel1", "") diff --git a/app/user.go b/app/user.go index 0b5b3b0551..64924a3b06 100644 --- a/app/user.go +++ b/app/user.go @@ -7,7 +7,6 @@ import ( "bytes" "context" "encoding/json" - "errors" "fmt" "io" "mime/multipart" @@ -16,6 +15,8 @@ import ( "strconv" "strings" + "github.com/pkg/errors" + "github.com/mattermost/mattermost-server/v6/app/email" "github.com/mattermost/mattermost-server/v6/app/imaging" "github.com/mattermost/mattermost-server/v6/app/request" @@ -27,6 +28,7 @@ import ( "github.com/mattermost/mattermost-server/v6/shared/mfa" "github.com/mattermost/mattermost-server/v6/shared/mlog" "github.com/mattermost/mattermost-server/v6/store" + "golang.org/x/sync/errgroup" ) const ( @@ -2291,15 +2293,61 @@ func (a *App) ConvertBotToUser(bot *model.Bot, userPatch *model.UserPatch, sysad } func (a *App) GetThreadsForUser(userID, teamID string, options model.GetUserThreadsOpts) (*model.Threads, *model.AppError) { - threads, err := a.Srv().Store.Thread().GetThreadsForUser(userID, teamID, options) - if err != nil { + var result model.Threads + var eg errgroup.Group + + eg.Go(func() error { + totalUnreadThreads, err := a.Srv().Store.Thread().GetTotalUnreadThreads(userID, teamID, options) + if err != nil { + return errors.Wrapf(err, "failed to count unread threads for user id=%s", userID) + } + result.TotalUnreadThreads = totalUnreadThreads + + return nil + }) + + eg.Go(func() error { + totalCount, err := a.Srv().Store.Thread().GetTotalThreads(userID, teamID, options) + if err != nil { + return errors.Wrapf(err, "failed to count threads for user id=%s", userID) + } + result.Total = totalCount + + return nil + }) + + eg.Go(func() error { + totalUnreadMentions, err := a.Srv().Store.Thread().GetTotalUnreadMentions(userID, teamID, options) + if err != nil { + return errors.Wrapf(err, "failed to count threads for user id=%s", userID) + } + result.TotalUnreadMentions = totalUnreadMentions + + return nil + }) + + if !options.TotalsOnly { + eg.Go(func() error { + threads, err := a.Srv().Store.Thread().GetThreadsForUser(userID, teamID, options) + if err != nil { + return errors.Wrapf(err, "failed to get threads for user id=%s", userID) + } + result.Threads = threads + + return nil + }) + } + + if err := eg.Wait(); err != nil { return nil, model.NewAppError("GetThreadsForUser", "app.user.get_threads_for_user.app_error", nil, err.Error(), http.StatusInternalServerError) } - for _, thread := range threads.Threads { + + for _, thread := range result.Threads { a.sanitizeProfiles(thread.Participants, false) thread.Post.SanitizeProps() } - return threads, nil + + return &result, nil } func (a *App) GetThreadMembershipForUser(userId, threadId string) (*model.ThreadMembership, *model.AppError) { diff --git a/store/opentracinglayer/opentracinglayer.go b/store/opentracinglayer/opentracinglayer.go index 0f2de5eb36..5d91851388 100644 --- a/store/opentracinglayer/opentracinglayer.go +++ b/store/opentracinglayer/opentracinglayer.go @@ -9393,7 +9393,7 @@ func (s *OpenTracingLayerThreadStore) GetThreadUnreadReplyCount(threadMembership return result, err } -func (s *OpenTracingLayerThreadStore) GetThreadsForUser(userId string, teamID string, opts model.GetUserThreadsOpts) (*model.Threads, error) { +func (s *OpenTracingLayerThreadStore) GetThreadsForUser(userId string, teamID string, opts model.GetUserThreadsOpts) ([]*model.ThreadResponse, error) { origCtx := s.Root.Store.Context() span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "ThreadStore.GetThreadsForUser") s.Root.Store.SetContext(newCtx) @@ -9411,6 +9411,60 @@ func (s *OpenTracingLayerThreadStore) GetThreadsForUser(userId string, teamID st return result, err } +func (s *OpenTracingLayerThreadStore) GetTotalThreads(userId string, teamID string, opts model.GetUserThreadsOpts) (int64, error) { + origCtx := s.Root.Store.Context() + span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "ThreadStore.GetTotalThreads") + s.Root.Store.SetContext(newCtx) + defer func() { + s.Root.Store.SetContext(origCtx) + }() + + defer span.Finish() + result, err := s.ThreadStore.GetTotalThreads(userId, teamID, opts) + if err != nil { + span.LogFields(spanlog.Error(err)) + ext.Error.Set(span, true) + } + + return result, err +} + +func (s *OpenTracingLayerThreadStore) GetTotalUnreadMentions(userId string, teamID string, opts model.GetUserThreadsOpts) (int64, error) { + origCtx := s.Root.Store.Context() + span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "ThreadStore.GetTotalUnreadMentions") + s.Root.Store.SetContext(newCtx) + defer func() { + s.Root.Store.SetContext(origCtx) + }() + + defer span.Finish() + result, err := s.ThreadStore.GetTotalUnreadMentions(userId, teamID, opts) + if err != nil { + span.LogFields(spanlog.Error(err)) + ext.Error.Set(span, true) + } + + return result, err +} + +func (s *OpenTracingLayerThreadStore) GetTotalUnreadThreads(userId string, teamID string, opts model.GetUserThreadsOpts) (int64, error) { + origCtx := s.Root.Store.Context() + span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "ThreadStore.GetTotalUnreadThreads") + s.Root.Store.SetContext(newCtx) + defer func() { + s.Root.Store.SetContext(origCtx) + }() + + defer span.Finish() + result, err := s.ThreadStore.GetTotalUnreadThreads(userId, teamID, opts) + if err != nil { + span.LogFields(spanlog.Error(err)) + ext.Error.Set(span, true) + } + + return result, err +} + func (s *OpenTracingLayerThreadStore) MaintainMembership(userID string, postID string, opts store.ThreadMembershipOpts) (*model.ThreadMembership, error) { origCtx := s.Root.Store.Context() span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "ThreadStore.MaintainMembership") diff --git a/store/retrylayer/retrylayer.go b/store/retrylayer/retrylayer.go index 50116b5d62..768338c8cd 100644 --- a/store/retrylayer/retrylayer.go +++ b/store/retrylayer/retrylayer.go @@ -10729,7 +10729,7 @@ func (s *RetryLayerThreadStore) GetThreadUnreadReplyCount(threadMembership *mode } -func (s *RetryLayerThreadStore) GetThreadsForUser(userId string, teamID string, opts model.GetUserThreadsOpts) (*model.Threads, error) { +func (s *RetryLayerThreadStore) GetThreadsForUser(userId string, teamID string, opts model.GetUserThreadsOpts) ([]*model.ThreadResponse, error) { tries := 0 for { @@ -10750,6 +10750,69 @@ func (s *RetryLayerThreadStore) GetThreadsForUser(userId string, teamID string, } +func (s *RetryLayerThreadStore) GetTotalThreads(userId string, teamID string, opts model.GetUserThreadsOpts) (int64, error) { + + tries := 0 + for { + result, err := s.ThreadStore.GetTotalThreads(userId, teamID, opts) + 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 *RetryLayerThreadStore) GetTotalUnreadMentions(userId string, teamID string, opts model.GetUserThreadsOpts) (int64, error) { + + tries := 0 + for { + result, err := s.ThreadStore.GetTotalUnreadMentions(userId, teamID, opts) + 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 *RetryLayerThreadStore) GetTotalUnreadThreads(userId string, teamID string, opts model.GetUserThreadsOpts) (int64, error) { + + tries := 0 + for { + result, err := s.ThreadStore.GetTotalUnreadThreads(userId, teamID, opts) + 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 *RetryLayerThreadStore) MaintainMembership(userID string, postID string, opts store.ThreadMembershipOpts) (*model.ThreadMembership, error) { tries := 0 diff --git a/store/sqlstore/thread_store.go b/store/sqlstore/thread_store.go index 602456cac2..d926b8accc 100644 --- a/store/sqlstore/thread_store.go +++ b/store/sqlstore/thread_store.go @@ -52,8 +52,146 @@ func (s *SqlThreadStore) Get(id string) (*model.Thread, error) { return &thread, nil } -func (s *SqlThreadStore) GetThreadsForUser(userId, teamId string, opts model.GetUserThreadsOpts) (*model.Threads, error) { - type JoinedThread struct { +// GetTotalUnreadThreads counts the number of unread threads for the given user, optionally +// constrained to the given team + DMs/GMs. +func (s *SqlThreadStore) GetTotalUnreadThreads(userId, teamId string, opts model.GetUserThreadsOpts) (int64, error) { + var totalUnreadThreads int64 + + query := s.getQueryBuilder(). + Select("COUNT(DISTINCT(Posts.RootId))"). + From("Posts"). + LeftJoin("ThreadMemberships ON Posts.RootId = ThreadMemberships.PostId"). + Where("Posts.CreateAt > ThreadMemberships.LastViewed"). + Where(sq.Eq{ + "ThreadMemberships.UserId": userId, + "ThreadMemberships.Following": true, + }) + + if teamId != "" { + query = query. + LeftJoin("Channels ON Posts.ChannelId = Channels.Id"). + Where(sq.Or{ + sq.Eq{"Channels.TeamId": teamId}, + sq.Eq{"Channels.TeamId": ""}, + }) + } + + if !opts.Deleted { + query = query.Where(sq.Eq{"COALESCE(Posts.DeleteAt, 0)": 0}) + } + + sql, args, err := query.ToSql() + if err != nil { + return 0, errors.Wrapf(err, "failed to build query to count unread threads for user id=%s", userId) + } + + err = s.GetMasterX().Get(&totalUnreadThreads, sql, args...) + if err != nil { + return 0, errors.Wrapf(err, "failed to count unread threads for user id=%s", userId) + } + + return totalUnreadThreads, nil +} + +// GetTotalUnreadThreads counts the number of threads for the given user, optionally constrained +// to the given team + DMs/GMs. +// +// TODO: Why do we support an Unread flag here? It's basically the same as GetTotalUnreadThreads, +// but with different comparison semantics. +func (s *SqlThreadStore) GetTotalThreads(userId, teamId string, opts model.GetUserThreadsOpts) (int64, error) { + var totalCount int64 + + query := s.getQueryBuilder(). + Select("COUNT(ThreadMemberships.PostId)"). + From("ThreadMemberships"). + LeftJoin("Threads ON Threads.PostId = ThreadMemberships.PostId"). + Where(sq.Eq{ + "ThreadMemberships.UserId": userId, + "ThreadMemberships.Following": true, + }) + + if teamId != "" { + query = query. + LeftJoin("Channels ON Threads.ChannelId = Channels.Id"). + Where(sq.Or{ + sq.Eq{"Channels.TeamId": teamId}, + sq.Eq{"Channels.TeamId": ""}, + }) + } + + if !opts.Deleted { + query = query. + LeftJoin("Posts ON Posts.Id = ThreadMemberships.PostId"). + Where(sq.Eq{"COALESCE(Posts.DeleteAt, 0)": 0}) + } + + if opts.Unread { + query = query. + Where(sq.Expr("ThreadMemberships.LastViewed < Threads.LastReplyAt")) + } + + sql, args, err := query.ToSql() + if err != nil { + return 0, errors.Wrapf(err, "failed to build query to count threads for user id=%s", userId) + } + + err = s.GetMasterX().Get(&totalCount, sql, args...) + if err != nil { + return 0, errors.Wrapf(err, "failed to count threads for user id=%s", userId) + } + + return totalCount, nil +} + +// GetTotalUnreadMentions counts the number of unread mentions for the given user, optionally +// constrained to the given team + DMs/GMs. +func (s *SqlThreadStore) GetTotalUnreadMentions(userId, teamId string, opts model.GetUserThreadsOpts) (int64, error) { + var totalUnreadMentions int64 + + query := s.getQueryBuilder(). + Select("COALESCE(SUM(ThreadMemberships.UnreadMentions),0)"). + From("ThreadMemberships"). + Where(sq.Eq{ + "ThreadMemberships.UserId": userId, + "ThreadMemberships.Following": true, + }) + + if teamId != "" { + query = query. + LeftJoin("Threads ON Threads.PostId = ThreadMemberships.PostId"). + LeftJoin("Channels ON Threads.ChannelId = Channels.Id"). + Where(sq.Or{ + sq.Eq{"Channels.TeamId": teamId}, + sq.Eq{"Channels.TeamId": ""}, + }) + } + + if !opts.Deleted { + query = query. + LeftJoin("Posts ON Posts.Id = ThreadMemberships.PostId"). + Where(sq.Eq{"COALESCE(Posts.DeleteAt, 0)": 0}) + } + + sql, args, err := query.ToSql() + if err != nil { + return 0, errors.Wrapf(err, "failed to build query to count unread mentions for user id=%s", userId) + } + + err = s.GetMasterX().Get(&totalUnreadMentions, sql, args...) + if err != nil { + return 0, errors.Wrapf(err, "failed to count unread mentions for user id=%s", userId) + } + + return totalUnreadMentions, nil +} + +func (s *SqlThreadStore) GetThreadsForUser(userId, teamId string, opts model.GetUserThreadsOpts) ([]*model.ThreadResponse, error) { + pageSize := uint64(30) + if opts.PageSize != 0 { + pageSize = opts.PageSize + } + + var threads []*struct { PostId string ReplyCount int64 LastReplyAt int64 @@ -64,236 +202,129 @@ func (s *SqlThreadStore) GetThreadsForUser(userId, teamId string, opts model.Get model.Post } - fetchConditions := sq.And{ - sq.Eq{"ThreadMemberships.UserId": userId}, - sq.Eq{"ThreadMemberships.Following": true}, - } + unreadRepliesQuery := sq. + Select("COUNT(Posts.Id)"). + From("Posts"). + Where(sq.Expr("Posts.RootId = ThreadMemberships.PostId")). + Where(sq.Expr("Posts.CreateAt > ThreadMemberships.LastViewed")) - if teamId != "" { - fetchConditions = sq.And{ - fetchConditions, - sq.Or{ - sq.Eq{"Channels.TeamId": teamId}, - sq.Eq{"Channels.TeamId": ""}, - }, - } - } if !opts.Deleted { - fetchConditions = sq.And{ - fetchConditions, - sq.Eq{"COALESCE(Posts.DeleteAt, 0)": 0}, - } + unreadRepliesQuery = unreadRepliesQuery.Where(sq.Eq{"Posts.DeleteAt": 0}) } - pageSize := uint64(30) - if opts.PageSize != 0 { - pageSize = opts.PageSize + unreadRepliesSql, unreadRepliesArgs, err := unreadRepliesQuery.ToSql() + if err != nil { + return nil, errors.Wrapf(err, "failed to build subquery to count unread replies when getting threads for user id=%s", userId) } - totalUnreadThreadsChan := make(chan store.StoreResult, 1) - totalCountChan := make(chan store.StoreResult, 1) - totalUnreadMentionsChan := make(chan store.StoreResult, 1) - var threadsChan chan store.StoreResult - if !opts.TotalsOnly { - threadsChan = make(chan store.StoreResult, 1) - } - - go func() { - repliesQuery, repliesQueryArgs, _ := s.getQueryBuilder(). - Select("COUNT(DISTINCT(Posts.RootId))"). - From("Posts"). - LeftJoin("ThreadMemberships ON Posts.RootId = ThreadMemberships.PostId"). - LeftJoin("Channels ON Posts.ChannelId = Channels.Id"). - Where(fetchConditions). - Where("Posts.CreateAt > ThreadMemberships.LastViewed").ToSql() - - var totalUnreadThreads int64 - err := s.GetMasterX().Get(&totalUnreadThreads, repliesQuery, repliesQueryArgs...) - totalUnreadThreadsChan <- store.StoreResult{Data: totalUnreadThreads, NErr: errors.Wrapf(err, "failed to get count unread on threads for user id=%s", userId)} - close(totalUnreadThreadsChan) - }() - go func() { - newFetchConditions := fetchConditions - - if opts.Unread { - newFetchConditions = sq.And{newFetchConditions, sq.Expr("ThreadMemberships.LastViewed < Threads.LastReplyAt")} - } - - threadsQuery, threadsQueryArgs, _ := s.getQueryBuilder(). - Select("COUNT(ThreadMemberships.PostId)"). - LeftJoin("Threads ON Threads.PostId = ThreadMemberships.PostId"). - LeftJoin("Channels ON Threads.ChannelId = Channels.Id"). - LeftJoin("Posts ON Posts.Id = ThreadMemberships.PostId"). - From("ThreadMemberships"). - Where(newFetchConditions).ToSql() - - var totalCount int64 - err := s.GetMasterX().Get(&totalCount, threadsQuery, threadsQueryArgs...) - totalCountChan <- store.StoreResult{Data: totalCount, NErr: err} - close(totalCountChan) - }() - go func() { - mentionsQuery, mentionsQueryArgs, _ := s.getQueryBuilder(). - Select("COALESCE(SUM(ThreadMemberships.UnreadMentions),0)"). - From("ThreadMemberships"). - LeftJoin("Threads ON Threads.PostId = ThreadMemberships.PostId"). - LeftJoin("Posts ON Posts.Id = ThreadMemberships.PostId"). - LeftJoin("Channels ON Threads.ChannelId = Channels.Id"). - Where(fetchConditions).ToSql() - - var totalUnreadMentions int64 - err := s.GetMasterX().Get(&totalUnreadMentions, mentionsQuery, mentionsQueryArgs...) - totalUnreadMentionsChan <- store.StoreResult{Data: totalUnreadMentions, NErr: err} - close(totalUnreadMentionsChan) - }() - - if !opts.TotalsOnly { - go func() { - newFetchConditions := fetchConditions - if opts.Since > 0 { - newFetchConditions = sq.And{newFetchConditions, sq.GtOrEq{"ThreadMemberships.LastUpdated": opts.Since}} - } - order := "DESC" - if opts.Before != "" { - newFetchConditions = sq.And{ - newFetchConditions, - sq.Expr(`LastReplyAt < (SELECT LastReplyAt FROM Threads WHERE PostId = ?)`, opts.Before), - } - } - if opts.After != "" { - order = "ASC" - newFetchConditions = sq.And{ - newFetchConditions, - sq.Expr(`LastReplyAt > (SELECT LastReplyAt FROM Threads WHERE PostId = ?)`, opts.After), - } - } - if opts.Unread { - newFetchConditions = sq.And{newFetchConditions, sq.Expr("ThreadMemberships.LastViewed < Threads.LastReplyAt")} - } - - unreadRepliesFetchConditions := sq.And{ - sq.Expr("Posts.RootId = ThreadMemberships.PostId"), - sq.Expr("Posts.CreateAt > ThreadMemberships.LastViewed"), - } - if !opts.Deleted { - unreadRepliesFetchConditions = sq.And{ - unreadRepliesFetchConditions, - sq.Expr("Posts.DeleteAt = 0"), - } - } - - unreadRepliesQuery, _ := sq. - Select("COUNT(Posts.Id)"). - From("Posts"). - Where(unreadRepliesFetchConditions). - MustSql() - - threads := []*JoinedThread{} - query, args, _ := s.getQueryBuilder(). - Select(`Threads.*, + query := s.getQueryBuilder(). + Select(`Threads.*, ` + postSliceCoalesceQuery() + `, ThreadMemberships.LastViewed as LastViewedAt, ThreadMemberships.UnreadMentions as UnreadMentions`). - From("Threads"). - Column(sq.Alias(sq.Expr(unreadRepliesQuery), "UnreadReplies")). - LeftJoin("Posts ON Posts.Id = Threads.PostId"). - LeftJoin("Channels ON Posts.ChannelId = Channels.Id"). - LeftJoin("ThreadMemberships ON ThreadMemberships.PostId = Threads.PostId"). - Where(newFetchConditions). - OrderBy("Threads.LastReplyAt " + order). - Limit(pageSize).ToSql() + From("Threads"). + Column(sq.Alias(sq.Expr(unreadRepliesSql, unreadRepliesArgs...), "UnreadReplies")). + Join("Posts ON Posts.Id = Threads.PostId"). + Join("ThreadMemberships ON ThreadMemberships.PostId = Threads.PostId") - err := s.GetReplicaX().Select(&threads, query, args...) - threadsChan <- store.StoreResult{Data: threads, NErr: err} - close(threadsChan) - }() - } + query = query. + Where(sq.Eq{"ThreadMemberships.UserId": userId}). + Where(sq.Eq{"ThreadMemberships.Following": true}) - totalUnreadMentionsResult := <-totalUnreadMentionsChan - if totalUnreadMentionsResult.NErr != nil { - return nil, totalUnreadMentionsResult.NErr - } - totalUnreadMentions := totalUnreadMentionsResult.Data.(int64) - - totalCountResult := <-totalCountChan - if totalCountResult.NErr != nil { - return nil, totalCountResult.NErr - } - totalCount := totalCountResult.Data.(int64) - - totalUnreadThreadsResult := <-totalUnreadThreadsChan - if totalUnreadThreadsResult.NErr != nil { - return nil, totalUnreadThreadsResult.NErr - } - totalUnreadThreads := totalUnreadThreadsResult.Data.(int64) - - // userIds is the de-duped list of participant ids from all threads. - userIds := []string{} - // userIdMap is the map of participant ids from all threads. - // Used to generate userIds - userIdMap := map[string]bool{} - - result := &model.Threads{ - Total: totalCount, - Threads: []*model.ThreadResponse{}, - TotalUnreadMentions: totalUnreadMentions, - TotalUnreadThreads: totalUnreadThreads, - } - - if !opts.TotalsOnly { - threadsResult := <-threadsChan - if threadsResult.NErr != nil { - return nil, threadsResult.NErr - } - threads := threadsResult.Data.([]*JoinedThread) - for _, thread := range threads { - for _, participantId := range thread.Participants { - if _, ok := userIdMap[participantId]; !ok { - userIdMap[participantId] = true - userIds = append(userIds, participantId) - } - } - } - // usersMap is the global profile map of all participants from all threads. - usersMap := make(map[string]*model.User, len(userIds)) - if opts.Extended { - users, err := s.User().GetProfileByIds(context.Background(), userIds, &store.UserGetByIdsOpts{}, true) - if err != nil { - return nil, errors.Wrapf(err, "failed to get threads for user id=%s", userId) - } - for _, user := range users { - usersMap[user.Id] = user - } - } else { - for _, userId := range userIds { - usersMap[userId] = &model.User{Id: userId} - } - } - - result.Threads = make([]*model.ThreadResponse, 0, len(threads)) - for _, thread := range threads { - participants := make([]*model.User, 0, len(thread.Participants)) - // We get the user profiles for only a single thread filtered from the - // global users map. - for _, participantId := range thread.Participants { - participant, ok := usersMap[participantId] - if !ok { - return nil, errors.New("cannot find thread participant with id=" + participantId) - } - participants = append(participants, participant) - } - result.Threads = append(result.Threads, &model.ThreadResponse{ - PostId: thread.PostId, - ReplyCount: thread.ReplyCount, - LastReplyAt: thread.LastReplyAt, - LastViewedAt: thread.LastViewedAt, - UnreadReplies: thread.UnreadReplies, - UnreadMentions: thread.UnreadMentions, - Participants: participants, - Post: thread.Post.ToNilIfInvalid(), + // If a team is specified, constrain to channels in that team or DMs/GMs without + // a team at all. + if teamId != "" { + query = query. + Join("Channels ON Posts.ChannelId = Channels.Id"). + Where(sq.Or{ + sq.Eq{"Channels.TeamId": teamId}, + sq.Eq{"Channels.TeamId": ""}, }) + } + + if !opts.Deleted { + query = query.Where(sq.Eq{"Posts.DeleteAt": 0}) + } + + if opts.Since > 0 { + query = query.Where(sq.GtOrEq{"ThreadMemberships.LastUpdated": opts.Since}) + } + + if opts.Unread { + query = query.Where(sq.Expr("ThreadMemberships.LastViewed < Threads.LastReplyAt")) + } + + order := "DESC" + if opts.Before != "" { + query = query.Where(sq.Expr(`LastReplyAt < (SELECT LastReplyAt FROM Threads WHERE PostId = ?)`, opts.Before)) + } + if opts.After != "" { + order = "ASC" + query = query.Where(sq.Expr(`LastReplyAt > (SELECT LastReplyAt FROM Threads WHERE PostId = ?)`, opts.After)) + } + + query = query. + OrderBy("Threads.LastReplyAt " + order). + Limit(pageSize) + + sql, args, err := query.ToSql() + if err != nil { + return nil, errors.Wrapf(err, "failed to build query to fetch threads for user id=%s", userId) + } + + err = s.GetReplicaX().Select(&threads, sql, args...) + if err != nil { + return nil, errors.Wrapf(err, "failed to fetch threads for user id=%s", userId) + } + + // Build the de-duplicated set of user ids representing participants across all threads. + var participantUserIds []string + for _, thread := range threads { + for _, participantUserId := range thread.Participants { + participantUserIds = append(participantUserIds, participantUserId) } } + participantUserIds = model.RemoveDuplicateStrings(participantUserIds) + + // Resolve the user objects for all participants, with extended metadata if requested. + allParticipants := make(map[string]*model.User, len(participantUserIds)) + if opts.Extended { + users, err := s.User().GetProfileByIds(context.Background(), participantUserIds, &store.UserGetByIdsOpts{}, true) + if err != nil { + return nil, errors.Wrapf(err, "failed to get %d thread profiles for user id=%s", len(participantUserIds), userId) + } + for _, user := range users { + allParticipants[user.Id] = user + } + } else { + for _, participantUserId := range participantUserIds { + allParticipants[participantUserId] = &model.User{Id: participantUserId} + } + } + + result := make([]*model.ThreadResponse, 0, len(threads)) + for _, thread := range threads { + // Find only this thread's participants + threadParticipants := make([]*model.User, 0, len(thread.Participants)) + for _, participantUserId := range thread.Participants { + participant, ok := allParticipants[participantUserId] + if !ok { + return nil, errors.Errorf("cannot find participant with user id=%s for thread id=%s", participantUserId, thread.PostId) + } + threadParticipants = append(threadParticipants, participant) + } + + result = append(result, &model.ThreadResponse{ + PostId: thread.PostId, + ReplyCount: thread.ReplyCount, + LastReplyAt: thread.LastReplyAt, + LastViewedAt: thread.LastViewedAt, + UnreadReplies: thread.UnreadReplies, + UnreadMentions: thread.UnreadMentions, + Participants: threadParticipants, + Post: thread.Post.ToNilIfInvalid(), + }) + } return result, nil } @@ -413,16 +444,20 @@ func (s *SqlThreadStore) GetThreadFollowers(threadID string, fetchOnlyActive boo } } - query, args, _ := s.getQueryBuilder(). + query, args, err := s.getQueryBuilder(). Select("ThreadMemberships.UserId"). From("ThreadMemberships"). Where(fetchConditions). ToSql() - err := s.GetReplicaX().Select(&users, query, args...) - if err != nil { - return nil, err + return nil, errors.Wrapf(err, "failed to build query to get thread followers for thread id=%s", threadID) } + + err = s.GetReplicaX().Select(&users, query, args...) + if err != nil { + return nil, errors.Wrapf(err, "failed to get thread followers for thread id=%s", threadID) + } + return users, nil } @@ -443,14 +478,17 @@ func (s *SqlThreadStore) GetThreadForUser(teamId string, threadMembership *model model.Post } - unreadRepliesQuery, unreadRepliesArgs := sq. + unreadRepliesQuery, unreadRepliesArgs, err := sq. Select("COUNT(Posts.Id)"). From("Posts"). Where(sq.And{ sq.Eq{"Posts.RootId": threadMembership.PostId}, sq.Gt{"Posts.CreateAt": threadMembership.LastViewed}, sq.Eq{"Posts.DeleteAt": 0}, - }).MustSql() + }).ToSql() + if err != nil { + return nil, errors.Wrapf(err, "failed to build subquery to count unread replies for getting thread for user id=%s, post id=%s", threadMembership.UserId, threadMembership.PostId) + } fetchConditions := sq.And{ sq.Or{sq.Eq{"Channels.TeamId": teamId}, sq.Eq{"Channels.TeamId": ""}}, @@ -458,22 +496,25 @@ func (s *SqlThreadStore) GetThreadForUser(teamId string, threadMembership *model } var thread JoinedThread - query, threadArgs, _ := s.getQueryBuilder(). + query, threadArgs, err := s.getQueryBuilder(). Select("Threads.*, Posts.*"). From("Threads"). Column(sq.Alias(sq.Expr(unreadRepliesQuery), "UnreadReplies")). LeftJoin("Posts ON Posts.Id = Threads.PostId"). LeftJoin("Channels ON Posts.ChannelId = Channels.Id"). Where(fetchConditions).ToSql() + if err != nil { + return nil, errors.Wrapf(err, "failed to build query to get thread for user id=%s, post id=%s", threadMembership.UserId, threadMembership.PostId) + } args := append(unreadRepliesArgs, threadArgs...) - err := s.GetReplicaX().Get(&thread, query, args...) + err = s.GetReplicaX().Get(&thread, query, args...) if err != nil { if err == sql.ErrNoRows { return nil, store.NewErrNotFound("Thread", threadMembership.PostId) } - return nil, err + return nil, errors.Wrapf(err, "failed to get thread for user id=%s, post id=%s", threadMembership.UserId, threadMembership.PostId) } thread.LastViewedAt = threadMembership.LastViewed @@ -569,7 +610,7 @@ func (s *SqlThreadStore) MarkAllAsReadByChannels(userID string, channelIDs []str func (s *SqlThreadStore) MarkAllAsRead(userId string, threadIds []string) error { timestamp := model.GetMillis() - query, args, _ := s.getQueryBuilder(). + query, args, err := s.getQueryBuilder(). Update("ThreadMemberships"). Where(sq.Eq{"UserId": userId}). Where(sq.Eq{"PostId": threadIds}). @@ -577,7 +618,12 @@ func (s *SqlThreadStore) MarkAllAsRead(userId string, threadIds []string) error Set("UnreadMentions", 0). Set("LastUpdated", model.GetMillis()). ToSql() - if _, err := s.GetMasterX().Exec(query, args...); err != nil { + if err != nil { + return errors.Wrapf(err, "failed to build query to mark %d threads as read for user id=%s", len(threadIds), userId) + } + + _, err = s.GetMasterX().Exec(query, args...) + if err != nil { return errors.Wrapf(err, "failed to mark %d threads as read for user id=%s", len(threadIds), userId) } @@ -596,7 +642,7 @@ func (s *SqlThreadStore) MarkAllAsReadByTeam(userId, teamId string) error { membershipIds = append(membershipIds, m.PostId) } timestamp := model.GetMillis() - query, args, _ := s.getQueryBuilder(). + query, args, err := s.getQueryBuilder(). Update("ThreadMemberships"). Where(sq.Eq{"PostId": membershipIds}). Where(sq.Eq{"UserId": userId}). @@ -604,7 +650,12 @@ func (s *SqlThreadStore) MarkAllAsReadByTeam(userId, teamId string) error { Set("UnreadMentions", 0). Set("LastUpdated", model.GetMillis()). ToSql() - if _, err := s.GetMasterX().Exec(query, args...); err != nil { + if err != nil { + return errors.Wrapf(err, "failed to build query to update thread read state for user id=%s", userId) + } + + _, err = s.GetMasterX().Exec(query, args...) + if err != nil { return errors.Wrapf(err, "failed to update thread read state for user id=%s", userId) } return nil @@ -612,14 +663,19 @@ func (s *SqlThreadStore) MarkAllAsReadByTeam(userId, teamId string) error { // MarkAsRead marks the given thread for the given user as unread from the given timestamp. func (s *SqlThreadStore) MarkAsRead(userId, threadId string, timestamp int64) error { - query, args, _ := s.getQueryBuilder(). + query, args, err := s.getQueryBuilder(). Update("ThreadMemberships"). Where(sq.Eq{"UserId": userId}). Where(sq.Eq{"PostId": threadId}). Set("LastViewed", timestamp). Set("LastUpdated", model.GetMillis()). ToSql() - if _, err := s.GetMasterX().Exec(query, args...); err != nil { + if err != nil { + return errors.Wrapf(err, "failed to build query to update thread read state for user id=%s thread_id=%v", userId, threadId) + } + + _, err = s.GetMasterX().Exec(query, args...) + if err != nil { return errors.Wrapf(err, "failed to update thread read state for user id=%s thread_id=%v", userId, threadId) } return nil @@ -632,9 +688,11 @@ func (s *SqlThreadStore) saveMembership(ex sqlxExecutor, membership *model.Threa Values(membership.PostId, membership.UserId, membership.Following, membership.LastViewed, membership.LastUpdated, membership.UnreadMentions). ToSql() if err != nil { - return nil, errors.Wrap(err, "threadmembership_tosql") + return nil, errors.Wrapf(err, "failed to build query to save thread membership with postid=%s userid=%s", membership.PostId, membership.UserId) } - if _, err := ex.Exec(query, args...); err != nil { + + _, err = ex.Exec(query, args...) + if err != nil { return nil, errors.Wrapf(err, "failed to save thread membership with postid=%s userid=%s", membership.PostId, membership.UserId) } @@ -658,9 +716,11 @@ func (s *SqlThreadStore) updateMembership(ex sqlxExecutor, membership *model.Thr }). ToSql() if err != nil { - return nil, errors.Wrap(err, "threadmembership_tosql") + return nil, errors.Wrapf(err, "failed to build query to update thread membership with postid=%s userid=%s", membership.PostId, membership.UserId) } - if _, err := ex.Exec(query, args...); err != nil { + + _, err = ex.Exec(query, args...) + if err != nil { return nil, errors.Wrapf(err, "failed to update thread membership with postid=%s userid=%s", membership.PostId, membership.UserId) } @@ -670,7 +730,7 @@ func (s *SqlThreadStore) updateMembership(ex sqlxExecutor, membership *model.Thr func (s *SqlThreadStore) GetMembershipsForUser(userId, teamId string) ([]*model.ThreadMembership, error) { memberships := []*model.ThreadMembership{} - query, args, _ := s.getQueryBuilder(). + query, args, err := s.getQueryBuilder(). Select("ThreadMemberships.*"). Join("Threads ON Threads.PostId = ThreadMemberships.PostId"). Join("Channels ON Threads.ChannelId = Channels.Id"). @@ -678,8 +738,11 @@ func (s *SqlThreadStore) GetMembershipsForUser(userId, teamId string) ([]*model. Where(sq.Or{sq.Eq{"Channels.TeamId": teamId}, sq.Eq{"Channels.TeamId": ""}}). Where(sq.Eq{"ThreadMemberships.UserId": userId}). ToSql() + if err != nil { + return nil, errors.Wrapf(err, "failed to build query to get thread membership with userid=%s", userId) + } - err := s.GetReplicaX().Select(&memberships, query, args...) + err = s.GetReplicaX().Select(&memberships, query, args...) if err != nil { return nil, errors.Wrapf(err, "failed to get thread membership with userid=%s", userId) } @@ -701,8 +764,9 @@ func (s *SqlThreadStore) getMembershipForUser(ex sqlxExecutor, userId, postId st }). ToSql() if err != nil { - return nil, errors.Wrap(err, "threadmembership_tosql") + return nil, errors.Wrapf(err, "failed to build query to get thread membership with userid=%s postid=%s", userId, postId) } + err = ex.Get(&membership, query, args...) if err != nil { if err == sql.ErrNoRows { @@ -710,6 +774,7 @@ func (s *SqlThreadStore) getMembershipForUser(ex sqlxExecutor, userId, postId st } return nil, errors.Wrapf(err, "failed to get thread membership with userid=%s postid=%s", userId, postId) } + return &membership, nil } @@ -722,9 +787,11 @@ func (s *SqlThreadStore) DeleteMembershipForUser(userId string, postId string) e }). ToSql() if err != nil { - return errors.Wrap(err, "threadmembership_tosql") + return errors.Wrap(err, "failed to build query to delete thread membership") } - if _, err := s.GetMasterX().Exec(query, args...); err != nil { + + _, err = s.GetMasterX().Exec(query, args...) + if err != nil { return errors.Wrap(err, "failed to delete thread membership") } @@ -825,16 +892,22 @@ func (s *SqlThreadStore) MaintainMembership(userId, postId string, opts store.Th } func (s *SqlThreadStore) GetPosts(threadId string, since int64) ([]*model.Post, error) { - query, args, _ := s.getQueryBuilder(). + query, args, err := s.getQueryBuilder(). Select("*"). From("Posts"). Where(sq.Eq{"RootId": threadId}). Where(sq.Eq{"DeleteAt": 0}). Where(sq.GtOrEq{"UpdateAt": since}).ToSql() + if err != nil { + return nil, errors.Wrap(err, "failed to build query to fetch thread posts") + } + result := []*model.Post{} - if err := s.GetReplicaX().Select(&result, query, args...); err != nil { + err = s.GetReplicaX().Select(&result, query, args...) + if err != nil { return nil, errors.Wrap(err, "failed to fetch thread posts") } + return result, nil } @@ -922,19 +995,21 @@ func (s *SqlThreadStore) DeleteOrphanedRows(limit int) (deleted int64, err error // return number of unread replies for a single thread func (s *SqlThreadStore) GetThreadUnreadReplyCount(threadMembership *model.ThreadMembership) (unreadReplies int64, err error) { - query, args := s.getQueryBuilder(). + query, args, err := s.getQueryBuilder(). Select("COUNT(Posts.Id)"). From("Posts"). Where(sq.And{ sq.Eq{"Posts.RootId": threadMembership.PostId}, sq.Gt{"Posts.CreateAt": threadMembership.LastViewed}, sq.Eq{"Posts.DeleteAt": 0}, - }).MustSql() + }).ToSql() + if err != nil { + return 0, errors.Wrapf(err, "failed to build query to count unread reply count for post id=%s", threadMembership.PostId) + } err = s.GetReplicaX().Get(&unreadReplies, query, args...) - if err != nil { - return + return 0, errors.Wrapf(err, "failed to count unread reply count for post id=%s", threadMembership.PostId) } return diff --git a/store/store.go b/store/store.go index de83e6e497..d9433dd574 100644 --- a/store/store.go +++ b/store/store.go @@ -293,7 +293,10 @@ type ThreadStore interface { GetThreadFollowers(threadID string, fetchOnlyActive bool) ([]string, error) Get(id string) (*model.Thread, error) - GetThreadsForUser(userId, teamID string, opts model.GetUserThreadsOpts) (*model.Threads, error) + GetTotalUnreadThreads(userId, teamID string, opts model.GetUserThreadsOpts) (int64, error) + GetTotalThreads(userId, teamID string, opts model.GetUserThreadsOpts) (int64, error) + GetTotalUnreadMentions(userId, teamID string, opts model.GetUserThreadsOpts) (int64, error) + GetThreadsForUser(userId, teamID string, opts model.GetUserThreadsOpts) ([]*model.ThreadResponse, error) GetThreadForUser(teamID string, threadMembership *model.ThreadMembership, extended bool) (*model.ThreadResponse, error) GetTeamsUnreadForUser(userID string, teamIDs []string) (map[string]*model.TeamUnread, error) GetPosts(threadID string, since int64) ([]*model.Post, error) diff --git a/store/storetest/mocks/ThreadStore.go b/store/storetest/mocks/ThreadStore.go index c936c4d2a3..f6992ff368 100644 --- a/store/storetest/mocks/ThreadStore.go +++ b/store/storetest/mocks/ThreadStore.go @@ -233,15 +233,15 @@ func (_m *ThreadStore) GetThreadUnreadReplyCount(threadMembership *model.ThreadM } // GetThreadsForUser provides a mock function with given fields: userId, teamID, opts -func (_m *ThreadStore) GetThreadsForUser(userId string, teamID string, opts model.GetUserThreadsOpts) (*model.Threads, error) { +func (_m *ThreadStore) GetThreadsForUser(userId string, teamID string, opts model.GetUserThreadsOpts) ([]*model.ThreadResponse, error) { ret := _m.Called(userId, teamID, opts) - var r0 *model.Threads - if rf, ok := ret.Get(0).(func(string, string, model.GetUserThreadsOpts) *model.Threads); ok { + var r0 []*model.ThreadResponse + if rf, ok := ret.Get(0).(func(string, string, model.GetUserThreadsOpts) []*model.ThreadResponse); ok { r0 = rf(userId, teamID, opts) } else { if ret.Get(0) != nil { - r0 = ret.Get(0).(*model.Threads) + r0 = ret.Get(0).([]*model.ThreadResponse) } } @@ -255,6 +255,69 @@ func (_m *ThreadStore) GetThreadsForUser(userId string, teamID string, opts mode return r0, r1 } +// GetTotalThreads provides a mock function with given fields: userId, teamID, opts +func (_m *ThreadStore) GetTotalThreads(userId string, teamID string, opts model.GetUserThreadsOpts) (int64, error) { + ret := _m.Called(userId, teamID, opts) + + var r0 int64 + if rf, ok := ret.Get(0).(func(string, string, model.GetUserThreadsOpts) int64); ok { + r0 = rf(userId, teamID, opts) + } else { + r0 = ret.Get(0).(int64) + } + + var r1 error + if rf, ok := ret.Get(1).(func(string, string, model.GetUserThreadsOpts) error); ok { + r1 = rf(userId, teamID, opts) + } else { + r1 = ret.Error(1) + } + + return r0, r1 +} + +// GetTotalUnreadMentions provides a mock function with given fields: userId, teamID, opts +func (_m *ThreadStore) GetTotalUnreadMentions(userId string, teamID string, opts model.GetUserThreadsOpts) (int64, error) { + ret := _m.Called(userId, teamID, opts) + + var r0 int64 + if rf, ok := ret.Get(0).(func(string, string, model.GetUserThreadsOpts) int64); ok { + r0 = rf(userId, teamID, opts) + } else { + r0 = ret.Get(0).(int64) + } + + var r1 error + if rf, ok := ret.Get(1).(func(string, string, model.GetUserThreadsOpts) error); ok { + r1 = rf(userId, teamID, opts) + } else { + r1 = ret.Error(1) + } + + return r0, r1 +} + +// GetTotalUnreadThreads provides a mock function with given fields: userId, teamID, opts +func (_m *ThreadStore) GetTotalUnreadThreads(userId string, teamID string, opts model.GetUserThreadsOpts) (int64, error) { + ret := _m.Called(userId, teamID, opts) + + var r0 int64 + if rf, ok := ret.Get(0).(func(string, string, model.GetUserThreadsOpts) int64); ok { + r0 = rf(userId, teamID, opts) + } else { + r0 = ret.Get(0).(int64) + } + + var r1 error + if rf, ok := ret.Get(1).(func(string, string, model.GetUserThreadsOpts) error); ok { + r1 = rf(userId, teamID, opts) + } else { + r1 = ret.Error(1) + } + + return r0, r1 +} + // MaintainMembership provides a mock function with given fields: userID, postID, opts func (_m *ThreadStore) MaintainMembership(userID string, postID string, opts store.ThreadMembershipOpts) (*model.ThreadMembership, error) { ret := _m.Called(userID, postID, opts) diff --git a/store/storetest/thread_store.go b/store/storetest/thread_store.go index 01110b3e14..1ef77780b1 100644 --- a/store/storetest/thread_store.go +++ b/store/storetest/thread_store.go @@ -5,7 +5,9 @@ package storetest import ( "context" + "sort" "testing" + "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -23,7 +25,7 @@ func TestThreadStore(t *testing.T, ss store.Store, s SqlStore) { testThreadStorePermanentDeleteBatchThreadMembershipsForRetentionPolicies(t, ss, s) }) t.Run("GetTeamsUnreadForUser", func(t *testing.T) { testGetTeamsUnreadForUser(t, ss) }) - t.Run("GetThreadsForUser", func(t *testing.T) { testGetThreadsForUser(t, ss) }) + t.Run("GetVarious", func(t *testing.T) { testVarious(t, ss) }) t.Run("MarkAllAsReadByChannels", func(t *testing.T) { testMarkAllAsReadByChannels(t, ss) }) } @@ -679,7 +681,35 @@ func testGetTeamsUnreadForUser(t *testing.T, ss store.Store) { assert.Equal(t, int64(1), teamsUnread[team2.Id].ThreadMentionCount) } -func testGetThreadsForUser(t *testing.T, ss store.Store) { +func testVarious(t *testing.T, ss store.Store) { + createThreadMembership := func(userID, postID string, isMention bool) { + t.Helper() + + opts := store.ThreadMembershipOpts{ + Following: true, + IncrementMentions: isMention, + UpdateFollowing: true, + UpdateViewedTimestamp: false, + UpdateParticipants: false, + } + _, err := ss.Thread().MaintainMembership(userID, postID, opts) + require.NoError(t, err) + } + + viewThread := func(userID, postID string) { + t.Helper() + + opts := store.ThreadMembershipOpts{ + Following: true, + IncrementMentions: false, + UpdateFollowing: true, + UpdateViewedTimestamp: true, + UpdateParticipants: false, + } + _, err := ss.Thread().MaintainMembership(userID, postID, opts) + require.NoError(t, err) + } + user1, err := ss.User().Save(&model.User{ Username: "user1" + model.NewId(), Email: MakeEmail(), @@ -750,6 +780,13 @@ func testGetThreadsForUser(t *testing.T, ss store.Store) { }) require.NoError(t, err) + team1channel1post3, err := ss.Post().Save(&model.Post{ + ChannelId: team1channel1.Id, + UserId: user1ID, + Message: model.NewRandomString(10), + }) + require.NoError(t, err) + team2channel1post1, err := ss.Post().Save(&model.Post{ ChannelId: team2channel1.Id, UserId: user1ID, @@ -757,6 +794,13 @@ func testGetThreadsForUser(t *testing.T, ss store.Store) { }) require.NoError(t, err) + team2channel1post2deleted, err := ss.Post().Save(&model.Post{ + ChannelId: team2channel1.Id, + UserId: user1ID, + Message: model.NewRandomString(10), + }) + require.NoError(t, err) + dm1post1, err := ss.Post().Save(&model.Post{ ChannelId: dm1.Id, UserId: user1ID, @@ -773,61 +817,206 @@ func testGetThreadsForUser(t *testing.T, ss store.Store) { threadStoreCreateReply(t, ss, team1channel1.Id, team1channel1post1.Id, user2ID, model.GetMillis()) threadStoreCreateReply(t, ss, team1channel1.Id, team1channel1post2.Id, user2ID, model.GetMillis()) + threadStoreCreateReply(t, ss, team1channel1.Id, team1channel1post3.Id, user2ID, model.GetMillis()) threadStoreCreateReply(t, ss, team2channel1.Id, team2channel1post1.Id, user2ID, model.GetMillis()) + threadStoreCreateReply(t, ss, team2channel1.Id, team2channel1post2deleted.Id, user2ID, model.GetMillis()) threadStoreCreateReply(t, ss, dm1.Id, dm1post1.Id, user2ID, model.GetMillis()) threadStoreCreateReply(t, ss, gm1.Id, gm1post1.Id, user2ID, model.GetMillis()) - createThreadMembership := func(userID, postID string) { - t.Helper() + // Create thread memberships, with simulated unread mentions. + createThreadMembership(user1ID, team1channel1post1.Id, false) + createThreadMembership(user1ID, team1channel1post2.Id, false) + createThreadMembership(user1ID, team1channel1post3.Id, true) + createThreadMembership(user1ID, team2channel1post1.Id, false) + createThreadMembership(user1ID, team2channel1post2deleted.Id, false) + createThreadMembership(user1ID, dm1post1.Id, false) + createThreadMembership(user1ID, gm1post1.Id, true) - opts := store.ThreadMembershipOpts{ - Following: true, - IncrementMentions: false, - UpdateFollowing: true, - UpdateViewedTimestamp: false, - UpdateParticipants: false, + // Have user1 view a subset of the threads + viewThread(user1ID, team1channel1post1.Id) + viewThread(user2ID, team1channel1post2.Id) + viewThread(user1ID, team2channel1post1.Id) + viewThread(user1ID, dm1post1.Id) + + // Add reply to a viewed thread to confirm it's unread again. + time.Sleep(1 * time.Millisecond) + threadStoreCreateReply(t, ss, team1channel1.Id, team1channel1post2.Id, user2ID, model.GetMillis()) + + err = ss.Post().Delete(team2channel1post2deleted.Id, model.GetMillis(), user1ID) + require.NoError(t, err) + + t.Run("GetTotalUnreadThreads", func(t *testing.T) { + testCases := []struct { + Description string + UserID string + TeamID string + Options model.GetUserThreadsOpts + + ExpectedThreads []*model.Post + }{ + {"all teams, user1", user1ID, "", model.GetUserThreadsOpts{}, []*model.Post{ + team1channel1post2, team1channel1post3, gm1post1, + }}, + {"team1, user1", user1ID, team1.Id, model.GetUserThreadsOpts{}, []*model.Post{ + team1channel1post2, team1channel1post3, gm1post1, + }}, + {"team1, user1, deleted", user1ID, team1.Id, model.GetUserThreadsOpts{Deleted: true}, []*model.Post{ + team1channel1post2, team1channel1post3, gm1post1, // (no deleted threads in team1) + }}, + {"team2, user1", user1ID, team2.Id, model.GetUserThreadsOpts{}, []*model.Post{ + gm1post1, // (no unread threads in team2) + }}, + {"team2, user1, deleted", user1ID, team2.Id, model.GetUserThreadsOpts{Deleted: true}, []*model.Post{ + team2channel1post2deleted, gm1post1, + }}, } - _, err := ss.Thread().MaintainMembership(userID, postID, opts) - require.NoError(t, err) - } - createThreadMembership(user1ID, team1channel1post1.Id) - createThreadMembership(user1ID, team1channel1post2.Id) - createThreadMembership(user1ID, team2channel1post1.Id) - createThreadMembership(user1ID, dm1post1.Id) - createThreadMembership(user1ID, gm1post1.Id) + for _, testCase := range testCases { + t.Run(testCase.Description, func(t *testing.T) { + totalUnreadThreads, err := ss.Thread().GetTotalUnreadThreads(testCase.UserID, testCase.TeamID, testCase.Options) + require.NoError(t, err) - t.Run("no team specified, user1", func(t *testing.T) { - threads, err := ss.Thread().GetThreadsForUser(user1ID, "", model.GetUserThreadsOpts{}) - require.NoError(t, err) - - // 2 threads from team1, 1 threads from team2, 1 dm thread, 1 gm thread - assert.EqualValues(t, 5, threads.Total) - assert.EqualValues(t, 5, threads.TotalUnreadThreads) - assert.EqualValues(t, 0, threads.TotalUnreadMentions) - assert.Len(t, threads.Threads, 5) + assert.EqualValues(t, int64(len(testCase.ExpectedThreads)), totalUnreadThreads) + }) + } }) - t.Run("team1 specified, user1", func(t *testing.T) { - threads, err := ss.Thread().GetThreadsForUser(user1ID, team1.Id, model.GetUserThreadsOpts{}) - require.NoError(t, err) + t.Run("GetTotalThreads", func(t *testing.T) { + testCases := []struct { + Description string + UserID string + TeamID string + Options model.GetUserThreadsOpts - // 2 threads from team1, 1 dm thread, 1 gm thread - assert.EqualValues(t, 4, threads.Total) - assert.EqualValues(t, 4, threads.TotalUnreadThreads) - assert.EqualValues(t, 0, threads.TotalUnreadMentions) - assert.Len(t, threads.Threads, 4) + ExpectedThreads []*model.Post + }{ + {"all teams, user1", user1ID, "", model.GetUserThreadsOpts{}, []*model.Post{ + team1channel1post1, team1channel1post2, team1channel1post3, team2channel1post1, dm1post1, gm1post1, + }}, + {"team1, user1", user1ID, team1.Id, model.GetUserThreadsOpts{}, []*model.Post{ + team1channel1post1, team1channel1post2, team1channel1post3, dm1post1, gm1post1, + }}, + {"team1, user1, unread", user1ID, team1.Id, model.GetUserThreadsOpts{Unread: true}, []*model.Post{ + team1channel1post2, team1channel1post3, gm1post1, + }}, + {"team1, user1, deleted", user1ID, team1.Id, model.GetUserThreadsOpts{Deleted: true}, []*model.Post{ + team1channel1post1, team1channel1post2, team1channel1post3, dm1post1, gm1post1, // (no deleted threads in team1) + }}, + {"team1, user1, unread + deleted", user1ID, team1.Id, model.GetUserThreadsOpts{Unread: true, Deleted: true}, []*model.Post{ + team1channel1post2, team1channel1post3, gm1post1, // (no deleted threads in team1) + }}, + {"team2, user1", user1ID, team2.Id, model.GetUserThreadsOpts{}, []*model.Post{ + team2channel1post1, dm1post1, gm1post1, + }}, + {"team2, user1, unread", user1ID, team2.Id, model.GetUserThreadsOpts{Unread: true}, []*model.Post{ + gm1post1, // (no unread in team2) + }}, + {"team2, user1, deleted", user1ID, team2.Id, model.GetUserThreadsOpts{Deleted: true}, []*model.Post{ + team2channel1post1, team2channel1post2deleted, dm1post1, gm1post1, + }}, + {"team2, user1, unread + deleted", user1ID, team2.Id, model.GetUserThreadsOpts{Unread: true, Deleted: true}, []*model.Post{ + team2channel1post2deleted, gm1post1, + }}, + } + + for _, testCase := range testCases { + t.Run(testCase.Description, func(t *testing.T) { + totalThreads, err := ss.Thread().GetTotalThreads(testCase.UserID, testCase.TeamID, testCase.Options) + require.NoError(t, err) + + assert.EqualValues(t, int64(len(testCase.ExpectedThreads)), totalThreads) + }) + } }) - t.Run("team2 specified, user1", func(t *testing.T) { - threads, err := ss.Thread().GetThreadsForUser(user1ID, team2.Id, model.GetUserThreadsOpts{}) - require.NoError(t, err) + t.Run("GetTotalUnreadMentions", func(t *testing.T) { + testCases := []struct { + Description string + UserID string + TeamID string + Options model.GetUserThreadsOpts - // 1 thread from team1, 1 dm thread, 1 gm thread - assert.EqualValues(t, 3, threads.Total) - assert.EqualValues(t, 3, threads.TotalUnreadThreads) - assert.EqualValues(t, 0, threads.TotalUnreadMentions) - assert.Len(t, threads.Threads, 3) + ExpectedThreads []*model.Post + }{ + {"all teams, user1", user1ID, "", model.GetUserThreadsOpts{}, []*model.Post{ + team1channel1post3, gm1post1, + }}, + {"team1, user1", user1ID, team1.Id, model.GetUserThreadsOpts{}, []*model.Post{ + team1channel1post3, gm1post1, + }}, + {"team2, user1", user1ID, team2.Id, model.GetUserThreadsOpts{}, []*model.Post{ + gm1post1, + }}, + } + + for _, testCase := range testCases { + t.Run(testCase.Description, func(t *testing.T) { + totalUnreadMentions, err := ss.Thread().GetTotalUnreadMentions(testCase.UserID, testCase.TeamID, testCase.Options) + require.NoError(t, err) + + assert.EqualValues(t, int64(len(testCase.ExpectedThreads)), totalUnreadMentions) + }) + } + }) + + t.Run("GetThreadsForUser", func(t *testing.T) { + testCases := []struct { + Description string + UserID string + TeamID string + Options model.GetUserThreadsOpts + + ExpectedThreads []*model.Post + }{ + {"all teams, user1", user1ID, "", model.GetUserThreadsOpts{}, []*model.Post{ + team1channel1post1, team1channel1post2, team1channel1post3, team2channel1post1, dm1post1, gm1post1, + }}, + {"team1, user1", user1ID, team1.Id, model.GetUserThreadsOpts{}, []*model.Post{ + team1channel1post1, team1channel1post2, team1channel1post3, dm1post1, gm1post1, + }}, + {"team1, user1, unread", user1ID, team1.Id, model.GetUserThreadsOpts{Unread: true}, []*model.Post{ + team1channel1post2, team1channel1post3, gm1post1, + }}, + {"team1, user1, deleted", user1ID, team1.Id, model.GetUserThreadsOpts{Deleted: true}, []*model.Post{ + team1channel1post1, team1channel1post2, team1channel1post3, dm1post1, gm1post1, // (no deleted threads in team1) + }}, + {"team1, user1, unread + deleted", user1ID, team1.Id, model.GetUserThreadsOpts{Unread: true, Deleted: true}, []*model.Post{ + team1channel1post2, team1channel1post3, gm1post1, // (no deleted threads in team1) + }}, + {"team2, user1", user1ID, team2.Id, model.GetUserThreadsOpts{}, []*model.Post{ + team2channel1post1, dm1post1, gm1post1, + }}, + {"team2, user1, unread", user1ID, team2.Id, model.GetUserThreadsOpts{Unread: true}, []*model.Post{ + gm1post1, // (no unread in team2) + }}, + {"team2, user1, deleted", user1ID, team2.Id, model.GetUserThreadsOpts{Deleted: true}, []*model.Post{ + team2channel1post1, team2channel1post2deleted, dm1post1, gm1post1, + }}, + {"team2, user1, unread + deleted", user1ID, team2.Id, model.GetUserThreadsOpts{Unread: true, Deleted: true}, []*model.Post{ + team2channel1post2deleted, gm1post1, + }}, + } + + for _, testCase := range testCases { + t.Run(testCase.Description, func(t *testing.T) { + threads, err := ss.Thread().GetThreadsForUser(testCase.UserID, testCase.TeamID, testCase.Options) + require.NoError(t, err) + + postIDs := make([]string, 0, len(threads)) + for _, thread := range threads { + postIDs = append(postIDs, thread.PostId) + } + sort.Strings(postIDs) + + expectedPostIDs := make([]string, 0, len(testCase.ExpectedThreads)) + for _, post := range testCase.ExpectedThreads { + expectedPostIDs = append(expectedPostIDs, post.Id) + } + sort.Strings(expectedPostIDs) + + assert.Equal(t, expectedPostIDs, postIDs) + }) + } }) } diff --git a/store/timerlayer/timerlayer.go b/store/timerlayer/timerlayer.go index c1ae4fbd64..f9574f0e2d 100644 --- a/store/timerlayer/timerlayer.go +++ b/store/timerlayer/timerlayer.go @@ -8455,7 +8455,7 @@ func (s *TimerLayerThreadStore) GetThreadUnreadReplyCount(threadMembership *mode return result, err } -func (s *TimerLayerThreadStore) GetThreadsForUser(userId string, teamID string, opts model.GetUserThreadsOpts) (*model.Threads, error) { +func (s *TimerLayerThreadStore) GetThreadsForUser(userId string, teamID string, opts model.GetUserThreadsOpts) ([]*model.ThreadResponse, error) { start := timemodule.Now() result, err := s.ThreadStore.GetThreadsForUser(userId, teamID, opts) @@ -8471,6 +8471,54 @@ func (s *TimerLayerThreadStore) GetThreadsForUser(userId string, teamID string, return result, err } +func (s *TimerLayerThreadStore) GetTotalThreads(userId string, teamID string, opts model.GetUserThreadsOpts) (int64, error) { + start := timemodule.Now() + + result, err := s.ThreadStore.GetTotalThreads(userId, teamID, opts) + + elapsed := float64(timemodule.Since(start)) / float64(timemodule.Second) + if s.Root.Metrics != nil { + success := "false" + if err == nil { + success = "true" + } + s.Root.Metrics.ObserveStoreMethodDuration("ThreadStore.GetTotalThreads", success, elapsed) + } + return result, err +} + +func (s *TimerLayerThreadStore) GetTotalUnreadMentions(userId string, teamID string, opts model.GetUserThreadsOpts) (int64, error) { + start := timemodule.Now() + + result, err := s.ThreadStore.GetTotalUnreadMentions(userId, teamID, opts) + + elapsed := float64(timemodule.Since(start)) / float64(timemodule.Second) + if s.Root.Metrics != nil { + success := "false" + if err == nil { + success = "true" + } + s.Root.Metrics.ObserveStoreMethodDuration("ThreadStore.GetTotalUnreadMentions", success, elapsed) + } + return result, err +} + +func (s *TimerLayerThreadStore) GetTotalUnreadThreads(userId string, teamID string, opts model.GetUserThreadsOpts) (int64, error) { + start := timemodule.Now() + + result, err := s.ThreadStore.GetTotalUnreadThreads(userId, teamID, opts) + + elapsed := float64(timemodule.Since(start)) / float64(timemodule.Second) + if s.Root.Metrics != nil { + success := "false" + if err == nil { + success = "true" + } + s.Root.Metrics.ObserveStoreMethodDuration("ThreadStore.GetTotalUnreadThreads", success, elapsed) + } + return result, err +} + func (s *TimerLayerThreadStore) MaintainMembership(userID string, postID string, opts store.ThreadMembershipOpts) (*model.ThreadMembership, error) { start := timemodule.Now()