1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package kvstore
4
5import (
6 "context"
7 "encoding/hex"
8 "errors"
9 "fmt"
10 "slices"
11 "strconv"
12 "strings"
13 "unicode/utf8"
14
15 kv "bnema.dev/ermite-kv"
16 "google.golang.org/protobuf/proto"
17
18 storagepb "bnema.dev/ermite/gen/ermite/storage"
19 "bnema.dev/ermite/internal/domain"
20 "bnema.dev/ermite/internal/ports"
21)
22
23// Key layout of the CI artifacts (CI-4 D3). New prefixes only: an older build
24// ignores them, so there is no schema change.
25//
26// artifact/<repo>/<run:020d>/<id:020d> CIArtifact
27// artifactid/<id:020d> index: "<repo>/<run:020d>"
28// artifactblk/<id:020d>/<hex(block id)> CIArtifactBlock: a staged block
29// artifactuse/<repo> CIArtifactUsage: bytes charged
30// artifactseq Sequence: last artifact id
31//
32// Every write of the artifacts of a repository requires and rewrites its
33// usage record (ports.CIArtifactWrite). The records of a repository go with it
34// (ciKeys); those of a run that retention removed are deleted by the sweep of
35// the artifacts, which refunds them.
36const (
37 artifactPrefix = "artifact/"
38 artifactIDPrefix = "artifactid/"
39 artifactBlkPrefix = "artifactblk/"
40 artifactUsePrefix = "artifactuse/"
41 artifactSeqKey = "artifactseq"
42)
43
44func artifactRunPrefix(repo domain.RepoName, runID int64) string {
45 return artifactPrefix + repo.String() + "/" + fmt.Sprintf(idFormat, runID) + "/"
46}
47func artifactKey(repo domain.RepoName, runID, id int64) []byte {
48 return []byte(artifactRunPrefix(repo, runID) + fmt.Sprintf(idFormat, id))
49}
50func artifactIDKey(id int64) []byte { return []byte(artifactIDPrefix + fmt.Sprintf(idFormat, id)) }
51func artifactBlocksPrefix(id int64) string {
52 return artifactBlkPrefix + fmt.Sprintf(idFormat, id) + "/"
53}
54func artifactBlockKey(id int64, block []byte) []byte {
55 return []byte(artifactBlocksPrefix(id) + hex.EncodeToString(block))
56}
57func artifactUseKey(repo domain.RepoName) []byte { return []byte(artifactUsePrefix + repo.String()) }
58
59// CIArtifacts is the ermite-kv backed ports.CIArtifactStore. It shares the
60// database of the Store it comes from.
61type CIArtifacts struct {
62 db *kv.DB
63}
64
65var _ ports.CIArtifactStore = (*CIArtifacts)(nil)
66
67// Artifacts returns the artifact store over the same database. It is valid
68// until the Store is closed.
69func (s *Store) Artifacts() *CIArtifacts { return &CIArtifacts{db: s.db} }
70
71// --- encoding ---
72
73var artifactStates = []struct {
74 d domain.CIArtifactState
75 p storagepb.CIArtifactState
76}{
77 {domain.CIArtifactPending, storagepb.CIArtifactState_CI_ARTIFACT_STATE_PENDING},
78 {domain.CIArtifactCommitted, storagepb.CIArtifactState_CI_ARTIFACT_STATE_COMMITTED},
79 {domain.CIArtifactFinal, storagepb.CIArtifactState_CI_ARTIFACT_STATE_FINAL},
80}
81
82func encodeArtifact(a domain.CIArtifact) ([]byte, error) {
83 if a.Repo.IsZero() || a.ID < 1 || a.RunID < 1 {
84 return nil, fmt.Errorf("%w: artifact without id, run or repository", domain.ErrInvalid)
85 }
86 if !utf8.ValidString(a.Name) || !utf8.ValidString(a.MediaType) || !utf8.ValidString(a.JobID) {
87 return nil, fmt.Errorf("%w: artifact strings must be valid UTF-8", domain.ErrInvalid)
88 }
89 pb := &storagepb.CIArtifact{
90 Id: a.ID, Repo: a.Repo.String(), RunId: a.RunID, Attempt: a.Attempt, JobId: a.JobID,
91 Name: a.Name, MediaType: a.MediaType, Size: a.Size, Digest: a.Digest,
92 CreatedAt: encodeTime(a.CreatedAt), ExpiresAt: encodeTime(a.ExpiresAt),
93 Charged: a.Charged, ListDigest: a.ListDigest,
94 }
95 if a.Assignment != (domain.TokenHash{}) {
96 pb.Assignment = a.Assignment[:]
97 }
98 for _, s := range artifactStates {
99 if s.d == a.State {
100 pb.State = s.p
101 }
102 }
103 if pb.State == storagepb.CIArtifactState_CI_ARTIFACT_STATE_UNSPECIFIED {
104 return nil, fmt.Errorf("%w: artifact state %q", domain.ErrInvalid, a.State)
105 }
106 return mustMarshal(pb), nil
107}
108
109//hexcheck:ignore suspicious-business-logic-in-adapter decodes a record into the domain type: a storage concern, no policy
110func decodeArtifact(item kv.Item) (domain.CIArtifact, error) {
111 var pb storagepb.CIArtifact
112 if err := proto.Unmarshal(item.Value, &pb); err != nil {
113 return domain.CIArtifact{}, fmt.Errorf("decode artifact: %w", err)
114 }
115 repo, err := domain.ParseRepoName(pb.GetRepo())
116 if err != nil {
117 return domain.CIArtifact{}, fmt.Errorf("decode artifact: %w", err)
118 }
119 a := domain.CIArtifact{
120 ID: pb.GetId(), Repo: repo, RunID: pb.GetRunId(), Attempt: pb.GetAttempt(), JobID: pb.GetJobId(),
121 Name: pb.GetName(), MediaType: pb.GetMediaType(), Size: pb.GetSize(), Digest: pb.GetDigest(),
122 CreatedAt: decodeTime(pb.GetCreatedAt()), ExpiresAt: decodeTime(pb.GetExpiresAt()),
123 Charged: pb.GetCharged(), ListDigest: pb.GetListDigest(), Version: item.Version,
124 }
125 copy(a.Assignment[:], pb.GetAssignment())
126 for _, s := range artifactStates {
127 if s.p == pb.GetState() {
128 a.State = s.d
129 }
130 }
131 if a.State == "" {
132 return domain.CIArtifact{}, fmt.Errorf("decode artifact %d: unknown state %v", a.ID, pb.GetState())
133 }
134 return a, nil
135}
136
137// --- reads ---
138
139// artifactIDTries bounds the attempts of NextArtifactID: each conflict means
140// another call took an id, and uploads begin many at once.
141const artifactIDTries = 1000
142
143// NextArtifactID takes the next id in its own batch.
144//
145//hexcheck:ignore suspicious-business-logic-in-adapter takes the next number under a version precondition: a storage concern, no policy
146func (c *CIArtifacts) NextArtifactID(ctx context.Context) (int64, error) {
147 ci := &CI{db: c.db}
148 for range artifactIDTries {
149 last, item, ok, err := ci.readSeq(ctx, []byte(artifactSeqKey))
150 if err != nil {
151 return 0, err
152 }
153 var b kv.Batch
154 stageSeq(&b, []byte(artifactSeqKey), item, ok, last+1)
155 _, err = c.db.Apply(ctx, &b)
156 if errors.Is(err, kv.ErrConflict) {
157 continue
158 }
159 if err != nil {
160 return 0, fmt.Errorf("take an artifact id: %w", err)
161 }
162 return last + 1, nil
163 }
164 return 0, fmt.Errorf("take an artifact id: %w", domain.ErrConflict)
165}
166
167// ArtifactUsage reads artifactuse/<repo>.
168func (c *CIArtifacts) ArtifactUsage(ctx context.Context, repo domain.RepoName) (int64, uint64, error) {
169 item, ok := c.db.Get(ctx, artifactUseKey(repo))
170 if !ok {
171 return 0, 0, nil
172 }
173 var pb storagepb.CIArtifactUsage
174 if err := proto.Unmarshal(item.Value, &pb); err != nil {
175 return 0, 0, fmt.Errorf("decode artifact usage: %w", err)
176 }
177 return pb.GetUsed(), item.Version, nil
178}
179
180// GetArtifact resolves the id index, then reads the record.
181func (c *CIArtifacts) GetArtifact(ctx context.Context, id int64) (domain.CIArtifact, error) {
182 snap := c.db.Snapshot(ctx)
183 idx, ok := snap.Get(ctx, artifactIDKey(id))
184 if !ok || id < 1 {
185 return domain.CIArtifact{}, domain.ErrNotFound
186 }
187 repo, runID, err := decodeArtifactIndex(idx.Value)
188 if err != nil {
189 return domain.CIArtifact{}, err
190 }
191 item, ok := snap.Get(ctx, artifactKey(repo, runID, id))
192 if !ok {
193 return domain.CIArtifact{}, domain.ErrNotFound
194 }
195 return decodeArtifact(item)
196}
197
198func encodeArtifactIndex(a domain.CIArtifact) []byte {
199 return []byte(a.Repo.String() + "/" + fmt.Sprintf(idFormat, a.RunID))
200}
201
202func decodeArtifactIndex(val []byte) (domain.RepoName, int64, error) {
203 name, run, ok := strings.Cut(string(val), "/")
204 if !ok {
205 return domain.RepoName{}, 0, fmt.Errorf("decode artifact index %q", val)
206 }
207 repo, err := domain.ParseRepoName(name)
208 if err != nil {
209 return domain.RepoName{}, 0, fmt.Errorf("decode artifact index: %w", err)
210 }
211 runID, err := strconv.ParseInt(run, 10, 64)
212 if err != nil {
213 return domain.RepoName{}, 0, fmt.Errorf("decode artifact index: %w", err)
214 }
215 return repo, runID, nil
216}
217
218// RunArtifacts lists artifact/<repo>/<run>/, which is in id order.
219func (c *CIArtifacts) RunArtifacts(ctx context.Context, repo domain.RepoName, runID int64) ([]domain.CIArtifact, error) {
220 return c.list(ctx, artifactRunPrefix(repo, runID))
221}
222
223// AllArtifacts lists artifact/.
224func (c *CIArtifacts) AllArtifacts(ctx context.Context) ([]domain.CIArtifact, error) {
225 return c.list(ctx, artifactPrefix)
226}
227
228func (c *CIArtifacts) list(ctx context.Context, prefix string) ([]domain.CIArtifact, error) {
229 var out []domain.CIArtifact
230 for key, item := range c.db.Snapshot(ctx).Prefix(ctx, []byte(prefix)) {
231 a, err := decodeArtifact(item)
232 if err != nil {
233 return nil, fmt.Errorf("decode %s: %w", key, err)
234 }
235 out = append(out, a)
236 }
237 return out, nil
238}
239
240// ArtifactBlocks lists artifactblk/<id>/.
241func (c *CIArtifacts) ArtifactBlocks(ctx context.Context, id int64) ([]ports.CIArtifactBlock, error) {
242 prefix := artifactBlocksPrefix(id)
243 var out []ports.CIArtifactBlock
244 for key, item := range c.db.Snapshot(ctx).Prefix(ctx, []byte(prefix)) {
245 block, err := hex.DecodeString(strings.TrimPrefix(string(key), prefix))
246 if err != nil {
247 return nil, fmt.Errorf("decode %s: %w", key, err)
248 }
249 var pb storagepb.CIArtifactBlock
250 if err := proto.Unmarshal(item.Value, &pb); err != nil {
251 return nil, fmt.Errorf("decode %s: %w", key, err)
252 }
253 out = append(out, ports.CIArtifactBlock{ArtifactID: id, ID: block, Size: pb.GetSize(), File: pb.GetFile()})
254 }
255 return out, nil
256}
257
258// BlockArtifactIDs lists the artifact ids under artifactblk/.
259func (c *CIArtifacts) BlockArtifactIDs(ctx context.Context) ([]int64, error) {
260 var ids []int64
261 for key := range c.db.Snapshot(ctx).Prefix(ctx, []byte(artifactBlkPrefix)) {
262 raw, _, _ := strings.Cut(strings.TrimPrefix(string(key), artifactBlkPrefix), "/")
263 id, err := strconv.ParseInt(raw, 10, 64)
264 if err != nil {
265 return nil, fmt.Errorf("decode %s: %w", key, err)
266 }
267 if len(ids) == 0 || ids[len(ids)-1] != id {
268 ids = append(ids, id)
269 }
270 }
271 return ids, nil
272}
273
274// --- writes ---
275
276// ApplyArtifacts writes w in one batch.
277//
278//hexcheck:ignore suspicious-business-logic-in-adapter stages records and their derived keys under version preconditions: a storage concern, no policy
279func (c *CIArtifacts) ApplyArtifacts(ctx context.Context, w ports.CIArtifactWrite) error {
280 var b kv.Batch
281 guard := false
282 if w.UsageVersion == 0 {
283 b.RequireAbsent(artifactUseKey(w.Repo))
284 guard = true
285 } else {
286 b.RequireVersion(artifactUseKey(w.Repo), w.UsageVersion)
287 }
288 b.Put(artifactUseKey(w.Repo), mustMarshal(&storagepb.CIArtifactUsage{Used: w.Used}))
289 snap := c.db.Snapshot(ctx)
290 for _, a := range w.Put {
291 if a.Repo != w.Repo {
292 return fmt.Errorf("%w: artifact %d of another repository", domain.ErrInvalid, a.ID)
293 }
294 val, err := encodeArtifact(a)
295 if err != nil {
296 return err
297 }
298 stage(&b, artifactKey(a.Repo, a.RunID, a.ID), val, a.Version)
299 if a.Version == 0 {
300 b.RequireAbsent(artifactIDKey(a.ID))
301 b.Put(artifactIDKey(a.ID), encodeArtifactIndex(a))
302 guard = true
303 }
304 }
305 for _, a := range w.Delete {
306 if a.Repo != w.Repo || a.Version == 0 {
307 return fmt.Errorf("%w: delete of artifact %d", domain.ErrInvalid, a.ID)
308 }
309 key := artifactKey(a.Repo, a.RunID, a.ID)
310 b.RequireVersion(key, a.Version)
311 b.Delete(key)
312 b.Delete(artifactIDKey(a.ID))
313 for k := range snap.Prefix(ctx, []byte(artifactBlocksPrefix(a.ID))) {
314 b.Delete(slices.Clone(k))
315 }
316 }
317 for _, id := range w.DropBlocks {
318 for k := range snap.Prefix(ctx, []byte(artifactBlocksPrefix(id))) {
319 b.Delete(slices.Clone(k))
320 }
321 }
322 for _, blk := range w.PutBlocks {
323 if !utf8.ValidString(blk.File) {
324 return fmt.Errorf("%w: block file name", domain.ErrInvalid)
325 }
326 b.Put(artifactBlockKey(blk.ArtifactID, blk.ID), mustMarshal(&storagepb.CIArtifactBlock{Size: blk.Size, File: blk.File}))
327 }
328 if guard {
329 if err := guardRepoKey(ctx, c.db, &b, w.Repo); err != nil {
330 return err
331 }
332 }
333 _, err := c.db.Apply(ctx, &b)
334 if errors.Is(err, kv.ErrConflict) {
335 return domain.ErrConflict
336 }
337 if err != nil {
338 return fmt.Errorf("write the artifacts of %q: %w", w.Repo, err)
339 }
340 return nil
341}
342
343// artifactKeys lists every key of the snapshot that belongs to the artifacts
344// of the repository: records, their id index and blocks, and the usage record.
345func artifactKeys(ctx context.Context, snap *kv.Snapshot, repo domain.RepoName) [][]byte {
346 keys := [][]byte{artifactUseKey(repo)}
347 for key, item := range snap.Prefix(ctx, []byte(artifactPrefix+repo.String()+"/")) {
348 keys = append(keys, slices.Clone(key))
349 if a, err := decodeArtifact(item); err == nil {
350 keys = append(keys, artifactIDKey(a.ID))
351 for k := range snap.Prefix(ctx, []byte(artifactBlocksPrefix(a.ID))) {
352 keys = append(keys, slices.Clone(k))
353 }
354 }
355 }
356 return keys
357}
358
359// DeleteBlocks removes artifactblk/<id>/.
360func (c *CIArtifacts) DeleteBlocks(ctx context.Context, id int64) error {
361 var b kv.Batch
362 for k := range c.db.Snapshot(ctx).Prefix(ctx, []byte(artifactBlocksPrefix(id))) {
363 b.Delete(slices.Clone(k))
364 }
365 if b.Len() == 0 {
366 return nil
367 }
368 if _, err := c.db.Apply(ctx, &b); err != nil {
369 return fmt.Errorf("delete the blocks of artifact %d: %w", id, err)
370 }
371 return nil
372}