1// SPDX-License-Identifier: AGPL-3.0-or-later
2package kvstore
3
4import (
5 kv "bnema.dev/ermite-kv"
6 storagepb "bnema.dev/ermite/gen/ermite/storage"
7 "bnema.dev/ermite/internal/domain"
8 "bnema.dev/ermite/internal/ports"
9 "context"
10 "errors"
11 "fmt"
12 "github.com/bnema/zerowrap"
13 "google.golang.org/protobuf/proto"
14 "slices"
15 "strconv"
16)
17
18// Releases implements ports.ReleaseStore over the shared metadata database.
19type Releases struct{ db *kv.DB }
20
21var _ ports.ReleaseStore = (*Releases)(nil)
22
23// Releases returns the release metadata store.
24func (s *Store) Releases() *Releases { return &Releases{db: s.db} }
25func relKey(repo domain.RepoName, id int64) []byte {
26 return []byte("rel/" + repo.String() + "/" + fmt.Sprintf(idFormat, id))
27}
28func relIDKey(id int64) []byte { return []byte("relid/" + fmt.Sprintf(idFormat, id)) }
29func relTagKey(repo domain.RepoName, tag string) []byte {
30 return []byte("reltag/" + repo.String() + "/" + tag)
31}
32func relAssetKey(id int64) []byte { return []byte("relasset/" + fmt.Sprintf(idFormat, id)) }
33func relNamePrefix(repo domain.RepoName, id int64) string {
34 return "relassets/" + repo.String() + "/" + fmt.Sprintf(idFormat, id) + "/"
35}
36func relNameKey(a domain.ReleaseAsset) []byte {
37 return []byte(relNamePrefix(a.Repo, a.ReleaseID) + a.Name)
38}
39func relLatestKey(repo domain.RepoName) []byte { return []byte("rellatest/" + repo.String()) }
40func relUseKey(repo domain.RepoName) []byte { return []byte("reluse/" + repo.String()) }
41
42// GetLatestRelease resolves the index and release from the same snapshot.
43func (c *Releases) GetLatestRelease(ctx context.Context, repo domain.RepoName) (domain.Release, error) {
44 snap := c.db.Snapshot(ctx)
45 item, ok := snap.Get(ctx, relLatestKey(repo))
46 if !ok {
47 return domain.Release{}, domain.ErrNotFound
48 }
49 id, err := strconv.ParseInt(string(item.Value), 10, 64)
50 if err != nil {
51 return domain.Release{}, err
52 }
53 item, ok = snap.Get(ctx, relKey(repo, id))
54 if !ok {
55 return domain.Release{}, domain.ErrNotFound
56 }
57 return decodeRelease(item)
58}
59func encodeRelease(r domain.Release) []byte {
60 return mustMarshal(&storagepb.Release{
61 Id: r.ID,
62 Repo: r.Repo.String(),
63 Tag: r.Tag,
64 Title: r.Title,
65 Notes: r.Notes,
66 Draft: r.Draft,
67 Prerelease: r.Prerelease,
68 MakeLatest: r.MakeLatest,
69 TargetCommitish: r.TargetCommitish,
70 Author: r.Author,
71 CreatedAt: encodeTime(r.CreatedAt),
72 PublishedAt: encodeTime(r.PublishedAt),
73 UpdatedAt: encodeTime(r.UpdatedAt),
74 })
75}
76func decodeRelease(item kv.Item) (domain.Release, error) {
77 var pb storagepb.Release
78 if err := proto.Unmarshal(item.Value, &pb); err != nil {
79 return domain.Release{}, err
80 }
81 repo, err := domain.ParseRepoName(pb.GetRepo())
82 if err != nil {
83 return domain.Release{}, err
84 }
85 return domain.Release{Version: item.Version,
86 ID: pb.GetId(),
87 Repo: repo,
88 Tag: pb.GetTag(),
89 Title: pb.GetTitle(),
90 Notes: pb.GetNotes(),
91 Draft: pb.GetDraft(),
92 Prerelease: pb.GetPrerelease(),
93 MakeLatest: pb.GetMakeLatest(),
94 TargetCommitish: pb.GetTargetCommitish(),
95 Author: pb.GetAuthor(),
96 CreatedAt: decodeTime(pb.GetCreatedAt()),
97 PublishedAt: decodeTime(pb.GetPublishedAt()),
98 UpdatedAt: decodeTime(pb.GetUpdatedAt()),
99 }, nil
100}
101func encodeReleaseAsset(r domain.ReleaseAsset) []byte {
102 return mustMarshal(&storagepb.ReleaseAsset{
103 Id: r.ID,
104 ReleaseId: r.ReleaseID,
105 Repo: r.Repo.String(),
106 Name: r.Name,
107 Label: r.Label,
108 ContentType: r.ContentType,
109 Size: r.Size,
110 Digest: r.Digest,
111 State: string(r.State),
112 Uploader: r.Uploader,
113 Charged: r.Charged,
114 MultipartId: r.MultipartID,
115 CreatedAt: encodeTime(r.CreatedAt),
116 UpdatedAt: encodeTime(r.UpdatedAt),
117 })
118}
119func decodeReleaseAsset(item kv.Item) (domain.ReleaseAsset, error) {
120 var pb storagepb.ReleaseAsset
121 if err := proto.Unmarshal(item.Value, &pb); err != nil {
122 return domain.ReleaseAsset{}, err
123 }
124 repo, err := domain.ParseRepoName(pb.GetRepo())
125 if err != nil {
126 return domain.ReleaseAsset{}, err
127 }
128 return domain.ReleaseAsset{Version: item.Version,
129 ID: pb.GetId(),
130 ReleaseID: pb.GetReleaseId(),
131 Repo: repo,
132 Name: pb.GetName(),
133 Label: pb.GetLabel(),
134 ContentType: pb.GetContentType(),
135 Size: pb.GetSize(),
136 Digest: pb.GetDigest(),
137 State: domain.ReleaseAssetState(pb.GetState()),
138 Uploader: pb.GetUploader(),
139 Charged: pb.GetCharged(),
140 MultipartID: pb.GetMultipartId(),
141 CreatedAt: decodeTime(pb.GetCreatedAt()),
142 UpdatedAt: decodeTime(pb.GetUpdatedAt()),
143 }, nil
144}
145
146// NextReleaseID allocates a global monotonically increasing id.
147func (c *Releases) NextReleaseID(ctx context.Context) (int64, error) {
148 for range artifactIDTries {
149 last, item, ok, err := (&CI{db: c.db}).readSeq(ctx, []byte("relseq"))
150 if err != nil {
151 return 0, err
152 }
153 var b kv.Batch
154 stageSeq(&b, []byte("relseq"), 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, err
161 }
162 return last + 1, nil
163 }
164 return 0, domain.ErrConflict
165}
166
167// ReleaseUsage reads the repository's quota gate.
168func (c *Releases) ReleaseUsage(ctx context.Context, repo domain.RepoName) (int64, uint64, error) {
169 item, ok := c.db.Get(ctx, relUseKey(repo))
170 if !ok {
171 return 0, 0, nil
172 }
173 var pb storagepb.ReleaseUsage
174 if err := proto.Unmarshal(item.Value, &pb); err != nil {
175 return 0, 0, err
176 }
177 return pb.GetUsed(), item.Version, nil
178}
179
180// GetRelease reads a release in its repository.
181func (c *Releases) GetRelease(ctx context.Context, repo domain.RepoName, id int64) (domain.Release, error) {
182 item, ok := c.db.Get(ctx, relKey(repo, id))
183 if !ok {
184 return domain.Release{}, domain.ErrNotFound
185 }
186 return decodeRelease(item)
187}
188
189// ReleaseRepo resolves the global id index.
190func (c *Releases) ReleaseRepo(ctx context.Context, id int64) (domain.RepoName, error) {
191 item, ok := c.db.Get(ctx, relIDKey(id))
192 if !ok {
193 return domain.RepoName{}, domain.ErrNotFound
194 }
195 return domain.ParseRepoName(string(item.Value))
196}
197
198// GetReleaseByTag resolves a repository tag.
199func (c *Releases) GetReleaseByTag(ctx context.Context, repo domain.RepoName, tag string) (domain.Release, error) {
200 snap := c.db.Snapshot(ctx)
201 idx, ok := snap.Get(ctx, relTagKey(repo, tag))
202 if !ok {
203 return domain.Release{}, domain.ErrNotFound
204 }
205 id, err := strconv.ParseInt(string(idx.Value), 10, 64)
206 if err != nil {
207 return domain.Release{}, err
208 }
209 item, ok := snap.Get(ctx, relKey(repo, id))
210 if !ok {
211 return domain.Release{}, domain.ErrNotFound
212 }
213 return decodeRelease(item)
214}
215
216// ListReleases returns all releases by descending id.
217func (c *Releases) ListReleases(ctx context.Context, repo domain.RepoName) ([]domain.Release, error) {
218 var out []domain.Release
219 for _, item := range c.db.Snapshot(ctx).Prefix(ctx, []byte("rel/"+repo.String()+"/")) {
220 r, err := decodeRelease(item)
221 if err != nil {
222 return nil, err
223 }
224 out = append(out, r)
225 }
226 slices.Reverse(out)
227 return out, nil
228}
229
230// GetAsset reads asset metadata in every state.
231func (c *Releases) GetAsset(ctx context.Context, id int64) (domain.ReleaseAsset, error) {
232 item, ok := c.db.Get(ctx, relAssetKey(id))
233 if !ok {
234 return domain.ReleaseAsset{}, domain.ErrNotFound
235 }
236 return decodeReleaseAsset(item)
237}
238
239// GetAssetByName resolves the unique name index.
240func (c *Releases) GetAssetByName(ctx context.Context, repo domain.RepoName, releaseID int64, name string) (domain.ReleaseAsset, error) {
241 snap := c.db.Snapshot(ctx)
242 idx, ok := snap.Get(ctx, relNameKey(domain.ReleaseAsset{Repo: repo, ReleaseID: releaseID, Name: name}))
243 if !ok {
244 return domain.ReleaseAsset{}, domain.ErrNotFound
245 }
246 id, err := strconv.ParseInt(string(idx.Value), 10, 64)
247 if err != nil {
248 return domain.ReleaseAsset{}, err
249 }
250 item, ok := snap.Get(ctx, relAssetKey(id))
251 if !ok {
252 return domain.ReleaseAsset{}, domain.ErrNotFound
253 }
254 return decodeReleaseAsset(item)
255}
256
257// ListAssets returns every state in name order.
258func (c *Releases) ListAssets(ctx context.Context, repo domain.RepoName, releaseID int64) ([]domain.ReleaseAsset, error) {
259 snap := c.db.Snapshot(ctx)
260 var out []domain.ReleaseAsset
261 for _, idx := range snap.Prefix(ctx, []byte(relNamePrefix(repo, releaseID))) {
262 id, err := strconv.ParseInt(string(idx.Value), 10, 64)
263 if err != nil {
264 return nil, err
265 }
266 item, ok := snap.Get(ctx, relAssetKey(id))
267 if !ok {
268 return nil, domain.ErrNotFound
269 }
270 a, err := decodeReleaseAsset(item)
271 if err != nil {
272 return nil, err
273 }
274 out = append(out, a)
275 }
276 return out, nil
277}
278
279// AllAssets returns every asset for garbage collection.
280func (c *Releases) AllAssets(ctx context.Context) ([]domain.ReleaseAsset, error) {
281 var out []domain.ReleaseAsset
282 for _, item := range c.db.Snapshot(ctx).Prefix(ctx, []byte("relasset/")) {
283 a, err := decodeReleaseAsset(item)
284 if err != nil {
285 return nil, err
286 }
287 out = append(out, a)
288 }
289 return out, nil
290}
291
292// ApplyReleases atomically changes metadata and its quota gate.
293//
294//hexcheck:ignore suspicious-business-logic-in-adapter enforces storage version preconditions and unique secondary indexes, not release policy
295func (c *Releases) ApplyReleases(ctx context.Context, w ports.ReleaseWrite) error {
296 log := zerowrap.FromCtx(ctx)
297 var b kv.Batch
298 if err := guardRepoKey(ctx, c.db, &b, w.Repo); err != nil {
299 return err
300 }
301 if w.UsageVersion == 0 {
302 b.RequireAbsent(relUseKey(w.Repo))
303 } else {
304 b.RequireVersion(relUseKey(w.Repo), w.UsageVersion)
305 }
306 b.Put(relUseKey(w.Repo), mustMarshal(&storagepb.ReleaseUsage{Used: w.Used}))
307 if w.SetLatest {
308 if w.LatestID > 0 {
309 b.Put(relLatestKey(w.Repo), []byte(strconv.FormatInt(w.LatestID, 10)))
310 } else {
311 b.Delete(relLatestKey(w.Repo))
312 }
313 }
314 snap := c.db.Snapshot(ctx)
315 for _, r := range w.DeleteReleases {
316 if r.Repo != w.Repo || r.Version == 0 {
317 return domain.ErrInvalid
318 }
319 b.RequireVersion(relKey(r.Repo, r.ID), r.Version)
320 b.Delete(relKey(r.Repo, r.ID))
321 b.Delete(relTagKey(r.Repo, r.Tag))
322 b.Delete(relIDKey(r.ID))
323 }
324 for _, r := range w.DeleteAssets {
325 if r.Repo != w.Repo || r.Version == 0 {
326 return domain.ErrInvalid
327 }
328 b.RequireVersion(relAssetKey(r.ID), r.Version)
329 b.Delete(relAssetKey(r.ID))
330 b.Delete(relNameKey(r))
331 }
332 seenReleases := map[string]int64{}
333 for _, r := range w.PutReleases {
334 if r.Repo != w.Repo || r.ID < 1 {
335 return domain.ErrInvalid
336 }
337 idxKey := relTagKey(r.Repo, r.Tag)
338 if id, ok := seenReleases[string(idxKey)]; ok && id != r.ID {
339 return domain.ErrAlreadyExists
340 }
341 seenReleases[string(idxKey)] = r.ID
342 if item, ok := snap.Get(ctx, idxKey); ok && string(item.Value) != fmt.Sprintf(idFormat, r.ID) {
343 removed := false
344 for _, d := range w.DeleteReleases {
345 if string(relTagKey(d.Repo, d.Tag)) == string(idxKey) {
346 removed = true
347 }
348 }
349 if !removed {
350 return domain.ErrAlreadyExists
351 }
352 }
353 if old, ok := snap.Get(ctx, relKey(r.Repo, r.ID)); ok {
354 previous, err := decodeRelease(old)
355 if err != nil {
356 return err
357 }
358 oldKey := relTagKey(previous.Repo, previous.Tag)
359 if string(oldKey) != string(idxKey) {
360 b.Delete(oldKey)
361 }
362 }
363 stage(&b, relKey(r.Repo, r.ID), encodeRelease(r), r.Version)
364 b.Put(idxKey, []byte(fmt.Sprintf(idFormat, r.ID)))
365 if r.Version == 0 {
366 b.RequireAbsent(relIDKey(r.ID))
367 }
368 b.Put(relIDKey(r.ID), []byte(r.Repo.String()))
369 }
370 seenAssets := map[string]int64{}
371 for _, r := range w.PutAssets {
372 if r.Repo != w.Repo || r.ID < 1 {
373 return domain.ErrInvalid
374 }
375 idxKey := relNameKey(r)
376 if id, ok := seenAssets[string(idxKey)]; ok && id != r.ID {
377 return domain.ErrAlreadyExists
378 }
379 seenAssets[string(idxKey)] = r.ID
380 if item, ok := snap.Get(ctx, idxKey); ok && string(item.Value) != fmt.Sprintf(idFormat, r.ID) {
381 removed := false
382 for _, d := range w.DeleteAssets {
383 if string(relNameKey(d)) == string(idxKey) {
384 removed = true
385 }
386 }
387 if !removed {
388 return domain.ErrAlreadyExists
389 }
390 }
391 if old, ok := snap.Get(ctx, relAssetKey(r.ID)); ok {
392 previous, err := decodeReleaseAsset(old)
393 if err != nil {
394 return err
395 }
396 oldKey := relNameKey(previous)
397 if string(oldKey) != string(idxKey) {
398 b.Delete(oldKey)
399 }
400 }
401 stage(&b, relAssetKey(r.ID), encodeReleaseAsset(r), r.Version)
402 b.Put(idxKey, []byte(fmt.Sprintf(idFormat, r.ID)))
403 }
404 _, err := c.db.Apply(ctx, &b)
405 if errors.Is(err, kv.ErrConflict) {
406 return domain.ErrConflict
407 }
408 if err != nil {
409 return fmt.Errorf("write releases: %w", err)
410 }
411 log.Debug().Str("repo", w.Repo.String()).Msg("release metadata updated")
412 return nil
413}
414
415// releaseKeys collects all repository-owned release keys, including global indexes.
416func releaseKeys(ctx context.Context, snap *kv.Snapshot, repo domain.RepoName) [][]byte {
417 keys := [][]byte{relUseKey(repo), relLatestKey(repo)}
418 for _, prefix := range []string{"rel/", "reltag/", "relassets/"} {
419 for key, item := range snap.Prefix(ctx, []byte(prefix+repo.String()+"/")) {
420 keys = append(keys, slices.Clone(key))
421 if prefix == "rel/" {
422 if r, err := decodeRelease(item); err == nil {
423 keys = append(keys, relIDKey(r.ID))
424 }
425 }
426 }
427 }
428 for key, item := range snap.Prefix(ctx, []byte("relasset/")) {
429 if a, err := decodeReleaseAsset(item); err == nil && a.Repo == repo {
430 keys = append(keys, slices.Clone(key))
431 }
432 }
433 return keys
434}