diff --git a/model/message_export.go b/model/message_export.go index 88108e2ef7..f94d861f8b 100644 --- a/model/message_export.go +++ b/model/message_export.go @@ -29,3 +29,8 @@ type MessageExport struct { PostOriginalId *string PostFileIds StringArray } + +type MessageExportCursor struct { + LastPostUpdateAt int64 + LastPostId string +} diff --git a/store/opentracinglayer/opentracinglayer.go b/store/opentracinglayer/opentracinglayer.go index cfea688307..3f747fe3b4 100644 --- a/store/opentracinglayer/opentracinglayer.go +++ b/store/opentracinglayer/opentracinglayer.go @@ -2801,7 +2801,7 @@ func (s *OpenTracingLayerComplianceStore) GetAll(offset int, limit int) (model.C return result, err } -func (s *OpenTracingLayerComplianceStore) MessageExport(after int64, limit int) ([]*model.MessageExport, error) { +func (s *OpenTracingLayerComplianceStore) MessageExport(cursor model.MessageExportCursor, limit int) ([]*model.MessageExport, model.MessageExportCursor, error) { origCtx := s.Root.Store.Context() span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "ComplianceStore.MessageExport") s.Root.Store.SetContext(newCtx) @@ -2810,13 +2810,13 @@ func (s *OpenTracingLayerComplianceStore) MessageExport(after int64, limit int) }() defer span.Finish() - result, err := s.ComplianceStore.MessageExport(after, limit) + result, resultVar1, err := s.ComplianceStore.MessageExport(cursor, limit) if err != nil { span.LogFields(spanlog.Error(err)) ext.Error.Set(span, true) } - return result, err + return result, resultVar1, err } func (s *OpenTracingLayerComplianceStore) Save(compliance *model.Compliance) (*model.Compliance, error) { diff --git a/store/retrylayer/retrylayer.go b/store/retrylayer/retrylayer.go index 227ddf63f4..1a92a9766f 100644 --- a/store/retrylayer/retrylayer.go +++ b/store/retrylayer/retrylayer.go @@ -2994,21 +2994,21 @@ func (s *RetryLayerComplianceStore) GetAll(offset int, limit int) (model.Complia } -func (s *RetryLayerComplianceStore) MessageExport(after int64, limit int) ([]*model.MessageExport, error) { +func (s *RetryLayerComplianceStore) MessageExport(cursor model.MessageExportCursor, limit int) ([]*model.MessageExport, model.MessageExportCursor, error) { tries := 0 for { - result, err := s.ComplianceStore.MessageExport(after, limit) + result, resultVar1, err := s.ComplianceStore.MessageExport(cursor, limit) if err == nil { - return result, nil + return result, resultVar1, nil } if !isRepeatableError(err) { - return result, err + return result, resultVar1, err } tries++ if tries >= 3 { err = errors.Wrap(err, "giving up after 3 consecutive repeatable transaction failures") - return result, err + return result, resultVar1, err } } diff --git a/store/sqlstore/compliance_store.go b/store/sqlstore/compliance_store.go index f200f953c7..b9548b5d28 100644 --- a/store/sqlstore/compliance_store.go +++ b/store/sqlstore/compliance_store.go @@ -242,8 +242,12 @@ func (s SqlComplianceStore) ComplianceExport(job *model.Compliance, cursor model return append(channelPosts, directMessagePosts...), cursor, nil } -func (s SqlComplianceStore) MessageExport(after int64, limit int) ([]*model.MessageExport, error) { - props := map[string]interface{}{"StartTime": after, "Limit": limit} +func (s SqlComplianceStore) MessageExport(cursor model.MessageExportCursor, limit int) ([]*model.MessageExport, model.MessageExportCursor, error) { + props := map[string]interface{}{ + "LastPostUpdateAt": cursor.LastPostUpdateAt, + "LastPostId": cursor.LastPostId, + "Limit": limit, + } query := `SELECT Posts.Id AS PostId, @@ -277,15 +281,23 @@ func (s SqlComplianceStore) MessageExport(after int64, limit int) ([]*model.Mess LEFT OUTER JOIN Teams ON Channels.TeamId = Teams.Id LEFT OUTER JOIN Users ON Posts.UserId = Users.Id LEFT JOIN Bots ON Bots.UserId = Posts.UserId - WHERE - Posts.UpdateAt > :StartTime AND - Posts.Type NOT LIKE 'system_%' - ORDER BY PostUpdateAt + WHERE ( + Posts.UpdateAt > :LastPostUpdateAt + OR ( + Posts.UpdateAt = :LastPostUpdateAt + AND Posts.Id > :LastPostId + ) + ) AND Posts.Type NOT LIKE 'system_%' + ORDER BY PostUpdateAt, PostId LIMIT :Limit` var cposts []*model.MessageExport if _, err := s.GetReplica().Select(&cposts, query, props); err != nil { - return nil, errors.Wrap(err, "unable to export messages") + return nil, cursor, errors.Wrap(err, "unable to export messages") } - return cposts, nil + if len(cposts) > 0 { + cursor.LastPostUpdateAt = *cposts[len(cposts)-1].PostUpdateAt + cursor.LastPostId = *cposts[len(cposts)-1].PostId + } + return cposts, cursor, nil } diff --git a/store/store.go b/store/store.go index 0729c5345e..a974de95d8 100644 --- a/store/store.go +++ b/store/store.go @@ -471,7 +471,7 @@ type ComplianceStore interface { Get(id string) (*model.Compliance, error) GetAll(offset, limit int) (model.Compliances, error) ComplianceExport(compliance *model.Compliance, cursor model.ComplianceExportCursor, limit int) ([]*model.CompliancePost, model.ComplianceExportCursor, error) - MessageExport(after int64, limit int) ([]*model.MessageExport, error) + MessageExport(cursor model.MessageExportCursor, limit int) ([]*model.MessageExport, model.MessageExportCursor, error) } type OAuthStore interface { diff --git a/store/storetest/compliance_store.go b/store/storetest/compliance_store.go index 6ee57b26e7..74d4cea9d5 100644 --- a/store/storetest/compliance_store.go +++ b/store/storetest/compliance_store.go @@ -394,7 +394,7 @@ func testMessageExportPublicChannel(t *testing.T, ss store.Store) { // get the starting number of message export entries startTime := model.GetMillis() - messages, err := ss.Compliance().MessageExport(startTime-10, 10) + messages, _, err := ss.Compliance().MessageExport(model.MessageExportCursor{LastPostUpdateAt: startTime - 10}, 10) require.NoError(t, err) assert.Equal(t, 0, len(messages)) @@ -464,7 +464,7 @@ func testMessageExportPublicChannel(t *testing.T, ss store.Store) { // fetch the message exports for both posts that user1 sent messageExportMap := map[string]model.MessageExport{} - messages, err = ss.Compliance().MessageExport(startTime-10, 10) + messages, _, err = ss.Compliance().MessageExport(model.MessageExportCursor{LastPostUpdateAt: startTime - 10}, 10) require.NoError(t, err) assert.Equal(t, 2, len(messages)) @@ -498,7 +498,7 @@ func testMessageExportPrivateChannel(t *testing.T, ss store.Store) { // get the starting number of message export entries startTime := model.GetMillis() - messages, err := ss.Compliance().MessageExport(startTime-10, 10) + messages, _, err := ss.Compliance().MessageExport(model.MessageExportCursor{LastPostUpdateAt: startTime - 10}, 10) require.NoError(t, err) assert.Equal(t, 0, len(messages)) @@ -568,7 +568,7 @@ func testMessageExportPrivateChannel(t *testing.T, ss store.Store) { // fetch the message exports for both posts that user1 sent messageExportMap := map[string]model.MessageExport{} - messages, err = ss.Compliance().MessageExport(startTime-10, 10) + messages, _, err = ss.Compliance().MessageExport(model.MessageExportCursor{LastPostUpdateAt: startTime - 10}, 10) require.NoError(t, err) assert.Equal(t, 2, len(messages)) @@ -604,7 +604,7 @@ func testMessageExportDirectMessageChannel(t *testing.T, ss store.Store) { // get the starting number of message export entries startTime := model.GetMillis() - messages, err := ss.Compliance().MessageExport(startTime-10, 10) + messages, _, err := ss.Compliance().MessageExport(model.MessageExportCursor{LastPostUpdateAt: startTime - 10}, 10) require.NoError(t, err) assert.Equal(t, 0, len(messages)) @@ -659,7 +659,7 @@ func testMessageExportDirectMessageChannel(t *testing.T, ss store.Store) { // fetch the message export for the post that user1 sent messageExportMap := map[string]model.MessageExport{} - messages, err = ss.Compliance().MessageExport(startTime-10, 10) + messages, _, err = ss.Compliance().MessageExport(model.MessageExportCursor{LastPostUpdateAt: startTime - 10}, 10) require.NoError(t, err) assert.Equal(t, 1, len(messages)) @@ -685,7 +685,7 @@ func testMessageExportGroupMessageChannel(t *testing.T, ss store.Store) { // get the starting number of message export entries startTime := model.GetMillis() - messages, err := ss.Compliance().MessageExport(startTime-10, 10) + messages, _, err := ss.Compliance().MessageExport(model.MessageExportCursor{LastPostUpdateAt: startTime - 10}, 10) require.NoError(t, err) assert.Equal(t, 0, len(messages)) @@ -757,7 +757,7 @@ func testMessageExportGroupMessageChannel(t *testing.T, ss store.Store) { // fetch the message export for the post that user1 sent messageExportMap := map[string]model.MessageExport{} - messages, err = ss.Compliance().MessageExport(startTime-10, 10) + messages, _, err = ss.Compliance().MessageExport(model.MessageExportCursor{LastPostUpdateAt: startTime - 10}, 10) require.NoError(t, err) assert.Equal(t, 1, len(messages)) @@ -782,7 +782,7 @@ func testEditExportMessage(t *testing.T, ss store.Store) { defer cleanupStoreState(t, ss) // get the starting number of message export entries startTime := model.GetMillis() - messages, err := ss.Compliance().MessageExport(startTime-1, 10) + messages, _, err := ss.Compliance().MessageExport(model.MessageExportCursor{LastPostUpdateAt: startTime - 1}, 10) require.NoError(t, err) assert.Equal(t, 0, len(messages)) @@ -837,7 +837,7 @@ func testEditExportMessage(t *testing.T, ss store.Store) { require.NoError(t, err) // fetch the message exports from the start - messages, err = ss.Compliance().MessageExport(startTime-1, 10) + messages, _, err = ss.Compliance().MessageExport(model.MessageExportCursor{LastPostUpdateAt: startTime - 1}, 10) require.NoError(t, err) assert.Equal(t, 2, len(messages)) @@ -874,7 +874,7 @@ func testEditAfterExportMessage(t *testing.T, ss store.Store) { defer cleanupStoreState(t, ss) // get the starting number of message export entries startTime := model.GetMillis() - messages, err := ss.Compliance().MessageExport(startTime-1, 10) + messages, _, err := ss.Compliance().MessageExport(model.MessageExportCursor{LastPostUpdateAt: startTime - 1}, 10) require.NoError(t, err) assert.Equal(t, 0, len(messages)) @@ -922,7 +922,7 @@ func testEditAfterExportMessage(t *testing.T, ss store.Store) { require.NoError(t, err) // fetch the message exports from the start - messages, err = ss.Compliance().MessageExport(startTime-1, 10) + messages, _, err = ss.Compliance().MessageExport(model.MessageExportCursor{LastPostUpdateAt: startTime - 1}, 10) require.NoError(t, err) assert.Equal(t, 1, len(messages)) @@ -948,7 +948,7 @@ func testEditAfterExportMessage(t *testing.T, ss store.Store) { require.NoError(t, err) // fetch the message exports after edit - messages, err = ss.Compliance().MessageExport(postEditTime-1, 10) + messages, _, err = ss.Compliance().MessageExport(model.MessageExportCursor{LastPostUpdateAt: postEditTime - 1}, 10) require.NoError(t, err) assert.Equal(t, 2, len(messages)) @@ -985,7 +985,7 @@ func testDeleteExportMessage(t *testing.T, ss store.Store) { defer cleanupStoreState(t, ss) // get the starting number of message export entries startTime := model.GetMillis() - messages, err := ss.Compliance().MessageExport(startTime-1, 10) + messages, _, err := ss.Compliance().MessageExport(model.MessageExportCursor{LastPostUpdateAt: startTime - 1}, 10) require.NoError(t, err) assert.Equal(t, 0, len(messages)) @@ -1038,7 +1038,7 @@ func testDeleteExportMessage(t *testing.T, ss store.Store) { require.NoError(t, err) // fetch the message exports from the start - messages, err = ss.Compliance().MessageExport(startTime-1, 10) + messages, _, err = ss.Compliance().MessageExport(model.MessageExportCursor{LastPostUpdateAt: startTime - 1}, 10) require.NoError(t, err) assert.Equal(t, 1, len(messages)) @@ -1070,7 +1070,7 @@ func testDeleteAfterExportMessage(t *testing.T, ss store.Store) { defer cleanupStoreState(t, ss) // get the starting number of message export entries startTime := model.GetMillis() - messages, err := ss.Compliance().MessageExport(startTime-1, 10) + messages, _, err := ss.Compliance().MessageExport(model.MessageExportCursor{LastPostUpdateAt: startTime - 1}, 10) require.NoError(t, err) assert.Equal(t, 0, len(messages)) @@ -1118,7 +1118,7 @@ func testDeleteAfterExportMessage(t *testing.T, ss store.Store) { require.NoError(t, err) // fetch the message exports from the start - messages, err = ss.Compliance().MessageExport(startTime-1, 10) + messages, _, err = ss.Compliance().MessageExport(model.MessageExportCursor{LastPostUpdateAt: startTime - 1}, 10) require.NoError(t, err) assert.Equal(t, 1, len(messages)) @@ -1141,7 +1141,7 @@ func testDeleteAfterExportMessage(t *testing.T, ss store.Store) { require.NoError(t, err) // fetch the message exports after delete - messages, err = ss.Compliance().MessageExport(postDeleteTime-1, 10) + messages, _, err = ss.Compliance().MessageExport(model.MessageExportCursor{LastPostUpdateAt: postDeleteTime - 1}, 10) require.NoError(t, err) assert.Equal(t, 1, len(messages)) diff --git a/store/storetest/mocks/ComplianceStore.go b/store/storetest/mocks/ComplianceStore.go index c901d12b1a..0fa369fe07 100644 --- a/store/storetest/mocks/ComplianceStore.go +++ b/store/storetest/mocks/ComplianceStore.go @@ -90,27 +90,34 @@ func (_m *ComplianceStore) GetAll(offset int, limit int) (model.Compliances, err return r0, r1 } -// MessageExport provides a mock function with given fields: after, limit -func (_m *ComplianceStore) MessageExport(after int64, limit int) ([]*model.MessageExport, error) { - ret := _m.Called(after, limit) +// MessageExport provides a mock function with given fields: cursor, limit +func (_m *ComplianceStore) MessageExport(cursor model.MessageExportCursor, limit int) ([]*model.MessageExport, model.MessageExportCursor, error) { + ret := _m.Called(cursor, limit) var r0 []*model.MessageExport - if rf, ok := ret.Get(0).(func(int64, int) []*model.MessageExport); ok { - r0 = rf(after, limit) + if rf, ok := ret.Get(0).(func(model.MessageExportCursor, int) []*model.MessageExport); ok { + r0 = rf(cursor, limit) } else { if ret.Get(0) != nil { r0 = ret.Get(0).([]*model.MessageExport) } } - var r1 error - if rf, ok := ret.Get(1).(func(int64, int) error); ok { - r1 = rf(after, limit) + var r1 model.MessageExportCursor + if rf, ok := ret.Get(1).(func(model.MessageExportCursor, int) model.MessageExportCursor); ok { + r1 = rf(cursor, limit) } else { - r1 = ret.Error(1) + r1 = ret.Get(1).(model.MessageExportCursor) } - return r0, r1 + var r2 error + if rf, ok := ret.Get(2).(func(model.MessageExportCursor, int) error); ok { + r2 = rf(cursor, limit) + } else { + r2 = ret.Error(2) + } + + return r0, r1, r2 } // Save provides a mock function with given fields: compliance diff --git a/store/timerlayer/timerlayer.go b/store/timerlayer/timerlayer.go index 8b84cc4986..41cb39caff 100644 --- a/store/timerlayer/timerlayer.go +++ b/store/timerlayer/timerlayer.go @@ -2573,10 +2573,10 @@ func (s *TimerLayerComplianceStore) GetAll(offset int, limit int) (model.Complia return result, err } -func (s *TimerLayerComplianceStore) MessageExport(after int64, limit int) ([]*model.MessageExport, error) { +func (s *TimerLayerComplianceStore) MessageExport(cursor model.MessageExportCursor, limit int) ([]*model.MessageExport, model.MessageExportCursor, error) { start := timemodule.Now() - result, err := s.ComplianceStore.MessageExport(after, limit) + result, resultVar1, err := s.ComplianceStore.MessageExport(cursor, limit) elapsed := float64(timemodule.Since(start)) / float64(timemodule.Second) if s.Root.Metrics != nil { @@ -2586,7 +2586,7 @@ func (s *TimerLayerComplianceStore) MessageExport(after int64, limit int) ([]*mo } s.Root.Metrics.ObserveStoreMethodDuration("ComplianceStore.MessageExport", success, elapsed) } - return result, err + return result, resultVar1, err } func (s *TimerLayerComplianceStore) Save(compliance *model.Compliance) (*model.Compliance, error) {