Refactor server/pipeline package

server/pipeline/approve.go

@@ -21,7 +21,6 @@ import (
 
 	"github.com/rs/zerolog/log"
 
-	"go.woodpecker-ci.org/woodpecker/v3/server"
 	forge_types "go.woodpecker-ci.org/woodpecker/v3/server/forge/types"
 	"go.woodpecker-ci.org/woodpecker/v3/server/model"
 	"go.woodpecker-ci.org/woodpecker/v3/server/store"
@@ -33,11 +32,9 @@ func Approve(ctx context.Context, store store.Store, currentPipeline *model.Pipe
 		return nil, ErrBadRequest{Msg: fmt.Sprintf("cannot approve a pipeline with status %s", currentPipeline.Status)}
 	}
 
-	forge, err := server.Config.Services.Manager.ForgeFromRepo(repo)
+	forge, err := loadForge(repo)
 	if err != nil {
-		msg := fmt.Sprintf("failure to load forge for repo '%s'", repo.FullName)
-		log.Error().Err(err).Str("repo", repo.FullName).Msg(msg)
-		return nil, errors.New(msg)
+		return nil, err
 	}
 
 	// fetch the pipeline file from the database
@@ -59,8 +56,8 @@ func Approve(ctx context.Context, store store.Store, currentPipeline *model.Pipe
 
 	currentPipeline, pipelineItems, parseErr, err := createPipelineItems(ctx, forge, store, currentPipeline, user, repo, yamls, nil, true)
 	if handleParseErrors(currentPipeline, parseErr) {
-		if err := updatePipelineWithErr(ctx, forge, store, currentPipeline, repo, user, parseErr); err != nil {
-			log.Error().Err(err).Msgf("error setting error status of pipeline for %s#%d after approval", repo.FullName, currentPipeline.Number)
+		if _, uErr := updatePipelineWithErr(ctx, forge, store, currentPipeline, repo, user, parseErr); uErr != nil {
+			log.Error().Err(uErr).Msgf("error setting error status of pipeline for %s#%d after approval", repo.FullName, currentPipeline.Number)
 		}
 		msg := fmt.Sprintf("failure to parse pipeline config for %s", repo.FullName)
 		log.Error().Err(parseErr).Msg(msg)
@@ -71,18 +68,9 @@ func Approve(ctx context.Context, store store.Store, currentPipeline *model.Pipe
 		return nil, err
 	}
 
-	if currentPipeline, err = UpdateToStatusPending(store, *currentPipeline, user.Login); err != nil {
+	if currentPipeline, err = UpdatePipelineToPending(store, *currentPipeline, user.Login); err != nil {
 		return nil, fmt.Errorf("error updating pipeline. %w", err)
 	}
 
-	publishPipeline(ctx, forge, currentPipeline, repo, user)
-
-	currentPipeline, err = start(ctx, forge, store, currentPipeline, user, repo, pipelineItems)
-	if err != nil {
-		msg := fmt.Sprintf("failure to start pipeline for %s: %v", repo.FullName, err)
-		log.Error().Err(err).Msg(msg)
-		return nil, errors.New(msg)
-	}
-
-	return currentPipeline, nil
+	return finishPipeline(ctx, forge, store, currentPipeline, user, repo, pipelineItems)
 }

server/pipeline/cancel.go

@@ -59,15 +59,15 @@ func Cancel(ctx context.Context, _forge forge.Forge, store store.Store, repo *mo
 	// Running ones will be set when the agents stop on the cancel signal
 	for _, workflow := range workflows {
 		if workflow.State == model.StatusPending {
-			if _, err = UpdateWorkflowToStatusSkipped(store, *workflow); err != nil {
+			if _, err = UpdateWorkflowToSkipped(store, *workflow); err != nil {
 				log.Error().Err(err).Msgf("cannot update workflow with id %d state", workflow.ID)
 			}
 		} else {
 			hasPendingOnly = false
 		}
 		for _, step := range workflow.Children {
 			if step.State == model.StatusPending {
-				if _, err = UpdateStepToStatusSkipped(store, *step, 0, model.StatusCanceled); err != nil {
+				if _, err = UpdateStepToSkipped(store, *step, 0, model.StatusCanceled); err != nil {
 					log.Error().Err(err).Msgf("cannot update workflow with id %d state", workflow.ID)
 				}
 			}
@@ -78,9 +78,9 @@ func Cancel(ctx context.Context, _forge forge.Forge, store store.Store, repo *mo
 	if hasPendingOnly {
 		plState = model.StatusCanceled
 	}
-	killedPipeline, err := UpdateToStatusKilled(store, *pipeline, cancelInfo, plState)
+	killedPipeline, err := UpdatePipelineToKilled(store, *pipeline, cancelInfo, plState)
 	if err != nil {
-		log.Error().Err(err).Msgf("UpdateToStatusKilled: %v", pipeline)
+		log.Error().Err(err).Msgf("UpdatePipelineToKilled: %v", pipeline)
 		return err
 	}
 

server/pipeline/create.go

@@ -49,11 +49,9 @@ func Create(ctx context.Context, _store store.Store, repo *model.Repo, pipeline
 		return nil, ErrFiltered
 	}
 
-	_forge, err := server.Config.Services.Manager.ForgeFromRepo(repo)
+	_forge, err := loadForge(repo)
 	if err != nil {
-		msg := fmt.Sprintf("failure to load forge for repo '%s'", repo.FullName)
-		log.Error().Err(err).Str("repo", repo.FullName).Msg(msg)
-		return nil, errors.New(msg)
+		return nil, err
 	}
 
 	// If the repoUser has a refresh token, the current access token
@@ -90,14 +88,21 @@ func Create(ctx context.Context, _store store.Store, repo *model.Repo, pipeline
 	case configFetchErr != nil:
 		// error while fetching config - not using the old config
 		log.Error().Str("repo", repo.FullName).Err(configFetchErr).Msgf("error while fetching config '%s' in '%s' with user: '%s', and did not get any config", repo.Config, pipeline.Ref, repoUser.Login)
-		return nil, updatePipelineWithErr(ctx, _forge, _store, pipeline, repo, repoUser, fmt.Errorf("could not load config from forge: %w", configFetchErr))
+		if _, err := updatePipelineWithErr(ctx, _forge, _store, pipeline, repo, repoUser, fmt.Errorf("could not load config from forge: %w", configFetchErr)); err != nil {
+			return nil, err
+		}
+		return nil, nil
 	}
 
 	currentPipeline, pipelineItems, parseErr, err := createPipelineItems(ctx, _forge, _store, pipeline, repoUser, repo, forgeYamlConfigs, nil, false)
-	*pipeline = *currentPipeline
+	pipeline = currentPipeline
 	if handleParseErrors(pipeline, parseErr) {
 		log.Debug().Str("repo", repo.FullName).Err(parseErr).Msg("failed to parse yaml")
-		return pipeline, updatePipelineWithErr(ctx, _forge, _store, pipeline, repo, repoUser, parseErr)
+		erroredPipeline, uErr := updatePipelineWithErr(ctx, _forge, _store, pipeline, repo, repoUser, parseErr)
+		if uErr != nil {
+			return pipeline, uErr
+		}
+		return erroredPipeline, nil
 	}
 	if err != nil {
 		return nil, fmt.Errorf("createPipelineItems failed: %w", err)
@@ -130,48 +135,40 @@ func Create(ctx context.Context, _store store.Store, repo *model.Repo, pipeline
 		return nil, errors.New(msg)
 	}
 
-	publishPipeline(ctx, _forge, pipeline, repo, repoUser)
-
-	if pipeline.Status == model.StatusBlocked {
-		return pipeline, nil
-	}
-
-	if err := updatePipelinePending(ctx, _forge, _store, pipeline, repo, repoUser); err != nil {
-		return nil, err
-	}
-
-	pipeline, err = start(ctx, _forge, _store, pipeline, repoUser, repo, pipelineItems)
-	if err != nil {
-		msg := fmt.Sprintf("failed to start pipeline for %s", repo.FullName)
-		log.Error().Err(err).Msg(msg)
-		return nil, errors.New(msg)
+	if pipeline.Status != model.StatusBlocked {
+		pipeline, err = updatePipelinePending(ctx, _forge, _store, pipeline, repo, repoUser)
+		if err != nil {
+			return nil, err
+		}
 	}
 
-	return pipeline, nil
+	return finishPipeline(ctx, _forge, _store, pipeline, repoUser, repo, pipelineItems)
 }
 
-func updatePipelineWithErr(ctx context.Context, _forge forge.Forge, _store store.Store, pipeline *model.Pipeline, repo *model.Repo, repoUser *model.User, err error) error {
-	_pipeline, err := UpdateToStatusError(_store, *pipeline, err)
-	if err != nil {
-		return err
+// updatePipelineWithErr moves the pipeline to the error state, persists it and
+// publishes the change. It returns the updated pipeline so the caller can use
+// the new value directly instead of relying on in-place mutation.
+func updatePipelineWithErr(ctx context.Context, _forge forge.Forge, _store store.Store, pipeline *model.Pipeline, repo *model.Repo, repoUser *model.User, err error) (*model.Pipeline, error) {
+	updated, uErr := UpdatePipelineToError(_store, *pipeline, err)
+	if uErr != nil {
+		return nil, uErr
 	}
-	// update value in ref
-	*pipeline = *_pipeline
 
-	publishPipeline(ctx, _forge, pipeline, repo, repoUser)
+	publishPipeline(ctx, _forge, updated, repo, repoUser)
 
-	return nil
+	return updated, nil
 }
 
-func updatePipelinePending(ctx context.Context, _forge forge.Forge, _store store.Store, pipeline *model.Pipeline, repo *model.Repo, repoUser *model.User) error {
-	_pipeline, err := UpdateToStatusPending(_store, *pipeline, "")
+// updatePipelinePending moves the pipeline to the pending state, persists it
+// and publishes the change. It returns the updated pipeline so the caller can
+// use the new value directly instead of relying on in-place mutation.
+func updatePipelinePending(ctx context.Context, _forge forge.Forge, _store store.Store, pipeline *model.Pipeline, repo *model.Repo, repoUser *model.User) (*model.Pipeline, error) {
+	updated, err := UpdatePipelineToPending(_store, *pipeline, "")
 	if err != nil {
-		return err
+		return nil, err
 	}
-	// update value in ref
-	*pipeline = *_pipeline
 
-	publishPipeline(ctx, _forge, pipeline, repo, repoUser)
+	publishPipeline(ctx, _forge, updated, repo, repoUser)
 
-	return nil
+	return updated, nil
 }

server/pipeline/decline.go

@@ -39,7 +39,7 @@ func Decline(ctx context.Context, store store.Store, pipeline *model.Pipeline, u
 		return nil, fmt.Errorf("cannot decline a pipeline with status %s", pipeline.Status)
 	}
 
-	pipeline, err = UpdateToStatusDeclined(store, *pipeline, user.Login)
+	pipeline, err = UpdatePipelineToDeclined(store, *pipeline, user.Login)
 	if err != nil {
 		return nil, fmt.Errorf("error updating pipeline. %w", err)
 	}
@@ -48,18 +48,8 @@ func Decline(ctx context.Context, store store.Store, pipeline *model.Pipeline, u
 		log.Error().Err(err).Msg("cannot build tree from step list")
 	}
 
-	for _, wf := range pipeline.Workflows {
-		wf.State = model.StatusDeclined
-		if err := store.WorkflowUpdate(wf); err != nil {
-			return nil, fmt.Errorf("error updating workflow. %w", err)
-		}
-
-		for _, step := range wf.Children {
-			step.State = model.StatusDeclined
-			if err := store.StepUpdate(step); err != nil {
-				return nil, fmt.Errorf("error updating step. %w", err)
-			}
-		}
+	if err := setWorkflowTreeState(store, pipeline, model.StatusDeclined); err != nil {
+		return nil, err
 	}
 
 	updatePipelineStatus(ctx, forge, pipeline, repo, user)

server/pipeline/pipeline_status.go

@@ -34,13 +34,13 @@ func PipelineStatus(workflows []*model.Workflow) model.StatusValue {
 	return status
 }
 
-func UpdateToStatusRunning(store store.Store, pipeline model.Pipeline, started int64) (*model.Pipeline, error) {
+func UpdatePipelineToRunning(store store.Store, pipeline model.Pipeline, started int64) (*model.Pipeline, error) {
 	pipeline.Status = model.StatusRunning
 	pipeline.Started = started
 	return &pipeline, store.UpdatePipeline(&pipeline)
 }
 
-func UpdateToStatusPending(store store.Store, pipeline model.Pipeline, reviewer string) (*model.Pipeline, error) {
+func UpdatePipelineToPending(store store.Store, pipeline model.Pipeline, reviewer string) (*model.Pipeline, error) {
 	if reviewer != "" {
 		pipeline.Reviewer = reviewer
 		pipeline.Reviewed = time.Now().Unix()
@@ -49,28 +49,28 @@ func UpdateToStatusPending(store store.Store, pipeline model.Pipeline, reviewer
 	return &pipeline, store.UpdatePipeline(&pipeline)
 }
 
-func UpdateToStatusDeclined(store store.Store, pipeline model.Pipeline, reviewer string) (*model.Pipeline, error) {
+func UpdatePipelineToDeclined(store store.Store, pipeline model.Pipeline, reviewer string) (*model.Pipeline, error) {
 	pipeline.Reviewer = reviewer
 	pipeline.Status = model.StatusDeclined
 	pipeline.Reviewed = time.Now().Unix()
 	return &pipeline, store.UpdatePipeline(&pipeline)
 }
 
-func UpdateStatusToDone(store store.Store, pipeline model.Pipeline, status model.StatusValue, stopped int64) (*model.Pipeline, error) {
+func UpdatePipelineToDone(store store.Store, pipeline model.Pipeline, status model.StatusValue, stopped int64) (*model.Pipeline, error) {
 	pipeline.Status = status
 	pipeline.Finished = stopped
 	return &pipeline, store.UpdatePipeline(&pipeline)
 }
 
-func UpdateToStatusError(store store.Store, pipeline model.Pipeline, err error) (*model.Pipeline, error) {
+func UpdatePipelineToError(store store.Store, pipeline model.Pipeline, err error) (*model.Pipeline, error) {
 	pipeline.Errors = errors.GetPipelineErrors(err)
 	pipeline.Status = model.StatusError
 	pipeline.Started = time.Now().Unix()
 	pipeline.Finished = pipeline.Started
 	return &pipeline, store.UpdatePipeline(&pipeline)
 }
 
-func UpdateToStatusKilled(store store.Store, pipeline model.Pipeline, cancelInfo *model.CancelInfo, state model.StatusValue) (*model.Pipeline, error) {
+func UpdatePipelineToKilled(store store.Store, pipeline model.Pipeline, cancelInfo *model.CancelInfo, state model.StatusValue) (*model.Pipeline, error) {
 	pipeline.Status = state
 	pipeline.Finished = time.Now().Unix()
 	pipeline.CancelInfo = cancelInfo

server/pipeline/pipeline_status_test.go

@@ -37,7 +37,7 @@ func mockStorePipeline(t *testing.T) store.Store {
 func TestUpdateToStatusRunning(t *testing.T) {
 	t.Parallel()
 
-	pipeline, _ := UpdateToStatusRunning(mockStorePipeline(t), model.Pipeline{}, int64(1))
+	pipeline, _ := UpdatePipelineToRunning(mockStorePipeline(t), model.Pipeline{}, int64(1))
 	assert.Equal(t, model.StatusRunning, pipeline.Status)
 	assert.EqualValues(t, 1, pipeline.Started)
 }
@@ -47,7 +47,7 @@ func TestUpdateToStatusPending(t *testing.T) {
 
 	now := time.Now().Unix()
 
-	pipeline, _ := UpdateToStatusPending(mockStorePipeline(t), model.Pipeline{}, "Reviewer")
+	pipeline, _ := UpdatePipelineToPending(mockStorePipeline(t), model.Pipeline{}, "Reviewer")
 
 	assert.Equal(t, model.StatusPending, pipeline.Status)
 	assert.Equal(t, "Reviewer", pipeline.Reviewer)
@@ -59,7 +59,7 @@ func TestUpdateToStatusDeclined(t *testing.T) {
 
 	now := time.Now().Unix()
 
-	pipeline, _ := UpdateToStatusDeclined(mockStorePipeline(t), model.Pipeline{}, "Reviewer")
+	pipeline, _ := UpdatePipelineToDeclined(mockStorePipeline(t), model.Pipeline{}, "Reviewer")
 
 	assert.Equal(t, model.StatusDeclined, pipeline.Status)
 	assert.Equal(t, "Reviewer", pipeline.Reviewer)
@@ -69,7 +69,7 @@ func TestUpdateToStatusDeclined(t *testing.T) {
 func TestUpdateToStatusToDone(t *testing.T) {
 	t.Parallel()
 
-	pipeline, _ := UpdateStatusToDone(mockStorePipeline(t), model.Pipeline{}, "status", int64(1))
+	pipeline, _ := UpdatePipelineToDone(mockStorePipeline(t), model.Pipeline{}, "status", int64(1))
 
 	assert.Equal(t, model.StatusValue("status"), pipeline.Status)
 	assert.EqualValues(t, 1, pipeline.Finished)
@@ -80,7 +80,7 @@ func TestUpdateToStatusError(t *testing.T) {
 
 	now := time.Now().Unix()
 
-	pipeline, _ := UpdateToStatusError(mockStorePipeline(t), model.Pipeline{}, errors.New("this is an error"))
+	pipeline, _ := UpdatePipelineToError(mockStorePipeline(t), model.Pipeline{}, errors.New("this is an error"))
 
 	assert.Len(t, pipeline.Errors, 1)
 	assert.Equal(t, "[generic] this is an error", pipeline.Errors[0].Error())
@@ -98,7 +98,7 @@ func TestUpdateToStatusKilled(t *testing.T) {
 		SupersededBy: 2,
 	}
 
-	pipeline, _ := UpdateToStatusKilled(mockStorePipeline(t), model.Pipeline{}, cancelInfo, model.StatusKilled)
+	pipeline, _ := UpdatePipelineToKilled(mockStorePipeline(t), model.Pipeline{}, cancelInfo, model.StatusKilled)
 
 	assert.Equal(t, model.StatusKilled, pipeline.Status)
 	assert.NotNil(t, pipeline.CancelInfo)

server/pipeline/restart.go

@@ -30,11 +30,9 @@ import (
 
 // Restart a pipeline by creating a new one out of the old and start it.
 func Restart(ctx context.Context, store store.Store, lastPipeline *model.Pipeline, user *model.User, repo *model.Repo, envs map[string]string) (*model.Pipeline, error) {
-	forge, err := server.Config.Services.Manager.ForgeFromRepo(repo)
+	forge, err := loadForge(repo)
 	if err != nil {
-		msg := fmt.Sprintf("failure to load forge for repo '%s'", repo.FullName)
-		log.Error().Err(err).Str("repo", repo.FullName).Msg(msg)
-		return nil, errors.New(msg)
+		return nil, err
 	}
 
 	if lastPipeline.Status == model.StatusBlocked {
@@ -75,7 +73,7 @@ func Restart(ctx context.Context, store store.Store, lastPipeline *model.Pipelin
 	}
 
 	if len(configs) == 0 {
-		newPipeline, uErr := UpdateToStatusError(store, *newPipeline, errors.New("pipeline definition not found"))
+		newPipeline, uErr := UpdatePipelineToError(store, *newPipeline, errors.New("pipeline definition not found"))
 		if uErr != nil {
 			log.Debug().Err(uErr).Msg("failure to update pipeline status")
 		} else {
@@ -91,7 +89,7 @@ func Restart(ctx context.Context, store store.Store, lastPipeline *model.Pipelin
 
 	newPipeline, pipelineItems, parseErr, err := createPipelineItems(ctx, forge, store, newPipeline, user, repo, pipelineFiles, envs, false)
 	if handleParseErrors(newPipeline, parseErr) {
-		if newPipeline, uErr := UpdateToStatusError(store, *newPipeline, parseErr); uErr != nil {
+		if newPipeline, uErr := UpdatePipelineToError(store, *newPipeline, parseErr); uErr != nil {
 			log.Error().Err(uErr).Msgf("error setting error status of pipeline for %s#%d", repo.FullName, newPipeline.Number)
 		} else {
 			updatePipelineStatus(ctx, forge, newPipeline, repo, user)
@@ -106,16 +104,7 @@ func Restart(ctx context.Context, store store.Store, lastPipeline *model.Pipelin
 		return nil, errors.New(msg)
 	}
 
-	publishPipeline(ctx, forge, newPipeline, repo, user)
-
-	newPipeline, err = start(ctx, forge, store, newPipeline, user, repo, pipelineItems)
-	if err != nil {
-		msg := fmt.Sprintf("failure to start pipeline for %s", repo.FullName)
-		log.Error().Err(err).Msg(msg)
-		return nil, errors.New(msg)
-	}
-
-	return newPipeline, nil
+	return finishPipeline(ctx, forge, store, newPipeline, user, repo, pipelineItems)
 }
 
 func linkPipelineConfigs(store store.Store, configs []*model.Config, pipelineID int64) error {