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 // ResolveActions and UploadSummary are job calls too.
37 ermiteconnect.RunnerServiceResolveActionsProcedure: true,
38 ermiteconnect.RunnerServiceUploadSummaryProcedure: true,
39}
40
41// ciError maps an error of the CI use cases. A call on a job that ended gets
42// failed_precondition with a JobEnded detail, as runner.proto says.
43func ciError(ctx context.Context, err error) error {
44 if ended, ok := domain.IsJobEnded(err); ok {
45 ce := connect.NewError(connect.CodeFailedPrecondition, errors.New("the job has ended"))
46 if d, derr := connect.NewErrorDetail(jobEndedToPB(ended)); derr == nil {
47 ce.AddDetail(d)
48 }
49 return ce
50 }
51 if errors.Is(err, domain.ErrOutOfRange) {
52 return connect.NewError(connect.CodeOutOfRange, err)
53 }
54 return connectError(ctx, err)
55}
56
57// runnerService implements ermiteconnect.RunnerServiceHandler.
58type runnerService struct {
59 runners ports.RunnerService
60 limiter ports.FailureLimiter
61}
62
63// failed counts a call that was refused for its credential against the client
64// address, like every other authentication failure, and maps the error.
65func (s *runnerService) failed(ctx context.Context, err error) error {
66 if errors.Is(err, domain.ErrUnauthenticated) && s.limiter != nil {
67 if addr := addrFromContext(ctx); addr != "" {
68 s.limiter.Fail(ctx, addr)
69 }
70 }
71 return ciError(ctx, err)
72}
73
74func (s *runnerService) RegisterRunner(ctx context.Context, req *connect.Request[pb.RegisterRunnerRequest]) (*connect.Response[pb.RegisterRunnerResponse], error) {
75 iso, err := isolationFromPB(req.Msg.GetIsolation())
76 if err != nil {
77 return nil, ciError(ctx, err)
78 }
79 out, err := s.runners.Register(ctx, caller(ctx), ports.RegisterRunnerParams{
80 Name: req.Msg.GetName(), Labels: req.Msg.GetLabels(), Version: req.Msg.GetVersion(), Isolation: iso,
81 Arch: req.Msg.GetArch(), OS: req.Msg.GetOs(), Replace: req.Msg.GetReplace(),
82 })
83 if err != nil {
84 return nil, ciError(ctx, err)
85 }
86 return connect.NewResponse(&pb.RegisterRunnerResponse{RunnerId: out.Runner.ID, RunnerToken: out.Token}), nil
87}
88
89func (s *runnerService) FetchJob(ctx context.Context, req *connect.Request[pb.FetchJobRequest]) (*connect.Response[pb.FetchJobResponse], error) {
90 res, err := s.runners.FetchJob(ctx, secretFrom(ctx), ports.FetchJobParams{Labels: req.Msg.GetLabels(), PollID: req.Msg.GetPollId()})
91 if err != nil {
92 return nil, s.failed(ctx, err)
93 }
94 out := &pb.FetchJobResponse{}
95 if res.Job != nil {
96 out.Job = runnerJobToPB(res.Job)
97 } else if res.RetryAfter > 0 {
98 out.RetryAfter = durationpb.New(res.RetryAfter)
99 }
100 return connect.NewResponse(out), nil
101}
102
103func (s *runnerService) UpdateJob(ctx context.Context, req *connect.Request[pb.UpdateJobRequest]) (*connect.Response[pb.UpdateJobResponse], error) {
104 steps, err := stepUpdatesFromPB(req.Msg.GetSteps())
105 if err != nil {
106 return nil, s.failed(ctx, err)
107 }
108 anns, err := annotationsFromPB(req.Msg.GetAnnotations())
109 if err != nil {
110 return nil, s.failed(ctx, err)
111 }
112 if err := s.runners.UpdateJob(ctx, secretFrom(ctx), req.Msg.GetJobId(), steps, anns); err != nil {
113 return nil, s.failed(ctx, err)
114 }
115 return connect.NewResponse(&pb.UpdateJobResponse{}), nil
116}
117
118func (s *runnerService) AppendLog(ctx context.Context, req *connect.Request[pb.AppendLogRequest]) (*connect.Response[pb.AppendLogResponse], error) {
119 lines := make([]domain.CILogLine, 0, len(req.Msg.GetLines()))
120 for _, l := range req.Msg.GetLines() {
121 lines = append(lines, domain.CILogLine{Time: goTime(l.GetTime()), Text: l.GetText(), Continued: l.GetContinued()})
122 }
123 next, err := s.runners.AppendLog(ctx, secretFrom(ctx), req.Msg.GetJobId(), req.Msg.GetStep(), req.Msg.GetOffset(), lines)
124 if err != nil {
125 var gap *ports.ErrCILogGap
126 if errors.As(err, &gap) {
127 return nil, connect.NewError(connect.CodeOutOfRange, err)
128 }
129 return nil, s.failed(ctx, err)
130 }
131 return connect.NewResponse(&pb.AppendLogResponse{NextOffset: next}), nil
132}
133
134func (s *runnerService) Heartbeat(ctx context.Context, req *connect.Request[pb.HeartbeatRequest]) (*connect.Response[pb.HeartbeatResponse], error) {
135 if err := s.runners.Heartbeat(ctx, secretFrom(ctx), req.Msg.GetJobId()); err != nil {
136 return nil, s.failed(ctx, err)
137 }
138 return connect.NewResponse(&pb.HeartbeatResponse{}), nil
139}
140
141func (s *runnerService) WatchJob(ctx context.Context, req *connect.Request[pb.WatchJobRequest], stream *connect.ServerStream[pb.WatchJobResponse]) error {
142 // A client's call returns when the response begins: send the headers now,
143 // not with the first message, which may be 30 seconds away. Connect
144 // flushes them for a nil message; a failure shows on the first real send.
145 ready := func() { _ = stream.Send(nil) }
146 err := s.runners.WatchJob(ctx, secretFrom(ctx), req.Msg.GetJobId(), ready, func(sig ports.JobSignal) error {
147 switch {
148 case sig.Cancel != nil:
149 return stream.Send(&pb.WatchJobResponse{Msg: &pb.WatchJobResponse_Cancel{Cancel: &pb.CancelRequest{
150 Reason: cancelReasonToPB(sig.Cancel.Reason), Message: sig.Cancel.Message, RequestedAt: timestamp(sig.Cancel.RequestedAt),
151 }}})
152 case sig.Ended != nil:
153 return stream.Send(&pb.WatchJobResponse{Msg: &pb.WatchJobResponse_Ended{Ended: jobEndedToPB(sig.Ended)}})
154 }
155 return stream.Send(&pb.WatchJobResponse{Msg: &pb.WatchJobResponse_Keepalive{Keepalive: &pb.Keepalive{}}})
156 })
157 return s.failed(ctx, err)
158}
159
160func (s *runnerService) CompleteJob(ctx context.Context, req *connect.Request[pb.CompleteJobRequest]) (*connect.Response[pb.CompleteJobResponse], error) {
161 concl, err := conclusionFromPB(req.Msg.GetConclusion())
162 if err != nil {
163 return nil, s.failed(ctx, err)
164 }
165 steps, err := stepUpdatesFromPB(req.Msg.GetSteps())
166 if err != nil {
167 return nil, s.failed(ctx, err)
168 }
169 anns, err := annotationsFromPB(req.Msg.GetAnnotations())
170 if err != nil {
171 return nil, s.failed(ctx, err)
172 }
173 res, err := s.runners.CompleteJob(ctx, secretFrom(ctx), ports.CompleteJobParams{
174 JobID: req.Msg.GetJobId(), Conclusion: concl, Outputs: req.Msg.GetOutputs(),
175 RunnerShutdown: req.Msg.GetRunnerShutdown(), Steps: steps, Annotations: anns,
176 })
177 if err != nil {
178 return nil, s.failed(ctx, err)
179 }
180 return connect.NewResponse(&pb.CompleteJobResponse{Conclusion: conclusionToPB(res.Conclusion), Error: res.Error, Applied: res.Applied}), nil
181}
182
183//
184//hexcheck:ignore suspicious-business-logic-in-adapter maps the references of the request to strings and the answers back to the contract; the rules are the use case's
185func (s *runnerService) ResolveActions(ctx context.Context, req *connect.Request[pb.ResolveActionsRequest]) (*connect.Response[pb.ResolveActionsResponse], error) {
186 uses := make([]string, 0, len(req.Msg.GetRefs()))
187 for _, r := range req.Msg.GetRefs() {
188 uses = append(uses, r.GetUses())
189 }
190 res, err := s.runners.ResolveActions(ctx, secretFrom(ctx), req.Msg.GetJobId(), uses)
191 if err != nil {
192 return nil, s.failed(ctx, err)
193 }
194 out := &pb.ResolveActionsResponse{Actions: make([]*pb.ResolvedAction, 0, len(res))}
195 for _, a := range res {
196 out.Actions = append(out.Actions, resolvedActionToPB(a))
197 }
198 return connect.NewResponse(out), nil
199}
200
201// UploadSummary stores the summary of a step.
202func (s *runnerService) UploadSummary(ctx context.Context, req *connect.Request[pb.UploadSummaryRequest]) (*connect.Response[pb.UploadSummaryResponse], error) {
203 if err := s.runners.UploadSummary(ctx, secretFrom(ctx), req.Msg.GetJobId(), req.Msg.GetStepNumber(), req.Msg.GetMarkdown()); err != nil {
204 return nil, s.failed(ctx, err)
205 }
206 return connect.NewResponse(&pb.UploadSummaryResponse{}), nil
207}
208
209// runService implements ermiteconnect.RunServiceHandler.
210type runService struct {
211 // DispatchWorkflow is not implemented yet: the feature "ci.dispatch" is not
212 // listed and the RPC answers unimplemented.
213 ermiteconnect.UnimplementedRunServiceHandler
214 runs ports.RunService
215}
216
217func repoOf(s string) (domain.RepoName, error) { return domain.ParseRepoName(s) }
218
219func (s *runService) ListRuns(ctx context.Context, req *connect.Request[pb.ListRunsRequest]) (*connect.Response[pb.ListRunsResponse], error) {
220 name, err := repoOf(req.Msg.GetRepo())
221 if err != nil {
222 return nil, ciError(ctx, err)
223 }
224 if _, err := statusFromPB(req.Msg.GetStatus()); err != nil {
225 return nil, ciError(ctx, err)
226 }
227 page, err := s.runs.ListRuns(ctx, caller(ctx), name, ports.ListRunsParams{
228 PageSize: int(req.Msg.GetPageSize()), PageToken: req.Msg.GetPageToken(),
229 Workflow: req.Msg.GetWorkflow(), Branch: req.Msg.GetBranch(), Status: statusOrZero(req.Msg.GetStatus()),
230 })
231 if err != nil {
232 return nil, ciError(ctx, err)
233 }
234 out := &pb.ListRunsResponse{NextPageToken: page.NextPageToken, Runs: make([]*pb.Run, 0, len(page.Runs))}
235 for _, r := range page.Runs {
236 out.Runs = append(out.Runs, runToPB(r))
237 }
238 return connect.NewResponse(out), nil
239}
240
241func statusOrZero(s pb.CiStatus) domain.CIStatus {
242 d, _ := statusFromPB(s)
243 return d
244}
245
246func (s *runService) GetRun(ctx context.Context, req *connect.Request[pb.GetRunRequest]) (*connect.Response[pb.GetRunResponse], error) {
247 name, err := repoOf(req.Msg.GetRepo())
248 if err != nil {
249 return nil, ciError(ctx, err)
250 }
251 d, err := s.runs.GetRun(ctx, caller(ctx), name, req.Msg.GetRunId(), req.Msg.GetAttempt())
252 if err != nil {
253 return nil, ciError(ctx, err)
254 }
255 out := &pb.GetRunResponse{Run: runToPB(d.Run), LatestAttempt: d.LatestAttempt}
256 for _, j := range d.Jobs {
257 out.Jobs = append(out.Jobs, jobViewToPB(j))
258 }
259 return connect.NewResponse(out), nil
260}
261
262func (s *runService) GetJob(ctx context.Context, req *connect.Request[pb.GetJobRequest]) (*connect.Response[pb.GetJobResponse], error) {
263 name, err := repoOf(req.Msg.GetRepo())
264 if err != nil {
265 return nil, ciError(ctx, err)
266 }
267 j, err := s.runs.GetJob(ctx, caller(ctx), name, req.Msg.GetJobId(), req.Msg.GetAttempt())
268 if err != nil {
269 return nil, ciError(ctx, err)
270 }
271 return connect.NewResponse(&pb.GetJobResponse{Job: jobViewToPB(j)}), nil
272}
273
274func (s *runService) GetJobLog(ctx context.Context, req *connect.Request[pb.GetJobLogRequest], stream *connect.ServerStream[pb.GetJobLogResponse]) error {
275 name, err := repoOf(req.Msg.GetRepo())
276 if err != nil {
277 return ciError(ctx, err)
278 }
279 err = s.runs.GetJobLog(ctx, caller(ctx), name, ports.GetJobLogParams{
280 JobID: req.Msg.GetJobId(), Attempt: req.Msg.GetAttempt(), Offset: req.Msg.GetOffset(), Follow: req.Msg.GetFollow(),
281 Client: clientFromContext(ctx),
282 Ready: func() { _ = stream.Send(nil) },
283 }, func(b ports.LogBatch) error {
284 out := &pb.GetJobLogResponse{Offset: b.Offset, Lines: make([]*pb.JobLogLine, 0, len(b.Lines))}
285 for _, l := range b.Lines {
286 out.Lines = append(out.Lines, logLineToPB(l))
287 }
288 return stream.Send(out)
289 })
290 return ciError(ctx, err)
291}
292
293func (s *runService) CancelRun(ctx context.Context, req *connect.Request[pb.CancelRunRequest]) (*connect.Response[pb.CancelRunResponse], error) {
294 name, err := repoOf(req.Msg.GetRepo())
295 if err != nil {
296 return nil, ciError(ctx, err)
297 }
298 run, err := s.runs.CancelRun(ctx, caller(ctx), name, req.Msg.GetRunId())
299 if err != nil {
300 return nil, ciError(ctx, err)
301 }
302 return connect.NewResponse(&pb.CancelRunResponse{Run: runToPB(run)}), nil
303}
304
305func (s *runService) RerunRun(ctx context.Context, req *connect.Request[pb.RerunRunRequest]) (*connect.Response[pb.RerunRunResponse], error) {
306 name, err := repoOf(req.Msg.GetRepo())
307 if err != nil {
308 return nil, ciError(ctx, err)
309 }
310 run, err := s.runs.RerunRun(ctx, caller(ctx), name, req.Msg.GetRunId(), req.Msg.GetFailedOnly())
311 if err != nil {
312 return nil, ciError(ctx, err)
313 }
314 return connect.NewResponse(&pb.RerunRunResponse{Run: runToPB(run)}), nil
315}
316
317func (s *runService) ListChecks(ctx context.Context, req *connect.Request[pb.ListChecksRequest]) (*connect.Response[pb.ListChecksResponse], error) {
318 name, err := repoOf(req.Msg.GetRepo())
319 if err != nil {
320 return nil, ciError(ctx, err)
321 }
322 res, err := s.runs.ListChecks(ctx, caller(ctx), name, req.Msg.GetSha())
323 if err != nil {
324 return nil, ciError(ctx, err)
325 }
326 out := &pb.ListChecksResponse{Truncated: res.Truncated}
327 for _, c := range res.Checks {
328 out.Checks = append(out.Checks, checkToPB(c.CICheck))
329 }
330 return connect.NewResponse(out), nil
331}
332
333func (s *runService) ListRunners(ctx context.Context, req *connect.Request[pb.ListRunnersRequest]) (*connect.Response[pb.ListRunnersResponse], error) {
334 page, err := s.runs.ListRunners(ctx, caller(ctx), int(req.Msg.GetPageSize()), req.Msg.GetPageToken())
335 if err != nil {
336 return nil, ciError(ctx, err)
337 }
338 out := &pb.ListRunnersResponse{NextPageToken: page.NextPageToken, Runners: make([]*pb.Runner, 0, len(page.Runners))}
339 for _, r := range page.Runners {
340 out.Runners = append(out.Runners, runnerToPB(r))
341 }
342 return connect.NewResponse(out), nil
343}
344
345func (s *runService) RemoveRunner(ctx context.Context, req *connect.Request[pb.RemoveRunnerRequest]) (*connect.Response[pb.RemoveRunnerResponse], error) {
346 if err := s.runs.RemoveRunner(ctx, caller(ctx), req.Msg.GetRunnerId()); err != nil {
347 return nil, ciError(ctx, err)
348 }
349 return connect.NewResponse(&pb.RemoveRunnerResponse{}), nil
350}
351
352// ListAnnotations lists the annotations of a run attempt.
353func (s *runService) ListAnnotations(ctx context.Context, req *connect.Request[pb.ListAnnotationsRequest]) (*connect.Response[pb.ListAnnotationsResponse], error) {
354 name, err := repoOf(req.Msg.GetRepo())
355 if err != nil {
356 return nil, ciError(ctx, err)
357 }
358 res, err := s.runs.ListAnnotations(ctx, caller(ctx), name, req.Msg.GetRunId(), req.Msg.GetAttempt(), req.Msg.GetJobId())
359 if err != nil {
360 return nil, ciError(ctx, err)
361 }
362 out := &pb.ListAnnotationsResponse{Truncated: res.Truncated, Annotations: make([]*pb.JobAnnotation, 0, len(res.Annotations))}
363 for _, a := range res.Annotations {
364 out.Annotations = append(out.Annotations, &pb.JobAnnotation{JobId: a.JobID, Annotation: annotationToPB(a.Annotation)})
365 }
366 return connect.NewResponse(out), nil
367}
368
369// GetStepSummaries returns the summaries of a job.
370func (s *runService) GetStepSummaries(ctx context.Context, req *connect.Request[pb.GetStepSummariesRequest]) (*connect.Response[pb.GetStepSummariesResponse], error) {
371 name, err := repoOf(req.Msg.GetRepo())
372 if err != nil {
373 return nil, ciError(ctx, err)
374 }
375 res, err := s.runs.GetStepSummaries(ctx, caller(ctx), name, req.Msg.GetJobId(), req.Msg.GetAttempt(), 0)
376 if err != nil {
377 return nil, ciError(ctx, err)
378 }
379 out := &pb.GetStepSummariesResponse{Truncated: res.Truncated, Summaries: make([]*pb.StepSummary, 0, len(res.Summaries))}
380 for _, m := range res.Summaries {
381 out.Summaries = append(out.Summaries, &pb.StepSummary{StepNumber: m.Step, Markdown: m.Markdown})
382 }
383 return connect.NewResponse(out), nil
384}