[MM-63557] mmctl: Add compliance export create cmd (#30594)

* Refactor job retrieval to support multiple statuses & multiple types - Updated job retrieval functions to handle multiple job statuses. - Renamed `GetJobsByTypeAndStatus` to `GetJobsByTypesAndStatuses` for consistency across the codebase. - Adjusted related function signatures and implementations in the job store and retry layer to accommodate the new method. - Updated tests to reflect changes in job retrieval logic and ensure proper functionality. * Add compliance export create command and tests - Introduced `ComplianceExportCreateCmd` to facilitate the creation of compliance export jobs with options for date, start, and end timestamps. - Added unit tests for the new command, covering various scenarios including valid and invalid inputs. - Updated documentation to include usage examples and options for the new command. - Enhanced existing tests to ensure proper functionality of compliance export job handling. * update docs * update tests for new logic * Refactor message export job tests to use DefaultPreviousJobPageSize - Updated all test cases in worker_test.go to replace hardcoded page size of 100 with DefaultPreviousJobPageSize for consistency. - Adjusted the worker.go file to define DefaultPreviousJobPageSize and use it in job retrieval logic. - Ensured that the changes maintain the functionality of job data initialization and retrieval tests. * PR comments * PR comments, simplifications, clarifications, formatting * prefer hypen over underscore in command names * merge conflict * update mmctl docs
2025-06-24 17:38:30 -04:00
--- a/server/enterprise/message_export/shared/shared.go
+++ b/server/enterprise/message_export/shared/shared.go
@@ -42,6 +42,7 @@ const (

 	JobDataJobStartId              = "job_start_id"
 	JobDataExportType              = "export_type"
+	JobDataInitiatedBy             = "initiated_by"
 	JobDataBatchSize               = "batch_size"
 	JobDataChannelBatchSize        = "channel_batch_size"
 	JobDataChannelHistoryBatchSize = "channel_history_batch_size"
--- a/server/enterprise/message_export/worker.go
+++ b/server/enterprise/message_export/worker.go
@@ -22,7 +22,10 @@ import (
 	"github.com/mattermost/mattermost/server/v8/platform/shared/templates"
 )

-const TimeBetweenBatchesMs = 100
+const (
+	TimeBetweenBatchesMs       = 100
+	DefaultPreviousJobPageSize = 5
+)

 // testEndOfBatchCb is only used for testing
 var testEndOfBatchCb func(worker *MessageExportWorker)
@@ -158,8 +161,9 @@ func (w *MessageExportWorker) DoJob(job *model.Job) {
 	go w.jobServer.CancellationWatcher(cancelContext, job.Id, cancelWatcherChan)
 	defer cancelCancelWatcher()

+	rctx := request.EmptyContext(logger).WithContext(w.context)
 	// if job data is missing, we'll do our best to recover
-	w.initJobData(logger, job, time.Now())
+	w.initJobData(rctx, logger, job, time.Now())
 	data, err := extractJobData(logger, job.Data)
 	if err != nil {
 		// Error in conversion. Not much we can do about that. But it shouldn't happen, unless someone edited the db.
@@ -167,7 +171,6 @@ func (w *MessageExportWorker) DoJob(job *model.Job) {
 		return
 	}

-	rctx := request.EmptyContext(logger).WithContext(w.context)
 	reportProgress := func(message string) {
 		logger.Debug(message)
 		// Don't fail because we couldn't update progress.
@@ -260,7 +263,7 @@ func (w *MessageExportWorker) finishExport(rctx request.CTX, logger *mlog.Logger
 }

 // initializes job data if it's missing, allows us to recover from failed or improperly configured jobs
-func (w *MessageExportWorker) initJobData(logger mlog.LoggerIFace, job *model.Job, now time.Time) {
+func (w *MessageExportWorker) initJobData(rctx request.CTX, logger mlog.LoggerIFace, job *model.Job, now time.Time) {
 	if job.Data == nil {
 		job.Data = make(map[string]string)
 	}
@@ -307,7 +310,8 @@ func (w *MessageExportWorker) initJobData(logger mlog.LoggerIFace, job *model.Jo
 	}

 	if _, exists := job.Data[shared.JobDataBatchStartTime]; !exists {
-		previousJob, err := w.jobServer.Store.Job().GetNewestJobByStatusesAndType([]string{model.JobStatusWarning, model.JobStatusSuccess}, model.JobTypeMessageExport)
+		previousJob, err := w.getPreviousNonCliJob(rctx)
+
 		if err != nil {
 			exportFromTimestamp := strconv.FormatInt(*w.jobServer.Config().MessageExportSettings.ExportFromTimestamp, 10)
 			logger.Info("Worker: No previously successful job found, falling back to configured MessageExportSettings.ExportFromTimestamp", mlog.String("export_from_timestamp", exportFromTimestamp))
@@ -362,6 +366,36 @@ func (w *MessageExportWorker) initJobData(logger mlog.LoggerIFace, job *model.Jo
 	job.Data[shared.JobDataExportDir] = getJobExportDir(logger, job.Data, job.Data[shared.JobDataJobStartTime], job.Data[shared.JobDataJobEndTime])
 }

+// getPreviousNonCliJob returns the most recent job that was not initiated by mmctl
+func (w *MessageExportWorker) getPreviousNonCliJob(rctx request.CTX) (*model.Job, error) {
+	offset := 0
+
+	for {
+		jobs, err := w.jobServer.Store.Job().GetAllByTypesAndStatusesPage(rctx,
+			[]string{model.JobTypeMessageExport},
+			[]string{model.JobStatusWarning, model.JobStatusSuccess},
+			offset, DefaultPreviousJobPageSize)
+		if err != nil {
+			return nil, err
+		}
+
+		// Find the first job not initiated by mmctl
+		for _, job := range jobs {
+			if job.Data == nil || job.Data[shared.JobDataInitiatedBy] != "mmctl" {
+				return job, nil
+			}
+		}
+
+		// If we didn't get a full page of jobs, we've reached the end
+		if len(jobs) < DefaultPreviousJobPageSize {
+			return nil, nil
+		}
+
+		// If we didn't find a non-mmctl job in this page, continue to the next page
+		offset += DefaultPreviousJobPageSize
+	}
+}
+
 func extractJobData(logger *mlog.Logger, strmap map[string]string) (shared.JobData, error) {
 	data, err := shared.StringMapToJobDataWithZeroValues(strmap)
 	if err != nil {
--- a/server/enterprise/message_export/worker_test.go
+++ b/server/enterprise/message_export/worker_test.go
@@ -43,7 +43,10 @@ func TestInitJobDataNoJobData(t *testing.T) {
 	}

 	// mock job store doesn't return a previously successful job, forcing fallback to config
-	mockStore.JobStore.On("GetNewestJobByStatusesAndType", []string{model.JobStatusWarning, model.JobStatusSuccess}, model.JobTypeMessageExport).Return(nil, errors.New("test"))
+	mockStore.JobStore.On("GetAllByTypesAndStatusesPage", mock.Anything,
+		[]string{model.JobTypeMessageExport},
+		[]string{model.JobStatusWarning, model.JobStatusSuccess},
+		0, DefaultPreviousJobPageSize).Return(nil, errors.New("test"))

 	worker := &MessageExportWorker{
 		jobServer: &jobs.JobServer{
@@ -67,7 +70,7 @@ func TestInitJobDataNoJobData(t *testing.T) {
 	}

 	now := time.Now()
-	worker.initJobData(logger, job, now)
+	worker.initJobData(request.EmptyContext(logger), logger, job, now)

 	assert.Equal(t, model.ComplianceExportTypeActiance, job.Data[shared.JobDataExportType])
 	assert.Equal(t, strconv.Itoa(*worker.jobServer.Config().MessageExportSettings.BatchSize), job.Data[shared.JobDataBatchSize])
@@ -98,7 +101,10 @@ func TestInitJobDataPreviousJobNoJobData(t *testing.T) {
 	}

 	// mock job store returns a previously successful job, but it doesn't have job data either, so we still fall back to config
-	mockStore.JobStore.On("GetNewestJobByStatusesAndType", []string{model.JobStatusWarning, model.JobStatusSuccess}, model.JobTypeMessageExport).Return(previousJob, nil)
+	mockStore.JobStore.On("GetAllByTypesAndStatusesPage", mock.Anything,
+		[]string{model.JobTypeMessageExport},
+		[]string{model.JobStatusWarning, model.JobStatusSuccess},
+		0, DefaultPreviousJobPageSize).Return([]*model.Job{previousJob}, nil)

 	worker := &MessageExportWorker{
 		jobServer: &jobs.JobServer{
@@ -122,7 +128,7 @@ func TestInitJobDataPreviousJobNoJobData(t *testing.T) {
 	}

 	now := time.Now()
-	worker.initJobData(logger, job, now)
+	worker.initJobData(request.EmptyContext(logger), logger, job, now)

 	assert.Equal(t, model.ComplianceExportTypeActiance, job.Data[shared.JobDataExportType])
 	assert.Equal(t, strconv.Itoa(*worker.jobServer.Config().MessageExportSettings.BatchSize), job.Data[shared.JobDataBatchSize])
@@ -155,7 +161,10 @@ func TestInitJobDataPreviousJobWithJobData(t *testing.T) {
 	}

 	// mock job store returns a previously successful job that has the config that we're looking for, so we use it
-	mockStore.JobStore.On("GetNewestJobByStatusesAndType", []string{model.JobStatusWarning, model.JobStatusSuccess}, model.JobTypeMessageExport).Return(previousJob, nil)
+	mockStore.JobStore.On("GetAllByTypesAndStatusesPage", mock.Anything,
+		[]string{model.JobTypeMessageExport},
+		[]string{model.JobStatusWarning, model.JobStatusSuccess},
+		0, DefaultPreviousJobPageSize).Return([]*model.Job{previousJob}, nil)

 	worker := &MessageExportWorker{
 		jobServer: &jobs.JobServer{
@@ -179,7 +188,7 @@ func TestInitJobDataPreviousJobWithJobData(t *testing.T) {
 	}

 	now := time.Now()
-	worker.initJobData(logger, job, now)
+	worker.initJobData(request.EmptyContext(logger), logger, job, now)

 	assert.Equal(t, model.ComplianceExportTypeActiance, job.Data[shared.JobDataExportType])
 	assert.Equal(t, strconv.Itoa(*worker.jobServer.Config().MessageExportSettings.BatchSize), job.Data[shared.JobDataBatchSize])
@@ -212,7 +221,10 @@ func TestInitJobDataPreviousJobWithJobDataPre105(t *testing.T) {
 	}

 	// mock job store returns a previously successful job that has the config that we're looking for, so we use it
-	mockStore.JobStore.On("GetNewestJobByStatusesAndType", []string{model.JobStatusWarning, model.JobStatusSuccess}, model.JobTypeMessageExport).Return(previousJob, nil)
+	mockStore.JobStore.On("GetAllByTypesAndStatusesPage", mock.Anything,
+		[]string{model.JobTypeMessageExport},
+		[]string{model.JobStatusWarning, model.JobStatusSuccess},
+		0, DefaultPreviousJobPageSize).Return([]*model.Job{previousJob}, nil)

 	worker := &MessageExportWorker{
 		jobServer: &jobs.JobServer{
@@ -236,7 +248,7 @@ func TestInitJobDataPreviousJobWithJobDataPre105(t *testing.T) {
 	}

 	now := time.Now()
-	worker.initJobData(logger, job, now)
+	worker.initJobData(request.EmptyContext(logger), logger, job, now)

 	assert.Equal(t, model.ComplianceExportTypeActiance, job.Data[shared.JobDataExportType])
 	assert.Equal(t, strconv.Itoa(*worker.jobServer.Config().MessageExportSettings.BatchSize), job.Data[shared.JobDataBatchSize])
@@ -273,7 +285,10 @@ func TestDoJobNoPostsToExport(t *testing.T) {
 	mockMetrics.On("IncrementJobActive", model.JobTypeMessageExport)

 	// no previous job, data will be loaded from config
-	mockStore.JobStore.On("GetNewestJobByStatusesAndType", []string{model.JobStatusWarning, model.JobStatusSuccess}, model.JobTypeMessageExport).Return(nil, errors.New("test"))
+	mockStore.JobStore.On("GetAllByTypesAndStatusesPage", mock.Anything,
+		[]string{model.JobTypeMessageExport},
+		[]string{model.JobStatusWarning, model.JobStatusSuccess},
+		0, DefaultPreviousJobPageSize).Return(nil, errors.New("test"))

 	// no channels with activity
 	mockStore.ChannelMemberHistoryStore.On("GetChannelsWithActivityDuring", mock.Anything, mock.Anything).
@@ -356,7 +371,10 @@ func TestDoJobWithDedicatedExportBackend(t *testing.T) {
 	mockMetrics.On("IncrementJobActive", model.JobTypeMessageExport)

 	// no previous job, data will be loaded from config
-	mockStore.JobStore.On("GetNewestJobByStatusesAndType", []string{model.JobStatusWarning, model.JobStatusSuccess}, model.JobTypeMessageExport).Return(nil, errors.New("test"))
+	mockStore.JobStore.On("GetAllByTypesAndStatusesPage", mock.Anything,
+		[]string{model.JobTypeMessageExport},
+		[]string{model.JobStatusWarning, model.JobStatusSuccess},
+		0, DefaultPreviousJobPageSize).Return(nil, errors.New("test"))

 	channelId := st.NewTestID()
 	channelName := st.NewTestID()
@@ -521,7 +539,10 @@ func TestDoJobCancel(t *testing.T) {
 	mockMetrics.On("IncrementJobActive", model.JobTypeMessageExport)

 	// No previous job, data will be loaded from config
-	mockStore.JobStore.On("GetNewestJobByStatusesAndType", []string{model.JobStatusWarning, model.JobStatusSuccess}, model.JobTypeMessageExport).Return(nil, errors.New("test"))
+	mockStore.JobStore.On("GetAllByTypesAndStatusesPage", mock.Anything,
+		[]string{model.JobTypeMessageExport},
+		[]string{model.JobStatusWarning, model.JobStatusSuccess},
+		0, DefaultPreviousJobPageSize).Return(nil, errors.New("test"))

 	// Job updates the system console UI, once for getting channels, once for getting activity
 	mockStore.JobStore.On("UpdateOptimistically", mock.AnythingOfType("*model.Job"), model.JobStatusInProgress).Return(true, nil).Times(2)
@@ -571,3 +592,159 @@ func TestDoJobCancel(t *testing.T) {
 	// Cleanup
 	worker.Stop()
 }
+
+func TestGetPreviousJobNoJobs(t *testing.T) {
+	logger := mlog.CreateConsoleTestLogger(t)
+	mockStore := &storetest.Store{}
+	defer mockStore.AssertExpectations(t)
+
+	// Mock the job store to return empty jobs list
+	mockStore.JobStore.On("GetAllByTypesAndStatusesPage", mock.Anything,
+		[]string{model.JobTypeMessageExport},
+		[]string{model.JobStatusWarning, model.JobStatusSuccess},
+		0, DefaultPreviousJobPageSize).Return([]*model.Job{}, nil).Once()
+
+	worker := &MessageExportWorker{
+		jobServer: &jobs.JobServer{
+			Store: mockStore,
+		},
+		logger: logger,
+	}
+
+	rctx := request.EmptyContext(logger)
+	job, err := worker.getPreviousNonCliJob(rctx)
+
+	require.NoError(t, err)
+	assert.Nil(t, job, "Expected nil job when no jobs are returned")
+}
+
+func TestGetPreviousJobOneRegularJob(t *testing.T) {
+	logger := mlog.CreateConsoleTestLogger(t)
+	mockStore := &storetest.Store{}
+	defer mockStore.AssertExpectations(t)
+
+	regularJob := &model.Job{
+		Id:     st.NewTestID(),
+		Status: model.JobStatusSuccess,
+		Type:   model.JobTypeMessageExport,
+		Data:   map[string]string{},
+	}
+
+	// Mock the job store to return one regular job
+	mockStore.JobStore.On("GetAllByTypesAndStatusesPage", mock.Anything,
+		[]string{model.JobTypeMessageExport},
+		[]string{model.JobStatusWarning, model.JobStatusSuccess},
+		0, DefaultPreviousJobPageSize).Return([]*model.Job{regularJob}, nil).Once()
+
+	worker := &MessageExportWorker{
+		jobServer: &jobs.JobServer{
+			Store: mockStore,
+		},
+		logger: logger,
+	}
+
+	rctx := request.EmptyContext(logger)
+	job, err := worker.getPreviousNonCliJob(rctx)
+
+	require.NoError(t, err)
+	assert.Equal(t, regularJob.Id, job.Id, "Expected to get the regular job")
+}
+
+func TestGetPreviousJobOneMmctlJob(t *testing.T) {
+	logger := mlog.CreateConsoleTestLogger(t)
+	mockStore := &storetest.Store{}
+	defer mockStore.AssertExpectations(t)
+
+	mmctlJob := &model.Job{
+		Id:     st.NewTestID(),
+		Status: model.JobStatusSuccess,
+		Type:   model.JobTypeMessageExport,
+		Data:   map[string]string{shared.JobDataInitiatedBy: "mmctl"},
+	}
+
+	// Mock the job store to return only mmctl jobs (4 jobs, not a full page)
+	mockStore.JobStore.On("GetAllByTypesAndStatusesPage", mock.Anything,
+		[]string{model.JobTypeMessageExport},
+		[]string{model.JobStatusWarning, model.JobStatusSuccess},
+		0, DefaultPreviousJobPageSize).Return([]*model.Job{mmctlJob, mmctlJob, mmctlJob, mmctlJob}, nil).Once()
+
+	worker := &MessageExportWorker{
+		jobServer: &jobs.JobServer{
+			Store: mockStore,
+		},
+		logger: logger,
+	}
+
+	rctx := request.EmptyContext(logger)
+	job, err := worker.getPreviousNonCliJob(rctx)
+
+	require.NoError(t, err)
+	assert.Nil(t, job, "Expected nil job when only mmctl jobs are found")
+}
+
+func TestGetPreviousJobManyJobs(t *testing.T) {
+	logger := mlog.CreateConsoleTestLogger(t)
+	mockStore := &storetest.Store{}
+	defer mockStore.AssertExpectations(t)
+
+	// Create DefaultPageSize mmctl jobs for first page
+	firstPageJobs := make([]*model.Job, DefaultPreviousJobPageSize)
+	for i := range DefaultPreviousJobPageSize {
+		firstPageJobs[i] = &model.Job{
+			Id:     st.NewTestID(),
+			Status: model.JobStatusSuccess,
+			Type:   model.JobTypeMessageExport,
+			Data:   map[string]string{shared.JobDataInitiatedBy: "mmctl"},
+		}
+	}
+
+	// Create DefaultPageSize mmctl jobs for second page
+	secondPageJobs := make([]*model.Job, DefaultPreviousJobPageSize)
+	for i := range DefaultPreviousJobPageSize {
+		secondPageJobs[i] = &model.Job{
+			Id:     st.NewTestID(),
+			Status: model.JobStatusSuccess,
+			Type:   model.JobTypeMessageExport,
+			Data:   map[string]string{shared.JobDataInitiatedBy: "mmctl"},
+		}
+	}
+
+	// Create 1 regular job for the third page (last job)
+	regularJob := &model.Job{
+		Id:     st.NewTestID(),
+		Status: model.JobStatusSuccess,
+		Type:   model.JobTypeMessageExport,
+		Data:   map[string]string{},
+	}
+	thirdPageJobs := []*model.Job{regularJob}
+
+	// Mock the job store to return the jobs in pages
+	mockStore.JobStore.On("GetAllByTypesAndStatusesPage", mock.Anything,
+		[]string{model.JobTypeMessageExport},
+		[]string{model.JobStatusWarning, model.JobStatusSuccess},
+		0, DefaultPreviousJobPageSize).Return(firstPageJobs, nil).Once()
+
+	mockStore.JobStore.On("GetAllByTypesAndStatusesPage", mock.Anything,
+		[]string{model.JobTypeMessageExport},
+		[]string{model.JobStatusWarning, model.JobStatusSuccess},
+		1*DefaultPreviousJobPageSize, DefaultPreviousJobPageSize).Return(secondPageJobs, nil).Once()
+
+	mockStore.JobStore.On("GetAllByTypesAndStatusesPage", mock.Anything,
+		[]string{model.JobTypeMessageExport},
+		[]string{model.JobStatusWarning, model.JobStatusSuccess},
+		2*DefaultPreviousJobPageSize, DefaultPreviousJobPageSize).Return(thirdPageJobs, nil).Once()
+
+	worker := &MessageExportWorker{
+		jobServer: &jobs.JobServer{
+			Store: mockStore,
+		},
+		logger: logger,
+	}
+
+	rctx := request.EmptyContext(logger)
+	job, err := worker.getPreviousNonCliJob(rctx)
+
+	require.NoError(t, err)
+	assert.NotNil(t, job)
+	assert.Equal(t, regularJob.Id, job.Id, "Expected to find the regular job at the end")
+}