1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package usecase_test
4
5import (
6 "context"
7 "errors"
8 "fmt"
9 "sync"
10 "sync/atomic"
11 "testing"
12 "testing/synctest"
13 "time"
14
15 "github.com/stretchr/testify/assert"
16 "github.com/stretchr/testify/mock"
17 "github.com/stretchr/testify/require"
18
19 "bnema.dev/ermite/internal/domain"
20 "bnema.dev/ermite/internal/ports"
21 "bnema.dev/ermite/internal/ports/mocks"
22 "bnema.dev/ermite/internal/usecase"
23)
24
25// eventRig is an Events use case fed by a channel the test writes to.
26type eventRig struct {
27 t *testing.T
28 meta *mocks.RepoMetaStore
29 git *mocks.GitRepos
30 feed chan domain.Event
31 auth *mocks.Revalidator
32 revoked sync.Map // token IDs revoked by the test
33 revalidations atomic.Int32 // calls to auth.Revalidate
34 events *usecase.Events
35}
36
37func newEventRig(t *testing.T, size int) *eventRig {
38 t.Helper()
39 return newEventRigConfig(t, usecase.EventsConfig{Size: size})
40}
41
42func newEventRigConfig(t *testing.T, cfg usecase.EventsConfig) *eventRig {
43 t.Helper()
44 r := &eventRig{
45 t: t,
46 meta: mocks.NewRepoMetaStore(t),
47 git: mocks.NewGitRepos(t),
48 feed: make(chan domain.Event, 4096),
49 auth: mocks.NewRevalidator(t),
50 }
51 // Tokens are valid until the test revokes them.
52 r.auth.EXPECT().Revalidate(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, p domain.Principal) error {
53 r.revalidations.Add(1)
54 if _, gone := r.revoked.Load(p.TokenID()); gone {
55 return domain.ErrUnauthenticated
56 }
57 return nil
58 }).Maybe()
59 cfg.Auth = r.auth
60 bus := mocks.NewEventBus(t)
61 bus.EXPECT().Subscribe(mock.Anything).RunAndReturn(func(ctx context.Context) <-chan domain.Event {
62 go func() { <-ctx.Done(); close(r.feed) }()
63 return r.feed
64 })
65 clock := mocks.NewClock(t)
66 clock.EXPECT().Now().Return(now).Maybe()
67 r.git.EXPECT().HeadBranch(mock.Anything, mock.Anything).Return("main", nil).Maybe()
68 repos := usecase.NewRepos(r.meta, r.git, bus, clock)
69 r.events = usecase.NewEvents(context.Background(), repos, bus, clock, cfg)
70 t.Cleanup(func() { r.events.Close(context.Background()) })
71 return r
72}
73
74// repo declares a repository the rig knows.
75func (r *eventRig) repo(s string, vis domain.Visibility) domain.RepoName {
76 n := name(r.t, s)
77 r.meta.EXPECT().Get(mock.Anything, n).Return(domain.Repo{Name: n, Visibility: vis, CreatedAt: now}, nil).Maybe()
78 return n
79}
80
81// pushEvent builds the PushEvent the Pushes use case would publish now: it
82// carries the repository as the store answers at this moment.
83func (r *eventRig) pushEvent(n domain.RepoName, oid, ref string) domain.PushEvent {
84 repo, err := r.meta.Get(context.Background(), n)
85 require.NoError(r.t, err)
86 return domain.PushEvent{
87 Repo: n, Updates: []domain.RefUpdate{{Old: domain.ZeroOID, New: oid, Ref: ref}}, At: now,
88 Visibility: repo.Visibility, CreatedAt: repo.CreatedAt,
89 }
90}
91
92func (r *eventRig) push(n domain.RepoName, i int) {
93 r.feed <- r.pushEvent(n, fmt.Sprintf("%040x", i+1), "refs/heads/main")
94}
95
96// watcher is a running Watch.
97type watcher struct {
98 got chan ports.WatchEvent
99 done chan error
100}
101
102func (r *eventRig) watch(ctx context.Context, p domain.Principal, params ports.WatchParams) *watcher {
103 w := &watcher{got: make(chan ports.WatchEvent, 4096), done: make(chan error, 1)}
104 go func() {
105 w.done <- r.events.Watch(ctx, p, params, func(e ports.WatchEvent) error {
106 w.got <- e
107 return nil
108 })
109 }()
110 return w
111}
112
113func (w *watcher) next(t *testing.T) ports.WatchEvent {
114 t.Helper()
115 synctest.Wait() // everything the event can reach has run
116 select {
117 case e := <-w.got:
118 return e
119 case err := <-w.done:
120 t.Fatalf("watch ended: %v", err)
121 default:
122 t.Fatal("no event")
123 }
124 return ports.WatchEvent{}
125}
126
127// sentinelOID marks the push events that only prove the ones before them were
128// handled.
129var sentinelOID = fmt.Sprintf("%040x", 0xfeed)
130
131// quiet proves that the watcher got nothing since its last read: it pushes an
132// event on n, which the watcher may see, and requires it to be the next one.
133// Events reach a watcher in order, so anything sent before would show up first.
134func (w *watcher) quiet(t *testing.T, r *eventRig, n domain.RepoName) {
135 t.Helper()
136 r.feed <- r.pushEvent(n, sentinelOID, "refs/heads/sentinel")
137 e := w.next(t)
138 push, ok := e.Event.(domain.PushEvent)
139 if !ok || len(push.Updates) != 1 || push.Updates[0].New != sentinelOID {
140 t.Fatalf("expected nothing before the sentinel, got %+v", e)
141 }
142}
143
144func TestWatchLiveAndFiltered(t *testing.T) {
145 t.Parallel()
146 synctest.Test(t, func(t *testing.T) {
147 r := newEventRig(t, 0)
148 pub := r.repo("pub", domain.VisibilityPublic)
149 priv := r.repo("priv", domain.VisibilityPrivate)
150 ctx, cancel := context.WithCancel(context.Background())
151 defer cancel()
152
153 anon := r.watch(ctx, domain.Anonymous(), ports.WatchParams{})
154 root := r.watch(ctx, domain.Root(), ports.WatchParams{Topics: []string{"home"}})
155 scoped := r.watch(ctx, tokenPrincipal(domain.ScopeRepoRead, "other"), ports.WatchParams{})
156 onlyPriv := r.watch(ctx, domain.Root(), ports.WatchParams{Topics: []string{"repo:priv"}})
157 synctest.Wait()
158 require.Equal(t, 4, r.events.Watchers())
159
160 r.push(priv, 1)
161 r.push(pub, 2)
162
163 // Anonymous and a token limited to another repository never see priv.
164 for name, w := range map[string]*watcher{"anonymous": anon, "scoped": scoped} {
165 e := w.next(t)
166 assert.Equal(t, pub, e.Event.(domain.PushEvent).Repo, name)
167 w.quiet(t, r, pub)
168 }
169 first, second := root.next(t), root.next(t)
170 assert.Equal(t, priv, first.Event.(domain.PushEvent).Repo)
171 assert.Equal(t, pub, second.Event.(domain.PushEvent).Repo)
172 assert.NotEmpty(t, first.ID)
173 assert.NotEqual(t, first.ID, second.ID)
174
175 assert.Equal(t, priv, onlyPriv.next(t).Event.(domain.PushEvent).Repo)
176 onlyPriv.quiet(t, r, priv)
177 })
178}
179
180func TestWatchRechecksAccessPerEvent(t *testing.T) {
181 t.Parallel()
182 synctest.Test(t, func(t *testing.T) {
183 r := newEventRig(t, 0)
184 n := name(t, "flip")
185 vis := domain.VisibilityPublic
186 var mu sync.Mutex
187 r.meta.EXPECT().Get(mock.Anything, n).RunAndReturn(func(context.Context, domain.RepoName) (domain.Repo, error) {
188 mu.Lock()
189 defer mu.Unlock()
190 return domain.Repo{Name: n, Visibility: vis, CreatedAt: now}, nil
191 })
192 ctx, cancel := context.WithCancel(context.Background())
193 defer cancel()
194 w := r.watch(ctx, domain.Anonymous(), ports.WatchParams{})
195 synctest.Wait()
196 require.Equal(t, 1, r.events.Watchers())
197
198 r.push(n, 1)
199 w.next(t)
200
201 mu.Lock()
202 vis = domain.VisibilityPrivate
203 mu.Unlock()
204 r.push(n, 2)
205 w.quiet(t, r, r.repo("open", domain.VisibilityPublic)) // push 2 is not delivered to anonymous
206 })
207}
208
209func TestWatchResumeAndGap(t *testing.T) {
210 t.Parallel()
211 const size = 8
212 // Each subtest gets its own bubble and a fresh run of 20 events recorded
213 // through a first watcher; the ring then holds the last 8.
214 setup := func(t *testing.T) (r *eventRig, n domain.RepoName, ctx context.Context, ids []string) {
215 r = newEventRig(t, size)
216 n = r.repo("pub", domain.VisibilityPublic)
217 ctx, cancel := context.WithCancel(context.Background())
218 t.Cleanup(cancel)
219
220 first := r.watch(ctx, domain.Root(), ports.WatchParams{})
221 synctest.Wait()
222 require.Equal(t, 1, r.events.Watchers())
223 // One at a time: this watcher must not fall behind the small ring.
224 for i := range 20 {
225 r.push(n, i)
226 ids = append(ids, first.next(t).ID)
227 }
228 require.Len(t, ids, 20)
229 return r, n, ctx, ids
230 }
231
232 t.Run("an id that fell out of the buffer gives a gap, then the oldest buffered event", func(t *testing.T) {
233 t.Parallel()
234 synctest.Test(t, func(t *testing.T) {
235 r, _, ctx, ids := setup(t)
236 // 20 events were taken, the buffer holds the last 8: ids[12] is the
237 // oldest still present; ids[0] is long gone.
238 w := r.watch(ctx, domain.Root(), ports.WatchParams{ResumeAfter: ids[0]})
239 gap := w.next(t)
240 assert.True(t, gap.Gap)
241 assert.Empty(t, gap.ID)
242 oldest := w.next(t)
243 assert.False(t, oldest.Gap)
244 assert.Equal(t, ids[12], oldest.ID)
245 })
246 })
247
248 t.Run("an unknown id gives a gap", func(t *testing.T) {
249 t.Parallel()
250 synctest.Test(t, func(t *testing.T) {
251 r, _, ctx, _ := setup(t)
252 for _, id := range []string{"nonsense", "zz-1", "-", "0-0"} {
253 w := r.watch(ctx, domain.Root(), ports.WatchParams{ResumeAfter: id})
254 assert.True(t, w.next(t).Gap, id)
255 }
256 })
257 })
258
259 t.Run("an id of a later run is a gap, not a newer event", func(t *testing.T) {
260 t.Parallel()
261 synctest.Test(t, func(t *testing.T) {
262 r, _, ctx, _ := setup(t)
263 w := r.watch(ctx, domain.Root(), ports.WatchParams{ResumeAfter: "otherboot-999"})
264 assert.True(t, w.next(t).Gap)
265 })
266 })
267
268 t.Run("resume replays the buffered events after the id, then goes live", func(t *testing.T) {
269 t.Parallel()
270 synctest.Test(t, func(t *testing.T) {
271 r, n, ctx, ids := setup(t)
272 w := r.watch(ctx, domain.Root(), ports.WatchParams{ResumeAfter: ids[16]})
273 for _, want := range ids[17:] {
274 e := w.next(t)
275 assert.False(t, e.Gap)
276 assert.Equal(t, want, e.ID)
277 }
278 // Nothing else was replayed: the next event is the one pushed now.
279 r.push(n, 98)
280 live := w.next(t)
281 assert.False(t, live.Gap)
282 assert.NotContains(t, ids, live.ID)
283 })
284 })
285
286 t.Run("resume at the last event replays nothing, then goes live", func(t *testing.T) {
287 t.Parallel()
288 synctest.Test(t, func(t *testing.T) {
289 r, n, ctx, ids := setup(t)
290 w := r.watch(ctx, domain.Root(), ports.WatchParams{ResumeAfter: ids[len(ids)-1]})
291 r.push(n, 99)
292 e := w.next(t)
293 assert.False(t, e.Gap)
294 assert.NotContains(t, ids, e.ID, "the first event is the new one, nothing was replayed")
295 })
296 })
297}
298
299func TestWatchTopicErrors(t *testing.T) {
300 t.Parallel()
301 r := newEventRig(t, 0)
302 r.repo("pub", domain.VisibilityPublic)
303 r.repo("priv", domain.VisibilityPrivate)
304 r.meta.EXPECT().Get(mock.Anything, name(t, "gone")).Return(domain.Repo{}, domain.ErrNotFound).Maybe()
305 noop := func(ports.WatchEvent) error { return nil }
306 ctx := context.Background()
307
308 for topic, want := range map[string]error{
309 "nope": domain.ErrInvalid,
310 "repo:Bad": domain.ErrInvalidName,
311 "repo:": domain.ErrInvalidName,
312 "repo:gone": domain.ErrNotFound,
313 "repo:priv": domain.ErrNotFound, // hidden from anonymous
314 "repo:pub:foo": domain.ErrInvalidName,
315 } {
316 err := r.events.Watch(ctx, domain.Anonymous(), ports.WatchParams{Topics: []string{topic}}, noop)
317 assert.ErrorIs(t, err, want, topic)
318 }
319}
320
321func TestWatchEndsCleanly(t *testing.T) {
322 t.Parallel()
323 t.Run("on context cancellation", func(t *testing.T) {
324 t.Parallel()
325 synctest.Test(t, func(t *testing.T) {
326 r := newEventRig(t, 0)
327 ctx, cancel := context.WithCancel(context.Background())
328 w := r.watch(ctx, domain.Root(), ports.WatchParams{})
329 synctest.Wait()
330 require.Equal(t, 1, r.events.Watchers())
331 cancel()
332 synctest.Wait()
333 select {
334 case err := <-w.done:
335 assert.NoError(t, err)
336 default:
337 t.Fatal("watch did not end")
338 }
339 assert.Zero(t, r.events.Watchers())
340 })
341 })
342
343 t.Run("on Close", func(t *testing.T) {
344 t.Parallel()
345 synctest.Test(t, func(t *testing.T) {
346 r := newEventRig(t, 0)
347 w := r.watch(context.Background(), domain.Root(), ports.WatchParams{})
348 synctest.Wait()
349 require.Equal(t, 1, r.events.Watchers())
350 r.events.Close(context.Background())
351 r.events.Close(context.Background()) // twice is fine
352 synctest.Wait()
353 select {
354 case err := <-w.done:
355 assert.NoError(t, err)
356 default:
357 t.Fatal("watch did not end")
358 }
359 })
360 })
361
362 t.Run("a failing send ends the watch with its error", func(t *testing.T) {
363 t.Parallel()
364 synctest.Test(t, func(t *testing.T) {
365 r := newEventRig(t, 0)
366 n := r.repo("pub", domain.VisibilityPublic)
367 boom := errors.New("client went away")
368 done := make(chan error, 1)
369 go func() {
370 done <- r.events.Watch(context.Background(), domain.Root(), ports.WatchParams{}, func(ports.WatchEvent) error { return boom })
371 }()
372 synctest.Wait()
373 require.Equal(t, 1, r.events.Watchers())
374 r.push(n, 1)
375 assert.ErrorIs(t, <-done, boom)
376 })
377 })
378}
379
380func TestWatchSlowWatcherGetsAGap(t *testing.T) {
381 t.Parallel()
382 synctest.Test(t, func(t *testing.T) {
383 const size = 4
384 r := newEventRig(t, size)
385 n := r.repo("pub", domain.VisibilityPublic)
386
387 release := make(chan struct{})
388 var mu sync.Mutex
389 var got []ports.WatchEvent
390 done := make(chan error, 1)
391 blocked := make(chan struct{})
392 first := true
393 go func() {
394 done <- r.events.Watch(context.Background(), domain.Root(), ports.WatchParams{}, func(e ports.WatchEvent) error {
395 if first { // block on the first event while the ring turns over
396 first = false
397 close(blocked)
398 <-release
399 }
400 mu.Lock()
401 got = append(got, e)
402 mu.Unlock()
403 return nil
404 })
405 }()
406 // A second, fast watcher tells when the use case has taken every event:
407 // it receives them in order.
408 fast := r.watch(context.Background(), domain.Root(), ports.WatchParams{})
409 synctest.Wait()
410 require.Equal(t, 2, r.events.Watchers())
411 r.push(n, 0)
412 <-blocked
413 for i := 1; i < 30; i++ {
414 r.push(n, i)
415 }
416 // The ring holds four events: the fast watcher may get a gap, but it ends
417 // on the last push, which proves the use case took all of them.
418 last := fmt.Sprintf("%040x", 29+1)
419 for {
420 e := fast.next(t)
421 if p, ok := e.Event.(domain.PushEvent); ok && p.Updates[0].New == last {
422 break
423 }
424 }
425 close(release)
426
427 synctest.Wait()
428 mu.Lock()
429 defer mu.Unlock()
430 require.GreaterOrEqual(t, len(got), 2)
431 hasGap := false
432 for _, e := range got[1:] {
433 hasGap = hasGap || e.Gap
434 }
435 assert.True(t, hasGap, "events were lost while the watcher was blocked: %+v", got)
436 })
437}
438
439func TestEventsRepoPayloads(t *testing.T) {
440 t.Parallel()
441 synctest.Test(t, func(t *testing.T) {
442 r := newEventRig(t, 0)
443 pub := r.repo("pub", domain.VisibilityPublic)
444 priv := r.repo("priv", domain.VisibilityPrivate)
445 ctx, cancel := context.WithCancel(context.Background())
446 defer cancel()
447 anon := r.watch(ctx, domain.Anonymous(), ports.WatchParams{})
448 root := r.watch(ctx, domain.Root(), ports.WatchParams{})
449 synctest.Wait()
450 require.Equal(t, 2, r.events.Watchers())
451
452 r.feed <- domain.RepoCreated{Repo: pub, Visibility: domain.VisibilityPublic, At: now}
453 r.feed <- domain.RepoUpdated{Repo: pub, At: now, Visibility: domain.VisibilityPublic, Previous: domain.VisibilityPublic, CreatedAt: now}
454 // A deleted repository cannot be read any more: its visibility decides.
455 r.feed <- domain.RepoDeleted{Repo: priv, Visibility: domain.VisibilityPrivate, At: now}
456 r.feed <- domain.RepoDeleted{Repo: pub, Visibility: domain.VisibilityPublic, At: now}
457
458 for _, w := range []*watcher{anon, root} {
459 created := w.next(t)
460 require.NotNil(t, created.Repo)
461 assert.Equal(t, "main", created.Repo.DefaultBranch)
462 assert.Equal(t, pub, created.Repo.Name)
463 require.NotNil(t, w.next(t).Repo)
464 }
465 assert.Equal(t, priv, root.next(t).Event.(domain.RepoDeleted).Repo)
466 assert.Equal(t, pub, root.next(t).Event.(domain.RepoDeleted).Repo)
467 assert.Equal(t, pub, anon.next(t).Event.(domain.RepoDeleted).Repo, "the private deletion was skipped")
468 anon.quiet(t, r, pub)
469 })
470}
471
472// tokenWith builds the principal of a stored token with an id and an expiry.
473func tokenWith(id string, expires time.Time, repos ...string) domain.Principal {
474 tok := domain.Token{ID: id, Scopes: []domain.Scope{domain.ScopeRepoRead}, ExpiresAt: expires}
475 for _, r := range repos {
476 n, err := domain.ParseRepoName(r)
477 if err != nil {
478 panic(err)
479 }
480 tok.Repos = append(tok.Repos, n)
481 }
482 return domain.TokenPrincipal(tok)
483}
484
485func TestWatchEndsWhenTheTokenIsRevoked(t *testing.T) {
486 t.Parallel()
487 synctest.Test(t, func(t *testing.T) {
488 // Asking at every turn keeps the test independent of the clock.
489 r := newEventRigConfig(t, usecase.EventsConfig{RevalidateEvery: -1})
490 pub := r.repo("pub", domain.VisibilityPublic)
491 priv := r.repo("priv", domain.VisibilityPrivate)
492 p := tokenWith("tok", time.Time{})
493 ctx, cancel := context.WithCancel(context.Background())
494 defer cancel()
495
496 w := r.watch(ctx, p, ports.WatchParams{})
497 synctest.Wait()
498 require.Equal(t, 1, r.events.Watchers())
499 r.push(pub, 1)
500 w.next(t)
501
502 r.revoked.Store("tok", true)
503 r.push(priv, 2) // would be delivered to a token that may read priv
504 synctest.Wait()
505 select {
506 case err := <-w.done:
507 assert.ErrorIs(t, err, domain.ErrUnauthenticated)
508 default:
509 t.Fatal("the watch of a revoked token did not end")
510 }
511 select {
512 case e := <-w.got:
513 t.Fatalf("an event was sent after the revocation: %+v", e)
514 default:
515 }
516 })
517}
518
519func TestWatchEndsWhenTheTokenExpires(t *testing.T) {
520 t.Parallel()
521 synctest.Test(t, func(t *testing.T) {
522 r := newEventRig(t, 0)
523 pub := r.repo("pub", domain.VisibilityPublic)
524 // The rig's clock stands at now: this token expired at now.
525 p := tokenWith("tok", now)
526 w := r.watch(context.Background(), p, ports.WatchParams{})
527 r.push(pub, 1) // whether or not the watch had started, it ends unauthenticated
528 synctest.Wait()
529 select {
530 case err := <-w.done:
531 assert.ErrorIs(t, err, domain.ErrUnauthenticated)
532 default:
533 t.Fatal("the watch of an expired token did not end")
534 }
535 assert.Empty(t, w.got)
536 })
537}
538
539func TestWatchRevalidatesIdleWatchersAndCachesTheAnswer(t *testing.T) {
540 t.Parallel()
541 t.Run("a quiet forge still notices a revocation", func(t *testing.T) {
542 t.Parallel()
543 synctest.Test(t, func(t *testing.T) {
544 r := newEventRigConfig(t, usecase.EventsConfig{}) // the production default interval
545 w := r.watch(context.Background(), tokenWith("tok", time.Time{}), ports.WatchParams{})
546 synctest.Wait()
547 require.Equal(t, 1, r.events.Watchers())
548 r.revoked.Store("tok", true)
549 synctest.Sleep(usecase.DefaultRevalidateEvery - time.Nanosecond)
550 select {
551 case err := <-w.done:
552 t.Fatalf("the watch ended before the interval: %v", err)
553 default:
554 }
555 assert.Zero(t, r.revalidations.Load(), "no store lookup before the interval")
556 synctest.Sleep(time.Nanosecond)
557 select {
558 case err := <-w.done:
559 assert.ErrorIs(t, err, domain.ErrUnauthenticated)
560 default:
561 t.Fatal("an idle watch of a revoked token did not end at the interval")
562 }
563 assert.Equal(t, int32(1), r.revalidations.Load())
564 })
565 })
566
567 t.Run("the answer is kept between checks", func(t *testing.T) {
568 t.Parallel()
569 synctest.Test(t, func(t *testing.T) {
570 r := newEventRigConfig(t, usecase.EventsConfig{RevalidateEvery: time.Hour})
571 pub := r.repo("pub", domain.VisibilityPublic)
572 w := r.watch(context.Background(), tokenWith("tok", time.Time{}), ports.WatchParams{})
573 synctest.Wait()
574 require.Equal(t, 1, r.events.Watchers())
575 for i := range 20 {
576 r.push(pub, i)
577 }
578 for range 20 {
579 w.next(t)
580 }
581 // Idle, the watcher wakes up every few seconds but keeps its answer
582 // until the interval is over.
583 synctest.Sleep(time.Hour - time.Nanosecond)
584 assert.Zero(t, r.revalidations.Load(), "no store lookup within the interval")
585 synctest.Sleep(time.Nanosecond)
586 assert.Equal(t, int32(1), r.revalidations.Load(), "one lookup when the interval is over")
587 })
588 })
589
590 t.Run("root and anonymous never ask", func(t *testing.T) {
591 t.Parallel()
592 synctest.Test(t, func(t *testing.T) {
593 r := newEventRigConfig(t, usecase.EventsConfig{RevalidateEvery: -1})
594 pub := r.repo("pub", domain.VisibilityPublic)
595 a := r.watch(context.Background(), domain.Anonymous(), ports.WatchParams{})
596 b := r.watch(context.Background(), domain.Root(), ports.WatchParams{})
597 synctest.Wait()
598 require.Equal(t, 2, r.events.Watchers())
599 r.push(pub, 1)
600 a.next(t)
601 b.next(t)
602 r.auth.AssertNotCalled(t, "Revalidate", mock.Anything, mock.Anything)
603 })
604 })
605}
606
607func TestWatchLimits(t *testing.T) {
608 t.Parallel()
609 noop := func(ports.WatchEvent) error { return nil }
610
611 t.Run("too many topics", func(t *testing.T) {
612 t.Parallel()
613 synctest.Test(t, func(t *testing.T) {
614 r := newEventRig(t, 0)
615 r.repo("pub", domain.VisibilityPublic)
616 ok := make([]string, usecase.MaxWatchTopics)
617 for i := range ok {
618 ok[i] = "home"
619 }
620 ctx, cancel := context.WithCancel(context.Background())
621 cancel() // returns at once once the topics are accepted
622 assert.NoError(t, r.events.Watch(ctx, domain.Root(), ports.WatchParams{Topics: ok}, noop))
623 err := r.events.Watch(context.Background(), domain.Root(), ports.WatchParams{Topics: append(ok, "home")}, noop)
624 assert.ErrorIs(t, err, domain.ErrInvalid)
625 })
626 })
627
628 t.Run("total watchers", func(t *testing.T) {
629 t.Parallel()
630 synctest.Test(t, func(t *testing.T) {
631 r := newEventRig(t, 0)
632 ctx, cancel := context.WithCancel(context.Background())
633 defer cancel()
634 for range usecase.MaxWatchers {
635 r.watch(ctx, domain.Root(), ports.WatchParams{})
636 }
637 synctest.Wait()
638 require.Equal(t, usecase.MaxWatchers, r.events.Watchers())
639 err := r.events.Watch(ctx, domain.Root(), ports.WatchParams{}, noop)
640 assert.ErrorIs(t, err, domain.ErrResourceExhausted)
641 assert.Equal(t, usecase.MaxWatchers, r.events.Watchers(), "a refused watch is not counted")
642
643 cancel() // all end; a new one is accepted again
644 synctest.Wait()
645 require.Equal(t, 0, r.events.Watchers())
646 again, stop := context.WithCancel(context.Background())
647 stop()
648 assert.NoError(t, r.events.Watch(again, domain.Root(), ports.WatchParams{}, noop))
649 })
650 })
651
652 t.Run("watchers per client", func(t *testing.T) {
653 t.Parallel()
654 synctest.Test(t, func(t *testing.T) {
655 r := newEventRig(t, 0)
656 ctx, cancel := context.WithCancel(context.Background())
657 defer cancel()
658 for range usecase.MaxWatchersPerClient {
659 r.watch(ctx, domain.Root(), ports.WatchParams{Client: "198.51.100.1"})
660 }
661 synctest.Wait()
662 require.Equal(t, usecase.MaxWatchersPerClient, r.events.Watchers())
663
664 err := r.events.Watch(ctx, domain.Root(), ports.WatchParams{Client: "198.51.100.1"}, noop)
665 assert.ErrorIs(t, err, domain.ErrResourceExhausted)
666
667 // Another client, and callers in process, are not affected.
668 r.watch(ctx, domain.Root(), ports.WatchParams{Client: "198.51.100.2"})
669 r.watch(ctx, domain.Root(), ports.WatchParams{})
670 synctest.Wait()
671 require.Equal(t, usecase.MaxWatchersPerClient+2, r.events.Watchers())
672
673 cancel()
674 synctest.Wait()
675 require.Equal(t, 0, r.events.Watchers())
676 again, stop := context.WithCancel(context.Background())
677 stop()
678 assert.NoError(t, r.events.Watch(again, domain.Root(), ports.WatchParams{Client: "198.51.100.1"}, noop), "the slots were released")
679 })
680 })
681}
682
683// repoAt declares a repository whose metadata the test changes.
684type liveRepo struct {
685 mu sync.Mutex
686 repo domain.Repo
687}
688
689func (l *liveRepo) set(f func(*domain.Repo)) {
690 l.mu.Lock()
691 defer l.mu.Unlock()
692 f(&l.repo)
693}
694
695func (r *eventRig) liveRepo(s string, vis domain.Visibility, created time.Time) (*liveRepo, domain.RepoName) {
696 n := name(r.t, s)
697 l := &liveRepo{repo: domain.Repo{Name: n, Visibility: vis, CreatedAt: created}}
698 r.meta.EXPECT().Get(mock.Anything, n).RunAndReturn(func(context.Context, domain.RepoName) (domain.Repo, error) {
699 l.mu.Lock()
700 defer l.mu.Unlock()
701 return l.repo, nil
702 }).Maybe()
703 return l, n
704}
705
706func TestEventsKeepTheVisibilityOfTheirTime(t *testing.T) {
707 t.Parallel()
708 t.Run("a repository made public does not reveal what happened while it was private", func(t *testing.T) {
709 t.Parallel()
710 synctest.Test(t, func(t *testing.T) {
711 ctx, cancel := context.WithCancel(context.Background())
712 defer cancel()
713 r := newEventRig(t, 0)
714 l, n := r.liveRepo("secret", domain.VisibilityPrivate, now)
715 root := r.watch(ctx, domain.Root(), ports.WatchParams{})
716 synctest.Wait()
717 require.Equal(t, 1, r.events.Watchers())
718 r.push(n, 1)
719 r.push(n, 2)
720 root.next(t)
721 root.next(t) // both are buffered now
722
723 l.set(func(repo *domain.Repo) { repo.Visibility = domain.VisibilityPublic })
724 r.feed <- domain.RepoUpdated{Repo: n, At: now, Visibility: domain.VisibilityPublic, Previous: domain.VisibilityPrivate, CreatedAt: now}
725 root.next(t)
726 r.push(n, 3) // public era
727
728 // An anonymous caller that resumes with an id nobody issued gets what
729 // is buffered, filtered: the public update and the public push, not the
730 // pushes of the private era.
731 anon := r.watch(ctx, domain.Anonymous(), ports.WatchParams{ResumeAfter: "bogus"})
732 assert.True(t, anon.next(t).Gap)
733 upd := anon.next(t)
734 require.IsType(t, domain.RepoUpdated{}, upd.Event)
735 push := anon.next(t)
736 require.IsType(t, domain.PushEvent{}, push.Event)
737 assert.Equal(t, fmt.Sprintf("%040x", 3+1), push.Event.(domain.PushEvent).Updates[0].New)
738 anon.quiet(t, r, n)
739 })
740 })
741
742 t.Run("a repository deleted and created again does not inherit old events", func(t *testing.T) {
743 t.Parallel()
744 synctest.Test(t, func(t *testing.T) {
745 ctx, cancel := context.WithCancel(context.Background())
746 defer cancel()
747 r := newEventRig(t, 0)
748 first := now.Add(-time.Hour)
749 l, n := r.liveRepo("again", domain.VisibilityPublic, first)
750 root := r.watch(ctx, domain.Root(), ports.WatchParams{})
751 synctest.Wait()
752 require.Equal(t, 1, r.events.Watchers())
753 r.push(n, 1)
754 root.next(t)
755
756 l.set(func(repo *domain.Repo) { repo.CreatedAt = now }) // deleted, created again
757 r.push(n, 2)
758 root.next(t)
759
760 anon := r.watch(ctx, domain.Anonymous(), ports.WatchParams{ResumeAfter: "bogus"})
761 assert.True(t, anon.next(t).Gap)
762 e := anon.next(t)
763 require.IsType(t, domain.PushEvent{}, e.Event)
764 assert.Equal(t, fmt.Sprintf("%040x", 2+1), e.Event.(domain.PushEvent).Updates[0].New, "only the push to the new repository")
765 anon.quiet(t, r, n)
766 })
767 })
768
769 t.Run("a repository made private tells those who could read it, once", func(t *testing.T) {
770 t.Parallel()
771 synctest.Test(t, func(t *testing.T) {
772 ctx, cancel := context.WithCancel(context.Background())
773 defer cancel()
774 r := newEventRig(t, 0)
775 l, n := r.liveRepo("hide", domain.VisibilityPublic, now)
776 anon := r.watch(ctx, domain.Anonymous(), ports.WatchParams{})
777 reader := r.watch(ctx, tokenPrincipal(domain.ScopeRepoRead, "hide"), ports.WatchParams{})
778 other := r.watch(ctx, tokenPrincipal(domain.ScopeRepoRead, "elsewhere"), ports.WatchParams{})
779 root := r.watch(ctx, domain.Root(), ports.WatchParams{})
780 synctest.Wait()
781 require.Equal(t, 4, r.events.Watchers())
782
783 l.set(func(repo *domain.Repo) { repo.Visibility = domain.VisibilityPrivate })
784 r.feed <- domain.RepoUpdated{Repo: n, At: now, Visibility: domain.VisibilityPrivate, Previous: domain.VisibilityPublic, CreatedAt: now}
785
786 for name, w := range map[string]*watcher{"anonymous": anon, "reader": reader, "other": other, "root": root} {
787 e := w.next(t)
788 up, ok := e.Event.(domain.RepoUpdated)
789 require.True(t, ok, name)
790 assert.Equal(t, n, up.Repo, name)
791 require.NotNil(t, e.Repo, name)
792 assert.Equal(t, domain.VisibilityPrivate, e.Repo.Visibility, name)
793 assert.Empty(t, e.Repo.Description, "nothing but the name and the visibility", name)
794 }
795
796 // Later events of the repository reach those who may read it now, only.
797 r.push(n, 1)
798 assert.IsType(t, domain.PushEvent{}, reader.next(t).Event)
799 assert.IsType(t, domain.PushEvent{}, root.next(t).Event)
800 anon.quiet(t, r, r.repo("open", domain.VisibilityPublic))
801 other.quiet(t, r, r.repo("open2", domain.VisibilityPublic))
802 })
803 })
804
805 t.Run("a private repository that never was public stays hidden", func(t *testing.T) {
806 t.Parallel()
807 synctest.Test(t, func(t *testing.T) {
808 ctx, cancel := context.WithCancel(context.Background())
809 defer cancel()
810 r := newEventRig(t, 0)
811 _, n := r.liveRepo("closed", domain.VisibilityPrivate, now)
812 anon := r.watch(ctx, domain.Anonymous(), ports.WatchParams{})
813 synctest.Wait()
814 require.Equal(t, 1, r.events.Watchers())
815 r.feed <- domain.RepoUpdated{Repo: n, At: now, Visibility: domain.VisibilityPrivate, Previous: domain.VisibilityPrivate, CreatedAt: now}
816 r.push(n, 1)
817 anon.quiet(t, r, r.repo("open", domain.VisibilityPublic))
818 })
819 })
820}
821
822// TestPushEventsKeepTheirSnapshot: a push is told by the repository as it was
823// when published, not as the store says when the feeder gets to it.
824func TestPushEventsKeepTheirSnapshot(t *testing.T) {
825 t.Parallel()
826 tests := []struct {
827 name string
828 swap func(repo *domain.Repo)
829 rootSees bool // the repository is still the one the push is about
830 }{
831 {"made public before the feeder appends", func(repo *domain.Repo) { repo.Visibility = domain.VisibilityPublic }, true},
832 {"recreated before the feeder appends", func(repo *domain.Repo) {
833 repo.Visibility = domain.VisibilityPublic
834 repo.CreatedAt = now.Add(time.Hour)
835 }, false},
836 }
837 for _, tt := range tests {
838 t.Run(tt.name, func(t *testing.T) {
839 t.Parallel()
840 synctest.Test(t, func(t *testing.T) {
841 ctx, cancel := context.WithCancel(context.Background())
842 defer cancel()
843 r := newEventRig(t, 0)
844 l, n := r.liveRepo("secret", domain.VisibilityPrivate, now)
845 anon := r.watch(ctx, domain.Anonymous(), ports.WatchParams{})
846 root := r.watch(ctx, domain.Root(), ports.WatchParams{})
847 synctest.Wait()
848 require.Equal(t, 2, r.events.Watchers())
849
850 // Published while private; the store changes before the feeder
851 // can have looked at it: the event is built, then the state is
852 // swapped, and only then is it handed to the feeder.
853 ev := r.pushEvent(n, fmt.Sprintf("%040x", 1), "refs/heads/main")
854 l.set(tt.swap)
855 r.feed <- ev
856 if tt.rootSees {
857 root.next(t)
858 }
859
860 anon.quiet(t, r, r.repo("open", domain.VisibilityPublic))
861
862 late := r.watch(ctx, domain.Anonymous(), ports.WatchParams{ResumeAfter: "bogus"})
863 assert.True(t, late.next(t).Gap)
864 late.quiet(t, r, r.repo("open2", domain.VisibilityPublic))
865 })
866 })
867 }
868
869 t.Run("a push of an unknown repository is told to nobody", func(t *testing.T) {
870 t.Parallel()
871 synctest.Test(t, func(t *testing.T) {
872 ctx, cancel := context.WithCancel(context.Background())
873 defer cancel()
874 r := newEventRig(t, 0)
875 n := r.repo("ghost", domain.VisibilityPublic)
876 root := r.watch(ctx, domain.Root(), ports.WatchParams{})
877 synctest.Wait()
878 require.Equal(t, 1, r.events.Watchers())
879 r.feed <- domain.PushEvent{Repo: n, Visibility: domain.VisibilityPublic, At: now}
880 root.quiet(t, r, n)
881 })
882 })
883}