1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package kvstore
4
5import (
6 "context"
7 "fmt"
8 "slices"
9 "strconv"
10 "strings"
11
12 kv "bnema.dev/ermite-kv"
13 "google.golang.org/protobuf/proto"
14
15 storagepb "bnema.dev/ermite/gen/ermite/storage"
16 "bnema.dev/ermite/internal/domain"
17 "bnema.dev/ermite/internal/ports"
18)
19
20// Key layout of the annotations, written by CI.Apply with the job they belong
21// to, under the lock of its run:
22//
23// annot/<run:020d>/<attempt:010d>/<job id>/<step:010d> CIStepAnnotations
24//
25// The attempt is the one in which the job executed (CIJob.Attempt, the one of
26// its log), not the attempt of the run that keeps the record. One value holds
27// all the annotations of one step, so that a report replaces them whole; a
28// step without annotation has no key.
29const annotPrefix = "annot/"
30
31func annotRunPrefix(runID int64) string { return fmt.Sprintf("%s%020d/", annotPrefix, runID) }
32func annotJobPrefix(runID int64, attempt int32, jobID string) string {
33 return fmt.Sprintf("%s%010d/%s/", annotRunPrefix(runID), attempt, jobID)
34}
35func annotKey(runID int64, attempt int32, jobID string, step int32) []byte {
36 return []byte(fmt.Sprintf("%s%010d", annotJobPrefix(runID, attempt, jobID), step))
37}
38
39var annotationLevels = []struct {
40 d domain.AnnotationLevel
41 p storagepb.CIAnnotationLevel
42}{
43 {domain.AnnotationNotice, storagepb.CIAnnotationLevel_CI_ANNOTATION_LEVEL_NOTICE},
44 {domain.AnnotationWarning, storagepb.CIAnnotationLevel_CI_ANNOTATION_LEVEL_WARNING},
45 {domain.AnnotationFailure, storagepb.CIAnnotationLevel_CI_ANNOTATION_LEVEL_FAILURE},
46}
47
48func encodeAnnotationLevel(l domain.AnnotationLevel) storagepb.CIAnnotationLevel {
49 for _, e := range annotationLevels {
50 if e.d == l {
51 return e.p
52 }
53 }
54 return storagepb.CIAnnotationLevel_CI_ANNOTATION_LEVEL_UNSPECIFIED
55}
56
57func decodeAnnotationLevel(l storagepb.CIAnnotationLevel) domain.AnnotationLevel {
58 for _, e := range annotationLevels {
59 if e.p == l {
60 return e.d
61 }
62 }
63 return domain.AnnotationNotice
64}
65
66func encodeAnnotations(items []domain.CIAnnotation) []byte {
67 pb := &storagepb.CIStepAnnotations{Annotations: make([]*storagepb.CIAnnotation, 0, len(items))}
68 for _, a := range items {
69 pb.Annotations = append(pb.Annotations, &storagepb.CIAnnotation{
70 Level: encodeAnnotationLevel(a.Level), Title: a.Title, Message: a.Message, File: a.File,
71 StartLine: a.StartLine, EndLine: a.EndLine, StartColumn: a.StartColumn, EndColumn: a.EndColumn,
72 })
73 }
74 return mustMarshal(pb)
75}
76
77func decodeAnnotations(val []byte, step int32) ([]domain.CIAnnotation, error) {
78 var pb storagepb.CIStepAnnotations
79 if err := proto.Unmarshal(val, &pb); err != nil {
80 return nil, fmt.Errorf("decode annotations: %w", err)
81 }
82 out := make([]domain.CIAnnotation, 0, len(pb.GetAnnotations()))
83 for _, a := range pb.GetAnnotations() {
84 out = append(out, domain.CIAnnotation{
85 Level: decodeAnnotationLevel(a.GetLevel()), Title: a.GetTitle(), Message: a.GetMessage(), File: a.GetFile(),
86 StartLine: a.GetStartLine(), EndLine: a.GetEndLine(), StartColumn: a.GetStartColumn(), EndColumn: a.GetEndColumn(),
87 Step: step,
88 })
89 }
90 return out, nil
91}
92
93// stageAnnotations stages the deletion of the annotations of the executions to
94// clear, then the replacement of the annotations of the steps to set.
95//
96// Annotations normally travel with the write of their job, whose version
97// precondition fails once the job is gone; guarding the repository key as well
98// would make the jobs of one repository conflict with one another at every
99// report. Only a batch without a job guards it. A report that races the
100// deletion of its repository can leave its annotation keys behind, under a run
101// id that is never reused: harmless, and a few kilobytes at most.
102func (c *CI) stageAnnotations(ctx context.Context, b *kv.Batch, w ports.CIWrite, guarded map[domain.RepoName]bool) {
103 if len(w.ClearAnnotations) > 0 {
104 snap := c.db.Snapshot(ctx)
105 for _, e := range w.ClearAnnotations {
106 for key := range snap.Prefix(ctx, []byte(annotJobPrefix(e.RunID, e.Attempt, e.JobID))) {
107 b.Delete(slices.Clone(key))
108 }
109 }
110 }
111 for _, s := range w.Annotations {
112 key := annotKey(s.Exec.RunID, s.Exec.Attempt, s.Exec.JobID, s.Step)
113 if len(s.Items) == 0 {
114 b.Delete(key)
115 continue
116 }
117 b.Put(key, encodeAnnotations(s.Items))
118 if len(w.Jobs) == 0 {
119 guarded[s.Repo] = true
120 }
121 }
122}
123
124// JobAnnotations returns the annotations of one execution of a job, ordered by
125// step then by the order the step produced them.
126func (c *CI) JobAnnotations(ctx context.Context, runID int64, attempt int32, jobID string) ([]domain.CIAnnotation, error) {
127 prefix := annotJobPrefix(runID, attempt, jobID)
128 var out []domain.CIAnnotation
129 for key, item := range c.db.Snapshot(ctx).Prefix(ctx, []byte(prefix)) {
130 n, err := strconv.ParseInt(strings.TrimPrefix(string(key), prefix), 10, 32)
131 if err != nil {
132 return nil, fmt.Errorf("decode %s: %w", key, err)
133 }
134 items, err := decodeAnnotations(item.Value, int32(n)) //nolint:gosec // parsed with a 32-bit size
135 if err != nil {
136 return nil, fmt.Errorf("decode %s: %w", key, err)
137 }
138 out = append(out, items...)
139 }
140 return out, nil
141}
142
143// annotationKeys lists the keys of the annotations of a run, for its deletion.
144func annotationKeys(ctx context.Context, snap *kv.Snapshot, runID int64) [][]byte {
145 var keys [][]byte
146 for key := range snap.Prefix(ctx, []byte(annotRunPrefix(runID))) {
147 keys = append(keys, slices.Clone(key))
148 }
149 return keys
150}