feat(events): owner topic across repositories

commit: ab38ed2eefa7888fda1065cc7ea9d81f55002fcd
author: brice · 3d ago · 2026-10-19 06:37 UTC
parent: 974935d
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

2 files · +113 -2

M internal/usecase/events.go +32 -2


@@ -38,6 +38,10 @@ const (
38	38		const (
39	39		// TopicHome follows every repository the caller may see.
40	40		TopicHome = "home"
	41	+	// TopicOwner follows the merge requests and issues of every repository,
	42	+	// for the owner only (the inbox). Anyone else is told it is an unknown
	43	+	// topic.
	44	+	TopicOwner = "owner"
41	45		// topicRepoPrefix starts the topic of one repository: "repo:<name>".
42	46		topicRepoPrefix = "repo:"
43	47		)

@@ -266,6 +270,7 @@ func (e *Events) parseID(id string) (uint64, bool) {
266	270		type watch struct {
267	271		p domain.Principal
268	272		home bool
	273	+	owner bool // the owner topic: merge request and issue events of every repository
269	274		repos map[domain.RepoName]bool
270	275
271	276		checked time.Time // when the token store last confirmed the principal

@@ -273,6 +278,12 @@ type watch struct {
273	278
274	279		func (w *watch) wants(name domain.RepoName) bool { return w.home \|\| w.repos[name] }
275	280
	281	+	// wantsEvent is wants for an event: the owner topic also follows the events
	282	+	// about merge requests and issues of every repository.
	283	+	func (w *watch) wantsEvent(name domain.RepoName, ev domain.Event) bool {
	284	+	return w.wants(name) \|\| (w.owner && isItemEvent(ev))
	285	+	}
	286	+
276	287		// parseTopics validates the topics of a watch. A repository topic needs a
277	288		// repository the principal may read (else domain.ErrNotFound); an unknown
278	289		// topic, or more than MaxWatchTopics, is domain.ErrInvalid.

@@ -289,8 +300,14 @@ func (e *Events) parseTopics(ctx context.Context, p domain.Principal, topics []s
289	300		w.home = true
290	301		continue
291	302		}
	303	+	if t == TopicOwner && p.HasOwnerRights() {
	304	+	w.owner = true
	305	+	continue
	306	+	}
292	307		raw, ok := strings.CutPrefix(t, topicRepoPrefix)
293	308		if !ok {
	309	+	// Includes "owner" for anyone but the owner: the same error as
	310	+	// any unknown topic.
294	311		return nil, fmt.Errorf("%w: unknown topic %q (want %q or \"repo:<name>\")", domain.ErrInvalid, t, TopicHome)
295	312		}
296	313		name, err := domain.ParseRepoName(raw)

@@ -427,7 +444,10 @@ func (e *Events) idleCheck() time.Duration {
427	444
428	445		// stillAllowed checks that the holder of a stored token may still watch: the
429	446		// token has not expired and, at most every few seconds, still exists. Root and
430		-	// anonymous callers have nothing to lose.
	447	+	// anonymous callers have nothing to lose. The owner topic needs nothing more:
	448	+	// a stored token with owner rights loses them only by ceasing to exist, which
	449	+	// is checked here, and the owner's web session, which this use case cannot see,
	450	+	// is checked by the web adapter before it sends anything.
431	451		func (e Events) stillAllowed(ctx context.Context, w watch) error {
432	452		if w.p.TokenID() == "" {
433	453		return nil

@@ -527,7 +547,7 @@ func (e Events) deliver(ctx context.Context, w watch, be bufferedEvent, send f
527	547		// read it before are told once, with its name and its new visibility only,
528	548		// so that what they show can be removed.
529	549		func (e Events) deliverRepo(ctx context.Context, w watch, name domain.RepoName, withRepo bool, be bufferedEvent, out ports.WatchEvent, send func(ports.WatchEvent) error) error {
530		-	if !w.wants(name) \|\| !be.known {
	550	+	if !w.wantsEvent(name, be.ev) \|\| !be.known {
531	551		return nil
532	552		}
533	553		repo, err := lookupRepo(ctx, e.repos.meta, name)

@@ -573,3 +593,13 @@ func isIssueEvent(ev domain.Event) bool {
573	593		}
574	594		return false
575	595		}
	596	+
	597	+	// isItemEvent reports whether ev is about a merge request or an issue.
	598	+	func isItemEvent(ev domain.Event) bool {
	599	+	switch ev.(type) {
	600	+	case domain.MergeRequestOpened, domain.MergeRequestMerged, domain.MergeRequestClosed,
	601	+	domain.MergeRequestReady, domain.CommentAdded:
	602	+	return true
	603	+	}
	604	+	return isIssueEvent(ev)
	605	+	}

 const (
 	// TopicHome follows every repository the caller may see.
 	TopicHome = "home"
+	// TopicOwner follows the merge requests and issues of every repository,
+	// for the owner only (the inbox). Anyone else is told it is an unknown
+	// topic.
+	TopicOwner = "owner"
 	// topicRepoPrefix starts the topic of one repository: "repo:<name>".
 	topicRepoPrefix = "repo:"
 )
 type watch struct {
 	p     domain.Principal
 	home  bool
+	owner bool // the owner topic: merge request and issue events of every repository
 	repos map[domain.RepoName]bool
 	checked time.Time // when the token store last confirmed the principal
 func (w *watch) wants(name domain.RepoName) bool { return w.home || w.repos[name] }
+// wantsEvent is wants for an event: the owner topic also follows the events
+// about merge requests and issues of every repository.
+func (w *watch) wantsEvent(name domain.RepoName, ev domain.Event) bool {
+	return w.wants(name) || (w.owner && isItemEvent(ev))
+}
 // parseTopics validates the topics of a watch. A repository topic needs a
 // repository the principal may read (else domain.ErrNotFound); an unknown
 // topic, or more than MaxWatchTopics, is domain.ErrInvalid.
 			w.home = true
 			continue
 		}
+		if t == TopicOwner && p.HasOwnerRights() {
+			w.owner = true
+			continue
+		}
 		raw, ok := strings.CutPrefix(t, topicRepoPrefix)
 		if !ok {
+			// Includes "owner" for anyone but the owner: the same error as
+			// any unknown topic.
 			return nil, fmt.Errorf("%w: unknown topic %q (want %q or \"repo:<name>\")", domain.ErrInvalid, t, TopicHome)
 		}
 		name, err := domain.ParseRepoName(raw)
 // stillAllowed checks that the holder of a stored token may still watch: the
 // token has not expired and, at most every few seconds, still exists. Root and
-// anonymous callers have nothing to lose.
+// anonymous callers have nothing to lose. The owner topic needs nothing more:
+// a stored token with owner rights loses them only by ceasing to exist, which
+// is checked here, and the owner's web session, which this use case cannot see,
+// is checked by the web adapter before it sends anything.
 func (e *Events) stillAllowed(ctx context.Context, w *watch) error {
 	if w.p.TokenID() == "" {
 		return nil
 // read it before are told once, with its name and its new visibility only,
 // so that what they show can be removed.
 func (e *Events) deliverRepo(ctx context.Context, w *watch, name domain.RepoName, withRepo bool, be bufferedEvent, out ports.WatchEvent, send func(ports.WatchEvent) error) error {
-	if !w.wants(name) || !be.known {
+	if !w.wantsEvent(name, be.ev) || !be.known {
 		return nil
 	}
 	repo, err := lookupRepo(ctx, e.repos.meta, name)
 	}
 	return false
 }
+// isItemEvent reports whether ev is about a merge request or an issue.
+func isItemEvent(ev domain.Event) bool {
+	switch ev.(type) {
+	case domain.MergeRequestOpened, domain.MergeRequestMerged, domain.MergeRequestClosed,
+		domain.MergeRequestReady, domain.CommentAdded:
+		return true
+	}
+	return isIssueEvent(ev)
+}

A internal/usecase/events_owner_test.go +81 -0

@@ -0,0 +1,81 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	package usecase_test
	4	+
	5	+	import (
	6	+	"context"
	7	+	"strings"
	8	+	"sync"
	9	+	"testing"
	10	+	"testing/synctest"
	11	+
	12	+	"github.com/stretchr/testify/assert"
	13	+	"github.com/stretchr/testify/mock"
	14	+	"github.com/stretchr/testify/require"
	15	+
	16	+	"bnema.dev/ermite/internal/domain"
	17	+	"bnema.dev/ermite/internal/ports"
	18	+	)
	19	+
	20	+	func TestWatchOwnerTopicIsUnknownToEveryoneElse(t *testing.T) {
	21	+	t.Parallel()
	22	+	r := newEventRig(t, 0)
	23	+	noop := func(ports.WatchEvent) error { return nil }
	24	+	unknown := r.events.Watch(context.Background(), domain.Anonymous(), ports.WatchParams{Topics: []string{"nope"}}, noop)
	25	+	require.ErrorIs(t, unknown, domain.ErrInvalid)
	26	+
	27	+	partial := tokenWith("t1", now.Add(0), "pub") // limited to a repository
	28	+	for _, p := range []domain.Principal{domain.Anonymous(), partial, issuePrincipal(domain.ScopeIssueWrite)} {
	29	+	err := r.events.Watch(context.Background(), p, ports.WatchParams{Topics: []string{"owner"}}, noop)
	30	+	require.ErrorIs(t, err, domain.ErrInvalid)
	31	+	assert.Equal(t, strings.ReplaceAll(unknown.Error(), "nope", "owner"), err.Error(), "same answer as an unknown topic")
	32	+	}
	33	+	}
	34	+
	35	+	func TestWatchOwnerTopicSpansRepositoriesAndHonoursIssuesOff(t *testing.T) {
	36	+	t.Parallel()
	37	+	synctest.Test(t, func(t *testing.T) {
	38	+	r := newEventRig(t, 16)
	39	+	a, b := name(t, "aaa"), name(t, "bbb")
	40	+	var mu sync.Mutex
	41	+	bMode := domain.IssuesOwner
	42	+	r.meta.EXPECT().Get(mock.Anything, a).Return(domain.Repo{Name: a, Visibility: domain.VisibilityPrivate, CreatedAt: now, Settings: withIssues(domain.IssuesOwner)}, nil)
	43	+	r.meta.EXPECT().Get(mock.Anything, b).RunAndReturn(func(context.Context, domain.RepoName) (domain.Repo, error) {
	44	+	mu.Lock()
	45	+	defer mu.Unlock()
	46	+	return domain.Repo{Name: b, Visibility: domain.VisibilityPrivate, CreatedAt: now, Settings: withIssues(bMode)}, nil
	47	+	})
	48	+	ctx, cancel := context.WithCancel(context.Background())
	49	+	defer cancel()
	50	+	w := r.watch(ctx, domain.Owner(), ports.WatchParams{Topics: []string{"owner"}})
	51	+	synctest.Wait()
	52	+
	53	+	r.feed <- domain.IssueOpened{Issue: domain.Issue{Repo: a, Number: 1}, At: now, Visibility: domain.VisibilityPrivate, CreatedAt: now}
	54	+	_, ok := w.next(t).Event.(domain.IssueOpened)
	55	+	require.True(t, ok)
	56	+	r.feed <- domain.MergeRequestOpened{MergeRequest: domain.MergeRequest{Repo: b, Number: 2}, At: now, Visibility: domain.VisibilityPrivate, CreatedAt: now}
	57	+	_, ok = w.next(t).Event.(domain.MergeRequestOpened)
	58	+	require.True(t, ok, "a second repository")
	59	+
	60	+	// Pushes are not inbox news.
	61	+	r.feed <- r.pushEvent(a, sentinelOID, "refs/heads/x")
	62	+	r.feed <- domain.IssueOpened{Issue: domain.Issue{Repo: b, Number: 3}, At: now, Visibility: domain.VisibilityPrivate, CreatedAt: now}
	63	+	_, ok = w.next(t).Event.(domain.IssueOpened)
	64	+	require.True(t, ok, "the push was skipped")
	65	+
	66	+	mu.Lock()
	67	+	bMode = domain.IssuesOff
	68	+	mu.Unlock()
	69	+	r.feed <- domain.IssueOpened{Issue: domain.Issue{Repo: b, Number: 4}, At: now, Visibility: domain.VisibilityPrivate, CreatedAt: now}
	70	+	r.feed <- domain.MergeRequestClosed{MergeRequest: domain.MergeRequest{Repo: b, Number: 2}, At: now, Visibility: domain.VisibilityPrivate, CreatedAt: now}
	71	+	e := w.next(t)
	72	+	_, ok = e.Event.(domain.MergeRequestClosed)
	73	+	require.True(t, ok, "the issue event of the issues-off repository is not delivered, got %T", e.Event)
	74	+	})
	75	+	}
	76	+
	77	+	func withIssues(m domain.IssuesMode) domain.RepoSettings {
	78	+	st := domain.DefaultRepoSettings()
	79	+	st.Issues = m
	80	+	return st
	81	+	}

@@ -0,0 +1,81 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	package usecase_test
	4	+
	5	+	import (
	6	+	"context"
	7	+	"strings"
	8	+	"sync"
	9	+	"testing"
	10	+	"testing/synctest"
	11	+
	12	+	"github.com/stretchr/testify/assert"
	13	+	"github.com/stretchr/testify/mock"
	14	+	"github.com/stretchr/testify/require"
	15	+
	16	+	"bnema.dev/ermite/internal/domain"
	17	+	"bnema.dev/ermite/internal/ports"
	18	+	)
	19	+
	20	+	func TestWatchOwnerTopicIsUnknownToEveryoneElse(t *testing.T) {
	21	+	t.Parallel()
	22	+	r := newEventRig(t, 0)
	23	+	noop := func(ports.WatchEvent) error { return nil }
	24	+	unknown := r.events.Watch(context.Background(), domain.Anonymous(), ports.WatchParams{Topics: []string{"nope"}}, noop)
	25	+	require.ErrorIs(t, unknown, domain.ErrInvalid)
	26	+
	27	+	partial := tokenWith("t1", now.Add(0), "pub") // limited to a repository
	28	+	for _, p := range []domain.Principal{domain.Anonymous(), partial, issuePrincipal(domain.ScopeIssueWrite)} {
	29	+	err := r.events.Watch(context.Background(), p, ports.WatchParams{Topics: []string{"owner"}}, noop)
	30	+	require.ErrorIs(t, err, domain.ErrInvalid)
	31	+	assert.Equal(t, strings.ReplaceAll(unknown.Error(), "nope", "owner"), err.Error(), "same answer as an unknown topic")
	32	+	}
	33	+	}
	34	+
	35	+	func TestWatchOwnerTopicSpansRepositoriesAndHonoursIssuesOff(t *testing.T) {
	36	+	t.Parallel()
	37	+	synctest.Test(t, func(t *testing.T) {
	38	+	r := newEventRig(t, 16)
	39	+	a, b := name(t, "aaa"), name(t, "bbb")
	40	+	var mu sync.Mutex
	41	+	bMode := domain.IssuesOwner
	42	+	r.meta.EXPECT().Get(mock.Anything, a).Return(domain.Repo{Name: a, Visibility: domain.VisibilityPrivate, CreatedAt: now, Settings: withIssues(domain.IssuesOwner)}, nil)
	43	+	r.meta.EXPECT().Get(mock.Anything, b).RunAndReturn(func(context.Context, domain.RepoName) (domain.Repo, error) {
	44	+	mu.Lock()
	45	+	defer mu.Unlock()
	46	+	return domain.Repo{Name: b, Visibility: domain.VisibilityPrivate, CreatedAt: now, Settings: withIssues(bMode)}, nil
	47	+	})
	48	+	ctx, cancel := context.WithCancel(context.Background())
	49	+	defer cancel()
	50	+	w := r.watch(ctx, domain.Owner(), ports.WatchParams{Topics: []string{"owner"}})
	51	+	synctest.Wait()
	52	+
	53	+	r.feed <- domain.IssueOpened{Issue: domain.Issue{Repo: a, Number: 1}, At: now, Visibility: domain.VisibilityPrivate, CreatedAt: now}
	54	+	_, ok := w.next(t).Event.(domain.IssueOpened)
	55	+	require.True(t, ok)
	56	+	r.feed <- domain.MergeRequestOpened{MergeRequest: domain.MergeRequest{Repo: b, Number: 2}, At: now, Visibility: domain.VisibilityPrivate, CreatedAt: now}
	57	+	_, ok = w.next(t).Event.(domain.MergeRequestOpened)
	58	+	require.True(t, ok, "a second repository")
	59	+
	60	+	// Pushes are not inbox news.
	61	+	r.feed <- r.pushEvent(a, sentinelOID, "refs/heads/x")
	62	+	r.feed <- domain.IssueOpened{Issue: domain.Issue{Repo: b, Number: 3}, At: now, Visibility: domain.VisibilityPrivate, CreatedAt: now}
	63	+	_, ok = w.next(t).Event.(domain.IssueOpened)
	64	+	require.True(t, ok, "the push was skipped")
	65	+
	66	+	mu.Lock()
	67	+	bMode = domain.IssuesOff
	68	+	mu.Unlock()
	69	+	r.feed <- domain.IssueOpened{Issue: domain.Issue{Repo: b, Number: 4}, At: now, Visibility: domain.VisibilityPrivate, CreatedAt: now}
	70	+	r.feed <- domain.MergeRequestClosed{MergeRequest: domain.MergeRequest{Repo: b, Number: 2}, At: now, Visibility: domain.VisibilityPrivate, CreatedAt: now}
	71	+	e := w.next(t)
	72	+	_, ok = e.Event.(domain.MergeRequestClosed)
	73	+	require.True(t, ok, "the issue event of the issues-off repository is not delivered, got %T", e.Event)
	74	+	})
	75	+	}
	76	+
	77	+	func withIssues(m domain.IssuesMode) domain.RepoSettings {
	78	+	st := domain.DefaultRepoSettings()
	79	+	st.Issues = m
	80	+	return st
	81	+	}