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 ermiteconnect.UnimplementedRunServiceHandler
212 runs ports.RunService
213}
214
215func repoOf(s string) (domain.RepoName, error) { return domain.ParseRepoName(s) }
216
217func (s *runService) ListRuns(ctx context.Context, req *connect.Request[pb.ListRunsRequest]) (*connect.Response[pb.ListRunsResponse], error) {
218 name, err := repoOf(req.Msg.GetRepo())
219 if err != nil {
220 return nil, ciError(ctx, err)
221 }
222 if _, err := statusFromPB(req.Msg.GetStatus()); err != nil {
223 return nil, ciError(ctx, err)
224 }
225 page, err := s.runs.ListRuns(ctx, caller(ctx), name, ports.ListRunsParams{
226 PageSize: int(req.Msg.GetPageSize()), PageToken: req.Msg.GetPageToken(),
227 Workflow: req.Msg.GetWorkflow(), Branch: req.Msg.GetBranch(), Status: statusOrZero(req.Msg.GetStatus()),
228 })
229 if err != nil {
230 return nil, ciError(ctx, err)
231 }
232 out := &pb.ListRunsResponse{NextPageToken: page.NextPageToken, Runs: make([]*pb.Run, 0, len(page.Runs))}
233 for _, r := range page.Runs {
234 out.Runs = append(out.Runs, runToPB(r))
235 }
236 return connect.NewResponse(out), nil
237}
238
239func statusOrZero(s pb.CiStatus) domain.CIStatus {
240 d, _ := statusFromPB(s)
241 return d
242}
243
244func (s *runService) GetRun(ctx context.Context, req *connect.Request[pb.GetRunRequest]) (*connect.Response[pb.GetRunResponse], error) {
245 name, err := repoOf(req.Msg.GetRepo())
246 if err != nil {
247 return nil, ciError(ctx, err)
248 }
249 d, err := s.runs.GetRun(ctx, caller(ctx), name, req.Msg.GetRunId(), req.Msg.GetAttempt())
250 if err != nil {
251 return nil, ciError(ctx, err)
252 }
253 out := &pb.GetRunResponse{Run: runToPB(d.Run), LatestAttempt: d.LatestAttempt}
254 for _, j := range d.Jobs {
255 out.Jobs = append(out.Jobs, jobViewToPB(j))
256 }
257 return connect.NewResponse(out), nil
258}
259
260func (s *runService) GetJob(ctx context.Context, req *connect.Request[pb.GetJobRequest]) (*connect.Response[pb.GetJobResponse], error) {
261 name, err := repoOf(req.Msg.GetRepo())
262 if err != nil {
263 return nil, ciError(ctx, err)
264 }
265 j, err := s.runs.GetJob(ctx, caller(ctx), name, req.Msg.GetJobId(), req.Msg.GetAttempt())
266 if err != nil {
267 return nil, ciError(ctx, err)
268 }
269 return connect.NewResponse(&pb.GetJobResponse{Job: jobViewToPB(j)}), nil
270}
271
272func (s *runService) GetJobLog(ctx context.Context, req *connect.Request[pb.GetJobLogRequest], stream *connect.ServerStream[pb.GetJobLogResponse]) error {
273 name, err := repoOf(req.Msg.GetRepo())
274 if err != nil {
275 return ciError(ctx, err)
276 }
277 err = s.runs.GetJobLog(ctx, caller(ctx), name, ports.GetJobLogParams{
278 JobID: req.Msg.GetJobId(), Attempt: req.Msg.GetAttempt(), Offset: req.Msg.GetOffset(), Follow: req.Msg.GetFollow(),
279 Client: clientFromContext(ctx),
280 Ready: func() { _ = stream.Send(nil) },
281 }, func(b ports.LogBatch) error {
282 out := &pb.GetJobLogResponse{Offset: b.Offset, Lines: make([]*pb.JobLogLine, 0, len(b.Lines))}
283 for _, l := range b.Lines {
284 out.Lines = append(out.Lines, logLineToPB(l))
285 }
286 return stream.Send(out)
287 })
288 return ciError(ctx, err)
289}
290
291func (s *runService) CancelRun(ctx context.Context, req *connect.Request[pb.CancelRunRequest]) (*connect.Response[pb.CancelRunResponse], error) {
292 name, err := repoOf(req.Msg.GetRepo())
293 if err != nil {
294 return nil, ciError(ctx, err)
295 }
296 run, err := s.runs.CancelRun(ctx, caller(ctx), name, req.Msg.GetRunId())
297 if err != nil {
298 return nil, ciError(ctx, err)
299 }
300 return connect.NewResponse(&pb.CancelRunResponse{Run: runToPB(run)}), nil
301}
302
303func (s *runService) RerunRun(ctx context.Context, req *connect.Request[pb.RerunRunRequest]) (*connect.Response[pb.RerunRunResponse], error) {
304 name, err := repoOf(req.Msg.GetRepo())
305 if err != nil {
306 return nil, ciError(ctx, err)
307 }
308 run, err := s.runs.RerunRun(ctx, caller(ctx), name, req.Msg.GetRunId(), req.Msg.GetFailedOnly())
309 if err != nil {
310 return nil, ciError(ctx, err)
311 }
312 return connect.NewResponse(&pb.RerunRunResponse{Run: runToPB(run)}), nil
313}
314
315func (s *runService) DispatchWorkflow(ctx context.Context, req *connect.Request[pb.DispatchWorkflowRequest]) (*connect.Response[pb.DispatchWorkflowResponse], error) {
316 name, err := repoOf(req.Msg.GetRepo())
317 if err != nil {
318 return nil, ciError(ctx, err)
319 }
320 run, err := s.runs.DispatchWorkflow(ctx, caller(ctx), name, ports.DispatchParams{
321 WorkflowPath: req.Msg.GetWorkflowPath(), Ref: req.Msg.GetRef(), Inputs: req.Msg.GetInputs(),
322 })
323 if err != nil {
324 return nil, ciError(ctx, err)
325 }
326 return connect.NewResponse(&pb.DispatchWorkflowResponse{Run: runToPB(run)}), nil
327}
328
329func (s *runService) ListChecks(ctx context.Context, req *connect.Request[pb.ListChecksRequest]) (*connect.Response[pb.ListChecksResponse], error) {
330 name, err := repoOf(req.Msg.GetRepo())
331 if err != nil {
332 return nil, ciError(ctx, err)
333 }
334 res, err := s.runs.ListChecks(ctx, caller(ctx), name, req.Msg.GetSha())
335 if err != nil {
336 return nil, ciError(ctx, err)
337 }
338 out := &pb.ListChecksResponse{Truncated: res.Truncated}
339 for _, c := range res.Checks {
340 out.Checks = append(out.Checks, checkToPB(c.CICheck))
341 }
342 return connect.NewResponse(out), nil
343}
344
345func (s *runService) ListRunners(ctx context.Context, req *connect.Request[pb.ListRunnersRequest]) (*connect.Response[pb.ListRunnersResponse], error) {
346 page, err := s.runs.ListRunners(ctx, caller(ctx), int(req.Msg.GetPageSize()), req.Msg.GetPageToken())
347 if err != nil {
348 return nil, ciError(ctx, err)
349 }
350 out := &pb.ListRunnersResponse{NextPageToken: page.NextPageToken, Runners: make([]*pb.Runner, 0, len(page.Runners))}
351 for _, r := range page.Runners {
352 out.Runners = append(out.Runners, runnerToPB(r))
353 }
354 return connect.NewResponse(out), nil
355}
356
357func (s *runService) RemoveRunner(ctx context.Context, req *connect.Request[pb.RemoveRunnerRequest]) (*connect.Response[pb.RemoveRunnerResponse], error) {
358 if err := s.runs.RemoveRunner(ctx, caller(ctx), req.Msg.GetRunnerId()); err != nil {
359 return nil, ciError(ctx, err)
360 }
361 return connect.NewResponse(&pb.RemoveRunnerResponse{}), nil
362}
363
364// ListAnnotations lists the annotations of a run attempt.
365func (s *runService) ListAnnotations(ctx context.Context, req *connect.Request[pb.ListAnnotationsRequest]) (*connect.Response[pb.ListAnnotationsResponse], error) {
366 name, err := repoOf(req.Msg.GetRepo())
367 if err != nil {
368 return nil, ciError(ctx, err)
369 }
370 res, err := s.runs.ListAnnotations(ctx, caller(ctx), name, req.Msg.GetRunId(), req.Msg.GetAttempt(), req.Msg.GetJobId())
371 if err != nil {
372 return nil, ciError(ctx, err)
373 }
374 out := &pb.ListAnnotationsResponse{Truncated: res.Truncated, Annotations: make([]*pb.JobAnnotation, 0, len(res.Annotations))}
375 for _, a := range res.Annotations {
376 out.Annotations = append(out.Annotations, &pb.JobAnnotation{JobId: a.JobID, Annotation: annotationToPB(a.Annotation)})
377 }
378 return connect.NewResponse(out), nil
379}
380
381// GetStepSummaries returns the summaries of a job.
382func (s *runService) GetStepSummaries(ctx context.Context, req *connect.Request[pb.GetStepSummariesRequest]) (*connect.Response[pb.GetStepSummariesResponse], error) {
383 name, err := repoOf(req.Msg.GetRepo())
384 if err != nil {
385 return nil, ciError(ctx, err)
386 }
387 res, err := s.runs.GetStepSummaries(ctx, caller(ctx), name, req.Msg.GetJobId(), req.Msg.GetAttempt(), 0)
388 if err != nil {
389 return nil, ciError(ctx, err)
390 }
391 out := &pb.GetStepSummariesResponse{Truncated: res.Truncated, Summaries: make([]*pb.StepSummary, 0, len(res.Summaries))}
392 for _, m := range res.Summaries {
393 out.Summaries = append(out.Summaries, &pb.StepSummary{StepNumber: m.Step, Markdown: m.Markdown})
394 }
395 return connect.NewResponse(out), nil
396}