1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package usecase
4
5import (
6 "context"
7 "encoding/base64"
8 "errors"
9 "fmt"
10 "path"
11 "slices"
12 "strconv"
13 "strings"
14 "time"
15
16 "bnema.dev/ermite/internal/domain"
17 "bnema.dev/ermite/internal/ports"
18)
19
20const runTokenPrefix = "ru1:"
21
22// --- runs and jobs ---
23
24// ListRuns returns a page of the runs of a repository, newest first. The
25// principal needs the access of reading the repository.
26func (c *CI) ListRuns(ctx context.Context, p domain.Principal, name domain.RepoName, params ports.ListRunsParams) (ports.RunPage, error) {
27 if _, err := c.readableRepo(ctx, p, name); err != nil {
28 return ports.RunPage{}, err
29 }
30 n, err := pageSize(params.PageSize)
31 if err != nil {
32 return ports.RunPage{}, err
33 }
34 var before int64
35 if params.PageToken != "" {
36 raw, ok := decodeToken(params.PageToken, runTokenPrefix)
37 id, perr := strconv.ParseInt(raw, 10, 64)
38 if !ok || perr != nil || id < 1 {
39 return ports.RunPage{}, fmt.Errorf("%w: runs", domain.ErrInvalidPageToken)
40 }
41 before = id
42 }
43 match := runFilter(params)
44 // The store returns runs by id; the filters are applied while paging, so a
45 // page of a filtered listing is full unless the runs ran out.
46 var kept []domain.CIRun
47 cursor := before
48 for len(kept) <= n {
49 batch, err := c.cfg.Store.ListRuns(ctx, name, cursor, n+1)
50 if err != nil {
51 return ports.RunPage{}, err
52 }
53 for _, r := range batch {
54 if match(r) {
55 kept = append(kept, r)
56 }
57 }
58 if len(batch) < n+1 {
59 break
60 }
61 cursor = batch[len(batch)-1].ID
62 }
63 page := ports.RunPage{Runs: kept}
64 if len(kept) > n {
65 page.Runs = kept[:n]
66 page.NextPageToken = base64.RawURLEncoding.EncodeToString([]byte(runTokenPrefix + strconv.FormatInt(kept[n-1].ID, 10)))
67 }
68 return page, nil
69}
70
71// runFilter returns the test of the filters of a listing.
72func runFilter(p ports.ListRunsParams) func(domain.CIRun) bool {
73 return func(r domain.CIRun) bool {
74 switch {
75 case p.Status != "" && r.Status != p.Status:
76 return false
77 case p.Branch != "" && r.Ref != "refs/heads/"+p.Branch:
78 return false
79 case p.Workflow != "" && r.WorkflowPath != p.Workflow && path.Base(r.WorkflowPath) != p.Workflow:
80 return false
81 }
82 return true
83 }
84}
85
86// runOf finds a run of the repository; a run of another repository is not
87// found.
88func (c *CI) runOf(ctx context.Context, name domain.RepoName, runID int64) (domain.CIRun, error) {
89 run, err := c.cfg.Store.GetRun(ctx, name, runID)
90 if err != nil {
91 return domain.CIRun{}, err
92 }
93 return run, nil
94}
95
96// view adds the live line count to a job.
97func (c *CI) view(ctx context.Context, j domain.CIJob) ports.JobView {
98 v := ports.JobView{CIJob: j}
99 if j.Phase != domain.CIPhaseWaiting && j.Phase != domain.CIPhaseReady {
100 v.Lines, _ = c.cfg.Logs.Count(ctx, logRef(j))
101 }
102 if j.RunnerID != "" {
103 if _, err := c.cfg.Store.GetRunner(ctx, j.RunnerID); errors.Is(err, domain.ErrNotFound) {
104 v.RunnerGone = true
105 }
106 }
107 return v
108}
109
110// GetRun returns a run with the jobs of an attempt (0 for the latest).
111func (c *CI) GetRun(ctx context.Context, p domain.Principal, name domain.RepoName, runID int64, attempt int32) (ports.RunDetail, error) {
112 if _, err := c.readableRepo(ctx, p, name); err != nil {
113 return ports.RunDetail{}, err
114 }
115 run, err := c.runOf(ctx, name, runID)
116 if err != nil {
117 return ports.RunDetail{}, err
118 }
119 latest := run.Attempt
120 if attempt < 0 || attempt > latest {
121 return ports.RunDetail{}, domain.ErrNotFound
122 }
123 if attempt == 0 {
124 attempt = latest
125 }
126 jobs, err := c.cfg.Store.ListJobs(ctx, runID, attempt)
127 if err != nil {
128 return ports.RunDetail{}, err
129 }
130 if attempt != latest {
131 run = c.runAtAttempt(run, jobs, attempt)
132 }
133 d := ports.RunDetail{Run: run, LatestAttempt: latest}
134 for _, j := range jobs {
135 d.Jobs = append(d.Jobs, c.view(ctx, j))
136 }
137 return d, nil
138}
139
140// runAtAttempt describes an earlier attempt of a run from its jobs: the run
141// record holds the latest attempt only.
142func (c *CI) runAtAttempt(run domain.CIRun, jobs []domain.CIJob, attempt int32) domain.CIRun {
143 rs := &runState{c: c, run: run, jobs: jobs, attempt: attempt}
144 r := rs.recompute(c.now())
145 r.Attempt, r.Version = attempt, run.Version
146 r.CancelRequested = false
147 return r
148}
149
150// jobOf finds a job of the repository by its id and attempt (0 for the
151// latest). The attempt is the one of the run it was kept by, or executed in.
152func (c *CI) jobOf(ctx context.Context, name domain.RepoName, jobID string, attempt int32) (domain.CIRun, domain.CIJob, error) {
153 runID, _, _, ok := domain.ParseJobID(jobID)
154 if !ok {
155 return domain.CIRun{}, domain.CIJob{}, domain.ErrNotFound
156 }
157 run, err := c.runOf(ctx, name, runID)
158 if err != nil {
159 return domain.CIRun{}, domain.CIJob{}, err
160 }
161 if attempt < 0 || attempt > run.Attempt {
162 return domain.CIRun{}, domain.CIJob{}, domain.ErrNotFound
163 }
164 if attempt == 0 {
165 attempt = run.Attempt
166 }
167 j, err := c.cfg.Store.GetJob(ctx, runID, attempt, jobID)
168 if err != nil {
169 return domain.CIRun{}, domain.CIJob{}, err
170 }
171 return run, j, nil
172}
173
174// GetJob returns one job with its steps.
175func (c *CI) GetJob(ctx context.Context, p domain.Principal, name domain.RepoName, jobID string, attempt int32) (ports.JobView, error) {
176 if _, err := c.readableRepo(ctx, p, name); err != nil {
177 return ports.JobView{}, err
178 }
179 _, j, err := c.jobOf(ctx, name, jobID, attempt)
180 if err != nil {
181 return ports.JobView{}, err
182 }
183 return c.view(ctx, j), nil
184}
185
186// --- log streaming ---
187
188// acquireStream counts a log stream against the limits.
189func (c *CI) acquireStream(client string) (release func(), err error) {
190 c.mu.Lock()
191 defer c.mu.Unlock()
192 if c.streams >= MaxCILogStreams || (client != "" && c.perClient[client] >= MaxCILogStreamsPerClient) {
193 return nil, fmt.Errorf("%w: too many log streams", domain.ErrResourceExhausted)
194 }
195 c.streams++
196 if client != "" {
197 c.perClient[client]++
198 }
199 return func() {
200 c.mu.Lock()
201 defer c.mu.Unlock()
202 c.streams--
203 if client != "" {
204 if c.perClient[client]--; c.perClient[client] <= 0 {
205 delete(c.perClient, client)
206 }
207 }
208 }, nil
209}
210
211// GetJobLog streams the lines of a job log from an offset, in batches of at
212// most 1,000 lines and 64 KiB. With Follow it goes on while the job runs, and
213// ends once the job completed and every line was sent.
214func (c *CI) GetJobLog(ctx context.Context, p domain.Principal, name domain.RepoName, params ports.GetJobLogParams, send func(ports.LogBatch) error) error {
215 if _, err := c.readableRepo(ctx, p, name); err != nil {
216 return err
217 }
218 if params.Offset < 0 {
219 return fmt.Errorf("%w: negative offset", domain.ErrInvalid)
220 }
221 _, j, err := c.jobOf(ctx, name, params.JobID, params.Attempt)
222 if err != nil {
223 return err
224 }
225 release, err := c.acquireStream(params.Client)
226 if err != nil {
227 return err
228 }
229 defer release()
230 if params.Ready != nil {
231 params.Ready()
232 }
233 ref := logRef(j)
234 jref := jobRef(j)
235 offset := params.Offset
236 checked := time.Now()
237 // revalidate asks again whether the token of a stored-token caller still
238 // exists, at most every few seconds, whether the stream is busy or idle: a
239 // token revoked while a long log is sent stops the stream.
240 revalidate := func() error {
241 if p.TokenID() == "" || c.cfg.Auth == nil || time.Since(checked) <= DefaultRevalidateEvery {
242 return nil
243 }
244 if err := c.cfg.Auth.Revalidate(ctx, p); err != nil {
245 return err
246 }
247 checked = time.Now()
248 return nil
249 }
250 for {
251 wakeJob, wakeLog := c.jobs.wait(jref), c.cfg.Logs.Changed(ref)
252 cur, err := c.cfg.Store.GetJob(ctx, j.RunID, j.RunAttempt, j.ID)
253 if err != nil {
254 return err
255 }
256 done := cur.Ended()
257 lines, err := c.cfg.Logs.Read(ctx, ref, offset, domain.CIMaxLogLines, domain.CIMaxLogChunkBytes)
258 switch {
259 case errors.Is(err, domain.ErrNotFound), errors.Is(err, domain.ErrOutOfRange):
260 // No line yet (a queued job has no log), or fewer lines than the
261 // offset. Without follow, and for a job that ended (follow changes
262 // nothing then), that is the answer (contract: not_found,
263 // out_of_range). A job that runs may still write the line: wait.
264 if !params.Follow || done {
265 if errors.Is(err, domain.ErrNotFound) {
266 return domain.ErrNotFound
267 }
268 return fmt.Errorf("%w: the log has fewer than %d lines", domain.ErrOutOfRange, offset)
269 }
270 lines = nil
271 case err != nil:
272 return err
273 }
274 if len(lines) > 0 {
275 if err := revalidate(); err != nil {
276 return err
277 }
278 batch := ports.LogBatch{Offset: offset, Lines: c.withSteps(ctx, cur, offset, lines)}
279 if err := send(batch); err != nil {
280 return err
281 }
282 offset += int64(len(lines))
283 continue // more may be waiting
284 }
285 if !params.Follow || done {
286 return nil
287 }
288 if err := revalidate(); err != nil {
289 return err
290 }
291 select {
292 case <-wakeJob:
293 case <-wakeLog:
294 case <-time.After(DefaultRevalidateEvery):
295 case <-ctx.Done():
296 return nil
297 case <-c.ctx.Done():
298 return nil
299 }
300 }
301}
302
303// withSteps tags the lines of a batch with their step from the step ranges.
304func (c *CI) withSteps(_ context.Context, j domain.CIJob, offset int64, lines []domain.CILogLine) []ports.JobLogLine {
305 out := make([]ports.JobLogLine, len(lines))
306 for i, l := range lines {
307 out[i] = ports.JobLogLine{Step: domain.StepOfLine(j.Steps, offset+int64(i)), CILogLine: l}
308 }
309 return out
310}
311
312// --- cancel and rerun ---
313
314// CancelRun cancels the latest attempt of a run. The principal needs ci:write
315// on the repository. Jobs not given to a runner end at once; the others get a
316// cancel request, and the run completes when they have ended.
317func (c *CI) CancelRun(ctx context.Context, p domain.Principal, name domain.RepoName, runID int64) (domain.CIRun, error) {
318 if _, err := c.writableRepo(ctx, p, name); err != nil {
319 return domain.CIRun{}, err
320 }
321 run, err := c.runOf(ctx, name, runID)
322 if err != nil {
323 return domain.CIRun{}, err
324 }
325 if err := runWriteAccess(p, run); err != nil {
326 return domain.CIRun{}, err
327 }
328 var out domain.CIRun
329 err = c.withRun(ctx, runID, 0, func(rs *runState) error {
330 if rs.run.Status == domain.CICompleted {
331 return fmt.Errorf("%w: run %d has completed", domain.ErrPreconditionFailed, runID)
332 }
333 if !rs.run.CancelRequested {
334 rs.run.CancelRequested, rs.run.CancelRequestedAt, rs.runDirty = true, c.now(), true
335 rs.cancelAll("Canceled by " + authorOf(p))
336 }
337 if err := rs.flush(ctx); err != nil {
338 return err
339 }
340 out = rs.run
341 return nil
342 })
343 if err != nil {
344 return domain.CIRun{}, err
345 }
346 ctxLog(ctx).Info().Str("repo", name.String()).Int64("run_id", runID).Str("principal", p.Label()).Str("token_id", p.TokenID()).Msg("ci run cancelled")
347 return out, nil
348}
349
350// RerunRun starts a new attempt of a completed run, from the stored workflow
351// file: all the jobs, or with failedOnly the jobs that did not succeed and
352// those that need them; the others are kept with their result, steps and log.
353func (c *CI) RerunRun(ctx context.Context, p domain.Principal, name domain.RepoName, runID int64, failedOnly bool) (domain.CIRun, error) {
354 repo, err := c.writableRepo(ctx, p, name)
355 if err != nil {
356 return domain.CIRun{}, err
357 }
358 run, err := c.runOf(ctx, name, runID)
359 if err != nil {
360 return domain.CIRun{}, err
361 }
362 if err := runWriteAccess(p, run); err != nil {
363 return domain.CIRun{}, err
364 }
365 err = c.withRun(ctx, runID, 0, func(rs *runState) error {
366 if rs.run.Status != domain.CICompleted {
367 return fmt.Errorf("%w: run %d has not completed", domain.ErrPreconditionFailed, runID)
368 }
369 if len(rs.jobs) == 0 {
370 return fmt.Errorf("%w: run %d has no job to run again", domain.ErrPreconditionFailed, runID)
371 }
372 redo, err := rerunSet(rs, failedOnly)
373 if err != nil {
374 return err
375 }
376 next := rs.run.Attempt + 1
377 now := c.now()
378 w := ports.CIWrite{}
379 var fresh []domain.CIJob
380 for _, j := range rs.jobs {
381 if !redo[j.Key] {
382 kept := j // kept with its own attempt, steps and log
383 kept.RunAttempt, kept.Version = next, 0
384 w.Jobs = append(w.Jobs, kept)
385 fresh = append(fresh, kept)
386 continue
387 }
388 if j.Combo > 0 {
389 continue // the matrix is expanded again when the job is planned
390 }
391 n := domain.CIJob{
392 ID: domain.NewJobID(runID, j.Index, 0), RunID: runID, Attempt: next, RunAttempt: next, Repo: rs.run.Repo,
393 Key: j.Key, Name: j.Key, Phase: domain.CIPhaseWaiting, Needs: slices.Clone(j.Needs), Index: j.Index, CreatedAt: now,
394 }
395 w.Jobs = append(w.Jobs, n)
396 fresh = append(fresh, n)
397 }
398 run := rs.run
399 run.Attempt, run.Status, run.Conclusion = next, domain.CIQueued, domain.CINoConclusion
400 run.StartedAt, run.CompletedAt, run.CreatedAt = time.Time{}, time.Time{}, now
401 run.CancelRequested, run.CancelRequestedAt = false, time.Time{}
402 run.Concurrency = domain.CIConcurrency{} // evaluated again: the re-run joins its group again
403 // Kept or lowered, never raised: the owner who runs an agent's run
404 // again does not give it secrets (D1).
405 run.Trusted = domain.RerunTrusted(rs.run.Trusted, p)
406 w.Runs = []domain.CIRun{run}
407 rs2 := &runState{c: c, run: run, attempt: next, jobs: fresh, src: rs.src, dirty: map[string]bool{}, checks: map[string]bool{}}
408 for _, j := range fresh {
409 w.Checks = append(w.Checks, rs2.checkOf(j))
410 }
411 v, err := c.cfg.Store.Apply(ctx, w)
412 if err != nil {
413 return err
414 }
415 run.Version = v
416 c.flows.drop(runID)
417 c.publishRun(ctx, repo, run)
418 for _, j := range w.Jobs {
419 j.Version = v
420 c.cfg.Bus.Publish(ctx, domain.JobUpdated{Job: j, At: now, Visibility: repo.Visibility, CreatedAt: repo.CreatedAt})
421 }
422 return nil
423 })
424 if err != nil {
425 return domain.CIRun{}, err
426 }
427 // Plan the new attempt: the kept jobs count as the results of their needs.
428 if err := c.advanceRun(ctx, runID); err != nil {
429 return domain.CIRun{}, err
430 }
431 run, err = c.runOf(ctx, name, runID)
432 if err != nil {
433 return domain.CIRun{}, err
434 }
435 ctxLog(ctx).Info().Str("repo", name.String()).Int64("run_id", runID).Int32("attempt", run.Attempt).Bool("failed_only", failedOnly).
436 Str("principal", p.Label()).Str("token_id", p.TokenID()).Msg("ci run started again")
437 return run, nil
438}
439
440// rerunSet returns the keys of the jobs to run again: all of them, or the ones
441// that did not succeed and those that need them, directly or not.
442func rerunSet(rs *runState, failedOnly bool) (map[string]bool, error) {
443 redo := map[string]bool{}
444 if !failedOnly {
445 for _, j := range rs.jobs {
446 redo[j.Key] = true
447 }
448 return redo, nil
449 }
450 for _, j := range rs.jobs {
451 switch j.Conclusion {
452 case domain.CIFailure, domain.CICancelled, domain.CITimedOut, domain.CIActionRequired:
453 redo[j.Key] = true
454 }
455 }
456 if len(redo) == 0 {
457 return nil, fmt.Errorf("%w: no job of run %d failed", domain.ErrPreconditionFailed, rs.run.ID)
458 }
459 for changed := true; changed; {
460 changed = false
461 for _, j := range rs.jobs {
462 if redo[j.Key] {
463 continue
464 }
465 if slices.ContainsFunc(j.Needs, func(n string) bool { return redo[n] }) {
466 redo[j.Key], changed = true, true
467 }
468 }
469 }
470 return redo, nil
471}
472
473func (w *wfCache) drop(runID int64) {
474 w.mu.Lock()
475 delete(w.m, runID)
476 delete(w.uses, runID)
477 w.mu.Unlock()
478}
479
480// --- checks ---
481
482// CommitChecks returns the checks of a commit, without access control. A
483// commit with more than MaxCIChecks checks is an error (domain.ErrResourceExhausted):
484// the merge rule cannot answer from a part of them.
485func (c *CI) CommitChecks(ctx context.Context, repo domain.RepoName, sha string) ([]domain.CICheck, error) {
486 checks, more, err := c.cfg.Store.ListChecks(ctx, repo, sha, MaxCIChecks)
487 if err != nil {
488 return nil, err
489 }
490 if more {
491 return nil, fmt.Errorf("%w: the commit has more than %d checks", domain.ErrResourceExhausted, MaxCIChecks)
492 }
493 return checks, nil
494}
495
496// ListChecks returns the checks of a commit: the one of each job of the latest
497// attempt of each run on it. A commit without a run has none.
498func (c *CI) ListChecks(ctx context.Context, p domain.Principal, name domain.RepoName, sha string) (ports.ChecksResult, error) {
499 if _, err := c.readableRepo(ctx, p, name); err != nil {
500 return ports.ChecksResult{}, err
501 }
502 if !isHex40(sha) {
503 return ports.ChecksResult{}, fmt.Errorf("%w: a commit is 40 lowercase hexadecimal characters", domain.ErrInvalidObjectID)
504 }
505 checks, more, err := c.cfg.Store.ListChecks(ctx, name, sha, MaxCIChecks)
506 if err != nil {
507 return ports.ChecksResult{}, err
508 }
509 res := ports.ChecksResult{Truncated: more}
510 var runners []domain.CIRunner
511 for _, ck := range checks {
512 v := ports.CheckView{CICheck: ck}
513 if ck.Status == domain.CIQueued {
514 if runners == nil {
515 runners, _ = c.cfg.Store.ListRunners(ctx)
516 }
517 v.WaitingForRunner = c.noRunnerFor(ctx, ck, runners)
518 }
519 res.Checks = append(res.Checks, v)
520 }
521 return res, nil
522}
523
524// noRunnerFor reports whether no registered runner can take a queued check's
525// job. A job that waits for its needs is not waiting for a runner.
526func (c *CI) noRunnerFor(ctx context.Context, ck domain.CICheck, runners []domain.CIRunner) bool {
527 if ck.JobID == "" {
528 return false
529 }
530 j, err := c.cfg.Store.GetJob(ctx, ck.RunID, ck.Attempt, ck.JobID)
531 if err != nil || j.Phase != domain.CIPhaseReady {
532 return false
533 }
534 return !slices.ContainsFunc(runners, func(r domain.CIRunner) bool { return domain.LabelsMatch(j.Labels, r.Labels) })
535}
536
537// --- runners ---
538
539// ListRunners returns a page of the runners ordered by name. The principal
540// needs runner:register on a token that is not restricted to repositories.
541func (c *CI) ListRunners(ctx context.Context, p domain.Principal, size int, token string) (ports.RunnerPage, error) {
542 if err := canManageRunners(p); err != nil {
543 return ports.RunnerPage{}, err
544 }
545 n, err := pageSize(size)
546 if err != nil {
547 return ports.RunnerPage{}, err
548 }
549 after := ""
550 if token != "" {
551 raw, ok := decodeToken(token, "rn1:")
552 if !ok || raw == "" {
553 return ports.RunnerPage{}, fmt.Errorf("%w: runners", domain.ErrInvalidPageToken)
554 }
555 after = raw
556 }
557 all, err := c.cfg.Store.ListRunners(ctx)
558 if err != nil {
559 return ports.RunnerPage{}, err
560 }
561 busy := map[string]bool{}
562 if active, err := c.cfg.Store.ListActiveJobs(ctx); err == nil {
563 for _, j := range active {
564 if j.Phase == domain.CIPhaseRunning {
565 busy[j.RunnerID] = true
566 }
567 }
568 }
569 var page ports.RunnerPage
570 var rows []domain.CIRunner
571 rows, page.NextPageToken = pageAfter(all, n,
572 func(r domain.CIRunner) bool { return after == "" || strings.ToLower(r.Name) > after },
573 func(r domain.CIRunner) string {
574 return base64.RawURLEncoding.EncodeToString([]byte("rn1:" + strings.ToLower(r.Name)))
575 })
576 c.mu.Lock()
577 defer c.mu.Unlock()
578 for _, r := range rows {
579 page.Runners = append(page.Runners, ports.RunnerView{CIRunner: r, LastSeen: c.seen[r.ID], Busy: busy[r.ID]})
580 }
581 return page, nil
582}
583
584// RemoveRunner removes a runner: its token stops working at once and a poll of
585// it in progress ends with unauthenticated. A job it runs is not affected.
586func (c *CI) RemoveRunner(ctx context.Context, p domain.Principal, id string) error {
587 if err := canManageRunners(p); err != nil {
588 return err
589 }
590 r, err := c.cfg.Store.RemoveRunner(ctx, id)
591 if err != nil {
592 return err
593 }
594 c.forgetRunners([]domain.CIRunner{r})
595 ctxLog(ctx).Info().Str("runner_id", r.ID).Str("runner", r.Name).Str("principal", p.Label()).Str("token_id", p.TokenID()).Msg("runner removed")
596 return nil
597}