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