1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package rpc
4
5import (
6 "context"
7 "errors"
8
9 "connectrpc.com/connect"
10 "google.golang.org/protobuf/types/known/durationpb"
11
12 pb "bnema.dev/ermite-api/gen/ermite"
13 "bnema.dev/ermite-api/gen/ermite/ermiteconnect"
14 "bnema.dev/ermite/internal/domain"
15 "bnema.dev/ermite/internal/ports"
16)
17
18// secretKey carries the bearer secret of a runner or job call to the handler:
19// those secrets are not stored tokens, the use case checks them.
20type secretKey struct{}
21
22func secretFrom(ctx context.Context) string {
23 s, _ := ctx.Value(secretKey{}).(string)
24 return s
25}
26
27// runnerSecretProcedures are the calls whose bearer secret is a runner token
28// or a job token: the interceptor does not authenticate them as stored tokens.
29var runnerSecretProcedures = map[string]bool{
30 ermiteconnect.RunnerServiceFetchJobProcedure: true,
31 ermiteconnect.RunnerServiceUpdateJobProcedure: true,
32 ermiteconnect.RunnerServiceAppendLogProcedure: true,
33 ermiteconnect.RunnerServiceHeartbeatProcedure: true,
34 ermiteconnect.RunnerServiceWatchJobProcedure: true,
35 ermiteconnect.RunnerServiceCompleteJobProcedure: true,
36}
37
38// ciError maps an error of the CI use cases. A call on a job that ended gets
39// failed_precondition with a JobEnded detail, as runner.proto says.
40func ciError(ctx context.Context, err error) error {
41 if ended, ok := domain.IsJobEnded(err); ok {
42 ce := connect.NewError(connect.CodeFailedPrecondition, errors.New("the job has ended"))
43 if d, derr := connect.NewErrorDetail(jobEndedToPB(ended)); derr == nil {
44 ce.AddDetail(d)
45 }
46 return ce
47 }
48 if errors.Is(err, domain.ErrOutOfRange) {
49 return connect.NewError(connect.CodeOutOfRange, err)
50 }
51 return connectError(ctx, err)
52}
53
54// runnerService implements ermiteconnect.RunnerServiceHandler.
55type runnerService struct {
56 runners ports.RunnerService
57 limiter ports.FailureLimiter
58}
59
60// failed counts a call that was refused for its credential against the client
61// address, like every other authentication failure, and maps the error.
62func (s *runnerService) failed(ctx context.Context, err error) error {
63 if errors.Is(err, domain.ErrUnauthenticated) && s.limiter != nil {
64 if addr := addrFromContext(ctx); addr != "" {
65 s.limiter.Fail(ctx, addr)
66 }
67 }
68 return ciError(ctx, err)
69}
70
71func (s *runnerService) RegisterRunner(ctx context.Context, req *connect.Request[pb.RegisterRunnerRequest]) (*connect.Response[pb.RegisterRunnerResponse], error) {
72 iso, err := isolationFromPB(req.Msg.GetIsolation())
73 if err != nil {
74 return nil, ciError(ctx, err)
75 }
76 out, err := s.runners.Register(ctx, caller(ctx), ports.RegisterRunnerParams{
77 Name: req.Msg.GetName(), Labels: req.Msg.GetLabels(), Version: req.Msg.GetVersion(), Isolation: iso,
78 Arch: req.Msg.GetArch(), OS: req.Msg.GetOs(), Replace: req.Msg.GetReplace(),
79 })
80 if err != nil {
81 return nil, ciError(ctx, err)
82 }
83 return connect.NewResponse(&pb.RegisterRunnerResponse{RunnerId: out.Runner.ID, RunnerToken: out.Token}), nil
84}
85
86func (s *runnerService) FetchJob(ctx context.Context, req *connect.Request[pb.FetchJobRequest]) (*connect.Response[pb.FetchJobResponse], error) {
87 res, err := s.runners.FetchJob(ctx, secretFrom(ctx), ports.FetchJobParams{Labels: req.Msg.GetLabels(), PollID: req.Msg.GetPollId()})
88 if err != nil {
89 return nil, s.failed(ctx, err)
90 }
91 out := &pb.FetchJobResponse{}
92 if res.Job != nil {
93 out.Job = runnerJobToPB(res.Job)
94 } else if res.RetryAfter > 0 {
95 out.RetryAfter = durationpb.New(res.RetryAfter)
96 }
97 return connect.NewResponse(out), nil
98}
99
100func (s *runnerService) UpdateJob(ctx context.Context, req *connect.Request[pb.UpdateJobRequest]) (*connect.Response[pb.UpdateJobResponse], error) {
101 steps, err := stepUpdatesFromPB(req.Msg.GetSteps())
102 if err != nil {
103 return nil, s.failed(ctx, err)
104 }
105 if err := s.runners.UpdateJob(ctx, secretFrom(ctx), req.Msg.GetJobId(), steps); err != nil {
106 return nil, s.failed(ctx, err)
107 }
108 return connect.NewResponse(&pb.UpdateJobResponse{}), nil
109}
110
111func (s *runnerService) AppendLog(ctx context.Context, req *connect.Request[pb.AppendLogRequest]) (*connect.Response[pb.AppendLogResponse], error) {
112 lines := make([]domain.CILogLine, 0, len(req.Msg.GetLines()))
113 for _, l := range req.Msg.GetLines() {
114 lines = append(lines, domain.CILogLine{Time: goTime(l.GetTime()), Text: l.GetText(), Continued: l.GetContinued()})
115 }
116 next, err := s.runners.AppendLog(ctx, secretFrom(ctx), req.Msg.GetJobId(), req.Msg.GetStep(), req.Msg.GetOffset(), lines)
117 if err != nil {
118 var gap *ports.ErrCILogGap
119 if errors.As(err, &gap) {
120 return nil, connect.NewError(connect.CodeOutOfRange, err)
121 }
122 return nil, s.failed(ctx, err)
123 }
124 return connect.NewResponse(&pb.AppendLogResponse{NextOffset: next}), nil
125}
126
127func (s *runnerService) Heartbeat(ctx context.Context, req *connect.Request[pb.HeartbeatRequest]) (*connect.Response[pb.HeartbeatResponse], error) {
128 if err := s.runners.Heartbeat(ctx, secretFrom(ctx), req.Msg.GetJobId()); err != nil {
129 return nil, s.failed(ctx, err)
130 }
131 return connect.NewResponse(&pb.HeartbeatResponse{}), nil
132}
133
134func (s *runnerService) WatchJob(ctx context.Context, req *connect.Request[pb.WatchJobRequest], stream *connect.ServerStream[pb.WatchJobResponse]) error {
135 // A client's call returns when the response begins: send the headers now,
136 // not with the first message, which may be 30 seconds away. Connect
137 // flushes them for a nil message; a failure shows on the first real send.
138 ready := func() { _ = stream.Send(nil) }
139 err := s.runners.WatchJob(ctx, secretFrom(ctx), req.Msg.GetJobId(), ready, func(sig ports.JobSignal) error {
140 switch {
141 case sig.Cancel != nil:
142 return stream.Send(&pb.WatchJobResponse{Msg: &pb.WatchJobResponse_Cancel{Cancel: &pb.CancelRequest{
143 Reason: cancelReasonToPB(sig.Cancel.Reason), Message: sig.Cancel.Message, RequestedAt: timestamp(sig.Cancel.RequestedAt),
144 }}})
145 case sig.Ended != nil:
146 return stream.Send(&pb.WatchJobResponse{Msg: &pb.WatchJobResponse_Ended{Ended: jobEndedToPB(sig.Ended)}})
147 }
148 return stream.Send(&pb.WatchJobResponse{Msg: &pb.WatchJobResponse_Keepalive{Keepalive: &pb.Keepalive{}}})
149 })
150 return s.failed(ctx, err)
151}
152
153func (s *runnerService) CompleteJob(ctx context.Context, req *connect.Request[pb.CompleteJobRequest]) (*connect.Response[pb.CompleteJobResponse], error) {
154 concl, err := conclusionFromPB(req.Msg.GetConclusion())
155 if err != nil {
156 return nil, s.failed(ctx, err)
157 }
158 steps, err := stepUpdatesFromPB(req.Msg.GetSteps())
159 if err != nil {
160 return nil, s.failed(ctx, err)
161 }
162 res, err := s.runners.CompleteJob(ctx, secretFrom(ctx), ports.CompleteJobParams{
163 JobID: req.Msg.GetJobId(), Conclusion: concl, Outputs: req.Msg.GetOutputs(),
164 RunnerShutdown: req.Msg.GetRunnerShutdown(), Steps: steps,
165 })
166 if err != nil {
167 return nil, s.failed(ctx, err)
168 }
169 return connect.NewResponse(&pb.CompleteJobResponse{Conclusion: conclusionToPB(res.Conclusion), Error: res.Error, Applied: res.Applied}), nil
170}
171
172// runService implements ermiteconnect.RunServiceHandler.
173type runService struct{ runs ports.RunService }
174
175func repoOf(s string) (domain.RepoName, error) { return domain.ParseRepoName(s) }
176
177func (s *runService) ListRuns(ctx context.Context, req *connect.Request[pb.ListRunsRequest]) (*connect.Response[pb.ListRunsResponse], error) {
178 name, err := repoOf(req.Msg.GetRepo())
179 if err != nil {
180 return nil, ciError(ctx, err)
181 }
182 if _, err := statusFromPB(req.Msg.GetStatus()); err != nil {
183 return nil, ciError(ctx, err)
184 }
185 page, err := s.runs.ListRuns(ctx, caller(ctx), name, ports.ListRunsParams{
186 PageSize: int(req.Msg.GetPageSize()), PageToken: req.Msg.GetPageToken(),
187 Workflow: req.Msg.GetWorkflow(), Branch: req.Msg.GetBranch(), Status: statusOrZero(req.Msg.GetStatus()),
188 })
189 if err != nil {
190 return nil, ciError(ctx, err)
191 }
192 out := &pb.ListRunsResponse{NextPageToken: page.NextPageToken, Runs: make([]*pb.Run, 0, len(page.Runs))}
193 for _, r := range page.Runs {
194 out.Runs = append(out.Runs, runToPB(r))
195 }
196 return connect.NewResponse(out), nil
197}
198
199func statusOrZero(s pb.CiStatus) domain.CIStatus {
200 d, _ := statusFromPB(s)
201 return d
202}
203
204func (s *runService) GetRun(ctx context.Context, req *connect.Request[pb.GetRunRequest]) (*connect.Response[pb.GetRunResponse], error) {
205 name, err := repoOf(req.Msg.GetRepo())
206 if err != nil {
207 return nil, ciError(ctx, err)
208 }
209 d, err := s.runs.GetRun(ctx, caller(ctx), name, req.Msg.GetRunId(), req.Msg.GetAttempt())
210 if err != nil {
211 return nil, ciError(ctx, err)
212 }
213 out := &pb.GetRunResponse{Run: runToPB(d.Run), LatestAttempt: d.LatestAttempt}
214 for _, j := range d.Jobs {
215 out.Jobs = append(out.Jobs, jobViewToPB(j))
216 }
217 return connect.NewResponse(out), nil
218}
219
220func (s *runService) GetJob(ctx context.Context, req *connect.Request[pb.GetJobRequest]) (*connect.Response[pb.GetJobResponse], error) {
221 name, err := repoOf(req.Msg.GetRepo())
222 if err != nil {
223 return nil, ciError(ctx, err)
224 }
225 j, err := s.runs.GetJob(ctx, caller(ctx), name, req.Msg.GetJobId(), req.Msg.GetAttempt())
226 if err != nil {
227 return nil, ciError(ctx, err)
228 }
229 return connect.NewResponse(&pb.GetJobResponse{Job: jobViewToPB(j)}), nil
230}
231
232func (s *runService) GetJobLog(ctx context.Context, req *connect.Request[pb.GetJobLogRequest], stream *connect.ServerStream[pb.GetJobLogResponse]) error {
233 name, err := repoOf(req.Msg.GetRepo())
234 if err != nil {
235 return ciError(ctx, err)
236 }
237 err = s.runs.GetJobLog(ctx, caller(ctx), name, ports.GetJobLogParams{
238 JobID: req.Msg.GetJobId(), Attempt: req.Msg.GetAttempt(), Offset: req.Msg.GetOffset(), Follow: req.Msg.GetFollow(),
239 Client: clientFromContext(ctx),
240 Ready: func() { _ = stream.Send(nil) },
241 }, func(b ports.LogBatch) error {
242 out := &pb.GetJobLogResponse{Offset: b.Offset, Lines: make([]*pb.JobLogLine, 0, len(b.Lines))}
243 for _, l := range b.Lines {
244 out.Lines = append(out.Lines, logLineToPB(l))
245 }
246 return stream.Send(out)
247 })
248 return ciError(ctx, err)
249}
250
251func (s *runService) CancelRun(ctx context.Context, req *connect.Request[pb.CancelRunRequest]) (*connect.Response[pb.CancelRunResponse], error) {
252 name, err := repoOf(req.Msg.GetRepo())
253 if err != nil {
254 return nil, ciError(ctx, err)
255 }
256 run, err := s.runs.CancelRun(ctx, caller(ctx), name, req.Msg.GetRunId())
257 if err != nil {
258 return nil, ciError(ctx, err)
259 }
260 return connect.NewResponse(&pb.CancelRunResponse{Run: runToPB(run)}), nil
261}
262
263func (s *runService) RerunRun(ctx context.Context, req *connect.Request[pb.RerunRunRequest]) (*connect.Response[pb.RerunRunResponse], error) {
264 name, err := repoOf(req.Msg.GetRepo())
265 if err != nil {
266 return nil, ciError(ctx, err)
267 }
268 run, err := s.runs.RerunRun(ctx, caller(ctx), name, req.Msg.GetRunId(), req.Msg.GetFailedOnly())
269 if err != nil {
270 return nil, ciError(ctx, err)
271 }
272 return connect.NewResponse(&pb.RerunRunResponse{Run: runToPB(run)}), nil
273}
274
275func (s *runService) ListChecks(ctx context.Context, req *connect.Request[pb.ListChecksRequest]) (*connect.Response[pb.ListChecksResponse], error) {
276 name, err := repoOf(req.Msg.GetRepo())
277 if err != nil {
278 return nil, ciError(ctx, err)
279 }
280 res, err := s.runs.ListChecks(ctx, caller(ctx), name, req.Msg.GetSha())
281 if err != nil {
282 return nil, ciError(ctx, err)
283 }
284 out := &pb.ListChecksResponse{Truncated: res.Truncated}
285 for _, c := range res.Checks {
286 out.Checks = append(out.Checks, checkToPB(c.CICheck))
287 }
288 return connect.NewResponse(out), nil
289}
290
291func (s *runService) ListRunners(ctx context.Context, req *connect.Request[pb.ListRunnersRequest]) (*connect.Response[pb.ListRunnersResponse], error) {
292 page, err := s.runs.ListRunners(ctx, caller(ctx), int(req.Msg.GetPageSize()), req.Msg.GetPageToken())
293 if err != nil {
294 return nil, ciError(ctx, err)
295 }
296 out := &pb.ListRunnersResponse{NextPageToken: page.NextPageToken, Runners: make([]*pb.Runner, 0, len(page.Runners))}
297 for _, r := range page.Runners {
298 out.Runners = append(out.Runners, runnerToPB(r))
299 }
300 return connect.NewResponse(out), nil
301}
302
303func (s *runService) RemoveRunner(ctx context.Context, req *connect.Request[pb.RemoveRunnerRequest]) (*connect.Response[pb.RemoveRunnerResponse], error) {
304 if err := s.runs.RemoveRunner(ctx, caller(ctx), req.Msg.GetRunnerId()); err != nil {
305 return nil, ciError(ctx, err)
306 }
307 return connect.NewResponse(&pb.RemoveRunnerResponse{}), nil
308}