diff --git a/app/syncables_test.go b/app/syncables_test.go index 623fde7707..3c2f950970 100644 --- a/app/syncables_test.go +++ b/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()) } diff --git a/i18n/en.json b/i18n/en.json index c43c15c776..e2738b54b0 100644 --- a/i18n/en.json +++ b/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" diff --git a/jobs/workers.go b/jobs/workers.go index 5bd5d1567f..660015d86a 100644 --- a/jobs/workers.go +++ b/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() } diff --git a/model/config.go b/model/config.go index ca82cef958..e6215f0fb9 100644 --- a/model/config.go +++ b/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 { diff --git a/model/data_retention_policy.go b/model/data_retention_policy.go index 2c0d6099a4..3f984d1bcf 100644 --- a/model/data_retention_policy.go +++ b/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 diff --git a/model/post.go b/model/post.go index dd66d69166..74bd1e813d 100644 --- a/model/post.go +++ 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 +} diff --git a/services/telemetry/telemetry.go b/services/telemetry/telemetry.go index 56bcf4c03c..a45eadd24b 100644 --- a/services/telemetry/telemetry.go +++ b/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{}{ diff --git a/store/opentracinglayer/opentracinglayer.go b/store/opentracinglayer/opentracinglayer.go index 5648c37fcf..70731b9df0 100644 --- a/store/opentracinglayer/opentracinglayer.go +++ b/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") diff --git a/store/retrylayer/retrylayer.go b/store/retrylayer/retrylayer.go index e8cc96e90f..e23b821fcf 100644 --- a/store/retrylayer/retrylayer.go +++ b/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 diff --git a/store/sqlstore/channel_member_history_store.go b/store/sqlstore/channel_member_history_store.go index 556d266a27..73f45cbe65 100644 --- a/store/sqlstore/channel_member_history_store.go +++ b/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 diff --git a/store/sqlstore/post_store.go b/store/sqlstore/post_store.go index bb675c6e89..087d4a1f24 100644 --- a/store/sqlstore/post_store.go +++ b/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" { diff --git a/store/sqlstore/preference_store.go b/store/sqlstore/preference_store.go index 6056724b72..e1a2a8aa98 100644 --- a/store/sqlstore/preference_store.go +++ b/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. diff --git a/store/sqlstore/reaction_store.go b/store/sqlstore/reaction_store.go index 6ef1ea2db3..40a375be91 100644 --- a/store/sqlstore/reaction_store.go +++ b/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" { diff --git a/store/sqlstore/retention_policy_store.go b/store/sqlstore/retention_policy_store.go index 946367d354..5a696c671f 100644 --- a/store/sqlstore/retention_policy_store.go +++ b/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 +} diff --git a/store/sqlstore/thread_store.go b/store/sqlstore/thread_store.go index 4606d93a34..46d4d3a8a3 100644 --- a/store/sqlstore/thread_store.go +++ b/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 +} diff --git a/store/store.go b/store/store.go index adcd0605a6..d6f53f7e21 100644 --- a/store/store.go +++ b/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 { diff --git a/store/storetest/channel_member_history_store.go b/store/storetest/channel_member_history_store.go index 8e14d47607..60cfed58f6 100644 --- a/store/storetest/channel_member_history_store.go +++ b/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") +} diff --git a/store/storetest/group_store.go b/store/storetest/group_store.go index 8b269026b0..499369bc00 100644 --- a/store/storetest/group_store.go +++ b/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) diff --git a/store/storetest/mocks/ChannelMemberHistoryStore.go b/store/storetest/mocks/ChannelMemberHistoryStore.go index 0ade64e1df..c274b470e5 100644 --- a/store/storetest/mocks/ChannelMemberHistoryStore.go +++ b/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 +} diff --git a/store/storetest/mocks/PostStore.go b/store/storetest/mocks/PostStore.go index 0625d4b47b..c384704d49 100644 --- a/store/storetest/mocks/PostStore.go +++ b/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) diff --git a/store/storetest/mocks/PreferenceStore.go b/store/storetest/mocks/PreferenceStore.go index 468aae8d5e..b1b3149fce 100644 --- a/store/storetest/mocks/PreferenceStore.go +++ b/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) diff --git a/store/storetest/mocks/ReactionStore.go b/store/storetest/mocks/ReactionStore.go index c9e78c5198..d7e10e2bc8 100644 --- a/store/storetest/mocks/ReactionStore.go +++ b/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) diff --git a/store/storetest/mocks/RetentionPolicyStore.go b/store/storetest/mocks/RetentionPolicyStore.go index 4b3dac2818..3a15db11ab 100644 --- a/store/storetest/mocks/RetentionPolicyStore.go +++ b/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) diff --git a/store/storetest/mocks/ThreadStore.go b/store/storetest/mocks/ThreadStore.go index d6c4b76764..317540e280 100644 --- a/store/storetest/mocks/ThreadStore.go +++ b/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) diff --git a/store/storetest/post_store.go b/store/storetest/post_store.go index 1ee3eb90e1..20b383cdfb 100644 --- a/store/storetest/post_store.go +++ b/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) { diff --git a/store/storetest/preference_store.go b/store/storetest/preference_store.go index 648b0b3234..2c8d38b07d 100644 --- a/store/storetest/preference_store.go +++ b/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") } diff --git a/store/storetest/reaction_store.go b/store/storetest/reaction_store.go index be1fc621c5..c331ce65c8 100644 --- a/store/storetest/reaction_store.go +++ b/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) { diff --git a/store/storetest/retention_policy_store.go b/store/storetest/retention_policy_store.go index 6fcc22d506..fc6f08eb16 100644 --- a/store/storetest/retention_policy_store.go +++ b/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) +} diff --git a/store/storetest/thread_store.go b/store/storetest/thread_store.go index d8f6d3a8be..46e380f282 100644 --- a/store/storetest/thread_store.go +++ b/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") +} diff --git a/store/timerlayer/timerlayer.go b/store/timerlayer/timerlayer.go index 18d5f532fb..3bae3c6031 100644 --- a/store/timerlayer/timerlayer.go +++ b/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()