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}
884
885// Home shows the CI state of each repository: a run update reaches it, for the
886// watchers who may read the repository, and so does a job that ended; the
887// progress of a job stays on the repository topic.
888func TestWatchHomeGetsRunUpdates(t *testing.T) {
889 t.Parallel()
890 synctest.Test(t, func(t *testing.T) {
891 r := newEventRig(t, 0)
892 pub := r.repo("pub", domain.VisibilityPublic)
893 priv := r.repo("priv", domain.VisibilityPrivate)
894 ctx, cancel := context.WithCancel(context.Background())
895 defer cancel()
896
897 anon := r.watch(ctx, domain.Anonymous(), ports.WatchParams{Topics: []string{"home"}})
898 root := r.watch(ctx, domain.Root(), ports.WatchParams{Topics: []string{"home"}})
899 synctest.Wait()
900 require.Equal(t, 2, r.events.Watchers())
901
902 run := func(n domain.RepoName, vis domain.Visibility) domain.RunUpdated {
903 return domain.RunUpdated{Run: domain.CIRun{Repo: n, ID: 1}, At: now, Visibility: vis, CreatedAt: now}
904 }
905 job := func(n domain.RepoName, vis domain.Visibility, phase domain.CIJobPhase) domain.JobUpdated {
906 return domain.JobUpdated{Job: domain.CIJob{Repo: n, Phase: phase}, At: now, Visibility: vis, CreatedAt: now}
907 }
908 r.feed <- job(pub, domain.VisibilityPublic, domain.CIPhaseRunning)
909 r.feed <- run(priv, domain.VisibilityPrivate)
910 r.feed <- job(priv, domain.VisibilityPrivate, domain.CIPhaseEnded)
911 r.feed <- run(pub, domain.VisibilityPublic)
912 r.feed <- job(pub, domain.VisibilityPublic, domain.CIPhaseEnded)
913
914 assert.Equal(t, pub, anon.next(t).Event.(domain.RunUpdated).Run.Repo, "anonymous: no private run, no running job")
915 assert.Equal(t, pub, anon.next(t).Event.(domain.JobUpdated).Job.Repo)
916 anon.quiet(t, r, pub)
917
918 assert.Equal(t, priv, root.next(t).Event.(domain.RunUpdated).Run.Repo)
919 assert.Equal(t, priv, root.next(t).Event.(domain.JobUpdated).Job.Repo)
920 assert.Equal(t, pub, root.next(t).Event.(domain.RunUpdated).Run.Repo)
921 assert.Equal(t, pub, root.next(t).Event.(domain.JobUpdated).Job.Repo)
922 root.quiet(t, r, pub)
923 })
924}