[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
2021-07-14 17:39:40 +05:30
--- 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",
--- 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
--- 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),
 	}
 }

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