Implement content extraction job (#18022)

app/enterprise.go

@@ -139,6 +139,12 @@ func RegisterJobsExportDeleteInterface(f func(*Server) tjobs.ExportDeleteInterfa
 	jobsExportDeleteInterface = f
 }
 
+var jobsExtractContentInterface func(*Server) tjobs.ExtractContentInterface
+
+func RegisterJobsExtractContentInterface(f func(*Server) tjobs.ExtractContentInterface) {
+	jobsExtractContentInterface = f
+}
+
 var productNoticesJobInterface func(*Server) tjobs.ProductNoticesJobInterface
 
 func RegisterProductNoticesJobInterface(f func(*Server) tjobs.ProductNoticesJobInterface) {

app/job.go

@@ -96,7 +96,8 @@ func (a *App) SessionHasPermissionToCreateJob(session model.Session, job *model.
 		model.JobTypeImportDelete,
 		model.JobTypeExportProcess,
 		model.JobTypeExportDelete,
-		model.JobTypeCloud:
+		model.JobTypeCloud,
+		model.JobTypeExtractContent:
 		return a.SessionHasPermissionTo(session, model.PermissionManageJobs), model.PermissionManageJobs
 	}
 
@@ -126,7 +127,8 @@ func (a *App) SessionHasPermissionToReadJob(session model.Session, jobType strin
 		model.JobTypeImportDelete,
 		model.JobTypeExportProcess,
 		model.JobTypeExportDelete,
-		model.JobTypeCloud:
+		model.JobTypeCloud,
+		model.JobTypeExtractContent:
 		return a.SessionHasPermissionTo(session, model.PermissionReadJobs), model.PermissionReadJobs
 	}
 

app/server.go

@@ -2061,6 +2061,10 @@ func (s *Server) initJobs() {
 		s.Jobs.ResendInvitationEmails = jobsResendInvitationEmailInterface(s)
 	}
 
+	if jobsExtractContentInterface != nil {
+		s.Jobs.ExtractContent = jobsExtractContentInterface(s)
+	}
+
 	s.Jobs.InitWorkers()
 	s.Jobs.InitSchedulers()
 }

i18n/en.json

@@ -7494,6 +7494,18 @@
     "id": "error",
     "translation": "Error"
   },
+  {
+    "id": "extrac_content.worker.do_job.invalid_input.from",
+    "translation": "Invalid input value 'from'"
+  },
+  {
+    "id": "extrac_content.worker.do_job.invalid_input.to",
+    "translation": "Invalid input value 'to'"
+  },
+  {
+    "id": "extract_content.worker.do_job.file_info",
+    "translation": "Failed to get file information for content extraction."
+  },
   {
     "id": "group_not_associated_to_synced_team",
     "translation": "Group cannot be associated to the channel until it is first associated to the parent group-synced team."

imports/placeholder.go

@@ -36,4 +36,7 @@ import (
 
 	// This is a placeholder so this package can be imported in Team Edition when it will be otherwise empty.
 	_ "github.com/mattermost/mattermost-server/v6/jobs/resend_invitation_email"
+
+	// This is a placeholder so this package can be imported in Team Edition when it will be otherwise empty.
+	_ "github.com/mattermost/mattermost-server/v6/jobs/extract_content"
 )

jobs/extract_content/worker.go

@@ -0,0 +1,176 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package extract_content
+
+import (
+	"net/http"
+	"strconv"
+
+	"github.com/mattermost/mattermost-server/v6/app"
+	"github.com/mattermost/mattermost-server/v6/app/request"
+	"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/shared/mlog"
+)
+
+var ignoredFiles = map[string]bool{
+	"png": true, "jpg": true, "jpeg": true, "gif": true, "wmv": true,
+	"mpg": true, "mpeg": true, "mp3": true, "mp4": true, "ogg": true,
+	"ogv": true, "mov": true, "apk": true, "svg": true, "webm": true,
+	"mkv": true,
+}
+
+func init() {
+	app.RegisterJobsExtractContentInterface(func(s *app.Server) tjobs.ExtractContentInterface {
+		a := app.New(app.ServerConnector(s))
+		return &ExtractContentInterfaceImpl{a}
+	})
+}
+
+type ExtractContentInterfaceImpl struct {
+	app *app.App
+}
+
+type ExtractContentWorker struct {
+	name        string
+	stopChan    chan struct{}
+	stoppedChan chan struct{}
+	jobsChan    chan model.Job
+	jobServer   *jobs.JobServer
+	app         *app.App
+	appContext  *request.Context
+}
+
+func (i *ExtractContentInterfaceImpl) MakeWorker() model.Worker {
+	return &ExtractContentWorker{
+		name:        "ExtractContent",
+		stopChan:    make(chan struct{}),
+		stoppedChan: make(chan struct{}),
+		jobsChan:    make(chan model.Job),
+		jobServer:   i.app.Srv().Jobs,
+		app:         i.app,
+		appContext:  &request.Context{},
+	}
+}
+
+func (w *ExtractContentWorker) JobChannel() chan<- model.Job {
+	return w.jobsChan
+}
+
+func (w *ExtractContentWorker) 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 (w *ExtractContentWorker) Stop() {
+	mlog.Debug("Worker stopping", mlog.String("worker", w.name))
+	close(w.stopChan)
+	<-w.stoppedChan
+}
+
+func (w *ExtractContentWorker) 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
+	}
+
+	var err error
+	var fromTS int64 = 0
+	var toTS int64 = model.GetMillis()
+	if fromStr, ok := job.Data["from"]; ok {
+		if fromTS, err = strconv.ParseInt(fromStr, 10, 64); err != nil {
+			w.setJobError(job, model.NewAppError("ExtractContentWorker", "extrac_content.worker.do_job.invalid_input.from", nil, "", http.StatusBadRequest))
+			return
+		}
+		fromTS *= 1000
+	}
+	if toStr, ok := job.Data["to"]; ok {
+		if toTS, err = strconv.ParseInt(toStr, 10, 64); err != nil {
+			w.setJobError(job, model.NewAppError("ExtractContentWorker", "extrac_content.worker.do_job.invalid_input.to", nil, "", http.StatusBadRequest))
+			return
+		}
+		toTS *= 1000
+	}
+
+	var nFiles int
+	var nErrs int
+	for {
+		opts := model.GetFileInfosOptions{
+			Since:          fromTS,
+			SortBy:         model.FileinfoSortByCreated,
+			IncludeDeleted: false,
+		}
+		fileInfos, err := w.app.Srv().Store.FileInfo().GetWithOptions(0, 1000, &opts)
+		if err != nil {
+			w.setJobError(job, model.NewAppError("ExtractContentWorker", "extract_content.worker.do_job.file_info", nil, err.Error(), http.StatusInternalServerError))
+			return
+		}
+		if len(fileInfos) == 0 {
+			break
+		}
+		for _, fileInfo := range fileInfos {
+			if !ignoredFiles[fileInfo.Extension] {
+				mlog.Debug("extracting file", mlog.String("filename", fileInfo.Name), mlog.String("filepath", fileInfo.Path))
+				err = w.app.ExtractContentFromFileInfo(fileInfo)
+				if err != nil {
+					mlog.Warn("Failed to extract file content", mlog.Err(err), mlog.String("file_info_id", fileInfo.Id))
+					nErrs++
+				}
+				nFiles++
+			}
+		}
+		lastFileInfo := fileInfos[len(fileInfos)-1]
+		if lastFileInfo.CreateAt > toTS {
+			break
+		}
+		fromTS = lastFileInfo.CreateAt + 1
+	}
+
+	job.Data["errors"] = strconv.Itoa(nErrs)
+	job.Data["processed"] = strconv.Itoa(nFiles)
+	w.updateData(job)
+
+	mlog.Info("Worker: Job is complete", mlog.String("worker", w.name), mlog.String("job_id", job.Id))
+	w.setJobSuccess(job)
+}
+
+func (w *ExtractContentWorker) 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 *ExtractContentWorker) 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()))
+	}
+}
+
+func (w *ExtractContentWorker) updateData(job *model.Job) {
+	if err := w.app.Srv().Jobs.UpdateInProgressJobData(job); err != nil {
+		mlog.Error("Worker: Failed to update job data", mlog.String("worker", w.name), mlog.String("job_id", job.Id), mlog.String("error", err.Error()))
+	}
+}

jobs/interfaces/extract_content_interface.go

@@ -0,0 +1,12 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package interfaces
+
+import (
+	"github.com/mattermost/mattermost-server/v6/model"
+)
+
+type ExtractContentInterface interface {
+	MakeWorker() model.Worker
+}

jobs/jobs_watcher.go

@@ -191,6 +191,13 @@ func (watcher *Watcher) PollAndNotify() {
 				default:
 				}
 			}
+		} else if job.Type == model.JobTypeExtractContent {
+			if watcher.workers.ExtractContent != nil {
+				select {
+				case watcher.workers.ExtractContent.JobChannel() <- *job:
+				default:
+				}
+			}
 		}
 	}
 }

jobs/server.go

@@ -36,6 +36,7 @@ type JobServer struct {
 	ExportDelete            tjobs.ExportDeleteInterface
 	Cloud                   ejobs.CloudJobInterface
 	ResendInvitationEmails  ejobs.ResendInvitationEmailJobInterface
+	ExtractContent          tjobs.ExtractContentInterface
 
 	// mut is used to protect the following fields from concurrent access.
 	mut        sync.Mutex

jobs/workers.go

@@ -32,6 +32,7 @@ type Workers struct {
 	ExportDelete             model.Worker
 	Cloud                    model.Worker
 	ResendInvitationEmail    model.Worker
+	ExtractContent           model.Worker
 
 	listenerId string
 	running    bool
@@ -124,6 +125,10 @@ func (srv *JobServer) InitWorkers() error {
 		workers.ResendInvitationEmail = resendInvitationEmailInterface.MakeWorker()
 	}
 
+	if extractContentInterface := srv.ExtractContent; extractContentInterface != nil {
+		workers.ExtractContent = extractContentInterface.MakeWorker()
+	}
+
 	srv.workers = workers
 
 	return nil
@@ -202,6 +207,10 @@ func (workers *Workers) Start() {
 		go workers.ResendInvitationEmail.Run()
 	}
 
+	if workers.ExtractContent != nil {
+		go workers.ExtractContent.Run()
+	}
+
 	go workers.Watcher.Start()
 
 	workers.listenerId = workers.ConfigService.AddConfigListener(workers.handleConfigChange)
@@ -335,6 +344,10 @@ func (workers *Workers) Stop() {
 		workers.ResendInvitationEmail.Stop()
 	}
 
+	if workers.ExtractContent != nil {
+		workers.ExtractContent.Stop()
+	}
+
 	workers.running = false
 
 	mlog.Info("Stopped workers")

model/job.go

@@ -28,6 +28,7 @@ const (
 	JobTypeExportDelete                 = "export_delete"
 	JobTypeCloud                        = "cloud"
 	JobTypeResendInvitationEmail        = "resend_invitation_email"
+	JobTypeExtractContent               = "extract_content"
 
 	JobStatusPending         = "pending"
 	JobStatusInProgress      = "in_progress"
@@ -55,6 +56,7 @@ var AllJobTypes = [...]string{
 	JobTypeExportProcess,
 	JobTypeExportDelete,
 	JobTypeCloud,
+	JobTypeExtractContent,
 }
 
 type Job struct {
@@ -96,6 +98,7 @@ func (j *Job) IsValid() *AppError {
 	case JobTypeExportDelete:
 	case JobTypeCloud:
 	case JobTypeResendInvitationEmail:
+	case JobTypeExtractContent:
 	default:
 		return NewAppError("Job.IsValid", "model.job.is_valid.type.app_error", nil, "id="+j.Id, http.StatusBadRequest)
 	}