MM-45009: Delete ThreadMemberships from "left" channels (#22559)

* MM-50550: Filter out threads from "left" channels v2

Currently leaving a channel doesn't affect the thread memberships of
that user/channel combination.
This PR aims to filter out all threads from those channels for the user.

Adds a DeleteAt column in the ThreadMemberships table, and filter out
all thread memberships that are "deleted".
Each time a user leaves a channel all thread memberships are going to be
marked as deleted, and when a user joins a channel again all those
existing thread memberships will be re-instantiated.

Adds a migration to mark all existing thread memberships as deleted
depending on whether there exists a channel membership for that
channel/user.

* Added migration files into list

* Fixes tests

* Fixes case where DeleteAt would be null

* Guard thread API endpoints with appropriate perms

* Deletes ThreadMembership rows upon leaving channel

* Minor style changes

* Use NoTranslation error

* Refactors tests

* Adds API tests to assert permissions on Team

* Adds tests, and fixes migrations

* Fixes test description

* Fix test

* Removes check on DM/GMs

* Change the MySQL query in the migration

---------

Co-authored-by: Mattermost Build <build@mattermost.com>

server/channels/store/opentracinglayer/opentracinglayer.go

@@ -10123,6 +10123,24 @@ func (s *OpenTracingLayerThreadStore) DeleteMembershipForUser(userId string, pos
 	return err
 }
 
+func (s *OpenTracingLayerThreadStore) DeleteMembershipsForChannel(userID string, channelID string) error {
+	origCtx := s.Root.Store.Context()
+	span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "ThreadStore.DeleteMembershipsForChannel")
+	s.Root.Store.SetContext(newCtx)
+	defer func() {
+		s.Root.Store.SetContext(origCtx)
+	}()
+
+	defer span.Finish()
+	err := s.ThreadStore.DeleteMembershipsForChannel(userID, channelID)
+	if err != nil {
+		span.LogFields(spanlog.Error(err))
+		ext.Error.Set(span, true)
+	}
+
+	return err
+}
+
 func (s *OpenTracingLayerThreadStore) DeleteOrphanedRows(limit int) (int64, error) {
 	origCtx := s.Root.Store.Context()
 	span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "ThreadStore.DeleteOrphanedRows")

server/channels/store/retrylayer/retrylayer.go

@@ -11563,6 +11563,27 @@ func (s *RetryLayerThreadStore) DeleteMembershipForUser(userId string, postID st
 
 }
 
+func (s *RetryLayerThreadStore) DeleteMembershipsForChannel(userID string, channelID string) error {
+
+	tries := 0
+	for {
+		err := s.ThreadStore.DeleteMembershipsForChannel(userID, channelID)
+		if err == nil {
+			return nil
+		}
+		if !isRepeatableError(err) {
+			return err
+		}
+		tries++
+		if tries >= 3 {
+			err = errors.Wrap(err, "giving up after 3 consecutive repeatable transaction failures")
+			return err
+		}
+		timepkg.Sleep(100 * timepkg.Millisecond)
+	}
+
+}
+
 func (s *RetryLayerThreadStore) DeleteOrphanedRows(limit int) (int64, error) {
 
 	tries := 0

server/channels/store/sqlstore/thread_store.go

@@ -688,6 +688,28 @@ func (s *SqlThreadStore) UpdateMembership(membership *model.ThreadMembership) (*
 	return s.updateMembership(s.GetMasterX(), membership)
 }
 
+func (s *SqlThreadStore) DeleteMembershipsForChannel(userID, channelID string) error {
+	subQuery := s.getSubQueryBuilder().
+		Select("1").
+		From("Threads").
+		Where(sq.And{
+			sq.Expr("Threads.PostId = ThreadMemberships.PostId"),
+			sq.Eq{"Threads.ChannelId": channelID},
+		})
+
+	query := s.getQueryBuilder().
+		Delete("ThreadMemberships").
+		Where(sq.Eq{"UserId": userID}).
+		Where(sq.Expr("EXISTS (?)", subQuery))
+
+	_, err := s.GetMasterX().ExecBuilder(query)
+	if err != nil {
+		return errors.Wrapf(err, "failed to remove thread memberships with userid=%s channelid=%s", userID, channelID)
+	}
+
+	return nil
+}
+
 func (s *SqlThreadStore) updateMembership(ex sqlxExecutor, membership *model.ThreadMembership) (*model.ThreadMembership, error) {
 	query := s.getQueryBuilder().
 		Update("ThreadMemberships").
@@ -712,7 +734,14 @@ func (s *SqlThreadStore) GetMembershipsForUser(userId, teamId string) ([]*model.
 	memberships := []*model.ThreadMembership{}
 
 	query := s.getQueryBuilder().
-		Select("ThreadMemberships.*").
+		Select(
+			"ThreadMemberships.PostId",
+			"ThreadMemberships.UserId",
+			"ThreadMemberships.Following",
+			"ThreadMemberships.LastUpdated",
+			"ThreadMemberships.LastViewed",
+			"ThreadMemberships.UnreadMentions",
+		).
 		Join("Threads ON Threads.PostId = ThreadMemberships.PostId").
 		From("ThreadMemberships").
 		Where(sq.Or{sq.Eq{"Threads.ThreadTeamId": teamId}, sq.Eq{"Threads.ThreadTeamId": ""}}).
@@ -732,7 +761,14 @@ func (s *SqlThreadStore) GetMembershipForUser(userId, postId string) (*model.Thr
 func (s *SqlThreadStore) getMembershipForUser(ex sqlxExecutor, userId, postId string) (*model.ThreadMembership, error) {
 	var membership model.ThreadMembership
 	query := s.getQueryBuilder().
-		Select("*").
+		Select(
+			"PostId",
+			"UserId",
+			"Following",
+			"LastUpdated",
+			"LastViewed",
+			"UnreadMentions",
+		).
 		From("ThreadMemberships").
 		Where(sq.And{
 			sq.Eq{"PostId": postId},

server/channels/store/store.go

@@ -344,6 +344,7 @@ type ThreadStore interface {
 	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)
+	DeleteMembershipsForChannel(userID, channelID string) error
 
 	// Insights - threads
 	GetTopThreadsForTeamSince(teamID string, userID string, since int64, offset int, limit int) (*model.TopThreadList, error)

server/channels/store/storetest/mocks/ThreadStore.go

@@ -29,6 +29,20 @@ func (_m *ThreadStore) DeleteMembershipForUser(userId string, postID string) err
 	return r0
 }
 
+// DeleteMembershipsForChannel provides a mock function with given fields: userID, channelID
+func (_m *ThreadStore) DeleteMembershipsForChannel(userID string, channelID string) error {
+	ret := _m.Called(userID, channelID)
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(string, string) error); ok {
+		r0 = rf(userID, channelID)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
 // DeleteOrphanedRows provides a mock function with given fields: limit
 func (_m *ThreadStore) DeleteOrphanedRows(limit int) (int64, error) {
 	ret := _m.Called(limit)

server/channels/store/storetest/thread_store.go

@@ -29,6 +29,7 @@ func TestThreadStore(t *testing.T, ss store.Store, s SqlStore) {
 	t.Run("MarkAllAsReadByChannels", func(t *testing.T) { testMarkAllAsReadByChannels(t, ss) })
 	t.Run("GetTopThreads", func(t *testing.T) { testGetTopThreads(t, ss) })
 	t.Run("MarkAllAsReadByTeam", func(t *testing.T) { testMarkAllAsReadByTeam(t, ss) })
+	t.Run("DeleteMembershipsForChannel", func(t *testing.T) { testDeleteMembershipsForChannel(t, ss) })
 }
 
 func testThreadStorePopulation(t *testing.T, ss store.Store) {
@@ -1914,3 +1915,121 @@ func testMarkAllAsReadByTeam(t *testing.T, ss store.Store) {
 		assertThreadReplyCount(t, userBID, team2.Id, 1, "expected 1 unread message in team2 for userB")
 	})
 }
+
+func testDeleteMembershipsForChannel(t *testing.T, ss store.Store) {
+	createThreadMembership := func(userID, postID string) (*model.ThreadMembership, func()) {
+		t.Helper()
+		opts := store.ThreadMembershipOpts{
+			Following:             true,
+			IncrementMentions:     false,
+			UpdateFollowing:       true,
+			UpdateViewedTimestamp: false,
+			UpdateParticipants:    false,
+		}
+		mem, err := ss.Thread().MaintainMembership(userID, postID, opts)
+		require.NoError(t, err)
+
+		return mem, func() {
+			err := ss.Thread().DeleteMembershipForUser(userID, postID)
+			require.NoError(t, err)
+		}
+	}
+
+	postingUserID := model.NewId()
+	userAID := model.NewId()
+	userBID := model.NewId()
+
+	team, err := ss.Team().Save(&model.Team{
+		DisplayName: "DisplayName",
+		Name:        "team" + model.NewId(),
+		Email:       MakeEmail(),
+		Type:        model.TeamOpen,
+	})
+	require.NoError(t, err)
+
+	channel1, err := ss.Channel().Save(&model.Channel{
+		TeamId:      team.Id,
+		DisplayName: "DisplayName",
+		Name:        "channel1" + model.NewId(),
+		Type:        model.ChannelTypeOpen,
+	}, -1)
+	require.NoError(t, err)
+	channel2, err := ss.Channel().Save(&model.Channel{
+		TeamId:      team.Id,
+		DisplayName: "DisplayName2",
+		Name:        "channel2" + model.NewId(),
+		Type:        model.ChannelTypeOpen,
+	}, -1)
+	require.NoError(t, err)
+
+	rootPost1, err := ss.Post().Save(&model.Post{
+		ChannelId: channel1.Id,
+		UserId:    postingUserID,
+		Message:   model.NewRandomString(10),
+	})
+	require.NoError(t, err)
+
+	_, err = ss.Post().Save(&model.Post{
+		ChannelId: channel1.Id,
+		UserId:    postingUserID,
+		Message:   model.NewRandomString(10),
+		RootId:    rootPost1.Id,
+	})
+	require.NoError(t, err)
+
+	rootPost2, err := ss.Post().Save(&model.Post{
+		ChannelId: channel2.Id,
+		UserId:    postingUserID,
+		Message:   model.NewRandomString(10),
+	})
+	require.NoError(t, err)
+	_, err = ss.Post().Save(&model.Post{
+		ChannelId: channel2.Id,
+		UserId:    postingUserID,
+		Message:   model.NewRandomString(10),
+		RootId:    rootPost2.Id,
+	})
+	require.NoError(t, err)
+
+	t.Run("should return memberships for user", func(t *testing.T) {
+		memA1, cleanupA1 := createThreadMembership(userAID, rootPost1.Id)
+		defer cleanupA1()
+		memA2, cleanupA2 := createThreadMembership(userAID, rootPost2.Id)
+		defer cleanupA2()
+
+		membershipsA, err := ss.Thread().GetMembershipsForUser(userAID, team.Id)
+		require.NoError(t, err)
+
+		require.Len(t, membershipsA, 2)
+		require.ElementsMatch(t, []*model.ThreadMembership{memA1, memA2}, membershipsA)
+	})
+
+	t.Run("should delete memberships for user for channel", func(t *testing.T) {
+		_, cleanupA1 := createThreadMembership(userAID, rootPost1.Id)
+		defer cleanupA1()
+		memA2, cleanupA2 := createThreadMembership(userAID, rootPost2.Id)
+		defer cleanupA2()
+
+		ss.Thread().DeleteMembershipsForChannel(userAID, channel1.Id)
+		membershipsA, err := ss.Thread().GetMembershipsForUser(userAID, team.Id)
+		require.NoError(t, err)
+
+		require.Len(t, membershipsA, 1)
+		require.ElementsMatch(t, []*model.ThreadMembership{memA2}, membershipsA)
+	})
+
+	t.Run("deleting memberships for channel for userA should not affect userB", func(t *testing.T) {
+		_, cleanupA1 := createThreadMembership(userAID, rootPost1.Id)
+		defer cleanupA1()
+		_, cleanupA2 := createThreadMembership(userAID, rootPost2.Id)
+		defer cleanupA2()
+		memB1, cleanupB2 := createThreadMembership(userBID, rootPost1.Id)
+		defer cleanupB2()
+
+		membershipsB, err := ss.Thread().GetMembershipsForUser(userBID, team.Id)
+		require.NoError(t, err)
+
+		require.Len(t, membershipsB, 1)
+		require.ElementsMatch(t, []*model.ThreadMembership{memB1}, membershipsB)
+	})
+}

server/channels/store/timerlayer/timerlayer.go

@@ -9112,6 +9112,22 @@ func (s *TimerLayerThreadStore) DeleteMembershipForUser(userId string, postID st
 	return err
 }
 
+func (s *TimerLayerThreadStore) DeleteMembershipsForChannel(userID string, channelID string) error {
+	start := time.Now()
+
+	err := s.ThreadStore.DeleteMembershipsForChannel(userID, channelID)
+
+	elapsed := float64(time.Since(start)) / float64(time.Second)
+	if s.Root.Metrics != nil {
+		success := "false"
+		if err == nil {
+			success = "true"
+		}
+		s.Root.Metrics.ObserveStoreMethodDuration("ThreadStore.DeleteMembershipsForChannel", success, elapsed)
+	}
+	return err
+}
+
 func (s *TimerLayerThreadStore) DeleteOrphanedRows(limit int) (int64, error) {
 	start := time.Now()