feat(ci): serve RunnerService and RunService, wire CI into the server

Add the Connect adapters for RunnerService and RunService (runner and
job tokens are passed to the use case as bearer secrets, failures count
against the client address, calls on an ended job answer
failed_precondition with the JobEnded detail), the RunUpdated and
JobUpdated events on the repository topic, the ci.runs, ci.runs.filters
and ci.runners features, the wiring in the composition root (store, log
files, engine started and stopped with the server, repository deletion
cleanup, merge rule and merge trigger), and in-process end-to-end tests
with a fake runner over the real handlers.

commit: 5fbf95474674f239310d4e2cbe0c95a466cf2db2
author: brice · 9d ago · 2026-10-12 18:17 UTC
parent: 2965018
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

15 files · +1137 -13

M internal/adapters/rpc/api.go +12 -1


@@ -26,7 +26,7 @@ const MaxRequestBytes = 1 << 20
26	26
27	27		// Features are the optional capabilities of the contract that this server
28	28		// implements, as GetVersion lists them.
29		-	var Features = []string{"repo.update", "events.resume", "merge_requests", "merge_requests.delete_branch", "signing_key", "session", "git.list_paths"}
	29	+	var Features = []string{"repo.update", "events.resume", "merge_requests", "merge_requests.delete_branch", "signing_key", "session", "git.list_paths", "ci.runs", "ci.runs.filters", "ci.runners"}
30	30
31	31		// Config configures the API.
32	32		type Config struct {

@@ -46,6 +46,11 @@ type Config struct {
46	46		// budget of failures. Leave it unset rather than assigning a typed nil
47	47		// pointer (a nil *T in the interface is not nil, and would be called).
48	48		Limiter ports.FailureLimiter
	49	+	// Runner serves RunnerService and Runs serves RunService; without them the
	50	+	// CI services are not served. Leave them unset rather than assigning a
	51	+	// typed nil pointer.
	52	+	Runner ports.RunnerService
	53	+	Runs ports.RunService
49	54		// Sessions are the owner's web sessions; nil serves no SessionService.
50	55		// Leave it unset rather than assigning a typed nil pointer (a nil *T in
51	56		// the interface is not nil, and the service would be served and called).

@@ -81,6 +86,12 @@ func New(cfg Config) *API {
81	86		if cfg.Sessions != nil {
82	87		mux.Handle(ermiteconnect.NewSessionServiceHandler(&sessionService{cfg.Sessions}, opts...))
83	88		}
	89	+	if cfg.Runner != nil {
	90	+	mux.Handle(ermiteconnect.NewRunnerServiceHandler(&runnerService{runners: cfg.Runner, limiter: cfg.Limiter}, opts...))
	91	+	}
	92	+	if cfg.Runs != nil {
	93	+	mux.Handle(ermiteconnect.NewRunServiceHandler(&runService{cfg.Runs}, opts...))
	94	+	}
84	95		return &API{handler: mux}
85	96		}
86	97

 // Features are the optional capabilities of the contract that this server
 // implements, as GetVersion lists them.
-var Features = []string{"repo.update", "events.resume", "merge_requests", "merge_requests.delete_branch", "signing_key", "session", "git.list_paths"}
+var Features = []string{"repo.update", "events.resume", "merge_requests", "merge_requests.delete_branch", "signing_key", "session", "git.list_paths", "ci.runs", "ci.runs.filters", "ci.runners"}
 // Config configures the API.
 type Config struct {
 	// budget of failures. Leave it unset rather than assigning a typed nil
 	// pointer (a nil *T in the interface is not nil, and would be called).
 	Limiter ports.FailureLimiter
+	// Runner serves RunnerService and Runs serves RunService; without them the
+	// CI services are not served. Leave them unset rather than assigning a
+	// typed nil pointer.
+	Runner ports.RunnerService
+	Runs   ports.RunService
 	// Sessions are the owner's web sessions; nil serves no SessionService.
 	// Leave it unset rather than assigning a typed nil pointer (a nil *T in
 	// the interface is not nil, and the service would be served and called).
 	if cfg.Sessions != nil {
 		mux.Handle(ermiteconnect.NewSessionServiceHandler(&sessionService{cfg.Sessions}, opts...))
 	}
+	if cfg.Runner != nil {
+		mux.Handle(ermiteconnect.NewRunnerServiceHandler(&runnerService{runners: cfg.Runner, limiter: cfg.Limiter}, opts...))
+	}
+	if cfg.Runs != nil {
+		mux.Handle(ermiteconnect.NewRunServiceHandler(&runService{cfg.Runs}, opts...))
+	}
 	return &API{handler: mux}
 }

A internal/adapters/rpc/ci.go +303 -0

@@ -0,0 +1,303 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	package rpc
	4	+
	5	+	import (
	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.
	20	+	type secretKey struct{}
	21	+
	22	+	func 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.
	29	+	var 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.
	40	+	func 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.
	55	+	type 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.
	62	+	func (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	+
	71	+	func (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	+
	86	+	func (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	+
	100	+	func (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	+
	111	+	func (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	+
	127	+	func (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	+
	134	+	func (s runnerService) WatchJob(ctx context.Context, req connect.Request[pb.WatchJobRequest], stream *connect.ServerStream[pb.WatchJobResponse]) error {
	135	+	err := s.runners.WatchJob(ctx, secretFrom(ctx), req.Msg.GetJobId(), func(sig ports.JobSignal) error {
	136	+	switch {
	137	+	case sig.Cancel != nil:
	138	+	return stream.Send(&pb.WatchJobResponse{Msg: &pb.WatchJobResponse_Cancel{Cancel: &pb.CancelRequest{
	139	+	Reason: cancelReasonToPB(sig.Cancel.Reason), Message: sig.Cancel.Message, RequestedAt: timestamp(sig.Cancel.RequestedAt),
	140	+	}}})
	141	+	case sig.Ended != nil:
	142	+	return stream.Send(&pb.WatchJobResponse{Msg: &pb.WatchJobResponse_Ended{Ended: jobEndedToPB(sig.Ended)}})
	143	+	}
	144	+	return stream.Send(&pb.WatchJobResponse{Msg: &pb.WatchJobResponse_Keepalive{Keepalive: &pb.Keepalive{}}})
	145	+	})
	146	+	return s.failed(ctx, err)
	147	+	}
	148	+
	149	+	func (s runnerService) CompleteJob(ctx context.Context, req connect.Request[pb.CompleteJobRequest]) (*connect.Response[pb.CompleteJobResponse], error) {
	150	+	concl, err := conclusionFromPB(req.Msg.GetConclusion())
	151	+	if err != nil {
	152	+	return nil, s.failed(ctx, err)
	153	+	}
	154	+	steps, err := stepUpdatesFromPB(req.Msg.GetSteps())
	155	+	if err != nil {
	156	+	return nil, s.failed(ctx, err)
	157	+	}
	158	+	res, err := s.runners.CompleteJob(ctx, secretFrom(ctx), ports.CompleteJobParams{
	159	+	JobID: req.Msg.GetJobId(), Conclusion: concl, Outputs: req.Msg.GetOutputs(),
	160	+	RunnerShutdown: req.Msg.GetRunnerShutdown(), Steps: steps,
	161	+	})
	162	+	if err != nil {
	163	+	return nil, s.failed(ctx, err)
	164	+	}
	165	+	return connect.NewResponse(&pb.CompleteJobResponse{Conclusion: conclusionToPB(res.Conclusion), Error: res.Error, Applied: res.Applied}), nil
	166	+	}
	167	+
	168	+	// runService implements ermiteconnect.RunServiceHandler.
	169	+	type runService struct{ runs ports.RunService }
	170	+
	171	+	func repoOf(s string) (domain.RepoName, error) { return domain.ParseRepoName(s) }
	172	+
	173	+	func (s runService) ListRuns(ctx context.Context, req connect.Request[pb.ListRunsRequest]) (*connect.Response[pb.ListRunsResponse], error) {
	174	+	name, err := repoOf(req.Msg.GetRepo())
	175	+	if err != nil {
	176	+	return nil, ciError(ctx, err)
	177	+	}
	178	+	if _, err := statusFromPB(req.Msg.GetStatus()); err != nil {
	179	+	return nil, ciError(ctx, err)
	180	+	}
	181	+	page, err := s.runs.ListRuns(ctx, caller(ctx), name, ports.ListRunsParams{
	182	+	PageSize: int(req.Msg.GetPageSize()), PageToken: req.Msg.GetPageToken(),
	183	+	Workflow: req.Msg.GetWorkflow(), Branch: req.Msg.GetBranch(), Status: statusOrZero(req.Msg.GetStatus()),
	184	+	})
	185	+	if err != nil {
	186	+	return nil, ciError(ctx, err)
	187	+	}
	188	+	out := &pb.ListRunsResponse{NextPageToken: page.NextPageToken, Runs: make([]*pb.Run, 0, len(page.Runs))}
	189	+	for _, r := range page.Runs {
	190	+	out.Runs = append(out.Runs, runToPB(r))
	191	+	}
	192	+	return connect.NewResponse(out), nil
	193	+	}
	194	+
	195	+	func statusOrZero(s pb.CiStatus) domain.CIStatus {
	196	+	d, _ := statusFromPB(s)
	197	+	return d
	198	+	}
	199	+
	200	+	func (s runService) GetRun(ctx context.Context, req connect.Request[pb.GetRunRequest]) (*connect.Response[pb.GetRunResponse], error) {
	201	+	name, err := repoOf(req.Msg.GetRepo())
	202	+	if err != nil {
	203	+	return nil, ciError(ctx, err)
	204	+	}
	205	+	d, err := s.runs.GetRun(ctx, caller(ctx), name, req.Msg.GetRunId(), req.Msg.GetAttempt())
	206	+	if err != nil {
	207	+	return nil, ciError(ctx, err)
	208	+	}
	209	+	out := &pb.GetRunResponse{Run: runToPB(d.Run), LatestAttempt: d.LatestAttempt}
	210	+	for _, j := range d.Jobs {
	211	+	out.Jobs = append(out.Jobs, jobViewToPB(j))
	212	+	}
	213	+	return connect.NewResponse(out), nil
	214	+	}
	215	+
	216	+	func (s runService) GetJob(ctx context.Context, req connect.Request[pb.GetJobRequest]) (*connect.Response[pb.GetJobResponse], error) {
	217	+	name, err := repoOf(req.Msg.GetRepo())
	218	+	if err != nil {
	219	+	return nil, ciError(ctx, err)
	220	+	}
	221	+	j, err := s.runs.GetJob(ctx, caller(ctx), name, req.Msg.GetJobId(), req.Msg.GetAttempt())
	222	+	if err != nil {
	223	+	return nil, ciError(ctx, err)
	224	+	}
	225	+	return connect.NewResponse(&pb.GetJobResponse{Job: jobViewToPB(j)}), nil
	226	+	}
	227	+
	228	+	func (s runService) GetJobLog(ctx context.Context, req connect.Request[pb.GetJobLogRequest], stream *connect.ServerStream[pb.GetJobLogResponse]) error {
	229	+	name, err := repoOf(req.Msg.GetRepo())
	230	+	if err != nil {
	231	+	return ciError(ctx, err)
	232	+	}
	233	+	err = s.runs.GetJobLog(ctx, caller(ctx), name, ports.GetJobLogParams{
	234	+	JobID: req.Msg.GetJobId(), Attempt: req.Msg.GetAttempt(), Offset: req.Msg.GetOffset(), Follow: req.Msg.GetFollow(),
	235	+	Client: clientFromContext(ctx),
	236	+	}, func(b ports.LogBatch) error {
	237	+	out := &pb.GetJobLogResponse{Offset: b.Offset, Lines: make([]*pb.JobLogLine, 0, len(b.Lines))}
	238	+	for _, l := range b.Lines {
	239	+	out.Lines = append(out.Lines, logLineToPB(l))
	240	+	}
	241	+	return stream.Send(out)
	242	+	})
	243	+	return ciError(ctx, err)
	244	+	}
	245	+
	246	+	func (s runService) CancelRun(ctx context.Context, req connect.Request[pb.CancelRunRequest]) (*connect.Response[pb.CancelRunResponse], error) {
	247	+	name, err := repoOf(req.Msg.GetRepo())
	248	+	if err != nil {
	249	+	return nil, ciError(ctx, err)
	250	+	}
	251	+	run, err := s.runs.CancelRun(ctx, caller(ctx), name, req.Msg.GetRunId())
	252	+	if err != nil {
	253	+	return nil, ciError(ctx, err)
	254	+	}
	255	+	return connect.NewResponse(&pb.CancelRunResponse{Run: runToPB(run)}), nil
	256	+	}
	257	+
	258	+	func (s runService) RerunRun(ctx context.Context, req connect.Request[pb.RerunRunRequest]) (*connect.Response[pb.RerunRunResponse], error) {
	259	+	name, err := repoOf(req.Msg.GetRepo())
	260	+	if err != nil {
	261	+	return nil, ciError(ctx, err)
	262	+	}
	263	+	run, err := s.runs.RerunRun(ctx, caller(ctx), name, req.Msg.GetRunId(), req.Msg.GetFailedOnly())
	264	+	if err != nil {
	265	+	return nil, ciError(ctx, err)
	266	+	}
	267	+	return connect.NewResponse(&pb.RerunRunResponse{Run: runToPB(run)}), nil
	268	+	}
	269	+
	270	+	func (s runService) ListChecks(ctx context.Context, req connect.Request[pb.ListChecksRequest]) (*connect.Response[pb.ListChecksResponse], error) {
	271	+	name, err := repoOf(req.Msg.GetRepo())
	272	+	if err != nil {
	273	+	return nil, ciError(ctx, err)
	274	+	}
	275	+	res, err := s.runs.ListChecks(ctx, caller(ctx), name, req.Msg.GetSha())
	276	+	if err != nil {
	277	+	return nil, ciError(ctx, err)
	278	+	}
	279	+	out := &pb.ListChecksResponse{Truncated: res.Truncated}
	280	+	for _, c := range res.Checks {
	281	+	out.Checks = append(out.Checks, checkToPB(c.CICheck))
	282	+	}
	283	+	return connect.NewResponse(out), nil
	284	+	}
	285	+
	286	+	func (s runService) ListRunners(ctx context.Context, req connect.Request[pb.ListRunnersRequest]) (*connect.Response[pb.ListRunnersResponse], error) {
	287	+	page, err := s.runs.ListRunners(ctx, caller(ctx), int(req.Msg.GetPageSize()), req.Msg.GetPageToken())
	288	+	if err != nil {
	289	+	return nil, ciError(ctx, err)
	290	+	}
	291	+	out := &pb.ListRunnersResponse{NextPageToken: page.NextPageToken, Runners: make([]*pb.Runner, 0, len(page.Runners))}
	292	+	for _, r := range page.Runners {
	293	+	out.Runners = append(out.Runners, runnerToPB(r))
	294	+	}
	295	+	return connect.NewResponse(out), nil
	296	+	}
	297	+
	298	+	func (s runService) RemoveRunner(ctx context.Context, req connect.Request[pb.RemoveRunnerRequest]) (*connect.Response[pb.RemoveRunnerResponse], error) {
	299	+	if err := s.runs.RemoveRunner(ctx, caller(ctx), req.Msg.GetRunnerId()); err != nil {
	300	+	return nil, ciError(ctx, err)
	301	+	}
	302	+	return connect.NewResponse(&pb.RemoveRunnerResponse{}), nil
	303	+	}

@@ -0,0 +1,303 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	package rpc
	4	+
	5	+	import (
	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.
	20	+	type secretKey struct{}
	21	+
	22	+	func 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.
	29	+	var 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.
	40	+	func 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.
	55	+	type 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.
	62	+	func (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	+
	71	+	func (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	+
	86	+	func (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	+
	100	+	func (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	+
	111	+	func (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	+
	127	+	func (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	+
	134	+	func (s runnerService) WatchJob(ctx context.Context, req connect.Request[pb.WatchJobRequest], stream *connect.ServerStream[pb.WatchJobResponse]) error {
	135	+	err := s.runners.WatchJob(ctx, secretFrom(ctx), req.Msg.GetJobId(), func(sig ports.JobSignal) error {
	136	+	switch {
	137	+	case sig.Cancel != nil:
	138	+	return stream.Send(&pb.WatchJobResponse{Msg: &pb.WatchJobResponse_Cancel{Cancel: &pb.CancelRequest{
	139	+	Reason: cancelReasonToPB(sig.Cancel.Reason), Message: sig.Cancel.Message, RequestedAt: timestamp(sig.Cancel.RequestedAt),
	140	+	}}})
	141	+	case sig.Ended != nil:
	142	+	return stream.Send(&pb.WatchJobResponse{Msg: &pb.WatchJobResponse_Ended{Ended: jobEndedToPB(sig.Ended)}})
	143	+	}
	144	+	return stream.Send(&pb.WatchJobResponse{Msg: &pb.WatchJobResponse_Keepalive{Keepalive: &pb.Keepalive{}}})
	145	+	})
	146	+	return s.failed(ctx, err)
	147	+	}
	148	+
	149	+	func (s runnerService) CompleteJob(ctx context.Context, req connect.Request[pb.CompleteJobRequest]) (*connect.Response[pb.CompleteJobResponse], error) {
	150	+	concl, err := conclusionFromPB(req.Msg.GetConclusion())
	151	+	if err != nil {
	152	+	return nil, s.failed(ctx, err)
	153	+	}
	154	+	steps, err := stepUpdatesFromPB(req.Msg.GetSteps())
	155	+	if err != nil {
	156	+	return nil, s.failed(ctx, err)
	157	+	}
	158	+	res, err := s.runners.CompleteJob(ctx, secretFrom(ctx), ports.CompleteJobParams{
	159	+	JobID: req.Msg.GetJobId(), Conclusion: concl, Outputs: req.Msg.GetOutputs(),
	160	+	RunnerShutdown: req.Msg.GetRunnerShutdown(), Steps: steps,
	161	+	})
	162	+	if err != nil {
	163	+	return nil, s.failed(ctx, err)
	164	+	}
	165	+	return connect.NewResponse(&pb.CompleteJobResponse{Conclusion: conclusionToPB(res.Conclusion), Error: res.Error, Applied: res.Applied}), nil
	166	+	}
	167	+
	168	+	// runService implements ermiteconnect.RunServiceHandler.
	169	+	type runService struct{ runs ports.RunService }
	170	+
	171	+	func repoOf(s string) (domain.RepoName, error) { return domain.ParseRepoName(s) }
	172	+
	173	+	func (s runService) ListRuns(ctx context.Context, req connect.Request[pb.ListRunsRequest]) (*connect.Response[pb.ListRunsResponse], error) {
	174	+	name, err := repoOf(req.Msg.GetRepo())
	175	+	if err != nil {
	176	+	return nil, ciError(ctx, err)
	177	+	}
	178	+	if _, err := statusFromPB(req.Msg.GetStatus()); err != nil {
	179	+	return nil, ciError(ctx, err)
	180	+	}
	181	+	page, err := s.runs.ListRuns(ctx, caller(ctx), name, ports.ListRunsParams{
	182	+	PageSize: int(req.Msg.GetPageSize()), PageToken: req.Msg.GetPageToken(),
	183	+	Workflow: req.Msg.GetWorkflow(), Branch: req.Msg.GetBranch(), Status: statusOrZero(req.Msg.GetStatus()),
	184	+	})
	185	+	if err != nil {
	186	+	return nil, ciError(ctx, err)
	187	+	}
	188	+	out := &pb.ListRunsResponse{NextPageToken: page.NextPageToken, Runs: make([]*pb.Run, 0, len(page.Runs))}
	189	+	for _, r := range page.Runs {
	190	+	out.Runs = append(out.Runs, runToPB(r))
	191	+	}
	192	+	return connect.NewResponse(out), nil
	193	+	}
	194	+
	195	+	func statusOrZero(s pb.CiStatus) domain.CIStatus {
	196	+	d, _ := statusFromPB(s)
	197	+	return d
	198	+	}
	199	+
	200	+	func (s runService) GetRun(ctx context.Context, req connect.Request[pb.GetRunRequest]) (*connect.Response[pb.GetRunResponse], error) {
	201	+	name, err := repoOf(req.Msg.GetRepo())
	202	+	if err != nil {
	203	+	return nil, ciError(ctx, err)
	204	+	}
	205	+	d, err := s.runs.GetRun(ctx, caller(ctx), name, req.Msg.GetRunId(), req.Msg.GetAttempt())
	206	+	if err != nil {
	207	+	return nil, ciError(ctx, err)
	208	+	}
	209	+	out := &pb.GetRunResponse{Run: runToPB(d.Run), LatestAttempt: d.LatestAttempt}
	210	+	for _, j := range d.Jobs {
	211	+	out.Jobs = append(out.Jobs, jobViewToPB(j))
	212	+	}
	213	+	return connect.NewResponse(out), nil
	214	+	}
	215	+
	216	+	func (s runService) GetJob(ctx context.Context, req connect.Request[pb.GetJobRequest]) (*connect.Response[pb.GetJobResponse], error) {
	217	+	name, err := repoOf(req.Msg.GetRepo())
	218	+	if err != nil {
	219	+	return nil, ciError(ctx, err)
	220	+	}
	221	+	j, err := s.runs.GetJob(ctx, caller(ctx), name, req.Msg.GetJobId(), req.Msg.GetAttempt())
	222	+	if err != nil {
	223	+	return nil, ciError(ctx, err)
	224	+	}
	225	+	return connect.NewResponse(&pb.GetJobResponse{Job: jobViewToPB(j)}), nil
	226	+	}
	227	+
	228	+	func (s runService) GetJobLog(ctx context.Context, req connect.Request[pb.GetJobLogRequest], stream *connect.ServerStream[pb.GetJobLogResponse]) error {
	229	+	name, err := repoOf(req.Msg.GetRepo())
	230	+	if err != nil {
	231	+	return ciError(ctx, err)
	232	+	}
	233	+	err = s.runs.GetJobLog(ctx, caller(ctx), name, ports.GetJobLogParams{
	234	+	JobID: req.Msg.GetJobId(), Attempt: req.Msg.GetAttempt(), Offset: req.Msg.GetOffset(), Follow: req.Msg.GetFollow(),
	235	+	Client: clientFromContext(ctx),
	236	+	}, func(b ports.LogBatch) error {
	237	+	out := &pb.GetJobLogResponse{Offset: b.Offset, Lines: make([]*pb.JobLogLine, 0, len(b.Lines))}
	238	+	for _, l := range b.Lines {
	239	+	out.Lines = append(out.Lines, logLineToPB(l))
	240	+	}
	241	+	return stream.Send(out)
	242	+	})
	243	+	return ciError(ctx, err)
	244	+	}
	245	+
	246	+	func (s runService) CancelRun(ctx context.Context, req connect.Request[pb.CancelRunRequest]) (*connect.Response[pb.CancelRunResponse], error) {
	247	+	name, err := repoOf(req.Msg.GetRepo())
	248	+	if err != nil {
	249	+	return nil, ciError(ctx, err)
	250	+	}
	251	+	run, err := s.runs.CancelRun(ctx, caller(ctx), name, req.Msg.GetRunId())
	252	+	if err != nil {
	253	+	return nil, ciError(ctx, err)
	254	+	}
	255	+	return connect.NewResponse(&pb.CancelRunResponse{Run: runToPB(run)}), nil
	256	+	}
	257	+
	258	+	func (s runService) RerunRun(ctx context.Context, req connect.Request[pb.RerunRunRequest]) (*connect.Response[pb.RerunRunResponse], error) {
	259	+	name, err := repoOf(req.Msg.GetRepo())
	260	+	if err != nil {
	261	+	return nil, ciError(ctx, err)
	262	+	}
	263	+	run, err := s.runs.RerunRun(ctx, caller(ctx), name, req.Msg.GetRunId(), req.Msg.GetFailedOnly())
	264	+	if err != nil {
	265	+	return nil, ciError(ctx, err)
	266	+	}
	267	+	return connect.NewResponse(&pb.RerunRunResponse{Run: runToPB(run)}), nil
	268	+	}
	269	+
	270	+	func (s runService) ListChecks(ctx context.Context, req connect.Request[pb.ListChecksRequest]) (*connect.Response[pb.ListChecksResponse], error) {
	271	+	name, err := repoOf(req.Msg.GetRepo())
	272	+	if err != nil {
	273	+	return nil, ciError(ctx, err)
	274	+	}
	275	+	res, err := s.runs.ListChecks(ctx, caller(ctx), name, req.Msg.GetSha())
	276	+	if err != nil {
	277	+	return nil, ciError(ctx, err)
	278	+	}
	279	+	out := &pb.ListChecksResponse{Truncated: res.Truncated}
	280	+	for _, c := range res.Checks {
	281	+	out.Checks = append(out.Checks, checkToPB(c.CICheck))
	282	+	}
	283	+	return connect.NewResponse(out), nil
	284	+	}
	285	+
	286	+	func (s runService) ListRunners(ctx context.Context, req connect.Request[pb.ListRunnersRequest]) (*connect.Response[pb.ListRunnersResponse], error) {
	287	+	page, err := s.runs.ListRunners(ctx, caller(ctx), int(req.Msg.GetPageSize()), req.Msg.GetPageToken())
	288	+	if err != nil {
	289	+	return nil, ciError(ctx, err)
	290	+	}
	291	+	out := &pb.ListRunnersResponse{NextPageToken: page.NextPageToken, Runners: make([]*pb.Runner, 0, len(page.Runners))}
	292	+	for _, r := range page.Runners {
	293	+	out.Runners = append(out.Runners, runnerToPB(r))
	294	+	}
	295	+	return connect.NewResponse(out), nil
	296	+	}
	297	+
	298	+	func (s runService) RemoveRunner(ctx context.Context, req connect.Request[pb.RemoveRunnerRequest]) (*connect.Response[pb.RemoveRunnerResponse], error) {
	299	+	if err := s.runs.RemoveRunner(ctx, caller(ctx), req.Msg.GetRunnerId()); err != nil {
	300	+	return nil, ciError(ctx, err)
	301	+	}
	302	+	return connect.NewResponse(&pb.RemoveRunnerResponse{}), nil
	303	+	}

A internal/adapters/rpc/ci_convert.go +228 -0

@@ -0,0 +1,228 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	package rpc
	4	+
	5	+	import (
	6	+	"fmt"
	7	+
	8	+	pb "bnema.dev/ermite-api/gen/ermite"
	9	+	"bnema.dev/ermite/internal/domain"
	10	+	"bnema.dev/ermite/internal/ports"
	11	+	)
	12	+
	13	+	func statusToPB(s domain.CIStatus) pb.CiStatus {
	14	+	switch s {
	15	+	case domain.CIQueued:
	16	+	return pb.CiStatus_CI_STATUS_QUEUED
	17	+	case domain.CIInProgress:
	18	+	return pb.CiStatus_CI_STATUS_IN_PROGRESS
	19	+	case domain.CICompleted:
	20	+	return pb.CiStatus_CI_STATUS_COMPLETED
	21	+	}
	22	+	return pb.CiStatus_CI_STATUS_UNSPECIFIED
	23	+	}
	24	+
	25	+	func statusFromPB(s pb.CiStatus) (domain.CIStatus, error) {
	26	+	switch s {
	27	+	case pb.CiStatus_CI_STATUS_UNSPECIFIED:
	28	+	return "", nil
	29	+	case pb.CiStatus_CI_STATUS_QUEUED:
	30	+	return domain.CIQueued, nil
	31	+	case pb.CiStatus_CI_STATUS_IN_PROGRESS:
	32	+	return domain.CIInProgress, nil
	33	+	case pb.CiStatus_CI_STATUS_COMPLETED:
	34	+	return domain.CICompleted, nil
	35	+	}
	36	+	return "", fmt.Errorf("%w: status %v", domain.ErrInvalid, s)
	37	+	}
	38	+
	39	+	var conclusionPairs = []struct {
	40	+	d domain.CIConclusion
	41	+	p pb.CiConclusion
	42	+	}{
	43	+	{domain.CISuccess, pb.CiConclusion_CI_CONCLUSION_SUCCESS},
	44	+	{domain.CIFailure, pb.CiConclusion_CI_CONCLUSION_FAILURE},
	45	+	{domain.CICancelled, pb.CiConclusion_CI_CONCLUSION_CANCELLED},
	46	+	{domain.CISkipped, pb.CiConclusion_CI_CONCLUSION_SKIPPED},
	47	+	{domain.CITimedOut, pb.CiConclusion_CI_CONCLUSION_TIMED_OUT},
	48	+	{domain.CINeutral, pb.CiConclusion_CI_CONCLUSION_NEUTRAL},
	49	+	{domain.CIActionRequired, pb.CiConclusion_CI_CONCLUSION_ACTION_REQUIRED},
	50	+	}
	51	+
	52	+	func conclusionToPB(c domain.CIConclusion) pb.CiConclusion {
	53	+	for _, e := range conclusionPairs {
	54	+	if e.d == c {
	55	+	return e.p
	56	+	}
	57	+	}
	58	+	return pb.CiConclusion_CI_CONCLUSION_UNSPECIFIED
	59	+	}
	60	+
	61	+	func conclusionFromPB(c pb.CiConclusion) (domain.CIConclusion, error) {
	62	+	if c == pb.CiConclusion_CI_CONCLUSION_UNSPECIFIED {
	63	+	return domain.CINoConclusion, nil
	64	+	}
	65	+	for _, e := range conclusionPairs {
	66	+	if e.p == c {
	67	+	return e.d, nil
	68	+	}
	69	+	}
	70	+	return "", fmt.Errorf("%w: conclusion %v", domain.ErrInvalid, c)
	71	+	}
	72	+
	73	+	func isolationToPB(i domain.CIIsolation) pb.RunnerIsolation {
	74	+	switch i {
	75	+	case domain.CIIsolationVM:
	76	+	return pb.RunnerIsolation_RUNNER_ISOLATION_VM
	77	+	case domain.CIIsolationContainerSysbox:
	78	+	return pb.RunnerIsolation_RUNNER_ISOLATION_CONTAINER_SYSBOX
	79	+	case domain.CIIsolationContainerPrivileged:
	80	+	return pb.RunnerIsolation_RUNNER_ISOLATION_CONTAINER_PRIVILEGED
	81	+	}
	82	+	return pb.RunnerIsolation_RUNNER_ISOLATION_UNSPECIFIED
	83	+	}
	84	+
	85	+	func isolationFromPB(i pb.RunnerIsolation) (domain.CIIsolation, error) {
	86	+	switch i {
	87	+	case pb.RunnerIsolation_RUNNER_ISOLATION_VM:
	88	+	return domain.CIIsolationVM, nil
	89	+	case pb.RunnerIsolation_RUNNER_ISOLATION_CONTAINER_SYSBOX:
	90	+	return domain.CIIsolationContainerSysbox, nil
	91	+	case pb.RunnerIsolation_RUNNER_ISOLATION_CONTAINER_PRIVILEGED:
	92	+	return domain.CIIsolationContainerPrivileged, nil
	93	+	}
	94	+	return "", fmt.Errorf("%w: isolation is required", domain.ErrInvalid)
	95	+	}
	96	+
	97	+	func stepStateToPB(s domain.CIStepState) pb.StepState {
	98	+	switch s {
	99	+	case domain.CIStepPending:
	100	+	return pb.StepState_STEP_STATE_PENDING
	101	+	case domain.CIStepInProgress:
	102	+	return pb.StepState_STEP_STATE_IN_PROGRESS
	103	+	case domain.CIStepCompleted:
	104	+	return pb.StepState_STEP_STATE_COMPLETED
	105	+	}
	106	+	return pb.StepState_STEP_STATE_UNSPECIFIED
	107	+	}
	108	+
	109	+	func stepStateFromPB(s pb.StepState) domain.CIStepState {
	110	+	switch s {
	111	+	case pb.StepState_STEP_STATE_PENDING:
	112	+	return domain.CIStepPending
	113	+	case pb.StepState_STEP_STATE_IN_PROGRESS:
	114	+	return domain.CIStepInProgress
	115	+	case pb.StepState_STEP_STATE_COMPLETED:
	116	+	return domain.CIStepCompleted
	117	+	}
	118	+	return ""
	119	+	}
	120	+
	121	+	func cancelReasonToPB(r domain.CICancelReason) pb.CancelReason {
	122	+	switch r {
	123	+	case domain.CICancelUser:
	124	+	return pb.CancelReason_CANCEL_REASON_USER
	125	+	case domain.CICancelFailFast:
	126	+	return pb.CancelReason_CANCEL_REASON_FAIL_FAST
	127	+	case domain.CICancelTimeout:
	128	+	return pb.CancelReason_CANCEL_REASON_TIMEOUT
	129	+	}
	130	+	return pb.CancelReason_CANCEL_REASON_UNSPECIFIED
	131	+	}
	132	+
	133	+	func runToPB(r domain.CIRun) *pb.Run {
	134	+	return &pb.Run{
	135	+	Repo: r.Repo.String(), Id: r.ID, Number: r.Number, Attempt: r.Attempt,
	136	+	WorkflowPath: r.WorkflowPath, WorkflowName: r.WorkflowName, Title: r.Title, Event: r.Event,
	137	+	Ref: r.Ref, Sha: r.Sha, Actor: r.Actor, Status: statusToPB(r.Status), Conclusion: conclusionToPB(r.Conclusion),
	138	+	Error: r.Error, CreatedAt: timestamp(r.CreatedAt), StartedAt: timestamp(r.StartedAt), CompletedAt: timestamp(r.CompletedAt),
	139	+	Version: int64(r.Version), //nolint:gosec // a store sequence, far below 2^63
	140	+	}
	141	+	}
	142	+
	143	+	func stepsToPB(steps []domain.CIStep, lines int64) []*pb.Step {
	144	+	ranges := domain.StepRanges(steps, lines)
	145	+	out := make([]*pb.Step, 0, len(steps))
	146	+	for i, s := range steps {
	147	+	out = append(out, &pb.Step{
	148	+	Number: s.Number, Name: s.Name, State: stepStateToPB(s.State), Conclusion: conclusionToPB(s.Conclusion),
	149	+	StartedAt: timestamp(s.StartedAt), CompletedAt: timestamp(s.CompletedAt),
	150	+	LogStart: ranges[i].Start, LogEnd: ranges[i].End,
	151	+	})
	152	+	}
	153	+	return out
	154	+	}
	155	+
	156	+	func jobToPB(j domain.CIJob, lines int64) *pb.RunJob {
	157	+	return &pb.RunJob{
	158	+	Id: j.ID, RunId: j.RunID, Attempt: j.Attempt, Repo: j.Repo.String(), Key: j.Key, Name: j.Name,
	159	+	Status: statusToPB(j.Status()), Conclusion: conclusionToPB(j.Conclusion), Error: j.Error,
	160	+	Labels: j.Labels, Needs: j.Needs, RunnerId: j.RunnerID, RunnerName: j.RunnerName, Isolation: isolationToPB(j.Isolation),
	161	+	Steps: stepsToPB(j.Steps, lines), CreatedAt: timestamp(j.CreatedAt), StartedAt: timestamp(j.StartedAt),
	162	+	CompletedAt: timestamp(j.CompletedAt),
	163	+	Version: int64(j.Version), //nolint:gosec // a store sequence, far below 2^63
	164	+	}
	165	+	}
	166	+
	167	+	func jobViewToPB(v ports.JobView) *pb.RunJob {
	168	+	out := jobToPB(v.CIJob, v.Lines)
	169	+	if v.RunnerGone {
	170	+	out.RunnerId = ""
	171	+	}
	172	+	return out
	173	+	}
	174	+
	175	+	func checkToPB(c domain.CICheck) *pb.Check {
	176	+	return &pb.Check{
	177	+	Id: c.ID, Repo: c.Repo.String(), Sha: c.Sha, Name: c.Name, Status: statusToPB(c.Status),
	178	+	Conclusion: conclusionToPB(c.Conclusion), RunId: c.RunID, JobId: c.JobID, Attempt: c.Attempt,
	179	+	WorkflowName: c.WorkflowName, StartedAt: timestamp(c.StartedAt), CompletedAt: timestamp(c.CompletedAt),
	180	+	}
	181	+	}
	182	+
	183	+	func runnerToPB(r ports.RunnerView) *pb.Runner {
	184	+	return &pb.Runner{
	185	+	Id: r.ID, Name: r.Name, Labels: r.Labels, Isolation: isolationToPB(r.Isolation), Version: r.Version,
	186	+	Arch: r.Arch, Os: r.OS, RegisteredAt: timestamp(r.RegisteredAt), LastSeen: timestamp(r.LastSeen), Busy: r.Busy,
	187	+	}
	188	+	}
	189	+
	190	+	func logLineToPB(l ports.JobLogLine) *pb.JobLogLine {
	191	+	return &pb.JobLogLine{Step: l.Step, Time: timestamp(l.Time), Text: l.Text, Continued: l.Continued}
	192	+	}
	193	+
	194	+	func runnerJobToPB(j ports.RunnerJob) pb.Job {
	195	+	return &pb.Job{
	196	+	JobId: j.Job.ID, RunId: j.Job.RunID, Attempt: j.Job.RunAttempt, JobToken: j.Token,
	197	+	WorkflowPath: j.WorkflowPath, WorkflowYaml: j.WorkflowYAML, JobKey: j.Job.Key, Matrix: j.Job.MatrixJSON,
	198	+	Contexts: j.Job.ContextsJSON, Labels: j.Job.Labels, TimeoutMinutes: j.Job.TimeoutMinutes,
	199	+	Repo: j.Repo.String(), Sha: j.Sha, Ref: j.Ref,
	200	+	}
	201	+	}
	202	+
	203	+	func stepUpdatesFromPB(steps []*pb.StepUpdate) ([]ports.StepUpdate, error) {
	204	+	out := make([]ports.StepUpdate, 0, len(steps))
	205	+	for _, s := range steps {
	206	+	concl, err := conclusionFromPB(s.GetConclusion())
	207	+	if err != nil {
	208	+	return nil, err
	209	+	}
	210	+	out = append(out, ports.StepUpdate{
	211	+	Number: s.GetNumber(), Name: s.GetName(), State: stepStateFromPB(s.GetState()), Conclusion: concl,
	212	+	StartedAt: goTime(s.GetStartedAt()), CompletedAt: goTime(s.GetCompletedAt()),
	213	+	})
	214	+	}
	215	+	return out, nil
	216	+	}
	217	+
	218	+	func jobEndedToPB(e domain.JobEndedError) pb.JobEnded {
	219	+	return &pb.JobEnded{Conclusion: conclusionToPB(e.Conclusion), Error: e.Message}
	220	+	}
	221	+
	222	+	func runEventToPB(ev domain.RunUpdated) *pb.Event_RunUpdated {
	223	+	return &pb.Event_RunUpdated{RunUpdated: &pb.RunUpdated{Run: runToPB(ev.Run)}}
	224	+	}
	225	+
	226	+	func jobEventToPB(ev domain.JobUpdated) *pb.Event_JobUpdated {
	227	+	return &pb.Event_JobUpdated{JobUpdated: &pb.JobUpdated{Job: jobToPB(ev.Job, ev.Lines)}}
	228	+	}

@@ -0,0 +1,228 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	package rpc
	4	+
	5	+	import (
	6	+	"fmt"
	7	+
	8	+	pb "bnema.dev/ermite-api/gen/ermite"
	9	+	"bnema.dev/ermite/internal/domain"
	10	+	"bnema.dev/ermite/internal/ports"
	11	+	)
	12	+
	13	+	func statusToPB(s domain.CIStatus) pb.CiStatus {
	14	+	switch s {
	15	+	case domain.CIQueued:
	16	+	return pb.CiStatus_CI_STATUS_QUEUED
	17	+	case domain.CIInProgress:
	18	+	return pb.CiStatus_CI_STATUS_IN_PROGRESS
	19	+	case domain.CICompleted:
	20	+	return pb.CiStatus_CI_STATUS_COMPLETED
	21	+	}
	22	+	return pb.CiStatus_CI_STATUS_UNSPECIFIED
	23	+	}
	24	+
	25	+	func statusFromPB(s pb.CiStatus) (domain.CIStatus, error) {
	26	+	switch s {
	27	+	case pb.CiStatus_CI_STATUS_UNSPECIFIED:
	28	+	return "", nil
	29	+	case pb.CiStatus_CI_STATUS_QUEUED:
	30	+	return domain.CIQueued, nil
	31	+	case pb.CiStatus_CI_STATUS_IN_PROGRESS:
	32	+	return domain.CIInProgress, nil
	33	+	case pb.CiStatus_CI_STATUS_COMPLETED:
	34	+	return domain.CICompleted, nil
	35	+	}
	36	+	return "", fmt.Errorf("%w: status %v", domain.ErrInvalid, s)
	37	+	}
	38	+
	39	+	var conclusionPairs = []struct {
	40	+	d domain.CIConclusion
	41	+	p pb.CiConclusion
	42	+	}{
	43	+	{domain.CISuccess, pb.CiConclusion_CI_CONCLUSION_SUCCESS},
	44	+	{domain.CIFailure, pb.CiConclusion_CI_CONCLUSION_FAILURE},
	45	+	{domain.CICancelled, pb.CiConclusion_CI_CONCLUSION_CANCELLED},
	46	+	{domain.CISkipped, pb.CiConclusion_CI_CONCLUSION_SKIPPED},
	47	+	{domain.CITimedOut, pb.CiConclusion_CI_CONCLUSION_TIMED_OUT},
	48	+	{domain.CINeutral, pb.CiConclusion_CI_CONCLUSION_NEUTRAL},
	49	+	{domain.CIActionRequired, pb.CiConclusion_CI_CONCLUSION_ACTION_REQUIRED},
	50	+	}
	51	+
	52	+	func conclusionToPB(c domain.CIConclusion) pb.CiConclusion {
	53	+	for _, e := range conclusionPairs {
	54	+	if e.d == c {
	55	+	return e.p
	56	+	}
	57	+	}
	58	+	return pb.CiConclusion_CI_CONCLUSION_UNSPECIFIED
	59	+	}
	60	+
	61	+	func conclusionFromPB(c pb.CiConclusion) (domain.CIConclusion, error) {
	62	+	if c == pb.CiConclusion_CI_CONCLUSION_UNSPECIFIED {
	63	+	return domain.CINoConclusion, nil
	64	+	}
	65	+	for _, e := range conclusionPairs {
	66	+	if e.p == c {
	67	+	return e.d, nil
	68	+	}
	69	+	}
	70	+	return "", fmt.Errorf("%w: conclusion %v", domain.ErrInvalid, c)
	71	+	}
	72	+
	73	+	func isolationToPB(i domain.CIIsolation) pb.RunnerIsolation {
	74	+	switch i {
	75	+	case domain.CIIsolationVM:
	76	+	return pb.RunnerIsolation_RUNNER_ISOLATION_VM
	77	+	case domain.CIIsolationContainerSysbox:
	78	+	return pb.RunnerIsolation_RUNNER_ISOLATION_CONTAINER_SYSBOX
	79	+	case domain.CIIsolationContainerPrivileged:
	80	+	return pb.RunnerIsolation_RUNNER_ISOLATION_CONTAINER_PRIVILEGED
	81	+	}
	82	+	return pb.RunnerIsolation_RUNNER_ISOLATION_UNSPECIFIED
	83	+	}
	84	+
	85	+	func isolationFromPB(i pb.RunnerIsolation) (domain.CIIsolation, error) {
	86	+	switch i {
	87	+	case pb.RunnerIsolation_RUNNER_ISOLATION_VM:
	88	+	return domain.CIIsolationVM, nil
	89	+	case pb.RunnerIsolation_RUNNER_ISOLATION_CONTAINER_SYSBOX:
	90	+	return domain.CIIsolationContainerSysbox, nil
	91	+	case pb.RunnerIsolation_RUNNER_ISOLATION_CONTAINER_PRIVILEGED:
	92	+	return domain.CIIsolationContainerPrivileged, nil
	93	+	}
	94	+	return "", fmt.Errorf("%w: isolation is required", domain.ErrInvalid)
	95	+	}
	96	+
	97	+	func stepStateToPB(s domain.CIStepState) pb.StepState {
	98	+	switch s {
	99	+	case domain.CIStepPending:
	100	+	return pb.StepState_STEP_STATE_PENDING
	101	+	case domain.CIStepInProgress:
	102	+	return pb.StepState_STEP_STATE_IN_PROGRESS
	103	+	case domain.CIStepCompleted:
	104	+	return pb.StepState_STEP_STATE_COMPLETED
	105	+	}
	106	+	return pb.StepState_STEP_STATE_UNSPECIFIED
	107	+	}
	108	+
	109	+	func stepStateFromPB(s pb.StepState) domain.CIStepState {
	110	+	switch s {
	111	+	case pb.StepState_STEP_STATE_PENDING:
	112	+	return domain.CIStepPending
	113	+	case pb.StepState_STEP_STATE_IN_PROGRESS:
	114	+	return domain.CIStepInProgress
	115	+	case pb.StepState_STEP_STATE_COMPLETED:
	116	+	return domain.CIStepCompleted
	117	+	}
	118	+	return ""
	119	+	}
	120	+
	121	+	func cancelReasonToPB(r domain.CICancelReason) pb.CancelReason {
	122	+	switch r {
	123	+	case domain.CICancelUser:
	124	+	return pb.CancelReason_CANCEL_REASON_USER
	125	+	case domain.CICancelFailFast:
	126	+	return pb.CancelReason_CANCEL_REASON_FAIL_FAST
	127	+	case domain.CICancelTimeout:
	128	+	return pb.CancelReason_CANCEL_REASON_TIMEOUT
	129	+	}
	130	+	return pb.CancelReason_CANCEL_REASON_UNSPECIFIED
	131	+	}
	132	+
	133	+	func runToPB(r domain.CIRun) *pb.Run {
	134	+	return &pb.Run{
	135	+	Repo: r.Repo.String(), Id: r.ID, Number: r.Number, Attempt: r.Attempt,
	136	+	WorkflowPath: r.WorkflowPath, WorkflowName: r.WorkflowName, Title: r.Title, Event: r.Event,
	137	+	Ref: r.Ref, Sha: r.Sha, Actor: r.Actor, Status: statusToPB(r.Status), Conclusion: conclusionToPB(r.Conclusion),
	138	+	Error: r.Error, CreatedAt: timestamp(r.CreatedAt), StartedAt: timestamp(r.StartedAt), CompletedAt: timestamp(r.CompletedAt),
	139	+	Version: int64(r.Version), //nolint:gosec // a store sequence, far below 2^63
	140	+	}
	141	+	}
	142	+
	143	+	func stepsToPB(steps []domain.CIStep, lines int64) []*pb.Step {
	144	+	ranges := domain.StepRanges(steps, lines)
	145	+	out := make([]*pb.Step, 0, len(steps))
	146	+	for i, s := range steps {
	147	+	out = append(out, &pb.Step{
	148	+	Number: s.Number, Name: s.Name, State: stepStateToPB(s.State), Conclusion: conclusionToPB(s.Conclusion),
	149	+	StartedAt: timestamp(s.StartedAt), CompletedAt: timestamp(s.CompletedAt),
	150	+	LogStart: ranges[i].Start, LogEnd: ranges[i].End,
	151	+	})
	152	+	}
	153	+	return out
	154	+	}
	155	+
	156	+	func jobToPB(j domain.CIJob, lines int64) *pb.RunJob {
	157	+	return &pb.RunJob{
	158	+	Id: j.ID, RunId: j.RunID, Attempt: j.Attempt, Repo: j.Repo.String(), Key: j.Key, Name: j.Name,
	159	+	Status: statusToPB(j.Status()), Conclusion: conclusionToPB(j.Conclusion), Error: j.Error,
	160	+	Labels: j.Labels, Needs: j.Needs, RunnerId: j.RunnerID, RunnerName: j.RunnerName, Isolation: isolationToPB(j.Isolation),
	161	+	Steps: stepsToPB(j.Steps, lines), CreatedAt: timestamp(j.CreatedAt), StartedAt: timestamp(j.StartedAt),
	162	+	CompletedAt: timestamp(j.CompletedAt),
	163	+	Version: int64(j.Version), //nolint:gosec // a store sequence, far below 2^63
	164	+	}
	165	+	}
	166	+
	167	+	func jobViewToPB(v ports.JobView) *pb.RunJob {
	168	+	out := jobToPB(v.CIJob, v.Lines)
	169	+	if v.RunnerGone {
	170	+	out.RunnerId = ""
	171	+	}
	172	+	return out
	173	+	}
	174	+
	175	+	func checkToPB(c domain.CICheck) *pb.Check {
	176	+	return &pb.Check{
	177	+	Id: c.ID, Repo: c.Repo.String(), Sha: c.Sha, Name: c.Name, Status: statusToPB(c.Status),
	178	+	Conclusion: conclusionToPB(c.Conclusion), RunId: c.RunID, JobId: c.JobID, Attempt: c.Attempt,
	179	+	WorkflowName: c.WorkflowName, StartedAt: timestamp(c.StartedAt), CompletedAt: timestamp(c.CompletedAt),
	180	+	}
	181	+	}
	182	+
	183	+	func runnerToPB(r ports.RunnerView) *pb.Runner {
	184	+	return &pb.Runner{
	185	+	Id: r.ID, Name: r.Name, Labels: r.Labels, Isolation: isolationToPB(r.Isolation), Version: r.Version,
	186	+	Arch: r.Arch, Os: r.OS, RegisteredAt: timestamp(r.RegisteredAt), LastSeen: timestamp(r.LastSeen), Busy: r.Busy,
	187	+	}
	188	+	}
	189	+
	190	+	func logLineToPB(l ports.JobLogLine) *pb.JobLogLine {
	191	+	return &pb.JobLogLine{Step: l.Step, Time: timestamp(l.Time), Text: l.Text, Continued: l.Continued}
	192	+	}
	193	+
	194	+	func runnerJobToPB(j ports.RunnerJob) pb.Job {
	195	+	return &pb.Job{
	196	+	JobId: j.Job.ID, RunId: j.Job.RunID, Attempt: j.Job.RunAttempt, JobToken: j.Token,
	197	+	WorkflowPath: j.WorkflowPath, WorkflowYaml: j.WorkflowYAML, JobKey: j.Job.Key, Matrix: j.Job.MatrixJSON,
	198	+	Contexts: j.Job.ContextsJSON, Labels: j.Job.Labels, TimeoutMinutes: j.Job.TimeoutMinutes,
	199	+	Repo: j.Repo.String(), Sha: j.Sha, Ref: j.Ref,
	200	+	}
	201	+	}
	202	+
	203	+	func stepUpdatesFromPB(steps []*pb.StepUpdate) ([]ports.StepUpdate, error) {
	204	+	out := make([]ports.StepUpdate, 0, len(steps))
	205	+	for _, s := range steps {
	206	+	concl, err := conclusionFromPB(s.GetConclusion())
	207	+	if err != nil {
	208	+	return nil, err
	209	+	}
	210	+	out = append(out, ports.StepUpdate{
	211	+	Number: s.GetNumber(), Name: s.GetName(), State: stepStateFromPB(s.GetState()), Conclusion: concl,
	212	+	StartedAt: goTime(s.GetStartedAt()), CompletedAt: goTime(s.GetCompletedAt()),
	213	+	})
	214	+	}
	215	+	return out, nil
	216	+	}
	217	+
	218	+	func jobEndedToPB(e domain.JobEndedError) pb.JobEnded {
	219	+	return &pb.JobEnded{Conclusion: conclusionToPB(e.Conclusion), Error: e.Message}
	220	+	}
	221	+
	222	+	func runEventToPB(ev domain.RunUpdated) *pb.Event_RunUpdated {
	223	+	return &pb.Event_RunUpdated{RunUpdated: &pb.RunUpdated{Run: runToPB(ev.Run)}}
	224	+	}
	225	+
	226	+	func jobEventToPB(ev domain.JobUpdated) *pb.Event_JobUpdated {
	227	+	return &pb.Event_JobUpdated{JobUpdated: &pb.JobUpdated{Job: jobToPB(ev.Job, ev.Lines)}}
	228	+	}

M internal/adapters/rpc/convert.go +6 -0


@@ -391,6 +391,12 @@ func watchEventToPB(we ports.WatchEvent) *pb.Event {
391	391		case domain.CommentAdded:
392	392		out.Time = timestamp(ev.At)
393	393		out.Kind = &pb.Event_CommentAdded{CommentAdded: &pb.CommentAdded{Repo: ev.Repo.String(), Number: ev.Number, Comment: commentToPB(ev.Comment)}}
	394	+	case domain.RunUpdated:
	395	+	out.Time = timestamp(ev.At)
	396	+	out.Kind = runEventToPB(ev)
	397	+	case domain.JobUpdated:
	398	+	out.Time = timestamp(ev.At)
	399	+	out.Kind = jobEventToPB(ev)
394	400		default:
395	401		return nil
396	402		}


@@ -391,6 +391,12 @@ func watchEventToPB(we ports.WatchEvent) *pb.Event {
391	case domain.CommentAdded:	391		case domain.CommentAdded:
392	out.Time = timestamp(ev.At)	392		out.Time = timestamp(ev.At)
393	out.Kind = &pb.Event_CommentAdded{CommentAdded: &pb.CommentAdded{Repo: ev.Repo.String(), Number: ev.Number, Comment: commentToPB(ev.Comment)}}	393		out.Kind = &pb.Event_CommentAdded{CommentAdded: &pb.CommentAdded{Repo: ev.Repo.String(), Number: ev.Number, Comment: commentToPB(ev.Comment)}}
		394	+	case domain.RunUpdated:
		395	+	out.Time = timestamp(ev.At)
		396	+	out.Kind = runEventToPB(ev)
		397	+	case domain.JobUpdated:
		398	+	out.Time = timestamp(ev.At)
		399	+	out.Kind = jobEventToPB(ev)
394	default:	400		default:
395	return nil	401		return nil
396	}	402		}

M internal/adapters/rpc/interceptor.go +16 -0


@@ -69,6 +69,13 @@ func (i *interceptor) authenticate(ctx context.Context, procedure, peerAddr stri
69	69		if !present {
70	70		return ctx, nil
71	71		}
	72	+	if runnerSecretProcedures[procedure] {
	73	+	// A runner or job token: not a stored token. The use case checks it
	74	+	// and answers unauthenticated; runnerService.failed counts the
	75	+	// failures against the client address.
	76	+	ctx = context.WithValue(ctx, addrKey{}, addr)
	77	+	return context.WithValue(ctx, secretKey{}, secret), nil
	78	+	}
72	79		p, err := i.auth.Authenticate(ctx, secret)
73	80		switch {
74	81		case errors.Is(err, domain.ErrUnauthenticated):

@@ -83,6 +90,15 @@ func (i *interceptor) authenticate(ctx context.Context, procedure, peerAddr stri
83	90
84	91		type clientKey struct{}
85	92
	93	+	// addrKey carries the normalized client address of a call whose credential the
	94	+	// use case checks.
	95	+	type addrKey struct{}
	96	+
	97	+	func addrFromContext(ctx context.Context) string {
	98	+	a, _ := ctx.Value(addrKey{}).(string)
	99	+	return a
	100	+	}
	101	+
86	102		// clientFromContext returns the normalized address of the client of a call:
87	103		// its peer, empty when there is none.
88	104		func clientFromContext(ctx context.Context) string {


@@ -69,6 +69,13 @@ func (i *interceptor) authenticate(ctx context.Context, procedure, peerAddr stri
69	if !present {	69		if !present {
70	return ctx, nil	70		return ctx, nil
71	}	71		}
		72	+	if runnerSecretProcedures[procedure] {
		73	+	// A runner or job token: not a stored token. The use case checks it
		74	+	// and answers unauthenticated; runnerService.failed counts the
		75	+	// failures against the client address.
		76	+	ctx = context.WithValue(ctx, addrKey{}, addr)
		77	+	return context.WithValue(ctx, secretKey{}, secret), nil
		78	+	}
72	p, err := i.auth.Authenticate(ctx, secret)	79		p, err := i.auth.Authenticate(ctx, secret)
73	switch {	80		switch {
74	case errors.Is(err, domain.ErrUnauthenticated):	81		case errors.Is(err, domain.ErrUnauthenticated):

@@ -83,6 +90,15 @@ func (i *interceptor) authenticate(ctx context.Context, procedure, peerAddr stri
83		90
84	type clientKey struct{}	91		type clientKey struct{}
85		92
		93	+	// addrKey carries the normalized client address of a call whose credential the
		94	+	// use case checks.
		95	+	type addrKey struct{}
		96	+
		97	+	func addrFromContext(ctx context.Context) string {
		98	+	a, _ := ctx.Value(addrKey{}).(string)
		99	+	return a
		100	+	}
		101	+
86	// clientFromContext returns the normalized address of the client of a call:	102		// clientFromContext returns the normalized address of the client of a call:
87	// its peer, empty when there is none.	103		// its peer, empty when there is none.
88	func clientFromContext(ctx context.Context) string {	104		func clientFromContext(ctx context.Context) string {

M internal/app/api_helpers_test.go +4 -0


@@ -33,6 +33,8 @@ type clients struct {
33	33		mr ermiteconnect.MergeRequestServiceClient
34	34
35	35		session ermiteconnect.SessionServiceClient
	36	+	runner ermiteconnect.RunnerServiceClient
	37	+	run ermiteconnect.RunServiceClient
36	38		}
37	39
38	40		// bearerTransport adds the Authorization header of a secret, if any.

@@ -81,6 +83,8 @@ func newClientsOn(hc connect.HTTPClient, base string, opts ...connect.ClientOpti
81	83		mr: ermiteconnect.NewMergeRequestServiceClient(hc, base, opts...),
82	84
83	85		session: ermiteconnect.NewSessionServiceClient(hc, base, opts...),
	86	+	runner: ermiteconnect.NewRunnerServiceClient(hc, base, opts...),
	87	+	run: ermiteconnect.NewRunServiceClient(hc, base, opts...),
84	88		}
85	89		}
86	90


@@ -33,6 +33,8 @@ type clients struct {
33	mr ermiteconnect.MergeRequestServiceClient	33		mr ermiteconnect.MergeRequestServiceClient
34		34
35	session ermiteconnect.SessionServiceClient	35		session ermiteconnect.SessionServiceClient
		36	+	runner ermiteconnect.RunnerServiceClient
		37	+	run ermiteconnect.RunServiceClient
36	}	38		}
37		39
38	// bearerTransport adds the Authorization header of a secret, if any.	40		// bearerTransport adds the Authorization header of a secret, if any.

@@ -81,6 +83,8 @@ func newClientsOn(hc connect.HTTPClient, base string, opts ...connect.ClientOpti
81	mr: ermiteconnect.NewMergeRequestServiceClient(hc, base, opts...),	83		mr: ermiteconnect.NewMergeRequestServiceClient(hc, base, opts...),
82		84
83	session: ermiteconnect.NewSessionServiceClient(hc, base, opts...),	85		session: ermiteconnect.NewSessionServiceClient(hc, base, opts...),
		86	+	runner: ermiteconnect.NewRunnerServiceClient(hc, base, opts...),
		87	+	run: ermiteconnect.NewRunServiceClient(hc, base, opts...),
84	}	88		}
85	}	89		}
86		90

A internal/app/ci_e2e_test.go +443 -0

@@ -0,0 +1,443 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	package app
	4	+
	5	+	import (
	6	+	"errors"
	7	+	"fmt"
	8	+	"strings"
	9	+	"sync"
	10	+	"testing"
	11	+	"time"
	12	+
	13	+	"connectrpc.com/connect"
	14	+	"github.com/stretchr/testify/assert"
	15	+	"github.com/stretchr/testify/require"
	16	+	"google.golang.org/protobuf/types/known/fieldmaskpb"
	17	+	"google.golang.org/protobuf/types/known/timestamppb"
	18	+
	19	+	pb "bnema.dev/ermite-api/gen/ermite"
	20	+	"bnema.dev/ermite/internal/domain"
	21	+	"bnema.dev/ermite/internal/ports"
	22	+	"bnema.dev/ermite/internal/usecase"
	23	+	)
	24	+
	25	+	// ciHarness is a server with a repository whose pushes run workflows, and a
	26	+	// fake runner that talks to the real Connect handlers.
	27	+	type ciHarness struct {
	28	+	*harness
	29	+	root *clients
	30	+	wt string
	31	+	repo string
	32	+	}
	33	+
	34	+	const ciWorkflow = `name: ci
	35	+	on:
	36	+	push:
	37	+	branches: [main, feat]
	38	+	jobs:
	39	+	build:
	40	+	runs-on: ubuntu-latest
	41	+	steps:
	42	+	- run: echo hello
	43	+	test:
	44	+	needs: build
	45	+	runs-on: ubuntu-latest
	46	+	steps:
	47	+	- run: echo test
	48	+	`
	49	+
	50	+	func newCIHarness(t testing.T, tweak func(usecase.CIConfig)) *ciHarness {
	51	+	t.Helper()
	52	+	h := newHarnessWith(t, harnessOpts{}, func(o *options) { o.ci = tweak })
	53	+	c := &ciHarness{harness: h, root: h.api(adminToken), repo: "proj"}
	54	+	_, err := h.forge.Repos.Create(h.ctx, domain.Root(), ports.CreateRepoParams{Name: h.name("proj"), Visibility: domain.VisibilityPrivate})
	55	+	require.NoError(t, err)
	56	+	c.wt = h.workTree("main")
	57	+	return c
	58	+	}
	59	+
	60	+	// push commits the files to branch and pushes it.
	61	+	func (c *ciHarness) push(branch string, files map[string]string, msg string) string {
	62	+	c.t.Helper()
	63	+	if cur, err := c.git(c.wt, "rev-parse", "--abbrev-ref", "HEAD"); err != nil \|\| strings.TrimSpace(cur) != branch {
	64	+	if _, err := c.git(c.wt, "checkout", "-q", branch); err != nil {
	65	+	c.mustGit(c.wt, "checkout", "-q", "-b", branch)
	66	+	}
	67	+	}
	68	+	for name, content := range files {
	69	+	c.writeFile(c.wt, name, []byte(content))
	70	+	}
	71	+	if msg != "" {
	72	+	c.mustGit(c.wt, "commit", "-q", "--amend", "-m", msg)
	73	+	}
	74	+	c.mustGit(c.wt, "push", "-q", c.urlAuth(adminToken, "/"+c.repo+".git"), branch)
	75	+	return strings.TrimSpace(c.mustGit(c.wt, "rev-parse", "HEAD"))
	76	+	}
	77	+
	78	+	// registerRunner registers a runner and returns a fake that polls with it.
	79	+	func (c ciHarness) registerRunner(name string, labels ...string) fakeRunner {
	80	+	c.t.Helper()
	81	+	res, err := c.root.runner.RegisterRunner(c.ctx, connect.NewRequest(&pb.RegisterRunnerRequest{
	82	+	Name: name, Labels: labels, Version: "devel", Isolation: pb.RunnerIsolation_RUNNER_ISOLATION_VM, Arch: "x64", Os: "linux",
	83	+	}))
	84	+	require.NoError(c.t, err)
	85	+	return &fakeRunner{c: c, id: res.Msg.GetRunnerId(), api: c.api(res.Msg.GetRunnerToken())}
	86	+	}
	87	+
	88	+	// fakeRunner is the runner side of the protocol, as a test.
	89	+	type fakeRunner struct {
	90	+	c *ciHarness
	91	+	id string
	92	+	api *clients
	93	+	}
	94	+
	95	+	// fetch polls once for a job.
	96	+	func (r fakeRunner) fetch(pollID string) (pb.Job, error) {
	97	+	res, err := r.api.runner.FetchJob(r.c.ctx, connect.NewRequest(&pb.FetchJobRequest{PollId: pollID}))
	98	+	if err != nil {
	99	+	return nil, err
	100	+	}
	101	+	return res.Msg.GetJob(), nil
	102	+	}
	103	+
	104	+	// await polls until a job is given.
	105	+	func (r fakeRunner) await() pb.Job {
	106	+	r.c.t.Helper()
	107	+	deadline := time.Now().Add(15 * time.Second)
	108	+	for time.Now().Before(deadline) {
	109	+	j, err := r.fetch("")
	110	+	require.NoError(r.c.t, err)
	111	+	if j != nil {
	112	+	return j
	113	+	}
	114	+	}
	115	+	r.c.t.Fatal("no job was given to the runner")
	116	+	return nil
	117	+	}
	118	+
	119	+	// jobAPI is the client of a job: its token is the bearer secret.
	120	+	func (r fakeRunner) jobAPI(j pb.Job) *clients { return r.c.api(j.GetJobToken()) }
	121	+
	122	+	func now() *timestamppb.Timestamp { return timestamppb.Now() }
	123	+
	124	+	// runJob executes a job as a runner does: heartbeat, steps, log, completion.
	125	+	func (r fakeRunner) runJob(j pb.Job, conclusion pb.CiConclusion, outputs map[string]string, text ...string) {
	126	+	r.c.t.Helper()
	127	+	ctx, api := r.c.ctx, r.jobAPI(j)
	128	+	_, err := api.runner.Heartbeat(ctx, connect.NewRequest(&pb.HeartbeatRequest{JobId: j.GetJobId()}))
	129	+	require.NoError(r.c.t, err)
	130	+	start := now()
	131	+	_, err = api.runner.UpdateJob(ctx, connect.NewRequest(&pb.UpdateJobRequest{JobId: j.GetJobId(), Steps: []*pb.StepUpdate{
	132	+	{Number: 1, Name: "Set up job", State: pb.StepState_STEP_STATE_IN_PROGRESS, StartedAt: start},
	133	+	}}))
	134	+	require.NoError(r.c.t, err)
	135	+	var lines []*pb.LogLine
	136	+	for _, t := range text {
	137	+	lines = append(lines, &pb.LogLine{Time: now(), Text: t})
	138	+	}
	139	+	if len(lines) > 0 {
	140	+	_, err = api.runner.AppendLog(ctx, connect.NewRequest(&pb.AppendLogRequest{JobId: j.GetJobId(), Step: 2, Offset: 0, Lines: lines}))
	141	+	require.NoError(r.c.t, err)
	142	+	}
	143	+	step := pb.CiConclusion_CI_CONCLUSION_SUCCESS
	144	+	if conclusion != pb.CiConclusion_CI_CONCLUSION_SUCCESS {
	145	+	step = conclusion
	146	+	}
	147	+	res, err := api.runner.CompleteJob(ctx, connect.NewRequest(&pb.CompleteJobRequest{
	148	+	JobId: j.GetJobId(), Conclusion: conclusion, Outputs: outputs,
	149	+	Steps: []*pb.StepUpdate{
	150	+	{Number: 1, Name: "Set up job", State: pb.StepState_STEP_STATE_COMPLETED, Conclusion: pb.CiConclusion_CI_CONCLUSION_SUCCESS, StartedAt: start, CompletedAt: now()},
	151	+	{Number: 2, Name: "Run echo", State: pb.StepState_STEP_STATE_COMPLETED, Conclusion: step, StartedAt: start, CompletedAt: now()},
	152	+	},
	153	+	}))
	154	+	require.NoError(r.c.t, err)
	155	+	assert.True(r.c.t, res.Msg.GetApplied())
	156	+	}
	157	+
	158	+	// waitFor polls cond until it holds.
	159	+	func (c *ciHarness) waitFor(what string, cond func() bool) {
	160	+	c.t.Helper()
	161	+	require.Eventually(c.t, cond, 15time.Second, 20time.Millisecond, what)
	162	+	}
	163	+
	164	+	func (c ciHarness) runs() []pb.Run {
	165	+	c.t.Helper()
	166	+	res, err := c.root.run.ListRuns(c.ctx, connect.NewRequest(&pb.ListRunsRequest{Repo: c.repo}))
	167	+	require.NoError(c.t, err)
	168	+	return res.Msg.GetRuns()
	169	+	}
	170	+
	171	+	func (c ciHarness) checks(sha string) []pb.Check {
	172	+	c.t.Helper()
	173	+	res, err := c.root.run.ListChecks(c.ctx, connect.NewRequest(&pb.ListChecksRequest{Repo: c.repo, Sha: sha}))
	174	+	require.NoError(c.t, err)
	175	+	return res.Msg.GetChecks()
	176	+	}
	177	+
	178	+	func (c ciHarness) getRun(id int64) pb.GetRunResponse {
	179	+	c.t.Helper()
	180	+	res, err := c.root.run.GetRun(c.ctx, connect.NewRequest(&pb.GetRunRequest{Repo: c.repo, RunId: id}))
	181	+	require.NoError(c.t, err)
	182	+	return res.Msg
	183	+	}
	184	+
	185	+	func ciErrCode(err error) connect.Code {
	186	+	var ce *connect.Error
	187	+	if errors.As(err, &ce) {
	188	+	return ce.Code()
	189	+	}
	190	+	return connect.Code(0)
	191	+	}
	192	+
	193	+	// AC3, AC5: a push runs the workflow on a runner; its jobs follow their needs;
	194	+	// the checks appear on the commit; the log is readable.
	195	+	func TestCIPushRunsAWorkflowOnARunner(t *testing.T) {
	196	+	c := newCIHarness(t, nil)
	197	+	runner := c.registerRunner("alpha", "ubuntu-latest")
	198	+	sha := c.push("main", map[string]string{".github/workflows/ci.yml": ciWorkflow}, "")
	199	+
	200	+	c.waitFor("a run is created", func() bool { return len(c.runs()) == 1 })
	201	+	run := c.runs()[0]
	202	+	assert.Equal(t, "ci", run.GetWorkflowName())
	203	+	assert.Equal(t, "refs/heads/main", run.GetRef())
	204	+	assert.Equal(t, sha, run.GetSha())
	205	+	assert.Equal(t, "push", run.GetEvent())
	206	+	assert.EqualValues(t, 1, run.GetNumber())
	207	+	assert.Equal(t, "root", run.GetActor())
	208	+
	209	+	// Only the first job is ready: the second needs it.
	210	+	job := runner.await()
	211	+	assert.Equal(t, "build", job.GetJobKey())
	212	+	assert.Equal(t, run.GetId(), job.GetRunId())
	213	+	assert.Equal(t, sha, job.GetSha())
	214	+	assert.Equal(t, "refs/heads/main", job.GetRef())
	215	+	assert.Equal(t, "proj", job.GetRepo())
	216	+	assert.Equal(t, ciWorkflow, job.GetWorkflowYaml())
	217	+	assert.Equal(t, []string{"ubuntu-latest"}, job.GetLabels())
	218	+	assert.EqualValues(t, 360, job.GetTimeoutMinutes())
	219	+	assert.Equal(t, "{}", job.GetMatrix())
	220	+	assert.True(t, strings.HasPrefix(job.GetJobToken(), "ermitejob_"))
	221	+	assert.Contains(t, job.GetContexts(), `"repository":"owner/proj"`)
	222	+	assert.Contains(t, job.GetContexts(), `"needs":{}`)
	223	+
	224	+	// Assigned but not acknowledged: still queued for the owner.
	225	+	assert.Equal(t, pb.CiStatus_CI_STATUS_QUEUED, c.getRun(run.GetId()).GetJobs()[0].GetStatus())
	226	+
	227	+	runner.runJob(job, pb.CiConclusion_CI_CONCLUSION_SUCCESS, map[string]string{"artifact": "out.bin"}, "hello", "world")
	228	+
	229	+	next := runner.await()
	230	+	assert.Equal(t, "test", next.GetJobKey())
	231	+	assert.Contains(t, next.GetContexts(), `"needs":{"build":{"result":"success","outputs":{"artifact":"out.bin"}}}`)
	232	+	runner.runJob(next, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil, "tested")
	233	+
	234	+	c.waitFor("the run completes", func() bool {
	235	+	return c.runs()[0].GetStatus() == pb.CiStatus_CI_STATUS_COMPLETED
	236	+	})
	237	+	assert.Equal(t, pb.CiConclusion_CI_CONCLUSION_SUCCESS, c.runs()[0].GetConclusion())
	238	+
	239	+	checks := c.checks(sha)
	240	+	require.Len(t, checks, 2)
	241	+	assert.Equal(t, "build", checks[0].GetName())
	242	+	assert.Equal(t, "test", checks[1].GetName())
	243	+	for _, ck := range checks {
	244	+	assert.Equal(t, pb.CiStatus_CI_STATUS_COMPLETED, ck.GetStatus())
	245	+	assert.Equal(t, pb.CiConclusion_CI_CONCLUSION_SUCCESS, ck.GetConclusion())
	246	+	}
	247	+
	248	+	detail := c.getRun(run.GetId())
	249	+	require.Len(t, detail.GetJobs(), 2)
	250	+	steps := detail.GetJobs()[0].GetSteps()
	251	+	require.Len(t, steps, 2)
	252	+	assert.EqualValues(t, 0, steps[1].GetLogStart(), "step 2 has the lines")
	253	+	assert.EqualValues(t, 2, steps[1].GetLogEnd())
	254	+
	255	+	// The log is streamed back with its steps.
	256	+	stream, err := c.root.run.GetJobLog(c.ctx, connect.NewRequest(&pb.GetJobLogRequest{Repo: "proj", JobId: job.GetJobId()}))
	257	+	require.NoError(t, err)
	258	+	var got []string
	259	+	for stream.Receive() {
	260	+	for _, l := range stream.Msg().GetLines() {
	261	+	got = append(got, l.GetText())
	262	+	assert.EqualValues(t, 2, l.GetStep())
	263	+	}
	264	+	}
	265	+	require.NoError(t, stream.Err())
	266	+	assert.Equal(t, []string{"hello", "world"}, got)
	267	+	}
	268	+
	269	+	// AC5: merge refused while checks are pending or failing, allowed when they
	270	+	// pass, and when failing_checks_block is off.
	271	+	func TestCIMergeIsBlockedByChecks(t *testing.T) {
	272	+	c := newCIHarness(t, nil)
	273	+	runner := c.registerRunner("alpha", "ubuntu-latest")
	274	+	c.push("main", map[string]string{"README": "x\n"}, "")
	275	+
	276	+	feat := c.push("feat", map[string]string{".github/workflows/ci.yml": strings.ReplaceAll(ciWorkflow, " test:\n needs: build\n runs-on: ubuntu-latest\n steps:\n - run: echo test\n", "")}, "")
	277	+	mr, err := c.root.mr.CreateMergeRequest(c.ctx, connect.NewRequest(&pb.CreateMergeRequestRequest{Repo: "proj", Title: "Add CI", SourceBranch: "feat"}))
	278	+	require.NoError(t, err)
	279	+	merge := func() error {
	280	+	_, err := c.root.mr.MergeMergeRequest(c.ctx, connect.NewRequest(&pb.MergeMergeRequestRequest{Repo: "proj", Number: mr.Msg.GetMergeRequest().GetNumber(), HeadId: feat}))
	281	+	return err
	282	+	}
	283	+
	284	+	c.waitFor("the feature run is queued", func() bool { return len(c.checks(feat)) == 1 })
	285	+	err = merge()
	286	+	require.Error(t, err)
	287	+	assert.Equal(t, connect.CodeFailedPrecondition, ciErrCode(err), "a pending check blocks the merge")
	288	+	assert.Contains(t, err.Error(), "have not finished")
	289	+
	290	+	job := runner.await()
	291	+	runner.runJob(job, pb.CiConclusion_CI_CONCLUSION_FAILURE, nil, "boom")
	292	+	c.waitFor("the check fails", func() bool {
	293	+	cs := c.checks(feat)
	294	+	return len(cs) == 1 && cs[0].GetStatus() == pb.CiStatus_CI_STATUS_COMPLETED
	295	+	})
	296	+	err = merge()
	297	+	require.Error(t, err)
	298	+	assert.Equal(t, connect.CodeFailedPrecondition, ciErrCode(err))
	299	+	assert.Contains(t, err.Error(), "failing checks block the merge")
	300	+
	301	+	// Off: the merge goes through.
	302	+	_, err = c.root.repo.UpdateRepo(c.ctx, connect.NewRequest(&pb.UpdateRepoRequest{
	303	+	Repo: "proj",
	304	+	Settings: &pb.RepoSettings{FailingChecksBlock: false},
	305	+	UpdateMask: &fieldmaskpb.FieldMask{Paths: []string{"settings.failing_checks_block"}},
	306	+	}))
	307	+	require.NoError(t, err)
	308	+	require.NoError(t, merge())
	309	+	}
	310	+
	311	+	// AC2: a job token works for its job only, and not once the job has ended.
	312	+	func TestCIJobTokensAreScopedToTheirJob(t *testing.T) {
	313	+	c := newCIHarness(t, nil)
	314	+	runner := c.registerRunner("alpha", "ubuntu-latest")
	315	+	c.push("main", map[string]string{".github/workflows/a.yml": ciWorkflow, ".github/workflows/b.yml": ciWorkflow}, "")
	316	+	c.waitFor("two runs", func() bool { return len(c.runs()) == 2 })
	317	+	j1, j2 := runner.await(), runner.await()
	318	+	require.NotEqual(t, j1.GetJobId(), j2.GetJobId())
	319	+
	320	+	// Another job's id with this job's token: permission_denied.
	321	+	_, err := runner.jobAPI(j1).runner.Heartbeat(c.ctx, connect.NewRequest(&pb.HeartbeatRequest{JobId: j2.GetJobId()}))
	322	+	assert.Equal(t, connect.CodePermissionDenied, ciErrCode(err))
	323	+	// A runner token is not a job token, and a stored token is neither.
	324	+	_, err = runner.api.runner.Heartbeat(c.ctx, connect.NewRequest(&pb.HeartbeatRequest{JobId: j1.GetJobId()}))
	325	+	assert.Equal(t, connect.CodeUnauthenticated, ciErrCode(err))
	326	+	_, err = c.root.runner.Heartbeat(c.ctx, connect.NewRequest(&pb.HeartbeatRequest{JobId: j1.GetJobId()}))
	327	+	assert.Equal(t, connect.CodeUnauthenticated, ciErrCode(err))
	328	+	// A job token does not poll.
	329	+	_, err = runner.jobAPI(j1).runner.FetchJob(c.ctx, connect.NewRequest(&pb.FetchJobRequest{}))
	330	+	assert.Equal(t, connect.CodeUnauthenticated, ciErrCode(err))
	331	+
	332	+	runner.runJob(j1, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil, "a")
	333	+	// After the end: failed_precondition with the JobEnded detail; CompleteJob
	334	+	// still answers with the stored result.
	335	+	_, err = runner.jobAPI(j1).runner.AppendLog(c.ctx, connect.NewRequest(&pb.AppendLogRequest{
	336	+	JobId: j1.GetJobId(), Step: 2, Offset: 1, Lines: []*pb.LogLine{{Time: now(), Text: "late"}},
	337	+	}))
	338	+	require.Error(t, err)
	339	+	assert.Equal(t, connect.CodeFailedPrecondition, ciErrCode(err))
	340	+	var ce *connect.Error
	341	+	require.True(t, errors.As(err, &ce))
	342	+	require.Len(t, ce.Details(), 1)
	343	+	msg, derr := ce.Details()[0].Value()
	344	+	require.NoError(t, derr)
	345	+	assert.Equal(t, pb.CiConclusion_CI_CONCLUSION_SUCCESS, msg.(*pb.JobEnded).GetConclusion())
	346	+
	347	+	res, err := runner.jobAPI(j1).runner.CompleteJob(c.ctx, connect.NewRequest(&pb.CompleteJobRequest{
	348	+	JobId: j1.GetJobId(), Conclusion: pb.CiConclusion_CI_CONCLUSION_FAILURE,
	349	+	}))
	350	+	require.NoError(t, err)
	351	+	assert.False(t, res.Msg.GetApplied(), "a second CompleteJob changes nothing")
	352	+	assert.Equal(t, pb.CiConclusion_CI_CONCLUSION_SUCCESS, res.Msg.GetConclusion(), "and answers the stored conclusion")
	353	+	}
	354	+
	355	+	// AC2: a job is given to exactly one runner, however many poll at once.
	356	+	func TestCIConcurrentFetchJobAssignsOnce(t *testing.T) {
	357	+	c := newCIHarness(t, func(cfg usecase.CIConfig) { cfg.PollTimeout = 300 time.Millisecond })
	358	+	var runners []*fakeRunner
	359	+	for i := range 8 {
	360	+	runners = append(runners, c.registerRunner(fmt.Sprintf("r%d", i), "ubuntu-latest"))
	361	+	}
	362	+	var wf strings.Builder
	363	+	wf.WriteString("on: push\njobs:\n")
	364	+	for i := range 3 {
	365	+	fmt.Fprintf(&wf, " j%d:\n runs-on: ubuntu-latest\n steps:\n - run: echo %d\n", i, i)
	366	+	}
	367	+	c.push("main", map[string]string{".github/workflows/many.yml": wf.String()}, "")
	368	+	c.waitFor("the run exists", func() bool { return len(c.runs()) == 1 })
	369	+
	370	+	var mu sync.Mutex
	371	+	got := map[string]int{}
	372	+	var wg sync.WaitGroup
	373	+	for _, r := range runners {
	374	+	wg.Add(1)
	375	+	go func() {
	376	+	defer wg.Done()
	377	+	for range 20 {
	378	+	j, err := r.fetch("")
	379	+	if err != nil {
	380	+	return
	381	+	}
	382	+	if j != nil {
	383	+	mu.Lock()
	384	+	got[j.GetJobId()]++
	385	+	mu.Unlock()
	386	+	}
	387	+	}
	388	+	}()
	389	+	}
	390	+	wg.Wait()
	391	+	assert.Len(t, got, 3, "every job was given")
	392	+	for id, n := range got {
	393	+	assert.Equal(t, 1, n, "job %s was given %d times", id, n)
	394	+	}
	395	+	}
	396	+
	397	+	// AC2: the first heartbeat acknowledges; without it the assignment is
	398	+	// cancelled and the job is queued again, three times in a row at most.
	399	+	func TestCIUnacknowledgedAssignmentGoesBackToTheQueue(t *testing.T) {
	400	+	c := newCIHarness(t, func(cfg *usecase.CIConfig) {
	401	+	cfg.AckTimeout, cfg.Tick, cfg.PollTimeout = 150time.Millisecond, 20time.Millisecond, 100*time.Millisecond
	402	+	})
	403	+	runner := c.registerRunner("alpha", "ubuntu-latest")
	404	+	c.push("main", map[string]string{".github/workflows/ci.yml": strings.Split(ciWorkflow, " test:")[0]}, "")
	405	+
	406	+	first := runner.await()
	407	+	// No heartbeat: the token is revoked and the job comes back with a new one.
	408	+	second := runner.await()
	409	+	assert.Equal(t, first.GetJobId(), second.GetJobId())
	410	+	assert.NotEqual(t, first.GetJobToken(), second.GetJobToken())
	411	+	_, err := runner.jobAPI(first).runner.Heartbeat(c.ctx, connect.NewRequest(&pb.HeartbeatRequest{JobId: first.GetJobId()}))
	412	+	assert.Equal(t, connect.CodeUnauthenticated, ciErrCode(err), "the replaced token is rejected")
	413	+
	414	+	// The third assignment is the last: ignoring it fails the job.
	415	+	runner.await()
	416	+	c.waitFor("the job fails", func() bool {
	417	+	return c.runs()[0].GetStatus() == pb.CiStatus_CI_STATUS_COMPLETED
	418	+	})
	419	+	detail := c.getRun(c.runs()[0].GetId())
	420	+	assert.Equal(t, pb.CiConclusion_CI_CONCLUSION_FAILURE, detail.GetJobs()[0].GetConclusion())
	421	+	assert.Contains(t, detail.GetJobs()[0].GetError(), "acknowledged")
	422	+	}
	423	+
	424	+	// A retried poll with the same poll_id returns the same assignment.
	425	+	func TestCIRetriedPollReturnsTheSameAssignment(t *testing.T) {
	426	+	c := newCIHarness(t, func(cfg usecase.CIConfig) { cfg.PollTimeout = 200 time.Millisecond })
	427	+	runner := c.registerRunner("alpha", "ubuntu-latest")
	428	+	c.push("main", map[string]string{".github/workflows/ci.yml": strings.Split(ciWorkflow, " test:")[0]}, "")
	429	+	c.waitFor("a run", func() bool { return len(c.runs()) == 1 })
	430	+
	431	+	var first *pb.Job
	432	+	c.waitFor("a job", func() bool {
	433	+	j, err := runner.fetch("poll-1")
	434	+	require.NoError(t, err)
	435	+	first = j
	436	+	return j != nil
	437	+	})
	438	+	again, err := runner.fetch("poll-1")
	439	+	require.NoError(t, err)
	440	+	require.NotNil(t, again)
	441	+	assert.Equal(t, first.GetJobId(), again.GetJobId())
	442	+	assert.Equal(t, first.GetJobToken(), again.GetJobToken())
	443	+	}

@@ -0,0 +1,443 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	package app
	4	+
	5	+	import (
	6	+	"errors"
	7	+	"fmt"
	8	+	"strings"
	9	+	"sync"
	10	+	"testing"
	11	+	"time"
	12	+
	13	+	"connectrpc.com/connect"
	14	+	"github.com/stretchr/testify/assert"
	15	+	"github.com/stretchr/testify/require"
	16	+	"google.golang.org/protobuf/types/known/fieldmaskpb"
	17	+	"google.golang.org/protobuf/types/known/timestamppb"
	18	+
	19	+	pb "bnema.dev/ermite-api/gen/ermite"
	20	+	"bnema.dev/ermite/internal/domain"
	21	+	"bnema.dev/ermite/internal/ports"
	22	+	"bnema.dev/ermite/internal/usecase"
	23	+	)
	24	+
	25	+	// ciHarness is a server with a repository whose pushes run workflows, and a
	26	+	// fake runner that talks to the real Connect handlers.
	27	+	type ciHarness struct {
	28	+	*harness
	29	+	root *clients
	30	+	wt string
	31	+	repo string
	32	+	}
	33	+
	34	+	const ciWorkflow = `name: ci
	35	+	on:
	36	+	push:
	37	+	branches: [main, feat]
	38	+	jobs:
	39	+	build:
	40	+	runs-on: ubuntu-latest
	41	+	steps:
	42	+	- run: echo hello
	43	+	test:
	44	+	needs: build
	45	+	runs-on: ubuntu-latest
	46	+	steps:
	47	+	- run: echo test
	48	+	`
	49	+
	50	+	func newCIHarness(t testing.T, tweak func(usecase.CIConfig)) *ciHarness {
	51	+	t.Helper()
	52	+	h := newHarnessWith(t, harnessOpts{}, func(o *options) { o.ci = tweak })
	53	+	c := &ciHarness{harness: h, root: h.api(adminToken), repo: "proj"}
	54	+	_, err := h.forge.Repos.Create(h.ctx, domain.Root(), ports.CreateRepoParams{Name: h.name("proj"), Visibility: domain.VisibilityPrivate})
	55	+	require.NoError(t, err)
	56	+	c.wt = h.workTree("main")
	57	+	return c
	58	+	}
	59	+
	60	+	// push commits the files to branch and pushes it.
	61	+	func (c *ciHarness) push(branch string, files map[string]string, msg string) string {
	62	+	c.t.Helper()
	63	+	if cur, err := c.git(c.wt, "rev-parse", "--abbrev-ref", "HEAD"); err != nil \|\| strings.TrimSpace(cur) != branch {
	64	+	if _, err := c.git(c.wt, "checkout", "-q", branch); err != nil {
	65	+	c.mustGit(c.wt, "checkout", "-q", "-b", branch)
	66	+	}
	67	+	}
	68	+	for name, content := range files {
	69	+	c.writeFile(c.wt, name, []byte(content))
	70	+	}
	71	+	if msg != "" {
	72	+	c.mustGit(c.wt, "commit", "-q", "--amend", "-m", msg)
	73	+	}
	74	+	c.mustGit(c.wt, "push", "-q", c.urlAuth(adminToken, "/"+c.repo+".git"), branch)
	75	+	return strings.TrimSpace(c.mustGit(c.wt, "rev-parse", "HEAD"))
	76	+	}
	77	+
	78	+	// registerRunner registers a runner and returns a fake that polls with it.
	79	+	func (c ciHarness) registerRunner(name string, labels ...string) fakeRunner {
	80	+	c.t.Helper()
	81	+	res, err := c.root.runner.RegisterRunner(c.ctx, connect.NewRequest(&pb.RegisterRunnerRequest{
	82	+	Name: name, Labels: labels, Version: "devel", Isolation: pb.RunnerIsolation_RUNNER_ISOLATION_VM, Arch: "x64", Os: "linux",
	83	+	}))
	84	+	require.NoError(c.t, err)
	85	+	return &fakeRunner{c: c, id: res.Msg.GetRunnerId(), api: c.api(res.Msg.GetRunnerToken())}
	86	+	}
	87	+
	88	+	// fakeRunner is the runner side of the protocol, as a test.
	89	+	type fakeRunner struct {
	90	+	c *ciHarness
	91	+	id string
	92	+	api *clients
	93	+	}
	94	+
	95	+	// fetch polls once for a job.
	96	+	func (r fakeRunner) fetch(pollID string) (pb.Job, error) {
	97	+	res, err := r.api.runner.FetchJob(r.c.ctx, connect.NewRequest(&pb.FetchJobRequest{PollId: pollID}))
	98	+	if err != nil {
	99	+	return nil, err
	100	+	}
	101	+	return res.Msg.GetJob(), nil
	102	+	}
	103	+
	104	+	// await polls until a job is given.
	105	+	func (r fakeRunner) await() pb.Job {
	106	+	r.c.t.Helper()
	107	+	deadline := time.Now().Add(15 * time.Second)
	108	+	for time.Now().Before(deadline) {
	109	+	j, err := r.fetch("")
	110	+	require.NoError(r.c.t, err)
	111	+	if j != nil {
	112	+	return j
	113	+	}
	114	+	}
	115	+	r.c.t.Fatal("no job was given to the runner")
	116	+	return nil
	117	+	}
	118	+
	119	+	// jobAPI is the client of a job: its token is the bearer secret.
	120	+	func (r fakeRunner) jobAPI(j pb.Job) *clients { return r.c.api(j.GetJobToken()) }
	121	+
	122	+	func now() *timestamppb.Timestamp { return timestamppb.Now() }
	123	+
	124	+	// runJob executes a job as a runner does: heartbeat, steps, log, completion.
	125	+	func (r fakeRunner) runJob(j pb.Job, conclusion pb.CiConclusion, outputs map[string]string, text ...string) {
	126	+	r.c.t.Helper()
	127	+	ctx, api := r.c.ctx, r.jobAPI(j)
	128	+	_, err := api.runner.Heartbeat(ctx, connect.NewRequest(&pb.HeartbeatRequest{JobId: j.GetJobId()}))
	129	+	require.NoError(r.c.t, err)
	130	+	start := now()
	131	+	_, err = api.runner.UpdateJob(ctx, connect.NewRequest(&pb.UpdateJobRequest{JobId: j.GetJobId(), Steps: []*pb.StepUpdate{
	132	+	{Number: 1, Name: "Set up job", State: pb.StepState_STEP_STATE_IN_PROGRESS, StartedAt: start},
	133	+	}}))
	134	+	require.NoError(r.c.t, err)
	135	+	var lines []*pb.LogLine
	136	+	for _, t := range text {
	137	+	lines = append(lines, &pb.LogLine{Time: now(), Text: t})
	138	+	}
	139	+	if len(lines) > 0 {
	140	+	_, err = api.runner.AppendLog(ctx, connect.NewRequest(&pb.AppendLogRequest{JobId: j.GetJobId(), Step: 2, Offset: 0, Lines: lines}))
	141	+	require.NoError(r.c.t, err)
	142	+	}
	143	+	step := pb.CiConclusion_CI_CONCLUSION_SUCCESS
	144	+	if conclusion != pb.CiConclusion_CI_CONCLUSION_SUCCESS {
	145	+	step = conclusion
	146	+	}
	147	+	res, err := api.runner.CompleteJob(ctx, connect.NewRequest(&pb.CompleteJobRequest{
	148	+	JobId: j.GetJobId(), Conclusion: conclusion, Outputs: outputs,
	149	+	Steps: []*pb.StepUpdate{
	150	+	{Number: 1, Name: "Set up job", State: pb.StepState_STEP_STATE_COMPLETED, Conclusion: pb.CiConclusion_CI_CONCLUSION_SUCCESS, StartedAt: start, CompletedAt: now()},
	151	+	{Number: 2, Name: "Run echo", State: pb.StepState_STEP_STATE_COMPLETED, Conclusion: step, StartedAt: start, CompletedAt: now()},
	152	+	},
	153	+	}))
	154	+	require.NoError(r.c.t, err)
	155	+	assert.True(r.c.t, res.Msg.GetApplied())
	156	+	}
	157	+
	158	+	// waitFor polls cond until it holds.
	159	+	func (c *ciHarness) waitFor(what string, cond func() bool) {
	160	+	c.t.Helper()
	161	+	require.Eventually(c.t, cond, 15time.Second, 20time.Millisecond, what)
	162	+	}
	163	+
	164	+	func (c ciHarness) runs() []pb.Run {
	165	+	c.t.Helper()
	166	+	res, err := c.root.run.ListRuns(c.ctx, connect.NewRequest(&pb.ListRunsRequest{Repo: c.repo}))
	167	+	require.NoError(c.t, err)
	168	+	return res.Msg.GetRuns()
	169	+	}
	170	+
	171	+	func (c ciHarness) checks(sha string) []pb.Check {
	172	+	c.t.Helper()
	173	+	res, err := c.root.run.ListChecks(c.ctx, connect.NewRequest(&pb.ListChecksRequest{Repo: c.repo, Sha: sha}))
	174	+	require.NoError(c.t, err)
	175	+	return res.Msg.GetChecks()
	176	+	}
	177	+
	178	+	func (c ciHarness) getRun(id int64) pb.GetRunResponse {
	179	+	c.t.Helper()
	180	+	res, err := c.root.run.GetRun(c.ctx, connect.NewRequest(&pb.GetRunRequest{Repo: c.repo, RunId: id}))
	181	+	require.NoError(c.t, err)
	182	+	return res.Msg
	183	+	}
	184	+
	185	+	func ciErrCode(err error) connect.Code {
	186	+	var ce *connect.Error
	187	+	if errors.As(err, &ce) {
	188	+	return ce.Code()
	189	+	}
	190	+	return connect.Code(0)
	191	+	}
	192	+
	193	+	// AC3, AC5: a push runs the workflow on a runner; its jobs follow their needs;
	194	+	// the checks appear on the commit; the log is readable.
	195	+	func TestCIPushRunsAWorkflowOnARunner(t *testing.T) {
	196	+	c := newCIHarness(t, nil)
	197	+	runner := c.registerRunner("alpha", "ubuntu-latest")
	198	+	sha := c.push("main", map[string]string{".github/workflows/ci.yml": ciWorkflow}, "")
	199	+
	200	+	c.waitFor("a run is created", func() bool { return len(c.runs()) == 1 })
	201	+	run := c.runs()[0]
	202	+	assert.Equal(t, "ci", run.GetWorkflowName())
	203	+	assert.Equal(t, "refs/heads/main", run.GetRef())
	204	+	assert.Equal(t, sha, run.GetSha())
	205	+	assert.Equal(t, "push", run.GetEvent())
	206	+	assert.EqualValues(t, 1, run.GetNumber())
	207	+	assert.Equal(t, "root", run.GetActor())
	208	+
	209	+	// Only the first job is ready: the second needs it.
	210	+	job := runner.await()
	211	+	assert.Equal(t, "build", job.GetJobKey())
	212	+	assert.Equal(t, run.GetId(), job.GetRunId())
	213	+	assert.Equal(t, sha, job.GetSha())
	214	+	assert.Equal(t, "refs/heads/main", job.GetRef())
	215	+	assert.Equal(t, "proj", job.GetRepo())
	216	+	assert.Equal(t, ciWorkflow, job.GetWorkflowYaml())
	217	+	assert.Equal(t, []string{"ubuntu-latest"}, job.GetLabels())
	218	+	assert.EqualValues(t, 360, job.GetTimeoutMinutes())
	219	+	assert.Equal(t, "{}", job.GetMatrix())
	220	+	assert.True(t, strings.HasPrefix(job.GetJobToken(), "ermitejob_"))
	221	+	assert.Contains(t, job.GetContexts(), `"repository":"owner/proj"`)
	222	+	assert.Contains(t, job.GetContexts(), `"needs":{}`)
	223	+
	224	+	// Assigned but not acknowledged: still queued for the owner.
	225	+	assert.Equal(t, pb.CiStatus_CI_STATUS_QUEUED, c.getRun(run.GetId()).GetJobs()[0].GetStatus())
	226	+
	227	+	runner.runJob(job, pb.CiConclusion_CI_CONCLUSION_SUCCESS, map[string]string{"artifact": "out.bin"}, "hello", "world")
	228	+
	229	+	next := runner.await()
	230	+	assert.Equal(t, "test", next.GetJobKey())
	231	+	assert.Contains(t, next.GetContexts(), `"needs":{"build":{"result":"success","outputs":{"artifact":"out.bin"}}}`)
	232	+	runner.runJob(next, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil, "tested")
	233	+
	234	+	c.waitFor("the run completes", func() bool {
	235	+	return c.runs()[0].GetStatus() == pb.CiStatus_CI_STATUS_COMPLETED
	236	+	})
	237	+	assert.Equal(t, pb.CiConclusion_CI_CONCLUSION_SUCCESS, c.runs()[0].GetConclusion())
	238	+
	239	+	checks := c.checks(sha)
	240	+	require.Len(t, checks, 2)
	241	+	assert.Equal(t, "build", checks[0].GetName())
	242	+	assert.Equal(t, "test", checks[1].GetName())
	243	+	for _, ck := range checks {
	244	+	assert.Equal(t, pb.CiStatus_CI_STATUS_COMPLETED, ck.GetStatus())
	245	+	assert.Equal(t, pb.CiConclusion_CI_CONCLUSION_SUCCESS, ck.GetConclusion())
	246	+	}
	247	+
	248	+	detail := c.getRun(run.GetId())
	249	+	require.Len(t, detail.GetJobs(), 2)
	250	+	steps := detail.GetJobs()[0].GetSteps()
	251	+	require.Len(t, steps, 2)
	252	+	assert.EqualValues(t, 0, steps[1].GetLogStart(), "step 2 has the lines")
	253	+	assert.EqualValues(t, 2, steps[1].GetLogEnd())
	254	+
	255	+	// The log is streamed back with its steps.
	256	+	stream, err := c.root.run.GetJobLog(c.ctx, connect.NewRequest(&pb.GetJobLogRequest{Repo: "proj", JobId: job.GetJobId()}))
	257	+	require.NoError(t, err)
	258	+	var got []string
	259	+	for stream.Receive() {
	260	+	for _, l := range stream.Msg().GetLines() {
	261	+	got = append(got, l.GetText())
	262	+	assert.EqualValues(t, 2, l.GetStep())
	263	+	}
	264	+	}
	265	+	require.NoError(t, stream.Err())
	266	+	assert.Equal(t, []string{"hello", "world"}, got)
	267	+	}
	268	+
	269	+	// AC5: merge refused while checks are pending or failing, allowed when they
	270	+	// pass, and when failing_checks_block is off.
	271	+	func TestCIMergeIsBlockedByChecks(t *testing.T) {
	272	+	c := newCIHarness(t, nil)
	273	+	runner := c.registerRunner("alpha", "ubuntu-latest")
	274	+	c.push("main", map[string]string{"README": "x\n"}, "")
	275	+
	276	+	feat := c.push("feat", map[string]string{".github/workflows/ci.yml": strings.ReplaceAll(ciWorkflow, " test:\n needs: build\n runs-on: ubuntu-latest\n steps:\n - run: echo test\n", "")}, "")
	277	+	mr, err := c.root.mr.CreateMergeRequest(c.ctx, connect.NewRequest(&pb.CreateMergeRequestRequest{Repo: "proj", Title: "Add CI", SourceBranch: "feat"}))
	278	+	require.NoError(t, err)
	279	+	merge := func() error {
	280	+	_, err := c.root.mr.MergeMergeRequest(c.ctx, connect.NewRequest(&pb.MergeMergeRequestRequest{Repo: "proj", Number: mr.Msg.GetMergeRequest().GetNumber(), HeadId: feat}))
	281	+	return err
	282	+	}
	283	+
	284	+	c.waitFor("the feature run is queued", func() bool { return len(c.checks(feat)) == 1 })
	285	+	err = merge()
	286	+	require.Error(t, err)
	287	+	assert.Equal(t, connect.CodeFailedPrecondition, ciErrCode(err), "a pending check blocks the merge")
	288	+	assert.Contains(t, err.Error(), "have not finished")
	289	+
	290	+	job := runner.await()
	291	+	runner.runJob(job, pb.CiConclusion_CI_CONCLUSION_FAILURE, nil, "boom")
	292	+	c.waitFor("the check fails", func() bool {
	293	+	cs := c.checks(feat)
	294	+	return len(cs) == 1 && cs[0].GetStatus() == pb.CiStatus_CI_STATUS_COMPLETED
	295	+	})
	296	+	err = merge()
	297	+	require.Error(t, err)
	298	+	assert.Equal(t, connect.CodeFailedPrecondition, ciErrCode(err))
	299	+	assert.Contains(t, err.Error(), "failing checks block the merge")
	300	+
	301	+	// Off: the merge goes through.
	302	+	_, err = c.root.repo.UpdateRepo(c.ctx, connect.NewRequest(&pb.UpdateRepoRequest{
	303	+	Repo: "proj",
	304	+	Settings: &pb.RepoSettings{FailingChecksBlock: false},
	305	+	UpdateMask: &fieldmaskpb.FieldMask{Paths: []string{"settings.failing_checks_block"}},
	306	+	}))
	307	+	require.NoError(t, err)
	308	+	require.NoError(t, merge())
	309	+	}
	310	+
	311	+	// AC2: a job token works for its job only, and not once the job has ended.
	312	+	func TestCIJobTokensAreScopedToTheirJob(t *testing.T) {
	313	+	c := newCIHarness(t, nil)
	314	+	runner := c.registerRunner("alpha", "ubuntu-latest")
	315	+	c.push("main", map[string]string{".github/workflows/a.yml": ciWorkflow, ".github/workflows/b.yml": ciWorkflow}, "")
	316	+	c.waitFor("two runs", func() bool { return len(c.runs()) == 2 })
	317	+	j1, j2 := runner.await(), runner.await()
	318	+	require.NotEqual(t, j1.GetJobId(), j2.GetJobId())
	319	+
	320	+	// Another job's id with this job's token: permission_denied.
	321	+	_, err := runner.jobAPI(j1).runner.Heartbeat(c.ctx, connect.NewRequest(&pb.HeartbeatRequest{JobId: j2.GetJobId()}))
	322	+	assert.Equal(t, connect.CodePermissionDenied, ciErrCode(err))
	323	+	// A runner token is not a job token, and a stored token is neither.
	324	+	_, err = runner.api.runner.Heartbeat(c.ctx, connect.NewRequest(&pb.HeartbeatRequest{JobId: j1.GetJobId()}))
	325	+	assert.Equal(t, connect.CodeUnauthenticated, ciErrCode(err))
	326	+	_, err = c.root.runner.Heartbeat(c.ctx, connect.NewRequest(&pb.HeartbeatRequest{JobId: j1.GetJobId()}))
	327	+	assert.Equal(t, connect.CodeUnauthenticated, ciErrCode(err))
	328	+	// A job token does not poll.
	329	+	_, err = runner.jobAPI(j1).runner.FetchJob(c.ctx, connect.NewRequest(&pb.FetchJobRequest{}))
	330	+	assert.Equal(t, connect.CodeUnauthenticated, ciErrCode(err))
	331	+
	332	+	runner.runJob(j1, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil, "a")
	333	+	// After the end: failed_precondition with the JobEnded detail; CompleteJob
	334	+	// still answers with the stored result.
	335	+	_, err = runner.jobAPI(j1).runner.AppendLog(c.ctx, connect.NewRequest(&pb.AppendLogRequest{
	336	+	JobId: j1.GetJobId(), Step: 2, Offset: 1, Lines: []*pb.LogLine{{Time: now(), Text: "late"}},
	337	+	}))
	338	+	require.Error(t, err)
	339	+	assert.Equal(t, connect.CodeFailedPrecondition, ciErrCode(err))
	340	+	var ce *connect.Error
	341	+	require.True(t, errors.As(err, &ce))
	342	+	require.Len(t, ce.Details(), 1)
	343	+	msg, derr := ce.Details()[0].Value()
	344	+	require.NoError(t, derr)
	345	+	assert.Equal(t, pb.CiConclusion_CI_CONCLUSION_SUCCESS, msg.(*pb.JobEnded).GetConclusion())
	346	+
	347	+	res, err := runner.jobAPI(j1).runner.CompleteJob(c.ctx, connect.NewRequest(&pb.CompleteJobRequest{
	348	+	JobId: j1.GetJobId(), Conclusion: pb.CiConclusion_CI_CONCLUSION_FAILURE,
	349	+	}))
	350	+	require.NoError(t, err)
	351	+	assert.False(t, res.Msg.GetApplied(), "a second CompleteJob changes nothing")
	352	+	assert.Equal(t, pb.CiConclusion_CI_CONCLUSION_SUCCESS, res.Msg.GetConclusion(), "and answers the stored conclusion")
	353	+	}
	354	+
	355	+	// AC2: a job is given to exactly one runner, however many poll at once.
	356	+	func TestCIConcurrentFetchJobAssignsOnce(t *testing.T) {
	357	+	c := newCIHarness(t, func(cfg usecase.CIConfig) { cfg.PollTimeout = 300 time.Millisecond })
	358	+	var runners []*fakeRunner
	359	+	for i := range 8 {
	360	+	runners = append(runners, c.registerRunner(fmt.Sprintf("r%d", i), "ubuntu-latest"))
	361	+	}
	362	+	var wf strings.Builder
	363	+	wf.WriteString("on: push\njobs:\n")
	364	+	for i := range 3 {
	365	+	fmt.Fprintf(&wf, " j%d:\n runs-on: ubuntu-latest\n steps:\n - run: echo %d\n", i, i)
	366	+	}
	367	+	c.push("main", map[string]string{".github/workflows/many.yml": wf.String()}, "")
	368	+	c.waitFor("the run exists", func() bool { return len(c.runs()) == 1 })
	369	+
	370	+	var mu sync.Mutex
	371	+	got := map[string]int{}
	372	+	var wg sync.WaitGroup
	373	+	for _, r := range runners {
	374	+	wg.Add(1)
	375	+	go func() {
	376	+	defer wg.Done()
	377	+	for range 20 {
	378	+	j, err := r.fetch("")
	379	+	if err != nil {
	380	+	return
	381	+	}
	382	+	if j != nil {
	383	+	mu.Lock()
	384	+	got[j.GetJobId()]++
	385	+	mu.Unlock()
	386	+	}
	387	+	}
	388	+	}()
	389	+	}
	390	+	wg.Wait()
	391	+	assert.Len(t, got, 3, "every job was given")
	392	+	for id, n := range got {
	393	+	assert.Equal(t, 1, n, "job %s was given %d times", id, n)
	394	+	}
	395	+	}
	396	+
	397	+	// AC2: the first heartbeat acknowledges; without it the assignment is
	398	+	// cancelled and the job is queued again, three times in a row at most.
	399	+	func TestCIUnacknowledgedAssignmentGoesBackToTheQueue(t *testing.T) {
	400	+	c := newCIHarness(t, func(cfg *usecase.CIConfig) {
	401	+	cfg.AckTimeout, cfg.Tick, cfg.PollTimeout = 150time.Millisecond, 20time.Millisecond, 100*time.Millisecond
	402	+	})
	403	+	runner := c.registerRunner("alpha", "ubuntu-latest")
	404	+	c.push("main", map[string]string{".github/workflows/ci.yml": strings.Split(ciWorkflow, " test:")[0]}, "")
	405	+
	406	+	first := runner.await()
	407	+	// No heartbeat: the token is revoked and the job comes back with a new one.
	408	+	second := runner.await()
	409	+	assert.Equal(t, first.GetJobId(), second.GetJobId())
	410	+	assert.NotEqual(t, first.GetJobToken(), second.GetJobToken())
	411	+	_, err := runner.jobAPI(first).runner.Heartbeat(c.ctx, connect.NewRequest(&pb.HeartbeatRequest{JobId: first.GetJobId()}))
	412	+	assert.Equal(t, connect.CodeUnauthenticated, ciErrCode(err), "the replaced token is rejected")
	413	+
	414	+	// The third assignment is the last: ignoring it fails the job.
	415	+	runner.await()
	416	+	c.waitFor("the job fails", func() bool {
	417	+	return c.runs()[0].GetStatus() == pb.CiStatus_CI_STATUS_COMPLETED
	418	+	})
	419	+	detail := c.getRun(c.runs()[0].GetId())
	420	+	assert.Equal(t, pb.CiConclusion_CI_CONCLUSION_FAILURE, detail.GetJobs()[0].GetConclusion())
	421	+	assert.Contains(t, detail.GetJobs()[0].GetError(), "acknowledged")
	422	+	}
	423	+
	424	+	// A retried poll with the same poll_id returns the same assignment.
	425	+	func TestCIRetriedPollReturnsTheSameAssignment(t *testing.T) {
	426	+	c := newCIHarness(t, func(cfg usecase.CIConfig) { cfg.PollTimeout = 200 time.Millisecond })
	427	+	runner := c.registerRunner("alpha", "ubuntu-latest")
	428	+	c.push("main", map[string]string{".github/workflows/ci.yml": strings.Split(ciWorkflow, " test:")[0]}, "")
	429	+	c.waitFor("a run", func() bool { return len(c.runs()) == 1 })
	430	+
	431	+	var first *pb.Job
	432	+	c.waitFor("a job", func() bool {
	433	+	j, err := runner.fetch("poll-1")
	434	+	require.NoError(t, err)
	435	+	first = j
	436	+	return j != nil
	437	+	})
	438	+	again, err := runner.fetch("poll-1")
	439	+	require.NoError(t, err)
	440	+	require.NotNil(t, again)
	441	+	assert.Equal(t, first.GetJobId(), again.GetJobId())
	442	+	assert.Equal(t, first.GetJobToken(), again.GetJobToken())
	443	+	}

M internal/app/e2e_helpers_test.go +11 -1


@@ -53,6 +53,12 @@ type harnessOpts struct {
53	53		// newHarness starts a forge. The data directory is short: the hook socket
54	54		// lives in it and Unix socket paths are limited to about 100 bytes.
55	55		func newHarness(t testing.T, o harnessOpts) harness {
	56	+	t.Helper()
	57	+	return newHarnessWith(t, o, nil)
	58	+	}
	59	+
	60	+	// newHarnessWith is newHarness with a change of the wiring options.
	61	+	func newHarnessWith(t testing.T, o harnessOpts, tweak func(options)) *harness {
56	62		t.Helper()
57	63		gitBin := requireGit(t)
58	64		data, err := os.MkdirTemp("", "ermite")

@@ -85,7 +91,11 @@ func newHarness(t testing.T, o harnessOpts) harness {
85	91		if hookExe == "" {
86	92		hookExe = binary(t)
87	93		}
88		-	f, err := newForge(ctx, cfg, options{hookExe: hookExe, maintenanceDelay: o.maintenance, maxBodyBytes: o.maxBody, eventBuffer: o.eventBuffer, revalidateEvery: o.revalidateEvery})
	94	+	opts := options{hookExe: hookExe, maintenanceDelay: o.maintenance, maxBodyBytes: o.maxBody, eventBuffer: o.eventBuffer, revalidateEvery: o.revalidateEvery}
	95	+	if tweak != nil {
	96	+	tweak(&opts)
	97	+	}
	98	+	f, err := newForge(ctx, cfg, opts)
89	99		require.NoError(t, err)
90	100
91	101		h := &harness{t: t, forge: f, data: data, logs: logs, ctx: ctx, cancel: cancel, hooksDone: make(chan error, 1)}

 // newHarness starts a forge. The data directory is short: the hook socket
 // lives in it and Unix socket paths are limited to about 100 bytes.
 func newHarness(t *testing.T, o harnessOpts) *harness {
+	t.Helper()
+	return newHarnessWith(t, o, nil)
+}
+// newHarnessWith is newHarness with a change of the wiring options.
+func newHarnessWith(t *testing.T, o harnessOpts, tweak func(*options)) *harness {
 	t.Helper()
 	gitBin := requireGit(t)
 	data, err := os.MkdirTemp("", "ermite")
 	if hookExe == "" {
 		hookExe = binary(t)
 	}
-	f, err := newForge(ctx, cfg, options{hookExe: hookExe, maintenanceDelay: o.maintenance, maxBodyBytes: o.maxBody, eventBuffer: o.eventBuffer, revalidateEvery: o.revalidateEvery})
+	opts := options{hookExe: hookExe, maintenanceDelay: o.maintenance, maxBodyBytes: o.maxBody, eventBuffer: o.eventBuffer, revalidateEvery: o.revalidateEvery}
+	if tweak != nil {
+		tweak(&opts)
+	}
+	f, err := newForge(ctx, cfg, opts)
 	require.NoError(t, err)
 	h := &harness{t: t, forge: f, data: data, logs: logs, ctx: ctx, cancel: cancel, hooksDone: make(chan error, 1)}

M internal/app/forge.go +30 -2


@@ -15,6 +15,7 @@ import (
15	15
16	16		"github.com/bnema/zerowrap"
17	17
	18	+	"bnema.dev/ermite/internal/adapters/cilogs"
18	19		"bnema.dev/ermite/internal/adapters/ermitegit"
19	20		"bnema.dev/ermite/internal/adapters/eventbus"
20	21		"bnema.dev/ermite/internal/adapters/gitcli"

@@ -48,6 +49,9 @@ type options struct {
48	49		eventBuffer int
49	50		// revalidateEvery overrides usecase.DefaultRevalidateEvery when not zero.
50	51		revalidateEvery time.Duration
	52	+	// ci overrides the settings of the CI engine (tests shorten its
	53	+	// durations).
	54	+	ci func(*usecase.CIConfig)
51	55		}
52	56
53	57		// forge is the wired server: every adapter and use case, ready to serve.

@@ -74,8 +78,11 @@ type forge struct {
74	78		SigningKey *usecase.SigningKey
75	79		Sessions *usecase.Sessions
76	80		EventLog *usecase.Events
77		-	API *rpc.API
78		-	Bus ports.EventBus
	81	+	// CI is the CI engine: workflow trigger, planner, runner protocol and the
	82	+	// owner side. It starts with the server and stops with it.
	83	+	CI *usecase.CI
	84	+	API *rpc.API
	85	+	Bus ports.EventBus
79	86		}
80	87
81	88		// newSigningSeed makes the seed of a new ed25519 signing key.

@@ -182,6 +189,21 @@ func newForge(ctx context.Context, cfg config.Config, opts options) (_ *forge, e
182	189		}
183	190		log.Info().Str("reader", cfg.GitReader).Msg("git reader selected")
184	191		f.Browse = usecase.NewBrowse(store, reader)
	192	+
	193	+	// CI (ADR 0025): the engine reads workflows through the reader, and plans
	194	+	// them in the background; a push only queues its refs.
	195	+	ciCfg := usecase.CIConfig{
	196	+	Store: store.CI(), Logs: cilogs.New(cilogs.Dir(cfg.DataDir)), Meta: store, Repos: git, Reader: reader,
	197	+	Git: git, Diff: git, Bus: f.Bus, Clock: clock, Auth: f.Auth,
	198	+	Owner: cfg.Owner, PublicURL: cfg.PublicURL, Retention: cfg.CIRetention,
	199	+	}
	200	+	if opts.ci != nil {
	201	+	opts.ci(&ciCfg)
	202	+	}
	203	+	f.CI = usecase.NewCI(ctx, ciCfg)
	204	+	f.Pushes.WithWorkflowTrigger(f.CI)
	205	+	f.Repos.WithCIRunCleaner(f.CI)
	206	+	f.CI.Start()
185	207		f.EventLog = usecase.NewEvents(ctx, f.Repos, f.Bus, clock, usecase.EventsConfig{
186	208		Size: opts.eventBuffer, Auth: f.Auth, RevalidateEvery: opts.revalidateEvery,
187	209		})

@@ -200,6 +222,7 @@ func newForge(ctx context.Context, cfg config.Config, opts options) (_ *forge, e
200	222		f.MergeRequests = usecase.NewMergeRequests(usecase.MergeRequestsConfig{
201	223		Meta: store, Store: store.MergeRequests(), Repos: git, Git: git, Signer: signer,
202	224		Committer: committer, Bus: f.Bus, Clock: clock, Maint: f.maint,
	225	+	CI: f.CI, Checks: f.CI,
203	226		})
204	227		log.Info().Str("fingerprint", signer.PublicKey().Fingerprint).Msg("commit signing key ready")
205	228

@@ -220,6 +243,8 @@ func newForge(ctx context.Context, cfg config.Config, opts options) (_ *forge, e
220	243		MergeRequests: f.MergeRequests,
221	244		SigningKey: f.SigningKey,
222	245		Sessions: f.Sessions,
	246	+	Runner: f.CI,
	247	+	Runs: f.CI,
223	248		Version: serverVersion(),
224	249		TrustedProxies: cfg.TrustedProxies,
225	250		Limiter: limiter,

@@ -297,6 +322,9 @@ func (f *forge) Close(ctx context.Context) {
297	322		if f.hooks != nil {
298	323		f.hooks.Close(ctx)
299	324		}
	325	+	if f.CI != nil {
	326	+	f.CI.Close(ctx)
	327	+	}
300	328		if f.EventLog != nil {
301	329		f.EventLog.Close(ctx)
302	330		}

 	"github.com/bnema/zerowrap"
+	"bnema.dev/ermite/internal/adapters/cilogs"
 	"bnema.dev/ermite/internal/adapters/ermitegit"
 	"bnema.dev/ermite/internal/adapters/eventbus"
 	"bnema.dev/ermite/internal/adapters/gitcli"
 	eventBuffer int
 	// revalidateEvery overrides usecase.DefaultRevalidateEvery when not zero.
 	revalidateEvery time.Duration
+	// ci overrides the settings of the CI engine (tests shorten its
+	// durations).
+	ci func(*usecase.CIConfig)
 }
 // forge is the wired server: every adapter and use case, ready to serve.
 	SigningKey    *usecase.SigningKey
 	Sessions      *usecase.Sessions
 	EventLog      *usecase.Events
-	API           *rpc.API
+	// CI is the CI engine: workflow trigger, planner, runner protocol and the
-	Bus           ports.EventBus
+	// owner side. It starts with the server and stops with it.
+	CI  *usecase.CI
+	API *rpc.API
+	Bus ports.EventBus
 }
 // newSigningSeed makes the seed of a new ed25519 signing key.
 	}
 	log.Info().Str("reader", cfg.GitReader).Msg("git reader selected")
 	f.Browse = usecase.NewBrowse(store, reader)
+	// CI (ADR 0025): the engine reads workflows through the reader, and plans
+	// them in the background; a push only queues its refs.
+	ciCfg := usecase.CIConfig{
+		Store: store.CI(), Logs: cilogs.New(cilogs.Dir(cfg.DataDir)), Meta: store, Repos: git, Reader: reader,
+		Git: git, Diff: git, Bus: f.Bus, Clock: clock, Auth: f.Auth,
+		Owner: cfg.Owner, PublicURL: cfg.PublicURL, Retention: cfg.CIRetention,
+	}
+	if opts.ci != nil {
+		opts.ci(&ciCfg)
+	}
+	f.CI = usecase.NewCI(ctx, ciCfg)
+	f.Pushes.WithWorkflowTrigger(f.CI)
+	f.Repos.WithCIRunCleaner(f.CI)
+	f.CI.Start()
 	f.EventLog = usecase.NewEvents(ctx, f.Repos, f.Bus, clock, usecase.EventsConfig{
 		Size: opts.eventBuffer, Auth: f.Auth, RevalidateEvery: opts.revalidateEvery,
 	})
 	f.MergeRequests = usecase.NewMergeRequests(usecase.MergeRequestsConfig{
 		Meta: store, Store: store.MergeRequests(), Repos: git, Git: git, Signer: signer,
 		Committer: committer, Bus: f.Bus, Clock: clock, Maint: f.maint,
+		CI: f.CI, Checks: f.CI,
 	})
 	log.Info().Str("fingerprint", signer.PublicKey().Fingerprint).Msg("commit signing key ready")
 		MergeRequests:  f.MergeRequests,
 		SigningKey:     f.SigningKey,
 		Sessions:       f.Sessions,
+		Runner:         f.CI,
+		Runs:           f.CI,
 		Version:        serverVersion(),
 		TrustedProxies: cfg.TrustedProxies,
 		Limiter:        limiter,
 	if f.hooks != nil {
 		f.hooks.Close(ctx)
 	}
+	if f.CI != nil {
+		f.CI.Close(ctx)
+	}
 	if f.EventLog != nil {
 		f.EventLog.Close(ctx)
 	}

M internal/ports/ci.go +6 -0


@@ -300,6 +300,12 @@ func (v JobView) StepRanges() []domain.StepRange { return domain.StepRanges(v.St
300	300		type ListRunsParams struct {
301	301		PageSize int
302	302		PageToken string
	303	+	// Workflow restricts the result to a workflow file: its path or its file
	304	+	// name. Branch restricts it to the runs of a branch, by short name. Status
	305	+	// restricts it to the runs of that status. Empty means no restriction.
	306	+	Workflow string
	307	+	Branch string
	308	+	Status domain.CIStatus
303	309		}
304	310
305	311		// RunPage is a page of runs, newest first.


@@ -300,6 +300,12 @@ func (v JobView) StepRanges() []domain.StepRange { return domain.StepRanges(v.St
300	type ListRunsParams struct {	300		type ListRunsParams struct {
301	PageSize int	301		PageSize int
302	PageToken string	302		PageToken string
		303	+	// Workflow restricts the result to a workflow file: its path or its file
		304	+	// name. Branch restricts it to the runs of a branch, by short name. Status
		305	+	// restricts it to the runs of that status. Empty means no restriction.
		306	+	Workflow string
		307	+	Branch string
		308	+	Status domain.CIStatus
303	}	309		}
304		310
305	// RunPage is a page of runs, newest first.	311		// RunPage is a page of runs, newest first.

M internal/ports/dto.go +2 -2


@@ -137,8 +137,8 @@ type WatchEvent struct {
137	137		// Event is nil then.
138	138		Gap bool
139	139		// Event is a domain.RepoCreated, RepoUpdated, RepoDeleted, PushEvent,
140		-	// MergeRequestOpened, MergeRequestMerged, MergeRequestClosed or
141		-	// CommentAdded.
	140	+	// MergeRequestOpened, MergeRequestMerged, MergeRequestClosed,
	141	+	// CommentAdded, RunUpdated or JobUpdated.
142	142		Event domain.Event
143	143		// Repo is the state of the repository at delivery, for RepoCreated and
144	144		// RepoUpdated.


@@ -137,8 +137,8 @@ type WatchEvent struct {
137		// Event is nil then.	137		// Event is nil then.
138		Gap bool	138		Gap bool
139		// Event is a domain.RepoCreated, RepoUpdated, RepoDeleted, PushEvent,	139		// Event is a domain.RepoCreated, RepoUpdated, RepoDeleted, PushEvent,
140	-	// MergeRequestOpened, MergeRequestMerged, MergeRequestClosed or	140	+	// MergeRequestOpened, MergeRequestMerged, MergeRequestClosed,
141	-	// CommentAdded.	141	+	// CommentAdded, RunUpdated or JobUpdated.
142		Event domain.Event	142		Event domain.Event
143		// Repo is the state of the repository at delivery, for RepoCreated and	143		// Repo is the state of the repository at delivery, for RepoCreated and
144		// RepoUpdated.	144		// RepoUpdated.

M internal/usecase/ci_runs.go +39 -7


@@ -7,6 +7,7 @@ import (
7	7		"encoding/base64"
8	8		"errors"
9	9		"fmt"
	10	+	"path"
10	11		"slices"
11	12		"strconv"
12	13		"strings"

@@ -39,18 +40,49 @@ func (c *CI) ListRuns(ctx context.Context, p domain.Principal, name domain.RepoN
39	40		}
40	41		before = id
41	42		}
42		-	runs, err := c.cfg.Store.ListRuns(ctx, name, before, n+1)
43		-	if err != nil {
44		-	return ports.RunPage{}, err
	43	+	match := runFilter(params)
	44	+	// The store returns runs by id; the filters are applied while paging, so a
	45	+	// page of a filtered listing is full unless the runs ran out.
	46	+	var kept []domain.CIRun
	47	+	cursor := before
	48	+	for len(kept) <= n {
	49	+	batch, err := c.cfg.Store.ListRuns(ctx, name, cursor, n+1)
	50	+	if err != nil {
	51	+	return ports.RunPage{}, err
	52	+	}
	53	+	for _, r := range batch {
	54	+	if match(r) {
	55	+	kept = append(kept, r)
	56	+	}
	57	+	}
	58	+	if len(batch) < n+1 {
	59	+	break
	60	+	}
	61	+	cursor = batch[len(batch)-1].ID
45	62		}
46		-	page := ports.RunPage{Runs: runs}
47		-	if len(runs) > n {
48		-	page.Runs = runs[:n]
49		-	page.NextPageToken = base64.RawURLEncoding.EncodeToString([]byte(runTokenPrefix + strconv.FormatInt(runs[n-1].ID, 10)))
	63	+	page := ports.RunPage{Runs: kept}
	64	+	if len(kept) > n {
	65	+	page.Runs = kept[:n]
	66	+	page.NextPageToken = base64.RawURLEncoding.EncodeToString([]byte(runTokenPrefix + strconv.FormatInt(kept[n-1].ID, 10)))
50	67		}
51	68		return page, nil
52	69		}
53	70
	71	+	// runFilter returns the test of the filters of a listing.
	72	+	func runFilter(p ports.ListRunsParams) func(domain.CIRun) bool {
	73	+	return func(r domain.CIRun) bool {
	74	+	switch {
	75	+	case p.Status != "" && r.Status != p.Status:
	76	+	return false
	77	+	case p.Branch != "" && r.Ref != "refs/heads/"+p.Branch:
	78	+	return false
	79	+	case p.Workflow != "" && r.WorkflowPath != p.Workflow && path.Base(r.WorkflowPath) != p.Workflow:
	80	+	return false
	81	+	}
	82	+	return true
	83	+	}
	84	+	}
	85	+
54	86		// runOf finds a run of the repository; a run of another repository is not
55	87		// found.
56	88		func (c *CI) runOf(ctx context.Context, name domain.RepoName, runID int64) (domain.CIRun, error) {


@@ -7,6 +7,7 @@ import (
7		"encoding/base64"	7		"encoding/base64"
8		"errors"	8		"errors"
9		"fmt"	9		"fmt"
			10	+	"path"
10		"slices"	11		"slices"
11		"strconv"	12		"strconv"
12		"strings"	13		"strings"

@@ -39,18 +40,49 @@ func (c *CI) ListRuns(ctx context.Context, p domain.Principal, name domain.RepoN
39		}	40		}
40		before = id	41		before = id
41		}	42		}
42	-	runs, err := c.cfg.Store.ListRuns(ctx, name, before, n+1)	43	+	match := runFilter(params)
43	-	if err != nil {	44	+	// The store returns runs by id; the filters are applied while paging, so a
44	-	return ports.RunPage{}, err	45	+	// page of a filtered listing is full unless the runs ran out.
			46	+	var kept []domain.CIRun
			47	+	cursor := before
			48	+	for len(kept) <= n {
			49	+	batch, err := c.cfg.Store.ListRuns(ctx, name, cursor, n+1)
			50	+	if err != nil {
			51	+	return ports.RunPage{}, err
			52	+	}
			53	+	for _, r := range batch {
			54	+	if match(r) {
			55	+	kept = append(kept, r)
			56	+	}
			57	+	}
			58	+	if len(batch) < n+1 {
			59	+	break
			60	+	}
			61	+	cursor = batch[len(batch)-1].ID
45		}	62		}
46	-	page := ports.RunPage{Runs: runs}	63	+	page := ports.RunPage{Runs: kept}
47	-	if len(runs) > n {	64	+	if len(kept) > n {
48	-	page.Runs = runs[:n]	65	+	page.Runs = kept[:n]
49	-	page.NextPageToken = base64.RawURLEncoding.EncodeToString([]byte(runTokenPrefix + strconv.FormatInt(runs[n-1].ID, 10)))	66	+	page.NextPageToken = base64.RawURLEncoding.EncodeToString([]byte(runTokenPrefix + strconv.FormatInt(kept[n-1].ID, 10)))
50		}	67		}
51		return page, nil	68		return page, nil
52		}	69		}
53			70
			71	+	// runFilter returns the test of the filters of a listing.
			72	+	func runFilter(p ports.ListRunsParams) func(domain.CIRun) bool {
			73	+	return func(r domain.CIRun) bool {
			74	+	switch {
			75	+	case p.Status != "" && r.Status != p.Status:
			76	+	return false
			77	+	case p.Branch != "" && r.Ref != "refs/heads/"+p.Branch:
			78	+	return false
			79	+	case p.Workflow != "" && r.WorkflowPath != p.Workflow && path.Base(r.WorkflowPath) != p.Workflow:
			80	+	return false
			81	+	}
			82	+	return true
			83	+	}
			84	+	}
			85	+
54		// runOf finds a run of the repository; a run of another repository is not	86		// runOf finds a run of the repository; a run of another repository is not
55		// found.	87		// found.
56		func (c *CI) runOf(ctx context.Context, name domain.RepoName, runID int64) (domain.CIRun, error) {	88		func (c *CI) runOf(ctx context.Context, name domain.RepoName, runID int64) (domain.CIRun, error) {

M internal/usecase/events.go +15 -0


@@ -176,6 +176,10 @@ func (e *Events) snapshot(ctx context.Context, ev domain.Event) bufferedEvent {
176	176		return repoSnapshot(ev.Visibility, ev.CreatedAt)
177	177		case domain.CommentAdded:
178	178		return repoSnapshot(ev.Visibility, ev.CreatedAt)
	179	+	case domain.RunUpdated:
	180	+	return repoSnapshot(ev.Visibility, ev.CreatedAt)
	181	+	case domain.JobUpdated:
	182	+	return repoSnapshot(ev.Visibility, ev.CreatedAt)
179	183		}
180	184		return bufferedEvent{known: true}
181	185		}

@@ -479,6 +483,17 @@ func (e Events) deliver(ctx context.Context, w watch, be bufferedEvent, send f
479	483		return e.deliverRepo(ctx, w, ev.MergeRequest.Repo, false, be, out, send)
480	484		case domain.CommentAdded:
481	485		return e.deliverRepo(ctx, w, ev.Repo, false, be, out, send)
	486	+	case domain.RunUpdated:
	487	+	// CI events go on the topic of their repository only, never on home.
	488	+	if !w.repos[ev.Run.Repo] {
	489	+	return nil
	490	+	}
	491	+	return e.deliverRepo(ctx, w, ev.Run.Repo, false, be, out, send)
	492	+	case domain.JobUpdated:
	493	+	if !w.repos[ev.Job.Repo] {
	494	+	return nil
	495	+	}
	496	+	return e.deliverRepo(ctx, w, ev.Job.Repo, false, be, out, send)
482	497		}
483	498		return nil // an event this version does not stream
484	499		}


@@ -176,6 +176,10 @@ func (e *Events) snapshot(ctx context.Context, ev domain.Event) bufferedEvent {
176	return repoSnapshot(ev.Visibility, ev.CreatedAt)	176		return repoSnapshot(ev.Visibility, ev.CreatedAt)
177	case domain.CommentAdded:	177		case domain.CommentAdded:
178	return repoSnapshot(ev.Visibility, ev.CreatedAt)	178		return repoSnapshot(ev.Visibility, ev.CreatedAt)
		179	+	case domain.RunUpdated:
		180	+	return repoSnapshot(ev.Visibility, ev.CreatedAt)
		181	+	case domain.JobUpdated:
		182	+	return repoSnapshot(ev.Visibility, ev.CreatedAt)
179	}	183		}
180	return bufferedEvent{known: true}	184		return bufferedEvent{known: true}
181	}	185		}

@@ -479,6 +483,17 @@ func (e Events) deliver(ctx context.Context, w watch, be bufferedEvent, send f
479	return e.deliverRepo(ctx, w, ev.MergeRequest.Repo, false, be, out, send)	483		return e.deliverRepo(ctx, w, ev.MergeRequest.Repo, false, be, out, send)
480	case domain.CommentAdded:	484		case domain.CommentAdded:
481	return e.deliverRepo(ctx, w, ev.Repo, false, be, out, send)	485		return e.deliverRepo(ctx, w, ev.Repo, false, be, out, send)
		486	+	case domain.RunUpdated:
		487	+	// CI events go on the topic of their repository only, never on home.
		488	+	if !w.repos[ev.Run.Repo] {
		489	+	return nil
		490	+	}
		491	+	return e.deliverRepo(ctx, w, ev.Run.Repo, false, be, out, send)
		492	+	case domain.JobUpdated:
		493	+	if !w.repos[ev.Job.Repo] {
		494	+	return nil
		495	+	}
		496	+	return e.deliverRepo(ctx, w, ev.Job.Repo, false, be, out, send)
482	}	497		}
483	return nil // an event this version does not stream	498		return nil // an event this version does not stream
484	}	499		}

M internal/usecase/repos.go +8 -0


@@ -21,6 +21,7 @@ type Repos struct {
21	21		git ports.GitRepos
22	22		bus ports.EventBus
23	23		clock ports.Clock
	24	+	ci ports.CIRunCleaner // nil: no CI logs to remove
24	25		}
25	26
26	27		var _ ports.RepoService = (*Repos)(nil)

@@ -31,6 +32,13 @@ func NewRepos(meta ports.RepoMetaStore, git ports.GitRepos, bus ports.EventBus,
31	32		return &Repos{meta: meta, git: git, bus: bus, clock: clock}
32	33		}
33	34
	35	+	// WithCIRunCleaner makes Delete remove the CI logs of the repository. Call it
	36	+	// before the first request; it returns r.
	37	+	func (r Repos) WithCIRunCleaner(c ports.CIRunCleaner) Repos {
	38	+	r.ci = c
	39	+	return r
	40	+	}
	41	+
34	42		// Create makes a repository and publishes RepoCreated: it stages a bare repository, records the
35	43		// metadata (which fails with domain.ErrAlreadyExists if the name is taken,
36	44		// also when a concurrent Create wins) and publishes the directory. A failure


@@ -21,6 +21,7 @@ type Repos struct {
21	git ports.GitRepos	21		git ports.GitRepos
22	bus ports.EventBus	22		bus ports.EventBus
23	clock ports.Clock	23		clock ports.Clock
		24	+	ci ports.CIRunCleaner // nil: no CI logs to remove
24	}	25		}
25		26
26	var _ ports.RepoService = (*Repos)(nil)	27		var _ ports.RepoService = (*Repos)(nil)

@@ -31,6 +32,13 @@ func NewRepos(meta ports.RepoMetaStore, git ports.GitRepos, bus ports.EventBus,
31	return &Repos{meta: meta, git: git, bus: bus, clock: clock}	32		return &Repos{meta: meta, git: git, bus: bus, clock: clock}
32	}	33		}
33		34
		35	+	// WithCIRunCleaner makes Delete remove the CI logs of the repository. Call it
		36	+	// before the first request; it returns r.
		37	+	func (r Repos) WithCIRunCleaner(c ports.CIRunCleaner) Repos {
		38	+	r.ci = c
		39	+	return r
		40	+	}
		41	+
34	// Create makes a repository and publishes RepoCreated: it stages a bare repository, records the	42		// Create makes a repository and publishes RepoCreated: it stages a bare repository, records the
35	// metadata (which fails with domain.ErrAlreadyExists if the name is taken,	43		// metadata (which fails with domain.ErrAlreadyExists if the name is taken,
36	// also when a concurrent Create wins) and publishes the directory. A failure	44		// also when a concurrent Create wins) and publishes the directory. A failure

M internal/usecase/repos_write.go +14 -0


@@ -191,9 +191,23 @@ func (r *Repos) Delete(ctx context.Context, p domain.Principal, name domain.Repo
191	191		case err != nil:
192	192		return fmt.Errorf("trash repository %q: %w", name, err)
193	193		}
	194	+	var runIDs []int64
	195	+	if r.ci != nil {
	196	+	// Listed before the records go: the logs are found by run id.
	197	+	if ids, err := r.ci.RunIDs(ctx, name); err == nil {
	198	+	runIDs = ids
	199	+	} else {
	200	+	log.Warn().Err(err).Str("repo", name.String()).Msg("list the CI runs of the repository; their logs are left to the sweep")
	201	+	}
	202	+	}
194	203		if err := r.meta.Delete(ctx, name); err != nil {
195	204		return fmt.Errorf("delete metadata of %q: %w", name, err)
196	205		}
	206	+	if len(runIDs) > 0 {
	207	+	if err := r.ci.RemoveLogs(context.WithoutCancel(ctx), runIDs); err != nil {
	208	+	log.Warn().Err(err).Str("repo", name.String()).Msg("remove the CI logs of the repository; the sweep will")
	209	+	}
	210	+	}
197	211		log.Info().Str("repo", name.String()).Str("principal", p.Label()).Str("token_id", p.TokenID()).Msg("repository deleted")
198	212		r.bus.Publish(ctx, domain.RepoDeleted{Repo: name, Visibility: repo.Visibility, At: r.clock.Now().UTC(), CreatedAt: repo.CreatedAt})
199	213		return nil


@@ -191,9 +191,23 @@ func (r *Repos) Delete(ctx context.Context, p domain.Principal, name domain.Repo
191	case err != nil:	191		case err != nil:
192	return fmt.Errorf("trash repository %q: %w", name, err)	192		return fmt.Errorf("trash repository %q: %w", name, err)
193	}	193		}
		194	+	var runIDs []int64
		195	+	if r.ci != nil {
		196	+	// Listed before the records go: the logs are found by run id.
		197	+	if ids, err := r.ci.RunIDs(ctx, name); err == nil {
		198	+	runIDs = ids
		199	+	} else {
		200	+	log.Warn().Err(err).Str("repo", name.String()).Msg("list the CI runs of the repository; their logs are left to the sweep")
		201	+	}
		202	+	}
194	if err := r.meta.Delete(ctx, name); err != nil {	203		if err := r.meta.Delete(ctx, name); err != nil {
195	return fmt.Errorf("delete metadata of %q: %w", name, err)	204		return fmt.Errorf("delete metadata of %q: %w", name, err)
196	}	205		}
		206	+	if len(runIDs) > 0 {
		207	+	if err := r.ci.RemoveLogs(context.WithoutCancel(ctx), runIDs); err != nil {
		208	+	log.Warn().Err(err).Str("repo", name.String()).Msg("remove the CI logs of the repository; the sweep will")
		209	+	}
		210	+	}
197	log.Info().Str("repo", name.String()).Str("principal", p.Label()).Str("token_id", p.TokenID()).Msg("repository deleted")	211		log.Info().Str("repo", name.String()).Str("principal", p.Label()).Str("token_id", p.TokenID()).Msg("repository deleted")
198	r.bus.Publish(ctx, domain.RepoDeleted{Repo: name, Visibility: repo.Visibility, At: r.clock.Now().UTC(), CreatedAt: repo.CreatedAt})	212		r.bus.Publish(ctx, domain.RepoDeleted{Repo: name, Visibility: repo.Visibility, At: r.clock.Now().UTC(), CreatedAt: repo.CreatedAt})
199	return nil	213		return nil