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 w.Runs = []domain.CIRun{run}
403 rs2 := &runState{c: c, run: run, attempt: next, jobs: fresh, src: rs.src, dirty: map[string]bool{}, checks: map[string]bool{}}
404 for _, j := range fresh {
405 w.Checks = append(w.Checks, rs2.checkOf(j))
406 }
407 v, err := c.cfg.Store.Apply(ctx, w)
408 if err != nil {
409 return err
410 }
411 run.Version = v
412 c.flows.drop(runID)
413 c.publishRun(ctx, repo, run)
414 for _, j := range w.Jobs {
415 j.Version = v
416 c.cfg.Bus.Publish(ctx, domain.JobUpdated{Job: j, At: now, Visibility: repo.Visibility, CreatedAt: repo.CreatedAt})
417 }
418 return nil
419 })
420 if err != nil {
421 return domain.CIRun{}, err
422 }
423 // Plan the new attempt: the kept jobs count as the results of their needs.
424 if err := c.advanceRun(ctx, runID); err != nil {
425 return domain.CIRun{}, err
426 }
427 run, err = c.runOf(ctx, name, runID)
428 if err != nil {
429 return domain.CIRun{}, err
430 }
431 ctxLog(ctx).Info().Str("repo", name.String()).Int64("run_id", runID).Int32("attempt", run.Attempt).Bool("failed_only", failedOnly).
432 Str("principal", p.Label()).Str("token_id", p.TokenID()).Msg("ci run started again")
433 return run, nil
434}
435
436// rerunSet returns the keys of the jobs to run again: all of them, or the ones
437// that did not succeed and those that need them, directly or not.
438func rerunSet(rs *runState, failedOnly bool) (map[string]bool, error) {
439 redo := map[string]bool{}
440 if !failedOnly {
441 for _, j := range rs.jobs {
442 redo[j.Key] = true
443 }
444 return redo, nil
445 }
446 for _, j := range rs.jobs {
447 switch j.Conclusion {
448 case domain.CIFailure, domain.CICancelled, domain.CITimedOut, domain.CIActionRequired:
449 redo[j.Key] = true
450 }
451 }
452 if len(redo) == 0 {
453 return nil, fmt.Errorf("%w: no job of run %d failed", domain.ErrPreconditionFailed, rs.run.ID)
454 }
455 for changed := true; changed; {
456 changed = false
457 for _, j := range rs.jobs {
458 if redo[j.Key] {
459 continue
460 }
461 if slices.ContainsFunc(j.Needs, func(n string) bool { return redo[n] }) {
462 redo[j.Key], changed = true, true
463 }
464 }
465 }
466 return redo, nil
467}
468
469func (w *wfCache) drop(runID int64) {
470 w.mu.Lock()
471 delete(w.m, runID)
472 w.mu.Unlock()
473}
474
475// --- checks ---
476
477// CommitChecks returns the checks of a commit, without access control. A
478// commit with more than MaxCIChecks checks is an error (domain.ErrResourceExhausted):
479// the merge rule cannot answer from a part of them.
480func (c *CI) CommitChecks(ctx context.Context, repo domain.RepoName, sha string) ([]domain.CICheck, error) {
481 checks, more, err := c.cfg.Store.ListChecks(ctx, repo, sha, MaxCIChecks)
482 if err != nil {
483 return nil, err
484 }
485 if more {
486 return nil, fmt.Errorf("%w: the commit has more than %d checks", domain.ErrResourceExhausted, MaxCIChecks)
487 }
488 return checks, nil
489}
490
491// ListChecks returns the checks of a commit: the one of each job of the latest
492// attempt of each run on it. A commit without a run has none.
493func (c *CI) ListChecks(ctx context.Context, p domain.Principal, name domain.RepoName, sha string) (ports.ChecksResult, error) {
494 if _, err := c.readableRepo(ctx, p, name); err != nil {
495 return ports.ChecksResult{}, err
496 }
497 if !isHex40(sha) {
498 return ports.ChecksResult{}, fmt.Errorf("%w: a commit is 40 lowercase hexadecimal characters", domain.ErrInvalidObjectID)
499 }
500 checks, more, err := c.cfg.Store.ListChecks(ctx, name, sha, MaxCIChecks)
501 if err != nil {
502 return ports.ChecksResult{}, err
503 }
504 res := ports.ChecksResult{Truncated: more}
505 var runners []domain.CIRunner
506 for _, ck := range checks {
507 v := ports.CheckView{CICheck: ck}
508 if ck.Status == domain.CIQueued {
509 if runners == nil {
510 runners, _ = c.cfg.Store.ListRunners(ctx)
511 }
512 v.WaitingForRunner = c.noRunnerFor(ctx, ck, runners)
513 }
514 res.Checks = append(res.Checks, v)
515 }
516 return res, nil
517}
518
519// noRunnerFor reports whether no registered runner can take a queued check's
520// job. A job that waits for its needs is not waiting for a runner.
521func (c *CI) noRunnerFor(ctx context.Context, ck domain.CICheck, runners []domain.CIRunner) bool {
522 if ck.JobID == "" {
523 return false
524 }
525 j, err := c.cfg.Store.GetJob(ctx, ck.RunID, ck.Attempt, ck.JobID)
526 if err != nil || j.Phase != domain.CIPhaseReady {
527 return false
528 }
529 return !slices.ContainsFunc(runners, func(r domain.CIRunner) bool { return domain.LabelsMatch(j.Labels, r.Labels) })
530}
531
532// --- runners ---
533
534// ListRunners returns a page of the runners ordered by name. The principal
535// needs runner:register on a token that is not restricted to repositories.
536func (c *CI) ListRunners(ctx context.Context, p domain.Principal, size int, token string) (ports.RunnerPage, error) {
537 if err := canManageRunners(p); err != nil {
538 return ports.RunnerPage{}, err
539 }
540 n, err := pageSize(size)
541 if err != nil {
542 return ports.RunnerPage{}, err
543 }
544 after := ""
545 if token != "" {
546 raw, ok := decodeToken(token, "rn1:")
547 if !ok || raw == "" {
548 return ports.RunnerPage{}, fmt.Errorf("%w: runners", domain.ErrInvalidPageToken)
549 }
550 after = raw
551 }
552 all, err := c.cfg.Store.ListRunners(ctx)
553 if err != nil {
554 return ports.RunnerPage{}, err
555 }
556 busy := map[string]bool{}
557 if active, err := c.cfg.Store.ListActiveJobs(ctx); err == nil {
558 for _, j := range active {
559 if j.Phase == domain.CIPhaseRunning {
560 busy[j.RunnerID] = true
561 }
562 }
563 }
564 var page ports.RunnerPage
565 var rows []domain.CIRunner
566 rows, page.NextPageToken = pageAfter(all, n,
567 func(r domain.CIRunner) bool { return after == "" || strings.ToLower(r.Name) > after },
568 func(r domain.CIRunner) string {
569 return base64.RawURLEncoding.EncodeToString([]byte("rn1:" + strings.ToLower(r.Name)))
570 })
571 c.mu.Lock()
572 defer c.mu.Unlock()
573 for _, r := range rows {
574 page.Runners = append(page.Runners, ports.RunnerView{CIRunner: r, LastSeen: c.seen[r.ID], Busy: busy[r.ID]})
575 }
576 return page, nil
577}
578
579// RemoveRunner removes a runner: its token stops working at once and a poll of
580// it in progress ends with unauthenticated. A job it runs is not affected.
581func (c *CI) RemoveRunner(ctx context.Context, p domain.Principal, id string) error {
582 if err := canManageRunners(p); err != nil {
583 return err
584 }
585 r, err := c.cfg.Store.RemoveRunner(ctx, id)
586 if err != nil {
587 return err
588 }
589 c.forgetRunners([]domain.CIRunner{r})
590 ctxLog(ctx).Info().Str("runner_id", r.ID).Str("runner", r.Name).Str("principal", p.Label()).Str("token_id", p.TokenID()).Msg("runner removed")
591 return nil
592}