Gitea/services/migrations/migrate.go

// Copyright 2019 The Gitea Authors. All rights reserved.
// Copyright 2018 Jonas Franz. All rights reserved.
// SPDX-License-Identifier: MIT

package migrations

import (
	"context"
	"errors"
	"fmt"
	"net/url"
	"path/filepath"
	"strings"

	repo_model "gitea.dev/models/repo"
	system_model "gitea.dev/models/system"
	user_model "gitea.dev/models/user"
	"gitea.dev/modules/container"
	"gitea.dev/modules/egress"
	"gitea.dev/modules/git"
	"gitea.dev/modules/log"
	base "gitea.dev/modules/migration"
	"gitea.dev/modules/setting"
	"gitea.dev/modules/util"
)

// MigrateOptions is equal to base.MigrateOptions
type MigrateOptions = base.MigrateOptions

var factories []base.DownloaderFactory

// RegisterDownloaderFactory registers a downloader factory
func RegisterDownloaderFactory(factory base.DownloaderFactory) {
	factories = append(factories, factory)
}

// IsMigrateURLAllowed checks if an URL is allowed to be migrated from
func IsMigrateURLAllowed(remoteURL string, doer *user_model.User) error {
	// Remote address can be HTTP/HTTPS/Git URL or local path.
	u, err := url.Parse(remoteURL)
	if err != nil {
		return &git.ErrInvalidCloneAddr{IsURLError: true, Host: remoteURL}
	}

	if u.Scheme == "file" || u.Scheme == "" {
		if !doer.CanImportLocal() {
			return &git.ErrInvalidCloneAddr{Host: "<LOCAL_FILESYSTEM>", IsPermissionDenied: true, LocalPath: true}
		}
		isAbs := filepath.IsAbs(u.Host + u.Path)
		if !isAbs {
			return &git.ErrInvalidCloneAddr{Host: "<LOCAL_FILESYSTEM>", IsInvalidPath: true, LocalPath: true}
		}
		isDir, err := util.IsDir(u.Host + u.Path)
		if err != nil {
			log.Error("Unable to check if %s is a directory: %v", u.Host+u.Path, err)
			return err
		}
		if !isDir {
			return &git.ErrInvalidCloneAddr{Host: "<LOCAL_FILESYSTEM>", IsInvalidPath: true, LocalPath: true}
		}

		return nil
	}

	if u.Scheme == "git" && u.Port() != "" && (strings.Contains(remoteURL, "%0d") || strings.Contains(remoteURL, "%0a")) {
		return &git.ErrInvalidCloneAddr{Host: u.Host, IsURLError: true}
	}

	if u.Opaque != "" || u.Scheme != "" && u.Scheme != "http" && u.Scheme != "https" && u.Scheme != "git" {
		return &git.ErrInvalidCloneAddr{Host: u.Host, IsProtocolInvalid: true, IsPermissionDenied: true, IsURLError: true}
	}

	if err := egress.NewMigrationPolicy().CheckHostIPs(u); err != nil {
		return &git.ErrInvalidCloneAddr{Host: u.Hostname(), IsPermissionDenied: true}
	}
	return nil
}

// MigrateRepository migrate repository according MigrateOptions
func MigrateRepository(ctx context.Context, doer *user_model.User, ownerName string, opts base.MigrateOptions, messenger base.Messenger) (*repo_model.Repository, error) {
	err := IsMigrateURLAllowed(opts.CloneAddr, doer)
	if err != nil {
		return nil, err
	}
	if opts.LFS && len(opts.LFSEndpoint) > 0 {
		err := IsMigrateURLAllowed(opts.LFSEndpoint, doer)
		if err != nil {
			return nil, err
		}
	}
	downloader, err := newDownloader(ctx, ownerName, opts)
	if err != nil {
		return nil, err
	}

	uploader := NewGiteaLocalUploader(ctx, doer, ownerName, opts.RepoName)
	uploader.gitServiceType = opts.GitServiceType

	if err := migrateRepository(ctx, doer, downloader, uploader, opts, messenger); err != nil {
		if err1 := uploader.Rollback(); err1 != nil {
			log.Error("rollback failed: %v", err1)
		}
		noticeMsg := fmt.Sprintf("Migrate repository (%s/%s) from %s failed: %v", ownerName, opts.RepoName, util.SanitizeCredentialURLs(opts.OriginalURL), util.SanitizeErrorCredentialURLs(err))
		if err2 := system_model.CreateRepositoryNotice(noticeMsg); err2 != nil {
			log.Error("create repository notice failed: ", err2)
		}
		return nil, err
	}
	return uploader.repo, nil
}

func newDownloader(ctx context.Context, ownerName string, opts base.MigrateOptions) (base.Downloader, error) {
	var (
		downloader base.Downloader
		err        error
	)

	for _, factory := range factories {
		if factory.GitServiceType() == opts.GitServiceType {
			downloader, err = factory.New(ctx, opts)
			if err != nil {
				return nil, err
			}
			break
		}
	}

	if downloader == nil {
		opts.Wiki = true
		opts.Milestones = false
		opts.Labels = false
		opts.Releases = false
		opts.Comments = false
		opts.Issues = false
		opts.PullRequests = false
		downloader = NewPlainGitDownloader(ownerName, opts.RepoName, opts.CloneAddr)
		log.Trace("Will migrate from git: %s", opts.OriginalURL)
	}

	if setting.Migrations.MaxAttempts > 1 {
		downloader = base.NewRetryDownloader(downloader, setting.Migrations.MaxAttempts, setting.Migrations.RetryBackoff)
	}
	return downloader, nil
}

// migrateRepository will download information and then upload it to Uploader, this is a simple
// process for small repository. For a big repository, save all the data to disk
// before upload is better
func migrateRepository(ctx context.Context, doer *user_model.User, downloader base.Downloader, uploader base.Uploader, opts base.MigrateOptions, messenger base.Messenger) error {
	if messenger == nil {
		messenger = base.NilMessenger
	}

	repo, err := downloader.GetRepoInfo(ctx)
	if err != nil {
		if !base.IsErrNotSupported(err) {
			return err
		}
		log.Info("migrating repo infos is not supported, ignored")
	}
	repo.IsPrivate = opts.Private
	repo.IsMirror = opts.Mirror
	if opts.Description != "" {
		repo.Description = opts.Description
	}
	if repo.CloneURL, err = downloader.FormatCloneURL(opts, repo.CloneURL); err != nil {
		return err
	}

	// SECURITY: If the downloader is not a RepositoryRestorer then we need to recheck the CloneURL
	if _, ok := downloader.(*RepositoryRestorer); !ok {
		// Now the clone URL can be rewritten by the downloader so we must recheck
		if err := IsMigrateURLAllowed(repo.CloneURL, doer); err != nil {
			return err
		}

		// SECURITY: Ensure that we haven't been redirected from an external to a local filesystem
		// Now we know all of these must parse
		cloneAddrURL, _ := url.Parse(opts.CloneAddr)
		cloneURL, _ := url.Parse(repo.CloneURL)

		if cloneURL.Scheme == "file" || cloneURL.Scheme == "" {
			if cloneAddrURL.Scheme != "file" && cloneAddrURL.Scheme != "" {
				return errors.New("repo info has changed from external to local filesystem")
			}
		}

		// We don't actually need to check the OriginalURL as it isn't used anywhere
	}

	log.Trace("migrating git data from %s", util.SanitizeCredentialURLs(repo.CloneURL))
	messenger("repo.migrate.migrating_git")
	if err = uploader.CreateRepo(ctx, repo, opts); err != nil {
		return err
	}
	defer uploader.Close()

	log.Trace("migrating topics")
	messenger("repo.migrate.migrating_topics")
	topics, err := downloader.GetTopics(ctx)
	if err != nil {
		if !base.IsErrNotSupported(err) {
			return err
		}
		log.Warn("migrating topics is not supported, ignored")
	}
	if len(topics) != 0 {
		if err = uploader.CreateTopics(ctx, topics...); err != nil {
			return err
		}
	}

	if opts.Milestones {
		log.Trace("migrating milestones")
		messenger("repo.migrate.migrating_milestones")
		milestones, err := downloader.GetMilestones(ctx)
		if err != nil {
			if !base.IsErrNotSupported(err) {
				return err
			}
			log.Warn("migrating milestones is not supported, ignored")
		}
		msBatchSize := uploader.MaxBatchInsertSize("milestone")
		for len(milestones) > 0 {
			if len(milestones) < msBatchSize {
				msBatchSize = len(milestones)
			}

			if err := uploader.CreateMilestones(ctx, milestones[:msBatchSize]...); err != nil {
				return err
			}
			milestones = milestones[msBatchSize:]
		}
	}

	if opts.Labels {
		log.Trace("migrating labels")
		messenger("repo.migrate.migrating_labels")
		labels, err := downloader.GetLabels(ctx)
		if err != nil {
			if !base.IsErrNotSupported(err) {
				return err
			}
			log.Warn("migrating labels is not supported, ignored")
		}

		lbBatchSize := uploader.MaxBatchInsertSize("label")
		for len(labels) > 0 {
			if len(labels) < lbBatchSize {
				lbBatchSize = len(labels)
			}

			if err := uploader.CreateLabels(ctx, labels[:lbBatchSize]...); err != nil {
				return err
			}
			labels = labels[lbBatchSize:]
		}
	}

	if opts.Releases {
		log.Trace("migrating releases")
		messenger("repo.migrate.migrating_releases")
		releases, err := downloader.GetReleases(ctx)
		if err != nil {
			if !base.IsErrNotSupported(err) {
				return err
			}
			log.Warn("migrating releases is not supported, ignored")
		}

		relBatchSize := uploader.MaxBatchInsertSize("release")
		for len(releases) > 0 {
			if len(releases) < relBatchSize {
				relBatchSize = len(releases)
			}

			if err = uploader.CreateReleases(ctx, releases[:relBatchSize]...); err != nil {
				return err
			}
			releases = releases[relBatchSize:]
		}

		// Once all releases (if any) are inserted, sync any remaining non-release tags
		if err = uploader.SyncTags(ctx); err != nil {
			return err
		}
	}

	var (
		commentBatchSize = uploader.MaxBatchInsertSize("comment")
		reviewBatchSize  = uploader.MaxBatchInsertSize("review")
	)

	supportAllComments := downloader.SupportGetRepoComments()

	if opts.Issues {
		log.Trace("migrating issues and comments")
		messenger("repo.migrate.migrating_issues")
		issueBatchSize := uploader.MaxBatchInsertSize("issue")

		// because when the migrating is running, some issues maybe removed, so after the next page
		// some of issue maybe duplicated, so we need to record the inserted issue indexes
		mapInsertedIssueIndexes := container.Set[int64]{}
		for i := 1; ; i++ {
			issues, isEnd, err := downloader.GetIssues(ctx, i, issueBatchSize)
			if err != nil {
				if !base.IsErrNotSupported(err) {
					return err
				}
				log.Warn("migrating issues is not supported, ignored")
				break
			}
			for i := 0; i < len(issues); i++ {
				if mapInsertedIssueIndexes.Contains(issues[i].Number) {
					issues = append(issues[:i], issues[i+1:]...)
					i--
					continue
				}
				mapInsertedIssueIndexes.Add(issues[i].Number)
			}

			if err := uploader.CreateIssues(ctx, issues...); err != nil {
				return err
			}

			if opts.Comments && !supportAllComments {
				allComments := make([]*base.Comment, 0, commentBatchSize)
				for _, issue := range issues {
					log.Trace("migrating issue %d's comments", issue.Number)
					comments, _, err := downloader.GetComments(ctx, issue)
					if err != nil {
						if !base.IsErrNotSupported(err) {
							return err
						}
						log.Warn("migrating comments is not supported, ignored")
					}

					allComments = append(allComments, comments...)

					if len(allComments) >= commentBatchSize {
						if err = uploader.CreateComments(ctx, allComments[:commentBatchSize]...); err != nil {
							return err
						}

						allComments = allComments[commentBatchSize:]
					}
				}

				if len(allComments) > 0 {
					if err = uploader.CreateComments(ctx, allComments...); err != nil {
						return err
					}
				}
			}

			if isEnd {
				break
			}
		}
	}

	if opts.PullRequests {
		log.Trace("migrating pull requests and comments")
		messenger("repo.migrate.migrating_pulls")
		prBatchSize := uploader.MaxBatchInsertSize("pullrequest")
		mapInsertedPRIndexes := container.Set[int64]{}
		for i := 1; ; i++ {
			prs, isEnd, err := downloader.GetPullRequests(ctx, i, prBatchSize)
			if err != nil {
				if !base.IsErrNotSupported(err) {
					return err
				}
				log.Warn("migrating pull requests is not supported, ignored")
				break
			}
			for i := 0; i < len(prs); i++ {
				if mapInsertedPRIndexes.Contains(prs[i].Number) {
					prs = append(prs[:i], prs[i+1:]...)
					i--
					continue
				}
				mapInsertedPRIndexes.Add(prs[i].Number)
			}

			if err := uploader.CreatePullRequests(ctx, prs...); err != nil {
				return err
			}

			if opts.Comments {
				if !supportAllComments {
					// plain comments
					allComments := make([]*base.Comment, 0, commentBatchSize)
					for _, pr := range prs {
						log.Trace("migrating pull request %d's comments", pr.Number)
						comments, _, err := downloader.GetComments(ctx, pr)
						if err != nil {
							if !base.IsErrNotSupported(err) {
								return err
							}
							log.Warn("migrating comments is not supported, ignored")
						}

						allComments = append(allComments, comments...)

						if len(allComments) >= commentBatchSize {
							if err = uploader.CreateComments(ctx, allComments[:commentBatchSize]...); err != nil {
								return err
							}
							allComments = allComments[commentBatchSize:]
						}
					}
					if len(allComments) > 0 {
						if err = uploader.CreateComments(ctx, allComments...); err != nil {
							return err
						}
					}
				}

				// migrate reviews
				allReviews := make([]*base.Review, 0, reviewBatchSize)
				for _, pr := range prs {
					reviews, err := downloader.GetReviews(ctx, pr)
					if err != nil {
						if !base.IsErrNotSupported(err) {
							return err
						}
						log.Warn("migrating reviews is not supported, ignored")
						break
					}

					allReviews = append(allReviews, reviews...)

					if len(allReviews) >= reviewBatchSize {
						if err = uploader.CreateReviews(ctx, allReviews[:reviewBatchSize]...); err != nil {
							return err
						}
						allReviews = allReviews[reviewBatchSize:]
					}
				}
				if len(allReviews) > 0 {
					if err = uploader.CreateReviews(ctx, allReviews...); err != nil {
						return err
					}
				}
			}

			if isEnd {
				break
			}
		}
		if len(mapInsertedPRIndexes) > 0 {
			// The pull requests migrating process may created head branches in the base repository
			// because head repository maybe a fork one which will not be migrated. So that we need
			// to sync branches again.
			log.Trace("syncing branches after migrating pull requests")
			if err = uploader.SyncBranches(ctx); err != nil {
				return err
			}
		}
	}

	if opts.Comments && supportAllComments {
		log.Trace("migrating comments")
		for i := 1; ; i++ {
			comments, isEnd, err := downloader.GetAllComments(ctx, i, commentBatchSize)
			if err != nil {
				return err
			}

			if err := uploader.CreateComments(ctx, comments...); err != nil {
				return err
			}

			if isEnd {
				break
			}
		}
	}

	return uploader.Finish(ctx)
}