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