1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package kvstore
4
5import (
6 "context"
7 "errors"
8 "fmt"
9 "slices"
10 "strconv"
11 "strings"
12 "time"
13
14 kv "bnema.dev/ermite-kv"
15 "google.golang.org/protobuf/proto"
16
17 storagepb "bnema.dev/ermite/gen/ermite/storage"
18 "bnema.dev/ermite/internal/domain"
19 "bnema.dev/ermite/internal/ports"
20)
21
22// Key layout of CI. Repository names hold no '/', so a prefix ending in "/"
23// never matches another repository; ids are zero-padded so that keys sort
24// numerically. Job records are keyed by run, attempt of the run and job id.
25//
26// run/<repo>/<id:020d> CIRun
27// runsrc/<id:020d> CIRunSource
28// job/<run:020d>/<attempt:010d>/<id> CIJob
29// check/<repo>/<sha>/<check id> CICheck
30// runseq Sequence: last run id
31// runnum/<repo>/<workflow path> Sequence: last run number
32// runner/<id>, runnername/<name>, runnertoken/<hash>
33//
34// Indexes, derived from the job and run records by every write:
35//
36// ciopen/<id:020d> run not completed (value: repo)
37// ciready/<ready_at:020d>/<job key> job waiting for a runner
38// ciactive/<job key> job assigned or running
39// jobtoken/<hash> job token (value: job key)
40// concur/<repo>/<key(group)> concurrency group (ci_concurrency.go)
41//
42// The job repository tokens and the action resolutions are in ci_token.go, the
43// annotations in ci_annotation.go.
44const (
45 runPrefix = "run/"
46 runSrcPrefix = "runsrc/"
47 jobPrefix = "job/"
48 checkPrefix = "check/"
49 runSeqKey = "runseq"
50 runNumPrefix = "runnum/"
51 runnerPrefix = "runner/"
52 runnerName = "runnername/"
53 runnerTokPref = "runnertoken/"
54 openPrefix = "ciopen/"
55 readyPrefix = "ciready/"
56 activePrefix = "ciactive/"
57 jobTokPrefix = "jobtoken/"
58 idFormat = "%020d"
59)
60
61// CI is the ermite-kv backed ports.CIStore. It shares the database of the
62// Store it comes from.
63type CI struct {
64 db *kv.DB
65}
66
67var _ ports.CIStore = (*CI)(nil)
68
69// CI returns the CI store over the same database. It is valid until the Store
70// is closed.
71func (s *Store) CI() *CI { return &CI{db: s.db} }
72
73func runKey(repo domain.RepoName, id int64) []byte {
74 return []byte(runPrefix + repo.String() + "/" + fmt.Sprintf(idFormat, id))
75}
76func runsPrefix(repo domain.RepoName) string { return runPrefix + repo.String() + "/" }
77func runSrcKey(id int64) []byte { return []byte(runSrcPrefix + fmt.Sprintf(idFormat, id)) }
78func openKey(id int64) []byte { return []byte(openPrefix + fmt.Sprintf(idFormat, id)) }
79func jobsPrefix(runID int64) string { return jobPrefix + fmt.Sprintf(idFormat, runID) + "/" }
80func jobAttemptPrefix(runID int64, attempt int32) string {
81 return jobsPrefix(runID) + fmt.Sprintf("%010d", attempt) + "/"
82}
83func jobRef(runID int64, attempt int32, id string) string {
84 return jobAttemptPrefix(runID, attempt) + id
85}
86func jobKey(runID int64, attempt int32, id string) []byte { return []byte(jobRef(runID, attempt, id)) }
87func checksPrefix(repo domain.RepoName, sha string) string {
88 return checkPrefix + repo.String() + "/" + sha + "/"
89}
90func checkKey(repo domain.RepoName, sha, id string) []byte {
91 return []byte(checksPrefix(repo, sha) + id)
92}
93func runNumKey(repo domain.RepoName, workflow string) []byte {
94 return []byte(runNumPrefix + repo.String() + "/" + workflow)
95}
96func readyKey(at time.Time, ref string) []byte {
97 return []byte(readyPrefix + fmt.Sprintf(idFormat, at.UnixNano()) + "/" + ref)
98}
99func activeKey(ref string) []byte { return []byte(activePrefix + ref) }
100func jobTokenKey(h domain.TokenHash) []byte {
101 return []byte(jobTokPrefix + h.String())
102}
103
104// --- encoding ---
105
106func encodeStatus(s domain.CIStatus) storagepb.CIStatus {
107 switch s {
108 case domain.CIQueued:
109 return storagepb.CIStatus_CI_STATUS_QUEUED
110 case domain.CIInProgress:
111 return storagepb.CIStatus_CI_STATUS_IN_PROGRESS
112 case domain.CICompleted:
113 return storagepb.CIStatus_CI_STATUS_COMPLETED
114 case domain.CIPending:
115 return storagepb.CIStatus_CI_STATUS_PENDING
116 }
117 return storagepb.CIStatus_CI_STATUS_UNSPECIFIED
118}
119
120func decodeStatus(s storagepb.CIStatus) domain.CIStatus {
121 switch s {
122 case storagepb.CIStatus_CI_STATUS_IN_PROGRESS:
123 return domain.CIInProgress
124 case storagepb.CIStatus_CI_STATUS_COMPLETED:
125 return domain.CICompleted
126 case storagepb.CIStatus_CI_STATUS_PENDING:
127 return domain.CIPending
128 }
129 return domain.CIQueued
130}
131
132var conclusions = []struct {
133 d domain.CIConclusion
134 p storagepb.CIConclusion
135}{
136 {domain.CISuccess, storagepb.CIConclusion_CI_CONCLUSION_SUCCESS},
137 {domain.CIFailure, storagepb.CIConclusion_CI_CONCLUSION_FAILURE},
138 {domain.CICancelled, storagepb.CIConclusion_CI_CONCLUSION_CANCELLED},
139 {domain.CISkipped, storagepb.CIConclusion_CI_CONCLUSION_SKIPPED},
140 {domain.CITimedOut, storagepb.CIConclusion_CI_CONCLUSION_TIMED_OUT},
141 {domain.CINeutral, storagepb.CIConclusion_CI_CONCLUSION_NEUTRAL},
142 {domain.CIActionRequired, storagepb.CIConclusion_CI_CONCLUSION_ACTION_REQUIRED},
143}
144
145func encodeConclusion(c domain.CIConclusion) storagepb.CIConclusion {
146 for _, e := range conclusions {
147 if e.d == c {
148 return e.p
149 }
150 }
151 return storagepb.CIConclusion_CI_CONCLUSION_UNSPECIFIED
152}
153
154func decodeConclusion(c storagepb.CIConclusion) domain.CIConclusion {
155 for _, e := range conclusions {
156 if e.p == c {
157 return e.d
158 }
159 }
160 return domain.CINoConclusion
161}
162
163var phases = []struct {
164 d domain.CIJobPhase
165 p storagepb.CIJobPhase
166}{
167 {domain.CIPhaseWaiting, storagepb.CIJobPhase_CI_JOB_PHASE_WAITING},
168 {domain.CIPhaseReady, storagepb.CIJobPhase_CI_JOB_PHASE_READY},
169 {domain.CIPhaseAssigned, storagepb.CIJobPhase_CI_JOB_PHASE_ASSIGNED},
170 {domain.CIPhaseRunning, storagepb.CIJobPhase_CI_JOB_PHASE_RUNNING},
171 {domain.CIPhaseEnded, storagepb.CIJobPhase_CI_JOB_PHASE_ENDED},
172}
173
174func encodePhase(p domain.CIJobPhase) (storagepb.CIJobPhase, error) {
175 for _, e := range phases {
176 if e.d == p {
177 return e.p, nil
178 }
179 }
180 return 0, fmt.Errorf("invalid job phase %q", p)
181}
182
183func decodePhase(p storagepb.CIJobPhase) domain.CIJobPhase {
184 for _, e := range phases {
185 if e.p == p {
186 return e.d
187 }
188 }
189 return domain.CIPhaseWaiting
190}
191
192var isolations = []struct {
193 d domain.CIIsolation
194 p storagepb.CIIsolation
195}{
196 {domain.CIIsolationVM, storagepb.CIIsolation_CI_ISOLATION_VM},
197 {domain.CIIsolationContainerSysbox, storagepb.CIIsolation_CI_ISOLATION_CONTAINER_SYSBOX},
198 {domain.CIIsolationContainerPrivileged, storagepb.CIIsolation_CI_ISOLATION_CONTAINER_PRIVILEGED},
199}
200
201func encodeIsolation(i domain.CIIsolation) storagepb.CIIsolation {
202 for _, e := range isolations {
203 if e.d == i {
204 return e.p
205 }
206 }
207 return storagepb.CIIsolation_CI_ISOLATION_UNSPECIFIED
208}
209
210func decodeIsolation(i storagepb.CIIsolation) domain.CIIsolation {
211 for _, e := range isolations {
212 if e.p == i {
213 return e.d
214 }
215 }
216 return domain.CIIsolationNone
217}
218
219var stepStates = []struct {
220 d domain.CIStepState
221 p storagepb.CIStepState
222}{
223 {domain.CIStepPending, storagepb.CIStepState_CI_STEP_STATE_PENDING},
224 {domain.CIStepInProgress, storagepb.CIStepState_CI_STEP_STATE_IN_PROGRESS},
225 {domain.CIStepCompleted, storagepb.CIStepState_CI_STEP_STATE_COMPLETED},
226}
227
228func encodeStepState(s domain.CIStepState) storagepb.CIStepState {
229 for _, e := range stepStates {
230 if e.d == s {
231 return e.p
232 }
233 }
234 return storagepb.CIStepState_CI_STEP_STATE_UNSPECIFIED
235}
236
237func decodeStepState(s storagepb.CIStepState) domain.CIStepState {
238 for _, e := range stepStates {
239 if e.p == s {
240 return e.d
241 }
242 }
243 return domain.CIStepPending
244}
245
246var cancelReasons = []struct {
247 d domain.CICancelReason
248 p storagepb.CICancelReason
249}{
250 {domain.CICancelUser, storagepb.CICancelReason_CI_CANCEL_REASON_USER},
251 {domain.CICancelFailFast, storagepb.CICancelReason_CI_CANCEL_REASON_FAIL_FAST},
252 {domain.CICancelTimeout, storagepb.CICancelReason_CI_CANCEL_REASON_TIMEOUT},
253}
254
255func encodeCancelReason(r domain.CICancelReason) storagepb.CICancelReason {
256 for _, e := range cancelReasons {
257 if e.d == r {
258 return e.p
259 }
260 }
261 return storagepb.CICancelReason_CI_CANCEL_REASON_UNSPECIFIED
262}
263
264func decodeCancelReason(r storagepb.CICancelReason) domain.CICancelReason {
265 for _, e := range cancelReasons {
266 if e.p == r {
267 return e.d
268 }
269 }
270 return domain.CICancelNone
271}
272
273func encodeRun(r domain.CIRun) ([]byte, error) {
274 if r.Repo.IsZero() {
275 return nil, fmt.Errorf("%w: empty name", domain.ErrInvalidName)
276 }
277 return mustMarshal(&storagepb.CIRun{
278 Repo: r.Repo.String(), Id: r.ID, Number: r.Number, Attempt: r.Attempt,
279 WorkflowPath: r.WorkflowPath, WorkflowName: r.WorkflowName, Name: r.Name, Title: r.Title,
280 Event: r.Event, Ref: r.Ref, Sha: r.Sha, HeadSha: r.HeadSha, Actor: r.Actor,
281 Status: encodeStatus(r.Status), Conclusion: encodeConclusion(r.Conclusion), Error: r.Error,
282 CreatedAt: encodeTime(r.CreatedAt), StartedAt: encodeTime(r.StartedAt), CompletedAt: encodeTime(r.CompletedAt),
283 CancelRequested: r.CancelRequested, CancelRequestedAt: encodeTime(r.CancelRequestedAt), Trusted: r.Trusted,
284 Concurrency: encodeConcurrency(r.Concurrency),
285 }), nil
286}
287
288func decodeRun(item kv.Item) (domain.CIRun, error) {
289 var pb storagepb.CIRun
290 if err := proto.Unmarshal(item.Value, &pb); err != nil {
291 return domain.CIRun{}, fmt.Errorf("decode run: %w", err)
292 }
293 repo, err := domain.ParseRepoName(pb.GetRepo())
294 if err != nil {
295 return domain.CIRun{}, fmt.Errorf("decode run: %w", err)
296 }
297 name := pb.GetName()
298 if name == "" {
299 name = pb.GetWorkflowName() // stored before the field existed
300 }
301 return domain.CIRun{
302 Repo: repo, ID: pb.GetId(), Number: pb.GetNumber(), Attempt: pb.GetAttempt(),
303 WorkflowPath: pb.GetWorkflowPath(), WorkflowName: pb.GetWorkflowName(), Name: name, Title: pb.GetTitle(),
304 Event: pb.GetEvent(), Ref: pb.GetRef(), Sha: pb.GetSha(), HeadSha: pb.GetHeadSha(), Actor: pb.GetActor(),
305 Status: decodeStatus(pb.GetStatus()), Conclusion: decodeConclusion(pb.GetConclusion()), Error: pb.GetError(),
306 CreatedAt: decodeTime(pb.GetCreatedAt()), StartedAt: decodeTime(pb.GetStartedAt()), CompletedAt: decodeTime(pb.GetCompletedAt()),
307 CancelRequested: pb.GetCancelRequested(), CancelRequestedAt: decodeTime(pb.GetCancelRequestedAt()),
308 Trusted: pb.GetTrusted(), Concurrency: decodeConcurrency(pb.GetConcurrency()), Version: item.Version,
309 }, nil
310}
311
312func encodeJob(j domain.CIJob) ([]byte, error) {
313 if j.Repo.IsZero() {
314 return nil, fmt.Errorf("%w: empty name", domain.ErrInvalidName)
315 }
316 phase, err := encodePhase(j.Phase)
317 if err != nil {
318 return nil, err
319 }
320 steps := make([]*storagepb.CIStep, 0, len(j.Steps))
321 for _, s := range j.Steps {
322 steps = append(steps, &storagepb.CIStep{
323 Number: s.Number, Name: s.Name, State: encodeStepState(s.State), Conclusion: encodeConclusion(s.Conclusion),
324 StartedAt: encodeTime(s.StartedAt), CompletedAt: encodeTime(s.CompletedAt),
325 LogStart: s.LogStart, HasLines: s.HasLines, Outcome: encodeConclusion(s.Outcome),
326 })
327 }
328 pb := &storagepb.CIJob{
329 Id: j.ID, RunId: j.RunID, Attempt: j.Attempt, RunAttempt: j.RunAttempt, Repo: j.Repo.String(),
330 Key: j.Key, Name: j.Name, Phase: phase, Conclusion: encodeConclusion(j.Conclusion), Error: j.Error,
331 Labels: j.Labels, Needs: j.Needs, RunnerId: j.RunnerID, RunnerName: j.RunnerName,
332 Isolation: encodeIsolation(j.Isolation), Steps: steps,
333 CreatedAt: encodeTime(j.CreatedAt), StartedAt: encodeTime(j.StartedAt), CompletedAt: encodeTime(j.CompletedAt),
334 Index: int32(j.Index), Combo: int32(j.Combo), MatrixJson: j.MatrixJSON, ContextsJson: j.ContextsJSON, ComboTotal: int32(j.ComboTotal), //nolint:gosec // bounded by the 256 job limit
335 TimeoutMinutes: j.TimeoutMinutes, FailFast: j.FailFast, ContinueOnError: j.ContinueOnError, MaxParallel: int32(j.MaxParallel), //nolint:gosec // validated at planning
336 Outputs: j.Outputs, ReadyAt: encodeTime(j.ReadyAt), AssignedAt: encodeTime(j.AssignedAt),
337 AckedAt: encodeTime(j.AckedAt), AssignCancels: int32(j.AssignCancels), //nolint:gosec // at most 3
338 TokenValidUntil: encodeTime(j.TokenValidUntil), CancelReason: encodeCancelReason(j.CancelReason), CancelMessage: j.CancelMessage,
339 CancelRequestedAt: encodeTime(j.CancelRequestedAt), Planned: j.Planned, Environment: j.Environment, Permissions: j.Permissions,
340 Concurrency: encodeConcurrency(j.Concurrency), Scope: j.Scope, CallId: j.CallID, Call: encodeCall(j.Call),
341 }
342 if j.HasToken() {
343 pb.TokenHash = j.TokenHash[:]
344 }
345 return mustMarshal(pb), nil
346}
347
348//hexcheck:ignore suspicious-business-logic-in-adapter maps the stored record to the domain job field by field: no rule is decided here
349func decodeJob(item kv.Item) (domain.CIJob, error) {
350 var pb storagepb.CIJob
351 if err := proto.Unmarshal(item.Value, &pb); err != nil {
352 return domain.CIJob{}, fmt.Errorf("decode job: %w", err)
353 }
354 repo, err := domain.ParseRepoName(pb.GetRepo())
355 if err != nil {
356 return domain.CIJob{}, fmt.Errorf("decode job: %w", err)
357 }
358 j := domain.CIJob{
359 ID: pb.GetId(), RunID: pb.GetRunId(), Attempt: pb.GetAttempt(), RunAttempt: pb.GetRunAttempt(), Repo: repo,
360 Key: pb.GetKey(), Name: pb.GetName(), Phase: decodePhase(pb.GetPhase()), Conclusion: decodeConclusion(pb.GetConclusion()),
361 Error: pb.GetError(), Labels: pb.GetLabels(), Needs: pb.GetNeeds(), RunnerID: pb.GetRunnerId(), RunnerName: pb.GetRunnerName(),
362 Isolation: decodeIsolation(pb.GetIsolation()),
363 CreatedAt: decodeTime(pb.GetCreatedAt()), StartedAt: decodeTime(pb.GetStartedAt()), CompletedAt: decodeTime(pb.GetCompletedAt()),
364 Index: int(pb.GetIndex()), Combo: int(pb.GetCombo()), Planned: pb.GetPlanned(), MatrixJSON: pb.GetMatrixJson(), ContextsJSON: pb.GetContextsJson(),
365 ComboTotal: int(pb.GetComboTotal()), TimeoutMinutes: pb.GetTimeoutMinutes(), FailFast: pb.GetFailFast(), ContinueOnError: pb.GetContinueOnError(),
366 MaxParallel: int(pb.GetMaxParallel()), Outputs: pb.GetOutputs(),
367 TokenValidUntil: decodeTime(pb.GetTokenValidUntil()), ReadyAt: decodeTime(pb.GetReadyAt()),
368 AssignedAt: decodeTime(pb.GetAssignedAt()), AckedAt: decodeTime(pb.GetAckedAt()), AssignCancels: int(pb.GetAssignCancels()),
369 CancelMessage: pb.GetCancelMessage(), CancelRequestedAt: decodeTime(pb.GetCancelRequestedAt()),
370 Environment: pb.GetEnvironment(), Permissions: pb.GetPermissions(), Concurrency: decodeConcurrency(pb.GetConcurrency()), Version: item.Version,
371 Scope: pb.GetScope(), CallID: pb.GetCallId(), Call: decodeCall(pb.GetCall()),
372 }
373 j.CancelReason = decodeCancelReason(pb.GetCancelReason())
374 copy(j.TokenHash[:], pb.GetTokenHash())
375 for _, s := range pb.GetSteps() {
376 j.Steps = append(j.Steps, domain.CIStep{
377 Number: s.GetNumber(), Name: s.GetName(), State: decodeStepState(s.GetState()), Conclusion: decodeConclusion(s.GetConclusion()),
378 StartedAt: decodeTime(s.GetStartedAt()), CompletedAt: decodeTime(s.GetCompletedAt()),
379 LogStart: s.GetLogStart(), HasLines: s.GetHasLines(), Outcome: decodeConclusion(s.GetOutcome()),
380 })
381 }
382 return j, nil
383}
384
385func encodeCheck(c domain.CICheck) []byte {
386 return mustMarshal(&storagepb.CICheck{
387 Id: c.ID, Repo: c.Repo.String(), Sha: c.Sha, Name: c.Name, Status: encodeStatus(c.Status),
388 Conclusion: encodeConclusion(c.Conclusion), RunId: c.RunID, JobId: c.JobID, Attempt: c.Attempt,
389 WorkflowName: c.WorkflowName, CreatedAt: encodeTime(c.CreatedAt), StartedAt: encodeTime(c.StartedAt),
390 CompletedAt: encodeTime(c.CompletedAt),
391 })
392}
393
394func decodeCheck(item kv.Item) (domain.CICheck, error) {
395 var pb storagepb.CICheck
396 if err := proto.Unmarshal(item.Value, &pb); err != nil {
397 return domain.CICheck{}, fmt.Errorf("decode check: %w", err)
398 }
399 repo, err := domain.ParseRepoName(pb.GetRepo())
400 if err != nil {
401 return domain.CICheck{}, fmt.Errorf("decode check: %w", err)
402 }
403 return domain.CICheck{
404 ID: pb.GetId(), Repo: repo, Sha: pb.GetSha(), Name: pb.GetName(), Status: decodeStatus(pb.GetStatus()),
405 Conclusion: decodeConclusion(pb.GetConclusion()), RunID: pb.GetRunId(), JobID: pb.GetJobId(), Attempt: pb.GetAttempt(),
406 WorkflowName: pb.GetWorkflowName(), CreatedAt: decodeTime(pb.GetCreatedAt()), StartedAt: decodeTime(pb.GetStartedAt()),
407 CompletedAt: decodeTime(pb.GetCompletedAt()), Version: item.Version,
408 }, nil
409}
410
411func encodeRunner(r domain.CIRunner) []byte {
412 return mustMarshal(&storagepb.CIRunner{
413 Id: r.ID, Name: r.Name, Labels: r.Labels, Isolation: encodeIsolation(r.Isolation), Version: r.Version,
414 Arch: r.Arch, Os: r.OS, RegisteredAt: encodeTime(r.RegisteredAt), TokenHash: r.TokenHash[:],
415 })
416}
417
418func decodeRunner(val []byte) (domain.CIRunner, error) {
419 var pb storagepb.CIRunner
420 if err := proto.Unmarshal(val, &pb); err != nil {
421 return domain.CIRunner{}, fmt.Errorf("decode runner: %w", err)
422 }
423 r := domain.CIRunner{
424 ID: pb.GetId(), Name: pb.GetName(), Labels: pb.GetLabels(), Isolation: decodeIsolation(pb.GetIsolation()),
425 Version: pb.GetVersion(), Arch: pb.GetArch(), OS: pb.GetOs(), RegisteredAt: decodeTime(pb.GetRegisteredAt()),
426 }
427 copy(r.TokenHash[:], pb.GetTokenHash())
428 return r, nil
429}
430
431// --- writes ---
432
433// stage adds the preconditions and writes of one record to the batch.
434func stage(b *kv.Batch, key, val []byte, version uint64) {
435 if version == 0 {
436 b.RequireAbsent(key)
437 } else {
438 b.RequireVersion(key, version)
439 }
440 b.Put(key, val)
441}
442
443// runIndexes stages the open-run index of the run.
444func runIndexes(b *kv.Batch, r domain.CIRun) {
445 if r.Status == domain.CICompleted {
446 b.Delete(openKey(r.ID))
447 return
448 }
449 b.Put(openKey(r.ID), []byte(r.Repo.String()))
450}
451
452// jobIndexes stages the ready, active and token indexes of the job. old is
453// the job as stored before, if any: its token index goes when the token
454// changed.
455func (c *CI) jobIndexes(b *kv.Batch, j domain.CIJob, old *domain.CIJob) {
456 ref := jobRef(j.RunID, j.RunAttempt, j.ID)
457 if j.Phase == domain.CIPhaseReady {
458 b.Put(readyKey(j.ReadyAt, ref), nil)
459 } else {
460 b.Delete(readyKey(j.ReadyAt, ref))
461 }
462 if old != nil && old.Phase == domain.CIPhaseReady && (j.Phase != domain.CIPhaseReady || !old.ReadyAt.Equal(j.ReadyAt)) {
463 b.Delete(readyKey(old.ReadyAt, ref))
464 }
465 if j.Active() {
466 b.Put(activeKey(ref), nil)
467 } else {
468 b.Delete(activeKey(ref))
469 }
470 if old != nil && old.HasToken() && old.TokenHash != j.TokenHash {
471 b.Delete(jobTokenKey(old.TokenHash))
472 }
473 if j.HasToken() {
474 b.Put(jobTokenKey(j.TokenHash), []byte(ref))
475 }
476}
477
478// guardRepo is guardRepoKey on the database of the CI store, for the writes that
479// create keys under a repository.
480func (c *CI) guardRepo(ctx context.Context, b *kv.Batch, repo domain.RepoName) error {
481 return guardRepoKey(ctx, c.db, b, repo)
482}
483
484// stageRecords stages the writes of w. guarded collects the repositories that
485// need guarding.
486func (c *CI) stageRecords(ctx context.Context, b *kv.Batch, w ports.CIWrite) (guarded map[domain.RepoName]bool, err error) {
487 guarded = map[domain.RepoName]bool{}
488 for _, r := range w.Runs {
489 val, err := encodeRun(r)
490 if err != nil {
491 return nil, err
492 }
493 stage(b, runKey(r.Repo, r.ID), val, r.Version)
494 runIndexes(b, r)
495 if r.Version == 0 {
496 guarded[r.Repo] = true
497 }
498 }
499 for _, j := range w.Jobs {
500 val, err := encodeJob(j)
501 if err != nil {
502 return nil, err
503 }
504 key := jobKey(j.RunID, j.RunAttempt, j.ID)
505 var old *domain.CIJob
506 if j.Version != 0 {
507 item, ok := c.db.Get(ctx, key)
508 if !ok {
509 return nil, domain.ErrConflict
510 }
511 if o, err := decodeJob(item); err == nil {
512 old = &o
513 }
514 }
515 stage(b, key, val, j.Version)
516 c.jobIndexes(b, j, old)
517 if j.Version == 0 {
518 guarded[j.Repo] = true
519 }
520 }
521 c.stageRepoTokens(ctx, b, w, guarded)
522 c.stageAnnotations(ctx, b, w, guarded)
523 // Checks are derived from the jobs, written in the same batch and under
524 // the lock of their run: they are upserted without a precondition.
525 for _, ck := range w.Checks {
526 b.Put(checkKey(ck.Repo, ck.Sha, ck.ID), encodeCheck(ck))
527 }
528 return guarded, nil
529}
530
531// Apply writes the records in one batch.
532//
533//hexcheck:ignore suspicious-business-logic-in-adapter stages records and their derived indexes under version preconditions: a storage concern, no policy
534func (c *CI) Apply(ctx context.Context, w ports.CIWrite) (uint64, error) {
535 var b kv.Batch
536 guarded, err := c.stageRecords(ctx, &b, w)
537 if err != nil {
538 return 0, err
539 }
540 for repo := range guarded {
541 if err := c.guardRepo(ctx, &b, repo); err != nil {
542 return 0, err
543 }
544 }
545 v, err := c.db.Apply(ctx, &b)
546 if errors.Is(err, kv.ErrConflict) {
547 return 0, domain.ErrConflict
548 }
549 if err != nil {
550 return 0, fmt.Errorf("write CI records: %w", err)
551 }
552 return v, nil
553}
554
555func (c *CI) readSeq(ctx context.Context, key []byte) (last int64, item kv.Item, ok bool, err error) {
556 item, ok = c.db.Get(ctx, key)
557 if !ok {
558 return 0, kv.Item{}, false, nil
559 }
560 var seq storagepb.Sequence
561 if err := proto.Unmarshal(item.Value, &seq); err != nil {
562 return 0, kv.Item{}, false, fmt.Errorf("decode sequence %s: %w", key, err)
563 }
564 return seq.GetLast(), item, true, nil
565}
566
567func stageSeq(b *kv.Batch, key []byte, item kv.Item, ok bool, last int64) {
568 if ok {
569 b.RequireVersion(key, item.Version)
570 } else {
571 b.RequireAbsent(key)
572 }
573 b.Put(key, mustMarshal(&storagepb.Sequence{Last: last}))
574}
575
576const createRunTries = 20
577
578// CreateRun takes the next run id and number and stores the draft.
579//
580//hexcheck:ignore suspicious-business-logic-in-adapter takes the next numbers under version preconditions: a storage concern, no policy
581func (c *CI) CreateRun(ctx context.Context, repo domain.RepoName, workflowPath string, build func(id, number int64) ports.CIRunDraft) (ports.CIRunDraft, error) {
582 for range createRunTries {
583 lastID, idItem, idOK, err := c.readSeq(ctx, []byte(runSeqKey))
584 if err != nil {
585 return ports.CIRunDraft{}, err
586 }
587 numKey := runNumKey(repo, workflowPath)
588 lastNum, numItem, numOK, err := c.readSeq(ctx, numKey)
589 if err != nil {
590 return ports.CIRunDraft{}, err
591 }
592 draft := build(lastID+1, lastNum+1)
593 var b kv.Batch
594 stageSeq(&b, []byte(runSeqKey), idItem, idOK, lastID+1)
595 stageSeq(&b, numKey, numItem, numOK, lastNum+1)
596 w := ports.CIWrite{Runs: []domain.CIRun{draft.Run}, Jobs: draft.Jobs, Checks: draft.Checks}
597 guarded, err := c.stageRecords(ctx, &b, w)
598 if err != nil {
599 return ports.CIRunDraft{}, err
600 }
601 src := mustMarshal(&storagepb.CIRunSource{
602 RunId: draft.Run.ID, Repo: repo.String(), WorkflowYaml: draft.Source.WorkflowYAML, GithubJson: draft.Source.GithubJSON,
603 PusherKind: string(draft.Source.Pusher.Kind), PusherTokenId: draft.Source.Pusher.TokenID,
604 })
605 b.RequireAbsent(runSrcKey(draft.Run.ID))
606 b.Put(runSrcKey(draft.Run.ID), src)
607 for r := range guarded {
608 if err := c.guardRepo(ctx, &b, r); err != nil {
609 return ports.CIRunDraft{}, err
610 }
611 }
612 v, err := c.db.Apply(ctx, &b)
613 if errors.Is(err, kv.ErrConflict) {
614 continue
615 }
616 if err != nil {
617 return ports.CIRunDraft{}, fmt.Errorf("create run of %q: %w", repo, err)
618 }
619 draft.Run.Version = v
620 for i := range draft.Jobs {
621 draft.Jobs[i].Version = v
622 }
623 for i := range draft.Checks {
624 draft.Checks[i].Version = v
625 }
626 return draft, nil
627 }
628 return ports.CIRunDraft{}, fmt.Errorf("create run of %q: %w", repo, domain.ErrConflict)
629}
630
631// --- reads ---
632
633// GetRun reads run/<repo>/<id>.
634func (c *CI) GetRun(ctx context.Context, repo domain.RepoName, id int64) (domain.CIRun, error) {
635 item, ok := c.db.Get(ctx, runKey(repo, id))
636 if !ok || id < 1 {
637 return domain.CIRun{}, domain.ErrNotFound
638 }
639 return decodeRun(item)
640}
641
642// GetRunSource reads runsrc/<id>.
643func (c *CI) GetRunSource(ctx context.Context, runID int64) (domain.CIRunSource, error) {
644 item, ok := c.db.Get(ctx, runSrcKey(runID))
645 if !ok || runID < 1 {
646 return domain.CIRunSource{}, domain.ErrNotFound
647 }
648 var pb storagepb.CIRunSource
649 if err := proto.Unmarshal(item.Value, &pb); err != nil {
650 return domain.CIRunSource{}, fmt.Errorf("decode run source: %w", err)
651 }
652 repo, err := domain.ParseRepoName(pb.GetRepo())
653 if err != nil {
654 return domain.CIRunSource{}, fmt.Errorf("decode run source: %w", err)
655 }
656 return domain.CIRunSource{
657 RunID: pb.GetRunId(), Repo: repo, WorkflowYAML: pb.GetWorkflowYaml(), GithubJSON: pb.GetGithubJson(),
658 Pusher: domain.CIPusher{Kind: domain.CIPusherKind(pb.GetPusherKind()), TokenID: pb.GetPusherTokenId()},
659 }, nil
660}
661
662// ListRuns returns the runs of the repository, newest first.
663//
664//hexcheck:ignore suspicious-business-logic-in-adapter port method name only: reads a key range and reverses it
665func (c *CI) ListRuns(ctx context.Context, repo domain.RepoName, before int64, limit int) ([]domain.CIRun, error) {
666 start := []byte(runsPrefix(repo))
667 end := []byte(runPrefix + repo.String() + "0") // '/' + 1
668 if before > 0 {
669 end = runKey(repo, before)
670 }
671 // The snapshot iterates in key order only: keep the last limit items in a
672 // ring and decode only those, so the cost of a page does not grow with the
673 // decoded history of the repository.
674 type entry struct {
675 key []byte
676 item kv.Item
677 }
678 var ring []entry
679 next := 0
680 for key, item := range c.db.Snapshot(ctx).Range(ctx, start, end) {
681 if limit <= 0 || len(ring) < limit {
682 ring = append(ring, entry{key, item})
683 continue
684 }
685 ring[next] = entry{key, item}
686 next = (next + 1) % limit
687 }
688 ordered := append(ring[next:len(ring):len(ring)], ring[:next]...)
689 all := make([]domain.CIRun, 0, len(ordered))
690 for i := len(ordered) - 1; i >= 0; i-- {
691 r, err := decodeRun(ordered[i].item)
692 if err != nil {
693 return nil, fmt.Errorf("decode %s: %w", ordered[i].key, err)
694 }
695 all = append(all, r)
696 }
697 return all, nil
698}
699
700// RunIDs returns the ids of the runs of the repository.
701func (c *CI) RunIDs(ctx context.Context, repo domain.RepoName) ([]int64, error) {
702 var ids []int64
703 prefix := runsPrefix(repo)
704 for key := range c.db.Snapshot(ctx).Prefix(ctx, []byte(prefix)) {
705 id, err := strconv.ParseInt(strings.TrimPrefix(string(key), prefix), 10, 64)
706 if err != nil {
707 return nil, fmt.Errorf("decode %s: %w", key, err)
708 }
709 ids = append(ids, id)
710 }
711 return ids, nil
712}
713
714// ListOpenRuns returns the runs that have not completed.
715func (c *CI) ListOpenRuns(ctx context.Context) ([]domain.CIRun, error) {
716 snap := c.db.Snapshot(ctx)
717 var out []domain.CIRun
718 for key, item := range snap.Prefix(ctx, []byte(openPrefix)) {
719 id, err := strconv.ParseInt(strings.TrimPrefix(string(key), openPrefix), 10, 64)
720 if err != nil {
721 return nil, fmt.Errorf("decode %s: %w", key, err)
722 }
723 repo, err := domain.ParseRepoName(string(item.Value))
724 if err != nil {
725 return nil, err
726 }
727 it, ok := snap.Get(ctx, runKey(repo, id))
728 if !ok {
729 continue
730 }
731 r, err := decodeRun(it)
732 if err != nil {
733 return nil, err
734 }
735 out = append(out, r)
736 }
737 return out, nil
738}
739
740// GetJob reads one job record.
741func (c *CI) GetJob(ctx context.Context, runID int64, runAttempt int32, jobID string) (domain.CIJob, error) {
742 item, ok := c.db.Get(ctx, jobKey(runID, runAttempt, jobID))
743 if !ok {
744 return domain.CIJob{}, domain.ErrNotFound
745 }
746 return decodeJob(item)
747}
748
749// ListJobs returns the jobs of a run at an attempt, ordered by index then
750// combination.
751func (c *CI) ListJobs(ctx context.Context, runID int64, runAttempt int32) ([]domain.CIJob, error) {
752 var out []domain.CIJob
753 for key, item := range c.db.Snapshot(ctx).Prefix(ctx, []byte(jobAttemptPrefix(runID, runAttempt))) {
754 j, err := decodeJob(item)
755 if err != nil {
756 return nil, fmt.Errorf("decode %s: %w", key, err)
757 }
758 out = append(out, j)
759 }
760 slices.SortFunc(out, func(a, b domain.CIJob) int {
761 if a.Index != b.Index {
762 return a.Index - b.Index
763 }
764 return a.Combo - b.Combo
765 })
766 return out, nil
767}
768
769func (c *CI) jobByRef(snap *kv.Snapshot, ctx context.Context, ref string) (domain.CIJob, error) {
770 item, ok := snap.Get(ctx, []byte(ref))
771 if !ok {
772 return domain.CIJob{}, domain.ErrNotFound
773 }
774 return decodeJob(item)
775}
776
777// FindJobByToken resolves the token index, then reads the job.
778func (c *CI) FindJobByToken(ctx context.Context, hash domain.TokenHash, now time.Time) (domain.CIJob, error) {
779 snap := c.db.Snapshot(ctx)
780 idx, ok := snap.Get(ctx, jobTokenKey(hash))
781 if !ok {
782 return domain.CIJob{}, domain.ErrNotFound
783 }
784 j, err := c.jobByRef(snap, ctx, string(idx.Value))
785 if err != nil {
786 return domain.CIJob{}, err
787 }
788 if j.TokenHash != hash || (!j.TokenValidUntil.IsZero() && !now.Before(j.TokenValidUntil)) {
789 return domain.CIJob{}, domain.ErrNotFound
790 }
791 return j, nil
792}
793
794// ListReadyJobs returns the jobs waiting for a runner, oldest first.
795func (c *CI) ListReadyJobs(ctx context.Context, limit int) ([]domain.CIJob, error) {
796 snap := c.db.Snapshot(ctx)
797 var out []domain.CIJob
798 for key := range snap.Prefix(ctx, []byte(readyPrefix)) {
799 ref := string(key)[len(readyPrefix)+len(fmt.Sprintf(idFormat, 0))+1:]
800 j, err := c.jobByRef(snap, ctx, ref)
801 if errors.Is(err, domain.ErrNotFound) {
802 continue
803 }
804 if err != nil {
805 return nil, err
806 }
807 out = append(out, j)
808 if limit > 0 && len(out) >= limit {
809 break
810 }
811 }
812 return out, nil
813}
814
815// ListActiveJobs returns the jobs given to a runner and not over.
816func (c *CI) ListActiveJobs(ctx context.Context) ([]domain.CIJob, error) {
817 snap := c.db.Snapshot(ctx)
818 var out []domain.CIJob
819 for key := range snap.Prefix(ctx, []byte(activePrefix)) {
820 j, err := c.jobByRef(snap, ctx, strings.TrimPrefix(string(key), activePrefix))
821 if errors.Is(err, domain.ErrNotFound) {
822 continue
823 }
824 if err != nil {
825 return nil, err
826 }
827 out = append(out, j)
828 }
829 return out, nil
830}
831
832// ListChecks returns the checks of a commit ordered by creation time then
833// name.
834func (c *CI) ListChecks(ctx context.Context, repo domain.RepoName, sha string, limit int) ([]domain.CICheck, bool, error) {
835 var out []domain.CICheck
836 for key, item := range c.db.Snapshot(ctx).Prefix(ctx, []byte(checksPrefix(repo, sha))) {
837 ck, err := decodeCheck(item)
838 if err != nil {
839 return nil, false, fmt.Errorf("decode %s: %w", key, err)
840 }
841 out = append(out, ck)
842 }
843 slices.SortStableFunc(out, func(a, b domain.CICheck) int {
844 if d := a.CreatedAt.Compare(b.CreatedAt); d != 0 {
845 return d
846 }
847 return strings.Compare(a.Name, b.Name)
848 })
849 more := limit > 0 && len(out) > limit
850 if more {
851 out = out[:limit]
852 }
853 return out, more, nil
854}
855
856// --- retention and deletion ---
857
858// runKeys lists every key of a run except its record: source, jobs, indexes.
859func (c *CI) runKeys(ctx context.Context, snap *kv.Snapshot, repo domain.RepoName, runID int64) [][]byte {
860 keys := [][]byte{runKey(repo, runID), runSrcKey(runID), openKey(runID)}
861 keys = append(keys, repoTokenKeys(ctx, snap, runID)...)
862 keys = append(keys, annotationKeys(ctx, snap, runID)...)
863 for key, item := range snap.Prefix(ctx, []byte(jobsPrefix(runID))) {
864 keys = append(keys, slices.Clone(key))
865 if j, err := decodeJob(item); err == nil {
866 ref := jobRef(j.RunID, j.RunAttempt, j.ID)
867 keys = append(keys, readyKey(j.ReadyAt, ref), activeKey(ref))
868 if j.HasToken() {
869 keys = append(keys, jobTokenKey(j.TokenHash))
870 }
871 }
872 }
873 return keys
874}
875
876// ciKeys lists every key of the snapshot that belongs to the repository's CI
877// runs, checks and numbers.
878func ciKeys(ctx context.Context, snap *kv.Snapshot, repo domain.RepoName) [][]byte {
879 c := &CI{}
880 var keys [][]byte
881 for key, item := range snap.Prefix(ctx, []byte(runsPrefix(repo))) {
882 if r, err := decodeRun(item); err == nil {
883 keys = append(keys, c.runKeys(ctx, snap, repo, r.ID)...)
884 }
885 keys = append(keys, slices.Clone(key))
886 }
887 keys = append(keys, repoReaderKeys(ctx, snap, repo)...)
888 for _, p := range []string{checkPrefix + repo.String() + "/", runNumPrefix + repo.String() + "/", concurPrefix + repo.String() + "/", concurMarkPrefixAll + repo.String() + "/"} {
889 for k := range snap.Prefix(ctx, []byte(p)) {
890 keys = append(keys, slices.Clone(k))
891 }
892 }
893 return keys
894}
895
896const sweepChunk = 500
897
898// Sweep removes the runs completed before the time.
899//
900//hexcheck:ignore suspicious-business-logic-in-adapter chunked delete of expired records: a storage concern, the retention is decided by the caller
901func (c *CI) Sweep(ctx context.Context, before time.Time) ([]int64, error) {
902 snap := c.db.Snapshot(ctx)
903 var ids []int64
904 var keys [][]byte
905 for _, rp := range []string{runPrefix} {
906 for _, item := range snap.Prefix(ctx, []byte(rp)) {
907 r, err := decodeRun(item)
908 if err != nil {
909 return nil, err
910 }
911 if r.Status != domain.CICompleted || r.CompletedAt.IsZero() || !r.CompletedAt.Before(before) {
912 continue
913 }
914 ids = append(ids, r.ID)
915 keys = append(keys, c.runKeys(ctx, snap, r.Repo, r.ID)...)
916 for ck := range snap.Prefix(ctx, []byte(checkPrefix+r.Repo.String()+"/"+r.CheckSha()+"/")) {
917 // A check belongs to the run only while it is the latest one
918 // for its job: leave it when another run replaced it.
919 if it, ok := snap.Get(ctx, ck); ok {
920 if chk, err := decodeCheck(it); err == nil && chk.RunID == r.ID {
921 keys = append(keys, slices.Clone(ck))
922 }
923 }
924 }
925 }
926 }
927 for len(keys) > 0 {
928 n := min(sweepChunk, len(keys))
929 var b kv.Batch
930 for _, k := range keys[:n] {
931 b.Delete(k)
932 }
933 if _, err := c.db.Apply(ctx, &b); err != nil {
934 return nil, fmt.Errorf("sweep CI runs: %w", err)
935 }
936 keys = keys[n:]
937 }
938 return ids, nil
939}
940
941// --- runners ---
942
943func runnerKey(id string) []byte { return []byte(runnerPrefix + id) }
944func runnerNameKey(n string) []byte { return []byte(runnerName + strings.ToLower(n)) }
945func runnerTokenKey(h domain.TokenHash) []byte { return []byte(runnerTokPref + h.String()) }
946
947// CreateRunner stores the runner, its name index and its token index.
948//
949//hexcheck:ignore suspicious-business-logic-in-adapter replaces a runner of the same name in one batch: a storage concern, the rule is decided by the caller
950func (c *CI) CreateRunner(ctx context.Context, r domain.CIRunner, replace bool) ([]domain.CIRunner, error) {
951 for range createRunTries {
952 var b kv.Batch
953 var removed []domain.CIRunner
954 if idx, ok := c.db.Get(ctx, runnerNameKey(r.Name)); ok {
955 if !replace {
956 return nil, domain.ErrAlreadyExists
957 }
958 old, err := c.GetRunner(ctx, string(idx.Value))
959 if err != nil && !errors.Is(err, domain.ErrNotFound) {
960 return nil, err
961 }
962 b.RequireVersion(runnerNameKey(r.Name), idx.Version)
963 if err == nil {
964 removed = append(removed, old)
965 b.Delete(runnerKey(old.ID))
966 b.Delete(runnerTokenKey(old.TokenHash))
967 }
968 } else {
969 b.RequireAbsent(runnerNameKey(r.Name))
970 }
971 b.RequireAbsent(runnerKey(r.ID))
972 b.RequireAbsent(runnerTokenKey(r.TokenHash))
973 b.Put(runnerKey(r.ID), encodeRunner(r))
974 b.Put(runnerNameKey(r.Name), []byte(r.ID))
975 b.Put(runnerTokenKey(r.TokenHash), []byte(r.ID))
976 _, err := c.db.Apply(ctx, &b)
977 if errors.Is(err, kv.ErrConflict) {
978 if !replace {
979 if _, ok := c.db.Get(ctx, runnerNameKey(r.Name)); ok {
980 return nil, domain.ErrAlreadyExists
981 }
982 }
983 continue
984 }
985 if err != nil {
986 return nil, fmt.Errorf("create runner %q: %w", r.Name, err)
987 }
988 return removed, nil
989 }
990 return nil, fmt.Errorf("create runner %q: %w", r.Name, domain.ErrConflict)
991}
992
993// GetRunner reads runner/<id>.
994func (c *CI) GetRunner(ctx context.Context, id string) (domain.CIRunner, error) {
995 item, ok := c.db.Get(ctx, runnerKey(id))
996 if !ok || id == "" || strings.Contains(id, "/") {
997 return domain.CIRunner{}, domain.ErrNotFound
998 }
999 return decodeRunner(item.Value)
1000}
1001
1002// FindRunnerByToken resolves the token index, then reads the runner.
1003func (c *CI) FindRunnerByToken(ctx context.Context, hash domain.TokenHash) (domain.CIRunner, error) {
1004 snap := c.db.Snapshot(ctx)
1005 idx, ok := snap.Get(ctx, runnerTokenKey(hash))
1006 if !ok {
1007 return domain.CIRunner{}, domain.ErrNotFound
1008 }
1009 item, ok := snap.Get(ctx, runnerKey(string(idx.Value)))
1010 if !ok {
1011 return domain.CIRunner{}, domain.ErrNotFound
1012 }
1013 r, err := decodeRunner(item.Value)
1014 if err != nil {
1015 return domain.CIRunner{}, err
1016 }
1017 if r.TokenHash != hash {
1018 return domain.CIRunner{}, fmt.Errorf("runner %q: token index does not match the stored hash", r.ID)
1019 }
1020 return r, nil
1021}
1022
1023// ListRunners returns every runner ordered by name.
1024func (c *CI) ListRunners(ctx context.Context) ([]domain.CIRunner, error) {
1025 var out []domain.CIRunner
1026 for key, item := range c.db.Snapshot(ctx).Prefix(ctx, []byte(runnerPrefix)) {
1027 r, err := decodeRunner(item.Value)
1028 if err != nil {
1029 return nil, fmt.Errorf("decode %s: %w", key, err)
1030 }
1031 out = append(out, r)
1032 }
1033 slices.SortFunc(out, func(a, b domain.CIRunner) int {
1034 return strings.Compare(strings.ToLower(a.Name), strings.ToLower(b.Name))
1035 })
1036 return out, nil
1037}
1038
1039// RemoveRunner removes the runner, its name and its token.
1040func (c *CI) RemoveRunner(ctx context.Context, id string) (domain.CIRunner, error) {
1041 for range createRunTries {
1042 item, ok := c.db.Get(ctx, runnerKey(id))
1043 if !ok || id == "" || strings.Contains(id, "/") {
1044 return domain.CIRunner{}, domain.ErrNotFound
1045 }
1046 r, err := decodeRunner(item.Value)
1047 if err != nil {
1048 return domain.CIRunner{}, err
1049 }
1050 var b kv.Batch
1051 b.RequireVersion(runnerKey(id), item.Version)
1052 b.Delete(runnerKey(id))
1053 b.Delete(runnerNameKey(r.Name))
1054 b.Delete(runnerTokenKey(r.TokenHash))
1055 _, err = c.db.Apply(ctx, &b)
1056 if errors.Is(err, kv.ErrConflict) {
1057 continue
1058 }
1059 if err != nil {
1060 return domain.CIRunner{}, fmt.Errorf("remove runner %q: %w", id, err)
1061 }
1062 return r, nil
1063 }
1064 return domain.CIRunner{}, fmt.Errorf("remove runner %q: %w", id, domain.ErrConflict)
1065}
1066
1067func encodeCall(c *domain.CICall) *storagepb.CICall {
1068 if c == nil {
1069 return nil
1070 }
1071 return &storagepb.CICall{
1072 Uses: c.Uses, Path: c.Path, Repository: c.Repository, Ref: c.Ref, Sha: c.Sha, Yaml: c.YAML, InputsJson: c.InputsJSON,
1073 Secrets: c.Secrets, Inherit: c.Inherit, Depth: int32(c.Depth), Expanded: c.Expanded, Concurrency: encodeConcurrency(c.Concurrency), Admitted: c.Admitted, SecretCtx: c.SecretCtx, //nolint:gosec // at most 4
1074 }
1075}
1076
1077//hexcheck:ignore suspicious-business-logic-in-adapter maps the stored call to the domain call field by field: no rule is decided here
1078func decodeCall(pb *storagepb.CICall) *domain.CICall {
1079 if pb == nil {
1080 return nil
1081 }
1082 return &domain.CICall{
1083 Uses: pb.GetUses(), Path: pb.GetPath(), Repository: pb.GetRepository(), Ref: pb.GetRef(), Sha: pb.GetSha(), YAML: pb.GetYaml(),
1084 InputsJSON: pb.GetInputsJson(), Secrets: pb.GetSecrets(), Inherit: pb.GetInherit(), Depth: int(pb.GetDepth()), Expanded: pb.GetExpanded(), Concurrency: decodeConcurrency(pb.GetConcurrency()),
1085 Admitted: pb.GetAdmitted(), SecretCtx: pb.GetSecretCtx(),
1086 }
1087}