[MM-60888] Fix: Retention can cause unrelated files to be deleted (#29897)

* fix for deleting non-bulk export files and directories

* add job worker setup and running to helper_test.go

* add infrastructure for e2e testing of workers; test ExportDelete

* linting

* use retention days in test instead of magic number

server/channels/jobs/export_delete/worker.go

@@ -5,6 +5,7 @@ package export_delete
 
 import (
 	"path/filepath"
+	"strings"
 	"time"
 
 	"github.com/wiggin77/merror"
@@ -41,6 +42,12 @@ func MakeWorker(jobServer *jobs.JobServer, app AppIface) *jobs.SimpleWorker {
 		errors := merror.New()
 		for i := range exports {
 			filename := filepath.Base(exports[i])
+
+			// Ignore files that were not created by the bulk export command
+			if !strings.HasSuffix(filename, "_export.zip") {
+				continue
+			}
+
 			modTime, appErr := app.ExportFileModTime(filepath.Join(exportPath, filename))
 			if appErr != nil {
 				logger.Debug("Worker: Failed to get file modification time",

server/channels/jobs/export_delete_test.go

@@ -0,0 +1,102 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package jobs_test
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/mattermost/mattermost/server/public/model"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestExportDelete(t *testing.T) {
+	// Create a temporary export directory
+	fileSettingsDir, err := os.MkdirTemp("", "")
+	require.NoError(t, err)
+	relExportDir := "./export"
+	exportDir := filepath.Join(fileSettingsDir, relExportDir)
+
+	t.Cleanup(func() {
+		err = os.RemoveAll(fileSettingsDir)
+		assert.NoError(t, err)
+	})
+
+	retentionDays := 1
+
+	updateConfig := func(cfg *model.Config) {
+		*cfg.FileSettings.DriverName = model.ImageDriverLocal
+		*cfg.FileSettings.Directory = fileSettingsDir
+		*cfg.ExportSettings.Directory = relExportDir
+		*cfg.ExportSettings.RetentionDays = retentionDays
+	}
+
+	th := SetupWithUpdateCfg(t, updateConfig)
+	defer th.TearDown()
+
+	// Create test files with different timestamps
+	files := []string{
+		"old_export.zip",
+		"recent_export.zip",
+		"normal.txt",
+		"test_export.zip",
+		"data.json",
+	}
+
+	// Create directories that should not be deleted
+	dirs := []string{
+		"data",
+		"data/subfolder",
+	}
+
+	for _, dir := range dirs {
+		err = os.MkdirAll(filepath.Join(exportDir, dir), 0755)
+		require.NoError(t, err)
+		// Add a file in each directory that should not be deleted
+		err = os.WriteFile(filepath.Join(exportDir, dir, "file.txt"), []byte("test"), 0644)
+		require.NoError(t, err)
+	}
+
+	for _, file := range files {
+		err = os.WriteFile(filepath.Join(exportDir, file), []byte("test"), 0644)
+		require.NoError(t, err)
+	}
+
+	// Set old timestamps for files that should be deleted
+	oldTime := time.Now().Add(-(time.Duration(retentionDays) * 24 * time.Hour) - 1*time.Hour)
+	err = os.Chtimes(filepath.Join(exportDir, "old_export.zip"), oldTime, oldTime)
+	require.NoError(t, err)
+	err = os.Chtimes(filepath.Join(exportDir, "test_export.zip"), oldTime, oldTime)
+	require.NoError(t, err)
+
+	// Start the workers
+	th.SetupWorkers(t)
+
+	// Run the export delete job
+	th.RunJob(t, model.JobTypeExportDelete, nil)
+
+	// Verify files that should still exist
+	for _, name := range []string{
+		"recent_export.zip",
+		"normal.txt",
+		"data.json",
+		"data/file.txt",
+		"data/subfolder/file.txt",
+	} {
+		_, err := os.Stat(filepath.Join(exportDir, name))
+		require.NoError(t, err, "Expected file/directory to exist: %s", name)
+	}
+
+	// Verify files that should be deleted
+	for _, name := range []string{
+		"old_export.zip",
+		"test_export.zip",
+	} {
+		_, err := os.Stat(filepath.Join(exportDir, name))
+		require.True(t, os.IsNotExist(err), "Expected file to be deleted: %s", name)
+	}
+}

server/channels/jobs/helper_test.go

@@ -11,6 +11,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/stretchr/testify/assert"
+
 	"github.com/mattermost/mattermost/server/public/model"
 	"github.com/mattermost/mattermost/server/public/shared/mlog"
 	"github.com/mattermost/mattermost/server/public/shared/request"
@@ -35,10 +37,12 @@ type TestHelper struct {
 	IncludeCacheLayer bool
 	ConfigStore       *config.Store
 
-	tempWorkspace string
+	tempWorkspace             string
+	oldWatcherPollingInterval int
 }
 
-func setupTestHelper(dbStore store.Store, enterprise bool, includeCacheLayer bool, options []app.Option, tb testing.TB) *TestHelper {
+func setupTestHelper(dbStore store.Store, enterprise bool, includeCacheLayer bool,
+	updateCfg func(cfg *model.Config), options []app.Option) *TestHelper {
 	tempWorkspace, err := os.MkdirTemp("", "jobstest")
 	if err != nil {
 		panic(err)
@@ -54,6 +58,11 @@ func setupTestHelper(dbStore store.Store, enterprise bool, includeCacheLayer boo
 	*memoryConfig.LogSettings.ConsoleLevel = mlog.LvlStdLog.Name
 	*memoryConfig.AnnouncementSettings.AdminNoticesEnabled = false
 	*memoryConfig.AnnouncementSettings.UserNoticesEnabled = false
+
+	if updateCfg != nil {
+		updateCfg(memoryConfig)
+	}
+
 	configStore.Set(memoryConfig)
 
 	buffer := &mlog.Buffer{}
@@ -108,15 +117,25 @@ func setupTestHelper(dbStore store.Store, enterprise bool, includeCacheLayer boo
 }
 
 func Setup(tb testing.TB, options ...app.Option) *TestHelper {
+	return SetupWithUpdateCfg(tb, nil, options...)
+}
+
+func SetupWithUpdateCfg(tb testing.TB, updateCfg func(cfg *model.Config), options ...app.Option) *TestHelper {
 	if testing.Short() {
 		tb.SkipNow()
 	}
+
+	oldWatcherPollingInterval := jobs.DefaultWatcherPollingInterval
+	jobs.DefaultWatcherPollingInterval = 100
+
 	dbStore := mainHelper.GetStore()
 	dbStore.DropAllTables()
 	dbStore.MarkSystemRanUnitTests()
 	mainHelper.PreloadMigrations()
 
-	return setupTestHelper(dbStore, false, true, options, tb)
+	th := setupTestHelper(dbStore, false, true, updateCfg, options)
+	th.oldWatcherPollingInterval = oldWatcherPollingInterval
+	return th
 }
 
 var initBasicOnce sync.Once
@@ -223,6 +242,10 @@ func (th *TestHelper) TearDown() {
 	if th.tempWorkspace != "" {
 		os.RemoveAll(th.tempWorkspace)
 	}
+
+	if th.oldWatcherPollingInterval != 0 {
+		jobs.DefaultWatcherPollingInterval = th.oldWatcherPollingInterval
+	}
 }
 
 func (th *TestHelper) SetupBatchWorker(t *testing.T, worker *jobs.BatchWorker) *model.Job {
@@ -294,3 +317,36 @@ func waitDone(t *testing.T, done chan bool, msg string) {
 		}
 	}, 5*time.Second, 100*time.Millisecond, msg)
 }
+
+func (th *TestHelper) SetupWorkers(t *testing.T) {
+	err := th.App.Srv().Jobs.StartWorkers()
+	require.NoError(t, err)
+}
+
+func (th *TestHelper) RunJob(t *testing.T, jobType string, jobData map[string]string) *model.Job {
+	t.Helper()
+
+	job, appErr := th.Server.Jobs.CreateJob(th.Context, jobType, jobData)
+	require.Nil(t, appErr)
+
+	// poll until completion
+	th.checkJobStatus(t, job.Id, model.JobStatusSuccess)
+	job, appErr = th.Server.Jobs.GetJob(th.Context, job.Id)
+	require.Nil(t, appErr)
+
+	return job
+}
+
+func (th *TestHelper) checkJobStatus(t *testing.T, jobId string, status string) {
+	t.Helper()
+
+	require.Eventuallyf(t, func() bool {
+		// it's ok if there's an error, it might take awhile for the job to finish.
+		job, err := th.Server.Jobs.GetJob(th.Context, jobId)
+		assert.Nil(t, err)
+		if jobId == job.Id {
+			return job.Status == status
+		}
+		return false
+	}, 15*time.Second, 100*time.Millisecond, "expected job's status to be %s", status)
+}