feat(rpc): serve the mirror service, convert mirror events and add the mirror client
- commit
- ed419c7c0a35bb6998c44565d18f86dfe5b5f33b
- author
- brice · · 2026-10-20 05:51 UTC
- committer
- brice · · 2026-10-20 10:02 UTC
- parent
- 26e2b66
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
7 files · +281 -6
M internal/adapters/client/client.go +5 -0
| @@ -60,6 +60,11 @@ func transport() *http.Transport { | |||
| 60 | 60 | // URL is the server address the client talks to. | |
| 61 | 61 | func (c *Client) URL() string { return c.cfg.URL } | |
| 62 | 62 | ||
| 63 | + | // Mirrors returns the MirrorService client. | |
| 64 | + | func (c *Client) Mirrors() ermiteconnect.MirrorServiceClient { | |
| 65 | + | return ermiteconnect.NewMirrorServiceClient(c.http, c.base, c.opts...) | |
| 66 | + | } | |
| 67 | + | ||
| 63 | 68 | // Repos returns the RepoService client. | |
| 64 | 69 | func (c *Client) Repos() ermiteconnect.RepoServiceClient { | |
| 65 | 70 | return ermiteconnect.NewRepoServiceClient(c.http, c.base, c.opts...) | |
| @@ -60,6 +60,11 @@ func transport() *http.Transport { | |||||
| 60 | // URL is the server address the client talks to. | 60 | // URL is the server address the client talks to. | ||
| 61 | func (c *Client) URL() string { return c.cfg.URL } | 61 | func (c *Client) URL() string { return c.cfg.URL } | ||
| 62 | 62 | ||||
| 63 | + | // Mirrors returns the MirrorService client. | |||
| 64 | + | func (c *Client) Mirrors() ermiteconnect.MirrorServiceClient { | |||
| 65 | + | return ermiteconnect.NewMirrorServiceClient(c.http, c.base, c.opts...) | |||
| 66 | + | } | |||
| 67 | + | ||||
| 63 | // Repos returns the RepoService client. | 68 | // Repos returns the RepoService client. | ||
| 64 | func (c *Client) Repos() ermiteconnect.RepoServiceClient { | 69 | func (c *Client) Repos() ermiteconnect.RepoServiceClient { | ||
| 65 | return ermiteconnect.NewRepoServiceClient(c.http, c.base, c.opts...) | 70 | return ermiteconnect.NewRepoServiceClient(c.http, c.base, c.opts...) | ||
M internal/adapters/rpc/api.go +11 -5
| @@ -47,11 +47,13 @@ const FeatureCIOIDC = "ci.oidc" | |||
| 47 | 47 | ||
| 48 | 48 | // Config configures the API. | |
| 49 | 49 | type Config struct { | |
| 50 | - | Auth ports.AuthService | |
| 51 | - | Repos ports.RepoService | |
| 52 | - | Tokens ports.TokenService | |
| 53 | - | Browse ports.BrowseService | |
| 54 | - | Events ports.EventService | |
| 50 | + | // Mirrors serves MirrorService when set; leave unset rather than a typed nil. | |
| 51 | + | Mirrors ports.MirrorService | |
| 52 | + | Auth ports.AuthService | |
| 53 | + | Repos ports.RepoService | |
| 54 | + | Tokens ports.TokenService | |
| 55 | + | Browse ports.BrowseService | |
| 56 | + | Events ports.EventService | |
| 55 | 57 | // MergeRequests serves MergeRequestService, SigningKey GetSigningKey. | |
| 56 | 58 | MergeRequests ports.MergeRequestService | |
| 57 | 59 | SigningKey ports.SigningKeyService | |
| @@ -125,6 +127,10 @@ func New(cfg Config) *API { | |||
| 125 | 127 | mux.Handle(ermiteconnect.NewTokenServiceHandler(&tokenService{cfg.Tokens}, opts...)) | |
| 126 | 128 | mux.Handle(ermiteconnect.NewEventServiceHandler(&eventService{cfg.Events}, opts...)) | |
| 127 | 129 | features := slices.Clone(Features) | |
| 130 | + | if cfg.Mirrors != nil { | |
| 131 | + | mux.Handle(ermiteconnect.NewMirrorServiceHandler(&mirrorService{cfg.Mirrors}, opts...)) | |
| 132 | + | features = append(features, FeatureMirror) | |
| 133 | + | } | |
| 128 | 134 | if cfg.Passkeys != nil { | |
| 129 | 135 | features = append(features, FeaturePasskeys) | |
| 130 | 136 | } | |
| @@ -47,11 +47,13 @@ const FeatureCIOIDC = "ci.oidc" | |||||
| 47 | 47 | ||||
| 48 | // Config configures the API. | 48 | // Config configures the API. | ||
| 49 | type Config struct { | 49 | type Config struct { | ||
| 50 | - | Auth ports.AuthService | 50 | + | // Mirrors serves MirrorService when set; leave unset rather than a typed nil. |
| 51 | - | Repos ports.RepoService | 51 | + | Mirrors ports.MirrorService |
| 52 | - | Tokens ports.TokenService | 52 | + | Auth ports.AuthService |
| 53 | - | Browse ports.BrowseService | 53 | + | Repos ports.RepoService |
| 54 | - | Events ports.EventService | 54 | + | Tokens ports.TokenService |
| 55 | + | Browse ports.BrowseService | |||
| 56 | + | Events ports.EventService | |||
| 55 | // MergeRequests serves MergeRequestService, SigningKey GetSigningKey. | 57 | // MergeRequests serves MergeRequestService, SigningKey GetSigningKey. | ||
| 56 | MergeRequests ports.MergeRequestService | 58 | MergeRequests ports.MergeRequestService | ||
| 57 | SigningKey ports.SigningKeyService | 59 | SigningKey ports.SigningKeyService | ||
| @@ -125,6 +127,10 @@ func New(cfg Config) *API { | |||||
| 125 | mux.Handle(ermiteconnect.NewTokenServiceHandler(&tokenService{cfg.Tokens}, opts...)) | 127 | mux.Handle(ermiteconnect.NewTokenServiceHandler(&tokenService{cfg.Tokens}, opts...)) | ||
| 126 | mux.Handle(ermiteconnect.NewEventServiceHandler(&eventService{cfg.Events}, opts...)) | 128 | mux.Handle(ermiteconnect.NewEventServiceHandler(&eventService{cfg.Events}, opts...)) | ||
| 127 | features := slices.Clone(Features) | 129 | features := slices.Clone(Features) | ||
| 130 | + | if cfg.Mirrors != nil { | |||
| 131 | + | mux.Handle(ermiteconnect.NewMirrorServiceHandler(&mirrorService{cfg.Mirrors}, opts...)) | |||
| 132 | + | features = append(features, FeatureMirror) | |||
| 133 | + | } | |||
| 128 | if cfg.Passkeys != nil { | 134 | if cfg.Passkeys != nil { | ||
| 129 | features = append(features, FeaturePasskeys) | 135 | features = append(features, FeaturePasskeys) | ||
| 130 | } | 136 | } | ||
M internal/adapters/rpc/convert.go +6 -0
| @@ -523,6 +523,12 @@ func watchEventToPB(we ports.WatchEvent) *pb.Event { | |||
| 523 | 523 | case domain.JobUpdated: | |
| 524 | 524 | out.Time = timestamp(ev.At) | |
| 525 | 525 | out.Kind = jobEventToPB(ev) | |
| 526 | + | case domain.MirrorFailed: | |
| 527 | + | out.Time = timestamp(ev.At) | |
| 528 | + | out.Kind = &pb.Event_MirrorFailed{MirrorFailed: &pb.MirrorFailed{Repo: ev.Repo.String(), State: mirrorStateToPB(ev.State), FailureClass: string(ev.Class), Reason: ev.Reason}} | |
| 529 | + | case domain.MirrorRecovered: | |
| 530 | + | out.Time = timestamp(ev.At) | |
| 531 | + | out.Kind = &pb.Event_MirrorRecovered{MirrorRecovered: &pb.MirrorRecovered{Repo: ev.Repo.String()}} | |
| 526 | 532 | default: | |
| 527 | 533 | // Web-only events, such as domain.IssuePendingRemoved (the API has | |
| 528 | 534 | // no message for it; watchers list pending issues on demand). | |
| @@ -523,6 +523,12 @@ func watchEventToPB(we ports.WatchEvent) *pb.Event { | |||||
| 523 | case domain.JobUpdated: | 523 | case domain.JobUpdated: | ||
| 524 | out.Time = timestamp(ev.At) | 524 | out.Time = timestamp(ev.At) | ||
| 525 | out.Kind = jobEventToPB(ev) | 525 | out.Kind = jobEventToPB(ev) | ||
| 526 | + | case domain.MirrorFailed: | |||
| 527 | + | out.Time = timestamp(ev.At) | |||
| 528 | + | out.Kind = &pb.Event_MirrorFailed{MirrorFailed: &pb.MirrorFailed{Repo: ev.Repo.String(), State: mirrorStateToPB(ev.State), FailureClass: string(ev.Class), Reason: ev.Reason}} | |||
| 529 | + | case domain.MirrorRecovered: | |||
| 530 | + | out.Time = timestamp(ev.At) | |||
| 531 | + | out.Kind = &pb.Event_MirrorRecovered{MirrorRecovered: &pb.MirrorRecovered{Repo: ev.Repo.String()}} | |||
| 526 | default: | 532 | default: | ||
| 527 | // Web-only events, such as domain.IssuePendingRemoved (the API has | 533 | // Web-only events, such as domain.IssuePendingRemoved (the API has | ||
| 528 | // no message for it; watchers list pending issues on demand). | 534 | // no message for it; watchers list pending issues on demand). | ||
A internal/adapters/rpc/mirror.go +185 -0
| @@ -0,0 +1,185 @@ | |||
| 1 | + | // SPDX-License-Identifier: AGPL-3.0-or-later | |
| 2 | + | ||
| 3 | + | package rpc | |
| 4 | + | ||
| 5 | + | import ( | |
| 6 | + | "context" | |
| 7 | + | "fmt" | |
| 8 | + | ||
| 9 | + | pb "bnema.dev/ermite-api/gen/ermite" | |
| 10 | + | "bnema.dev/ermite-api/gen/ermite/ermiteconnect" | |
| 11 | + | "bnema.dev/ermite/internal/domain" | |
| 12 | + | "bnema.dev/ermite/internal/ports" | |
| 13 | + | "connectrpc.com/connect" | |
| 14 | + | ) | |
| 15 | + | ||
| 16 | + | const FeatureMirror = "mirror" | |
| 17 | + | ||
| 18 | + | type mirrorService struct{ mirrors ports.MirrorService } | |
| 19 | + | ||
| 20 | + | var _ ermiteconnect.MirrorServiceHandler = (*mirrorService)(nil) | |
| 21 | + | ||
| 22 | + | func (s *mirrorService) GetMirrorKey(ctx context.Context, _ *connect.Request[pb.GetMirrorKeyRequest]) (*connect.Response[pb.GetMirrorKeyResponse], error) { | |
| 23 | + | k, err := s.mirrors.ShowKey(ctx, caller(ctx)) | |
| 24 | + | if err != nil { | |
| 25 | + | return nil, connectError(ctx, err) | |
| 26 | + | } | |
| 27 | + | return connect.NewResponse(&pb.GetMirrorKeyResponse{Key: mirrorKeyToPB(k)}), nil | |
| 28 | + | } | |
| 29 | + | func (s *mirrorService) RotateMirrorKey(ctx context.Context, _ *connect.Request[pb.RotateMirrorKeyRequest]) (*connect.Response[pb.RotateMirrorKeyResponse], error) { | |
| 30 | + | // The use case owns authorization for both operations. | |
| 31 | + | previous, err := s.mirrors.ShowKey(ctx, caller(ctx)) | |
| 32 | + | if err != nil { | |
| 33 | + | return nil, connectError(ctx, err) | |
| 34 | + | } | |
| 35 | + | k, err := s.mirrors.RotateKey(ctx, caller(ctx)) | |
| 36 | + | if err != nil { | |
| 37 | + | return nil, connectError(ctx, err) | |
| 38 | + | } | |
| 39 | + | return connect.NewResponse(&pb.RotateMirrorKeyResponse{Key: mirrorKeyToPB(k), PreviousFingerprint: previous.Fingerprint}), nil | |
| 40 | + | } | |
| 41 | + | func (s *mirrorService) GetMirrorConfig(ctx context.Context, _ *connect.Request[pb.GetMirrorConfigRequest]) (*connect.Response[pb.GetMirrorConfigResponse], error) { | |
| 42 | + | g, err := s.mirrors.GetGlobalConfig(ctx, caller(ctx)) | |
| 43 | + | if err != nil { | |
| 44 | + | return nil, connectError(ctx, err) | |
| 45 | + | } | |
| 46 | + | return connect.NewResponse(&pb.GetMirrorConfigResponse{Config: mirrorConfigToPB(g)}), nil | |
| 47 | + | } | |
| 48 | + | func (s *mirrorService) SetMirrorConfig(ctx context.Context, req *connect.Request[pb.SetMirrorConfigRequest]) (*connect.Response[pb.SetMirrorConfigResponse], error) { | |
| 49 | + | m := req.Msg | |
| 50 | + | g, err := s.mirrors.SetGlobalConfig(ctx, caller(ctx), ports.MirrorGlobalPatch{Enabled: m.Enabled, URLTemplate: m.UrlTemplate, Cron: m.Cron, AllBranches: m.AllBranches}) | |
| 51 | + | if err != nil { | |
| 52 | + | return nil, connectError(ctx, err) | |
| 53 | + | } | |
| 54 | + | return connect.NewResponse(&pb.SetMirrorConfigResponse{Config: mirrorConfigToPB(g)}), nil | |
| 55 | + | } | |
| 56 | + | func (s *mirrorService) SetRepoMirror(ctx context.Context, req *connect.Request[pb.SetRepoMirrorRequest]) (*connect.Response[pb.SetRepoMirrorResponse], error) { | |
| 57 | + | m := req.Msg | |
| 58 | + | name, err := domain.ParseRepoName(m.GetRepo()) | |
| 59 | + | if err != nil { | |
| 60 | + | return nil, connectError(ctx, err) | |
| 61 | + | } | |
| 62 | + | t, err := s.mirrors.SetRepoMirror(ctx, caller(ctx), domain.MirrorRepoConfig{Repo: name, URL: m.GetUrl(), Cron: m.GetCron(), AllBranches: m.AllBranches}) | |
| 63 | + | if err != nil { | |
| 64 | + | return nil, connectError(ctx, err) | |
| 65 | + | } | |
| 66 | + | return connect.NewResponse(&pb.SetRepoMirrorResponse{Target: mirrorTargetToPB(t)}), nil | |
| 67 | + | } | |
| 68 | + | func (s *mirrorService) RemoveRepoMirror(ctx context.Context, req *connect.Request[pb.RemoveRepoMirrorRequest]) (*connect.Response[pb.RemoveRepoMirrorResponse], error) { | |
| 69 | + | name, err := domain.ParseRepoName(req.Msg.GetRepo()) | |
| 70 | + | if err != nil { | |
| 71 | + | return nil, connectError(ctx, err) | |
| 72 | + | } | |
| 73 | + | if err := s.mirrors.RemoveRepoMirror(ctx, caller(ctx), name); err != nil { | |
| 74 | + | return nil, connectError(ctx, err) | |
| 75 | + | } | |
| 76 | + | return connect.NewResponse(&pb.RemoveRepoMirrorResponse{}), nil | |
| 77 | + | } | |
| 78 | + | func (s *mirrorService) SetRepoMirrorExcluded(ctx context.Context, req *connect.Request[pb.SetRepoMirrorExcludedRequest]) (*connect.Response[pb.SetRepoMirrorExcludedResponse], error) { | |
| 79 | + | name, err := domain.ParseRepoName(req.Msg.GetRepo()) | |
| 80 | + | if err != nil { | |
| 81 | + | return nil, connectError(ctx, err) | |
| 82 | + | } | |
| 83 | + | t, err := s.mirrors.SetRepoExcluded(ctx, caller(ctx), name, req.Msg.GetExcluded()) | |
| 84 | + | if err != nil { | |
| 85 | + | return nil, connectError(ctx, err) | |
| 86 | + | } | |
| 87 | + | return connect.NewResponse(&pb.SetRepoMirrorExcludedResponse{Target: mirrorTargetToPB(t)}), nil | |
| 88 | + | } | |
| 89 | + | func (s *mirrorService) GetMirrorStatus(ctx context.Context, req *connect.Request[pb.GetMirrorStatusRequest]) (*connect.Response[pb.GetMirrorStatusResponse], error) { | |
| 90 | + | name, err := domain.ParseRepoName(req.Msg.GetRepo()) | |
| 91 | + | if err != nil { | |
| 92 | + | return nil, connectError(ctx, err) | |
| 93 | + | } | |
| 94 | + | v, err := s.mirrors.GetStatus(ctx, caller(ctx), name) | |
| 95 | + | if err != nil { | |
| 96 | + | return nil, connectError(ctx, err) | |
| 97 | + | } | |
| 98 | + | return connect.NewResponse(&pb.GetMirrorStatusResponse{Status: mirrorStatusToPB(v)}), nil | |
| 99 | + | } | |
| 100 | + | func (s *mirrorService) ListMirrorStatus(ctx context.Context, _ *connect.Request[pb.ListMirrorStatusRequest]) (*connect.Response[pb.ListMirrorStatusResponse], error) { | |
| 101 | + | list, err := s.mirrors.ListStatus(ctx, caller(ctx)) | |
| 102 | + | if err != nil { | |
| 103 | + | return nil, connectError(ctx, err) | |
| 104 | + | } | |
| 105 | + | out := make([]*pb.MirrorStatus, 0, len(list)) | |
| 106 | + | for _, v := range list { | |
| 107 | + | out = append(out, mirrorStatusToPB(v)) | |
| 108 | + | } | |
| 109 | + | return connect.NewResponse(&pb.ListMirrorStatusResponse{Statuses: out}), nil | |
| 110 | + | } | |
| 111 | + | func (s *mirrorService) SyncMirror(ctx context.Context, req *connect.Request[pb.SyncMirrorRequest]) (*connect.Response[pb.SyncMirrorResponse], error) { | |
| 112 | + | m := req.Msg | |
| 113 | + | if m.GetAll() == (m.GetRepo() != "") { | |
| 114 | + | return nil, connectError(ctx, fmt.Errorf("%w: specify either repo or all", domain.ErrInvalid)) | |
| 115 | + | } | |
| 116 | + | var results []ports.MirrorSyncResult | |
| 117 | + | if m.GetAll() { | |
| 118 | + | list, err := s.mirrors.SyncAll(ctx, caller(ctx), m.GetForce()) | |
| 119 | + | if err != nil { | |
| 120 | + | return nil, connectError(ctx, err) | |
| 121 | + | } | |
| 122 | + | results = list | |
| 123 | + | } else { | |
| 124 | + | name, err := domain.ParseRepoName(m.GetRepo()) | |
| 125 | + | if err != nil { | |
| 126 | + | return nil, connectError(ctx, err) | |
| 127 | + | } | |
| 128 | + | result, err := s.mirrors.Sync(ctx, caller(ctx), name, m.GetForce()) | |
| 129 | + | if err != nil { | |
| 130 | + | return nil, connectError(ctx, err) | |
| 131 | + | } | |
| 132 | + | results = []ports.MirrorSyncResult{result} | |
| 133 | + | } | |
| 134 | + | out := make([]*pb.SyncMirrorResult, 0, len(results)) | |
| 135 | + | for _, result := range results { | |
| 136 | + | v, err := s.mirrors.GetStatus(ctx, caller(ctx), result.Status.Repo) | |
| 137 | + | if err != nil { | |
| 138 | + | return nil, connectError(ctx, err) | |
| 139 | + | } | |
| 140 | + | if !result.Queued { | |
| 141 | + | v.Status = result.Status | |
| 142 | + | } | |
| 143 | + | out = append(out, &pb.SyncMirrorResult{Repo: result.Status.Repo.String(), Queued: result.Queued, NoOp: result.NoOp, Status: mirrorStatusToPB(v)}) | |
| 144 | + | } | |
| 145 | + | return connect.NewResponse(&pb.SyncMirrorResponse{Results: out}), nil | |
| 146 | + | } | |
| 147 | + | ||
| 148 | + | func mirrorKeyToPB(k domain.MirrorPublicKey) *pb.MirrorKey { | |
| 149 | + | return &pb.MirrorKey{AuthorizedKey: k.AuthorizedKey, Fingerprint: k.Fingerprint, CreatedAt: timestamp(k.CreatedAt)} | |
| 150 | + | } | |
| 151 | + | func mirrorConfigToPB(g domain.MirrorGlobalConfig) *pb.MirrorConfig { | |
| 152 | + | return &pb.MirrorConfig{Enabled: g.Enabled, UrlTemplate: g.URLTemplate, Cron: g.Cron, AllBranches: g.AllBranches, UpdatedAt: timestamp(g.UpdatedAt)} | |
| 153 | + | } | |
| 154 | + | func mirrorStateToPB(s domain.MirrorState) pb.MirrorState { | |
| 155 | + | switch s { | |
| 156 | + | case domain.MirrorDisabled: | |
| 157 | + | return pb.MirrorState_MIRROR_STATE_DISABLED | |
| 158 | + | case domain.MirrorOK: | |
| 159 | + | return pb.MirrorState_MIRROR_STATE_OK | |
| 160 | + | case domain.MirrorFailing: | |
| 161 | + | return pb.MirrorState_MIRROR_STATE_FAILING | |
| 162 | + | case domain.MirrorPaused: | |
| 163 | + | return pb.MirrorState_MIRROR_STATE_PAUSED | |
| 164 | + | default: | |
| 165 | + | return pb.MirrorState_MIRROR_STATE_UNSPECIFIED | |
| 166 | + | } | |
| 167 | + | } | |
| 168 | + | func mirrorTargetToPB(t domain.MirrorTarget) *pb.MirrorTarget { | |
| 169 | + | source := pb.MirrorSource_MIRROR_SOURCE_UNSPECIFIED | |
| 170 | + | switch t.Source { | |
| 171 | + | case domain.MirrorSourceNone: | |
| 172 | + | source = pb.MirrorSource_MIRROR_SOURCE_DISABLED | |
| 173 | + | case domain.MirrorSourceGlobal: | |
| 174 | + | source = pb.MirrorSource_MIRROR_SOURCE_GLOBAL | |
| 175 | + | case domain.MirrorSourceOverride: | |
| 176 | + | source = pb.MirrorSource_MIRROR_SOURCE_OVERRIDE | |
| 177 | + | case domain.MirrorSourceExcluded: | |
| 178 | + | source = pb.MirrorSource_MIRROR_SOURCE_EXCLUDED | |
| 179 | + | } | |
| 180 | + | return &pb.MirrorTarget{Repo: t.Repo.String(), Enabled: t.Enabled, Url: t.URL, Cron: t.CronSpec, AllBranches: t.AllBranches, Source: source} | |
| 181 | + | } | |
| 182 | + | func mirrorStatusToPB(v ports.MirrorStatusView) *pb.MirrorStatus { | |
| 183 | + | s := v.Status | |
| 184 | + | return &pb.MirrorStatus{Target: mirrorTargetToPB(v.Target), State: mirrorStateToPB(s.State), FailureClass: string(s.Class), Reason: s.Reason, Detail: s.Detail, ConsecutiveFailures: int32(s.ConsecutiveFailures), LastAttemptAt: timestamp(s.LastAttemptAt), LastSuccessAt: timestamp(s.LastSuccessAt), NextRetryAt: timestamp(s.NextRetryAt), NextRunAt: timestamp(v.NextRunAt), PausedAt: timestamp(s.PausedAt)} | |
| 185 | + | } | |
| @@ -0,0 +1,185 @@ | |||||
| 1 | + | // SPDX-License-Identifier: AGPL-3.0-or-later | |||
| 2 | + | ||||
| 3 | + | package rpc | |||
| 4 | + | ||||
| 5 | + | import ( | |||
| 6 | + | "context" | |||
| 7 | + | "fmt" | |||
| 8 | + | ||||
| 9 | + | pb "bnema.dev/ermite-api/gen/ermite" | |||
| 10 | + | "bnema.dev/ermite-api/gen/ermite/ermiteconnect" | |||
| 11 | + | "bnema.dev/ermite/internal/domain" | |||
| 12 | + | "bnema.dev/ermite/internal/ports" | |||
| 13 | + | "connectrpc.com/connect" | |||
| 14 | + | ) | |||
| 15 | + | ||||
| 16 | + | const FeatureMirror = "mirror" | |||
| 17 | + | ||||
| 18 | + | type mirrorService struct{ mirrors ports.MirrorService } | |||
| 19 | + | ||||
| 20 | + | var _ ermiteconnect.MirrorServiceHandler = (*mirrorService)(nil) | |||
| 21 | + | ||||
| 22 | + | func (s *mirrorService) GetMirrorKey(ctx context.Context, _ *connect.Request[pb.GetMirrorKeyRequest]) (*connect.Response[pb.GetMirrorKeyResponse], error) { | |||
| 23 | + | k, err := s.mirrors.ShowKey(ctx, caller(ctx)) | |||
| 24 | + | if err != nil { | |||
| 25 | + | return nil, connectError(ctx, err) | |||
| 26 | + | } | |||
| 27 | + | return connect.NewResponse(&pb.GetMirrorKeyResponse{Key: mirrorKeyToPB(k)}), nil | |||
| 28 | + | } | |||
| 29 | + | func (s *mirrorService) RotateMirrorKey(ctx context.Context, _ *connect.Request[pb.RotateMirrorKeyRequest]) (*connect.Response[pb.RotateMirrorKeyResponse], error) { | |||
| 30 | + | // The use case owns authorization for both operations. | |||
| 31 | + | previous, err := s.mirrors.ShowKey(ctx, caller(ctx)) | |||
| 32 | + | if err != nil { | |||
| 33 | + | return nil, connectError(ctx, err) | |||
| 34 | + | } | |||
| 35 | + | k, err := s.mirrors.RotateKey(ctx, caller(ctx)) | |||
| 36 | + | if err != nil { | |||
| 37 | + | return nil, connectError(ctx, err) | |||
| 38 | + | } | |||
| 39 | + | return connect.NewResponse(&pb.RotateMirrorKeyResponse{Key: mirrorKeyToPB(k), PreviousFingerprint: previous.Fingerprint}), nil | |||
| 40 | + | } | |||
| 41 | + | func (s *mirrorService) GetMirrorConfig(ctx context.Context, _ *connect.Request[pb.GetMirrorConfigRequest]) (*connect.Response[pb.GetMirrorConfigResponse], error) { | |||
| 42 | + | g, err := s.mirrors.GetGlobalConfig(ctx, caller(ctx)) | |||
| 43 | + | if err != nil { | |||
| 44 | + | return nil, connectError(ctx, err) | |||
| 45 | + | } | |||
| 46 | + | return connect.NewResponse(&pb.GetMirrorConfigResponse{Config: mirrorConfigToPB(g)}), nil | |||
| 47 | + | } | |||
| 48 | + | func (s *mirrorService) SetMirrorConfig(ctx context.Context, req *connect.Request[pb.SetMirrorConfigRequest]) (*connect.Response[pb.SetMirrorConfigResponse], error) { | |||
| 49 | + | m := req.Msg | |||
| 50 | + | g, err := s.mirrors.SetGlobalConfig(ctx, caller(ctx), ports.MirrorGlobalPatch{Enabled: m.Enabled, URLTemplate: m.UrlTemplate, Cron: m.Cron, AllBranches: m.AllBranches}) | |||
| 51 | + | if err != nil { | |||
| 52 | + | return nil, connectError(ctx, err) | |||
| 53 | + | } | |||
| 54 | + | return connect.NewResponse(&pb.SetMirrorConfigResponse{Config: mirrorConfigToPB(g)}), nil | |||
| 55 | + | } | |||
| 56 | + | func (s *mirrorService) SetRepoMirror(ctx context.Context, req *connect.Request[pb.SetRepoMirrorRequest]) (*connect.Response[pb.SetRepoMirrorResponse], error) { | |||
| 57 | + | m := req.Msg | |||
| 58 | + | name, err := domain.ParseRepoName(m.GetRepo()) | |||
| 59 | + | if err != nil { | |||
| 60 | + | return nil, connectError(ctx, err) | |||
| 61 | + | } | |||
| 62 | + | t, err := s.mirrors.SetRepoMirror(ctx, caller(ctx), domain.MirrorRepoConfig{Repo: name, URL: m.GetUrl(), Cron: m.GetCron(), AllBranches: m.AllBranches}) | |||
| 63 | + | if err != nil { | |||
| 64 | + | return nil, connectError(ctx, err) | |||
| 65 | + | } | |||
| 66 | + | return connect.NewResponse(&pb.SetRepoMirrorResponse{Target: mirrorTargetToPB(t)}), nil | |||
| 67 | + | } | |||
| 68 | + | func (s *mirrorService) RemoveRepoMirror(ctx context.Context, req *connect.Request[pb.RemoveRepoMirrorRequest]) (*connect.Response[pb.RemoveRepoMirrorResponse], error) { | |||
| 69 | + | name, err := domain.ParseRepoName(req.Msg.GetRepo()) | |||
| 70 | + | if err != nil { | |||
| 71 | + | return nil, connectError(ctx, err) | |||
| 72 | + | } | |||
| 73 | + | if err := s.mirrors.RemoveRepoMirror(ctx, caller(ctx), name); err != nil { | |||
| 74 | + | return nil, connectError(ctx, err) | |||
| 75 | + | } | |||
| 76 | + | return connect.NewResponse(&pb.RemoveRepoMirrorResponse{}), nil | |||
| 77 | + | } | |||
| 78 | + | func (s *mirrorService) SetRepoMirrorExcluded(ctx context.Context, req *connect.Request[pb.SetRepoMirrorExcludedRequest]) (*connect.Response[pb.SetRepoMirrorExcludedResponse], error) { | |||
| 79 | + | name, err := domain.ParseRepoName(req.Msg.GetRepo()) | |||
| 80 | + | if err != nil { | |||
| 81 | + | return nil, connectError(ctx, err) | |||
| 82 | + | } | |||
| 83 | + | t, err := s.mirrors.SetRepoExcluded(ctx, caller(ctx), name, req.Msg.GetExcluded()) | |||
| 84 | + | if err != nil { | |||
| 85 | + | return nil, connectError(ctx, err) | |||
| 86 | + | } | |||
| 87 | + | return connect.NewResponse(&pb.SetRepoMirrorExcludedResponse{Target: mirrorTargetToPB(t)}), nil | |||
| 88 | + | } | |||
| 89 | + | func (s *mirrorService) GetMirrorStatus(ctx context.Context, req *connect.Request[pb.GetMirrorStatusRequest]) (*connect.Response[pb.GetMirrorStatusResponse], error) { | |||
| 90 | + | name, err := domain.ParseRepoName(req.Msg.GetRepo()) | |||
| 91 | + | if err != nil { | |||
| 92 | + | return nil, connectError(ctx, err) | |||
| 93 | + | } | |||
| 94 | + | v, err := s.mirrors.GetStatus(ctx, caller(ctx), name) | |||
| 95 | + | if err != nil { | |||
| 96 | + | return nil, connectError(ctx, err) | |||
| 97 | + | } | |||
| 98 | + | return connect.NewResponse(&pb.GetMirrorStatusResponse{Status: mirrorStatusToPB(v)}), nil | |||
| 99 | + | } | |||
| 100 | + | func (s *mirrorService) ListMirrorStatus(ctx context.Context, _ *connect.Request[pb.ListMirrorStatusRequest]) (*connect.Response[pb.ListMirrorStatusResponse], error) { | |||
| 101 | + | list, err := s.mirrors.ListStatus(ctx, caller(ctx)) | |||
| 102 | + | if err != nil { | |||
| 103 | + | return nil, connectError(ctx, err) | |||
| 104 | + | } | |||
| 105 | + | out := make([]*pb.MirrorStatus, 0, len(list)) | |||
| 106 | + | for _, v := range list { | |||
| 107 | + | out = append(out, mirrorStatusToPB(v)) | |||
| 108 | + | } | |||
| 109 | + | return connect.NewResponse(&pb.ListMirrorStatusResponse{Statuses: out}), nil | |||
| 110 | + | } | |||
| 111 | + | func (s *mirrorService) SyncMirror(ctx context.Context, req *connect.Request[pb.SyncMirrorRequest]) (*connect.Response[pb.SyncMirrorResponse], error) { | |||
| 112 | + | m := req.Msg | |||
| 113 | + | if m.GetAll() == (m.GetRepo() != "") { | |||
| 114 | + | return nil, connectError(ctx, fmt.Errorf("%w: specify either repo or all", domain.ErrInvalid)) | |||
| 115 | + | } | |||
| 116 | + | var results []ports.MirrorSyncResult | |||
| 117 | + | if m.GetAll() { | |||
| 118 | + | list, err := s.mirrors.SyncAll(ctx, caller(ctx), m.GetForce()) | |||
| 119 | + | if err != nil { | |||
| 120 | + | return nil, connectError(ctx, err) | |||
| 121 | + | } | |||
| 122 | + | results = list | |||
| 123 | + | } else { | |||
| 124 | + | name, err := domain.ParseRepoName(m.GetRepo()) | |||
| 125 | + | if err != nil { | |||
| 126 | + | return nil, connectError(ctx, err) | |||
| 127 | + | } | |||
| 128 | + | result, err := s.mirrors.Sync(ctx, caller(ctx), name, m.GetForce()) | |||
| 129 | + | if err != nil { | |||
| 130 | + | return nil, connectError(ctx, err) | |||
| 131 | + | } | |||
| 132 | + | results = []ports.MirrorSyncResult{result} | |||
| 133 | + | } | |||
| 134 | + | out := make([]*pb.SyncMirrorResult, 0, len(results)) | |||
| 135 | + | for _, result := range results { | |||
| 136 | + | v, err := s.mirrors.GetStatus(ctx, caller(ctx), result.Status.Repo) | |||
| 137 | + | if err != nil { | |||
| 138 | + | return nil, connectError(ctx, err) | |||
| 139 | + | } | |||
| 140 | + | if !result.Queued { | |||
| 141 | + | v.Status = result.Status | |||
| 142 | + | } | |||
| 143 | + | out = append(out, &pb.SyncMirrorResult{Repo: result.Status.Repo.String(), Queued: result.Queued, NoOp: result.NoOp, Status: mirrorStatusToPB(v)}) | |||
| 144 | + | } | |||
| 145 | + | return connect.NewResponse(&pb.SyncMirrorResponse{Results: out}), nil | |||
| 146 | + | } | |||
| 147 | + | ||||
| 148 | + | func mirrorKeyToPB(k domain.MirrorPublicKey) *pb.MirrorKey { | |||
| 149 | + | return &pb.MirrorKey{AuthorizedKey: k.AuthorizedKey, Fingerprint: k.Fingerprint, CreatedAt: timestamp(k.CreatedAt)} | |||
| 150 | + | } | |||
| 151 | + | func mirrorConfigToPB(g domain.MirrorGlobalConfig) *pb.MirrorConfig { | |||
| 152 | + | return &pb.MirrorConfig{Enabled: g.Enabled, UrlTemplate: g.URLTemplate, Cron: g.Cron, AllBranches: g.AllBranches, UpdatedAt: timestamp(g.UpdatedAt)} | |||
| 153 | + | } | |||
| 154 | + | func mirrorStateToPB(s domain.MirrorState) pb.MirrorState { | |||
| 155 | + | switch s { | |||
| 156 | + | case domain.MirrorDisabled: | |||
| 157 | + | return pb.MirrorState_MIRROR_STATE_DISABLED | |||
| 158 | + | case domain.MirrorOK: | |||
| 159 | + | return pb.MirrorState_MIRROR_STATE_OK | |||
| 160 | + | case domain.MirrorFailing: | |||
| 161 | + | return pb.MirrorState_MIRROR_STATE_FAILING | |||
| 162 | + | case domain.MirrorPaused: | |||
| 163 | + | return pb.MirrorState_MIRROR_STATE_PAUSED | |||
| 164 | + | default: | |||
| 165 | + | return pb.MirrorState_MIRROR_STATE_UNSPECIFIED | |||
| 166 | + | } | |||
| 167 | + | } | |||
| 168 | + | func mirrorTargetToPB(t domain.MirrorTarget) *pb.MirrorTarget { | |||
| 169 | + | source := pb.MirrorSource_MIRROR_SOURCE_UNSPECIFIED | |||
| 170 | + | switch t.Source { | |||
| 171 | + | case domain.MirrorSourceNone: | |||
| 172 | + | source = pb.MirrorSource_MIRROR_SOURCE_DISABLED | |||
| 173 | + | case domain.MirrorSourceGlobal: | |||
| 174 | + | source = pb.MirrorSource_MIRROR_SOURCE_GLOBAL | |||
| 175 | + | case domain.MirrorSourceOverride: | |||
| 176 | + | source = pb.MirrorSource_MIRROR_SOURCE_OVERRIDE | |||
| 177 | + | case domain.MirrorSourceExcluded: | |||
| 178 | + | source = pb.MirrorSource_MIRROR_SOURCE_EXCLUDED | |||
| 179 | + | } | |||
| 180 | + | return &pb.MirrorTarget{Repo: t.Repo.String(), Enabled: t.Enabled, Url: t.URL, Cron: t.CronSpec, AllBranches: t.AllBranches, Source: source} | |||
| 181 | + | } | |||
| 182 | + | func mirrorStatusToPB(v ports.MirrorStatusView) *pb.MirrorStatus { | |||
| 183 | + | s := v.Status | |||
| 184 | + | return &pb.MirrorStatus{Target: mirrorTargetToPB(v.Target), State: mirrorStateToPB(s.State), FailureClass: string(s.Class), Reason: s.Reason, Detail: s.Detail, ConsecutiveFailures: int32(s.ConsecutiveFailures), LastAttemptAt: timestamp(s.LastAttemptAt), LastSuccessAt: timestamp(s.LastSuccessAt), NextRetryAt: timestamp(s.NextRetryAt), NextRunAt: timestamp(v.NextRunAt), PausedAt: timestamp(s.PausedAt)} | |||
| 185 | + | } | |||
A internal/adapters/rpc/mirror_test.go +72 -0
| @@ -0,0 +1,72 @@ | |||
| 1 | + | // SPDX-License-Identifier: AGPL-3.0-or-later | |
| 2 | + | ||
| 3 | + | package rpc | |
| 4 | + | ||
| 5 | + | import ( | |
| 6 | + | "context" | |
| 7 | + | "testing" | |
| 8 | + | "time" | |
| 9 | + | ||
| 10 | + | pb "bnema.dev/ermite-api/gen/ermite" | |
| 11 | + | "bnema.dev/ermite/internal/domain" | |
| 12 | + | "bnema.dev/ermite/internal/ports" | |
| 13 | + | "bnema.dev/ermite/internal/ports/mocks" | |
| 14 | + | "connectrpc.com/connect" | |
| 15 | + | "github.com/stretchr/testify/assert" | |
| 16 | + | "github.com/stretchr/testify/mock" | |
| 17 | + | "github.com/stretchr/testify/require" | |
| 18 | + | ) | |
| 19 | + | ||
| 20 | + | func TestMirrorConfigPatchPresence(t *testing.T) { | |
| 21 | + | service := mocks.NewMirrorService(t) | |
| 22 | + | disabled := false | |
| 23 | + | empty := "" | |
| 24 | + | patch := ports.MirrorGlobalPatch{Enabled: &disabled, URLTemplate: &empty} | |
| 25 | + | service.EXPECT().SetGlobalConfig(mock.Anything, domain.Principal{}, patch).Return(domain.MirrorGlobalConfig{}, nil).Once() | |
| 26 | + | s := &mirrorService{service} | |
| 27 | + | res, err := s.SetMirrorConfig(context.Background(), connect.NewRequest(&pb.SetMirrorConfigRequest{Enabled: &disabled, UrlTemplate: &empty})) | |
| 28 | + | require.NoError(t, err) | |
| 29 | + | assert.False(t, res.Msg.GetConfig().GetEnabled()) | |
| 30 | + | } | |
| 31 | + | ||
| 32 | + | func TestMirrorSyncSelector(t *testing.T) { | |
| 33 | + | s := &mirrorService{mocks.NewMirrorService(t)} | |
| 34 | + | for _, req := range []*pb.SyncMirrorRequest{{}, {Repo: "repo", All: true}, {Repo: "../invalid"}} { | |
| 35 | + | _, err := s.SyncMirror(context.Background(), connect.NewRequest(req)) | |
| 36 | + | assert.Equal(t, connect.CodeInvalidArgument, connect.CodeOf(err)) | |
| 37 | + | } | |
| 38 | + | } | |
| 39 | + | ||
| 40 | + | func TestMirrorSyncResponse(t *testing.T) { | |
| 41 | + | service := mocks.NewMirrorService(t) | |
| 42 | + | name, err := domain.ParseRepoName("repo") | |
| 43 | + | require.NoError(t, err) | |
| 44 | + | status := domain.MirrorStatus{Repo: name, State: domain.MirrorPaused, Class: domain.MirrorClassDiverged, Reason: "remote diverged", ConsecutiveFailures: 1} | |
| 45 | + | target := domain.MirrorTarget{Repo: name, Enabled: true, Source: domain.MirrorSourceOverride} | |
| 46 | + | service.EXPECT().Sync(mock.Anything, domain.Principal{}, name, true).Return(ports.MirrorSyncResult{Status: status}, nil).Once() | |
| 47 | + | service.EXPECT().GetStatus(mock.Anything, domain.Principal{}, name).Return(ports.MirrorStatusView{Target: target, Status: status}, nil).Once() | |
| 48 | + | s := &mirrorService{service} | |
| 49 | + | res, err := s.SyncMirror(context.Background(), connect.NewRequest(&pb.SyncMirrorRequest{Repo: "repo", Force: true})) | |
| 50 | + | require.NoError(t, err) | |
| 51 | + | require.Len(t, res.Msg.Results, 1) | |
| 52 | + | assert.Equal(t, "repo", res.Msg.Results[0].Repo) | |
| 53 | + | assert.Equal(t, pb.MirrorState_MIRROR_STATE_PAUSED, res.Msg.Results[0].Status.State) | |
| 54 | + | assert.Equal(t, "diverged", res.Msg.Results[0].Status.FailureClass) | |
| 55 | + | } | |
| 56 | + | ||
| 57 | + | func TestMirrorEvents(t *testing.T) { | |
| 58 | + | name, err := domain.ParseRepoName("repo") | |
| 59 | + | require.NoError(t, err) | |
| 60 | + | now := time.Now().UTC() | |
| 61 | + | failed := watchEventToPB(ports.WatchEvent{Event: domain.MirrorFailed{Repo: name, State: domain.MirrorPaused, Class: domain.MirrorClassDiverged, Reason: "remote diverged", At: now}}) | |
| 62 | + | require.NotNil(t, failed) | |
| 63 | + | assert.Equal(t, now, failed.Time.AsTime()) | |
| 64 | + | assert.Equal(t, &pb.MirrorFailed{Repo: "repo", State: pb.MirrorState_MIRROR_STATE_PAUSED, FailureClass: "diverged", Reason: "remote diverged"}, failed.GetMirrorFailed()) | |
| 65 | + | recovered := watchEventToPB(ports.WatchEvent{Event: domain.MirrorRecovered{Repo: name, At: now}}) | |
| 66 | + | require.NotNil(t, recovered) | |
| 67 | + | assert.Equal(t, "repo", recovered.GetMirrorRecovered().Repo) | |
| 68 | + | // The public event messages deliberately expose neither URL nor stderr detail. | |
| 69 | + | fields := failed.GetMirrorFailed().ProtoReflect().Descriptor().Fields() | |
| 70 | + | assert.Nil(t, fields.ByName("url")) | |
| 71 | + | assert.Nil(t, fields.ByName("detail")) | |
| 72 | + | } | |
| @@ -0,0 +1,72 @@ | |||||
| 1 | + | // SPDX-License-Identifier: AGPL-3.0-or-later | |||
| 2 | + | ||||
| 3 | + | package rpc | |||
| 4 | + | ||||
| 5 | + | import ( | |||
| 6 | + | "context" | |||
| 7 | + | "testing" | |||
| 8 | + | "time" | |||
| 9 | + | ||||
| 10 | + | pb "bnema.dev/ermite-api/gen/ermite" | |||
| 11 | + | "bnema.dev/ermite/internal/domain" | |||
| 12 | + | "bnema.dev/ermite/internal/ports" | |||
| 13 | + | "bnema.dev/ermite/internal/ports/mocks" | |||
| 14 | + | "connectrpc.com/connect" | |||
| 15 | + | "github.com/stretchr/testify/assert" | |||
| 16 | + | "github.com/stretchr/testify/mock" | |||
| 17 | + | "github.com/stretchr/testify/require" | |||
| 18 | + | ) | |||
| 19 | + | ||||
| 20 | + | func TestMirrorConfigPatchPresence(t *testing.T) { | |||
| 21 | + | service := mocks.NewMirrorService(t) | |||
| 22 | + | disabled := false | |||
| 23 | + | empty := "" | |||
| 24 | + | patch := ports.MirrorGlobalPatch{Enabled: &disabled, URLTemplate: &empty} | |||
| 25 | + | service.EXPECT().SetGlobalConfig(mock.Anything, domain.Principal{}, patch).Return(domain.MirrorGlobalConfig{}, nil).Once() | |||
| 26 | + | s := &mirrorService{service} | |||
| 27 | + | res, err := s.SetMirrorConfig(context.Background(), connect.NewRequest(&pb.SetMirrorConfigRequest{Enabled: &disabled, UrlTemplate: &empty})) | |||
| 28 | + | require.NoError(t, err) | |||
| 29 | + | assert.False(t, res.Msg.GetConfig().GetEnabled()) | |||
| 30 | + | } | |||
| 31 | + | ||||
| 32 | + | func TestMirrorSyncSelector(t *testing.T) { | |||
| 33 | + | s := &mirrorService{mocks.NewMirrorService(t)} | |||
| 34 | + | for _, req := range []*pb.SyncMirrorRequest{{}, {Repo: "repo", All: true}, {Repo: "../invalid"}} { | |||
| 35 | + | _, err := s.SyncMirror(context.Background(), connect.NewRequest(req)) | |||
| 36 | + | assert.Equal(t, connect.CodeInvalidArgument, connect.CodeOf(err)) | |||
| 37 | + | } | |||
| 38 | + | } | |||
| 39 | + | ||||
| 40 | + | func TestMirrorSyncResponse(t *testing.T) { | |||
| 41 | + | service := mocks.NewMirrorService(t) | |||
| 42 | + | name, err := domain.ParseRepoName("repo") | |||
| 43 | + | require.NoError(t, err) | |||
| 44 | + | status := domain.MirrorStatus{Repo: name, State: domain.MirrorPaused, Class: domain.MirrorClassDiverged, Reason: "remote diverged", ConsecutiveFailures: 1} | |||
| 45 | + | target := domain.MirrorTarget{Repo: name, Enabled: true, Source: domain.MirrorSourceOverride} | |||
| 46 | + | service.EXPECT().Sync(mock.Anything, domain.Principal{}, name, true).Return(ports.MirrorSyncResult{Status: status}, nil).Once() | |||
| 47 | + | service.EXPECT().GetStatus(mock.Anything, domain.Principal{}, name).Return(ports.MirrorStatusView{Target: target, Status: status}, nil).Once() | |||
| 48 | + | s := &mirrorService{service} | |||
| 49 | + | res, err := s.SyncMirror(context.Background(), connect.NewRequest(&pb.SyncMirrorRequest{Repo: "repo", Force: true})) | |||
| 50 | + | require.NoError(t, err) | |||
| 51 | + | require.Len(t, res.Msg.Results, 1) | |||
| 52 | + | assert.Equal(t, "repo", res.Msg.Results[0].Repo) | |||
| 53 | + | assert.Equal(t, pb.MirrorState_MIRROR_STATE_PAUSED, res.Msg.Results[0].Status.State) | |||
| 54 | + | assert.Equal(t, "diverged", res.Msg.Results[0].Status.FailureClass) | |||
| 55 | + | } | |||
| 56 | + | ||||
| 57 | + | func TestMirrorEvents(t *testing.T) { | |||
| 58 | + | name, err := domain.ParseRepoName("repo") | |||
| 59 | + | require.NoError(t, err) | |||
| 60 | + | now := time.Now().UTC() | |||
| 61 | + | failed := watchEventToPB(ports.WatchEvent{Event: domain.MirrorFailed{Repo: name, State: domain.MirrorPaused, Class: domain.MirrorClassDiverged, Reason: "remote diverged", At: now}}) | |||
| 62 | + | require.NotNil(t, failed) | |||
| 63 | + | assert.Equal(t, now, failed.Time.AsTime()) | |||
| 64 | + | assert.Equal(t, &pb.MirrorFailed{Repo: "repo", State: pb.MirrorState_MIRROR_STATE_PAUSED, FailureClass: "diverged", Reason: "remote diverged"}, failed.GetMirrorFailed()) | |||
| 65 | + | recovered := watchEventToPB(ports.WatchEvent{Event: domain.MirrorRecovered{Repo: name, At: now}}) | |||
| 66 | + | require.NotNil(t, recovered) | |||
| 67 | + | assert.Equal(t, "repo", recovered.GetMirrorRecovered().Repo) | |||
| 68 | + | // The public event messages deliberately expose neither URL nor stderr detail. | |||
| 69 | + | fields := failed.GetMirrorFailed().ProtoReflect().Descriptor().Fields() | |||
| 70 | + | assert.Nil(t, fields.ByName("url")) | |||
| 71 | + | assert.Nil(t, fields.ByName("detail")) | |||
| 72 | + | } | |||
M internal/usecase/mirrors_sync.go +1 -1
| @@ -37,7 +37,7 @@ func (m *Mirrors) Sync(ctx context.Context, p domain.Principal, repo domain.Repo | |||
| 37 | 37 | return ports.MirrorSyncResult{}, err | |
| 38 | 38 | } | |
| 39 | 39 | if queued { | |
| 40 | - | return ports.MirrorSyncResult{Queued: true}, nil | |
| 40 | + | return ports.MirrorSyncResult{Status: domain.MirrorStatus{Repo: repo}, Queued: true}, nil | |
| 41 | 41 | } | |
| 42 | 42 | select { | |
| 43 | 43 | case <-m.ctx.Done(): | |
| @@ -37,7 +37,7 @@ func (m *Mirrors) Sync(ctx context.Context, p domain.Principal, repo domain.Repo | |||||
| 37 | return ports.MirrorSyncResult{}, err | 37 | return ports.MirrorSyncResult{}, err | ||
| 38 | } | 38 | } | ||
| 39 | if queued { | 39 | if queued { | ||
| 40 | - | return ports.MirrorSyncResult{Queued: true}, nil | 40 | + | return ports.MirrorSyncResult{Status: domain.MirrorStatus{Repo: repo}, Queued: true}, nil |
| 41 | } | 41 | } | ||
| 42 | select { | 42 | select { | ||
| 43 | case <-m.ctx.Done(): | 43 | case <-m.ctx.Done(): | ||
M internal/usecase/mirrors_test.go +1 -0
| @@ -136,6 +136,7 @@ func TestMirrorsSingleFlight(t *testing.T) { | |||
| 136 | 136 | v, err := f.m.Sync(context.Background(), domain.Root(), f.repo, true) | |
| 137 | 137 | require.NoError(t, err) | |
| 138 | 138 | require.True(t, v.Queued) | |
| 139 | + | require.Equal(t, f.repo, v.Status.Repo) | |
| 139 | 140 | }() | |
| 140 | 141 | } | |
| 141 | 142 | wg.Wait() | |
| @@ -136,6 +136,7 @@ func TestMirrorsSingleFlight(t *testing.T) { | |||||
| 136 | v, err := f.m.Sync(context.Background(), domain.Root(), f.repo, true) | 136 | v, err := f.m.Sync(context.Background(), domain.Root(), f.repo, true) | ||
| 137 | require.NoError(t, err) | 137 | require.NoError(t, err) | ||
| 138 | require.True(t, v.Queued) | 138 | require.True(t, v.Queued) | ||
| 139 | + | require.Equal(t, f.repo, v.Status.Repo) | |||
| 139 | }() | 140 | }() | ||
| 140 | } | 141 | } | ||
| 141 | wg.Wait() | 142 | wg.Wait() | ||