1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package rpc
4
5import (
6 "context"
7 "fmt"
8 "io"
9
10 "connectrpc.com/connect"
11
12 pb "bnema.dev/ermite-api/gen/ermite"
13 "bnema.dev/ermite/internal/domain"
14 "bnema.dev/ermite/internal/ports"
15)
16
17// FeatureCIArtifacts is listed by GetVersion when the artifact services are
18// served.
19const FeatureCIArtifacts = "ci.artifacts"
20
21func artifactToPB(a domain.CIArtifact) *pb.Artifact {
22 return &pb.Artifact{
23 Id: a.ID, Repo: a.Repo.String(), RunId: a.RunID, Attempt: a.Attempt, JobId: a.JobID,
24 Name: a.Name, Size: a.Size, Digest: a.Digest, MediaType: a.MediaType,
25 CreatedAt: timestamp(a.CreatedAt), ExpiresAt: timestamp(a.ExpiresAt),
26 }
27}
28
29// jobArtifactService implements ermiteconnect.JobArtifactServiceHandler.
30type jobArtifactService struct {
31 artifacts ports.JobArtifactService
32 runner *runnerService // failed: the failures of job tokens count like the runner's
33}
34
35func (s *jobArtifactService) CreateJobArtifact(ctx context.Context, req *connect.Request[pb.CreateJobArtifactRequest]) (*connect.Response[pb.CreateJobArtifactResponse], error) {
36 out, err := s.artifacts.CreateJobArtifact(ctx, secretFrom(ctx), ports.CreateJobArtifactParams{
37 JobID: req.Msg.GetJobId(), Name: req.Msg.GetName(), MediaType: req.Msg.GetMediaType(), ExpiresAt: goTime(req.Msg.GetExpiresAt()),
38 })
39 if err != nil {
40 return nil, s.runner.failed(ctx, err)
41 }
42 return connect.NewResponse(&pb.CreateJobArtifactResponse{ArtifactId: out.ID, ExpiresAt: timestamp(out.ExpiresAt)}), nil
43}
44
45// blockReader reads the data of a block stream: the first message's, then
46// the following ones', which must carry data only.
47type blockReader struct {
48 stream *connect.ClientStream[pb.UploadJobArtifactBlockRequest]
49 buf []byte
50}
51
52func checkBlockData(data []byte) error {
53 if len(data) > domain.CIArtifactChunk {
54 return fmt.Errorf("%w: a message carries at most %d bytes of data", domain.ErrInvalid, domain.CIArtifactChunk)
55 }
56 return nil
57}
58
59func (r *blockReader) Read(p []byte) (int, error) {
60 for len(r.buf) == 0 {
61 if !r.stream.Receive() {
62 if err := r.stream.Err(); err != nil {
63 return 0, err
64 }
65 return 0, io.EOF
66 }
67 m := r.stream.Msg()
68 if m.GetJobId() != "" || m.GetArtifactId() != 0 || len(m.GetBlockId()) > 0 {
69 return 0, fmt.Errorf("%w: job_id, artifact_id and block_id are in the first message only", domain.ErrInvalid)
70 }
71 if err := checkBlockData(m.GetData()); err != nil {
72 return 0, err
73 }
74 r.buf = m.GetData()
75 }
76 n := copy(p, r.buf)
77 r.buf = r.buf[n:]
78 return n, nil
79}
80
81func (s *jobArtifactService) UploadJobArtifactBlock(ctx context.Context, stream *connect.ClientStream[pb.UploadJobArtifactBlockRequest]) (*connect.Response[pb.UploadJobArtifactBlockResponse], error) {
82 if !stream.Receive() {
83 err := stream.Err()
84 if err == nil {
85 err = fmt.Errorf("%w: the stream has no message", domain.ErrInvalid)
86 }
87 return nil, s.runner.failed(ctx, err)
88 }
89 first := stream.Msg()
90 if err := checkBlockData(first.GetData()); err != nil {
91 return nil, s.runner.failed(ctx, err)
92 }
93 r := &blockReader{stream: stream, buf: first.GetData()}
94 size, err := s.artifacts.UploadJobArtifactBlock(ctx, secretFrom(ctx), first.GetJobId(), first.GetArtifactId(), first.GetBlockId(), r)
95 if err != nil {
96 return nil, s.runner.failed(ctx, err)
97 }
98 return connect.NewResponse(&pb.UploadJobArtifactBlockResponse{Size: size}), nil
99}
100
101func (s *jobArtifactService) CommitJobArtifact(ctx context.Context, req *connect.Request[pb.CommitJobArtifactRequest]) (*connect.Response[pb.CommitJobArtifactResponse], error) {
102 size, digest, err := s.artifacts.CommitJobArtifact(ctx, secretFrom(ctx), req.Msg.GetJobId(), req.Msg.GetArtifactId(), req.Msg.GetBlockIds())
103 if err != nil {
104 return nil, s.runner.failed(ctx, err)
105 }
106 return connect.NewResponse(&pb.CommitJobArtifactResponse{Size: size, Digest: digest}), nil
107}
108
109func (s *jobArtifactService) FinalizeJobArtifact(ctx context.Context, req *connect.Request[pb.FinalizeJobArtifactRequest]) (*connect.Response[pb.FinalizeJobArtifactResponse], error) {
110 a, err := s.artifacts.FinalizeJobArtifact(ctx, secretFrom(ctx), req.Msg.GetJobId(), req.Msg.GetName(), req.Msg.GetSize(), req.Msg.GetDigest())
111 if err != nil {
112 return nil, s.runner.failed(ctx, err)
113 }
114 return connect.NewResponse(&pb.FinalizeJobArtifactResponse{Artifact: artifactToPB(a)}), nil
115}
116
117func (s *jobArtifactService) ListJobArtifacts(ctx context.Context, req *connect.Request[pb.ListJobArtifactsRequest]) (*connect.Response[pb.ListJobArtifactsResponse], error) {
118 arts, err := s.artifacts.ListJobArtifacts(ctx, secretFrom(ctx), req.Msg.GetJobId(), req.Msg.GetName(), req.Msg.GetArtifactId())
119 if err != nil {
120 return nil, s.runner.failed(ctx, err)
121 }
122 out := &pb.ListJobArtifactsResponse{Artifacts: make([]*pb.Artifact, 0, len(arts))}
123 for _, a := range arts {
124 out.Artifacts = append(out.Artifacts, artifactToPB(a))
125 }
126 return connect.NewResponse(out), nil
127}
128
129func (s *jobArtifactService) DownloadJobArtifact(ctx context.Context, req *connect.Request[pb.DownloadJobArtifactRequest], stream *connect.ServerStream[pb.DownloadJobArtifactResponse]) error {
130 err := s.artifacts.DownloadJobArtifact(ctx, secretFrom(ctx), req.Msg.GetJobId(), req.Msg.GetArtifactId(), req.Msg.GetOffset(), func(c ports.ArtifactChunk) error {
131 m := &pb.DownloadJobArtifactResponse{Chunk: c.Data}
132 if c.Artifact != nil {
133 m.Artifact = artifactToPB(*c.Artifact)
134 }
135 return stream.Send(m)
136 })
137 return s.runner.failed(ctx, err)
138}
139
140func (s *jobArtifactService) DeleteJobArtifact(ctx context.Context, req *connect.Request[pb.DeleteJobArtifactRequest]) (*connect.Response[pb.DeleteJobArtifactResponse], error) {
141 id, err := s.artifacts.DeleteJobArtifact(ctx, secretFrom(ctx), req.Msg.GetJobId(), req.Msg.GetName())
142 if err != nil {
143 return nil, s.runner.failed(ctx, err)
144 }
145 return connect.NewResponse(&pb.DeleteJobArtifactResponse{ArtifactId: id}), nil
146}
147
148// artifactService implements ermiteconnect.ArtifactServiceHandler.
149type artifactService struct {
150 artifacts ports.ArtifactService
151}
152
153func (s *artifactService) ListArtifacts(ctx context.Context, req *connect.Request[pb.ListArtifactsRequest]) (*connect.Response[pb.ListArtifactsResponse], error) {
154 name, err := repoOf(req.Msg.GetRepo())
155 if err != nil {
156 return nil, ciError(ctx, err)
157 }
158 res, err := s.artifacts.ListArtifacts(ctx, caller(ctx), name, req.Msg.GetRunId())
159 if err != nil {
160 return nil, ciError(ctx, err)
161 }
162 out := &pb.ListArtifactsResponse{UsedBytes: res.Used, QuotaBytes: res.Quota, Artifacts: make([]*pb.Artifact, 0, len(res.Artifacts))}
163 for _, a := range res.Artifacts {
164 out.Artifacts = append(out.Artifacts, artifactToPB(a))
165 }
166 return connect.NewResponse(out), nil
167}
168
169func (s *artifactService) DownloadArtifact(ctx context.Context, req *connect.Request[pb.DownloadArtifactRequest], stream *connect.ServerStream[pb.DownloadArtifactResponse]) error {
170 name, err := repoOf(req.Msg.GetRepo())
171 if err != nil {
172 return ciError(ctx, err)
173 }
174 err = s.artifacts.DownloadArtifact(ctx, caller(ctx), name, req.Msg.GetArtifactId(), req.Msg.GetOffset(), func(c ports.ArtifactChunk) error {
175 m := &pb.DownloadArtifactResponse{Chunk: c.Data}
176 if c.Artifact != nil {
177 m.Artifact = artifactToPB(*c.Artifact)
178 }
179 return stream.Send(m)
180 })
181 return ciError(ctx, err)
182}
183
184func (s *artifactService) DeleteArtifact(ctx context.Context, req *connect.Request[pb.DeleteArtifactRequest]) (*connect.Response[pb.DeleteArtifactResponse], error) {
185 name, err := repoOf(req.Msg.GetRepo())
186 if err != nil {
187 return nil, ciError(ctx, err)
188 }
189 if err := s.artifacts.DeleteArtifact(ctx, caller(ctx), name, req.Msg.GetArtifactId()); err != nil {
190 return nil, ciError(ctx, err)
191 }
192 return connect.NewResponse(&pb.DeleteArtifactResponse{}), nil
193}