MM-53747: Create job to encode older image paths (#24073)

Bifrost now encodes all image paths. Due to this
one-way translation, we need to encode all the older
image paths as well.

After this is done, we can remove the double-lookup.

https://mattermost.atlassian.net/browse/MM-53747

```release-note
NONE
```

Co-authored-by: Mattermost Build <build@mattermost.com>

server/channels/app/migrations.go

@@ -6,6 +6,7 @@ package app
 import (
 	"context"
 	"fmt"
+	"os"
 	"reflect"
 
 	"github.com/mattermost/mattermost/server/public/model"
@@ -571,6 +572,35 @@ func (s *Server) doElasticsearchFixChannelIndex() {
 	}
 }
 
+func (s *Server) doCloudS3PathMigrations() {
+	// This migration is only applicable for cloud environments
+	if os.Getenv("MM_CLOUD_FILESTORE_BIFROST") == "" {
+		return
+	}
+
+	// If the migration is already marked as completed, don't do it again.
+	if _, err := s.Store().System().GetByName(model.MigrationKeyS3Path); err == nil {
+		return
+	}
+
+	// If there is a job already pending, no need to schedule again.
+	// This is possible if the pod was rolled over.
+	jobs, err := s.Store().Job().GetAllByTypeAndStatus(model.JobTypeS3PathMigration, model.JobStatusPending)
+	if err != nil {
+		mlog.Fatal("failed to get jobs by type and status", mlog.Err(err))
+		return
+	}
+	if len(jobs) > 0 {
+		return
+	}
+
+	if _, appErr := s.Jobs.CreateJob(model.JobTypeS3PathMigration, nil); appErr != nil {
+		mlog.Fatal("failed to start job for migrating s3 file paths", mlog.Err(appErr))
+		return
+	}
+
+}
+
 func (a *App) DoAppMigrations() {
 	a.Srv().doAppMigrations()
 }
@@ -593,4 +623,5 @@ func (s *Server) doAppMigrations() {
 	s.doRemainingSchemaMigrations()
 	s.doPostPriorityConfigDefaultTrueMigration()
 	s.doElasticsearchFixChannelIndex()
+	s.doCloudS3PathMigrations()
 }

server/channels/app/server.go

@@ -54,6 +54,7 @@ import (
 	"github.com/mattermost/mattermost/server/v8/channels/jobs/post_persistent_notifications"
 	"github.com/mattermost/mattermost/server/v8/channels/jobs/product_notices"
 	"github.com/mattermost/mattermost/server/v8/channels/jobs/resend_invitation_email"
+	"github.com/mattermost/mattermost/server/v8/channels/jobs/s3_path_migration"
 	"github.com/mattermost/mattermost/server/v8/channels/product"
 	"github.com/mattermost/mattermost/server/v8/channels/store"
 	"github.com/mattermost/mattermost/server/v8/channels/utils"
@@ -1568,6 +1569,11 @@ func (s *Server) initJobs() {
 		import_delete.MakeScheduler(s.Jobs),
 	)
 
+	s.Jobs.RegisterJobType(
+		model.JobTypeS3PathMigration,
+		s3_path_migration.MakeWorker(s.Jobs, s.Store(), s.FileBackend()),
+		nil)
+
 	s.Jobs.RegisterJobType(
 		model.JobTypeExportDelete,
 		export_delete.MakeWorker(s.Jobs, New(ServerConnector(s.Channels()))),

server/channels/jobs/s3_path_migration/s3_path_migration.go

@@ -0,0 +1,265 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package s3_path_migration
+
+import (
+	"errors"
+	"net/http"
+	"os"
+	"strconv"
+	"time"
+
+	"github.com/mattermost/mattermost/server/public/model"
+	"github.com/mattermost/mattermost/server/public/shared/mlog"
+	"github.com/mattermost/mattermost/server/v8/channels/jobs"
+	"github.com/mattermost/mattermost/server/v8/channels/store"
+	"github.com/mattermost/mattermost/server/v8/platform/shared/filestore"
+)
+
+const (
+	JobName = "S3PathMigration"
+
+	timeBetweenBatches = 1 * time.Second
+)
+
+type S3PathMigrationWorker struct {
+	name        string
+	jobServer   *jobs.JobServer
+	store       store.Store
+	fileBackend *filestore.S3FileBackend
+
+	stop    chan bool
+	stopped chan bool
+	jobs    chan model.Job
+}
+
+func MakeWorker(jobServer *jobs.JobServer, store store.Store, fileBackend filestore.FileBackend) model.Worker {
+	// If the type cast fails, it will be nil
+	// which is checked later.
+	s3Backend, _ := fileBackend.(*filestore.S3FileBackend)
+	worker := &S3PathMigrationWorker{
+		jobServer:   jobServer,
+		store:       store,
+		fileBackend: s3Backend,
+		name:        JobName,
+		stop:        make(chan bool, 1),
+		stopped:     make(chan bool, 1),
+		jobs:        make(chan model.Job),
+	}
+	return worker
+}
+
+func (worker *S3PathMigrationWorker) Run() {
+	mlog.Debug("Worker started", mlog.String("worker", worker.name))
+
+	defer func() {
+		mlog.Debug("Worker finished", mlog.String("worker", worker.name))
+		worker.stopped <- true
+	}()
+
+	for {
+		select {
+		case <-worker.stop:
+			mlog.Debug("Worker received stop signal", mlog.String("worker", worker.name))
+			return
+		case job := <-worker.jobs:
+			mlog.Debug("Worker received a new candidate job.", mlog.String("worker", worker.name))
+			worker.DoJob(&job)
+		}
+	}
+}
+
+func (worker *S3PathMigrationWorker) Stop() {
+	mlog.Debug("Worker stopping", mlog.String("worker", worker.name))
+	close(worker.stop)
+	<-worker.stopped
+}
+
+func (worker *S3PathMigrationWorker) JobChannel() chan<- model.Job {
+	return worker.jobs
+}
+
+func (worker *S3PathMigrationWorker) IsEnabled(_ *model.Config) bool {
+	return os.Getenv("MM_CLOUD_FILESTORE_BIFROST") != ""
+}
+
+func (worker *S3PathMigrationWorker) getJobMetadata(job *model.Job, key string) (int, *model.AppError) {
+	countStr := job.Data[key]
+	count := 0
+	var err error
+	if countStr != "" {
+		count, err = strconv.Atoi(countStr)
+		if err != nil {
+			return 0, model.NewAppError("getJobMetadata", model.NoTranslation, nil, "", http.StatusInternalServerError).Wrap(err)
+		}
+	}
+	return count, nil
+}
+
+func (worker *S3PathMigrationWorker) DoJob(job *model.Job) {
+	defer worker.jobServer.HandleJobPanic(job)
+
+	if claimed, err := worker.jobServer.ClaimJob(job); err != nil {
+		mlog.Warn("S3PathMigrationWorker experienced an error while trying to claim job",
+			mlog.String("worker", worker.name),
+			mlog.String("job_id", job.Id),
+			mlog.Err(err))
+		return
+	} else if !claimed {
+		return
+	}
+
+	if worker.fileBackend == nil {
+		err := errors.New("no S3 file backend found")
+		mlog.Error("S3PathMigrationWorker: ", mlog.Err(err))
+		worker.setJobError(job, model.NewAppError("DoJob", model.NoTranslation, nil, "", http.StatusInternalServerError).Wrap(err))
+		return
+	}
+
+	var appErr *model.AppError
+	// We get the job again because ClaimJob changes the job status.
+	job, appErr = worker.jobServer.GetJob(job.Id)
+	if appErr != nil {
+		mlog.Error("S3PathMigrationWorker: job execution error", mlog.String("worker", worker.name), mlog.String("job_id", job.Id), mlog.Err(appErr))
+		worker.setJobError(job, appErr)
+		return
+	}
+
+	// Check if there is metadata for that job.
+	// If there isn't, it will be empty by default, which is the right value.
+	startFileID := job.Data["start_file_id"]
+
+	doneCount, appErr := worker.getJobMetadata(job, "done_file_count")
+	if appErr != nil {
+		mlog.Error("S3PathMigrationWorker: failed to get done file count", mlog.String("worker", worker.name), mlog.String("job_id", job.Id), mlog.Err(appErr))
+		worker.setJobError(job, appErr)
+		return
+	}
+
+	startTime, appErr := worker.getJobMetadata(job, "start_create_at")
+	if appErr != nil {
+		mlog.Error("S3PathMigrationWorker: failed to get start create_at", mlog.String("worker", worker.name), mlog.String("job_id", job.Id), mlog.Err(appErr))
+		worker.setJobError(job, appErr)
+		return
+	}
+	if startTime == 0 {
+		// Time of the commit because we know no files older than that are affected.
+		// Exact commit was done on 09:54AM June 27, IST.
+		// We take June 26 as an approximation.
+		startTime = int(time.Date(2023, time.June, 26, 0, 0, 0, 0, time.UTC).UnixMilli())
+	}
+
+	const pageSize = 100
+
+	for {
+		select {
+		case <-worker.stop:
+			mlog.Info("Worker: S3 Migration has been canceled via Worker Stop. Setting the job back to pending.",
+				mlog.String("workername", worker.name),
+				mlog.String("job_id", job.Id))
+			if err := worker.jobServer.SetJobPending(job); err != nil {
+				mlog.Error("Worker: Failed to mark job as pending",
+					mlog.String("workername", worker.name),
+					mlog.String("job_id", job.Id),
+					mlog.Err(err))
+			}
+			return
+		case <-time.After(timeBetweenBatches):
+			var files []*model.FileForIndexing
+			tries := 0
+			for files == nil {
+				var err error
+				// Take batches of `pageSize`
+				files, err = worker.store.FileInfo().GetFilesBatchForIndexing(int64(startTime), startFileID, true, pageSize)
+				if err != nil {
+					if tries > 3 {
+						mlog.Error("Worker: Failed to get files after multiple retries. Exiting")
+						worker.setJobError(job, model.NewAppError("DoJob", model.NoTranslation, nil, "", http.StatusInternalServerError).Wrap(err))
+						return
+					}
+					mlog.Warn("Failed to get file info for s3 migration. Retrying .. ", mlog.Err(err))
+
+					// Wait a bit before trying again.
+					time.Sleep(15 * time.Second)
+				}
+
+				tries++
+			}
+
+			if len(files) == 0 {
+				mlog.Info("S3PathMigrationWorker: Job is complete", mlog.String("worker", worker.name), mlog.String("job_id", job.Id))
+				worker.setJobSuccess(job)
+				worker.markAsComplete()
+				return
+			}
+
+			// Iterate through the rows in each page.
+			for _, f := range files {
+				mlog.Debug("Processing file ID", mlog.String("id", f.Id), mlog.String("worker", worker.name), mlog.String("job_id", job.Id))
+				// We do not fail the job if a single image failed to encode.
+				if f.Path != "" {
+					if err := worker.fileBackend.DecodeFilePathIfNeeded(f.Path); err != nil {
+						mlog.Warn("Failed to encode S3 file path", mlog.String("path", f.Path), mlog.String("id", f.Id), mlog.Err(err))
+					}
+				}
+				if f.PreviewPath != "" {
+					if err := worker.fileBackend.DecodeFilePathIfNeeded(f.PreviewPath); err != nil {
+						mlog.Warn("Failed to encode S3 file path", mlog.String("path", f.PreviewPath), mlog.String("id", f.Id), mlog.Err(err))
+					}
+				}
+				if f.ThumbnailPath != "" {
+					if err := worker.fileBackend.DecodeFilePathIfNeeded(f.ThumbnailPath); err != nil {
+						mlog.Warn("Failed to encode S3 file path", mlog.String("path", f.ThumbnailPath), mlog.String("id", f.Id), mlog.Err(err))
+					}
+				}
+			}
+
+			// Work on each batch and save the batch starting ID in metadata
+			lastFile := files[len(files)-1]
+			startFileID = lastFile.Id
+			startTime = int(lastFile.CreateAt)
+
+			if job.Data == nil {
+				job.Data = make(model.StringMap)
+			}
+			job.Data["start_file_id"] = startFileID
+			job.Data["start_create_at"] = strconv.Itoa(startTime)
+			doneCount += len(files)
+			job.Data["done_file_count"] = strconv.Itoa(doneCount)
+		}
+	}
+}
+
+func (worker *S3PathMigrationWorker) markAsComplete() {
+	system := model.System{
+		Name:  model.MigrationKeyS3Path,
+		Value: "true",
+	}
+
+	// Note that if this fails, then the job would have still succeeded.
+	// So it will try to run the same job again next time, but then
+	// it will just fall through everything because all files would have
+	// converted. The actual job is idempotent, so there won't be a problem.
+	if err := worker.jobServer.Store.System().Save(&system); err != nil {
+		mlog.Error("Worker: Failed to mark s3 path migration as completed in the systems table.", mlog.String("workername", worker.name), mlog.Err(err))
+	}
+}
+
+func (worker *S3PathMigrationWorker) setJobSuccess(job *model.Job) {
+	if err := worker.jobServer.SetJobProgress(job, 100); err != nil {
+		mlog.Error("Worker: Failed to update progress for job", mlog.String("worker", worker.name), mlog.String("job_id", job.Id), mlog.Err(err))
+		worker.setJobError(job, err)
+	}
+
+	if err := worker.jobServer.SetJobSuccess(job); err != nil {
+		mlog.Error("S3PathMigrationWorker: Failed to set success for job", mlog.String("worker", worker.name), mlog.String("job_id", job.Id), mlog.Err(err))
+		worker.setJobError(job, err)
+	}
+}
+
+func (worker *S3PathMigrationWorker) setJobError(job *model.Job, appError *model.AppError) {
+	if err := worker.jobServer.SetJobError(job, appError); err != nil {
+		mlog.Error("S3PathMigrationWorker: Failed to set job error", mlog.String("worker", worker.name), mlog.String("job_id", job.Id), mlog.Err(err))
+	}
+}

server/channels/store/opentracinglayer/opentracinglayer.go

@@ -3608,7 +3608,7 @@ func (s *OpenTracingLayerFileInfoStore) GetByPath(path string) (*model.FileInfo,
 	return result, err
 }
 
-func (s *OpenTracingLayerFileInfoStore) GetFilesBatchForIndexing(startTime int64, startFileID string, limit int) ([]*model.FileForIndexing, error) {
+func (s *OpenTracingLayerFileInfoStore) GetFilesBatchForIndexing(startTime int64, startFileID string, includeDeleted bool, limit int) ([]*model.FileForIndexing, error) {
 	origCtx := s.Root.Store.Context()
 	span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "FileInfoStore.GetFilesBatchForIndexing")
 	s.Root.Store.SetContext(newCtx)
@@ -3617,7 +3617,7 @@ func (s *OpenTracingLayerFileInfoStore) GetFilesBatchForIndexing(startTime int64
 	}()
 
 	defer span.Finish()
-	result, err := s.FileInfoStore.GetFilesBatchForIndexing(startTime, startFileID, limit)
+	result, err := s.FileInfoStore.GetFilesBatchForIndexing(startTime, startFileID, includeDeleted, limit)
 	if err != nil {
 		span.LogFields(spanlog.Error(err))
 		ext.Error.Set(span, true)

server/channels/store/retrylayer/retrylayer.go

@@ -4034,11 +4034,11 @@ func (s *RetryLayerFileInfoStore) GetByPath(path string) (*model.FileInfo, error
 
 }
 
-func (s *RetryLayerFileInfoStore) GetFilesBatchForIndexing(startTime int64, startFileID string, limit int) ([]*model.FileForIndexing, error) {
+func (s *RetryLayerFileInfoStore) GetFilesBatchForIndexing(startTime int64, startFileID string, includeDeleted bool, limit int) ([]*model.FileForIndexing, error) {
 
 	tries := 0
 	for {
-		result, err := s.FileInfoStore.GetFilesBatchForIndexing(startTime, startFileID, limit)
+		result, err := s.FileInfoStore.GetFilesBatchForIndexing(startTime, startFileID, includeDeleted, limit)
 		if err == nil {
 			return result, nil
 		}

server/channels/store/sqlstore/file_info_store.go

@@ -691,9 +691,10 @@ func (fs SqlFileInfoStore) CountAll() (int64, error) {
 	return count, nil
 }
 
-func (fs SqlFileInfoStore) GetFilesBatchForIndexing(startTime int64, startFileID string, limit int) ([]*model.FileForIndexing, error) {
+func (fs SqlFileInfoStore) GetFilesBatchForIndexing(startTime int64, startFileID string, includeDeleted bool, limit int) ([]*model.FileForIndexing, error) {
 	files := []*model.FileForIndexing{}
-	sql, args, _ := fs.getQueryBuilder().
+
+	query := fs.getQueryBuilder().
 		Select(fs.queryFields...).
 		From("FileInfo").
 		Where(sq.Or{
@@ -704,10 +705,13 @@ func (fs SqlFileInfoStore) GetFilesBatchForIndexing(startTime int64, startFileID
 			},
 		}).
 		OrderBy("FileInfo.CreateAt ASC, FileInfo.Id ASC").
-		Limit(uint64(limit)).
-		ToSql()
+		Limit(uint64(limit))
 
-	err := fs.GetSearchReplicaX().Select(&files, sql, args...)
+	if !includeDeleted {
+		query = query.Where(sq.Eq{"FileInfo.DeleteAt": 0})
+	}
+
+	err := fs.GetSearchReplicaX().SelectBuilder(&files, query)
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to find Files")
 	}

server/channels/store/store.go

@@ -703,7 +703,7 @@ type FileInfoStore interface {
 	SetContent(fileID, content string) error
 	Search(paramsList []*model.SearchParams, userID, teamID string, page, perPage int) (*model.FileInfoList, error)
 	CountAll() (int64, error)
-	GetFilesBatchForIndexing(startTime int64, startFileID string, limit int) ([]*model.FileForIndexing, error)
+	GetFilesBatchForIndexing(startTime int64, startFileID string, includeDeleted bool, 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.

server/channels/store/storetest/file_info_store.go

@@ -748,21 +748,29 @@ func testFileInfoStoreGetFilesBatchForIndexing(t *testing.T, ss store.Store) {
 		ss.FileInfo().PermanentDelete(f3.Id)
 	}()
 
+	// Soft-deleting one file info
+	_, err = ss.FileInfo().DeleteForPost(f1.PostId)
+	require.NoError(t, err)
+
 	// Getting all
-	r, err := ss.FileInfo().GetFilesBatchForIndexing(f1.CreateAt-1, "", 100)
+	r, err := ss.FileInfo().GetFilesBatchForIndexing(f1.CreateAt-1, "", true, 100)
 	require.NoError(t, err)
 	require.Len(t, r, 3, "Expected 3 posts in results. Got %v", len(r))
 
-	// Testing pagination
-	r, err = ss.FileInfo().GetFilesBatchForIndexing(f1.CreateAt-1, "", 2)
+	r, err = ss.FileInfo().GetFilesBatchForIndexing(f1.CreateAt-1, "", false, 100)
 	require.NoError(t, err)
 	require.Len(t, r, 2, "Expected 2 posts in results. Got %v", len(r))
 
-	r, err = ss.FileInfo().GetFilesBatchForIndexing(r[1].CreateAt, r[1].Id, 2)
+	// Testing pagination
+	r, err = ss.FileInfo().GetFilesBatchForIndexing(f1.CreateAt-1, "", true, 2)
+	require.NoError(t, err)
+	require.Len(t, r, 2, "Expected 2 posts in results. Got %v", len(r))
+
+	r, err = ss.FileInfo().GetFilesBatchForIndexing(r[1].CreateAt, r[1].Id, true, 2)
 	require.NoError(t, err)
 	require.Len(t, r, 1, "Expected 1 post in results. Got %v", len(r))
 
-	r, err = ss.FileInfo().GetFilesBatchForIndexing(r[0].CreateAt, r[0].Id, 2)
+	r, err = ss.FileInfo().GetFilesBatchForIndexing(r[0].CreateAt, r[0].Id, true, 2)
 	require.NoError(t, err)
 	require.Len(t, r, 0, "Expected 0 posts in results. Got %v", len(r))
 }

server/channels/store/storetest/mocks/FileInfoStore.go

@@ -159,25 +159,25 @@ func (_m *FileInfoStore) GetByPath(path string) (*model.FileInfo, error) {
 	return r0, r1
 }
 
-// GetFilesBatchForIndexing provides a mock function with given fields: startTime, startFileID, limit
-func (_m *FileInfoStore) GetFilesBatchForIndexing(startTime int64, startFileID string, limit int) ([]*model.FileForIndexing, error) {
-	ret := _m.Called(startTime, startFileID, limit)
+// GetFilesBatchForIndexing provides a mock function with given fields: startTime, startFileID, includeDeleted, limit
+func (_m *FileInfoStore) GetFilesBatchForIndexing(startTime int64, startFileID string, includeDeleted bool, limit int) ([]*model.FileForIndexing, error) {
+	ret := _m.Called(startTime, startFileID, includeDeleted, limit)
 
 	var r0 []*model.FileForIndexing
 	var r1 error
-	if rf, ok := ret.Get(0).(func(int64, string, int) ([]*model.FileForIndexing, error)); ok {
-		return rf(startTime, startFileID, limit)
+	if rf, ok := ret.Get(0).(func(int64, string, bool, int) ([]*model.FileForIndexing, error)); ok {
+		return rf(startTime, startFileID, includeDeleted, limit)
 	}
-	if rf, ok := ret.Get(0).(func(int64, string, int) []*model.FileForIndexing); ok {
-		r0 = rf(startTime, startFileID, limit)
+	if rf, ok := ret.Get(0).(func(int64, string, bool, int) []*model.FileForIndexing); ok {
+		r0 = rf(startTime, startFileID, includeDeleted, limit)
 	} else {
 		if ret.Get(0) != nil {
 			r0 = ret.Get(0).([]*model.FileForIndexing)
 		}
 	}
 
-	if rf, ok := ret.Get(1).(func(int64, string, int) error); ok {
-		r1 = rf(startTime, startFileID, limit)
+	if rf, ok := ret.Get(1).(func(int64, string, bool, int) error); ok {
+		r1 = rf(startTime, startFileID, includeDeleted, limit)
 	} else {
 		r1 = ret.Error(1)
 	}

server/channels/store/timerlayer/timerlayer.go

@@ -3303,10 +3303,10 @@ func (s *TimerLayerFileInfoStore) GetByPath(path string) (*model.FileInfo, error
 	return result, err
 }
 
-func (s *TimerLayerFileInfoStore) GetFilesBatchForIndexing(startTime int64, startFileID string, limit int) ([]*model.FileForIndexing, error) {
+func (s *TimerLayerFileInfoStore) GetFilesBatchForIndexing(startTime int64, startFileID string, includeDeleted bool, limit int) ([]*model.FileForIndexing, error) {
 	start := time.Now()
 
-	result, err := s.FileInfoStore.GetFilesBatchForIndexing(startTime, startFileID, limit)
+	result, err := s.FileInfoStore.GetFilesBatchForIndexing(startTime, startFileID, includeDeleted, limit)
 
 	elapsed := float64(time.Since(start)) / float64(time.Second)
 	if s.Root.Metrics != nil {