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 err = r.store.ApplyReleases(ctx, w)
96 if errors.Is(err, domain.ErrConflict) {
97 continue
98 }
99 return err
100 }
101 return fmt.Errorf("%w: releases of %s keep changing, try again", domain.ErrUnavailable, repo)
102}
103func (r *Releases) resolve(ctx context.Context, repo domain.RepoName, rev string) (domain.ObjectID, error) {
104 v, err := domain.ParseRevision(rev)
105 if err != nil {
106 return domain.ObjectID{}, err
107 }
108 got, err := r.reader.ResolveRevision(ctx, repo, v)
109 return got.CommitID, err
110}
111func (r *Releases) bind(ctx context.Context, p domain.Principal, rel *domain.Release, target string, create bool) error {
112 oid, err := r.resolve(ctx, rel.Repo, "refs/tags/"+rel.Tag)
113 if errors.Is(err, domain.ErrNotFound) {
114 if rel.Draft {
115 rel.TargetCommitish = ""
116 return nil
117 }
118 if !create || target == "" {
119 return fmt.Errorf("%w: tag %q does not exist; give target_commitish to create it", domain.ErrPreconditionFailed, rel.Tag)
120 }
121 if r.pushes == nil {
122 return fmt.Errorf("%w: release tag notifications are not configured", domain.ErrUnavailable)
123 }
124 oid, err = r.resolve(ctx, rel.Repo, target)
125 if err != nil {
126 return err
127 }
128 err = r.tags.CreateTag(ctx, rel.Repo, rel.Tag, oid)
129 if errors.Is(err, domain.ErrAlreadyExists) {
130 oid, err = r.resolve(ctx, rel.Repo, "refs/tags/"+rel.Tag)
131 } else if err == nil {
132 err = r.pushes.Record(ctx, p, rel.Repo, []domain.RefUpdate{{Ref: "refs/tags/" + rel.Tag, Old: domain.ZeroOID, New: oid.String()}})
133 }
134 }
135 if err != nil {
136 return err
137 }
138 rel.TargetCommitish = oid.String()
139 return nil
140}
141func (r *Releases) pin(ctx context.Context, w *ports.ReleaseWrite, rel domain.Release) error {
142 if !rel.MakeLatest {
143 return nil
144 }
145 all, err := r.store.ListReleases(ctx, rel.Repo)
146 if err != nil {
147 return err
148 }
149 for _, other := range all {
150 if other.ID != rel.ID && other.MakeLatest {
151 other.MakeLatest = false
152 w.PutReleases = append(w.PutReleases, other)
153 }
154 }
155 return nil
156}
157func (r *Releases) emit(ctx context.Context, repo domain.Repo, rel domain.Release, kind string) {
158 at := r.clock.Now()
159 var ev domain.Event
160 switch kind {
161 case "published":
162 ev = domain.ReleasePublished{Release: rel, At: at, Visibility: repo.Visibility, CreatedAt: repo.CreatedAt}
163 case "updated":
164 ev = domain.ReleaseUpdated{Release: rel, At: at, Visibility: repo.Visibility, CreatedAt: repo.CreatedAt}
165 case "deleted":
166 ev = domain.ReleaseDeleted{Release: rel, At: at, Visibility: repo.Visibility, CreatedAt: repo.CreatedAt}
167 case "unpublished":
168 ev = domain.ReleaseUnpublished{Release: rel, At: at, Visibility: repo.Visibility, CreatedAt: repo.CreatedAt}
169 }
170 r.bus.Publish(ctx, ev)
171}
172
173// CreateRelease creates a release, optionally creating its tag when publishing.
174func (r *Releases) CreateRelease(ctx context.Context, p domain.Principal, name domain.RepoName, params ports.CreateReleaseParams) (domain.Release, error) {
175 repo, err := r.writable(ctx, p, name)
176 if err != nil {
177 return domain.Release{}, err
178 }
179 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()}
180 if rel.Tag, err = domain.ParseReleaseTag(params.Tag); err != nil {
181 return rel, err
182 }
183 if rel.Title, err = domain.ParseReleaseTitle(params.Title); err != nil {
184 return rel, err
185 }
186 if rel.Notes, err = domain.ParseReleaseNotes(params.Notes); err != nil {
187 return rel, err
188 }
189 if _, err = r.store.GetReleaseByTag(ctx, name, rel.Tag); err == nil {
190 return rel, domain.ErrAlreadyExists
191 } else if !errors.Is(err, domain.ErrNotFound) {
192 return rel, err
193 }
194 if err = r.bind(ctx, p, &rel, params.TargetCommitish, true); err != nil {
195 return rel, err
196 }
197 if rel.Draft {
198 rel.MakeLatest = false
199 } else {
200 rel.PublishedAt = r.clock.Now()
201 }
202 rel.ID, err = r.store.NextReleaseID(ctx)
203 if err != nil {
204 return rel, err
205 }
206 err = r.updateReleases(ctx, name, func(w *ports.ReleaseWrite) error {
207 if err := r.pin(ctx, w, rel); err != nil {
208 return err
209 }
210 w.PutReleases = append(w.PutReleases, rel)
211 return nil
212 })
213 if err != nil {
214 return domain.Release{}, err
215 }
216 if !rel.Draft {
217 r.emit(ctx, repo, rel, "published")
218 }
219 return r.store.GetRelease(ctx, name, rel.ID)
220}
221
222// GetRelease resolves a visible release by exactly one selector.
223func (r *Releases) GetRelease(ctx context.Context, p domain.Principal, name domain.RepoName, ref ports.ReleaseRef) (domain.Release, error) {
224 if _, err := r.readable(ctx, p, name); err != nil {
225 return domain.Release{}, err
226 }
227 n := 0
228 if ref.ID != 0 {
229 n++
230 }
231 if ref.Tag != "" {
232 n++
233 }
234 if ref.Latest {
235 n++
236 }
237 if n != 1 {
238 return domain.Release{}, domain.ErrInvalid
239 }
240 var rel domain.Release
241 var err error
242 switch {
243 case ref.ID != 0:
244 rel, err = r.store.GetRelease(ctx, name, ref.ID)
245 case ref.Tag != "":
246 rel, err = r.store.GetReleaseByTag(ctx, name, ref.Tag)
247 default:
248 var all []domain.Release
249 all, err = r.store.ListReleases(ctx, name)
250 if err == nil {
251 var ok bool
252 rel, ok = domain.LatestRelease(all)
253 if !ok {
254 err = domain.ErrNotFound
255 }
256 }
257 }
258 if err == nil && rel.Draft && !releaseWriter(p, name) {
259 err = domain.ErrNotFound
260 }
261 return rel, err
262}
263
264// ListReleases lists visible releases newest first, with an ID cursor.
265func (r *Releases) ListReleases(ctx context.Context, p domain.Principal, name domain.RepoName, size int, token string) (ports.ReleasePage, error) {
266 var out ports.ReleasePage
267 if _, err := r.readable(ctx, p, name); err != nil {
268 return out, err
269 }
270 if size == 0 {
271 size = 30
272 }
273 n, err := pageSize(size)
274 if err != nil {
275 return out, err
276 }
277 var cursor int64
278 if token != "" {
279 cursor, err = strconv.ParseInt(token, 10, 64)
280 if err != nil || cursor <= 0 {
281 return out, domain.ErrInvalidPageToken
282 }
283 }
284 all, err := r.store.ListReleases(ctx, name)
285 if err != nil {
286 return out, err
287 }
288 slices.SortFunc(all, func(a, b domain.Release) int {
289 if c := b.CreatedAt.Compare(a.CreatedAt); c != 0 {
290 return c
291 }
292 if a.ID > b.ID {
293 return -1
294 }
295 if a.ID < b.ID {
296 return 1
297 }
298 return 0
299 })
300 start := token == ""
301 writer := releaseWriter(p, name)
302 for _, rel := range all {
303 if rel.Draft && !writer {
304 continue // a hidden draft is not even a valid cursor
305 }
306 if !start {
307 if rel.ID == cursor {
308 start = true
309 }
310 continue
311 }
312 if len(out.Releases) == n {
313 out.NextPageToken = strconv.FormatInt(out.Releases[n-1].ID, 10)
314 break
315 }
316 out.Releases = append(out.Releases, rel)
317 }
318 if !start {
319 return out, domain.ErrInvalidPageToken
320 }
321 return out, nil
322}
323
324// UpdateRelease changes only fields whose pointers are non-nil.
325func (r *Releases) UpdateRelease(ctx context.Context, p domain.Principal, name domain.RepoName, id int64, params ports.UpdateReleaseParams) (domain.Release, error) {
326 repo, err := r.writable(ctx, p, name)
327 if err != nil {
328 return domain.Release{}, err
329 }
330 var rel domain.Release
331 kind := "updated"
332 err = r.updateReleases(ctx, name, func(w *ports.ReleaseWrite) error {
333 var err error
334 rel, err = r.store.GetRelease(ctx, name, id)
335 if err != nil {
336 return err
337 }
338 wasDraft := rel.Draft
339 if params.Tag != nil {
340 rel.Tag, err = domain.ParseReleaseTag(*params.Tag)
341 if err != nil {
342 return err
343 }
344 }
345 if params.Title != nil {
346 rel.Title, err = domain.ParseReleaseTitle(*params.Title)
347 if err != nil {
348 return err
349 }
350 }
351 if params.Notes != nil {
352 rel.Notes, err = domain.ParseReleaseNotes(*params.Notes)
353 if err != nil {
354 return err
355 }
356 }
357 if params.Draft != nil {
358 rel.Draft = *params.Draft
359 }
360 if params.Prerelease != nil {
361 rel.Prerelease = *params.Prerelease
362 }
363 if params.MakeLatest != nil {
364 rel.MakeLatest = *params.MakeLatest
365 }
366 target := ""
367 if params.TargetCommitish != nil {
368 target = *params.TargetCommitish
369 }
370 if err = r.bind(ctx, p, &rel, target, target != ""); err != nil {
371 return err
372 }
373 kind = "updated"
374 if rel.Draft {
375 rel.PublishedAt = time.Time{}
376 rel.MakeLatest = false
377 } else if wasDraft {
378 rel.PublishedAt = r.clock.Now()
379 kind = "published"
380 }
381 rel.UpdatedAt = r.clock.Now()
382 if err = r.pin(ctx, w, rel); err != nil {
383 return err
384 }
385 w.PutReleases = append(w.PutReleases, rel)
386 return nil
387 })
388 if err != nil {
389 return domain.Release{}, err
390 }
391 r.emit(ctx, repo, rel, kind)
392 return r.store.GetRelease(ctx, name, id)
393}
394
395// DeleteRelease removes metadata first, then its contents.
396func (r *Releases) DeleteRelease(ctx context.Context, p domain.Principal, name domain.RepoName, id int64) error {
397 repo, err := r.writable(ctx, p, name)
398 if err != nil {
399 return err
400 }
401 var rel domain.Release
402 var assets []domain.ReleaseAsset
403 err = r.updateReleases(ctx, name, func(w *ports.ReleaseWrite) error {
404 var err error
405 rel, err = r.store.GetRelease(ctx, name, id)
406 if err != nil {
407 return err
408 }
409 assets, err = r.store.ListAssets(ctx, name, id)
410 if err != nil {
411 return err
412 }
413 w.DeleteReleases = []domain.Release{rel}
414 w.DeleteAssets = assets
415 for _, a := range assets {
416 w.Used -= a.Charged
417 }
418 return nil
419 })
420 if err != nil {
421 return err
422 }
423 r.removeAssetBlobs(context.WithoutCancel(ctx), assets)
424 r.emit(ctx, repo, rel, "deleted")
425 return nil
426}
427
428// OnRefsMoved unpublishes releases of deleted tags; recreated tags stay drafts.
429func (r *Releases) OnRefsMoved(ctx context.Context, name domain.RepoName, updates []domain.RefUpdate) {
430 for _, u := range updates {
431 if !strings.HasPrefix(u.Ref, "refs/tags/") || u.New != domain.ZeroOID {
432 continue
433 }
434 var changed bool
435 var rel domain.Release
436 err := r.updateReleases(ctx, name, func(w *ports.ReleaseWrite) error {
437 changed = false
438 var err error
439 rel, err = r.store.GetReleaseByTag(ctx, name, strings.TrimPrefix(u.Ref, "refs/tags/"))
440 if err != nil {
441 return err
442 }
443 if rel.Draft {
444 return nil
445 }
446 rel.Draft = true
447 rel.MakeLatest = false
448 rel.PublishedAt = time.Time{}
449 rel.UpdatedAt = r.clock.Now()
450 w.PutReleases = []domain.Release{rel}
451 changed = true
452 return nil
453 })
454 if errors.Is(err, domain.ErrNotFound) {
455 continue
456 }
457 if err != nil {
458 ctxLog(ctx).Error().Err(err).Msg("unpublish release after tag deletion")
459 continue
460 }
461 if changed {
462 repo, err := lookupRepo(ctx, r.repos, name)
463 if err == nil {
464 r.emit(ctx, repo, rel, "unpublished")
465 }
466 }
467 }
468}
469
470// AssetsOf snapshots assets before the repository's metadata is removed.
471func (r *Releases) AssetsOf(ctx context.Context, name domain.RepoName) ([]domain.ReleaseAsset, error) {
472 all, err := r.store.AllAssets(ctx)
473 if err != nil {
474 return nil, err
475 }
476 return slices.DeleteFunc(all, func(a domain.ReleaseAsset) bool { return a.Repo != name }), nil
477}
478
479// OnRepoDeleted removes contents after repository metadata has been deleted.
480func (r *Releases) OnRepoDeleted(ctx context.Context, _ domain.RepoName, assets []domain.ReleaseAsset) {
481 r.removeAssetBlobs(ctx, assets)
482}
483func (r *Releases) removeAssetBlobs(ctx context.Context, assets []domain.ReleaseAsset) {
484 for _, a := range assets {
485 if a.MultipartID != "" {
486 if err := r.blobs.AbortMultipart(ctx, a.BlobKey(), a.MultipartID); err != nil {
487 ctxLog(ctx).Error().Err(err).Int64("asset", a.ID).Msg("abort release asset multipart")
488 }
489 }
490 if err := r.blobs.Delete(ctx, a.BlobKey()); err != nil {
491 ctxLog(ctx).Error().Err(err).Int64("asset", a.ID).Msg("remove release asset blob")
492 }
493 }
494}