1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package usecase
4
5import (
6 "context"
7 "encoding/json"
8 "errors"
9 "fmt"
10 "runtime/debug"
11 "slices"
12 "strings"
13
14 "github.com/bnema/zerowrap"
15
16 "bnema.dev/ermite-actions/workflow"
17 "bnema.dev/ermite/internal/domain"
18)
19
20// The workflow_run event (CI-3, D5). A run that is created ("requested") or
21// completed ("completed") triggers the workflows of the default branch that list
22// its workflow by name under workflow_run.workflows.
23//
24// The triggered run is read from the default branch's latest commit and runs on
25// it. It inherits the actor, the pusher and the trusted flag of the triggering
26// run, and never gets more: an agent's untrusted run cannot launder secrets
27// through a workflow_run (D1). A job token's push starts no run, but a run
28// started by a push of anyone else triggers workflow_run workflows even when the
29// triggering run was itself a workflow_run, up to the chain limit.
30const (
31 // maxWorkflowRunDepth is the number of workflow_run levels in a chain:
32 // docs.github.com ("Events that trigger workflows", workflow_run): "You can't
33 // use workflow_run to chain together more than three levels of workflows"
34 // (A -> B -> C -> D runs, E does not). Recalled from memory, not fetched.
35 maxWorkflowRunDepth = 3
36 runRequested = "requested"
37 runCompleted = "completed"
38)
39
40// runEvent is a change of a run that workflow_run workflows may be waiting for.
41type runEvent struct {
42 run domain.CIRun
43 action string
44 log zerowrap.Logger
45}
46
47// onRunEvent queues the "requested" or "completed" event of a run and returns:
48// a worker handles it, so the run's lock is never held while workflows are read.
49// A full queue drops it, logged, like the queue of pushes.
50func (c *CI) onRunEvent(ctx context.Context, run domain.CIRun, action string) {
51 c.mu.Lock()
52 closed := c.closed
53 c.mu.Unlock()
54 if closed {
55 return
56 }
57 e := runEvent{run: run, action: action, log: zerowrap.FromCtx(ctx)}
58 select {
59 case c.runEvents <- e:
60 default:
61 e.log.Error().Int64("run_id", run.ID).Str("action", action).Msg("ci workflow_run queue is full: the workflows waiting for this run are not run")
62 }
63}
64
65func (c *CI) runEventLoop() {
66 defer c.wg.Done()
67 for {
68 select {
69 case <-c.ctx.Done():
70 return
71 case e := <-c.runEvents:
72 c.planRunEvent(e)
73 }
74 }
75}
76
77// planRunEvent handles one event; a panic or an error is logged and never
78// escapes.
79func (c *CI) planRunEvent(e runEvent) {
80 ctx := zerowrap.WithCtx(c.ctx, e.log)
81 defer func() {
82 if r := recover(); r != nil {
83 e.log.Error().Interface("panic", r).Bytes("stack", debug.Stack()).Int64("run_id", e.run.ID).Msg("ci workflow_run panicked")
84 }
85 if c.cfg.RunEventHandled != nil {
86 c.cfg.RunEventHandled(e.run.ID, e.action)
87 }
88 }()
89 if err := c.handleRunEvent(ctx, e); err != nil && ctx.Err() == nil {
90 e.log.Error().Err(err).Int64("run_id", e.run.ID).Str("action", e.action).Msg("ci workflow_run failed")
91 }
92}
93
94// runLinks are the two facts of the github context of a run that workflow_run
95// reads back: the head branch of a merge request, and the run that triggered a
96// workflow_run run.
97type runLinks struct {
98 HeadRef string `json:"head_ref"`
99 Event struct {
100 WorkflowRun struct {
101 ID int64 `json:"id"`
102 } `json:"workflow_run"`
103 } `json:"event"`
104}
105
106func (c *CI) runLinks(ctx context.Context, runID int64) (runLinks, domain.CIRunSource, error) {
107 src, err := c.cfg.Store.GetRunSource(ctx, runID)
108 if err != nil {
109 return runLinks{}, src, err
110 }
111 var l runLinks
112 if err := json.Unmarshal([]byte(src.GithubJSON), &l); err != nil {
113 return runLinks{}, src, nil // a context that does not read: no link
114 }
115 return l, src, nil
116}
117
118// headBranchOf is workflow_run.head_branch: the source branch of a merge
119// request, else the branch or tag of the run.
120func headBranchOf(run domain.CIRun, l runLinks) string {
121 if l.HeadRef != "" {
122 return l.HeadRef
123 }
124 return strings.TrimPrefix(strings.TrimPrefix(run.Ref, "refs/heads/"), "refs/tags/")
125}
126
127// workflowRunDepth is the number of workflow_run levels above and including run:
128// 0 for a run that is not a workflow_run run. The walk stops at a triggering
129// run that cannot be found, which counts for nothing more.
130func (c *CI) workflowRunDepth(ctx context.Context, run domain.CIRun) int {
131 depth := 0
132 for cur := run; cur.Event == eventWorkflowRun && depth <= maxWorkflowRunDepth; {
133 depth++
134 l, _, err := c.runLinks(ctx, cur.ID)
135 if err != nil || l.Event.WorkflowRun.ID == 0 {
136 break
137 }
138 parent, err := c.cfg.Store.GetRun(ctx, cur.Repo, l.Event.WorkflowRun.ID)
139 if err != nil {
140 break
141 }
142 cur = parent
143 }
144 return depth
145}
146
147// matchWorkflowRun reports whether the workflow runs for the action of a run of
148// the workflow called name on headBranch: the event is listed, the name is in
149// "workflows", the action is one of its "types" (requested and completed when it
150// lists none; ermite has no in_progress activity), and the branches or
151// branches-ignore filter, which applies to the head branch of the triggering
152// run, lets it through (the filters of the push event, under another name).
153func matchWorkflowRun(wf *workflow.Workflow, name, action, headBranch string) bool {
154 if wf == nil || wf.On == nil {
155 return false
156 }
157 ev := wf.On.Event(eventWorkflowRun)
158 if ev == nil || !slices.Contains(ev.Workflows, name) {
159 return false
160 }
161 types := ev.Types
162 if types == nil {
163 types = []string{runRequested, runCompleted}
164 }
165 if !slices.Contains(types, action) {
166 return false
167 }
168 as := *ev
169 as.Name = "push"
170 probe := &workflow.Workflow{On: &workflow.On{Events: []*workflow.Event{&as}}}
171 return probe.MatchPush(workflow.PushInfo{Ref: "refs/heads/" + headBranch})
172}
173
174// handleRunEvent starts the workflow_run workflows of the default branch that
175// match the event.
176func (c *CI) handleRunEvent(ctx context.Context, e runEvent) error {
177 runners, err := c.cfg.Store.ListRunners(ctx)
178 if err != nil {
179 return fmt.Errorf("list the runners: %w", err)
180 }
181 if len(runners) == 0 {
182 return nil
183 }
184 if c.workflowRunDepth(ctx, e.run) >= maxWorkflowRunDepth {
185 e.log.Debug().Int64("run_id", e.run.ID).Msg("workflow_run chain limit reached: no run is triggered")
186 return nil
187 }
188 repo, err := c.cfg.Meta.Get(ctx, e.run.Repo)
189 if err != nil {
190 return nil // deleted meanwhile
191 }
192 def, exists, err := c.cfg.Repos.DefaultBranch(ctx, repo.Name)
193 if err != nil || !exists {
194 return nil
195 }
196 tip, err := c.cfg.Git.BranchHead(ctx, repo.Name, def)
197 if err != nil {
198 return nil
199 }
200 files, err := c.workflowFiles(ctx, repo.Name, tip)
201 if err != nil || len(files) == 0 {
202 return err
203 }
204 links, src, err := c.runLinks(ctx, e.run.ID)
205 if err != nil {
206 return nil // the run is gone
207 }
208 head := headBranchOf(e.run, links)
209
210 var matching []workflowFile
211 var parsed []*workflow.Workflow
212 for _, f := range files {
213 if f.problem != "" {
214 continue
215 }
216 wf, perrs := parseWorkflow(f.path, f.content)
217 if len(perrs) > 0 || !matchWorkflowRun(wf, e.run.WorkflowName, e.action, head) {
218 continue
219 }
220 matching, parsed = append(matching, f), append(parsed, wf)
221 }
222 if len(matching) == 0 {
223 return nil
224 }
225 commits, _, err := c.cfg.Git.ListRange(ctx, repo.Name, domain.ObjectID{}, tip, 1)
226 if err != nil || len(commits) == 0 {
227 return fmt.Errorf("read the head commit of %s: %w", def, err)
228 }
229 facts := pushFacts{
230 update: domain.RefUpdate{Ref: "refs/heads/" + def, Old: domain.ZeroOID, New: tip.String()}, commit: tip, head: commits[0],
231 defaultBranch: def,
232 custom: &customEvent{name: eventWorkflowRun, triggered: &workflowRunFacts{run: e.run, action: e.action, headBranch: head}},
233 }
234 // The actor, the pusher and the trust of the triggering run, never more.
235 t := ciTrigger{repo: repo.Name, actor: e.run.Actor, pusher: src.Pusher, trusted: e.run.Trusted, at: c.now(), log: e.log}
236 var errs []error
237 for i, f := range matching {
238 wf := parsed[i]
239 if problem := c.matrixProblem(ctx, repo, t, facts, f.path, wf); problem.text != "" {
240 name := f.path
241 if problem.named {
242 name = workflowName(wf, f.path)
243 }
244 errs = append(errs, c.startInvalid(ctx, repo, t, facts, f.path, name, problem.text))
245 continue
246 }
247 if _, err := c.launchRun(ctx, repo, t, facts, f.path, string(f.content), wf); err != nil && !errors.Is(err, context.Canceled) && !errors.Is(err, domain.ErrWorkflowDisabled) {
248 errs = append(errs, err)
249 }
250 }
251 return errors.Join(errs...)
252}