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 TestWatchDropsIssueEventsOnceIssuesAreOff(t *testing.T) {
210 t.Parallel()
211 synctest.Test(t, func(t *testing.T) {
212 r := newEventRig(t, 16)
213 n := name(t, "iss")
214 var mu sync.Mutex
215 mode := domain.IssuesOwner
216 r.meta.EXPECT().Get(mock.Anything, n).RunAndReturn(func(context.Context, domain.RepoName) (domain.Repo, error) {
217 mu.Lock()
218 defer mu.Unlock()
219 st := domain.DefaultRepoSettings()
220 st.Issues = mode
221 return domain.Repo{Name: n, Visibility: domain.VisibilityPublic, CreatedAt: now, Settings: st}, nil
222 })
223 ctx, cancel := context.WithCancel(context.Background())
224 defer cancel()
225 first := r.watch(ctx, domain.Root(), ports.WatchParams{})
226 synctest.Wait()
227
228 r.push(n, 1)
229 pushID := first.next(t).ID
230 r.feed <- domain.IssueCommented{Repo: n, Number: 1, Comment: domain.IssueComment{ID: 1}, At: now, Visibility: domain.VisibilityPublic, CreatedAt: now}
231 _, ok := first.next(t).Event.(domain.IssueCommented)
232 require.True(t, ok, "delivered while issues are on")
233 r.push(n, 2)
234 first.next(t)
235
236 mu.Lock()
237 mode = domain.IssuesOff
238 mu.Unlock()
239
240 // The replay holds the issue event: it is dropped, the other ones stay.
241 w := r.watch(ctx, domain.Root(), ports.WatchParams{ResumeAfter: pushID})
242 e := w.next(t)
243 _, isPush := e.Event.(domain.PushEvent)
244 require.True(t, isPush, "got %T", e.Event)
245 w.quiet(t, r, n)
246
247 r.feed <- domain.IssueOpened{Issue: domain.Issue{Repo: n, Number: 2}, At: now, Visibility: domain.VisibilityPublic, CreatedAt: now}
248 w.quiet(t, r, n)
249 })
250}
251
252func TestWatchResumeAndGap(t *testing.T) {
253 t.Parallel()
254 const size = 8
255 // Each subtest gets its own bubble and a fresh run of 20 events recorded
256 // through a first watcher; the ring then holds the last 8.
257 setup := func(t *testing.T) (r *eventRig, n domain.RepoName, ctx context.Context, ids []string) {
258 r = newEventRig(t, size)
259 n = r.repo("pub", domain.VisibilityPublic)
260 ctx, cancel := context.WithCancel(context.Background())
261 t.Cleanup(cancel)
262
263 first := r.watch(ctx, domain.Root(), ports.WatchParams{})
264 synctest.Wait()
265 require.Equal(t, 1, r.events.Watchers())
266 // One at a time: this watcher must not fall behind the small ring.
267 for i := range 20 {
268 r.push(n, i)
269 ids = append(ids, first.next(t).ID)
270 }
271 require.Len(t, ids, 20)
272 return r, n, ctx, ids
273 }
274
275 t.Run("an id that fell out of the buffer gives a gap, then the oldest buffered event", func(t *testing.T) {
276 t.Parallel()
277 synctest.Test(t, func(t *testing.T) {
278 r, _, ctx, ids := setup(t)
279 // 20 events were taken, the buffer holds the last 8: ids[12] is the
280 // oldest still present; ids[0] is long gone.
281 w := r.watch(ctx, domain.Root(), ports.WatchParams{ResumeAfter: ids[0]})
282 gap := w.next(t)
283 assert.True(t, gap.Gap)
284 assert.Empty(t, gap.ID)
285 oldest := w.next(t)
286 assert.False(t, oldest.Gap)
287 assert.Equal(t, ids[12], oldest.ID)
288 })
289 })
290
291 t.Run("an unknown id gives a gap", func(t *testing.T) {
292 t.Parallel()
293 synctest.Test(t, func(t *testing.T) {
294 r, _, ctx, _ := setup(t)
295 for _, id := range []string{"nonsense", "zz-1", "-", "0-0"} {
296 w := r.watch(ctx, domain.Root(), ports.WatchParams{ResumeAfter: id})
297 assert.True(t, w.next(t).Gap, id)
298 }
299 })
300 })
301
302 t.Run("an id of a later run is a gap, not a newer event", func(t *testing.T) {
303 t.Parallel()
304 synctest.Test(t, func(t *testing.T) {
305 r, _, ctx, _ := setup(t)
306 w := r.watch(ctx, domain.Root(), ports.WatchParams{ResumeAfter: "otherboot-999"})
307 assert.True(t, w.next(t).Gap)
308 })
309 })
310
311 t.Run("resume replays the buffered events after the id, then goes live", func(t *testing.T) {
312 t.Parallel()
313 synctest.Test(t, func(t *testing.T) {
314 r, n, ctx, ids := setup(t)
315 w := r.watch(ctx, domain.Root(), ports.WatchParams{ResumeAfter: ids[16]})
316 for _, want := range ids[17:] {
317 e := w.next(t)
318 assert.False(t, e.Gap)
319 assert.Equal(t, want, e.ID)
320 }
321 // Nothing else was replayed: the next event is the one pushed now.
322 r.push(n, 98)
323 live := w.next(t)
324 assert.False(t, live.Gap)
325 assert.NotContains(t, ids, live.ID)
326 })
327 })
328
329 t.Run("resume at the last event replays nothing, then goes live", func(t *testing.T) {
330 t.Parallel()
331 synctest.Test(t, func(t *testing.T) {
332 r, n, ctx, ids := setup(t)
333 w := r.watch(ctx, domain.Root(), ports.WatchParams{ResumeAfter: ids[len(ids)-1]})
334 r.push(n, 99)
335 e := w.next(t)
336 assert.False(t, e.Gap)
337 assert.NotContains(t, ids, e.ID, "the first event is the new one, nothing was replayed")
338 })
339 })
340}
341
342func TestWatchTopicErrors(t *testing.T) {
343 t.Parallel()
344 r := newEventRig(t, 0)
345 r.repo("pub", domain.VisibilityPublic)
346 r.repo("priv", domain.VisibilityPrivate)
347 r.meta.EXPECT().Get(mock.Anything, name(t, "gone")).Return(domain.Repo{}, domain.ErrNotFound).Maybe()
348 noop := func(ports.WatchEvent) error { return nil }
349 ctx := context.Background()
350
351 for topic, want := range map[string]error{
352 "nope": domain.ErrInvalid,
353 "repo:Bad": domain.ErrInvalidName,
354 "repo:": domain.ErrInvalidName,
355 "repo:gone": domain.ErrNotFound,
356 "repo:priv": domain.ErrNotFound, // hidden from anonymous
357 "repo:pub:foo": domain.ErrInvalidName,
358 } {
359 err := r.events.Watch(ctx, domain.Anonymous(), ports.WatchParams{Topics: []string{topic}}, noop)
360 assert.ErrorIs(t, err, want, topic)
361 }
362}
363
364func TestWatchEndsCleanly(t *testing.T) {
365 t.Parallel()
366 t.Run("on context cancellation", func(t *testing.T) {
367 t.Parallel()
368 synctest.Test(t, func(t *testing.T) {
369 r := newEventRig(t, 0)
370 ctx, cancel := context.WithCancel(context.Background())
371 w := r.watch(ctx, domain.Root(), ports.WatchParams{})
372 synctest.Wait()
373 require.Equal(t, 1, r.events.Watchers())
374 cancel()
375 synctest.Wait()
376 select {
377 case err := <-w.done:
378 assert.NoError(t, err)
379 default:
380 t.Fatal("watch did not end")
381 }
382 assert.Zero(t, r.events.Watchers())
383 })
384 })
385
386 t.Run("on Close", func(t *testing.T) {
387 t.Parallel()
388 synctest.Test(t, func(t *testing.T) {
389 r := newEventRig(t, 0)
390 w := r.watch(context.Background(), domain.Root(), ports.WatchParams{})
391 synctest.Wait()
392 require.Equal(t, 1, r.events.Watchers())
393 r.events.Close(context.Background())
394 r.events.Close(context.Background()) // twice is fine
395 synctest.Wait()
396 select {
397 case err := <-w.done:
398 assert.NoError(t, err)
399 default:
400 t.Fatal("watch did not end")
401 }
402 })
403 })
404
405 t.Run("a failing send ends the watch with its error", func(t *testing.T) {
406 t.Parallel()
407 synctest.Test(t, func(t *testing.T) {
408 r := newEventRig(t, 0)
409 n := r.repo("pub", domain.VisibilityPublic)
410 boom := errors.New("client went away")
411 done := make(chan error, 1)
412 go func() {
413 done <- r.events.Watch(context.Background(), domain.Root(), ports.WatchParams{}, func(ports.WatchEvent) error { return boom })
414 }()
415 synctest.Wait()
416 require.Equal(t, 1, r.events.Watchers())
417 r.push(n, 1)
418 assert.ErrorIs(t, <-done, boom)
419 })
420 })
421}
422
423func TestWatchSlowWatcherGetsAGap(t *testing.T) {
424 t.Parallel()
425 synctest.Test(t, func(t *testing.T) {
426 const size = 4
427 r := newEventRig(t, size)
428 n := r.repo("pub", domain.VisibilityPublic)
429
430 release := make(chan struct{})
431 var mu sync.Mutex
432 var got []ports.WatchEvent
433 done := make(chan error, 1)
434 blocked := make(chan struct{})
435 first := true
436 go func() {
437 done <- r.events.Watch(context.Background(), domain.Root(), ports.WatchParams{}, func(e ports.WatchEvent) error {
438 if first { // block on the first event while the ring turns over
439 first = false
440 close(blocked)
441 <-release
442 }
443 mu.Lock()
444 got = append(got, e)
445 mu.Unlock()
446 return nil
447 })
448 }()
449 // A second, fast watcher tells when the use case has taken every event:
450 // it receives them in order.
451 fast := r.watch(context.Background(), domain.Root(), ports.WatchParams{})
452 synctest.Wait()
453 require.Equal(t, 2, r.events.Watchers())
454 r.push(n, 0)
455 <-blocked
456 for i := 1; i < 30; i++ {
457 r.push(n, i)
458 }
459 // The ring holds four events: the fast watcher may get a gap, but it ends
460 // on the last push, which proves the use case took all of them.
461 last := fmt.Sprintf("%040x", 29+1)
462 for {
463 e := fast.next(t)
464 if p, ok := e.Event.(domain.PushEvent); ok && p.Updates[0].New == last {
465 break
466 }
467 }
468 close(release)
469
470 synctest.Wait()
471 mu.Lock()
472 defer mu.Unlock()
473 require.GreaterOrEqual(t, len(got), 2)
474 hasGap := false
475 for _, e := range got[1:] {
476 hasGap = hasGap || e.Gap
477 }
478 assert.True(t, hasGap, "events were lost while the watcher was blocked: %+v", got)
479 })
480}
481
482func TestEventsRepoPayloads(t *testing.T) {
483 t.Parallel()
484 synctest.Test(t, func(t *testing.T) {
485 r := newEventRig(t, 0)
486 pub := r.repo("pub", domain.VisibilityPublic)
487 priv := r.repo("priv", domain.VisibilityPrivate)
488 ctx, cancel := context.WithCancel(context.Background())
489 defer cancel()
490 anon := r.watch(ctx, domain.Anonymous(), ports.WatchParams{})
491 root := r.watch(ctx, domain.Root(), ports.WatchParams{})
492 synctest.Wait()
493 require.Equal(t, 2, r.events.Watchers())
494
495 r.feed <- domain.RepoCreated{Repo: pub, Visibility: domain.VisibilityPublic, At: now}
496 r.feed <- domain.RepoUpdated{Repo: pub, At: now, Visibility: domain.VisibilityPublic, Previous: domain.VisibilityPublic, CreatedAt: now}
497 // A deleted repository cannot be read any more: its visibility decides.
498 r.feed <- domain.RepoDeleted{Repo: priv, Visibility: domain.VisibilityPrivate, At: now}
499 r.feed <- domain.RepoDeleted{Repo: pub, Visibility: domain.VisibilityPublic, At: now}
500
501 for _, w := range []*watcher{anon, root} {
502 created := w.next(t)
503 require.NotNil(t, created.Repo)
504 assert.Equal(t, "main", created.Repo.DefaultBranch)
505 assert.Equal(t, pub, created.Repo.Name)
506 require.NotNil(t, w.next(t).Repo)
507 }
508 assert.Equal(t, priv, root.next(t).Event.(domain.RepoDeleted).Repo)
509 assert.Equal(t, pub, root.next(t).Event.(domain.RepoDeleted).Repo)
510 assert.Equal(t, pub, anon.next(t).Event.(domain.RepoDeleted).Repo, "the private deletion was skipped")
511 anon.quiet(t, r, pub)
512 })
513}
514
515// tokenWith builds the principal of a stored token with an id and an expiry.
516func tokenWith(id string, expires time.Time, repos ...string) domain.Principal {
517 tok := domain.Token{ID: id, Scopes: []domain.Scope{domain.ScopeRepoRead}, ExpiresAt: expires}
518 for _, r := range repos {
519 n, err := domain.ParseRepoName(r)
520 if err != nil {
521 panic(err)
522 }
523 tok.Repos = append(tok.Repos, n)
524 }
525 return domain.TokenPrincipal(tok)
526}
527
528func TestWatchEndsWhenTheTokenIsRevoked(t *testing.T) {
529 t.Parallel()
530 synctest.Test(t, func(t *testing.T) {
531 // Asking at every turn keeps the test independent of the clock.
532 r := newEventRigConfig(t, usecase.EventsConfig{RevalidateEvery: -1})
533 pub := r.repo("pub", domain.VisibilityPublic)
534 priv := r.repo("priv", domain.VisibilityPrivate)
535 p := tokenWith("tok", time.Time{})
536 ctx, cancel := context.WithCancel(context.Background())
537 defer cancel()
538
539 w := r.watch(ctx, p, ports.WatchParams{})
540 synctest.Wait()
541 require.Equal(t, 1, r.events.Watchers())
542 r.push(pub, 1)
543 w.next(t)
544
545 r.revoked.Store("tok", true)
546 r.push(priv, 2) // would be delivered to a token that may read priv
547 synctest.Wait()
548 select {
549 case err := <-w.done:
550 assert.ErrorIs(t, err, domain.ErrUnauthenticated)
551 default:
552 t.Fatal("the watch of a revoked token did not end")
553 }
554 select {
555 case e := <-w.got:
556 t.Fatalf("an event was sent after the revocation: %+v", e)
557 default:
558 }
559 })
560}
561
562func TestWatchEndsWhenTheTokenExpires(t *testing.T) {
563 t.Parallel()
564 synctest.Test(t, func(t *testing.T) {
565 r := newEventRig(t, 0)
566 pub := r.repo("pub", domain.VisibilityPublic)
567 // The rig's clock stands at now: this token expired at now.
568 p := tokenWith("tok", now)
569 w := r.watch(context.Background(), p, ports.WatchParams{})
570 r.push(pub, 1) // whether or not the watch had started, it ends unauthenticated
571 synctest.Wait()
572 select {
573 case err := <-w.done:
574 assert.ErrorIs(t, err, domain.ErrUnauthenticated)
575 default:
576 t.Fatal("the watch of an expired token did not end")
577 }
578 assert.Empty(t, w.got)
579 })
580}
581
582func TestWatchRevalidatesIdleWatchersAndCachesTheAnswer(t *testing.T) {
583 t.Parallel()
584 t.Run("a quiet forge still notices a revocation", func(t *testing.T) {
585 t.Parallel()
586 synctest.Test(t, func(t *testing.T) {
587 r := newEventRigConfig(t, usecase.EventsConfig{}) // the production default interval
588 w := r.watch(context.Background(), tokenWith("tok", time.Time{}), ports.WatchParams{})
589 synctest.Wait()
590 require.Equal(t, 1, r.events.Watchers())
591 r.revoked.Store("tok", true)
592 synctest.Sleep(usecase.DefaultRevalidateEvery - time.Nanosecond)
593 select {
594 case err := <-w.done:
595 t.Fatalf("the watch ended before the interval: %v", err)
596 default:
597 }
598 assert.Zero(t, r.revalidations.Load(), "no store lookup before the interval")
599 synctest.Sleep(time.Nanosecond)
600 select {
601 case err := <-w.done:
602 assert.ErrorIs(t, err, domain.ErrUnauthenticated)
603 default:
604 t.Fatal("an idle watch of a revoked token did not end at the interval")
605 }
606 assert.Equal(t, int32(1), r.revalidations.Load())
607 })
608 })
609
610 t.Run("the answer is kept between checks", func(t *testing.T) {
611 t.Parallel()
612 synctest.Test(t, func(t *testing.T) {
613 r := newEventRigConfig(t, usecase.EventsConfig{RevalidateEvery: time.Hour})
614 pub := r.repo("pub", domain.VisibilityPublic)
615 w := r.watch(context.Background(), tokenWith("tok", time.Time{}), ports.WatchParams{})
616 synctest.Wait()
617 require.Equal(t, 1, r.events.Watchers())
618 for i := range 20 {
619 r.push(pub, i)
620 }
621 for range 20 {
622 w.next(t)
623 }
624 // Idle, the watcher wakes up every few seconds but keeps its answer
625 // until the interval is over.
626 synctest.Sleep(time.Hour - time.Nanosecond)
627 assert.Zero(t, r.revalidations.Load(), "no store lookup within the interval")
628 synctest.Sleep(time.Nanosecond)
629 assert.Equal(t, int32(1), r.revalidations.Load(), "one lookup when the interval is over")
630 })
631 })
632
633 t.Run("root and anonymous never ask", func(t *testing.T) {
634 t.Parallel()
635 synctest.Test(t, func(t *testing.T) {
636 r := newEventRigConfig(t, usecase.EventsConfig{RevalidateEvery: -1})
637 pub := r.repo("pub", domain.VisibilityPublic)
638 a := r.watch(context.Background(), domain.Anonymous(), ports.WatchParams{})
639 b := r.watch(context.Background(), domain.Root(), ports.WatchParams{})
640 synctest.Wait()
641 require.Equal(t, 2, r.events.Watchers())
642 r.push(pub, 1)
643 a.next(t)
644 b.next(t)
645 r.auth.AssertNotCalled(t, "Revalidate", mock.Anything, mock.Anything)
646 })
647 })
648}
649
650func TestWatchLimits(t *testing.T) {
651 t.Parallel()
652 noop := func(ports.WatchEvent) error { return nil }
653
654 t.Run("too many topics", func(t *testing.T) {
655 t.Parallel()
656 synctest.Test(t, func(t *testing.T) {
657 r := newEventRig(t, 0)
658 r.repo("pub", domain.VisibilityPublic)
659 ok := make([]string, usecase.MaxWatchTopics)
660 for i := range ok {
661 ok[i] = "home"
662 }
663 ctx, cancel := context.WithCancel(context.Background())
664 cancel() // returns at once once the topics are accepted
665 assert.NoError(t, r.events.Watch(ctx, domain.Root(), ports.WatchParams{Topics: ok}, noop))
666 err := r.events.Watch(context.Background(), domain.Root(), ports.WatchParams{Topics: append(ok, "home")}, noop)
667 assert.ErrorIs(t, err, domain.ErrInvalid)
668 })
669 })
670
671 t.Run("total watchers", func(t *testing.T) {
672 t.Parallel()
673 synctest.Test(t, func(t *testing.T) {
674 r := newEventRig(t, 0)
675 ctx, cancel := context.WithCancel(context.Background())
676 defer cancel()
677 for range usecase.MaxWatchers {
678 r.watch(ctx, domain.Root(), ports.WatchParams{})
679 }
680 synctest.Wait()
681 require.Equal(t, usecase.MaxWatchers, r.events.Watchers())
682 err := r.events.Watch(ctx, domain.Root(), ports.WatchParams{}, noop)
683 assert.ErrorIs(t, err, domain.ErrResourceExhausted)
684 assert.Equal(t, usecase.MaxWatchers, r.events.Watchers(), "a refused watch is not counted")
685
686 cancel() // all end; a new one is accepted again
687 synctest.Wait()
688 require.Equal(t, 0, r.events.Watchers())
689 again, stop := context.WithCancel(context.Background())
690 stop()
691 assert.NoError(t, r.events.Watch(again, domain.Root(), ports.WatchParams{}, noop))
692 })
693 })
694
695 t.Run("watchers per client", func(t *testing.T) {
696 t.Parallel()
697 synctest.Test(t, func(t *testing.T) {
698 r := newEventRig(t, 0)
699 ctx, cancel := context.WithCancel(context.Background())
700 defer cancel()
701 for range usecase.MaxWatchersPerClient {
702 r.watch(ctx, domain.Root(), ports.WatchParams{Client: "198.51.100.1"})
703 }
704 synctest.Wait()
705 require.Equal(t, usecase.MaxWatchersPerClient, r.events.Watchers())
706
707 err := r.events.Watch(ctx, domain.Root(), ports.WatchParams{Client: "198.51.100.1"}, noop)
708 assert.ErrorIs(t, err, domain.ErrResourceExhausted)
709
710 // Another client, and callers in process, are not affected.
711 r.watch(ctx, domain.Root(), ports.WatchParams{Client: "198.51.100.2"})
712 r.watch(ctx, domain.Root(), ports.WatchParams{})
713 synctest.Wait()
714 require.Equal(t, usecase.MaxWatchersPerClient+2, r.events.Watchers())
715
716 cancel()
717 synctest.Wait()
718 require.Equal(t, 0, r.events.Watchers())
719 again, stop := context.WithCancel(context.Background())
720 stop()
721 assert.NoError(t, r.events.Watch(again, domain.Root(), ports.WatchParams{Client: "198.51.100.1"}, noop), "the slots were released")
722 })
723 })
724}
725
726// repoAt declares a repository whose metadata the test changes.
727type liveRepo struct {
728 mu sync.Mutex
729 repo domain.Repo
730}
731
732func (l *liveRepo) set(f func(*domain.Repo)) {
733 l.mu.Lock()
734 defer l.mu.Unlock()
735 f(&l.repo)
736}
737
738func (r *eventRig) liveRepo(s string, vis domain.Visibility, created time.Time) (*liveRepo, domain.RepoName) {
739 n := name(r.t, s)
740 l := &liveRepo{repo: domain.Repo{Name: n, Visibility: vis, CreatedAt: created}}
741 r.meta.EXPECT().Get(mock.Anything, n).RunAndReturn(func(context.Context, domain.RepoName) (domain.Repo, error) {
742 l.mu.Lock()
743 defer l.mu.Unlock()
744 return l.repo, nil
745 }).Maybe()
746 return l, n
747}
748
749func TestEventsKeepTheVisibilityOfTheirTime(t *testing.T) {
750 t.Parallel()
751 t.Run("a repository made public does not reveal what happened while it was private", func(t *testing.T) {
752 t.Parallel()
753 synctest.Test(t, func(t *testing.T) {
754 ctx, cancel := context.WithCancel(context.Background())
755 defer cancel()
756 r := newEventRig(t, 0)
757 l, n := r.liveRepo("secret", domain.VisibilityPrivate, now)
758 root := r.watch(ctx, domain.Root(), ports.WatchParams{})
759 synctest.Wait()
760 require.Equal(t, 1, r.events.Watchers())
761 r.push(n, 1)
762 r.push(n, 2)
763 root.next(t)
764 root.next(t) // both are buffered now
765
766 l.set(func(repo *domain.Repo) { repo.Visibility = domain.VisibilityPublic })
767 r.feed <- domain.RepoUpdated{Repo: n, At: now, Visibility: domain.VisibilityPublic, Previous: domain.VisibilityPrivate, CreatedAt: now}
768 root.next(t)
769 r.push(n, 3) // public era
770
771 // An anonymous caller that resumes with an id nobody issued gets what
772 // is buffered, filtered: the public update and the public push, not the
773 // pushes of the private era.
774 anon := r.watch(ctx, domain.Anonymous(), ports.WatchParams{ResumeAfter: "bogus"})
775 assert.True(t, anon.next(t).Gap)
776 upd := anon.next(t)
777 require.IsType(t, domain.RepoUpdated{}, upd.Event)
778 push := anon.next(t)
779 require.IsType(t, domain.PushEvent{}, push.Event)
780 assert.Equal(t, fmt.Sprintf("%040x", 3+1), push.Event.(domain.PushEvent).Updates[0].New)
781 anon.quiet(t, r, n)
782 })
783 })
784
785 t.Run("a repository deleted and created again does not inherit old events", func(t *testing.T) {
786 t.Parallel()
787 synctest.Test(t, func(t *testing.T) {
788 ctx, cancel := context.WithCancel(context.Background())
789 defer cancel()
790 r := newEventRig(t, 0)
791 first := now.Add(-time.Hour)
792 l, n := r.liveRepo("again", domain.VisibilityPublic, first)
793 root := r.watch(ctx, domain.Root(), ports.WatchParams{})
794 synctest.Wait()
795 require.Equal(t, 1, r.events.Watchers())
796 r.push(n, 1)
797 root.next(t)
798
799 l.set(func(repo *domain.Repo) { repo.CreatedAt = now }) // deleted, created again
800 r.push(n, 2)
801 root.next(t)
802
803 anon := r.watch(ctx, domain.Anonymous(), ports.WatchParams{ResumeAfter: "bogus"})
804 assert.True(t, anon.next(t).Gap)
805 e := anon.next(t)
806 require.IsType(t, domain.PushEvent{}, e.Event)
807 assert.Equal(t, fmt.Sprintf("%040x", 2+1), e.Event.(domain.PushEvent).Updates[0].New, "only the push to the new repository")
808 anon.quiet(t, r, n)
809 })
810 })
811
812 t.Run("a repository made private tells those who could read it, once", func(t *testing.T) {
813 t.Parallel()
814 synctest.Test(t, func(t *testing.T) {
815 ctx, cancel := context.WithCancel(context.Background())
816 defer cancel()
817 r := newEventRig(t, 0)
818 l, n := r.liveRepo("hide", domain.VisibilityPublic, now)
819 anon := r.watch(ctx, domain.Anonymous(), ports.WatchParams{})
820 reader := r.watch(ctx, tokenPrincipal(domain.ScopeRepoRead, "hide"), ports.WatchParams{})
821 other := r.watch(ctx, tokenPrincipal(domain.ScopeRepoRead, "elsewhere"), ports.WatchParams{})
822 root := r.watch(ctx, domain.Root(), ports.WatchParams{})
823 synctest.Wait()
824 require.Equal(t, 4, r.events.Watchers())
825
826 l.set(func(repo *domain.Repo) { repo.Visibility = domain.VisibilityPrivate })
827 r.feed <- domain.RepoUpdated{Repo: n, At: now, Visibility: domain.VisibilityPrivate, Previous: domain.VisibilityPublic, CreatedAt: now}
828
829 for name, w := range map[string]*watcher{"anonymous": anon, "reader": reader, "other": other, "root": root} {
830 e := w.next(t)
831 up, ok := e.Event.(domain.RepoUpdated)
832 require.True(t, ok, name)
833 assert.Equal(t, n, up.Repo, name)
834 require.NotNil(t, e.Repo, name)
835 assert.Equal(t, domain.VisibilityPrivate, e.Repo.Visibility, name)
836 assert.Empty(t, e.Repo.Description, "nothing but the name and the visibility", name)
837 }
838
839 // Later events of the repository reach those who may read it now, only.
840 r.push(n, 1)
841 assert.IsType(t, domain.PushEvent{}, reader.next(t).Event)
842 assert.IsType(t, domain.PushEvent{}, root.next(t).Event)
843 anon.quiet(t, r, r.repo("open", domain.VisibilityPublic))
844 other.quiet(t, r, r.repo("open2", domain.VisibilityPublic))
845 })
846 })
847
848 t.Run("a private repository that never was public stays hidden", func(t *testing.T) {
849 t.Parallel()
850 synctest.Test(t, func(t *testing.T) {
851 ctx, cancel := context.WithCancel(context.Background())
852 defer cancel()
853 r := newEventRig(t, 0)
854 _, n := r.liveRepo("closed", domain.VisibilityPrivate, now)
855 anon := r.watch(ctx, domain.Anonymous(), ports.WatchParams{})
856 synctest.Wait()
857 require.Equal(t, 1, r.events.Watchers())
858 r.feed <- domain.RepoUpdated{Repo: n, At: now, Visibility: domain.VisibilityPrivate, Previous: domain.VisibilityPrivate, CreatedAt: now}
859 r.push(n, 1)
860 anon.quiet(t, r, r.repo("open", domain.VisibilityPublic))
861 })
862 })
863}
864
865// TestPushEventsKeepTheirSnapshot: a push is told by the repository as it was
866// when published, not as the store says when the feeder gets to it.
867func TestPushEventsKeepTheirSnapshot(t *testing.T) {
868 t.Parallel()
869 tests := []struct {
870 name string
871 swap func(repo *domain.Repo)
872 rootSees bool // the repository is still the one the push is about
873 }{
874 {"made public before the feeder appends", func(repo *domain.Repo) { repo.Visibility = domain.VisibilityPublic }, true},
875 {"recreated before the feeder appends", func(repo *domain.Repo) {
876 repo.Visibility = domain.VisibilityPublic
877 repo.CreatedAt = now.Add(time.Hour)
878 }, false},
879 }
880 for _, tt := range tests {
881 t.Run(tt.name, func(t *testing.T) {
882 t.Parallel()
883 synctest.Test(t, func(t *testing.T) {
884 ctx, cancel := context.WithCancel(context.Background())
885 defer cancel()
886 r := newEventRig(t, 0)
887 l, n := r.liveRepo("secret", domain.VisibilityPrivate, now)
888 anon := r.watch(ctx, domain.Anonymous(), ports.WatchParams{})
889 root := r.watch(ctx, domain.Root(), ports.WatchParams{})
890 synctest.Wait()
891 require.Equal(t, 2, r.events.Watchers())
892
893 // Published while private; the store changes before the feeder
894 // can have looked at it: the event is built, then the state is
895 // swapped, and only then is it handed to the feeder.
896 ev := r.pushEvent(n, fmt.Sprintf("%040x", 1), "refs/heads/main")
897 l.set(tt.swap)
898 r.feed <- ev
899 if tt.rootSees {
900 root.next(t)
901 }
902
903 anon.quiet(t, r, r.repo("open", domain.VisibilityPublic))
904
905 late := r.watch(ctx, domain.Anonymous(), ports.WatchParams{ResumeAfter: "bogus"})
906 assert.True(t, late.next(t).Gap)
907 late.quiet(t, r, r.repo("open2", domain.VisibilityPublic))
908 })
909 })
910 }
911
912 t.Run("a push of an unknown repository is told to nobody", func(t *testing.T) {
913 t.Parallel()
914 synctest.Test(t, func(t *testing.T) {
915 ctx, cancel := context.WithCancel(context.Background())
916 defer cancel()
917 r := newEventRig(t, 0)
918 n := r.repo("ghost", domain.VisibilityPublic)
919 root := r.watch(ctx, domain.Root(), ports.WatchParams{})
920 synctest.Wait()
921 require.Equal(t, 1, r.events.Watchers())
922 r.feed <- domain.PushEvent{Repo: n, Visibility: domain.VisibilityPublic, At: now}
923 root.quiet(t, r, n)
924 })
925 })
926}
927
928// Home shows the CI state of each repository: a run update reaches it, for the
929// watchers who may read the repository, and so does a job that ended; the
930// progress of a job stays on the repository topic.
931func TestWatchHomeGetsRunUpdates(t *testing.T) {
932 t.Parallel()
933 synctest.Test(t, func(t *testing.T) {
934 r := newEventRig(t, 0)
935 pub := r.repo("pub", domain.VisibilityPublic)
936 priv := r.repo("priv", domain.VisibilityPrivate)
937 ctx, cancel := context.WithCancel(context.Background())
938 defer cancel()
939
940 anon := r.watch(ctx, domain.Anonymous(), ports.WatchParams{Topics: []string{"home"}})
941 root := r.watch(ctx, domain.Root(), ports.WatchParams{Topics: []string{"home"}})
942 synctest.Wait()
943 require.Equal(t, 2, r.events.Watchers())
944
945 run := func(n domain.RepoName, vis domain.Visibility) domain.RunUpdated {
946 return domain.RunUpdated{Run: domain.CIRun{Repo: n, ID: 1}, At: now, Visibility: vis, CreatedAt: now}
947 }
948 job := func(n domain.RepoName, vis domain.Visibility, phase domain.CIJobPhase) domain.JobUpdated {
949 return domain.JobUpdated{Job: domain.CIJob{Repo: n, Phase: phase}, At: now, Visibility: vis, CreatedAt: now}
950 }
951 r.feed <- job(pub, domain.VisibilityPublic, domain.CIPhaseRunning)
952 r.feed <- run(priv, domain.VisibilityPrivate)
953 r.feed <- job(priv, domain.VisibilityPrivate, domain.CIPhaseEnded)
954 r.feed <- run(pub, domain.VisibilityPublic)
955 r.feed <- job(pub, domain.VisibilityPublic, domain.CIPhaseEnded)
956
957 assert.Equal(t, pub, anon.next(t).Event.(domain.RunUpdated).Run.Repo, "anonymous: no private run, no running job")
958 assert.Equal(t, pub, anon.next(t).Event.(domain.JobUpdated).Job.Repo)
959 anon.quiet(t, r, pub)
960
961 assert.Equal(t, priv, root.next(t).Event.(domain.RunUpdated).Run.Repo)
962 assert.Equal(t, priv, root.next(t).Event.(domain.JobUpdated).Job.Repo)
963 assert.Equal(t, pub, root.next(t).Event.(domain.RunUpdated).Run.Repo)
964 assert.Equal(t, pub, root.next(t).Event.(domain.JobUpdated).Job.Repo)
965 root.quiet(t, r, pub)
966 })
967}
968
969func TestNoteMutationEventVisibility(t *testing.T) {
970 synctest.Test(t, func(t *testing.T) {
971 r := newEventRig(t, 0)
972 pub := r.repo("pub", domain.VisibilityPublic)
973 priv := r.repo("priv", domain.VisibilityPrivate)
974 ctx, cancel := context.WithCancel(context.Background())
975 defer cancel()
976 anon := r.watch(ctx, domain.Anonymous(), ports.WatchParams{})
977 owner := r.watch(ctx, domain.Root(), ports.WatchParams{Topics: []string{"repo:priv"}})
978 synctest.Wait()
979 for _, kind := range []string{"edited", "deleted", "resolved"} {
980 for _, n := range []domain.RepoName{priv, pub} {
981 vis := domain.VisibilityPublic
982 if n == priv {
983 vis = domain.VisibilityPrivate
984 }
985 switch kind {
986 case "edited":
987 r.feed <- domain.CommentEdited{Repo: n, Visibility: vis, CreatedAt: now, At: now}
988 case "deleted":
989 r.feed <- domain.CommentDeleted{Repo: n, Visibility: vis, CreatedAt: now, At: now}
990 case "resolved":
991 r.feed <- domain.CommentResolved{Repo: n, Visibility: vis, CreatedAt: now, At: now}
992 }
993 }
994 assert.Equal(t, "mr.comment_"+kind, anon.next(t).Event.EventKind())
995 assert.Equal(t, "mr.comment_"+kind, owner.next(t).Event.EventKind())
996 anon.quiet(t, r, pub)
997 }
998 })
999}
1000
1001func TestMergeabilityEventReachesOwnerTopic(t *testing.T) {
1002 synctest.Test(t, func(t *testing.T) {
1003 r := newEventRig(t, 0)
1004 priv := r.repo("priv", domain.VisibilityPrivate)
1005 ctx, cancel := context.WithCancel(context.Background())
1006 defer cancel()
1007 owner := r.watch(ctx, domain.Root(), ports.WatchParams{Topics: []string{"repo:priv"}})
1008 synctest.Wait()
1009 r.feed <- domain.MergeRequestMergeability{Repo: priv, Number: 1, Visibility: domain.VisibilityPrivate, CreatedAt: now, At: now}
1010 assert.Equal(t, priv, owner.next(t).Event.(domain.MergeRequestMergeability).Repo)
1011 })
1012}