Add workflow_run api + webhook (#33964)

Implements 
- https://docs.github.com/en/rest/actions/workflow-jobs?apiVersion=2022-11-28#list-jobs-for-a-workflow-run--code-samples
- https://docs.github.com/en/rest/actions/workflow-jobs?apiVersion=2022-11-28#get-a-job-for-a-workflow-run--code-samples
- https://docs.github.com/en/rest/actions/workflow-runs?apiVersion=2022-11-28#list-workflow-runs-for-a-repository
- https://docs.github.com/en/rest/actions/workflow-runs?apiVersion=2022-11-28#get-a-workflow-run
  - `/actions/runs` for global + user + org (Gitea only)
  - `/actions/jobs` for global + user + org + repository (Gitea only)
  - workflow_run webhook + action trigger
    - limitations
- workflow id is assigned to a string, this may result into problems in
strongly typed clients

Fixes
- workflow_job webhook url to no longer contain the `runs/<run>` part to
align with api
- workflow instance does now use it's name inside the file instead of
filename if set

Refactoring
- Moved a lot of logic from workflows/workflow_job into a shared module
used by both webhook and api

TODO
- [x] Verify Keda Compatibility
- [x] Edit Webhook API bug is resolved
 
Closes https://github.com/go-gitea/gitea/issues/23670
Closes https://github.com/go-gitea/gitea/issues/23796
Closes https://github.com/go-gitea/gitea/issues/24898
Replaces https://github.com/go-gitea/gitea/pull/28047 and is much more
complete

---------

Co-authored-by: wxiaoguang <wxiaoguang@gmail.com>

services/actions/clear_tasks.go

@@ -42,6 +42,10 @@ func notifyWorkflowJobStatusUpdate(ctx context.Context, jobs []*actions_model.Ac
 			_ = job.LoadAttributes(ctx)
 			notify_service.WorkflowJobStatusUpdate(ctx, job.Run.Repo, job.Run.TriggerUser, job, nil)
 		}
+		if len(jobs) > 0 {
+			job := jobs[0]
+			notify_service.WorkflowRunStatusUpdate(ctx, job.Run.Repo, job.Run.TriggerUser, job.Run)
+		}
 	}
 }
 
@@ -123,7 +127,7 @@ func CancelAbandonedJobs(ctx context.Context) error {
 		}
 		CreateCommitStatus(ctx, job)
 		if updated {
-			_ = job.LoadAttributes(ctx)
+			NotifyWorkflowRunStatusUpdateWithReload(ctx, job)
 			notify_service.WorkflowJobStatusUpdate(ctx, job.Run.Repo, job.Run.TriggerUser, job, nil)
 		}
 	}

services/actions/interface.go

@@ -33,4 +33,8 @@ type API interface {
 	GetRunner(*context.APIContext)
 	// DeleteRunner delete runner
 	DeleteRunner(*context.APIContext)
+	// ListWorkflowJobs list jobs
+	ListWorkflowJobs(*context.APIContext)
+	// ListWorkflowRuns list runs
+	ListWorkflowRuns(*context.APIContext)
 }

services/actions/job_emitter.go

@@ -11,6 +11,7 @@ import (
 	actions_model "code.gitea.io/gitea/models/actions"
 	"code.gitea.io/gitea/models/db"
 	"code.gitea.io/gitea/modules/graceful"
+	"code.gitea.io/gitea/modules/log"
 	"code.gitea.io/gitea/modules/queue"
 	notify_service "code.gitea.io/gitea/services/notify"
 
@@ -78,9 +79,30 @@ func checkJobsOfRun(ctx context.Context, runID int64) error {
 		_ = job.LoadAttributes(ctx)
 		notify_service.WorkflowJobStatusUpdate(ctx, job.Run.Repo, job.Run.TriggerUser, job, nil)
 	}
+	if len(jobs) > 0 {
+		runUpdated := true
+		for _, job := range jobs {
+			if !job.Status.IsDone() {
+				runUpdated = false
+				break
+			}
+		}
+		if runUpdated {
+			NotifyWorkflowRunStatusUpdateWithReload(ctx, jobs[0])
+		}
+	}
 	return nil
 }
 
+func NotifyWorkflowRunStatusUpdateWithReload(ctx context.Context, job *actions_model.ActionRunJob) {
+	job.Run = nil
+	if err := job.LoadAttributes(ctx); err != nil {
+		log.Error("LoadAttributes: %v", err)
+		return
+	}
+	notify_service.WorkflowRunStatusUpdate(ctx, job.Run.Repo, job.Run.TriggerUser, job.Run)
+}
+
 type jobStatusResolver struct {
 	statuses map[int64]actions_model.Status
 	needs    map[int64][]int64

services/actions/notifier.go

@@ -6,13 +6,16 @@ package actions
 import (
 	"context"
 
+	actions_model "code.gitea.io/gitea/models/actions"
 	issues_model "code.gitea.io/gitea/models/issues"
+	"code.gitea.io/gitea/models/organization"
 	packages_model "code.gitea.io/gitea/models/packages"
 	perm_model "code.gitea.io/gitea/models/perm"
 	access_model "code.gitea.io/gitea/models/perm/access"
 	repo_model "code.gitea.io/gitea/models/repo"
 	user_model "code.gitea.io/gitea/models/user"
 	"code.gitea.io/gitea/modules/git"
+	"code.gitea.io/gitea/modules/gitrepo"
 	"code.gitea.io/gitea/modules/log"
 	"code.gitea.io/gitea/modules/repository"
 	"code.gitea.io/gitea/modules/setting"
@@ -762,3 +765,41 @@ func (n *actionsNotifier) MigrateRepository(ctx context.Context, doer, u *user_m
 		Sender:       convert.ToUser(ctx, doer, nil),
 	}).Notify(ctx)
 }
+
+func (n *actionsNotifier) WorkflowRunStatusUpdate(ctx context.Context, repo *repo_model.Repository, sender *user_model.User, run *actions_model.ActionRun) {
+	ctx = withMethod(ctx, "WorkflowRunStatusUpdate")
+
+	var org *api.Organization
+	if repo.Owner.IsOrganization() {
+		org = convert.ToOrganization(ctx, organization.OrgFromUser(repo.Owner))
+	}
+
+	status := convert.ToWorkflowRunAction(run.Status)
+
+	gitRepo, err := gitrepo.OpenRepository(ctx, repo)
+	if err != nil {
+		log.Error("OpenRepository: %v", err)
+		return
+	}
+	defer gitRepo.Close()
+
+	convertedWorkflow, err := convert.GetActionWorkflow(ctx, gitRepo, repo, run.WorkflowID)
+	if err != nil {
+		log.Error("GetActionWorkflow: %v", err)
+		return
+	}
+	convertedRun, err := convert.ToActionWorkflowRun(ctx, repo, run)
+	if err != nil {
+		log.Error("ToActionWorkflowRun: %v", err)
+		return
+	}
+
+	newNotifyInput(repo, sender, webhook_module.HookEventWorkflowRun).WithPayload(&api.WorkflowRunPayload{
+		Action:       status,
+		Workflow:     convertedWorkflow,
+		WorkflowRun:  convertedRun,
+		Organization: org,
+		Repo:         convert.ToRepo(ctx, repo, access_model.Permission{AccessMode: perm_model.AccessModeOwner}),
+		Sender:       convert.ToUser(ctx, sender, nil),
+	}).Notify(ctx)
+}

services/actions/notifier_helper.go

@@ -178,7 +178,7 @@ func notify(ctx context.Context, input *notifyInput) error {
 		return fmt.Errorf("gitRepo.GetCommit: %w", err)
 	}
 
-	if skipWorkflows(input, commit) {
+	if skipWorkflows(ctx, input, commit) {
 		return nil
 	}
 
@@ -243,7 +243,7 @@ func notify(ctx context.Context, input *notifyInput) error {
 	return handleWorkflows(ctx, detectedWorkflows, commit, input, ref.String())
 }
 
-func skipWorkflows(input *notifyInput, commit *git.Commit) bool {
+func skipWorkflows(ctx context.Context, input *notifyInput, commit *git.Commit) bool {
 	// skip workflow runs with a configured skip-ci string in commit message or pr title if the event is push or pull_request(_sync)
 	// https://docs.github.com/en/actions/managing-workflow-runs/skipping-workflow-runs
 	skipWorkflowEvents := []webhook_module.HookEventType{
@@ -263,6 +263,27 @@ func skipWorkflows(input *notifyInput, commit *git.Commit) bool {
 			}
 		}
 	}
+	if input.Event == webhook_module.HookEventWorkflowRun {
+		wrun, ok := input.Payload.(*api.WorkflowRunPayload)
+		for i := 0; i < 5 && ok && wrun.WorkflowRun != nil; i++ {
+			if wrun.WorkflowRun.Event != "workflow_run" {
+				return false
+			}
+			r, err := actions_model.GetRunByRepoAndID(ctx, input.Repo.ID, wrun.WorkflowRun.ID)
+			if err != nil {
+				log.Error("GetRunByRepoAndID: %v", err)
+				return true
+			}
+			wrun, err = r.GetWorkflowRunEventPayload()
+			if err != nil {
+				log.Error("GetWorkflowRunEventPayload: %v", err)
+				return true
+			}
+		}
+		// skip workflow runs events exceeding the maxiumum of 5 recursive events
+		log.Debug("repo %s: skipped workflow_run because of recursive event of 5", input.Repo.RepoPath())
+		return true
+	}
 	return false
 }
 
@@ -372,6 +393,15 @@ func handleWorkflows(
 			continue
 		}
 		CreateCommitStatus(ctx, alljobs...)
+		if len(alljobs) > 0 {
+			job := alljobs[0]
+			err := job.LoadRun(ctx)
+			if err != nil {
+				log.Error("LoadRun: %v", err)
+				continue
+			}
+			notify_service.WorkflowRunStatusUpdate(ctx, job.Run.Repo, job.Run.TriggerUser, job.Run)
+		}
 		for _, job := range alljobs {
 			notify_service.WorkflowJobStatusUpdate(ctx, input.Repo, input.Doer, job, nil)
 		}

services/actions/schedule_tasks.go

@@ -157,6 +157,7 @@ func CreateScheduleTask(ctx context.Context, cron *actions_model.ActionSchedule)
 	if err != nil {
 		log.Error("LoadAttributes: %v", err)
 	}
+	notify_service.WorkflowRunStatusUpdate(ctx, run.Repo, run.TriggerUser, run)
 	for _, job := range allJobs {
 		notify_service.WorkflowJobStatusUpdate(ctx, run.Repo, run.TriggerUser, job, nil)
 	}

services/actions/workflow.go

@@ -5,9 +5,6 @@ package actions
 
 import (
 	"fmt"
-	"net/http"
-	"net/url"
-	"path"
 	"strings"
 
 	actions_model "code.gitea.io/gitea/models/actions"
@@ -31,51 +28,8 @@ import (
 	"github.com/nektos/act/pkg/model"
 )
 
-func getActionWorkflowEntry(ctx *context.APIContext, commit *git.Commit, folder string, entry *git.TreeEntry) *api.ActionWorkflow {
-	cfgUnit := ctx.Repo.Repository.MustGetUnit(ctx, unit.TypeActions)
-	cfg := cfgUnit.ActionsConfig()
-
-	defaultBranch, _ := commit.GetBranchName()
-
-	workflowURL := fmt.Sprintf("%s/actions/workflows/%s", ctx.Repo.Repository.APIURL(), url.PathEscape(entry.Name()))
-	workflowRepoURL := fmt.Sprintf("%s/src/branch/%s/%s/%s", ctx.Repo.Repository.HTMLURL(ctx), util.PathEscapeSegments(defaultBranch), util.PathEscapeSegments(folder), url.PathEscape(entry.Name()))
-	badgeURL := fmt.Sprintf("%s/actions/workflows/%s/badge.svg?branch=%s", ctx.Repo.Repository.HTMLURL(ctx), url.PathEscape(entry.Name()), url.QueryEscape(ctx.Repo.Repository.DefaultBranch))
-
-	// See https://docs.github.com/en/rest/actions/workflows?apiVersion=2022-11-28#get-a-workflow
-	// State types:
-	// - active
-	// - deleted
-	// - disabled_fork
-	// - disabled_inactivity
-	// - disabled_manually
-	state := "active"
-	if cfg.IsWorkflowDisabled(entry.Name()) {
-		state = "disabled_manually"
-	}
-
-	// The CreatedAt and UpdatedAt fields currently reflect the timestamp of the latest commit, which can later be refined
-	// by retrieving the first and last commits for the file history. The first commit would indicate the creation date,
-	// while the last commit would represent the modification date. The DeletedAt could be determined by identifying
-	// the last commit where the file existed. However, this implementation has not been done here yet, as it would likely
-	// cause a significant performance degradation.
-	createdAt := commit.Author.When
-	updatedAt := commit.Author.When
-
-	return &api.ActionWorkflow{
-		ID:        entry.Name(),
-		Name:      entry.Name(),
-		Path:      path.Join(folder, entry.Name()),
-		State:     state,
-		CreatedAt: createdAt,
-		UpdatedAt: updatedAt,
-		URL:       workflowURL,
-		HTMLURL:   workflowRepoURL,
-		BadgeURL:  badgeURL,
-	}
-}
-
 func EnableOrDisableWorkflow(ctx *context.APIContext, workflowID string, isEnable bool) error {
-	workflow, err := GetActionWorkflow(ctx, workflowID)
+	workflow, err := convert.GetActionWorkflow(ctx, ctx.Repo.GitRepo, ctx.Repo.Repository, workflowID)
 	if err != nil {
 		return err
 	}
@@ -92,42 +46,6 @@ func EnableOrDisableWorkflow(ctx *context.APIContext, workflowID string, isEnabl
 	return repo_model.UpdateRepoUnit(ctx, cfgUnit)
 }
 
-func ListActionWorkflows(ctx *context.APIContext) ([]*api.ActionWorkflow, error) {
-	defaultBranchCommit, err := ctx.Repo.GitRepo.GetBranchCommit(ctx.Repo.Repository.DefaultBranch)
-	if err != nil {
-		ctx.APIErrorInternal(err)
-		return nil, err
-	}
-
-	folder, entries, err := actions.ListWorkflows(defaultBranchCommit)
-	if err != nil {
-		ctx.APIError(http.StatusNotFound, err.Error())
-		return nil, err
-	}
-
-	workflows := make([]*api.ActionWorkflow, len(entries))
-	for i, entry := range entries {
-		workflows[i] = getActionWorkflowEntry(ctx, defaultBranchCommit, folder, entry)
-	}
-
-	return workflows, nil
-}
-
-func GetActionWorkflow(ctx *context.APIContext, workflowID string) (*api.ActionWorkflow, error) {
-	entries, err := ListActionWorkflows(ctx)
-	if err != nil {
-		return nil, err
-	}
-
-	for _, entry := range entries {
-		if entry.Name == workflowID {
-			return entry, nil
-		}
-	}
-
-	return nil, util.NewNotExistErrorf("workflow %q not found", workflowID)
-}
-
 func DispatchActionWorkflow(ctx reqctx.RequestContext, doer *user_model.User, repo *repo_model.Repository, gitRepo *git.Repository, workflowID, ref string, processInputs func(model *model.WorkflowDispatch, inputs map[string]any) error) error {
 	if workflowID == "" {
 		return util.ErrorWrapLocale(
@@ -285,6 +203,15 @@ func DispatchActionWorkflow(ctx reqctx.RequestContext, doer *user_model.User, re
 		log.Error("FindRunJobs: %v", err)
 	}
 	CreateCommitStatus(ctx, allJobs...)
+	if len(allJobs) > 0 {
+		job := allJobs[0]
+		err := job.LoadRun(ctx)
+		if err != nil {
+			log.Error("LoadRun: %v", err)
+		} else {
+			notify_service.WorkflowRunStatusUpdate(ctx, job.Run.Repo, job.Run.TriggerUser, job.Run)
+		}
+	}
 	for _, job := range allJobs {
 		notify_service.WorkflowJobStatusUpdate(ctx, repo, doer, job, nil)
 	}