fix(events): stop streaming issue events and views once issues are off or the repo is not visible
- commit
- 02692c451d45024000071a39129dbeeaa57a3bc8
- author
- brice · · 2026-10-19 06:08 UTC
- parent
- 4f0da00
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
4 files · +117 -0
M internal/adapters/web/events.go +25 -0
| @@ -116,6 +116,11 @@ func (s *Server) events(w http.ResponseWriter, r *http.Request) { | |||
| 116 | 116 | // A repository the visitor cannot see is a missing one (AC2). | |
| 117 | 117 | cctx, ccancel := context.WithTimeout(ctx, requestTimeout) | |
| 118 | 118 | err := s.canSee(cctx, repo) | |
| 119 | + | if err == nil { | |
| 120 | + | if v := q.Get("view"); v == "issues" || v == "issue" { | |
| 121 | + | err = s.issuesOn(cctx, repo) | |
| 122 | + | } | |
| 123 | + | } | |
| 119 | 124 | ccancel() | |
| 120 | 125 | if err != nil { | |
| 121 | 126 | s.denyEvents(ctx, w, err) | |
| @@ -300,6 +305,26 @@ func (s *Server) canSee(ctx context.Context, rawName string) error { | |||
| 300 | 305 | return err | |
| 301 | 306 | } | |
| 302 | 307 | ||
| 308 | + | // issuesOn is domain.ErrNotFound unless the repository has issues on and | |
| 309 | + | // s.cfg.Issues is set: the same answer as for a repository the visitor cannot see. | |
| 310 | + | func (s *Server) issuesOn(ctx context.Context, rawName string) error { | |
| 311 | + | if s.cfg.Issues == nil { | |
| 312 | + | return domain.ErrNotFound | |
| 313 | + | } | |
| 314 | + | name, err := domain.ParseRepoName(rawName) | |
| 315 | + | if err != nil { | |
| 316 | + | return err | |
| 317 | + | } | |
| 318 | + | v, err := s.cfg.Repos.Get(ctx, principal(ctx), name) | |
| 319 | + | if err != nil { | |
| 320 | + | return err | |
| 321 | + | } | |
| 322 | + | if v.Settings.Issues.OrDefault() == domain.IssuesOff { | |
| 323 | + | return domain.ErrNotFound | |
| 324 | + | } | |
| 325 | + | return nil | |
| 326 | + | } | |
| 327 | + | ||
| 303 | 328 | // denyStream answers a request that gets no stream. | |
| 304 | 329 | func denyStream(w http.ResponseWriter, status int, msg string) { | |
| 305 | 330 | w.Header().Set("Cache-Control", "no-store") | |
| @@ -116,6 +116,11 @@ func (s *Server) events(w http.ResponseWriter, r *http.Request) { | |||||
| 116 | // A repository the visitor cannot see is a missing one (AC2). | 116 | // A repository the visitor cannot see is a missing one (AC2). | ||
| 117 | cctx, ccancel := context.WithTimeout(ctx, requestTimeout) | 117 | cctx, ccancel := context.WithTimeout(ctx, requestTimeout) | ||
| 118 | err := s.canSee(cctx, repo) | 118 | err := s.canSee(cctx, repo) | ||
| 119 | + | if err == nil { | |||
| 120 | + | if v := q.Get("view"); v == "issues" || v == "issue" { | |||
| 121 | + | err = s.issuesOn(cctx, repo) | |||
| 122 | + | } | |||
| 123 | + | } | |||
| 119 | ccancel() | 124 | ccancel() | ||
| 120 | if err != nil { | 125 | if err != nil { | ||
| 121 | s.denyEvents(ctx, w, err) | 126 | s.denyEvents(ctx, w, err) | ||
| @@ -300,6 +305,26 @@ func (s *Server) canSee(ctx context.Context, rawName string) error { | |||||
| 300 | return err | 305 | return err | ||
| 301 | } | 306 | } | ||
| 302 | 307 | ||||
| 308 | + | // issuesOn is domain.ErrNotFound unless the repository has issues on and | |||
| 309 | + | // s.cfg.Issues is set: the same answer as for a repository the visitor cannot see. | |||
| 310 | + | func (s *Server) issuesOn(ctx context.Context, rawName string) error { | |||
| 311 | + | if s.cfg.Issues == nil { | |||
| 312 | + | return domain.ErrNotFound | |||
| 313 | + | } | |||
| 314 | + | name, err := domain.ParseRepoName(rawName) | |||
| 315 | + | if err != nil { | |||
| 316 | + | return err | |||
| 317 | + | } | |||
| 318 | + | v, err := s.cfg.Repos.Get(ctx, principal(ctx), name) | |||
| 319 | + | if err != nil { | |||
| 320 | + | return err | |||
| 321 | + | } | |||
| 322 | + | if v.Settings.Issues.OrDefault() == domain.IssuesOff { | |||
| 323 | + | return domain.ErrNotFound | |||
| 324 | + | } | |||
| 325 | + | return nil | |||
| 326 | + | } | |||
| 327 | + | ||||
| 303 | // denyStream answers a request that gets no stream. | 328 | // denyStream answers a request that gets no stream. | ||
| 304 | func denyStream(w http.ResponseWriter, status int, msg string) { | 329 | func denyStream(w http.ResponseWriter, status int, msg string) { | ||
| 305 | w.Header().Set("Cache-Control", "no-store") | 330 | w.Header().Set("Cache-Control", "no-store") | ||
M internal/adapters/web/events_test.go +37 -0
| @@ -428,3 +428,40 @@ func TestEventsInvalidCookieStream(t *testing.T) { | |||
| 428 | 428 | assert.Contains(t, resp.Header.Values("Vary"), "Cookie") | |
| 429 | 429 | assert.Contains(t, resp.Header.Get("Set-Cookie"), "Max-Age=0") | |
| 430 | 430 | } | |
| 431 | + | ||
| 432 | + | // An issues stream on a repository with issues off, unknown or private is | |
| 433 | + | // refused the same way: 404 with the same body. | |
| 434 | + | func TestEventsIssuesViewsAreDeniedLikeAMissingRepo(t *testing.T) { | |
| 435 | + | t.Parallel() | |
| 436 | + | priv := mocks.NewRepoService(t) | |
| 437 | + | priv.EXPECT().Get(mock.Anything, mock.Anything, mock.Anything).Return(domain.Repo{}, domain.ErrNotFound) | |
| 438 | + | for _, view := range []string{"issues", "issue&n=1"} { | |
| 439 | + | answers := map[string]string{} | |
| 440 | + | for name, repos := range map[string]ports.RepoService{ | |
| 441 | + | "off": reposServing(t, issueRepoView(t, domain.IssuesOff)), | |
| 442 | + | "unset": reposServing(t, issueRepoView(t, "")), | |
| 443 | + | "unknown": reposServing(t), | |
| 444 | + | "private": priv, | |
| 445 | + | } { | |
| 446 | + | f := &watchProbe{release: make(chan struct{})} | |
| 447 | + | srv := newTestServer(t, Config{Repos: repos, Browse: mocks.NewBrowseService(t), Events: f.service(t), Issues: mocks.NewIssueService(t), Heartbeat: time.Hour}) | |
| 448 | + | ts := httptest.NewServer(srv) | |
| 449 | + | t.Cleanup(ts.Close) | |
| 450 | + | resp, _ := open(t, ts, "topic=repo:proj&view="+view, nil) | |
| 451 | + | body, _ := io.ReadAll(resp.Body) | |
| 452 | + | assert.Equal(t, http.StatusNotFound, resp.StatusCode, "%s %s", view, name) | |
| 453 | + | answers[name] = resp.Header.Get("Cache-Control") + "|" + string(body) | |
| 454 | + | assert.Zero(t, f.opened.Load(), "%s %s: no watch is opened", view, name) | |
| 455 | + | } | |
| 456 | + | for name, a := range answers { | |
| 457 | + | assert.Equal(t, answers["unknown"], a, "%s %s", view, name) | |
| 458 | + | } | |
| 459 | + | } | |
| 460 | + | // With issues on, the stream opens. | |
| 461 | + | f := &watchProbe{release: make(chan struct{})} | |
| 462 | + | srv := newTestServer(t, Config{Repos: reposServing(t, issueRepoView(t, domain.IssuesOwner)), Browse: mocks.NewBrowseService(t), Events: f.service(t), Issues: mocks.NewIssueService(t), Heartbeat: time.Hour}) | |
| 463 | + | ts := httptest.NewServer(srv) | |
| 464 | + | t.Cleanup(ts.Close) | |
| 465 | + | resp, _ := open(t, ts, "topic=repo:proj&view=issues", nil) | |
| 466 | + | assert.Equal(t, http.StatusOK, resp.StatusCode) | |
| 467 | + | } | |
| @@ -428,3 +428,40 @@ func TestEventsInvalidCookieStream(t *testing.T) { | |||||
| 428 | assert.Contains(t, resp.Header.Values("Vary"), "Cookie") | 428 | assert.Contains(t, resp.Header.Values("Vary"), "Cookie") | ||
| 429 | assert.Contains(t, resp.Header.Get("Set-Cookie"), "Max-Age=0") | 429 | assert.Contains(t, resp.Header.Get("Set-Cookie"), "Max-Age=0") | ||
| 430 | } | 430 | } | ||
| 431 | + | ||||
| 432 | + | // An issues stream on a repository with issues off, unknown or private is | |||
| 433 | + | // refused the same way: 404 with the same body. | |||
| 434 | + | func TestEventsIssuesViewsAreDeniedLikeAMissingRepo(t *testing.T) { | |||
| 435 | + | t.Parallel() | |||
| 436 | + | priv := mocks.NewRepoService(t) | |||
| 437 | + | priv.EXPECT().Get(mock.Anything, mock.Anything, mock.Anything).Return(domain.Repo{}, domain.ErrNotFound) | |||
| 438 | + | for _, view := range []string{"issues", "issue&n=1"} { | |||
| 439 | + | answers := map[string]string{} | |||
| 440 | + | for name, repos := range map[string]ports.RepoService{ | |||
| 441 | + | "off": reposServing(t, issueRepoView(t, domain.IssuesOff)), | |||
| 442 | + | "unset": reposServing(t, issueRepoView(t, "")), | |||
| 443 | + | "unknown": reposServing(t), | |||
| 444 | + | "private": priv, | |||
| 445 | + | } { | |||
| 446 | + | f := &watchProbe{release: make(chan struct{})} | |||
| 447 | + | srv := newTestServer(t, Config{Repos: repos, Browse: mocks.NewBrowseService(t), Events: f.service(t), Issues: mocks.NewIssueService(t), Heartbeat: time.Hour}) | |||
| 448 | + | ts := httptest.NewServer(srv) | |||
| 449 | + | t.Cleanup(ts.Close) | |||
| 450 | + | resp, _ := open(t, ts, "topic=repo:proj&view="+view, nil) | |||
| 451 | + | body, _ := io.ReadAll(resp.Body) | |||
| 452 | + | assert.Equal(t, http.StatusNotFound, resp.StatusCode, "%s %s", view, name) | |||
| 453 | + | answers[name] = resp.Header.Get("Cache-Control") + "|" + string(body) | |||
| 454 | + | assert.Zero(t, f.opened.Load(), "%s %s: no watch is opened", view, name) | |||
| 455 | + | } | |||
| 456 | + | for name, a := range answers { | |||
| 457 | + | assert.Equal(t, answers["unknown"], a, "%s %s", view, name) | |||
| 458 | + | } | |||
| 459 | + | } | |||
| 460 | + | // With issues on, the stream opens. | |||
| 461 | + | f := &watchProbe{release: make(chan struct{})} | |||
| 462 | + | srv := newTestServer(t, Config{Repos: reposServing(t, issueRepoView(t, domain.IssuesOwner)), Browse: mocks.NewBrowseService(t), Events: f.service(t), Issues: mocks.NewIssueService(t), Heartbeat: time.Hour}) | |||
| 463 | + | ts := httptest.NewServer(srv) | |||
| 464 | + | t.Cleanup(ts.Close) | |||
| 465 | + | resp, _ := open(t, ts, "topic=repo:proj&view=issues", nil) | |||
| 466 | + | assert.Equal(t, http.StatusOK, resp.StatusCode) | |||
| 467 | + | } | |||
M internal/usecase/events.go +12 -0
| @@ -552,6 +552,9 @@ func (e *Events) deliverRepo(ctx context.Context, w *watch, name domain.RepoName | |||
| 552 | 552 | if !canRead(w.p, repo) { | |
| 553 | 553 | return nil | |
| 554 | 554 | } | |
| 555 | + | if isIssueEvent(be.ev) && repo.Settings.Issues.OrDefault() == domain.IssuesOff { | |
| 556 | + | return nil // issues were turned off since: a replay must not show them | |
| 557 | + | } | |
| 555 | 558 | if withRepo { | |
| 556 | 559 | view, err := e.repos.View(ctx, repo) | |
| 557 | 560 | if err != nil { | |
| @@ -561,3 +564,12 @@ func (e *Events) deliverRepo(ctx context.Context, w *watch, name domain.RepoName | |||
| 561 | 564 | } | |
| 562 | 565 | return send(out) | |
| 563 | 566 | } | |
| 567 | + | ||
| 568 | + | // isIssueEvent reports whether ev is about an issue. | |
| 569 | + | func isIssueEvent(ev domain.Event) bool { | |
| 570 | + | switch ev.(type) { | |
| 571 | + | case domain.IssueOpened, domain.IssueClosed, domain.IssueReopened, domain.IssueCommented: | |
| 572 | + | return true | |
| 573 | + | } | |
| 574 | + | return false | |
| 575 | + | } | |
| @@ -552,6 +552,9 @@ func (e *Events) deliverRepo(ctx context.Context, w *watch, name domain.RepoName | |||||
| 552 | if !canRead(w.p, repo) { | 552 | if !canRead(w.p, repo) { | ||
| 553 | return nil | 553 | return nil | ||
| 554 | } | 554 | } | ||
| 555 | + | if isIssueEvent(be.ev) && repo.Settings.Issues.OrDefault() == domain.IssuesOff { | |||
| 556 | + | return nil // issues were turned off since: a replay must not show them | |||
| 557 | + | } | |||
| 555 | if withRepo { | 558 | if withRepo { | ||
| 556 | view, err := e.repos.View(ctx, repo) | 559 | view, err := e.repos.View(ctx, repo) | ||
| 557 | if err != nil { | 560 | if err != nil { | ||
| @@ -561,3 +564,12 @@ func (e *Events) deliverRepo(ctx context.Context, w *watch, name domain.RepoName | |||||
| 561 | } | 564 | } | ||
| 562 | return send(out) | 565 | return send(out) | ||
| 563 | } | 566 | } | ||
| 567 | + | ||||
| 568 | + | // isIssueEvent reports whether ev is about an issue. | |||
| 569 | + | func isIssueEvent(ev domain.Event) bool { | |||
| 570 | + | switch ev.(type) { | |||
| 571 | + | case domain.IssueOpened, domain.IssueClosed, domain.IssueReopened, domain.IssueCommented: | |||
| 572 | + | return true | |||
| 573 | + | } | |||
| 574 | + | return false | |||
| 575 | + | } | |||
M internal/usecase/events_test.go +43 -0
| @@ -206,6 +206,49 @@ func TestWatchRechecksAccessPerEvent(t *testing.T) { | |||
| 206 | 206 | }) | |
| 207 | 207 | } | |
| 208 | 208 | ||
| 209 | + | func 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 | + | ||
| 209 | 252 | func TestWatchResumeAndGap(t *testing.T) { | |
| 210 | 253 | t.Parallel() | |
| 211 | 254 | const size = 8 | |
| @@ -206,6 +206,49 @@ func TestWatchRechecksAccessPerEvent(t *testing.T) { | |||||
| 206 | }) | 206 | }) | ||
| 207 | } | 207 | } | ||
| 208 | 208 | ||||
| 209 | + | func 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 | + | ||||
| 209 | func TestWatchResumeAndGap(t *testing.T) { | 252 | func TestWatchResumeAndGap(t *testing.T) { | ||
| 210 | t.Parallel() | 253 | t.Parallel() | ||
| 211 | const size = 8 | 254 | const size = 8 | ||