1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package usecase
4
5import (
6 "context"
7 "crypto/rand"
8 "encoding/base64"
9 "encoding/hex"
10 "fmt"
11 "slices"
12 "strings"
13 "sync"
14 "time"
15
16 "github.com/bnema/zerowrap"
17
18 "bnema.dev/ermite/internal/domain"
19 "bnema.dev/ermite/internal/ports"
20)
21
22// Defaults of the CI engine.
23const (
24 // DefaultCIPollTimeout is how long FetchJob holds a call: the contract's 50 s.
25 DefaultCIPollTimeout = 50 * time.Second
26 // DefaultCITick is how often the watchdog looks at the jobs in flight.
27 DefaultCITick = time.Second
28 // DefaultCIKeepalive is how often WatchJob sends a keepalive.
29 DefaultCIKeepalive = 30 * time.Second
30 // DefaultCITriggerQueue is the size of the queue of pushes to plan.
31 DefaultCITriggerQueue = 256
32 // DefaultCIRemoteConcurrency is the git ls-remote processes ResolveActions
33 // runs at once, whatever the number of jobs that resolve.
34 DefaultCIRemoteConcurrency = 4
35 // DefaultCIResolveTimeout bounds one ResolveActions call.
36 DefaultCIResolveTimeout = 60 * time.Second
37 // DefaultCIConcurrencyPruneBatch is the stale marks of a concurrency group a
38 // transaction removes at most.
39 DefaultCIConcurrencyPruneBatch = 256
40 // MaxCIChecks is the most checks ListChecks returns.
41 MaxCIChecks = 1000
42 // MaxCILogStreams and MaxCILogStreamsPerClient bound the concurrent log
43 // streams (GetJobLog).
44 MaxCILogStreams = 64
45 MaxCILogStreamsPerClient = 8
46 // maxCIWorkflowBytes is the largest workflow file (the parser's limit).
47 maxCIWorkflowBytes = 1 << 20
48 // maxCIWorkflowFiles bounds the workflow files read per push.
49 maxCIWorkflowFiles = 256
50)
51
52// CIConfig holds what the CI use cases need. Store, Logs, Summaries, Meta, Repos, Reader,
53// Git, Diff, Bus, Clock, Owner and PublicURL are required (the config supplies
54// the defaults of Owner and Retention).
55type CIConfig struct {
56 Store ports.CIStore
57 Logs ports.CILogStore
58 // Summaries keeps the job summaries, deleted with the logs of their run.
59 Summaries ports.CISummaryStore
60 Meta ports.RepoMetaStore
61 Repos ports.GitRepos
62 Reader ports.RepoReader
63 Git ports.MergeGit
64 Diff ports.CIGit
65 // MRs, Pull and Committer give the workflows their pull_request event and
66 // the refs/pull/<n>/{head,merge} refs of the merge requests (CI-3, D5);
67 // without MRs or Pull the merge requests start no run and have no such
68 // refs. Committer is the identity of the test merge commits (default
69 // "ermite"). Fresh, when set, is told when the refs of a repository changed
70 // so that the reader sees them at once.
71 MRs ports.MergeRequestStore
72 Pull ports.PullGit
73 Committer domain.Identity
74 Fresh ports.RepoFreshener
75 // PullStage, when set, is called by the workers of the merge requests at
76 // the stages "event" (an activity is taken), "computed" (its merge is
77 // computed, before it is published) and "refresh" (a refresh is taken).
78 // Tests use it to interleave other changes.
79 PullStage func(stage string)
80 // PullRefreshes bounds the merge requests waiting for a refresh of their refs
81 // (default DefaultCIPullRefreshes); past it repositories are rescanned. It
82 // does not bound the activities (synchronize runs), which have their own queue
83 // with the overload policy of the push triggers.
84 PullRefreshes int
85 // PullAttempts bounds the merge requests whose last conflict is remembered
86 // (default DefaultCIPullAttempts), so that a rescan does not try a conflicting
87 // merge again until a branch moves.
88 PullAttempts int
89 // PullSnapshot bounds the merge requests one push takes the snapshot of, in the
90 // push handler (default DefaultCIPullSnapshot): past it the others are skipped
91 // with an error in the log and the repository is rescanned for its refs.
92 PullSnapshot int
93 Bus ports.EventBus
94 Clock ports.Clock
95 // Auth re-validates the token of a log stream; nil skips it.
96 Auth ports.Revalidator
97 // Remote lists the refs of github.com repositories for ResolveActions; nil
98 // answers every github.com action with unavailable.
99 Remote ports.GitRemote
100 // Workflows downloads reusable workflow files from github.com at a resolved
101 // commit; nil fails such a call as unavailable.
102 Workflows ports.WorkflowFetcher
103 // Archive serves the tarballs of the REST subset.
104 Archive ports.GitArchive
105 // Tokens finds the stored token of the pusher of a run, to ask what it may
106 // read when an action of another repository is resolved for the run. Nil
107 // reads nothing for a token (fail closed).
108 Tokens ports.TokenStore
109 // Settings resolves the variables and secrets of a job (CI-3). Nil gives
110 // jobs none, and an empty "vars" context. Leave it unset rather than
111 // assigning a typed nil pointer.
112 Settings ports.CIJobSettings
113 // RemoteConcurrency bounds the git ls-remote processes at once (default
114 // DefaultCIRemoteConcurrency); ResolveTimeout bounds one ResolveActions call
115 // (default DefaultCIResolveTimeout).
116 RemoteConcurrency int
117 ResolveTimeout time.Duration
118 // Owner is the owner login of the github context.
119 Owner string
120 // PublicURL is the base URL of the instance (github.server_url).
121 PublicURL string
122 // Retention is how long runs, jobs, checks and logs are kept.
123 Retention time.Duration
124
125 // The durations below default to the contract's; tests shorten them.
126 PollTimeout time.Duration
127 Tick time.Duration
128 Keepalive time.Duration
129 AckTimeout time.Duration
130 HeartbeatDeadline time.Duration
131 CancelGrace time.Duration
132 TimeoutSlack time.Duration
133 QueueTimeout time.Duration
134 TriggerQueue int
135 // ConcurrencyRetries bounds the retries of a delivery of the concurrency
136 // groups that failed (a cancel, the planning of a promoted entry): 0 is the
137 // default (8), a negative number none, which the tests use to leave the
138 // recovery to the reconciliation.
139 ConcurrencyRetries int
140 // ConcurrencyPruneBatch bounds the stale marks of a concurrency group one
141 // transaction removes (default 256, far below the size limit of a kv batch);
142 // the others go with the next decisions and the reconciliation.
143 ConcurrencyPruneBatch int
144}
145
146func (c CIConfig) withDefaults() CIConfig {
147 def := func(d *time.Duration, v time.Duration) {
148 if *d <= 0 {
149 *d = v
150 }
151 }
152 def(&c.PollTimeout, DefaultCIPollTimeout)
153 def(&c.Tick, DefaultCITick)
154 def(&c.Keepalive, DefaultCIKeepalive)
155 def(&c.AckTimeout, domain.CIAckTimeout)
156 def(&c.HeartbeatDeadline, domain.CIHeartbeatDeadline)
157 def(&c.CancelGrace, domain.CICancelGrace)
158 def(&c.TimeoutSlack, domain.CITimeoutSlack)
159 def(&c.QueueTimeout, domain.CIQueueTimeout)
160 def(&c.Retention, domain.CIRetentionDays*24*time.Hour)
161 if c.RemoteConcurrency <= 0 {
162 c.RemoteConcurrency = DefaultCIRemoteConcurrency
163 }
164 def(&c.ResolveTimeout, DefaultCIResolveTimeout)
165 if c.PullSnapshot <= 0 {
166 c.PullSnapshot = DefaultCIPullSnapshot
167 }
168 if c.PullRefreshes <= 0 {
169 c.PullRefreshes = DefaultCIPullRefreshes
170 }
171 if c.TriggerQueue <= 0 {
172 c.TriggerQueue = DefaultCITriggerQueue
173 }
174 if c.ConcurrencyPruneBatch <= 0 {
175 c.ConcurrencyPruneBatch = DefaultCIConcurrencyPruneBatch
176 }
177 return c
178}
179
180// CI is the CI use cases: the workflow trigger, the planner, the assignment of
181// jobs to runners, the runner protocol and the owner side (ADR 0025).
182//
183// Every change of a run goes through its lock (withRun): the records of a run
184// are read, changed and written again by one goroutine at a time, so that two
185// calls (a runner's update and an owner's cancel) never lose one another's
186// write. Assignment takes the assignment lock before the lock of the run.
187type CI struct {
188 cfg CIConfig
189 boot time.Time
190 queue chan ciTrigger
191 pull *pullQueue // the work of the merge requests (ci_pull.go)
192
193 pullFileLists pullFilesCache // the files of the merge requests (ci_pullfiles.go)
194
195 stop context.CancelFunc
196 ctx context.Context
197 wg sync.WaitGroup
198
199 remoteSlots chan struct{} // slots of the git ls-remote processes
200 locks sync.Map // run id -> *sync.Mutex
201 assign sync.Mutex
202 jobs notifier // woken when a job changes: key is the job reference
203 pending notifier // woken when a job may be ready or a runner was removed
204 flows *wfCache
205
206 mu sync.Mutex
207 heartbeats map[string]time.Time // job reference -> last heartbeat
208 seen map[string]time.Time // runner id -> last seen
209 polls map[string]pollRecord
210 // archiveTokens are the secrets of the tokens given for the archives of
211 // ermite actions, per job reference then "<repository>@<sha>", so that a job
212 // that resolves the same commit twice gets one token (memory only: a
213 // restart issues new ones).
214 archiveTokens map[string]map[string]string
215 streams int
216 perClient map[string]int
217 closed bool
218}
219
220var (
221 _ ports.RunnerService = (*CI)(nil)
222 _ ports.RunService = (*CI)(nil)
223 _ ports.WorkflowTrigger = (*CI)(nil)
224 _ ports.CIChecksReader = (*CI)(nil)
225)
226
227// pollRecord remembers the assignment a poll made, for the retry of a lost
228// response (FetchJobRequest.poll_id). The token is kept in memory only.
229type pollRecord struct {
230 pollID string
231 ref string
232 token string
233 ghToken string
234 hash domain.TokenHash
235}
236
237// NewCI builds the CI use cases. ctx carries the logger; its cancellation is
238// not followed: call Start to run the workers and Close to stop them.
239func NewCI(ctx context.Context, cfg CIConfig) *CI {
240 if cfg.Owner == "" {
241 panic("usecase: CIConfig.Owner is required")
242 }
243 cfg = cfg.withDefaults()
244 run, stop := context.WithCancel(context.WithoutCancel(ctx))
245 return &CI{
246 cfg: cfg, boot: cfg.Clock.Now(), queue: make(chan ciTrigger, cfg.TriggerQueue), pull: newPullQueue(cfg.TriggerQueue, cfg.PullRefreshes, cfg.PullAttempts),
247 ctx: run, stop: stop, flows: newWfCache(64), remoteSlots: make(chan struct{}, cfg.RemoteConcurrency),
248 heartbeats: map[string]time.Time{}, seen: map[string]time.Time{}, polls: map[string]pollRecord{},
249 archiveTokens: map[string]map[string]string{},
250 perClient: map[string]int{},
251 }
252}
253
254// Start runs the trigger worker and the watchdog, and plans again the runs
255// that were open when the server stopped.
256func (c *CI) Start() {
257 c.wg.Add(4)
258 go c.pullLoop()
259 go c.refreshLoop()
260 go c.triggerLoop()
261 go c.watchdog()
262 c.wg.Add(1)
263 go func() {
264 defer c.wg.Done()
265 c.resume()
266 }()
267}
268
269// Close stops the workers and waits for them.
270func (c *CI) Close(_ context.Context) {
271 c.mu.Lock()
272 c.closed = true
273 c.mu.Unlock()
274 c.stop()
275 c.wg.Wait()
276}
277
278func (c *CI) now() time.Time { return c.cfg.Clock.Now().UTC() }
279
280// --- identifiers and secrets ---
281
282func newToken(prefix string) string {
283 var raw [32]byte
284 _, _ = rand.Read(raw[:])
285 return prefix + base64.RawURLEncoding.EncodeToString(raw[:])
286}
287
288func newRunnerID() string {
289 var raw [8]byte
290 _, _ = rand.Read(raw[:])
291 return hex.EncodeToString(raw[:])
292}
293
294func jobRef(j domain.CIJob) string { return fmt.Sprintf("%d/%d/%s", j.RunID, j.RunAttempt, j.ID) }
295
296// --- locks and notifications ---
297
298// notifier wakes the goroutines waiting on a key. A channel exists only while
299// somebody waits on it.
300type notifier struct {
301 mu sync.Mutex
302 ch map[string]chan struct{}
303}
304
305// wait returns a channel closed by the next notify of key.
306func (n *notifier) wait(key string) <-chan struct{} {
307 n.mu.Lock()
308 defer n.mu.Unlock()
309 if n.ch == nil {
310 n.ch = map[string]chan struct{}{}
311 }
312 c, ok := n.ch[key]
313 if !ok {
314 c = make(chan struct{})
315 n.ch[key] = c
316 }
317 return c
318}
319
320func (n *notifier) notify(key string) {
321 n.mu.Lock()
322 defer n.mu.Unlock()
323 if c, ok := n.ch[key]; ok {
324 close(c)
325 delete(n.ch, key)
326 }
327}
328
329const queueKey = "q"
330
331func (c *CI) wakeQueue() { c.pending.notify(queueKey) }
332
333func (c *CI) runLock(runID int64) *sync.Mutex {
334 m, _ := c.locks.LoadOrStore(runID, &sync.Mutex{})
335 return m.(*sync.Mutex)
336}
337
338// --- run state ---
339
340// runState is a run being changed under its lock: the run, the jobs of one of
341// its attempts, and what must be written. flush writes everything in one batch
342// and publishes the events.
343type runState struct {
344 c *CI
345 run domain.CIRun
346 attempt int32
347 jobs []domain.CIJob
348 src *domain.CIRunSource
349
350 runDirty bool
351 dirty map[string]bool
352 // chk marks jobs whose check must be written even when the job itself did
353 // not change (a rerun moves every check to the new attempt).
354 checks map[string]bool
355 // tokens are the job repository tokens to create, revoke the jobs whose
356 // tokens are deleted, with the next flush.
357 tokens []domain.CIRepoToken
358 revoke []ports.CIJobRef
359 // resolutions are the action resolutions to keep with the next flush.
360 resolutions []domain.CIActionResolution
361 // annotations replace the annotations of steps, clear drops all those of
362 // executions of jobs, with the next flush.
363 annotations []ports.CIStepAnnotations
364 clear []ports.CILogRef
365}
366
367// withRun locks the run, loads it with the jobs of attempt (0 for the latest)
368// and calls fn. The run is found by id alone through its source.
369func (c *CI) withRun(ctx context.Context, runID int64, attempt int32, fn func(rs *runState) error) error {
370 lock := c.runLock(runID)
371 lock.Lock()
372 defer lock.Unlock()
373 rs, err := c.loadRun(ctx, runID, attempt)
374 if err != nil {
375 return err
376 }
377 return fn(rs)
378}
379
380func (c *CI) loadRun(ctx context.Context, runID int64, attempt int32) (*runState, error) {
381 src, err := c.cfg.Store.GetRunSource(ctx, runID)
382 if err != nil {
383 return nil, err
384 }
385 run, err := c.cfg.Store.GetRun(ctx, src.Repo, runID)
386 if err != nil {
387 return nil, err
388 }
389 if attempt <= 0 {
390 attempt = run.Attempt
391 }
392 jobs, err := c.cfg.Store.ListJobs(ctx, runID, attempt)
393 if err != nil {
394 return nil, err
395 }
396 return &runState{c: c, run: run, attempt: attempt, jobs: jobs, src: &src, dirty: map[string]bool{}, checks: map[string]bool{}}, nil
397}
398
399// latest reports whether the state holds the latest attempt of the run.
400func (rs *runState) latest() bool { return rs.attempt == rs.run.Attempt }
401
402func (rs *runState) find(id string) *domain.CIJob {
403 for i := range rs.jobs {
404 if rs.jobs[i].ID == id {
405 return &rs.jobs[i]
406 }
407 }
408 return nil
409}
410
411func (rs *runState) touch(j *domain.CIJob) { rs.dirty[j.ID] = true }
412
413// checkOf builds the check of a job.
414func (rs *runState) checkOf(j domain.CIJob) domain.CICheck {
415 return domain.CICheck{
416 ID: j.ID, Repo: rs.run.Repo, Sha: rs.run.CheckSha(), Name: j.Name, Status: j.Status(), Conclusion: j.Conclusion,
417 RunID: rs.run.ID, JobID: j.ID, Attempt: rs.run.Attempt, WorkflowName: rs.run.WorkflowName,
418 CreatedAt: j.CreatedAt, StartedAt: j.StartedAt, CompletedAt: j.CompletedAt,
419 }
420}
421
422// flush recomputes the status of the run, writes the changed jobs, their
423// checks and the run in one batch, and publishes RunUpdated and JobUpdated.
424func (rs *runState) flush(ctx context.Context) error {
425 c := rs.c
426 now := c.now()
427 var w ports.CIWrite
428 var changedJobs []int
429 for i := range rs.jobs {
430 if rs.dirty[rs.jobs[i].ID] {
431 w.Jobs = append(w.Jobs, rs.jobs[i])
432 changedJobs = append(changedJobs, i)
433 }
434 }
435 if rs.latest() {
436 for i := range rs.jobs {
437 if rs.dirty[rs.jobs[i].ID] || rs.checks[rs.jobs[i].ID] {
438 w.Checks = append(w.Checks, rs.checkOf(rs.jobs[i]))
439 }
440 }
441 if next := rs.recompute(now); next != rs.run {
442 rs.run, rs.runDirty = next, true
443 }
444 }
445 if rs.runDirty {
446 w.Runs = []domain.CIRun{rs.run}
447 }
448 w.RepoTokens, w.RevokeJobs, w.Resolutions = rs.tokens, rs.revoke, rs.resolutions
449 w.Annotations, w.ClearAnnotations = rs.annotations, rs.clear
450 if len(w.Jobs) == 0 && len(w.Runs) == 0 && len(w.Checks) == 0 && len(w.RepoTokens) == 0 && len(w.RevokeJobs) == 0 &&
451 len(w.Resolutions) == 0 && len(w.Annotations) == 0 && len(w.ClearAnnotations) == 0 {
452 return nil
453 }
454 v, err := c.cfg.Store.Apply(ctx, w)
455 if err != nil {
456 return fmt.Errorf("write run %d: %w", rs.run.ID, err)
457 }
458 for _, i := range changedJobs {
459 rs.jobs[i].Version = v
460 }
461 if rs.runDirty {
462 rs.run.Version = v
463 }
464 changedRun := rs.runDirty
465 rs.releaseEnded(ctx, changedJobs, changedRun && rs.run.Status == domain.CICompleted)
466 rs.dirty, rs.checks, rs.runDirty = map[string]bool{}, map[string]bool{}, false
467 rs.tokens, rs.revoke, rs.resolutions = nil, nil, nil
468 rs.annotations, rs.clear = nil, nil
469
470 repo, err := c.cfg.Meta.Get(ctx, rs.run.Repo)
471 if err != nil {
472 for _, i := range changedJobs {
473 c.jobs.notify(jobRef(rs.jobs[i]))
474 }
475 return nil // the repository is gone: nobody to tell
476 }
477 for _, i := range changedJobs {
478 j := rs.jobs[i]
479 lines, _ := c.cfg.Logs.Count(ctx, ports.CILogRef{RunID: j.RunID, JobID: j.ID, Attempt: j.Attempt})
480 c.cfg.Bus.Publish(ctx, domain.JobUpdated{Job: j, Lines: lines, At: now, Visibility: repo.Visibility, CreatedAt: repo.CreatedAt})
481 c.jobs.notify(jobRef(j))
482 if j.Phase == domain.CIPhaseReady {
483 c.wakeQueue()
484 }
485 }
486 if changedRun {
487 c.cfg.Bus.Publish(ctx, domain.RunUpdated{Run: rs.run, At: now, Visibility: repo.Visibility, CreatedAt: repo.CreatedAt})
488 }
489 return nil
490}
491
492// recompute returns the run with the status, the conclusion and the
493// timestamps its jobs imply.
494func (rs *runState) recompute(now time.Time) domain.CIRun {
495 run := rs.run
496 if run.Status == domain.CICompleted && len(rs.jobs) == 0 {
497 return run // an invalid workflow file: failed from the start
498 }
499 allEnded := len(rs.jobs) > 0
500 started := time.Time{}
501 var failed, cancelled bool
502 for _, j := range rs.jobs {
503 if !j.Ended() {
504 allEnded = false
505 }
506 // A job kept from an earlier attempt did not start in this one.
507 if j.Attempt == rs.attempt && !j.StartedAt.IsZero() && (started.IsZero() || j.StartedAt.Before(started)) {
508 started = j.StartedAt
509 }
510 switch {
511 case j.Conclusion.IsFailure() && j.ContinueOnError:
512 // continue-on-error: the job failed, the run does not.
513 case j.Conclusion.IsFailure():
514 failed = true
515 case j.Conclusion == domain.CICancelled:
516 cancelled = true
517 }
518 }
519 if run.StartedAt.IsZero() && !started.IsZero() {
520 run.StartedAt = started
521 }
522 switch {
523 case allEnded:
524 run.Status, run.CompletedAt = domain.CICompleted, now
525 switch {
526 case run.CancelRequested:
527 run.Conclusion = domain.CICancelled
528 case failed:
529 run.Conclusion = domain.CIFailure
530 case cancelled:
531 run.Conclusion = domain.CICancelled
532 default:
533 run.Conclusion = domain.CISuccess
534 }
535 if rs.run.Status == domain.CICompleted {
536 run.CompletedAt = rs.run.CompletedAt
537 }
538 case !run.StartedAt.IsZero():
539 run.Status, run.Conclusion, run.CompletedAt = domain.CIInProgress, domain.CINoConclusion, time.Time{}
540 default:
541 run.Status, run.Conclusion, run.CompletedAt = domain.CIQueued, domain.CINoConclusion, time.Time{}
542 }
543 return run
544}
545
546// --- access ---
547
548// readableRepo looks the repository up and applies the read rule.
549func (c *CI) readableRepo(ctx context.Context, p domain.Principal, name domain.RepoName) (domain.Repo, error) {
550 repo, err := lookupRepo(ctx, c.cfg.Meta, name)
551 if err != nil {
552 return domain.Repo{}, err
553 }
554 if !canRead(p, repo) {
555 return domain.Repo{}, domain.ErrNotFound
556 }
557 return repo, nil
558}
559
560// writableRepo is the access rule of cancelling and re-running: a principal
561// whose token holds ci:write on the repository sees the repository for this
562// action even without repo:read, as for merge requests.
563func (c *CI) writableRepo(ctx context.Context, p domain.Principal, name domain.RepoName) (domain.Repo, error) {
564 repo, err := lookupRepo(ctx, c.cfg.Meta, name)
565 if err != nil {
566 return domain.Repo{}, err
567 }
568 if !canRead(p, repo) && !p.Allows(domain.ScopeCIWrite, name) {
569 return domain.Repo{}, domain.ErrNotFound
570 }
571 if err := scopeAccess(p, repo, domain.ScopeCIWrite); err != nil {
572 return domain.Repo{}, err
573 }
574 return repo, nil
575}
576
577// runWriteAccess is the branch rule of the write actions on a run, cancel and
578// re-run, which come after the ci:write check. A run is tied to the ref that
579// made it, and re-running it executes the workflow of that ref again while
580// cancelling it stops the checks of that ref. So, like a push, a merge or a
581// branch delete, these actions fall under the branch prefixes of a limited
582// token (ADR 0027): it may act only on the runs of branches under its prefixes,
583// and never on a run of a tag or of any other ref. Without prefixes, every run.
584//
585// Reading runs, jobs, logs and checks is not a write: it needs only the access
586// to read the repository, as for the code, whatever the prefixes. The caller
587// has seen the run by then, so the refusal is forbidden, not not found.
588func runWriteAccess(p domain.Principal, run domain.CIRun) error {
589 if !p.AllowsRef(run.Ref) {
590 return fmt.Errorf("%w: this token may not write %s", domain.ErrForbidden, run.Ref)
591 }
592 return nil
593}
594
595// canManageRunners is the access rule of runners: runner:register on a token
596// that is not restricted to repositories (a runner serves every repository),
597// or root. A token limited to branch prefixes never manages runners, whatever
598// scopes it holds: a runner runs the workflows of every branch of every
599// repository, with their code and secrets, so registering or removing one
600// reaches far beyond the prefixes (a replaced runner could also be a way to
601// take over the jobs of main). Like repository administration and owner
602// rights, it is out of reach of a limited token.
603func canManageRunners(p domain.Principal) error {
604 switch {
605 case p.IsAnonymous():
606 return domain.ErrUnauthenticated
607 case !p.HasScope(domain.ScopeRunnerRegister):
608 return fmt.Errorf("%w: this needs the %s scope", domain.ErrForbidden, domain.ScopeRunnerRegister)
609 case len(p.Repos()) > 0:
610 return fmt.Errorf("%w: a runner serves every repository: the token must not be restricted to some", domain.ErrForbidden)
611 case p.IsBranchLimited():
612 return fmt.Errorf("%w: a token limited to branch prefixes may not manage runners", domain.ErrForbidden)
613 }
614 return nil
615}
616
617// --- small helpers ---
618
619func truncateRunes(s string, n int) string {
620 if len(s) <= n {
621 return s
622 }
623 r := []rune(s)
624 if len(r) <= n {
625 return s
626 }
627 return string(r[:n])
628}
629
630// cutText cuts s to n bytes at a character boundary.
631func cutText(s string, n int) string {
632 if len(s) <= n {
633 return s
634 }
635 for n > 0 && n < len(s) && s[n]&0xC0 == 0x80 {
636 n--
637 }
638 return s[:n]
639}
640
641func sortedKeys[V any](m map[string]V) []string {
642 keys := make([]string, 0, len(m))
643 for k := range m {
644 keys = append(keys, k)
645 }
646 slices.Sort(keys)
647 return keys
648}
649
650// ctxLog returns the logger of the context, addressable so that the level
651// methods can be chained on it.
652func ctxLog(ctx context.Context) *zerowrap.Logger {
653 l := zerowrap.FromCtx(ctx)
654 return &l
655}
656
657// logRun returns the logger of the context with the fields of the run.
658func logRun(ctx context.Context, run domain.CIRun) zerowrap.Logger {
659 return zerowrap.FromCtx(ctx).WithFields(map[string]any{"repo": run.Repo.String(), "run_id": run.ID})
660}
661
662func isHex40(s string) bool {
663 if len(s) != 40 {
664 return false
665 }
666 return strings.Trim(s, "0123456789abcdef") == ""
667}