1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package usecase
4
5import (
6 "context"
7 "encoding/json"
8 "errors"
9 "fmt"
10 "io"
11 "path"
12 "runtime/debug"
13 "slices"
14 "strconv"
15 "strings"
16 "time"
17
18 "github.com/bnema/zerowrap"
19
20 "bnema.dev/ermite-actions/workflow"
21 "bnema.dev/ermite/internal/domain"
22 "bnema.dev/ermite/internal/ports"
23)
24
25const (
26 workflowDir = ".github/workflows"
27 // maxRunError bounds Run.error (8 KiB, run.proto).
28 maxRunError = 8 << 10
29 // maxPushCommits is the number of commits of the push event payload.
30 maxPushCommits = 20
31)
32
33// ciTrigger is a push waiting to be planned.
34type ciTrigger struct {
35 repo domain.RepoName
36 updates []domain.RefUpdate
37 actor string
38 at time.Time
39 log zerowrap.Logger
40}
41
42// OnPush queues the refs of a push or a merge for planning and returns at once;
43// a worker plans them. A full queue drops the trigger, logged at error: a push
44// is never slowed down or failed by CI.
45func (c *CI) OnPush(ctx context.Context, repo domain.RepoName, updates []domain.RefUpdate, pusher domain.Principal) {
46 var kept []domain.RefUpdate
47 for _, u := range updates {
48 if !u.IsDelete() && (strings.HasPrefix(u.Ref, "refs/heads/") || strings.HasPrefix(u.Ref, "refs/tags/")) {
49 kept = append(kept, u)
50 }
51 }
52 if len(kept) == 0 {
53 return
54 }
55 t := ciTrigger{repo: repo, updates: kept, actor: authorOf(pusher), at: c.now(), log: zerowrap.FromCtx(ctx)}
56 c.mu.Lock()
57 closed := c.closed
58 c.mu.Unlock()
59 if closed {
60 return
61 }
62 select {
63 case c.queue <- t:
64 default:
65 t.log.Error().Str("repo", repo.String()).Int("refs", len(kept)).Msg("ci trigger queue is full: the workflows of this push are not run")
66 }
67}
68
69func (c *CI) triggerLoop() {
70 defer c.wg.Done()
71 for {
72 select {
73 case <-c.ctx.Done():
74 return
75 case t := <-c.queue:
76 c.planPush(t)
77 }
78 }
79}
80
81// planPush plans one push; a panic or an error is logged and never escapes.
82func (c *CI) planPush(t ciTrigger) {
83 ctx := zerowrap.WithCtx(c.ctx, t.log)
84 defer func() {
85 if r := recover(); r != nil {
86 t.log.Error().Interface("panic", r).Bytes("stack", debug.Stack()).Str("repo", t.repo.String()).Msg("ci trigger panicked")
87 }
88 }()
89 if err := c.handlePush(ctx, t); err != nil && ctx.Err() == nil {
90 t.log.Error().Err(err).Str("repo", t.repo.String()).Msg("ci trigger failed")
91 }
92}
93
94// handlePush starts the workflows that match each ref of the push.
95func (c *CI) handlePush(ctx context.Context, t ciTrigger) error {
96 // CI is on once a runner is registered (plan, "Config"): without one a push
97 // makes no run and no check, so a forge that does not use CI shows nothing.
98 runners, err := c.cfg.Store.ListRunners(ctx)
99 if err != nil {
100 return fmt.Errorf("list the runners: %w", err)
101 }
102 if len(runners) == 0 {
103 return nil
104 }
105 tags := 0
106 for _, u := range t.updates {
107 if strings.HasPrefix(u.Ref, "refs/tags/") {
108 tags++
109 }
110 }
111 var errs []error
112 for _, u := range t.updates {
113 if err := c.handleRef(ctx, t, u, tags); err != nil {
114 errs = append(errs, fmt.Errorf("%s: %w", u.Ref, err))
115 }
116 }
117 return errors.Join(errs...)
118}
119
120// pushFacts is what is known of one pushed ref.
121type pushFacts struct {
122 update domain.RefUpdate
123 commit domain.ObjectID // the commit the ref points to, tags followed
124 head domain.Commit
125 commits []domain.Commit // oldest first, at most maxPushCommits (branches only)
126 forced bool
127 created bool
128 files []string
129 // filesTruncated: the list of changed files is not available.
130 filesTruncated bool
131 tagCount int
132 isTag bool
133 // defaultBranch is the default branch of the repository, "" when unknown.
134 defaultBranch string
135}
136
137func (c *CI) handleRef(ctx context.Context, t ciTrigger, u domain.RefUpdate, tagCount int) error {
138 isTag := strings.HasPrefix(u.Ref, "refs/tags/")
139 if isTag && tagCount > workflow.MaxTagsPerPush {
140 return nil // GitHub creates no event for the tags of such a push
141 }
142 repo, err := c.cfg.Meta.Get(ctx, t.repo)
143 if err != nil {
144 return nil // deleted meanwhile
145 }
146 facts, err := c.collect(ctx, t.repo, u, isTag, tagCount)
147 if err != nil {
148 return err
149 }
150 files, err := c.workflowFiles(ctx, t.repo, facts.commit)
151 if err != nil || len(files) == 0 {
152 return err
153 }
154 skip := workflow.SkipCI(facts.head.Message)
155 var errs []error
156 for _, f := range files {
157 wf, perrs := workflow.Parse(f.path, f.content)
158 switch {
159 case f.problem != "":
160 errs = append(errs, c.startInvalid(ctx, repo, t, facts, f.path, f.problem))
161 case len(perrs) > 0:
162 var lines []string
163 for _, e := range perrs {
164 lines = append(lines, e.Error())
165 }
166 errs = append(errs, c.startInvalid(ctx, repo, t, facts, f.path, strings.Join(lines, "\n")))
167 case skip || !wf.MatchPush(workflow.PushInfo{
168 Ref: u.Ref, Created: facts.created, ChangedFiles: facts.files, FilesTruncated: facts.filesTruncated, TagCount: tagCount,
169 }):
170 default:
171 errs = append(errs, c.startRun(ctx, repo, t, facts, f.path, string(f.content), wf))
172 }
173 }
174 return errors.Join(errs...)
175}
176
177// collect reads what the payload and the filters need from Git.
178func (c *CI) collect(ctx context.Context, repo domain.RepoName, u domain.RefUpdate, isTag bool, tagCount int) (pushFacts, error) {
179 newID, err := domain.ParseObjectID(u.New)
180 if err != nil {
181 return pushFacts{}, err
182 }
183 f := pushFacts{update: u, created: u.IsCreate(), isTag: isTag, tagCount: tagCount}
184 if def, _, err := c.cfg.Repos.DefaultBranch(ctx, repo); err == nil {
185 f.defaultBranch = def
186 }
187 res, err := c.cfg.Reader.ResolveRevision(ctx, repo, domain.RevisionOf(newID))
188 if err != nil {
189 return f, fmt.Errorf("resolve %s: %w", u.New, err)
190 }
191 f.commit = res.CommitID
192 if isTag {
193 cs, _, err := c.cfg.Git.ListRange(ctx, repo, domain.ObjectID{}, f.commit, 1)
194 if err != nil {
195 return f, fmt.Errorf("read the commit of the tag: %w", err)
196 }
197 if len(cs) == 0 {
198 return f, fmt.Errorf("read the commit of the tag: %w", domain.ErrNotFound)
199 }
200 f.head = cs[0]
201 return f, nil
202 }
203
204 var exclude domain.ObjectID
205 if !f.created {
206 if old, err := domain.ParseObjectID(u.Old); err == nil {
207 exclude = old
208 if ok, err := c.cfg.Git.IsAncestor(ctx, repo, old, f.commit); err == nil && !ok {
209 f.forced = true
210 }
211 }
212 }
213 base := exclude
214 if f.created {
215 base = c.defaultBase(ctx, repo, strings.TrimPrefix(u.Ref, "refs/heads/"), f.commit)
216 exclude = base
217 }
218 cs, more, err := c.cfg.Git.ListRange(ctx, repo, exclude, f.commit, ports.MaxCommitLimit)
219 if err != nil {
220 return f, fmt.Errorf("read the commits of the push: %w", err)
221 }
222 if len(cs) == 0 {
223 // A new branch at a commit the default branch has, or a branch rewound
224 // to an ancestor: no new commit, no changed file. GitHub still runs
225 // the push workflows that have no paths filter, for the head commit.
226 head, _, err := c.cfg.Git.ListRange(ctx, repo, domain.ObjectID{}, f.commit, 1)
227 if err != nil {
228 return f, fmt.Errorf("read the head commit of the push: %w", err)
229 }
230 if len(head) == 0 {
231 return f, fmt.Errorf("read the head commit of the push: %w", domain.ErrNotFound)
232 }
233 f.head = head[0]
234 return f, nil
235 }
236 f.head = cs[0]
237 n := min(len(cs), maxPushCommits)
238 f.commits = slices.Clone(cs[:n])
239 slices.Reverse(f.commits)
240 if more || base == f.commit {
241 // More than 1,000 commits: the diff is not looked at (it runs).
242 f.filesTruncated = more
243 }
244 if !more && base != f.commit {
245 paths, _, err := c.cfg.Diff.ChangedPaths(ctx, repo, base, f.commit, workflow.MaxFilesEvaluated)
246 if err != nil {
247 f.filesTruncated = true // diff unavailable: runs
248 } else {
249 f.files = paths
250 }
251 }
252 return f, nil
253}
254
255// defaultBase is the merge base of a new branch with the default branch, the
256// zero id when there is none (every file of the branch counts).
257func (c *CI) defaultBase(ctx context.Context, repo domain.RepoName, branch string, head domain.ObjectID) domain.ObjectID {
258 def, exists, err := c.cfg.Repos.DefaultBranch(ctx, repo)
259 if err != nil || !exists || def == branch {
260 return domain.ObjectID{}
261 }
262 tip, err := c.cfg.Git.BranchHead(ctx, repo, def)
263 if err != nil {
264 return domain.ObjectID{}
265 }
266 base, err := c.cfg.Git.MergeBase(ctx, repo, tip, head)
267 if err != nil {
268 return domain.ObjectID{}
269 }
270 return base
271}
272
273type workflowFile struct {
274 path string
275 content []byte
276 // problem is set when the file cannot even be parsed (too large).
277 problem string
278}
279
280// workflowFiles reads .github/workflows/*.yml|*.yaml at the commit.
281func (c *CI) workflowFiles(ctx context.Context, repo domain.RepoName, commit domain.ObjectID) ([]workflowFile, error) {
282 dir, err := domain.ParseTreePath(workflowDir)
283 if err != nil {
284 return nil, err
285 }
286 tree, err := c.cfg.Reader.GetTree(ctx, repo, domain.RevisionOf(commit), dir, false)
287 if errors.Is(err, domain.ErrNotFound) || errors.Is(err, domain.ErrNotDirectory) {
288 return nil, nil
289 }
290 if err != nil {
291 return nil, fmt.Errorf("list %s: %w", workflowDir, err)
292 }
293 var out []workflowFile
294 for _, e := range tree.Entries {
295 ext := strings.ToLower(path.Ext(e.Name))
296 if (e.Kind != domain.EntryFile && e.Kind != domain.EntryExecutable) || (ext != ".yml" && ext != ".yaml") {
297 continue
298 }
299 if len(out) == maxCIWorkflowFiles {
300 break
301 }
302 wf := workflowFile{path: e.Path}
303 if wf.path == "" {
304 wf.path = workflowDir + "/" + e.Name
305 }
306 if e.Size > maxCIWorkflowBytes {
307 wf.problem = fmt.Sprintf("%s: the workflow file is larger than 1 MiB", wf.path)
308 out = append(out, wf)
309 continue
310 }
311 tp, err := domain.ParseTreePath(wf.path)
312 if err != nil {
313 continue
314 }
315 _, rc, err := c.cfg.Reader.OpenBlob(ctx, repo, domain.RevisionOf(commit), tp)
316 if err != nil {
317 return nil, fmt.Errorf("read %s: %w", wf.path, err)
318 }
319 wf.content, err = io.ReadAll(io.LimitReader(rc, maxCIWorkflowBytes+1))
320 _ = rc.Close()
321 if err != nil {
322 return nil, fmt.Errorf("read %s: %w", wf.path, err)
323 }
324 if len(wf.content) > maxCIWorkflowBytes {
325 wf.content, wf.problem = nil, fmt.Sprintf("%s: the workflow file is larger than 1 MiB", wf.path)
326 }
327 out = append(out, wf)
328 }
329 return out, nil
330}
331
332// --- the run ---
333
334func firstLine(s string) string {
335 line, _, _ := strings.Cut(s, "\n")
336 return truncateRunes(strings.TrimSpace(line), 256)
337}
338
339// startInvalid records the failed run of an invalid workflow file: no job, the
340// errors as Run.error, like GitHub's "Invalid workflow file".
341func (c *CI) startInvalid(ctx context.Context, repo domain.Repo, t ciTrigger, f pushFacts, wfPath, problem string) error {
342 now := c.now()
343 draft, err := c.cfg.Store.CreateRun(ctx, repo.Name, wfPath, func(id, number int64) ports.CIRunDraft {
344 run := domain.CIRun{
345 Repo: repo.Name, ID: id, Number: number, Attempt: 1, WorkflowPath: wfPath, WorkflowName: wfPath,
346 Title: firstLine(f.head.Message), Event: "push", Ref: f.update.Ref, Sha: f.commit.String(), Actor: t.actor,
347 Status: domain.CICompleted, Conclusion: domain.CIFailure, Error: cutText(problem, maxRunError),
348 CreatedAt: now, StartedAt: now, CompletedAt: now,
349 }
350 return ports.CIRunDraft{
351 Run: run, Source: domain.CIRunSource{Repo: repo.Name, WorkflowYAML: "", GithubJSON: "{}"},
352 Checks: []domain.CICheck{{
353 ID: "run-" + strconv.FormatInt(id, 10), Repo: repo.Name, Sha: run.Sha, Name: wfPath, Status: domain.CICompleted,
354 Conclusion: domain.CIFailure, RunID: id, Attempt: 1, WorkflowName: wfPath, CreatedAt: now, StartedAt: now, CompletedAt: now,
355 }},
356 }
357 })
358 if errors.Is(err, domain.ErrNotFound) {
359 return nil
360 }
361 if err != nil {
362 return err
363 }
364 c.publishRun(ctx, repo, draft.Run)
365 log := logRun(ctx, draft.Run)
366 log.Info().Str("workflow", wfPath).Msg("ci run failed: invalid workflow file")
367 return nil
368}
369
370func (c *CI) publishRun(ctx context.Context, repo domain.Repo, run domain.CIRun) {
371 c.cfg.Bus.Publish(ctx, domain.RunUpdated{Run: run, At: c.now(), Visibility: repo.Visibility, CreatedAt: repo.CreatedAt})
372}
373
374// startRun creates the run of a matching workflow with one waiting record per
375// job of the file, then plans the jobs that can start.
376func (c *CI) startRun(ctx context.Context, repo domain.Repo, t ciTrigger, f pushFacts, wfPath, yaml string, wf *workflow.Workflow) error {
377 now := c.now()
378 name := wf.Name
379 if name == "" {
380 name = wfPath
381 }
382 draft, err := c.cfg.Store.CreateRun(ctx, repo.Name, wfPath, func(id, number int64) ports.CIRunDraft {
383 run := domain.CIRun{
384 Repo: repo.Name, ID: id, Number: number, Attempt: 1, WorkflowPath: wfPath, WorkflowName: name,
385 Title: firstLine(f.head.Message), Event: "push", Ref: f.update.Ref, Sha: f.commit.String(), Actor: t.actor,
386 Status: domain.CIQueued, CreatedAt: now,
387 }
388 d := ports.CIRunDraft{Run: run, Source: domain.CIRunSource{
389 Repo: repo.Name, WorkflowYAML: yaml, GithubJSON: c.githubJSON(ctx, repo, run, f, wf),
390 }}
391 for i, j := range wf.Jobs {
392 job := domain.CIJob{
393 ID: domain.NewJobID(id, i, 0), RunID: id, Attempt: 1, RunAttempt: 1, Repo: repo.Name, Key: j.ID, Name: j.ID,
394 Phase: domain.CIPhaseWaiting, Needs: slices.Clone(j.Needs), Index: i, CreatedAt: now,
395 }
396 d.Jobs = append(d.Jobs, job)
397 d.Checks = append(d.Checks, domain.CICheck{
398 ID: job.ID, Repo: repo.Name, Sha: run.Sha, Name: job.Name, Status: domain.CIQueued, RunID: id, JobID: job.ID,
399 Attempt: 1, WorkflowName: name, CreatedAt: now,
400 })
401 }
402 return d
403 })
404 if errors.Is(err, domain.ErrNotFound) {
405 return nil
406 }
407 if err != nil {
408 return err
409 }
410 c.publishRun(ctx, repo, draft.Run)
411 for _, j := range draft.Jobs {
412 c.cfg.Bus.Publish(ctx, domain.JobUpdated{Job: j, At: now, Visibility: repo.Visibility, CreatedAt: repo.CreatedAt})
413 }
414 log := logRun(ctx, draft.Run)
415 log.Info().Str("workflow", wfPath).Str("ref", f.update.Ref).Int("jobs", len(draft.Jobs)).Msg("ci run created")
416 return c.advanceRun(ctx, draft.Run.ID)
417}
418
419// --- the github context ---
420
421// shortID is the 12-character form of an object id that GitHub puts in the
422// compare URL of a push.
423func shortID(id string) string {
424 if len(id) > 12 {
425 return id[:12]
426 }
427 return id
428}
429
430// retentionDays is the retention in days, as github.retention_days shows it
431// (at least 1).
432func (c *CI) retentionDays() int {
433 return max(int(c.cfg.Retention/(24*time.Hour)), 1)
434}
435
436type obj = map[string]any
437
438func person(name, email string) obj { return obj{"name": name, "email": email} }
439
440func (c *CI) commitJSON(ctx context.Context, repo domain.Repo, cm domain.Commit) obj {
441 ts := cm.Committer.Time.UTC().Format(time.RFC3339)
442 tree := ""
443 if id, err := c.cfg.Git.CommitTree(ctx, repo.Name, cm.ID); err == nil {
444 tree = id.String()
445 }
446 return obj{
447 "id": cm.ID.String(), "tree_id": tree, "distinct": true, "message": cm.Message, "timestamp": ts,
448 "url": c.cfg.PublicURL + "/" + repo.Name.String() + "/commit/" + cm.ID.String(),
449 "author": obj{"name": cm.Author.Name, "email": cm.Author.Email, "username": ""},
450 "committer": obj{"name": cm.Committer.Name, "email": cm.Committer.Email, "username": ""},
451 "added": []string{}, "removed": []string{}, "modified": []string{},
452 }
453}
454
455// githubJSON builds the "github" context of the run as GitHub does for a push
456// (the job-dependent keys are added when a job is planned). Values that
457// GitHub's documentation does not settle are marked for the recordings.
458//
459// Checked against the recordings of ermite-conformance (push-branch-create,
460// push-tag, push-force, ctx-github-names): ids and counters are strings, so is
461// retention_days; "compare" has 12-character ids, the commit page for a created
462// ref and "<before>...<after>" for an update; repository.default_branch and
463// master_branch are the default branch.
464//
465// TODO(verify with recording): the empty added/removed/modified lists of the
466// push commits.
467func (c *CI) githubJSON(ctx context.Context, repo domain.Repo, run domain.CIRun, f pushFacts, wf *workflow.Workflow) string {
468 full := c.cfg.Owner + "/" + repo.Name.String()
469 refName, refType := strings.TrimPrefix(run.Ref, "refs/heads/"), "branch"
470 if f.isTag {
471 refName, refType = strings.TrimPrefix(run.Ref, "refs/tags/"), "tag"
472 }
473 repoURL := c.cfg.PublicURL + "/" + repo.Name.String()
474 before := f.update.Old
475 if f.created {
476 before = domain.ZeroOID
477 }
478 compare := repoURL + "/compare/" + shortID(before) + "..." + shortID(f.update.New)
479 if f.created {
480 compare = repoURL + "/commit/" + shortID(f.update.New)
481 }
482 commits := make([]any, 0, len(f.commits))
483 for _, cm := range f.commits {
484 commits = append(commits, c.commitJSON(ctx, repo, cm))
485 }
486 event := obj{
487 "ref": run.Ref, "before": before, "after": f.update.New, "created": f.created, "deleted": false, "forced": f.forced,
488 "base_ref": nil, "compare": compare, "commits": commits, "head_commit": c.commitJSON(ctx, repo, f.head),
489 "pusher": person(run.Actor, ""),
490 "repository": obj{
491 "name": repo.Name.String(), "full_name": full, "private": repo.Visibility == domain.VisibilityPrivate,
492 "owner": obj{"login": c.cfg.Owner, "name": c.cfg.Owner}, "html_url": repoURL, "clone_url": repoURL + ".git",
493 "default_branch": f.defaultBranch, "master_branch": f.defaultBranch, "url": repoURL,
494 },
495 "sender": obj{"login": run.Actor, "type": "User"},
496 }
497 wfRef := full + "/" + run.WorkflowPath + "@" + run.Ref
498 gh := obj{
499 "event_name": "push", "event": event, "ref": run.Ref, "ref_name": refName, "ref_type": refType, "ref_protected": false,
500 "sha": run.Sha, "repository": full, "repository_owner": c.cfg.Owner, "repositoryUrl": repoURL + ".git",
501 "actor": run.Actor, "triggering_actor": run.Actor, "workflow": run.WorkflowName, "workflow_ref": wfRef,
502 "workflow_sha": run.Sha, "run_id": strconv.FormatInt(run.ID, 10), "run_number": strconv.FormatInt(run.Number, 10),
503 "server_url": c.cfg.PublicURL, "api_url": c.cfg.PublicURL + "/-/github/api/v3",
504 "graphql_url": c.cfg.PublicURL + "/-/github/api/graphql", "retention_days": strconv.Itoa(c.retentionDays()),
505 "head_ref": "", "base_ref": "",
506 }
507 b, err := json.Marshal(gh)
508 if err != nil {
509 return "{}"
510 }
511 return string(b)
512}