1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package usecase
4
5import (
6 "bytes"
7 "context"
8 "crypto/sha256"
9 "errors"
10 "fmt"
11 "io"
12 "slices"
13 "strings"
14 "time"
15
16 "bnema.dev/ermite/internal/domain"
17 "bnema.dev/ermite/internal/ports"
18)
19
20// UploadAsset reserves quota before reading, stores bytes, then publishes metadata.
21func (r *Releases) UploadAsset(ctx context.Context, p domain.Principal, name domain.RepoName, params ports.UploadAssetParams, body io.Reader) (result domain.ReleaseAsset, err error) {
22 repo, err := r.writable(ctx, p, name)
23 if err != nil {
24 return result, err
25 }
26 if params.Size < 0 || params.Size > r.limits.AssetMax {
27 return result, fmt.Errorf("%w: asset size must be between zero and %s", domain.ErrResourceExhausted, sizeText(r.limits.AssetMax))
28 }
29 a := domain.ReleaseAsset{Repo: name, ReleaseID: params.ReleaseID, Size: params.Size, Charged: params.Size, State: domain.ReleaseAssetUploading, Uploader: domain.AuthorName(p), CreatedAt: r.clock.Now(), UpdatedAt: r.clock.Now()}
30 if a.Name, err = domain.ParseReleaseAssetName(params.Name); err != nil {
31 return result, err
32 }
33 if a.Label, err = domain.ParseReleaseAssetLabel(params.Label); err != nil {
34 return result, err
35 }
36 if a.ContentType, err = domain.ParseReleaseContentType(params.ContentType); err != nil {
37 return result, err
38 }
39 select {
40 case r.uploads <- struct{}{}:
41 defer func() { <-r.uploads }()
42 default:
43 return result, fmt.Errorf("%w: too many uploads in progress, retry", domain.ErrResourceExhausted)
44 }
45 a.ID, err = r.store.NextReleaseID(ctx)
46 if err != nil {
47 return result, err
48 }
49 var replaced []domain.ReleaseAsset
50 err = r.updateReleases(ctx, name, func(w *ports.ReleaseWrite) error {
51 replaced = nil
52 if _, err := r.store.GetRelease(ctx, name, a.ReleaseID); err != nil {
53 if errors.Is(err, domain.ErrNotFound) {
54 owner, lookupErr := r.store.ReleaseRepo(ctx, a.ReleaseID)
55 if lookupErr == nil && owner != name {
56 return fmt.Errorf("%w: release belongs to another repository", domain.ErrPreconditionFailed)
57 }
58 }
59 return err
60 }
61 all, err := r.store.ListAssets(ctx, name, a.ReleaseID)
62 if err != nil {
63 return err
64 }
65 for _, old := range all {
66 if old.Name != a.Name {
67 continue
68 }
69 if !(old.State == domain.ReleaseAssetUploading && r.clock.Now().Sub(old.UpdatedAt) > 60*time.Second) && !(old.State == domain.ReleaseAssetUploaded && params.Replace) {
70 return fmt.Errorf("%w: an asset named %q already exists", domain.ErrAlreadyExists, a.Name)
71 }
72 replaced = append(replaced, old)
73 w.DeleteAssets = append(w.DeleteAssets, old)
74 w.Used -= old.Charged
75 }
76 if int64(len(all)-len(replaced)) >= r.limits.AssetsMax {
77 return fmt.Errorf("%w: release asset count limit is %d", domain.ErrResourceExhausted, r.limits.AssetsMax)
78 }
79 if a.Size > r.limits.Quota-w.Used {
80 return fmt.Errorf("%w: release quota is %s", domain.ErrResourceExhausted, sizeText(r.limits.Quota))
81 }
82 w.Used += a.Charged
83 w.PutAssets = []domain.ReleaseAsset{a}
84 return nil
85 })
86 if err != nil {
87 return result, err
88 }
89 r.removeAssetBlobs(context.WithoutCancel(ctx), replaced)
90 body = releaseContextReader{ctx: ctx, reader: body}
91 // From here every failure removes the reservation and any completed blob.
92 defer func() {
93 if err != nil {
94 clean := context.WithoutCancel(ctx)
95 if e := r.removeAssetRecord(clean, a, nil); e != nil && !errors.Is(e, domain.ErrNotFound) {
96 ctxLog(ctx).Error().Err(e).Int64("asset", a.ID).Msg("refund failed release upload")
97 }
98 r.removeAssetBlobs(clean, []domain.ReleaseAsset{a})
99 }
100 }()
101 if a.Size <= ports.BlobPartSize {
102 var info ports.BlobInfo
103 limited := &io.LimitedReader{R: body, N: a.Size}
104 info, err = r.blobs.Put(ctx, a.BlobKey(), limited, a.Size)
105 if err != nil {
106 if limited.N > 0 && errors.Is(err, domain.ErrInvalid) {
107 err = fmt.Errorf("%w: body shorter than size", domain.ErrInvalid)
108 }
109 return result, err
110 }
111 a.Digest = info.Digest
112 if !validReleaseDigest(a.Digest) {
113 return result, fmt.Errorf("%w: blob store returned no SHA-256 digest", domain.ErrInvalid)
114 }
115 } else {
116 var upload ports.MultipartUpload
117 upload, err = r.blobs.NewMultipart(ctx, a.BlobKey())
118 if err != nil {
119 return result, err
120 }
121 completed := false
122 defer func() {
123 if !completed {
124 if e := upload.Abort(context.WithoutCancel(ctx)); e != nil {
125 ctxLog(ctx).Error().Err(e).Msg("abort failed release upload")
126 }
127 }
128 }()
129 a.MultipartID = upload.ID()
130 err = r.changeAsset(ctx, a.ID, name, func(current *domain.ReleaseAsset) error { current.MultipartID = a.MultipartID; return nil })
131 if err != nil {
132 return result, err
133 }
134 hash := sha256.New()
135 buf := make([]byte, ports.BlobPartSize)
136 remaining := a.Size
137 for part := 1; remaining > 0; part++ {
138 if part > ports.BlobMaxParts {
139 return result, domain.ErrResourceExhausted
140 }
141 want := min(remaining, int64(len(buf)))
142 var n int
143 n, err = io.ReadFull(body, buf[:want])
144 if err != nil {
145 if errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF) {
146 err = fmt.Errorf("%w: body shorter than size", domain.ErrInvalid)
147 }
148 return result, err
149 }
150 hash.Write(buf[:n])
151 err = upload.UploadPart(ctx, part, bytes.NewReader(buf[:n]), int64(n))
152 if err != nil {
153 return result, err
154 }
155 remaining -= int64(n)
156 }
157 if err = releaseBodyEnd(body); err != nil {
158 return result, err
159 }
160 _, err = upload.Complete(ctx)
161 if err != nil {
162 return result, err
163 }
164 completed = true
165 a.Digest = fmt.Sprintf("sha256:%x", hash.Sum(nil))
166 a.MultipartID = ""
167 }
168 if a.Size <= ports.BlobPartSize {
169 if err = releaseBodyEnd(body); err != nil {
170 return result, err
171 }
172 }
173 err = r.changeAsset(ctx, a.ID, name, func(current *domain.ReleaseAsset) error {
174 if current.State != domain.ReleaseAssetUploading {
175 return domain.ErrConflict
176 }
177 current.State = domain.ReleaseAssetUploaded
178 current.Digest = a.Digest
179 current.MultipartID = ""
180 current.UpdatedAt = r.clock.Now()
181 return nil
182 })
183 if err != nil {
184 return result, err
185 }
186 result, err = r.store.GetAsset(ctx, a.ID)
187 if err != nil {
188 return result, err
189 }
190 rel, e := r.store.GetRelease(ctx, name, a.ReleaseID)
191 if e == nil {
192 r.emit(ctx, repo, rel, "updated")
193 }
194 ctxLog(ctx).Info().Int64("release", a.ReleaseID).Int64("asset", a.ID).Int64("size", a.Size).Msg("release asset uploaded")
195 return result, nil
196}
197func validReleaseDigest(s string) bool {
198 if len(s) != 71 || s[:7] != "sha256:" {
199 return false
200 }
201 for _, c := range s[7:] {
202 if !(c >= '0' && c <= '9' || c >= 'a' && c <= 'f') {
203 return false
204 }
205 }
206 return true
207}
208
209// releaseContextReader checks cancellation between reads, including multipart
210// buffering and the final extra-byte check. The transport owns interrupting a
211// body.Read already in progress.
212type releaseContextReader struct {
213 ctx context.Context
214 reader io.Reader
215}
216
217func (r releaseContextReader) Read(p []byte) (int, error) {
218 if err := r.ctx.Err(); err != nil {
219 return 0, err
220 }
221 return r.reader.Read(p)
222}
223
224func releaseBodyEnd(body io.Reader) error {
225 var extra [1]byte
226 n, err := io.ReadFull(body, extra[:])
227 if n > 0 {
228 return fmt.Errorf("%w: body longer than size", domain.ErrInvalid)
229 }
230 if errors.Is(err, io.EOF) {
231 return nil
232 }
233 return err
234}
235func (r *Releases) changeAsset(ctx context.Context, id int64, repo domain.RepoName, fn func(*domain.ReleaseAsset) error) error {
236 return r.updateReleases(ctx, repo, func(w *ports.ReleaseWrite) error {
237 a, err := r.store.GetAsset(ctx, id)
238 if err != nil {
239 return err
240 }
241 if a.Repo != repo {
242 return domain.ErrNotFound
243 }
244 if err = fn(&a); err != nil {
245 return err
246 }
247 w.PutAssets = []domain.ReleaseAsset{a}
248 return nil
249 })
250}
251
252// removeAssetRecord conditionally removes the current record; sweep predicates
253// are evaluated again after a conflict so a just-completed upload is protected.
254func (r *Releases) removeAssetRecord(ctx context.Context, a domain.ReleaseAsset, predicate func(domain.ReleaseAsset) bool) error {
255 return r.updateReleases(ctx, a.Repo, func(w *ports.ReleaseWrite) error {
256 current, err := r.store.GetAsset(ctx, a.ID)
257 if err != nil {
258 return err
259 }
260 if current.Repo != a.Repo {
261 return domain.ErrNotFound
262 }
263 if predicate != nil && !predicate(current) {
264 return domain.ErrConflict
265 }
266 w.DeleteAssets = []domain.ReleaseAsset{current}
267 w.Used -= current.Charged
268 return nil
269 })
270}
271
272// ListAssets returns completed assets only, ordered by name.
273func (r *Releases) ListAssets(ctx context.Context, p domain.Principal, repo domain.RepoName, id int64) ([]domain.ReleaseAsset, error) {
274 if _, err := r.GetRelease(ctx, p, repo, ports.ReleaseRef{ID: id}); err != nil {
275 return nil, err
276 }
277 all, err := r.store.ListAssets(ctx, repo, id)
278 if err != nil {
279 return nil, err
280 }
281 return slices.DeleteFunc(all, func(a domain.ReleaseAsset) bool { return a.State != domain.ReleaseAssetUploaded }), nil
282}
283
284// GetAsset resolves an uploaded asset, enforcing its release's draft rule.
285func (r *Releases) GetAsset(ctx context.Context, p domain.Principal, repo domain.RepoName, ref ports.AssetRef) (domain.ReleaseAsset, error) {
286 var a domain.ReleaseAsset
287 var err error
288 if ref.ID != 0 {
289 if ref.Tag != "" || ref.Latest || ref.Name != "" {
290 return a, domain.ErrInvalid
291 }
292 if _, err = r.readable(ctx, p, repo); err != nil {
293 return a, err
294 }
295 a, err = r.store.GetAsset(ctx, ref.ID)
296 if err != nil {
297 return a, err
298 }
299 if a.Repo != repo {
300 return domain.ReleaseAsset{}, domain.ErrNotFound
301 }
302 _, err = r.GetRelease(ctx, p, repo, ports.ReleaseRef{ID: a.ReleaseID})
303 } else {
304 if ref.Name == "" {
305 return a, domain.ErrInvalid
306 }
307 var rel domain.Release
308 rel, err = r.GetRelease(ctx, p, repo, ports.ReleaseRef{Tag: ref.Tag, Latest: ref.Latest})
309 if err != nil {
310 return a, err
311 }
312 a, err = r.store.GetAssetByName(ctx, repo, rel.ID, ref.Name)
313 }
314 if err == nil && a.State != domain.ReleaseAssetUploaded {
315 err = domain.ErrNotFound
316 }
317 return a, err
318}
319
320// UpdateAsset changes an uploaded asset's name and label without moving bytes.
321func (r *Releases) UpdateAsset(ctx context.Context, p domain.Principal, repo domain.RepoName, id int64, params ports.UpdateAssetParams) (domain.ReleaseAsset, error) {
322 meta, err := r.writable(ctx, p, repo)
323 if err != nil {
324 return domain.ReleaseAsset{}, err
325 }
326 a, err := r.GetAsset(ctx, p, repo, ports.AssetRef{ID: id})
327 if err != nil {
328 return a, err
329 }
330 var newName, newLabel string
331 if params.Name != nil {
332 newName, err = domain.ParseReleaseAssetName(*params.Name)
333 if err != nil {
334 return a, err
335 }
336 }
337 if params.Label != nil {
338 newLabel, err = domain.ParseReleaseAssetLabel(*params.Label)
339 if err != nil {
340 return a, err
341 }
342 }
343 err = r.changeAsset(ctx, id, repo, func(current *domain.ReleaseAsset) error {
344 if current.State != domain.ReleaseAssetUploaded {
345 return domain.ErrNotFound
346 }
347 if params.Name != nil {
348 current.Name = newName
349 }
350 if params.Label != nil {
351 current.Label = newLabel
352 }
353 current.UpdatedAt = r.clock.Now()
354 return nil
355 })
356 if err != nil {
357 return domain.ReleaseAsset{}, err
358 }
359 a, err = r.store.GetAsset(ctx, id)
360 if err != nil {
361 return a, err
362 }
363 if rel, e := r.store.GetRelease(ctx, repo, a.ReleaseID); e == nil {
364 r.emit(ctx, meta, rel, "updated")
365 }
366 return a, nil
367}
368
369// DeleteAsset refunds quota before deleting the blob.
370func (r *Releases) DeleteAsset(ctx context.Context, p domain.Principal, repo domain.RepoName, id int64) error {
371 meta, err := r.writable(ctx, p, repo)
372 if err != nil {
373 return err
374 }
375 a, err := r.GetAsset(ctx, p, repo, ports.AssetRef{ID: id})
376 if err != nil {
377 return err
378 }
379 if err = r.removeAssetRecord(ctx, a, nil); err != nil {
380 return err
381 }
382 r.removeAssetBlobs(context.WithoutCancel(ctx), []domain.ReleaseAsset{a})
383 if rel, e := r.store.GetRelease(ctx, repo, a.ReleaseID); e == nil {
384 r.emit(ctx, meta, rel, "updated")
385 }
386 return nil
387}
388
389// DownloadAsset sends metadata followed by bounded chunks of the requested range.
390func (r *Releases) DownloadAsset(ctx context.Context, p domain.Principal, repo domain.RepoName, ref ports.AssetRef, offset, length int64, send func(ports.AssetChunk) error) error {
391 a, err := r.GetAsset(ctx, p, repo, ref)
392 if err != nil {
393 return err
394 }
395 var body io.ReadCloser
396 if a.Size == 0 && offset == 0 {
397 // Blob stores reject Open on an empty object: stat it instead.
398 if _, err = r.blobs.Stat(ctx, a.BlobKey()); err == nil {
399 body = io.NopCloser(strings.NewReader(""))
400 }
401 } else {
402 body, _, err = r.blobs.Open(ctx, a.BlobKey(), offset, length)
403 }
404 if err != nil {
405 if errors.Is(err, domain.ErrNotFound) {
406 ctxLog(ctx).Error().Int64("asset", a.ID).Msg("asset blob missing")
407 }
408 return err
409 }
410 defer body.Close()
411 if err = send(ports.AssetChunk{Asset: &a}); err != nil {
412 return err
413 }
414 buf := make([]byte, domain.CIArtifactChunk)
415 for {
416 if err = ctx.Err(); err != nil {
417 return err
418 }
419 n, e := body.Read(buf)
420 if n > 0 {
421 if err = send(ports.AssetChunk{Data: buf[:n]}); err != nil {
422 return err
423 }
424 }
425 if errors.Is(e, io.EOF) {
426 return nil
427 }
428 if e != nil {
429 return e
430 }
431 }
432}
433
434// Usage reports quota and configured limits to a repository reader.
435func (r *Releases) Usage(ctx context.Context, p domain.Principal, repo domain.RepoName) (ports.ReleaseUsage, error) {
436 if _, err := r.readable(ctx, p, repo); err != nil {
437 return ports.ReleaseUsage{}, err
438 }
439 used, _, err := r.store.ReleaseUsage(ctx, repo)
440 return ports.ReleaseUsage{Used: used, Limits: r.limits}, err
441}