1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package usecase
4
5import (
6 "context"
7 "errors"
8 "fmt"
9 "slices"
10 "strconv"
11 "strings"
12 "sync"
13 "time"
14
15 "bnema.dev/ermite/internal/domain"
16 "bnema.dev/ermite/internal/ports"
17)
18
19// Releases manages release metadata and blob contents. Construct once per server.
20type Releases struct {
21 store ports.ReleaseStore
22 blobs ports.BlobStore
23 repos ports.RepoMetaStore
24 reader ports.RepoReader
25 tags ports.ReleaseTagWriter
26 pushes ports.PushHookService
27 bus ports.EventBus
28 clock ports.Clock
29 limits ports.ReleaseLimits
30 uploads chan struct{}
31 sweepMu sync.Mutex
32 misses map[int64]int
33}
34
35var _ ports.ReleaseService = (*Releases)(nil)
36var _ ports.ReleaseTagTrigger = (*Releases)(nil)
37var _ ports.ReleaseRepoCleaner = (*Releases)(nil)
38var _ ports.ReleaseSweeper = (*Releases)(nil)
39
40// NewReleases constructs the release service. Zero limits select the defaults.
41// pushes must be the shared Pushes instance, so created tags notify every hook.
42func NewReleases(store ports.ReleaseStore, blobs ports.BlobStore, repos ports.RepoMetaStore, reader ports.RepoReader, tags ports.ReleaseTagWriter, pushes ports.PushHookService, bus ports.EventBus, clock ports.Clock, limits ports.ReleaseLimits) *Releases {
43 if limits.AssetMax == 0 {
44 limits.AssetMax = domain.DefaultReleaseAssetMax
45 }
46 if limits.AssetsMax == 0 {
47 limits.AssetsMax = domain.DefaultReleaseAssetsMax
48 }
49 if limits.Quota == 0 {
50 limits.Quota = domain.DefaultReleaseQuota
51 }
52 return &Releases{store: store, blobs: blobs, repos: repos, reader: reader, tags: tags, pushes: pushes, bus: bus, clock: clock, limits: limits, uploads: make(chan struct{}, 16), misses: make(map[int64]int)}
53}
54
55func (r *Releases) readable(ctx context.Context, p domain.Principal, name domain.RepoName) (domain.Repo, error) {
56 repo, err := lookupRepo(ctx, r.repos, name)
57 if err == nil && !canRead(p, repo) {
58 err = domain.ErrNotFound
59 }
60 return repo, err
61}
62func releaseWriter(p domain.Principal, name domain.RepoName) bool {
63 return !p.IsBranchLimited() && (p.Allows(domain.ScopeReleaseWrite, name) || p.JobPermits(domain.PermContents, domain.PermissionWrite, name))
64}
65func (r *Releases) writable(ctx context.Context, p domain.Principal, name domain.RepoName) (domain.Repo, error) {
66 repo, err := r.readable(ctx, p, name)
67 if err != nil {
68 return repo, err
69 }
70 if p.IsBranchLimited() {
71 return repo, fmt.Errorf("%w: a branch-limited token cannot manage releases", domain.ErrForbidden)
72 }
73 if releaseWriter(p, name) {
74 return repo, nil
75 }
76 if p.IsAnonymous() {
77 return repo, domain.ErrUnauthenticated
78 }
79 return repo, domain.ErrForbidden
80}
81func (r *Releases) updateReleases(ctx context.Context, repo domain.RepoName, fn func(*ports.ReleaseWrite) error) error {
82 const releaseTries = 256
83 for range releaseTries {
84 if err := ctx.Err(); err != nil {
85 return err
86 }
87 used, v, err := r.store.ReleaseUsage(ctx, repo)
88 if err != nil {
89 return err
90 }
91 w := ports.ReleaseWrite{Repo: repo, Used: used, UsageVersion: v}
92 if err = fn(&w); err != nil {
93 return err
94 }
95 if len(w.PutReleases) > 0 || len(w.DeleteReleases) > 0 {
96 w.SetLatest = true
97 all, err := r.store.ListReleases(ctx, repo)
98 if err != nil {
99 return err
100 }
101 byID := make(map[int64]domain.Release, len(all)+len(w.PutReleases))
102 for _, rel := range all {
103 byID[rel.ID] = rel
104 }
105 for _, rel := range w.DeleteReleases {
106 delete(byID, rel.ID)
107 }
108 for _, rel := range w.PutReleases {
109 byID[rel.ID] = rel
110 }
111 all = all[:0]
112 for _, rel := range byID {
113 all = append(all, rel)
114 }
115 if latest, ok := domain.LatestRelease(all); ok {
116 w.LatestID = latest.ID
117 }
118 }
119 err = r.store.ApplyReleases(ctx, w)
120 if errors.Is(err, domain.ErrConflict) {
121 continue
122 }
123 return err
124 }
125 return fmt.Errorf("%w: releases of %s keep changing, try again", domain.ErrUnavailable, repo)
126}
127func (r *Releases) resolve(ctx context.Context, repo domain.RepoName, rev string) (domain.ObjectID, error) {
128 v, err := domain.ParseRevision(rev)
129 if err != nil {
130 return domain.ObjectID{}, err
131 }
132 got, err := r.reader.ResolveRevision(ctx, repo, v)
133 return got.CommitID, err
134}
135func (r *Releases) bind(ctx context.Context, p domain.Principal, rel *domain.Release, target string, create bool) error {
136 oid, err := r.resolve(ctx, rel.Repo, "refs/tags/"+rel.Tag)
137 if errors.Is(err, domain.ErrNotFound) {
138 if rel.Draft {
139 rel.TargetCommitish = ""
140 return nil
141 }
142 if !create || target == "" {
143 return fmt.Errorf("%w: tag %q does not exist; give target_commitish to create it", domain.ErrPreconditionFailed, rel.Tag)
144 }
145 if r.pushes == nil {
146 return fmt.Errorf("%w: release tag notifications are not configured", domain.ErrUnavailable)
147 }
148 oid, err = r.resolve(ctx, rel.Repo, target)
149 if err != nil {
150 return err
151 }
152 err = r.tags.CreateTag(ctx, rel.Repo, rel.Tag, oid)
153 if errors.Is(err, domain.ErrAlreadyExists) {
154 oid, err = r.resolve(ctx, rel.Repo, "refs/tags/"+rel.Tag)
155 } else if err == nil {
156 err = r.pushes.Record(ctx, p, rel.Repo, []domain.RefUpdate{{Ref: "refs/tags/" + rel.Tag, Old: domain.ZeroOID, New: oid.String()}})
157 }
158 }
159 if err != nil {
160 return err
161 }
162 rel.TargetCommitish = oid.String()
163 return nil
164}
165func (r *Releases) pin(ctx context.Context, w *ports.ReleaseWrite, rel domain.Release) error {
166 if !rel.MakeLatest {
167 return nil
168 }
169 all, err := r.store.ListReleases(ctx, rel.Repo)
170 if err != nil {
171 return err
172 }
173 for _, other := range all {
174 if other.ID != rel.ID && other.MakeLatest {
175 other.MakeLatest = false
176 w.PutReleases = append(w.PutReleases, other)
177 }
178 }
179 return nil
180}
181func (r *Releases) emit(ctx context.Context, repo domain.Repo, rel domain.Release, kind string) {
182 at := r.clock.Now()
183 var ev domain.Event
184 switch kind {
185 case "published":
186 ev = domain.ReleasePublished{Release: rel, At: at, Visibility: repo.Visibility, CreatedAt: repo.CreatedAt}
187 case "updated":
188 ev = domain.ReleaseUpdated{Release: rel, At: at, Visibility: repo.Visibility, CreatedAt: repo.CreatedAt}
189 case "deleted":
190 ev = domain.ReleaseDeleted{Release: rel, At: at, Visibility: repo.Visibility, CreatedAt: repo.CreatedAt}
191 case "unpublished":
192 ev = domain.ReleaseUnpublished{Release: rel, At: at, Visibility: repo.Visibility, CreatedAt: repo.CreatedAt}
193 }
194 r.bus.Publish(ctx, ev)
195}
196
197// CreateRelease creates a release, optionally creating its tag when publishing.
198func (r *Releases) CreateRelease(ctx context.Context, p domain.Principal, name domain.RepoName, params ports.CreateReleaseParams) (domain.Release, error) {
199 repo, err := r.writable(ctx, p, name)
200 if err != nil {
201 return domain.Release{}, err
202 }
203 rel := domain.Release{Repo: name, Draft: params.Draft, Prerelease: params.Prerelease, MakeLatest: params.MakeLatest, Author: domain.AuthorName(p), CreatedAt: r.clock.Now(), UpdatedAt: r.clock.Now()}
204 if rel.Tag, err = domain.ParseReleaseTag(params.Tag); err != nil {
205 return rel, err
206 }
207 if rel.Title, err = domain.ParseReleaseTitle(params.Title); err != nil {
208 return rel, err
209 }
210 if rel.Notes, err = domain.ParseReleaseNotes(params.Notes); err != nil {
211 return rel, err
212 }
213 if _, err = r.store.GetReleaseByTag(ctx, name, rel.Tag); err == nil {
214 return rel, domain.ErrAlreadyExists
215 } else if !errors.Is(err, domain.ErrNotFound) {
216 return rel, err
217 }
218 if err = r.bind(ctx, p, &rel, params.TargetCommitish, true); err != nil {
219 return rel, err
220 }
221 if rel.Draft {
222 rel.MakeLatest = false
223 } else {
224 rel.PublishedAt = r.clock.Now()
225 }
226 rel.ID, err = r.store.NextReleaseID(ctx)
227 if err != nil {
228 return rel, err
229 }
230 err = r.updateReleases(ctx, name, func(w *ports.ReleaseWrite) error {
231 if err := r.pin(ctx, w, rel); err != nil {
232 return err
233 }
234 w.PutReleases = append(w.PutReleases, rel)
235 return nil
236 })
237 if err != nil {
238 return domain.Release{}, err
239 }
240 if !rel.Draft {
241 r.emit(ctx, repo, rel, "published")
242 }
243 return r.store.GetRelease(ctx, name, rel.ID)
244}
245
246// GetRelease resolves a visible release by exactly one selector.
247func (r *Releases) GetRelease(ctx context.Context, p domain.Principal, name domain.RepoName, ref ports.ReleaseRef) (domain.Release, error) {
248 if _, err := r.readable(ctx, p, name); err != nil {
249 return domain.Release{}, err
250 }
251 n := 0
252 if ref.ID != 0 {
253 n++
254 }
255 if ref.Tag != "" {
256 n++
257 }
258 if ref.Latest {
259 n++
260 }
261 if n != 1 {
262 return domain.Release{}, domain.ErrInvalid
263 }
264 var rel domain.Release
265 var err error
266 switch {
267 case ref.ID != 0:
268 rel, err = r.store.GetRelease(ctx, name, ref.ID)
269 case ref.Tag != "":
270 rel, err = r.store.GetReleaseByTag(ctx, name, ref.Tag)
271 default:
272 rel, err = r.store.GetLatestRelease(ctx, name)
273 }
274 if err == nil && rel.Draft && !releaseWriter(p, name) {
275 err = domain.ErrNotFound
276 }
277 return rel, err
278}
279
280// ListReleases lists visible releases newest first, with an ID cursor.
281func (r *Releases) ListReleases(ctx context.Context, p domain.Principal, name domain.RepoName, size int, token string) (ports.ReleasePage, error) {
282 var out ports.ReleasePage
283 if _, err := r.readable(ctx, p, name); err != nil {
284 return out, err
285 }
286 if size == 0 {
287 size = 30
288 }
289 n, err := pageSize(size)
290 if err != nil {
291 return out, err
292 }
293 var cursor int64
294 if token != "" {
295 cursor, err = strconv.ParseInt(token, 10, 64)
296 if err != nil || cursor <= 0 {
297 return out, domain.ErrInvalidPageToken
298 }
299 }
300 all, err := r.store.ListReleases(ctx, name)
301 if err != nil {
302 return out, err
303 }
304 slices.SortFunc(all, func(a, b domain.Release) int {
305 if c := b.CreatedAt.Compare(a.CreatedAt); c != 0 {
306 return c
307 }
308 if a.ID > b.ID {
309 return -1
310 }
311 if a.ID < b.ID {
312 return 1
313 }
314 return 0
315 })
316 start := token == ""
317 writer := releaseWriter(p, name)
318 for _, rel := range all {
319 if rel.Draft && !writer {
320 continue // a hidden draft is not even a valid cursor
321 }
322 if !start {
323 if rel.ID == cursor {
324 start = true
325 }
326 continue
327 }
328 if len(out.Releases) == n {
329 out.NextPageToken = strconv.FormatInt(out.Releases[n-1].ID, 10)
330 break
331 }
332 out.Releases = append(out.Releases, rel)
333 }
334 if !start {
335 return out, domain.ErrInvalidPageToken
336 }
337 return out, nil
338}
339
340// UpdateRelease changes only fields whose pointers are non-nil.
341func (r *Releases) UpdateRelease(ctx context.Context, p domain.Principal, name domain.RepoName, id int64, params ports.UpdateReleaseParams) (domain.Release, error) {
342 repo, err := r.writable(ctx, p, name)
343 if err != nil {
344 return domain.Release{}, err
345 }
346 var rel domain.Release
347 kind := "updated"
348 err = r.updateReleases(ctx, name, func(w *ports.ReleaseWrite) error {
349 var err error
350 rel, err = r.store.GetRelease(ctx, name, id)
351 if err != nil {
352 return err
353 }
354 wasDraft := rel.Draft
355 if params.Tag != nil {
356 rel.Tag, err = domain.ParseReleaseTag(*params.Tag)
357 if err != nil {
358 return err
359 }
360 }
361 if params.Title != nil {
362 rel.Title, err = domain.ParseReleaseTitle(*params.Title)
363 if err != nil {
364 return err
365 }
366 }
367 if params.Notes != nil {
368 rel.Notes, err = domain.ParseReleaseNotes(*params.Notes)
369 if err != nil {
370 return err
371 }
372 }
373 if params.Draft != nil {
374 rel.Draft = *params.Draft
375 }
376 if params.Prerelease != nil {
377 rel.Prerelease = *params.Prerelease
378 }
379 if params.MakeLatest != nil {
380 rel.MakeLatest = *params.MakeLatest
381 }
382 target := ""
383 if params.TargetCommitish != nil {
384 target = *params.TargetCommitish
385 }
386 if err = r.bind(ctx, p, &rel, target, target != ""); err != nil {
387 return err
388 }
389 kind = "updated"
390 if rel.Draft {
391 rel.PublishedAt = time.Time{}
392 rel.MakeLatest = false
393 } else if wasDraft {
394 rel.PublishedAt = r.clock.Now()
395 kind = "published"
396 }
397 rel.UpdatedAt = r.clock.Now()
398 if err = r.pin(ctx, w, rel); err != nil {
399 return err
400 }
401 w.PutReleases = append(w.PutReleases, rel)
402 return nil
403 })
404 if err != nil {
405 return domain.Release{}, err
406 }
407 r.emit(ctx, repo, rel, kind)
408 return r.store.GetRelease(ctx, name, id)
409}
410
411// DeleteRelease removes metadata first, then its contents.
412func (r *Releases) DeleteRelease(ctx context.Context, p domain.Principal, name domain.RepoName, id int64) error {
413 repo, err := r.writable(ctx, p, name)
414 if err != nil {
415 return err
416 }
417 var rel domain.Release
418 var assets []domain.ReleaseAsset
419 err = r.updateReleases(ctx, name, func(w *ports.ReleaseWrite) error {
420 var err error
421 rel, err = r.store.GetRelease(ctx, name, id)
422 if err != nil {
423 return err
424 }
425 assets, err = r.store.ListAssets(ctx, name, id)
426 if err != nil {
427 return err
428 }
429 w.DeleteReleases = []domain.Release{rel}
430 w.DeleteAssets = assets
431 for _, a := range assets {
432 w.Used -= a.Charged
433 }
434 return nil
435 })
436 if err != nil {
437 return err
438 }
439 r.removeAssetBlobs(context.WithoutCancel(ctx), assets)
440 r.emit(ctx, repo, rel, "deleted")
441 return nil
442}
443
444// OnRefsMoved unpublishes releases of deleted tags; recreated tags stay drafts.
445func (r *Releases) OnRefsMoved(ctx context.Context, name domain.RepoName, updates []domain.RefUpdate) {
446 for _, u := range updates {
447 if !strings.HasPrefix(u.Ref, "refs/tags/") || u.New != domain.ZeroOID {
448 continue
449 }
450 var changed bool
451 var rel domain.Release
452 err := r.updateReleases(ctx, name, func(w *ports.ReleaseWrite) error {
453 changed = false
454 var err error
455 rel, err = r.store.GetReleaseByTag(ctx, name, strings.TrimPrefix(u.Ref, "refs/tags/"))
456 if err != nil {
457 return err
458 }
459 if rel.Draft {
460 return nil
461 }
462 rel.Draft = true
463 rel.MakeLatest = false
464 rel.PublishedAt = time.Time{}
465 rel.UpdatedAt = r.clock.Now()
466 w.PutReleases = []domain.Release{rel}
467 changed = true
468 return nil
469 })
470 if errors.Is(err, domain.ErrNotFound) {
471 continue
472 }
473 if err != nil {
474 ctxLog(ctx).Error().Err(err).Msg("unpublish release after tag deletion")
475 continue
476 }
477 if changed {
478 repo, err := lookupRepo(ctx, r.repos, name)
479 if err == nil {
480 r.emit(ctx, repo, rel, "unpublished")
481 }
482 }
483 }
484}
485
486// AssetsOf snapshots assets before the repository's metadata is removed.
487func (r *Releases) AssetsOf(ctx context.Context, name domain.RepoName) ([]domain.ReleaseAsset, error) {
488 all, err := r.store.AllAssets(ctx)
489 if err != nil {
490 return nil, err
491 }
492 return slices.DeleteFunc(all, func(a domain.ReleaseAsset) bool { return a.Repo != name }), nil
493}
494
495// OnRepoDeleted removes contents after repository metadata has been deleted.
496func (r *Releases) OnRepoDeleted(ctx context.Context, _ domain.RepoName, assets []domain.ReleaseAsset) {
497 r.removeAssetBlobs(ctx, assets)
498}
499func (r *Releases) removeAssetBlobs(ctx context.Context, assets []domain.ReleaseAsset) {
500 for _, a := range assets {
501 if a.MultipartID != "" {
502 if err := r.blobs.AbortMultipart(ctx, a.BlobKey(), a.MultipartID); err != nil {
503 ctxLog(ctx).Error().Err(err).Int64("asset", a.ID).Msg("abort release asset multipart")
504 }
505 }
506 if err := r.blobs.Delete(ctx, a.BlobKey()); err != nil {
507 ctxLog(ctx).Error().Err(err).Int64("asset", a.ID).Msg("remove release asset blob")
508 }
509 }
510}