[MM-44489] Cloud limits: enforcing files (#20703)

* Update last accessible file time * Filter fileInfos * Set inaccessible header * Fix lint issue * Fix lint issue * Fix i18n * add nil check * Fix merge conflicts * Add helper functions to clear out inaccessible files content * Remove content for inaccessible files * Fix typo * wip * Remove InaccessibleContent field, instead use Archived * Add store tests * Add tests * Add separate funcs to ignore cloud limits * Use separate query for MySql * Use GetReplicaX Co-authored-by: Mattermod <mattermod@users.noreply.github.com>
2022-09-28 22:22:53 +05:30
--- a/store/opentracinglayer/opentracinglayer.go
+++ b/store/opentracinglayer/opentracinglayer.go
@@ -3527,6 +3527,24 @@ func (s *OpenTracingLayerFileInfoStore) GetStorageUsage(allowFromCache bool, inc
 	return result, err
 }

+func (s *OpenTracingLayerFileInfoStore) GetUptoNSizeFileTime(n int64) (int64, error) {
+	origCtx := s.Root.Store.Context()
+	span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "FileInfoStore.GetUptoNSizeFileTime")
+	s.Root.Store.SetContext(newCtx)
+	defer func() {
+		s.Root.Store.SetContext(origCtx)
+	}()
+
+	defer span.Finish()
+	result, err := s.FileInfoStore.GetUptoNSizeFileTime(n)
+	if err != nil {
+		span.LogFields(spanlog.Error(err))
+		ext.Error.Set(span, true)
+	}
+
+	return result, err
+}
+
 func (s *OpenTracingLayerFileInfoStore) GetWithOptions(page int, perPage int, opt *model.GetFileInfosOptions) ([]*model.FileInfo, error) {
 	origCtx := s.Root.Store.Context()
 	span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "FileInfoStore.GetWithOptions")
--- a/store/retrylayer/retrylayer.go
+++ b/store/retrylayer/retrylayer.go
@@ -3957,6 +3957,27 @@ func (s *RetryLayerFileInfoStore) GetStorageUsage(allowFromCache bool, includeDe

 }

+func (s *RetryLayerFileInfoStore) GetUptoNSizeFileTime(n int64) (int64, error) {
+
+	tries := 0
+	for {
+		result, err := s.FileInfoStore.GetUptoNSizeFileTime(n)
+		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
+		}
+		timepkg.Sleep(100 * timepkg.Millisecond)
+	}
+
+}
+
 func (s *RetryLayerFileInfoStore) GetWithOptions(page int, perPage int, opt *model.GetFileInfosOptions) ([]*model.FileInfo, error) {

 	tries := 0
--- a/store/sqlstore/file_info_store.go
+++ b/store/sqlstore/file_info_store.go
@@ -753,3 +753,49 @@ func (fs SqlFileInfoStore) GetStorageUsage(allowFromCache, includeDeleted bool)
 	}
 	return size, nil
 }
+
+// GetUptoNSizeFileTime returns the CreateAt time of the last accessible file with a running-total size upto n bytes.
+func (fs *SqlFileInfoStore) GetUptoNSizeFileTime(n int64) (int64, error) {
+	if n <= 0 {
+		return 0, errors.New("n can't be less than 1")
+	}
+
+	var sizeSubQuery sq.SelectBuilder
+	// Separate query for MySql, as current min-version 5.x doesn't support window-functions
+	if fs.DriverName() == model.DatabaseDriverMysql {
+		sizeSubQuery = sq.
+			Select("(@runningSum := @runningSum + fi.Size) RunningTotal", "fi.CreateAt").
+			From("FileInfo fi").
+			Join("(SELECT @runningSum := 0) as tmp").
+			Where(sq.Eq{"fi.DeleteAt": 0}).
+			OrderBy("fi.CreateAt DESC, fi.Id")
+	} else {
+		sizeSubQuery = sq.
+			Select("SUM(fi.Size) OVER(ORDER BY CreateAt DESC, fi.Id) RunningTotal", "fi.CreateAt").
+			From("FileInfo fi").
+			Where(sq.Eq{"fi.DeleteAt": 0})
+	}
+
+	builder := fs.getQueryBuilder().
+		Select("fi2.CreateAt").
+		FromSelect(sizeSubQuery, "fi2").
+		Where(sq.LtOrEq{"fi2.RunningTotal": n}).
+		OrderBy("fi2.CreateAt").
+		Limit(1)
+
+	query, queryArgs, err := builder.ToSql()
+	if err != nil {
+		return 0, errors.Wrap(err, "GetUptoNSizeFileTime_tosql")
+	}
+
+	var createAt int64
+	if err := fs.GetReplicaX().Get(&createAt, query, queryArgs...); err != nil {
+		if err == sql.ErrNoRows {
+			return 0, store.NewErrNotFound("File", "none")
+		}
+
+		return 0, errors.Wrapf(err, "failed to get the File for size upto=%d", n)
+	}
+
+	return createAt, nil
+}
--- a/store/store.go
+++ b/store/store.go
@@ -698,6 +698,8 @@ type FileInfoStore interface {
 	GetFilesBatchForIndexing(startTime int64, startFileID string, limit int) ([]*model.FileForIndexing, error)
 	ClearCaches()
 	GetStorageUsage(allowFromCache, includeDeleted bool) (int64, error)
+	// GetUptoNSizeFileTime returns the CreateAt time of the last accessible file with a running-total size upto n bytes.
+	GetUptoNSizeFileTime(n int64) (int64, error)
 }

 type UploadSessionStore interface {
--- a/store/storetest/file_info_store.go
+++ b/store/storetest/file_info_store.go
@@ -11,6 +11,7 @@ import (

 	"github.com/mattermost/mattermost-server/v6/model"
 	"github.com/mattermost/mattermost-server/v6/store"
+	"github.com/mattermost/mattermost-server/v6/utils"

 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -30,6 +31,7 @@ func TestFileInfoStore(t *testing.T, ss store.Store) {
 	t.Run("GetFilesBatchForIndexing", func(t *testing.T) { testFileInfoStoreGetFilesBatchForIndexing(t, ss) })
 	t.Run("CountAll", func(t *testing.T) { testFileInfoStoreCountAll(t, ss) })
 	t.Run("GetStorageUsage", func(t *testing.T) { testFileInfoGetStorageUsage(t, ss) })
+	t.Run("GetUptoNSizeFileTime", func(t *testing.T) { testGetUptoNSizeFileTime(t, ss) })
 }

 func testFileInfoSaveGet(t *testing.T, ss store.Store) {
@@ -772,3 +774,71 @@ func testFileInfoGetStorageUsage(t *testing.T, ss store.Store) {
 	require.NoError(t, err)
 	require.Equal(t, int64(30), usage)
 }
+
+func testGetUptoNSizeFileTime(t *testing.T, ss store.Store) {
+	_, err := ss.FileInfo().GetUptoNSizeFileTime(0)
+	assert.Error(t, err)
+	_, err = ss.FileInfo().GetUptoNSizeFileTime(-1)
+	assert.Error(t, err)
+
+	_, err = ss.FileInfo().PermanentDeleteBatch(model.GetMillis(), 100000)
+	require.NoError(t, err)
+
+	diff := int64(10000)
+	now := utils.MillisFromTime(time.Now()) + diff
+
+	f1, err := ss.FileInfo().Save(&model.FileInfo{
+		PostId:    model.NewId(),
+		CreatorId: model.NewId(),
+		Size:      10,
+		Path:      "file1.txt",
+		CreateAt:  now,
+	})
+	require.NoError(t, err)
+	now = now + diff
+	f2, err := ss.FileInfo().Save(&model.FileInfo{
+		PostId:    model.NewId(),
+		CreatorId: model.NewId(),
+		Size:      10,
+		Path:      "file2.txt",
+		CreateAt:  now,
+	})
+	require.NoError(t, err)
+	now = now + diff
+	f3, err := ss.FileInfo().Save(&model.FileInfo{
+		PostId:    model.NewId(),
+		CreatorId: model.NewId(),
+		Size:      10,
+		Path:      "file3.txt",
+		CreateAt:  now,
+	})
+	require.NoError(t, err)
+	now = now + diff
+	_, err = ss.FileInfo().Save(&model.FileInfo{
+		PostId:    model.NewId(),
+		CreatorId: model.NewId(),
+		Size:      10,
+		Path:      "file4.txt",
+		CreateAt:  now,
+	})
+	require.NoError(t, err)
+
+	createAt, err := ss.FileInfo().GetUptoNSizeFileTime(20)
+	require.NoError(t, err)
+	assert.Equal(t, f3.CreateAt, createAt)
+
+	_, err = ss.FileInfo().GetUptoNSizeFileTime(5)
+	assert.Error(t, err)
+	assert.IsType(t, &store.ErrNotFound{}, err)
+
+	createAt, err = ss.FileInfo().GetUptoNSizeFileTime(1000)
+	require.NoError(t, err)
+	assert.Equal(t, f1.CreateAt, createAt)
+
+	_, err = ss.FileInfo().DeleteForPost(f3.PostId)
+	require.NoError(t, err)
+
+	createAt, err = ss.FileInfo().GetUptoNSizeFileTime(20)
+	require.NoError(t, err)
+	assert.Equal(t, f2.CreateAt, createAt)
+}
--- a/store/storetest/mocks/FileInfoStore.go
+++ b/store/storetest/mocks/FileInfoStore.go
@@ -257,6 +257,27 @@ func (_m *FileInfoStore) GetStorageUsage(allowFromCache bool, includeDeleted boo
 	return r0, r1
 }

+// GetUptoNSizeFileTime provides a mock function with given fields: n
+func (_m *FileInfoStore) GetUptoNSizeFileTime(n int64) (int64, error) {
+	ret := _m.Called(n)
+
+	var r0 int64
+	if rf, ok := ret.Get(0).(func(int64) int64); ok {
+		r0 = rf(n)
+	} else {
+		r0 = ret.Get(0).(int64)
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(int64) error); ok {
+		r1 = rf(n)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
 // GetWithOptions provides a mock function with given fields: page, perPage, opt
 func (_m *FileInfoStore) GetWithOptions(page int, perPage int, opt *model.GetFileInfosOptions) ([]*model.FileInfo, error) {
 	ret := _m.Called(page, perPage, opt)
--- a/store/timerlayer/timerlayer.go
+++ b/store/timerlayer/timerlayer.go
@@ -3222,6 +3222,22 @@ func (s *TimerLayerFileInfoStore) GetStorageUsage(allowFromCache bool, includeDe
 	return result, err
 }

+func (s *TimerLayerFileInfoStore) GetUptoNSizeFileTime(n int64) (int64, error) {
+	start := time.Now()
+
+	result, err := s.FileInfoStore.GetUptoNSizeFileTime(n)
+
+	elapsed := float64(time.Since(start)) / float64(time.Second)
+	if s.Root.Metrics != nil {
+		success := "false"
+		if err == nil {
+			success = "true"
+		}
+		s.Root.Metrics.ObserveStoreMethodDuration("FileInfoStore.GetUptoNSizeFileTime", success, elapsed)
+	}
+	return result, err
+}
+
 func (s *TimerLayerFileInfoStore) GetWithOptions(page int, perPage int, opt *model.GetFileInfosOptions) ([]*model.FileInfo, error) {
 	start := time.Now()