detached 5dbd27f

#24 feat(releases): GitHub-like releases and S3 blob storage

state
merged · 5dbd27f · fast-forward
branch
feat/releases → main
opened
bnema ·
updated

GitHub-like releases for ermite, plus a pluggable blob store (ADR 0034).

What it adds

  • Releases bound to a Git tag: title, Markdown notes, assets, draft/prerelease, latest (make_latest pin). Deleting the tag turns the release back into a draft.
  • Interfaces: GitHub-compatible REST under /-/github/api (also /api/v3 for go-github/GoReleaser), ConnectRPC ReleaseService, ermite release CLI, read-only web pages.
  • Stable links: //releases/latest/download/ and //releases/download//. Downloads are always relayed by ermite (Range/If-Range supported).
  • Scope release:write (in the cli preset), or a job token with contents: write. Branch-limited tokens are refused.
  • Blob store: disk ($ERMITE_DATA_DIR/blobs) or any S3-compatible provider via ERMITE_S3_* (common S3 subset, ermite never creates buckets). CI artifacts move onto it; legacy disk artifacts stay readable.
  • Storage schema 6: unrestricted tokens holding every schema-5 scope gain release:write.
  • Requires ermite-api v0.16.0 (SCOPE_RELEASE_WRITE = 13, release events 27-30).

Checks

  • go build, vet, hexcheck, go test -race ./... green.
  • e2e on the local dev stack: disk and MinIO backends, Firecracker runner with action-gh-release and upload-artifact (20 MiB), Playwright on the web pages. S3 contract suite against MinIO and a real Scaleway bucket.
  • Wrap-up review: 13 findings, all fixed with regression tests.

Deployment (after merge): Gordon secrets and S3 config for staging then prod, re-enable release.yml workflows.

CommitSubjectAuthorDateSignatureStatus
5dbd27ffix(lint): check ignored close and remove errors, simplify release conditionsbriceGPG Verified
b869944chore(deps): require ermite-api v0.16.0 and drop the local replacebriceGPG Verified
1428b20fix(releases): hide draft events and cursors, honour update targets, share range handlingbriceGPG Verified
97e2b33fix(blobs): reclaim crashed temp files, skip broken multipart entries, sync new directoriesbriceGPG Verified
8361f60fix(ci): bound concurrent artifact block uploads and reuse their buffersbriceGPG Verified
6abe390test(blobs): run the blob contract on MinIO, which lists unfinished uploads by keybriceGPG Verified
27d9aa4fix(githubapi): accept the api/v3 base that GoReleaser appends to the API URLbriceGPG Verified
d50b7cdfeat(app): wire releases and the blob store, S3 configuration, MinIO compose profile and docsbriceGPG Verified
0933a48feat(cli): ermite release commands and release:write in the cli presetbriceGPG Verified
f67dc6afeat(githubapi): GitHub-compatible REST API for releases and assetsbriceGPG Verified
eab40eefeat(web): releases tab, release pages and stable asset download linksbriceGPG Verified
bfabb0cfeat(rpc): ConnectRPC ReleaseService with streamed asset upload and downloadbriceGPG Verified
b8fe0a4fix(rpc): map the release:write scope to the contractbriceGPG Verified
f9e6941feat(ci): store CI artifacts on the blob store, keep legacy disk artifacts readablebriceGPG Verified
eb72c2ffeat(releases): release, asset and sweep use casesbriceGPG Verified
0c4e1e3feat(releases): git tag writer and release hooks on pushes and repository deletionbriceGPG Verified
8fd2b19feat(blobs): blob store port with disk and provider-agnostic S3 adaptersbriceGPG Verified
c57f9ecfeat(releases): release domain, kv storage, ports and schema 6 with release:writebriceGPG Verified

Checks 8

149 files · +17035 -809
Unified
UnifiedSplit
8/8
⋯

Merge settings

modes
fast-forward, squash, merge commit
checks
failing or unfinished checks block the merge
M .hexcheck.yaml +12 -2
show this file
M README.md +1 -0
show this file
M compose.yaml +39 -0
show this file
M docs/ci-artifacts.md +2 -1
show this file
M docs/cli.md +45 -0
show this file
M docs/configuration.md +29 -1
show this file
A docs/releases.md +99 -0
show this file
A docs/storage.md +92 -0
show this file
M docs/tokens.md +9 -0
show this file
M gen/ermite/storage/ci_artifact.pb.go +15 -5
show this file
A gen/ermite/storage/release.pb.go +463 -0
show this file
M go.mod +16 -1
show this file
M go.sum +40 -2
show this file
A internal/adapters/blobdisk/blobdisk.go +572 -0
show this file
A internal/adapters/blobdisk/blobdisk_internal_test.go +87 -0
show this file
A internal/adapters/blobdisk/blobdisk_test.go +29 -0
show this file
A internal/adapters/blobs3/blobs3.go +461 -0
show this file
A internal/adapters/blobs3/blobs3_test.go +85 -0
show this file
A internal/adapters/blobs3/config.go +13 -0
show this file
A internal/adapters/blobs3/integration_test.go +44 -0
show this file
A internal/adapters/blobtest/suite.go +256 -0
show this file
M internal/adapters/ciartifacts/ciartifacts.go +5 -171
show this file
M internal/adapters/ciartifacts/ciartifacts_test.go +8 -105
show this file
M internal/adapters/client/client.go +5 -0
show this file
M internal/adapters/client/client_test.go +3 -3
show this file
M internal/adapters/client/features.go +3 -1
show this file
A internal/adapters/client/release_test.go +43 -0
show this file
M internal/adapters/client/scopes_test.go +1 -0
show this file
A internal/adapters/gitcli/tags.go +42 -0
show this file
A internal/adapters/gitcli/tags_test.go +31 -0
show this file
M internal/adapters/githubapi/handler.go +23 -2
show this file
A internal/adapters/githubapi/releases.go +413 -0
show this file
A internal/adapters/githubapi/releases_test.go +466 -0
show this file
A internal/adapters/githubapi/testdata/releases/draft.json +40 -0
show this file
A internal/adapters/githubapi/testdata/releases/release.json +113 -0
show this file
M internal/adapters/kvstore/ci_artifact.go +2 -1
show this file
M internal/adapters/kvstore/ci_artifact_test.go +5 -0
show this file
M internal/adapters/kvstore/issues_test.go +16 -8
show this file
A internal/adapters/kvstore/releases.go +408 -0
show this file
A internal/adapters/kvstore/releases_test.go +131 -0
show this file
M internal/adapters/kvstore/repos.go +1 -0
show this file
M internal/adapters/kvstore/store.go +25 -4
show this file
M internal/adapters/kvstore/store_test.go +57 -0
show this file
M internal/adapters/rpc/api.go +9 -0
show this file
M internal/adapters/rpc/convert.go +28 -3
show this file
M internal/adapters/rpc/errors.go +2 -0
show this file
M internal/adapters/rpc/event.go +26 -0
show this file
A internal/adapters/rpc/release.go +358 -0
show this file
A internal/adapters/rpc/release_test.go +472 -0
show this file
M internal/adapters/web/fragments.go +8 -2
show this file
M internal/adapters/web/fragments_test.go +20 -0
show this file
A internal/adapters/web/releases.go +244 -0
show this file
A internal/adapters/web/releases.templ +67 -0
show this file
A internal/adapters/web/releases_templ.go +439 -0
show this file
A internal/adapters/web/releases_test.go +298 -0
show this file
M internal/adapters/web/repo.go +3 -0
show this file
M internal/adapters/web/web.go +11 -0
show this file
M internal/app/api_test.go +1 -1
show this file
M internal/app/ci_artifacts_test.go +110 -12
show this file
M internal/app/ci_e2e_test.go +2 -0
show this file
M internal/app/cli_e2e_test.go +4 -3
show this file
A internal/app/cli_release.go +603 -0
show this file
A internal/app/cli_release_e2e_test.go +65 -0
show this file
A internal/app/cli_release_test.go +388 -0
show this file
M internal/app/cli_token.go +9 -0
show this file
M internal/app/cli_token_test.go +40 -1
show this file
M internal/app/cli_version.go +1 -0
show this file
M internal/app/cli_version_test.go +5 -5
show this file
M internal/app/forge.go +49 -4
show this file
A internal/app/release_e2e_test.go +160 -0
show this file
M internal/app/root.go +1 -1
show this file
M internal/app/testdata/web/blob-binary.html +15 -5
show this file
M internal/app/testdata/web/blob-code.html +15 -5
show this file
M internal/app/testdata/web/blob-empty.html +15 -5
show this file
M internal/app/testdata/web/blob-image.html +15 -5
show this file
M internal/app/testdata/web/blob-markdown.html +15 -5
show this file
M internal/app/testdata/web/blob-svg.html +15 -5
show this file
M internal/app/testdata/web/blob-text.html +15 -5
show this file
M internal/app/testdata/web/blob-too-large.html +15 -5
show this file
M internal/app/testdata/web/blob-trojan.html +15 -5
show this file
M internal/app/testdata/web/commit-big.html +15 -5
show this file
M internal/app/testdata/web/commit-binary.html +15 -5
show this file
M internal/app/testdata/web/commit-merge.html +15 -5
show this file
M internal/app/testdata/web/commit-modify-split.html +15 -5
show this file
M internal/app/testdata/web/commit-modify.html +15 -5
show this file
M internal/app/testdata/web/commit-rename.html +15 -5
show this file
M internal/app/testdata/web/commit-root.html +15 -5
show this file
M internal/app/testdata/web/commit-trojan.html +15 -5
show this file
M internal/app/testdata/web/commits-page2.html +14 -4
show this file
M internal/app/testdata/web/commits-path.html +14 -4
show this file
M internal/app/testdata/web/commits-root.html +14 -4
show this file
M internal/app/testdata/web/commits-small.html +16 -6
show this file
M internal/app/testdata/web/refs.html +17 -7
show this file
M internal/app/testdata/web/summary.html +17 -7
show this file
M internal/app/testdata/web/tree-nested.html +15 -5
show this file
M internal/app/testdata/web/tree-root.html +15 -5
show this file
M internal/app/testdata/web/tree-slash-ref.html +15 -5
show this file
M internal/app/testdata/web/tree-subdir.html +15 -5
show this file
M internal/app/web.go +1 -0
show this file
A internal/app/web_release_test.go +111 -0
show this file
M internal/boundary/boundary_test.go +14 -2
show this file
M internal/config/config.go +117 -0
show this file
A internal/config/storage_test.go +111 -0
show this file
M internal/domain/ci_artifact.go +21 -0
show this file
M internal/domain/ci_artifact_test.go +7 -0
show this file
M internal/domain/event.go +44 -0
show this file
A internal/domain/release.go +165 -0
show this file
A internal/domain/release_test.go +59 -0
show this file
M internal/domain/token.go +4 -2
show this file
M internal/domain/token_test.go +5 -1
show this file
A internal/httprange/httprange.go +77 -0
show this file
A internal/httprange/httprange_test.go +72 -0
show this file
A internal/ports/blob.go +72 -0
show this file
A internal/ports/blob_test.go +19 -0
show this file
M internal/ports/ci_artifact.go +3 -13
show this file
A internal/ports/mocks/BlobStore.go +644 -0
show this file
M internal/ports/mocks/CIArtifactFiles.go +0 -276
show this file
A internal/ports/mocks/GitHubReleaseAPI.go +1044 -0
show this file
A internal/ports/mocks/MultipartUpload.go +273 -0
show this file
A internal/ports/mocks/ReleaseRepoCleaner.go +168 -0
show this file
A internal/ports/mocks/ReleaseService.go +997 -0
show this file
A internal/ports/mocks/ReleaseStore.go +796 -0
show this file
A internal/ports/mocks/ReleaseSweeper.go +87 -0
show this file
A internal/ports/mocks/ReleaseTagTrigger.go +100 -0
show this file
A internal/ports/mocks/ReleaseTagWriter.go +117 -0
show this file
A internal/ports/release.go +119 -0
@@ -0,0 +1,119 @@
1+// SPDX-License-Identifier: AGPL-3.0-or-later
2+
3+package ports
4+
5+import (
6+ "bnema.dev/ermite/internal/domain"
7+ "context"
8+ "io"
9+)
10+
11+// ReleaseWrite atomically changes one repository's metadata. Every write
12+// requires UsageVersion (zero requires absence), and each record's Version.
13+// Used includes both uploading and uploaded assets.
14+type ReleaseWrite struct {
15+ Repo domain.RepoName
16+ Used int64
17+ UsageVersion uint64
18+ PutReleases, DeleteReleases []domain.Release
19+ PutAssets, DeleteAssets []domain.ReleaseAsset
20+}
21+
22+// ReleaseStore keeps metadata, never contents. *kvstore.Releases implements it.
23+type ReleaseStore interface {
24+ NextReleaseID(ctx context.Context) (int64, error)
25+ ReleaseUsage(ctx context.Context, repo domain.RepoName) (used int64, version uint64, err error)
26+ GetRelease(ctx context.Context, repo domain.RepoName, id int64) (domain.Release, error)
27+ GetReleaseByTag(ctx context.Context, repo domain.RepoName, tag string) (domain.Release, error)
28+ ReleaseRepo(ctx context.Context, id int64) (domain.RepoName, error)
29+ ListReleases(ctx context.Context, repo domain.RepoName) ([]domain.Release, error)
30+ GetAsset(ctx context.Context, id int64) (domain.ReleaseAsset, error)
31+ GetAssetByName(ctx context.Context, repo domain.RepoName, releaseID int64, name string) (domain.ReleaseAsset, error)
32+ ListAssets(ctx context.Context, repo domain.RepoName, releaseID int64) ([]domain.ReleaseAsset, error)
33+ AllAssets(ctx context.Context) ([]domain.ReleaseAsset, error)
34+ ApplyReleases(ctx context.Context, w ReleaseWrite) error
35+}
36+
37+// ReleaseLimits defaults to domain.DefaultReleaseAssetMax (ERMITE_RELEASE_ASSET_MAX),
38+// domain.DefaultReleaseAssetsMax (ERMITE_RELEASE_ASSETS_MAX), and
39+// domain.DefaultReleaseQuota (ERMITE_RELEASE_QUOTA).
40+type ReleaseLimits struct{ AssetMax, AssetsMax, Quota int64 }
41+type CreateReleaseParams struct {
42+ Tag, TargetCommitish, Title, Notes string
43+ Draft, Prerelease, MakeLatest bool
44+}
45+type UpdateReleaseParams struct {
46+ Tag, TargetCommitish, Title, Notes *string
47+ Draft, Prerelease, MakeLatest *bool
48+}
49+
50+// ReleaseRef selects exactly one of ID, Tag, Latest.
51+type ReleaseRef struct {
52+ ID int64
53+ Tag string
54+ Latest bool
55+}
56+
57+// AssetRef selects ID, (Tag, Name), or (Latest, Name).
58+type AssetRef struct {
59+ ID int64
60+ Tag, Name string
61+ Latest bool
62+}
63+type UploadAssetParams struct {
64+ ReleaseID int64
65+ Name, Label, ContentType string
66+ Size int64
67+ Replace bool
68+}
69+type UpdateAssetParams struct{ Name, Label *string }
70+type ReleaseUsage struct {
71+ Used int64
72+ Limits ReleaseLimits
73+}
74+type AssetChunk struct {
75+ Asset *domain.ReleaseAsset
76+ Data []byte
77+}
78+type ReleasePage struct {
79+ Releases []domain.Release
80+ NextPageToken string
81+}
82+
83+// ReleaseService manages releases and streams assets; *usecase.Releases implements it.
84+type ReleaseService interface {
85+ CreateRelease(ctx context.Context, p domain.Principal, repo domain.RepoName, params CreateReleaseParams) (domain.Release, error)
86+ GetRelease(ctx context.Context, p domain.Principal, repo domain.RepoName, ref ReleaseRef) (domain.Release, error)
87+ ListReleases(ctx context.Context, p domain.Principal, repo domain.RepoName, pageSize int, pageToken string) (ReleasePage, error)
88+ UpdateRelease(ctx context.Context, p domain.Principal, repo domain.RepoName, id int64, params UpdateReleaseParams) (domain.Release, error)
89+ DeleteRelease(ctx context.Context, p domain.Principal, repo domain.RepoName, id int64) error
90+ ListAssets(ctx context.Context, p domain.Principal, repo domain.RepoName, releaseID int64) ([]domain.ReleaseAsset, error)
91+ UploadAsset(ctx context.Context, p domain.Principal, repo domain.RepoName, params UploadAssetParams, body io.Reader) (domain.ReleaseAsset, error)
92+ UpdateAsset(ctx context.Context, p domain.Principal, repo domain.RepoName, assetID int64, params UpdateAssetParams) (domain.ReleaseAsset, error)
93+ DeleteAsset(ctx context.Context, p domain.Principal, repo domain.RepoName, assetID int64) error
94+ GetAsset(ctx context.Context, p domain.Principal, repo domain.RepoName, ref AssetRef) (domain.ReleaseAsset, error)
95+ // DownloadAsset sends metadata first, then chunks at most domain.CIArtifactChunk;
96+ // length less than zero reads to the end.
97+ DownloadAsset(ctx context.Context, p domain.Principal, repo domain.RepoName, ref AssetRef, offset, length int64, send func(AssetChunk) error) error
98+ Usage(ctx context.Context, p domain.Principal, repo domain.RepoName) (ReleaseUsage, error)
99+}
100+
101+// ReleaseTagWriter creates lightweight tags without moving existing refs.
102+// *gitcli.Git implements it. An existing ref returns domain.ErrAlreadyExists.
103+type ReleaseTagWriter interface {
104+ CreateTag(ctx context.Context, repo domain.RepoName, tag string, target domain.ObjectID) error
105+}
106+
107+// ReleaseTagTrigger handles tag deletion; *usecase.Releases implements it.
108+type ReleaseTagTrigger interface {
109+ OnRefsMoved(ctx context.Context, repo domain.RepoName, updates []domain.RefUpdate)
110+}
111+
112+// ReleaseSweeper removes stale assets and blobs; *usecase.Releases implements it.
113+type ReleaseSweeper interface{ Sweep(ctx context.Context) }
114+
115+// ReleaseRepoCleaner removes contents after metadata deletion; *usecase.Releases implements it.
116+type ReleaseRepoCleaner interface {
117+ AssetsOf(ctx context.Context, repo domain.RepoName) ([]domain.ReleaseAsset, error)
118+ OnRepoDeleted(ctx context.Context, repo domain.RepoName, assets []domain.ReleaseAsset)
119+}
M internal/ports/services.go +17 -0
@@ -390,3 +390,20 @@ type EnvironmentService interface {
390390 ListEnvironments(ctx context.Context, p domain.Principal, repo domain.RepoName, size int, token string) (EnvironmentPage, error)
391391 DeleteEnvironment(ctx context.Context, p domain.Principal, repo domain.RepoName, name string) error
392392 }
393+
394+// GitHubReleaseAPI is the Releases part of the REST subset. owner/repo follow
395+// GitHubAPIService; bodies are GitHub JSON. *usecase.ReleasesGitHub implements it.
396+type GitHubReleaseAPI interface {
397+ CreateRelease(ctx context.Context, p domain.Principal, owner, repo string, body []byte) ([]byte, error)
398+ GetRelease(ctx context.Context, p domain.Principal, owner, repo string, ref ReleaseRef) ([]byte, error)
399+ ListReleases(ctx context.Context, p domain.Principal, owner, repo string, page, perPage int) (body []byte, link string, err error)
400+ UpdateRelease(ctx context.Context, p domain.Principal, owner, repo string, id int64, body []byte) ([]byte, error)
401+ DeleteRelease(ctx context.Context, p domain.Principal, owner, repo string, id int64) error
402+ ListAssets(ctx context.Context, p domain.Principal, owner, repo string, id int64, page, perPage int) (body []byte, link string, err error)
403+ UploadAsset(ctx context.Context, p domain.Principal, owner, repo string, id int64, name, label, contentType string, size int64, r io.Reader) ([]byte, error)
404+ GetAsset(ctx context.Context, p domain.Principal, owner, repo string, assetID int64) ([]byte, error)
405+ UpdateAsset(ctx context.Context, p domain.Principal, owner, repo string, assetID int64, body []byte) ([]byte, error)
406+ DeleteAsset(ctx context.Context, p domain.Principal, owner, repo string, assetID int64) error
407+ // AssetByID resolves metadata for a download through ReleaseService.
408+ AssetByID(ctx context.Context, p domain.Principal, owner, repo string, assetID int64) (domain.ReleaseAsset, error)
409+}
M internal/usecase/ci.go +15 -7
@@ -125,14 +125,17 @@ type CIConfig struct {
125125 PublicURL string
126126 // Retention is how long runs, jobs, checks and logs are kept.
127127 Retention time.Duration
128- // Artifacts and ArtifactFiles keep the artifacts of the runs (CI-4 D3):
129- // the records and the contents. Without both the artifact services refuse
128+ // Artifacts and Blobs keep the records and contents of the artifacts.
129+ // Without both the artifact services refuse
130130 // every call. ArtifactLimits are their settings; a zero field takes its
131131 // default (the run retention of domain.CIRetentionDays,
132132 // domain.DefaultCIArtifactQuota, domain.DefaultCIArtifactMaxSize). Leave
133133 // them unset rather than assigning typed nil pointers.
134- Artifacts ports.CIArtifactStore
134+ Artifacts ports.CIArtifactStore
135+ // ArtifactFiles is the legacy reader; nil when $ERMITE_DATA_DIR/ci/artifacts
136+ // does not exist at startup; P8 wires it.
135137 ArtifactFiles ports.CIArtifactFiles
138+ Blobs ports.BlobStore
136139 ArtifactLimits ports.ArtifactLimits
137140
138141 // The durations below default to the contract's; tests shorten them.
@@ -186,6 +189,8 @@ type CIConfig struct {
186189 // ScheduleVisit, when set, is called with each repository the scheduler
187190 // looks at. Tests use it.
188191 ScheduleVisit func(repo domain.RepoName)
192+ // ReleaseSweeper collects abandoned release uploads during the daily sweep.
193+ ReleaseSweeper ports.ReleaseSweeper
189194 }
190195
191196 func (c CIConfig) withDefaults() CIConfig {
@@ -244,9 +249,12 @@ func (c CIConfig) withDefaults() CIConfig {
244249 // calls (a runner's update and an owner's cancel) never lose one another's
245250 // write. Assignment takes the assignment lock before the lock of the run.
246251 type CI struct {
247- cfg CIConfig
248- boot time.Time
249- queue chan ciTrigger
252+ cfg CIConfig
253+ // uploadSlots bounds the artifact uploads and assemblies buffering a
254+ // block in memory at once; nil (hand-built CI in tests) means unbounded.
255+ uploadSlots chan struct{}
256+ boot time.Time
257+ queue chan ciTrigger
250258 // runEvents are the requested and completed events of runs, for the
251259 // workflow_run workflows (ci_workflowrun.go).
252260 runEvents chan runEvent
@@ -315,7 +323,7 @@ func NewCI(ctx context.Context, cfg CIConfig) *CI {
315323 cfg = cfg.withDefaults()
316324 run, stop := context.WithCancel(context.WithoutCancel(ctx))
317325 return &CI{
318- pushes: map[domain.RepoName]int{}, pulls: map[pullKey]int{}, autoMerge: cfg.AutoMerge,
326+ pushes: map[domain.RepoName]int{}, uploadSlots: make(chan struct{}, ciArtifactUploadSlots), pulls: map[pullKey]int{}, autoMerge: cfg.AutoMerge,
319327 cfg: cfg, boot: cfg.Clock.Now(), queue: make(chan ciTrigger, cfg.TriggerQueue), runEvents: make(chan runEvent, cfg.TriggerQueue), sched: newScheduler(cfg.Clock.Now()), pull: newPullQueue(cfg.TriggerQueue, cfg.PullRefreshes, cfg.PullAttempts),
320328 ctx: run, stop: stop, flows: newWfCache(64), remoteSlots: make(chan struct{}, cfg.RemoteConcurrency), remoteCache: newRefsCache(),
321329 heartbeats: map[string]time.Time{}, seen: map[string]time.Time{}, polls: map[string]pollRecord{},
M internal/usecase/ci_artifacts.go +325 -12
@@ -13,6 +13,7 @@ import (
1313 "fmt"
1414 "io"
1515 "strconv"
16+ "strings"
1617 "sync"
1718 "time"
1819
@@ -25,7 +26,7 @@ import (
2526 // The artifacts of the runs (artifacts v4, CI-4 D3). The runner translates the
2627 // toolkit's protocol and calls JobArtifactService with the job token; owners
2728 // use ArtifactService. The records are in CIConfig.Artifacts, the contents in
28-// CIConfig.ArtifactFiles, named by ids only.
29+// CIConfig.Blobs (ArtifactFiles for legacy records), named by ids only.
2930 //
3031 // Every write of the artifacts of a repository goes through its usage record
3132 // (ports.CIArtifactWrite): a decision is made on the records read after it and
@@ -50,6 +51,44 @@ const (
5051 commitTries = 5
5152 )
5253
54+// ciArtifactUploadSlots bounds the artifact block uploads and assemblies that
55+// hold an 8 MiB buffer at the same time, server-wide.
56+const ciArtifactUploadSlots = 16
57+
58+// blockBufs pools the 8 MiB buffers of uploads and assemblies. CIArtifactMaxBlockSize
59+// and ports.BlobPartSize are both 8 MiB.
60+var blockBufs = sync.Pool{New: func() any {
61+ b := make([]byte, max(domain.CIArtifactMaxBlockSize, ports.BlobPartSize))
62+ return &b
63+}}
64+
65+// acquireUploadSlot takes an upload slot without waiting; release frees it.
66+func (c *CI) acquireUploadSlot() (release func(), err error) {
67+ if c.uploadSlots == nil {
68+ return func() {}, nil
69+ }
70+ select {
71+ case c.uploadSlots <- struct{}{}:
72+ return func() { <-c.uploadSlots }, nil
73+ default:
74+ return nil, fmt.Errorf("%w: too many artifact uploads in progress, retry", domain.ErrResourceExhausted)
75+ }
76+}
77+
78+// waitUploadSlot takes an upload slot, waiting for one until ctx ends: a
79+// commit must not fail because block uploads are in flight.
80+func (c *CI) waitUploadSlot(ctx context.Context) (release func(), err error) {
81+ if c.uploadSlots == nil {
82+ return func() {}, nil
83+ }
84+ select {
85+ case c.uploadSlots <- struct{}{}:
86+ return func() { <-c.uploadSlots }, nil
87+ case <-ctx.Done():
88+ return nil, ctx.Err()
89+ }
90+}
91+
5392 var (
5493 // errNoArtifactWrite ends updateArtifacts without a write.
5594 errNoArtifactWrite = errors.New("no write")
@@ -57,7 +96,7 @@ var (
5796 errArtifactChanged = errors.New("artifact changed")
5897 )
5998
60-func (c *CI) artifactsOn() bool { return c.cfg.Artifacts != nil && c.cfg.ArtifactFiles != nil }
99+func (c *CI) artifactsOn() bool { return c.cfg.Artifacts != nil && c.cfg.Blobs != nil }
61100
62101 func artifactRef(a domain.CIArtifact) ports.CIArtifactRef {
63102 return ports.CIArtifactRef{RunID: a.RunID, ID: a.ID}
@@ -309,7 +348,13 @@ func (c *CI) removeArtifacts(ctx context.Context, repo domain.RepoName, runID in
309348 // logged: the sweep removes the files that no record names.
310349 func (c *CI) removeArtifactFiles(ctx context.Context, arts []domain.CIArtifact) {
311350 for _, a := range arts {
312- if err := c.cfg.ArtifactFiles.Remove(ctx, artifactRef(a)); err != nil {
351+ var err error
352+ if a.Store == domain.CIArtifactStoreBlob {
353+ err = errors.Join(c.cfg.Blobs.Delete(ctx, a.BlobKey()), c.removeBlobBlocks(ctx, a))
354+ } else if c.cfg.ArtifactFiles != nil {
355+ err = c.cfg.ArtifactFiles.Remove(ctx, artifactRef(a))
356+ }
357+ if err != nil {
313358 ctxLog(ctx).Warn().Err(err).Int64("artifact_id", a.ID).Msg("ci: remove the files of an artifact; the sweep will")
314359 }
315360 }
@@ -373,7 +418,7 @@ func (c *CI) CreateJobArtifact(ctx context.Context, secret string, params ports.
373418 }
374419 a := domain.CIArtifact{
375420 ID: id, Repo: j.Repo, RunID: j.RunID, Attempt: j.RunAttempt, JobID: j.ID, Assignment: j.TokenHash,
376- Name: params.Name, MediaType: mediaType, State: domain.CIArtifactPending,
421+ Name: params.Name, MediaType: mediaType, State: domain.CIArtifactPending, Store: domain.CIArtifactStoreBlob,
377422 CreatedAt: now, ExpiresAt: domain.ArtifactExpiry(params.ExpiresAt, now, c.cfg.ArtifactLimits.Retention),
378423 }
379424 var discarded []domain.CIArtifact
@@ -460,15 +505,24 @@ func (c *CI) UploadJobArtifactBlock(ctx context.Context, secret, jobID string, a
460505 if a.State != domain.CIArtifactPending {
461506 return 0, errCommitted(a.ID)
462507 }
508+ // The slot is taken before anything is reserved, read or allocated.
509+ freeSlot, err := c.acquireUploadSlot()
510+ if err != nil {
511+ return 0, err
512+ }
513+ defer freeSlot()
463514 // Every byte is reserved before it is written, against the charges and
464515 // the other writes in flight (a block it replaces counts until the record
465516 // drops it): nothing beyond the limits ever reaches the disk.
466517 src := &reservingReader{ctx: ctx, c: c, a: a, r: data}
467518 defer func() { c.release(a, src.reserved) }()
468519 limits := c.cfg.ArtifactLimits
469- ref, file := artifactRef(a), newBlockFile(blockID)
470- size, err := c.cfg.ArtifactFiles.PutBlock(ctx, ref, file, src)
520+ file := newBlockFile(blockID)
521+ size, err := c.putUnsized(ctx, a.BlockKey(file), src)
471522 if err != nil {
523+ if derr := c.cfg.Blobs.Delete(context.WithoutCancel(ctx), a.BlockKey(file)); derr != nil {
524+ ctxLog(ctx).Warn().Err(derr).Msg("ci: remove a refused block; the sweep will")
525+ }
472526 return 0, err
473527 }
474528 var old *ports.CIArtifactBlock
@@ -509,13 +563,13 @@ func (c *CI) UploadJobArtifactBlock(ctx context.Context, secret, jobID string, a
509563 return nil
510564 })
511565 if err != nil {
512- if rerr := c.cfg.ArtifactFiles.RemoveBlock(ctx, ref, file); rerr != nil {
566+ if rerr := c.cfg.Blobs.Delete(context.WithoutCancel(ctx), a.BlockKey(file)); rerr != nil {
513567 ctxLog(ctx).Warn().Err(rerr).Int64("artifact_id", a.ID).Msg("ci: remove a refused block; the sweep will")
514568 }
515569 return 0, err
516570 }
517571 if old != nil {
518- if err := c.cfg.ArtifactFiles.RemoveBlock(ctx, ref, old.File); err != nil {
572+ if err := c.cfg.Blobs.Delete(ctx, a.BlockKey(old.File)); err != nil {
519573 ctxLog(ctx).Warn().Err(err).Int64("artifact_id", a.ID).Msg("ci: remove a replaced block")
520574 }
521575 }
@@ -602,7 +656,7 @@ func (c *CI) CommitJobArtifact(ctx context.Context, secret, jobID string, artifa
602656 return 0, "", err
603657 }
604658 }
605- size, digest, err := c.cfg.ArtifactFiles.Assemble(ctx, artifactRef(a), files)
659+ size, digest, err := c.assembleBlobs(ctx, a, files, total)
606660 if err != nil {
607661 c.release(a, excess)
608662 }
@@ -638,7 +692,7 @@ func (c *CI) CommitJobArtifact(ctx context.Context, secret, jobID string, artifa
638692 if err != nil {
639693 return 0, "", err
640694 }
641- if err := c.cfg.ArtifactFiles.RemoveBlocks(ctx, artifactRef(a)); err != nil {
695+ if err := c.removeBlobBlocks(ctx, a); err != nil {
642696 ctxLog(ctx).Warn().Err(err).Int64("artifact_id", a.ID).Msg("ci: remove the blocks of a committed artifact")
643697 }
644698 c.dropIfEnded(ctx, j)
@@ -782,7 +836,23 @@ func (c *CI) streamArtifact(ctx context.Context, a domain.CIArtifact, offset int
782836 if offset < 0 || offset > a.Size {
783837 return fmt.Errorf("%w: offset %d is beyond the size %d", domain.ErrOutOfRange, offset, a.Size)
784838 }
785- f, err := c.cfg.ArtifactFiles.Open(ctx, artifactRef(a), offset)
839+ var f io.ReadCloser
840+ var err error
841+ if a.Store == domain.CIArtifactStoreBlob {
842+ // BlobStore rejects offsets at EOF, while artifact downloads permit
843+ // an empty resumed stream (including zero-byte artifacts).
844+ if offset == a.Size {
845+ _, err = c.cfg.Blobs.Stat(ctx, a.BlobKey())
846+ f = io.NopCloser(bytes.NewReader(nil))
847+ } else {
848+ f, _, err = c.cfg.Blobs.Open(ctx, a.BlobKey(), offset, -1)
849+ }
850+ } else if c.cfg.ArtifactFiles != nil {
851+ f, err = c.cfg.ArtifactFiles.Open(ctx, artifactRef(a), offset)
852+ } else {
853+ ctxLog(ctx).Error().Int64("artifact_id", a.ID).Msg("ci: legacy artifact reader is unavailable")
854+ return domain.ErrNotFound
855+ }
786856 if err != nil {
787857 return err
788858 }
@@ -1008,6 +1078,10 @@ func (c *CI) sweepArtifactOrphans(ctx context.Context) {
10081078 }
10091079 }
10101080 }
1081+ c.sweepArtifactBlobOrphans(ctx)
1082+ if c.cfg.ArtifactFiles == nil {
1083+ return
1084+ }
10111085 c.removeOrphans(ctx, "artifacts", "artifact directories", c.cfg.ArtifactFiles.RunIDs, c.cfg.ArtifactFiles.RemoveRuns)
10121086 runIDs, err := c.cfg.ArtifactFiles.RunIDs(ctx)
10131087 if err != nil {
@@ -1022,7 +1096,7 @@ func (c *CI) sweepArtifactOrphans(ctx context.Context) {
10221096 }
10231097 for _, id := range ids {
10241098 a, err := c.cfg.Artifacts.GetArtifact(ctx, id)
1025- if err == nil && a.RunID == run {
1099+ if err == nil && a.RunID == run && a.Store != domain.CIArtifactStoreBlob {
10261100 continue
10271101 }
10281102 if err != nil && !errors.Is(err, domain.ErrNotFound) {
@@ -1035,3 +1109,242 @@ func (c *CI) sweepArtifactOrphans(ctx context.Context) {
10351109 }
10361110 }
10371111 }
1112+
1113+// putUnsized buffers at most one toolkit block; Put requires an exact size.
1114+func (c *CI) putUnsized(ctx context.Context, key string, r io.Reader) (int64, error) {
1115+ // A fixed backing array prevents bytes.Buffer's geometric growth from
1116+ // allocating beyond the block bound. Probe the next byte separately.
1117+ bp := blockBufs.Get().(*[]byte)
1118+ defer blockBufs.Put(bp)
1119+ buf := (*bp)[:domain.CIArtifactMaxBlockSize]
1120+ n, err := io.ReadFull(r, buf)
1121+ if err != nil && err != io.EOF && err != io.ErrUnexpectedEOF {
1122+ return 0, err
1123+ }
1124+ if n == len(buf) {
1125+ var extra [1]byte
1126+ m, err := io.ReadFull(r, extra[:])
1127+ if m != 0 {
1128+ return 0, fmt.Errorf("%w: a block has at most 8 MiB", domain.ErrInvalid)
1129+ }
1130+ if err != io.EOF {
1131+ return 0, err
1132+ }
1133+ }
1134+ if err := ctx.Err(); err != nil {
1135+ return 0, err
1136+ }
1137+ _, err = c.cfg.Blobs.Put(ctx, key, bytes.NewReader(buf[:n]), int64(n))
1138+ return int64(n), err
1139+}
1140+
1141+func (c *CI) removeBlobBlocks(ctx context.Context, a domain.CIArtifact) error {
1142+ return c.cfg.Blobs.List(ctx, a.BlobKey()+".blocks/", func(b ports.BlobInfo) error {
1143+ return c.cfg.Blobs.Delete(ctx, b.Key)
1144+ })
1145+}
1146+
1147+// assembleBlobs keeps one 8 MiB part in memory, even when a block crosses
1148+// part boundaries or is repeated in the commit list.
1149+func (c *CI) assembleBlobs(ctx context.Context, a domain.CIArtifact, files []string, total int64) (int64, string, error) {
1150+ freeSlot, err := c.waitUploadSlot(ctx)
1151+ if err != nil {
1152+ return 0, "", err
1153+ }
1154+ defer freeSlot()
1155+ var upload ports.MultipartUpload
1156+ if total > ports.BlobPartSize {
1157+ var err error
1158+ upload, err = c.cfg.Blobs.NewMultipart(ctx, a.BlobKey())
1159+ if err != nil {
1160+ return 0, "", err
1161+ }
1162+ }
1163+ completed := false
1164+ defer func() {
1165+ if upload != nil && !completed {
1166+ if err := upload.Abort(context.WithoutCancel(ctx)); err != nil {
1167+ ctxLog(ctx).Warn().Err(err).Msg("ci: abort artifact assembly; the sweep will")
1168+ }
1169+ }
1170+ }()
1171+ bp := blockBufs.Get().(*[]byte)
1172+ defer blockBufs.Put(bp)
1173+ buf := (*bp)[:ports.BlobPartSize]
1174+ used, part := 0, 1
1175+ var size int64
1176+ h := sha256.New()
1177+ flush := func() error {
1178+ if err := upload.UploadPart(ctx, part, bytes.NewReader(buf[:used]), int64(used)); err != nil {
1179+ return err
1180+ }
1181+ used = 0
1182+ part++
1183+ return nil
1184+ }
1185+ for _, file := range files {
1186+ r, info, err := c.cfg.Blobs.Open(ctx, a.BlockKey(file), 0, -1)
1187+ // Empty blocks cannot be opened under BlobStore's EOF contract.
1188+ if errors.Is(err, domain.ErrOutOfRange) {
1189+ info, err = c.cfg.Blobs.Stat(ctx, a.BlockKey(file))
1190+ if err == nil && info.Size == 0 {
1191+ continue
1192+ }
1193+ if err == nil {
1194+ return 0, "", domain.ErrOutOfRange
1195+ }
1196+ }
1197+ if err != nil {
1198+ return 0, "", err
1199+ }
1200+ if info.Size > total-size {
1201+ _ = r.Close()
1202+ return 0, "", fmt.Errorf("%w: artifact block size changed", domain.ErrInvalid)
1203+ }
1204+ left := info.Size
1205+ for left > 0 {
1206+ if err := ctx.Err(); err != nil {
1207+ _ = r.Close()
1208+ return 0, "", err
1209+ }
1210+ want := min(left, int64(len(buf)-used))
1211+ if want == 0 {
1212+ _ = r.Close()
1213+ return 0, "", fmt.Errorf("%w: artifact block size changed", domain.ErrInvalid)
1214+ }
1215+ n, err := io.ReadFull(r, buf[used:used+int(want)])
1216+ if err != nil {
1217+ _ = r.Close()
1218+ return 0, "", err
1219+ }
1220+ h.Write(buf[used : used+n])
1221+ used += n
1222+ size += int64(n)
1223+ left -= int64(n)
1224+ if used == len(buf) && upload != nil {
1225+ if err := flush(); err != nil {
1226+ _ = r.Close()
1227+ return 0, "", err
1228+ }
1229+ }
1230+ }
1231+ var extra [1]byte
1232+ n, err := io.ReadFull(r, extra[:])
1233+ cerr := r.Close()
1234+ if n != 0 || err != io.EOF {
1235+ return 0, "", fmt.Errorf("%w: artifact block size changed", domain.ErrInvalid)
1236+ }
1237+ if cerr != nil {
1238+ return 0, "", cerr
1239+ }
1240+ }
1241+ if size != total {
1242+ return 0, "", fmt.Errorf("%w: artifact block size changed", domain.ErrInvalid)
1243+ }
1244+ if upload == nil {
1245+ if _, err := c.cfg.Blobs.Put(ctx, a.BlobKey(), bytes.NewReader(buf[:used]), size); err != nil {
1246+ return 0, "", err
1247+ }
1248+ } else {
1249+ if used > 0 {
1250+ if err := flush(); err != nil {
1251+ return 0, "", err
1252+ }
1253+ }
1254+ if _, err := upload.Complete(ctx); err != nil {
1255+ return 0, "", err
1256+ }
1257+ completed = true
1258+ }
1259+ return size, "sha256:" + hex.EncodeToString(h.Sum(nil)), nil
1260+}
1261+
1262+func (c *CI) sweepArtifactBlobOrphans(ctx context.Context) {
1263+ log := ctxLog(ctx)
1264+ cutoff := c.now().Add(-domain.CIArtifactPendingMax)
1265+ err := c.cfg.Blobs.List(ctx, "ci/", func(b ports.BlobInfo) error {
1266+ if !b.ModTime.Before(cutoff) {
1267+ return nil
1268+ }
1269+ run, id, block, ok := artifactBlobIDs(b.Key)
1270+ if !ok {
1271+ return nil
1272+ }
1273+ a, err := c.cfg.Artifacts.GetArtifact(ctx, id)
1274+ if err != nil && !errors.Is(err, domain.ErrNotFound) {
1275+ return nil
1276+ }
1277+ if err == nil && a.RunID == run && a.Store == domain.CIArtifactStoreBlob && (!block || a.State == domain.CIArtifactPending) {
1278+ return nil
1279+ }
1280+ if err := c.cfg.Blobs.Delete(ctx, b.Key); err != nil {
1281+ log.Error().Err(err).Msg("ci: remove orphan artifact blob")
1282+ }
1283+ return nil
1284+ })
1285+ if err != nil {
1286+ log.Error().Err(err).Msg("ci: list artifact blobs")
1287+ }
1288+ err = c.cfg.Blobs.ListMultipart(ctx, "ci/", func(m ports.MultipartInfo) error {
1289+ if m.Started.Before(cutoff) {
1290+ if err := c.cfg.Blobs.AbortMultipart(ctx, m.Key, m.ID); err != nil {
1291+ log.Error().Err(err).Msg("ci: abort abandoned artifact upload")
1292+ }
1293+ }
1294+ return nil
1295+ })
1296+ if err != nil {
1297+ log.Error().Err(err).Msg("ci: list artifact multipart uploads")
1298+ }
1299+}
1300+
1301+func artifactBlobIDs(key string) (run, id int64, block, ok bool) {
1302+ rest, found := strings.CutPrefix(key, "ci/")
1303+ if !found {
1304+ return
1305+ }
1306+ r, tail, found := strings.Cut(rest, "/")
1307+ if !found || len(r) != 20 {
1308+ return
1309+ }
1310+ name, file, block := strings.Cut(tail, ".blocks/")
1311+ if len(name) != 20 || (block && file == "") {
1312+ return 0, 0, false, false
1313+ }
1314+ run, err := strconv.ParseInt(r, 10, 64)
1315+ if err != nil || run <= 0 {
1316+ return 0, 0, false, false
1317+ }
1318+ id, err = strconv.ParseInt(name, 10, 64)
1319+ return run, id, block, err == nil && id > 0
1320+}
1321+
1322+// removeArtifactRuns follows repository deletion, when records are already
1323+// gone. Run ids are global, so their blob prefixes cannot name another repo.
1324+func (c *CI) removeArtifactRuns(ctx context.Context, runIDs []int64) error {
1325+ var errs []error
1326+ if c.cfg.ArtifactFiles != nil {
1327+ errs = append(errs, c.cfg.ArtifactFiles.RemoveRuns(ctx, runIDs))
1328+ }
1329+ if c.cfg.Blobs != nil {
1330+ for _, run := range runIDs {
1331+ if run <= 0 {
1332+ continue
1333+ }
1334+ prefix := fmt.Sprintf("ci/%020d/", run)
1335+ errs = append(errs, c.cfg.Blobs.List(ctx, prefix, func(b ports.BlobInfo) error {
1336+ if err := c.cfg.Blobs.Delete(ctx, b.Key); err != nil {
1337+ errs = append(errs, err)
1338+ }
1339+ return nil
1340+ }))
1341+ errs = append(errs, c.cfg.Blobs.ListMultipart(ctx, prefix, func(m ports.MultipartInfo) error {
1342+ if err := c.cfg.Blobs.AbortMultipart(ctx, m.Key, m.ID); err != nil {
1343+ errs = append(errs, err)
1344+ }
1345+ return nil
1346+ }))
1347+ }
1348+ }
1349+ return errors.Join(errs...)
1350+}
A internal/usecase/ci_artifacts_blob_test.go +301 -0
@@ -0,0 +1,301 @@
1+// SPDX-License-Identifier: AGPL-3.0-or-later
2+
3+package usecase
4+
5+import (
6+ "bytes"
7+ "context"
8+ "crypto/sha256"
9+ "encoding/hex"
10+ "errors"
11+ "io"
12+ "strconv"
13+ "strings"
14+ "testing"
15+ "time"
16+
17+ "github.com/stretchr/testify/assert"
18+ "github.com/stretchr/testify/mock"
19+ "github.com/stretchr/testify/require"
20+
21+ "bnema.dev/ermite/internal/domain"
22+ "bnema.dev/ermite/internal/ports"
23+ "bnema.dev/ermite/internal/ports/mocks"
24+)
25+
26+// artifactBlobMock uses generated port mocks with byte storage in callbacks.
27+// Real-disk end-to-end behavior is exercised by internal/app's artifact rig.
28+func artifactBlobMock(t *testing.T, failPart bool) (*mocks.BlobStore, *int, *int) {
29+ t.Helper()
30+ b := mocks.NewBlobStore(t)
31+ objects := map[string][]byte{}
32+ active := map[string]ports.MultipartInfo{}
33+ puts, multiparts := new(int), new(int)
34+ b.EXPECT().Put(mock.Anything, mock.Anything, mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, key string, r io.Reader, size int64) (ports.BlobInfo, error) {
35+ *puts++
36+ data, err := io.ReadAll(r)
37+ if err != nil {
38+ return ports.BlobInfo{}, err
39+ }
40+ if int64(len(data)) != size {
41+ return ports.BlobInfo{}, domain.ErrInvalid
42+ }
43+ objects[key] = data
44+ return ports.BlobInfo{Key: key, Size: size}, nil
45+ }).Maybe()
46+ b.EXPECT().Open(mock.Anything, mock.Anything, mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, key string, offset, length int64) (io.ReadCloser, ports.BlobInfo, error) {
47+ data, ok := objects[key]
48+ if !ok {
49+ return nil, ports.BlobInfo{}, domain.ErrNotFound
50+ }
51+ info := ports.BlobInfo{Key: key, Size: int64(len(data))}
52+ if offset >= info.Size {
53+ return nil, info, domain.ErrOutOfRange
54+ }
55+ end := info.Size
56+ if length >= 0 {
57+ end = min(end, offset+length)
58+ }
59+ return io.NopCloser(bytes.NewReader(data[offset:end])), info, nil
60+ }).Maybe()
61+ b.EXPECT().Stat(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, key string) (ports.BlobInfo, error) {
62+ data, ok := objects[key]
63+ if !ok {
64+ return ports.BlobInfo{}, domain.ErrNotFound
65+ }
66+ return ports.BlobInfo{Key: key, Size: int64(len(data))}, nil
67+ }).Maybe()
68+ b.EXPECT().Delete(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, key string) error { delete(objects, key); return nil }).Maybe()
69+ b.EXPECT().List(mock.Anything, mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, prefix string, fn func(ports.BlobInfo) error) error {
70+ for key, data := range objects {
71+ if strings.HasPrefix(key, prefix) {
72+ if err := fn(ports.BlobInfo{Key: key, Size: int64(len(data))}); err != nil {
73+ return err
74+ }
75+ }
76+ }
77+ return nil
78+ }).Maybe()
79+ b.EXPECT().ListMultipart(mock.Anything, mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, prefix string, fn func(ports.MultipartInfo) error) error {
80+ for key, m := range active {
81+ if strings.HasPrefix(key, prefix) {
82+ if err := fn(m); err != nil {
83+ return err
84+ }
85+ }
86+ }
87+ return nil
88+ }).Maybe()
89+ b.EXPECT().AbortMultipart(mock.Anything, mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, key, id string) error { delete(active, key); return nil }).Maybe()
90+ b.EXPECT().NewMultipart(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, key string) (ports.MultipartUpload, error) {
91+ *multiparts++
92+ active[key] = ports.MultipartInfo{Key: key, ID: "upload"}
93+ u := mocks.NewMultipartUpload(t)
94+ var data []byte
95+ u.EXPECT().UploadPart(mock.Anything, mock.Anything, mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, _ int, r io.Reader, size int64) error {
96+ if failPart {
97+ return errors.New("part failed")
98+ }
99+ part, err := io.ReadAll(r)
100+ data = append(data, part...)
101+ return err
102+ }).Maybe()
103+ u.EXPECT().Complete(mock.Anything).RunAndReturn(func(context.Context) (ports.BlobInfo, error) {
104+ objects[key] = data
105+ delete(active, key)
106+ return ports.BlobInfo{Key: key, Size: int64(len(data))}, nil
107+ }).Maybe()
108+ u.EXPECT().Abort(mock.Anything).RunAndReturn(func(context.Context) error { delete(active, key); return nil }).Maybe()
109+ return u, nil
110+ }).Maybe()
111+ return b, puts, multiparts
112+}
113+
114+func TestArtifactBlobAssembly(t *testing.T) {
115+ for _, size := range []int{0, 1, ports.BlobPartSize, ports.BlobPartSize + 1, 2*ports.BlobPartSize + 17} {
116+ t.Run(strconv.Itoa(size), func(t *testing.T) {
117+ ctx := t.Context()
118+ b, puts, multiparts := artifactBlobMock(t, false)
119+ c := &CI{cfg: CIConfig{Blobs: b}}
120+ a := domain.CIArtifact{RunID: 1, ID: 2, Store: domain.CIArtifactStoreBlob}
121+ data := bytes.Repeat([]byte("x"), size)
122+ // Parts straddle block boundaries, with an empty block in the list.
123+ cut := min(size, 3<<20)
124+ _, err := b.Put(ctx, a.BlockKey("a"), bytes.NewReader(data[:cut]), int64(cut))
125+ require.NoError(t, err)
126+ _, err = b.Put(ctx, a.BlockKey("b"), bytes.NewReader(data[cut:]), int64(size-cut))
127+ require.NoError(t, err)
128+ _, err = b.Put(ctx, a.BlockKey("empty"), bytes.NewReader(nil), 0)
129+ require.NoError(t, err)
130+ *puts = 0
131+ n, digest, err := c.assembleBlobs(ctx, a, []string{"a", "empty", "b"}, int64(size))
132+ require.NoError(t, err)
133+ assert.EqualValues(t, size, n)
134+ sum := sha256.Sum256(data)
135+ assert.Equal(t, "sha256:"+hex.EncodeToString(sum[:]), digest)
136+ if size <= ports.BlobPartSize {
137+ assert.Equal(t, 1, *puts)
138+ assert.Zero(t, *multiparts)
139+ } else {
140+ assert.Zero(t, *puts)
141+ assert.Equal(t, 1, *multiparts)
142+ }
143+ if size > 0 {
144+ r, _, err := b.Open(ctx, a.BlobKey(), 0, -1)
145+ require.NoError(t, err)
146+ got, err := io.ReadAll(r)
147+ require.NoError(t, err)
148+ require.NoError(t, r.Close())
149+ assert.Equal(t, data, got)
150+ } else {
151+ info, err := b.Stat(ctx, a.BlobKey())
152+ require.NoError(t, err)
153+ assert.Zero(t, info.Size)
154+ }
155+ // No legacy reader is needed for a blob download, including EOF resumes.
156+ a.Size = int64(size)
157+ var got []byte
158+ require.NoError(t, c.streamArtifact(ctx, a, 0, func(ch ports.ArtifactChunk) error { got = append(got, ch.Data...); return nil }))
159+ assert.Equal(t, data, append([]byte{}, got...))
160+ require.NoError(t, c.streamArtifact(ctx, a, a.Size, func(ports.ArtifactChunk) error { return nil }))
161+ c.removeArtifactFiles(ctx, []domain.CIArtifact{a})
162+ _, err = b.Stat(ctx, a.BlobKey())
163+ assert.ErrorIs(t, err, domain.ErrNotFound)
164+ require.NoError(t, b.List(ctx, a.BlobKey()+".blocks/", func(ports.BlobInfo) error { t.Error("block survived deletion"); return nil }))
165+ })
166+ }
167+}
168+
169+func TestArtifactBlobAssemblyFailureAborts(t *testing.T) {
170+ for _, failPart := range []bool{false, true} {
171+ t.Run(map[bool]string{false: "missing block", true: "failed part"}[failPart], func(t *testing.T) {
172+ b, _, multiparts := artifactBlobMock(t, failPart)
173+ c := &CI{cfg: CIConfig{Blobs: b}}
174+ a := domain.CIArtifact{RunID: 1, ID: 2}
175+ data := bytes.Repeat([]byte("a"), ports.BlobPartSize)
176+ _, err := b.Put(t.Context(), a.BlockKey("a"), bytes.NewReader(data), int64(len(data)))
177+ require.NoError(t, err)
178+ _, _, err = c.assembleBlobs(t.Context(), a, []string{"a", "missing"}, ports.BlobPartSize+1)
179+ require.Error(t, err)
180+ assert.Equal(t, 1, *multiparts)
181+ _, err = b.Stat(t.Context(), a.BlobKey())
182+ assert.ErrorIs(t, err, domain.ErrNotFound)
183+ require.NoError(t, b.ListMultipart(t.Context(), "ci/", func(ports.MultipartInfo) error { t.Error("multipart survived failure"); return nil }))
184+ })
185+ }
186+}
187+
188+func TestArtifactPutUnsizedBound(t *testing.T) {
189+ b, _, _ := artifactBlobMock(t, false)
190+ c := &CI{cfg: CIConfig{Blobs: b}}
191+ for _, size := range []int{0, 1, domain.CIArtifactMaxBlockSize, domain.CIArtifactMaxBlockSize + 1} {
192+ n, err := c.putUnsized(t.Context(), "ci/1/2", bytes.NewReader(make([]byte, size)))
193+ if size > domain.CIArtifactMaxBlockSize {
194+ assert.ErrorIs(t, err, domain.ErrInvalid)
195+ assert.Contains(t, err.Error(), "a block has at most 8 MiB")
196+ } else {
197+ require.NoError(t, err)
198+ assert.EqualValues(t, size, n)
199+ }
200+ }
201+ ctx, cancel := context.WithCancel(t.Context())
202+ cancel()
203+ _, err := c.putUnsized(ctx, "ci/1/cancelled", bytes.NewReader([]byte("x")))
204+ assert.ErrorIs(t, err, context.Canceled)
205+ _, err = b.Stat(t.Context(), "ci/1/cancelled")
206+ assert.ErrorIs(t, err, domain.ErrNotFound)
207+}
208+
209+func TestArtifactLegacyStream(t *testing.T) {
210+ a := domain.CIArtifact{RunID: 1, ID: 2, Size: 6}
211+ legacy := mocks.NewCIArtifactFiles(t)
212+ legacy.EXPECT().Open(mock.Anything, artifactRef(a), int64(2)).RunAndReturn(func(context.Context, ports.CIArtifactRef, int64) (io.ReadCloser, error) {
213+ return io.NopCloser(bytes.NewReader([]byte("gacy"))), nil
214+ }).Twice()
215+ c := &CI{cfg: CIConfig{ArtifactFiles: legacy}}
216+ for _, store := range []domain.CIArtifactStore{domain.CIArtifactStoreLegacy, "disk"} {
217+ a.Store = store
218+ var got []byte
219+ require.NoError(t, c.streamArtifact(t.Context(), a, 2, func(ch ports.ArtifactChunk) error { got = append(got, ch.Data...); return nil }))
220+ assert.Equal(t, "gacy", string(got))
221+ }
222+ c.cfg.ArtifactFiles = nil
223+ assert.ErrorIs(t, c.streamArtifact(t.Context(), a, 0, func(ports.ArtifactChunk) error { return nil }), domain.ErrNotFound)
224+}
225+
226+func TestArtifactBlobOrphanSweep(t *testing.T) {
227+ ctx := t.Context()
228+ b := mocks.NewBlobStore(t)
229+ s := mocks.NewCIArtifactStore(t)
230+ c := &CI{cfg: CIConfig{Blobs: b, Artifacts: s, Clock: stillClock{}}}
231+ now := c.now()
232+ old := now.Add(-25 * time.Hour)
233+ pending := domain.CIArtifact{RunID: 1, ID: 2, Store: domain.CIArtifactStoreBlob, State: domain.CIArtifactPending}
234+ committed := pending
235+ committed.ID = 3
236+ committed.State = domain.CIArtifactCommitted
237+ missing := pending
238+ missing.ID = 4
239+ b.EXPECT().List(mock.Anything, "ci/", mock.Anything).RunAndReturn(func(_ context.Context, _ string, fn func(ports.BlobInfo) error) error {
240+ for _, info := range []ports.BlobInfo{
241+ {Key: missing.BlobKey(), ModTime: now}, // young: no lookup, no deletion
242+ {Key: missing.BlobKey(), ModTime: old},
243+ {Key: pending.BlockKey("a"), ModTime: old},
244+ {Key: committed.BlockKey("a"), ModTime: old},
245+ {Key: committed.BlobKey(), ModTime: old},
246+ } {
247+ require.NoError(t, fn(info))
248+ }
249+ return nil
250+ })
251+ s.EXPECT().GetArtifact(mock.Anything, missing.ID).Return(domain.CIArtifact{}, domain.ErrNotFound).Once()
252+ s.EXPECT().GetArtifact(mock.Anything, pending.ID).Return(pending, nil).Once()
253+ s.EXPECT().GetArtifact(mock.Anything, committed.ID).Return(committed, nil).Twice()
254+ b.EXPECT().Delete(mock.Anything, missing.BlobKey()).Return(nil).Once()
255+ b.EXPECT().Delete(mock.Anything, committed.BlockKey("a")).Return(nil).Once()
256+ b.EXPECT().ListMultipart(mock.Anything, "ci/", mock.Anything).RunAndReturn(func(_ context.Context, _ string, fn func(ports.MultipartInfo) error) error {
257+ require.NoError(t, fn(ports.MultipartInfo{Key: missing.BlobKey(), ID: "old", Started: old}))
258+ require.NoError(t, fn(ports.MultipartInfo{Key: missing.BlobKey(), ID: "young", Started: now}))
259+ return nil
260+ })
261+ b.EXPECT().AbortMultipart(mock.Anything, missing.BlobKey(), "old").Return(nil).Once()
262+ c.sweepArtifactBlobOrphans(ctx)
263+}
264+
265+func TestArtifactRemoveRunsWithoutLegacyReader(t *testing.T) {
266+ ctx := t.Context()
267+ b, _, _ := artifactBlobMock(t, false)
268+ c := &CI{cfg: CIConfig{Blobs: b}}
269+ a := domain.CIArtifact{RunID: 1, ID: 2}
270+ other := a
271+ other.RunID = 2
272+ for _, key := range []string{a.BlobKey(), a.BlockKey("a"), other.BlobKey()} {
273+ _, err := b.Put(ctx, key, bytes.NewReader([]byte("x")), 1)
274+ require.NoError(t, err)
275+ }
276+ _, err := b.NewMultipart(ctx, a.BlobKey())
277+ require.NoError(t, err)
278+ require.NoError(t, c.removeArtifactRuns(ctx, []int64{1}))
279+ _, err = b.Stat(ctx, a.BlobKey())
280+ assert.ErrorIs(t, err, domain.ErrNotFound)
281+ _, err = b.Stat(ctx, a.BlockKey("a"))
282+ assert.ErrorIs(t, err, domain.ErrNotFound)
283+ _, err = b.Stat(ctx, other.BlobKey())
284+ require.NoError(t, err)
285+ require.NoError(t, b.ListMultipart(ctx, "ci/", func(ports.MultipartInfo) error { t.Error("upload survived run deletion"); return nil }))
286+}
287+
288+func TestArtifactBlobIDs(t *testing.T) {
289+ a := domain.CIArtifact{RunID: 7, ID: 42}
290+ for _, key := range []string{a.BlobKey(), a.BlockKey("abcdef.0123")} {
291+ run, id, block, ok := artifactBlobIDs(key)
292+ assert.True(t, ok)
293+ assert.EqualValues(t, 7, run)
294+ assert.EqualValues(t, 42, id)
295+ assert.Equal(t, strings.Contains(key, ".blocks/"), block)
296+ }
297+ for _, key := range []string{"release/1", "ci/1/2", a.BlobKey() + ".blocks/", a.BlobKey() + "/wrong"} {
298+ _, _, _, ok := artifactBlobIDs(key)
299+ assert.False(t, ok, key)
300+ }
301+}
A internal/usecase/ci_artifacts_slots_test.go +179 -0
@@ -0,0 +1,179 @@
1+// SPDX-License-Identifier: AGPL-3.0-or-later
2+
3+package usecase
4+
5+import (
6+ "bytes"
7+ "context"
8+ "errors"
9+ "io"
10+ "strings"
11+ "sync"
12+ "testing"
13+ "time"
14+
15+ "github.com/stretchr/testify/mock"
16+ "github.com/stretchr/testify/require"
17+
18+ "bnema.dev/ermite/internal/domain"
19+ "bnema.dev/ermite/internal/ports"
20+ "bnema.dev/ermite/internal/ports/mocks"
21+)
22+
23+// gateReader signals its first Read on started, then blocks until release is
24+// closed or ctx ends; it then returns data (or err when set).
25+type gateReader struct {
26+ ctx context.Context
27+ started chan struct{}
28+ release chan struct{}
29+ data []byte
30+ err error
31+ once sync.Once
32+ reads int
33+ r *bytes.Reader
34+}
35+
36+func (g *gateReader) Read(p []byte) (int, error) {
37+ g.reads++
38+ g.once.Do(func() {
39+ if g.started != nil {
40+ close(g.started)
41+ }
42+ })
43+ if g.release != nil {
44+ select {
45+ case <-g.release:
46+ case <-g.ctx.Done():
47+ return 0, g.ctx.Err()
48+ }
49+ }
50+ if g.err != nil {
51+ return 0, g.err
52+ }
53+ if g.r == nil {
54+ g.r = bytes.NewReader(g.data)
55+ }
56+ return g.r.Read(p)
57+}
58+
59+type slotFixture struct {
60+ c *CI
61+ secret string
62+ job domain.CIJob
63+ puts *int
64+}
65+
66+func newSlotFixture(t *testing.T, slots int) *slotFixture {
67+ t.Helper()
68+ repo, err := domain.ParseRepoName("proj")
69+ require.NoError(t, err)
70+ secret := domain.JobTokenPrefix + strings.Repeat("a", domain.CITokenRandomLen)
71+ hash := domain.HashTokenSecret(secret)
72+ j := domain.CIJob{ID: "1-0-0", RunID: 1, RunAttempt: 1, Repo: repo, TokenHash: hash, Phase: domain.CIPhaseRunning}
73+ a := domain.CIArtifact{ID: 7, Repo: repo, RunID: 1, Attempt: 1, JobID: j.ID, Assignment: hash, State: domain.CIArtifactPending}
74+
75+ store := mocks.NewCIStore(t)
76+ store.EXPECT().FindJobByToken(mock.Anything, hash, mock.Anything).Return(j, nil).Maybe()
77+ store.EXPECT().GetJob(mock.Anything, j.RunID, j.RunAttempt, j.ID).Return(j, nil).Maybe()
78+ arts := mocks.NewCIArtifactStore(t)
79+ arts.EXPECT().ArtifactUsage(mock.Anything, repo).Return(0, 1, nil).Maybe()
80+ arts.EXPECT().GetArtifact(mock.Anything, a.ID).Return(a, nil).Maybe()
81+ arts.EXPECT().ArtifactBlocks(mock.Anything, a.ID).Return(nil, nil).Maybe()
82+ arts.EXPECT().ApplyArtifacts(mock.Anything, mock.Anything).Return(nil).Maybe()
83+ clock := mocks.NewClock(t)
84+ clock.EXPECT().Now().Return(time.Unix(1000, 0)).Maybe()
85+ blobs, puts, _ := artifactBlobMock(t, false)
86+ c := &CI{
87+ cfg: CIConfig{
88+ Store: store, Artifacts: arts, Blobs: blobs, Clock: clock,
89+ ArtifactLimits: ports.ArtifactLimits{Quota: 1 << 30, MaxSize: 1 << 30},
90+ },
91+ uploadSlots: make(chan struct{}, slots),
92+ }
93+ return &slotFixture{c: c, secret: secret, job: j, puts: puts}
94+}
95+
96+func (f *slotFixture) upload(ctx context.Context, r io.Reader) (int64, error) {
97+ return f.c.UploadJobArtifactBlock(ctx, f.secret, f.job.ID, 7, []byte("blk"), r)
98+}
99+
100+func TestArtifactUploadSlotsBound(t *testing.T) {
101+ for _, mode := range []string{"success", "body error", "ctx cancel"} {
102+ t.Run(mode, func(t *testing.T) {
103+ f := newSlotFixture(t, 1)
104+ ctx, cancel := context.WithCancel(t.Context())
105+ defer cancel()
106+ hold := &gateReader{ctx: ctx, started: make(chan struct{}), release: make(chan struct{}), data: []byte("held")}
107+ if mode == "body error" {
108+ hold.err = errors.New("connection reset")
109+ }
110+ type result struct {
111+ n int64
112+ err error
113+ }
114+ done := make(chan result, 1)
115+ go func() { n, err := f.upload(ctx, hold); done <- result{n, err} }()
116+ <-hold.started
117+
118+ // The slot is taken: the next upload is refused at once, before
119+ // reading its body, and nothing is reserved or stored.
120+ refused := &gateReader{ctx: t.Context(), data: []byte("x")}
121+ _, err := f.upload(t.Context(), refused)
122+ require.ErrorIs(t, err, domain.ErrResourceExhausted)
123+ require.Zero(t, refused.reads)
124+ require.Zero(t, *f.puts)
125+ require.Zero(t, f.c.reservations.of(f.job.Repo).total)
126+
127+ switch mode {
128+ case "ctx cancel":
129+ cancel()
130+ default:
131+ close(hold.release)
132+ }
133+ res := <-done
134+ if mode == "success" {
135+ require.NoError(t, res.err)
136+ require.EqualValues(t, 4, res.n)
137+ } else {
138+ require.Error(t, res.err)
139+ }
140+ require.Empty(t, f.c.uploadSlots, "the slot is released")
141+
142+ n, err := f.upload(t.Context(), &gateReader{ctx: t.Context(), data: []byte("again")})
143+ require.NoError(t, err)
144+ require.EqualValues(t, 5, n)
145+ require.Empty(t, f.c.uploadSlots)
146+ })
147+ }
148+}
149+
150+func TestArtifactAssemblyTakesASlot(t *testing.T) {
151+ b, _, _ := artifactBlobMock(t, false)
152+ c := &CI{cfg: CIConfig{Blobs: b}, uploadSlots: make(chan struct{}, 1)}
153+ a := domain.CIArtifact{RunID: 1, ID: 2}
154+ _, err := b.Put(t.Context(), a.BlockKey("a"), strings.NewReader("hi"), 2)
155+ require.NoError(t, err)
156+
157+ // A full set of slots makes the assembly wait, not fail; it gives up
158+ // only when its context ends.
159+ c.uploadSlots <- struct{}{}
160+ ctx, cancel := context.WithCancel(t.Context())
161+ cancel()
162+ _, _, err = c.assembleBlobs(ctx, a, []string{"a"}, 2)
163+ require.ErrorIs(t, err, context.Canceled)
164+
165+ type result struct {
166+ n int64
167+ err error
168+ }
169+ done := make(chan result, 1)
170+ go func() {
171+ n, _, err := c.assembleBlobs(t.Context(), a, []string{"a"}, 2)
172+ done <- result{n, err}
173+ }()
174+ <-c.uploadSlots // free the slot: the waiting assembly proceeds
175+ res := <-done
176+ require.NoError(t, res.err)
177+ require.EqualValues(t, 2, res.n)
178+ require.Empty(t, c.uploadSlots)
179+}
M internal/usecase/ci_watch.go +4 -3
@@ -276,6 +276,9 @@ func (c *CI) sweep(ctx context.Context) {
276276 c.removeOrphans(ctx, "logs", "log directories", c.cfg.Logs.RunIDs, c.cfg.Logs.RemoveRuns)
277277 c.removeOrphans(ctx, "summaries", "summary directories", c.cfg.Summaries.RunIDs, c.cfg.Summaries.RemoveRuns)
278278 c.sweepArtifacts(ctx)
279+ if c.cfg.ReleaseSweeper != nil {
280+ c.cfg.ReleaseSweeper.Sweep(ctx)
281+ }
279282 }
280283
281284 // removeOrphans deletes what a file store (logs, summaries) holds for runs that
@@ -323,8 +326,6 @@ func (c *CI) RemoveLogs(ctx context.Context, runIDs []int64) error {
323326 c.locks.Delete(id)
324327 }
325328 err := errors.Join(c.cfg.Logs.RemoveRuns(ctx, runIDs), c.cfg.Summaries.RemoveRuns(ctx, runIDs))
326- if c.artifactsOn() {
327- err = errors.Join(err, c.cfg.ArtifactFiles.RemoveRuns(ctx, runIDs))
328- }
329+ err = errors.Join(err, c.removeArtifactRuns(ctx, runIDs))
329330 return err
330331 }
M internal/usecase/events.go +25 -0
@@ -171,6 +171,14 @@ func (e *Events) snapshot(ctx context.Context, ev domain.Event) bufferedEvent {
171171 return repoSnapshot(ev.Visibility, ev.CreatedAt)
172172 case domain.MirrorRecovered:
173173 return repoSnapshot(ev.Visibility, ev.CreatedAt)
174+ case domain.ReleasePublished:
175+ return repoSnapshot(ev.Visibility, ev.CreatedAt)
176+ case domain.ReleaseUpdated:
177+ return repoSnapshot(ev.Visibility, ev.CreatedAt)
178+ case domain.ReleaseDeleted:
179+ return repoSnapshot(ev.Visibility, ev.CreatedAt)
180+ case domain.ReleaseUnpublished:
181+ return repoSnapshot(ev.Visibility, ev.CreatedAt)
174182 case domain.PushEvent:
175183 if ev.CreatedAt.IsZero() {
176184 return bufferedEvent{} // repository unknown at publish time: nobody is told
@@ -531,6 +539,14 @@ func (e *Events) deliver(ctx context.Context, w *watch, be bufferedEvent, send f
531539 return e.deliverRepo(ctx, w, ev.Repo, false, be, out, send)
532540 case domain.MirrorRecovered:
533541 return e.deliverRepo(ctx, w, ev.Repo, false, be, out, send)
542+ case domain.ReleasePublished:
543+ return e.deliverRelease(ctx, w, ev.Release, be, out, send)
544+ case domain.ReleaseUpdated:
545+ return e.deliverRelease(ctx, w, ev.Release, be, out, send)
546+ case domain.ReleaseDeleted:
547+ return e.deliverRelease(ctx, w, ev.Release, be, out, send)
548+ case domain.ReleaseUnpublished:
549+ return e.deliverRelease(ctx, w, ev.Release, be, out, send)
534550 case domain.PushEvent:
535551 return e.deliverRepo(ctx, w, ev.Repo, false, be, out, send)
536552 case domain.MergeRequestOpened:
@@ -646,6 +662,15 @@ func (e *Events) deliverRepo(ctx context.Context, w *watch, name domain.RepoName
646662 return send(out)
647663 }
648664
665+// deliverRelease sends a release event. The event of a draft goes only to
666+// release writers: a draft is invisible to every other reader.
667+func (e *Events) deliverRelease(ctx context.Context, w *watch, rel domain.Release, be bufferedEvent, out ports.WatchEvent, send func(ports.WatchEvent) error) error {
668+ if rel.Draft && !releaseWriter(w.p, rel.Repo) {
669+ return nil
670+ }
671+ return e.deliverRepo(ctx, w, rel.Repo, false, be, out, send)
672+}
673+
649674 // isIssueEvent reports whether ev is about an issue.
650675 func isIssueEvent(ev domain.Event) bool {
651676 switch ev.(type) {
M internal/usecase/events_test.go +95 -0
@@ -1034,3 +1034,98 @@ func TestIssueReferencedRepoOnly(t *testing.T) {
10341034 }
10351035 })
10361036 }
1037+
1038+func TestDraftReleaseEventsReachOnlyReleaseWriters(t *testing.T) {
1039+ t.Parallel()
1040+ synctest.Test(t, func(t *testing.T) {
1041+ r := newEventRig(t, 64)
1042+ pub := r.repo("pub", domain.VisibilityPublic)
1043+ other := r.repo("other", domain.VisibilityPublic)
1044+ ctx, cancel := context.WithCancel(context.Background())
1045+ defer cancel()
1046+
1047+ job := func(repo domain.RepoName, level string) domain.Principal {
1048+ return domain.RepoTokenPrincipal(domain.CIRepoToken{ID: "j", Repo: repo, Permissions: map[string]string{"contents": level}})
1049+ }
1050+ writerBranch := scopedToken(domain.ScopeReleaseWrite, []string{"agent/"}, "pub")
1051+ cases := []struct {
1052+ name string
1053+ p domain.Principal
1054+ want bool
1055+ }{
1056+ {"anonymous", domain.Anonymous(), false},
1057+ {"reader", tokenPrincipal(domain.ScopeRepoRead), false},
1058+ {"branch-limited writer", writerBranch, false},
1059+ {"job token of another repo", job(other, "write"), false},
1060+ {"job token contents:read", job(pub, "read"), false},
1061+ {"release:write token", tokenPrincipal(domain.ScopeReleaseWrite, "pub"), true},
1062+ {"job token contents:write", job(pub, "write"), true},
1063+ }
1064+
1065+ anchor := r.watch(ctx, domain.Root(), ports.WatchParams{})
1066+ synctest.Wait()
1067+ r.push(pub, 1)
1068+ anchorID := anchor.next(t).ID
1069+
1070+ draft := domain.Release{ID: 1, Repo: pub, Tag: "d", Draft: true}
1071+ published := domain.Release{ID: 2, Repo: pub, Tag: "v1"}
1072+ events := func(rel domain.Release) []domain.Event {
1073+ return []domain.Event{
1074+ domain.ReleasePublished{Release: rel, At: now, CreatedAt: now, Visibility: domain.VisibilityPublic},
1075+ domain.ReleaseUpdated{Release: rel, At: now, CreatedAt: now, Visibility: domain.VisibilityPublic},
1076+ domain.ReleaseDeleted{Release: rel, At: now, CreatedAt: now, Visibility: domain.VisibilityPublic},
1077+ domain.ReleaseUnpublished{Release: rel, At: now, CreatedAt: now, Visibility: domain.VisibilityPublic},
1078+ }
1079+ }
1080+ check := func(t *testing.T, w *watcher, want bool) {
1081+ t.Helper()
1082+ if want {
1083+ for _, ev := range events(draft) {
1084+ require.Equal(t, ev.EventKind(), w.next(t).Event.EventKind())
1085+ }
1086+ }
1087+ for _, ev := range events(published) {
1088+ e := w.next(t).Event
1089+ require.Equal(t, ev.EventKind(), e.EventKind())
1090+ require.EqualValues(t, 2, releaseOf(e))
1091+ }
1092+ w.quiet(t, r, pub)
1093+ }
1094+
1095+ var live []*watcher
1096+ for _, tc := range cases {
1097+ live = append(live, r.watch(ctx, tc.p, ports.WatchParams{}))
1098+ }
1099+ synctest.Wait()
1100+ for _, ev := range events(draft) {
1101+ r.feed <- ev
1102+ }
1103+ for _, ev := range events(published) {
1104+ r.feed <- ev
1105+ }
1106+ for i, tc := range cases {
1107+ t.Log("live", tc.name)
1108+ check(t, live[i], tc.want)
1109+ }
1110+ for _, tc := range cases {
1111+ t.Log("replay", tc.name)
1112+ w := r.watch(ctx, tc.p, ports.WatchParams{ResumeAfter: anchorID})
1113+ check(t, w, tc.want)
1114+ }
1115+ })
1116+}
1117+
1118+// releaseOf returns the id of the release an event is about.
1119+func releaseOf(ev domain.Event) int64 {
1120+ switch ev := ev.(type) {
1121+ case domain.ReleasePublished:
1122+ return ev.Release.ID
1123+ case domain.ReleaseUpdated:
1124+ return ev.Release.ID
1125+ case domain.ReleaseDeleted:
1126+ return ev.Release.ID
1127+ case domain.ReleaseUnpublished:
1128+ return ev.Release.ID
1129+ }
1130+ return 0
1131+}
M internal/usecase/pushes.go +21 -14
@@ -17,20 +17,21 @@ import (
1717
1818 // Pushes are the use cases around a Git push.
1919 type Pushes struct {
20- links ports.IssueLinkTrigger
21- mirror ports.MirrorTrigger // nil: no outbound mirror
22- meta ports.RepoMetaStore
23- repos *Repos
24- git ports.GitRepos
25- bus ports.EventBus
26- clock ports.Clock
27- maint ports.MaintenanceScheduler
28- auth ports.Revalidator
29- fresh ports.RepoFreshener // nil: the reader needs no telling
30- ci ports.WorkflowTrigger // nil: no CI
31- trees ports.PushTreeReader // nil: pushes by job tokens are refused
32- tips ports.CIRefTipStore // nil: who pushes a ref is not recorded
33- mergeab ports.MergeabilityTrigger // nil: mergeability is not tracked
20+ releases ports.ReleaseTagTrigger
21+ links ports.IssueLinkTrigger
22+ mirror ports.MirrorTrigger // nil: no outbound mirror
23+ meta ports.RepoMetaStore
24+ repos *Repos
25+ git ports.GitRepos
26+ bus ports.EventBus
27+ clock ports.Clock
28+ maint ports.MaintenanceScheduler
29+ auth ports.Revalidator
30+ fresh ports.RepoFreshener // nil: the reader needs no telling
31+ ci ports.WorkflowTrigger // nil: no CI
32+ trees ports.PushTreeReader // nil: pushes by job tokens are refused
33+ tips ports.CIRefTipStore // nil: who pushes a ref is not recorded
34+ mergeab ports.MergeabilityTrigger // nil: mergeability is not tracked
3435 }
3536
3637 var (
@@ -45,6 +46,9 @@ func NewPushes(meta ports.RepoMetaStore, repos *Repos, git ports.GitRepos, bus p
4546 return &Pushes{meta: meta, repos: repos, git: git, bus: bus, clock: clock, maint: maint, auth: auth}
4647 }
4748
49+// WithReleases enables synchronous release tag-deletion handling before the first push.
50+func (u *Pushes) WithReleases(t ports.ReleaseTagTrigger) *Pushes { u.releases = t; return u }
51+
4852 // WithIssueLinks enables synchronous best-effort issue indexing before the first push.
4953 func (u *Pushes) WithIssueLinks(t ports.IssueLinkTrigger) *Pushes { u.links = t; return u }
5054
@@ -272,6 +276,9 @@ func (u *Pushes) Record(ctx context.Context, p domain.Principal, name domain.Rep
272276 if u.links != nil {
273277 u.links.OnBranchesMoved(ctx, name, updates)
274278 }
279+ if u.releases != nil {
280+ u.releases.OnRefsMoved(ctx, name, updates)
281+ }
275282 u.bus.Publish(ctx, domain.PushEvent{
276283 Repo: name, Updates: append([]domain.RefUpdate(nil), updates...), At: now,
277284 Visibility: repo.Visibility, CreatedAt: repo.CreatedAt,
A internal/usecase/releases.go +494 -0
@@ -0,0 +1,494 @@
1+// SPDX-License-Identifier: AGPL-3.0-or-later
2+
3+package usecase
4+
5+import (
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.
20+type 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+
35+var _ ports.ReleaseService = (*Releases)(nil)
36+var _ ports.ReleaseTagTrigger = (*Releases)(nil)
37+var _ ports.ReleaseRepoCleaner = (*Releases)(nil)
38+var _ 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.
42+func 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+
55+func (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+}
62+func 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+}
65+func (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+}
81+func (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+}
103+func (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+}
111+func (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+}
141+func (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+}
157+func (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.
174+func (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.
223+func (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.
265+func (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.
325+func (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.
396+func (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.
429+func (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.
471+func (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.
480+func (r *Releases) OnRepoDeleted(ctx context.Context, _ domain.RepoName, assets []domain.ReleaseAsset) {
481+ r.removeAssetBlobs(ctx, assets)
482+}
483+func (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+}
A internal/usecase/releases_assets.go +441 -0
@@ -0,0 +1,441 @@
1+// SPDX-License-Identifier: AGPL-3.0-or-later
2+
3+package usecase
4+
5+import (
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.
21+func (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+}
197+func 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.
212+type releaseContextReader struct {
213+ ctx context.Context
214+ reader io.Reader
215+}
216+
217+func (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+
224+func 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+}
235+func (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.
254+func (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.
273+func (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.
285+func (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.
321+func (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.
370+func (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.
390+func (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 func() { _ = 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.
435+func (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+}
A internal/usecase/releases_assets_test.go +308 -0
@@ -0,0 +1,308 @@
1+// SPDX-License-Identifier: AGPL-3.0-or-later
2+
3+package usecase_test
4+
5+import (
6+ "bytes"
7+ "context"
8+ "crypto/sha256"
9+ "errors"
10+ "fmt"
11+ "io"
12+ "sync"
13+ "testing"
14+ "time"
15+
16+ "bnema.dev/ermite/internal/domain"
17+ "bnema.dev/ermite/internal/ports"
18+ "bnema.dev/ermite/internal/ports/mocks"
19+ "bnema.dev/ermite/internal/usecase"
20+ "github.com/stretchr/testify/mock"
21+ "github.com/stretchr/testify/require"
22+)
23+
24+func (r *releaseRig) upload(t *testing.T, rel domain.Release, n string, data []byte) domain.ReleaseAsset {
25+ t.Helper()
26+ a, err := r.svc.UploadAsset(r.ctx, domain.Root(), r.repo, ports.UploadAssetParams{ReleaseID: rel.ID, Name: n, Size: int64(len(data))}, bytes.NewReader(data))
27+ require.NoError(t, err)
28+ return a
29+}
30+func TestReleaseAssetRoundTrip(t *testing.T) {
31+ r := newReleaseRig(t, ports.ReleaseLimits{AssetMax: 10, Quota: 10})
32+ rel := r.create(t, "v1", false)
33+ a := r.upload(t, rel, "binary", []byte("0123456789"))
34+ require.Equal(t, fmt.Sprintf("sha256:%x", sha256.Sum256([]byte("0123456789"))), a.Digest)
35+ usage, err := r.svc.Usage(r.ctx, domain.Anonymous(), r.repo)
36+ require.NoError(t, err)
37+ require.EqualValues(t, 10, usage.Used)
38+ for _, tc := range []struct {
39+ offset, length int64
40+ want string
41+ err error
42+ }{{0, -1, "0123456789", nil}, {3, 4, "3456", nil}, {9, 1, "9", nil}, {10, -1, "", domain.ErrOutOfRange}, {11, -1, "", domain.ErrOutOfRange}} {
43+ var got bytes.Buffer
44+ metadata := false
45+ err := r.svc.DownloadAsset(r.ctx, domain.Anonymous(), r.repo, ports.AssetRef{Latest: true, Name: a.Name}, tc.offset, tc.length, func(c ports.AssetChunk) error {
46+ if c.Asset != nil {
47+ require.False(t, metadata)
48+ metadata = true
49+ require.Equal(t, a.ID, c.Asset.ID)
50+ } else {
51+ require.True(t, metadata)
52+ got.Write(c.Data)
53+ }
54+ return nil
55+ })
56+ require.ErrorIs(t, err, tc.err)
57+ require.Equal(t, tc.want, got.String())
58+ }
59+ _, err = r.svc.UploadAsset(r.ctx, domain.Root(), r.repo, ports.UploadAssetParams{ReleaseID: rel.ID, Name: "large", Size: 11}, bytes.NewReader(nil))
60+ require.ErrorIs(t, err, domain.ErrResourceExhausted)
61+ _, err = r.svc.UploadAsset(r.ctx, domain.Root(), r.repo, ports.UploadAssetParams{ReleaseID: rel.ID, Name: "quota", Size: 1}, bytes.NewReader([]byte("x")))
62+ require.ErrorIs(t, err, domain.ErrResourceExhausted)
63+ _, err = r.svc.UploadAsset(r.ctx, domain.Root(), r.repo, ports.UploadAssetParams{ReleaseID: rel.ID, Name: a.Name, Size: 0}, bytes.NewReader(nil))
64+ require.ErrorIs(t, err, domain.ErrAlreadyExists)
65+ replacement, err := r.svc.UploadAsset(r.ctx, domain.Root(), r.repo, ports.UploadAssetParams{ReleaseID: rel.ID, Name: a.Name, Size: 2, Replace: true}, bytes.NewReader([]byte("hi")))
66+ require.NoError(t, err)
67+ require.NotEqual(t, a.ID, replacement.ID)
68+ _, err = r.blobs.Stat(r.ctx, a.BlobKey())
69+ require.ErrorIs(t, err, domain.ErrNotFound)
70+ renamed := "new"
71+ label := "label"
72+ replacement, err = r.svc.UpdateAsset(r.ctx, domain.Root(), r.repo, replacement.ID, ports.UpdateAssetParams{Name: &renamed, Label: &label})
73+ require.NoError(t, err)
74+ require.Equal(t, label, replacement.Label)
75+ list, err := r.svc.ListAssets(r.ctx, domain.Anonymous(), r.repo, rel.ID)
76+ require.NoError(t, err)
77+ require.Len(t, list, 1)
78+ require.Equal(t, renamed, list[0].Name)
79+ require.NoError(t, r.svc.DeleteAsset(r.ctx, domain.Root(), r.repo, replacement.ID))
80+ usage, err = r.svc.Usage(r.ctx, domain.Root(), r.repo)
81+ require.NoError(t, err)
82+ require.Zero(t, usage.Used)
83+}
84+func TestReleaseAssetCountBoundary(t *testing.T) {
85+ r := newReleaseRig(t, ports.ReleaseLimits{})
86+ rel := r.create(t, "v1", false)
87+ // Seed 999 records in one batch, then exercise the exact public boundary.
88+ used, v, err := r.store.Releases().ReleaseUsage(r.ctx, r.repo)
89+ require.NoError(t, err)
90+ w := ports.ReleaseWrite{Repo: r.repo, Used: used, UsageVersion: v}
91+ for range 999 {
92+ id, err := r.store.Releases().NextReleaseID(r.ctx)
93+ require.NoError(t, err)
94+ w.PutAssets = append(w.PutAssets, domain.ReleaseAsset{ID: id, ReleaseID: rel.ID, Repo: r.repo, Name: fmt.Sprint(id), State: domain.ReleaseAssetUploaded})
95+ }
96+ require.NoError(t, r.store.Releases().ApplyReleases(r.ctx, w))
97+ r.upload(t, rel, "thousandth", nil)
98+ _, err = r.svc.UploadAsset(r.ctx, domain.Root(), r.repo, ports.UploadAssetParams{ReleaseID: rel.ID, Name: "1001"}, bytes.NewReader(nil))
99+ require.ErrorIs(t, err, domain.ErrResourceExhausted)
100+}
101+func TestReleaseAssetBadBodies(t *testing.T) {
102+ bodyErr := errors.New("reader failed")
103+ for _, size := range []int64{4, 9 << 20} {
104+ for _, kind := range []string{"short", "long", "error", "cancel"} {
105+ t.Run(fmt.Sprint(size, "/", kind), func(t *testing.T) {
106+ r := newReleaseRig(t, ports.ReleaseLimits{})
107+ rel := r.create(t, "v1", false)
108+ var body io.Reader
109+ switch kind {
110+ case "short":
111+ body = bytes.NewReader(make([]byte, size-1))
112+ case "long":
113+ body = bytes.NewReader(make([]byte, size+1))
114+ case "error":
115+ body = errReader{bodyErr}
116+ case "cancel":
117+ body = errReader{context.Canceled}
118+ }
119+ _, err := r.svc.UploadAsset(r.ctx, domain.Root(), r.repo, ports.UploadAssetParams{ReleaseID: rel.ID, Name: "bad", Size: size}, body)
120+ switch kind {
121+ case "error":
122+ require.ErrorIs(t, err, bodyErr)
123+ case "cancel":
124+ require.ErrorIs(t, err, context.Canceled)
125+ default:
126+ require.ErrorIs(t, err, domain.ErrInvalid)
127+ }
128+ all, err := r.store.Releases().AllAssets(r.ctx)
129+ require.NoError(t, err)
130+ require.Empty(t, all)
131+ u, err := r.svc.Usage(r.ctx, domain.Root(), r.repo)
132+ require.NoError(t, err)
133+ require.Zero(t, u.Used)
134+ require.NoError(t, r.blobs.List(r.ctx, "release/", func(ports.BlobInfo) error { t.Error("blob left behind"); return nil }))
135+ require.NoError(t, r.blobs.ListMultipart(r.ctx, "release/", func(ports.MultipartInfo) error { t.Error("multipart left behind"); return nil }))
136+ })
137+ }
138+ }
139+}
140+
141+type releaseBlobSpy struct {
142+ ports.BlobStore
143+ put, multipart int
144+ parts []int64
145+}
146+
147+func (s *releaseBlobSpy) Put(ctx context.Context, key string, b io.Reader, size int64) (ports.BlobInfo, error) {
148+ s.put++
149+ return s.BlobStore.Put(ctx, key, b, size)
150+}
151+func (s *releaseBlobSpy) NewMultipart(ctx context.Context, key string) (ports.MultipartUpload, error) {
152+ s.multipart++
153+ u, e := s.BlobStore.NewMultipart(ctx, key)
154+ return &releasePartSpy{MultipartUpload: u, spy: s}, e
155+}
156+
157+type releasePartSpy struct {
158+ ports.MultipartUpload
159+ spy *releaseBlobSpy
160+}
161+
162+func (s *releasePartSpy) UploadPart(ctx context.Context, n int, b io.Reader, size int64) error {
163+ s.spy.parts = append(s.spy.parts, size)
164+ return s.MultipartUpload.UploadPart(ctx, n, b, size)
165+}
166+func TestReleaseAssetUploadPaths(t *testing.T) {
167+ for _, size := range []int{1 << 20, 9 << 20} {
168+ t.Run(fmt.Sprint(size), func(t *testing.T) {
169+ r := newReleaseRig(t, ports.ReleaseLimits{})
170+ rel := r.create(t, "v1", false)
171+ spy := &releaseBlobSpy{BlobStore: r.blobs}
172+ bus := mocks.NewEventBus(t)
173+ bus.EXPECT().Publish(mock.Anything, mock.Anything).Return().Maybe()
174+ r.svc = usecase.NewReleases(r.store.Releases(), spy, r.store, r.reader, r.tags, r.pushes, bus, r.clock, ports.ReleaseLimits{})
175+ data := bytes.Repeat([]byte("z"), size)
176+ a := r.upload(t, rel, "blob", data)
177+ require.Equal(t, fmt.Sprintf("sha256:%x", sha256.Sum256(data)), a.Digest)
178+ if size == 1<<20 {
179+ require.Equal(t, 1, spy.put)
180+ require.Zero(t, spy.multipart)
181+ } else {
182+ require.Zero(t, spy.put)
183+ require.Equal(t, 1, spy.multipart)
184+ require.Equal(t, []int64{8 << 20, 1 << 20}, spy.parts)
185+ }
186+ })
187+ }
188+}
189+func TestReleaseAssetStaleReplacementAndReleaseDelete(t *testing.T) {
190+ r := newReleaseRig(t, ports.ReleaseLimits{})
191+ rel := r.create(t, "v1", false)
192+ a := r.upload(t, rel, "pending", []byte("old"))
193+ current, err := r.store.Releases().GetAsset(r.ctx, a.ID)
194+ require.NoError(t, err)
195+ current.State = domain.ReleaseAssetUploading
196+ current.UpdatedAt = now.Add(-61 * time.Second)
197+ used, v, err := r.store.Releases().ReleaseUsage(r.ctx, r.repo)
198+ require.NoError(t, err)
199+ require.NoError(t, r.store.Releases().ApplyReleases(r.ctx, ports.ReleaseWrite{Repo: r.repo, Used: used, UsageVersion: v, PutAssets: []domain.ReleaseAsset{current}}))
200+ b := r.upload(t, rel, "pending", []byte("new"))
201+ require.NotEqual(t, a.ID, b.ID)
202+ require.NoError(t, r.svc.DeleteRelease(r.ctx, domain.Root(), r.repo, rel.ID))
203+ _, err = r.blobs.Stat(r.ctx, b.BlobKey())
204+ require.ErrorIs(t, err, domain.ErrNotFound)
205+ u, err := r.svc.Usage(r.ctx, domain.Root(), r.repo)
206+ require.NoError(t, err)
207+ require.Zero(t, u.Used)
208+}
209+
210+type releaseBlockingReader struct {
211+ entered chan<- struct{}
212+ release <-chan struct{}
213+}
214+
215+func (b releaseBlockingReader) Read([]byte) (int, error) {
216+ b.entered <- struct{}{}
217+ <-b.release
218+ return 0, io.EOF
219+}
220+func TestReleaseAssetConcurrencyCap(t *testing.T) {
221+ r := newReleaseRig(t, ports.ReleaseLimits{})
222+ rel := r.create(t, "v1", false)
223+ entered := make(chan struct{}, 16)
224+ release := make(chan struct{})
225+ var wg sync.WaitGroup
226+ errs := make(chan error, 16)
227+ for i := range 16 {
228+ wg.Go(func() {
229+ _, err := r.svc.UploadAsset(r.ctx, domain.Root(), r.repo, ports.UploadAssetParams{ReleaseID: rel.ID, Name: fmt.Sprint(i), Size: 1}, releaseBlockingReader{entered, release})
230+ errs <- err
231+ })
232+ }
233+ // Every reader has a reservation and holds its slot before the 17th call.
234+ for range 16 {
235+ <-entered
236+ }
237+ _, err := r.svc.UploadAsset(r.ctx, domain.Root(), r.repo, ports.UploadAssetParams{ReleaseID: rel.ID, Name: "excess", Size: 0}, bytes.NewReader(nil))
238+ require.ErrorIs(t, err, domain.ErrResourceExhausted)
239+ close(release)
240+ wg.Wait()
241+ close(errs)
242+ for err := range errs {
243+ require.ErrorIs(t, err, domain.ErrInvalid)
244+ }
245+}
246+
247+type errReader struct{ err error }
248+
249+func (r errReader) Read([]byte) (int, error) { return 0, r.err }
250+
251+func TestReleaseAssetDraftAccessAndRenameCollision(t *testing.T) {
252+ r := newReleaseRig(t, ports.ReleaseLimits{})
253+ rel := r.create(t, "draft", true)
254+ a := r.upload(t, rel, "a", []byte("abc"))
255+ b := r.upload(t, rel, "b", nil)
256+ _, err := r.svc.GetAsset(r.ctx, domain.Anonymous(), r.repo, ports.AssetRef{ID: a.ID})
257+ require.ErrorIs(t, err, domain.ErrNotFound)
258+ _, err = r.svc.ListAssets(r.ctx, domain.Anonymous(), r.repo, rel.ID)
259+ require.ErrorIs(t, err, domain.ErrNotFound)
260+ _, err = r.svc.UpdateAsset(r.ctx, domain.Root(), r.repo, b.ID, ports.UpdateAssetParams{Name: &a.Name})
261+ require.ErrorIs(t, err, domain.ErrAlreadyExists)
262+ got, err := r.svc.GetAsset(r.ctx, domain.Root(), r.repo, ports.AssetRef{Tag: rel.Tag, Name: a.Name})
263+ require.NoError(t, err)
264+ require.Equal(t, a.ID, got.ID)
265+ assets, err := r.svc.AssetsOf(r.ctx, r.repo)
266+ require.NoError(t, err)
267+ require.Len(t, assets, 2)
268+ require.NoError(t, r.store.Delete(r.ctx, r.repo))
269+ r.svc.OnRepoDeleted(r.ctx, r.repo, assets)
270+ _, err = r.blobs.Stat(r.ctx, a.BlobKey())
271+ require.ErrorIs(t, err, domain.ErrNotFound)
272+ all, err := r.store.Releases().AllAssets(r.ctx)
273+ require.NoError(t, err)
274+ require.Empty(t, all)
275+}
276+
277+func TestReleaseAssetQuotaAcrossReleasesAndWrongRepository(t *testing.T) {
278+ r := newReleaseRig(t, ports.ReleaseLimits{AssetMax: 3, Quota: 5})
279+ a := r.create(t, "a", false)
280+ b := r.create(t, "b", false)
281+ r.upload(t, a, "a", []byte("abc"))
282+ r.upload(t, b, "b", []byte("de"))
283+ _, err := r.svc.UploadAsset(r.ctx, domain.Root(), r.repo, ports.UploadAssetParams{ReleaseID: b.ID, Name: "extra", Size: 1}, bytes.NewReader([]byte("x")))
284+ require.ErrorIs(t, err, domain.ErrResourceExhausted)
285+ other := name(t, "other")
286+ require.NoError(t, r.store.Create(r.ctx, domain.Repo{Name: other, Visibility: domain.VisibilityPublic, CreatedAt: now}))
287+ _, err = r.svc.UploadAsset(r.ctx, domain.Root(), other, ports.UploadAssetParams{ReleaseID: b.ID, Name: "wrong"}, bytes.NewReader(nil))
288+ require.ErrorIs(t, err, domain.ErrPreconditionFailed)
289+}
290+
291+func TestReleaseAssetDownloadZeroBytes(t *testing.T) {
292+ r := newReleaseRig(t, ports.ReleaseLimits{AssetMax: 10, Quota: 10})
293+ rel := r.create(t, "v1", false)
294+ a := r.upload(t, rel, "empty", nil)
295+ var chunks []ports.AssetChunk
296+ err := r.svc.DownloadAsset(r.ctx, domain.Anonymous(), r.repo, ports.AssetRef{ID: a.ID}, 0, -1, func(c ports.AssetChunk) error {
297+ chunks = append(chunks, c)
298+ return nil
299+ })
300+ require.NoError(t, err)
301+ require.Len(t, chunks, 1)
302+ require.Equal(t, a.ID, chunks[0].Asset.ID)
303+
304+ // A missing blob stays an error.
305+ require.NoError(t, r.blobs.Delete(r.ctx, a.BlobKey()))
306+ err = r.svc.DownloadAsset(r.ctx, domain.Anonymous(), r.repo, ports.AssetRef{ID: a.ID}, 0, -1, func(ports.AssetChunk) error { return nil })
307+ require.ErrorIs(t, err, domain.ErrNotFound)
308+}
A internal/usecase/releases_github.go +387 -0
@@ -0,0 +1,387 @@
1+// SPDX-License-Identifier: AGPL-3.0-or-later
2+
3+package usecase
4+
5+import (
6+ "context"
7+ "encoding/json"
8+ "fmt"
9+ "io"
10+ "net/url"
11+ "slices"
12+ "strconv"
13+ "strings"
14+
15+ "bnema.dev/ermite/internal/domain"
16+ "bnema.dev/ermite/internal/ports"
17+)
18+
19+// ReleasesGitHub builds the GitHub REST representations of releases. Authorization
20+// and all mutations remain in Releases; it owns no metadata or contents.
21+type ReleasesGitHub struct {
22+ rel *Releases
23+ owner, publicURL string
24+}
25+
26+var _ ports.GitHubReleaseAPI = (*ReleasesGitHub)(nil)
27+
28+// NewReleasesGitHub wraps the shared release service. publicURL is the external
29+// server URL (without the API prefix), used in upload and browser download URLs.
30+func NewReleasesGitHub(rel *Releases, owner, publicURL string) *ReleasesGitHub {
31+ return &ReleasesGitHub{rel: rel, owner: owner, publicURL: strings.TrimRight(publicURL, "/")}
32+}
33+
34+func (g *ReleasesGitHub) repo(owner, name string) (domain.RepoName, error) {
35+ if !domain.OwnerMatches(g.owner, owner) {
36+ return domain.RepoName{}, domain.ErrNotFound
37+ }
38+ rn, err := domain.ParseRepoName(strings.ToLower(name))
39+ if err != nil {
40+ return domain.RepoName{}, domain.ErrNotFound
41+ }
42+ return rn, nil
43+}
44+func (g *ReleasesGitHub) base(repo domain.RepoName) string {
45+ return g.publicURL + apiPath + "/repos/" + url.PathEscape(g.owner) + "/" + url.PathEscape(repo.String())
46+}
47+func (g *ReleasesGitHub) user(label string) map[string]any {
48+ if label == "" || label == "root" || label == "owner" {
49+ label = g.owner
50+ }
51+ u := ghURLs{server: g.publicURL, api: g.publicURL + apiPath}
52+ return u.userObj(ghIdentity{login: label, id: stableID("user", label)}, false)
53+}
54+
55+// escapeReleaseTag preserves tag separators: the web routes join the tag path.
56+func escapeReleaseTag(tag string) string {
57+ parts := strings.Split(tag, "/")
58+ for i := range parts {
59+ parts[i] = url.PathEscape(parts[i])
60+ }
61+ return strings.Join(parts, "/")
62+}
63+func (g *ReleasesGitHub) assetObj(a domain.ReleaseAsset, tag string) ordered {
64+ return ordered{
65+ {"url", g.base(a.Repo) + "/releases/assets/" + strconv.FormatInt(a.ID, 10)},
66+ {"id", a.ID}, {"node_id", ""}, {"name", a.Name}, {"label", a.Label},
67+ {"uploader", g.user(a.Uploader)}, {"content_type", a.ContentType}, {"state", "uploaded"},
68+ {"size", a.Size}, {"digest", a.Digest}, {"download_count", 0},
69+ {"created_at", a.CreatedAt.UTC().Format(timeLayout)}, {"updated_at", a.UpdatedAt.UTC().Format(timeLayout)},
70+ {"browser_download_url", g.publicURL + "/" + a.Repo.String() + "/releases/download/" + url.PathEscape(tag) + "/" + url.PathEscape(a.Name)},
71+ }
72+}
73+func (g *ReleasesGitHub) releaseObj(ctx context.Context, p domain.Principal, r domain.Release) (ordered, error) {
74+ assets, err := g.rel.ListAssets(ctx, p, r.Repo, r.ID)
75+ if err != nil {
76+ return nil, err
77+ }
78+ objects := make([]ordered, 0, len(assets))
79+ for _, a := range assets {
80+ objects = append(objects, g.assetObj(a, r.Tag))
81+ }
82+ base := g.base(r.Repo) + "/releases/" + strconv.FormatInt(r.ID, 10)
83+ var published any
84+ if !r.Draft && !r.PublishedAt.IsZero() {
85+ published = r.PublishedAt.UTC().Format(timeLayout)
86+ }
87+ return ordered{
88+ {"url", base}, {"assets_url", base + "/assets"},
89+ {"upload_url", g.publicURL + apiPath + "/uploads/repos/" + url.PathEscape(g.owner) + "/" + r.Repo.String() + "/releases/" + strconv.FormatInt(r.ID, 10) + "/assets{?name,label}"},
90+ {"html_url", g.publicURL + "/" + r.Repo.String() + "/releases/tag/" + escapeReleaseTag(r.Tag)},
91+ {"id", r.ID}, {"author", g.user(r.Author)}, {"node_id", ""}, {"tag_name", r.Tag}, {"target_commitish", r.TargetCommitish},
92+ {"name", r.Title}, {"draft", r.Draft}, {"prerelease", r.Prerelease}, {"created_at", r.CreatedAt.UTC().Format(timeLayout)},
93+ {"published_at", published}, {"assets", objects}, {"tarball_url", g.base(r.Repo) + "/tarball/" + escapeReleaseTag(r.Tag)},
94+ {"zipball_url", nil}, {"body", r.Notes},
95+ }, nil
96+}
97+func (g *ReleasesGitHub) marshalRelease(ctx context.Context, p domain.Principal, r domain.Release) ([]byte, error) {
98+ o, err := g.releaseObj(ctx, p, r)
99+ if err != nil {
100+ return nil, err
101+ }
102+ return json.Marshal(o)
103+}
104+func releaseFields(body []byte) (map[string]json.RawMessage, error) {
105+ var fields map[string]json.RawMessage
106+ if err := json.Unmarshal(body, &fields); err != nil || fields == nil {
107+ return nil, fmt.Errorf("%w: JSON object required", domain.ErrInvalid)
108+ }
109+ return fields, nil
110+}
111+func releaseField[T any](fields map[string]json.RawMessage, key string) (*T, error) {
112+ b, ok := fields[key]
113+ if !ok {
114+ return nil, nil
115+ }
116+ var v T
117+ if string(b) == "null" || json.Unmarshal(b, &v) != nil {
118+ return nil, fmt.Errorf("%w: %s", domain.ErrInvalid, key)
119+ }
120+ return &v, nil
121+}
122+func releaseParams(body []byte) (ports.UpdateReleaseParams, error) {
123+ var p ports.UpdateReleaseParams
124+ f, err := releaseFields(body)
125+ if err != nil {
126+ return p, err
127+ }
128+ for key, dst := range map[string]**string{"tag_name": &p.Tag, "target_commitish": &p.TargetCommitish, "name": &p.Title, "body": &p.Notes} {
129+ *dst, err = releaseField[string](f, key)
130+ if err != nil {
131+ return p, err
132+ }
133+ }
134+ for key, dst := range map[string]**bool{"draft": &p.Draft, "prerelease": &p.Prerelease} {
135+ *dst, err = releaseField[bool](f, key)
136+ if err != nil {
137+ return p, err
138+ }
139+ }
140+ latest, err := releaseField[string](f, "make_latest")
141+ if err != nil {
142+ return p, err
143+ }
144+ if latest != nil {
145+ switch *latest {
146+ case "true":
147+ v := true
148+ p.MakeLatest = &v
149+ case "false":
150+ v := false
151+ p.MakeLatest = &v
152+ case "legacy":
153+ default:
154+ return p, fmt.Errorf("%w: make_latest", domain.ErrInvalid)
155+ }
156+ }
157+ return p, nil
158+}
159+func value[T any](p *T) (v T) {
160+ if p != nil {
161+ return *p
162+ }
163+ return v
164+}
165+
166+// CreateRelease accepts GitHub's create body and returns its representation.
167+func (g *ReleasesGitHub) CreateRelease(ctx context.Context, p domain.Principal, owner, repo string, body []byte) ([]byte, error) {
168+ rn, err := g.repo(owner, repo)
169+ if err != nil {
170+ return nil, err
171+ }
172+ params, err := releaseParams(body)
173+ if err != nil {
174+ return nil, err
175+ }
176+ if params.Tag == nil || *params.Tag == "" {
177+ return nil, fmt.Errorf("%w: tag_name", domain.ErrInvalid)
178+ }
179+ r, err := g.rel.CreateRelease(ctx, p, rn, ports.CreateReleaseParams{Tag: *params.Tag, TargetCommitish: value(params.TargetCommitish), Title: value(params.Title), Notes: value(params.Notes), Draft: value(params.Draft), Prerelease: value(params.Prerelease), MakeLatest: value(params.MakeLatest)})
180+ if err != nil {
181+ return nil, err
182+ }
183+ ctxLog(ctx).Info().Str("repo", rn.String()).Int64("release", r.ID).Msg("github release created")
184+ return g.marshalRelease(ctx, p, r)
185+}
186+
187+// GetRelease resolves an id, tag or latest selector.
188+func (g *ReleasesGitHub) GetRelease(ctx context.Context, p domain.Principal, owner, repo string, ref ports.ReleaseRef) ([]byte, error) {
189+ rn, err := g.repo(owner, repo)
190+ if err != nil {
191+ return nil, err
192+ }
193+ r, err := g.rel.GetRelease(ctx, p, rn, ref)
194+ if err != nil {
195+ return nil, err
196+ }
197+ return g.marshalRelease(ctx, p, r)
198+}
199+
200+// UpdateRelease changes only present fields, including explicit false and empty strings.
201+func (g *ReleasesGitHub) UpdateRelease(ctx context.Context, p domain.Principal, owner, repo string, id int64, body []byte) ([]byte, error) {
202+ rn, err := g.repo(owner, repo)
203+ if err != nil {
204+ return nil, err
205+ }
206+ params, err := releaseParams(body)
207+ if err != nil {
208+ return nil, err
209+ }
210+ r, err := g.rel.UpdateRelease(ctx, p, rn, id, params)
211+ if err != nil {
212+ return nil, err
213+ }
214+ ctxLog(ctx).Info().Str("repo", rn.String()).Int64("release", id).Msg("github release updated")
215+ return g.marshalRelease(ctx, p, r)
216+}
217+
218+// DeleteRelease removes a release and all its assets.
219+func (g *ReleasesGitHub) DeleteRelease(ctx context.Context, p domain.Principal, owner, repo string, id int64) error {
220+ rn, err := g.repo(owner, repo)
221+ if err != nil {
222+ return err
223+ }
224+ if err := g.rel.DeleteRelease(ctx, p, rn, id); err != nil {
225+ return err
226+ }
227+ ctxLog(ctx).Info().Str("repo", rn.String()).Int64("release", id).Msg("github release deleted")
228+ return nil
229+}
230+func releasePagination(page, perPage int) (int, int) {
231+ if page < 1 {
232+ page = 1
233+ }
234+ if perPage < 1 {
235+ perPage = 30
236+ }
237+ return page, min(perPage, 100)
238+}
239+func releasePageBounds(page, perPage, total int) (int, int) {
240+ // Avoid overflow on arbitrary query parameters.
241+ if page-1 > total/perPage {
242+ return total, total
243+ }
244+ start := min((page-1)*perPage, total)
245+ return start, min(start+perPage, total)
246+}
247+
248+// ListReleases returns visible releases newest first and GitHub Link pagination.
249+func (g *ReleasesGitHub) ListReleases(ctx context.Context, p domain.Principal, owner, repo string, page, perPage int) ([]byte, string, error) {
250+ rn, err := g.repo(owner, repo)
251+ if err != nil {
252+ return nil, "", err
253+ }
254+ if _, err = g.rel.readable(ctx, p, rn); err != nil {
255+ return nil, "", err
256+ }
257+ all, err := g.rel.store.ListReleases(ctx, rn)
258+ if err != nil {
259+ return nil, "", err
260+ }
261+ all = slices.DeleteFunc(all, func(r domain.Release) bool { return r.Draft && !releaseWriter(p, rn) })
262+ slices.SortFunc(all, func(a, b domain.Release) int {
263+ if c := b.CreatedAt.Compare(a.CreatedAt); c != 0 {
264+ return c
265+ }
266+ if a.ID > b.ID {
267+ return -1
268+ }
269+ if a.ID < b.ID {
270+ return 1
271+ }
272+ return 0
273+ })
274+ page, perPage = releasePagination(page, perPage)
275+ start, end := releasePageBounds(page, perPage, len(all))
276+ objects := make([]ordered, 0, end-start)
277+ for _, r := range all[start:end] {
278+ o, e := g.releaseObj(ctx, p, r)
279+ if e != nil {
280+ return nil, "", e
281+ }
282+ objects = append(objects, o)
283+ }
284+ b, err := json.Marshal(objects)
285+ return b, pullFilesLink(g.base(rn)+"/releases", page, perPage, len(all)), err
286+}
287+
288+// ListAssets lists uploaded assets with the same pagination conventions.
289+func (g *ReleasesGitHub) ListAssets(ctx context.Context, p domain.Principal, owner, repo string, id int64, page, perPage int) ([]byte, string, error) {
290+ rn, err := g.repo(owner, repo)
291+ if err != nil {
292+ return nil, "", err
293+ }
294+ r, err := g.rel.GetRelease(ctx, p, rn, ports.ReleaseRef{ID: id})
295+ if err != nil {
296+ return nil, "", err
297+ }
298+ all, err := g.rel.ListAssets(ctx, p, rn, id)
299+ if err != nil {
300+ return nil, "", err
301+ }
302+ page, perPage = releasePagination(page, perPage)
303+ start, end := releasePageBounds(page, perPage, len(all))
304+ objects := make([]ordered, 0, end-start)
305+ for _, a := range all[start:end] {
306+ objects = append(objects, g.assetObj(a, r.Tag))
307+ }
308+ b, err := json.Marshal(objects)
309+ return b, pullFilesLink(g.base(rn)+"/releases/"+strconv.FormatInt(id, 10)+"/assets", page, perPage, len(all)), err
310+}
311+func (g *ReleasesGitHub) marshalAsset(ctx context.Context, p domain.Principal, a domain.ReleaseAsset) ([]byte, error) {
312+ r, err := g.rel.GetRelease(ctx, p, a.Repo, ports.ReleaseRef{ID: a.ReleaseID})
313+ if err != nil {
314+ return nil, err
315+ }
316+ return json.Marshal(g.assetObj(a, r.Tag))
317+}
318+
319+// UploadAsset streams an exact-sized raw body; duplicate names are not replaced.
320+func (g *ReleasesGitHub) UploadAsset(ctx context.Context, p domain.Principal, owner, repo string, id int64, name, label, contentType string, size int64, body io.Reader) ([]byte, error) {
321+ rn, err := g.repo(owner, repo)
322+ if err != nil {
323+ return nil, err
324+ }
325+ a, err := g.rel.UploadAsset(ctx, p, rn, ports.UploadAssetParams{ReleaseID: id, Name: name, Label: label, ContentType: contentType, Size: size}, body)
326+ if err != nil {
327+ return nil, err
328+ }
329+ return g.marshalAsset(ctx, p, a)
330+}
331+
332+// AssetByID resolves a visible asset, including its repository ownership.
333+func (g *ReleasesGitHub) AssetByID(ctx context.Context, p domain.Principal, owner, repo string, id int64) (domain.ReleaseAsset, error) {
334+ rn, err := g.repo(owner, repo)
335+ if err != nil {
336+ return domain.ReleaseAsset{}, err
337+ }
338+ return g.rel.GetAsset(ctx, p, rn, ports.AssetRef{ID: id})
339+}
340+
341+// GetAsset returns the JSON representation, not the contents.
342+func (g *ReleasesGitHub) GetAsset(ctx context.Context, p domain.Principal, owner, repo string, id int64) ([]byte, error) {
343+ a, err := g.AssetByID(ctx, p, owner, repo, id)
344+ if err != nil {
345+ return nil, err
346+ }
347+ return g.marshalAsset(ctx, p, a)
348+}
349+
350+// UpdateAsset changes the name and label fields that are present.
351+func (g *ReleasesGitHub) UpdateAsset(ctx context.Context, p domain.Principal, owner, repo string, id int64, body []byte) ([]byte, error) {
352+ rn, err := g.repo(owner, repo)
353+ if err != nil {
354+ return nil, err
355+ }
356+ f, err := releaseFields(body)
357+ if err != nil {
358+ return nil, err
359+ }
360+ name, err := releaseField[string](f, "name")
361+ if err != nil {
362+ return nil, err
363+ }
364+ label, err := releaseField[string](f, "label")
365+ if err != nil {
366+ return nil, err
367+ }
368+ a, err := g.rel.UpdateAsset(ctx, p, rn, id, ports.UpdateAssetParams{Name: name, Label: label})
369+ if err != nil {
370+ return nil, err
371+ }
372+ ctxLog(ctx).Info().Str("repo", rn.String()).Int64("asset", id).Msg("github release asset updated")
373+ return g.marshalAsset(ctx, p, a)
374+}
375+
376+// DeleteAsset removes the asset record and its blob.
377+func (g *ReleasesGitHub) DeleteAsset(ctx context.Context, p domain.Principal, owner, repo string, id int64) error {
378+ rn, err := g.repo(owner, repo)
379+ if err != nil {
380+ return err
381+ }
382+ if err := g.rel.DeleteAsset(ctx, p, rn, id); err != nil {
383+ return err
384+ }
385+ ctxLog(ctx).Info().Str("repo", rn.String()).Int64("asset", id).Msg("github release asset deleted")
386+ return nil
387+}
A internal/usecase/releases_github_test.go +360 -0
@@ -0,0 +1,360 @@
1+// SPDX-License-Identifier: AGPL-3.0-or-later
2+
3+package usecase_test
4+
5+import (
6+ "bytes"
7+ "encoding/json"
8+ "fmt"
9+ "io"
10+ "math"
11+ "net/http"
12+ "net/http/httptest"
13+ "os"
14+ "path/filepath"
15+ "strconv"
16+ "strings"
17+ "testing"
18+
19+ "bnema.dev/ermite/internal/adapters/githubapi"
20+ "bnema.dev/ermite/internal/domain"
21+ "bnema.dev/ermite/internal/ports"
22+ "bnema.dev/ermite/internal/ports/mocks"
23+ "bnema.dev/ermite/internal/usecase"
24+ "github.com/stretchr/testify/mock"
25+ "github.com/stretchr/testify/require"
26+)
27+
28+func releaseGitHub(t *testing.T, r *releaseRig) *usecase.ReleasesGitHub {
29+ t.Helper()
30+ return usecase.NewReleasesGitHub(r.svc, "acme", "http://localhost:7341/")
31+}
32+func releaseGolden(t *testing.T, file string, body []byte) {
33+ t.Helper()
34+ var pretty bytes.Buffer
35+ require.NoError(t, json.Indent(&pretty, body, "", " "))
36+ pretty.WriteByte('\n')
37+ path := filepath.Join("..", "adapters", "githubapi", "testdata", "releases", file+".json")
38+ if os.Getenv("UPDATE_RELEASE_GOLDENS") == "1" {
39+ require.NoError(t, os.MkdirAll(filepath.Dir(path), 0755))
40+ require.NoError(t, os.WriteFile(path, pretty.Bytes(), 0644))
41+ }
42+ want, err := os.ReadFile(path)
43+ require.NoError(t, err)
44+ require.Equal(t, string(want), pretty.String())
45+}
46+func TestGitHubReleaseGoldens(t *testing.T) {
47+ r := newReleaseRig(t, ports.ReleaseLimits{})
48+ g := releaseGitHub(t, r)
49+ rel := r.create(t, "v1", false)
50+ for _, name := range []string{"ermite.tar.gz", "checksums.txt"} {
51+ _, err := g.UploadAsset(r.ctx, domain.Root(), "acme", r.repo.String(), rel.ID, name, "", "application/octet-stream", 3, strings.NewReader("abc"))
52+ require.NoError(t, err)
53+ }
54+ body, err := g.GetRelease(r.ctx, domain.Anonymous(), "ACME", "RELEASES", ports.ReleaseRef{Tag: "v1"})
55+ require.NoError(t, err)
56+ releaseGolden(t, "release", body)
57+ draft := r.create(t, "next", true)
58+ body, err = g.GetRelease(r.ctx, domain.Root(), "acme", r.repo.String(), ports.ReleaseRef{ID: draft.ID})
59+ require.NoError(t, err)
60+ releaseGolden(t, "draft", body)
61+ _, err = g.GetRelease(r.ctx, domain.Anonymous(), "acme", r.repo.String(), ports.ReleaseRef{ID: draft.ID})
62+ require.ErrorIs(t, err, domain.ErrNotFound)
63+}
64+func TestGitHubReleaseRequests(t *testing.T) {
65+ r := newReleaseRig(t, ports.ReleaseLimits{})
66+ g := releaseGitHub(t, r)
67+ repo := r.repo.String()
68+ p := domain.Root()
69+ r.resolve(t, "v1", nil)
70+ body, err := g.CreateRelease(r.ctx, p, "acme", repo, []byte(`{"tag_name":"v1","name":"Title","body":"Notes","draft":true,"prerelease":true,"discussion_category_name":"ignored","make_latest":"legacy"}`))
71+ require.NoError(t, err)
72+ var rel struct {
73+ ID int64 `json:"id"`
74+ }
75+ require.NoError(t, json.Unmarshal(body, &rel))
76+ body, err = g.UpdateRelease(r.ctx, p, "acme", repo, rel.ID, []byte(`{"name":"","draft":false,"prerelease":false,"make_latest":"true","unknown":1}`))
77+ require.NoError(t, err)
78+ var got map[string]any
79+ require.NoError(t, json.Unmarshal(body, &got))
80+ require.Equal(t, "", got["name"])
81+ require.Equal(t, "Notes", got["body"])
82+ require.Equal(t, false, got["draft"])
83+ require.Equal(t, false, got["prerelease"])
84+ require.NotContains(t, got, "make_latest")
85+ stored, err := r.svc.GetRelease(r.ctx, p, r.repo, ports.ReleaseRef{ID: rel.ID})
86+ require.NoError(t, err)
87+ require.True(t, stored.MakeLatest)
88+ _, err = g.UpdateRelease(r.ctx, p, "acme", repo, rel.ID, []byte(`{"make_latest":"legacy"}`))
89+ require.NoError(t, err)
90+ stored, err = r.svc.GetRelease(r.ctx, p, r.repo, ports.ReleaseRef{ID: rel.ID})
91+ require.NoError(t, err)
92+ require.True(t, stored.MakeLatest)
93+ _, err = g.UpdateRelease(r.ctx, p, "acme", repo, rel.ID, []byte(`{"make_latest":"false"}`))
94+ require.NoError(t, err)
95+ stored, err = r.svc.GetRelease(r.ctx, p, r.repo, ports.ReleaseRef{ID: rel.ID})
96+ require.NoError(t, err)
97+ require.False(t, stored.MakeLatest)
98+ for _, bad := range []string{`{}`, `null`, `[]`, `{`, `{"tag_name":2}`, `{"tag_name":"v1","draft":"true"}`, `{"tag_name":"v1","name":null}`, `{"tag_name":"v1","make_latest":true}`, `{"tag_name":"v1","make_latest":"yes"}`} {
99+ _, err = g.CreateRelease(r.ctx, p, "acme", repo, []byte(bad))
100+ require.ErrorIs(t, err, domain.ErrInvalid, bad)
101+ }
102+ _, err = g.CreateRelease(r.ctx, p, "acme", repo, []byte(`{"tag_name":"v1"}`))
103+ require.ErrorIs(t, err, domain.ErrAlreadyExists)
104+ _, err = g.GetRelease(r.ctx, p, "other", repo, ports.ReleaseRef{ID: rel.ID})
105+ require.ErrorIs(t, err, domain.ErrNotFound)
106+ _, err = g.GetRelease(r.ctx, p, "acme", "bad/name", ports.ReleaseRef{ID: rel.ID})
107+ require.ErrorIs(t, err, domain.ErrNotFound)
108+}
109+func TestGitHubReleasePagination(t *testing.T) {
110+ r := newReleaseRig(t, ports.ReleaseLimits{})
111+ g := releaseGitHub(t, r)
112+ for i := 0; i < 5; i++ {
113+ r.create(t, "v"+strconv.Itoa(i), i == 4)
114+ }
115+ for page := 1; page <= 3; page++ {
116+ b, link, err := g.ListReleases(r.ctx, domain.Root(), "acme", r.repo.String(), page, 2)
117+ require.NoError(t, err)
118+ var rows []struct {
119+ ID int64 `json:"id"`
120+ }
121+ require.NoError(t, json.Unmarshal(b, &rows))
122+ require.Len(t, rows, min(2, 5-(page-1)*2))
123+ require.Equal(t, int64(5-(page-1)*2), rows[0].ID)
124+ base := "http://localhost:7341/-/github/api/repos/acme/releases/releases?per_page=2&page="
125+ switch page {
126+ case 1:
127+ require.Equal(t, `<`+base+`2>; rel="next", <`+base+`3>; rel="last"`, link)
128+ case 2:
129+ require.Equal(t, `<`+base+`3>; rel="next", <`+base+`3>; rel="last", <`+base+`1>; rel="first", <`+base+`1>; rel="prev"`, link)
130+ case 3:
131+ require.Equal(t, `<`+base+`1>; rel="first", <`+base+`2>; rel="prev"`, link)
132+ }
133+ }
134+ b, _, err := g.ListReleases(r.ctx, domain.Anonymous(), "acme", r.repo.String(), 0, 0)
135+ require.NoError(t, err)
136+ var rows []any
137+ require.NoError(t, json.Unmarshal(b, &rows))
138+ require.Len(t, rows, 4)
139+ for _, page := range []int{4, math.MaxInt} {
140+ b, _, err = g.ListReleases(r.ctx, domain.Root(), "acme", r.repo.String(), page, 2)
141+ require.NoError(t, err)
142+ require.Equal(t, "[]", string(b))
143+ }
144+ b, _, err = g.ListReleases(r.ctx, domain.Root(), "acme", r.repo.String(), 1, 1000)
145+ require.NoError(t, err)
146+ require.NoError(t, json.Unmarshal(b, &rows))
147+ require.Len(t, rows, 5)
148+}
149+func TestGitHubReleaseAssetsAndRetry(t *testing.T) {
150+ r := newReleaseRig(t, ports.ReleaseLimits{AssetMax: 10})
151+ g := releaseGitHub(t, r)
152+ p := domain.Root()
153+ repo := r.repo.String()
154+ rel := r.create(t, "v1/slash", false)
155+ upload := func(name string) ([]byte, error) {
156+ return g.UploadAsset(r.ctx, p, "acme", repo, rel.ID, name, "label", "text/plain", 3, strings.NewReader("abc"))
157+ }
158+ b, err := upload("é #?.txt")
159+ require.NoError(t, err)
160+ var asset struct {
161+ ID int64 `json:"id"`
162+ URL string `json:"browser_download_url"`
163+ }
164+ require.NoError(t, json.Unmarshal(b, &asset))
165+ require.Equal(t, "http://localhost:7341/releases/releases/download/v1%2Fslash/%C3%A9%20%23%3F.txt", asset.URL)
166+ _, err = upload("é #?.txt")
167+ require.ErrorIs(t, err, domain.ErrAlreadyExists)
168+ b, err = g.UpdateAsset(r.ctx, p, "acme", repo, asset.ID, []byte(`{"label":""}`))
169+ require.NoError(t, err)
170+ require.Contains(t, string(b), `"label":""`)
171+ require.Contains(t, string(b), `"name":"é #?.txt"`)
172+ b, err = g.GetAsset(r.ctx, domain.Anonymous(), "acme", repo, asset.ID)
173+ require.NoError(t, err)
174+ require.Contains(t, string(b), `"state":"uploaded"`)
175+ for i := 0; i < 2; i++ {
176+ _, err = upload(fmt.Sprint(i))
177+ require.NoError(t, err)
178+ }
179+ b, link, err := g.ListAssets(r.ctx, p, "acme", repo, rel.ID, 1, 2)
180+ require.NoError(t, err)
181+ require.Contains(t, link, `rel="next"`)
182+ var assets []any
183+ require.NoError(t, json.Unmarshal(b, &assets))
184+ require.Len(t, assets, 2)
185+ require.NoError(t, g.DeleteAsset(r.ctx, p, "acme", repo, asset.ID))
186+ _, err = g.AssetByID(r.ctx, p, "acme", repo, asset.ID)
187+ require.ErrorIs(t, err, domain.ErrNotFound)
188+ _, err = upload("é #?.txt")
189+ require.NoError(t, err)
190+ _, err = g.UploadAsset(r.ctx, p, "acme", repo, rel.ID, "huge", "", "", 11, strings.NewReader("ignored"))
191+ require.ErrorIs(t, err, domain.ErrResourceExhausted)
192+ require.NoError(t, g.DeleteRelease(r.ctx, p, "acme", repo, rel.ID))
193+ _, err = g.GetRelease(r.ctx, p, "acme", repo, ports.ReleaseRef{ID: rel.ID})
194+ require.ErrorIs(t, err, domain.ErrNotFound)
195+}
196+
197+func TestGitHubReleaseAuthorizationAndLatest(t *testing.T) {
198+ r := newReleaseRig(t, ports.ReleaseLimits{})
199+ g := releaseGitHub(t, r)
200+ repo := r.repo.String()
201+ stable := r.create(t, "v1", false)
202+ r.create(t, "draft", true)
203+ b, err := g.GetRelease(r.ctx, domain.Anonymous(), "acme", repo, ports.ReleaseRef{Latest: true})
204+ require.NoError(t, err)
205+ var got struct {
206+ ID int64 `json:"id"`
207+ }
208+ require.NoError(t, json.Unmarshal(b, &got))
209+ require.Equal(t, stable.ID, got.ID)
210+ writer := domain.RepoTokenPrincipal(domain.CIRepoToken{Repo: r.repo, Permissions: map[string]string{domain.PermContents: domain.PermissionWrite}})
211+ reader := domain.RepoTokenPrincipal(domain.CIRepoToken{Repo: r.repo, Permissions: map[string]string{domain.PermContents: domain.PermissionRead}})
212+ branched := domain.TokenPrincipal(domain.Token{Scopes: []domain.Scope{domain.ScopeRepoRead, domain.ScopeReleaseWrite}, BranchPrefixes: []string{"feature/"}})
213+ for _, tc := range []struct {
214+ p domain.Principal
215+ tag string
216+ want error
217+ }{
218+ {domain.Anonymous(), "anonymous", domain.ErrUnauthenticated},
219+ {reader, "reader", domain.ErrForbidden},
220+ {branched, "branch", domain.ErrForbidden},
221+ {writer, "job", nil},
222+ } {
223+ if tc.want == nil {
224+ r.resolve(t, tc.tag, nil)
225+ }
226+ b, err = g.CreateRelease(r.ctx, tc.p, "acme", repo, []byte(`{"tag_name":"`+tc.tag+`"}`))
227+ if tc.want != nil {
228+ require.ErrorIs(t, err, tc.want)
229+ continue
230+ }
231+ require.NoError(t, err)
232+ require.NoError(t, json.Unmarshal(b, &got))
233+ _, err = g.UploadAsset(r.ctx, tc.p, "acme", repo, got.ID, "ci.tar.gz", "", "", 3, strings.NewReader("abc"))
234+ require.NoError(t, err)
235+ _, err = g.UploadAsset(r.ctx, reader, "acme", repo, got.ID, "read.tar.gz", "", "", 3, strings.NewReader("abc"))
236+ require.ErrorIs(t, err, domain.ErrForbidden)
237+ }
238+ other, err := domain.ParseRepoName("other")
239+ require.NoError(t, err)
240+ require.NoError(t, r.store.Create(r.ctx, domain.Repo{Name: other, Visibility: domain.VisibilityPublic, CreatedAt: now}))
241+ b, _, err = g.ListAssets(r.ctx, domain.Root(), "acme", other.String(), stable.ID, 1, 30)
242+ require.ErrorIs(t, err, domain.ErrNotFound)
243+ require.Nil(t, b)
244+}
245+
246+func TestGitHubReleaseTargetAndMissingTag(t *testing.T) {
247+ r := newReleaseRig(t, ports.ReleaseLimits{})
248+ g := releaseGitHub(t, r)
249+ p := domain.Root()
250+ repo := r.repo.String()
251+ r.resolve(t, "missing", domain.ErrNotFound)
252+ _, err := g.CreateRelease(r.ctx, p, "acme", repo, []byte(`{"tag_name":"missing"}`))
253+ require.ErrorIs(t, err, domain.ErrPreconditionFailed)
254+ b, err := g.CreateRelease(r.ctx, p, "acme", repo, []byte(`{"tag_name":"missing","draft":true}`))
255+ require.NoError(t, err)
256+ var got struct {
257+ ID int64 `json:"id"`
258+ Target string `json:"target_commitish"`
259+ }
260+ require.NoError(t, json.Unmarshal(b, &got))
261+ require.Empty(t, got.Target)
262+ _, err = g.UpdateRelease(r.ctx, p, "acme", repo, got.ID, []byte(`{"draft":false}`))
263+ require.ErrorIs(t, err, domain.ErrPreconditionFailed)
264+ r.resolve(t, "new", domain.ErrNotFound)
265+ v, err := domain.ParseRevision("main")
266+ require.NoError(t, err)
267+ oid, err := domain.ParseObjectID(oidA)
268+ require.NoError(t, err)
269+ r.reader.EXPECT().ResolveRevision(r.ctx, r.repo, v).Return(domain.ResolvedRevision{CommitID: oid}, nil).Once()
270+ r.tags.EXPECT().CreateTag(r.ctx, r.repo, "new", oid).Return(nil).Once()
271+ r.pushes.EXPECT().Record(r.ctx, p, r.repo, []domain.RefUpdate{{Ref: "refs/tags/new", Old: domain.ZeroOID, New: oidA}}).Return(nil).Once()
272+ b, err = g.CreateRelease(r.ctx, p, "acme", repo, []byte(`{"tag_name":"new","target_commitish":"main"}`))
273+ require.NoError(t, err)
274+ require.NoError(t, json.Unmarshal(b, &got))
275+ require.Equal(t, oidA, got.Target)
276+
277+ // PATCH with target_commitish publishes a draft whose tag is missing.
278+ r.resolve(t, "patched", domain.ErrNotFound)
279+ b, err = g.CreateRelease(r.ctx, p, "acme", repo, []byte(`{"tag_name":"patched","draft":true}`))
280+ require.NoError(t, err)
281+ require.NoError(t, json.Unmarshal(b, &got))
282+ r.reader.EXPECT().ResolveRevision(r.ctx, r.repo, v).Return(domain.ResolvedRevision{CommitID: oid}, nil).Once()
283+ r.tags.EXPECT().CreateTag(r.ctx, r.repo, "patched", oid).Return(nil).Once()
284+ r.pushes.EXPECT().Record(r.ctx, p, r.repo, []domain.RefUpdate{{Ref: "refs/tags/patched", Old: domain.ZeroOID, New: oidA}}).Return(nil).Once()
285+ b, err = g.UpdateRelease(r.ctx, p, "acme", repo, got.ID, []byte(`{"draft":false,"target_commitish":"main"}`))
286+ require.NoError(t, err)
287+ require.Contains(t, string(b), `"draft":false`)
288+ require.Contains(t, string(b), `"target_commitish":"`+oidA+`"`)
289+}
290+
291+// A real HTTP round trip through the facade, metadata store and disk blob
292+// adapter protects the goreleaser/softprops create, retry, overwrite, publish
293+// sequence, not just each side's mocked contract.
294+func TestGitHubReleaseHTTPWorkflow(t *testing.T) {
295+ r := newReleaseRig(t, ports.ReleaseLimits{})
296+ g := releaseGitHub(t, r)
297+ auth := mocks.NewAuthService(t)
298+ auth.EXPECT().Authenticate(mock.Anything, "dev-token").Return(domain.Root(), nil).Maybe()
299+ srv := httptest.NewServer(githubapi.New(githubapi.Config{Auth: auth, Releases: g, ReleaseDownload: r.svc}))
300+ defer srv.Close()
301+ do := func(method, path, body string, headers map[string]string) (*http.Response, string) {
302+ req, err := http.NewRequestWithContext(r.ctx, method, srv.URL+githubapi.Prefix+path, strings.NewReader(body))
303+ require.NoError(t, err)
304+ req.Header.Set("Authorization", "Bearer dev-token")
305+ for k, v := range headers {
306+ req.Header.Set(k, v)
307+ }
308+ resp, err := srv.Client().Do(req)
309+ require.NoError(t, err)
310+ defer func() { _ = resp.Body.Close() }()
311+ b, err := io.ReadAll(resp.Body)
312+ require.NoError(t, err)
313+ return resp, string(b)
314+ }
315+ base := "/repos/acme/" + r.repo.String() + "/releases"
316+ resp, _ := do("GET", base+"/tags/v1", "", nil)
317+ require.Equal(t, 404, resp.StatusCode)
318+ r.resolve(t, "v1", nil)
319+ resp, b := do("POST", base, `{"tag_name":"v1","draft":true}`, nil)
320+ require.Equal(t, 201, resp.StatusCode)
321+ var rel struct {
322+ ID int64 `json:"id"`
323+ UploadURL string `json:"upload_url"`
324+ }
325+ require.NoError(t, json.Unmarshal([]byte(b), &rel))
326+ upload := strings.TrimPrefix(strings.Split(rel.UploadURL, "{")[0], "http://localhost:7341"+githubapi.Prefix) + "?name=archive.tar.gz"
327+ resp, b = do("POST", upload, "0123456789", map[string]string{"Content-Type": "application/gzip"})
328+ require.Equal(t, 201, resp.StatusCode)
329+ var a struct {
330+ ID int64 `json:"id"`
331+ }
332+ require.NoError(t, json.Unmarshal([]byte(b), &a))
333+ resp, b = do("POST", upload, "replacement", nil)
334+ require.Equal(t, 422, resp.StatusCode)
335+ require.Contains(t, b, `"code":"already_exists"`)
336+ resp, b = do("GET", base+"/"+strconv.FormatInt(rel.ID, 10)+"/assets", "", nil)
337+ require.Equal(t, 200, resp.StatusCode)
338+ require.Contains(t, b, `"name":"archive.tar.gz"`)
339+ assetURL := base + "/assets/" + strconv.FormatInt(a.ID, 10)
340+ resp, _ = do("DELETE", assetURL, "", nil)
341+ require.Equal(t, 204, resp.StatusCode)
342+ resp, b = do("POST", upload, "abcdefghij", nil)
343+ require.Equal(t, 201, resp.StatusCode)
344+ require.NoError(t, json.Unmarshal([]byte(b), &a))
345+ resp, _ = do("PATCH", base+"/"+strconv.FormatInt(rel.ID, 10), `{"draft":false}`, nil)
346+ require.Equal(t, 200, resp.StatusCode)
347+ assetURL = base + "/assets/" + strconv.FormatInt(a.ID, 10)
348+ resp, b = do("GET", assetURL, "", map[string]string{"Accept": "application/octet-stream", "Range": "bytes=2-5"})
349+ require.Equal(t, 206, resp.StatusCode)
350+ require.Equal(t, "cdef", b)
351+ require.Equal(t, "bytes 2-5/10", resp.Header.Get("Content-Range"))
352+ require.Equal(t, "4", resp.Header.Get("Content-Length"))
353+ require.Empty(t, resp.Header.Get("Location"))
354+ resp, b = do("HEAD", assetURL, "", map[string]string{"Accept": "application/octet-stream"})
355+ require.Equal(t, 200, resp.StatusCode)
356+ require.Empty(t, b)
357+ require.Equal(t, "10", resp.Header.Get("Content-Length"))
358+ resp, _ = do("GET", assetURL, "", map[string]string{"Accept": "application/octet-stream", "If-None-Match": resp.Header.Get("ETag")})
359+ require.Equal(t, 304, resp.StatusCode)
360+}
A internal/usecase/releases_integration_test.go +336 -0
@@ -0,0 +1,336 @@
1+// SPDX-License-Identifier: AGPL-3.0-or-later
2+
3+package usecase_test
4+
5+import (
6+ "context"
7+ "fmt"
8+ "testing"
9+ "testing/synctest"
10+
11+ "bnema.dev/ermite/internal/adapters/blobdisk"
12+ "bnema.dev/ermite/internal/adapters/eventbus"
13+ "bnema.dev/ermite/internal/adapters/kvstore"
14+ "bnema.dev/ermite/internal/domain"
15+ "bnema.dev/ermite/internal/ports"
16+ "bnema.dev/ermite/internal/ports/mocks"
17+ "bnema.dev/ermite/internal/usecase"
18+ "github.com/stretchr/testify/mock"
19+ "github.com/stretchr/testify/require"
20+)
21+
22+type releaseRig struct {
23+ ctx context.Context
24+ repo domain.RepoName
25+ store *kvstore.Store
26+ blobs ports.BlobStore
27+ reader *mocks.RepoReader
28+ tags *mocks.ReleaseTagWriter
29+ pushes *mocks.PushHookService
30+ clock *mocks.Clock
31+ events <-chan domain.Event
32+ svc *usecase.Releases
33+}
34+
35+func newReleaseRig(t *testing.T, limits ports.ReleaseLimits) *releaseRig {
36+ t.Helper()
37+ ctx := t.Context()
38+ store, err := kvstore.Open(ctx, t.TempDir())
39+ require.NoError(t, err)
40+ t.Cleanup(func() { require.NoError(t, store.Close(context.Background())) })
41+ n := name(t, "releases")
42+ require.NoError(t, store.Create(ctx, domain.Repo{Name: n, Visibility: domain.VisibilityPublic, CreatedAt: now}))
43+ clock := mocks.NewClock(t)
44+ clock.EXPECT().Now().Return(now).Maybe()
45+ reader := mocks.NewRepoReader(t)
46+ tags := mocks.NewReleaseTagWriter(t)
47+ pushes := mocks.NewPushHookService(t)
48+ bus := eventbus.NewWithBuffer(1024)
49+ b := blobdisk.New(t.TempDir())
50+ r := &releaseRig{ctx: ctx, repo: n, store: store, blobs: b, reader: reader, tags: tags, pushes: pushes, clock: clock, events: bus.Subscribe(ctx)}
51+ r.svc = usecase.NewReleases(store.Releases(), b, store, reader, tags, pushes, bus, clock, limits)
52+ return r
53+}
54+func (r *releaseRig) resolve(t *testing.T, tag string, err error) {
55+ t.Helper()
56+ v, e := domain.ParseRevision("refs/tags/" + tag)
57+ require.NoError(t, e)
58+ oid, e := domain.ParseObjectID(oidA)
59+ require.NoError(t, e)
60+ r.reader.EXPECT().ResolveRevision(mock.Anything, r.repo, v).Return(domain.ResolvedRevision{CommitID: oid}, err).Maybe()
61+}
62+func (r *releaseRig) create(t *testing.T, tag string, draft bool) domain.Release {
63+ t.Helper()
64+ r.resolve(t, tag, nil)
65+ rel, err := r.svc.CreateRelease(r.ctx, domain.Root(), r.repo, ports.CreateReleaseParams{Tag: tag, Draft: draft})
66+ require.NoError(t, err)
67+ return rel
68+}
69+func TestReleaseAccessMatrix(t *testing.T) {
70+ for _, private := range []bool{false, true} {
71+ t.Run(fmt.Sprint("private=", private), func(t *testing.T) {
72+ r := newReleaseRig(t, ports.ReleaseLimits{})
73+ published := r.create(t, "v1", false)
74+ draft := r.create(t, "draft", true)
75+ if private {
76+ _, err := r.store.Mutate(r.ctx, r.repo, func(repo *domain.Repo) (bool, error) { repo.Visibility = domain.VisibilityPrivate; return true, nil })
77+ require.NoError(t, err)
78+ }
79+ token := func(scopes []domain.Scope, branches ...string) domain.Principal {
80+ return domain.TokenPrincipal(domain.Token{ID: "t", Scopes: scopes, BranchPrefixes: branches})
81+ }
82+ for _, tc := range []struct {
83+ name string
84+ p domain.Principal
85+ read, write bool
86+ }{
87+ {"anonymous", domain.Anonymous(), !private, false},
88+ {"reader", token([]domain.Scope{domain.ScopeRepoRead}), true, false},
89+ {"writer", token([]domain.Scope{domain.ScopeRepoRead, domain.ScopeReleaseWrite}), true, true},
90+ {"write-only", token([]domain.Scope{domain.ScopeReleaseWrite}), !private, !private},
91+ {"branch", token([]domain.Scope{domain.ScopeRepoRead, domain.ScopeReleaseWrite}, "x/"), true, false},
92+ {"job-read", domain.RepoTokenPrincipal(domain.CIRepoToken{ID: "j", Repo: r.repo, Permissions: map[string]string{"contents": "read"}}), true, false},
93+ {"job-write", domain.RepoTokenPrincipal(domain.CIRepoToken{ID: "j", Repo: r.repo, Permissions: map[string]string{"contents": "write"}}), true, true},
94+ {"owner", domain.Owner(), true, true},
95+ } {
96+ t.Run(tc.name, func(t *testing.T) {
97+ _, err := r.svc.GetRelease(r.ctx, tc.p, r.repo, ports.ReleaseRef{ID: published.ID})
98+ if tc.read {
99+ require.NoError(t, err)
100+ } else {
101+ require.ErrorIs(t, err, domain.ErrNotFound)
102+ }
103+ _, err = r.svc.GetRelease(r.ctx, tc.p, r.repo, ports.ReleaseRef{ID: draft.ID})
104+ if tc.write {
105+ require.NoError(t, err)
106+ } else {
107+ require.ErrorIs(t, err, domain.ErrNotFound)
108+ }
109+ title := "changed"
110+ _, err = r.svc.UpdateRelease(r.ctx, tc.p, r.repo, published.ID, ports.UpdateReleaseParams{Title: &title})
111+ switch {
112+ case tc.write:
113+ require.NoError(t, err)
114+ case !tc.read:
115+ require.ErrorIs(t, err, domain.ErrNotFound)
116+ case tc.p.IsAnonymous():
117+ require.ErrorIs(t, err, domain.ErrUnauthenticated)
118+ default:
119+ require.ErrorIs(t, err, domain.ErrForbidden)
120+ }
121+ })
122+ }
123+ })
124+ }
125+}
126+func TestReleaseLifecycle(t *testing.T) {
127+ r := newReleaseRig(t, ports.ReleaseLimits{})
128+ first := r.create(t, "v1", false)
129+ second := r.create(t, "v2", false)
130+ r.create(t, "draft", true)
131+ got, err := r.svc.GetRelease(r.ctx, domain.Anonymous(), r.repo, ports.ReleaseRef{Latest: true})
132+ require.NoError(t, err)
133+ require.Equal(t, second.ID, got.ID)
134+ yes := true
135+ _, err = r.svc.UpdateRelease(r.ctx, domain.Root(), r.repo, first.ID, ports.UpdateReleaseParams{MakeLatest: &yes})
136+ require.NoError(t, err)
137+ got, err = r.svc.GetRelease(r.ctx, domain.Root(), r.repo, ports.ReleaseRef{Latest: true})
138+ require.NoError(t, err)
139+ require.Equal(t, first.ID, got.ID)
140+ _, err = r.svc.CreateRelease(r.ctx, domain.Root(), r.repo, ports.CreateReleaseParams{Tag: "v1"})
141+ require.ErrorIs(t, err, domain.ErrAlreadyExists)
142+ page, err := r.svc.ListReleases(r.ctx, domain.Anonymous(), r.repo, 1, "")
143+ require.NoError(t, err)
144+ require.Len(t, page.Releases, 1)
145+ require.NotEmpty(t, page.NextPageToken)
146+ next, err := r.svc.ListReleases(r.ctx, domain.Anonymous(), r.repo, 1, page.NextPageToken)
147+ require.NoError(t, err)
148+ require.Equal(t, first.ID, next.Releases[0].ID)
149+ require.Empty(t, next.NextPageToken)
150+ r.svc.OnRefsMoved(r.ctx, r.repo, []domain.RefUpdate{{Ref: "refs/tags/v1", New: domain.ZeroOID}})
151+ got, err = r.svc.GetRelease(r.ctx, domain.Root(), r.repo, ports.ReleaseRef{ID: first.ID})
152+ require.NoError(t, err)
153+ require.True(t, got.Draft)
154+ require.True(t, got.PublishedAt.IsZero())
155+ require.False(t, got.MakeLatest)
156+ r.svc.OnRefsMoved(r.ctx, r.repo, []domain.RefUpdate{{Ref: "refs/tags/v1", New: oidA}})
157+ got, err = r.svc.GetRelease(r.ctx, domain.Root(), r.repo, ports.ReleaseRef{ID: first.ID})
158+ require.NoError(t, err)
159+ require.True(t, got.Draft)
160+ var kinds []string
161+ for len(r.events) > 0 {
162+ kinds = append(kinds, (<-r.events).EventKind())
163+ }
164+ require.Contains(t, kinds, "release.unpublished")
165+ require.NoError(t, r.svc.DeleteRelease(r.ctx, domain.Root(), r.repo, first.ID))
166+ _, err = r.svc.GetRelease(r.ctx, domain.Root(), r.repo, ports.ReleaseRef{ID: first.ID})
167+ require.ErrorIs(t, err, domain.ErrNotFound)
168+}
169+func TestReleaseMissingTag(t *testing.T) {
170+ r := newReleaseRig(t, ports.ReleaseLimits{})
171+ r.resolve(t, "missing", domain.ErrNotFound)
172+ draft, err := r.svc.CreateRelease(r.ctx, domain.Root(), r.repo, ports.CreateReleaseParams{Tag: "missing", Draft: true})
173+ require.NoError(t, err)
174+ require.Empty(t, draft.TargetCommitish)
175+ no := false
176+ _, err = r.svc.UpdateRelease(r.ctx, domain.Root(), r.repo, draft.ID, ports.UpdateReleaseParams{Draft: &no})
177+ require.ErrorIs(t, err, domain.ErrPreconditionFailed)
178+ r.resolve(t, "new", domain.ErrNotFound)
179+ _, err = r.svc.CreateRelease(r.ctx, domain.Root(), r.repo, ports.CreateReleaseParams{Tag: "new"})
180+ require.ErrorIs(t, err, domain.ErrPreconditionFailed)
181+ v, err := domain.ParseRevision("main")
182+ require.NoError(t, err)
183+ oid, err := domain.ParseObjectID(oidA)
184+ require.NoError(t, err)
185+ r.reader.EXPECT().ResolveRevision(r.ctx, r.repo, v).Return(domain.ResolvedRevision{CommitID: oid}, nil).Once()
186+ r.tags.EXPECT().CreateTag(r.ctx, r.repo, "new", oid).Return(nil).Once()
187+ r.pushes.EXPECT().Record(r.ctx, domain.Root(), r.repo, []domain.RefUpdate{{Ref: "refs/tags/new", Old: domain.ZeroOID, New: oidA}}).Return(nil).Once()
188+ rel, err := r.svc.CreateRelease(r.ctx, domain.Root(), r.repo, ports.CreateReleaseParams{Tag: "new", TargetCommitish: "main"})
189+ require.NoError(t, err)
190+ require.Equal(t, oidA, rel.TargetCommitish)
191+}
192+
193+func TestReleaseLatestPinsAndPrereleases(t *testing.T) {
194+ r := newReleaseRig(t, ports.ReleaseLimits{})
195+ a := r.create(t, "a", false)
196+ b := r.create(t, "b", false)
197+ yes := true
198+ _, err := r.svc.UpdateRelease(r.ctx, domain.Root(), r.repo, b.ID, ports.UpdateReleaseParams{Prerelease: &yes})
199+ require.NoError(t, err)
200+ latest, err := r.svc.GetRelease(r.ctx, domain.Anonymous(), r.repo, ports.ReleaseRef{Latest: true})
201+ require.NoError(t, err)
202+ require.Equal(t, a.ID, latest.ID)
203+ _, err = r.svc.UpdateRelease(r.ctx, domain.Root(), r.repo, a.ID, ports.UpdateReleaseParams{MakeLatest: &yes})
204+ require.NoError(t, err)
205+ r.resolve(t, "c", nil)
206+ c, err := r.svc.CreateRelease(r.ctx, domain.Root(), r.repo, ports.CreateReleaseParams{Tag: "c", MakeLatest: true})
207+ require.NoError(t, err)
208+ a, err = r.svc.GetRelease(r.ctx, domain.Root(), r.repo, ports.ReleaseRef{ID: a.ID})
209+ require.NoError(t, err)
210+ require.False(t, a.MakeLatest)
211+ target := "b"
212+ _, err = r.svc.UpdateRelease(r.ctx, domain.Root(), r.repo, c.ID, ports.UpdateReleaseParams{Tag: &target})
213+ require.ErrorIs(t, err, domain.ErrAlreadyExists)
214+ require.NoError(t, r.svc.DeleteRelease(r.ctx, domain.Root(), r.repo, c.ID))
215+ latest, err = r.svc.GetRelease(r.ctx, domain.Root(), r.repo, ports.ReleaseRef{Latest: true})
216+ require.NoError(t, err)
217+ require.Equal(t, a.ID, latest.ID)
218+}
219+
220+func TestReleaseEventsReachHomeAndRepositoryTopics(t *testing.T) {
221+ synctest.Test(t, func(t *testing.T) {
222+ r := newEventRig(t, 0)
223+ pub := r.repo("pub", domain.VisibilityPublic)
224+ priv := r.repo("priv", domain.VisibilityPrivate)
225+ ctx, cancel := context.WithCancel(context.Background())
226+ defer cancel()
227+ anon := r.watch(ctx, domain.Anonymous(), ports.WatchParams{Topics: []string{"home"}})
228+ owner := r.watch(ctx, domain.Root(), ports.WatchParams{Topics: []string{"repo:priv"}})
229+ synctest.Wait()
230+ for _, n := range []domain.RepoName{priv, pub} {
231+ vis := domain.VisibilityPublic
232+ if n == priv {
233+ vis = domain.VisibilityPrivate
234+ }
235+ rel := domain.Release{ID: 1, Repo: n, Tag: "v1"}
236+ r.feed <- domain.ReleasePublished{Release: rel, At: now, CreatedAt: now, Visibility: vis}
237+ r.feed <- domain.ReleaseUpdated{Release: rel, At: now, CreatedAt: now, Visibility: vis}
238+ r.feed <- domain.ReleaseDeleted{Release: rel, At: now, CreatedAt: now, Visibility: vis}
239+ r.feed <- domain.ReleaseUnpublished{Release: rel, At: now, CreatedAt: now, Visibility: vis}
240+ }
241+ for _, w := range []*watcher{anon, owner} {
242+ for _, kind := range []string{"release.published", "release.updated", "release.deleted", "release.unpublished"} {
243+ require.Equal(t, kind, w.next(t).Event.EventKind())
244+ }
245+ }
246+ anon.quiet(t, r, pub)
247+ owner.quiet(t, r, priv)
248+ })
249+}
250+
251+func TestReleaseWriteRetriesUsageConflict(t *testing.T) {
252+ ctx := t.Context()
253+ repo := name(t, "retry")
254+ store := mocks.NewReleaseStore(t)
255+ meta := mocks.NewRepoMetaStore(t)
256+ reader := mocks.NewRepoReader(t)
257+ clock := mocks.NewClock(t)
258+ bus := mocks.NewEventBus(t)
259+ meta.EXPECT().Get(ctx, repo).Return(domain.Repo{Name: repo, Visibility: domain.VisibilityPublic, CreatedAt: now}, nil)
260+ clock.EXPECT().Now().Return(now).Maybe()
261+ rev, err := domain.ParseRevision("refs/tags/v1")
262+ require.NoError(t, err)
263+ reader.EXPECT().ResolveRevision(ctx, repo, rev).Return(domain.ResolvedRevision{}, domain.ErrNotFound)
264+ store.EXPECT().GetReleaseByTag(ctx, repo, "v1").Return(domain.Release{}, domain.ErrNotFound)
265+ store.EXPECT().NextReleaseID(ctx).Return(int64(1), nil)
266+ store.EXPECT().ReleaseUsage(ctx, repo).Return(int64(0), uint64(1), nil).Once()
267+ store.EXPECT().ApplyReleases(ctx, mock.MatchedBy(func(w ports.ReleaseWrite) bool { return w.UsageVersion == 1 && len(w.PutReleases) == 1 })).Return(domain.ErrConflict).Once()
268+ store.EXPECT().ReleaseUsage(ctx, repo).Return(int64(0), uint64(2), nil).Once()
269+ store.EXPECT().ApplyReleases(ctx, mock.MatchedBy(func(w ports.ReleaseWrite) bool { return w.UsageVersion == 2 && len(w.PutReleases) == 1 })).Return(nil).Once()
270+ store.EXPECT().GetRelease(ctx, repo, int64(1)).Return(domain.Release{ID: 1, Repo: repo, Tag: "v1", Draft: true}, nil)
271+ svc := usecase.NewReleases(store, nil, meta, reader, nil, nil, bus, clock, ports.ReleaseLimits{})
272+ rel, err := svc.CreateRelease(ctx, domain.Root(), repo, ports.CreateReleaseParams{Tag: "v1", Draft: true})
273+ require.NoError(t, err)
274+ require.EqualValues(t, 1, rel.ID)
275+}
276+
277+func TestReleaseListHidesDraftCursors(t *testing.T) {
278+ r := newReleaseRig(t, ports.ReleaseLimits{})
279+ pub := r.create(t, "v1", false)
280+ d1 := r.create(t, "d1", true)
281+ d2 := r.create(t, "d2", true)
282+ anon := domain.Anonymous()
283+
284+ _, errDraft := r.svc.ListReleases(r.ctx, anon, r.repo, 1, fmt.Sprint(d1.ID))
285+ _, errUnknown := r.svc.ListReleases(r.ctx, anon, r.repo, 1, "999999")
286+ require.ErrorIs(t, errUnknown, domain.ErrInvalidPageToken)
287+ require.Equal(t, errUnknown, errDraft)
288+
289+ // A writer still pages through drafts, newest first.
290+ var got []int64
291+ token := ""
292+ for {
293+ page, err := r.svc.ListReleases(r.ctx, domain.Root(), r.repo, 1, token)
294+ require.NoError(t, err)
295+ for _, rel := range page.Releases {
296+ got = append(got, rel.ID)
297+ }
298+ if page.NextPageToken == "" {
299+ break
300+ }
301+ token = page.NextPageToken
302+ }
303+ require.ElementsMatch(t, []int64{pub.ID, d1.ID, d2.ID}, got)
304+}
305+
306+func TestReleaseUpdateTargetCreatesMissingTag(t *testing.T) {
307+ r := newReleaseRig(t, ports.ReleaseLimits{})
308+ r.resolve(t, "missing", domain.ErrNotFound)
309+ draft, err := r.svc.CreateRelease(r.ctx, domain.Root(), r.repo, ports.CreateReleaseParams{Tag: "missing", Draft: true})
310+ require.NoError(t, err)
311+ no := false
312+ _, err = r.svc.UpdateRelease(r.ctx, domain.Root(), r.repo, draft.ID, ports.UpdateReleaseParams{Draft: &no})
313+ require.ErrorIs(t, err, domain.ErrPreconditionFailed)
314+
315+ v, err := domain.ParseRevision("main")
316+ require.NoError(t, err)
317+ oid, err := domain.ParseObjectID(oidA)
318+ require.NoError(t, err)
319+ r.reader.EXPECT().ResolveRevision(r.ctx, r.repo, v).Return(domain.ResolvedRevision{CommitID: oid}, nil).Once()
320+ r.tags.EXPECT().CreateTag(r.ctx, r.repo, "missing", oid).Return(nil).Once()
321+ r.pushes.EXPECT().Record(r.ctx, domain.Root(), r.repo, []domain.RefUpdate{{Ref: "refs/tags/missing", Old: domain.ZeroOID, New: oidA}}).Return(nil).Once()
322+ main := "main"
323+ rel, err := r.svc.UpdateRelease(r.ctx, domain.Root(), r.repo, draft.ID, ports.UpdateReleaseParams{Draft: &no, TargetCommitish: &main})
324+ require.NoError(t, err)
325+ require.False(t, rel.Draft)
326+ require.Equal(t, oidA, rel.TargetCommitish)
327+}
328+
329+func TestReleaseUpdateTargetNeverMovesExistingTag(t *testing.T) {
330+ r := newReleaseRig(t, ports.ReleaseLimits{})
331+ rel := r.create(t, "v1", false)
332+ other := "other" // never resolved: the mock would fail the test
333+ got, err := r.svc.UpdateRelease(r.ctx, domain.Root(), r.repo, rel.ID, ports.UpdateReleaseParams{TargetCommitish: &other})
334+ require.NoError(t, err)
335+ require.Equal(t, oidA, got.TargetCommitish)
336+}
A internal/usecase/releases_sweep.go +106 -0
@@ -0,0 +1,106 @@
1+// SPDX-License-Identifier: AGPL-3.0-or-later
2+
3+package usecase
4+
5+import (
6+ "context"
7+ "errors"
8+ "strconv"
9+ "strings"
10+
11+ "bnema.dev/ermite/internal/domain"
12+ "bnema.dev/ermite/internal/ports"
13+)
14+
15+// Sweep collects abandoned uploads, missing contents, and old orphan blobs.
16+// A failed metadata scan never authorizes blob deletion. Sweeps are serialized
17+// so the two observations of a missing blob are always distinct sweeps.
18+func (r *Releases) Sweep(ctx context.Context) {
19+ r.sweepMu.Lock()
20+ defer r.sweepMu.Unlock()
21+ log := ctxLog(ctx)
22+ now := r.clock.Now()
23+ all, err := r.store.AllAssets(ctx)
24+ if err != nil {
25+ log.Error().Err(err).Msg("scan release assets")
26+ return
27+ }
28+ seen := make(map[int64]bool, len(all))
29+ for _, a := range all {
30+ seen[a.ID] = true
31+ remove := false
32+ switch a.State {
33+ case domain.ReleaseAssetUploading:
34+ remove = now.Sub(a.UpdatedAt) > domain.ReleaseAssetPendingMax
35+ case domain.ReleaseAssetUploaded:
36+ _, err := r.blobs.Stat(ctx, a.BlobKey())
37+ switch {
38+ case errors.Is(err, domain.ErrNotFound):
39+ r.misses[a.ID]++
40+ remove = r.misses[a.ID] >= 2
41+ if !remove {
42+ log.Error().Int64("asset", a.ID).Msg("asset blob missing")
43+ }
44+ case err != nil:
45+ log.Error().Err(err).Int64("asset", a.ID).Msg("stat release asset")
46+ default:
47+ delete(r.misses, a.ID)
48+ }
49+ }
50+ if !remove {
51+ continue
52+ }
53+ err := r.removeAssetRecord(ctx, a, func(current domain.ReleaseAsset) bool { return current.Version == a.Version })
54+ if err != nil {
55+ if !errors.Is(err, domain.ErrNotFound) && !errors.Is(err, domain.ErrConflict) {
56+ log.Error().Err(err).Int64("asset", a.ID).Msg("remove stale release asset")
57+ }
58+ continue
59+ }
60+ delete(r.misses, a.ID)
61+ r.removeAssetBlobs(ctx, []domain.ReleaseAsset{a})
62+ }
63+ for id := range r.misses {
64+ if !seen[id] {
65+ delete(r.misses, id)
66+ }
67+ }
68+ // Re-check the store for each old key: concurrent reservations must not be
69+ // classified as orphans from a stale snapshot.
70+ err = r.blobs.List(ctx, "release/", func(info ports.BlobInfo) error {
71+ if now.Sub(info.ModTime) <= domain.ReleaseAssetPendingMax {
72+ return nil
73+ }
74+ // Only a canonical key can name an asset. Check its record directly
75+ // rather than repeating the full metadata scan for every object.
76+ id, parseErr := strconv.ParseInt(strings.TrimPrefix(info.Key, "release/"), 10, 64)
77+ if parseErr == nil && (domain.ReleaseAsset{ID: id}).BlobKey() == info.Key {
78+ _, e := r.store.GetAsset(ctx, id)
79+ if e == nil {
80+ return nil
81+ }
82+ if !errors.Is(e, domain.ErrNotFound) {
83+ log.Error().Err(e).Msg("check orphan release blob")
84+ return nil
85+ }
86+ }
87+ if e := r.blobs.Delete(ctx, info.Key); e != nil {
88+ log.Error().Err(e).Str("key", info.Key).Msg("remove orphan release blob")
89+ }
90+ return nil
91+ })
92+ if err != nil {
93+ log.Error().Err(err).Msg("list release blobs")
94+ }
95+ err = r.blobs.ListMultipart(ctx, "release/", func(info ports.MultipartInfo) error {
96+ if now.Sub(info.Started) > domain.ReleaseAssetPendingMax {
97+ if e := r.blobs.AbortMultipart(ctx, info.Key, info.ID); e != nil {
98+ log.Error().Err(e).Msg("abort abandoned release multipart")
99+ }
100+ }
101+ return nil
102+ })
103+ if err != nil {
104+ log.Error().Err(err).Msg("list release multipart uploads")
105+ }
106+}
A internal/usecase/releases_sweep_test.go +147 -0
@@ -0,0 +1,147 @@
1+// SPDX-License-Identifier: AGPL-3.0-or-later
2+
3+package usecase_test
4+
5+import (
6+ "bytes"
7+ "context"
8+ "errors"
9+ "testing"
10+ "time"
11+
12+ "bnema.dev/ermite/internal/domain"
13+ "bnema.dev/ermite/internal/ports"
14+ "bnema.dev/ermite/internal/ports/mocks"
15+ "bnema.dev/ermite/internal/usecase"
16+ "github.com/stretchr/testify/mock"
17+ "github.com/stretchr/testify/require"
18+)
19+
20+func (r *releaseRig) setClock(t *testing.T, at time.Time, blobs ports.BlobStore) {
21+ t.Helper()
22+ clock := mocks.NewClock(t)
23+ clock.EXPECT().Now().Return(at).Maybe()
24+ bus := mocks.NewEventBus(t)
25+ bus.EXPECT().Publish(mock.Anything, mock.Anything).Return().Maybe()
26+ r.clock = clock
27+ r.svc = usecase.NewReleases(r.store.Releases(), blobs, r.store, r.reader, r.tags, r.pushes, bus, clock, ports.ReleaseLimits{})
28+}
29+func TestReleaseSweepPendingAndOrphans(t *testing.T) {
30+ r := newReleaseRig(t, ports.ReleaseLimits{})
31+ rel := r.create(t, "v1", false)
32+ pending := r.upload(t, rel, "pending", []byte("abc"))
33+ good := r.upload(t, rel, "good", []byte("xyz"))
34+ fresh := r.upload(t, rel, "fresh", nil)
35+ old, err := r.store.Releases().GetAsset(r.ctx, pending.ID)
36+ require.NoError(t, err)
37+ old.State = domain.ReleaseAssetUploading
38+ mp, err := r.blobs.NewMultipart(r.ctx, old.BlobKey())
39+ require.NoError(t, err)
40+ old.MultipartID = mp.ID()
41+ fresh.State = domain.ReleaseAssetUploading
42+ fresh.UpdatedAt = time.Now().Add(25 * time.Hour)
43+ used, v, err := r.store.Releases().ReleaseUsage(r.ctx, r.repo)
44+ require.NoError(t, err)
45+ require.NoError(t, r.store.Releases().ApplyReleases(r.ctx, ports.ReleaseWrite{Repo: r.repo, Used: used, UsageVersion: v, PutAssets: []domain.ReleaseAsset{old, fresh}}))
46+ _, err = r.blobs.Put(r.ctx, "release/99999999999999999999", bytes.NewReader([]byte("orphan")), 6)
47+ require.NoError(t, err)
48+ // The clock is advanced beyond the disk adapter's real modification times.
49+ at := time.Now().Add(26 * time.Hour)
50+ r.setClock(t, at, r.blobs)
51+ r.svc.Sweep(r.ctx)
52+ _, err = r.store.Releases().GetAsset(r.ctx, pending.ID)
53+ require.ErrorIs(t, err, domain.ErrNotFound)
54+ _, err = r.blobs.Stat(r.ctx, pending.BlobKey())
55+ require.ErrorIs(t, err, domain.ErrNotFound)
56+ _, err = r.blobs.Stat(r.ctx, "release/99999999999999999999")
57+ require.ErrorIs(t, err, domain.ErrNotFound)
58+ _, err = r.store.Releases().GetAsset(r.ctx, good.ID)
59+ require.NoError(t, err)
60+ // The pending record inside the grace period is retained.
61+ _, err = r.store.Releases().GetAsset(r.ctx, fresh.ID)
62+ require.NoError(t, err)
63+ require.NoError(t, r.blobs.ListMultipart(r.ctx, "release/", func(ports.MultipartInfo) error { t.Error("multipart survived"); return nil }))
64+}
65+func TestReleaseSweepMissingBlobTwoObservations(t *testing.T) {
66+ r := newReleaseRig(t, ports.ReleaseLimits{})
67+ rel := r.create(t, "v1", false)
68+ a := r.upload(t, rel, "binary", []byte("abc"))
69+ require.NoError(t, r.blobs.Delete(r.ctx, a.BlobKey()))
70+ r.svc.Sweep(r.ctx)
71+ _, err := r.store.Releases().GetAsset(r.ctx, a.ID)
72+ require.NoError(t, err)
73+ r.svc.Sweep(r.ctx)
74+ _, err = r.store.Releases().GetAsset(r.ctx, a.ID)
75+ require.ErrorIs(t, err, domain.ErrNotFound)
76+ usage, err := r.svc.Usage(r.ctx, domain.Root(), r.repo)
77+ require.NoError(t, err)
78+ require.Zero(t, usage.Used)
79+}
80+
81+type releaseStatOverride struct {
82+ ports.BlobStore
83+ err error
84+}
85+
86+func (b *releaseStatOverride) Stat(ctx context.Context, key string) (ports.BlobInfo, error) {
87+ if b.err != nil {
88+ return ports.BlobInfo{}, b.err
89+ }
90+ return b.BlobStore.Stat(ctx, key)
91+}
92+func TestReleaseSweepTransientStatAndRecovery(t *testing.T) {
93+ for _, recover := range []bool{false, true} {
94+ t.Run(map[bool]string{false: "transient preserves miss", true: "success resets miss"}[recover], func(t *testing.T) {
95+ r := newReleaseRig(t, ports.ReleaseLimits{})
96+ rel := r.create(t, "v1", false)
97+ a := r.upload(t, rel, "binary", []byte("abc"))
98+ b := &releaseStatOverride{BlobStore: r.blobs, err: domain.ErrNotFound}
99+ r.setClock(t, now, b)
100+ r.svc.Sweep(r.ctx)
101+ b.err = errors.New("outage")
102+ r.svc.Sweep(r.ctx)
103+ _, err := r.store.Releases().GetAsset(r.ctx, a.ID)
104+ require.NoError(t, err)
105+ if recover {
106+ b.err = nil
107+ r.svc.Sweep(r.ctx)
108+ }
109+ b.err = domain.ErrNotFound
110+ r.svc.Sweep(r.ctx)
111+ _, err = r.store.Releases().GetAsset(r.ctx, a.ID)
112+ if recover {
113+ require.NoError(t, err)
114+ } else {
115+ require.ErrorIs(t, err, domain.ErrNotFound)
116+ }
117+ })
118+ }
119+}
120+func TestReleaseSweepGraceAndFailureIsolation(t *testing.T) {
121+ r := newReleaseRig(t, ports.ReleaseLimits{})
122+ rel := r.create(t, "v1", false)
123+ a := r.upload(t, rel, "pending", nil)
124+ a.State = domain.ReleaseAssetUploading
125+ used, v, err := r.store.Releases().ReleaseUsage(r.ctx, r.repo)
126+ require.NoError(t, err)
127+ require.NoError(t, r.store.Releases().ApplyReleases(r.ctx, ports.ReleaseWrite{Repo: r.repo, Used: used, UsageVersion: v, PutAssets: []domain.ReleaseAsset{a}}))
128+ b := mocks.NewBlobStore(t)
129+ b.EXPECT().List(r.ctx, "release/", mock.Anything).RunAndReturn(func(_ context.Context, _ string, fn func(ports.BlobInfo) error) error {
130+ require.NoError(t, fn(ports.BlobInfo{Key: "release/old", ModTime: now.Add(-25 * time.Hour)}))
131+ require.NoError(t, fn(ports.BlobInfo{Key: "release/new", ModTime: now.Add(23 * time.Hour)}))
132+ return nil
133+ }).Once()
134+ b.EXPECT().Delete(r.ctx, "release/old").Return(errors.New("failed deletion")).Once()
135+ b.EXPECT().ListMultipart(r.ctx, "release/", mock.Anything).RunAndReturn(func(_ context.Context, _ string, fn func(ports.MultipartInfo) error) error {
136+ require.NoError(t, fn(ports.MultipartInfo{Key: "release/old", ID: "old", Started: now.Add(-25 * time.Hour)}))
137+ require.NoError(t, fn(ports.MultipartInfo{Key: "release/new", ID: "new", Started: now}))
138+ return nil
139+ }).Once()
140+ b.EXPECT().AbortMultipart(r.ctx, "release/old", "old").Return(nil).Once()
141+ r.setClock(t, now.Add(24*time.Hour), b)
142+ // Use times relative to the sweep, keeping both new objects within grace.
143+ // pending exactly 24h old survives; failures in orphan deletion do not stop multipart GC.
144+ r.svc.Sweep(r.ctx)
145+ _, err = r.store.Releases().GetAsset(r.ctx, a.ID)
146+ require.NoError(t, err)
147+}
A internal/usecase/releases_test.go +57 -0
@@ -0,0 +1,57 @@
1+// SPDX-License-Identifier: AGPL-3.0-or-later
2+
3+package usecase_test
4+
5+import (
6+ "context"
7+ "testing"
8+
9+ "bnema.dev/ermite/internal/domain"
10+ "bnema.dev/ermite/internal/ports/mocks"
11+ "github.com/stretchr/testify/mock"
12+ "github.com/stretchr/testify/require"
13+)
14+
15+func TestReleaseTagDeletionHook(t *testing.T) {
16+ ctx := context.Background()
17+ n := name(t, "tools")
18+ f := newPushFixture(t)
19+ trigger := mocks.NewReleaseTagTrigger(t)
20+ f.pushes.WithReleases(trigger)
21+ updates := []domain.RefUpdate{{Old: oidA, New: domain.ZeroOID, Ref: "refs/tags/v1"}}
22+ expectPushMutation(t, f.meta, domain.Repo{Name: n}, nil)
23+ f.git.EXPECT().DefaultBranch(ctx, n).Return("main", true, nil)
24+ trigger.EXPECT().OnRefsMoved(ctx, n, updates).Return().Once()
25+ f.bus.EXPECT().Publish(ctx, mock.Anything).Return().Once()
26+ f.sched.EXPECT().Schedule(ctx, n).Return().Once()
27+ require.NoError(t, f.pushes.Record(ctx, domain.Root(), n, updates))
28+}
29+
30+func TestReleaseRepositoryDeletionHook(t *testing.T) {
31+ for _, failed := range []bool{false, true} {
32+ t.Run(map[bool]string{false: "deleted", true: "metadata failure"}[failed], func(t *testing.T) {
33+ ctx := context.Background()
34+ n := name(t, "tools")
35+ f := newFixture(t)
36+ cleaner := mocks.NewReleaseRepoCleaner(t)
37+ f.repos.WithReleases(cleaner)
38+ assets := []domain.ReleaseAsset{{ID: 42, Repo: n, ReleaseID: 1}}
39+ f.meta.EXPECT().Get(ctx, n).Return(domain.Repo{Name: n}, nil)
40+ f.git.EXPECT().Trash(ctx, n).Return(nil)
41+ cleaner.EXPECT().AssetsOf(ctx, n).Return(assets, nil).Once()
42+ if failed {
43+ f.meta.EXPECT().Delete(ctx, n).Return(domain.ErrUnavailable).Once()
44+ require.ErrorIs(t, f.repos.Delete(ctx, domain.Root(), n), domain.ErrUnavailable)
45+ } else {
46+ deleted := false
47+ f.meta.EXPECT().Delete(ctx, n).Run(func(context.Context, domain.RepoName) { deleted = true }).Return(nil).Once()
48+ cleaner.EXPECT().OnRepoDeleted(mock.Anything, n, assets).Run(func(ctx context.Context, _ domain.RepoName, _ []domain.ReleaseAsset) {
49+ require.True(t, deleted)
50+ require.Nil(t, ctx.Done())
51+ }).Return().Once()
52+ f.bus.EXPECT().Publish(ctx, mock.Anything).Return().Once()
53+ require.NoError(t, f.repos.Delete(ctx, domain.Root(), n))
54+ }
55+ })
56+ }
57+}
M internal/usecase/repos.go +9 -1
@@ -24,7 +24,8 @@ type Repos struct {
2424 ci ports.CIRunCleaner // nil: no CI logs to remove
2525 autoMerge ports.AutoMergeTrigger
2626 mergeGate ports.MergeGate
27- mirror ports.MirrorTrigger // nil: no outbound mirror
27+ mirror ports.MirrorTrigger // nil: no outbound mirror
28+ releases ports.ReleaseRepoCleaner // nil: no release blobs to remove
2829 }
2930
3031 var _ ports.RepoService = (*Repos)(nil)
@@ -42,6 +43,13 @@ func (r *Repos) WithMirror(t ports.MirrorTrigger) *Repos {
4243 return r
4344 }
4445
46+// WithReleases makes Delete remove release blobs after metadata deletion.
47+// Call it before the first request; it returns r.
48+func (r *Repos) WithReleases(c ports.ReleaseRepoCleaner) *Repos {
49+ r.releases = c
50+ return r
51+}
52+
4553 // WithCIRunCleaner makes Delete remove the CI logs of the repository. Call it
4654 // before the first request; it returns r.
4755 func (r *Repos) WithCIRunCleaner(c ports.CIRunCleaner) *Repos {
M internal/usecase/repos_write.go +11 -0
@@ -226,9 +226,20 @@ func (r *Repos) Delete(ctx context.Context, p domain.Principal, name domain.Repo
226226 log.Warn().Err(err).Str("repo", name.String()).Msg("list the CI runs of the repository; their logs are left to the sweep")
227227 }
228228 }
229+ var assets []domain.ReleaseAsset
230+ if r.releases != nil {
231+ var err error
232+ assets, err = r.releases.AssetsOf(ctx, name)
233+ if err != nil {
234+ log.Warn().Err(err).Str("repo", name.String()).Msg("list release assets; orphan sweep will remove their blobs")
235+ }
236+ }
229237 if err := r.meta.Delete(ctx, name); err != nil {
230238 return fmt.Errorf("delete metadata of %q: %w", name, err)
231239 }
240+ if r.releases != nil {
241+ r.releases.OnRepoDeleted(context.WithoutCancel(ctx), name, assets)
242+ }
232243 if r.ci != nil {
233244 r.ci.ForgetRepo(name)
234245 }
M proto/ermite/storage/ci_artifact.proto +4 -2
@@ -1,8 +1,8 @@
11 // SPDX-License-Identifier: AGPL-3.0-or-later
22
33 // Internal, unversioned storage schema of the CI artifacts (artifacts v4,
4-// CI-4 D3). The contents are not stored here: they are files under
5-// $ERMITE_DATA_DIR/ci/artifacts.
4+// CI-4 D3). Contents are stored separately, in the blob store or (for
5+// legacy records) under $ERMITE_DATA_DIR/ci/artifacts.
66
77 syntax = "proto3";
88
@@ -49,6 +49,8 @@ message CIArtifact {
4949 // list_digest: the SHA-256 of the block list of the commit, to recognize a
5050 // repeated commit.
5151 bytes list_digest = 15;
52+ // "" or "disk": legacy files under ci/artifacts; "blob": the blob store.
53+ string store = 16;
5254 }
5355
5456 // CIArtifactBlock is stored under "artifactblk/<artifact:020d>/<hex(block id)>":
A proto/ermite/storage/release.proto +46 -0
@@ -0,0 +1,46 @@
1+// SPDX-License-Identifier: AGPL-3.0-or-later
2+
3+syntax = "proto3";
4+package ermite.storage;
5+import "google/protobuf/timestamp.proto";
6+option go_package = "bnema.dev/ermite/gen/ermite/storage;storagepb";
7+
8+// Release stores release metadata, not object contents.
9+message Release {
10+ int64 id = 1;
11+ string repo = 2;
12+ string tag = 3;
13+ string title = 4;
14+ string notes = 5;
15+ bool draft = 6;
16+ bool prerelease = 7;
17+ bool make_latest = 8;
18+ string target_commitish = 9;
19+ string author = 10;
20+ google.protobuf.Timestamp created_at = 11;
21+ google.protobuf.Timestamp published_at = 12;
22+ google.protobuf.Timestamp updated_at = 13;
23+}
24+
25+// ReleaseAsset stores release metadata, not object contents.
26+message ReleaseAsset {
27+ int64 id = 1;
28+ int64 release_id = 2;
29+ string repo = 3;
30+ string name = 4;
31+ string label = 5;
32+ string content_type = 6;
33+ int64 size = 7;
34+ string digest = 8;
35+ string state = 9;
36+ string uploader = 10;
37+ int64 charged = 11;
38+ string multipart_id = 12;
39+ google.protobuf.Timestamp created_at = 13;
40+ google.protobuf.Timestamp updated_at = 14;
41+}
42+
43+// ReleaseUsage stores release metadata, not object contents.
44+message ReleaseUsage {
45+ int64 used = 1;
46+}
M skills/ermite/CLI.md +20 -0
@@ -44,6 +44,7 @@ Scopes:
4444 | `mr:merge` | merge, and delete the merged source branch with `--delete-branch` |
4545 | `mr:force-merge` | bypass failing or unfinished checks with `--ignore-checks` |
4646 | `issue:write` | open, comment on, close and reopen issues |
47+| `release:write` | create, edit and delete releases and their assets; included in `cli`; refused for branch-limited tokens |
4748 | `token:admin` | create, list and revoke tokens |
4849 | `ci:write`, `secrets:write`, `runner:register` | reserved for CI |
4950
@@ -93,6 +94,25 @@ Issues share the numbering of merge requests and must be enabled per repository:
9394
9495 Issue JSON: `repo`, `number`, `title`, `body`, `state`, `author` (token label), `created_at`, `updated_at`, `comment_count`. `issue show --json` adds `comments` (`id`, `author`, `text`, `created_at`). `repo show --json` has `issues`.
9596
97+## Releases
98+
99+Needs the `releases` server feature. Read access follows repository visibility;
100+drafts need write access. Writes need `release:write`, never a branch-limited
101+token, or a CI job's `contents: write` permission. Select a release by tag or id;
102+`view` and `download` also accept `latest`.
103+
104+| Command | Notes |
105+|---|---|
106+| `release create <repo> <tag> [--title T] [--notes TEXT\|--notes-file F\|-] [--target REF] [--draft] [--prerelease] [--latest] [--asset FILE[#LABEL]] [--json]` | existing tags never move; target creates a missing tag |
107+| `release list <repo> [--limit N] [--json]` | newest first |
108+| `release view <repo> <tag\|id\|latest> [--json\|--notes]` | metadata and assets |
109+| `release edit <repo> <tag\|id> [--title T] [--notes TEXT\|--notes-file F\|-] [--tag T] [--target REF] [--draft\|--publish] [--prerelease\|--no-prerelease] [--latest]` | only supplied fields change |
110+| `release delete <repo> <tag\|id> [--yes]` | leaves the Git tag |
111+| `release upload <repo> <tag\|id> <file[#label]>… [--clobber] [--content-type TYPE]` | clobber replaces same-named assets |
112+| `release download <repo> <tag\|id\|latest> [--name N]… [--pattern GLOB]… [--dir D]` | refuses existing local files |
113+
114+See [publishing and storage](../../docs/releases.md).
115+
96116 ## Workflows
97117
98118 | Command | Scope | Notes |

Comments0 open / 0

no comments yet.

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
l
merge request: the list of files (j k move, enter opens, esc back)
M
merge request: the merge menu; there 1–3 pick the mode, D deletes the branch, enter merges
R
merge request: mark the head read
esc
close a dialog, leave a field

Branches 3

Tags 41

↑↓ moveenter openesc close