[MM-64360] Data retention: optionally preserve pinned posts (#31165)

* add new config to preserve pinned posts during data retention

* more graceful error if the pinned post was a reply in a deleted thread

server/channels/app/syncables_test.go

@@ -303,9 +303,15 @@ func TestCreateDefaultMemberships(t *testing.T) {
 
 	timeAfterLeaving := model.GetMillis() + 1
 
+	retentionPolicyBatchConfigs := model.RetentionPolicyBatchConfigs{
+		Now:                 0,
+		GlobalPolicyEndTime: timeBeforeLeaving,
+		Limit:               1000,
+	}
+
 	// Purging channelmemberhistory doesn't re-add user to channel
 	deletedCount, _, nErr := th.App.Srv().Store().ChannelMemberHistory().PermanentDeleteBatchForRetentionPolicies(
-		0, timeBeforeLeaving, 1000, model.RetentionPolicyCursor{})
+		retentionPolicyBatchConfigs, model.RetentionPolicyCursor{})
 	if nErr != nil {
 		t.Errorf("error permanently deleting channelmemberhistory: %s", nErr.Error())
 	}
@@ -321,9 +327,15 @@ func TestCreateDefaultMemberships(t *testing.T) {
 		t.Error("Expected channel member to remain deleted")
 	}
 
+	retentionPolicyBatchConfigs = model.RetentionPolicyBatchConfigs{
+		Now:                 0,
+		GlobalPolicyEndTime: timeAfterLeaving,
+		Limit:               1000,
+	}
+
 	// Purging channelmemberhistory doesn't re-add user to channel
 	deletedCount, _, nErr = th.App.Srv().Store().ChannelMemberHistory().PermanentDeleteBatchForRetentionPolicies(
-		0, timeAfterLeaving, 1000, model.RetentionPolicyCursor{})
+		retentionPolicyBatchConfigs, model.RetentionPolicyCursor{})
 	if nErr != nil {
 		t.Errorf("error permanently deleting channelmemberhistory: %s", nErr.Error())
 	}

server/channels/store/retrylayer/retrylayer.go

@@ -3656,11 +3656,11 @@ func (s *RetryLayerChannelMemberHistoryStore) PermanentDeleteBatch(endTime int64
 
 }
 
-func (s *RetryLayerChannelMemberHistoryStore) PermanentDeleteBatchForRetentionPolicies(now int64, globalPolicyEndTime int64, limit int64, cursor model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error) {
+func (s *RetryLayerChannelMemberHistoryStore) PermanentDeleteBatchForRetentionPolicies(retentionPolicyBatchConfigs model.RetentionPolicyBatchConfigs, cursor model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error) {
 
 	tries := 0
 	for {
-		result, resultVar1, err := s.ChannelMemberHistoryStore.PermanentDeleteBatchForRetentionPolicies(now, globalPolicyEndTime, limit, cursor)
+		result, resultVar1, err := s.ChannelMemberHistoryStore.PermanentDeleteBatchForRetentionPolicies(retentionPolicyBatchConfigs, cursor)
 		if err == nil {
 			return result, resultVar1, nil
 		}
@@ -8360,11 +8360,11 @@ func (s *RetryLayerPostStore) PermanentDeleteBatch(endTime int64, limit int64) (
 
 }
 
-func (s *RetryLayerPostStore) PermanentDeleteBatchForRetentionPolicies(now int64, globalPolicyEndTime int64, limit int64, cursor model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error) {
+func (s *RetryLayerPostStore) PermanentDeleteBatchForRetentionPolicies(retentionPolicyBatchConfigs model.RetentionPolicyBatchConfigs, cursor model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error) {
 
 	tries := 0
 	for {
-		result, resultVar1, err := s.PostStore.PermanentDeleteBatchForRetentionPolicies(now, globalPolicyEndTime, limit, cursor)
+		result, resultVar1, err := s.PostStore.PermanentDeleteBatchForRetentionPolicies(retentionPolicyBatchConfigs, cursor)
 		if err == nil {
 			return result, resultVar1, nil
 		}
@@ -13796,11 +13796,11 @@ func (s *RetryLayerThreadStore) MarkAsRead(userID string, threadID string, times
 
 }
 
-func (s *RetryLayerThreadStore) PermanentDeleteBatchForRetentionPolicies(now int64, globalPolicyEndTime int64, limit int64, cursor model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error) {
+func (s *RetryLayerThreadStore) PermanentDeleteBatchForRetentionPolicies(retentionPolicyBatchConfigs model.RetentionPolicyBatchConfigs, cursor model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error) {
 
 	tries := 0
 	for {
-		result, resultVar1, err := s.ThreadStore.PermanentDeleteBatchForRetentionPolicies(now, globalPolicyEndTime, limit, cursor)
+		result, resultVar1, err := s.ThreadStore.PermanentDeleteBatchForRetentionPolicies(retentionPolicyBatchConfigs, cursor)
 		if err == nil {
 			return result, resultVar1, nil
 		}
@@ -13817,11 +13817,11 @@ func (s *RetryLayerThreadStore) PermanentDeleteBatchForRetentionPolicies(now int
 
 }
 
-func (s *RetryLayerThreadStore) PermanentDeleteBatchThreadMembershipsForRetentionPolicies(now int64, globalPolicyEndTime int64, limit int64, cursor model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error) {
+func (s *RetryLayerThreadStore) PermanentDeleteBatchThreadMembershipsForRetentionPolicies(retentionPolicyBatchConfigs model.RetentionPolicyBatchConfigs, cursor model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error) {
 
 	tries := 0
 	for {
-		result, resultVar1, err := s.ThreadStore.PermanentDeleteBatchThreadMembershipsForRetentionPolicies(now, globalPolicyEndTime, limit, cursor)
+		result, resultVar1, err := s.ThreadStore.PermanentDeleteBatchThreadMembershipsForRetentionPolicies(retentionPolicyBatchConfigs, cursor)
 		if err == nil {
 			return result, resultVar1, nil
 		}

server/channels/store/sqlstore/channel_member_history_store.go

@@ -229,7 +229,7 @@ func (s SqlChannelMemberHistoryStore) getFromChannelMembersTable(startTime int64
 // PermanentDeleteBatchForRetentionPolicies deletes a batch of records which are affected by
 // the global or a granular retention policy.
 // See `genericPermanentDeleteBatchForRetentionPolicies` for details.
-func (s SqlChannelMemberHistoryStore) PermanentDeleteBatchForRetentionPolicies(now, globalPolicyEndTime, limit int64, cursor model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error) {
+func (s SqlChannelMemberHistoryStore) PermanentDeleteBatchForRetentionPolicies(retentionPolicyBatchConfigs model.RetentionPolicyBatchConfigs, cursor model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error) {
 	builder := s.getQueryBuilder().
 		Select("ChannelMemberHistory.ChannelId, ChannelMemberHistory.UserId, ChannelMemberHistory.JoinTime").
 		From("ChannelMemberHistory")
@@ -239,9 +239,9 @@ func (s SqlChannelMemberHistoryStore) PermanentDeleteBatchForRetentionPolicies(n
 		TimeColumn:          "LeaveTime",
 		PrimaryKeys:         []string{"ChannelId", "UserId", "JoinTime"},
 		ChannelIDTable:      "ChannelMemberHistory",
-		NowMillis:           now,
+		NowMillis:           retentionPolicyBatchConfigs.Now,
-		GlobalPolicyEndTime: globalPolicyEndTime,
+		GlobalPolicyEndTime: retentionPolicyBatchConfigs.GlobalPolicyEndTime,
-		Limit:               limit,
+		Limit:               retentionPolicyBatchConfigs.Limit,
 		StoreDeletedIds:     false,
 	}, s.SqlStore, cursor)
 }

server/channels/store/sqlstore/post_store.go

@@ -2681,19 +2681,30 @@ func (s *SqlPostStore) GetPostsBatchForIndexing(startTime int64, startPostID str
 // PermanentDeleteBatchForRetentionPolicies deletes a batch of records which are affected by
 // the global or a granular retention policy.
 // See `genericPermanentDeleteBatchForRetentionPolicies` for details.
-func (s *SqlPostStore) PermanentDeleteBatchForRetentionPolicies(now, globalPolicyEndTime, limit int64, cursor model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error) {
+func (s *SqlPostStore) PermanentDeleteBatchForRetentionPolicies(retentionPolicyBatchConfigs model.RetentionPolicyBatchConfigs, cursor model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error) {
 	builder := s.getQueryBuilder().
 		Select("Posts.Id").
 		From("Posts")
+
+	if retentionPolicyBatchConfigs.PreservePinnedPosts {
+		builder = builder.Where(sq.Or{
+			sq.Eq{"Posts.IsPinned": false},
+			sq.And{
+				sq.Eq{"Posts.IsPinned": true},
+				sq.Gt{"Posts.DeleteAt": 0},
+			},
+		})
+	}
+
 	return genericPermanentDeleteBatchForRetentionPolicies(RetentionPolicyBatchDeletionInfo{
 		BaseBuilder:         builder,
 		Table:               "Posts",
 		TimeColumn:          "CreateAt",
 		PrimaryKeys:         []string{"Id"},
 		ChannelIDTable:      "Posts",
-		NowMillis:           now,
+		NowMillis:           retentionPolicyBatchConfigs.Now,
-		GlobalPolicyEndTime: globalPolicyEndTime,
+		GlobalPolicyEndTime: retentionPolicyBatchConfigs.GlobalPolicyEndTime,
-		Limit:               limit,
+		Limit:               retentionPolicyBatchConfigs.Limit,
 		StoreDeletedIds:     true,
 	}, s.SqlStore, cursor)
 }

server/channels/store/sqlstore/thread_store.go

@@ -987,7 +987,7 @@ func (s *SqlThreadStore) maintainMembershipTx(trx *sqlxTxWrapper, userID, postID
 // PermanentDeleteBatchForRetentionPolicies deletes a batch of records which are affected by
 // the global or a granular retention policy.
 // See `genericPermanentDeleteBatchForRetentionPolicies` for details.
-func (s *SqlThreadStore) PermanentDeleteBatchForRetentionPolicies(now, globalPolicyEndTime, limit int64, cursor model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error) {
+func (s *SqlThreadStore) PermanentDeleteBatchForRetentionPolicies(retentionPolicyBatchConfigs model.RetentionPolicyBatchConfigs, cursor model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error) {
 	builder := s.getQueryBuilder().
 		Select("Threads.PostId").
 		From("Threads")
@@ -997,9 +997,9 @@ func (s *SqlThreadStore) PermanentDeleteBatchForRetentionPolicies(now, globalPol
 		TimeColumn:          "LastReplyAt",
 		PrimaryKeys:         []string{"PostId"},
 		ChannelIDTable:      "Threads",
-		NowMillis:           now,
+		NowMillis:           retentionPolicyBatchConfigs.Now,
-		GlobalPolicyEndTime: globalPolicyEndTime,
+		GlobalPolicyEndTime: retentionPolicyBatchConfigs.GlobalPolicyEndTime,
-		Limit:               limit,
+		Limit:               retentionPolicyBatchConfigs.Limit,
 		StoreDeletedIds:     false,
 	}, s.SqlStore, cursor)
 }
@@ -1007,7 +1007,7 @@ func (s *SqlThreadStore) PermanentDeleteBatchForRetentionPolicies(now, globalPol
 // PermanentDeleteBatchThreadMembershipsForRetentionPolicies deletes a batch of records
 // which are affected by the global or a granular retention policy.
 // See `genericPermanentDeleteBatchForRetentionPolicies` for details.
-func (s *SqlThreadStore) PermanentDeleteBatchThreadMembershipsForRetentionPolicies(now, globalPolicyEndTime, limit int64, cursor model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error) {
+func (s *SqlThreadStore) PermanentDeleteBatchThreadMembershipsForRetentionPolicies(retentionPolicyBatchConfigs model.RetentionPolicyBatchConfigs, cursor model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error) {
 	builder := s.getQueryBuilder().
 		Select("ThreadMemberships.PostId").
 		From("ThreadMemberships").
@@ -1018,9 +1018,9 @@ func (s *SqlThreadStore) PermanentDeleteBatchThreadMembershipsForRetentionPolici
 		TimeColumn:          "LastUpdated",
 		PrimaryKeys:         []string{"PostId"},
 		ChannelIDTable:      "Threads",
-		NowMillis:           now,
+		NowMillis:           retentionPolicyBatchConfigs.Now,
-		GlobalPolicyEndTime: globalPolicyEndTime,
+		GlobalPolicyEndTime: retentionPolicyBatchConfigs.GlobalPolicyEndTime,
-		Limit:               limit,
+		Limit:               retentionPolicyBatchConfigs.Limit,
 		StoreDeletedIds:     false,
 	}, s.SqlStore, cursor)
 }

server/channels/store/store.go

@@ -327,7 +327,7 @@ type ChannelMemberHistoryStore interface {
 	LogLeaveEvent(userID string, channelID string, leaveTime int64) error
 	GetUsersInChannelDuring(startTime int64, endTime int64, channelID []string) ([]*model.ChannelMemberHistoryResult, error)
 	GetChannelsWithActivityDuring(startTime int64, endTime int64) ([]string, error)
-	PermanentDeleteBatchForRetentionPolicies(now, globalPolicyEndTime, limit int64, cursor model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error)
+	PermanentDeleteBatchForRetentionPolicies(retentionPolicyBatchConfigs model.RetentionPolicyBatchConfigs, cursor model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error)
 	DeleteOrphanedRows(limit int) (deleted int64, err error)
 	PermanentDeleteBatch(endTime int64, limit int64) (int64, error)
 	GetChannelsLeftSince(userID string, since int64) ([]string, error)
@@ -355,8 +355,8 @@ type ThreadStore interface {
 	GetMembershipForUser(userID, postID string) (*model.ThreadMembership, error)
 	DeleteMembershipForUser(userID, postID string) error
 	MaintainMembership(userID, postID string, opts ThreadMembershipOpts) (*model.ThreadMembership, error)
-	PermanentDeleteBatchForRetentionPolicies(now, globalPolicyEndTime, limit int64, cursor model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error)
+	PermanentDeleteBatchForRetentionPolicies(retentionPolicyBatchConfigs model.RetentionPolicyBatchConfigs, cursor model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error)
-	PermanentDeleteBatchThreadMembershipsForRetentionPolicies(now, globalPolicyEndTime, limit int64, cursor model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error)
+	PermanentDeleteBatchThreadMembershipsForRetentionPolicies(retentionPolicyBatchConfigs model.RetentionPolicyBatchConfigs, 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
@@ -401,7 +401,7 @@ type PostStore interface {
 	GetPostsByIds(postIds []string) ([]*model.Post, error)
 	GetEditHistoryForPost(postID string) ([]*model.Post, error)
 	GetPostsBatchForIndexing(startTime int64, startPostID string, limit int) ([]*model.PostForIndexing, error)
-	PermanentDeleteBatchForRetentionPolicies(now, globalPolicyEndTime, limit int64, cursor model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error)
+	PermanentDeleteBatchForRetentionPolicies(retentionPolicyBatchConfigs model.RetentionPolicyBatchConfigs, cursor model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error)
 	PermanentDeleteBatch(endTime int64, limit int64) (int64, error)
 	GetOldest() (*model.Post, error)
 	GetMaxPostSize() int

server/channels/store/storetest/channel_member_history_store.go

@@ -358,8 +358,11 @@ func testGetUsersInChannelAtChannelMembers(t *testing.T, rctx request.CTX, ss st
 	tableDataTruncated := false
 	for !tableDataTruncated {
 		var count int64
-		count, _, err = ss.ChannelMemberHistory().PermanentDeleteBatchForRetentionPolicies(
+		count, _, err = ss.ChannelMemberHistory().PermanentDeleteBatchForRetentionPolicies(model.RetentionPolicyBatchConfigs{
-			0, model.GetMillis(), 1000, model.RetentionPolicyCursor{})
+			Now:                 0,
+			GlobalPolicyEndTime: model.GetMillis(),
+			Limit:               1000,
+		}, model.RetentionPolicyCursor{})
 		require.NoError(t, err, "Failed to truncate ChannelMemberHistory contents")
 		tableDataTruncated = count == int64(0)
 	}
@@ -493,8 +496,11 @@ func testPermanentDeleteBatch(t *testing.T, rctx request.CTX, ss store.Store) {
 	assert.Len(t, channelMembers, 2)
 
 	// the permanent delete should delete at least one record
-	rowsDeleted, _, err := ss.ChannelMemberHistory().PermanentDeleteBatchForRetentionPolicies(
+	rowsDeleted, _, err := ss.ChannelMemberHistory().PermanentDeleteBatchForRetentionPolicies(model.RetentionPolicyBatchConfigs{
-		0, leaveTime+1, math.MaxInt64, model.RetentionPolicyCursor{})
+		Now:                 0,
+		GlobalPolicyEndTime: leaveTime + 1,
+		Limit:               math.MaxInt64,
+	}, model.RetentionPolicyCursor{})
 	require.NoError(t, err)
 	assert.NotEqual(t, int64(0), rowsDeleted)
 
@@ -540,8 +546,11 @@ func testPermanentDeleteBatchForRetentionPolicies(t *testing.T, rctx request.CTX
 	require.NoError(t, err)
 
 	nowMillis := leaveTime + *channelPolicy.PostDurationDays*model.DayInMilliseconds + 1
-	_, _, err = ss.ChannelMemberHistory().PermanentDeleteBatchForRetentionPolicies(
+	_, _, err = ss.ChannelMemberHistory().PermanentDeleteBatchForRetentionPolicies(model.RetentionPolicyBatchConfigs{
-		nowMillis, 0, limit, model.RetentionPolicyCursor{})
+		Now:                 nowMillis,
+		GlobalPolicyEndTime: 0,
+		Limit:               limit,
+	}, model.RetentionPolicyCursor{})
 	require.NoError(t, err)
 	result, err := ss.ChannelMemberHistory().GetUsersInChannelDuring(joinTime, leaveTime, []string{channel.Id})
 	require.NoError(t, err)

server/channels/store/storetest/group_store.go

@@ -2345,8 +2345,11 @@ func testChannelMembersToAdd(t *testing.T, rctx request.CTX, ss store.Store) {
 	require.Empty(t, channelMembers)
 
 	// Purging ChannelMemberHistory re-returns the result
-	_, _, nErr = ss.ChannelMemberHistory().PermanentDeleteBatchForRetentionPolicies(
+	_, _, nErr = ss.ChannelMemberHistory().PermanentDeleteBatchForRetentionPolicies(model.RetentionPolicyBatchConfigs{
-		0, model.GetMillis()+1, 100, model.RetentionPolicyCursor{})
+		Now:                 0,
+		GlobalPolicyEndTime: model.GetMillis() + 1,
+		Limit:               100,
+	}, model.RetentionPolicyCursor{})
 	require.NoError(t, nErr)
 	channelMembers, err = ss.Group().ChannelMembersToAdd(0, nil, false)
 	require.NoError(t, err)

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

@@ -196,9 +196,9 @@ func (_m *ChannelMemberHistoryStore) PermanentDeleteBatch(endTime int64, limit i
 	return r0, r1
 }
 
-// PermanentDeleteBatchForRetentionPolicies provides a mock function with given fields: now, globalPolicyEndTime, limit, cursor
+// PermanentDeleteBatchForRetentionPolicies provides a mock function with given fields: retentionPolicyBatchConfigs, cursor
-func (_m *ChannelMemberHistoryStore) PermanentDeleteBatchForRetentionPolicies(now int64, globalPolicyEndTime int64, limit int64, cursor model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error) {
+func (_m *ChannelMemberHistoryStore) PermanentDeleteBatchForRetentionPolicies(retentionPolicyBatchConfigs model.RetentionPolicyBatchConfigs, cursor model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error) {
-	ret := _m.Called(now, globalPolicyEndTime, limit, cursor)
+	ret := _m.Called(retentionPolicyBatchConfigs, cursor)
 
 	if len(ret) == 0 {
 		panic("no return value specified for PermanentDeleteBatchForRetentionPolicies")
@@ -207,23 +207,23 @@ func (_m *ChannelMemberHistoryStore) PermanentDeleteBatchForRetentionPolicies(no
 	var r0 int64
 	var r1 model.RetentionPolicyCursor
 	var r2 error
-	if rf, ok := ret.Get(0).(func(int64, int64, int64, model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error)); ok {
+	if rf, ok := ret.Get(0).(func(model.RetentionPolicyBatchConfigs, model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error)); ok {
-		return rf(now, globalPolicyEndTime, limit, cursor)
+		return rf(retentionPolicyBatchConfigs, cursor)
 	}
-	if rf, ok := ret.Get(0).(func(int64, int64, int64, model.RetentionPolicyCursor) int64); ok {
+	if rf, ok := ret.Get(0).(func(model.RetentionPolicyBatchConfigs, model.RetentionPolicyCursor) int64); ok {
-		r0 = rf(now, globalPolicyEndTime, limit, cursor)
+		r0 = rf(retentionPolicyBatchConfigs, cursor)
 	} else {
 		r0 = ret.Get(0).(int64)
 	}
 
-	if rf, ok := ret.Get(1).(func(int64, int64, int64, model.RetentionPolicyCursor) model.RetentionPolicyCursor); ok {
+	if rf, ok := ret.Get(1).(func(model.RetentionPolicyBatchConfigs, model.RetentionPolicyCursor) model.RetentionPolicyCursor); ok {
-		r1 = rf(now, globalPolicyEndTime, limit, cursor)
+		r1 = rf(retentionPolicyBatchConfigs, cursor)
 	} else {
 		r1 = ret.Get(1).(model.RetentionPolicyCursor)
 	}
 
-	if rf, ok := ret.Get(2).(func(int64, int64, int64, model.RetentionPolicyCursor) error); ok {
+	if rf, ok := ret.Get(2).(func(model.RetentionPolicyBatchConfigs, model.RetentionPolicyCursor) error); ok {
-		r2 = rf(now, globalPolicyEndTime, limit, cursor)
+		r2 = rf(retentionPolicyBatchConfigs, cursor)
 	} else {
 		r2 = ret.Error(2)
 	}

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

@@ -1120,9 +1120,9 @@ func (_m *PostStore) PermanentDeleteBatch(endTime int64, limit int64) (int64, er
 	return r0, r1
 }
 
-// PermanentDeleteBatchForRetentionPolicies provides a mock function with given fields: now, globalPolicyEndTime, limit, cursor
+// PermanentDeleteBatchForRetentionPolicies provides a mock function with given fields: retentionPolicyBatchConfigs, cursor
-func (_m *PostStore) PermanentDeleteBatchForRetentionPolicies(now int64, globalPolicyEndTime int64, limit int64, cursor model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error) {
+func (_m *PostStore) PermanentDeleteBatchForRetentionPolicies(retentionPolicyBatchConfigs model.RetentionPolicyBatchConfigs, cursor model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error) {
-	ret := _m.Called(now, globalPolicyEndTime, limit, cursor)
+	ret := _m.Called(retentionPolicyBatchConfigs, cursor)
 
 	if len(ret) == 0 {
 		panic("no return value specified for PermanentDeleteBatchForRetentionPolicies")
@@ -1131,23 +1131,23 @@ func (_m *PostStore) PermanentDeleteBatchForRetentionPolicies(now int64, globalP
 	var r0 int64
 	var r1 model.RetentionPolicyCursor
 	var r2 error
-	if rf, ok := ret.Get(0).(func(int64, int64, int64, model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error)); ok {
+	if rf, ok := ret.Get(0).(func(model.RetentionPolicyBatchConfigs, model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error)); ok {
-		return rf(now, globalPolicyEndTime, limit, cursor)
+		return rf(retentionPolicyBatchConfigs, cursor)
 	}
-	if rf, ok := ret.Get(0).(func(int64, int64, int64, model.RetentionPolicyCursor) int64); ok {
+	if rf, ok := ret.Get(0).(func(model.RetentionPolicyBatchConfigs, model.RetentionPolicyCursor) int64); ok {
-		r0 = rf(now, globalPolicyEndTime, limit, cursor)
+		r0 = rf(retentionPolicyBatchConfigs, cursor)
 	} else {
 		r0 = ret.Get(0).(int64)
 	}
 
-	if rf, ok := ret.Get(1).(func(int64, int64, int64, model.RetentionPolicyCursor) model.RetentionPolicyCursor); ok {
+	if rf, ok := ret.Get(1).(func(model.RetentionPolicyBatchConfigs, model.RetentionPolicyCursor) model.RetentionPolicyCursor); ok {
-		r1 = rf(now, globalPolicyEndTime, limit, cursor)
+		r1 = rf(retentionPolicyBatchConfigs, cursor)
 	} else {
 		r1 = ret.Get(1).(model.RetentionPolicyCursor)
 	}
 
-	if rf, ok := ret.Get(2).(func(int64, int64, int64, model.RetentionPolicyCursor) error); ok {
+	if rf, ok := ret.Get(2).(func(model.RetentionPolicyBatchConfigs, model.RetentionPolicyCursor) error); ok {
-		r2 = rf(now, globalPolicyEndTime, limit, cursor)
+		r2 = rf(retentionPolicyBatchConfigs, cursor)
 	} else {
 		r2 = ret.Error(2)
 	}

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

@@ -591,9 +591,9 @@ func (_m *ThreadStore) MarkAsRead(userID string, threadID string, timestamp int6
 	return r0
 }
 
-// PermanentDeleteBatchForRetentionPolicies provides a mock function with given fields: now, globalPolicyEndTime, limit, cursor
+// PermanentDeleteBatchForRetentionPolicies provides a mock function with given fields: retentionPolicyBatchConfigs, cursor
-func (_m *ThreadStore) PermanentDeleteBatchForRetentionPolicies(now int64, globalPolicyEndTime int64, limit int64, cursor model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error) {
+func (_m *ThreadStore) PermanentDeleteBatchForRetentionPolicies(retentionPolicyBatchConfigs model.RetentionPolicyBatchConfigs, cursor model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error) {
-	ret := _m.Called(now, globalPolicyEndTime, limit, cursor)
+	ret := _m.Called(retentionPolicyBatchConfigs, cursor)
 
 	if len(ret) == 0 {
 		panic("no return value specified for PermanentDeleteBatchForRetentionPolicies")
@@ -602,23 +602,23 @@ func (_m *ThreadStore) PermanentDeleteBatchForRetentionPolicies(now int64, globa
 	var r0 int64
 	var r1 model.RetentionPolicyCursor
 	var r2 error
-	if rf, ok := ret.Get(0).(func(int64, int64, int64, model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error)); ok {
+	if rf, ok := ret.Get(0).(func(model.RetentionPolicyBatchConfigs, model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error)); ok {
-		return rf(now, globalPolicyEndTime, limit, cursor)
+		return rf(retentionPolicyBatchConfigs, cursor)
 	}
-	if rf, ok := ret.Get(0).(func(int64, int64, int64, model.RetentionPolicyCursor) int64); ok {
+	if rf, ok := ret.Get(0).(func(model.RetentionPolicyBatchConfigs, model.RetentionPolicyCursor) int64); ok {
-		r0 = rf(now, globalPolicyEndTime, limit, cursor)
+		r0 = rf(retentionPolicyBatchConfigs, cursor)
 	} else {
 		r0 = ret.Get(0).(int64)
 	}
 
-	if rf, ok := ret.Get(1).(func(int64, int64, int64, model.RetentionPolicyCursor) model.RetentionPolicyCursor); ok {
+	if rf, ok := ret.Get(1).(func(model.RetentionPolicyBatchConfigs, model.RetentionPolicyCursor) model.RetentionPolicyCursor); ok {
-		r1 = rf(now, globalPolicyEndTime, limit, cursor)
+		r1 = rf(retentionPolicyBatchConfigs, cursor)
 	} else {
 		r1 = ret.Get(1).(model.RetentionPolicyCursor)
 	}
 
-	if rf, ok := ret.Get(2).(func(int64, int64, int64, model.RetentionPolicyCursor) error); ok {
+	if rf, ok := ret.Get(2).(func(model.RetentionPolicyBatchConfigs, model.RetentionPolicyCursor) error); ok {
-		r2 = rf(now, globalPolicyEndTime, limit, cursor)
+		r2 = rf(retentionPolicyBatchConfigs, cursor)
 	} else {
 		r2 = ret.Error(2)
 	}
@@ -626,9 +626,9 @@ func (_m *ThreadStore) PermanentDeleteBatchForRetentionPolicies(now int64, globa
 	return r0, r1, r2
 }
 
-// PermanentDeleteBatchThreadMembershipsForRetentionPolicies provides a mock function with given fields: now, globalPolicyEndTime, limit, cursor
+// PermanentDeleteBatchThreadMembershipsForRetentionPolicies provides a mock function with given fields: retentionPolicyBatchConfigs, cursor
-func (_m *ThreadStore) PermanentDeleteBatchThreadMembershipsForRetentionPolicies(now int64, globalPolicyEndTime int64, limit int64, cursor model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error) {
+func (_m *ThreadStore) PermanentDeleteBatchThreadMembershipsForRetentionPolicies(retentionPolicyBatchConfigs model.RetentionPolicyBatchConfigs, cursor model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error) {
-	ret := _m.Called(now, globalPolicyEndTime, limit, cursor)
+	ret := _m.Called(retentionPolicyBatchConfigs, cursor)
 
 	if len(ret) == 0 {
 		panic("no return value specified for PermanentDeleteBatchThreadMembershipsForRetentionPolicies")
@@ -637,23 +637,23 @@ func (_m *ThreadStore) PermanentDeleteBatchThreadMembershipsForRetentionPolicies
 	var r0 int64
 	var r1 model.RetentionPolicyCursor
 	var r2 error
-	if rf, ok := ret.Get(0).(func(int64, int64, int64, model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error)); ok {
+	if rf, ok := ret.Get(0).(func(model.RetentionPolicyBatchConfigs, model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error)); ok {
-		return rf(now, globalPolicyEndTime, limit, cursor)
+		return rf(retentionPolicyBatchConfigs, cursor)
 	}
-	if rf, ok := ret.Get(0).(func(int64, int64, int64, model.RetentionPolicyCursor) int64); ok {
+	if rf, ok := ret.Get(0).(func(model.RetentionPolicyBatchConfigs, model.RetentionPolicyCursor) int64); ok {
-		r0 = rf(now, globalPolicyEndTime, limit, cursor)
+		r0 = rf(retentionPolicyBatchConfigs, cursor)
 	} else {
 		r0 = ret.Get(0).(int64)
 	}
 
-	if rf, ok := ret.Get(1).(func(int64, int64, int64, model.RetentionPolicyCursor) model.RetentionPolicyCursor); ok {
+	if rf, ok := ret.Get(1).(func(model.RetentionPolicyBatchConfigs, model.RetentionPolicyCursor) model.RetentionPolicyCursor); ok {
-		r1 = rf(now, globalPolicyEndTime, limit, cursor)
+		r1 = rf(retentionPolicyBatchConfigs, cursor)
 	} else {
 		r1 = ret.Get(1).(model.RetentionPolicyCursor)
 	}
 
-	if rf, ok := ret.Get(2).(func(int64, int64, int64, model.RetentionPolicyCursor) error); ok {
+	if rf, ok := ret.Get(2).(func(model.RetentionPolicyBatchConfigs, model.RetentionPolicyCursor) error); ok {
-		r2 = rf(now, globalPolicyEndTime, limit, cursor)
+		r2 = rf(retentionPolicyBatchConfigs, cursor)
 	} else {
 		r2 = ret.Error(2)
 	}

server/channels/store/storetest/post_store.go

@@ -4220,7 +4220,11 @@ func testPostStorePermanentDeleteBatch(t *testing.T, rctx request.CTX, ss store.
 	o3, err = ss.Post().Save(rctx, o3)
 	require.NoError(t, err)
 
-	deleted, _, err := ss.Post().PermanentDeleteBatchForRetentionPolicies(0, 2000, 1000, model.RetentionPolicyCursor{})
+	deleted, _, err := ss.Post().PermanentDeleteBatchForRetentionPolicies(model.RetentionPolicyBatchConfigs{
+		Now:                 0,
+		GlobalPolicyEndTime: 2000,
+		Limit:               1000,
+	}, model.RetentionPolicyCursor{})
 	require.NoError(t, err)
 	require.Equal(t, int64(2), deleted)
 
@@ -4254,7 +4258,11 @@ func testPostStorePermanentDeleteBatch(t *testing.T, rctx request.CTX, ss store.
 		}
 		cursor := model.RetentionPolicyCursor{}
 
-		deleted, cursor, err = ss.Post().PermanentDeleteBatchForRetentionPolicies(0, 2, 2, cursor)
+		deleted, cursor, err = ss.Post().PermanentDeleteBatchForRetentionPolicies(model.RetentionPolicyBatchConfigs{
+			Now:                 0,
+			GlobalPolicyEndTime: 2,
+			Limit:               2,
+		}, cursor)
 		require.NoError(t, err)
 		require.Equal(t, int64(2), deleted)
 
@@ -4268,7 +4276,11 @@ func testPostStorePermanentDeleteBatch(t *testing.T, rctx request.CTX, ss store.
 		require.NoError(t, err)
 		require.Equal(t, int64(0), deleted)
 
-		deleted, _, err = ss.Post().PermanentDeleteBatchForRetentionPolicies(0, 2, 2, cursor)
+		deleted, _, err = ss.Post().PermanentDeleteBatchForRetentionPolicies(model.RetentionPolicyBatchConfigs{
+			Now:                 0,
+			GlobalPolicyEndTime: 2,
+			Limit:               2,
+		}, cursor)
 		require.NoError(t, err)
 		require.Equal(t, int64(1), deleted)
 
@@ -4301,13 +4313,21 @@ func testPostStorePermanentDeleteBatch(t *testing.T, rctx request.CTX, ss store.
 		post, err2 = ss.Post().Save(rctx, post)
 		require.NoError(t, err2)
 
-		_, _, err2 = ss.Post().PermanentDeleteBatchForRetentionPolicies(0, 2000, 1000, model.RetentionPolicyCursor{})
+		_, _, err2 = ss.Post().PermanentDeleteBatchForRetentionPolicies(model.RetentionPolicyBatchConfigs{
+			Now:                 0,
+			GlobalPolicyEndTime: 2000,
+			Limit:               1000,
+		}, model.RetentionPolicyCursor{})
 		require.NoError(t, err2)
 		_, err2 = ss.Post().Get(context.Background(), post.Id, model.GetPostsOptions{}, "", map[string]bool{})
 		require.NoError(t, err2, "global policy should have been ignored due to granular policy")
 
 		nowMillis := post.CreateAt + *channelPolicy.PostDurationDays*model.DayInMilliseconds + 1
-		_, _, err2 = ss.Post().PermanentDeleteBatchForRetentionPolicies(nowMillis, 0, 1000, model.RetentionPolicyCursor{})
+		_, _, err2 = ss.Post().PermanentDeleteBatchForRetentionPolicies(model.RetentionPolicyBatchConfigs{
+			Now:                 nowMillis,
+			GlobalPolicyEndTime: 0,
+			Limit:               1000,
+		}, model.RetentionPolicyCursor{})
 		require.NoError(t, err2)
 		_, err2 = ss.Post().Get(context.Background(), post.Id, model.GetPostsOptions{}, "", map[string]bool{})
 		require.Error(t, err2, "post should have been deleted by channel policy")
@@ -4326,7 +4346,11 @@ func testPostStorePermanentDeleteBatch(t *testing.T, rctx request.CTX, ss store.
 		require.NoError(t, err2)
 
 		nowMillis = post.CreateAt + *teamPolicy.PostDurationDays*model.DayInMilliseconds + 1
-		_, _, err2 = ss.Post().PermanentDeleteBatchForRetentionPolicies(nowMillis, 0, 1000, model.RetentionPolicyCursor{})
+		_, _, err2 = ss.Post().PermanentDeleteBatchForRetentionPolicies(model.RetentionPolicyBatchConfigs{
+			Now:                 nowMillis,
+			GlobalPolicyEndTime: 0,
+			Limit:               1000,
+		}, model.RetentionPolicyCursor{})
 		require.NoError(t, err2)
 		_, err2 = ss.Post().Get(context.Background(), post.Id, model.GetPostsOptions{}, "", map[string]bool{})
 		require.NoError(t, err2, "channel policy should have overridden team policy")
@@ -4338,7 +4362,11 @@ func testPostStorePermanentDeleteBatch(t *testing.T, rctx request.CTX, ss store.
 		err2 = ss.RetentionPolicy().Delete(channelPolicy.ID)
 		require.NoError(t, err2)
 
-		_, _, err2 = ss.Post().PermanentDeleteBatchForRetentionPolicies(nowMillis, 0, 1000, model.RetentionPolicyCursor{})
+		_, _, err2 = ss.Post().PermanentDeleteBatchForRetentionPolicies(model.RetentionPolicyBatchConfigs{
+			Now:                 nowMillis,
+			GlobalPolicyEndTime: 0,
+			Limit:               1000,
+		}, model.RetentionPolicyCursor{})
 		require.NoError(t, err2)
 		_, err2 = ss.Post().Get(context.Background(), post.Id, model.GetPostsOptions{}, "", map[string]bool{})
 		require.Error(t, err2, "post should have been deleted by team policy")
@@ -4419,7 +4447,11 @@ func testPostStorePermanentDeleteBatch(t *testing.T, rctx request.CTX, ss store.
 		require.NoError(t, err2)
 
 		nowMillis := int64(1 + 30*model.DayInMilliseconds + 1)
-		deleted, _, err2 := ss.Post().PermanentDeleteBatchForRetentionPolicies(nowMillis, 2, 1000, model.RetentionPolicyCursor{})
+		deleted, _, err2 = ss.Post().PermanentDeleteBatchForRetentionPolicies(model.RetentionPolicyBatchConfigs{
+			Now:                 nowMillis,
+			GlobalPolicyEndTime: 2,
+			Limit:               1000,
+		}, model.RetentionPolicyCursor{})
 		require.NoError(t, err2)
 		require.Equal(t, int64(3), deleted)
 
@@ -4435,6 +4467,151 @@ func testPostStorePermanentDeleteBatch(t *testing.T, rctx request.CTX, ss store.
 			require.Equal(t, int64(0), deleted)
 		}
 	})
+
+	t.Run("with preserve pinned posts true", func(t *testing.T) {
+		p1 := &model.Post{}
+		p1.ChannelId = channel.Id
+		p1.UserId = model.NewId()
+		p1.IsPinned = false
+		p1.Message = NewTestID()
+		p1.CreateAt = 1000
+		p1, err = ss.Post().Save(rctx, p1)
+		require.NoError(t, err)
+
+		p2 := &model.Post{}
+		p2.ChannelId = channel.Id
+		p2.UserId = model.NewId()
+		p2.IsPinned = false
+		p2.Message = NewTestID()
+		p2.CreateAt = 1000
+		p2, err = ss.Post().Save(rctx, p2)
+		require.NoError(t, err)
+
+		p3 := &model.Post{}
+		p3.ChannelId = channel.Id
+		p3.UserId = model.NewId()
+		p3.IsPinned = false
+		p3.Message = NewTestID()
+		p3.CreateAt = 1000
+		p3, err = ss.Post().Save(rctx, p3)
+		require.NoError(t, err)
+
+		np3 := p3.Clone()
+		np3.IsPinned = true
+
+		np3, err = ss.Post().Update(rctx, np3, p3)
+
+		deleted, _, err = ss.Post().PermanentDeleteBatchForRetentionPolicies(model.RetentionPolicyBatchConfigs{
+			Now:                 0,
+			GlobalPolicyEndTime: 2000,
+			Limit:               1000,
+			PreservePinnedPosts: true,
+		}, model.RetentionPolicyCursor{})
+		require.NoError(t, err)
+		require.Equal(t, int64(3), deleted)
+
+		_, err = ss.Post().Get(context.Background(), p1.Id, model.GetPostsOptions{}, "", map[string]bool{})
+		require.Error(t, err, "Should have not found post 1 after purge")
+
+		_, err = ss.Post().Get(context.Background(), p2.Id, model.GetPostsOptions{}, "", map[string]bool{})
+		require.Error(t, err, "Should have not found post 2 after purge")
+
+		_, err = ss.Post().Get(context.Background(), p3.Id, model.GetPostsOptions{}, "", map[string]bool{})
+		require.Error(t, err, "Should have not found post 3 before update after purge")
+
+		_, err = ss.Post().Get(context.Background(), np3.Id, model.GetPostsOptions{}, "", map[string]bool{})
+		require.NoError(t, err, "Should have found updated post 3 after purge")
+
+		rows, err = ss.RetentionPolicy().GetIdsForDeletionByTableName("Posts", 1000)
+		require.NoError(t, err)
+		require.Equal(t, 1, len(rows))
+		require.Equal(t, 3, len(rows[0].Ids))
+		// Clean up retention ids table
+		deleted, err = ss.Reaction().DeleteOrphanedRowsByIds(rows[0])
+		require.NoError(t, err)
+		require.Equal(t, int64(0), deleted)
+
+		// Clean up pinned post
+		_, _, err = ss.Post().PermanentDeleteBatchForRetentionPolicies(model.RetentionPolicyBatchConfigs{
+			Now:                 0,
+			GlobalPolicyEndTime: 2000,
+			Limit:               1000,
+			PreservePinnedPosts: false,
+		}, model.RetentionPolicyCursor{})
+		require.NoError(t, err)
+
+		rows, err = ss.RetentionPolicy().GetIdsForDeletionByTableName("Posts", 1000)
+		require.NoError(t, err)
+		require.Equal(t, 1, len(rows))
+
+		// Clean up retention ids table
+		_, err = ss.Reaction().DeleteOrphanedRowsByIds(rows[0])
+		require.NoError(t, err)
+	})
+
+	t.Run("with preserve pinned posts false", func(t *testing.T) {
+		p1 := &model.Post{}
+		p1.ChannelId = channel.Id
+		p1.UserId = model.NewId()
+		p1.IsPinned = false
+		p1.Message = NewTestID()
+		p1.CreateAt = 1000
+		p1, err = ss.Post().Save(rctx, p1)
+		require.NoError(t, err)
+
+		p2 := &model.Post{}
+		p2.ChannelId = channel.Id
+		p2.UserId = model.NewId()
+		p2.IsPinned = false
+		p2.Message = NewTestID()
+		p2.CreateAt = 1000
+		p2, err = ss.Post().Save(rctx, p2)
+		require.NoError(t, err)
+
+		p3 := &model.Post{}
+		p3.ChannelId = channel.Id
+		p3.UserId = model.NewId()
+		p3.IsPinned = false
+		p3.Message = NewTestID()
+		p3.CreateAt = 1000
+		p3, err = ss.Post().Save(rctx, p3)
+		require.NoError(t, err)
+
+		np3 := p3.Clone()
+		np3.IsPinned = true
+
+		np3, err = ss.Post().Update(rctx, np3, p3)
+
+		deleted, _, err = ss.Post().PermanentDeleteBatchForRetentionPolicies(model.RetentionPolicyBatchConfigs{
+			Now:                 0,
+			GlobalPolicyEndTime: 2000,
+			Limit:               1000,
+			PreservePinnedPosts: false,
+		}, model.RetentionPolicyCursor{})
+		require.NoError(t, err)
+		require.Equal(t, int64(4), deleted)
+
+		_, err = ss.Post().Get(context.Background(), p1.Id, model.GetPostsOptions{}, "", map[string]bool{})
+		require.Error(t, err, "Should have not found post 1 after purge")
+
+		_, err = ss.Post().Get(context.Background(), p2.Id, model.GetPostsOptions{}, "", map[string]bool{})
+		require.Error(t, err, "Should have not found post 2 after purge")
+
+		_, err = ss.Post().Get(context.Background(), p3.Id, model.GetPostsOptions{}, "", map[string]bool{})
+		require.Error(t, err, "Should have not found post 3 before update after purge")
+
+		_, err = ss.Post().Get(context.Background(), np3.Id, model.GetPostsOptions{}, "", map[string]bool{})
+		require.Error(t, err, "Should have not found updated post 3 after purge")
+
+		rows, err = ss.RetentionPolicy().GetIdsForDeletionByTableName("Posts", 1000)
+		require.NoError(t, err)
+		require.Equal(t, 1, len(rows))
+		require.Equal(t, 4, len(rows[0].Ids))
+		// Clean up retention ids table
+		deleted, err = ss.Reaction().DeleteOrphanedRowsByIds(rows[0])
+		require.NoError(t, err)
+		require.Equal(t, int64(0), deleted)
+	})
 }
 
 func testPostStoreGetOldest(t *testing.T, rctx request.CTX, ss store.Store) {

server/channels/store/storetest/preference_store.go

@@ -448,7 +448,11 @@ func testPreferenceDeleteOrphanedRows(t *testing.T, rctx request.CTX, ss store.S
 	nErr := ss.Preference().Save(model.Preferences{preference1, preference2})
 	require.NoError(t, nErr)
 
-	_, _, nErr = ss.Post().PermanentDeleteBatchForRetentionPolicies(0, 2000, limit, model.RetentionPolicyCursor{})
+	_, _, nErr = ss.Post().PermanentDeleteBatchForRetentionPolicies(model.RetentionPolicyBatchConfigs{
+		Now:                 0,
+		GlobalPolicyEndTime: 2000,
+		Limit:               limit,
+	}, model.RetentionPolicyCursor{})
 	assert.NoError(t, nErr)
 
 	rows, err := ss.RetentionPolicy().GetIdsForDeletionByTableName("Posts", 1000)

server/channels/store/storetest/reaction_store.go

@@ -710,7 +710,11 @@ func testReactionStorePermanentDeleteBatch(t *testing.T, rctx request.CTX, ss st
 		require.NoError(t, err)
 	}
 
-	_, _, err = ss.Post().PermanentDeleteBatchForRetentionPolicies(0, 2000, limit, model.RetentionPolicyCursor{})
+	_, _, err = ss.Post().PermanentDeleteBatchForRetentionPolicies(model.RetentionPolicyBatchConfigs{
+		Now:                 0,
+		GlobalPolicyEndTime: 2000,
+		Limit:               limit,
+	}, model.RetentionPolicyCursor{})
 	require.NoError(t, err)
 
 	rows, err := ss.RetentionPolicy().GetIdsForDeletionByTableName("Posts", 1000)

server/channels/store/storetest/thread_store.go

@@ -530,7 +530,11 @@ func testThreadStorePermanentDeleteBatchForRetentionPolicies(t *testing.T, rctx
 	require.NoError(t, err)
 
 	nowMillis := thread.LastReplyAt + *channelPolicy.PostDurationDays*model.DayInMilliseconds + 1
-	_, _, err = ss.Thread().PermanentDeleteBatchForRetentionPolicies(nowMillis, 0, limit, model.RetentionPolicyCursor{})
+	_, _, err = ss.Thread().PermanentDeleteBatchForRetentionPolicies(model.RetentionPolicyBatchConfigs{
+		Now:                 nowMillis,
+		GlobalPolicyEndTime: 0,
+		Limit:               limit,
+	}, model.RetentionPolicyCursor{})
 	require.NoError(t, err)
 	thread, err = ss.Thread().Get(post.Id)
 	assert.NoError(t, err)
@@ -552,7 +556,11 @@ func testThreadStorePermanentDeleteBatchForRetentionPolicies(t *testing.T, rctx
 	require.NoError(t, err)
 
 	nowMillis = thread.LastReplyAt + *teamPolicy.PostDurationDays*model.DayInMilliseconds + 1
-	_, _, err = ss.Thread().PermanentDeleteBatchForRetentionPolicies(nowMillis, 0, limit, model.RetentionPolicyCursor{})
+	_, _, err = ss.Thread().PermanentDeleteBatchForRetentionPolicies(model.RetentionPolicyBatchConfigs{
+		Now:                 nowMillis,
+		GlobalPolicyEndTime: 0,
+		Limit:               limit,
+	}, model.RetentionPolicyCursor{})
 	require.NoError(t, err)
 	_, err = ss.Thread().Get(post.Id)
 	require.NoError(t, err, "channel policy should have overridden team policy")
@@ -560,7 +568,11 @@ func testThreadStorePermanentDeleteBatchForRetentionPolicies(t *testing.T, rctx
 	// Delete channel policy and re-run team policy
 	err = ss.RetentionPolicy().Delete(channelPolicy.ID)
 	require.NoError(t, err)
-	_, _, err = ss.Thread().PermanentDeleteBatchForRetentionPolicies(nowMillis, 0, limit, model.RetentionPolicyCursor{})
+	_, _, err = ss.Thread().PermanentDeleteBatchForRetentionPolicies(model.RetentionPolicyBatchConfigs{
+		Now:                 nowMillis,
+		GlobalPolicyEndTime: 0,
+		Limit:               limit,
+	}, model.RetentionPolicyCursor{})
 	require.NoError(t, err)
 	thread, err = ss.Thread().Get(post.Id)
 	assert.NoError(t, err)
@@ -617,7 +629,11 @@ func testThreadStorePermanentDeleteBatchThreadMembershipsForRetentionPolicies(t
 	require.NoError(t, err)
 
 	nowMillis := threadMembership.LastUpdated + *channelPolicy.PostDurationDays*model.DayInMilliseconds + 1
-	_, _, err = ss.Thread().PermanentDeleteBatchThreadMembershipsForRetentionPolicies(nowMillis, 0, limit, model.RetentionPolicyCursor{})
+	_, _, err = ss.Thread().PermanentDeleteBatchThreadMembershipsForRetentionPolicies(model.RetentionPolicyBatchConfigs{
+		Now:                 nowMillis,
+		GlobalPolicyEndTime: 0,
+		Limit:               limit,
+	}, model.RetentionPolicyCursor{})
 	require.NoError(t, err)
 	_, err = ss.Thread().GetMembershipForUser(userID, post.Id)
 	require.Error(t, err, "thread membership should have been deleted by channel policy")
@@ -636,7 +652,11 @@ func testThreadStorePermanentDeleteBatchThreadMembershipsForRetentionPolicies(t
 	require.NoError(t, err)
 
 	nowMillis = threadMembership.LastUpdated + *teamPolicy.PostDurationDays*model.DayInMilliseconds + 1
-	_, _, err = ss.Thread().PermanentDeleteBatchThreadMembershipsForRetentionPolicies(nowMillis, 0, limit, model.RetentionPolicyCursor{})
+	_, _, err = ss.Thread().PermanentDeleteBatchThreadMembershipsForRetentionPolicies(model.RetentionPolicyBatchConfigs{
+		Now:                 nowMillis,
+		GlobalPolicyEndTime: 0,
+		Limit:               limit,
+	}, model.RetentionPolicyCursor{})
 	require.NoError(t, err)
 	_, err = ss.Thread().GetMembershipForUser(userID, post.Id)
 	require.NoError(t, err, "channel policy should have overridden team policy")
@@ -644,7 +664,11 @@ func testThreadStorePermanentDeleteBatchThreadMembershipsForRetentionPolicies(t
 	// Delete channel policy and re-run team policy
 	err = ss.RetentionPolicy().Delete(channelPolicy.ID)
 	require.NoError(t, err)
-	_, _, err = ss.Thread().PermanentDeleteBatchThreadMembershipsForRetentionPolicies(nowMillis, 0, limit, model.RetentionPolicyCursor{})
+	_, _, err = ss.Thread().PermanentDeleteBatchThreadMembershipsForRetentionPolicies(model.RetentionPolicyBatchConfigs{
+		Now:                 nowMillis,
+		GlobalPolicyEndTime: 0,
+		Limit:               limit,
+	}, model.RetentionPolicyCursor{})
 	require.NoError(t, err)
 	_, err = ss.Thread().GetMembershipForUser(userID, post.Id)
 	require.Error(t, err, "thread membership should have been deleted by team policy")

server/channels/store/timerlayer/timerlayer.go

@@ -3002,10 +3002,10 @@ func (s *TimerLayerChannelMemberHistoryStore) PermanentDeleteBatch(endTime int64
 	return result, err
 }
 
-func (s *TimerLayerChannelMemberHistoryStore) PermanentDeleteBatchForRetentionPolicies(now int64, globalPolicyEndTime int64, limit int64, cursor model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error) {
+func (s *TimerLayerChannelMemberHistoryStore) PermanentDeleteBatchForRetentionPolicies(retentionPolicyBatchConfigs model.RetentionPolicyBatchConfigs, cursor model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error) {
 	start := time.Now()
 
-	result, resultVar1, err := s.ChannelMemberHistoryStore.PermanentDeleteBatchForRetentionPolicies(now, globalPolicyEndTime, limit, cursor)
+	result, resultVar1, err := s.ChannelMemberHistoryStore.PermanentDeleteBatchForRetentionPolicies(retentionPolicyBatchConfigs, cursor)
 
 	elapsed := float64(time.Since(start)) / float64(time.Second)
 	if s.Root.Metrics != nil {
@@ -6661,10 +6661,10 @@ func (s *TimerLayerPostStore) PermanentDeleteBatch(endTime int64, limit int64) (
 	return result, err
 }
 
-func (s *TimerLayerPostStore) PermanentDeleteBatchForRetentionPolicies(now int64, globalPolicyEndTime int64, limit int64, cursor model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error) {
+func (s *TimerLayerPostStore) PermanentDeleteBatchForRetentionPolicies(retentionPolicyBatchConfigs model.RetentionPolicyBatchConfigs, cursor model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error) {
 	start := time.Now()
 
-	result, resultVar1, err := s.PostStore.PermanentDeleteBatchForRetentionPolicies(now, globalPolicyEndTime, limit, cursor)
+	result, resultVar1, err := s.PostStore.PermanentDeleteBatchForRetentionPolicies(retentionPolicyBatchConfigs, cursor)
 
 	elapsed := float64(time.Since(start)) / float64(time.Second)
 	if s.Root.Metrics != nil {
@@ -10835,10 +10835,10 @@ func (s *TimerLayerThreadStore) MarkAsRead(userID string, threadID string, times
 	return err
 }
 
-func (s *TimerLayerThreadStore) PermanentDeleteBatchForRetentionPolicies(now int64, globalPolicyEndTime int64, limit int64, cursor model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error) {
+func (s *TimerLayerThreadStore) PermanentDeleteBatchForRetentionPolicies(retentionPolicyBatchConfigs model.RetentionPolicyBatchConfigs, cursor model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error) {
 	start := time.Now()
 
-	result, resultVar1, err := s.ThreadStore.PermanentDeleteBatchForRetentionPolicies(now, globalPolicyEndTime, limit, cursor)
+	result, resultVar1, err := s.ThreadStore.PermanentDeleteBatchForRetentionPolicies(retentionPolicyBatchConfigs, cursor)
 
 	elapsed := float64(time.Since(start)) / float64(time.Second)
 	if s.Root.Metrics != nil {
@@ -10851,10 +10851,10 @@ func (s *TimerLayerThreadStore) PermanentDeleteBatchForRetentionPolicies(now int
 	return result, resultVar1, err
 }
 
-func (s *TimerLayerThreadStore) PermanentDeleteBatchThreadMembershipsForRetentionPolicies(now int64, globalPolicyEndTime int64, limit int64, cursor model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error) {
+func (s *TimerLayerThreadStore) PermanentDeleteBatchThreadMembershipsForRetentionPolicies(retentionPolicyBatchConfigs model.RetentionPolicyBatchConfigs, cursor model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error) {
 	start := time.Now()
 
-	result, resultVar1, err := s.ThreadStore.PermanentDeleteBatchThreadMembershipsForRetentionPolicies(now, globalPolicyEndTime, limit, cursor)
+	result, resultVar1, err := s.ThreadStore.PermanentDeleteBatchThreadMembershipsForRetentionPolicies(retentionPolicyBatchConfigs, cursor)
 
 	elapsed := float64(time.Since(start)) / float64(time.Second)
 	if s.Root.Metrics != nil {

server/public/model/config.go

@@ -3119,6 +3119,7 @@ type DataRetentionSettings struct {
 	BatchSize                      *int    `access:"compliance_data_retention_policy"`
 	TimeBetweenBatchesMilliseconds *int    `access:"compliance_data_retention_policy"`
 	RetentionIdsBatchSize          *int    `access:"compliance_data_retention_policy"`
+	PreservePinnedPosts            *bool   `access:"compliance_data_retention_policy"`
 }
 
 func (s *DataRetentionSettings) SetDefaults() {
@@ -3168,6 +3169,10 @@ func (s *DataRetentionSettings) SetDefaults() {
 	if s.RetentionIdsBatchSize == nil {
 		s.RetentionIdsBatchSize = NewPointer(DataRetentionSettingsDefaultRetentionIdsBatchSize)
 	}
+
+	if s.PreservePinnedPosts == nil {
+		s.PreservePinnedPosts = NewPointer(false)
+	}
 }
 
 // GetMessageRetentionHours returns the message retention time as an int.

server/public/model/data_retention_policy.go

@@ -91,6 +91,13 @@ type RetentionIdsForDeletion struct {
 	Ids       []string
 }
 
+type RetentionPolicyBatchConfigs struct {
+	Now                 int64
+	GlobalPolicyEndTime int64
+	Limit               int64
+	PreservePinnedPosts bool
+}
+
 func (r *RetentionIdsForDeletion) PreSave() {
 	if r.Id == "" {
 		r.Id = NewId()

webapp/channels/src/components/error_page/error_message.tsx

@@ -60,6 +60,16 @@ const ErrorMessage: React.FC<Props> = ({type, message, service, isGuest}: Props)
                 </p>
             );
             break;
+        case ErrorPageTypes.POST_NOT_FOUND:
+            errorMessage = (
+                <p>
+                    <FormattedMessage
+                        id='post.error.access'
+                        defaultMessage="The post you're requesting is private or does not exist."
+                    />
+                </p>
+            );
+            break;
         case ErrorPageTypes.CLOUD_ARCHIVED:
             errorMessage = (
                 <p>

webapp/channels/src/components/error_page/error_title.tsx

@@ -31,6 +31,14 @@ const ErrorTitle: React.FC<Props> = ({type, title}: Props) => {
                 />
             );
             break;
+        case ErrorPageTypes.POST_NOT_FOUND:
+            errorTitle = (
+                <FormattedMessage
+                    id='post.error.title'
+                    defaultMessage='Post Not Found'
+                />
+            );
+            break;
         case ErrorPageTypes.CLOUD_ARCHIVED:
             errorTitle = (
                 <FormattedMessage

webapp/channels/src/components/permalink_view/actions.ts

@@ -52,6 +52,11 @@ function focusReplyPost(post: Post, channel: Channel, teamId: string, returnTo:
     return async (dispatch, getState) => {
         const {data} = await dispatch(getPostThread(post.root_id));
 
+        if (!data) {
+            getHistory().replace(`/error?type=${ErrorPageTypes.POST_NOT_FOUND}&returnTo=${returnTo}`);
+            return {data: false};
+        }
+
         if (data!.first_inaccessible_post_time) {
             getHistory().replace(`/error?type=${ErrorPageTypes.CLOUD_ARCHIVED}&returnTo=${returnTo}`);
             return {data: false};

webapp/channels/src/i18n/en.json

@@ -5008,6 +5008,8 @@
   "post.ariaLabel.reaction": ", 1 reaction",
   "post.ariaLabel.reactionMultiple": ", {reactionCount} reactions",
   "post.ariaLabel.replyMessage": "At {time} {date}, {authorName} replied, {message}",
+  "post.error.access": "The post you're requesting is private or does not exist.",
+  "post.error.title": "Post Not Found",
   "post.reminder.acknowledgement": "You will be reminded at {reminderTime}, {reminderDate} about this message from {username}: {permaLink}",
   "post.reminder.systemBot": "Hi there, here's your reminder about this message from {username}: {permaLink}",
   "post.renderError.message": "An error occurred while rendering this post.",

webapp/channels/src/utils/constants.tsx

@@ -927,6 +927,7 @@ export const ErrorPageTypes = {
     PERMALINK_NOT_FOUND: 'permalink_not_found',
     TEAM_NOT_FOUND: 'team_not_found',
     CHANNEL_NOT_FOUND: 'channel_not_found',
+    POST_NOT_FOUND: 'post_not_found',
     CLOUD_ARCHIVED: 'cloud_archived',
 };