[MM-35345]: fix unreads in threads and migration (#17938)

Summary
Fix unreads query for threads to check deleted option
Normalize unreads query to look for threads after given time
Fix migration to only run once, and increment timestamp by one so that threads are definitely marked as read.

Ticket Link
Fixes issues with https://mattermost.atlassian.net/browse/MM-35345
And possibly some issues with thread unreads.

store/sqlstore/thread_store.go

@@ -130,7 +130,6 @@ func (s *SqlThreadStore) GetThreadsForUser(userId, teamId string, opts model.Get
 		model.Post
 	}
 
-	unreadRepliesQuery := "SELECT COUNT(Posts.Id) From Posts Where Posts.RootId=ThreadMemberships.PostId AND Posts.CreateAt >= ThreadMemberships.LastViewed"
 	fetchConditions := sq.And{
 		sq.Or{sq.Eq{"Channels.TeamId": teamId}, sq.Eq{"Channels.TeamId": ""}},
 		sq.Eq{"ThreadMemberships.UserId": userId},
@@ -159,7 +158,7 @@ func (s *SqlThreadStore) GetThreadsForUser(userId, teamId string, opts model.Get
 			LeftJoin("ThreadMemberships ON Posts.RootId = ThreadMemberships.PostId").
 			LeftJoin("Channels ON Posts.ChannelId = Channels.Id").
 			Where(fetchConditions).
-			Where("Posts.CreateAt >= ThreadMemberships.LastViewed").ToSql()
+			Where("Posts.CreateAt > ThreadMemberships.LastViewed").ToSql()
 
 		totalUnreadThreads, err := s.GetMaster().SelectInt(repliesQuery, repliesQueryArgs...)
 		totalUnreadThreadsChan <- store.StoreResult{Data: totalUnreadThreads, NErr: errors.Wrapf(err, "failed to get count unread on threads for user id=%s", userId)}
@@ -218,6 +217,24 @@ func (s *SqlThreadStore) GetThreadsForUser(userId, teamId string, opts model.Get
 		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()
+
 		var threads []*JoinedThread
 		query, args, _ := s.getQueryBuilder().
 			Select(`Threads.*,
@@ -358,7 +375,7 @@ func (s *SqlThreadStore) GetThreadForUser(teamId string, threadMembership *model
 		From("Posts").
 		Where(sq.And{
 			sq.Eq{"Posts.RootId": threadMembership.PostId},
-			sq.GtOrEq{"Posts.CreateAt": threadMembership.LastViewed},
+			sq.Gt{"Posts.CreateAt": threadMembership.LastViewed},
 			sq.Eq{"Posts.DeleteAt": 0},
 		}).MustSql()
 
@@ -651,8 +668,8 @@ func (s *SqlThreadStore) CollectThreadsWithNewerReplies(userId string, channelId
 			sq.Eq{"Threads.ChannelId": channelIds},
 			sq.Eq{"ChannelMembers.UserId": userId},
 			sq.Or{
-				sq.Expr("Threads.LastReplyAt >= ChannelMembers.LastViewedAt"),
-				sq.GtOrEq{"Threads.LastReplyAt": timestamp},
+				sq.Expr("Threads.LastReplyAt > ChannelMembers.LastViewedAt"),
+				sq.Gt{"Threads.LastReplyAt": timestamp},
 			},
 		}).
 		ToSql()

store/sqlstore/upgrade.go

@@ -1230,6 +1230,11 @@ func upgradeDatabaseToVersion538(sqlStore *SqlStore) {
 // last viewed at time of the the thread as the last viewed at time
 // of the channel
 func fixCRTThreadCountsAndUnreads(sqlStore *SqlStore) {
+	var system model.System
+	if err := sqlStore.GetMaster().SelectOne(&system, "SELECT * FROM Systems WHERE Name = 'CRTThreadCountsAndUnreadsMigrationComplete'"); err == nil {
+		return
+	}
+
 	threadMembershipsCTE := `
 		SELECT PostId, UserId, ChannelMembers.LastViewedAt as CM_LastViewedAt, Threads.LastReplyAt
 		FROM Threads
@@ -1238,7 +1243,7 @@ func fixCRTThreadCountsAndUnreads(sqlStore *SqlStore) {
 	`
 	updateThreadMembershipQuery := `
 		WITH q as (` + threadMembershipsCTE + `)
-		UPDATE ThreadMemberships set LastViewed = q.CM_LastViewedAt, UnreadMentions = 0, LastUpdated = :Now
+		UPDATE ThreadMemberships set LastViewed = q.CM_LastViewedAt + 1, UnreadMentions = 0, LastUpdated = :Now
 		FROM q WHERE ThreadMemberships.Postid = q.PostId AND ThreadMemberships.UserId = q.UserId
 	`
 	if sqlStore.DriverName() == model.DATABASE_DRIVER_MYSQL {
@@ -1246,12 +1251,17 @@ func fixCRTThreadCountsAndUnreads(sqlStore *SqlStore) {
 			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
+			SET LastViewed = q.CM_LastViewedAt + 1, 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))
+		return
+	}
+
+	if _, err := sqlStore.GetMaster().ExecNoTimeout("INSERT INTO Systems VALUES ('CRTThreadCountsAndUnreadsMigrationComplete', 'true')"); err != nil {
+		mlog.Error("Error marking migration as done", mlog.Err(err))
 	}
 }
 
@@ -1259,6 +1269,10 @@ func fixCRTThreadCountsAndUnreads(sqlStore *SqlStore) {
 // 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) {
+	var system model.System
+	if err := sqlStore.GetMaster().SelectOne(&system, "SELECT * FROM Systems WHERE Name = 'CRTChannelMembershipCountsMigrationComplete'"); err == nil {
+		return
+	}
 	channelMembershipsCountsAndMentions := `
 		UPDATE ChannelMembers
 		SET MentionCount=0, MentionCountRoot=0, MsgCount=Channels.TotalMsgCount, MsgCountRoot=Channels.TotalMsgCountRoot, LastUpdateAt = :Now
@@ -1277,5 +1291,9 @@ func fixCRTChannelMembershipCounts(sqlStore *SqlStore) {
 
 	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))
+		return
+	}
+	if _, err := sqlStore.GetMaster().ExecNoTimeout("INSERT INTO Systems VALUES ('CRTChannelMembershipCountsMigrationComplete', 'true')"); err != nil {
+		mlog.Error("Error marking migration as done", mlog.Err(err))
 	}
 }

store/sqlstore/upgrade_test.go

@@ -331,12 +331,13 @@ func TestFixCRTCountsAndUnreads(t *testing.T) {
 		require.NoError(t, err)
 
 		// Run migration to fix threads and memberships
+		ss.System().PermanentDeleteByName("CRTThreadCountsAndUnreadsMigrationComplete")
 		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, lastReplyAt+1, fixedThreadMembership1.LastViewed)
 		require.EqualValues(t, int64(0), fixedThreadMembership1.UnreadMentions)
 		require.NotEqual(t, goodThreadMembership1.LastUpdated, fixedThreadMembership1.LastUpdated)
 
@@ -387,6 +388,7 @@ func TestFixCRTChannelUnreads(t *testing.T) {
 		})
 		require.NoError(t, err)
 
+		ss.System().PermanentDeleteByName("CRTChannelMembershipCountsMigrationComplete")
 		fixCRTChannelMembershipCounts(sqlStore)
 
 		cm1AfterFix, err := ss.Channel().GetMember(context.Background(), c1.Id, uId1)