Move task from modules to services (#17680)

1 files changed, 130 insertions, 0 deletions

diff --git a/services/task/task.go b/services/task/task.go
new file mode 100644
index 0000000000..f538b36efc
--- /dev/null
+++ b/services/task/task.go
@@ -0,0 +1,130 @@
+// Copyright 2019 Gitea. All rights reserved.
+// Use of this source code is governed by a MIT-style
+// license that can be found in the LICENSE file.
+
+package task
+
+import (
+	"fmt"
+
+	"code.gitea.io/gitea/models"
+	"code.gitea.io/gitea/modules/graceful"
+	"code.gitea.io/gitea/modules/json"
+	"code.gitea.io/gitea/modules/log"
+	base "code.gitea.io/gitea/modules/migration"
+	"code.gitea.io/gitea/modules/queue"
+	repo_module "code.gitea.io/gitea/modules/repository"
+	"code.gitea.io/gitea/modules/secret"
+	"code.gitea.io/gitea/modules/setting"
+	"code.gitea.io/gitea/modules/structs"
+	"code.gitea.io/gitea/modules/timeutil"
+	"code.gitea.io/gitea/modules/util"
+)
+
+// taskQueue is a global queue of tasks
+var taskQueue queue.Queue
+
+// Run a task
+func Run(t *models.Task) error {
+	switch t.Type {
+	case structs.TaskTypeMigrateRepo:
+		return runMigrateTask(t)
+	default:
+		return fmt.Errorf("Unknown task type: %d", t.Type)
+	}
+}
+
+// Init will start the service to get all unfinished tasks and run them
+func Init() error {
+	taskQueue = queue.CreateQueue("task", handle, &models.Task{})
+
+	if taskQueue == nil {
+		return fmt.Errorf("Unable to create Task Queue")
+	}
+
+	go graceful.GetManager().RunWithShutdownFns(taskQueue.Run)
+
+	return nil
+}
+
+func handle(data ...queue.Data) {
+	for _, datum := range data {
+		task := datum.(*models.Task)
+		if err := Run(task); err != nil {
+			log.Error("Run task failed: %v", err)
+		}
+	}
+}
+
+// MigrateRepository add migration repository to task
+func MigrateRepository(doer, u *models.User, opts base.MigrateOptions) error {
+	task, err := CreateMigrateTask(doer, u, opts)
+	if err != nil {
+		return err
+	}
+
+	return taskQueue.Push(task)
+}
+
+// CreateMigrateTask creates a migrate task
+func CreateMigrateTask(doer, u *models.User, opts base.MigrateOptions) (*models.Task, error) {
+	// encrypt credentials for persistence
+	var err error
+	opts.CloneAddrEncrypted, err = secret.EncryptSecret(setting.SecretKey, opts.CloneAddr)
+	if err != nil {
+		return nil, err
+	}
+	opts.CloneAddr = util.NewStringURLSanitizer(opts.CloneAddr, true).Replace(opts.CloneAddr)
+	opts.AuthPasswordEncrypted, err = secret.EncryptSecret(setting.SecretKey, opts.AuthPassword)
+	if err != nil {
+		return nil, err
+	}
+	opts.AuthPassword = ""
+	opts.AuthTokenEncrypted, err = secret.EncryptSecret(setting.SecretKey, opts.AuthToken)
+	if err != nil {
+		return nil, err
+	}
+	opts.AuthToken = ""
+	bs, err := json.Marshal(&opts)
+	if err != nil {
+		return nil, err
+	}
+
+	var task = &models.Task{
+		DoerID:         doer.ID,
+		OwnerID:        u.ID,
+		Type:           structs.TaskTypeMigrateRepo,
+		Status:         structs.TaskStatusQueue,
+		PayloadContent: string(bs),
+	}
+
+	if err := models.CreateTask(task); err != nil {
+		return nil, err
+	}
+
+	repo, err := repo_module.CreateRepository(doer, u, models.CreateRepoOptions{
+		Name:           opts.RepoName,
+		Description:    opts.Description,
+		OriginalURL:    opts.OriginalURL,
+		GitServiceType: opts.GitServiceType,
+		IsPrivate:      opts.Private,
+		IsMirror:       opts.Mirror,
+		Status:         models.RepositoryBeingMigrated,
+	})
+	if err != nil {
+		task.EndTime = timeutil.TimeStampNow()
+		task.Status = structs.TaskStatusFailed
+		err2 := task.UpdateCols("end_time", "status")
+		if err2 != nil {
+			log.Error("UpdateCols Failed: %v", err2.Error())
+		}
+		return nil, err
+	}
+
+	task.RepoID = repo.ID
+	if err = task.UpdateCols("repo_id"); err != nil {
+		return nil, err
+	}
+
+	return task, nil
+}