From 457aeb8e7e0a54a97b35ced864dd741f8914ce36 Mon Sep 17 00:00:00 2001 From: Kyriakos Z <3829551+koox00@users.noreply.github.com> Date: Wed, 9 Feb 2022 22:41:30 +0200 Subject: [PATCH] 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 --- api4/apitestlib.go | 14 ++ api4/user_test.go | 149 +++++++++++++++++++-- app/channel.go | 19 ++- app/notification.go | 41 +++++- app/user.go | 11 +- store/opentracinglayer/opentracinglayer.go | 18 +++ store/retrylayer/retrylayer.go | 21 +++ store/sqlstore/thread_store.go | 26 +++- store/store.go | 1 + store/storetest/mocks/ThreadStore.go | 21 +++ store/storetest/thread_store.go | 29 ++++ store/timerlayer/timerlayer.go | 16 +++ 12 files changed, 343 insertions(+), 23 deletions(-) diff --git a/api4/apitestlib.go b/api4/apitestlib.go index 27b59fad74..c9165fead9 100644 --- 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") diff --git a/api4/user_test.go b/api4/user_test.go index 69b370f4f7..a754add770 100644 --- 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) { diff --git a/app/channel.go b/app/channel.go index 25d7e944df..ec9fcd29c8 100644 --- 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) - } } } diff --git a/app/notification.go b/app/notification.go index bb9c642a3d..d572066891 100644 --- 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) diff --git a/app/user.go b/app/user.go index 2d94355fdb..2c725ec432 100644 --- 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 diff --git a/store/opentracinglayer/opentracinglayer.go b/store/opentracinglayer/opentracinglayer.go index a069e779cc..6e9f59e9f8 100644 --- 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") diff --git a/store/retrylayer/retrylayer.go b/store/retrylayer/retrylayer.go index e71001e715..565995d2e7 100644 --- 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 diff --git a/store/sqlstore/thread_store.go b/store/sqlstore/thread_store.go index 2ccce96e4b..b79d051ca6 100644 --- 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 +} diff --git a/store/store.go b/store/store.go index fc86edd452..6594ebc51e 100644 --- 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 { diff --git a/store/storetest/mocks/ThreadStore.go b/store/storetest/mocks/ThreadStore.go index 9b2512557b..4745fe7e05 100644 --- 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) diff --git a/store/storetest/thread_store.go b/store/storetest/thread_store.go index 017acd5dbf..6ab858a826 100644 --- 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) { diff --git a/store/timerlayer/timerlayer.go b/store/timerlayer/timerlayer.go index a05f323f68..77cb50f547 100644 --- 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()