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