MM-36862: removes participant from thread (#18795)

* MM-36862: removes participant from thread

Removing a participant upon last reply deleted from thread didn't work
reliably, a suspect on this is the replica lag, since we are first
deleting the post and then counting non-deleted posts of the participant
to decide on whether to delete or not.

The findings that led to this conclusion is that the reply count gets
updated but the participant is not removed (participant removal depends
on the number of replies this participant has in the thread.)

This commit fixes that by removing first the participant and then
deleting the post.
So we delete the participant if they have 1 post in that thread, and
then we delete the post, so now they have no posts in the thread.

* Makes deleting posts transactional

This commit makes deleting a post transactional and also tries to fix
permanent deletion of posts.

Currently when we permanently delete all posts by a user we don't update
the threads reply count nor the participant's array. This commit tries
to fix that.

* Adds comments on deleting posts

Co-authored-by: Mattermod <mattermod@users.noreply.github.com>

model/utils.go

@@ -117,6 +117,24 @@ func (m *StringMap) Scan(value interface{}) error {
 	return errors.New("received value is neither a byte slice nor string")
 }
 
+func (si *StringInterface) Scan(value interface{}) error {
+	if value == nil {
+		return nil
+	}
+
+	buf, ok := value.([]byte)
+	if ok {
+		return json.Unmarshal(buf, si)
+	}
+
+	str, ok := value.(string)
+	if ok {
+		return json.Unmarshal([]byte(str), si)
+	}
+
+	return errors.New("received value is neither a byte slice nor string")
+}
+
 var translateFunc i18n.TranslateFunc
 var translateFuncOnce sync.Once
 

store/sqlstore/post_store.go

@@ -675,16 +675,34 @@ func (s *SqlPostStore) GetEtag(channelId string, allowFromCache, collapsedThread
 	return result
 }
 
+// Soft deletes a post
+// and cleans up the thread if it's a comment
 func (s *SqlPostStore) Delete(postID string, time int64, deleteByID string) error {
-	var err error
+	transaction, err := s.GetMasterX().Beginx()
+	if err != nil {
+		return errors.Wrap(err, "begin_transaction")
+	}
+	defer finalizeTransactionX(transaction)
+
+	id := postIds{}
+	// TODO: change this to later delete thread directly from postID
+	err = transaction.Get(&id, "SELECT RootId, UserId FROM Posts WHERE Id = ?", postID)
+	if err != nil {
+		if err == sql.ErrNoRows {
+			return store.NewErrNotFound("Post", postID)
+		}
+
+		return errors.Wrapf(err, "failed to delete Post with id=%s", postID)
+	}
+
 	if s.DriverName() == model.DatabaseDriverPostgres {
-		_, err = s.GetMaster().Exec(`UPDATE Posts
+		_, err = transaction.Exec(`UPDATE Posts
 			SET DeleteAt = $1,
 				UpdateAt = $1,
 				Props = jsonb_set(Props, $2, $3)
 			WHERE Id = $4 OR RootId = $4`, time, jsonKeyPath(model.PostPropsDeleteBy), jsonStringVal(deleteByID), postID)
 	} else {
-		_, err = s.GetMaster().Exec(`UPDATE Posts
+		_, err = transaction.Exec(`UPDATE Posts
 			SET DeleteAt = ?,
 			UpdateAt = ?,
 			Props = JSON_SET(Props, ?, ?)
@@ -695,34 +713,43 @@ func (s *SqlPostStore) Delete(postID string, time int64, deleteByID string) erro
 		return errors.Wrap(err, "failed to update Posts")
 	}
 
-	ids := postIds{}
-	// TODO: change this to later delete thread directly from postID
-	err = s.GetReplica().SelectOne(&ids, "SELECT RootId, UserId FROM Posts WHERE Id = :Id", map[string]interface{}{"Id": postID})
-	if err != nil {
-		if err == sql.ErrNoRows {
-			return store.NewErrNotFound("Post", postID)
-		}
+	err = s.cleanupThreadComments(transaction, postID, id.RootId, id.UserId)
 
-		return errors.Wrapf(err, "failed to delete Post with id=%s", postID)
+	if err != nil {
+		return errors.Wrapf(err, "failed to cleanup Thread with postid=%s", id.RootId)
 	}
 
-	return s.cleanupThreads(postID, ids.RootId, false, ids.UserId)
+	if err = transaction.Commit(); err != nil {
+		return errors.Wrap(err, "commit_transaction")
+	}
+
+	return nil
 }
 
 func (s *SqlPostStore) permanentDelete(postId string) error {
 	var post model.Post
-	err := s.GetReplica().SelectOne(&post, "SELECT * FROM Posts WHERE Id = :Id", map[string]interface{}{"Id": postId})
+	transaction, err := s.GetMasterX().Beginx()
+	if err != nil {
+		return errors.Wrap(err, "begin_transaction")
+	}
+	defer finalizeTransactionX(transaction)
+
+	err = transaction.Get(&post, "SELECT * FROM Posts WHERE Id = ?", postId)
 	if err != nil && err != sql.ErrNoRows {
 		return errors.Wrapf(err, "failed to get Post with id=%s", postId)
 	}
-	if err = s.cleanupThreads(post.Id, post.RootId, true, post.UserId); err != nil {
+	if err = s.permanentDeleteThreads(transaction, post.Id); err != nil {
 		return errors.Wrapf(err, "failed to cleanup threads for Post with id=%s", postId)
 	}
 
-	if _, err = s.GetMaster().Exec("DELETE FROM Posts WHERE Id = :Id OR RootId = :RootId", map[string]interface{}{"Id": postId, "RootId": postId}); err != nil {
+	if _, err = transaction.NamedExec("DELETE FROM Posts WHERE Id = :id OR RootId = :rootid", map[string]interface{}{"id": postId, "rootid": postId}); err != nil {
 		return errors.Wrapf(err, "failed to delete Post with id=%s", postId)
 	}
 
+	if err = transaction.Commit(); err != nil {
+		return errors.Wrap(err, "commit_transaction")
+	}
+
 	return nil
 }
 
@@ -734,24 +761,40 @@ type postIds struct {
 
 func (s *SqlPostStore) permanentDeleteAllCommentByUser(userId string) error {
 	results := []postIds{}
-	_, err := s.GetMaster().Select(&results, "Select Id, RootId FROM Posts WHERE UserId = :UserId AND RootId != ''", map[string]interface{}{"UserId": userId})
+	transaction, err := s.GetMasterX().Beginx()
+	if err != nil {
+		return errors.Wrap(err, "begin_transaction")
+	}
+	defer finalizeTransactionX(transaction)
+
+	err = transaction.Select(&results, "Select Id, RootId FROM Posts WHERE UserId = ? AND RootId != ''", userId)
 	if err != nil {
 		return errors.Wrapf(err, "failed to fetch Posts with userId=%s", userId)
 	}
 
+	_, err = transaction.Exec("DELETE FROM Posts WHERE UserId = ? AND RootId != ''", userId)
+
+	if err != nil {
+		return errors.Wrapf(err, "failed to delete Posts with userId=%s", userId)
+	}
+
 	for _, ids := range results {
-		if err = s.cleanupThreads(ids.Id, ids.RootId, true, userId); err != nil {
+		if err = s.cleanupThreadComments(transaction, ids.Id, ids.RootId, userId); err != nil {
 			return err
 		}
 	}
 
-	_, err = s.GetMaster().Exec("DELETE FROM Posts WHERE UserId = :UserId AND RootId != ''", map[string]interface{}{"UserId": userId})
-	if err != nil {
-		return errors.Wrapf(err, "failed to delete Posts with userId=%s", userId)
+	if err = transaction.Commit(); err != nil {
+		return errors.Wrap(err, "commit_transaction")
 	}
+
 	return nil
 }
 
+// Permanently deletes all comments by user,
+// cleans up threads (removes said user from participants and decreases reply count),
+// permanent delete all root posts by user,
+// and delete threads and thread memberships for those root posts
 func (s *SqlPostStore) PermanentDeleteByUser(userId string) error {
 	// First attempt to delete all the comments for a user
 	if err := s.permanentDeleteAllCommentByUser(userId); err != nil {
@@ -788,22 +831,36 @@ func (s *SqlPostStore) PermanentDeleteByUser(userId string) error {
 	return nil
 }
 
+// Permanent deletes all channel root posts and comments,
+// deletes all threads and thread memberships
+// no thread comment cleanup needed, since we are deleting threads and thread memberships
 func (s *SqlPostStore) PermanentDeleteByChannel(channelId string) error {
+	transaction, err := s.GetMasterX().Beginx()
+	if err != nil {
+		return errors.Wrap(err, "begin_transaction")
+	}
+	defer finalizeTransactionX(transaction)
+
 	results := []postIds{}
-	_, err := s.GetMaster().Select(&results, "SELECT Id, RootId, UserId FROM Posts WHERE ChannelId = :ChannelId", map[string]interface{}{"ChannelId": channelId})
+	err = transaction.Select(&results, "SELECT Id, RootId, UserId FROM Posts WHERE ChannelId = ?", channelId)
 	if err != nil {
 		return errors.Wrapf(err, "failed to fetch Posts with channelId=%s", channelId)
 	}
 
 	for _, ids := range results {
-		if err = s.cleanupThreads(ids.Id, ids.RootId, true, ids.UserId); err != nil {
+		if err = s.permanentDeleteThreads(transaction, ids.Id); err != nil {
 			return err
 		}
 	}
 
-	if _, err := s.GetMaster().Exec("DELETE FROM Posts WHERE ChannelId = :ChannelId", map[string]interface{}{"ChannelId": channelId}); err != nil {
+	if _, err = transaction.Exec("DELETE FROM Posts WHERE ChannelId = ?", channelId); err != nil {
 		return errors.Wrapf(err, "failed to delete Posts with channelId=%s", channelId)
 	}
+
+	if err = transaction.Commit(); err != nil {
+		return errors.Wrap(err, "commit_transaction")
+	}
+
 	return nil
 }
 
@@ -2377,16 +2434,22 @@ func (s *SqlPostStore) GetOldestEntityCreationTime() (int64, error) {
 	return oldest, nil
 }
 
-func (s *SqlPostStore) cleanupThreads(postId, rootId string, permanent bool, userId string) error {
-	if permanent {
-		if _, err := s.GetMaster().Exec("DELETE FROM Threads WHERE PostId = :Id", map[string]interface{}{"Id": postId}); err != nil {
-			return errors.Wrap(err, "failed to delete Threads")
-		}
-		if _, err := s.GetMaster().Exec("DELETE FROM ThreadMemberships WHERE PostId = :Id", map[string]interface{}{"Id": postId}); err != nil {
-			return errors.Wrap(err, "failed to delete ThreadMemberships")
-		}
-		return nil
+// Deletes a thread and a thread membership if the postId is a root post
+func (s *SqlPostStore) permanentDeleteThreads(transaction *sqlxTxWrapper, postId string) error {
+	if _, err := transaction.Exec("DELETE FROM Threads WHERE PostId = ?", postId); err != nil {
+		return errors.Wrap(err, "failed to delete Threads")
 	}
+	if _, err := transaction.Exec("DELETE FROM ThreadMemberships WHERE PostId = ?", postId); err != nil {
+		return errors.Wrap(err, "failed to delete ThreadMemberships")
+	}
+	return nil
+}
+
+// Thread cleanup upon post deletion
+// if the post is a comment
+// reply count is reduced by 1 and,
+// the user is removed from participants if the comment deleted is the last reply from said user.
+func (s *SqlPostStore) cleanupThreadComments(transaction *sqlxTxWrapper, postId, rootId string, userId string) error {
 	if rootId != "" {
 		queryString, args, err := s.getQueryBuilder().
 			Select("COUNT(Id)").
@@ -2402,24 +2465,24 @@ func (s *SqlPostStore) cleanupThreads(postId, rootId string, permanent bool, use
 			return errors.Wrap(err, "failed to create SQL query to count user's posts")
 		}
 
-		count, err := s.GetReplica().SelectInt(queryString, args...)
+		var count int64
+		err = transaction.Get(&count, queryString, args...)
 
 		if err != nil {
 			return errors.Wrap(err, "failed to count user's posts in thread")
 		}
 
+		// Updating replyCount, and reducing participants if this was the last post in the thread for the user
 		updateQuery := s.getQueryBuilder().Update("Threads")
 
 		if count == 0 {
 			if s.DriverName() == model.DatabaseDriverPostgres {
 				updateQuery = updateQuery.Set("Participants", sq.Expr("Participants - ?", userId))
 			} else {
-				// The .Where is because JSON_REMOVE returns null if the element to remove wasn't present
 				updateQuery = updateQuery.
 					Set("Participants", sq.Expr(
-						`JSON_REMOVE(Participants, JSON_UNQUOTE(JSON_SEARCH(Participants, 'one', ?)))`, userId,
-					)).
-					Where(sq.Expr(`JSON_CONTAINS(Participants, ?)`, strconv.Quote(userId)))
+						`IFNULL(JSON_REMOVE(Participants, JSON_UNQUOTE(JSON_SEARCH(Participants, 'one', ?))), Participants)`, userId,
+					))
 			}
 		}
 
@@ -2435,7 +2498,7 @@ func (s *SqlPostStore) cleanupThreads(postId, rootId string, permanent bool, use
 			return errors.Wrap(err, "failed to create SQL query to update thread")
 		}
 
-		_, err = s.GetMaster().Exec(updateQueryString, updateArgs...)
+		_, err = transaction.Exec(updateQueryString, updateArgs...)
 
 		if err != nil {
 			return errors.Wrap(err, "failed to update Threads")

store/storetest/post_store.go

@@ -848,20 +848,73 @@ func testPostStorePermDelete1Level(t *testing.T, ss store.Store) {
 	o3, err = ss.Post().Save(o3)
 	require.NoError(t, err)
 
+	o4 := &model.Post{}
+	o4.ChannelId = model.NewId()
+	o4.RootId = o1.Id
+	o4.UserId = o2.UserId
+	o4.Message = NewTestId()
+	o4, err = ss.Post().Save(o4)
+	require.NoError(t, err)
+
+	o5 := &model.Post{}
+	o5.ChannelId = o3.ChannelId
+	o5.UserId = model.NewId()
+	o5.Message = NewTestId()
+	o5, err = ss.Post().Save(o5)
+	require.NoError(t, err)
+
+	o6 := &model.Post{}
+	o6.ChannelId = o3.ChannelId
+	o6.RootId = o5.Id
+	o6.UserId = model.NewId()
+	o6.Message = NewTestId()
+	o6, err = ss.Post().Save(o6)
+	require.NoError(t, err)
+
+	var thread *model.Thread
+	thread, err = ss.Thread().Get(o1.Id)
+	require.NoError(t, err)
+
+	require.EqualValues(t, 2, thread.ReplyCount)
+	require.EqualValues(t, model.StringArray{o2.UserId}, thread.Participants)
+
 	err2 := ss.Post().PermanentDeleteByUser(o2.UserId)
 	require.NoError(t, err2)
 
+	thread, err = ss.Thread().Get(o1.Id)
+	require.NoError(t, err)
+
+	require.EqualValues(t, 0, thread.ReplyCount)
+	require.EqualValues(t, model.StringArray{}, thread.Participants)
+
 	_, err = ss.Post().Get(context.Background(), o1.Id, false, false, false, "")
 	require.NoError(t, err, "Deleted id shouldn't have failed")
 
 	_, err = ss.Post().Get(context.Background(), o2.Id, false, false, false, "")
 	require.Error(t, err, "Deleted id should have failed")
 
+	thread, err = ss.Thread().Get(o5.Id)
+	require.NoError(t, err)
+	require.NotEmpty(t, thread)
+
 	err = ss.Post().PermanentDeleteByChannel(o3.ChannelId)
 	require.NoError(t, err)
 
+	thread, err = ss.Thread().Get(o5.Id)
+	require.NoError(t, err)
+	require.Nil(t, thread)
+
 	_, err = ss.Post().Get(context.Background(), o3.Id, false, false, false, "")
 	require.Error(t, err, "Deleted id should have failed")
+
+	_, err = ss.Post().Get(context.Background(), o4.Id, false, false, false, "")
+	require.Error(t, err, "Deleted id should have failed")
+
+	_, err = ss.Post().Get(context.Background(), o5.Id, false, false, false, "")
+	require.Error(t, err, "Deleted id should have failed")
+
+	_, err = ss.Post().Get(context.Background(), o6.Id, false, false, false, "")
+	require.Error(t, err, "Deleted id should have failed")
 }
 
 func testPostStorePermDelete1Level2(t *testing.T, ss store.Store) {