[MM-30831] granular data retention wireup (#17417)

* pre-checkout commit

* add API endpoints for retention policies

* allow deleting multiple teams/channels from a policy in a single request

* pre-checkout commit

* add auditing in API functions

* add permission checks

* update the store layers

* update storetest

* add check constraint on PostDuration column

* pre-checkout commit

* add query to delete posts under the scope of a granular retention policy

* add suggestions from sbishel

* allow clients to specify channels/teams when creating a new policy

* remove foreign keys referencing Channels and Teams tables

* add checks for whether teams and channels exist

* pre-checkout commit

* remove data referencing the Posts table

* pre-checkout commit

* write data store tests

* sort results of buildGetPoliciesQuery

* add missing test cases for teams

* pre-checkout commit

* add Client4 methods for data retention policy endpoints

* add uint and uint64 to app/layer_generators

* make granular policies override global policies

* fix lint errors

* pre-checkout commit

* add license to top of files

* add tests for data store layer

* add missing test cases for store layer

* run make i18n-extract

* add query to delete ChannelMemberHistory

* work in progress

* add test for old reply to old post

* fix lint error

* use COALESCE on each Posts column

* begin implementing orphaned rows worker

* split PR

* pre-checkout commit

* use RetentionPolicyWithTeamAndChannelCounts

* update app and api layers

* run make i18n-extract

* add RetentionPolicy to retrylayer_test.go

* Revert "split PR"

This reverts commit b316f03dd307a30deae931944ca7e4a1cc904605.

* fix errors caused by revert

* add suggestions from sbishel

* fix copy-paste error

* fix lint errors

* pre-checkout commit

* add function to delete orphaned rows

* use -1 for infinite retention

* remove check constraint

* copy i18n entries from master

* re-run tests with newer enterprise branch

* add team data to channel list

* add search for channels and teams in a policy

* add store tests for channel and team search

* add suggestions from mkraft

* run make einterfaces-mocks

* fix lint errors

* add suggestions from mkraft

* move removeOrphanedRows method to wireup branch

* Revert "move removeOrphanedRows method to wireup branch"

This reverts commit 94605c9b4a5378ffa44a3dec4d3f8e3306b9d33e.

* use DeleteOrphanedRows where possible

* run make i18n-extract

* use COMPLIANCE permissions

* run make migrations-bindadta

* clean up teams before test

* fix tests for TestRetentionPolicyStore

* add API endpoints for mobile

* fix lint error

* fix some of the lint errors

* move user/data_retention endpoints to data_retention.go

* Revert "fix some of the lint errors"

This reverts commit b5b2dc27566c427187db942c5c0afe319e7679c4.

* add exclude_policy_constrained parameter for /channels and /teams

* fix lint errors

* add policy_id field to GET endpoints for channels and teams

* use PolicyWithTeamID in RetentionPolicy layer

* fix lint errors

* run make i18n-extract

* update mock call in telemetry_test.go

* return status:OK in JSON instead of 204

* pre-checkout commit

* add policy_id field on channels/teams

* fix lint errors

* use sq.Eq instead of '?'

* use new subsection permissions

* update channels and teams endpoints to use new subsection permissions

* add extra search opts for channels in a policy

* fix lint errors

* allow negative post duration in patch

* remove DELETE FROM query in retention policy tests

* use *int64 for PostDuration

* re-run CI tests

* use 3-step deletion strategy for each table

* fix lint errors

* run make store-layers

* re-run CI tests

* add test with channel, team and global policies

* use common function for SQL queries

* add pagination test

* use struct for args to common SQL function

* fix lint errors

* run make i18n-extract

* check if Channels.TeamId is "" or nil

* use three OR clauses

* write separate genericRetentionPoliciesDeletion function

* add config setting for BatchSize

* add telemetry for BatchSize

* use feature flag

* add old i18n messages back in

* re-run CI tests

* update call signature in storetest

* MM-30831: Adds constant for retention default batch size.

* MM-30831: Removes comment re: optimization.

* MM-30831: Converts days to milliseconds.

* MM-30831: Reverts change to test.

* Revert "MM-30831: Reverts change to test."

This reverts commit 6d14275a1ceae682bb9e17ec69b39252b44e0c0c.

* Revert "MM-30831: Converts days to milliseconds."

This reverts commit a0cb6ec09d854a05194c1daee1c5333f260231c3.

* MM-30831: Fixes tests.

* MM-30381: Fix for change to method sig.

Co-authored-by: Max Erenberg <max.erenberg@mattermost.com>
Co-authored-by: Mattermod <mattermod@users.noreply.github.com>
Co-authored-by: Martin Kraft <martin@upspin.org>

app/syncables_test.go

@@ -299,7 +299,8 @@ func TestCreateDefaultMemberships(t *testing.T) {
 	timeAfterLeaving := model.GetMillis()
 
 	// Purging channelmemberhistory doesn't re-add user to channel
-	_, nErr := th.App.Srv().Store.ChannelMemberHistory().PermanentDeleteBatch(timeBeforeLeaving, 1000)
+	_, _, nErr := th.App.Srv().Store.ChannelMemberHistory().PermanentDeleteBatchForRetentionPolicies(
+		0, timeBeforeLeaving, 1000, model.RetentionPolicyCursor{})
 	if nErr != nil {
 		t.Errorf("error permanently deleting channelmemberhistory: %s", nErr.Error())
 	}
@@ -315,7 +316,8 @@ func TestCreateDefaultMemberships(t *testing.T) {
 	}
 
 	// Purging channelmemberhistory doesn't re-add user to channel
-	_, nErr = th.App.Srv().Jobs.Store.ChannelMemberHistory().PermanentDeleteBatch(timeAfterLeaving, 1000)
+	_, _, nErr = th.App.Srv().Store.ChannelMemberHistory().PermanentDeleteBatchForRetentionPolicies(
+		0, timeAfterLeaving, 1000, model.RetentionPolicyCursor{})
 	if nErr != nil {
 		t.Errorf("error permanently deleting channelmemberhistory: %s", nErr.Error())
 	}

i18n/en.json

@@ -6958,6 +6958,10 @@
     "id": "ent.data_retention.reactions_batch.internal_error",
     "translation": "We encountered an error permanently deleting the batch of reactions."
   },
+  {
+    "id": "ent.data_retention.run_failed.error",
+    "translation": "Data retention job failed."
+  },
   {
     "id": "ent.elasticsearch.aggregator_worker.create_index_job.error",
     "translation": "Elasticsearch aggregator worker failed to create the indexing job"

jobs/workers.go

@@ -133,7 +133,8 @@ func (srv *JobServer) InitWorkers() error {
 // Synchronization should be implemented by the caller.
 func (workers *Workers) Start() {
 	mlog.Info("Starting workers")
-	if workers.DataRetention != nil && (*workers.ConfigService.Config().DataRetentionSettings.EnableMessageDeletion || *workers.ConfigService.Config().DataRetentionSettings.EnableFileDeletion) {
+
+	if workers.DataRetention != nil {
 		go workers.DataRetention.Run()
 	}
 

model/config.go

@@ -206,6 +206,7 @@ const (
 	DATA_RETENTION_SETTINGS_DEFAULT_MESSAGE_RETENTION_DAYS  = 365
 	DATA_RETENTION_SETTINGS_DEFAULT_FILE_RETENTION_DAYS     = 365
 	DATA_RETENTION_SETTINGS_DEFAULT_DELETION_JOB_START_TIME = "02:00"
+	DATA_RETENTION_SETTINGS_DEFAULT_BATCH_SIZE              = 3000
 
 	PLUGIN_SETTINGS_DEFAULT_DIRECTORY          = "./plugins"
 	PLUGIN_SETTINGS_DEFAULT_CLIENT_DIRECTORY   = "./client/plugins"
@@ -2700,6 +2701,7 @@ type DataRetentionSettings struct {
 	MessageRetentionDays  *int    `access:"compliance_data_retention_policy"`
 	FileRetentionDays     *int    `access:"compliance_data_retention_policy"`
 	DeletionJobStartTime  *string `access:"compliance_data_retention_policy"`
+	BatchSize             *int    `access:"compliance_data_retention_policy"`
 }
 
 func (s *DataRetentionSettings) SetDefaults() {
@@ -2722,6 +2724,10 @@ func (s *DataRetentionSettings) SetDefaults() {
 	if s.DeletionJobStartTime == nil {
 		s.DeletionJobStartTime = NewString(DATA_RETENTION_SETTINGS_DEFAULT_DELETION_JOB_START_TIME)
 	}
+
+	if s.BatchSize == nil {
+		s.BatchSize = NewInt(DATA_RETENTION_SETTINGS_DEFAULT_BATCH_SIZE)
+	}
 }
 
 type JobSettings struct {

model/data_retention_policy.go

@@ -68,6 +68,12 @@ type RetentionPolicyForChannelList struct {
 	TotalCount int64                        `json:"total_count"`
 }
 
+type RetentionPolicyCursor struct {
+	ChannelPoliciesDone bool
+	TeamPoliciesDone    bool
+	GlobalPoliciesDone  bool
+}
+
 func (rp *GlobalRetentionPolicy) ToJson() []byte {
 	b, _ := json.Marshal(rp)
 	return b

model/post.go

@@ -371,6 +371,9 @@ func (o *Post) IsValid(maxPostSize int) *AppError {
 }
 
 func (o *Post) SanitizeProps() {
+	if o == nil {
+		return
+	}
 	membersToSanitize := []string{
 		PROPS_ADD_CHANNEL_MEMBER,
 	}
@@ -728,3 +731,10 @@ func (o *Post) IsFromOAuthBot() bool {
 	props := o.GetProps()
 	return props["from_webhook"] == "true" && props["override_username"] != ""
 }
+
+func (o *Post) ToNilIfInvalid() *Post {
+	if o.Id == "" {
+		return nil
+	}
+	return o
+}

services/telemetry/telemetry.go

@@ -781,6 +781,7 @@ func (ts *TelemetryService) trackConfig() {
 		"message_retention_days":  *cfg.DataRetentionSettings.MessageRetentionDays,
 		"file_retention_days":     *cfg.DataRetentionSettings.FileRetentionDays,
 		"deletion_job_start_time": *cfg.DataRetentionSettings.DeletionJobStartTime,
+		"batch_size":              *cfg.DataRetentionSettings.BatchSize,
 	})
 
 	ts.sendTelemetry(TrackConfigMessageExport, map[string]interface{}{

store/opentracinglayer/opentracinglayer.go

@@ -2338,6 +2338,24 @@ func (s *OpenTracingLayerChannelStore) UserBelongsToChannels(userID string, chan
 	return result, err
 }
 
+func (s *OpenTracingLayerChannelMemberHistoryStore) DeleteOrphanedRows(limit int) (int64, error) {
+	origCtx := s.Root.Store.Context()
+	span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "ChannelMemberHistoryStore.DeleteOrphanedRows")
+	s.Root.Store.SetContext(newCtx)
+	defer func() {
+		s.Root.Store.SetContext(origCtx)
+	}()
+
+	defer span.Finish()
+	result, err := s.ChannelMemberHistoryStore.DeleteOrphanedRows(limit)
+	if err != nil {
+		span.LogFields(spanlog.Error(err))
+		ext.Error.Set(span, true)
+	}
+
+	return result, err
+}
+
 func (s *OpenTracingLayerChannelMemberHistoryStore) GetUsersInChannelDuring(startTime int64, endTime int64, channelID string) ([]*model.ChannelMemberHistoryResult, error) {
 	origCtx := s.Root.Store.Context()
 	span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "ChannelMemberHistoryStore.GetUsersInChannelDuring")
@@ -2410,6 +2428,24 @@ func (s *OpenTracingLayerChannelMemberHistoryStore) PermanentDeleteBatch(endTime
 	return result, err
 }
 
+func (s *OpenTracingLayerChannelMemberHistoryStore) PermanentDeleteBatchForRetentionPolicies(now int64, globalPolicyEndTime int64, limit int64, cursor model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error) {
+	origCtx := s.Root.Store.Context()
+	span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "ChannelMemberHistoryStore.PermanentDeleteBatchForRetentionPolicies")
+	s.Root.Store.SetContext(newCtx)
+	defer func() {
+		s.Root.Store.SetContext(origCtx)
+	}()
+
+	defer span.Finish()
+	result, resultVar1, err := s.ChannelMemberHistoryStore.PermanentDeleteBatchForRetentionPolicies(now, globalPolicyEndTime, limit, cursor)
+	if err != nil {
+		span.LogFields(spanlog.Error(err))
+		ext.Error.Set(span, true)
+	}
+
+	return result, resultVar1, err
+}
+
 func (s *OpenTracingLayerClusterDiscoveryStore) Cleanup() error {
 	origCtx := s.Root.Store.Context()
 	span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "ClusterDiscoveryStore.Cleanup")
@@ -5054,6 +5090,24 @@ func (s *OpenTracingLayerPostStore) Delete(postID string, time int64, deleteByID
 	return err
 }
 
+func (s *OpenTracingLayerPostStore) DeleteOrphanedRows(limit int) (int64, error) {
+	origCtx := s.Root.Store.Context()
+	span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "PostStore.DeleteOrphanedRows")
+	s.Root.Store.SetContext(newCtx)
+	defer func() {
+		s.Root.Store.SetContext(origCtx)
+	}()
+
+	defer span.Finish()
+	result, err := s.PostStore.DeleteOrphanedRows(limit)
+	if err != nil {
+		span.LogFields(spanlog.Error(err))
+		ext.Error.Set(span, true)
+	}
+
+	return result, err
+}
+
 func (s *OpenTracingLayerPostStore) Get(ctx context.Context, id string, skipFetchThreads bool, collapsedThreads bool, collapsedThreadsExtended bool, userID string) (*model.PostList, error) {
 	origCtx := s.Root.Store.Context()
 	span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "PostStore.Get")
@@ -5551,6 +5605,24 @@ func (s *OpenTracingLayerPostStore) PermanentDeleteBatch(endTime int64, limit in
 	return result, err
 }
 
+func (s *OpenTracingLayerPostStore) PermanentDeleteBatchForRetentionPolicies(now int64, globalPolicyEndTime int64, limit int64, cursor model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error) {
+	origCtx := s.Root.Store.Context()
+	span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "PostStore.PermanentDeleteBatchForRetentionPolicies")
+	s.Root.Store.SetContext(newCtx)
+	defer func() {
+		s.Root.Store.SetContext(origCtx)
+	}()
+
+	defer span.Finish()
+	result, resultVar1, err := s.PostStore.PermanentDeleteBatchForRetentionPolicies(now, globalPolicyEndTime, limit, cursor)
+	if err != nil {
+		span.LogFields(spanlog.Error(err))
+		ext.Error.Set(span, true)
+	}
+
+	return result, resultVar1, err
+}
+
 func (s *OpenTracingLayerPostStore) PermanentDeleteByChannel(channelID string) error {
 	origCtx := s.Root.Store.Context()
 	span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "PostStore.PermanentDeleteByChannel")
@@ -5749,6 +5821,24 @@ func (s *OpenTracingLayerPreferenceStore) DeleteCategoryAndName(category string,
 	return err
 }
 
+func (s *OpenTracingLayerPreferenceStore) DeleteOrphanedRows(limit int) (int64, error) {
+	origCtx := s.Root.Store.Context()
+	span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "PreferenceStore.DeleteOrphanedRows")
+	s.Root.Store.SetContext(newCtx)
+	defer func() {
+		s.Root.Store.SetContext(origCtx)
+	}()
+
+	defer span.Finish()
+	result, err := s.PreferenceStore.DeleteOrphanedRows(limit)
+	if err != nil {
+		span.LogFields(spanlog.Error(err))
+		ext.Error.Set(span, true)
+	}
+
+	return result, err
+}
+
 func (s *OpenTracingLayerPreferenceStore) Get(userID string, category string, name string) (*model.Preference, error) {
 	origCtx := s.Root.Store.Context()
 	span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "PreferenceStore.Get")
@@ -5965,6 +6055,24 @@ func (s *OpenTracingLayerReactionStore) DeleteAllWithEmojiName(emojiName string)
 	return err
 }
 
+func (s *OpenTracingLayerReactionStore) DeleteOrphanedRows(limit int) (int64, error) {
+	origCtx := s.Root.Store.Context()
+	span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "ReactionStore.DeleteOrphanedRows")
+	s.Root.Store.SetContext(newCtx)
+	defer func() {
+		s.Root.Store.SetContext(origCtx)
+	}()
+
+	defer span.Finish()
+	result, err := s.ReactionStore.DeleteOrphanedRows(limit)
+	if err != nil {
+		span.LogFields(spanlog.Error(err))
+		ext.Error.Set(span, true)
+	}
+
+	return result, err
+}
+
 func (s *OpenTracingLayerReactionStore) GetForPost(postID string, allowFromCache bool) ([]*model.Reaction, error) {
 	origCtx := s.Root.Store.Context()
 	span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "ReactionStore.GetForPost")
@@ -6217,6 +6325,24 @@ func (s *OpenTracingLayerRetentionPolicyStore) Delete(id string) error {
 	return err
 }
 
+func (s *OpenTracingLayerRetentionPolicyStore) DeleteOrphanedRows(limit int) (int64, error) {
+	origCtx := s.Root.Store.Context()
+	span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "RetentionPolicyStore.DeleteOrphanedRows")
+	s.Root.Store.SetContext(newCtx)
+	defer func() {
+		s.Root.Store.SetContext(origCtx)
+	}()
+
+	defer span.Finish()
+	result, err := s.RetentionPolicyStore.DeleteOrphanedRows(limit)
+	if err != nil {
+		span.LogFields(spanlog.Error(err))
+		ext.Error.Set(span, true)
+	}
+
+	return result, err
+}
+
 func (s *OpenTracingLayerRetentionPolicyStore) Get(id string) (*model.RetentionPolicyWithTeamAndChannelCounts, error) {
 	origCtx := s.Root.Store.Context()
 	span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "RetentionPolicyStore.Get")
@@ -8812,6 +8938,24 @@ func (s *OpenTracingLayerThreadStore) DeleteMembershipForUser(userId string, pos
 	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")
+	s.Root.Store.SetContext(newCtx)
+	defer func() {
+		s.Root.Store.SetContext(origCtx)
+	}()
+
+	defer span.Finish()
+	result, err := s.ThreadStore.DeleteOrphanedRows(limit)
+	if err != nil {
+		span.LogFields(spanlog.Error(err))
+		ext.Error.Set(span, true)
+	}
+
+	return result, err
+}
+
 func (s *OpenTracingLayerThreadStore) Get(id string) (*model.Thread, error) {
 	origCtx := s.Root.Store.Context()
 	span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "ThreadStore.Get")
@@ -9010,6 +9154,42 @@ func (s *OpenTracingLayerThreadStore) MarkAsRead(userID string, threadID string,
 	return err
 }
 
+func (s *OpenTracingLayerThreadStore) PermanentDeleteBatchForRetentionPolicies(now int64, globalPolicyEndTime int64, limit int64, cursor model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error) {
+	origCtx := s.Root.Store.Context()
+	span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "ThreadStore.PermanentDeleteBatchForRetentionPolicies")
+	s.Root.Store.SetContext(newCtx)
+	defer func() {
+		s.Root.Store.SetContext(origCtx)
+	}()
+
+	defer span.Finish()
+	result, resultVar1, err := s.ThreadStore.PermanentDeleteBatchForRetentionPolicies(now, globalPolicyEndTime, limit, cursor)
+	if err != nil {
+		span.LogFields(spanlog.Error(err))
+		ext.Error.Set(span, true)
+	}
+
+	return result, resultVar1, err
+}
+
+func (s *OpenTracingLayerThreadStore) PermanentDeleteBatchThreadMembershipsForRetentionPolicies(now int64, globalPolicyEndTime int64, limit int64, cursor model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error) {
+	origCtx := s.Root.Store.Context()
+	span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "ThreadStore.PermanentDeleteBatchThreadMembershipsForRetentionPolicies")
+	s.Root.Store.SetContext(newCtx)
+	defer func() {
+		s.Root.Store.SetContext(origCtx)
+	}()
+
+	defer span.Finish()
+	result, resultVar1, err := s.ThreadStore.PermanentDeleteBatchThreadMembershipsForRetentionPolicies(now, globalPolicyEndTime, limit, cursor)
+	if err != nil {
+		span.LogFields(spanlog.Error(err))
+		ext.Error.Set(span, true)
+	}
+
+	return result, resultVar1, err
+}
+
 func (s *OpenTracingLayerThreadStore) Save(thread *model.Thread) (*model.Thread, error) {
 	origCtx := s.Root.Store.Context()
 	span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "ThreadStore.Save")

store/retrylayer/retrylayer.go

@@ -2488,6 +2488,26 @@ func (s *RetryLayerChannelStore) UserBelongsToChannels(userID string, channelIds
 
 }
 
+func (s *RetryLayerChannelMemberHistoryStore) DeleteOrphanedRows(limit int) (int64, error) {
+
+	tries := 0
+	for {
+		result, err := s.ChannelMemberHistoryStore.DeleteOrphanedRows(limit)
+		if err == nil {
+			return result, nil
+		}
+		if !isRepeatableError(err) {
+			return result, err
+		}
+		tries++
+		if tries >= 3 {
+			err = errors.Wrap(err, "giving up after 3 consecutive repeatable transaction failures")
+			return result, err
+		}
+	}
+
+}
+
 func (s *RetryLayerChannelMemberHistoryStore) GetUsersInChannelDuring(startTime int64, endTime int64, channelID string) ([]*model.ChannelMemberHistoryResult, error) {
 
 	tries := 0
@@ -2568,6 +2588,26 @@ func (s *RetryLayerChannelMemberHistoryStore) PermanentDeleteBatch(endTime int64
 
 }
 
+func (s *RetryLayerChannelMemberHistoryStore) PermanentDeleteBatchForRetentionPolicies(now int64, globalPolicyEndTime int64, limit int64, cursor model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error) {
+
+	tries := 0
+	for {
+		result, resultVar1, err := s.ChannelMemberHistoryStore.PermanentDeleteBatchForRetentionPolicies(now, globalPolicyEndTime, limit, cursor)
+		if err == nil {
+			return result, resultVar1, nil
+		}
+		if !isRepeatableError(err) {
+			return result, resultVar1, err
+		}
+		tries++
+		if tries >= 3 {
+			err = errors.Wrap(err, "giving up after 3 consecutive repeatable transaction failures")
+			return result, resultVar1, err
+		}
+	}
+
+}
+
 func (s *RetryLayerClusterDiscoveryStore) Cleanup() error {
 
 	tries := 0
@@ -5472,6 +5512,26 @@ func (s *RetryLayerPostStore) Delete(postID string, time int64, deleteByID strin
 
 }
 
+func (s *RetryLayerPostStore) DeleteOrphanedRows(limit int) (int64, error) {
+
+	tries := 0
+	for {
+		result, err := s.PostStore.DeleteOrphanedRows(limit)
+		if err == nil {
+			return result, nil
+		}
+		if !isRepeatableError(err) {
+			return result, err
+		}
+		tries++
+		if tries >= 3 {
+			err = errors.Wrap(err, "giving up after 3 consecutive repeatable transaction failures")
+			return result, err
+		}
+	}
+
+}
+
 func (s *RetryLayerPostStore) Get(ctx context.Context, id string, skipFetchThreads bool, collapsedThreads bool, collapsedThreadsExtended bool, userID string) (*model.PostList, error) {
 
 	tries := 0
@@ -5990,6 +6050,26 @@ 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) {
+
+	tries := 0
+	for {
+		result, resultVar1, err := s.PostStore.PermanentDeleteBatchForRetentionPolicies(now, globalPolicyEndTime, limit, cursor)
+		if err == nil {
+			return result, resultVar1, nil
+		}
+		if !isRepeatableError(err) {
+			return result, resultVar1, err
+		}
+		tries++
+		if tries >= 3 {
+			err = errors.Wrap(err, "giving up after 3 consecutive repeatable transaction failures")
+			return result, resultVar1, err
+		}
+	}
+
+}
+
 func (s *RetryLayerPostStore) PermanentDeleteByChannel(channelID string) error {
 
 	tries := 0
@@ -6210,6 +6290,26 @@ func (s *RetryLayerPreferenceStore) DeleteCategoryAndName(category string, name
 
 }
 
+func (s *RetryLayerPreferenceStore) DeleteOrphanedRows(limit int) (int64, error) {
+
+	tries := 0
+	for {
+		result, err := s.PreferenceStore.DeleteOrphanedRows(limit)
+		if err == nil {
+			return result, nil
+		}
+		if !isRepeatableError(err) {
+			return result, err
+		}
+		tries++
+		if tries >= 3 {
+			err = errors.Wrap(err, "giving up after 3 consecutive repeatable transaction failures")
+			return result, err
+		}
+	}
+
+}
+
 func (s *RetryLayerPreferenceStore) Get(userID string, category string, name string) (*model.Preference, error) {
 
 	tries := 0
@@ -6450,6 +6550,26 @@ func (s *RetryLayerReactionStore) DeleteAllWithEmojiName(emojiName string) error
 
 }
 
+func (s *RetryLayerReactionStore) DeleteOrphanedRows(limit int) (int64, error) {
+
+	tries := 0
+	for {
+		result, err := s.ReactionStore.DeleteOrphanedRows(limit)
+		if err == nil {
+			return result, nil
+		}
+		if !isRepeatableError(err) {
+			return result, err
+		}
+		tries++
+		if tries >= 3 {
+			err = errors.Wrap(err, "giving up after 3 consecutive repeatable transaction failures")
+			return result, err
+		}
+	}
+
+}
+
 func (s *RetryLayerReactionStore) GetForPost(postID string, allowFromCache bool) ([]*model.Reaction, error) {
 
 	tries := 0
@@ -6730,6 +6850,26 @@ func (s *RetryLayerRetentionPolicyStore) Delete(id string) error {
 
 }
 
+func (s *RetryLayerRetentionPolicyStore) DeleteOrphanedRows(limit int) (int64, error) {
+
+	tries := 0
+	for {
+		result, err := s.RetentionPolicyStore.DeleteOrphanedRows(limit)
+		if err == nil {
+			return result, nil
+		}
+		if !isRepeatableError(err) {
+			return result, err
+		}
+		tries++
+		if tries >= 3 {
+			err = errors.Wrap(err, "giving up after 3 consecutive repeatable transaction failures")
+			return result, err
+		}
+	}
+
+}
+
 func (s *RetryLayerRetentionPolicyStore) Get(id string) (*model.RetentionPolicyWithTeamAndChannelCounts, error) {
 
 	tries := 0
@@ -9588,6 +9728,26 @@ func (s *RetryLayerThreadStore) DeleteMembershipForUser(userId string, postID st
 
 }
 
+func (s *RetryLayerThreadStore) DeleteOrphanedRows(limit int) (int64, error) {
+
+	tries := 0
+	for {
+		result, err := s.ThreadStore.DeleteOrphanedRows(limit)
+		if err == nil {
+			return result, nil
+		}
+		if !isRepeatableError(err) {
+			return result, err
+		}
+		tries++
+		if tries >= 3 {
+			err = errors.Wrap(err, "giving up after 3 consecutive repeatable transaction failures")
+			return result, err
+		}
+	}
+
+}
+
 func (s *RetryLayerThreadStore) Get(id string) (*model.Thread, error) {
 
 	tries := 0
@@ -9808,6 +9968,46 @@ 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) {
+
+	tries := 0
+	for {
+		result, resultVar1, err := s.ThreadStore.PermanentDeleteBatchForRetentionPolicies(now, globalPolicyEndTime, limit, cursor)
+		if err == nil {
+			return result, resultVar1, nil
+		}
+		if !isRepeatableError(err) {
+			return result, resultVar1, err
+		}
+		tries++
+		if tries >= 3 {
+			err = errors.Wrap(err, "giving up after 3 consecutive repeatable transaction failures")
+			return result, resultVar1, err
+		}
+	}
+
+}
+
+func (s *RetryLayerThreadStore) PermanentDeleteBatchThreadMembershipsForRetentionPolicies(now int64, globalPolicyEndTime int64, limit int64, cursor model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error) {
+
+	tries := 0
+	for {
+		result, resultVar1, err := s.ThreadStore.PermanentDeleteBatchThreadMembershipsForRetentionPolicies(now, globalPolicyEndTime, limit, cursor)
+		if err == nil {
+			return result, resultVar1, nil
+		}
+		if !isRepeatableError(err) {
+			return result, resultVar1, err
+		}
+		tries++
+		if tries >= 3 {
+			err = errors.Wrap(err, "giving up after 3 consecutive repeatable transaction failures")
+			return result, resultVar1, err
+		}
+	}
+
+}
+
 func (s *RetryLayerThreadStore) Save(thread *model.Thread) (*model.Thread, error) {
 
 	tries := 0

store/sqlstore/channel_member_history_store.go

@@ -165,6 +165,46 @@ func (s SqlChannelMemberHistoryStore) getFromChannelMembersTable(startTime int64
 	return histories, nil
 }
 
+// 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) {
+	builder := s.getQueryBuilder().
+		Select("ChannelMemberHistory.ChannelId, ChannelMemberHistory.UserId, ChannelMemberHistory.JoinTime").
+		From("ChannelMemberHistory")
+	return genericPermanentDeleteBatchForRetentionPolicies(RetentionPolicyBatchDeletionInfo{
+		BaseBuilder:         builder,
+		Table:               "ChannelMemberHistory",
+		TimeColumn:          "LeaveTime",
+		PrimaryKeys:         []string{"ChannelId", "UserId", "JoinTime"},
+		ChannelIDTable:      "ChannelMemberHistory",
+		NowMillis:           now,
+		GlobalPolicyEndTime: globalPolicyEndTime,
+		Limit:               limit,
+	}, s.SqlStore, cursor)
+}
+
+// DeleteOrphanedRows removes entries from ChannelMemberHistory when a corresponding channel no longer exists.
+func (s SqlChannelMemberHistoryStore) DeleteOrphanedRows(limit int) (deleted int64, err error) {
+	// We need the extra level of nesting to deal with MySQL's locking
+	const query = `
+	DELETE FROM ChannelMemberHistory WHERE (ChannelId, UserId, JoinTime) IN (
+		SELECT * FROM (
+			SELECT ChannelId, UserId, JoinTime FROM ChannelMemberHistory
+			LEFT JOIN Channels ON ChannelMemberHistory.ChannelId = Channels.Id
+			WHERE Channels.Id IS NULL
+			LIMIT :Limit
+		) AS A
+	)`
+	props := map[string]interface{}{"Limit": limit}
+	result, err := s.GetMaster().Exec(query, props)
+	if err != nil {
+		return
+	}
+	deleted, err = result.RowsAffected()
+	return
+}
+
 func (s SqlChannelMemberHistoryStore) PermanentDeleteBatch(endTime int64, limit int64) (int64, error) {
 	var (
 		query string

store/sqlstore/post_store.go

@@ -7,6 +7,7 @@ import (
 	"context"
 	"database/sql"
 	"fmt"
+	"reflect"
 	"regexp"
 	"strconv"
 	"strings"
@@ -42,8 +43,34 @@ type postWithExtra struct {
 func (s *SqlPostStore) ClearCaches() {
 }
 
-func postSliceColumns() []string {
-	return []string{"Id", "CreateAt", "UpdateAt", "EditAt", "DeleteAt", "IsPinned", "UserId", "ChannelId", "RootId", "ParentId", "OriginalId", "Message", "Type", "Props", "Hashtags", "Filenames", "FileIds", "HasReactions", "RemoteId"}
+func postSliceColumnsWithTypes() []struct {
+	Name string
+	Type reflect.Kind
+} {
+	return []struct {
+		Name string
+		Type reflect.Kind
+	}{
+		{"Id", reflect.String},
+		{"CreateAt", reflect.Int64},
+		{"UpdateAt", reflect.Int64},
+		{"EditAt", reflect.Int64},
+		{"DeleteAt", reflect.Int64},
+		{"IsPinned", reflect.Bool},
+		{"UserId", reflect.String},
+		{"ChannelId", reflect.String},
+		{"RootId", reflect.String},
+		{"ParentId", reflect.String},
+		{"OriginalId", reflect.String},
+		{"Message", reflect.String},
+		{"Type", reflect.String},
+		{"Props", reflect.Map},
+		{"Hashtags", reflect.String},
+		{"Filenames", reflect.Slice},
+		{"FileIds", reflect.Slice},
+		{"HasReactions", reflect.Bool},
+		{"RemoteId", reflect.String},
+	}
 }
 
 func postToSlice(post *model.Post) []interface{} {
@@ -70,6 +97,37 @@ func postToSlice(post *model.Post) []interface{} {
 	}
 }
 
+func postSliceColumns() []string {
+	colInfos := postSliceColumnsWithTypes()
+	cols := make([]string, len(colInfos))
+	for i, colInfo := range colInfos {
+		cols[i] = colInfo.Name
+	}
+	return cols
+}
+
+func postSliceCoalesceQuery() string {
+	colInfos := postSliceColumnsWithTypes()
+	cols := make([]string, len(colInfos))
+	for i, colInfo := range colInfos {
+		var defaultValue string
+		switch colInfo.Type {
+		case reflect.String:
+			defaultValue = "''"
+		case reflect.Int64:
+			defaultValue = "0"
+		case reflect.Bool:
+			defaultValue = "false"
+		case reflect.Map:
+			defaultValue = "'{}'"
+		case reflect.Slice:
+			defaultValue = "'[]'"
+		}
+		cols[i] = "COALESCE(Posts." + colInfo.Name + "," + defaultValue + ") AS " + colInfo.Name
+	}
+	return strings.Join(cols, ",")
+}
+
 func newSqlPostStore(sqlStore *SqlStore, metrics einterfaces.MetricsInterface) store.PostStore {
 	s := &SqlPostStore{
 		SqlStore:          sqlStore,
@@ -1927,6 +1985,46 @@ func (s *SqlPostStore) GetPostsBatchForIndexing(startTime int64, endTime int64,
 	return posts, nil
 }
 
+// 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) {
+	builder := s.getQueryBuilder().
+		Select("Posts.Id").
+		From("Posts")
+	return genericPermanentDeleteBatchForRetentionPolicies(RetentionPolicyBatchDeletionInfo{
+		BaseBuilder:         builder,
+		Table:               "Posts",
+		TimeColumn:          "CreateAt",
+		PrimaryKeys:         []string{"Id"},
+		ChannelIDTable:      "Posts",
+		NowMillis:           now,
+		GlobalPolicyEndTime: globalPolicyEndTime,
+		Limit:               limit,
+	}, s.SqlStore, cursor)
+}
+
+// DeleteOrphanedRows removes entries from Posts when a corresponding channel no longer exists.
+func (s *SqlPostStore) DeleteOrphanedRows(limit int) (deleted int64, err error) {
+	// We need the extra level of nesting to deal with MySQL's locking
+	const query = `
+	DELETE FROM Posts WHERE Id IN (
+		SELECT * FROM (
+			SELECT Posts.Id FROM Posts
+			LEFT JOIN Channels ON Posts.ChannelId = Channels.Id
+			WHERE Channels.Id IS NULL
+			LIMIT :Limit
+		) AS A
+	)`
+	props := map[string]interface{}{"Limit": limit}
+	result, err := s.GetMaster().Exec(query, props)
+	if err != nil {
+		return
+	}
+	deleted, err = result.RowsAffected()
+	return
+}
+
 func (s *SqlPostStore) PermanentDeleteBatch(endTime int64, limit int64) (int64, error) {
 	var query string
 	if s.DriverName() == "postgres" {

store/sqlstore/preference_store.go

@@ -265,6 +265,28 @@ func (s SqlPreferenceStore) DeleteCategoryAndName(category string, name string)
 	return nil
 }
 
+// DeleteOrphanedRows removes entries from Preferences (flagged post) when a
+// corresponding post no longer exists.
+func (s *SqlPreferenceStore) DeleteOrphanedRows(limit int) (deleted int64, err error) {
+	// We need the extra level of nesting to deal with MySQL's locking
+	const query = `
+	DELETE FROM Preferences WHERE Name IN (
+		SELECT * FROM (
+			SELECT Preferences.Name FROM Preferences
+			LEFT JOIN Posts ON Preferences.Name = Posts.Id
+			WHERE Posts.Id IS NULL AND Category = :Category
+			LIMIT :Limit
+		) AS A
+	)`
+	props := map[string]interface{}{"Limit": limit, "Category": model.PREFERENCE_CATEGORY_FLAGGED_POST}
+	result, err := s.GetMaster().Exec(query, props)
+	if err != nil {
+		return
+	}
+	deleted, err = result.RowsAffected()
+	return
+}
+
 func (s SqlPreferenceStore) CleanupFlagsBatch(limit int64) (int64, error) {
 	if limit < 0 {
 		// uint64 does not throw an error, it overflows if it is negative.

store/sqlstore/reaction_store.go

@@ -209,6 +209,27 @@ func (s *SqlReactionStore) DeleteAllWithEmojiName(emojiName string) error {
 	return nil
 }
 
+// DeleteOrphanedRows removes entries from Reactions when a corresponding post no longer exists.
+func (s *SqlReactionStore) DeleteOrphanedRows(limit int) (deleted int64, err error) {
+	// We need the extra level of nesting to deal with MySQL's locking
+	const query = `
+	DELETE FROM Reactions WHERE PostId IN (
+		SELECT * FROM (
+			SELECT PostId FROM Reactions
+			LEFT JOIN Posts ON Reactions.PostId = Posts.Id
+			WHERE Posts.Id IS NULL
+			LIMIT :Limit
+		) AS A
+	)`
+	props := map[string]interface{}{"Limit": limit}
+	result, err := s.GetMaster().Exec(query, props)
+	if err != nil {
+		return
+	}
+	deleted, err = result.RowsAffected()
+	return
+}
+
 func (s *SqlReactionStore) PermanentDeleteBatch(endTime int64, limit int64) (int64, error) {
 	var query string
 	if s.DriverName() == "postgres" {

store/sqlstore/retention_policy_store.go

@@ -5,6 +5,8 @@ package sqlstore
 
 import (
 	"database/sql"
+	"strconv"
+	"strings"
 
 	sq "github.com/Masterminds/squirrel"
 	"github.com/go-sql-driver/mysql"
@@ -541,6 +543,49 @@ func (s *SqlRetentionPolicyStore) RemoveTeams(policyId string, teamIds []string)
 	return err
 }
 
+// DeleteOrphanedRows removes entries from RetentionPoliciesChannels and RetentionPoliciesTeams
+// where a channel or team no longer exists.
+func (s *SqlRetentionPolicyStore) DeleteOrphanedRows(limit int) (deleted int64, err error) {
+	// We need the extra level of nesting to deal with MySQL's locking
+	const rpcDeleteQuery = `
+	DELETE FROM RetentionPoliciesChannels WHERE ChannelId IN (
+		SELECT * FROM (
+			SELECT ChannelId FROM RetentionPoliciesChannels
+			LEFT JOIN Channels ON RetentionPoliciesChannels.ChannelId = Channels.Id
+			WHERE Channels.Id IS NULL
+			LIMIT :Limit
+		) AS A
+	)`
+	const rptDeleteQuery = `
+	DELETE FROM RetentionPoliciesTeams WHERE TeamId IN (
+		SELECT * FROM (
+			SELECT TeamId FROM RetentionPoliciesTeams
+			LEFT JOIN Teams ON RetentionPoliciesTeams.TeamId = Teams.Id
+			WHERE Teams.Id IS NULL
+			LIMIT :Limit
+		) AS A
+	)`
+	props := map[string]interface{}{"Limit": limit}
+	result, err := s.GetMaster().Exec(rpcDeleteQuery, props)
+	if err != nil {
+		return
+	}
+	rpcDeleted, err := result.RowsAffected()
+	if err != nil {
+		return
+	}
+	result, err = s.GetMaster().Exec(rptDeleteQuery, props)
+	if err != nil {
+		return
+	}
+	rptDeleted, err := result.RowsAffected()
+	if err != nil {
+		return
+	}
+	deleted = rpcDeleted + rptDeleted
+	return
+}
+
 func (s *SqlRetentionPolicyStore) GetTeamPoliciesForUser(userID string, offset, limit int) (policies []*model.RetentionPolicyForTeam, err error) {
 	const query = `
 	SELECT Teams.Id, RetentionPolicies.PostDuration
@@ -606,3 +651,173 @@ func (s *SqlRetentionPolicyStore) GetChannelPoliciesCountForUser(userID string)
 	props := map[string]interface{}{"UserId": userID}
 	return s.GetReplica().SelectInt(query, props)
 }
+
+// RetentionPolicyBatchDeletionInfo gives information on how to delete records
+// under a retention policy; see `genericPermanentDeleteBatchForRetentionPolicies`.
+//
+// `BaseBuilder` should already have selected the primary key(s) for the main table
+// and should be joined to a table with a ChannelId column, which will be used to join
+// on the Channels table.
+// `Table` is the name of the table from which records are being deleted.
+// `TimeColumn` is the name of the column which contains the timestamp of the record.
+// `PrimaryKeys` contains the primary keys of `table`. It should be the same as the
+// `From` clause in `baseBuilder`.
+// `ChannelIDTable` is the table which contains the ChannelId column, it may be the
+// same as `table`, or will be different if a join was used.
+// `NowMillis` must be a Unix timestamp in milliseconds and is used by the granular
+// policies; if `nowMillis - timestamp(record)` is greater than
+// the post duration of a granular policy, than the record will be deleted.
+// `GlobalPolicyEndTime` is used by the global policy; any record older than this time
+// will be deleted by the global policy if it does not fall under a granular policy.
+// To disable the granular policies, set `NowMillis` to 0.
+// To disable the global policy, set `GlobalPolicyEndTime` to 0.
+type RetentionPolicyBatchDeletionInfo struct {
+	BaseBuilder         sq.SelectBuilder
+	Table               string
+	TimeColumn          string
+	PrimaryKeys         []string
+	ChannelIDTable      string
+	NowMillis           int64
+	GlobalPolicyEndTime int64
+	Limit               int64
+}
+
+// genericPermanentDeleteBatchForRetentionPolicies is a helper function for tables
+// which need to delete records for granular and global policies.
+func genericPermanentDeleteBatchForRetentionPolicies(
+	r RetentionPolicyBatchDeletionInfo,
+	s *SqlStore,
+	cursor model.RetentionPolicyCursor,
+) (int64, model.RetentionPolicyCursor, error) {
+	baseBuilder := r.BaseBuilder.InnerJoin("Channels ON " + r.ChannelIDTable + ".ChannelId = Channels.Id")
+
+	scopedTimeColumn := r.Table + "." + r.TimeColumn
+	nowStr := strconv.FormatInt(r.NowMillis, 10)
+	// A record falls under the scope of a granular retention policy if:
+	// 1. The policy's post duration is >= 0
+	// 2. The record's lifespan has not exceeded the policy's post duration
+	const millisecondsInADay = 24 * 60 * 60 * 1000
+	fallsUnderGranularPolicy := sq.And{
+		sq.GtOrEq{"RetentionPolicies.PostDuration": 0},
+		sq.Expr(nowStr + " - " + scopedTimeColumn + " > RetentionPolicies.PostDuration * " + strconv.FormatInt(millisecondsInADay, 10)),
+	}
+
+	// If the caller wants to disable the global policy from running
+	if r.GlobalPolicyEndTime <= 0 {
+		cursor.GlobalPoliciesDone = true
+	}
+	// If the caller wants to disable the granular policies from running
+	if r.NowMillis <= 0 {
+		cursor.ChannelPoliciesDone = true
+		cursor.TeamPoliciesDone = true
+	}
+
+	var totalRowsAffected int64
+
+	// First, delete all of the records which fall under the scope of a channel-specific policy
+	if !cursor.ChannelPoliciesDone {
+		channelPoliciesBuilder := baseBuilder.
+			InnerJoin("RetentionPoliciesChannels ON " + r.ChannelIDTable + ".ChannelId = RetentionPoliciesChannels.ChannelId").
+			InnerJoin("RetentionPolicies ON RetentionPoliciesChannels.PolicyId = RetentionPolicies.Id").
+			Where(fallsUnderGranularPolicy).
+			Limit(uint64(r.Limit))
+		rowsAffected, err := genericRetentionPoliciesDeletion(channelPoliciesBuilder, r, s)
+		if err != nil {
+			return 0, cursor, err
+		}
+		if rowsAffected < r.Limit {
+			cursor.ChannelPoliciesDone = true
+		}
+		totalRowsAffected += rowsAffected
+		r.Limit -= rowsAffected
+	}
+
+	// Next, delete all of the records which fall under the scope of a team-specific policy
+	if cursor.ChannelPoliciesDone && !cursor.TeamPoliciesDone {
+		// Channel-specific policies override team-specific policies.
+		teamPoliciesBuilder := baseBuilder.
+			LeftJoin("RetentionPoliciesChannels ON " + r.ChannelIDTable + ".ChannelId = RetentionPoliciesChannels.ChannelId").
+			InnerJoin("RetentionPoliciesTeams ON Channels.TeamId = RetentionPoliciesTeams.TeamId").
+			InnerJoin("RetentionPolicies ON RetentionPoliciesTeams.PolicyId = RetentionPolicies.Id").
+			Where(sq.And{
+				sq.Eq{"RetentionPoliciesChannels.PolicyId": nil},
+				sq.Expr("RetentionPoliciesTeams.PolicyId = RetentionPolicies.Id"),
+			}).
+			Where(fallsUnderGranularPolicy).
+			Limit(uint64(r.Limit))
+		rowsAffected, err := genericRetentionPoliciesDeletion(teamPoliciesBuilder, r, s)
+		if err != nil {
+			return 0, cursor, err
+		}
+		if rowsAffected < r.Limit {
+			cursor.TeamPoliciesDone = true
+		}
+		totalRowsAffected += rowsAffected
+		r.Limit -= rowsAffected
+	}
+
+	// Finally, delete all of the records which fall under the scope of the global policy
+	if cursor.ChannelPoliciesDone && cursor.TeamPoliciesDone && !cursor.GlobalPoliciesDone {
+		// Granular policies override the global policy.
+		globalPolicyBuilder := baseBuilder.
+			LeftJoin("RetentionPoliciesChannels ON " + r.ChannelIDTable + ".ChannelId = RetentionPoliciesChannels.ChannelId").
+			LeftJoin("RetentionPoliciesTeams ON Channels.TeamId = RetentionPoliciesTeams.TeamId").
+			LeftJoin("RetentionPolicies ON RetentionPoliciesChannels.PolicyId = RetentionPolicies.Id").
+			Where(sq.And{
+				sq.Eq{"RetentionPoliciesChannels.PolicyId": nil},
+				sq.Eq{"RetentionPoliciesTeams.PolicyId": nil},
+			}).
+			Where(sq.Lt{scopedTimeColumn: r.GlobalPolicyEndTime}).
+			Limit(uint64(r.Limit))
+		rowsAffected, err := genericRetentionPoliciesDeletion(globalPolicyBuilder, r, s)
+		if err != nil {
+			return 0, cursor, err
+		}
+		if rowsAffected < r.Limit {
+			cursor.GlobalPoliciesDone = true
+		}
+		totalRowsAffected += rowsAffected
+	}
+
+	return totalRowsAffected, cursor, nil
+}
+
+// genericRetentionPoliciesDeletion actually executes the DELETE query using a sq.SelectBuilder
+// which selects the rows to delete.
+func genericRetentionPoliciesDeletion(
+	builder sq.SelectBuilder,
+	r RetentionPolicyBatchDeletionInfo,
+	s *SqlStore,
+) (rowsAffected int64, err error) {
+	query, args, err := builder.ToSql()
+	if err != nil {
+		return 0, errors.Wrap(err, r.Table+"_tosql")
+	}
+	if s.DriverName() == model.DATABASE_DRIVER_POSTGRES {
+		primaryKeysStr := "(" + strings.Join(r.PrimaryKeys, ",") + ")"
+		query = `
+		DELETE FROM ` + r.Table + ` WHERE ` + primaryKeysStr + ` IN (
+		` + query + `
+		)`
+	} else {
+		// MySQL does not support the LIMIT clause in a subquery with IN
+		clauses := make([]string, len(r.PrimaryKeys))
+		for i, key := range r.PrimaryKeys {
+			clauses[i] = r.Table + "." + key + " = A." + key
+		}
+		joinClause := strings.Join(clauses, " AND ")
+		query = `
+		DELETE ` + r.Table + ` FROM ` + r.Table + ` INNER JOIN (
+		` + query + `
+		) AS A ON ` + joinClause
+	}
+	result, err := s.GetMaster().Exec(query, args...)
+	if err != nil {
+		return 0, errors.Wrap(err, "failed to delete "+r.Table)
+	}
+	rowsAffected, err = result.RowsAffected()
+	if err != nil {
+		return 0, errors.Wrap(err, "failed to get rows affected for "+r.Table)
+	}
+	return
+}

store/sqlstore/thread_store.go

@@ -132,12 +132,15 @@ func (s *SqlThreadStore) GetThreadsForUser(userId, teamId string, opts model.Get
 
 	unreadRepliesQuery := "SELECT COUNT(Posts.Id) From Posts Where Posts.RootId=ThreadMemberships.PostId AND Posts.CreateAt >= ThreadMemberships.LastViewed"
 	fetchConditions := sq.And{
-		sq.Or{sq.Eq{"Channels.TeamId": teamId}, sq.Eq{"Channels.TeamId": ""}},
+		sq.Or{sq.Eq{"Channels.TeamId": teamId}, sq.Eq{"Channels.TeamId": ""}, sq.Eq{"Channels.TeamId": nil}},
 		sq.Eq{"ThreadMemberships.UserId": userId},
 		sq.Eq{"ThreadMemberships.Following": true},
 	}
 	if !opts.Deleted {
-		fetchConditions = sq.And{fetchConditions, sq.Eq{"Posts.DeleteAt": 0}}
+		fetchConditions = sq.And{
+			fetchConditions,
+			sq.Eq{"COALESCE(Posts.DeleteAt, 0)": 0},
+		}
 	}
 
 	pageSize := uint64(30)
@@ -217,7 +220,10 @@ func (s *SqlThreadStore) GetThreadsForUser(userId, teamId string, opts model.Get
 		}
 		var threads []*JoinedThread
 		query, args, _ := s.getQueryBuilder().
-			Select("Threads.*, Posts.*, ThreadMemberships.LastViewed as LastViewedAt, ThreadMemberships.UnreadMentions as UnreadMentions").
+			Select(`Threads.*,
+				` + postSliceCoalesceQuery() + `,
+				ThreadMemberships.LastViewed as LastViewedAt,
+				ThreadMemberships.UnreadMentions as UnreadMentions`).
 			From("Threads").
 			Column(sq.Alias(sq.Expr(unreadRepliesQuery), "UnreadReplies")).
 			LeftJoin("Posts ON Posts.Id = Threads.PostId").
@@ -309,7 +315,7 @@ func (s *SqlThreadStore) GetThreadsForUser(userId, teamId string, opts model.Get
 			UnreadReplies:  thread.UnreadReplies,
 			UnreadMentions: thread.UnreadMentions,
 			Participants:   participants,
-			Post:           &thread.Post,
+			Post:           thread.Post.ToNilIfInvalid(),
 		})
 	}
 
@@ -401,7 +407,7 @@ func (s *SqlThreadStore) GetThreadForUser(teamId string, threadMembership *model
 		UnreadReplies:  thread.UnreadReplies,
 		UnreadMentions: thread.UnreadMentions,
 		Participants:   users,
-		Post:           &thread.Post,
+		Post:           thread.Post.ToNilIfInvalid(),
 	}
 
 	return result, nil
@@ -691,3 +697,86 @@ func (s *SqlThreadStore) GetPosts(threadId string, since int64) ([]*model.Post,
 	}
 	return result, nil
 }
+
+// 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) {
+	builder := s.getQueryBuilder().
+		Select("Threads.PostId").
+		From("Threads")
+	return genericPermanentDeleteBatchForRetentionPolicies(RetentionPolicyBatchDeletionInfo{
+		BaseBuilder:         builder,
+		Table:               "Threads",
+		TimeColumn:          "LastReplyAt",
+		PrimaryKeys:         []string{"PostId"},
+		ChannelIDTable:      "Threads",
+		NowMillis:           now,
+		GlobalPolicyEndTime: globalPolicyEndTime,
+		Limit:               limit,
+	}, s.SqlStore, cursor)
+}
+
+// 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) {
+	builder := s.getQueryBuilder().
+		Select("ThreadMemberships.PostId").
+		From("ThreadMemberships").
+		InnerJoin("Threads ON ThreadMemberships.PostId = Threads.PostId")
+	return genericPermanentDeleteBatchForRetentionPolicies(RetentionPolicyBatchDeletionInfo{
+		BaseBuilder:         builder,
+		Table:               "ThreadMemberships",
+		TimeColumn:          "LastUpdated",
+		PrimaryKeys:         []string{"PostId"},
+		ChannelIDTable:      "Threads",
+		NowMillis:           now,
+		GlobalPolicyEndTime: globalPolicyEndTime,
+		Limit:               limit,
+	}, s.SqlStore, cursor)
+}
+
+// DeleteOrphanedRows removes orphaned rows from Threads and ThreadMemberships
+func (s *SqlThreadStore) DeleteOrphanedRows(limit int) (deleted int64, err error) {
+	// We need the extra level of nesting to deal with MySQL's locking
+	const threadsQuery = `
+	DELETE FROM Threads WHERE PostId IN (
+		SELECT * FROM (
+			SELECT Threads.PostId FROM Threads
+			LEFT JOIN Channels ON Threads.ChannelId = Channels.Id
+			WHERE Channels.Id IS NULL
+			LIMIT :Limit
+		) AS A
+	)`
+	// We only delete a thread membership if the entire thread no longer exists,
+	// not if the root post has been deleted
+	const threadMembershipsQuery = `
+	DELETE FROM ThreadMemberships WHERE PostId IN (
+		SELECT * FROM (
+			SELECT ThreadMemberships.PostId FROM ThreadMemberships
+			LEFT JOIN Threads ON ThreadMemberships.PostId = Threads.PostId
+			WHERE Threads.PostId IS NULL
+			LIMIT :Limit
+		) AS A
+	)`
+	props := map[string]interface{}{"Limit": limit}
+	result, err := s.GetMaster().Exec(threadsQuery, props)
+	if err != nil {
+		return
+	}
+	rpcDeleted, err := result.RowsAffected()
+	if err != nil {
+		return
+	}
+	result, err = s.GetMaster().Exec(threadMembershipsQuery, props)
+	if err != nil {
+		return
+	}
+	rptDeleted, err := result.RowsAffected()
+	if err != nil {
+		return
+	}
+	deleted = rpcDeleted + rptDeleted
+	return
+}

store/store.go

@@ -90,6 +90,7 @@ type RetentionPolicyStore interface {
 	GetTeamsCount(policyId string) (int64, error)
 	AddTeams(policyId string, teamIds []string) error
 	RemoveTeams(policyId string, teamIds []string) error
+	DeleteOrphanedRows(limit int) (int64, error)
 	GetTeamPoliciesForUser(userID string, offset, limit int) ([]*model.RetentionPolicyForTeam, error)
 	GetTeamPoliciesCountForUser(userID string) (int64, error)
 	GetChannelPoliciesForUser(userID string, offset, limit int) ([]*model.RetentionPolicyForChannel, error)
@@ -273,6 +274,8 @@ type ChannelMemberHistoryStore interface {
 	LogJoinEvent(userID string, channelID string, joinTime int64) error
 	LogLeaveEvent(userID string, channelID string, leaveTime int64) error
 	GetUsersInChannelDuring(startTime int64, endTime int64, channelID string) ([]*model.ChannelMemberHistoryResult, error)
+	PermanentDeleteBatchForRetentionPolicies(now, globalPolicyEndTime, limit int64, cursor model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error)
+	DeleteOrphanedRows(limit int) (deleted int64, err error)
 	PermanentDeleteBatch(endTime int64, limit int64) (int64, error)
 }
 type ThreadStore interface {
@@ -299,6 +302,9 @@ type ThreadStore interface {
 	MaintainMembership(userID, postID string, opts ThreadMembershipOpts) (*model.ThreadMembership, error)
 	CollectThreadsWithNewerReplies(userId string, channelIds []string, timestamp int64) ([]string, error)
 	UpdateUnreadsByChannel(userId string, changedThreads []string, timestamp int64, updateViewedTimestamp bool) error
+	PermanentDeleteBatchForRetentionPolicies(now, globalPolicyEndTime, limit int64, cursor model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error)
+	PermanentDeleteBatchThreadMembershipsForRetentionPolicies(now, globalPolicyEndTime, limit int64, cursor model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error)
+	DeleteOrphanedRows(limit int) (deleted int64, err error)
 }
 
 type PostStore interface {
@@ -333,6 +339,8 @@ type PostStore interface {
 	OverwriteMultiple(posts []*model.Post) ([]*model.Post, int, error)
 	GetPostsByIds(postIds []string) ([]*model.Post, error)
 	GetPostsBatchForIndexing(startTime int64, endTime int64, limit int) ([]*model.PostForIndexing, error)
+	PermanentDeleteBatchForRetentionPolicies(now, globalPolicyEndTime, limit int64, cursor model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error)
+	DeleteOrphanedRows(limit int) (deleted int64, err error)
 	PermanentDeleteBatch(endTime int64, limit int64) (int64, error)
 	GetOldest() (*model.Post, error)
 	GetMaxPostSize() int
@@ -572,6 +580,7 @@ type PreferenceStore interface {
 	DeleteCategory(userID string, category string) error
 	DeleteCategoryAndName(category string, name string) error
 	PermanentDeleteByUser(userID string) error
+	DeleteOrphanedRows(limit int) (deleted int64, err error)
 	CleanupFlagsBatch(limit int64) (int64, error)
 }
 
@@ -646,8 +655,9 @@ type ReactionStore interface {
 	GetForPost(postID string, allowFromCache bool) ([]*model.Reaction, error)
 	GetForPostSince(postId string, since int64, excludeRemoteId string, inclDeleted bool) ([]*model.Reaction, error)
 	DeleteAllWithEmojiName(emojiName string) error
-	PermanentDeleteBatch(endTime int64, limit int64) (int64, error)
 	BulkGetForPosts(postIds []string) ([]*model.Reaction, error)
+	DeleteOrphanedRows(limit int) (int64, error)
+	PermanentDeleteBatch(endTime int64, limit int64) (int64, error)
 }
 
 type JobStore interface {

store/storetest/channel_member_history_store.go

@@ -21,6 +21,7 @@ func TestChannelMemberHistoryStore(t *testing.T, ss store.Store) {
 	t.Run("TestGetUsersInChannelAtChannelMemberHistory", func(t *testing.T) { testGetUsersInChannelAtChannelMemberHistory(t, ss) })
 	t.Run("TestGetUsersInChannelAtChannelMembers", func(t *testing.T) { testGetUsersInChannelAtChannelMembers(t, ss) })
 	t.Run("TestPermanentDeleteBatch", func(t *testing.T) { testPermanentDeleteBatch(t, ss) })
+	t.Run("TestPermanentDeleteBatchForRetentionPolicies", func(t *testing.T) { testPermanentDeleteBatchForRetentionPolicies(t, ss) })
 }
 
 func testLogJoinEvent(t *testing.T, ss store.Store) {
@@ -199,7 +200,8 @@ func testGetUsersInChannelAtChannelMembers(t *testing.T, ss store.Store) {
 	var tableDataTruncated = false
 	for !tableDataTruncated {
 		var count int64
-		count, err = ss.ChannelMemberHistory().PermanentDeleteBatch(model.GetMillis(), 1000)
+		count, _, err = ss.ChannelMemberHistory().PermanentDeleteBatchForRetentionPolicies(
+			0, model.GetMillis(), 1000, model.RetentionPolicyCursor{})
 		require.NoError(t, err, "Failed to truncate ChannelMemberHistory contents")
 		tableDataTruncated = count == int64(0)
 	}
@@ -333,7 +335,8 @@ func testPermanentDeleteBatch(t *testing.T, ss store.Store) {
 	assert.Len(t, channelMembers, 2)
 
 	// the permanent delete should delete at least one record
-	rowsDeleted, err := ss.ChannelMemberHistory().PermanentDeleteBatch(leaveTime, math.MaxInt64)
+	rowsDeleted, _, err := ss.ChannelMemberHistory().PermanentDeleteBatchForRetentionPolicies(
+		0, leaveTime+1, math.MaxInt64, model.RetentionPolicyCursor{})
 	require.NoError(t, err)
 	assert.NotEqual(t, int64(0), rowsDeleted)
 
@@ -343,3 +346,46 @@ func testPermanentDeleteBatch(t *testing.T, ss store.Store) {
 	assert.Len(t, channelMembers, 1)
 	assert.Equal(t, user2.Id, channelMembers[0].UserId)
 }
+
+func testPermanentDeleteBatchForRetentionPolicies(t *testing.T, ss store.Store) {
+	const limit = 1000
+	team, err := ss.Team().Save(&model.Team{
+		DisplayName: "DisplayName",
+		Name:        "team" + model.NewId(),
+		Email:       MakeEmail(),
+		Type:        model.TEAM_OPEN,
+	})
+	require.NoError(t, err)
+	channel, err := ss.Channel().Save(&model.Channel{
+		TeamId:      team.Id,
+		DisplayName: "DisplayName",
+		Name:        "channel" + model.NewId(),
+		Type:        model.CHANNEL_OPEN,
+	}, -1)
+	require.NoError(t, err)
+	userID := model.NewId()
+
+	joinTime := int64(1000)
+	leaveTime := int64(1500)
+	err = ss.ChannelMemberHistory().LogJoinEvent(userID, channel.Id, joinTime)
+	require.NoError(t, err)
+	err = ss.ChannelMemberHistory().LogLeaveEvent(userID, channel.Id, leaveTime)
+	require.NoError(t, err)
+
+	channelPolicy, err := ss.RetentionPolicy().Save(&model.RetentionPolicyWithTeamAndChannelIDs{
+		RetentionPolicy: model.RetentionPolicy{
+			DisplayName:  "DisplayName",
+			PostDuration: model.NewInt64(30),
+		},
+		ChannelIDs: []string{channel.Id},
+	})
+	require.NoError(t, err)
+
+	nowMillis := leaveTime + *channelPolicy.PostDuration*24*60*60*1000 + 1
+	_, _, err = ss.ChannelMemberHistory().PermanentDeleteBatchForRetentionPolicies(
+		nowMillis, 0, limit, model.RetentionPolicyCursor{})
+	require.NoError(t, err)
+	result, err := ss.ChannelMemberHistory().GetUsersInChannelDuring(joinTime, leaveTime, channel.Id)
+	require.NoError(t, err)
+	require.Empty(t, result, "history should have been deleted by channel policy")
+}

store/storetest/group_store.go

@@ -1757,7 +1757,8 @@ func testChannelMembersToAdd(t *testing.T, ss store.Store) {
 	require.Empty(t, channelMembers)
 
 	// Purging ChannelMemberHistory re-returns the result
-	_, nErr = ss.ChannelMemberHistory().PermanentDeleteBatch(model.GetMillis()+1, 100)
+	_, _, nErr = ss.ChannelMemberHistory().PermanentDeleteBatchForRetentionPolicies(
+		0, model.GetMillis()+1, 100, model.RetentionPolicyCursor{})
 	require.NoError(t, nErr)
 	channelMembers, err = ss.Group().ChannelMembersToAdd(0, nil, false)
 	require.NoError(t, err)

store/storetest/mocks/ChannelMemberHistoryStore.go

@@ -14,6 +14,27 @@ type ChannelMemberHistoryStore struct {
 	mock.Mock
 }
 
+// DeleteOrphanedRows provides a mock function with given fields: limit
+func (_m *ChannelMemberHistoryStore) DeleteOrphanedRows(limit int) (int64, error) {
+	ret := _m.Called(limit)
+
+	var r0 int64
+	if rf, ok := ret.Get(0).(func(int) int64); ok {
+		r0 = rf(limit)
+	} else {
+		r0 = ret.Get(0).(int64)
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(int) error); ok {
+		r1 = rf(limit)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
 // GetUsersInChannelDuring provides a mock function with given fields: startTime, endTime, channelID
 func (_m *ChannelMemberHistoryStore) GetUsersInChannelDuring(startTime int64, endTime int64, channelID string) ([]*model.ChannelMemberHistoryResult, error) {
 	ret := _m.Called(startTime, endTime, channelID)
@@ -85,3 +106,31 @@ func (_m *ChannelMemberHistoryStore) PermanentDeleteBatch(endTime int64, limit i
 
 	return r0, r1
 }
+
+// PermanentDeleteBatchForRetentionPolicies provides a mock function with given fields: now, globalPolicyEndTime, limit, cursor
+func (_m *ChannelMemberHistoryStore) PermanentDeleteBatchForRetentionPolicies(now int64, globalPolicyEndTime int64, limit int64, cursor model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error) {
+	ret := _m.Called(now, globalPolicyEndTime, limit, cursor)
+
+	var r0 int64
+	if rf, ok := ret.Get(0).(func(int64, int64, int64, model.RetentionPolicyCursor) int64); ok {
+		r0 = rf(now, globalPolicyEndTime, limit, cursor)
+	} else {
+		r0 = ret.Get(0).(int64)
+	}
+
+	var r1 model.RetentionPolicyCursor
+	if rf, ok := ret.Get(1).(func(int64, int64, int64, model.RetentionPolicyCursor) model.RetentionPolicyCursor); ok {
+		r1 = rf(now, globalPolicyEndTime, limit, cursor)
+	} else {
+		r1 = ret.Get(1).(model.RetentionPolicyCursor)
+	}
+
+	var r2 error
+	if rf, ok := ret.Get(2).(func(int64, int64, int64, model.RetentionPolicyCursor) error); ok {
+		r2 = rf(now, globalPolicyEndTime, limit, cursor)
+	} else {
+		r2 = ret.Error(2)
+	}
+
+	return r0, r1, r2
+}

store/storetest/mocks/PostStore.go

@@ -102,6 +102,27 @@ func (_m *PostStore) Delete(postID string, time int64, deleteByID string) error
 	return r0
 }
 
+// DeleteOrphanedRows provides a mock function with given fields: limit
+func (_m *PostStore) DeleteOrphanedRows(limit int) (int64, error) {
+	ret := _m.Called(limit)
+
+	var r0 int64
+	if rf, ok := ret.Get(0).(func(int) int64); ok {
+		r0 = rf(limit)
+	} else {
+		r0 = ret.Get(0).(int64)
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(int) error); ok {
+		r1 = rf(limit)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
 // Get provides a mock function with given fields: ctx, id, skipFetchThreads, collapsedThreads, collapsedThreadsExtended, userID
 func (_m *PostStore) Get(ctx context.Context, id string, skipFetchThreads bool, collapsedThreads bool, collapsedThreadsExtended bool, userID string) (*model.PostList, error) {
 	ret := _m.Called(ctx, id, skipFetchThreads, collapsedThreads, collapsedThreadsExtended, userID)
@@ -714,6 +735,34 @@ 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
+func (_m *PostStore) PermanentDeleteBatchForRetentionPolicies(now int64, globalPolicyEndTime int64, limit int64, cursor model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error) {
+	ret := _m.Called(now, globalPolicyEndTime, limit, cursor)
+
+	var r0 int64
+	if rf, ok := ret.Get(0).(func(int64, int64, int64, model.RetentionPolicyCursor) int64); ok {
+		r0 = rf(now, globalPolicyEndTime, limit, cursor)
+	} else {
+		r0 = ret.Get(0).(int64)
+	}
+
+	var r1 model.RetentionPolicyCursor
+	if rf, ok := ret.Get(1).(func(int64, int64, int64, model.RetentionPolicyCursor) model.RetentionPolicyCursor); ok {
+		r1 = rf(now, globalPolicyEndTime, limit, cursor)
+	} else {
+		r1 = ret.Get(1).(model.RetentionPolicyCursor)
+	}
+
+	var r2 error
+	if rf, ok := ret.Get(2).(func(int64, int64, int64, model.RetentionPolicyCursor) error); ok {
+		r2 = rf(now, globalPolicyEndTime, limit, cursor)
+	} else {
+		r2 = ret.Error(2)
+	}
+
+	return r0, r1, r2
+}
+
 // PermanentDeleteByChannel provides a mock function with given fields: channelID
 func (_m *PostStore) PermanentDeleteByChannel(channelID string) error {
 	ret := _m.Called(channelID)

store/storetest/mocks/PreferenceStore.go

@@ -77,6 +77,27 @@ func (_m *PreferenceStore) DeleteCategoryAndName(category string, name string) e
 	return r0
 }
 
+// DeleteOrphanedRows provides a mock function with given fields: limit
+func (_m *PreferenceStore) DeleteOrphanedRows(limit int) (int64, error) {
+	ret := _m.Called(limit)
+
+	var r0 int64
+	if rf, ok := ret.Get(0).(func(int) int64); ok {
+		r0 = rf(limit)
+	} else {
+		r0 = ret.Get(0).(int64)
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(int) error); ok {
+		r1 = rf(limit)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
 // Get provides a mock function with given fields: userID, category, name
 func (_m *PreferenceStore) Get(userID string, category string, name string) (*model.Preference, error) {
 	ret := _m.Called(userID, category, name)

store/storetest/mocks/ReactionStore.go

@@ -74,6 +74,27 @@ func (_m *ReactionStore) DeleteAllWithEmojiName(emojiName string) error {
 	return r0
 }
 
+// DeleteOrphanedRows provides a mock function with given fields: limit
+func (_m *ReactionStore) DeleteOrphanedRows(limit int) (int64, error) {
+	ret := _m.Called(limit)
+
+	var r0 int64
+	if rf, ok := ret.Get(0).(func(int) int64); ok {
+		r0 = rf(limit)
+	} else {
+		r0 = ret.Get(0).(int64)
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(int) error); ok {
+		r1 = rf(limit)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
 // GetForPost provides a mock function with given fields: postID, allowFromCache
 func (_m *ReactionStore) GetForPost(postID string, allowFromCache bool) ([]*model.Reaction, error) {
 	ret := _m.Called(postID, allowFromCache)

store/storetest/mocks/RetentionPolicyStore.go

@@ -56,6 +56,27 @@ func (_m *RetentionPolicyStore) Delete(id string) error {
 	return r0
 }
 
+// DeleteOrphanedRows provides a mock function with given fields: limit
+func (_m *RetentionPolicyStore) DeleteOrphanedRows(limit int) (int64, error) {
+	ret := _m.Called(limit)
+
+	var r0 int64
+	if rf, ok := ret.Get(0).(func(int) int64); ok {
+		r0 = rf(limit)
+	} else {
+		r0 = ret.Get(0).(int64)
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(int) error); ok {
+		r1 = rf(limit)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
 // Get provides a mock function with given fields: id
 func (_m *RetentionPolicyStore) Get(id string) (*model.RetentionPolicyWithTeamAndChannelCounts, error) {
 	ret := _m.Called(id)

store/storetest/mocks/ThreadStore.go

@@ -66,6 +66,27 @@ func (_m *ThreadStore) DeleteMembershipForUser(userId string, postID string) err
 	return r0
 }
 
+// DeleteOrphanedRows provides a mock function with given fields: limit
+func (_m *ThreadStore) DeleteOrphanedRows(limit int) (int64, error) {
+	ret := _m.Called(limit)
+
+	var r0 int64
+	if rf, ok := ret.Get(0).(func(int) int64); ok {
+		r0 = rf(limit)
+	} else {
+		r0 = ret.Get(0).(int64)
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(int) error); ok {
+		r1 = rf(limit)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
 // Get provides a mock function with given fields: id
 func (_m *ThreadStore) Get(id string) (*model.Thread, error) {
 	ret := _m.Called(id)
@@ -292,6 +313,62 @@ func (_m *ThreadStore) MarkAsRead(userID string, threadID string, timestamp int6
 	return r0
 }
 
+// PermanentDeleteBatchForRetentionPolicies provides a mock function with given fields: now, globalPolicyEndTime, limit, cursor
+func (_m *ThreadStore) PermanentDeleteBatchForRetentionPolicies(now int64, globalPolicyEndTime int64, limit int64, cursor model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error) {
+	ret := _m.Called(now, globalPolicyEndTime, limit, cursor)
+
+	var r0 int64
+	if rf, ok := ret.Get(0).(func(int64, int64, int64, model.RetentionPolicyCursor) int64); ok {
+		r0 = rf(now, globalPolicyEndTime, limit, cursor)
+	} else {
+		r0 = ret.Get(0).(int64)
+	}
+
+	var r1 model.RetentionPolicyCursor
+	if rf, ok := ret.Get(1).(func(int64, int64, int64, model.RetentionPolicyCursor) model.RetentionPolicyCursor); ok {
+		r1 = rf(now, globalPolicyEndTime, limit, cursor)
+	} else {
+		r1 = ret.Get(1).(model.RetentionPolicyCursor)
+	}
+
+	var r2 error
+	if rf, ok := ret.Get(2).(func(int64, int64, int64, model.RetentionPolicyCursor) error); ok {
+		r2 = rf(now, globalPolicyEndTime, limit, cursor)
+	} else {
+		r2 = ret.Error(2)
+	}
+
+	return r0, r1, r2
+}
+
+// PermanentDeleteBatchThreadMembershipsForRetentionPolicies provides a mock function with given fields: now, globalPolicyEndTime, limit, cursor
+func (_m *ThreadStore) PermanentDeleteBatchThreadMembershipsForRetentionPolicies(now int64, globalPolicyEndTime int64, limit int64, cursor model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error) {
+	ret := _m.Called(now, globalPolicyEndTime, limit, cursor)
+
+	var r0 int64
+	if rf, ok := ret.Get(0).(func(int64, int64, int64, model.RetentionPolicyCursor) int64); ok {
+		r0 = rf(now, globalPolicyEndTime, limit, cursor)
+	} else {
+		r0 = ret.Get(0).(int64)
+	}
+
+	var r1 model.RetentionPolicyCursor
+	if rf, ok := ret.Get(1).(func(int64, int64, int64, model.RetentionPolicyCursor) model.RetentionPolicyCursor); ok {
+		r1 = rf(now, globalPolicyEndTime, limit, cursor)
+	} else {
+		r1 = ret.Get(1).(model.RetentionPolicyCursor)
+	}
+
+	var r2 error
+	if rf, ok := ret.Get(2).(func(int64, int64, int64, model.RetentionPolicyCursor) error); ok {
+		r2 = rf(now, globalPolicyEndTime, limit, cursor)
+	} else {
+		r2 = ret.Error(2)
+	}
+
+	return r0, r1, r2
+}
+
 // Save provides a mock function with given fields: thread
 func (_m *ThreadStore) Save(thread *model.Thread) (*model.Thread, error) {
 	ret := _m.Called(thread)

store/storetest/post_store.go

@@ -2586,16 +2586,31 @@ func testPostStoreGetPostsBatchForIndexing(t *testing.T, ss store.Store) {
 }
 
 func testPostStorePermanentDeleteBatch(t *testing.T, ss store.Store) {
+	team, err := ss.Team().Save(&model.Team{
+		DisplayName: "DisplayName",
+		Name:        "team" + model.NewId(),
+		Email:       MakeEmail(),
+		Type:        model.TEAM_OPEN,
+	})
+	require.NoError(t, err)
+	channel, err := ss.Channel().Save(&model.Channel{
+		TeamId:      team.Id,
+		DisplayName: "DisplayName",
+		Name:        "channel" + model.NewId(),
+		Type:        model.CHANNEL_OPEN,
+	}, -1)
+	require.NoError(t, err)
+
 	o1 := &model.Post{}
-	o1.ChannelId = model.NewId()
+	o1.ChannelId = channel.Id
 	o1.UserId = model.NewId()
 	o1.Message = "zz" + model.NewId() + "AAAAAAAAAAA"
 	o1.CreateAt = 1000
-	o1, err := ss.Post().Save(o1)
+	o1, err = ss.Post().Save(o1)
 	require.NoError(t, err)
 
 	o2 := &model.Post{}
-	o2.ChannelId = model.NewId()
+	o2.ChannelId = channel.Id
 	o2.UserId = model.NewId()
 	o2.Message = "zz" + model.NewId() + "AAAAAAAAAAA"
 	o2.CreateAt = 1000
@@ -2603,14 +2618,14 @@ func testPostStorePermanentDeleteBatch(t *testing.T, ss store.Store) {
 	require.NoError(t, err)
 
 	o3 := &model.Post{}
-	o3.ChannelId = model.NewId()
+	o3.ChannelId = channel.Id
 	o3.UserId = model.NewId()
 	o3.Message = "zz" + model.NewId() + "AAAAAAAAAAA"
 	o3.CreateAt = 100000
 	o3, err = ss.Post().Save(o3)
 	require.NoError(t, err)
 
-	_, err = ss.Post().PermanentDeleteBatch(2000, 1000)
+	_, _, err = ss.Post().PermanentDeleteBatchForRetentionPolicies(0, 2000, 1000, model.RetentionPolicyCursor{})
 	require.NoError(t, err)
 
 	_, err = ss.Post().Get(context.Background(), o1.Id, false, false, false, "")
@@ -2620,7 +2635,160 @@ func testPostStorePermanentDeleteBatch(t *testing.T, ss store.Store) {
 	require.Error(t, err, "Should have not found post 2 after purge")
 
 	_, err = ss.Post().Get(context.Background(), o3.Id, false, false, false, "")
-	require.NoError(t, err, "Should have not found post 3 after purge")
+	require.NoError(t, err, "Should have found post 3 after purge")
+
+	t.Run("with pagination", func(t *testing.T) {
+		for i := 0; i < 3; i++ {
+			_, err = ss.Post().Save(&model.Post{
+				ChannelId: channel.Id,
+				UserId:    model.NewId(),
+				Message:   "message",
+				CreateAt:  1,
+			})
+			require.NoError(t, err)
+		}
+		cursor := model.RetentionPolicyCursor{}
+
+		deleted, cursor, err := ss.Post().PermanentDeleteBatchForRetentionPolicies(0, 2, 2, cursor)
+		require.NoError(t, err)
+		require.Equal(t, int64(2), deleted)
+
+		deleted, _, err = ss.Post().PermanentDeleteBatchForRetentionPolicies(0, 2, 2, cursor)
+		require.NoError(t, err)
+		require.Equal(t, int64(1), deleted)
+	})
+
+	t.Run("with data retention policies", func(t *testing.T) {
+		channelPolicy, err2 := ss.RetentionPolicy().Save(&model.RetentionPolicyWithTeamAndChannelIDs{
+			RetentionPolicy: model.RetentionPolicy{
+				DisplayName:  "DisplayName",
+				PostDuration: model.NewInt64(30),
+			},
+			ChannelIDs: []string{channel.Id},
+		})
+		require.NoError(t, err2)
+		post := &model.Post{
+			ChannelId: channel.Id,
+			UserId:    model.NewId(),
+			Message:   "message",
+			CreateAt:  1,
+		}
+		post, err2 = ss.Post().Save(post)
+		require.NoError(t, err2)
+
+		_, _, err2 = ss.Post().PermanentDeleteBatchForRetentionPolicies(0, 2000, 1000, model.RetentionPolicyCursor{})
+		require.NoError(t, err2)
+		_, err2 = ss.Post().Get(context.Background(), post.Id, false, false, false, "")
+		require.NoError(t, err2, "global policy should have been ignored due to granular policy")
+
+		nowMillis := post.CreateAt + *channelPolicy.PostDuration*24*60*60*1000 + 1
+		_, _, err2 = ss.Post().PermanentDeleteBatchForRetentionPolicies(nowMillis, 0, 1000, model.RetentionPolicyCursor{})
+		require.NoError(t, err2)
+		_, err2 = ss.Post().Get(context.Background(), post.Id, false, false, false, "")
+		require.Error(t, err2, "post should have been deleted by channel policy")
+
+		// Create a team policy which is stricter than the channel policy
+		teamPolicy, err2 := ss.RetentionPolicy().Save(&model.RetentionPolicyWithTeamAndChannelIDs{
+			RetentionPolicy: model.RetentionPolicy{
+				DisplayName:  "DisplayName",
+				PostDuration: model.NewInt64(20),
+			},
+			TeamIDs: []string{team.Id},
+		})
+		require.NoError(t, err2)
+		post.Id = ""
+		post, err2 = ss.Post().Save(post)
+		require.NoError(t, err2)
+
+		nowMillis = post.CreateAt + *teamPolicy.PostDuration*24*60*60*1000 + 1
+		_, _, err2 = ss.Post().PermanentDeleteBatchForRetentionPolicies(nowMillis, 0, 1000, model.RetentionPolicyCursor{})
+		require.NoError(t, err2)
+		_, err2 = ss.Post().Get(context.Background(), post.Id, false, false, false, "")
+		require.NoError(t, err2, "channel policy should have overridden team policy")
+
+		// Delete channel policy and re-run team policy
+		err2 = ss.RetentionPolicy().RemoveChannels(channelPolicy.ID, []string{channel.Id})
+		require.NoError(t, err2)
+
+		err2 = ss.RetentionPolicy().Delete(channelPolicy.ID)
+		require.NoError(t, err2)
+
+		_, _, err2 = ss.Post().PermanentDeleteBatchForRetentionPolicies(nowMillis, 0, 1000, model.RetentionPolicyCursor{})
+		require.NoError(t, err2)
+		_, err2 = ss.Post().Get(context.Background(), post.Id, false, false, false, "")
+		require.Error(t, err2, "post should have been deleted by team policy")
+
+		err2 = ss.RetentionPolicy().RemoveTeams(teamPolicy.ID, []string{team.Id})
+		require.NoError(t, err2)
+
+		err2 = ss.RetentionPolicy().Delete(teamPolicy.ID)
+		require.NoError(t, err2)
+	})
+
+	t.Run("with channel, team and global policies", func(t *testing.T) {
+		c1 := &model.Channel{}
+		c1.TeamId = model.NewId()
+		c1.DisplayName = "Channel1"
+		c1.Name = "zz" + model.NewId() + "b"
+		c1.Type = model.CHANNEL_OPEN
+		c1, _ = ss.Channel().Save(c1, -1)
+
+		c2 := &model.Channel{}
+		c2.TeamId = model.NewId()
+		c2.DisplayName = "Channel2"
+		c2.Name = "zz" + model.NewId() + "b"
+		c2.Type = model.CHANNEL_OPEN
+		c2, _ = ss.Channel().Save(c2, -1)
+
+		channelPolicy, err2 := ss.RetentionPolicy().Save(&model.RetentionPolicyWithTeamAndChannelIDs{
+			RetentionPolicy: model.RetentionPolicy{
+				DisplayName:  "DisplayName",
+				PostDuration: model.NewInt64(30),
+			},
+			ChannelIDs: []string{c1.Id},
+		})
+		require.NoError(t, err2)
+		defer ss.RetentionPolicy().Delete(channelPolicy.ID)
+		teamPolicy, err2 := ss.RetentionPolicy().Save(&model.RetentionPolicyWithTeamAndChannelIDs{
+			RetentionPolicy: model.RetentionPolicy{
+				DisplayName:  "DisplayName",
+				PostDuration: model.NewInt64(30),
+			},
+			TeamIDs: []string{team.Id},
+		})
+		require.NoError(t, err2)
+		defer ss.RetentionPolicy().Delete(teamPolicy.ID)
+
+		// This one should be deleted by the channel policy
+		_, err2 = ss.Post().Save(&model.Post{
+			ChannelId: c1.Id,
+			UserId:    model.NewId(),
+			Message:   "message",
+			CreateAt:  1,
+		})
+		require.NoError(t, err2)
+		// This one, by the team policy
+		_, err2 = ss.Post().Save(&model.Post{
+			ChannelId: channel.Id,
+			UserId:    model.NewId(),
+			Message:   "message",
+			CreateAt:  1,
+		})
+		require.NoError(t, err2)
+		// This one, by the global policy
+		_, err2 = ss.Post().Save(&model.Post{
+			ChannelId: c2.Id,
+			UserId:    model.NewId(),
+			Message:   "message",
+			CreateAt:  1,
+		})
+		require.NoError(t, err2)
+
+		nowMillis := int64(1 + 30*24*60*60*1000 + 1)
+		deleted, _, err2 := ss.Post().PermanentDeleteBatchForRetentionPolicies(nowMillis, 2, 1000, model.RetentionPolicyCursor{})
+		require.NoError(t, err2)
+		require.Equal(t, int64(3), deleted)
+	})
 }
 
 func testPostStoreGetOldest(t *testing.T, ss store.Store) {

store/storetest/preference_store.go

@@ -22,7 +22,7 @@ func TestPreferenceStore(t *testing.T, ss store.Store) {
 	t.Run("PreferenceDelete", func(t *testing.T) { testPreferenceDelete(t, ss) })
 	t.Run("PreferenceDeleteCategory", func(t *testing.T) { testPreferenceDeleteCategory(t, ss) })
 	t.Run("PreferenceDeleteCategoryAndName", func(t *testing.T) { testPreferenceDeleteCategoryAndName(t, ss) })
-	t.Run("PreferenceCleanupFlagsBatch", func(t *testing.T) { testPreferenceCleanupFlagsBatch(t, ss) })
+	t.Run("PreferenceDeleteOrphanedRows", func(t *testing.T) { testPreferenceDeleteOrphanedRows(t, ss) })
 }
 
 func testPreferenceSave(t *testing.T, ss store.Store) {
@@ -330,44 +330,66 @@ func testPreferenceDeleteCategoryAndName(t *testing.T, ss store.Store) {
 	assert.Empty(t, preferences, "should've returned no preference")
 }
 
-func testPreferenceCleanupFlagsBatch(t *testing.T, ss store.Store) {
+func testPreferenceDeleteOrphanedRows(t *testing.T, ss store.Store) {
+	const limit = 1000
+	team, err := ss.Team().Save(&model.Team{
+		DisplayName: "DisplayName",
+		Name:        "team" + model.NewId(),
+		Email:       MakeEmail(),
+		Type:        model.TEAM_OPEN,
+	})
+	require.NoError(t, err)
+	channel, err := ss.Channel().Save(&model.Channel{
+		TeamId:      team.Id,
+		DisplayName: "DisplayName",
+		Name:        "channel" + model.NewId(),
+		Type:        model.CHANNEL_OPEN,
+	}, -1)
+	require.NoError(t, err)
 	category := model.PREFERENCE_CATEGORY_FLAGGED_POST
 	userId := model.NewId()
 
-	o1 := &model.Post{}
-	o1.ChannelId = model.NewId()
-	o1.UserId = userId
-	o1.Message = "zz" + model.NewId() + "AAAAAAAAAAA"
-	o1.CreateAt = 1000
-	o1, err := ss.Post().Save(o1)
+	olderPost, err := ss.Post().Save(&model.Post{
+		ChannelId: channel.Id,
+		UserId:    userId,
+		Message:   "message",
+		CreateAt:  1000,
+	})
+	require.NoError(t, err)
+	newerPost, err := ss.Post().Save(&model.Post{
+		ChannelId: channel.Id,
+		UserId:    userId,
+		Message:   "message",
+		CreateAt:  3000,
+	})
 	require.NoError(t, err)
 
 	preference1 := model.Preference{
 		UserId:   userId,
 		Category: category,
-		Name:     o1.Id,
+		Name:     olderPost.Id,
 		Value:    "true",
 	}
 
 	preference2 := model.Preference{
 		UserId:   userId,
 		Category: category,
-		Name:     model.NewId(),
+		Name:     newerPost.Id,
 		Value:    "true",
 	}
 
 	nErr := ss.Preference().Save(&model.Preferences{preference1, preference2})
 	require.NoError(t, nErr)
 
-	_, nErr = ss.Preference().CleanupFlagsBatch(-1)
-	require.Error(t, nErr)
+	_, _, nErr = ss.Post().PermanentDeleteBatchForRetentionPolicies(0, 2000, limit, model.RetentionPolicyCursor{})
+	assert.NoError(t, nErr)
 
-	_, nErr = ss.Preference().CleanupFlagsBatch(10000)
+	_, nErr = ss.Preference().DeleteOrphanedRows(limit)
 	assert.NoError(t, nErr)
 
 	_, nErr = ss.Preference().Get(userId, category, preference1.Name)
-	assert.NoError(t, nErr)
+	assert.Error(t, nErr, "older preference should have been deleted")
 
 	_, nErr = ss.Preference().Get(userId, category, preference2.Name)
-	assert.Error(t, nErr)
+	assert.NoError(t, nErr, "newer preference should not have been deleted")
 }

store/storetest/reaction_store.go

@@ -522,61 +522,72 @@ func testReactionDeleteAllWithEmojiName(t *testing.T, ss store.Store, s SqlStore
 }
 
 func testReactionStorePermanentDeleteBatch(t *testing.T, ss store.Store) {
-	post, err1 := ss.Post().Save(&model.Post{
-		ChannelId: model.NewId(),
-		UserId:    model.NewId(),
+	const limit = 1000
+	team, err := ss.Team().Save(&model.Team{
+		DisplayName: "DisplayName",
+		Name:        "team" + model.NewId(),
+		Email:       MakeEmail(),
+		Type:        model.TEAM_OPEN,
 	})
-	require.NoError(t, err1)
+	require.NoError(t, err)
+	channel, err := ss.Channel().Save(&model.Channel{
+		TeamId:      team.Id,
+		DisplayName: "DisplayName",
+		Name:        "channel" + model.NewId(),
+		Type:        model.CHANNEL_OPEN,
+	}, -1)
+	require.NoError(t, err)
+	olderPost, err := ss.Post().Save(&model.Post{
+		ChannelId: channel.Id,
+		UserId:    model.NewId(),
+		CreateAt:  1000,
+	})
+	require.NoError(t, err)
+	newerPost, err := ss.Post().Save(&model.Post{
+		ChannelId: channel.Id,
+		UserId:    model.NewId(),
+		CreateAt:  3000,
+	})
+	require.NoError(t, err)
 
+	// Reactions will be deleted based on the timestamp of their post. So the time at
+	// which a reaction was created doesn't matter.
 	reactions := []*model.Reaction{
 		{
 			UserId:    model.NewId(),
-			PostId:    post.Id,
+			PostId:    olderPost.Id,
 			EmojiName: "sad",
-			CreateAt:  1000,
 		},
 		{
 			UserId:    model.NewId(),
-			PostId:    post.Id,
+			PostId:    olderPost.Id,
 			EmojiName: "sad",
-			CreateAt:  1500,
 		},
 		{
 			UserId:    model.NewId(),
-			PostId:    post.Id,
-			EmojiName: "sad",
-			CreateAt:  2000,
-		},
-		{
-			UserId:    model.NewId(),
-			PostId:    post.Id,
-			EmojiName: "sad",
-			CreateAt:  2000,
+			PostId:    newerPost.Id,
+			EmojiName: "smile",
 		},
 	}
 
-	// Need to hang on to a reaction to delete later in order to clear the cache, as "allowFromCache" isn't honoured any more.
-	var lastReaction *model.Reaction
 	for _, reaction := range reactions {
-		var nErr error
-		lastReaction, nErr = ss.Reaction().Save(reaction)
-		require.NoError(t, nErr)
+		_, err = ss.Reaction().Save(reaction)
+		require.NoError(t, err)
 	}
 
-	returned, err := ss.Reaction().GetForPost(post.Id, false)
-	require.NoError(t, err)
-	require.Len(t, returned, 4, "expected 4 reactions")
-
-	_, err = ss.Reaction().PermanentDeleteBatch(1800, 1000)
+	_, _, err = ss.Post().PermanentDeleteBatchForRetentionPolicies(0, 2000, limit, model.RetentionPolicyCursor{})
 	require.NoError(t, err)
 
-	// This is to force a clear of the cache.
-	_, err = ss.Reaction().Delete(lastReaction)
+	_, err = ss.Reaction().DeleteOrphanedRows(limit)
 	require.NoError(t, err)
 
-	returned, err = ss.Reaction().GetForPost(post.Id, false)
+	returned, err := ss.Reaction().GetForPost(olderPost.Id, false)
 	require.NoError(t, err)
-	require.Len(t, returned, 1, "expected 1 reaction. Got: %v", len(returned))
+	require.Len(t, returned, 0, "reactions for older post should have been deleted")
+
+	returned, err = ss.Reaction().GetForPost(newerPost.Id, false)
+	require.NoError(t, err)
+	require.Len(t, returned, 1, "reactions for newer post should not have been deleted")
 }
 
 func testReactionBulkGetForPosts(t *testing.T, ss store.Store) {

store/storetest/retention_policy_store.go

@@ -25,6 +25,7 @@ func TestRetentionPolicyStore(t *testing.T, ss store.Store, s SqlStore) {
 	t.Run("GetTeams", func(t *testing.T) { testRetentionPolicyStoreGetTeams(t, ss, s) })
 	t.Run("AddTeams", func(t *testing.T) { testRetentionPolicyStoreAddTeams(t, ss, s) })
 	t.Run("RemoveTeams", func(t *testing.T) { testRetentionPolicyStoreRemoveTeams(t, ss, s) })
+	t.Run("RemoveOrphanedRows", func(t *testing.T) { testRetentionPolicyStoreRemoveOrphanedRows(t, ss, s) })
 	t.Run("GetPoliciesForUser", func(t *testing.T) { testRetentionPolicyStoreGetPoliciesForUser(t, ss, s) })
 }
 
@@ -656,3 +657,21 @@ func testRetentionPolicyStoreGetPoliciesForUser(t *testing.T, ss store.Store, s
 		require.Equal(t, int64(len(channelIDs)), count)
 	})
 }
+
+func testRetentionPolicyStoreRemoveOrphanedRows(t *testing.T, ss store.Store, s SqlStore) {
+	teamID := createTeamsForRetentionPolicy(t, ss, 1)[0]
+	channelID := createChannelsForRetentionPolicy(t, ss, teamID, 1)[0]
+	policy := saveRetentionPolicyWithTeamAndChannelIds(t, ss, "Policy 1",
+		[]string{teamID}, []string{channelID})
+
+	err := ss.Channel().PermanentDelete(channelID)
+	require.NoError(t, err)
+	err = ss.Team().PermanentDelete(teamID)
+	require.NoError(t, err)
+	_, err = ss.RetentionPolicy().DeleteOrphanedRows(1000)
+	require.NoError(t, err)
+
+	policy.ChannelIDs = make([]string, 0)
+	policy.TeamIDs = make([]string, 0)
+	checkRetentionPolicyLikeThisExists(t, ss, policy)
+}

store/storetest/thread_store.go

@@ -17,6 +17,12 @@ import (
 
 func TestThreadStore(t *testing.T, ss store.Store, s SqlStore) {
 	t.Run("ThreadStorePopulation", func(t *testing.T) { testThreadStorePopulation(t, ss) })
+	t.Run("ThreadStorePermanentDeleteBatchForRetentionPolicies", func(t *testing.T) {
+		testThreadStorePermanentDeleteBatchForRetentionPolicies(t, ss)
+	})
+	t.Run("ThreadStorePermanentDeleteBatchThreadMembershipsForRetentionPolicies", func(t *testing.T) {
+		testThreadStorePermanentDeleteBatchThreadMembershipsForRetentionPolicies(t, ss)
+	})
 }
 
 func testThreadStorePopulation(t *testing.T, ss store.Store) {
@@ -411,3 +417,185 @@ func testThreadStorePopulation(t *testing.T, ss store.Store) {
 		require.Equal(t, int64(0), th.UnreadReplies)
 	})
 }
+
+func threadStoreCreateReply(t *testing.T, ss store.Store, channelID, postID string, createAt int64) *model.Post {
+	reply, err := ss.Post().Save(&model.Post{
+		ChannelId: channelID,
+		UserId:    model.NewId(),
+		CreateAt:  createAt,
+		RootId:    postID,
+		ParentId:  postID,
+	})
+	require.NoError(t, err)
+	return reply
+}
+
+func testThreadStorePermanentDeleteBatchForRetentionPolicies(t *testing.T, ss store.Store) {
+	const limit = 1000
+	team, err := ss.Team().Save(&model.Team{
+		DisplayName: "DisplayName",
+		Name:        "team" + model.NewId(),
+		Email:       MakeEmail(),
+		Type:        model.TEAM_OPEN,
+	})
+	require.NoError(t, err)
+	channel, err := ss.Channel().Save(&model.Channel{
+		TeamId:      team.Id,
+		DisplayName: "DisplayName",
+		Name:        "channel" + model.NewId(),
+		Type:        model.CHANNEL_OPEN,
+	}, -1)
+	require.NoError(t, err)
+
+	post, err := ss.Post().Save(&model.Post{
+		ChannelId: channel.Id,
+		UserId:    model.NewId(),
+	})
+	require.NoError(t, err)
+	threadStoreCreateReply(t, ss, channel.Id, post.Id, 2000)
+
+	thread, err := ss.Thread().Get(post.Id)
+	require.NoError(t, err)
+
+	channelPolicy, err := ss.RetentionPolicy().Save(&model.RetentionPolicyWithTeamAndChannelIDs{
+		RetentionPolicy: model.RetentionPolicy{
+			DisplayName:  "DisplayName",
+			PostDuration: model.NewInt64(30),
+		},
+		ChannelIDs: []string{channel.Id},
+	})
+	require.NoError(t, err)
+
+	nowMillis := thread.LastReplyAt + *channelPolicy.PostDuration*24*60*60*1000 + 1
+	_, _, err = ss.Thread().PermanentDeleteBatchForRetentionPolicies(nowMillis, 0, limit, model.RetentionPolicyCursor{})
+	require.NoError(t, err)
+	_, err = ss.Thread().Get(post.Id)
+	require.Error(t, err, "thread should have been deleted by channel policy")
+
+	// create a new thread
+	threadStoreCreateReply(t, ss, channel.Id, post.Id, 2000)
+	thread, err = ss.Thread().Get(post.Id)
+	require.NoError(t, err)
+
+	// Create a team policy which is stricter than the channel policy
+	teamPolicy, err := ss.RetentionPolicy().Save(&model.RetentionPolicyWithTeamAndChannelIDs{
+		RetentionPolicy: model.RetentionPolicy{
+			DisplayName:  "DisplayName",
+			PostDuration: model.NewInt64(20),
+		},
+		TeamIDs: []string{team.Id},
+	})
+	require.NoError(t, err)
+
+	nowMillis = thread.LastReplyAt + *teamPolicy.PostDuration*24*60*60*1000 + 1
+	_, _, err = ss.Thread().PermanentDeleteBatchForRetentionPolicies(nowMillis, 0, limit, model.RetentionPolicyCursor{})
+	require.NoError(t, err)
+	_, err = ss.Thread().Get(post.Id)
+	require.NoError(t, err, "channel policy should have overridden team policy")
+
+	// 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{})
+	require.NoError(t, err)
+	_, err = ss.Thread().Get(post.Id)
+	require.Error(t, err, "thread should have been deleted by team policy")
+}
+
+func testThreadStorePermanentDeleteBatchThreadMembershipsForRetentionPolicies(t *testing.T, ss store.Store) {
+	const limit = 1000
+	userID := model.NewId()
+	createThreadMembership := func(userID, postID string) *model.ThreadMembership {
+		opts := store.ThreadMembershipOpts{
+			Following:             true,
+			IncrementMentions:     false,
+			UpdateFollowing:       true,
+			UpdateViewedTimestamp: false,
+			UpdateParticipants:    false,
+		}
+		_, err := ss.Thread().MaintainMembership(userID, postID, opts)
+		require.NoError(t, err)
+		threadMembership, err := ss.Thread().GetMembershipForUser(userID, postID)
+		require.NoError(t, err)
+		return threadMembership
+	}
+	team, err := ss.Team().Save(&model.Team{
+		DisplayName: "DisplayName",
+		Name:        "team" + model.NewId(),
+		Email:       MakeEmail(),
+		Type:        model.TEAM_OPEN,
+	})
+	require.NoError(t, err)
+	channel, err := ss.Channel().Save(&model.Channel{
+		TeamId:      team.Id,
+		DisplayName: "DisplayName",
+		Name:        "channel" + model.NewId(),
+		Type:        model.CHANNEL_OPEN,
+	}, -1)
+	require.NoError(t, err)
+	post, err := ss.Post().Save(&model.Post{
+		ChannelId: channel.Id,
+		UserId:    model.NewId(),
+	})
+	require.NoError(t, err)
+	threadStoreCreateReply(t, ss, channel.Id, post.Id, 2000)
+
+	threadMembership := createThreadMembership(userID, post.Id)
+
+	channelPolicy, err := ss.RetentionPolicy().Save(&model.RetentionPolicyWithTeamAndChannelIDs{
+		RetentionPolicy: model.RetentionPolicy{
+			DisplayName:  "DisplayName",
+			PostDuration: model.NewInt64(30),
+		},
+		ChannelIDs: []string{channel.Id},
+	})
+	require.NoError(t, err)
+
+	nowMillis := threadMembership.LastUpdated + *channelPolicy.PostDuration*24*60*60*1000 + 1
+	_, _, err = ss.Thread().PermanentDeleteBatchThreadMembershipsForRetentionPolicies(nowMillis, 0, 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")
+
+	// create a new thread membership
+	threadMembership = createThreadMembership(userID, post.Id)
+
+	// Create a team policy which is stricter than the channel policy
+	teamPolicy, err := ss.RetentionPolicy().Save(&model.RetentionPolicyWithTeamAndChannelIDs{
+		RetentionPolicy: model.RetentionPolicy{
+			DisplayName:  "DisplayName",
+			PostDuration: model.NewInt64(20),
+		},
+		TeamIDs: []string{team.Id},
+	})
+	require.NoError(t, err)
+
+	nowMillis = threadMembership.LastUpdated + *teamPolicy.PostDuration*24*60*60*1000 + 1
+	_, _, err = ss.Thread().PermanentDeleteBatchThreadMembershipsForRetentionPolicies(nowMillis, 0, 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")
+
+	// 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{})
+	require.NoError(t, err)
+	_, err = ss.Thread().GetMembershipForUser(userID, post.Id)
+	require.Error(t, err, "thread membership should have been deleted by team policy")
+
+	// create a new thread membership
+	threadMembership = createThreadMembership(userID, post.Id)
+
+	// Delete team policy and thread
+	err = ss.RetentionPolicy().Delete(teamPolicy.ID)
+	require.NoError(t, err)
+	err = ss.Thread().Delete(post.Id)
+	require.NoError(t, err)
+
+	deleted, err := ss.Thread().DeleteOrphanedRows(1000)
+	require.NoError(t, err)
+	require.NotZero(t, deleted)
+	_, err = ss.Thread().GetMembershipForUser(userID, post.Id)
+	require.Error(t, err, "thread membership should have been deleted because thread no longer exists")
+}

store/timerlayer/timerlayer.go

@@ -2158,6 +2158,22 @@ func (s *TimerLayerChannelStore) UserBelongsToChannels(userID string, channelIds
 	return result, err
 }
 
+func (s *TimerLayerChannelMemberHistoryStore) DeleteOrphanedRows(limit int) (int64, error) {
+	start := timemodule.Now()
+
+	result, err := s.ChannelMemberHistoryStore.DeleteOrphanedRows(limit)
+
+	elapsed := float64(timemodule.Since(start)) / float64(timemodule.Second)
+	if s.Root.Metrics != nil {
+		success := "false"
+		if err == nil {
+			success = "true"
+		}
+		s.Root.Metrics.ObserveStoreMethodDuration("ChannelMemberHistoryStore.DeleteOrphanedRows", success, elapsed)
+	}
+	return result, err
+}
+
 func (s *TimerLayerChannelMemberHistoryStore) GetUsersInChannelDuring(startTime int64, endTime int64, channelID string) ([]*model.ChannelMemberHistoryResult, error) {
 	start := timemodule.Now()
 
@@ -2222,6 +2238,22 @@ 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) {
+	start := timemodule.Now()
+
+	result, resultVar1, err := s.ChannelMemberHistoryStore.PermanentDeleteBatchForRetentionPolicies(now, globalPolicyEndTime, limit, cursor)
+
+	elapsed := float64(timemodule.Since(start)) / float64(timemodule.Second)
+	if s.Root.Metrics != nil {
+		success := "false"
+		if err == nil {
+			success = "true"
+		}
+		s.Root.Metrics.ObserveStoreMethodDuration("ChannelMemberHistoryStore.PermanentDeleteBatchForRetentionPolicies", success, elapsed)
+	}
+	return result, resultVar1, err
+}
+
 func (s *TimerLayerClusterDiscoveryStore) Cleanup() error {
 	start := timemodule.Now()
 
@@ -4586,6 +4618,22 @@ func (s *TimerLayerPostStore) Delete(postID string, time int64, deleteByID strin
 	return err
 }
 
+func (s *TimerLayerPostStore) DeleteOrphanedRows(limit int) (int64, error) {
+	start := timemodule.Now()
+
+	result, err := s.PostStore.DeleteOrphanedRows(limit)
+
+	elapsed := float64(timemodule.Since(start)) / float64(timemodule.Second)
+	if s.Root.Metrics != nil {
+		success := "false"
+		if err == nil {
+			success = "true"
+		}
+		s.Root.Metrics.ObserveStoreMethodDuration("PostStore.DeleteOrphanedRows", success, elapsed)
+	}
+	return result, err
+}
+
 func (s *TimerLayerPostStore) Get(ctx context.Context, id string, skipFetchThreads bool, collapsedThreads bool, collapsedThreadsExtended bool, userID string) (*model.PostList, error) {
 	start := timemodule.Now()
 
@@ -5033,6 +5081,22 @@ 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) {
+	start := timemodule.Now()
+
+	result, resultVar1, err := s.PostStore.PermanentDeleteBatchForRetentionPolicies(now, globalPolicyEndTime, limit, cursor)
+
+	elapsed := float64(timemodule.Since(start)) / float64(timemodule.Second)
+	if s.Root.Metrics != nil {
+		success := "false"
+		if err == nil {
+			success = "true"
+		}
+		s.Root.Metrics.ObserveStoreMethodDuration("PostStore.PermanentDeleteBatchForRetentionPolicies", success, elapsed)
+	}
+	return result, resultVar1, err
+}
+
 func (s *TimerLayerPostStore) PermanentDeleteByChannel(channelID string) error {
 	start := timemodule.Now()
 
@@ -5209,6 +5273,22 @@ func (s *TimerLayerPreferenceStore) DeleteCategoryAndName(category string, name
 	return err
 }
 
+func (s *TimerLayerPreferenceStore) DeleteOrphanedRows(limit int) (int64, error) {
+	start := timemodule.Now()
+
+	result, err := s.PreferenceStore.DeleteOrphanedRows(limit)
+
+	elapsed := float64(timemodule.Since(start)) / float64(timemodule.Second)
+	if s.Root.Metrics != nil {
+		success := "false"
+		if err == nil {
+			success = "true"
+		}
+		s.Root.Metrics.ObserveStoreMethodDuration("PreferenceStore.DeleteOrphanedRows", success, elapsed)
+	}
+	return result, err
+}
+
 func (s *TimerLayerPreferenceStore) Get(userID string, category string, name string) (*model.Preference, error) {
 	start := timemodule.Now()
 
@@ -5401,6 +5481,22 @@ func (s *TimerLayerReactionStore) DeleteAllWithEmojiName(emojiName string) error
 	return err
 }
 
+func (s *TimerLayerReactionStore) DeleteOrphanedRows(limit int) (int64, error) {
+	start := timemodule.Now()
+
+	result, err := s.ReactionStore.DeleteOrphanedRows(limit)
+
+	elapsed := float64(timemodule.Since(start)) / float64(timemodule.Second)
+	if s.Root.Metrics != nil {
+		success := "false"
+		if err == nil {
+			success = "true"
+		}
+		s.Root.Metrics.ObserveStoreMethodDuration("ReactionStore.DeleteOrphanedRows", success, elapsed)
+	}
+	return result, err
+}
+
 func (s *TimerLayerReactionStore) GetForPost(postID string, allowFromCache bool) ([]*model.Reaction, error) {
 	start := timemodule.Now()
 
@@ -5625,6 +5721,22 @@ func (s *TimerLayerRetentionPolicyStore) Delete(id string) error {
 	return err
 }
 
+func (s *TimerLayerRetentionPolicyStore) DeleteOrphanedRows(limit int) (int64, error) {
+	start := timemodule.Now()
+
+	result, err := s.RetentionPolicyStore.DeleteOrphanedRows(limit)
+
+	elapsed := float64(timemodule.Since(start)) / float64(timemodule.Second)
+	if s.Root.Metrics != nil {
+		success := "false"
+		if err == nil {
+			success = "true"
+		}
+		s.Root.Metrics.ObserveStoreMethodDuration("RetentionPolicyStore.DeleteOrphanedRows", success, elapsed)
+	}
+	return result, err
+}
+
 func (s *TimerLayerRetentionPolicyStore) Get(id string) (*model.RetentionPolicyWithTeamAndChannelCounts, error) {
 	start := timemodule.Now()
 
@@ -7942,6 +8054,22 @@ func (s *TimerLayerThreadStore) DeleteMembershipForUser(userId string, postID st
 	return err
 }
 
+func (s *TimerLayerThreadStore) DeleteOrphanedRows(limit int) (int64, error) {
+	start := timemodule.Now()
+
+	result, err := s.ThreadStore.DeleteOrphanedRows(limit)
+
+	elapsed := float64(timemodule.Since(start)) / float64(timemodule.Second)
+	if s.Root.Metrics != nil {
+		success := "false"
+		if err == nil {
+			success = "true"
+		}
+		s.Root.Metrics.ObserveStoreMethodDuration("ThreadStore.DeleteOrphanedRows", success, elapsed)
+	}
+	return result, err
+}
+
 func (s *TimerLayerThreadStore) Get(id string) (*model.Thread, error) {
 	start := timemodule.Now()
 
@@ -8118,6 +8246,38 @@ 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) {
+	start := timemodule.Now()
+
+	result, resultVar1, err := s.ThreadStore.PermanentDeleteBatchForRetentionPolicies(now, globalPolicyEndTime, limit, cursor)
+
+	elapsed := float64(timemodule.Since(start)) / float64(timemodule.Second)
+	if s.Root.Metrics != nil {
+		success := "false"
+		if err == nil {
+			success = "true"
+		}
+		s.Root.Metrics.ObserveStoreMethodDuration("ThreadStore.PermanentDeleteBatchForRetentionPolicies", success, elapsed)
+	}
+	return result, resultVar1, err
+}
+
+func (s *TimerLayerThreadStore) PermanentDeleteBatchThreadMembershipsForRetentionPolicies(now int64, globalPolicyEndTime int64, limit int64, cursor model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error) {
+	start := timemodule.Now()
+
+	result, resultVar1, err := s.ThreadStore.PermanentDeleteBatchThreadMembershipsForRetentionPolicies(now, globalPolicyEndTime, limit, cursor)
+
+	elapsed := float64(timemodule.Since(start)) / float64(timemodule.Second)
+	if s.Root.Metrics != nil {
+		success := "false"
+		if err == nil {
+			success = "true"
+		}
+		s.Root.Metrics.ObserveStoreMethodDuration("ThreadStore.PermanentDeleteBatchThreadMembershipsForRetentionPolicies", success, elapsed)
+	}
+	return result, resultVar1, err
+}
+
 func (s *TimerLayerThreadStore) Save(thread *model.Thread) (*model.Thread, error) {
 	start := timemodule.Now()