fix(usecase): refresh the inbox after a rejection
- commit
- b0db7ade024de25e7d9010c18caf3390b0eb0d9e
- author
- brice · · 2026-10-19 08:50 UTC
- parent
- 03a1ab8
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
8 files · +80 -19
M internal/adapters/rpc/convert.go +2 -0
| @@ -494,6 +494,8 @@ func watchEventToPB(we ports.WatchEvent) *pb.Event { | |||
| 494 | 494 | out.Time = timestamp(ev.At) | |
| 495 | 495 | out.Kind = jobEventToPB(ev) | |
| 496 | 496 | default: | |
| 497 | + | // Web-only events, such as domain.IssuePendingRemoved (the API has | |
| 498 | + | // no message for it; watchers list pending issues on demand). | |
| 497 | 499 | return nil | |
| 498 | 500 | } | |
| 499 | 501 | return out | |
| @@ -494,6 +494,8 @@ func watchEventToPB(we ports.WatchEvent) *pb.Event { | |||||
| 494 | out.Time = timestamp(ev.At) | 494 | out.Time = timestamp(ev.At) | ||
| 495 | out.Kind = jobEventToPB(ev) | 495 | out.Kind = jobEventToPB(ev) | ||
| 496 | default: | 496 | default: | ||
| 497 | + | // Web-only events, such as domain.IssuePendingRemoved (the API has | |||
| 498 | + | // no message for it; watchers list pending issues on demand). | |||
| 497 | return nil | 499 | return nil | ||
| 498 | } | 500 | } | ||
| 499 | return out | 501 | return out | ||
M internal/adapters/rpc/convert_test.go +5 -0
| @@ -370,3 +370,8 @@ func TestIssueMappers(t *testing.T) { | |||
| 370 | 370 | assert.NotNil(t, watchEventToPB(ports.WatchEvent{Event: domain.IssueOpened{}}).GetIssueOpened()) | |
| 371 | 371 | assert.NotNil(t, watchEventToPB(ports.WatchEvent{Event: domain.IssueReopened{}}).GetIssueReopened()) | |
| 372 | 372 | } | |
| 373 | + | ||
| 374 | + | func TestWatchEventToPBSkipsPendingRemoved(t *testing.T) { | |
| 375 | + | t.Parallel() | |
| 376 | + | assert.Nil(t, watchEventToPB(ports.WatchEvent{Event: domain.IssuePendingRemoved{}})) | |
| 377 | + | } | |
| @@ -370,3 +370,8 @@ func TestIssueMappers(t *testing.T) { | |||||
| 370 | assert.NotNil(t, watchEventToPB(ports.WatchEvent{Event: domain.IssueOpened{}}).GetIssueOpened()) | 370 | assert.NotNil(t, watchEventToPB(ports.WatchEvent{Event: domain.IssueOpened{}}).GetIssueOpened()) | ||
| 371 | assert.NotNil(t, watchEventToPB(ports.WatchEvent{Event: domain.IssueReopened{}}).GetIssueReopened()) | 371 | assert.NotNil(t, watchEventToPB(ports.WatchEvent{Event: domain.IssueReopened{}}).GetIssueReopened()) | ||
| 372 | } | 372 | } | ||
| 373 | + | ||||
| 374 | + | func TestWatchEventToPBSkipsPendingRemoved(t *testing.T) { | |||
| 375 | + | t.Parallel() | |||
| 376 | + | assert.Nil(t, watchEventToPB(ports.WatchEvent{Event: domain.IssuePendingRemoved{}})) | |||
| 377 | + | } | |||
M internal/adapters/web/fragments.go +1 -1
| @@ -62,7 +62,7 @@ func liveEvent(we ports.WatchEvent) bool { | |||
| 62 | 62 | case domain.RepoCreated, domain.RepoUpdated, domain.RepoDeleted, domain.PushEvent, | |
| 63 | 63 | domain.MergeRequestOpened, domain.MergeRequestMerged, domain.MergeRequestClosed, domain.MergeRequestReady, domain.CommentAdded, | |
| 64 | 64 | domain.RunUpdated, domain.JobUpdated, | |
| 65 | - | domain.IssueOpened, domain.IssueClosed, domain.IssueReopened, domain.IssueCommented, domain.IssueSubmitted: | |
| 65 | + | domain.IssueOpened, domain.IssueClosed, domain.IssueReopened, domain.IssueCommented, domain.IssueSubmitted, domain.IssuePendingRemoved: | |
| 66 | 66 | return true | |
| 67 | 67 | } | |
| 68 | 68 | return false | |
| @@ -62,7 +62,7 @@ func liveEvent(we ports.WatchEvent) bool { | |||||
| 62 | case domain.RepoCreated, domain.RepoUpdated, domain.RepoDeleted, domain.PushEvent, | 62 | case domain.RepoCreated, domain.RepoUpdated, domain.RepoDeleted, domain.PushEvent, | ||
| 63 | domain.MergeRequestOpened, domain.MergeRequestMerged, domain.MergeRequestClosed, domain.MergeRequestReady, domain.CommentAdded, | 63 | domain.MergeRequestOpened, domain.MergeRequestMerged, domain.MergeRequestClosed, domain.MergeRequestReady, domain.CommentAdded, | ||
| 64 | domain.RunUpdated, domain.JobUpdated, | 64 | domain.RunUpdated, domain.JobUpdated, | ||
| 65 | - | domain.IssueOpened, domain.IssueClosed, domain.IssueReopened, domain.IssueCommented, domain.IssueSubmitted: | 65 | + | domain.IssueOpened, domain.IssueClosed, domain.IssueReopened, domain.IssueCommented, domain.IssueSubmitted, domain.IssuePendingRemoved: |
| 66 | return true | 66 | return true | ||
| 67 | } | 67 | } | ||
| 68 | return false | 68 | return false | ||
M internal/adapters/web/inbox.go +1 -1
| @@ -265,7 +265,7 @@ func (s *Server) inboxCount(ctx context.Context) (n int, show bool) { | |||
| 265 | 265 | func (l *liveStream) inbox(ctx context.Context, we ports.WatchEvent) ([]fragment, error) { | |
| 266 | 266 | switch we.Event.(type) { | |
| 267 | 267 | case domain.MergeRequestOpened, domain.MergeRequestMerged, domain.MergeRequestClosed, domain.MergeRequestReady, domain.CommentAdded, | |
| 268 | - | domain.IssueOpened, domain.IssueClosed, domain.IssueReopened, domain.IssueCommented, domain.IssueSubmitted, | |
| 268 | + | domain.IssueOpened, domain.IssueClosed, domain.IssueReopened, domain.IssueCommented, domain.IssueSubmitted, domain.IssuePendingRemoved, | |
| 269 | 269 | domain.RepoUpdated, domain.RepoDeleted: | |
| 270 | 270 | default: | |
| 271 | 271 | return nil, nil | |
| @@ -265,7 +265,7 @@ func (s *Server) inboxCount(ctx context.Context) (n int, show bool) { | |||||
| 265 | func (l *liveStream) inbox(ctx context.Context, we ports.WatchEvent) ([]fragment, error) { | 265 | func (l *liveStream) inbox(ctx context.Context, we ports.WatchEvent) ([]fragment, error) { | ||
| 266 | switch we.Event.(type) { | 266 | switch we.Event.(type) { | ||
| 267 | case domain.MergeRequestOpened, domain.MergeRequestMerged, domain.MergeRequestClosed, domain.MergeRequestReady, domain.CommentAdded, | 267 | case domain.MergeRequestOpened, domain.MergeRequestMerged, domain.MergeRequestClosed, domain.MergeRequestReady, domain.CommentAdded, | ||
| 268 | - | domain.IssueOpened, domain.IssueClosed, domain.IssueReopened, domain.IssueCommented, domain.IssueSubmitted, | 268 | + | domain.IssueOpened, domain.IssueClosed, domain.IssueReopened, domain.IssueCommented, domain.IssueSubmitted, domain.IssuePendingRemoved, |
| 269 | domain.RepoUpdated, domain.RepoDeleted: | 269 | domain.RepoUpdated, domain.RepoDeleted: | ||
| 270 | default: | 270 | default: | ||
| 271 | return nil, nil | 271 | return nil, nil | ||
M internal/domain/event.go +12 -0
| @@ -233,6 +233,18 @@ type IssueSubmitted struct { | |||
| 233 | 233 | // EventKind implements Event. | |
| 234 | 234 | func (IssueSubmitted) EventKind() string { return "issue.submitted" } | |
| 235 | 235 | ||
| 236 | + | // IssuePendingRemoved is published when a pending submission is rejected, so | |
| 237 | + | // the other open inboxes drop it. Like IssueSubmitted it carries no content | |
| 238 | + | // and is delivered to the owner topic only. | |
| 239 | + | type IssuePendingRemoved struct { | |
| 240 | + | Repo RepoName | |
| 241 | + | ID PendingIssueID | |
| 242 | + | At time.Time | |
| 243 | + | } | |
| 244 | + | ||
| 245 | + | // EventKind implements Event. | |
| 246 | + | func (IssuePendingRemoved) EventKind() string { return "issue.pending_removed" } | |
| 247 | + | ||
| 236 | 248 | // EventGap tells a subscriber that it missed events because it was too slow; | |
| 237 | 249 | // it must refresh its whole view. | |
| 238 | 250 | type EventGap struct{} | |
| @@ -233,6 +233,18 @@ type IssueSubmitted struct { | |||||
| 233 | // EventKind implements Event. | 233 | // EventKind implements Event. | ||
| 234 | func (IssueSubmitted) EventKind() string { return "issue.submitted" } | 234 | func (IssueSubmitted) EventKind() string { return "issue.submitted" } | ||
| 235 | 235 | ||||
| 236 | + | // IssuePendingRemoved is published when a pending submission is rejected, so | |||
| 237 | + | // the other open inboxes drop it. Like IssueSubmitted it carries no content | |||
| 238 | + | // and is delivered to the owner topic only. | |||
| 239 | + | type IssuePendingRemoved struct { | |||
| 240 | + | Repo RepoName | |||
| 241 | + | ID PendingIssueID | |||
| 242 | + | At time.Time | |||
| 243 | + | } | |||
| 244 | + | ||||
| 245 | + | // EventKind implements Event. | |||
| 246 | + | func (IssuePendingRemoved) EventKind() string { return "issue.pending_removed" } | |||
| 247 | + | ||||
| 236 | // EventGap tells a subscriber that it missed events because it was too slow; | 248 | // EventGap tells a subscriber that it missed events because it was too slow; | ||
| 237 | // it must refresh its whole view. | 249 | // it must refresh its whole view. | ||
| 238 | type EventGap struct{} | 250 | type EventGap struct{} | ||
M internal/usecase/events.go +23 -17
| @@ -190,7 +190,7 @@ func (e *Events) snapshot(ctx context.Context, ev domain.Event) bufferedEvent { | |||
| 190 | 190 | return repoSnapshot(ev.Visibility, ev.CreatedAt) | |
| 191 | 191 | case domain.IssueCommented: | |
| 192 | 192 | return repoSnapshot(ev.Visibility, ev.CreatedAt) | |
| 193 | - | case domain.IssueSubmitted: | |
| 193 | + | case domain.IssueSubmitted, domain.IssuePendingRemoved: | |
| 194 | 194 | return bufferedEvent{known: true} // owner topic only: no repository snapshot | |
| 195 | 195 | case domain.RunUpdated: | |
| 196 | 196 | return repoSnapshot(ev.Visibility, ev.CreatedAt) | |
| @@ -502,22 +502,9 @@ func (e *Events) deliver(ctx context.Context, w *watch, be bufferedEvent, send f | |||
| 502 | 502 | } | |
| 503 | 503 | return send(out) | |
| 504 | 504 | case domain.IssueSubmitted: | |
| 505 | - | // Owner topic only: never home, never a repository topic, not even | |
| 506 | - | // for an owner who watches the repository. No content travels. | |
| 507 | - | if !w.owner { | |
| 508 | - | return nil | |
| 509 | - | } | |
| 510 | - | repo, err := lookupRepo(ctx, e.repos.meta, ev.Repo) | |
| 511 | - | switch { | |
| 512 | - | case errors.Is(err, domain.ErrNotFound), ctx.Err() != nil: | |
| 513 | - | return nil | |
| 514 | - | case err != nil: | |
| 515 | - | return fmt.Errorf("check access to %q: %w", ev.Repo, err) | |
| 516 | - | } | |
| 517 | - | if repo.Settings.Issues.OrDefault() == domain.IssuesOff { | |
| 518 | - | return nil | |
| 519 | - | } | |
| 520 | - | return send(out) | |
| 505 | + | return e.deliverOwnerOnly(ctx, w, ev.Repo, out, send) | |
| 506 | + | case domain.IssuePendingRemoved: | |
| 507 | + | return e.deliverOwnerOnly(ctx, w, ev.Repo, out, send) | |
| 521 | 508 | case domain.RepoCreated: | |
| 522 | 509 | return e.deliverRepo(ctx, w, ev.Repo, true, be, out, send) | |
| 523 | 510 | case domain.RepoUpdated: | |
| @@ -557,6 +544,25 @@ func (e *Events) deliver(ctx context.Context, w *watch, be bufferedEvent, send f | |||
| 557 | 544 | return nil // an event this version does not stream | |
| 558 | 545 | } | |
| 559 | 546 | ||
| 547 | + | // deliverOwnerOnly sends an owner-topic event: never home, never a repository | |
| 548 | + | // topic, not even to an owner who watches the repository. No content travels. | |
| 549 | + | func (e *Events) deliverOwnerOnly(ctx context.Context, w *watch, name domain.RepoName, out ports.WatchEvent, send func(ports.WatchEvent) error) error { | |
| 550 | + | if !w.owner { | |
| 551 | + | return nil | |
| 552 | + | } | |
| 553 | + | repo, err := lookupRepo(ctx, e.repos.meta, name) | |
| 554 | + | switch { | |
| 555 | + | case errors.Is(err, domain.ErrNotFound), ctx.Err() != nil: | |
| 556 | + | return nil | |
| 557 | + | case err != nil: | |
| 558 | + | return fmt.Errorf("check access to %q: %w", name, err) | |
| 559 | + | } | |
| 560 | + | if repo.Settings.Issues.OrDefault() == domain.IssuesOff { | |
| 561 | + | return nil | |
| 562 | + | } | |
| 563 | + | return send(out) | |
| 564 | + | } | |
| 565 | + | ||
| 560 | 566 | // deliverRepo sends an event about a repository if the watcher could read the | |
| 561 | 567 | // repository when the event happened and can read it now, and the repository | |
| 562 | 568 | // is still the one the event is about. A repository that no longer exists is | |
| @@ -190,7 +190,7 @@ func (e *Events) snapshot(ctx context.Context, ev domain.Event) bufferedEvent { | |||||
| 190 | return repoSnapshot(ev.Visibility, ev.CreatedAt) | 190 | return repoSnapshot(ev.Visibility, ev.CreatedAt) | ||
| 191 | case domain.IssueCommented: | 191 | case domain.IssueCommented: | ||
| 192 | return repoSnapshot(ev.Visibility, ev.CreatedAt) | 192 | return repoSnapshot(ev.Visibility, ev.CreatedAt) | ||
| 193 | - | case domain.IssueSubmitted: | 193 | + | case domain.IssueSubmitted, domain.IssuePendingRemoved: |
| 194 | return bufferedEvent{known: true} // owner topic only: no repository snapshot | 194 | return bufferedEvent{known: true} // owner topic only: no repository snapshot | ||
| 195 | case domain.RunUpdated: | 195 | case domain.RunUpdated: | ||
| 196 | return repoSnapshot(ev.Visibility, ev.CreatedAt) | 196 | return repoSnapshot(ev.Visibility, ev.CreatedAt) | ||
| @@ -502,22 +502,9 @@ func (e *Events) deliver(ctx context.Context, w *watch, be bufferedEvent, send f | |||||
| 502 | } | 502 | } | ||
| 503 | return send(out) | 503 | return send(out) | ||
| 504 | case domain.IssueSubmitted: | 504 | case domain.IssueSubmitted: | ||
| 505 | - | // Owner topic only: never home, never a repository topic, not even | 505 | + | return e.deliverOwnerOnly(ctx, w, ev.Repo, out, send) |
| 506 | - | // for an owner who watches the repository. No content travels. | 506 | + | case domain.IssuePendingRemoved: |
| 507 | - | if !w.owner { | 507 | + | return e.deliverOwnerOnly(ctx, w, ev.Repo, out, send) |
| 508 | - | return nil | |||
| 509 | - | } | |||
| 510 | - | repo, err := lookupRepo(ctx, e.repos.meta, ev.Repo) | |||
| 511 | - | switch { | |||
| 512 | - | case errors.Is(err, domain.ErrNotFound), ctx.Err() != nil: | |||
| 513 | - | return nil | |||
| 514 | - | case err != nil: | |||
| 515 | - | return fmt.Errorf("check access to %q: %w", ev.Repo, err) | |||
| 516 | - | } | |||
| 517 | - | if repo.Settings.Issues.OrDefault() == domain.IssuesOff { | |||
| 518 | - | return nil | |||
| 519 | - | } | |||
| 520 | - | return send(out) | |||
| 521 | case domain.RepoCreated: | 508 | case domain.RepoCreated: | ||
| 522 | return e.deliverRepo(ctx, w, ev.Repo, true, be, out, send) | 509 | return e.deliverRepo(ctx, w, ev.Repo, true, be, out, send) | ||
| 523 | case domain.RepoUpdated: | 510 | case domain.RepoUpdated: | ||
| @@ -557,6 +544,25 @@ func (e *Events) deliver(ctx context.Context, w *watch, be bufferedEvent, send f | |||||
| 557 | return nil // an event this version does not stream | 544 | return nil // an event this version does not stream | ||
| 558 | } | 545 | } | ||
| 559 | 546 | ||||
| 547 | + | // deliverOwnerOnly sends an owner-topic event: never home, never a repository | |||
| 548 | + | // topic, not even to an owner who watches the repository. No content travels. | |||
| 549 | + | func (e *Events) deliverOwnerOnly(ctx context.Context, w *watch, name domain.RepoName, out ports.WatchEvent, send func(ports.WatchEvent) error) error { | |||
| 550 | + | if !w.owner { | |||
| 551 | + | return nil | |||
| 552 | + | } | |||
| 553 | + | repo, err := lookupRepo(ctx, e.repos.meta, name) | |||
| 554 | + | switch { | |||
| 555 | + | case errors.Is(err, domain.ErrNotFound), ctx.Err() != nil: | |||
| 556 | + | return nil | |||
| 557 | + | case err != nil: | |||
| 558 | + | return fmt.Errorf("check access to %q: %w", name, err) | |||
| 559 | + | } | |||
| 560 | + | if repo.Settings.Issues.OrDefault() == domain.IssuesOff { | |||
| 561 | + | return nil | |||
| 562 | + | } | |||
| 563 | + | return send(out) | |||
| 564 | + | } | |||
| 565 | + | ||||
| 560 | // deliverRepo sends an event about a repository if the watcher could read the | 566 | // deliverRepo sends an event about a repository if the watcher could read the | ||
| 561 | // repository when the event happened and can read it now, and the repository | 567 | // repository when the event happened and can read it now, and the repository | ||
| 562 | // is still the one the event is about. A repository that no longer exists is | 568 | // is still the one the event is about. A repository that no longer exists is | ||
M internal/usecase/issue_moderation.go +1 -0
| @@ -114,5 +114,6 @@ func (m *IssueModeration) Reject(ctx context.Context, p domain.Principal, repo d | |||
| 114 | 114 | log := zerowrap.FromCtx(ctx) | |
| 115 | 115 | log.Info().Str("repo", repo.String()).Bool("ban", ban). | |
| 116 | 116 | Str("principal", p.Label()).Str("token_id", p.TokenID()).Msg("pending issue rejected") | |
| 117 | + | m.bus.Publish(ctx, domain.IssuePendingRemoved{Repo: repo, ID: id, At: m.clock.Now().UTC()}) | |
| 117 | 118 | return nil | |
| 118 | 119 | } | |
| @@ -114,5 +114,6 @@ func (m *IssueModeration) Reject(ctx context.Context, p domain.Principal, repo d | |||||
| 114 | log := zerowrap.FromCtx(ctx) | 114 | log := zerowrap.FromCtx(ctx) | ||
| 115 | log.Info().Str("repo", repo.String()).Bool("ban", ban). | 115 | log.Info().Str("repo", repo.String()).Bool("ban", ban). | ||
| 116 | Str("principal", p.Label()).Str("token_id", p.TokenID()).Msg("pending issue rejected") | 116 | Str("principal", p.Label()).Str("token_id", p.TokenID()).Msg("pending issue rejected") | ||
| 117 | + | m.bus.Publish(ctx, domain.IssuePendingRemoved{Repo: repo, ID: id, At: m.clock.Now().UTC()}) | |||
| 117 | return nil | 118 | return nil | ||
| 118 | } | 119 | } | ||
M internal/usecase/issue_moderation_test.go +35 -0
| @@ -117,12 +117,14 @@ func TestIssueModerationReject(t *testing.T) { | |||
| 117 | 117 | u.pending.EXPECT().Get(mock.Anything, u.repo.Name, u.id).Return(domain.PendingIssue{ID: u.id, Repo: u.repo.Name, IPHash: hash}, nil).Once() | |
| 118 | 118 | u.bans.EXPECT().Ban(mock.Anything, hash, now.Add(domain.BanDuration)).Return(nil).Once() | |
| 119 | 119 | u.pending.EXPECT().Delete(mock.Anything, u.repo.Name, u.id).Return(nil).Once() | |
| 120 | + | u.bus.EXPECT().Publish(mock.Anything, domain.IssuePendingRemoved{Repo: u.repo.Name, ID: u.id, At: now}).Once() | |
| 120 | 121 | require.NoError(t, u.mod.Reject(ctx, domain.Owner(), u.repo.Name, u.id, true)) | |
| 121 | 122 | }) | |
| 122 | 123 | t.Run("without a ban", func(t *testing.T) { | |
| 123 | 124 | u := newModerationUnit(t, domain.IssuesPublic) | |
| 124 | 125 | u.pending.EXPECT().Get(mock.Anything, u.repo.Name, u.id).Return(domain.PendingIssue{ID: u.id, IPHash: hash}, nil).Once() | |
| 125 | 126 | u.pending.EXPECT().Delete(mock.Anything, u.repo.Name, u.id).Return(nil).Once() | |
| 127 | + | u.bus.EXPECT().Publish(mock.Anything, domain.IssuePendingRemoved{Repo: u.repo.Name, ID: u.id, At: now}).Once() | |
| 126 | 128 | require.NoError(t, u.mod.Reject(ctx, domain.Owner(), u.repo.Name, u.id, false)) | |
| 127 | 129 | }) | |
| 128 | 130 | t.Run("unknown id", func(t *testing.T) { | |
| @@ -207,3 +209,36 @@ func TestIssueSubmittedReachesOnlyTheOwnerTopic(t *testing.T) { | |||
| 207 | 209 | } | |
| 208 | 210 | }) | |
| 209 | 211 | } | |
| 212 | + | ||
| 213 | + | func TestIssuePendingRemovedReachesOnlyOwnerTopics(t *testing.T) { | |
| 214 | + | t.Parallel() | |
| 215 | + | synctest.Test(t, func(t *testing.T) { | |
| 216 | + | r := newEventRig(t, 16) | |
| 217 | + | a := r.repo("aaa", domain.VisibilityPublic) | |
| 218 | + | r.meta.EXPECT().Get(mock.Anything, a).Unset() | |
| 219 | + | r.meta.EXPECT().Get(mock.Anything, a).Return(domain.Repo{Name: a, Visibility: domain.VisibilityPublic, CreatedAt: now, Settings: withIssues(domain.IssuesPublic)}, nil).Maybe() | |
| 220 | + | ctx, cancel := context.WithCancel(context.Background()) | |
| 221 | + | defer cancel() | |
| 222 | + | ||
| 223 | + | owner1 := r.watch(ctx, domain.Owner(), ports.WatchParams{Topics: []string{"owner"}}) | |
| 224 | + | owner2 := r.watch(ctx, domain.Owner(), ports.WatchParams{Topics: []string{"owner"}}) | |
| 225 | + | home := r.watch(ctx, domain.Owner(), ports.WatchParams{Topics: []string{"home"}}) | |
| 226 | + | repoTopic := r.watch(ctx, domain.Owner(), ports.WatchParams{Topics: []string{"repo:aaa"}}) | |
| 227 | + | anon := r.watch(ctx, domain.Anonymous(), ports.WatchParams{Topics: []string{"repo:aaa", "home"}}) | |
| 228 | + | synctest.Wait() | |
| 229 | + | ||
| 230 | + | r.feed <- domain.IssuePendingRemoved{Repo: a, ID: "x", At: now} | |
| 231 | + | r.feed <- domain.IssueOpened{Issue: domain.Issue{Repo: a, Number: 1}, At: now, Visibility: domain.VisibilityPublic, CreatedAt: now} | |
| 232 | + | ||
| 233 | + | for name, w := range map[string]*watcher{"owner 1": owner1, "owner 2": owner2} { | |
| 234 | + | _, ok := w.next(t).Event.(domain.IssuePendingRemoved) | |
| 235 | + | require.True(t, ok, "%s gets the refresh", name) | |
| 236 | + | _, ok = w.next(t).Event.(domain.IssueOpened) | |
| 237 | + | require.True(t, ok) | |
| 238 | + | } | |
| 239 | + | for name, w := range map[string]*watcher{"home": home, "repo topic": repoTopic, "anonymous": anon} { | |
| 240 | + | e, ok := w.next(t).Event.(domain.IssueOpened) | |
| 241 | + | require.True(t, ok, "%s: the first event is the sentinel (got %T)", name, e) | |
| 242 | + | } | |
| 243 | + | }) | |
| 244 | + | } | |
| @@ -117,12 +117,14 @@ func TestIssueModerationReject(t *testing.T) { | |||||
| 117 | u.pending.EXPECT().Get(mock.Anything, u.repo.Name, u.id).Return(domain.PendingIssue{ID: u.id, Repo: u.repo.Name, IPHash: hash}, nil).Once() | 117 | u.pending.EXPECT().Get(mock.Anything, u.repo.Name, u.id).Return(domain.PendingIssue{ID: u.id, Repo: u.repo.Name, IPHash: hash}, nil).Once() | ||
| 118 | u.bans.EXPECT().Ban(mock.Anything, hash, now.Add(domain.BanDuration)).Return(nil).Once() | 118 | u.bans.EXPECT().Ban(mock.Anything, hash, now.Add(domain.BanDuration)).Return(nil).Once() | ||
| 119 | u.pending.EXPECT().Delete(mock.Anything, u.repo.Name, u.id).Return(nil).Once() | 119 | u.pending.EXPECT().Delete(mock.Anything, u.repo.Name, u.id).Return(nil).Once() | ||
| 120 | + | u.bus.EXPECT().Publish(mock.Anything, domain.IssuePendingRemoved{Repo: u.repo.Name, ID: u.id, At: now}).Once() | |||
| 120 | require.NoError(t, u.mod.Reject(ctx, domain.Owner(), u.repo.Name, u.id, true)) | 121 | require.NoError(t, u.mod.Reject(ctx, domain.Owner(), u.repo.Name, u.id, true)) | ||
| 121 | }) | 122 | }) | ||
| 122 | t.Run("without a ban", func(t *testing.T) { | 123 | t.Run("without a ban", func(t *testing.T) { | ||
| 123 | u := newModerationUnit(t, domain.IssuesPublic) | 124 | u := newModerationUnit(t, domain.IssuesPublic) | ||
| 124 | u.pending.EXPECT().Get(mock.Anything, u.repo.Name, u.id).Return(domain.PendingIssue{ID: u.id, IPHash: hash}, nil).Once() | 125 | u.pending.EXPECT().Get(mock.Anything, u.repo.Name, u.id).Return(domain.PendingIssue{ID: u.id, IPHash: hash}, nil).Once() | ||
| 125 | u.pending.EXPECT().Delete(mock.Anything, u.repo.Name, u.id).Return(nil).Once() | 126 | u.pending.EXPECT().Delete(mock.Anything, u.repo.Name, u.id).Return(nil).Once() | ||
| 127 | + | u.bus.EXPECT().Publish(mock.Anything, domain.IssuePendingRemoved{Repo: u.repo.Name, ID: u.id, At: now}).Once() | |||
| 126 | require.NoError(t, u.mod.Reject(ctx, domain.Owner(), u.repo.Name, u.id, false)) | 128 | require.NoError(t, u.mod.Reject(ctx, domain.Owner(), u.repo.Name, u.id, false)) | ||
| 127 | }) | 129 | }) | ||
| 128 | t.Run("unknown id", func(t *testing.T) { | 130 | t.Run("unknown id", func(t *testing.T) { | ||
| @@ -207,3 +209,36 @@ func TestIssueSubmittedReachesOnlyTheOwnerTopic(t *testing.T) { | |||||
| 207 | } | 209 | } | ||
| 208 | }) | 210 | }) | ||
| 209 | } | 211 | } | ||
| 212 | + | ||||
| 213 | + | func TestIssuePendingRemovedReachesOnlyOwnerTopics(t *testing.T) { | |||
| 214 | + | t.Parallel() | |||
| 215 | + | synctest.Test(t, func(t *testing.T) { | |||
| 216 | + | r := newEventRig(t, 16) | |||
| 217 | + | a := r.repo("aaa", domain.VisibilityPublic) | |||
| 218 | + | r.meta.EXPECT().Get(mock.Anything, a).Unset() | |||
| 219 | + | r.meta.EXPECT().Get(mock.Anything, a).Return(domain.Repo{Name: a, Visibility: domain.VisibilityPublic, CreatedAt: now, Settings: withIssues(domain.IssuesPublic)}, nil).Maybe() | |||
| 220 | + | ctx, cancel := context.WithCancel(context.Background()) | |||
| 221 | + | defer cancel() | |||
| 222 | + | ||||
| 223 | + | owner1 := r.watch(ctx, domain.Owner(), ports.WatchParams{Topics: []string{"owner"}}) | |||
| 224 | + | owner2 := r.watch(ctx, domain.Owner(), ports.WatchParams{Topics: []string{"owner"}}) | |||
| 225 | + | home := r.watch(ctx, domain.Owner(), ports.WatchParams{Topics: []string{"home"}}) | |||
| 226 | + | repoTopic := r.watch(ctx, domain.Owner(), ports.WatchParams{Topics: []string{"repo:aaa"}}) | |||
| 227 | + | anon := r.watch(ctx, domain.Anonymous(), ports.WatchParams{Topics: []string{"repo:aaa", "home"}}) | |||
| 228 | + | synctest.Wait() | |||
| 229 | + | ||||
| 230 | + | r.feed <- domain.IssuePendingRemoved{Repo: a, ID: "x", At: now} | |||
| 231 | + | r.feed <- domain.IssueOpened{Issue: domain.Issue{Repo: a, Number: 1}, At: now, Visibility: domain.VisibilityPublic, CreatedAt: now} | |||
| 232 | + | ||||
| 233 | + | for name, w := range map[string]*watcher{"owner 1": owner1, "owner 2": owner2} { | |||
| 234 | + | _, ok := w.next(t).Event.(domain.IssuePendingRemoved) | |||
| 235 | + | require.True(t, ok, "%s gets the refresh", name) | |||
| 236 | + | _, ok = w.next(t).Event.(domain.IssueOpened) | |||
| 237 | + | require.True(t, ok) | |||
| 238 | + | } | |||
| 239 | + | for name, w := range map[string]*watcher{"home": home, "repo topic": repoTopic, "anonymous": anon} { | |||
| 240 | + | e, ok := w.next(t).Event.(domain.IssueOpened) | |||
| 241 | + | require.True(t, ok, "%s: the first event is the sentinel (got %T)", name, e) | |||
| 242 | + | } | |||
| 243 | + | }) | |||
| 244 | + | } | |||