Move mirror push functions to a pushmirror service

The functions related push mirrors will be called by notification module, when sync-on-push is supported.
So relevant functions are moved to pushmirror service package to avoid circular imports.

Author: harryzcy

services/mirror/mirror_push.go

@@ -5,252 +5,16 @@
 package mirror
 
 import (
-	"context"
-	"errors"
-	"fmt"
-	"io"
-	"regexp"
-	"strings"
-	"time"
-
-	repo_model "code.gitea.io/gitea/models/repo"
-	"code.gitea.io/gitea/modules/git"
-	"code.gitea.io/gitea/modules/lfs"
-	"code.gitea.io/gitea/modules/log"
-	"code.gitea.io/gitea/modules/process"
-	"code.gitea.io/gitea/modules/repository"
-	"code.gitea.io/gitea/modules/setting"
-	"code.gitea.io/gitea/modules/timeutil"
-	"code.gitea.io/gitea/modules/util"
+	// Implementations related to push mirrors are in `services/pushmirror` package
+	// to avoid circular imports
+	"code.gitea.io/gitea/services/pushmirror"
 )
 
-var stripExitStatus = regexp.MustCompile(`exit status \d+ - `)
-
 // AddPushMirrorRemote registers the push mirror remote.
-func AddPushMirrorRemote(ctx context.Context, m *repo_model.PushMirror, addr string) error {
-	addRemoteAndConfig := func(addr, path string) error {
-		cmd := git.NewCommand(ctx, "remote", "add", "--mirror=push", m.RemoteName, addr)
-		if strings.Contains(addr, "://") && strings.Contains(addr, "@") {
-			cmd.SetDescription(fmt.Sprintf("remote add %s --mirror=push %s [repo_path: %s]", m.RemoteName, util.SanitizeCredentialURLs(addr), path))
-		} else {
-			cmd.SetDescription(fmt.Sprintf("remote add %s --mirror=push %s [repo_path: %s]", m.RemoteName, addr, path))
-		}
-		if _, _, err := cmd.RunStdString(&git.RunOpts{Dir: path}); err != nil {
-			return err
-		}
-		if _, _, err := git.NewCommand(ctx, "config", "--add", "remote."+m.RemoteName+".push", "+refs/heads/*:refs/heads/*").RunStdString(&git.RunOpts{Dir: path}); err != nil {
-			return err
-		}
-		if _, _, err := git.NewCommand(ctx, "config", "--add", "remote."+m.RemoteName+".push", "+refs/tags/*:refs/tags/*").RunStdString(&git.RunOpts{Dir: path}); err != nil {
-			return err
-		}
-		return nil
-	}
-
-	if err := addRemoteAndConfig(addr, m.Repo.RepoPath()); err != nil {
-		return err
-	}
-
-	if m.Repo.HasUncyclo() {
-		wikiRemoteURL := repository.UncycloRemoteURL(ctx, addr)
-		if len(wikiRemoteURL) > 0 {
-			if err := addRemoteAndConfig(wikiRemoteURL, m.Repo.UncycloPath()); err != nil {
-				return err
-			}
-		}
-	}
-
-	return nil
-}
+var AddPushMirrorRemote = pushmirror.AddPushMirrorRemote
 
 // RemovePushMirrorRemote removes the push mirror remote.
-func RemovePushMirrorRemote(ctx context.Context, m *repo_model.PushMirror) error {
-	cmd := git.NewCommand(ctx, "remote", "rm", m.RemoteName)
-	_ = m.GetRepository()
-
-	if _, _, err := cmd.RunStdString(&git.RunOpts{Dir: m.Repo.RepoPath()}); err != nil {
-		return err
-	}
-
-	if m.Repo.HasUncyclo() {
-		if _, _, err := cmd.RunStdString(&git.RunOpts{Dir: m.Repo.UncycloPath()}); err != nil {
-			// The wiki remote may not exist
-			log.Warn("Uncyclo Remote[%d] could not be removed: %v", m.ID, err)
-		}
-	}
-
-	return nil
-}
+var RemovePushMirrorRemote = pushmirror.RemovePushMirrorRemote
 
 // SyncPushMirror starts the sync of the push mirror and schedules the next run.
-func SyncPushMirror(ctx context.Context, mirrorID int64) bool {
-	log.Trace("SyncPushMirror [mirror: %d]", mirrorID)
-	defer func() {
-		err := recover()
-		if err == nil {
-			return
-		}
-		// There was a panic whilst syncPushMirror...
-		log.Error("PANIC whilst syncPushMirror[%d] Panic: %v\nStacktrace: %s", mirrorID, err, log.Stack(2))
-	}()
-
-	m, err := repo_model.GetPushMirrorByID(mirrorID)
-	if err != nil {
-		log.Error("GetPushMirrorByID [%d]: %v", mirrorID, err)
-		return false
-	}
-
-	_ = m.GetRepository()
-
-	m.LastError = ""
-
-	ctx, _, finished := process.GetManager().AddContext(ctx, fmt.Sprintf("Syncing PushMirror %s/%s to %s", m.Repo.OwnerName, m.Repo.Name, m.RemoteName))
-	defer finished()
-
-	log.Trace("SyncPushMirror [mirror: %d][repo: %-v]: Running Sync", m.ID, m.Repo)
-	err = runPushSync(ctx, m)
-	if err != nil {
-		log.Error("SyncPushMirror [mirror: %d][repo: %-v]: %v", m.ID, m.Repo, err)
-		m.LastError = stripExitStatus.ReplaceAllLiteralString(err.Error(), "")
-	}
-
-	m.LastUpdateUnix = timeutil.TimeStampNow()
-
-	if err := repo_model.UpdatePushMirror(m); err != nil {
-		log.Error("UpdatePushMirror [%d]: %v", m.ID, err)
-
-		return false
-	}
-
-	log.Trace("SyncPushMirror [mirror: %d][repo: %-v]: Finished", m.ID, m.Repo)
-
-	return err == nil
-}
-
-func runPushSync(ctx context.Context, m *repo_model.PushMirror) error {
-	timeout := time.Duration(setting.Git.Timeout.Mirror) * time.Second
-
-	performPush := func(path string) error {
-		remoteURL, err := git.GetRemoteURL(ctx, path, m.RemoteName)
-		if err != nil {
-			log.Error("GetRemoteAddress(%s) Error %v", path, err)
-			return errors.New("Unexpected error")
-		}
-
-		if setting.LFS.StartServer {
-			log.Trace("SyncMirrors [repo: %-v]: syncing LFS objects...", m.Repo)
-
-			gitRepo, err := git.OpenRepository(ctx, path)
-			if err != nil {
-				log.Error("OpenRepository: %v", err)
-				return errors.New("Unexpected error")
-			}
-			defer gitRepo.Close()
-
-			endpoint := lfs.DetermineEndpoint(remoteURL.String(), "")
-			lfsClient := lfs.NewClient(endpoint, nil)
-			if err := pushAllLFSObjects(ctx, gitRepo, lfsClient); err != nil {
-				return util.SanitizeErrorCredentialURLs(err)
-			}
-		}
-
-		log.Trace("Pushing %s mirror[%d] remote %s", path, m.ID, m.RemoteName)
-
-		if err := git.Push(ctx, path, git.PushOptions{
-			Remote:  m.RemoteName,
-			Force:   true,
-			Mirror:  true,
-			Timeout: timeout,
-		}); err != nil {
-			log.Error("Error pushing %s mirror[%d] remote %s: %v", path, m.ID, m.RemoteName, err)
-
-			return util.SanitizeErrorCredentialURLs(err)
-		}
-
-		return nil
-	}
-
-	err := performPush(m.Repo.RepoPath())
-	if err != nil {
-		return err
-	}
-
-	if m.Repo.HasUncyclo() {
-		wikiPath := m.Repo.UncycloPath()
-		_, err := git.GetRemoteAddress(ctx, wikiPath, m.RemoteName)
-		if err == nil {
-			err := performPush(wikiPath)
-			if err != nil {
-				return err
-			}
-		} else {
-			log.Trace("Skipping wiki: No remote configured")
-		}
-	}
-
-	return nil
-}
-
-func pushAllLFSObjects(ctx context.Context, gitRepo *git.Repository, lfsClient lfs.Client) error {
-	contentStore := lfs.NewContentStore()
-
-	pointerChan := make(chan lfs.PointerBlob)
-	errChan := make(chan error, 1)
-	go lfs.SearchPointerBlobs(ctx, gitRepo, pointerChan, errChan)
-
-	uploadObjects := func(pointers []lfs.Pointer) error {
-		err := lfsClient.Upload(ctx, pointers, func(p lfs.Pointer, objectError error) (io.ReadCloser, error) {
-			if objectError != nil {
-				return nil, objectError
-			}
-
-			content, err := contentStore.Get(p)
-			if err != nil {
-				log.Error("Error reading LFS object %v: %v", p, err)
-			}
-			return content, err
-		})
-		if err != nil {
-			select {
-			case <-ctx.Done():
-				return nil
-			default:
-			}
-		}
-		return err
-	}
-
-	var batch []lfs.Pointer
-	for pointerBlob := range pointerChan {
-		exists, err := contentStore.Exists(pointerBlob.Pointer)
-		if err != nil {
-			log.Error("Error checking if LFS object %v exists: %v", pointerBlob.Pointer, err)
-			return err
-		}
-		if !exists {
-			log.Trace("Skipping missing LFS object %v", pointerBlob.Pointer)
-			continue
-		}
-
-		batch = append(batch, pointerBlob.Pointer)
-		if len(batch) >= lfsClient.BatchSize() {
-			if err := uploadObjects(batch); err != nil {
-				return err
-			}
-			batch = nil
-		}
-	}
-	if len(batch) > 0 {
-		if err := uploadObjects(batch); err != nil {
-			return err
-		}
-	}
-
-	err, has := <-errChan
-	if has {
-		log.Error("Error enumerating LFS objects for repository: %v", err)
-		return err
-	}
-
-	return nil
-}
+var SyncPushMirror = pushmirror.SyncPushMirror