Working on refactoring jobs service (#19205)

* Working on refactoring jobs service * Making more consistent with the previous existing code * Remove no longer needed functions * Making a base PeridicScheduler to use it in most of the schedulers implementations * Removing accidental complexity from on of the jobs * Removing accidental complexity from expirynotify * Fixing compilation from previous commit * Remove accidental complexity from the export_delete job * Simplifying the workers by making a reusable worker * Using simple worker for export_delete job * Simpliying export process job * Simpliying extract content job * Simpliying import delete job * Simpliying import process job * Simpliying product noticies job * Simpliying fix crt channel unreads job (only removing the uneeded register function) * Simpliying migrations job (only removing the uneeded register function) * fixup * Simpliying plugins job (only removing the uneeded register function) * Simpliying bleve indexing job (only removing the uneeded register function) * Simpliying resend invitation email job (only removing the uneeded register function) * Fixing tests * Simplifying migration tests infrastructure * Adding missed license to files * Adding an empty file to imports package to ensure this package exist even without enterprise repo * Regenerating einterfaces mocks * Adding missed license to files * Updating i18n/en.json file * help fixing enterprise tests compilation * Adding new DailyScheduler * Fixing typo and changing the waitTime type for periodic sechduler * Making the daily scheduler more generic * Adding comments to clarify not used parameters in interface scheduler interface implementations * Using merror to handle multiple errors in jobs workers * Fixing linter errors * Addressing PR review comments * Reverting go.tools.mod changes * Removing the static check for worker type in the model (moving it to the insertion of new jobs * Moving migrations job to the jobs directory * Fixing (and improving a bit) tests * Apply suggestions from code review Co-authored-by: Doug Lauder <wiggin77@warpmail.net> * Fixing enterprise tests * Removing unneeded InitWorkers/InitSchedulers calls * Fix expirenotify job when error happens * Fixing govet errors Co-authored-by: Mattermod <mattermod@users.noreply.github.com> Co-authored-by: Doug Lauder <wiggin77@warpmail.net>
2022-02-14 18:21:18 +01:00
--- a/jobs/export_process/worker.go
+++ b/jobs/export_process/worker.go
@@ -7,133 +7,58 @@ import (
 	"io"
 	"path/filepath"

-	"github.com/mattermost/mattermost-server/v6/app"
 	"github.com/mattermost/mattermost-server/v6/jobs"
-	tjobs "github.com/mattermost/mattermost-server/v6/jobs/interfaces"
 	"github.com/mattermost/mattermost-server/v6/model"
+	"github.com/mattermost/mattermost-server/v6/services/configservice"
 	"github.com/mattermost/mattermost-server/v6/shared/mlog"
 )

-func init() {
-	app.RegisterJobsExportProcessInterface(func(s *app.Server) tjobs.ExportProcessInterface {
-		a := app.New(app.ServerConnector(s.Channels()))
-		return &ExportProcessInterfaceImpl{a}
-	})
+const jobName = "ExportProcess"
+
+type AppIface interface {
+	configservice.ConfigService
+	WriteFile(fr io.Reader, path string) (int64, *model.AppError)
+	BulkExport(writer io.Writer, outPath string, opts model.BulkExportOpts) *model.AppError
 }

-type ExportProcessInterfaceImpl struct {
-	app *app.App
-}
-
-type ExportProcessWorker struct {
-	name        string
-	stopChan    chan struct{}
-	stoppedChan chan struct{}
-	jobsChan    chan model.Job
-	jobServer   *jobs.JobServer
-	app         *app.App
-}
-
-func (i *ExportProcessInterfaceImpl) MakeWorker() model.Worker {
-	return &ExportProcessWorker{
-		name:        "ExportProcess",
-		stopChan:    make(chan struct{}),
-		stoppedChan: make(chan struct{}),
-		jobsChan:    make(chan model.Job),
-		jobServer:   i.app.Srv().Jobs,
-		app:         i.app,
-	}
-}
-
-func (w *ExportProcessWorker) JobChannel() chan<- model.Job {
-	return w.jobsChan
-}
-
-func (w *ExportProcessWorker) Run() {
-	mlog.Debug("Worker started", mlog.String("worker", w.name))
-
-	defer func() {
-		mlog.Debug("Worker finished", mlog.String("worker", w.name))
-		close(w.stoppedChan)
-	}()
-
-	for {
-		select {
-		case <-w.stopChan:
-			mlog.Debug("Worker received stop signal", mlog.String("worker", w.name))
-			return
-		case job := <-w.jobsChan:
-			mlog.Debug("Worker received a new candidate job.", mlog.String("worker", w.name))
-			w.doJob(&job)
+func MakeWorker(jobServer *jobs.JobServer, app AppIface) model.Worker {
+	isEnabled := func(cfg *model.Config) bool { return true }
+	execute := func(job *model.Job) error {
+		opts := model.BulkExportOpts{
+			CreateArchive: true,
 		}
+
+		includeAttachments, ok := job.Data["include_attachments"]
+		if ok && includeAttachments == "true" {
+			opts.IncludeAttachments = true
+		}
+
+		outPath := *app.Config().ExportSettings.Directory
+		exportFilename := model.NewId() + "_export.zip"
+
+		rd, wr := io.Pipe()
+
+		errCh := make(chan *model.AppError, 1)
+		go func() {
+			defer close(errCh)
+			_, appErr := app.WriteFile(rd, filepath.Join(outPath, exportFilename))
+			errCh <- appErr
+		}()
+
+		appErr := app.BulkExport(wr, outPath, opts)
+		if err := wr.Close(); err != nil {
+			mlog.Warn("Worker: error closing writer")
+		}
+
+		if appErr != nil {
+			return appErr
+		}
+
+		if appErr := <-errCh; appErr != nil {
+			return appErr
+		}
+		return nil
 	}
-}
-
-func (w *ExportProcessWorker) Stop() {
-	mlog.Debug("Worker stopping", mlog.String("worker", w.name))
-	close(w.stopChan)
-	<-w.stoppedChan
-}
-
-func (w *ExportProcessWorker) doJob(job *model.Job) {
-	if claimed, err := w.jobServer.ClaimJob(job); err != nil {
-		mlog.Warn("Worker experienced an error while trying to claim job",
-			mlog.String("worker", w.name),
-			mlog.String("job_id", job.Id),
-			mlog.String("error", err.Error()))
-		return
-	} else if !claimed {
-		return
-	}
-
-	opts := app.BulkExportOpts{
-		CreateArchive: true,
-	}
-
-	includeAttachments, ok := job.Data["include_attachments"]
-	if ok && includeAttachments == "true" {
-		opts.IncludeAttachments = true
-	}
-
-	outPath := *w.app.Config().ExportSettings.Directory
-	exportFilename := model.NewId() + "_export.zip"
-
-	rd, wr := io.Pipe()
-
-	errCh := make(chan *model.AppError, 1)
-	go func() {
-		defer close(errCh)
-		_, appErr := w.app.WriteFile(rd, filepath.Join(outPath, exportFilename))
-		errCh <- appErr
-	}()
-
-	appErr := w.app.BulkExport(wr, outPath, opts)
-	if err := wr.Close(); err != nil {
-		mlog.Warn("Worker: error closing writer")
-	}
-	if appErr != nil {
-		w.setJobError(job, appErr)
-		return
-	}
-
-	if appErr := <-errCh; appErr != nil {
-		w.setJobError(job, appErr)
-		return
-	}
-
-	mlog.Info("Worker: Job is complete", mlog.String("worker", w.name), mlog.String("job_id", job.Id))
-	w.setJobSuccess(job)
-}
-
-func (w *ExportProcessWorker) setJobSuccess(job *model.Job) {
-	if err := w.app.Srv().Jobs.SetJobSuccess(job); err != nil {
-		mlog.Error("Worker: Failed to set success for job", mlog.String("worker", w.name), mlog.String("job_id", job.Id), mlog.String("error", err.Error()))
-		w.setJobError(job, err)
-	}
-}
-
-func (w *ExportProcessWorker) setJobError(job *model.Job, appError *model.AppError) {
-	if err := w.app.Srv().Jobs.SetJobError(job, appError); err != nil {
-		mlog.Error("Worker: Failed to set job error", mlog.String("worker", w.name), mlog.String("job_id", job.Id), mlog.String("error", err.Error()))
-	}
+	worker := jobs.NewSimpleWorker(jobName, jobServer, execute, isEnabled)
+	return worker
 }