MM-36589: provide previous values for unreads (#18492)

* MM-36589: provide previous values for unreads To successfully figure out the new counts of mentions or unread replies for CRT we need to provide previous values alongside with the new. This is needed so we'll know how many to subtract from the total. This commit provides those numbers upon publishing websocket ThreadUpdated and ThreadReadChanged events. * Fixes errors * Removes unneeded lines * Adds GetThreadUnreadReplyCount store method Uses the new store method to get unread replies instead of GetThreadForUser. * Tests, and some changes - Adds api4 tests to test ws events - Uses sqlx instead of gorp for new store method - Fixes case where previous_unread_replies could be a negative value * Refactors tests and adds more cases * Fixes previous and current unread counts for commenter When a user posts a reply to a thread the unread counts had a couple of issues. UnreadMentions where not zeroed out, and previous unread counts where not set correctly. This commit tries to fix that by marking the thread as read for the current poster after we set previous unread counts to the websocket event data. Also MaintainMembership should zero out UnreadMentions when we are setting the thread as read. * Oops * Fixes tests by updating when maintaining membership OK, so some tests broke because we zero UnreadMentions in the membership when we UpdateViewedTimestamp, since the new timestamp is always now. Some tests broke because MaintainMembership for the commenter so that the thread is read each time commenter posts moved further down in the SendNotification method. BOTH those test cases are fixed with this commit. To be sincere though I don't understand why the second one is fixed by this. * SystemAdminUser was not part of the channel Some tests are failing because SystemAdminUser is not part of the team and channels. This commit adds user to team and channels, in an effort to fix api4/user_tests * Fixes tests * Fixes tests * Addresses review comments * Fix if clause Co-authored-by: Mattermod <mattermod@users.noreply.github.com>
2022-02-09 22:41:30 +02:00
--- a/api4/apitestlib.go
+++ b/api4/apitestlib.go
@@ -835,6 +835,13 @@ func (th *TestHelper) LinkUserToTeam(user *model.User, team *model.Team) {
 	}
 }

+func (th *TestHelper) UnlinkUserFromTeam(user *model.User, team *model.Team) {
+	err := th.App.RemoveUserFromTeam(th.Context, team.Id, user.Id, "")
+	if err != nil {
+		panic(err)
+	}
+}
+
 func (th *TestHelper) AddUserToChannel(user *model.User, channel *model.Channel) *model.ChannelMember {
 	member, err := th.App.AddUserToChannel(user, channel, false)
 	if err != nil {
@@ -843,6 +850,13 @@ func (th *TestHelper) AddUserToChannel(user *model.User, channel *model.Channel)
 	return member
 }

+func (th *TestHelper) RemoveUserFromChannel(user *model.User, channel *model.Channel) {
+	err := th.App.RemoveUserFromChannel(th.Context, user.Id, "", channel)
+	if err != nil {
+		panic(err)
+	}
+}
+
 func (th *TestHelper) GenerateTestEmail() string {
 	if *th.App.Config().EmailSettings.SMTPServer != "localhost" && os.Getenv("CI_INBUCKET_PORT") == "" {
 		return strings.ToLower("success+" + model.NewId() + "@simulator.amazonses.com")
--- a/api4/user_test.go
+++ b/api4/user_test.go
@@ -5865,7 +5865,7 @@ func TestThreadSocketEvents(t *testing.T) {
 	require.NoError(t, err)
 	CheckCreatedStatus(t, resp)

-	_, appErr := th.App.CreatePostAsUser(th.Context, &model.Post{ChannelId: th.BasicChannel.Id, Message: "testReply", UserId: th.BasicUser2.Id, RootId: rpost.Id}, th.Context.Session().Id, false)
+	replyPost, appErr := th.App.CreatePostAsUser(th.Context, &model.Post{ChannelId: th.BasicChannel.Id, Message: "testReply @" + th.BasicUser.Username, UserId: th.BasicUser2.Id, RootId: rpost.Id}, th.Context.Session().Id, false)
 	require.Nil(t, appErr)
 	defer th.App.Srv().Store.Post().PermanentDeleteByUser(th.BasicUser.Id)
 	defer th.App.Srv().Store.Post().PermanentDeleteByUser(th.BasicUser2.Id)
@@ -5918,7 +5918,7 @@ func TestThreadSocketEvents(t *testing.T) {
 		require.Truef(t, caught, "User should have received %s event", model.WebsocketEventThreadFollowChanged)
 	})

-	_, resp, err = th.Client.UpdateThreadReadForUser(th.BasicUser.Id, th.BasicTeam.Id, rpost.Id, 123)
+	_, resp, err = th.Client.UpdateThreadReadForUser(th.BasicUser.Id, th.BasicTeam.Id, rpost.Id, replyPost.CreateAt+1)
 	require.NoError(t, err)
 	CheckOKStatus(t, resp)

@@ -5930,7 +5930,14 @@ func TestThreadSocketEvents(t *testing.T) {
 				case ev := <-userWSClient.EventChannel:
 					if ev.EventType() == model.WebsocketEventThreadReadChanged {
 						caught = true
-						require.EqualValues(t, ev.GetData()["timestamp"], 123)
+
+						data := ev.GetData()
+						require.EqualValues(t, replyPost.CreateAt+1, data["timestamp"])
+						require.EqualValues(t, float64(1), data["previous_unread_replies"])
+						require.EqualValues(t, float64(1), data["previous_unread_mentions"])
+						require.EqualValues(t, float64(0), data["unread_replies"])
+						require.EqualValues(t, float64(0), data["unread_mentions"])
+
 					}
 				case <-time.After(1 * time.Second):
 					return
@@ -5941,6 +5948,124 @@ func TestThreadSocketEvents(t *testing.T) {
 		require.Truef(t, caught, "User should have received %s event", model.WebsocketEventThreadReadChanged)
 	})

+	_, resp, err = th.Client.UpdateThreadReadForUser(th.BasicUser.Id, th.BasicTeam.Id, rpost.Id, rpost.CreateAt)
+	require.NoError(t, err)
+	CheckOKStatus(t, resp)
+
+	t.Run("Listen for read event 2", func(t *testing.T) {
+		var caught bool
+		func() {
+			for {
+				select {
+				case ev := <-userWSClient.EventChannel:
+					if ev.EventType() == model.WebsocketEventThreadReadChanged {
+						caught = true
+
+						data := ev.GetData()
+						require.EqualValues(t, rpost.CreateAt, data["timestamp"])
+						require.EqualValues(t, float64(0), data["previous_unread_replies"])
+						require.EqualValues(t, float64(0), data["previous_unread_mentions"])
+						require.EqualValues(t, float64(1), data["unread_replies"])
+						require.EqualValues(t, float64(1), data["unread_mentions"])
+
+					}
+				case <-time.After(1 * time.Second):
+					return
+				}
+			}
+		}()
+
+		require.Truef(t, caught, "User should have received %s event", model.WebsocketEventThreadReadChanged)
+	})
+
+	// read the thread
+	_, resp, err = th.Client.UpdateThreadReadForUser(th.BasicUser.Id, th.BasicTeam.Id, rpost.Id, replyPost.CreateAt+1)
+	require.NoError(t, err)
+	CheckOKStatus(t, resp)
+
+	t.Run("Listen for thread updated event after create post", func(t *testing.T) {
+		testCases := []struct {
+			post        *model.Post
+			preReplies  int64
+			preMentions int64
+			replies     int64
+			mentions    int64
+		}{
+			{
+				post:        &model.Post{ChannelId: th.BasicChannel.Id, Message: "simple reply", UserId: th.BasicUser2.Id, RootId: rpost.Id},
+				preReplies:  0,
+				preMentions: 0,
+				replies:     1,
+				mentions:    0,
+			},
+			{
+				post:        &model.Post{ChannelId: th.BasicChannel.Id, Message: "mention reply 1 @" + th.BasicUser.Username, UserId: th.BasicUser2.Id, RootId: rpost.Id},
+				preReplies:  1,
+				preMentions: 0,
+				replies:     2,
+				mentions:    1,
+			},
+			{
+				post:        &model.Post{ChannelId: th.BasicChannel.Id, Message: "mention reply 2 @" + th.BasicUser.Username, UserId: th.BasicUser2.Id, RootId: rpost.Id},
+				preReplies:  2,
+				preMentions: 1,
+				replies:     3,
+				mentions:    2,
+			},
+			{
+				// posting as current user will read the thread
+				post:        &model.Post{ChannelId: th.BasicChannel.Id, Message: "self reply", UserId: th.BasicUser.Id, RootId: rpost.Id},
+				preReplies:  3,
+				preMentions: 2,
+				replies:     0,
+				mentions:    0,
+			}, {
+				post:        &model.Post{ChannelId: th.BasicChannel.Id, Message: "simple reply", UserId: th.BasicUser2.Id, RootId: rpost.Id},
+				preReplies:  0,
+				preMentions: 0,
+				replies:     1,
+				mentions:    0,
+			},
+			{
+				post:        &model.Post{ChannelId: th.BasicChannel.Id, Message: "mention reply 3 @" + th.BasicUser.Username, UserId: th.BasicUser2.Id, RootId: rpost.Id},
+				preReplies:  1,
+				preMentions: 0,
+				replies:     2,
+				mentions:    1,
+			},
+		}
+
+		for _, tc := range testCases {
+			// post a reply on the thread
+			_, appErr = th.App.CreatePostAsUser(th.Context, tc.post, th.Context.Session().Id, false)
+			require.Nil(t, appErr)
+
+			var caught bool
+			func() {
+				for {
+					select {
+					case ev := <-userWSClient.EventChannel:
+						if ev.EventType() == model.WebsocketEventThreadUpdated {
+							caught = true
+							data := ev.GetData()
+							var thread model.ThreadResponse
+							jsonErr := json.Unmarshal([]byte(data["thread"].(string)), &thread)
+							require.NoError(t, jsonErr)
+
+							require.Equal(t, tc.preReplies, int64(data["previous_unread_replies"].(float64)))
+							require.Equal(t, tc.preMentions, int64(data["previous_unread_mentions"].(float64)))
+							require.Equal(t, tc.replies, thread.UnreadReplies)
+							require.Equal(t, tc.mentions, thread.UnreadMentions)
+						}
+					case <-time.After(1 * time.Second):
+						return
+					}
+				}
+			}()
+
+			require.Truef(t, caught, "User should have received %s event", model.WebsocketEventThreadUpdated)
+		}
+	})
 }

 func TestFollowThreads(t *testing.T) {
@@ -6052,6 +6177,10 @@ func postAndCheck(t *testing.T, client *model.Client4, post *model.Post) (*model
 func TestMaintainUnreadRepliesInThread(t *testing.T) {
 	th := Setup(t).InitBasic()
 	defer th.TearDown()
+	th.LinkUserToTeam(th.SystemAdminUser, th.BasicTeam)
+	defer th.UnlinkUserFromTeam(th.SystemAdminUser, th.BasicTeam)
+	th.AddUserToChannel(th.SystemAdminUser, th.BasicChannel)
+	defer th.RemoveUserFromChannel(th.SystemAdminUser, th.BasicChannel)
 	os.Setenv("MM_FEATUREFLAGS_COLLAPSEDTHREADS", "true")
 	defer os.Unsetenv("MM_FEATUREFLAGS_COLLAPSEDTHREADS")
 	th.App.UpdateConfig(func(cfg *model.Config) {
@@ -6189,6 +6318,10 @@ func TestSingleThreadGet(t *testing.T) {
 func TestMaintainUnreadMentionsInThread(t *testing.T) {
 	th := Setup(t).InitBasic()
 	defer th.TearDown()
+	th.LinkUserToTeam(th.SystemAdminUser, th.BasicTeam)
+	defer th.UnlinkUserFromTeam(th.SystemAdminUser, th.BasicTeam)
+	th.AddUserToChannel(th.SystemAdminUser, th.BasicChannel)
+	defer th.RemoveUserFromChannel(th.SystemAdminUser, th.BasicChannel)
 	client := th.Client
 	os.Setenv("MM_FEATUREFLAGS_COLLAPSEDTHREADS", "true")
 	defer os.Unsetenv("MM_FEATUREFLAGS_COLLAPSEDTHREADS")
@@ -6228,27 +6361,27 @@ func TestMaintainUnreadMentionsInThread(t *testing.T) {

 	// test self mention, shouldn't increase mention count
 	postAndCheck(t, client, &model.Post{ChannelId: th.BasicChannel.Id, Message: "testReply @" + th.BasicUser.Username, RootId: rpost.Id})
-	// count shouldn't increase
-	checkThreadList(th.Client, th.BasicUser.Id, 1, 1)
+	// mention should be 0 after self reply
+	checkThreadList(th.Client, th.BasicUser.Id, 0, 1)

 	// test DM
 	dm := th.CreateDmChannel(th.SystemAdminUser)
 	dm_root_post, _ := postAndCheck(t, client, &model.Post{ChannelId: dm.Id, Message: "hi @" + th.SystemAdminUser.Username})

 	// no changes
-	checkThreadList(th.Client, th.BasicUser.Id, 1, 1)
+	checkThreadList(th.Client, th.BasicUser.Id, 0, 1)

 	// post reply by the same user
 	postAndCheck(t, client, &model.Post{ChannelId: dm.Id, Message: "how are you", RootId: dm_root_post.Id})

 	// thread created
-	checkThreadList(th.Client, th.BasicUser.Id, 1, 2)
+	checkThreadList(th.Client, th.BasicUser.Id, 0, 2)

 	// post two replies by another user, without mentions. mention count should still increase since this is a DM
 	postAndCheck(t, th.SystemAdminClient, &model.Post{ChannelId: dm.Id, Message: "msg1", RootId: dm_root_post.Id})
 	postAndCheck(t, th.SystemAdminClient, &model.Post{ChannelId: dm.Id, Message: "msg2", RootId: dm_root_post.Id})
 	// expect increment by two mentions
-	checkThreadList(th.Client, th.BasicUser.Id, 3, 2)
+	checkThreadList(th.Client, th.BasicUser.Id, 2, 2)
 }

 func TestReadThreads(t *testing.T) {
--- a/app/channel.go
+++ b/app/channel.go
@@ -2553,15 +2553,23 @@ func (a *App) MarkChannelAsUnreadFromPost(postID string, userID string, collapse
 		if storeErr != nil && !errors.As(storeErr, &nfErr) {
 			return nil, model.NewAppError("MarkChannelAsUnreadFromPost", "app.channel.update_last_viewed_at_post.app_error", nil, storeErr.Error(), http.StatusInternalServerError)
 		}
+		var opts store.ThreadMembershipOpts
 		// if this post was not followed before, create thread membership and update mention count
 		if threadMembership == nil {
-			opts := store.ThreadMembershipOpts{
+			opts = store.ThreadMembershipOpts{
 				Following:             followThread,
 				IncrementMentions:     false,
 				UpdateFollowing:       true,
 				UpdateViewedTimestamp: true,
 				UpdateParticipants:    false,
 			}
+		} else if !threadMembership.Following && followThread {
+			opts = store.ThreadMembershipOpts{
+				Following:       true,
+				UpdateFollowing: true,
+			}
+		}
+		if opts.UpdateFollowing || threadMembership == nil {
 			threadMembership, storeErr = a.Srv().Store.Thread().MaintainMembership(user.Id, threadId, opts)
 			if storeErr != nil && !errors.As(storeErr, &nfErr) {
 				return nil, model.NewAppError("MarkChannelAsUnreadFromPost", "app.channel.update_last_viewed_at_post.app_error", nil, storeErr.Error(), http.StatusInternalServerError)
@@ -2597,15 +2605,6 @@ func (a *App) MarkChannelAsUnreadFromPost(postID string, userID string, collapse
 				message.Add("thread", string(payload))
 				a.Publish(message)
 			}
-		} else if !threadMembership.Following && followThread {
-			opts := store.ThreadMembershipOpts{
-				Following:       true,
-				UpdateFollowing: true,
-			}
-			_, storeErr = a.Srv().Store.Thread().MaintainMembership(user.Id, threadId, opts)
-			if storeErr != nil && !errors.As(storeErr, &nfErr) {
-				return nil, model.NewAppError("MarkChannelAsUnreadFromPost", "app.channel.update_last_viewed_at_post.app_error", nil, storeErr.Error(), http.StatusInternalServerError)
-			}
 		}
 	}

--- a/app/notification.go
+++ b/app/notification.go
@@ -263,7 +263,7 @@ func (a *App) SendNotifications(post *model.Post, team *model.Team, channel *mod
 					Following:             true,
 					IncrementMentions:     incrementMentions,
 					UpdateFollowing:       updateFollowing,
-					UpdateViewedTimestamp: userID == post.UserId,
+					UpdateViewedTimestamp: false,
 					UpdateParticipants:    userID == post.UserId,
 				}
 				threadMembership, err := a.Srv().Store.Thread().MaintainMembership(userID, post.RootId, opts)
@@ -612,6 +612,24 @@ func (a *App) SendNotifications(post *model.Post, team *model.Team, channel *mod
 					return nil, errors.Wrapf(err, "cannot get thread %q for user %q", post.RootId, uid)
 				}
 				if userThread != nil {
+					previousUnreadMentions := userThread.UnreadMentions
+					previousUnreadReplies := max(userThread.UnreadReplies-1, 0)
+					if mentions.isUserMentioned(uid) {
+						previousUnreadMentions = max(userThread.UnreadMentions-1, 0)
+					}
+					// set LastViewed to now for commenter
+					if uid == post.UserId {
+						opts := store.ThreadMembershipOpts{
+							UpdateViewedTimestamp: true,
+						}
+						// should set unread mentions, and unread replies to 0
+						_, err = a.Srv().Store.Thread().MaintainMembership(uid, post.RootId, opts)
+						if err != nil {
+							return nil, errors.Wrapf(err, "cannot maintain thread membership %q for user %q", post.RootId, uid)
+						}
+						userThread.UnreadMentions = 0
+						userThread.UnreadReplies = 0
+					}
 					a.sanitizeProfiles(userThread.Participants, false)
 					userThread.Post.SanitizeProps()

@@ -626,6 +644,8 @@ func (a *App) SendNotifications(post *model.Post, team *model.Team, channel *mod
 						mlog.Warn("Failed to encode thread to JSON")
 					}
 					message.Add("thread", string(payload))
+					message.Add("previous_unread_mentions", previousUnreadMentions)
+					message.Add("previous_unread_replies", previousUnreadReplies)

 					a.Publish(message)
 				}
@@ -635,6 +655,13 @@ func (a *App) SendNotifications(post *model.Post, team *model.Team, channel *mod
 	return mentionedUsersList, nil
 }

+func max(a, b int64) int64 {
+	if a < b {
+		return b
+	}
+	return a
+}
+
 func (a *App) userAllowsEmail(user *model.User, channelMemberNotificationProps model.StringMap, post *model.Post) bool {
 	userAllowsEmails := user.NotifyProps[model.EmailNotifyProp] != "false"

@@ -891,6 +918,18 @@ const (
 	GroupMention
 )

+func (m *ExplicitMentions) isUserMentioned(userID string) bool {
+	if _, ok := m.Mentions[userID]; ok {
+		return true
+	}
+
+	if _, ok := m.GroupMentions[userID]; ok {
+		return true
+	}
+
+	return m.HereMentioned || m.AllMentioned || m.ChannelMentioned
+}
+
 func (m *ExplicitMentions) addMention(userID string, mentionType MentionType) {
 	if m.Mentions == nil {
 		m.Mentions = make(map[string]MentionType)
--- a/app/user.go
+++ b/app/user.go
@@ -2393,6 +2393,12 @@ func (a *App) UpdateThreadReadForUser(currentSessionId, userID, teamID, threadID
 		return nil, model.NewAppError("UpdateThreadReadForUser", "app.user.update_thread_read_for_user.app_error", nil, storeErr.Error(), http.StatusInternalServerError)
 	}

+	previousUnreadMentions := membership.UnreadMentions
+	previousUnreadReplies, nErr := a.Srv().Store.Thread().GetThreadUnreadReplyCount(membership)
+	if nErr != nil {
+		return nil, model.NewAppError("UpdateThreadReadForUser", "app.user.update_thread_read_for_user.app_error", nil, nErr.Error(), http.StatusInternalServerError)
+	}
+
 	post, err := a.GetSinglePost(threadID)
 	if err != nil {
 		return nil, err
@@ -2401,12 +2407,13 @@ func (a *App) UpdateThreadReadForUser(currentSessionId, userID, teamID, threadID
 	if err != nil {
 		return nil, err
 	}
-	_, nErr := a.Srv().Store.Thread().UpdateMembership(membership)
+	_, nErr = a.Srv().Store.Thread().UpdateMembership(membership)
 	if nErr != nil {
 		return nil, model.NewAppError("UpdateThreadReadForUser", "app.user.update_thread_read_for_user.app_error", nil, nErr.Error(), http.StatusInternalServerError)
 	}

 	membership.LastViewed = timestamp
+
 	nErr = a.Srv().Store.Thread().MarkAsRead(userID, threadID, timestamp)
 	if nErr != nil {
 		return nil, model.NewAppError("UpdateThreadReadForUser", "app.user.update_thread_read_for_user.app_error", nil, nErr.Error(), http.StatusInternalServerError)
@@ -2426,6 +2433,8 @@ func (a *App) UpdateThreadReadForUser(currentSessionId, userID, teamID, threadID
 	message.Add("timestamp", timestamp)
 	message.Add("unread_mentions", membership.UnreadMentions)
 	message.Add("unread_replies", thread.UnreadReplies)
+	message.Add("previous_unread_mentions", previousUnreadMentions)
+	message.Add("previous_unread_replies", previousUnreadReplies)
 	message.Add("channel_id", post.ChannelId)
 	a.Publish(message)
 	return thread, nil
--- a/store/opentracinglayer/opentracinglayer.go
+++ b/store/opentracinglayer/opentracinglayer.go
@@ -9249,6 +9249,24 @@ func (s *OpenTracingLayerThreadStore) GetThreadForUser(teamID string, threadMemb
 	return result, err
 }

+func (s *OpenTracingLayerThreadStore) GetThreadUnreadReplyCount(threadMembership *model.ThreadMembership) (int64, error) {
+	origCtx := s.Root.Store.Context()
+	span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "ThreadStore.GetThreadUnreadReplyCount")
+	s.Root.Store.SetContext(newCtx)
+	defer func() {
+		s.Root.Store.SetContext(origCtx)
+	}()
+
+	defer span.Finish()
+	result, err := s.ThreadStore.GetThreadUnreadReplyCount(threadMembership)
+	if err != nil {
+		span.LogFields(spanlog.Error(err))
+		ext.Error.Set(span, true)
+	}
+
+	return result, err
+}
+
 func (s *OpenTracingLayerThreadStore) GetThreadsForUser(userId string, teamID string, opts model.GetUserThreadsOpts) (*model.Threads, error) {
 	origCtx := s.Root.Store.Context()
 	span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "ThreadStore.GetThreadsForUser")
--- a/store/retrylayer/retrylayer.go
+++ b/store/retrylayer/retrylayer.go
@@ -10561,6 +10561,27 @@ func (s *RetryLayerThreadStore) GetThreadForUser(teamID string, threadMembership

 }

+func (s *RetryLayerThreadStore) GetThreadUnreadReplyCount(threadMembership *model.ThreadMembership) (int64, error) {
+
+	tries := 0
+	for {
+		result, err := s.ThreadStore.GetThreadUnreadReplyCount(threadMembership)
+		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) GetThreadsForUser(userId string, teamID string, opts model.GetUserThreadsOpts) (*model.Threads, error) {

 	tries := 0
--- a/store/sqlstore/thread_store.go
+++ b/store/sqlstore/thread_store.go
@@ -644,11 +644,11 @@ func (s *SqlThreadStore) MaintainMembership(userId, postId string, opts store.Th
 			}
 			if opts.UpdateViewedTimestamp {
 				membership.LastViewed = now
-			}
-			membership.LastUpdated = now
-			if opts.IncrementMentions {
+				membership.UnreadMentions = 0
+			} else if opts.IncrementMentions {
 				membership.UnreadMentions += 1
 			}
+			membership.LastUpdated = now
 			if _, err = s.updateMembership(trx, membership); err != nil {
 				return nil, err
 			}
@@ -849,3 +849,23 @@ func (s *SqlThreadStore) DeleteOrphanedRows(limit int) (deleted int64, err error
 	deleted = rpcDeleted + rptDeleted
 	return
 }
+
+// return number of unread replies for a single thread
+func (s *SqlThreadStore) GetThreadUnreadReplyCount(threadMembership *model.ThreadMembership) (unreadReplies int64, err error) {
+	query, args := 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()
+
+	err = s.GetReplicaX().Get(&unreadReplies, query, args...)
+
+	if err != nil {
+		return
+	}
+
+	return
+}
--- a/store/store.go
+++ b/store/store.go
@@ -316,6 +316,7 @@ type ThreadStore interface {
 	PermanentDeleteBatchForRetentionPolicies(now, globalPolicyEndTime, limit int64, cursor model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error)
 	PermanentDeleteBatchThreadMembershipsForRetentionPolicies(now, globalPolicyEndTime, limit int64, cursor model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error)
 	DeleteOrphanedRows(limit int) (deleted int64, err error)
+	GetThreadUnreadReplyCount(threadMembership *model.ThreadMembership) (int64, error)
 }

 type PostStore interface {
--- a/store/storetest/mocks/ThreadStore.go
+++ b/store/storetest/mocks/ThreadStore.go
@@ -225,6 +225,27 @@ func (_m *ThreadStore) GetThreadForUser(teamID string, threadMembership *model.T
 	return r0, r1
 }

+// GetThreadUnreadReplyCount provides a mock function with given fields: threadMembership
+func (_m *ThreadStore) GetThreadUnreadReplyCount(threadMembership *model.ThreadMembership) (int64, error) {
+	ret := _m.Called(threadMembership)
+
+	var r0 int64
+	if rf, ok := ret.Get(0).(func(*model.ThreadMembership) int64); ok {
+		r0 = rf(threadMembership)
+	} else {
+		r0 = ret.Get(0).(int64)
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(*model.ThreadMembership) error); ok {
+		r1 = rf(threadMembership)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
 // 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) {
 	ret := _m.Called(userId, teamID, opts)
--- a/store/storetest/thread_store.go
+++ b/store/storetest/thread_store.go
@@ -470,6 +470,35 @@ func testThreadStorePopulation(t *testing.T, ss store.Store) {
 			require.NotNil(t, user)
 		}
 	})
+	t.Run("Get unread reply counts for thread", func(t *testing.T) {
+		newPosts := makeSomePosts()
+		opts := store.ThreadMembershipOpts{
+			Following:             true,
+			IncrementMentions:     false,
+			UpdateFollowing:       true,
+			UpdateViewedTimestamp: true,
+			UpdateParticipants:    false,
+		}
+
+		_, e := ss.Thread().MaintainMembership(newPosts[0].UserId, newPosts[0].Id, opts)
+		require.NoError(t, e)
+
+		m, err1 := ss.Thread().GetMembershipForUser(newPosts[0].UserId, newPosts[0].Id)
+		require.NoError(t, err1)
+
+		unreads, err := ss.Thread().GetThreadUnreadReplyCount(m)
+		require.NoError(t, err)
+		require.Equal(t, int64(0), unreads)
+
+		err = ss.Thread().MarkAsRead(newPosts[0].UserId, newPosts[0].Id, newPosts[0].CreateAt)
+		require.NoError(t, err)
+		m, err = ss.Thread().GetMembershipForUser(newPosts[0].UserId, newPosts[0].Id)
+		require.NoError(t, err)
+
+		unreads, err = ss.Thread().GetThreadUnreadReplyCount(m)
+		require.NoError(t, err)
+		require.Equal(t, int64(2), unreads)
+	})
 }

 func testThreadSQLOperations(t *testing.T, ss store.Store, s SqlStore) {
--- a/store/timerlayer/timerlayer.go
+++ b/store/timerlayer/timerlayer.go
@@ -8327,6 +8327,22 @@ func (s *TimerLayerThreadStore) GetThreadForUser(teamID string, threadMembership
 	return result, err
 }

+func (s *TimerLayerThreadStore) GetThreadUnreadReplyCount(threadMembership *model.ThreadMembership) (int64, error) {
+	start := timemodule.Now()
+
+	result, err := s.ThreadStore.GetThreadUnreadReplyCount(threadMembership)
+
+	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.GetThreadUnreadReplyCount", success, elapsed)
+	}
+	return result, err
+}
+
 func (s *TimerLayerThreadStore) GetThreadsForUser(userId string, teamID string, opts model.GetUserThreadsOpts) (*model.Threads, error) {
 	start := timemodule.Now()