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