MM-40302: CRT, fix performance of MarkAllAsReadInChannels (#19566)

* deadcode: remove UpdateChannelLastViewedAt

* deadcode: remove ThreadStore.(Save(Multiple)|Update|Delete)

* deadcode: followThead in App.MarkChannelAsUnreadFromPost

* document ThreadMembership, Thread structs

* maintain LastUpdated consistently

Whenever we touch a `ThreadMembership` record, we should be setting `LastUpdated` to the current timestamp. The mobile client relies on this to detect changes to these records.

* simplify: never updateThreads from `App.MarkChannelAsUnreadFromPost`

Change all invocations of `ChannelStore.UpdateLastViewedAtPost` from `App.MarkChannelAsUnreadFromPost` to pass `updateThreads` as `false`. When `ChannelStore.UpdateLastViewedAtPost` was invoked with `updateThreads` as `true`, it would in turn call `ThreadStore.UpdateUnreadsByChannel` but pass `updateViewedTimestamp` as `false`. This effectively updated the `LastUpdated` field of the corresponding thread memberships but never touched any of the actual data (such as `LastViewed`).

The overall CRT feature continued to work, because `App.MarkChannelAsUnreadFromPost` directly updates the relevant thread memberships via `ThreadStore.MaintainMembership`.

* deadcode: updateThreads in ChannelStore.UpdateLastViewedAtPost

* simplify: never updateThreads from App.SendNotifications

Change all invocations of `ChannelStore.IncrementMentionCount` from
`App.SendNotifications` to pass `updateThreads` as `false`. When `ChannelStore.IncrementMentionCount` was invoked with `updateThreads` as `true`, it would in turn call `ThreadStore.UpdateUnreadsByChannel` but pass `updateViewedTimestamp` as `false`. This effectively updated the `LastUpdated` field of the corresponding thread memberships but never touched any of the actual data (such as `UnreadMentions`).

The overall CRT feature continued to work, because `App.SendNotifications` directly updates the relevant thread memberships mention counts via `ThreadStore.MaintainMembership`.

* deadcode: updateThreads in ChannelStore.IncrementMentionCount

* fix & rename ThreadStore.UpdateUnreadsByChannel

Rename `ThreadStore.UpdateUnreadsByChannel` to `ThreadStore.UpdateLastViewedByThreadIds`, making it unconditionally set the `LastViewed` for the given threads (as well as `LastUpdated`).

All previous invocations of this method that passed `updateViewedTimestamp` have been previously removed.

* unrelated gofmt -w -s changes to satisfy linter

* always set LastUpdated to model.GetMillis()

* deadcode: ThreadStore.SaveMembership

* fix TestMarkUnreadWithThreads

* MM-40302: CRT, use updateThreads param vs. MarkAllAsReadInChannels

`MarkAllAsReadInChannels` was the subject of a significant performance regression in v5.37 and is known to be very inefficient, by virtue of always writing to an ever increasing number of rows, and doing so on common events like simply viewing a channel.

Fortunately, `ChannelStore.UpdateLastViewedAt` already supported an `updateThreads` parameter that implemented the start of an improved algorithm: query the set of threads with newer posts, and then update only /those/. Missing was the need to reset the `UnreadMentions`, but thanks to the previous simplifications in #19523, we can make this change largely without impacting other semantics.

Fixes: https://mattermost.atlassian.net/browse/MM-40302

* fix MySQL

* remove another JOIN

* remove outdated comment

* unit tests

store/sqlstore/channel_store.go

@@ -2311,17 +2311,7 @@ func (s SqlChannelStore) PermanentDeleteMembersByUser(userId string) error {
 	return nil
 }
 
-func (s SqlChannelStore) UpdateLastViewedAt(channelIds []string, userId string, updateThreads bool) (map[string]int64, error) {
-	var threadsToUpdate []string
-	now := model.GetMillis()
-	if updateThreads {
-		var err error
-		threadsToUpdate, err = s.Thread().CollectThreadsWithNewerReplies(userId, channelIds, now)
-		if err != nil {
-			return nil, err
-		}
-	}
-
+func (s SqlChannelStore) UpdateLastViewedAt(channelIds []string, userId string) (map[string]int64, error) {
 	lastPostAtTimes := []struct {
 		Id                string
 		LastPostAt        int64
@@ -2379,9 +2369,6 @@ func (s SqlChannelStore) UpdateLastViewedAt(channelIds []string, userId string,
 		for _, t := range lastPostAtTimes {
 			times[t.Id] = t.LastPostAt
 		}
-		if updateThreads {
-			s.Thread().UpdateLastViewedByThreadIds(userId, threadsToUpdate, now)
-		}
 		return times, nil
 	}
 
@@ -2424,9 +2411,6 @@ func (s SqlChannelStore) UpdateLastViewedAt(channelIds []string, userId string,
 		return nil, errors.Wrapf(err, "failed to update ChannelMembers with userId=%s and channelId in %v", userId, channelIds)
 	}
 
-	if updateThreads {
-		s.Thread().UpdateLastViewedByThreadIds(userId, threadsToUpdate, now)
-	}
 	return times, nil
 }
 

store/sqlstore/thread_store.go

@@ -511,44 +511,73 @@ func (s *SqlThreadStore) GetThreadForUser(teamId string, threadMembership *model
 	return result, nil
 }
 
-// MarkAllAsReadInChannels marks all threads for the given user in the given channels as read from
-// the current time.
-func (s *SqlThreadStore) MarkAllAsReadInChannels(userID string, channelIDs []string) error {
-	threadIDs := []string{}
-
-	query, args, _ := s.getQueryBuilder().
-		Select("ThreadMemberships.PostId").
-		Join("Threads ON Threads.PostId = ThreadMemberships.PostId").
-		Join("Channels ON Threads.ChannelId = Channels.Id").
-		From("ThreadMemberships").
-		Where(sq.Eq{"Threads.ChannelId": channelIDs}).
-		Where(sq.Eq{"ThreadMemberships.UserId": userID}).
-		ToSql()
-
-	err := s.GetReplicaX().Select(&threadIDs, query, args...)
-	if err != nil {
-		return errors.Wrapf(err, "failed to get thread membership with userid=%s", userID)
+// MarkAllAsReadByChannels marks thread membership for the given users in the given channels
+// as read. This is used by the application layer to keep threads up-to-date when CRT is disabled
+// for the enduser, avoiding an influx of unread threads when first turning the feature on.
+func (s *SqlThreadStore) MarkAllAsReadByChannels(userID string, channelIDs []string) error {
+	if len(channelIDs) == 0 {
+		return nil
 	}
 
+	now := model.GetMillis()
+
+	// TODO: Fork squirrel to include https://github.com/Masterminds/squirrel/pull/256 and
+	// support FROM in an UPDATE query.
+	channelIDsSql, channelIDsArgs := constructArrayArgs(channelIDs)
+
+	var query string
+	if s.DriverName() == model.DatabaseDriverPostgres {
+		query = `
+			UPDATE ThreadMemberships
+			SET LastViewed = ?, UnreadMentions = ?, LastUpdated = ?
+			FROM Threads
+			WHERE ThreadMemberships.UserId = ?
+			AND Threads.PostId = ThreadMemberships.PostId
+			AND Threads.ChannelID IN ` + channelIDsSql + `
+			AND Threads.LastReplyAt > ThreadMemberships.LastViewed
+		`
+	} else {
+		query = `
+			UPDATE ThreadMemberships, Threads
+			SET ThreadMemberships.LastViewed = ?, ThreadMemberships.UnreadMentions = ?, ThreadMemberships.LastUpdated = ?
+			WHERE ThreadMemberships.UserId = ?
+			AND Threads.PostId = ThreadMemberships.PostId
+			AND Threads.ChannelID IN ` + channelIDsSql + `
+			AND Threads.LastReplyAt > ThreadMemberships.LastViewed
+		`
+	}
+
+	args := []interface{}{now, 0, now, userID}
+	args = append(args, channelIDsArgs...)
+
+	if _, err := s.GetMasterX().Exec(query, args...); err != nil {
+		return errors.Wrapf(err, "failed to mark all threads as read by channels for user id=%s", userID)
+	}
+
+	return nil
+}
+
+func (s *SqlThreadStore) MarkAllAsRead(userId string, threadIds []string) error {
 	timestamp := model.GetMillis()
-	query, args, _ = s.getQueryBuilder().
+
+	query, args, _ := s.getQueryBuilder().
 		Update("ThreadMemberships").
-		Where(sq.Eq{"PostId": threadIDs}).
-		Where(sq.Eq{"UserId": userID}).
+		Where(sq.Eq{"UserId": userId}).
+		Where(sq.Eq{"PostId": threadIds}).
 		Set("LastViewed", timestamp).
 		Set("UnreadMentions", 0).
 		Set("LastUpdated", model.GetMillis()).
 		ToSql()
 	if _, err := s.GetMasterX().Exec(query, args...); err != nil {
-		return errors.Wrapf(err, "failed to update thread read state for user id=%s", userID)
+		return errors.Wrapf(err, "failed to mark %d threads as read for user id=%s", len(threadIds), userId)
 	}
-	return nil
 
+	return nil
 }
 
-// MarkAllAsRead marks all threads for the given user in the given team as read from the current
-// time.
-func (s *SqlThreadStore) MarkAllAsRead(userId, teamId string) error {
+// MarkAllAsReadByTeam marks all threads for the given user in the given team as read from the
+// current time.
+func (s *SqlThreadStore) MarkAllAsReadByTeam(userId, teamId string) error {
 	memberships, err := s.GetMembershipsForUser(userId, teamId)
 	if err != nil {
 		return err
@@ -786,50 +815,6 @@ func (s *SqlThreadStore) MaintainMembership(userId, postId string, opts store.Th
 	return membership, err
 }
 
-func (s *SqlThreadStore) CollectThreadsWithNewerReplies(userId string, channelIds []string, timestamp int64) ([]string, error) {
-	changedThreads := []string{}
-	query, args, _ := s.getQueryBuilder().
-		Select("Threads.PostId").
-		From("Threads").
-		LeftJoin("ChannelMembers ON ChannelMembers.ChannelId=Threads.ChannelId").
-		Where(sq.And{
-			sq.Eq{"Threads.ChannelId": channelIds},
-			sq.Eq{"ChannelMembers.UserId": userId},
-			sq.Or{
-				sq.Expr("Threads.LastReplyAt > ChannelMembers.LastViewedAt"),
-				sq.Gt{"Threads.LastReplyAt": timestamp},
-			},
-		}).
-		ToSql()
-	if err := s.GetReplicaX().Select(&changedThreads, query, args...); err != nil {
-		return nil, errors.Wrap(err, "failed to fetch threads")
-	}
-	return changedThreads, nil
-}
-
-// UpdateLastViewedByThreadIds marks the given threads as read up to the given timestamp. If there
-// are no newer posts, it effectively marks the thread as read. If there are newer posts, say
-// because the user explicitly marked a past post as unread, the thread will be considered unread
-// past the given timestamp.
-func (s *SqlThreadStore) UpdateLastViewedByThreadIds(userId string, threadIds []string, timestamp int64) error {
-	if len(threadIds) == 0 {
-		return nil
-	}
-
-	qb := s.getQueryBuilder().
-		Update("ThreadMemberships").
-		Where(sq.Eq{"UserId": userId, "PostId": threadIds}).
-		Set("LastViewed", timestamp).
-		Set("LastUpdated", model.GetMillis())
-	updateQuery, updateArgs, _ := qb.ToSql()
-
-	if _, err := s.GetMasterX().Exec(updateQuery, updateArgs...); err != nil {
-		return errors.Wrap(err, "failed to update thread membership")
-	}
-
-	return nil
-}
-
 func (s *SqlThreadStore) GetPosts(threadId string, since int64) ([]*model.Post, error) {
 	query, args, _ := s.getQueryBuilder().
 		Select("*").