break up getThreadsForUser, leverage errgroup (#19709)

* MM-42282: handle teamId parameter correctly

As per https://community-daily.mattermost.com/core/pl/ugs7ue6e4j8a7cgegk1bxje8to, `ThreadStore.GetThreadsForUser` accepts a `teamId` parameter, but incorrectly handles an empty value of `""` as looking only for channels with an empty `teamId` (aka DMs and GMs) instead of finding all channels and effectively ignoring the team property.

Fixes: https://mattermost.atlassian.net/browse/MM-42282

* break up getThreadsForUser, leverage errgroup

This change breaks up `GetThreadsForUser` in the `ThreadStore` into its constituent `GetTotalUnreadThreads`, `GetTotalThreads`, `GetTotalUnreadMentions`, and the original `GetThreadsForUser` but now solely returning the thread structures. Instead of a monolithic method at the store level, the application layer now handles calling bulk requests, leveraging `errgroup` for simpler parallelization.

This change brings with it a few benefits:
* Simpler code, including more idiomatic usage of squirrel
* Simpler SQL, joining tables only when configured conditions require same. (No performance benefit here, since an unused LEFT JOIN generally has no overhead.)
* Discrete Grafana metrics for each store method, giving us better insight into the performance characteristics in play.
* **Performance boost**: reduced overhead when clearing push notifications.

This last point is what prompted the re-re-reactoring in this PR. As I broke things up, I realized that `clearPushNotificationSync` only used the `TotalUnreadMentions`, but asked for the count of total threads and total unread threads. By exposing the discrete methods, this code path avoids two aggregate queries. We clear notifications when marking a thread as read, and when marking a channel with unread mentions as viewed, so I expect we'll see at least a modest boost to performance from simply not wasting these cycles anymore.

No performance improvements are expected from this PR for the general case of using `GetThreadsForUser` to populate the threads view.

* never discard errors from building queries

* no MustSql

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

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)
 }

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", "")

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) {

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")

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

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

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)

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)

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)
+			})
+		}
 	})
 }
 

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()