Refactor pipeline parsing and forge refreshing (#2527)

- refactor pipeline parsing
- do not parse the pipeline multiple times to perform filter checks, do
this once and perform checks on the result directly
    - code deduplication
- refactor forge token refreshing
    - move refreshing to a helper func to reduce code

---------

Co-authored-by: Anbraten <anton@ju60.de>

cmd/server/docs/docs.go

@@ -4323,7 +4323,7 @@ const docTemplate = `{
             "type": "object",
             "properties": {
                 "admin": {
-                    "description": "Admin indicates the user is a system administrator.\n\nNOTE: If the username is part of the WOODPECKER_ADMIN\nenvironment variable this value will be set to true on login.",
+                    "description": "Admin indicates the user is a system administrator.\n\nNOTE: If the username is part of the WOODPECKER_ADMIN\nenvironment variable, this value will be set to true on login.",
                     "type": "boolean"
                 },
                 "avatar_url": {

pipeline/stepBuilder.go

@@ -79,11 +79,10 @@ func (b *StepBuilder) Build() ([]*Item, error) {
 
 		for _, axis := range axes {
 			workflow := &model.Workflow{
-				PipelineID: b.Curr.ID,
+				PID:     pidSequence,
-				PID:        pidSequence,
+				State:   model.StatusPending,
-				State:      model.StatusPending,
+				Environ: axis,
-				Environ:    axis,
+				Name:    SanitizePath(y.Name),
-				Name:       SanitizePath(y.Name),
 			}
 			item, err := b.genItemForWorkflow(workflow, axis, string(y.Data))
 			if err != nil {
@@ -102,7 +101,7 @@ func (b *StepBuilder) Build() ([]*Item, error) {
 
 	items = filterItemsWithMissingDependencies(items)
 
-	// check if at least one step can start, if list is not empty
+	// check if at least one step can start if slice is not empty
 	if len(items) > 0 && !stepListContainsItemsToRun(items) {
 		return nil, fmt.Errorf("pipeline has no startpoint")
 	}
@@ -145,9 +144,9 @@ func (b *StepBuilder) genItemForWorkflow(workflow *model.Workflow, axis matrix.A
 	// checking if filtered.
 	if match, err := parsed.When.Match(workflowMetadata, true, environ); !match && err == nil {
 		log.Debug().Str("pipeline", workflow.Name).Msg(
-			"Marked as skipped, dose not match metadata",
+			"Marked as skipped, does not match metadata",
 		)
-		workflow.State = model.StatusSkipped
+		return nil, nil
 	} else if err != nil {
 		log.Debug().Str("pipeline", workflow.Name).Msg(
 			"Pipeline config could not be parsed",
@@ -288,47 +287,6 @@ func (b *StepBuilder) toInternalRepresentation(parsed *yaml_types.Workflow, envi
 	).Compile(parsed)
 }
 
-// SetPipelineStepsOnPipeline is the link between pipeline representation in "pipeline package" and server
-// to be specific this func currently is used to convert the pipeline.Item list (crafted by StepBuilder.Build()) into
-// a pipeline that can be stored in the database by the server
-func SetPipelineStepsOnPipeline(pipeline *model.Pipeline, pipelineItems []*Item) *model.Pipeline {
-	var pidSequence int
-	for _, item := range pipelineItems {
-		if pidSequence < item.Workflow.PID {
-			pidSequence = item.Workflow.PID
-		}
-	}
-
-	for _, item := range pipelineItems {
-		for _, stage := range item.Config.Stages {
-			var gid int
-			for _, step := range stage.Steps {
-				pidSequence++
-				if gid == 0 {
-					gid = pidSequence
-				}
-				step := &model.Step{
-					Name:       step.Alias,
-					UUID:       step.UUID,
-					PipelineID: pipeline.ID,
-					PID:        pidSequence,
-					PPID:       item.Workflow.PID,
-					State:      model.StatusPending,
-					Failure:    step.Failure,
-					Type:       model.StepType(step.Type),
-				}
-				if item.Workflow.State == model.StatusSkipped {
-					step.State = model.StatusSkipped
-				}
-				item.Workflow.Children = append(item.Workflow.Children, step)
-			}
-		}
-		pipeline.Workflows = append(pipeline.Workflows, item.Workflow)
-	}
-
-	return pipeline
-}
-
 func SanitizePath(path string) string {
 	path = filepath.Base(path)
 	path = strings.TrimSuffix(path, ".yml")

pipeline/stepBuilder_test.go

@@ -516,44 +516,6 @@ depends_on: [ shouldbefiltered ]
 	}
 }
 
-func TestTree(t *testing.T) {
-	t.Parallel()
-
-	pipeline := &model.Pipeline{
-		Event: model.EventPush,
-	}
-
-	b := StepBuilder{
-		Forge: getMockForge(t),
-		Repo:  &model.Repo{},
-		Curr:  pipeline,
-		Last:  &model.Pipeline{},
-		Netrc: &model.Netrc{},
-		Secs:  []*model.Secret{},
-		Regs:  []*model.Registry{},
-		Link:  "",
-		Yamls: []*forge_types.FileMeta{
-			{Data: []byte(`
-steps:
-  build:
-    image: scratch
-`)},
-		},
-	}
-
-	pipelineItems, err := b.Build()
-	pipeline = SetPipelineStepsOnPipeline(pipeline, pipelineItems)
-	if err != nil {
-		t.Fatal(err)
-	}
-	if len(pipeline.Workflows) != 1 {
-		t.Fatal("Should generate three in total")
-	}
-	if len(pipeline.Workflows[0].Children) != 2 {
-		t.Fatal("Workflow should have two children")
-	}
-}
-
 func TestSanitizePath(t *testing.T) {
 	t.Parallel()
 

server/api/helper.go

@@ -19,7 +19,6 @@ import (
 	"net/http"
 
 	"github.com/gin-gonic/gin"
-	"github.com/rs/zerolog/log"
 
 	"github.com/woodpecker-ci/woodpecker/server"
 	"github.com/woodpecker-ci/woodpecker/server/forge"
@@ -34,7 +33,7 @@ func handlePipelineErr(c *gin.Context, err error) {
 		c.String(http.StatusNotFound, "%s", err)
 	} else if errors.Is(err, &pipeline.ErrBadRequest{}) {
 		c.String(http.StatusBadRequest, "%s", err)
-	} else if errors.Is(err, &pipeline.ErrFiltered{}) {
+	} else if errors.Is(err, pipeline.ErrFiltered) {
 		c.Status(http.StatusNoContent)
 	} else {
 		_ = c.AbortWithError(http.StatusInternalServerError, err)
@@ -49,19 +48,10 @@ func handleDbError(c *gin.Context, err error) {
 	_ = c.AbortWithError(http.StatusInternalServerError, err)
 }
 
-// if the forge has a refresh token, the current access token may be stale.
+// If the forge has a refresh token, the current access token may be stale.
 // Therefore, we should refresh prior to dispatching the job.
 func refreshUserToken(c *gin.Context, user *model.User) {
 	_forge := server.Config.Services.Forge
 	_store := store.FromContext(c)
-	if refresher, ok := _forge.(forge.Refresher); ok {
+	forge.Refresh(c, _forge, _store, user)
-		ok, err := refresher.Refresh(c, user)
-		if err != nil {
-			log.Error().Err(err).Msgf("refresh oauth token of user '%s' failed", user.Login)
-		} else if ok {
-			if err := _store.UpdateUser(user); err != nil {
-				log.Error().Err(err).Msg("fail to save user to store after refresh oauth token")
-			}
-		}
-	}
 }

server/cron/cron.go

@@ -121,22 +121,10 @@ func CreatePipeline(ctx context.Context, store store.Store, f forge.Forge, cron
 		return nil, nil, err
 	}
 
-	// if the forge has a refresh token, the current access token
+	// If the forge has a refresh token, the current access token
 	// may be stale. Therefore, we should refresh prior to dispatching
 	// the pipeline.
-	if refresher, ok := f.(forge.Refresher); ok {
+	forge.Refresh(ctx, f, store, creator)
-		refreshed, err := refresher.Refresh(ctx, creator)
-		if err != nil {
-			log.Error().Err(err).Msgf("failed to refresh oauth2 token for creator: %s", creator.Login)
-		} else if refreshed {
-			if err := store.UpdateUser(creator); err != nil {
-				log.Error().Err(err).Msgf("error while updating creator: %s", creator.Login)
-				// move forward
-			} else {
-				log.Debug().Msgf("token refreshed for creator: %s", creator.Login)
-			}
-		}
-	}
 
 	commit, err := f.BranchHead(ctx, creator, repo, cron.Branch)
 	if err != nil {

server/forge/forge.go

@@ -94,10 +94,3 @@ type Forge interface {
 	// Org fetches the organization from the forge by name. If the name is a user an org with type user is returned.
 	Org(ctx context.Context, u *model.User, org string) (*model.Org, error)
 }
-
-// Refresher refreshes an oauth token and expiration for the given user. It
-// returns true if the token was refreshed, false if the token was not refreshed,
-// and error if it failed to refresh.
-type Refresher interface {
-	Refresh(context.Context, *model.User) (bool, error)
-}

server/forge/gitea/gitea.go

@@ -173,7 +173,7 @@ func (c *Gitea) Auth(ctx context.Context, token, _ string) (string, error) {
 }
 
 // Refresh refreshes the Gitea oauth2 access token. If the token is
-// refreshed the user is updated and a true value is returned.
+// refreshed, the user is updated and a true value is returned.
 func (c *Gitea) Refresh(ctx context.Context, user *model.User) (bool, error) {
 	config, oauth2Ctx := c.oauth2Config(ctx)
 	config.RedirectURL = ""

server/forge/refresh.go

@@ -0,0 +1,52 @@
+// Copyright 2023 Woodpecker Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package forge
+
+import (
+	"context"
+	"time"
+
+	"github.com/rs/zerolog/log"
+
+	"github.com/woodpecker-ci/woodpecker/server/model"
+	"github.com/woodpecker-ci/woodpecker/server/store"
+)
+
+// Refresher refreshes an oauth token and expiration for the given user. It
+// returns true if the token was refreshed, false if the token was not refreshed,
+// and error if it failed to refresh.
+type Refresher interface {
+	Refresh(context.Context, *model.User) (bool, error)
+}
+
+func Refresh(c context.Context, forge Forge, _store store.Store, user *model.User) {
+	if refresher, ok := forge.(Refresher); ok {
+		// Check to see if the user token is expired or
+		// will expire within the next 30 minutes (1800 seconds).
+		// If not, there is nothing we really need to do here.
+		if time.Now().UTC().Unix() < (user.Expiry - 1800) {
+			return
+		}
+
+		ok, err := refresher.Refresh(c, user)
+		if err != nil {
+			log.Error().Err(err).Msgf("refresh oauth token of user '%s' failed", user.Login)
+		} else if ok {
+			if err := _store.UpdateUser(user); err != nil {
+				log.Error().Err(err).Msg("fail to save user to store after refresh oauth token")
+			}
+		}
+	}
+}

server/grpc/rpc.go

@@ -388,16 +388,7 @@ func (s *RPC) updateForgeStatus(ctx context.Context, repo *model.Repo, pipeline
 		return
 	}
 
-	if refresher, ok := s.forge.(forge.Refresher); ok {
+	forge.Refresh(ctx, s.forge, s.store, user)
-		ok, err := refresher.Refresh(ctx, user)
-		if err != nil {
-			log.Error().Err(err).Msgf("grpc: refresh oauth token of user '%s' failed", user.Login)
-		} else if ok {
-			if err := s.store.UpdateUser(user); err != nil {
-				log.Error().Err(err).Msg("fail to save user to store after refresh oauth token")
-			}
-		}
-	}
 
 	// only do status updates for parent steps
 	if workflow != nil {

server/model/user.go

@@ -23,7 +23,7 @@ import (
 // validate a username (e.g. from github)
 var reUsername = regexp.MustCompile("^[a-zA-Z0-9-_.]+$")
 
-var errUserLoginInvalid = errors.New("Invalid User Login")
+var errUserLoginInvalid = errors.New("invalid user login")
 
 // User represents a registered user.
 type User struct {
@@ -59,7 +59,7 @@ type User struct {
 	// Admin indicates the user is a system administrator.
 	//
 	// NOTE: If the username is part of the WOODPECKER_ADMIN
-	// environment variable this value will be set to true on login.
+	// environment variable, this value will be set to true on login.
 	Admin bool `json:"admin,omitempty" xorm:"user_admin"`
 
 	// Hash is a unique token used to sign tokens.

server/pipeline/approve.go

@@ -25,7 +25,7 @@ import (
 	"github.com/woodpecker-ci/woodpecker/server/store"
 )
 
-// Approve update the status to pending for blocked pipeline because of a gated repo
+// Approve update the status to pending for a blocked pipeline because of a gated repo
 // and start them afterward
 func Approve(ctx context.Context, store store.Store, currentPipeline *model.Pipeline, user *model.User, repo *model.Repo) (*model.Pipeline, error) {
 	if currentPipeline.Status != model.StatusBlocked {

server/pipeline/config.go

@@ -44,13 +44,5 @@ func findOrPersistPipelineConfig(store store.Store, currentPipeline *model.Pipel
 		}
 	}
 
-	pipelineConfig := &model.PipelineConfig{
-		ConfigID:   conf.ID,
-		PipelineID: currentPipeline.ID,
-	}
-	if err := store.PipelineConfigCreate(pipelineConfig); err != nil {
-		return nil, err
-	}
-
 	return conf, nil
 }

server/pipeline/create.go

@@ -23,7 +23,6 @@ import (
 
 	"github.com/woodpecker-ci/woodpecker/server"
 	"github.com/woodpecker-ci/woodpecker/server/forge"
-	"github.com/woodpecker-ci/woodpecker/server/forge/types"
 	"github.com/woodpecker-ci/woodpecker/server/model"
 	"github.com/woodpecker-ci/woodpecker/server/store"
 )
@@ -37,118 +36,92 @@ func Create(ctx context.Context, _store store.Store, repo *model.Repo, pipeline
 		return nil, fmt.Errorf(msg)
 	}
 
-	// if the forge has a refresh token, the current access token
+	// If the forge has a refresh token, the current access token
 	// may be stale. Therefore, we should refresh prior to dispatching
 	// the pipeline.
-	if refresher, ok := server.Config.Services.Forge.(forge.Refresher); ok {
+	forge.Refresh(ctx, server.Config.Services.Forge, _store, repoUser)
-		refreshed, err := refresher.Refresh(ctx, repoUser)
-		if err != nil {
-			log.Error().Err(err).Msgf("failed to refresh oauth2 token for repoUser: %s", repoUser.Login)
-		} else if refreshed {
-			if err := _store.UpdateUser(repoUser); err != nil {
-				log.Error().Err(err).Msgf("error while updating repoUser: %s", repoUser.Login)
-				// move forward
-			}
-		}
-	}
-
-	var (
-		forgeYamlConfigs []*types.FileMeta
-		configFetchErr   error
-		filtered         bool
-		parseErr         error
-	)
-
-	// fetch the pipeline file from the forge
-	configFetcher := forge.NewConfigFetcher(server.Config.Services.Forge, server.Config.Services.Timeout, server.Config.Services.ConfigService, repoUser, repo, pipeline)
-	forgeYamlConfigs, configFetchErr = configFetcher.Fetch(ctx)
-	if configFetchErr == nil {
-		filtered, parseErr = checkIfFiltered(repo, pipeline, forgeYamlConfigs)
-		if parseErr == nil {
-			if filtered {
-				err := ErrFiltered{Msg: "global when filter of all workflows do skip this pipeline"}
-				log.Debug().Str("repo", repo.FullName).Msgf("%v", err)
-				return nil, err
-			}
-
-			if zeroSteps(pipeline, forgeYamlConfigs) {
-				err := ErrFiltered{Msg: "step conditions yield zero runnable steps"}
-				log.Debug().Str("repo", repo.FullName).Msgf("%v", err)
-				return nil, err
-			}
-		}
-	}
 
 	// update some pipeline fields
 	pipeline.RepoID = repo.ID
 	pipeline.Status = model.StatusPending
 
+	// fetch the pipeline file from the forge
+	configFetcher := forge.NewConfigFetcher(server.Config.Services.Forge, server.Config.Services.Timeout, server.Config.Services.ConfigService, repoUser, repo, pipeline)
+	forgeYamlConfigs, configFetchErr := configFetcher.Fetch(ctx)
+
 	if configFetchErr != nil {
 		log.Debug().Str("repo", repo.FullName).Err(configFetchErr).Msgf("cannot find config '%s' in '%s' with user: '%s'", repo.Config, pipeline.Ref, repoUser.Login)
-		pipeline.Started = time.Now().Unix()
+		return nil, persistPipelineWithErr(ctx, _store, pipeline, repo, repoUser, fmt.Sprintf("pipeline definition not found in %s", repo.FullName))
-		pipeline.Finished = pipeline.Started
-		pipeline.Status = model.StatusError
-		pipeline.Error = fmt.Sprintf("pipeline definition not found in %s", repo.FullName)
-	} else if parseErr != nil {
-		log.Debug().Str("repo", repo.FullName).Err(parseErr).Msg("failed to parse yaml")
-		pipeline.Started = time.Now().Unix()
-		pipeline.Finished = pipeline.Started
-		pipeline.Status = model.StatusError
-		pipeline.Error = fmt.Sprintf("failed to parse pipeline: %s", parseErr.Error())
-	} else {
-		setGatedState(repo, pipeline)
 	}
 
+	pipelineItems, parseErr := parsePipeline(_store, pipeline, repoUser, repo, forgeYamlConfigs, nil)
+	if parseErr != nil {
+		log.Debug().Str("repo", repo.FullName).Err(parseErr).Msg("failed to parse yaml")
+		return nil, persistPipelineWithErr(ctx, _store, pipeline, repo, repoUser, fmt.Sprintf("failed to parse pipeline: %s", parseErr.Error()))
+	}
+
+	if len(pipelineItems) == 0 {
+		log.Debug().Str("repo", repo.FullName).Msg(ErrFiltered.Error())
+		return nil, ErrFiltered
+	}
+
+	setGatedState(repo, pipeline)
+
 	err = _store.CreatePipeline(pipeline)
 	if err != nil {
-		msg := fmt.Sprintf("failure to save pipeline for %s", repo.FullName)
+		msg := fmt.Errorf("failed to save pipeline for %s", repo.FullName)
-		log.Error().Err(err).Msg(msg)
+		log.Error().Err(err).Msg(msg.Error())
-		return nil, fmt.Errorf(msg)
+		return nil, msg
 	}
 
+	pipeline = setPipelineStepsOnPipeline(pipeline, pipelineItems)
+
 	// persist the pipeline config for historical correctness, restarts, etc
+	var configs []*model.Config
 	for _, forgeYamlConfig := range forgeYamlConfigs {
-		_, err := findOrPersistPipelineConfig(_store, pipeline, forgeYamlConfig)
+		config, err := findOrPersistPipelineConfig(_store, pipeline, forgeYamlConfig)
 		if err != nil {
-			msg := fmt.Sprintf("failure to find or persist pipeline config for %s", repo.FullName)
+			msg := fmt.Sprintf("failed to find or persist pipeline config for %s", repo.FullName)
 			log.Error().Err(err).Msg(msg)
 			return nil, fmt.Errorf(msg)
 		}
+		configs = append(configs, config)
 	}
-
+	// link pipeline to persisted configs
-	if pipeline.Status == model.StatusError {
+	if err := linkPipelineConfigs(_store, configs, pipeline.ID); err != nil {
-		if err := publishToTopic(ctx, pipeline, repo); err != nil {
+		msg := fmt.Sprintf("failed to find or persist pipeline config for %s", repo.FullName)
-			log.Error().Err(err).Msg("publishToTopic")
-		}
-
-		updatePipelineStatus(ctx, pipeline, repo, repoUser)
-
-		return pipeline, nil
-	}
-
-	pipeline, pipelineItems, err := createPipelineItems(ctx, _store, pipeline, repoUser, repo, forgeYamlConfigs, nil)
-	if err != nil {
-		msg := fmt.Sprintf("failure to createPipelineItems for %s", repo.FullName)
 		log.Error().Err(err).Msg(msg)
 		return nil, fmt.Errorf(msg)
 	}
 
 	if pipeline.Status == model.StatusBlocked {
-		if err := publishToTopic(ctx, pipeline, repo); err != nil {
+		publishPipeline(ctx, pipeline, repo, repoUser)
-			log.Error().Err(err).Msg("publishToTopic")
-		}
-
-		updatePipelineStatus(ctx, pipeline, repo, repoUser)
-
 		return pipeline, nil
 	}
 
 	pipeline, err = start(ctx, _store, pipeline, repoUser, repo, pipelineItems)
 	if err != nil {
-		msg := fmt.Sprintf("failure to start pipeline for %s", repo.FullName)
+		msg := fmt.Sprintf("failed to start pipeline for %s", repo.FullName)
 		log.Error().Err(err).Msg(msg)
 		return nil, fmt.Errorf(msg)
 	}
 
 	return pipeline, nil
 }
+
+func persistPipelineWithErr(ctx context.Context, _store store.Store, pipeline *model.Pipeline, repo *model.Repo, repoUser *model.User, err string) error {
+	pipeline.Started = time.Now().Unix()
+	pipeline.Finished = pipeline.Started
+	pipeline.Status = model.StatusError
+	pipeline.Error = err
+	dbErr := _store.CreatePipeline(pipeline)
+	if dbErr != nil {
+		msg := fmt.Errorf("failed to save pipeline for %s", repo.FullName)
+		log.Error().Err(dbErr).Msg(msg.Error())
+		return msg
+	}
+
+	publishPipeline(ctx, pipeline, repo, repoUser)
+
+	return nil
+}

server/pipeline/errors.go

@@ -14,6 +14,8 @@
 
 package pipeline
 
+import "errors"
+
 type ErrNotFound struct {
 	Msg string
 }
@@ -46,18 +48,4 @@ func (e ErrBadRequest) Is(target error) bool {
 	return ok
 }
 
-type ErrFiltered struct {
+var ErrFiltered = errors.New("ignoring hook: 'when' filters filtered out all steps")
-	Msg string
-}
-
-func (e ErrFiltered) Error() string {
-	return "ignoring hook: " + e.Msg
-}
-
-func (e *ErrFiltered) Is(target error) bool {
-	_, ok := target.(ErrFiltered) //nolint:errorlint
-	if !ok {
-		_, ok = target.(*ErrFiltered) //nolint:errorlint
-	}
-	return ok
-}

server/pipeline/filter.go

@@ -1,87 +0,0 @@
-// Copyright 2022 Woodpecker Authors
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//      http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package pipeline
-
-// TODO(770): pipeline filter should not belong here
-
-import (
-	"github.com/rs/zerolog/log"
-
-	"github.com/woodpecker-ci/woodpecker/pipeline"
-	"github.com/woodpecker-ci/woodpecker/pipeline/frontend"
-	"github.com/woodpecker-ci/woodpecker/pipeline/frontend/yaml"
-	"github.com/woodpecker-ci/woodpecker/server"
-	forge_types "github.com/woodpecker-ci/woodpecker/server/forge/types"
-	"github.com/woodpecker-ci/woodpecker/server/model"
-)
-
-func zeroSteps(currentPipeline *model.Pipeline, forgeYamlConfigs []*forge_types.FileMeta) bool {
-	b := pipeline.StepBuilder{
-		Repo:  &model.Repo{},
-		Curr:  currentPipeline,
-		Last:  &model.Pipeline{},
-		Netrc: &model.Netrc{},
-		Secs:  []*model.Secret{},
-		Regs:  []*model.Registry{},
-		Link:  "",
-		Yamls: forgeYamlConfigs,
-		Forge: server.Config.Services.Forge,
-	}
-
-	pipelineItems, err := b.Build()
-	if err != nil {
-		return false
-	}
-	if len(pipelineItems) == 0 {
-		return true
-	}
-
-	return false
-}
-
-// TODO: parse yaml once and not for each filter function (-> move server/pipeline/filter* into pipeline/step_builder)
-// Check if at least one pipeline step will be execute otherwise we will just ignore this webhook
-func checkIfFiltered(repo *model.Repo, p *model.Pipeline, forgeYamlConfigs []*forge_types.FileMeta) (bool, error) {
-	log.Trace().Msgf("hook.branchFiltered(): pipeline branch: '%s' pipeline event: '%s' config count: %d", p.Branch, p.Event, len(forgeYamlConfigs))
-
-	matchMetadata := frontend.MetadataFromStruct(server.Config.Services.Forge, repo, p, nil, nil, "")
-
-	for _, forgeYamlConfig := range forgeYamlConfigs {
-		substitutedConfigData, err := frontend.EnvVarSubst(string(forgeYamlConfig.Data), matchMetadata.Environ())
-		if err != nil {
-			log.Trace().Err(err).Msgf("failed to substitute config '%s'", forgeYamlConfig.Name)
-			return false, err
-		}
-		parsedPipelineConfig, err := yaml.ParseString(substitutedConfigData)
-		if err != nil {
-			log.Trace().Err(err).Msgf("failed to parse config '%s'", forgeYamlConfig.Name)
-			return false, err
-		}
-		log.Trace().Msgf("config '%s': %#v", forgeYamlConfig.Name, parsedPipelineConfig)
-
-		// ignore if the pipeline was filtered by matched constraints
-		if match, err := parsedPipelineConfig.When.Match(matchMetadata, true, p.AdditionalVariables); !match && err == nil {
-			continue
-		} else if err != nil {
-			return false, err
-		}
-
-		// at least one config yielded in a valid run.
-		return false, nil
-	}
-
-	// no configs yielded a valid run.
-	return true, nil
-}

server/pipeline/gated.go

@@ -16,11 +16,11 @@ package pipeline
 
 import "github.com/woodpecker-ci/woodpecker/server/model"
 
-func setGatedState(repo *model.Repo, pipe *model.Pipeline) {
+func setGatedState(repo *model.Repo, pipeline *model.Pipeline) {
 	// TODO(336): extend gated feature with an allow/block List
 	if repo.IsGated &&
 		// events created by woodpecker itself should run right away
-		pipe.Event != model.EventCron && pipe.Event != model.EventManual {
+		pipeline.Event != model.EventCron && pipeline.Event != model.EventManual {
-		pipe.Status = model.StatusBlocked
+		pipeline.Status = model.StatusBlocked
 	}
 }

server/pipeline/items.go

@@ -29,10 +29,7 @@ import (
 	"github.com/woodpecker-ci/woodpecker/server/store"
 )
 
-func createPipelineItems(c context.Context, store store.Store,
+func parsePipeline(store store.Store, currentPipeline *model.Pipeline, user *model.User, repo *model.Repo, yamls []*forge_types.FileMeta, envs map[string]string) ([]*pipeline.Item, error) {
-	currentPipeline *model.Pipeline, user *model.User, repo *model.Repo,
-	yamls []*forge_types.FileMeta, envs map[string]string,
-) (*model.Pipeline, []*pipeline.Item, error) {
 	netrc, err := server.Config.Services.Forge.Netrc(user, repo)
 	if err != nil {
 		log.Error().Err(err).Msg("Failed to generate netrc file")
@@ -86,17 +83,71 @@ func createPipelineItems(c context.Context, store store.Store,
 		},
 	}
 	pipelineItems, err := b.Build()
+	if err != nil {
+		return nil, err
+	}
+
+	return pipelineItems, nil
+}
+
+func createPipelineItems(c context.Context, store store.Store,
+	currentPipeline *model.Pipeline, user *model.User, repo *model.Repo,
+	yamls []*forge_types.FileMeta, envs map[string]string,
+) (*model.Pipeline, []*pipeline.Item, error) {
+	pipelineItems, err := parsePipeline(store, currentPipeline, user, repo, yamls, envs)
 	if err != nil {
 		currentPipeline, uerr := UpdateToStatusError(store, *currentPipeline, err)
 		if uerr != nil {
-			log.Error().Err(err).Msgf("Error setting error status of pipeline for %s#%d", repo.FullName, currentPipeline.Number)
+			log.Error().Err(uerr).Msgf("Error setting error status of pipeline for %s#%d", repo.FullName, currentPipeline.Number)
 		} else {
 			updatePipelineStatus(c, currentPipeline, repo, user)
 		}
 		return currentPipeline, nil, err
 	}
 
-	currentPipeline = pipeline.SetPipelineStepsOnPipeline(b.Curr, pipelineItems)
+	currentPipeline = setPipelineStepsOnPipeline(currentPipeline, pipelineItems)
 
 	return currentPipeline, pipelineItems, nil
 }
+
+// setPipelineStepsOnPipeline is the link between pipeline representation in "pipeline package" and server
+// to be specific this func currently is used to convert the pipeline.Item list (crafted by StepBuilder.Build()) into
+// a pipeline that can be stored in the database by the server
+func setPipelineStepsOnPipeline(pipeline *model.Pipeline, pipelineItems []*pipeline.Item) *model.Pipeline {
+	var pidSequence int
+	for _, item := range pipelineItems {
+		if pidSequence < item.Workflow.PID {
+			pidSequence = item.Workflow.PID
+		}
+	}
+
+	for _, item := range pipelineItems {
+		for _, stage := range item.Config.Stages {
+			var gid int
+			for _, step := range stage.Steps {
+				pidSequence++
+				if gid == 0 {
+					gid = pidSequence
+				}
+				step := &model.Step{
+					Name:       step.Alias,
+					UUID:       step.UUID,
+					PipelineID: pipeline.ID,
+					PID:        pidSequence,
+					PPID:       item.Workflow.PID,
+					State:      model.StatusPending,
+					Failure:    step.Failure,
+					Type:       model.StepType(step.Type),
+				}
+				if item.Workflow.State == model.StatusSkipped {
+					step.State = model.StatusSkipped
+				}
+				item.Workflow.Children = append(item.Workflow.Children, step)
+			}
+		}
+		item.Workflow.PipelineID = pipeline.ID
+		pipeline.Workflows = append(pipeline.Workflows, item.Workflow)
+	}
+
+	return pipeline
+}

server/pipeline/items_test.go

@@ -0,0 +1,52 @@
+package pipeline
+
+import (
+	"testing"
+
+	sharedPipeline "github.com/woodpecker-ci/woodpecker/pipeline"
+	"github.com/woodpecker-ci/woodpecker/pipeline/backend/types"
+	"github.com/woodpecker-ci/woodpecker/server/model"
+)
+
+func TestSetPipelineStepsOnPipeline(t *testing.T) {
+	t.Parallel()
+
+	pipeline := &model.Pipeline{
+		ID:    1,
+		Event: model.EventPush,
+	}
+
+	pipelineItems := []*sharedPipeline.Item{{
+		Workflow: &model.Workflow{
+			PID: 1,
+		},
+		Config: &types.Config{
+			Stages: []*types.Stage{
+				{
+					Steps: []*types.Step{
+						{
+							Name: "clone",
+						},
+					},
+				},
+				{
+					Steps: []*types.Step{
+						{
+							Name: "step",
+						},
+					},
+				},
+			},
+		},
+	}}
+	pipeline = setPipelineStepsOnPipeline(pipeline, pipelineItems)
+	if len(pipeline.Workflows) != 1 {
+		t.Fatal("Should generate three in total")
+	}
+	if pipeline.Workflows[0].PipelineID != 1 {
+		t.Fatal("Should set workflow's pipeline ID")
+	}
+	if pipeline.Workflows[0].Children[0].PPID != 1 {
+		t.Fatal("Should set step PPID")
+	}
+}

server/pipeline/restart.go

@@ -39,7 +39,7 @@ func Restart(ctx context.Context, store store.Store, lastPipeline *model.Pipelin
 
 	var pipelineFiles []*forge_types.FileMeta
 
-	// fetch the old pipeline config from database
+	// fetch the old pipeline config from the database
 	configs, err := store.ConfigsForPipeline(lastPipeline.ID)
 	if err != nil {
 		msg := fmt.Sprintf("failure to get pipeline config for %s. %s", repo.FullName, err)
@@ -88,7 +88,7 @@ func Restart(ctx context.Context, store store.Store, lastPipeline *model.Pipelin
 		}
 		return newPipeline, nil
 	}
-	if err := persistPipelineConfigs(store, configs, newPipeline.ID); err != nil {
+	if err := linkPipelineConfigs(store, configs, newPipeline.ID); err != nil {
 		msg := fmt.Sprintf("failure to persist pipeline config for %s.", repo.FullName)
 		log.Error().Err(err).Msg(msg)
 		return nil, fmt.Errorf(msg)
@@ -114,8 +114,7 @@ func Restart(ctx context.Context, store store.Store, lastPipeline *model.Pipelin
 	return newPipeline, nil
 }
 
-// TODO: reuse at create.go too
+func linkPipelineConfigs(store store.Store, configs []*model.Config, pipelineID int64) error {
-func persistPipelineConfigs(store store.Store, configs []*model.Config, pipelineID int64) error {
 	for _, conf := range configs {
 		pipelineConfig := &model.PipelineConfig{
 			ConfigID:   conf.ID,

server/pipeline/start.go

@@ -38,9 +38,7 @@ func start(ctx context.Context, store store.Store, activePipeline *model.Pipelin
 		return nil, err
 	}
 
-	if err := publishToTopic(ctx, activePipeline, repo); err != nil {
+	publishPipeline(ctx, activePipeline, repo, user)
-		log.Error().Err(err).Msg("publishToTopic")
-	}
 
 	if err := queuePipeline(repo, pipelineItems); err != nil {
 		log.Error().Err(err).Msg("queuePipeline")
@@ -59,7 +57,13 @@ func start(ctx context.Context, store store.Store, activePipeline *model.Pipelin
 		}
 	}
 
-	updatePipelineStatus(ctx, activePipeline, repo, user)
-
 	return activePipeline, nil
 }
+
+func publishPipeline(ctx context.Context, pipeline *model.Pipeline, repo *model.Repo, repoUser *model.User) {
+	if err := publishToTopic(ctx, pipeline, repo); err != nil {
+		log.Error().Err(err).Msg("publishToTopic")
+	}
+
+	updatePipelineStatus(ctx, pipeline, repo, repoUser)
+}

server/router/middleware/token/token.go

@@ -15,10 +15,6 @@
 package token
 
 import (
-	"time"
-
-	"github.com/rs/zerolog/log"
-
 	"github.com/gin-gonic/gin"
 
 	"github.com/woodpecker-ci/woodpecker/server"
@@ -29,43 +25,8 @@ import (
 
 func Refresh(c *gin.Context) {
 	user := session.User(c)
-	if user == nil {
+	if user != nil {
-		c.Next()
+		forge.Refresh(c, server.Config.Services.Forge, store.FromContext(c), user)
-		return
-	}
-
-	// check if the forge includes the ability to
-	// refresh the user token.
-	_forge := server.Config.Services.Forge
-	refresher, ok := _forge.(forge.Refresher)
-	if !ok {
-		c.Next()
-		return
-	}
-
-	// check to see if the user token is expired or
-	// will expire within the next 30 minutes (1800 seconds).
-	// If not, there is nothing we really need to do here.
-	if time.Now().UTC().Unix() < (user.Expiry - 1800) {
-		c.Next()
-		return
-	}
-
-	// attempts to refresh the access token. If the
-	// token is refreshed, we must also persist to the
-	// database.
-	ok, err := refresher.Refresh(c, user)
-	if err != nil {
-		log.Error().Err(err).Msgf("refresh oauth token of user '%s' failed", user.Login)
-	} else if ok {
-		err := store.FromContext(c).UpdateUser(user)
-		if err != nil {
-			// we only log the error at this time. not sure
-			// if we really want to fail the request, do we?
-			log.Error().Msgf("cannot refresh access token for %s. %s", user.Login, err)
-		} else {
-			log.Debug().Msgf("refreshed access token for %s", user.Login)
-		}
 	}
 
 	c.Next()

web/src/compositions/useEvents.ts

@@ -29,12 +29,5 @@ export default () => {
     }
     const { pipeline } = data;
     pipelineStore.setPipeline(repo.id, pipeline);
-
-    // contains step update
-    if (!data.step) {
-      return;
-    }
-    const { step } = data;
-    pipelineStore.setWorkflow(repo.id, pipeline.number, step);
   });
 };

web/src/lib/api/index.ts

@@ -8,7 +8,6 @@ import {
   PipelineConfig,
   PipelineFeed,
   PipelineLog,
-  PipelineWorkflow,
   PullRequest,
   QueueInfo,
   Registry,
@@ -316,7 +315,7 @@ export default class WoodpeckerClient extends ApiClient {
   }
 
   // eslint-disable-next-line promise/prefer-await-to-callbacks
-  on(callback: (data: { pipeline?: Pipeline; repo?: Repo; step?: PipelineWorkflow }) => void): EventSource {
+  on(callback: (data: { pipeline?: Pipeline; repo?: Repo }) => void): EventSource {
     return this._subscribe('/api/stream/events', callback, {
       reconnect: true,
     });