detached 26e2b66

merge: push mirror use case into feat/push-mirror

commit
26e2b668f55ef2de94314b6cbc46849e51178168
author
brice · · 2026-10-20 05:43 UTC
committer
brice · · 2026-10-20 10:02 UTC
merge of
3fb99b3 + a40b26f · diff against the first parent
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
8 files · +1260 -0
Unified
UnifiedSplit
M internal/usecase/events.go +8 -0
@@ -167,6 +167,10 @@ func (e *Events) snapshot(ctx context.Context, ev domain.Event) bufferedEvent {
167167 return bufferedEvent{known: true, vis: ev.Visibility, created: ev.CreatedAt}
168168 case domain.RepoDeleted:
169169 return bufferedEvent{known: true, vis: ev.Visibility, created: ev.CreatedAt}
170+ case domain.MirrorFailed:
171+ return repoSnapshot(ev.Visibility, ev.CreatedAt)
172+ case domain.MirrorRecovered:
173+ return repoSnapshot(ev.Visibility, ev.CreatedAt)
170174 case domain.PushEvent:
171175 if ev.CreatedAt.IsZero() {
172176 return bufferedEvent{} // repository unknown at publish time: nobody is told
@@ -517,6 +521,10 @@ func (e *Events) deliver(ctx context.Context, w *watch, be bufferedEvent, send f
517521 return e.deliverRepo(ctx, w, ev.Repo, true, be, out, send)
518522 case domain.RepoUpdated:
519523 return e.deliverRepo(ctx, w, ev.Repo, true, be, out, send)
524+ case domain.MirrorFailed:
525+ return e.deliverRepo(ctx, w, ev.Repo, false, be, out, send)
526+ case domain.MirrorRecovered:
527+ return e.deliverRepo(ctx, w, ev.Repo, false, be, out, send)
520528 case domain.PushEvent:
521529 return e.deliverRepo(ctx, w, ev.Repo, false, be, out, send)
522530 case domain.MergeRequestOpened:
@@ -167,6 +167,10 @@ func (e *Events) snapshot(ctx context.Context, ev domain.Event) bufferedEvent {
167 return bufferedEvent{known: true, vis: ev.Visibility, created: ev.CreatedAt}167 return bufferedEvent{known: true, vis: ev.Visibility, created: ev.CreatedAt}
168 case domain.RepoDeleted:168 case domain.RepoDeleted:
169 return bufferedEvent{known: true, vis: ev.Visibility, created: ev.CreatedAt}169 return bufferedEvent{known: true, vis: ev.Visibility, created: ev.CreatedAt}
170+ case domain.MirrorFailed:
171+ return repoSnapshot(ev.Visibility, ev.CreatedAt)
172+ case domain.MirrorRecovered:
173+ return repoSnapshot(ev.Visibility, ev.CreatedAt)
170 case domain.PushEvent:174 case domain.PushEvent:
171 if ev.CreatedAt.IsZero() {175 if ev.CreatedAt.IsZero() {
172 return bufferedEvent{} // repository unknown at publish time: nobody is told176 return bufferedEvent{} // repository unknown at publish time: nobody is told
@@ -517,6 +521,10 @@ func (e *Events) deliver(ctx context.Context, w *watch, be bufferedEvent, send f
517 return e.deliverRepo(ctx, w, ev.Repo, true, be, out, send)521 return e.deliverRepo(ctx, w, ev.Repo, true, be, out, send)
518 case domain.RepoUpdated:522 case domain.RepoUpdated:
519 return e.deliverRepo(ctx, w, ev.Repo, true, be, out, send)523 return e.deliverRepo(ctx, w, ev.Repo, true, be, out, send)
524+ case domain.MirrorFailed:
525+ return e.deliverRepo(ctx, w, ev.Repo, false, be, out, send)
526+ case domain.MirrorRecovered:
527+ return e.deliverRepo(ctx, w, ev.Repo, false, be, out, send)
520 case domain.PushEvent:528 case domain.PushEvent:
521 return e.deliverRepo(ctx, w, ev.Repo, false, be, out, send)529 return e.deliverRepo(ctx, w, ev.Repo, false, be, out, send)
522 case domain.MergeRequestOpened:530 case domain.MergeRequestOpened:
A internal/usecase/mirrors.go +73 -0
@@ -0,0 +1,73 @@
1+// SPDX-License-Identifier: AGPL-3.0-or-later
2+
3+package usecase
4+
5+import (
6+ "context"
7+ "crypto/ed25519"
8+ "sync"
9+ "time"
10+
11+ "bnema.dev/ermite/internal/domain"
12+ "bnema.dev/ermite/internal/ports"
13+)
14+
15+// MirrorsConfig supplies the mirror use case's dependencies.
16+type MirrorsConfig struct {
17+ Store ports.MirrorStore
18+ Keys ports.MirrorKeyStore
19+ Sealer ports.SecretSealer
20+ Pusher ports.MirrorPusher
21+ Meta ports.RepoMetaStore
22+ Git ports.GitRepos
23+ Reader ports.RepoReader
24+ Bus ports.EventBus
25+ Clock ports.Clock
26+ PublicURL string
27+ ScheduleTick time.Duration
28+ Workers int
29+ AllowAnyURL bool
30+}
31+
32+// Mirrors manages outbound repository mirrors.
33+type Mirrors struct {
34+ cfg MirrorsConfig
35+ configMu sync.RWMutex
36+ mu sync.Mutex
37+ dirty map[domain.RepoName]uint64
38+ clean map[domain.RepoName]uint64
39+ flights map[domain.RepoName]*mirrorFlight
40+ ctx context.Context
41+ cancel context.CancelFunc
42+ wg sync.WaitGroup
43+ start sync.Once
44+ closed bool
45+ jobs chan domain.RepoName
46+ scheduleMu sync.Mutex
47+ lastMinute time.Time
48+ keyMu sync.Mutex
49+ key ed25519.PrivateKey
50+ keyCreated time.Time
51+}
52+
53+// NewMirrors builds the mirror use case.
54+func NewMirrors(cfg MirrorsConfig) *Mirrors {
55+ if cfg.ScheduleTick <= 0 {
56+ cfg.ScheduleTick = time.Second
57+ }
58+ if cfg.Workers <= 0 {
59+ cfg.Workers = 2
60+ }
61+ ctx, cancel := context.WithCancel(context.Background())
62+ return &Mirrors{cfg: cfg, ctx: ctx, cancel: cancel, dirty: make(map[domain.RepoName]uint64), clean: make(map[domain.RepoName]uint64), flights: make(map[domain.RepoName]*mirrorFlight), jobs: make(chan domain.RepoName, 64), lastMinute: cfg.Clock.Now().UTC().Truncate(time.Minute)}
63+}
64+
65+// privateKey returns a copy so a transport cannot mutate the cached key.
66+func (m *Mirrors) privateKey(ctx context.Context) (ed25519.PrivateKey, error) {
67+ m.keyMu.Lock()
68+ defer m.keyMu.Unlock()
69+ if err := m.loadKey(ctx, false); err != nil {
70+ return nil, err
71+ }
72+ return append(ed25519.PrivateKey(nil), m.key...), nil
73+}
@@ -0,0 +1,73 @@
1+// SPDX-License-Identifier: AGPL-3.0-or-later
2+
3+package usecase
4+
5+import (
6+ "context"
7+ "crypto/ed25519"
8+ "sync"
9+ "time"
10+
11+ "bnema.dev/ermite/internal/domain"
12+ "bnema.dev/ermite/internal/ports"
13+)
14+
15+// MirrorsConfig supplies the mirror use case's dependencies.
16+type MirrorsConfig struct {
17+ Store ports.MirrorStore
18+ Keys ports.MirrorKeyStore
19+ Sealer ports.SecretSealer
20+ Pusher ports.MirrorPusher
21+ Meta ports.RepoMetaStore
22+ Git ports.GitRepos
23+ Reader ports.RepoReader
24+ Bus ports.EventBus
25+ Clock ports.Clock
26+ PublicURL string
27+ ScheduleTick time.Duration
28+ Workers int
29+ AllowAnyURL bool
30+}
31+
32+// Mirrors manages outbound repository mirrors.
33+type Mirrors struct {
34+ cfg MirrorsConfig
35+ configMu sync.RWMutex
36+ mu sync.Mutex
37+ dirty map[domain.RepoName]uint64
38+ clean map[domain.RepoName]uint64
39+ flights map[domain.RepoName]*mirrorFlight
40+ ctx context.Context
41+ cancel context.CancelFunc
42+ wg sync.WaitGroup
43+ start sync.Once
44+ closed bool
45+ jobs chan domain.RepoName
46+ scheduleMu sync.Mutex
47+ lastMinute time.Time
48+ keyMu sync.Mutex
49+ key ed25519.PrivateKey
50+ keyCreated time.Time
51+}
52+
53+// NewMirrors builds the mirror use case.
54+func NewMirrors(cfg MirrorsConfig) *Mirrors {
55+ if cfg.ScheduleTick <= 0 {
56+ cfg.ScheduleTick = time.Second
57+ }
58+ if cfg.Workers <= 0 {
59+ cfg.Workers = 2
60+ }
61+ ctx, cancel := context.WithCancel(context.Background())
62+ return &Mirrors{cfg: cfg, ctx: ctx, cancel: cancel, dirty: make(map[domain.RepoName]uint64), clean: make(map[domain.RepoName]uint64), flights: make(map[domain.RepoName]*mirrorFlight), jobs: make(chan domain.RepoName, 64), lastMinute: cfg.Clock.Now().UTC().Truncate(time.Minute)}
63+}
64+
65+// privateKey returns a copy so a transport cannot mutate the cached key.
66+func (m *Mirrors) privateKey(ctx context.Context) (ed25519.PrivateKey, error) {
67+ m.keyMu.Lock()
68+ defer m.keyMu.Unlock()
69+ if err := m.loadKey(ctx, false); err != nil {
70+ return nil, err
71+ }
72+ return append(ed25519.PrivateKey(nil), m.key...), nil
73+}
A internal/usecase/mirrors_config.go +293 -0
@@ -0,0 +1,293 @@
1+// SPDX-License-Identifier: AGPL-3.0-or-later
2+
3+package usecase
4+
5+import (
6+ "context"
7+ "errors"
8+ "strings"
9+ "time"
10+
11+ "bnema.dev/ermite/internal/domain"
12+ "bnema.dev/ermite/internal/ports"
13+)
14+
15+func mirrorOwner(p domain.Principal) error {
16+ if p.IsAnonymous() {
17+ return domain.ErrUnauthenticated
18+ }
19+ if !p.HasOwnerRights() {
20+ return domain.ErrForbidden
21+ }
22+ return nil
23+}
24+func (m *Mirrors) admin(ctx context.Context, p domain.Principal, repo domain.RepoName) error {
25+ if p.IsAnonymous() {
26+ return domain.ErrUnauthenticated
27+ }
28+ r, err := m.cfg.Meta.Get(ctx, repo)
29+ if err != nil {
30+ return err
31+ }
32+ if !p.Allows(domain.ScopeRepoAdmin, repo) {
33+ if !canRead(p, r) {
34+ return domain.ErrNotFound
35+ }
36+ return domain.ErrForbidden
37+ }
38+ return nil
39+}
40+func (m *Mirrors) target(ctx context.Context, repo domain.RepoName) (domain.MirrorTarget, error) {
41+ g, err := m.cfg.Store.GetGlobal(ctx)
42+ if err != nil {
43+ return domain.MirrorTarget{}, err
44+ }
45+ r, err := m.cfg.Store.GetRepoConfig(ctx, repo)
46+ if errors.Is(err, domain.ErrNotFound) {
47+ return m.resolve(g, nil, repo)
48+ }
49+ if err != nil {
50+ return domain.MirrorTarget{}, err
51+ }
52+ return m.resolve(g, &r, repo)
53+}
54+func (m *Mirrors) status(ctx context.Context, repo domain.RepoName) (domain.MirrorStatus, error) {
55+ s, err := m.cfg.Store.GetStatus(ctx, repo)
56+ if errors.Is(err, domain.ErrNotFound) {
57+ return domain.MirrorStatus{Repo: repo, State: domain.MirrorDisabled}, nil
58+ }
59+ return s, err
60+}
61+
62+// GetGlobalConfig returns server-wide policy to an owner.
63+func (m *Mirrors) GetGlobalConfig(ctx context.Context, p domain.Principal) (domain.MirrorGlobalConfig, error) {
64+ if err := mirrorOwner(p); err != nil {
65+ return domain.MirrorGlobalConfig{}, err
66+ }
67+ return m.cfg.Store.GetGlobal(ctx)
68+}
69+
70+// SetGlobalConfig patches server-wide policy and resets affected mirror incidents.
71+func (m *Mirrors) SetGlobalConfig(ctx context.Context, p domain.Principal, patch ports.MirrorGlobalPatch) (domain.MirrorGlobalConfig, error) {
72+ if err := mirrorOwner(p); err != nil {
73+ return domain.MirrorGlobalConfig{}, err
74+ }
75+ m.configMu.Lock()
76+ defer m.configMu.Unlock()
77+ g, err := m.cfg.Store.GetGlobal(ctx)
78+ if err != nil {
79+ return g, err
80+ }
81+ if patch.Enabled != nil {
82+ g.Enabled = *patch.Enabled
83+ }
84+ if patch.URLTemplate != nil {
85+ g.URLTemplate = *patch.URLTemplate
86+ }
87+ if patch.Cron != nil {
88+ g.Cron = *patch.Cron
89+ }
90+ if patch.AllBranches != nil {
91+ g.AllBranches = *patch.AllBranches
92+ }
93+ if err := m.validateGlobal(g); err != nil {
94+ return g, err
95+ }
96+ g.UpdatedAt = m.cfg.Clock.Now().UTC()
97+ if err := m.cfg.Store.PutGlobal(ctx, g); err != nil {
98+ return g, err
99+ }
100+ repos, err := m.cfg.Meta.List(ctx)
101+ if err != nil {
102+ return g, err
103+ }
104+ for _, r := range repos {
105+ c, e := m.cfg.Store.GetRepoConfig(ctx, r.Name)
106+ if e != nil && !errors.Is(e, domain.ErrNotFound) {
107+ return g, e
108+ }
109+ if c.URL != "" {
110+ continue
111+ }
112+ if err := m.reset(ctx, r.Name); err != nil {
113+ return g, err
114+ }
115+ }
116+ return g, nil
117+}
118+func (m *Mirrors) reset(ctx context.Context, repo domain.RepoName) error {
119+ t, err := m.target(ctx, repo)
120+ if err != nil {
121+ return err
122+ }
123+ s, err := m.status(ctx, repo)
124+ if err != nil {
125+ return err
126+ }
127+ now := m.cfg.Clock.Now().UTC()
128+ if t.Enabled {
129+ s = domain.MirrorReset(s, now)
130+ } else {
131+ s = domain.MirrorDisable(s, now)
132+ }
133+ m.mu.Lock()
134+ m.dirty[repo]++
135+ m.mu.Unlock()
136+ return m.cfg.Store.PutStatus(ctx, s)
137+}
138+
139+// SetRepoMirror sets a repository override.
140+func (m *Mirrors) SetRepoMirror(ctx context.Context, p domain.Principal, c domain.MirrorRepoConfig) (domain.MirrorTarget, error) {
141+ if err := m.admin(ctx, p, c.Repo); err != nil {
142+ return domain.MirrorTarget{}, err
143+ }
144+ if err := m.validateRepo(c); err != nil {
145+ return domain.MirrorTarget{}, err
146+ }
147+ m.configMu.Lock()
148+ defer m.configMu.Unlock()
149+ c.UpdatedAt = m.cfg.Clock.Now().UTC()
150+ if err := m.cfg.Store.PutRepoConfig(ctx, c); err != nil {
151+ return domain.MirrorTarget{}, err
152+ }
153+ if err := m.reset(ctx, c.Repo); err != nil {
154+ return domain.MirrorTarget{}, err
155+ }
156+ return m.target(ctx, c.Repo)
157+}
158+
159+// RemoveRepoMirror restores inherited policy.
160+func (m *Mirrors) RemoveRepoMirror(ctx context.Context, p domain.Principal, repo domain.RepoName) error {
161+ if err := m.admin(ctx, p, repo); err != nil {
162+ return err
163+ }
164+ m.configMu.Lock()
165+ defer m.configMu.Unlock()
166+ if err := m.cfg.Store.DeleteRepoConfig(ctx, repo); err != nil {
167+ return err
168+ }
169+ return m.reset(ctx, repo)
170+}
171+
172+// SetRepoExcluded updates exclusion without discarding an override.
173+func (m *Mirrors) SetRepoExcluded(ctx context.Context, p domain.Principal, repo domain.RepoName, excluded bool) (domain.MirrorTarget, error) {
174+ if err := m.admin(ctx, p, repo); err != nil {
175+ return domain.MirrorTarget{}, err
176+ }
177+ c, err := m.cfg.Store.GetRepoConfig(ctx, repo)
178+ if err != nil && !errors.Is(err, domain.ErrNotFound) {
179+ return domain.MirrorTarget{}, err
180+ }
181+ c.Repo, c.Excluded = repo, excluded
182+ return m.SetRepoMirror(ctx, p, c)
183+}
184+
185+// GetStatus returns a read-authorized status, redacting administrative diagnostics.
186+func (m *Mirrors) GetStatus(ctx context.Context, p domain.Principal, repo domain.RepoName) (ports.MirrorStatusView, error) {
187+ r, err := m.cfg.Meta.Get(ctx, repo)
188+ if err != nil {
189+ return ports.MirrorStatusView{}, err
190+ }
191+ if !canRead(p, r) {
192+ return ports.MirrorStatusView{}, domain.ErrNotFound
193+ }
194+ t, err := m.target(ctx, repo)
195+ if err != nil {
196+ return ports.MirrorStatusView{}, err
197+ }
198+ s, err := m.status(ctx, repo)
199+ if err != nil {
200+ return ports.MirrorStatusView{}, err
201+ }
202+ var next time.Time
203+ if t.Enabled && s.State != domain.MirrorPaused {
204+ if s.State == domain.MirrorFailing {
205+ next = s.NextRetryAt
206+ } else {
207+ next, _ = domain.NextCron(domain.CronSchedule{Cron: t.Cron, Loc: time.UTC}, m.cfg.Clock.Now())
208+ }
209+ }
210+ if !p.Allows(domain.ScopeRepoAdmin, repo) {
211+ s.Detail, s.Reason = "", ""
212+ t.URL = ""
213+ }
214+ return ports.MirrorStatusView{Target: t, Status: s, NextRunAt: next}, nil
215+}
216+
217+// ListStatus returns all statuses to an owner.
218+func (m *Mirrors) ListStatus(ctx context.Context, p domain.Principal) ([]ports.MirrorStatusView, error) {
219+ if err := mirrorOwner(p); err != nil {
220+ return nil, err
221+ }
222+ repos, err := m.cfg.Meta.List(ctx)
223+ if err != nil {
224+ return nil, err
225+ }
226+ out := make([]ports.MirrorStatusView, 0, len(repos))
227+ for _, r := range repos {
228+ v, err := m.GetStatus(ctx, p, r.Name)
229+ if err != nil {
230+ return nil, err
231+ }
232+ out = append(out, v)
233+ }
234+ return out, nil
235+}
236+
237+// Warning provides an administrative post-receive diagnostic.
238+func (m *Mirrors) Warning(ctx context.Context, repo domain.RepoName) string {
239+ s, err := m.status(ctx, repo)
240+ if err != nil || (s.State != domain.MirrorFailing && s.State != domain.MirrorPaused) {
241+ return ""
242+ }
243+ t, err := m.target(ctx, repo)
244+ if err != nil {
245+ return ""
246+ }
247+ return "ermite: mirror to " + domain.RedactMirrorURL(t.URL) + " is " + string(s.State) + ": " + s.Reason
248+}
249+
250+// Badge returns a viewer-safe state and optional administrative reason.
251+func (m *Mirrors) Badge(ctx context.Context, p domain.Principal, repo domain.RepoName) (domain.MirrorState, string) {
252+ v, err := m.GetStatus(ctx, p, repo)
253+ if err != nil {
254+ return domain.MirrorDisabled, ""
255+ }
256+ return v.Status.State, v.Status.Reason
257+}
258+
259+// testURL substitutes only during validation/resolution; stored and pushed URLs remain intact.
260+func (m *Mirrors) testURL(url string) string {
261+ if m.cfg.AllowAnyURL && strings.HasPrefix(url, "file://") {
262+ return "git@mirror-test:" + strings.TrimPrefix(url, "file://")
263+ }
264+ return url
265+}
266+func (m *Mirrors) validateGlobal(g domain.MirrorGlobalConfig) error {
267+ g.URLTemplate = m.testURL(g.URLTemplate)
268+ return domain.ValidateMirrorGlobalConfig(g)
269+}
270+func (m *Mirrors) validateRepo(c domain.MirrorRepoConfig) error {
271+ c.URL = m.testURL(c.URL)
272+ return domain.ValidateMirrorRepoConfig(c)
273+}
274+func (m *Mirrors) resolve(g domain.MirrorGlobalConfig, c *domain.MirrorRepoConfig, repo domain.RepoName) (domain.MirrorTarget, error) {
275+ originalGlobal := g.URLTemplate
276+ originalRepo := ""
277+ g.URLTemplate = m.testURL(g.URLTemplate)
278+ if c != nil {
279+ copyConfig := *c
280+ originalRepo = c.URL
281+ copyConfig.URL = m.testURL(c.URL)
282+ c = &copyConfig
283+ }
284+ t, err := domain.ResolveMirror(g, c, repo)
285+ if err == nil && t.Enabled && m.cfg.AllowAnyURL {
286+ if originalRepo != "" {
287+ t.URL = originalRepo
288+ } else if strings.HasPrefix(originalGlobal, "file://") {
289+ t.URL, err = domain.ExpandMirrorURL(originalGlobal, repo)
290+ }
291+ }
292+ return t, err
293+}
@@ -0,0 +1,293 @@
1+// SPDX-License-Identifier: AGPL-3.0-or-later
2+
3+package usecase
4+
5+import (
6+ "context"
7+ "errors"
8+ "strings"
9+ "time"
10+
11+ "bnema.dev/ermite/internal/domain"
12+ "bnema.dev/ermite/internal/ports"
13+)
14+
15+func mirrorOwner(p domain.Principal) error {
16+ if p.IsAnonymous() {
17+ return domain.ErrUnauthenticated
18+ }
19+ if !p.HasOwnerRights() {
20+ return domain.ErrForbidden
21+ }
22+ return nil
23+}
24+func (m *Mirrors) admin(ctx context.Context, p domain.Principal, repo domain.RepoName) error {
25+ if p.IsAnonymous() {
26+ return domain.ErrUnauthenticated
27+ }
28+ r, err := m.cfg.Meta.Get(ctx, repo)
29+ if err != nil {
30+ return err
31+ }
32+ if !p.Allows(domain.ScopeRepoAdmin, repo) {
33+ if !canRead(p, r) {
34+ return domain.ErrNotFound
35+ }
36+ return domain.ErrForbidden
37+ }
38+ return nil
39+}
40+func (m *Mirrors) target(ctx context.Context, repo domain.RepoName) (domain.MirrorTarget, error) {
41+ g, err := m.cfg.Store.GetGlobal(ctx)
42+ if err != nil {
43+ return domain.MirrorTarget{}, err
44+ }
45+ r, err := m.cfg.Store.GetRepoConfig(ctx, repo)
46+ if errors.Is(err, domain.ErrNotFound) {
47+ return m.resolve(g, nil, repo)
48+ }
49+ if err != nil {
50+ return domain.MirrorTarget{}, err
51+ }
52+ return m.resolve(g, &r, repo)
53+}
54+func (m *Mirrors) status(ctx context.Context, repo domain.RepoName) (domain.MirrorStatus, error) {
55+ s, err := m.cfg.Store.GetStatus(ctx, repo)
56+ if errors.Is(err, domain.ErrNotFound) {
57+ return domain.MirrorStatus{Repo: repo, State: domain.MirrorDisabled}, nil
58+ }
59+ return s, err
60+}
61+
62+// GetGlobalConfig returns server-wide policy to an owner.
63+func (m *Mirrors) GetGlobalConfig(ctx context.Context, p domain.Principal) (domain.MirrorGlobalConfig, error) {
64+ if err := mirrorOwner(p); err != nil {
65+ return domain.MirrorGlobalConfig{}, err
66+ }
67+ return m.cfg.Store.GetGlobal(ctx)
68+}
69+
70+// SetGlobalConfig patches server-wide policy and resets affected mirror incidents.
71+func (m *Mirrors) SetGlobalConfig(ctx context.Context, p domain.Principal, patch ports.MirrorGlobalPatch) (domain.MirrorGlobalConfig, error) {
72+ if err := mirrorOwner(p); err != nil {
73+ return domain.MirrorGlobalConfig{}, err
74+ }
75+ m.configMu.Lock()
76+ defer m.configMu.Unlock()
77+ g, err := m.cfg.Store.GetGlobal(ctx)
78+ if err != nil {
79+ return g, err
80+ }
81+ if patch.Enabled != nil {
82+ g.Enabled = *patch.Enabled
83+ }
84+ if patch.URLTemplate != nil {
85+ g.URLTemplate = *patch.URLTemplate
86+ }
87+ if patch.Cron != nil {
88+ g.Cron = *patch.Cron
89+ }
90+ if patch.AllBranches != nil {
91+ g.AllBranches = *patch.AllBranches
92+ }
93+ if err := m.validateGlobal(g); err != nil {
94+ return g, err
95+ }
96+ g.UpdatedAt = m.cfg.Clock.Now().UTC()
97+ if err := m.cfg.Store.PutGlobal(ctx, g); err != nil {
98+ return g, err
99+ }
100+ repos, err := m.cfg.Meta.List(ctx)
101+ if err != nil {
102+ return g, err
103+ }
104+ for _, r := range repos {
105+ c, e := m.cfg.Store.GetRepoConfig(ctx, r.Name)
106+ if e != nil && !errors.Is(e, domain.ErrNotFound) {
107+ return g, e
108+ }
109+ if c.URL != "" {
110+ continue
111+ }
112+ if err := m.reset(ctx, r.Name); err != nil {
113+ return g, err
114+ }
115+ }
116+ return g, nil
117+}
118+func (m *Mirrors) reset(ctx context.Context, repo domain.RepoName) error {
119+ t, err := m.target(ctx, repo)
120+ if err != nil {
121+ return err
122+ }
123+ s, err := m.status(ctx, repo)
124+ if err != nil {
125+ return err
126+ }
127+ now := m.cfg.Clock.Now().UTC()
128+ if t.Enabled {
129+ s = domain.MirrorReset(s, now)
130+ } else {
131+ s = domain.MirrorDisable(s, now)
132+ }
133+ m.mu.Lock()
134+ m.dirty[repo]++
135+ m.mu.Unlock()
136+ return m.cfg.Store.PutStatus(ctx, s)
137+}
138+
139+// SetRepoMirror sets a repository override.
140+func (m *Mirrors) SetRepoMirror(ctx context.Context, p domain.Principal, c domain.MirrorRepoConfig) (domain.MirrorTarget, error) {
141+ if err := m.admin(ctx, p, c.Repo); err != nil {
142+ return domain.MirrorTarget{}, err
143+ }
144+ if err := m.validateRepo(c); err != nil {
145+ return domain.MirrorTarget{}, err
146+ }
147+ m.configMu.Lock()
148+ defer m.configMu.Unlock()
149+ c.UpdatedAt = m.cfg.Clock.Now().UTC()
150+ if err := m.cfg.Store.PutRepoConfig(ctx, c); err != nil {
151+ return domain.MirrorTarget{}, err
152+ }
153+ if err := m.reset(ctx, c.Repo); err != nil {
154+ return domain.MirrorTarget{}, err
155+ }
156+ return m.target(ctx, c.Repo)
157+}
158+
159+// RemoveRepoMirror restores inherited policy.
160+func (m *Mirrors) RemoveRepoMirror(ctx context.Context, p domain.Principal, repo domain.RepoName) error {
161+ if err := m.admin(ctx, p, repo); err != nil {
162+ return err
163+ }
164+ m.configMu.Lock()
165+ defer m.configMu.Unlock()
166+ if err := m.cfg.Store.DeleteRepoConfig(ctx, repo); err != nil {
167+ return err
168+ }
169+ return m.reset(ctx, repo)
170+}
171+
172+// SetRepoExcluded updates exclusion without discarding an override.
173+func (m *Mirrors) SetRepoExcluded(ctx context.Context, p domain.Principal, repo domain.RepoName, excluded bool) (domain.MirrorTarget, error) {
174+ if err := m.admin(ctx, p, repo); err != nil {
175+ return domain.MirrorTarget{}, err
176+ }
177+ c, err := m.cfg.Store.GetRepoConfig(ctx, repo)
178+ if err != nil && !errors.Is(err, domain.ErrNotFound) {
179+ return domain.MirrorTarget{}, err
180+ }
181+ c.Repo, c.Excluded = repo, excluded
182+ return m.SetRepoMirror(ctx, p, c)
183+}
184+
185+// GetStatus returns a read-authorized status, redacting administrative diagnostics.
186+func (m *Mirrors) GetStatus(ctx context.Context, p domain.Principal, repo domain.RepoName) (ports.MirrorStatusView, error) {
187+ r, err := m.cfg.Meta.Get(ctx, repo)
188+ if err != nil {
189+ return ports.MirrorStatusView{}, err
190+ }
191+ if !canRead(p, r) {
192+ return ports.MirrorStatusView{}, domain.ErrNotFound
193+ }
194+ t, err := m.target(ctx, repo)
195+ if err != nil {
196+ return ports.MirrorStatusView{}, err
197+ }
198+ s, err := m.status(ctx, repo)
199+ if err != nil {
200+ return ports.MirrorStatusView{}, err
201+ }
202+ var next time.Time
203+ if t.Enabled && s.State != domain.MirrorPaused {
204+ if s.State == domain.MirrorFailing {
205+ next = s.NextRetryAt
206+ } else {
207+ next, _ = domain.NextCron(domain.CronSchedule{Cron: t.Cron, Loc: time.UTC}, m.cfg.Clock.Now())
208+ }
209+ }
210+ if !p.Allows(domain.ScopeRepoAdmin, repo) {
211+ s.Detail, s.Reason = "", ""
212+ t.URL = ""
213+ }
214+ return ports.MirrorStatusView{Target: t, Status: s, NextRunAt: next}, nil
215+}
216+
217+// ListStatus returns all statuses to an owner.
218+func (m *Mirrors) ListStatus(ctx context.Context, p domain.Principal) ([]ports.MirrorStatusView, error) {
219+ if err := mirrorOwner(p); err != nil {
220+ return nil, err
221+ }
222+ repos, err := m.cfg.Meta.List(ctx)
223+ if err != nil {
224+ return nil, err
225+ }
226+ out := make([]ports.MirrorStatusView, 0, len(repos))
227+ for _, r := range repos {
228+ v, err := m.GetStatus(ctx, p, r.Name)
229+ if err != nil {
230+ return nil, err
231+ }
232+ out = append(out, v)
233+ }
234+ return out, nil
235+}
236+
237+// Warning provides an administrative post-receive diagnostic.
238+func (m *Mirrors) Warning(ctx context.Context, repo domain.RepoName) string {
239+ s, err := m.status(ctx, repo)
240+ if err != nil || (s.State != domain.MirrorFailing && s.State != domain.MirrorPaused) {
241+ return ""
242+ }
243+ t, err := m.target(ctx, repo)
244+ if err != nil {
245+ return ""
246+ }
247+ return "ermite: mirror to " + domain.RedactMirrorURL(t.URL) + " is " + string(s.State) + ": " + s.Reason
248+}
249+
250+// Badge returns a viewer-safe state and optional administrative reason.
251+func (m *Mirrors) Badge(ctx context.Context, p domain.Principal, repo domain.RepoName) (domain.MirrorState, string) {
252+ v, err := m.GetStatus(ctx, p, repo)
253+ if err != nil {
254+ return domain.MirrorDisabled, ""
255+ }
256+ return v.Status.State, v.Status.Reason
257+}
258+
259+// testURL substitutes only during validation/resolution; stored and pushed URLs remain intact.
260+func (m *Mirrors) testURL(url string) string {
261+ if m.cfg.AllowAnyURL && strings.HasPrefix(url, "file://") {
262+ return "git@mirror-test:" + strings.TrimPrefix(url, "file://")
263+ }
264+ return url
265+}
266+func (m *Mirrors) validateGlobal(g domain.MirrorGlobalConfig) error {
267+ g.URLTemplate = m.testURL(g.URLTemplate)
268+ return domain.ValidateMirrorGlobalConfig(g)
269+}
270+func (m *Mirrors) validateRepo(c domain.MirrorRepoConfig) error {
271+ c.URL = m.testURL(c.URL)
272+ return domain.ValidateMirrorRepoConfig(c)
273+}
274+func (m *Mirrors) resolve(g domain.MirrorGlobalConfig, c *domain.MirrorRepoConfig, repo domain.RepoName) (domain.MirrorTarget, error) {
275+ originalGlobal := g.URLTemplate
276+ originalRepo := ""
277+ g.URLTemplate = m.testURL(g.URLTemplate)
278+ if c != nil {
279+ copyConfig := *c
280+ originalRepo = c.URL
281+ copyConfig.URL = m.testURL(c.URL)
282+ c = &copyConfig
283+ }
284+ t, err := domain.ResolveMirror(g, c, repo)
285+ if err == nil && t.Enabled && m.cfg.AllowAnyURL {
286+ if originalRepo != "" {
287+ t.URL = originalRepo
288+ } else if strings.HasPrefix(originalGlobal, "file://") {
289+ t.URL, err = domain.ExpandMirrorURL(originalGlobal, repo)
290+ }
291+ }
292+ return t, err
293+}
A internal/usecase/mirrors_key.go +85 -0
@@ -0,0 +1,85 @@
1+// SPDX-License-Identifier: AGPL-3.0-or-later
2+
3+package usecase
4+
5+import (
6+ "context"
7+ "crypto/ed25519"
8+ "crypto/rand"
9+ "errors"
10+ "fmt"
11+ "strings"
12+
13+ "bnema.dev/ermite/internal/domain"
14+)
15+
16+// ShowKey lazily generates the server's mirror key for authenticated callers.
17+func (m *Mirrors) ShowKey(ctx context.Context, p domain.Principal) (domain.MirrorPublicKey, error) {
18+ if p.IsAnonymous() {
19+ return domain.MirrorPublicKey{}, domain.ErrUnauthenticated
20+ }
21+ return m.publicKey(ctx, false)
22+}
23+
24+// RotateKey immediately replaces the mirror key. Remote authorization must be updated separately.
25+func (m *Mirrors) RotateKey(ctx context.Context, p domain.Principal) (domain.MirrorPublicKey, error) {
26+ if p.IsAnonymous() {
27+ return domain.MirrorPublicKey{}, domain.ErrUnauthenticated
28+ }
29+ if !p.HasOwnerRights() {
30+ return domain.MirrorPublicKey{}, domain.ErrForbidden
31+ }
32+ return m.publicKey(ctx, true)
33+}
34+
35+func (m *Mirrors) publicKey(ctx context.Context, rotate bool) (domain.MirrorPublicKey, error) {
36+ m.keyMu.Lock()
37+ defer m.keyMu.Unlock()
38+ if err := m.loadKey(ctx, rotate); err != nil {
39+ return domain.MirrorPublicKey{}, err
40+ }
41+ host := strings.TrimPrefix(domain.MirrorKeyComment(m.cfg.PublicURL), "ermite-mirror@")
42+ return domain.MirrorPublicKeyOf(m.key.Public().(ed25519.PublicKey), host, m.keyCreated), nil
43+}
44+
45+// loadKey is called with keyMu held. Only successfully persisted material is cached.
46+func (m *Mirrors) loadKey(ctx context.Context, rotate bool) error {
47+ if m.cfg.Sealer == nil {
48+ return fmt.Errorf("%w: the server has no ERMITE_SECRETS_KEY, so it cannot store the mirror key", domain.ErrPreconditionFailed)
49+ }
50+ if !rotate && m.key != nil {
51+ return nil
52+ }
53+ if !rotate {
54+ sealed, created, err := m.cfg.Keys.GetMirrorKey(ctx)
55+ if err == nil {
56+ seed, err := m.cfg.Sealer.Open(sealed, []byte(domain.MirrorKeyPath))
57+ if err != nil {
58+ return fmt.Errorf("%w: cannot open the stored mirror key", domain.ErrPreconditionFailed)
59+ }
60+ if len(seed) != ed25519.SeedSize {
61+ return fmt.Errorf("%w: invalid mirror key seed", domain.ErrPreconditionFailed)
62+ }
63+ m.key, m.keyCreated = ed25519.NewKeyFromSeed(seed), created
64+ return nil
65+ }
66+ if !errors.Is(err, domain.ErrNotFound) {
67+ return fmt.Errorf("read mirror key: %w", err)
68+ }
69+ }
70+ _, key, err := ed25519.GenerateKey(rand.Reader)
71+ if err != nil {
72+ return fmt.Errorf("generate mirror key: %w", err)
73+ }
74+ created := m.cfg.Clock.Now().UTC()
75+ sealed := m.cfg.Sealer.Seal(key.Seed(), []byte(domain.MirrorKeyPath))
76+ err = m.cfg.Keys.PutMirrorKey(ctx, sealed, created, !rotate)
77+ if !rotate && errors.Is(err, domain.ErrAlreadyExists) {
78+ return m.loadKey(ctx, false)
79+ }
80+ if err != nil {
81+ return fmt.Errorf("store mirror key: %w", err)
82+ }
83+ m.key, m.keyCreated = key, created
84+ return nil
85+}
@@ -0,0 +1,85 @@
1+// SPDX-License-Identifier: AGPL-3.0-or-later
2+
3+package usecase
4+
5+import (
6+ "context"
7+ "crypto/ed25519"
8+ "crypto/rand"
9+ "errors"
10+ "fmt"
11+ "strings"
12+
13+ "bnema.dev/ermite/internal/domain"
14+)
15+
16+// ShowKey lazily generates the server's mirror key for authenticated callers.
17+func (m *Mirrors) ShowKey(ctx context.Context, p domain.Principal) (domain.MirrorPublicKey, error) {
18+ if p.IsAnonymous() {
19+ return domain.MirrorPublicKey{}, domain.ErrUnauthenticated
20+ }
21+ return m.publicKey(ctx, false)
22+}
23+
24+// RotateKey immediately replaces the mirror key. Remote authorization must be updated separately.
25+func (m *Mirrors) RotateKey(ctx context.Context, p domain.Principal) (domain.MirrorPublicKey, error) {
26+ if p.IsAnonymous() {
27+ return domain.MirrorPublicKey{}, domain.ErrUnauthenticated
28+ }
29+ if !p.HasOwnerRights() {
30+ return domain.MirrorPublicKey{}, domain.ErrForbidden
31+ }
32+ return m.publicKey(ctx, true)
33+}
34+
35+func (m *Mirrors) publicKey(ctx context.Context, rotate bool) (domain.MirrorPublicKey, error) {
36+ m.keyMu.Lock()
37+ defer m.keyMu.Unlock()
38+ if err := m.loadKey(ctx, rotate); err != nil {
39+ return domain.MirrorPublicKey{}, err
40+ }
41+ host := strings.TrimPrefix(domain.MirrorKeyComment(m.cfg.PublicURL), "ermite-mirror@")
42+ return domain.MirrorPublicKeyOf(m.key.Public().(ed25519.PublicKey), host, m.keyCreated), nil
43+}
44+
45+// loadKey is called with keyMu held. Only successfully persisted material is cached.
46+func (m *Mirrors) loadKey(ctx context.Context, rotate bool) error {
47+ if m.cfg.Sealer == nil {
48+ return fmt.Errorf("%w: the server has no ERMITE_SECRETS_KEY, so it cannot store the mirror key", domain.ErrPreconditionFailed)
49+ }
50+ if !rotate && m.key != nil {
51+ return nil
52+ }
53+ if !rotate {
54+ sealed, created, err := m.cfg.Keys.GetMirrorKey(ctx)
55+ if err == nil {
56+ seed, err := m.cfg.Sealer.Open(sealed, []byte(domain.MirrorKeyPath))
57+ if err != nil {
58+ return fmt.Errorf("%w: cannot open the stored mirror key", domain.ErrPreconditionFailed)
59+ }
60+ if len(seed) != ed25519.SeedSize {
61+ return fmt.Errorf("%w: invalid mirror key seed", domain.ErrPreconditionFailed)
62+ }
63+ m.key, m.keyCreated = ed25519.NewKeyFromSeed(seed), created
64+ return nil
65+ }
66+ if !errors.Is(err, domain.ErrNotFound) {
67+ return fmt.Errorf("read mirror key: %w", err)
68+ }
69+ }
70+ _, key, err := ed25519.GenerateKey(rand.Reader)
71+ if err != nil {
72+ return fmt.Errorf("generate mirror key: %w", err)
73+ }
74+ created := m.cfg.Clock.Now().UTC()
75+ sealed := m.cfg.Sealer.Seal(key.Seed(), []byte(domain.MirrorKeyPath))
76+ err = m.cfg.Keys.PutMirrorKey(ctx, sealed, created, !rotate)
77+ if !rotate && errors.Is(err, domain.ErrAlreadyExists) {
78+ return m.loadKey(ctx, false)
79+ }
80+ if err != nil {
81+ return fmt.Errorf("store mirror key: %w", err)
82+ }
83+ m.key, m.keyCreated = key, created
84+ return nil
85+}
A internal/usecase/mirrors_schedule.go +135 -0
@@ -0,0 +1,135 @@
1+// SPDX-License-Identifier: AGPL-3.0-or-later
2+
3+package usecase
4+
5+import (
6+ "context"
7+ "time"
8+
9+ "bnema.dev/ermite/internal/domain"
10+ "github.com/bnema/zerowrap"
11+)
12+
13+// Start starts the bounded worker pool and UTC scheduler exactly once.
14+func (m *Mirrors) Start() {
15+ m.start.Do(func() {
16+ m.mu.Lock()
17+ defer m.mu.Unlock()
18+ if m.closed {
19+ return
20+ }
21+ m.wg.Add(m.cfg.Workers + 1)
22+ for i := 0; i < m.cfg.Workers; i++ {
23+ go func() {
24+ defer m.wg.Done()
25+ for {
26+ select {
27+ case <-m.ctx.Done():
28+ return
29+ case repo := <-m.jobs:
30+ m.run(repo)
31+ }
32+ }
33+ }()
34+ }
35+ go func() {
36+ defer m.wg.Done()
37+ ticker := time.NewTicker(m.cfg.ScheduleTick)
38+ defer ticker.Stop()
39+ for {
40+ select {
41+ case <-m.ctx.Done():
42+ return
43+ case <-ticker.C:
44+ m.ScheduleNow(m.ctx)
45+ }
46+ }
47+ }()
48+ })
49+}
50+
51+// Close cancels workers and waits until they stop or the caller's deadline expires.
52+func (m *Mirrors) Close(ctx context.Context) {
53+ m.mu.Lock()
54+ m.closed = true
55+ m.cancel()
56+ m.mu.Unlock()
57+ done := make(chan struct{})
58+ go func() { m.wg.Wait(); close(done) }()
59+ select {
60+ case <-ctx.Done():
61+ case <-done:
62+ }
63+}
64+
65+// OnPush marks the repository dirty without I/O or waiting for a worker.
66+func (m *Mirrors) OnPush(_ context.Context, repo domain.RepoName, _ []domain.RefUpdate) {
67+ m.mu.Lock()
68+ defer m.mu.Unlock()
69+ if !m.closed {
70+ m.dirty[repo]++
71+ }
72+}
73+
74+// ScheduleNow examines one current minute; missed minutes are not replayed.
75+func (m *Mirrors) ScheduleNow(ctx context.Context) {
76+ m.Start()
77+ m.scheduleMu.Lock()
78+ defer m.scheduleMu.Unlock()
79+ now := m.cfg.Clock.Now().UTC()
80+ minute := now.Truncate(time.Minute)
81+ if !minute.After(m.lastMinute) {
82+ return
83+ }
84+ m.lastMinute = minute
85+ repos, err := m.cfg.Meta.List(ctx)
86+ if err != nil {
87+ return
88+ }
89+ live := make(map[domain.RepoName]bool, len(repos))
90+ for _, r := range repos {
91+ if ctx.Err() != nil {
92+ return
93+ }
94+ live[r.Name] = true
95+ t, err := m.target(ctx, r.Name)
96+ if err != nil || !t.Enabled {
97+ continue
98+ }
99+ s, err := m.status(ctx, r.Name)
100+ if err != nil || s.State == domain.MirrorPaused {
101+ continue
102+ }
103+ due := false
104+ if s.State == domain.MirrorFailing {
105+ due = !now.Before(s.NextRetryAt)
106+ } else {
107+ due = (domain.CronSchedule{Cron: t.Cron, Loc: time.UTC}).Due(minute)
108+ }
109+ if !due {
110+ continue
111+ }
112+ m.mu.Lock()
113+ clean, known := m.clean[r.Name]
114+ dirty := m.dirty[r.Name]
115+ m.mu.Unlock()
116+ if s.State == domain.MirrorOK && known && clean == dirty {
117+ snapshot, e := m.cfg.Store.GetSnapshot(ctx, r.Name)
118+ if e == nil && snapshot.URL == t.URL && snapshot.AllBranches == t.AllBranches {
119+ continue
120+ }
121+ }
122+ if _, _, err := m.enqueue(r.Name, false, false); err != nil {
123+ log := zerowrap.FromCtx(ctx)
124+ log.Warn().Str("repo", r.Name.String()).Msg("mirror queue full or closed")
125+ }
126+ }
127+ m.mu.Lock()
128+ for repo := range m.dirty {
129+ if !live[repo] {
130+ delete(m.dirty, repo)
131+ delete(m.clean, repo)
132+ }
133+ }
134+ m.mu.Unlock()
135+}
@@ -0,0 +1,135 @@
1+// SPDX-License-Identifier: AGPL-3.0-or-later
2+
3+package usecase
4+
5+import (
6+ "context"
7+ "time"
8+
9+ "bnema.dev/ermite/internal/domain"
10+ "github.com/bnema/zerowrap"
11+)
12+
13+// Start starts the bounded worker pool and UTC scheduler exactly once.
14+func (m *Mirrors) Start() {
15+ m.start.Do(func() {
16+ m.mu.Lock()
17+ defer m.mu.Unlock()
18+ if m.closed {
19+ return
20+ }
21+ m.wg.Add(m.cfg.Workers + 1)
22+ for i := 0; i < m.cfg.Workers; i++ {
23+ go func() {
24+ defer m.wg.Done()
25+ for {
26+ select {
27+ case <-m.ctx.Done():
28+ return
29+ case repo := <-m.jobs:
30+ m.run(repo)
31+ }
32+ }
33+ }()
34+ }
35+ go func() {
36+ defer m.wg.Done()
37+ ticker := time.NewTicker(m.cfg.ScheduleTick)
38+ defer ticker.Stop()
39+ for {
40+ select {
41+ case <-m.ctx.Done():
42+ return
43+ case <-ticker.C:
44+ m.ScheduleNow(m.ctx)
45+ }
46+ }
47+ }()
48+ })
49+}
50+
51+// Close cancels workers and waits until they stop or the caller's deadline expires.
52+func (m *Mirrors) Close(ctx context.Context) {
53+ m.mu.Lock()
54+ m.closed = true
55+ m.cancel()
56+ m.mu.Unlock()
57+ done := make(chan struct{})
58+ go func() { m.wg.Wait(); close(done) }()
59+ select {
60+ case <-ctx.Done():
61+ case <-done:
62+ }
63+}
64+
65+// OnPush marks the repository dirty without I/O or waiting for a worker.
66+func (m *Mirrors) OnPush(_ context.Context, repo domain.RepoName, _ []domain.RefUpdate) {
67+ m.mu.Lock()
68+ defer m.mu.Unlock()
69+ if !m.closed {
70+ m.dirty[repo]++
71+ }
72+}
73+
74+// ScheduleNow examines one current minute; missed minutes are not replayed.
75+func (m *Mirrors) ScheduleNow(ctx context.Context) {
76+ m.Start()
77+ m.scheduleMu.Lock()
78+ defer m.scheduleMu.Unlock()
79+ now := m.cfg.Clock.Now().UTC()
80+ minute := now.Truncate(time.Minute)
81+ if !minute.After(m.lastMinute) {
82+ return
83+ }
84+ m.lastMinute = minute
85+ repos, err := m.cfg.Meta.List(ctx)
86+ if err != nil {
87+ return
88+ }
89+ live := make(map[domain.RepoName]bool, len(repos))
90+ for _, r := range repos {
91+ if ctx.Err() != nil {
92+ return
93+ }
94+ live[r.Name] = true
95+ t, err := m.target(ctx, r.Name)
96+ if err != nil || !t.Enabled {
97+ continue
98+ }
99+ s, err := m.status(ctx, r.Name)
100+ if err != nil || s.State == domain.MirrorPaused {
101+ continue
102+ }
103+ due := false
104+ if s.State == domain.MirrorFailing {
105+ due = !now.Before(s.NextRetryAt)
106+ } else {
107+ due = (domain.CronSchedule{Cron: t.Cron, Loc: time.UTC}).Due(minute)
108+ }
109+ if !due {
110+ continue
111+ }
112+ m.mu.Lock()
113+ clean, known := m.clean[r.Name]
114+ dirty := m.dirty[r.Name]
115+ m.mu.Unlock()
116+ if s.State == domain.MirrorOK && known && clean == dirty {
117+ snapshot, e := m.cfg.Store.GetSnapshot(ctx, r.Name)
118+ if e == nil && snapshot.URL == t.URL && snapshot.AllBranches == t.AllBranches {
119+ continue
120+ }
121+ }
122+ if _, _, err := m.enqueue(r.Name, false, false); err != nil {
123+ log := zerowrap.FromCtx(ctx)
124+ log.Warn().Str("repo", r.Name.String()).Msg("mirror queue full or closed")
125+ }
126+ }
127+ m.mu.Lock()
128+ for repo := range m.dirty {
129+ if !live[repo] {
130+ delete(m.dirty, repo)
131+ delete(m.clean, repo)
132+ }
133+ }
134+ m.mu.Unlock()
135+}
A internal/usecase/mirrors_sync.go +222 -0
@@ -0,0 +1,222 @@
1+// SPDX-License-Identifier: AGPL-3.0-or-later
2+
3+package usecase
4+
5+import (
6+ "context"
7+ "errors"
8+ "fmt"
9+
10+ "bnema.dev/ermite/internal/domain"
11+ "bnema.dev/ermite/internal/ports"
12+)
13+
14+type mirrorFlight struct {
15+ pending, force, manual bool
16+ pendingForce bool
17+ done chan struct{}
18+ completed bool
19+ result ports.MirrorSyncResult
20+ err error
21+}
22+
23+var (
24+ _ ports.MirrorService = (*Mirrors)(nil)
25+ _ ports.MirrorTrigger = (*Mirrors)(nil)
26+ _ ports.MirrorWarner = (*Mirrors)(nil)
27+)
28+
29+// Sync waits for its own run; overlapping requests coalesce into one sticky rerun.
30+func (m *Mirrors) Sync(ctx context.Context, p domain.Principal, repo domain.RepoName, force bool) (ports.MirrorSyncResult, error) {
31+ if err := m.admin(ctx, p, repo); err != nil {
32+ return ports.MirrorSyncResult{}, err
33+ }
34+ m.Start()
35+ f, queued, err := m.enqueue(repo, force, true)
36+ if err != nil {
37+ return ports.MirrorSyncResult{}, err
38+ }
39+ if queued {
40+ return ports.MirrorSyncResult{Queued: true}, nil
41+ }
42+ select {
43+ case <-m.ctx.Done():
44+ return ports.MirrorSyncResult{}, m.ctx.Err()
45+ case <-ctx.Done():
46+ return ports.MirrorSyncResult{}, ctx.Err()
47+ case <-f.done:
48+ return f.result, f.err
49+ }
50+}
51+
52+// SyncAll requests a sync for every repository as an owner.
53+func (m *Mirrors) SyncAll(ctx context.Context, p domain.Principal, force bool) ([]ports.MirrorSyncResult, error) {
54+ if err := mirrorOwner(p); err != nil {
55+ return nil, err
56+ }
57+ repos, err := m.cfg.Meta.List(ctx)
58+ if err != nil {
59+ return nil, err
60+ }
61+ out := make([]ports.MirrorSyncResult, 0, len(repos))
62+ for _, r := range repos {
63+ v, err := m.Sync(ctx, p, r.Name, force)
64+ if err != nil {
65+ return out, err
66+ }
67+ out = append(out, v)
68+ }
69+ return out, nil
70+}
71+func (m *Mirrors) enqueue(repo domain.RepoName, force, manual bool) (*mirrorFlight, bool, error) {
72+ m.mu.Lock()
73+ defer m.mu.Unlock()
74+ if m.closed {
75+ return nil, false, context.Canceled
76+ }
77+ if f := m.flights[repo]; f != nil {
78+ f.pending = true
79+ f.pendingForce = f.pendingForce || force
80+ return f, true, nil
81+ }
82+ f := &mirrorFlight{force: force, manual: manual, done: make(chan struct{})}
83+ m.flights[repo] = f
84+ select {
85+ case m.jobs <- repo:
86+ return f, false, nil
87+ default:
88+ delete(m.flights, repo)
89+ return nil, false, domain.ErrResourceExhausted
90+ }
91+}
92+func (m *Mirrors) run(repo domain.RepoName) {
93+ m.mu.Lock()
94+ f := m.flights[repo]
95+ m.mu.Unlock()
96+ for {
97+ result, err := m.attempt(m.ctx, repo, f.force, f.manual)
98+ m.mu.Lock()
99+ if !f.completed {
100+ f.result, f.err = result, err
101+ f.completed = true
102+ close(f.done)
103+ }
104+ if !f.pending || m.closed {
105+ delete(m.flights, repo)
106+ m.mu.Unlock()
107+ return
108+ }
109+ f.force, f.manual = f.pendingForce, false
110+ f.pending, f.pendingForce = false, false
111+ m.mu.Unlock()
112+ }
113+}
114+
115+func (m *Mirrors) syncOnce(ctx context.Context, repo domain.RepoName, force bool) (ports.MirrorSyncResult, error) {
116+ return m.attempt(ctx, repo, force, true)
117+}
118+func (m *Mirrors) attempt(ctx context.Context, repo domain.RepoName, force, manual bool) (ports.MirrorSyncResult, error) {
119+ m.configMu.RLock()
120+ defer m.configMu.RUnlock()
121+ meta, err := m.cfg.Meta.Get(ctx, repo)
122+ if err != nil {
123+ return ports.MirrorSyncResult{}, err
124+ }
125+ t, err := m.target(ctx, repo)
126+ if err != nil {
127+ return ports.MirrorSyncResult{}, err
128+ }
129+ prev, err := m.status(ctx, repo)
130+ if err != nil {
131+ return ports.MirrorSyncResult{}, err
132+ }
133+ now := m.cfg.Clock.Now().UTC()
134+ if !t.Enabled {
135+ s := domain.MirrorDisable(prev, now)
136+ if prev.State != domain.MirrorDisabled {
137+ err = m.cfg.Store.PutStatus(ctx, s)
138+ }
139+ return ports.MirrorSyncResult{Status: s}, err
140+ }
141+ if !force && (prev.State == domain.MirrorPaused || (!manual && prev.State == domain.MirrorFailing && now.Before(prev.NextRetryAt))) {
142+ return ports.MirrorSyncResult{Status: prev}, nil
143+ }
144+ if manual && !force && !prev.LastAttemptAt.IsZero() && now.Sub(prev.LastAttemptAt) < domain.MirrorManualInterval {
145+ return ports.MirrorSyncResult{Status: prev}, fmt.Errorf("%w: last sync %s ago; wait 60s or use --force", domain.ErrResourceExhausted, now.Sub(prev.LastAttemptAt))
146+ }
147+ m.mu.Lock()
148+ generation := m.dirty[repo]
149+ m.mu.Unlock()
150+ refs, runErr := m.cfg.Reader.ListRefs(ctx, repo)
151+ var selected []domain.MirrorRef
152+ if runErr == nil {
153+ branch, _, e := m.cfg.Git.DefaultBranch(ctx, repo)
154+ runErr = e
155+ if e == nil {
156+ selected = domain.MirrorRefspecs(refs, branch, t.AllBranches)
157+ }
158+ }
159+ noOp := false
160+ if runErr == nil && !force {
161+ snapshot, e := m.cfg.Store.GetSnapshot(ctx, repo)
162+ if e != nil && !errors.Is(e, domain.ErrNotFound) {
163+ return ports.MirrorSyncResult{}, e
164+ }
165+ noOp = e == nil && snapshot.Equal(t.URL, selected, t.AllBranches)
166+ }
167+ if runErr == nil && !noOp {
168+ key, e := m.privateKey(ctx)
169+ runErr = e
170+ if e == nil {
171+ specs := make([]string, len(selected))
172+ for i, ref := range selected {
173+ specs[i] = ref.Name + ":" + ref.Name
174+ }
175+ // An empty ref list must not invoke git's implicit default push behavior.
176+ if len(specs) > 0 {
177+ _, runErr = m.cfg.Pusher.Push(ctx, ports.MirrorPush{Repo: repo, URL: t.URL, Refspecs: specs, PrivateKey: key})
178+ }
179+ }
180+ }
181+ if ctx.Err() != nil {
182+ return ports.MirrorSyncResult{}, ctx.Err()
183+ }
184+ var failure *domain.MirrorError
185+ if runErr != nil && !errors.As(runErr, &failure) {
186+ detail := "local mirror operation failed"
187+ if m.cfg.Sealer == nil {
188+ detail = "the server has no ERMITE_SECRETS_KEY, so it cannot store the mirror key"
189+ }
190+ failure = &domain.MirrorError{Class: domain.MirrorClassInternal, Detail: detail}
191+ runErr = failure
192+ if m.cfg.Sealer == nil {
193+ runErr = fmt.Errorf("%w: %s", domain.ErrPreconditionFailed, detail)
194+ }
195+ }
196+ transition := domain.ApplyMirrorOutcome(prev, domain.MirrorOutcome{Err: failure, NoOp: noOp}, now)
197+ if failure != nil && m.cfg.Sealer == nil {
198+ transition.Status.Reason = failure.Detail
199+ }
200+ if failure == nil {
201+ if err := m.cfg.Store.PutSnapshot(ctx, repo, domain.MirrorSnapshot{URL: t.URL, Refs: selected, AllBranches: t.AllBranches, At: now}); err != nil {
202+ return ports.MirrorSyncResult{}, err
203+ }
204+ }
205+ if err := m.cfg.Store.PutStatus(ctx, transition.Status); err != nil {
206+ return ports.MirrorSyncResult{}, err
207+ }
208+ if failure == nil {
209+ m.mu.Lock()
210+ m.clean[repo] = generation
211+ m.mu.Unlock()
212+ }
213+ if m.cfg.Bus != nil {
214+ if transition.FailedEvent {
215+ m.cfg.Bus.Publish(ctx, domain.MirrorFailed{Repo: repo, State: transition.Status.State, Class: transition.Status.Class, Reason: transition.Status.Reason, At: now, Visibility: meta.Visibility, CreatedAt: meta.CreatedAt})
216+ }
217+ if transition.Recovered {
218+ m.cfg.Bus.Publish(ctx, domain.MirrorRecovered{Repo: repo, At: now, Visibility: meta.Visibility, CreatedAt: meta.CreatedAt})
219+ }
220+ }
221+ return ports.MirrorSyncResult{Status: transition.Status, NoOp: noOp}, runErr
222+}
@@ -0,0 +1,222 @@
1+// SPDX-License-Identifier: AGPL-3.0-or-later
2+
3+package usecase
4+
5+import (
6+ "context"
7+ "errors"
8+ "fmt"
9+
10+ "bnema.dev/ermite/internal/domain"
11+ "bnema.dev/ermite/internal/ports"
12+)
13+
14+type mirrorFlight struct {
15+ pending, force, manual bool
16+ pendingForce bool
17+ done chan struct{}
18+ completed bool
19+ result ports.MirrorSyncResult
20+ err error
21+}
22+
23+var (
24+ _ ports.MirrorService = (*Mirrors)(nil)
25+ _ ports.MirrorTrigger = (*Mirrors)(nil)
26+ _ ports.MirrorWarner = (*Mirrors)(nil)
27+)
28+
29+// Sync waits for its own run; overlapping requests coalesce into one sticky rerun.
30+func (m *Mirrors) Sync(ctx context.Context, p domain.Principal, repo domain.RepoName, force bool) (ports.MirrorSyncResult, error) {
31+ if err := m.admin(ctx, p, repo); err != nil {
32+ return ports.MirrorSyncResult{}, err
33+ }
34+ m.Start()
35+ f, queued, err := m.enqueue(repo, force, true)
36+ if err != nil {
37+ return ports.MirrorSyncResult{}, err
38+ }
39+ if queued {
40+ return ports.MirrorSyncResult{Queued: true}, nil
41+ }
42+ select {
43+ case <-m.ctx.Done():
44+ return ports.MirrorSyncResult{}, m.ctx.Err()
45+ case <-ctx.Done():
46+ return ports.MirrorSyncResult{}, ctx.Err()
47+ case <-f.done:
48+ return f.result, f.err
49+ }
50+}
51+
52+// SyncAll requests a sync for every repository as an owner.
53+func (m *Mirrors) SyncAll(ctx context.Context, p domain.Principal, force bool) ([]ports.MirrorSyncResult, error) {
54+ if err := mirrorOwner(p); err != nil {
55+ return nil, err
56+ }
57+ repos, err := m.cfg.Meta.List(ctx)
58+ if err != nil {
59+ return nil, err
60+ }
61+ out := make([]ports.MirrorSyncResult, 0, len(repos))
62+ for _, r := range repos {
63+ v, err := m.Sync(ctx, p, r.Name, force)
64+ if err != nil {
65+ return out, err
66+ }
67+ out = append(out, v)
68+ }
69+ return out, nil
70+}
71+func (m *Mirrors) enqueue(repo domain.RepoName, force, manual bool) (*mirrorFlight, bool, error) {
72+ m.mu.Lock()
73+ defer m.mu.Unlock()
74+ if m.closed {
75+ return nil, false, context.Canceled
76+ }
77+ if f := m.flights[repo]; f != nil {
78+ f.pending = true
79+ f.pendingForce = f.pendingForce || force
80+ return f, true, nil
81+ }
82+ f := &mirrorFlight{force: force, manual: manual, done: make(chan struct{})}
83+ m.flights[repo] = f
84+ select {
85+ case m.jobs <- repo:
86+ return f, false, nil
87+ default:
88+ delete(m.flights, repo)
89+ return nil, false, domain.ErrResourceExhausted
90+ }
91+}
92+func (m *Mirrors) run(repo domain.RepoName) {
93+ m.mu.Lock()
94+ f := m.flights[repo]
95+ m.mu.Unlock()
96+ for {
97+ result, err := m.attempt(m.ctx, repo, f.force, f.manual)
98+ m.mu.Lock()
99+ if !f.completed {
100+ f.result, f.err = result, err
101+ f.completed = true
102+ close(f.done)
103+ }
104+ if !f.pending || m.closed {
105+ delete(m.flights, repo)
106+ m.mu.Unlock()
107+ return
108+ }
109+ f.force, f.manual = f.pendingForce, false
110+ f.pending, f.pendingForce = false, false
111+ m.mu.Unlock()
112+ }
113+}
114+
115+func (m *Mirrors) syncOnce(ctx context.Context, repo domain.RepoName, force bool) (ports.MirrorSyncResult, error) {
116+ return m.attempt(ctx, repo, force, true)
117+}
118+func (m *Mirrors) attempt(ctx context.Context, repo domain.RepoName, force, manual bool) (ports.MirrorSyncResult, error) {
119+ m.configMu.RLock()
120+ defer m.configMu.RUnlock()
121+ meta, err := m.cfg.Meta.Get(ctx, repo)
122+ if err != nil {
123+ return ports.MirrorSyncResult{}, err
124+ }
125+ t, err := m.target(ctx, repo)
126+ if err != nil {
127+ return ports.MirrorSyncResult{}, err
128+ }
129+ prev, err := m.status(ctx, repo)
130+ if err != nil {
131+ return ports.MirrorSyncResult{}, err
132+ }
133+ now := m.cfg.Clock.Now().UTC()
134+ if !t.Enabled {
135+ s := domain.MirrorDisable(prev, now)
136+ if prev.State != domain.MirrorDisabled {
137+ err = m.cfg.Store.PutStatus(ctx, s)
138+ }
139+ return ports.MirrorSyncResult{Status: s}, err
140+ }
141+ if !force && (prev.State == domain.MirrorPaused || (!manual && prev.State == domain.MirrorFailing && now.Before(prev.NextRetryAt))) {
142+ return ports.MirrorSyncResult{Status: prev}, nil
143+ }
144+ if manual && !force && !prev.LastAttemptAt.IsZero() && now.Sub(prev.LastAttemptAt) < domain.MirrorManualInterval {
145+ return ports.MirrorSyncResult{Status: prev}, fmt.Errorf("%w: last sync %s ago; wait 60s or use --force", domain.ErrResourceExhausted, now.Sub(prev.LastAttemptAt))
146+ }
147+ m.mu.Lock()
148+ generation := m.dirty[repo]
149+ m.mu.Unlock()
150+ refs, runErr := m.cfg.Reader.ListRefs(ctx, repo)
151+ var selected []domain.MirrorRef
152+ if runErr == nil {
153+ branch, _, e := m.cfg.Git.DefaultBranch(ctx, repo)
154+ runErr = e
155+ if e == nil {
156+ selected = domain.MirrorRefspecs(refs, branch, t.AllBranches)
157+ }
158+ }
159+ noOp := false
160+ if runErr == nil && !force {
161+ snapshot, e := m.cfg.Store.GetSnapshot(ctx, repo)
162+ if e != nil && !errors.Is(e, domain.ErrNotFound) {
163+ return ports.MirrorSyncResult{}, e
164+ }
165+ noOp = e == nil && snapshot.Equal(t.URL, selected, t.AllBranches)
166+ }
167+ if runErr == nil && !noOp {
168+ key, e := m.privateKey(ctx)
169+ runErr = e
170+ if e == nil {
171+ specs := make([]string, len(selected))
172+ for i, ref := range selected {
173+ specs[i] = ref.Name + ":" + ref.Name
174+ }
175+ // An empty ref list must not invoke git's implicit default push behavior.
176+ if len(specs) > 0 {
177+ _, runErr = m.cfg.Pusher.Push(ctx, ports.MirrorPush{Repo: repo, URL: t.URL, Refspecs: specs, PrivateKey: key})
178+ }
179+ }
180+ }
181+ if ctx.Err() != nil {
182+ return ports.MirrorSyncResult{}, ctx.Err()
183+ }
184+ var failure *domain.MirrorError
185+ if runErr != nil && !errors.As(runErr, &failure) {
186+ detail := "local mirror operation failed"
187+ if m.cfg.Sealer == nil {
188+ detail = "the server has no ERMITE_SECRETS_KEY, so it cannot store the mirror key"
189+ }
190+ failure = &domain.MirrorError{Class: domain.MirrorClassInternal, Detail: detail}
191+ runErr = failure
192+ if m.cfg.Sealer == nil {
193+ runErr = fmt.Errorf("%w: %s", domain.ErrPreconditionFailed, detail)
194+ }
195+ }
196+ transition := domain.ApplyMirrorOutcome(prev, domain.MirrorOutcome{Err: failure, NoOp: noOp}, now)
197+ if failure != nil && m.cfg.Sealer == nil {
198+ transition.Status.Reason = failure.Detail
199+ }
200+ if failure == nil {
201+ if err := m.cfg.Store.PutSnapshot(ctx, repo, domain.MirrorSnapshot{URL: t.URL, Refs: selected, AllBranches: t.AllBranches, At: now}); err != nil {
202+ return ports.MirrorSyncResult{}, err
203+ }
204+ }
205+ if err := m.cfg.Store.PutStatus(ctx, transition.Status); err != nil {
206+ return ports.MirrorSyncResult{}, err
207+ }
208+ if failure == nil {
209+ m.mu.Lock()
210+ m.clean[repo] = generation
211+ m.mu.Unlock()
212+ }
213+ if m.cfg.Bus != nil {
214+ if transition.FailedEvent {
215+ m.cfg.Bus.Publish(ctx, domain.MirrorFailed{Repo: repo, State: transition.Status.State, Class: transition.Status.Class, Reason: transition.Status.Reason, At: now, Visibility: meta.Visibility, CreatedAt: meta.CreatedAt})
216+ }
217+ if transition.Recovered {
218+ m.cfg.Bus.Publish(ctx, domain.MirrorRecovered{Repo: repo, At: now, Visibility: meta.Visibility, CreatedAt: meta.CreatedAt})
219+ }
220+ }
221+ return ports.MirrorSyncResult{Status: transition.Status, NoOp: noOp}, runErr
222+}
A internal/usecase/mirrors_test.go +433 -0
@@ -0,0 +1,433 @@
1+// SPDX-License-Identifier: AGPL-3.0-or-later
2+
3+package usecase_test
4+
5+import (
6+ "bytes"
7+ "context"
8+ "crypto/ed25519"
9+ "errors"
10+ "fmt"
11+ "github.com/bnema/zerowrap"
12+ "sync"
13+ "testing"
14+ "time"
15+
16+ "bnema.dev/ermite/internal/domain"
17+ "bnema.dev/ermite/internal/ports"
18+ "bnema.dev/ermite/internal/ports/mocks"
19+ "bnema.dev/ermite/internal/usecase"
20+ "github.com/stretchr/testify/mock"
21+ "github.com/stretchr/testify/require"
22+)
23+
24+type mirrorFixture struct {
25+ m *usecase.Mirrors
26+ store *mocks.MirrorStore
27+ push *mocks.MirrorPusher
28+ reader *mocks.RepoReader
29+ bus *mocks.EventBus
30+ keys *mocks.MirrorKeyStore
31+ sealer *mocks.SecretSealer
32+ mu sync.Mutex
33+ at time.Time
34+ status domain.MirrorStatus
35+ snapshot domain.MirrorSnapshot
36+ events []domain.Event
37+ repo domain.RepoName
38+}
39+
40+func newMirrorFixture(t *testing.T) *mirrorFixture {
41+ t.Helper()
42+ f := &mirrorFixture{store: mocks.NewMirrorStore(t), push: mocks.NewMirrorPusher(t), reader: mocks.NewRepoReader(t), bus: mocks.NewEventBus(t), keys: mocks.NewMirrorKeyStore(t), sealer: mocks.NewSecretSealer(t), at: now, repo: name(t, "tools")}
43+ meta, git, clock := mocks.NewRepoMetaStore(t), mocks.NewGitRepos(t), mocks.NewClock(t)
44+ clock.On("Now").Return(func() time.Time { f.mu.Lock(); defer f.mu.Unlock(); return f.at })
45+ meta.On("Get", mock.Anything, f.repo).Return(domain.Repo{Name: f.repo, Visibility: domain.VisibilityPublic, CreatedAt: now}, nil).Maybe()
46+ meta.On("List", mock.Anything).Return([]domain.Repo{{Name: f.repo}}, nil).Maybe()
47+ f.store.On("GetGlobal", mock.Anything).Return(domain.MirrorGlobalConfig{Enabled: true, URLTemplate: "git@example.com:acme/{repo}", Cron: "*/5 * * * *"}, nil)
48+ f.store.On("GetRepoConfig", mock.Anything, f.repo).Return(domain.MirrorRepoConfig{}, domain.ErrNotFound)
49+ f.store.On("GetStatus", mock.Anything, f.repo).Return(func(context.Context, domain.RepoName) domain.MirrorStatus {
50+ f.mu.Lock()
51+ defer f.mu.Unlock()
52+ return f.status
53+ }, nil)
54+ f.store.On("PutStatus", mock.Anything, mock.Anything).Run(func(a mock.Arguments) { f.mu.Lock(); defer f.mu.Unlock(); f.status = a.Get(1).(domain.MirrorStatus) }).Return(nil).Maybe()
55+ f.store.On("GetSnapshot", mock.Anything, f.repo).Return(func(context.Context, domain.RepoName) domain.MirrorSnapshot {
56+ f.mu.Lock()
57+ defer f.mu.Unlock()
58+ return f.snapshot
59+ }, nil).Maybe()
60+ f.store.On("PutSnapshot", mock.Anything, f.repo, mock.Anything).Run(func(a mock.Arguments) {
61+ f.mu.Lock()
62+ defer f.mu.Unlock()
63+ f.snapshot = a.Get(2).(domain.MirrorSnapshot)
64+ }).Return(nil).Maybe()
65+ f.reader.On("ListRefs", mock.Anything, f.repo).Return([]domain.Ref{{Name: "refs/heads/main", ID: mirrorOID(t)}}, nil).Maybe()
66+ git.On("DefaultBranch", mock.Anything, f.repo).Return("main", true, nil).Maybe()
67+ f.bus.On("Publish", mock.Anything, mock.Anything).Run(func(a mock.Arguments) {
68+ f.mu.Lock()
69+ defer f.mu.Unlock()
70+ f.events = append(f.events, a.Get(1).(domain.Event))
71+ }).Maybe()
72+ f.keys.On("GetMirrorKey", mock.Anything).Return([]byte("sealed"), now, nil).Maybe()
73+ f.sealer.On("Open", []byte("sealed"), []byte(domain.MirrorKeyPath)).Return(make([]byte, ed25519.SeedSize), nil).Maybe()
74+ f.status = domain.MirrorStatus{Repo: f.repo, State: domain.MirrorOK}
75+ f.m = usecase.NewMirrors(usecase.MirrorsConfig{Store: f.store, Keys: f.keys, Sealer: f.sealer, Pusher: f.push, Meta: meta, Git: git, Reader: f.reader, Bus: f.bus, Clock: clock, ScheduleTick: time.Hour})
76+ t.Cleanup(func() {
77+ ctx, cancel := context.WithTimeout(context.Background(), time.Second)
78+ defer cancel()
79+ f.m.Close(ctx)
80+ })
81+ return f
82+}
83+func (f *mirrorFixture) advance(d time.Duration) { f.mu.Lock(); f.at = f.at.Add(d); f.mu.Unlock() }
84+func TestMirrorsNoOpThrottleForce(t *testing.T) {
85+ f := newMirrorFixture(t)
86+ f.snapshot = domain.MirrorSnapshot{URL: "git@example.com:acme/tools", Refs: []domain.MirrorRef{{Name: "refs/heads/main", ID: mirrorOID(t)}}}
87+ v, err := f.m.Sync(context.Background(), domain.Root(), f.repo, false)
88+ require.NoError(t, err)
89+ require.True(t, v.NoOp)
90+ _, err = f.m.Sync(context.Background(), domain.Root(), f.repo, false)
91+ require.ErrorIs(t, err, domain.ErrResourceExhausted)
92+ f.push.On("Push", mock.Anything, mock.MatchedBy(func(p ports.MirrorPush) bool {
93+ return len(p.Refspecs) == 1 && p.Refspecs[0] == "refs/heads/main:refs/heads/main"
94+ })).Return(ports.MirrorPushResult{}, nil).Once()
95+ v, err = f.m.Sync(context.Background(), domain.Root(), f.repo, true)
96+ require.NoError(t, err)
97+ require.False(t, v.NoOp)
98+}
99+func TestMirrorsIncidentAndRecovery(t *testing.T) {
100+ f := newMirrorFixture(t)
101+ f.push.On("Push", mock.Anything, mock.Anything).Return(ports.MirrorPushResult{}, &domain.MirrorError{Class: domain.MirrorClassTransient}).Times(5)
102+ for i := 1; i <= 5; i++ {
103+ v, err := f.m.Sync(context.Background(), domain.Root(), f.repo, true)
104+ require.Error(t, err)
105+ require.Equal(t, i, v.Status.ConsecutiveFailures)
106+ if i < 5 {
107+ require.Equal(t, domain.MirrorFailing, v.Status.State)
108+ require.Equal(t, domain.MirrorBackoff(i), v.Status.NextRetryAt.Sub(v.Status.LastAttemptAt))
109+ } else {
110+ require.Equal(t, domain.MirrorPaused, v.Status.State)
111+ }
112+ }
113+ require.Len(t, f.events, 1)
114+ require.IsType(t, domain.MirrorFailed{}, f.events[0])
115+ f.advance(24 * time.Hour)
116+ f.m.ScheduleNow(context.Background())
117+ f.push.On("Push", mock.Anything, mock.Anything).Return(ports.MirrorPushResult{}, nil).Once()
118+ v, err := f.m.Sync(context.Background(), domain.Root(), f.repo, true)
119+ require.NoError(t, err)
120+ require.Equal(t, domain.MirrorOK, v.Status.State)
121+ require.Len(t, f.events, 2)
122+ require.IsType(t, domain.MirrorRecovered{}, f.events[1])
123+}
124+func TestMirrorsSingleFlight(t *testing.T) {
125+ f := newMirrorFixture(t)
126+ entered, release := make(chan struct{}), make(chan struct{})
127+ f.push.On("Push", mock.Anything, mock.Anything).Run(func(mock.Arguments) { close(entered); <-release }).Return(ports.MirrorPushResult{}, nil).Once()
128+ done := make(chan error, 1)
129+ go func() { _, err := f.m.Sync(context.Background(), domain.Root(), f.repo, true); done <- err }()
130+ <-entered
131+ var wg sync.WaitGroup
132+ for i := 0; i < 20; i++ {
133+ wg.Add(1)
134+ go func() {
135+ defer wg.Done()
136+ v, err := f.m.Sync(context.Background(), domain.Root(), f.repo, true)
137+ require.NoError(t, err)
138+ require.True(t, v.Queued)
139+ }()
140+ }
141+ wg.Wait()
142+ rerun := make(chan struct{})
143+ f.push.On("Push", mock.Anything, mock.Anything).Run(func(mock.Arguments) { close(rerun) }).Return(ports.MirrorPushResult{}, nil).Once()
144+ close(release)
145+ require.NoError(t, <-done)
146+ select {
147+ case <-rerun:
148+ case <-time.After(time.Second):
149+ t.Fatal("coalesced force was lost")
150+ }
151+}
152+func TestMirrorsAuthorizationAndMissingSecretsKey(t *testing.T) {
153+ m := usecase.NewMirrors(usecase.MirrorsConfig{Clock: func() *mocks.Clock { c := mocks.NewClock(t); c.On("Now").Return(now); return c }()})
154+ _, err := m.ShowKey(context.Background(), domain.Anonymous())
155+ require.ErrorIs(t, err, domain.ErrUnauthenticated)
156+ _, err = m.ShowKey(context.Background(), domain.Root())
157+ require.ErrorIs(t, err, domain.ErrPreconditionFailed)
158+ require.Contains(t, err.Error(), "ERMITE_SECRETS_KEY")
159+ _, err = m.RotateKey(context.Background(), tokenPrincipal(domain.ScopeRepoRead))
160+ require.True(t, errors.Is(err, domain.ErrForbidden))
161+}
162+
163+func mirrorOID(t *testing.T) domain.ObjectID {
164+ t.Helper()
165+ id, err := domain.ParseObjectID(oidA)
166+ require.NoError(t, err)
167+ return id
168+}
169+
170+func TestMirrorsDivergedPausedAndConfigReset(t *testing.T) {
171+ f := newMirrorFixture(t)
172+ f.push.On("Push", mock.Anything, mock.Anything).Return(ports.MirrorPushResult{}, &domain.MirrorError{Class: domain.MirrorClassDiverged}).Once()
173+ v, err := f.m.Sync(context.Background(), domain.Root(), f.repo, false)
174+ require.Error(t, err)
175+ require.Equal(t, domain.MirrorPaused, v.Status.State)
176+ for i := 0; i < 3; i++ {
177+ f.advance(time.Hour)
178+ f.m.ScheduleNow(context.Background())
179+ }
180+ f.store.On("PutRepoConfig", mock.Anything, mock.Anything).Return(nil).Once()
181+ _, err = f.m.SetRepoMirror(context.Background(), domain.Root(), domain.MirrorRepoConfig{Repo: f.repo, URL: "git@example.com:acme/tools", Cron: "*/5 * * * *"})
182+ require.NoError(t, err)
183+ f.mu.Lock()
184+ s := f.status
185+ events := len(f.events)
186+ f.mu.Unlock()
187+ require.Equal(t, domain.MirrorOK, s.State)
188+ require.Zero(t, s.ConsecutiveFailures)
189+ require.Equal(t, 1, events)
190+}
191+
192+func TestMirrorsKeyLazyGenerationAndRotation(t *testing.T) {
193+ keys, sealer, clock := mocks.NewMirrorKeyStore(t), mocks.NewSecretSealer(t), mocks.NewClock(t)
194+ clock.On("Now").Return(now)
195+ keys.On("GetMirrorKey", mock.Anything).Return([]byte(nil), time.Time{}, domain.ErrNotFound).Once()
196+ var seeds [][]byte
197+ sealer.On("Seal", mock.Anything, []byte(domain.MirrorKeyPath)).Return(func(seed, aad []byte) []byte {
198+ seeds = append(seeds, append([]byte(nil), seed...))
199+ return append([]byte("sealed:"), seed...)
200+ }).Twice()
201+ keys.On("PutMirrorKey", mock.Anything, mock.Anything, now, true).Return(nil).Once()
202+ keys.On("PutMirrorKey", mock.Anything, mock.Anything, now, false).Return(nil).Once()
203+ m := usecase.NewMirrors(usecase.MirrorsConfig{Keys: keys, Sealer: sealer, Clock: clock, PublicURL: "https://forge.example.com"})
204+ first, err := m.ShowKey(context.Background(), domain.Root())
205+ require.NoError(t, err)
206+ cached, err := m.ShowKey(context.Background(), domain.Root())
207+ require.NoError(t, err)
208+ require.Equal(t, first, cached)
209+ rotated, err := m.RotateKey(context.Background(), domain.Root())
210+ require.NoError(t, err)
211+ require.NotEqual(t, first.Fingerprint, rotated.Fingerprint)
212+ require.Contains(t, rotated.AuthorizedKey, "ermite-mirror@forge.example.com")
213+ require.Len(t, seeds, 2)
214+ require.NotEqual(t, seeds[0], seeds[1])
215+ require.Len(t, seeds[0], ed25519.SeedSize)
216+}
217+
218+func TestMirrorsPushRecordNotification(t *testing.T) {
219+ f := newPushFixture(t)
220+ n := name(t, "tools")
221+ trigger := mocks.NewMirrorTrigger(t)
222+ updates := []domain.RefUpdate{}
223+ f.meta.On("Mutate", mock.Anything, n, mock.Anything).Return(domain.Repo{Name: n}, nil)
224+ f.git.On("DefaultBranch", mock.Anything, n).Return("main", true, nil)
225+ f.bus.On("Publish", mock.Anything, mock.Anything).Return()
226+ f.sched.On("Schedule", mock.Anything, n).Return()
227+ notified := make(chan struct{})
228+ trigger.On("OnPush", mock.Anything, n, updates).Run(func(mock.Arguments) { close(notified) }).Return().Once()
229+ require.NoError(t, f.pushes.WithMirror(trigger).Record(context.Background(), domain.Root(), n, updates))
230+ <-notified
231+}
232+
233+func TestMirrorsCronDueAndDirtySkip(t *testing.T) {
234+ f := newMirrorFixture(t)
235+ // Establish a successful snapshot; subsequent clean cron ticks need no refs.
236+ f.push.On("Push", mock.Anything, mock.Anything).Return(ports.MirrorPushResult{}, nil).Once()
237+ _, err := f.m.Sync(context.Background(), domain.Root(), f.repo, true)
238+ require.NoError(t, err)
239+ f.reader.AssertNumberOfCalls(t, "ListRefs", 1)
240+ f.mu.Lock()
241+ f.at = f.at.Add(24 * time.Hour).Truncate(time.Hour).Add(time.Minute)
242+ f.mu.Unlock()
243+ f.m.ScheduleNow(context.Background())
244+ f.advance(4 * time.Minute)
245+ f.m.ScheduleNow(context.Background())
246+ f.reader.AssertNumberOfCalls(t, "ListRefs", 1)
247+ f.m.OnPush(context.Background(), f.repo, nil)
248+ read := make(chan struct{})
249+ f.reader.ExpectedCalls = nil
250+ f.reader.On("ListRefs", mock.Anything, f.repo).Run(func(mock.Arguments) { close(read) }).Return([]domain.Ref{{Name: "refs/heads/main", ID: mirrorOID(t)}}, nil).Once()
251+ // Same minute does not run, even though dirty.
252+ f.m.ScheduleNow(context.Background())
253+ select {
254+ case <-read:
255+ t.Fatal("same-minute cron reran")
256+ default:
257+ }
258+ f.advance(5 * time.Minute)
259+ f.m.ScheduleNow(context.Background())
260+ select {
261+ case <-read:
262+ case <-time.After(time.Second):
263+ t.Fatal("due dirty repository not examined")
264+ }
265+}
266+
267+func TestMirrorsAuthorizationMatrix(t *testing.T) {
268+ f := newMirrorFixture(t)
269+ ctx := context.Background()
270+ reader := tokenPrincipal(domain.ScopeRepoRead)
271+ admin := tokenPrincipal(domain.ScopeRepoAdmin)
272+ _, err := f.m.GetGlobalConfig(ctx, reader)
273+ require.ErrorIs(t, err, domain.ErrForbidden)
274+ _, err = f.m.SetGlobalConfig(ctx, reader, ports.MirrorGlobalPatch{})
275+ require.ErrorIs(t, err, domain.ErrForbidden)
276+ _, err = f.m.ListStatus(ctx, reader)
277+ require.ErrorIs(t, err, domain.ErrForbidden)
278+ _, err = f.m.RotateKey(ctx, admin)
279+ require.ErrorIs(t, err, domain.ErrForbidden)
280+ // Key show deliberately requires authentication, not owner rights.
281+ _, err = f.m.ShowKey(ctx, reader)
282+ require.NoError(t, err)
283+ _, err = f.m.ShowKey(ctx, domain.Anonymous())
284+ require.ErrorIs(t, err, domain.ErrUnauthenticated)
285+ _, err = f.m.SetRepoMirror(ctx, reader, domain.MirrorRepoConfig{Repo: f.repo})
286+ require.ErrorIs(t, err, domain.ErrForbidden)
287+ err = f.m.RemoveRepoMirror(ctx, reader, f.repo)
288+ require.ErrorIs(t, err, domain.ErrForbidden)
289+ _, err = f.m.SetRepoExcluded(ctx, reader, f.repo, true)
290+ require.ErrorIs(t, err, domain.ErrForbidden)
291+ _, err = f.m.Sync(ctx, reader, f.repo, false)
292+ require.ErrorIs(t, err, domain.ErrForbidden)
293+ f.mu.Lock()
294+ f.status.Detail, f.status.Reason = "admin detail", "admin reason"
295+ f.mu.Unlock()
296+ view, err := f.m.GetStatus(ctx, domain.Anonymous(), f.repo)
297+ require.NoError(t, err)
298+ require.Empty(t, view.Status.Detail)
299+ require.Empty(t, view.Status.Reason)
300+ require.Empty(t, view.Target.URL)
301+ view, err = f.m.GetStatus(ctx, admin, f.repo)
302+ require.NoError(t, err)
303+ require.Equal(t, "admin detail", view.Status.Detail)
304+ f.store.On("PutRepoConfig", mock.Anything, mock.Anything).Return(nil).Once()
305+ _, err = f.m.SetRepoMirror(ctx, admin, domain.MirrorRepoConfig{Repo: f.repo})
306+ require.NoError(t, err)
307+}
308+
309+func TestMirrorsPrivateStatusAuthorization(t *testing.T) {
310+ f := newMirrorFixture(t)
311+ meta := mocks.NewRepoMetaStore(t)
312+ meta.On("Get", mock.Anything, f.repo).Return(domain.Repo{Name: f.repo, Visibility: domain.VisibilityPrivate}, nil)
313+ clock := mocks.NewClock(t)
314+ clock.On("Now").Return(now)
315+ m := usecase.NewMirrors(usecase.MirrorsConfig{Meta: meta, Store: f.store, Clock: clock})
316+ _, err := m.GetStatus(context.Background(), domain.Anonymous(), f.repo)
317+ require.ErrorIs(t, err, domain.ErrNotFound)
318+ _, err = m.GetStatus(context.Background(), tokenPrincipal(domain.ScopeRepoRead, "other"), f.repo)
319+ require.ErrorIs(t, err, domain.ErrNotFound)
320+ _, err = m.GetStatus(context.Background(), tokenPrincipal(domain.ScopeRepoRead, "tools"), f.repo)
321+ require.NoError(t, err)
322+}
323+
324+func TestMirrorsCloseCancelsWorker(t *testing.T) {
325+ f := newMirrorFixture(t)
326+ entered, exited := make(chan struct{}), make(chan struct{})
327+ f.push.On("Push", mock.Anything, mock.Anything).Run(func(a mock.Arguments) { close(entered); <-a.Get(0).(context.Context).Done(); close(exited) }).Return(ports.MirrorPushResult{}, context.Canceled).Once()
328+ done := make(chan error, 1)
329+ go func() { _, err := f.m.Sync(context.Background(), domain.Root(), f.repo, true); done <- err }()
330+ <-entered
331+ ctx, cancel := context.WithTimeout(context.Background(), time.Second)
332+ defer cancel()
333+ f.m.Close(ctx)
334+ require.NoError(t, ctx.Err())
335+ select {
336+ case <-exited:
337+ default:
338+ t.Fatal("Close returned before worker exited")
339+ }
340+ require.ErrorIs(t, <-done, context.Canceled)
341+ f.m.Close(ctx) // idempotent; Start cannot resurrect workers.
342+ f.m.Start()
343+ _, err := f.m.Sync(context.Background(), domain.Root(), f.repo, true)
344+ require.ErrorIs(t, err, context.Canceled)
345+}
346+
347+func TestMirrorsCronBackoff(t *testing.T) {
348+ f := newMirrorFixture(t)
349+ f.push.On("Push", mock.Anything, mock.Anything).Return(ports.MirrorPushResult{}, &domain.MirrorError{Class: domain.MirrorClassTransient}).Once()
350+ _, err := f.m.Sync(context.Background(), domain.Root(), f.repo, false)
351+ require.Error(t, err)
352+ f.advance(time.Minute)
353+ f.m.ScheduleNow(context.Background())
354+ f.push.AssertNumberOfCalls(t, "Push", 1)
355+ attempted := make(chan struct{})
356+ f.push.On("Push", mock.Anything, mock.Anything).Run(func(mock.Arguments) { close(attempted) }).Return(ports.MirrorPushResult{}, &domain.MirrorError{Class: domain.MirrorClassTransient}).Once()
357+ f.advance(4 * time.Minute)
358+ f.m.ScheduleNow(context.Background())
359+ select {
360+ case <-attempted:
361+ case <-time.After(time.Second):
362+ t.Fatal("retry did not fire at backoff deadline")
363+ }
364+ // Cleanup joins the worker before mock verification.
365+}
366+
367+func TestMirrorsRecordDoesNotWaitForTrigger(t *testing.T) {
368+ f := newPushFixture(t)
369+ n := name(t, "tools")
370+ trigger := mocks.NewMirrorTrigger(t)
371+ entered, release, exited := make(chan struct{}), make(chan struct{}), make(chan struct{})
372+ trigger.On("OnPush", mock.Anything, n, mock.Anything).Run(func(mock.Arguments) { close(entered); <-release; close(exited) }).Return().Once()
373+ f.meta.On("Mutate", mock.Anything, n, mock.Anything).Return(domain.Repo{Name: n}, nil)
374+ f.git.On("DefaultBranch", mock.Anything, n).Return("main", true, nil)
375+ f.bus.On("Publish", mock.Anything, mock.Anything).Return()
376+ f.sched.On("Schedule", mock.Anything, n).Return()
377+ returned := make(chan error, 1)
378+ go func() { returned <- f.pushes.WithMirror(trigger).Record(context.Background(), domain.Root(), n, nil) }()
379+ <-entered
380+ select {
381+ case err := <-returned:
382+ require.NoError(t, err)
383+ case <-time.After(time.Second):
384+ close(release)
385+ <-exited
386+ t.Fatal("Record blocked on mirror trigger")
387+ }
388+ close(release)
389+ <-exited
390+}
391+
392+func TestMirrorsErrorsAndLogsDoNotExposeSecrets(t *testing.T) {
393+ f := newMirrorFixture(t)
394+ var logs bytes.Buffer
395+ ctx := zerowrap.WithCtx(context.Background(), zerowrap.New(zerowrap.Config{Level: "debug", Format: "json", Output: &logs}))
396+ credentialURL := "ssh://alice:secret-password@example.com/acme/tools"
397+ _, err := f.m.SetRepoMirror(ctx, domain.Root(), domain.MirrorRepoConfig{Repo: f.repo, URL: credentialURL})
398+ require.Error(t, err)
399+ require.NotContains(t, err.Error(), "secret-password")
400+ require.NotContains(t, err.Error(), "alice")
401+ f.push.On("Push", mock.Anything, mock.Anything).Return(ports.MirrorPushResult{}, errors.New("local key PRIVATE-KEY-MARKER ssh://alice:secret-password@example.com/acme/tools")).Once()
402+ _, err = f.m.Sync(ctx, domain.Root(), f.repo, true)
403+ require.Error(t, err)
404+ require.NotContains(t, err.Error(), "PRIVATE-KEY-MARKER")
405+ require.NotContains(t, err.Error(), "secret-password")
406+ require.NotContains(t, logs.String(), "PRIVATE-KEY-MARKER")
407+ require.NotContains(t, logs.String(), "secret-password")
408+}
409+
410+func TestMirrorsFileURLTestOption(t *testing.T) {
411+ for _, allowed := range []bool{false, true} {
412+ t.Run(fmt.Sprint(allowed), func(t *testing.T) {
413+ store, meta, clock := mocks.NewMirrorStore(t), mocks.NewRepoMetaStore(t), mocks.NewClock(t)
414+ repo := name(t, "tools")
415+ clock.On("Now").Return(now)
416+ meta.On("Get", mock.Anything, repo).Return(domain.Repo{Name: repo}, nil)
417+ global := domain.MirrorGlobalConfig{Enabled: true, URLTemplate: "file:///tmp/remotes/{repo}", Cron: "*/5 * * * *"}
418+ store.On("GetGlobal", mock.Anything).Return(global, nil).Maybe()
419+ store.On("GetRepoConfig", mock.Anything, repo).Return(domain.MirrorRepoConfig{}, domain.ErrNotFound).Maybe()
420+ store.On("GetStatus", mock.Anything, repo).Return(domain.MirrorStatus{Repo: repo}, nil).Maybe()
421+ store.On("PutRepoConfig", mock.Anything, mock.Anything).Return(nil).Maybe()
422+ store.On("PutStatus", mock.Anything, mock.Anything).Return(nil).Maybe()
423+ m := usecase.NewMirrors(usecase.MirrorsConfig{Store: store, Meta: meta, Clock: clock, AllowAnyURL: allowed})
424+ target, err := m.SetRepoMirror(context.Background(), domain.Root(), domain.MirrorRepoConfig{Repo: repo, URL: "file:///tmp/remotes/tools", Cron: "*/5 * * * *"})
425+ if allowed {
426+ require.NoError(t, err)
427+ require.Equal(t, "file:///tmp/remotes/tools", target.URL)
428+ } else {
429+ require.ErrorIs(t, err, domain.ErrInvalid)
430+ }
431+ })
432+ }
433+}
@@ -0,0 +1,433 @@
1+// SPDX-License-Identifier: AGPL-3.0-or-later
2+
3+package usecase_test
4+
5+import (
6+ "bytes"
7+ "context"
8+ "crypto/ed25519"
9+ "errors"
10+ "fmt"
11+ "github.com/bnema/zerowrap"
12+ "sync"
13+ "testing"
14+ "time"
15+
16+ "bnema.dev/ermite/internal/domain"
17+ "bnema.dev/ermite/internal/ports"
18+ "bnema.dev/ermite/internal/ports/mocks"
19+ "bnema.dev/ermite/internal/usecase"
20+ "github.com/stretchr/testify/mock"
21+ "github.com/stretchr/testify/require"
22+)
23+
24+type mirrorFixture struct {
25+ m *usecase.Mirrors
26+ store *mocks.MirrorStore
27+ push *mocks.MirrorPusher
28+ reader *mocks.RepoReader
29+ bus *mocks.EventBus
30+ keys *mocks.MirrorKeyStore
31+ sealer *mocks.SecretSealer
32+ mu sync.Mutex
33+ at time.Time
34+ status domain.MirrorStatus
35+ snapshot domain.MirrorSnapshot
36+ events []domain.Event
37+ repo domain.RepoName
38+}
39+
40+func newMirrorFixture(t *testing.T) *mirrorFixture {
41+ t.Helper()
42+ f := &mirrorFixture{store: mocks.NewMirrorStore(t), push: mocks.NewMirrorPusher(t), reader: mocks.NewRepoReader(t), bus: mocks.NewEventBus(t), keys: mocks.NewMirrorKeyStore(t), sealer: mocks.NewSecretSealer(t), at: now, repo: name(t, "tools")}
43+ meta, git, clock := mocks.NewRepoMetaStore(t), mocks.NewGitRepos(t), mocks.NewClock(t)
44+ clock.On("Now").Return(func() time.Time { f.mu.Lock(); defer f.mu.Unlock(); return f.at })
45+ meta.On("Get", mock.Anything, f.repo).Return(domain.Repo{Name: f.repo, Visibility: domain.VisibilityPublic, CreatedAt: now}, nil).Maybe()
46+ meta.On("List", mock.Anything).Return([]domain.Repo{{Name: f.repo}}, nil).Maybe()
47+ f.store.On("GetGlobal", mock.Anything).Return(domain.MirrorGlobalConfig{Enabled: true, URLTemplate: "git@example.com:acme/{repo}", Cron: "*/5 * * * *"}, nil)
48+ f.store.On("GetRepoConfig", mock.Anything, f.repo).Return(domain.MirrorRepoConfig{}, domain.ErrNotFound)
49+ f.store.On("GetStatus", mock.Anything, f.repo).Return(func(context.Context, domain.RepoName) domain.MirrorStatus {
50+ f.mu.Lock()
51+ defer f.mu.Unlock()
52+ return f.status
53+ }, nil)
54+ f.store.On("PutStatus", mock.Anything, mock.Anything).Run(func(a mock.Arguments) { f.mu.Lock(); defer f.mu.Unlock(); f.status = a.Get(1).(domain.MirrorStatus) }).Return(nil).Maybe()
55+ f.store.On("GetSnapshot", mock.Anything, f.repo).Return(func(context.Context, domain.RepoName) domain.MirrorSnapshot {
56+ f.mu.Lock()
57+ defer f.mu.Unlock()
58+ return f.snapshot
59+ }, nil).Maybe()
60+ f.store.On("PutSnapshot", mock.Anything, f.repo, mock.Anything).Run(func(a mock.Arguments) {
61+ f.mu.Lock()
62+ defer f.mu.Unlock()
63+ f.snapshot = a.Get(2).(domain.MirrorSnapshot)
64+ }).Return(nil).Maybe()
65+ f.reader.On("ListRefs", mock.Anything, f.repo).Return([]domain.Ref{{Name: "refs/heads/main", ID: mirrorOID(t)}}, nil).Maybe()
66+ git.On("DefaultBranch", mock.Anything, f.repo).Return("main", true, nil).Maybe()
67+ f.bus.On("Publish", mock.Anything, mock.Anything).Run(func(a mock.Arguments) {
68+ f.mu.Lock()
69+ defer f.mu.Unlock()
70+ f.events = append(f.events, a.Get(1).(domain.Event))
71+ }).Maybe()
72+ f.keys.On("GetMirrorKey", mock.Anything).Return([]byte("sealed"), now, nil).Maybe()
73+ f.sealer.On("Open", []byte("sealed"), []byte(domain.MirrorKeyPath)).Return(make([]byte, ed25519.SeedSize), nil).Maybe()
74+ f.status = domain.MirrorStatus{Repo: f.repo, State: domain.MirrorOK}
75+ f.m = usecase.NewMirrors(usecase.MirrorsConfig{Store: f.store, Keys: f.keys, Sealer: f.sealer, Pusher: f.push, Meta: meta, Git: git, Reader: f.reader, Bus: f.bus, Clock: clock, ScheduleTick: time.Hour})
76+ t.Cleanup(func() {
77+ ctx, cancel := context.WithTimeout(context.Background(), time.Second)
78+ defer cancel()
79+ f.m.Close(ctx)
80+ })
81+ return f
82+}
83+func (f *mirrorFixture) advance(d time.Duration) { f.mu.Lock(); f.at = f.at.Add(d); f.mu.Unlock() }
84+func TestMirrorsNoOpThrottleForce(t *testing.T) {
85+ f := newMirrorFixture(t)
86+ f.snapshot = domain.MirrorSnapshot{URL: "git@example.com:acme/tools", Refs: []domain.MirrorRef{{Name: "refs/heads/main", ID: mirrorOID(t)}}}
87+ v, err := f.m.Sync(context.Background(), domain.Root(), f.repo, false)
88+ require.NoError(t, err)
89+ require.True(t, v.NoOp)
90+ _, err = f.m.Sync(context.Background(), domain.Root(), f.repo, false)
91+ require.ErrorIs(t, err, domain.ErrResourceExhausted)
92+ f.push.On("Push", mock.Anything, mock.MatchedBy(func(p ports.MirrorPush) bool {
93+ return len(p.Refspecs) == 1 && p.Refspecs[0] == "refs/heads/main:refs/heads/main"
94+ })).Return(ports.MirrorPushResult{}, nil).Once()
95+ v, err = f.m.Sync(context.Background(), domain.Root(), f.repo, true)
96+ require.NoError(t, err)
97+ require.False(t, v.NoOp)
98+}
99+func TestMirrorsIncidentAndRecovery(t *testing.T) {
100+ f := newMirrorFixture(t)
101+ f.push.On("Push", mock.Anything, mock.Anything).Return(ports.MirrorPushResult{}, &domain.MirrorError{Class: domain.MirrorClassTransient}).Times(5)
102+ for i := 1; i <= 5; i++ {
103+ v, err := f.m.Sync(context.Background(), domain.Root(), f.repo, true)
104+ require.Error(t, err)
105+ require.Equal(t, i, v.Status.ConsecutiveFailures)
106+ if i < 5 {
107+ require.Equal(t, domain.MirrorFailing, v.Status.State)
108+ require.Equal(t, domain.MirrorBackoff(i), v.Status.NextRetryAt.Sub(v.Status.LastAttemptAt))
109+ } else {
110+ require.Equal(t, domain.MirrorPaused, v.Status.State)
111+ }
112+ }
113+ require.Len(t, f.events, 1)
114+ require.IsType(t, domain.MirrorFailed{}, f.events[0])
115+ f.advance(24 * time.Hour)
116+ f.m.ScheduleNow(context.Background())
117+ f.push.On("Push", mock.Anything, mock.Anything).Return(ports.MirrorPushResult{}, nil).Once()
118+ v, err := f.m.Sync(context.Background(), domain.Root(), f.repo, true)
119+ require.NoError(t, err)
120+ require.Equal(t, domain.MirrorOK, v.Status.State)
121+ require.Len(t, f.events, 2)
122+ require.IsType(t, domain.MirrorRecovered{}, f.events[1])
123+}
124+func TestMirrorsSingleFlight(t *testing.T) {
125+ f := newMirrorFixture(t)
126+ entered, release := make(chan struct{}), make(chan struct{})
127+ f.push.On("Push", mock.Anything, mock.Anything).Run(func(mock.Arguments) { close(entered); <-release }).Return(ports.MirrorPushResult{}, nil).Once()
128+ done := make(chan error, 1)
129+ go func() { _, err := f.m.Sync(context.Background(), domain.Root(), f.repo, true); done <- err }()
130+ <-entered
131+ var wg sync.WaitGroup
132+ for i := 0; i < 20; i++ {
133+ wg.Add(1)
134+ go func() {
135+ defer wg.Done()
136+ v, err := f.m.Sync(context.Background(), domain.Root(), f.repo, true)
137+ require.NoError(t, err)
138+ require.True(t, v.Queued)
139+ }()
140+ }
141+ wg.Wait()
142+ rerun := make(chan struct{})
143+ f.push.On("Push", mock.Anything, mock.Anything).Run(func(mock.Arguments) { close(rerun) }).Return(ports.MirrorPushResult{}, nil).Once()
144+ close(release)
145+ require.NoError(t, <-done)
146+ select {
147+ case <-rerun:
148+ case <-time.After(time.Second):
149+ t.Fatal("coalesced force was lost")
150+ }
151+}
152+func TestMirrorsAuthorizationAndMissingSecretsKey(t *testing.T) {
153+ m := usecase.NewMirrors(usecase.MirrorsConfig{Clock: func() *mocks.Clock { c := mocks.NewClock(t); c.On("Now").Return(now); return c }()})
154+ _, err := m.ShowKey(context.Background(), domain.Anonymous())
155+ require.ErrorIs(t, err, domain.ErrUnauthenticated)
156+ _, err = m.ShowKey(context.Background(), domain.Root())
157+ require.ErrorIs(t, err, domain.ErrPreconditionFailed)
158+ require.Contains(t, err.Error(), "ERMITE_SECRETS_KEY")
159+ _, err = m.RotateKey(context.Background(), tokenPrincipal(domain.ScopeRepoRead))
160+ require.True(t, errors.Is(err, domain.ErrForbidden))
161+}
162+
163+func mirrorOID(t *testing.T) domain.ObjectID {
164+ t.Helper()
165+ id, err := domain.ParseObjectID(oidA)
166+ require.NoError(t, err)
167+ return id
168+}
169+
170+func TestMirrorsDivergedPausedAndConfigReset(t *testing.T) {
171+ f := newMirrorFixture(t)
172+ f.push.On("Push", mock.Anything, mock.Anything).Return(ports.MirrorPushResult{}, &domain.MirrorError{Class: domain.MirrorClassDiverged}).Once()
173+ v, err := f.m.Sync(context.Background(), domain.Root(), f.repo, false)
174+ require.Error(t, err)
175+ require.Equal(t, domain.MirrorPaused, v.Status.State)
176+ for i := 0; i < 3; i++ {
177+ f.advance(time.Hour)
178+ f.m.ScheduleNow(context.Background())
179+ }
180+ f.store.On("PutRepoConfig", mock.Anything, mock.Anything).Return(nil).Once()
181+ _, err = f.m.SetRepoMirror(context.Background(), domain.Root(), domain.MirrorRepoConfig{Repo: f.repo, URL: "git@example.com:acme/tools", Cron: "*/5 * * * *"})
182+ require.NoError(t, err)
183+ f.mu.Lock()
184+ s := f.status
185+ events := len(f.events)
186+ f.mu.Unlock()
187+ require.Equal(t, domain.MirrorOK, s.State)
188+ require.Zero(t, s.ConsecutiveFailures)
189+ require.Equal(t, 1, events)
190+}
191+
192+func TestMirrorsKeyLazyGenerationAndRotation(t *testing.T) {
193+ keys, sealer, clock := mocks.NewMirrorKeyStore(t), mocks.NewSecretSealer(t), mocks.NewClock(t)
194+ clock.On("Now").Return(now)
195+ keys.On("GetMirrorKey", mock.Anything).Return([]byte(nil), time.Time{}, domain.ErrNotFound).Once()
196+ var seeds [][]byte
197+ sealer.On("Seal", mock.Anything, []byte(domain.MirrorKeyPath)).Return(func(seed, aad []byte) []byte {
198+ seeds = append(seeds, append([]byte(nil), seed...))
199+ return append([]byte("sealed:"), seed...)
200+ }).Twice()
201+ keys.On("PutMirrorKey", mock.Anything, mock.Anything, now, true).Return(nil).Once()
202+ keys.On("PutMirrorKey", mock.Anything, mock.Anything, now, false).Return(nil).Once()
203+ m := usecase.NewMirrors(usecase.MirrorsConfig{Keys: keys, Sealer: sealer, Clock: clock, PublicURL: "https://forge.example.com"})
204+ first, err := m.ShowKey(context.Background(), domain.Root())
205+ require.NoError(t, err)
206+ cached, err := m.ShowKey(context.Background(), domain.Root())
207+ require.NoError(t, err)
208+ require.Equal(t, first, cached)
209+ rotated, err := m.RotateKey(context.Background(), domain.Root())
210+ require.NoError(t, err)
211+ require.NotEqual(t, first.Fingerprint, rotated.Fingerprint)
212+ require.Contains(t, rotated.AuthorizedKey, "ermite-mirror@forge.example.com")
213+ require.Len(t, seeds, 2)
214+ require.NotEqual(t, seeds[0], seeds[1])
215+ require.Len(t, seeds[0], ed25519.SeedSize)
216+}
217+
218+func TestMirrorsPushRecordNotification(t *testing.T) {
219+ f := newPushFixture(t)
220+ n := name(t, "tools")
221+ trigger := mocks.NewMirrorTrigger(t)
222+ updates := []domain.RefUpdate{}
223+ f.meta.On("Mutate", mock.Anything, n, mock.Anything).Return(domain.Repo{Name: n}, nil)
224+ f.git.On("DefaultBranch", mock.Anything, n).Return("main", true, nil)
225+ f.bus.On("Publish", mock.Anything, mock.Anything).Return()
226+ f.sched.On("Schedule", mock.Anything, n).Return()
227+ notified := make(chan struct{})
228+ trigger.On("OnPush", mock.Anything, n, updates).Run(func(mock.Arguments) { close(notified) }).Return().Once()
229+ require.NoError(t, f.pushes.WithMirror(trigger).Record(context.Background(), domain.Root(), n, updates))
230+ <-notified
231+}
232+
233+func TestMirrorsCronDueAndDirtySkip(t *testing.T) {
234+ f := newMirrorFixture(t)
235+ // Establish a successful snapshot; subsequent clean cron ticks need no refs.
236+ f.push.On("Push", mock.Anything, mock.Anything).Return(ports.MirrorPushResult{}, nil).Once()
237+ _, err := f.m.Sync(context.Background(), domain.Root(), f.repo, true)
238+ require.NoError(t, err)
239+ f.reader.AssertNumberOfCalls(t, "ListRefs", 1)
240+ f.mu.Lock()
241+ f.at = f.at.Add(24 * time.Hour).Truncate(time.Hour).Add(time.Minute)
242+ f.mu.Unlock()
243+ f.m.ScheduleNow(context.Background())
244+ f.advance(4 * time.Minute)
245+ f.m.ScheduleNow(context.Background())
246+ f.reader.AssertNumberOfCalls(t, "ListRefs", 1)
247+ f.m.OnPush(context.Background(), f.repo, nil)
248+ read := make(chan struct{})
249+ f.reader.ExpectedCalls = nil
250+ f.reader.On("ListRefs", mock.Anything, f.repo).Run(func(mock.Arguments) { close(read) }).Return([]domain.Ref{{Name: "refs/heads/main", ID: mirrorOID(t)}}, nil).Once()
251+ // Same minute does not run, even though dirty.
252+ f.m.ScheduleNow(context.Background())
253+ select {
254+ case <-read:
255+ t.Fatal("same-minute cron reran")
256+ default:
257+ }
258+ f.advance(5 * time.Minute)
259+ f.m.ScheduleNow(context.Background())
260+ select {
261+ case <-read:
262+ case <-time.After(time.Second):
263+ t.Fatal("due dirty repository not examined")
264+ }
265+}
266+
267+func TestMirrorsAuthorizationMatrix(t *testing.T) {
268+ f := newMirrorFixture(t)
269+ ctx := context.Background()
270+ reader := tokenPrincipal(domain.ScopeRepoRead)
271+ admin := tokenPrincipal(domain.ScopeRepoAdmin)
272+ _, err := f.m.GetGlobalConfig(ctx, reader)
273+ require.ErrorIs(t, err, domain.ErrForbidden)
274+ _, err = f.m.SetGlobalConfig(ctx, reader, ports.MirrorGlobalPatch{})
275+ require.ErrorIs(t, err, domain.ErrForbidden)
276+ _, err = f.m.ListStatus(ctx, reader)
277+ require.ErrorIs(t, err, domain.ErrForbidden)
278+ _, err = f.m.RotateKey(ctx, admin)
279+ require.ErrorIs(t, err, domain.ErrForbidden)
280+ // Key show deliberately requires authentication, not owner rights.
281+ _, err = f.m.ShowKey(ctx, reader)
282+ require.NoError(t, err)
283+ _, err = f.m.ShowKey(ctx, domain.Anonymous())
284+ require.ErrorIs(t, err, domain.ErrUnauthenticated)
285+ _, err = f.m.SetRepoMirror(ctx, reader, domain.MirrorRepoConfig{Repo: f.repo})
286+ require.ErrorIs(t, err, domain.ErrForbidden)
287+ err = f.m.RemoveRepoMirror(ctx, reader, f.repo)
288+ require.ErrorIs(t, err, domain.ErrForbidden)
289+ _, err = f.m.SetRepoExcluded(ctx, reader, f.repo, true)
290+ require.ErrorIs(t, err, domain.ErrForbidden)
291+ _, err = f.m.Sync(ctx, reader, f.repo, false)
292+ require.ErrorIs(t, err, domain.ErrForbidden)
293+ f.mu.Lock()
294+ f.status.Detail, f.status.Reason = "admin detail", "admin reason"
295+ f.mu.Unlock()
296+ view, err := f.m.GetStatus(ctx, domain.Anonymous(), f.repo)
297+ require.NoError(t, err)
298+ require.Empty(t, view.Status.Detail)
299+ require.Empty(t, view.Status.Reason)
300+ require.Empty(t, view.Target.URL)
301+ view, err = f.m.GetStatus(ctx, admin, f.repo)
302+ require.NoError(t, err)
303+ require.Equal(t, "admin detail", view.Status.Detail)
304+ f.store.On("PutRepoConfig", mock.Anything, mock.Anything).Return(nil).Once()
305+ _, err = f.m.SetRepoMirror(ctx, admin, domain.MirrorRepoConfig{Repo: f.repo})
306+ require.NoError(t, err)
307+}
308+
309+func TestMirrorsPrivateStatusAuthorization(t *testing.T) {
310+ f := newMirrorFixture(t)
311+ meta := mocks.NewRepoMetaStore(t)
312+ meta.On("Get", mock.Anything, f.repo).Return(domain.Repo{Name: f.repo, Visibility: domain.VisibilityPrivate}, nil)
313+ clock := mocks.NewClock(t)
314+ clock.On("Now").Return(now)
315+ m := usecase.NewMirrors(usecase.MirrorsConfig{Meta: meta, Store: f.store, Clock: clock})
316+ _, err := m.GetStatus(context.Background(), domain.Anonymous(), f.repo)
317+ require.ErrorIs(t, err, domain.ErrNotFound)
318+ _, err = m.GetStatus(context.Background(), tokenPrincipal(domain.ScopeRepoRead, "other"), f.repo)
319+ require.ErrorIs(t, err, domain.ErrNotFound)
320+ _, err = m.GetStatus(context.Background(), tokenPrincipal(domain.ScopeRepoRead, "tools"), f.repo)
321+ require.NoError(t, err)
322+}
323+
324+func TestMirrorsCloseCancelsWorker(t *testing.T) {
325+ f := newMirrorFixture(t)
326+ entered, exited := make(chan struct{}), make(chan struct{})
327+ f.push.On("Push", mock.Anything, mock.Anything).Run(func(a mock.Arguments) { close(entered); <-a.Get(0).(context.Context).Done(); close(exited) }).Return(ports.MirrorPushResult{}, context.Canceled).Once()
328+ done := make(chan error, 1)
329+ go func() { _, err := f.m.Sync(context.Background(), domain.Root(), f.repo, true); done <- err }()
330+ <-entered
331+ ctx, cancel := context.WithTimeout(context.Background(), time.Second)
332+ defer cancel()
333+ f.m.Close(ctx)
334+ require.NoError(t, ctx.Err())
335+ select {
336+ case <-exited:
337+ default:
338+ t.Fatal("Close returned before worker exited")
339+ }
340+ require.ErrorIs(t, <-done, context.Canceled)
341+ f.m.Close(ctx) // idempotent; Start cannot resurrect workers.
342+ f.m.Start()
343+ _, err := f.m.Sync(context.Background(), domain.Root(), f.repo, true)
344+ require.ErrorIs(t, err, context.Canceled)
345+}
346+
347+func TestMirrorsCronBackoff(t *testing.T) {
348+ f := newMirrorFixture(t)
349+ f.push.On("Push", mock.Anything, mock.Anything).Return(ports.MirrorPushResult{}, &domain.MirrorError{Class: domain.MirrorClassTransient}).Once()
350+ _, err := f.m.Sync(context.Background(), domain.Root(), f.repo, false)
351+ require.Error(t, err)
352+ f.advance(time.Minute)
353+ f.m.ScheduleNow(context.Background())
354+ f.push.AssertNumberOfCalls(t, "Push", 1)
355+ attempted := make(chan struct{})
356+ f.push.On("Push", mock.Anything, mock.Anything).Run(func(mock.Arguments) { close(attempted) }).Return(ports.MirrorPushResult{}, &domain.MirrorError{Class: domain.MirrorClassTransient}).Once()
357+ f.advance(4 * time.Minute)
358+ f.m.ScheduleNow(context.Background())
359+ select {
360+ case <-attempted:
361+ case <-time.After(time.Second):
362+ t.Fatal("retry did not fire at backoff deadline")
363+ }
364+ // Cleanup joins the worker before mock verification.
365+}
366+
367+func TestMirrorsRecordDoesNotWaitForTrigger(t *testing.T) {
368+ f := newPushFixture(t)
369+ n := name(t, "tools")
370+ trigger := mocks.NewMirrorTrigger(t)
371+ entered, release, exited := make(chan struct{}), make(chan struct{}), make(chan struct{})
372+ trigger.On("OnPush", mock.Anything, n, mock.Anything).Run(func(mock.Arguments) { close(entered); <-release; close(exited) }).Return().Once()
373+ f.meta.On("Mutate", mock.Anything, n, mock.Anything).Return(domain.Repo{Name: n}, nil)
374+ f.git.On("DefaultBranch", mock.Anything, n).Return("main", true, nil)
375+ f.bus.On("Publish", mock.Anything, mock.Anything).Return()
376+ f.sched.On("Schedule", mock.Anything, n).Return()
377+ returned := make(chan error, 1)
378+ go func() { returned <- f.pushes.WithMirror(trigger).Record(context.Background(), domain.Root(), n, nil) }()
379+ <-entered
380+ select {
381+ case err := <-returned:
382+ require.NoError(t, err)
383+ case <-time.After(time.Second):
384+ close(release)
385+ <-exited
386+ t.Fatal("Record blocked on mirror trigger")
387+ }
388+ close(release)
389+ <-exited
390+}
391+
392+func TestMirrorsErrorsAndLogsDoNotExposeSecrets(t *testing.T) {
393+ f := newMirrorFixture(t)
394+ var logs bytes.Buffer
395+ ctx := zerowrap.WithCtx(context.Background(), zerowrap.New(zerowrap.Config{Level: "debug", Format: "json", Output: &logs}))
396+ credentialURL := "ssh://alice:secret-password@example.com/acme/tools"
397+ _, err := f.m.SetRepoMirror(ctx, domain.Root(), domain.MirrorRepoConfig{Repo: f.repo, URL: credentialURL})
398+ require.Error(t, err)
399+ require.NotContains(t, err.Error(), "secret-password")
400+ require.NotContains(t, err.Error(), "alice")
401+ f.push.On("Push", mock.Anything, mock.Anything).Return(ports.MirrorPushResult{}, errors.New("local key PRIVATE-KEY-MARKER ssh://alice:secret-password@example.com/acme/tools")).Once()
402+ _, err = f.m.Sync(ctx, domain.Root(), f.repo, true)
403+ require.Error(t, err)
404+ require.NotContains(t, err.Error(), "PRIVATE-KEY-MARKER")
405+ require.NotContains(t, err.Error(), "secret-password")
406+ require.NotContains(t, logs.String(), "PRIVATE-KEY-MARKER")
407+ require.NotContains(t, logs.String(), "secret-password")
408+}
409+
410+func TestMirrorsFileURLTestOption(t *testing.T) {
411+ for _, allowed := range []bool{false, true} {
412+ t.Run(fmt.Sprint(allowed), func(t *testing.T) {
413+ store, meta, clock := mocks.NewMirrorStore(t), mocks.NewRepoMetaStore(t), mocks.NewClock(t)
414+ repo := name(t, "tools")
415+ clock.On("Now").Return(now)
416+ meta.On("Get", mock.Anything, repo).Return(domain.Repo{Name: repo}, nil)
417+ global := domain.MirrorGlobalConfig{Enabled: true, URLTemplate: "file:///tmp/remotes/{repo}", Cron: "*/5 * * * *"}
418+ store.On("GetGlobal", mock.Anything).Return(global, nil).Maybe()
419+ store.On("GetRepoConfig", mock.Anything, repo).Return(domain.MirrorRepoConfig{}, domain.ErrNotFound).Maybe()
420+ store.On("GetStatus", mock.Anything, repo).Return(domain.MirrorStatus{Repo: repo}, nil).Maybe()
421+ store.On("PutRepoConfig", mock.Anything, mock.Anything).Return(nil).Maybe()
422+ store.On("PutStatus", mock.Anything, mock.Anything).Return(nil).Maybe()
423+ m := usecase.NewMirrors(usecase.MirrorsConfig{Store: store, Meta: meta, Clock: clock, AllowAnyURL: allowed})
424+ target, err := m.SetRepoMirror(context.Background(), domain.Root(), domain.MirrorRepoConfig{Repo: repo, URL: "file:///tmp/remotes/tools", Cron: "*/5 * * * *"})
425+ if allowed {
426+ require.NoError(t, err)
427+ require.Equal(t, "file:///tmp/remotes/tools", target.URL)
428+ } else {
429+ require.ErrorIs(t, err, domain.ErrInvalid)
430+ }
431+ })
432+ }
433+}
M internal/usecase/pushes.go +11 -0
@@ -17,6 +17,7 @@ import (
1717
1818 // Pushes are the use cases around a Git push.
1919 type Pushes struct {
20+ mirror ports.MirrorTrigger // nil: no outbound mirror
2021 meta ports.RepoMetaStore
2122 repos *Repos
2223 git ports.GitRepos
@@ -43,6 +44,12 @@ func NewPushes(meta ports.RepoMetaStore, repos *Repos, git ports.GitRepos, bus p
4344 return &Pushes{meta: meta, repos: repos, git: git, bus: bus, clock: clock, maint: maint, auth: auth}
4445 }
4546
47+// WithMirror makes Record notify outbound mirrors. Call before the first push.
48+func (u *Pushes) WithMirror(t ports.MirrorTrigger) *Pushes {
49+ u.mirror = t
50+ return u
51+}
52+
4653 // WithFreshener makes Record tell the reader that a repository changed, so that
4754 // it serves the push at once. Call it before the first push; it returns u.
4855 func (u *Pushes) WithFreshener(f ports.RepoFreshener) *Pushes {
@@ -262,6 +269,10 @@ func (u *Pushes) Record(ctx context.Context, p domain.Principal, name domain.Rep
262269 Repo: name, Updates: append([]domain.RefUpdate(nil), updates...), At: now,
263270 Visibility: repo.Visibility, CreatedAt: repo.CreatedAt,
264271 })
272+ if u.mirror != nil {
273+ updates := append(make([]domain.RefUpdate, 0, len(updates)), updates...)
274+ go u.mirror.OnPush(ctx, name, updates)
275+ }
265276 u.maint.Schedule(ctx, name)
266277 log.Info().
267278 Str("repo", name.String()).Int("refs", len(updates)).
@@ -17,6 +17,7 @@ import (
17 17
18 // Pushes are the use cases around a Git push.18 // Pushes are the use cases around a Git push.
19 type Pushes struct {19 type Pushes struct {
20+ mirror ports.MirrorTrigger // nil: no outbound mirror
20 meta ports.RepoMetaStore21 meta ports.RepoMetaStore
21 repos *Repos22 repos *Repos
22 git ports.GitRepos23 git ports.GitRepos
@@ -43,6 +44,12 @@ func NewPushes(meta ports.RepoMetaStore, repos *Repos, git ports.GitRepos, bus p
43 return &Pushes{meta: meta, repos: repos, git: git, bus: bus, clock: clock, maint: maint, auth: auth}44 return &Pushes{meta: meta, repos: repos, git: git, bus: bus, clock: clock, maint: maint, auth: auth}
44 }45 }
45 46
47+// WithMirror makes Record notify outbound mirrors. Call before the first push.
48+func (u *Pushes) WithMirror(t ports.MirrorTrigger) *Pushes {
49+ u.mirror = t
50+ return u
51+}
52+
46 // WithFreshener makes Record tell the reader that a repository changed, so that53 // WithFreshener makes Record tell the reader that a repository changed, so that
47 // it serves the push at once. Call it before the first push; it returns u.54 // it serves the push at once. Call it before the first push; it returns u.
48 func (u *Pushes) WithFreshener(f ports.RepoFreshener) *Pushes {55 func (u *Pushes) WithFreshener(f ports.RepoFreshener) *Pushes {
@@ -262,6 +269,10 @@ func (u *Pushes) Record(ctx context.Context, p domain.Principal, name domain.Rep
262 Repo: name, Updates: append([]domain.RefUpdate(nil), updates...), At: now,269 Repo: name, Updates: append([]domain.RefUpdate(nil), updates...), At: now,
263 Visibility: repo.Visibility, CreatedAt: repo.CreatedAt,270 Visibility: repo.Visibility, CreatedAt: repo.CreatedAt,
264 })271 })
272+ if u.mirror != nil {
273+ updates := append(make([]domain.RefUpdate, 0, len(updates)), updates...)
274+ go u.mirror.OnPush(ctx, name, updates)
275+ }
265 u.maint.Schedule(ctx, name)276 u.maint.Schedule(ctx, name)
266 log.Info().277 log.Info().
267 Str("repo", name.String()).Int("refs", len(updates)).278 Str("repo", name.String()).Int("refs", len(updates)).

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 2

Tags 44

↑↓ moveenter openesc close