From e50cfca2ea44afef66aa87065592da43832adfb7 Mon Sep 17 00:00:00 2001 From: Ashish Bhate Date: Wed, 14 Jul 2021 17:39:40 +0530 Subject: [PATCH] [MM-35345][MM-35494] fixes for incorrect mentions and unreads for threads and channels (#17803) Summary: The CRT backend was first released in version 5.29.0. Since then, the behaviour of the CRT feature has been refined, several bugs have been fixed, and a few sql columns have been added. Before these various fixes went in, the threads and channel membership tables have accumulated incorrect mention and unreads data. This PR fixes some of this bad historical data. Summary of fixes: - Marks threads as read for users where the last reply time of the thread is earlier than the time the user viewed the channel. Marking a thread means setting the mention count to zero and setting the last viewed at time of the the thread as the last viewed at time of the channel. This is done through a "sql migration" - Fix channel counts, i.e. the total message count, total root message count, mention count, and mention count in root messages for users who have viewed the channel after the last post in the channel. This is done as a "sql migration" Ticket Link: Parts of https://mattermost.atlassian.net/browse/MM-35494 https://mattermost.atlassian.net/browse/MM-35345 --- model/version.go | 1 + store/sqlstore/post_store.go | 12 ++- store/sqlstore/thread_store.go | 2 +- store/sqlstore/upgrade.go | 68 +++++++++++++++ store/sqlstore/upgrade_test.go | 149 ++++++++++++++++++++++++++++++++ store/storetest/thread_store.go | 21 ++++- 6 files changed, 248 insertions(+), 5 deletions(-) diff --git a/model/version.go b/model/version.go index 44ad579043..cd2591e2f8 100644 --- a/model/version.go +++ b/model/version.go @@ -13,6 +13,7 @@ import ( // It should be maintained in chronological order with most current // release at the front of the list. var versions = []string{ + "5.38.0", "5.37.0", "5.36.0", "5.35.0", diff --git a/store/sqlstore/post_store.go b/store/sqlstore/post_store.go index 087d4a1f24..b9b45efa77 100644 --- a/store/sqlstore/post_store.go +++ b/store/sqlstore/post_store.go @@ -2383,7 +2383,6 @@ func (s *SqlPostStore) updateThreadsFromPosts(transaction *gorp.Transaction, pos if len(rootIds) == 0 { return nil } - now := model.GetMillis() threadsByRootsSql, threadsByRootsArgs, _ := s.getQueryBuilder().Select("*").From("Threads").Where(sq.Eq{"PostId": rootIds}).ToSql() var threadsByRoots []*model.Thread if _, err := transaction.Select(&threadsByRoots, threadsByRootsSql, threadsByRootsArgs...); err != nil { @@ -2407,24 +2406,31 @@ func (s *SqlPostStore) updateThreadsFromPosts(transaction *gorp.Transaction, pos if err != nil { return err } + // calculate last reply at + lastReplyAt, err := transaction.SelectInt("SELECT COALESCE(MAX(Posts.CreateAt), 0) FROM Posts WHERE RootID=:RootId and DeleteAt=0", map[string]interface{}{"RootId": rootId}) + if err != nil { + return err + } // no metadata entry, create one if err := transaction.Insert(&model.Thread{ PostId: rootId, ChannelId: posts[0].ChannelId, ReplyCount: count, - LastReplyAt: now, + LastReplyAt: lastReplyAt, Participants: participants, }); err != nil { return err } } else { // metadata exists, update it - thread.LastReplyAt = now for _, post := range posts { thread.ReplyCount += 1 if !thread.Participants.Contains(post.UserId) { thread.Participants = append(thread.Participants, post.UserId) } + if post.CreateAt > thread.LastReplyAt { + thread.LastReplyAt = post.CreateAt + } } if _, err := transaction.Update(thread); err != nil { return err diff --git a/store/sqlstore/thread_store.go b/store/sqlstore/thread_store.go index 5eae3c8564..02c6d0fa0e 100644 --- a/store/sqlstore/thread_store.go +++ b/store/sqlstore/thread_store.go @@ -52,7 +52,7 @@ func threadToSlice(thread *model.Thread) []interface{} { thread.ChannelId, thread.LastReplyAt, thread.ReplyCount, - thread.Participants, + model.ArrayToJson(thread.Participants), } } diff --git a/store/sqlstore/upgrade.go b/store/sqlstore/upgrade.go index 21fc619149..d2ddfe78b1 100644 --- a/store/sqlstore/upgrade.go +++ b/store/sqlstore/upgrade.go @@ -20,6 +20,7 @@ import ( const ( CurrentSchemaVersion = Version5370 + Version5380 = "5.38.0" Version5370 = "5.37.0" Version5360 = "5.36.0" Version5350 = "5.35.0" @@ -210,6 +211,7 @@ func upgradeDatabase(sqlStore *SqlStore, currentModelVersionString string) error upgradeDatabaseToVersion535(sqlStore) upgradeDatabaseToVersion536(sqlStore) upgradeDatabaseToVersion537(sqlStore) + upgradeDatabaseToVersion538(sqlStore) return nil } @@ -1211,3 +1213,69 @@ func upgradeDatabaseToVersion537(sqlStore *SqlStore) { saveSchemaVersion(sqlStore, Version5370) } } + +func upgradeDatabaseToVersion538(sqlStore *SqlStore) { + // TODO: uncomment when the time arrive to upgrade the DB for 5.38 + // if shouldPerformUpgrade(sqlStore, Version5370, Version5380) { + fixCRTChannelMembershipCounts(sqlStore) + fixCRTThreadCountsAndUnreads(sqlStore) + + // saveSchemaVersion(sqlStore, Version5380) + // } +} + +// fixCRTThreadCountsAndUnreads Marks threads as read for users where the last +// reply time of the thread is earlier than the time the user viewed the channel. +// Marking a thread means setting the mention count to zero and setting the +// last viewed at time of the the thread as the last viewed at time +// of the channel +func fixCRTThreadCountsAndUnreads(sqlStore *SqlStore) { + threadMembershipsCTE := ` + SELECT PostId, UserId, ChannelMembers.LastViewedAt as CM_LastViewedAt, Threads.LastReplyAt + FROM Threads + INNER JOIN ChannelMembers on ChannelMembers.ChannelId = Threads.ChannelId + WHERE Threads.LastReplyAt <= ChannelMembers.LastViewedAt + ` + updateThreadMembershipQuery := ` + WITH q as (` + threadMembershipsCTE + `) + UPDATE ThreadMemberships set LastViewed = q.CM_LastViewedAt, UnreadMentions = 0, LastUpdated = :Now + FROM q WHERE ThreadMemberships.Postid = q.PostId AND ThreadMemberships.UserId = q.UserId + ` + if sqlStore.DriverName() == model.DATABASE_DRIVER_MYSQL { + updateThreadMembershipQuery = ` + UPDATE ThreadMemberships + INNER JOIN (` + threadMembershipsCTE + `) as q + ON ThreadMemberships.Postid = q.PostId AND ThreadMemberships.UserId = q.UserId + SET LastViewed = q.CM_LastViewedAt, UnreadMentions = 0, LastUpdated = :Now + ` + } + + if _, err := sqlStore.GetMaster().ExecNoTimeout(updateThreadMembershipQuery, map[string]interface{}{"Now": model.GetMillis()}); err != nil { + mlog.Error("Error updating lastviewedat and unreadmentions of threadmemberships", mlog.Err(err)) + } +} + +// fixCRTChannelMembershipCounts fixes the channel counts, i.e. the total message count, +// total root message count, mention count, and mention count in root messages for users +// who have viewed the channel after the last post in the channel +func fixCRTChannelMembershipCounts(sqlStore *SqlStore) { + channelMembershipsCountsAndMentions := ` + UPDATE ChannelMembers + SET MentionCount=0, MentionCountRoot=0, MsgCount=Channels.TotalMsgCount, MsgCountRoot=Channels.TotalMsgCountRoot, LastUpdateAt = :Now + FROM Channels + WHERE ChannelMembers.Channelid = Channels.Id AND ChannelMembers.LastViewedAt >= Channels.LastPostAt; + ` + + if sqlStore.DriverName() == model.DATABASE_DRIVER_MYSQL { + channelMembershipsCountsAndMentions = ` + UPDATE ChannelMembers + INNER JOIN Channels on Channels.Id = ChannelMembers.ChannelId + SET MentionCount=0, MentionCountRoot=0, MsgCount=Channels.TotalMsgCount, MsgCountRoot=Channels.TotalMsgCountRoot, LastUpdateAt = :Now + WHERE ChannelMembers.LastViewedAt >= Channels.LastPostAt; + ` + } + + if _, err := sqlStore.GetMaster().ExecNoTimeout(channelMembershipsCountsAndMentions, map[string]interface{}{"Now": model.GetMillis()}); err != nil { + mlog.Error("Error updating counts and unreads for channelmemberships", mlog.Err(err)) + } +} diff --git a/store/sqlstore/upgrade_test.go b/store/sqlstore/upgrade_test.go index 9623e01c37..6772268d4d 100644 --- a/store/sqlstore/upgrade_test.go +++ b/store/sqlstore/upgrade_test.go @@ -4,6 +4,7 @@ package sqlstore import ( + "context" "testing" "github.com/stretchr/testify/assert" @@ -256,3 +257,151 @@ func TestMsgCountRootMigration(t *testing.T) { }) } } + +func TestFixCRTCountsAndUnreads(t *testing.T) { + StoreTest(t, func(t *testing.T, ss store.Store) { + sqlStore := ss.(*SqlStore) + + team := createTeam(ss) + uId1 := model.NewId() + uId2 := model.NewId() + c1, err := createChannelWithLastPostAt(ss, team.Id, uId1, 0, 0, 0) + require.NoError(t, err) + createChannelMemberWithLastViewAt(ss, c1.Id, uId1, 0) + createChannelMemberWithLastViewAt(ss, c1.Id, uId2, 0) + + // Create a thread + // user2: root post 1 + // - user1: reply 1 to root post 1 + // - user2: reply 2 to root post 1 + // - user1: reply 3 to root post 1 + // - user2: reply 4 to root post 1 + rootPost1 := createPostWithTimestamp(ss, c1.Id, uId2, "", "", 1) + lastReplyAt := int64(40) + _ = createPostWithTimestamp(ss, c1.Id, uId1, rootPost1.Id, rootPost1.Id, 10) + _ = createPostWithTimestamp(ss, c1.Id, uId2, rootPost1.Id, rootPost1.Id, 20) + _ = createPostWithTimestamp(ss, c1.Id, uId1, rootPost1.Id, rootPost1.Id, 30) + _ = createPostWithTimestamp(ss, c1.Id, uId2, rootPost1.Id, rootPost1.Id, lastReplyAt) + + // Check created thread is good + goodThread1, err := ss.Thread().Get(rootPost1.Id) + require.NoError(t, err) + require.EqualValues(t, 4, goodThread1.ReplyCount) + require.EqualValues(t, lastReplyAt, goodThread1.LastReplyAt) + require.ElementsMatch(t, model.StringArray{uId1, uId2}, goodThread1.Participants) + + // Create ThreadMembership + goodThreadMembership1, err := ss.Thread().SaveMembership(&model.ThreadMembership{ + PostId: rootPost1.Id, + UserId: uId1, + Following: true, + LastViewed: lastReplyAt + 1, + LastUpdated: lastReplyAt + 1, + UnreadMentions: 0, + }) + require.NoError(t, err) + goodThreadMembership2, err := ss.Thread().SaveMembership(&model.ThreadMembership{ + PostId: rootPost1.Id, + UserId: uId2, + Following: true, + LastViewed: lastReplyAt + 1, + LastUpdated: lastReplyAt + 1, + UnreadMentions: 0, + }) + require.NoError(t, err) + + // Update channel last viewed at + // set channel as fully read for user1 + _, err = ss.Channel().UpdateLastViewedAt([]string{c1.Id}, uId1, true) + require.NoError(t, err) + // for user2 set channel as read before the last post in thread + // user2's threadmembership wont change + cm2, err := ss.Channel().GetMember(context.Background(), c1.Id, uId2) + require.NoError(t, err) + cm2.LastViewedAt = lastReplyAt - 10 + cm2, err = ss.Channel().UpdateMember(cm2) + require.NoError(t, err) + + // Update ThreadMembership with bad data, as we might expect because + // of previous bugs or changed behaviour + badThreadMembership1 := *goodThreadMembership1 + badThreadMembership1.LastViewed = 30 + badThreadMembership1.UnreadMentions = 4 + _, err = ss.Thread().UpdateMembership(&badThreadMembership1) + require.NoError(t, err) + + // Run migration to fix threads and memberships + fixCRTThreadCountsAndUnreads(sqlStore) + + // Check bad threadMemberships is fixed + fixedThreadMembership1, err := ss.Thread().GetMembershipForUser(uId1, rootPost1.Id) + require.NoError(t, err) + require.EqualValues(t, lastReplyAt, fixedThreadMembership1.LastViewed) + require.EqualValues(t, int64(0), fixedThreadMembership1.UnreadMentions) + require.NotEqual(t, goodThreadMembership1.LastUpdated, fixedThreadMembership1.LastUpdated) + + // check good threadMembership is unchanged + fixedThreadMembership2, err := ss.Thread().GetMembershipForUser(uId2, rootPost1.Id) + require.NoError(t, err) + require.Equal(t, goodThreadMembership2, fixedThreadMembership2) + }) +} + +func TestFixCRTChannelUnreads(t *testing.T) { + StoreTest(t, func(t *testing.T, ss store.Store) { + sqlStore := ss.(*SqlStore) + + team := createTeam(ss) + uId1 := model.NewId() + uId2 := model.NewId() + channelLastPostAt := int64(100) + channelMsgCount := int64(200) + channelMsgCountRoot := int64(100) + c1, err := createChannelWithLastPostAt(ss, team.Id, uId1, channelLastPostAt, channelMsgCount, channelMsgCountRoot) + require.NoError(t, err) + + // Make a membership entry + cm1, err := ss.Channel().SaveMember(&model.ChannelMember{ + ChannelId: c1.Id, + UserId: uId1, + LastViewedAt: channelLastPostAt - 50, + NotifyProps: model.GetDefaultChannelNotifyProps(), + MsgCount: 80, + MsgCountRoot: 40, + MentionCount: 5, + MentionCountRoot: 5, + }) + require.NoError(t, err) + + // make a bad membership entry + // LastViewed at newer than channel LastPostAt and with unreads and mentions + cm2, err := ss.Channel().SaveMember(&model.ChannelMember{ + ChannelId: c1.Id, + UserId: uId2, + LastViewedAt: channelLastPostAt + 50, + NotifyProps: model.GetDefaultChannelNotifyProps(), + MsgCount: 80, + MsgCountRoot: 40, + MentionCount: 5, + MentionCountRoot: 5, + }) + require.NoError(t, err) + + fixCRTChannelMembershipCounts(sqlStore) + + cm1AfterFix, err := ss.Channel().GetMember(context.Background(), c1.Id, uId1) + require.NoError(t, err) + // Migration should not affect this channelmembership + require.Equal(t, *cm1, *cm1AfterFix) + + cm2AfterFix, err := ss.Channel().GetMember(context.Background(), c1.Id, uId2) + require.NoError(t, err) + // Check that the channelmembership is fixed + require.NotEqual(t, *cm2, *cm2AfterFix) + require.EqualValues(t, 0, cm2AfterFix.MentionCount) + require.EqualValues(t, 0, cm2AfterFix.MentionCountRoot) + require.Equal(t, channelMsgCount, cm2AfterFix.MsgCount) + require.Equal(t, channelMsgCountRoot, cm2AfterFix.MsgCountRoot) + require.NotEqual(t, cm2.LastUpdateAt, cm2AfterFix.LastUpdateAt) + }) +} diff --git a/store/storetest/thread_store.go b/store/storetest/thread_store.go index 1eaba518b5..ecd445d565 100644 --- a/store/storetest/thread_store.go +++ b/store/storetest/thread_store.go @@ -16,6 +16,7 @@ import ( ) func TestThreadStore(t *testing.T, ss store.Store, s SqlStore) { + t.Run("ThreadSQLOperations", func(t *testing.T) { testThreadSQLOperations(t, ss, s) }) t.Run("ThreadStorePopulation", func(t *testing.T) { testThreadStorePopulation(t, ss) }) t.Run("ThreadStorePermanentDeleteBatchForRetentionPolicies", func(t *testing.T) { testThreadStorePermanentDeleteBatchForRetentionPolicies(t, ss) @@ -418,6 +419,24 @@ func testThreadStorePopulation(t *testing.T, ss store.Store) { }) } +func testThreadSQLOperations(t *testing.T, ss store.Store, s SqlStore) { + t.Run("Save", func(t *testing.T) { + threadToSave := &model.Thread{ + PostId: model.NewId(), + ChannelId: model.NewId(), + LastReplyAt: 10, + ReplyCount: 5, + Participants: model.StringArray{model.NewId(), model.NewId()}, + } + _, err := ss.Thread().Save(threadToSave) + require.NoError(t, err) + + th, err := ss.Thread().Get(threadToSave.PostId) + require.NoError(t, err) + require.Equal(t, threadToSave, th) + }) +} + func threadStoreCreateReply(t *testing.T, ss store.Store, channelID, postID string, createAt int64) *model.Post { reply, err := ss.Post().Save(&model.Post{ ChannelId: channelID, @@ -587,7 +606,7 @@ func testThreadStorePermanentDeleteBatchThreadMembershipsForRetentionPolicies(t require.Error(t, err, "thread membership should have been deleted by team policy") // create a new thread membership - threadMembership = createThreadMembership(userID, post.Id) + createThreadMembership(userID, post.Id) // Delete team policy and thread err = ss.RetentionPolicy().Delete(teamPolicy.ID)