1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package rpc
4
5import (
6 "context"
7 "errors"
8 "fmt"
9 "io"
10
11 "connectrpc.com/connect"
12
13 pb "bnema.dev/ermite-api/gen/ermite"
14 "bnema.dev/ermite-api/gen/ermite/ermiteconnect"
15 "bnema.dev/ermite/internal/domain"
16 "bnema.dev/ermite/internal/ports"
17)
18
19// FeatureReleases is listed by GetVersion when ReleaseService is served.
20const FeatureReleases = "releases"
21
22// releaseService implements ermiteconnect.ReleaseServiceHandler.
23type releaseService struct{ rel ports.ReleaseService }
24
25var _ ermiteconnect.ReleaseServiceHandler = (*releaseService)(nil)
26
27// releaseMessage fills the computed latest flag and uploaded assets using the
28// same caller's read permissions. No latest release is a normal repository state.
29func (s *releaseService) releaseMessage(ctx context.Context, repo domain.RepoName, r domain.Release) (*pb.Release, error) {
30 latest, err := s.latestID(ctx, repo)
31 if err != nil {
32 return nil, err
33 }
34 return s.releaseRow(ctx, repo, r, latest)
35}
36
37// latestID is the id of the latest release, or 0 when the repository has none.
38func (s *releaseService) latestID(ctx context.Context, repo domain.RepoName) (int64, error) {
39 latest, err := s.rel.GetRelease(ctx, caller(ctx), repo, ports.ReleaseRef{Latest: true})
40 if err != nil && !errors.Is(err, domain.ErrNotFound) {
41 return 0, err
42 }
43 return latest.ID, nil
44}
45
46// releaseRow converts one release given the id of the latest one, so a list
47// resolves the latest release once.
48func (s *releaseService) releaseRow(ctx context.Context, repo domain.RepoName, r domain.Release, latest int64) (*pb.Release, error) {
49 assets, err := s.rel.ListAssets(ctx, caller(ctx), repo, r.ID)
50 if err != nil {
51 return nil, err
52 }
53 out := releaseToPB(r)
54 out.Latest = latest != 0 && latest == r.ID
55 out.Assets = assetsToPB(assets)
56 return out, nil
57}
58
59func (s *releaseService) CreateRelease(ctx context.Context, req *connect.Request[pb.CreateReleaseRequest]) (*connect.Response[pb.CreateReleaseResponse], error) {
60 m := req.Msg
61 repo, err := repoOf(m.GetRepo())
62 if err != nil {
63 return nil, connectError(ctx, err)
64 }
65 r, err := s.rel.CreateRelease(ctx, caller(ctx), repo, ports.CreateReleaseParams{Tag: m.GetTag(), TargetCommitish: m.GetTargetCommitish(), Title: m.GetTitle(), Notes: m.GetNotes(), Draft: m.GetDraft(), Prerelease: m.GetPrerelease(), MakeLatest: m.GetMakeLatest()})
66 if err != nil {
67 return nil, connectError(ctx, err)
68 }
69 out, err := s.releaseMessage(ctx, repo, r)
70 if err != nil {
71 return nil, connectError(ctx, err)
72 }
73 return connect.NewResponse(&pb.CreateReleaseResponse{Release: out}), nil
74}
75
76func releaseRef(m *pb.GetReleaseRequest) (ports.ReleaseRef, error) {
77 switch v := m.GetSelector().(type) {
78 case *pb.GetReleaseRequest_Id:
79 if v.Id > 0 {
80 return ports.ReleaseRef{ID: v.Id}, nil
81 }
82 case *pb.GetReleaseRequest_Tag:
83 if v.Tag != "" {
84 return ports.ReleaseRef{Tag: v.Tag}, nil
85 }
86 case *pb.GetReleaseRequest_Latest:
87 if v.Latest {
88 return ports.ReleaseRef{Latest: true}, nil
89 }
90 }
91 return ports.ReleaseRef{}, fmt.Errorf("%w: select an id, tag or latest=true", domain.ErrInvalid)
92}
93
94func (s *releaseService) GetRelease(ctx context.Context, req *connect.Request[pb.GetReleaseRequest]) (*connect.Response[pb.GetReleaseResponse], error) {
95 repo, err := repoOf(req.Msg.GetRepo())
96 if err != nil {
97 return nil, connectError(ctx, err)
98 }
99 ref, err := releaseRef(req.Msg)
100 if err != nil {
101 return nil, connectError(ctx, err)
102 }
103 r, err := s.rel.GetRelease(ctx, caller(ctx), repo, ref)
104 if err != nil {
105 return nil, connectError(ctx, err)
106 }
107 out, err := s.releaseMessage(ctx, repo, r)
108 if err != nil {
109 return nil, connectError(ctx, err)
110 }
111 return connect.NewResponse(&pb.GetReleaseResponse{Release: out}), nil
112}
113
114func (s *releaseService) ListReleases(ctx context.Context, req *connect.Request[pb.ListReleasesRequest]) (*connect.Response[pb.ListReleasesResponse], error) {
115 repo, err := repoOf(req.Msg.GetRepo())
116 if err != nil {
117 return nil, connectError(ctx, err)
118 }
119 page, err := s.rel.ListReleases(ctx, caller(ctx), repo, int(req.Msg.GetPageSize()), req.Msg.GetPageToken())
120 if err != nil {
121 return nil, connectError(ctx, err)
122 }
123 out := &pb.ListReleasesResponse{Releases: make([]*pb.Release, 0, len(page.Releases)), NextPageToken: page.NextPageToken}
124 latest, err := s.latestID(ctx, repo)
125 if err != nil {
126 return nil, connectError(ctx, err)
127 }
128 for _, r := range page.Releases {
129 m, err := s.releaseRow(ctx, repo, r, latest)
130 if err != nil {
131 return nil, connectError(ctx, err)
132 }
133 out.Releases = append(out.Releases, m)
134 }
135 return connect.NewResponse(out), nil
136}
137
138func releaseUpdateParams(m *pb.UpdateReleaseRequest) (ports.UpdateReleaseParams, error) {
139 var p ports.UpdateReleaseParams
140 if len(m.GetUpdateMask().GetPaths()) == 0 {
141 return p, fmt.Errorf("%w: update_mask is empty", domain.ErrInvalid)
142 }
143 for _, path := range m.GetUpdateMask().GetPaths() {
144 switch path {
145 case "tag":
146 p.Tag = new(m.GetTag())
147 case "target_commitish":
148 p.TargetCommitish = new(m.GetTargetCommitish())
149 case "title":
150 p.Title = new(m.GetTitle())
151 case "notes":
152 p.Notes = new(m.GetNotes())
153 case "draft":
154 p.Draft = new(m.GetDraft())
155 case "prerelease":
156 p.Prerelease = new(m.GetPrerelease())
157 case "make_latest":
158 p.MakeLatest = new(m.GetMakeLatest())
159 default:
160 return p, fmt.Errorf("%w: unknown update_mask path %q", domain.ErrInvalid, path)
161 }
162 }
163 return p, nil
164}
165
166func (s *releaseService) UpdateRelease(ctx context.Context, req *connect.Request[pb.UpdateReleaseRequest]) (*connect.Response[pb.UpdateReleaseResponse], error) {
167 repo, err := repoOf(req.Msg.GetRepo())
168 if err != nil {
169 return nil, connectError(ctx, err)
170 }
171 p, err := releaseUpdateParams(req.Msg)
172 if err != nil {
173 return nil, connectError(ctx, err)
174 }
175 r, err := s.rel.UpdateRelease(ctx, caller(ctx), repo, req.Msg.GetId(), p)
176 if err != nil {
177 return nil, connectError(ctx, err)
178 }
179 out, err := s.releaseMessage(ctx, repo, r)
180 if err != nil {
181 return nil, connectError(ctx, err)
182 }
183 return connect.NewResponse(&pb.UpdateReleaseResponse{Release: out}), nil
184}
185
186func (s *releaseService) DeleteRelease(ctx context.Context, req *connect.Request[pb.DeleteReleaseRequest]) (*connect.Response[pb.DeleteReleaseResponse], error) {
187 repo, err := repoOf(req.Msg.GetRepo())
188 if err != nil {
189 return nil, connectError(ctx, err)
190 }
191 if err := s.rel.DeleteRelease(ctx, caller(ctx), repo, req.Msg.GetId()); err != nil {
192 return nil, connectError(ctx, err)
193 }
194 return connect.NewResponse(&pb.DeleteReleaseResponse{}), nil
195}
196
197// chunkReader retains at most one message; metadata belongs to the first only.
198type chunkReader struct {
199 stream *connect.ClientStream[pb.UploadReleaseAssetRequest]
200 buf []byte
201}
202
203func (r *chunkReader) Read(p []byte) (int, error) {
204 if len(p) == 0 {
205 return 0, nil
206 }
207 for len(r.buf) == 0 {
208 if !r.stream.Receive() {
209 if err := r.stream.Err(); err != nil {
210 return 0, err
211 }
212 return 0, io.EOF
213 }
214 m := r.stream.Msg()
215 if m.GetRepo() != "" || m.GetReleaseId() != 0 || m.GetName() != "" || m.GetLabel() != "" || m.GetContentType() != "" || m.GetSize() != 0 || m.GetReplace() {
216 return 0, fmt.Errorf("%w: upload metadata is in the first message only", domain.ErrInvalid)
217 }
218 if err := checkBlockData(m.GetData()); err != nil {
219 return 0, err
220 }
221 r.buf = m.GetData()
222 }
223 n := copy(p, r.buf)
224 r.buf = r.buf[n:]
225 return n, nil
226}
227
228func (s *releaseService) UploadReleaseAsset(ctx context.Context, stream *connect.ClientStream[pb.UploadReleaseAssetRequest]) (*connect.Response[pb.UploadReleaseAssetResponse], error) {
229 if !stream.Receive() {
230 err := stream.Err()
231 if err == nil {
232 err = fmt.Errorf("%w: the stream has no message", domain.ErrInvalid)
233 }
234 return nil, connectError(ctx, err)
235 }
236 m := stream.Msg()
237 repo, err := repoOf(m.GetRepo())
238 if err != nil {
239 return nil, connectError(ctx, err)
240 }
241 if m.GetSize() < 0 {
242 return nil, connectError(ctx, fmt.Errorf("%w: size must be nonnegative", domain.ErrInvalid))
243 }
244 if err := checkBlockData(m.GetData()); err != nil {
245 return nil, connectError(ctx, err)
246 }
247 a, err := s.rel.UploadAsset(ctx, caller(ctx), repo, ports.UploadAssetParams{ReleaseID: m.GetReleaseId(), Name: m.GetName(), Label: m.GetLabel(), ContentType: m.GetContentType(), Size: m.GetSize(), Replace: m.GetReplace()}, &chunkReader{stream: stream, buf: m.GetData()})
248 if err != nil {
249 return nil, connectError(ctx, err)
250 }
251 return connect.NewResponse(&pb.UploadReleaseAssetResponse{Asset: assetToPB(a)}), nil
252}
253
254func assetUpdateParams(m *pb.UpdateReleaseAssetRequest) (ports.UpdateAssetParams, error) {
255 var p ports.UpdateAssetParams
256 if len(m.GetUpdateMask().GetPaths()) == 0 {
257 return p, fmt.Errorf("%w: update_mask is empty", domain.ErrInvalid)
258 }
259 for _, path := range m.GetUpdateMask().GetPaths() {
260 switch path {
261 case "name":
262 p.Name = new(m.GetName())
263 case "label":
264 p.Label = new(m.GetLabel())
265 default:
266 return p, fmt.Errorf("%w: unknown update_mask path %q", domain.ErrInvalid, path)
267 }
268 }
269 return p, nil
270}
271
272func (s *releaseService) UpdateReleaseAsset(ctx context.Context, req *connect.Request[pb.UpdateReleaseAssetRequest]) (*connect.Response[pb.UpdateReleaseAssetResponse], error) {
273 repo, err := repoOf(req.Msg.GetRepo())
274 if err != nil {
275 return nil, connectError(ctx, err)
276 }
277 p, err := assetUpdateParams(req.Msg)
278 if err != nil {
279 return nil, connectError(ctx, err)
280 }
281 a, err := s.rel.UpdateAsset(ctx, caller(ctx), repo, req.Msg.GetAssetId(), p)
282 if err != nil {
283 return nil, connectError(ctx, err)
284 }
285 return connect.NewResponse(&pb.UpdateReleaseAssetResponse{Asset: assetToPB(a)}), nil
286}
287
288func (s *releaseService) DeleteReleaseAsset(ctx context.Context, req *connect.Request[pb.DeleteReleaseAssetRequest]) (*connect.Response[pb.DeleteReleaseAssetResponse], error) {
289 repo, err := repoOf(req.Msg.GetRepo())
290 if err != nil {
291 return nil, connectError(ctx, err)
292 }
293 if err := s.rel.DeleteAsset(ctx, caller(ctx), repo, req.Msg.GetAssetId()); err != nil {
294 return nil, connectError(ctx, err)
295 }
296 return connect.NewResponse(&pb.DeleteReleaseAssetResponse{}), nil
297}
298
299func assetRef(m *pb.DownloadReleaseAssetRequest) (ports.AssetRef, error) {
300 switch v := m.GetSelector().(type) {
301 case *pb.DownloadReleaseAssetRequest_AssetId:
302 if v.AssetId > 0 {
303 return ports.AssetRef{ID: v.AssetId}, nil
304 }
305 case *pb.DownloadReleaseAssetRequest_Tag:
306 if v.Tag.GetTag() != "" && v.Tag.GetName() != "" {
307 return ports.AssetRef{Tag: v.Tag.GetTag(), Name: v.Tag.GetName()}, nil
308 }
309 case *pb.DownloadReleaseAssetRequest_Latest:
310 if v.Latest.GetLatest() && v.Latest.GetName() != "" {
311 return ports.AssetRef{Latest: true, Name: v.Latest.GetName()}, nil
312 }
313 }
314 return ports.AssetRef{}, fmt.Errorf("%w: select an asset id, tag/name or latest=true/name", domain.ErrInvalid)
315}
316
317func (s *releaseService) DownloadReleaseAsset(ctx context.Context, req *connect.Request[pb.DownloadReleaseAssetRequest], stream *connect.ServerStream[pb.DownloadReleaseAssetResponse]) error {
318 repo, err := repoOf(req.Msg.GetRepo())
319 if err != nil {
320 return connectError(ctx, err)
321 }
322 ref, err := assetRef(req.Msg)
323 if err != nil {
324 return connectError(ctx, err)
325 }
326 err = s.rel.DownloadAsset(ctx, caller(ctx), repo, ref, req.Msg.GetOffset(), -1, func(c ports.AssetChunk) error {
327 m := &pb.DownloadReleaseAssetResponse{Chunk: c.Data}
328 if c.Asset != nil {
329 m.Asset = assetToPB(*c.Asset)
330 }
331 return stream.Send(m)
332 })
333 return connectError(ctx, err)
334}
335
336func (s *releaseService) ListReleaseAssets(ctx context.Context, req *connect.Request[pb.ListReleaseAssetsRequest]) (*connect.Response[pb.ListReleaseAssetsResponse], error) {
337 repo, err := repoOf(req.Msg.GetRepo())
338 if err != nil {
339 return nil, connectError(ctx, err)
340 }
341 assets, err := s.rel.ListAssets(ctx, caller(ctx), repo, req.Msg.GetReleaseId())
342 if err != nil {
343 return nil, connectError(ctx, err)
344 }
345 return connect.NewResponse(&pb.ListReleaseAssetsResponse{Assets: assetsToPB(assets)}), nil
346}
347
348func (s *releaseService) GetReleaseUsage(ctx context.Context, req *connect.Request[pb.GetReleaseUsageRequest]) (*connect.Response[pb.GetReleaseUsageResponse], error) {
349 repo, err := repoOf(req.Msg.GetRepo())
350 if err != nil {
351 return nil, connectError(ctx, err)
352 }
353 u, err := s.rel.Usage(ctx, caller(ctx), repo)
354 if err != nil {
355 return nil, connectError(ctx, err)
356 }
357 return connect.NewResponse(&pb.GetReleaseUsageResponse{UsedBytes: u.Used, QuotaBytes: u.Limits.Quota, AssetMaxBytes: u.Limits.AssetMax, AssetsMax: int32(u.Limits.AssetsMax)}), nil
358}