[MM-53428] Delete empty drafts on upsert (#24046)

* [MM-53428] Delete empty drafts on upsert * Add migrations to fix existing drafts * Fix CI * Delete empty drafts entirely from the DB * Fix lint * Implement batch migration for deleting drafts * Missing store layers * Add updated mock * Remove unnecessary test * PR feedback * Add check for cluster migration * Fix MySQL * Don't check for len<2 * Bit of PR feedback * Use query builder for parameters * PR feedback * More PR feedback * Merge'd * unit test GetLastCreateAtAndUserIdValuesForEmptyDraftsMigration * simplified builder interface * fix DeleteEmptyDraftsByCreateAtAndUserId for MySQL * rework as batch migration worker * fix typo * log ip address on version mismatches too * simplify reset semantics * remove trace log in favour of low spam * document parameters for clarity --------- Co-authored-by: Mattermost Build <build@mattermost.com> Co-authored-by: Jesse Hallam <jesse.hallam@gmail.com>
2023-10-12 10:52:10 -04:00
--- a/server/channels/jobs/batch_migration_worker.go
+++ b/server/channels/jobs/batch_migration_worker.go
@@ -0,0 +1,236 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package jobs
+
+import (
+	"net/http"
+	"time"
+
+	"github.com/mattermost/mattermost/server/public/model"
+	"github.com/mattermost/mattermost/server/public/shared/mlog"
+	"github.com/mattermost/mattermost/server/public/shared/request"
+	"github.com/mattermost/mattermost/server/v8/channels/store"
+)
+
+type BatchMigrationWorkerAppIFace interface {
+	GetClusterStatus() []*model.ClusterInfo
+}
+
+// BatchMigrationWorker processes database migration jobs in batches to help avoid table locks.
+//
+// It uses the jobs infrastructure to ensure only one node in the cluster runs the migration at
+// any given time, avoids running the migration until the cluster is uniform, and automatically
+// resets the migration if the cluster version diverges after starting.
+//
+// In principle, the job infrastructure is overkill for this kind of work, as there's a worker
+// created per migration. There's also complication with edge cases, like having to restart the
+// server in order to retry a failed migration job. Refactoring the job infrastructure is left as
+// a future exercise.
+type BatchMigrationWorker struct {
+	jobServer *JobServer
+	logger    mlog.LoggerIFace
+	store     store.Store
+	app       BatchMigrationWorkerAppIFace
+
+	stop    chan bool
+	stopped chan bool
+	jobs    chan model.Job
+
+	migrationKey       string
+	timeBetweenBatches time.Duration
+	doMigrationBatch   func(data model.StringMap, store store.Store) (model.StringMap, bool, error)
+}
+
+// MakeBatchMigrationWorker creates a worker to process the given migration batch function.
+func MakeBatchMigrationWorker(jobServer *JobServer, store store.Store, app BatchMigrationWorkerAppIFace, migrationKey string, timeBetweenBatches time.Duration, doMigrationBatch func(data model.StringMap, store store.Store) (model.StringMap, bool, error)) model.Worker {
+	worker := &BatchMigrationWorker{
+		jobServer:          jobServer,
+		logger:             jobServer.Logger().With(mlog.String("worker_name", migrationKey)),
+		store:              store,
+		app:                app,
+		stop:               make(chan bool, 1),
+		stopped:            make(chan bool, 1),
+		jobs:               make(chan model.Job),
+		migrationKey:       migrationKey,
+		timeBetweenBatches: timeBetweenBatches,
+		doMigrationBatch:   doMigrationBatch,
+	}
+	return worker
+}
+
+// Run starts the worker dedicated to the unique migration batch job it will be given to process.
+func (worker *BatchMigrationWorker) Run() {
+	worker.logger.Debug("Worker started")
+	// We have to re-assign the stop channel again, because
+	// it might happen that the job was restarted due to a config change.
+	worker.stop = make(chan bool, 1)
+
+	defer func() {
+		worker.logger.Debug("Worker finished")
+		worker.stopped <- true
+	}()
+
+	for {
+		select {
+		case <-worker.stop:
+			worker.logger.Debug("Worker received stop signal")
+			return
+		case job := <-worker.jobs:
+			worker.DoJob(&job)
+		}
+	}
+}
+
+// Stop interrupts the worker even if the migration has not yet completed.
+func (worker *BatchMigrationWorker) Stop() {
+	worker.logger.Debug("Worker stopping")
+	close(worker.stop)
+	<-worker.stopped
+}
+
+// JobChannel is the means by which the jobs infrastructure provides the worker the job to execute.
+func (worker *BatchMigrationWorker) JobChannel() chan<- model.Job {
+	return worker.jobs
+}
+
+// IsEnabled is always true for batch migrations.
+func (worker *BatchMigrationWorker) IsEnabled(_ *model.Config) bool {
+	return true
+}
+
+// checkIsClusterInSync returns true if all nodes in the cluster are running the same version,
+// logging a warning on the first mismatch found.
+func (worker *BatchMigrationWorker) checkIsClusterInSync() bool {
+	clusterStatus := worker.app.GetClusterStatus()
+	for i := 1; i < len(clusterStatus); i++ {
+		if clusterStatus[i].SchemaVersion != clusterStatus[0].SchemaVersion {
+			worker.logger.Warn(
+				"Worker: cluster not in sync",
+				mlog.String("schema_version_a", clusterStatus[0].SchemaVersion),
+				mlog.String("schema_version_b", clusterStatus[1].SchemaVersion),
+				mlog.String("server_ip_a", clusterStatus[0].IPAddress),
+				mlog.String("server_ip_b", clusterStatus[1].IPAddress),
+			)
+			return false
+		}
+	}
+
+	return true
+}
+
+// DoJob executes the job picked up through the job channel.
+//
+// Note that this is a lot of distracting machinery here to claim the job, then double check the
+// status, and keep the status up to date in line with job infrastrcuture semantics. Unless an
+// error occurs, this worker should hold onto the job until its completed.
+func (worker *BatchMigrationWorker) DoJob(job *model.Job) {
+	logger := worker.logger.With(mlog.Any("job", job))
+	logger.Debug("Worker received a new candidate job.")
+	defer worker.jobServer.HandleJobPanic(logger, job)
+
+	if claimed, err := worker.jobServer.ClaimJob(job); err != nil {
+		logger.Warn("Worker experienced an error while trying to claim job", mlog.Err(err))
+		return
+	} else if !claimed {
+		return
+	}
+
+	c := request.EmptyContext(logger)
+	var appErr *model.AppError
+
+	// We get the job again because ClaimJob changes the job status.
+	job, appErr = worker.jobServer.GetJob(c, job.Id)
+	if appErr != nil {
+		worker.logger.Error("Worker: job execution error", mlog.Err(appErr))
+		worker.setJobError(logger, job, appErr)
+		return
+	}
+
+	if job.Data == nil {
+		job.Data = make(model.StringMap)
+	}
+
+	for {
+		select {
+		case <-worker.stop:
+			logger.Info("Worker: Migration has been canceled via Worker Stop. Setting the job back to pending.")
+			if err := worker.jobServer.SetJobPending(job); err != nil {
+				worker.logger.Error("Worker: Failed to mark job as pending", mlog.Err(err))
+			}
+			return
+		case <-time.After(worker.timeBetweenBatches):
+			// Ensure the cluster remains in sync, otherwise we restart the job to
+			// ensure a complete migration. Technically, the cluster could go out of
+			// sync briefly within a batch, but we accept that risk.
+			if !worker.checkIsClusterInSync() {
+				worker.logger.Warn("Worker: Resetting job")
+				worker.resetJob(logger, job)
+				return
+			}
+
+			nextData, done, err := worker.doMigrationBatch(job.Data, worker.store)
+			if err != nil {
+				worker.logger.Error("Worker: Failed to do migration batch. Exiting", mlog.Err(err))
+				worker.setJobError(logger, job, model.NewAppError("doMigrationBatch", model.NoTranslation, nil, "", http.StatusInternalServerError).Wrap(err))
+				return
+			} else if done {
+				logger.Info("Worker: Job is complete")
+				worker.setJobSuccess(logger, job)
+				worker.markAsComplete()
+				return
+			}
+
+			job.Data = nextData
+
+			// Migrations currently don't support reporting meaningful progress.
+			worker.jobServer.SetJobProgress(job, 0)
+		}
+	}
+}
+
+// resetJob erases the data tracking the next batch to execute and returns the job status to
+// pending to allow the job infrastructure to requeue it.
+func (worker *BatchMigrationWorker) resetJob(logger mlog.LoggerIFace, job *model.Job) {
+	job.Data = nil
+	job.Progress = 0
+	job.Status = model.JobStatusPending
+
+	if _, err := worker.store.Job().UpdateOptimistically(job, model.JobStatusInProgress); err != nil {
+		worker.logger.Error("Worker: Failed to reset job data. May resume instead of restarting.", mlog.Err(err))
+	}
+}
+
+// setJobSuccess records the job as successful.
+func (worker *BatchMigrationWorker) setJobSuccess(logger mlog.LoggerIFace, job *model.Job) {
+	if err := worker.jobServer.SetJobProgress(job, 100); err != nil {
+		logger.Error("Worker: Failed to update progress for job", mlog.Err(err))
+		worker.setJobError(logger, job, err)
+	}
+
+	if err := worker.jobServer.SetJobSuccess(job); err != nil {
+		logger.Error("Worker: Failed to set success for job", mlog.Err(err))
+		worker.setJobError(logger, job, err)
+	}
+}
+
+// setJobError puts the job into an error state, preventing the job from running again.
+func (worker *BatchMigrationWorker) setJobError(logger mlog.LoggerIFace, job *model.Job, appError *model.AppError) {
+	if err := worker.jobServer.SetJobError(job, appError); err != nil {
+		logger.Error("Worker: Failed to set job error", mlog.Err(err))
+	}
+}
+
+// markAsComplete records a discrete migration key to prevent this job from ever running again.
+func (worker *BatchMigrationWorker) markAsComplete() {
+	system := model.System{
+		Name:  worker.migrationKey,
+		Value: "true",
+	}
+
+	// Note that if this fails, then the job would have still succeeded. We will spuriously
+	// run the job again in the future, but as migrations are idempotent it won't be an issue.
+	if err := worker.jobServer.Store.System().Save(&system); err != nil {
+		worker.logger.Error("Worker: Failed to mark migration as completed in the systems table.", mlog.String("migration_key", worker.migrationKey), mlog.Err(err))
+	}
+}
--- a/server/channels/jobs/batch_migration_worker_test.go
+++ b/server/channels/jobs/batch_migration_worker_test.go
@@ -0,0 +1,310 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package jobs_test
+
+import (
+	"strconv"
+	"testing"
+	"time"
+
+	"github.com/mattermost/mattermost/server/public/model"
+	"github.com/mattermost/mattermost/server/public/shared/mlog"
+	"github.com/mattermost/mattermost/server/public/shared/request"
+	"github.com/mattermost/mattermost/server/v8/channels/jobs"
+	"github.com/mattermost/mattermost/server/v8/channels/store"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+type MockApp struct {
+	clusterInfo []*model.ClusterInfo
+}
+
+func (ma MockApp) GetClusterStatus() []*model.ClusterInfo {
+	return ma.clusterInfo
+}
+
+func (ma *MockApp) SetInSync() {
+	ma.clusterInfo = nil
+	ma.clusterInfo = append(ma.clusterInfo, &model.ClusterInfo{
+		SchemaVersion: "a",
+	})
+	ma.clusterInfo = append(ma.clusterInfo, &model.ClusterInfo{
+		SchemaVersion: "a",
+	})
+}
+
+func (ma *MockApp) SetOutOfSync() {
+	ma.clusterInfo = nil
+	ma.clusterInfo = append(ma.clusterInfo, &model.ClusterInfo{
+		SchemaVersion: "a",
+	})
+	ma.clusterInfo = append(ma.clusterInfo, &model.ClusterInfo{
+		SchemaVersion: "b",
+	})
+}
+
+func TestBatchMigrationWorker(t *testing.T) {
+	waitDone := func(t *testing.T, done chan bool, msg string) {
+		t.Helper()
+
+		require.Eventually(t, func() bool {
+			select {
+			case <-done:
+				return true
+			default:
+				return false
+			}
+		}, 5*time.Second, 100*time.Millisecond, msg)
+	}
+
+	setupBatchWorker := func(t *testing.T, th *TestHelper, mockApp *MockApp, doMigrationBatch func(model.StringMap, store.Store) (model.StringMap, bool, error)) (model.Worker, *model.Job) {
+		t.Helper()
+
+		migrationKey := model.NewId()
+		timeBetweenBatches := 1 * time.Second
+
+		worker := jobs.MakeBatchMigrationWorker(
+			th.Server.Jobs,
+			th.Server.Store(),
+			mockApp,
+			migrationKey,
+			timeBetweenBatches,
+			doMigrationBatch,
+		)
+		th.Server.Jobs.RegisterJobType(migrationKey, worker, nil)
+
+		ctx := request.EmptyContext(mlog.CreateConsoleTestLogger(t))
+		job, appErr := th.Server.Jobs.CreateJob(ctx, migrationKey, nil)
+		require.Nil(t, appErr)
+
+		done := make(chan bool)
+		go func() {
+			defer close(done)
+			worker.Run()
+		}()
+
+		// When ending the test, ensure we wait for the worker to finish.
+		t.Cleanup(func() {
+			waitDone(t, done, "worker did not stop running")
+		})
+
+		// Give the worker time to start running
+		time.Sleep(500 * time.Millisecond)
+
+		return worker, job
+	}
+
+	stopWorker := func(t *testing.T, worker model.Worker) {
+		t.Helper()
+
+		stopped := make(chan bool, 1)
+		go func() {
+			worker.Stop()
+			close(stopped)
+		}()
+
+		waitDone(t, stopped, "worker did not stop")
+	}
+
+	waitForJobStatus := func(t *testing.T, th *TestHelper, job *model.Job, status string) {
+		t.Helper()
+
+		require.Eventuallyf(t, func() bool {
+			ctx := request.EmptyContext(mlog.CreateConsoleTestLogger(t))
+			actualJob, appErr := th.Server.Jobs.GetJob(ctx, job.Id)
+			require.Nil(t, appErr)
+			require.Equal(t, job.Id, actualJob.Id)
+
+			return actualJob.Status == status
+		}, 5*time.Second, 250*time.Millisecond, "job never transitioned to %s", status)
+	}
+
+	assertJobReset := func(t *testing.T, th *TestHelper, job *model.Job) {
+		ctx := request.EmptyContext(mlog.CreateConsoleTestLogger(t))
+		actualJob, appErr := th.Server.Jobs.GetJob(ctx, job.Id)
+		require.Nil(t, appErr)
+		assert.Empty(t, actualJob.Progress)
+		assert.Empty(t, actualJob.Data)
+	}
+
+	getBatchNumberFromData := func(t *testing.T, data model.StringMap) int {
+		t.Helper()
+
+		if data["batch_number"] == "" {
+			data["batch_number"] = "1"
+		}
+		batchNumber, err := strconv.Atoi(data["batch_number"])
+		require.NoError(t, err)
+
+		return batchNumber
+	}
+
+	getDataFromBatchNumber := func(batchNumber int) model.StringMap {
+		data := make(model.StringMap)
+		data["batch_number"] = strconv.Itoa(batchNumber)
+
+		return data
+	}
+
+	t.Run("clusters not in sync before first batch", func(t *testing.T) {
+		th := Setup(t).InitBasic()
+		defer th.TearDown()
+
+		mockApp := &MockApp{}
+		mockApp.SetOutOfSync()
+
+		var worker model.Worker
+		var job *model.Job
+		worker, job = setupBatchWorker(t, th, mockApp, func(model.StringMap, store.Store) (model.StringMap, bool, error) {
+			require.Fail(t, "migration batch should never run while clusters not in sync")
+
+			return nil, false, nil
+		})
+
+		// Give the worker time to start running
+		time.Sleep(500 * time.Millisecond)
+
+		// Queue the work to be done
+		worker.JobChannel() <- *job
+
+		waitForJobStatus(t, th, job, model.JobStatusPending)
+		assertJobReset(t, th, job)
+
+		stopWorker(t, worker)
+	})
+
+	t.Run("stop after first batch", func(t *testing.T) {
+		th := Setup(t).InitBasic()
+		defer th.TearDown()
+
+		mockApp := &MockApp{}
+
+		var worker model.Worker
+		var job *model.Job
+		worker, job = setupBatchWorker(t, th, mockApp, func(data model.StringMap, s store.Store) (model.StringMap, bool, error) {
+			batchNumber := getBatchNumberFromData(t, data)
+
+			require.Equal(t, 1, batchNumber, "only batch 1 should have run")
+
+			// Shut down the worker after the first batch to prevent subsequent ones.
+			go worker.Stop()
+
+			batchNumber++
+
+			return getDataFromBatchNumber(batchNumber), false, nil
+		})
+
+		// Queue the work to be done
+		worker.JobChannel() <- *job
+
+		waitForJobStatus(t, th, job, model.JobStatusPending)
+	})
+
+	t.Run("stop after second batch", func(t *testing.T) {
+		th := Setup(t).InitBasic()
+		defer th.TearDown()
+
+		mockApp := &MockApp{}
+
+		var worker model.Worker
+		var job *model.Job
+		worker, job = setupBatchWorker(t, th, mockApp, func(data model.StringMap, s store.Store) (model.StringMap, bool, error) {
+			batchNumber := getBatchNumberFromData(t, data)
+
+			require.LessOrEqual(t, batchNumber, 2, "only batches 1 and 2 should have run")
+
+			// Shut down the worker after the first batch to prevent subsequent ones.
+			go worker.Stop()
+			batchNumber++
+
+			return getDataFromBatchNumber(batchNumber), false, nil
+		})
+
+		// Queue the work to be done
+		worker.JobChannel() <- *job
+
+		waitForJobStatus(t, th, job, model.JobStatusPending)
+	})
+
+	t.Run("clusters not in sync after first batch", func(t *testing.T) {
+		th := Setup(t).InitBasic()
+		defer th.TearDown()
+
+		mockApp := &MockApp{}
+
+		var worker model.Worker
+		var job *model.Job
+		worker, job = setupBatchWorker(t, th, mockApp, func(data model.StringMap, s store.Store) (model.StringMap, bool, error) {
+			batchNumber := getBatchNumberFromData(t, data)
+			require.Equal(t, 1, batchNumber, "only batch 1 should have run")
+
+			mockApp.SetOutOfSync()
+			batchNumber++
+
+			return getDataFromBatchNumber(batchNumber), false, nil
+		})
+
+		// Give the worker time to start running
+		time.Sleep(500 * time.Millisecond)
+
+		// Queue the work to be done
+		worker.JobChannel() <- *job
+
+		waitForJobStatus(t, th, job, model.JobStatusPending)
+		assertJobReset(t, th, job)
+
+		stopWorker(t, worker)
+	})
+
+	t.Run("done after first batch", func(t *testing.T) {
+		th := Setup(t).InitBasic()
+		defer th.TearDown()
+
+		mockApp := &MockApp{}
+
+		var worker model.Worker
+		var job *model.Job
+		worker, job = setupBatchWorker(t, th, mockApp, func(data model.StringMap, s store.Store) (model.StringMap, bool, error) {
+			batchNumber := getBatchNumberFromData(t, data)
+			require.Equal(t, 1, batchNumber, "only batch 1 should have run")
+
+			// Shut down the worker after the first batch to prevent subsequent ones.
+			go worker.Stop()
+			batchNumber++
+
+			return getDataFromBatchNumber(batchNumber), true, nil
+		})
+
+		// Queue the work to be done
+		worker.JobChannel() <- *job
+
+		waitForJobStatus(t, th, job, model.JobStatusSuccess)
+	})
+
+	t.Run("done after three batches", func(t *testing.T) {
+		th := Setup(t).InitBasic()
+		defer th.TearDown()
+
+		mockApp := &MockApp{}
+
+		var worker model.Worker
+		var job *model.Job
+		worker, job = setupBatchWorker(t, th, mockApp, func(data model.StringMap, s store.Store) (model.StringMap, bool, error) {
+			batchNumber := getBatchNumberFromData(t, data)
+			require.LessOrEqual(t, batchNumber, 3, "only 3 batches should have run")
+
+			// Shut down the worker after the first batch to prevent subsequent ones.
+			go worker.Stop()
+			batchNumber++
+
+			return getDataFromBatchNumber(batchNumber), true, nil
+		})
+
+		// Queue the work to be done
+		worker.JobChannel() <- *job
+
+		waitForJobStatus(t, th, job, model.JobStatusSuccess)
+	})
+}
--- a/server/channels/jobs/delete_empty_drafts_migration/delete_empty_drafts_migration.go
+++ b/server/channels/jobs/delete_empty_drafts_migration/delete_empty_drafts_migration.go
@@ -0,0 +1,85 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package delete_empty_drafts_migration
+
+import (
+	"strconv"
+	"time"
+
+	"github.com/mattermost/mattermost/server/public/model"
+	"github.com/mattermost/mattermost/server/v8/channels/jobs"
+	"github.com/mattermost/mattermost/server/v8/channels/store"
+	"github.com/pkg/errors"
+)
+
+const (
+	timeBetweenBatches = 1 * time.Second
+)
+
+// MakeWorker creates a batch migration worker to delete empty drafts.
+func MakeWorker(jobServer *jobs.JobServer, store store.Store, app jobs.BatchMigrationWorkerAppIFace) model.Worker {
+	return jobs.MakeBatchMigrationWorker(
+		jobServer,
+		store,
+		app,
+		model.MigrationKeyDeleteEmptyDrafts,
+		timeBetweenBatches,
+		doDeleteEmptyDraftsMigrationBatch,
+	)
+}
+
+// parseJobMetadata parses the opaque job metadata to return the information needed to decide which
+// batch to process next.
+func parseJobMetadata(data model.StringMap) (int64, string, error) {
+	createAt := int64(0)
+	if data["create_at"] != "" {
+		parsedCreateAt, parseErr := strconv.ParseInt(data["create_at"], 10, 64)
+		if parseErr != nil {
+			return 0, "", errors.Wrap(parseErr, "failed to parse create_at")
+		}
+		createAt = parsedCreateAt
+	}
+
+	userID := data["user_id"]
+
+	return createAt, userID, nil
+}
+
+// makeJobMetadata encodes the information needed to decide which batch to process next back into
+// the opaque job metadata.
+func makeJobMetadata(createAt int64, userID string) model.StringMap {
+	data := make(model.StringMap)
+	data["create_at"] = strconv.FormatInt(createAt, 10)
+	data["user_id"] = userID
+
+	return data
+}
+
+// doDeleteEmptyDraftsMigrationBatch iterates through all drafts, deleting empty drafts within each
+// batch keyed by the compound primary key (createAt, userID)
+func doDeleteEmptyDraftsMigrationBatch(data model.StringMap, store store.Store) (model.StringMap, bool, error) {
+	createAt, userID, err := parseJobMetadata(data)
+	if err != nil {
+		return nil, false, errors.Wrap(err, "failed to parse job metadata")
+	}
+
+	// Determine the /next/ (createAt, userId) by finding the last record in the batch we're
+	// about to delete.
+	nextCreateAt, nextUserID, err := store.Draft().GetLastCreateAtAndUserIdValuesForEmptyDraftsMigration(createAt, userID)
+	if err != nil {
+		return nil, false, errors.Wrapf(err, "failed to get the next batch (create_at=%v, user_id=%v)", createAt, userID)
+	}
+
+	// If we get the nil values, it means the batch was empty and we're done.
+	if nextCreateAt == 0 && nextUserID == "" {
+		return nil, true, nil
+	}
+
+	err = store.Draft().DeleteEmptyDraftsByCreateAtAndUserId(createAt, userID)
+	if err != nil {
+		return nil, false, errors.Wrapf(err, "failed to delete empty drafts (create_at=%v, user_id=%v)", createAt, userID)
+	}
+
+	return makeJobMetadata(nextCreateAt, nextUserID), false, nil
+}
--- a/server/channels/jobs/delete_empty_drafts_migration/delete_empty_drafts_migration_test.go
+++ b/server/channels/jobs/delete_empty_drafts_migration/delete_empty_drafts_migration_test.go
@@ -0,0 +1,180 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package delete_empty_drafts_migration
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/mattermost/mattermost/server/public/model"
+	"github.com/mattermost/mattermost/server/v8/channels/store/storetest"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestJobMetadata(t *testing.T) {
+	t.Run("parse nil data", func(t *testing.T) {
+		var data model.StringMap
+		createAt, userID, err := parseJobMetadata(data)
+		require.NoError(t, err)
+		assert.Empty(t, createAt)
+		assert.Empty(t, userID)
+	})
+
+	t.Run("parse invalid create_at", func(t *testing.T) {
+		data := make(model.StringMap)
+		data["user_id"] = "user_id"
+		data["create_at"] = "invalid"
+		_, _, err := parseJobMetadata(data)
+		require.Error(t, err)
+	})
+
+	t.Run("parse valid", func(t *testing.T) {
+		data := make(model.StringMap)
+		data["user_id"] = "user_id"
+		data["create_at"] = "1695918431"
+
+		createAt, userID, err := parseJobMetadata(data)
+		require.NoError(t, err)
+		assert.EqualValues(t, 1695918431, createAt)
+		assert.Equal(t, "user_id", userID)
+	})
+
+	t.Run("parse/make", func(t *testing.T) {
+		data := makeJobMetadata(1695918431, "user_id")
+		assert.Equal(t, "1695918431", data["create_at"])
+		assert.Equal(t, "user_id", data["user_id"])
+
+		createAt, userID, err := parseJobMetadata(data)
+		require.NoError(t, err)
+		assert.EqualValues(t, 1695918431, createAt)
+		assert.Equal(t, "user_id", userID)
+	})
+}
+
+func TestDoDeleteEmptyDraftsMigrationBatch(t *testing.T) {
+	t.Run("invalid job metadata", func(t *testing.T) {
+		mockStore := &storetest.Store{}
+		t.Cleanup(func() {
+			mockStore.AssertExpectations(t)
+		})
+
+		data := make(model.StringMap)
+		data["user_id"] = "user_id"
+		data["create_at"] = "invalid"
+		data, done, err := doDeleteEmptyDraftsMigrationBatch(data, mockStore)
+		require.Error(t, err)
+		assert.False(t, done)
+		assert.Nil(t, data)
+	})
+
+	t.Run("failure getting next offset", func(t *testing.T) {
+		mockStore := &storetest.Store{}
+		t.Cleanup(func() {
+			mockStore.AssertExpectations(t)
+		})
+
+		createAt, userID := int64(1695920000), "user_id_1"
+		nextCreateAt, nextUserID := int64(0), ""
+
+		mockStore.DraftStore.On("GetLastCreateAtAndUserIdValuesForEmptyDraftsMigration", createAt, userID).Return(nextCreateAt, nextUserID, errors.New("failure"))
+
+		data, done, err := doDeleteEmptyDraftsMigrationBatch(makeJobMetadata(createAt, userID), mockStore)
+		require.EqualError(t, err, "failed to get the next batch (create_at=1695920000, user_id=user_id_1): failure")
+		assert.False(t, done)
+		assert.Nil(t, data)
+	})
+
+	t.Run("failure deleting batch", func(t *testing.T) {
+		mockStore := &storetest.Store{}
+		t.Cleanup(func() {
+			mockStore.AssertExpectations(t)
+		})
+
+		createAt, userID := int64(1695920000), "user_id_1"
+		nextCreateAt, nextUserID := int64(1695922034), "user_id_2"
+
+		mockStore.DraftStore.On("GetLastCreateAtAndUserIdValuesForEmptyDraftsMigration", createAt, userID).Return(nextCreateAt, nextUserID, nil)
+		mockStore.DraftStore.On("DeleteEmptyDraftsByCreateAtAndUserId", createAt, userID).Return(errors.New("failure"))
+
+		data, done, err := doDeleteEmptyDraftsMigrationBatch(makeJobMetadata(createAt, userID), mockStore)
+		require.EqualError(t, err, "failed to delete empty drafts (create_at=1695920000, user_id=user_id_1): failure")
+		assert.False(t, done)
+		assert.Nil(t, data)
+	})
+
+	t.Run("do first batch (nil job metadata)", func(t *testing.T) {
+		mockStore := &storetest.Store{}
+		t.Cleanup(func() {
+			mockStore.AssertExpectations(t)
+		})
+
+		createAt, userID := int64(0), ""
+		nextCreateAt, nextUserID := int64(1695922034), "user_id_2"
+
+		mockStore.DraftStore.On("GetLastCreateAtAndUserIdValuesForEmptyDraftsMigration", createAt, userID).Return(nextCreateAt, nextUserID, nil)
+		mockStore.DraftStore.On("DeleteEmptyDraftsByCreateAtAndUserId", createAt, userID).Return(nil)
+
+		data, done, err := doDeleteEmptyDraftsMigrationBatch(nil, mockStore)
+		require.NoError(t, err)
+		assert.False(t, done)
+		assert.Equal(t, model.StringMap{
+			"create_at": "1695922034",
+			"user_id":   "user_id_2",
+		}, data)
+	})
+
+	t.Run("do first batch (empty job metadata)", func(t *testing.T) {
+		mockStore := &storetest.Store{}
+		t.Cleanup(func() {
+			mockStore.AssertExpectations(t)
+		})
+
+		createAt, userID := int64(0), ""
+		nextCreateAt, nextUserID := int64(1695922034), "user_id_2"
+
+		mockStore.DraftStore.On("GetLastCreateAtAndUserIdValuesForEmptyDraftsMigration", createAt, userID).Return(nextCreateAt, nextUserID, nil)
+		mockStore.DraftStore.On("DeleteEmptyDraftsByCreateAtAndUserId", createAt, userID).Return(nil)
+
+		data, done, err := doDeleteEmptyDraftsMigrationBatch(model.StringMap{}, mockStore)
+		require.NoError(t, err)
+		assert.False(t, done)
+		assert.Equal(t, makeJobMetadata(nextCreateAt, nextUserID), data)
+	})
+
+	t.Run("do batch", func(t *testing.T) {
+		mockStore := &storetest.Store{}
+		t.Cleanup(func() {
+			mockStore.AssertExpectations(t)
+		})
+
+		createAt, userID := int64(1695922000), "user_id_1"
+		nextCreateAt, nextUserID := int64(1695922034), "user_id_2"
+
+		mockStore.DraftStore.On("GetLastCreateAtAndUserIdValuesForEmptyDraftsMigration", createAt, userID).Return(nextCreateAt, nextUserID, nil)
+		mockStore.DraftStore.On("DeleteEmptyDraftsByCreateAtAndUserId", createAt, userID).Return(nil)
+
+		data, done, err := doDeleteEmptyDraftsMigrationBatch(makeJobMetadata(createAt, userID), mockStore)
+		require.NoError(t, err)
+		assert.False(t, done)
+		assert.Equal(t, makeJobMetadata(nextCreateAt, nextUserID), data)
+	})
+
+	t.Run("done batches", func(t *testing.T) {
+		mockStore := &storetest.Store{}
+		t.Cleanup(func() {
+			mockStore.AssertExpectations(t)
+		})
+
+		createAt, userID := int64(1695922000), "user_id_1"
+		nextCreateAt, nextUserID := int64(0), ""
+
+		mockStore.DraftStore.On("GetLastCreateAtAndUserIdValuesForEmptyDraftsMigration", createAt, userID).Return(nextCreateAt, nextUserID, nil)
+
+		data, done, err := doDeleteEmptyDraftsMigrationBatch(makeJobMetadata(createAt, userID), mockStore)
+		require.NoError(t, err)
+		assert.True(t, done)
+		assert.Nil(t, data)
+	})
+}
--- a/server/channels/jobs/helper_test.go
+++ b/server/channels/jobs/helper_test.go
@@ -0,0 +1,223 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package jobs_test
+
+import (
+	"os"
+	"path/filepath"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/mattermost/mattermost/server/public/model"
+	"github.com/mattermost/mattermost/server/public/shared/mlog"
+	"github.com/mattermost/mattermost/server/public/shared/request"
+	"github.com/mattermost/mattermost/server/v8/channels/app"
+	"github.com/mattermost/mattermost/server/v8/channels/store"
+	"github.com/mattermost/mattermost/server/v8/config"
+)
+
+type TestHelper struct {
+	App        *app.App
+	Context    *request.Context
+	Server     *app.Server
+	BasicTeam  *model.Team
+	BasicUser  *model.User
+	BasicUser2 *model.User
+
+	SystemAdminUser   *model.User
+	LogBuffer         *mlog.Buffer
+	TestLogger        *mlog.Logger
+	IncludeCacheLayer bool
+	ConfigStore       *config.Store
+
+	tempWorkspace string
+}
+
+func setupTestHelper(dbStore store.Store, enterprise bool, includeCacheLayer bool, options []app.Option, tb testing.TB) *TestHelper {
+	tempWorkspace, err := os.MkdirTemp("", "jobstest")
+	if err != nil {
+		panic(err)
+	}
+
+	configStore := config.NewTestMemoryStore()
+	memoryConfig := configStore.Get()
+	memoryConfig.SqlSettings = *mainHelper.GetSQLSettings()
+	*memoryConfig.PluginSettings.Directory = filepath.Join(tempWorkspace, "plugins")
+	*memoryConfig.PluginSettings.ClientDirectory = filepath.Join(tempWorkspace, "webapp")
+	*memoryConfig.PluginSettings.AutomaticPrepackagedPlugins = false
+	*memoryConfig.LogSettings.EnableSentry = false // disable error reporting during tests
+	*memoryConfig.AnnouncementSettings.AdminNoticesEnabled = false
+	*memoryConfig.AnnouncementSettings.UserNoticesEnabled = false
+	configStore.Set(memoryConfig)
+
+	buffer := &mlog.Buffer{}
+
+	options = append(options, app.ConfigStore(configStore))
+	if includeCacheLayer {
+		// Adds the cache layer to the test store
+		options = append(options, app.StoreOverrideWithCache(dbStore))
+	} else {
+		options = append(options, app.StoreOverride(dbStore))
+	}
+
+	testLogger, _ := mlog.NewLogger()
+	logCfg, _ := config.MloggerConfigFromLoggerConfig(&memoryConfig.LogSettings, nil, config.GetLogFileLocation)
+	if errCfg := testLogger.ConfigureTargets(logCfg, nil); errCfg != nil {
+		panic("failed to configure test logger: " + errCfg.Error())
+	}
+	if errW := mlog.AddWriterTarget(testLogger, buffer, true, mlog.StdAll...); errW != nil {
+		panic("failed to add writer target to test logger: " + errW.Error())
+	}
+	// lock logger config so server init cannot override it during testing.
+	testLogger.LockConfiguration()
+	options = append(options, app.SetLogger(testLogger))
+
+	s, err := app.NewServer(options...)
+	if err != nil {
+		panic(err)
+	}
+
+	th := &TestHelper{
+		App:               app.New(app.ServerConnector(s.Channels())),
+		Context:           request.EmptyContext(testLogger),
+		Server:            s,
+		LogBuffer:         buffer,
+		TestLogger:        testLogger,
+		IncludeCacheLayer: includeCacheLayer,
+		ConfigStore:       configStore,
+	}
+	th.Context.SetLogger(testLogger)
+
+	prevListenAddress := *th.App.Config().ServiceSettings.ListenAddress
+	th.App.UpdateConfig(func(cfg *model.Config) { *cfg.ServiceSettings.ListenAddress = "localhost:0" })
+	serverErr := th.Server.Start()
+	if serverErr != nil {
+		panic(serverErr)
+	}
+
+	th.App.UpdateConfig(func(cfg *model.Config) { *cfg.ServiceSettings.ListenAddress = prevListenAddress })
+
+	th.App.Srv().Store().MarkSystemRanUnitTests()
+
+	return th
+}
+
+func Setup(tb testing.TB, options ...app.Option) *TestHelper {
+	if testing.Short() {
+		tb.SkipNow()
+	}
+	dbStore := mainHelper.GetStore()
+	dbStore.DropAllTables()
+	dbStore.MarkSystemRanUnitTests()
+	mainHelper.PreloadMigrations()
+
+	return setupTestHelper(dbStore, false, true, options, tb)
+}
+
+var initBasicOnce sync.Once
+var userCache struct {
+	SystemAdminUser *model.User
+	BasicUser       *model.User
+	BasicUser2      *model.User
+}
+
+func (th *TestHelper) InitBasic() *TestHelper {
+	// create users once and cache them because password hashing is slow
+	initBasicOnce.Do(func() {
+		th.SystemAdminUser = th.CreateUser()
+		th.App.UpdateUserRoles(th.Context, th.SystemAdminUser.Id, model.SystemUserRoleId+" "+model.SystemAdminRoleId, false)
+		th.SystemAdminUser, _ = th.App.GetUser(th.SystemAdminUser.Id)
+		userCache.SystemAdminUser = th.SystemAdminUser.DeepCopy()
+
+		th.BasicUser = th.CreateUser()
+		th.BasicUser, _ = th.App.GetUser(th.BasicUser.Id)
+		userCache.BasicUser = th.BasicUser.DeepCopy()
+
+		th.BasicUser2 = th.CreateUser()
+		th.BasicUser2, _ = th.App.GetUser(th.BasicUser2.Id)
+		userCache.BasicUser2 = th.BasicUser2.DeepCopy()
+	})
+	// restore cached users
+	th.SystemAdminUser = userCache.SystemAdminUser.DeepCopy()
+	th.BasicUser = userCache.BasicUser.DeepCopy()
+	th.BasicUser2 = userCache.BasicUser2.DeepCopy()
+
+	users := []*model.User{th.SystemAdminUser, th.BasicUser, th.BasicUser2}
+	mainHelper.GetSQLStore().User().InsertUsers(users)
+
+	th.BasicTeam = th.CreateTeam()
+	return th
+}
+
+func (th *TestHelper) CreateTeam() *model.Team {
+	id := model.NewId()
+	team := &model.Team{
+		DisplayName: "dn_" + id,
+		Name:        "name" + id,
+		Email:       "success+" + id + "@simulator.amazonses.com",
+		Type:        model.TeamOpen,
+	}
+
+	var err *model.AppError
+	if team, err = th.App.CreateTeam(th.Context, team); err != nil {
+		panic(err)
+	}
+	return team
+}
+
+func (th *TestHelper) CreateUser() *model.User {
+	return th.CreateUserOrGuest(false)
+}
+
+func (th *TestHelper) CreateUserOrGuest(guest bool) *model.User {
+	id := model.NewId()
+
+	user := &model.User{
+		Email:         "success+" + id + "@simulator.amazonses.com",
+		Username:      "un_" + id,
+		Nickname:      "nn_" + id,
+		Password:      "Password1",
+		EmailVerified: true,
+	}
+
+	var err *model.AppError
+	if guest {
+		if user, err = th.App.CreateGuest(th.Context, user); err != nil {
+			panic(err)
+		}
+	} else {
+		if user, err = th.App.CreateUser(th.Context, user); err != nil {
+			panic(err)
+		}
+	}
+	return user
+}
+
+func (th *TestHelper) ShutdownApp() {
+	done := make(chan bool)
+	go func() {
+		th.Server.Shutdown()
+		close(done)
+	}()
+
+	select {
+	case <-done:
+	case <-time.After(30 * time.Second):
+		// panic instead of fatal to terminate all tests in this package, otherwise the
+		// still running App could spuriously fail subsequent tests.
+		panic("failed to shutdown App within 30 seconds")
+	}
+}
+
+func (th *TestHelper) TearDown() {
+	if th.IncludeCacheLayer {
+		// Clean all the caches
+		th.App.Srv().InvalidateAllCaches()
+	}
+	th.ShutdownApp()
+	if th.tempWorkspace != "" {
+		os.RemoveAll(th.tempWorkspace)
+	}
+}
--- a/server/channels/jobs/main_test.go
+++ b/server/channels/jobs/main_test.go
@@ -0,0 +1,24 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package jobs_test
+
+import (
+	"testing"
+
+	"github.com/mattermost/mattermost/server/v8/channels/testlib"
+)
+
+var mainHelper *testlib.MainHelper
+
+func TestMain(m *testing.M) {
+	var options = testlib.HelperOptions{
+		EnableStore:     true,
+		EnableResources: true,
+	}
+
+	mainHelper = testlib.NewMainHelperWithOptions(&options)
+	defer mainHelper.Close()
+
+	mainHelper.Main(m)
+}