PLT-8233: Purge ChannelMemberHistory Data when Data Retention Job runs (#7937)

* Changed the channel member history purge function to match pattern established by other data retention purge methods

* Simplifying the channel member history store and associated tests

* Regenerating mocks

* Using a constant

i18n/en.json

@@ -4443,7 +4443,7 @@
     "translation": "Failed to get records"
   },
   {
-    "id": "store.sql_channel_member_history.purge_history_before.app_error",
+    "id": "store.sql_channel_member_history.permanent_delete_batch.app_error",
     "translation": "Failed to purge records"
   },
   {

store/sqlstore/channel_member_history_store.go

@@ -39,7 +39,7 @@ func (s SqlChannelMemberHistoryStore) LogJoinEvent(userId string, channelId stri
 		}
 
 		if err := s.GetMaster().Insert(channelMemberHistory); err != nil {
-			result.Err = model.NewAppError("SqlChannelMemberHistoryStore.LogJoinEvent", "store.sql_channel_member_history.log_join_event.app_error", map[string]interface{}{"ChannelMemberHistory": channelMemberHistory}, err.Error(), http.StatusInternalServerError)
+			result.Err = model.NewAppError("SqlChannelMemberHistoryStore.LogJoinEvent", "store.sql_channel_member_history.log_join_event.app_error", nil, err.Error(), http.StatusInternalServerError)
 		}
 	})
 }
@@ -55,9 +55,9 @@ func (s SqlChannelMemberHistoryStore) LogLeaveEvent(userId string, channelId str
 
 		params := map[string]interface{}{"UserId": userId, "ChannelId": channelId, "LeaveTime": leaveTime}
 		if sqlResult, err := s.GetMaster().Exec(query, params); err != nil {
-			result.Err = model.NewAppError("SqlChannelMemberHistoryStore.LogLeaveEvent", "store.sql_channel_member_history.log_leave_event.update_error", nil, err.Error(), http.StatusInternalServerError)
+			result.Err = model.NewAppError("SqlChannelMemberHistoryStore.LogLeaveEvent", "store.sql_channel_member_history.log_leave_event.update_error", params, err.Error(), http.StatusInternalServerError)
 		} else if rows, err := sqlResult.RowsAffected(); err == nil && rows != 1 {
-			// there was no join event to update
+			// there was no join event to update - this is best effort, so no need to raise an error
 			l4g.Warn("Channel join event for user %v and channel %v not found", userId, channelId)
 		}
 	})
@@ -86,17 +86,36 @@ func (s SqlChannelMemberHistoryStore) GetUsersInChannelDuring(startTime int64, e
 	})
 }
 
-func (s SqlChannelMemberHistoryStore) PurgeHistoryBefore(time int64, channelId string) store.StoreChannel {
+func (s SqlChannelMemberHistoryStore) PermanentDeleteBatch(endTime int64, limit int64) store.StoreChannel {
 	return store.Do(func(result *store.StoreResult) {
-		query := `
-			DELETE FROM ChannelMemberHistory
-			WHERE ChannelId = :ChannelId
-			AND LeaveTime IS NOT NULL
-			AND LeaveTime <= :AtTime`
+		var query string
+		if s.DriverName() == model.DATABASE_DRIVER_POSTGRES {
+			query =
+				`DELETE FROM ChannelMemberHistory
+				 WHERE ctid IN (
+				 	SELECT ctid FROM ChannelMemberHistory
+					WHERE LeaveTime IS NOT NULL
+					AND LeaveTime <= :EndTime
+					LIMIT :Limit
+				);`
+		} else {
+			query =
+				`DELETE FROM ChannelMemberHistory
+				 WHERE LeaveTime IS NOT NULL
+				 AND LeaveTime <= :EndTime
+				 LIMIT :Limit`
+		}
 
-		params := map[string]interface{}{"AtTime": time, "ChannelId": channelId}
-		if _, err := s.GetMaster().Exec(query, params); err != nil {
-			result.Err = model.NewAppError("SqlChannelMemberHistoryStore.PurgeHistoryBefore", "store.sql_channel_member_history.purge_history_before.app_error", params, err.Error(), http.StatusInternalServerError)
+		params := map[string]interface{}{"EndTime": endTime, "Limit": limit}
+		if sqlResult, err := s.GetMaster().Exec(query, params); err != nil {
+			result.Err = model.NewAppError("SqlChannelMemberHistoryStore.PermanentDeleteBatchForChannel", "store.sql_channel_member_history.permanent_delete_batch.app_error", params, err.Error(), http.StatusInternalServerError)
+		} else {
+			if rowsAffected, err1 := sqlResult.RowsAffected(); err1 != nil {
+				result.Err = model.NewAppError("SqlChannelMemberHistoryStore.PermanentDeleteBatchForChannel", "store.sql_channel_member_history.permanent_delete_batch.app_error", params, err.Error(), http.StatusInternalServerError)
+				result.Data = int64(0)
+			} else {
+				result.Data = rowsAffected
+			}
 		}
 	})
 }

store/store.go

@@ -165,7 +165,7 @@ type ChannelMemberHistoryStore interface {
 	LogJoinEvent(userId string, channelId string, joinTime int64) StoreChannel
 	LogLeaveEvent(userId string, channelId string, leaveTime int64) StoreChannel
 	GetUsersInChannelDuring(startTime int64, endTime int64, channelId string) StoreChannel
-	PurgeHistoryBefore(time int64, channelId string) StoreChannel
+	PermanentDeleteBatch(endTime int64, limit int64) StoreChannel
 }
 
 type PostStore interface {

store/storetest/channel_member_history_store.go

@@ -6,6 +6,8 @@ package storetest
 import (
 	"testing"
 
+	"math"
+
 	"github.com/mattermost/mattermost-server/model"
 	"github.com/mattermost/mattermost-server/store"
 	"github.com/stretchr/testify/assert"
@@ -15,7 +17,7 @@ func TestChannelMemberHistoryStore(t *testing.T, ss store.Store) {
 	t.Run("Log Join Event", func(t *testing.T) { testLogJoinEvent(t, ss) })
 	t.Run("Log Leave Event", func(t *testing.T) { testLogLeaveEvent(t, ss) })
 	t.Run("Get Users In Channel At Time", func(t *testing.T) { testGetUsersInChannelAt(t, ss) })
-	t.Run("Purge History", func(t *testing.T) { testPurgeHistoryBefore(t, ss) })
+	t.Run("Purge History", func(t *testing.T) { testPermanentDeleteBatch(t, ss) })
 }
 
 func testLogJoinEvent(t *testing.T, ss store.Store) {
@@ -135,7 +137,7 @@ func testGetUsersInChannelAt(t *testing.T, ss store.Store) {
 	assert.Len(t, channelMembers, 0)
 }
 
-func testPurgeHistoryBefore(t *testing.T, ss store.Store) {
+func testPermanentDeleteBatch(t *testing.T, ss store.Store) {
 	// create a test channel
 	channel := model.Channel{
 		TeamId:      model.NewId(),
@@ -171,8 +173,11 @@ func testPurgeHistoryBefore(t *testing.T, ss store.Store) {
 	channelMembers := store.Must(ss.ChannelMemberHistory().GetUsersInChannelDuring(joinTime+10, leaveTime-10, channel.Id)).([]*model.ChannelMemberHistory)
 	assert.Len(t, channelMembers, 2)
 
-	// but if we purge the old data, only the user that didn't leave is left
-	store.Must(ss.ChannelMemberHistory().PurgeHistoryBefore(leaveTime, channel.Id))
+	// the permanent delete should delete at least one record
+	rowsDeleted := store.Must(ss.ChannelMemberHistory().PermanentDeleteBatch(leaveTime, math.MaxInt64)).(int64)
+	assert.NotEqual(t, int64(0), rowsDeleted)
+
+	// after the delete, there should be one less member in the channel
 	channelMembers = store.Must(ss.ChannelMemberHistory().GetUsersInChannelDuring(joinTime+10, leaveTime-10, channel.Id)).([]*model.ChannelMemberHistory)
 	assert.Len(t, channelMembers, 1)
 	assert.Equal(t, user2.Id, channelMembers[0].UserId)

store/storetest/mocks/ChannelMemberHistoryStore.go

@@ -60,13 +60,13 @@ func (_m *ChannelMemberHistoryStore) LogLeaveEvent(userId string, channelId stri
 	return r0
 }
 
-// PurgeHistoryBefore provides a mock function with given fields: time, channelId
-func (_m *ChannelMemberHistoryStore) PurgeHistoryBefore(time int64, channelId string) store.StoreChannel {
-	ret := _m.Called(time, channelId)
+// PermanentDeleteBatch provides a mock function with given fields: endTime, limit
+func (_m *ChannelMemberHistoryStore) PermanentDeleteBatch(endTime int64, limit int64) store.StoreChannel {
+	ret := _m.Called(endTime, limit)
 
 	var r0 store.StoreChannel
-	if rf, ok := ret.Get(0).(func(int64, string) store.StoreChannel); ok {
-		r0 = rf(time, channelId)
+	if rf, ok := ret.Get(0).(func(int64, int64) store.StoreChannel); ok {
+		r0 = rf(endTime, limit)
 	} else {
 		if ret.Get(0) != nil {
 			r0 = ret.Get(0).(store.StoreChannel)