fix(events): end Watch for revoked tokens, bound watchers, and keep the visibility of event time

- Watch re-checks token expiry for every event and asks Auth.Revalidate at
  most every 5 s per watcher (also when idle); a revoked or expired token
  ends the stream with unauthenticated.
- At most 64 topics (invalid_argument), 256 watchers overall and 16 per
  client address (resource_exhausted); in-process callers count globally.
- Buffered events record the visibility and generation of their repository:
  a repository made public later does not reveal its private-era events, a
  repository deleted and created again does not inherit the old ones, and a
  repository made private sends one last repo_updated to those who could
  read it before.
- Events.Close honours its context; forge.Events is renamed forge.Bus.
- Event tests use sentinel events instead of fixed negative waits.

commit: 1dc333ecdfe74d6d13a515e46edacda7966cbb16
author: brice · 12d ago · 2026-10-10 11:10 UTC
parent: f305b1c
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

13 files · +678 -70

M cmd/ermite/e2e_helpers_test.go +3 -1


@@ -46,6 +46,8 @@ type harnessOpts struct {
46	46		maxBody int64
47	47		// eventBuffer overrides the size of the event ring when positive.
48	48		eventBuffer int
	49	+	// revalidateEvery overrides how often a watcher's token is re-checked.
	50	+	revalidateEvery time.Duration
49	51		}
50	52
51	53		// newHarness starts a forge. The data directory is short: the hook socket

@@ -79,7 +81,7 @@ func newHarness(t testing.T, o harnessOpts) harness {
79	81		if hookExe == "" {
80	82		hookExe = binary(t)
81	83		}
82		-	f, err := newForge(ctx, cfg, options{hookExe: hookExe, maintenanceDelay: o.maintenance, maxBodyBytes: o.maxBody, eventBuffer: o.eventBuffer})
	84	+	f, err := newForge(ctx, cfg, options{hookExe: hookExe, maintenanceDelay: o.maintenance, maxBodyBytes: o.maxBody, eventBuffer: o.eventBuffer, revalidateEvery: o.revalidateEvery})
83	85		require.NoError(t, err)
84	86
85	87		h := &harness{t: t, forge: f, data: data, logs: logs, ctx: ctx, cancel: cancel, hooksDone: make(chan error, 1)}

 	maxBody int64
 	// eventBuffer overrides the size of the event ring when positive.
 	eventBuffer int
+	// revalidateEvery overrides how often a watcher's token is re-checked.
+	revalidateEvery time.Duration
 }
 // newHarness starts a forge. The data directory is short: the hook socket
 	if hookExe == "" {
 		hookExe = binary(t)
 	}
-	f, err := newForge(ctx, cfg, options{hookExe: hookExe, maintenanceDelay: o.maintenance, maxBodyBytes: o.maxBody, eventBuffer: o.eventBuffer})
+	f, err := newForge(ctx, cfg, options{hookExe: hookExe, maintenanceDelay: o.maintenance, maxBodyBytes: o.maxBody, eventBuffer: o.eventBuffer, revalidateEvery: o.revalidateEvery})
 	require.NoError(t, err)
 	h := &harness{t: t, forge: f, data: data, logs: logs, ctx: ctx, cancel: cancel, hooksDone: make(chan error, 1)}

M cmd/ermite/e2e_test.go +2 -2


@@ -210,7 +210,7 @@ func TestAC2PushAuthorization(t *testing.T) {
210	210		// AC3: first push creates a private repository with HEAD on the pushed branch.
211	211		func TestAC3CreateOnPush(t *testing.T) {
212	212		h := newHarness(t, harnessOpts{})
213		-	events := h.forge.Events.Subscribe(h.ctx)
	213	+	events := h.forge.Bus.Subscribe(h.ctx)
214	214
215	215		// Only "dev" is pushed: HEAD (initially main) moves to it.
216	216		wt := h.workTree("dev", "a")

@@ -316,7 +316,7 @@ func TestAC4HooksUnknownTokenAndMissingBinary(t *testing.T) {
316	316
317	317		func TestAC4PostReceiveAttribution(t *testing.T) {
318	318		h := newHarness(t, harnessOpts{})
319		-	events := h.forge.Events.Subscribe(h.ctx)
	319	+	events := h.forge.Bus.Subscribe(h.ctx)
320	320
321	321		// Two repositories, pushed at the same time with different refs.
322	322		const n = 4


@@ -210,7 +210,7 @@ func TestAC2PushAuthorization(t *testing.T) {
210		// AC3: first push creates a private repository with HEAD on the pushed branch.	210		// AC3: first push creates a private repository with HEAD on the pushed branch.
211		func TestAC3CreateOnPush(t *testing.T) {	211		func TestAC3CreateOnPush(t *testing.T) {
212		h := newHarness(t, harnessOpts{})	212		h := newHarness(t, harnessOpts{})
213	-	events := h.forge.Events.Subscribe(h.ctx)	213	+	events := h.forge.Bus.Subscribe(h.ctx)
214			214
215		// Only "dev" is pushed: HEAD (initially main) moves to it.	215		// Only "dev" is pushed: HEAD (initially main) moves to it.
216		wt := h.workTree("dev", "a")	216		wt := h.workTree("dev", "a")

@@ -316,7 +316,7 @@ func TestAC4HooksUnknownTokenAndMissingBinary(t *testing.T) {
316			316
317		func TestAC4PostReceiveAttribution(t *testing.T) {	317		func TestAC4PostReceiveAttribution(t *testing.T) {
318		h := newHarness(t, harnessOpts{})	318		h := newHarness(t, harnessOpts{})
319	-	events := h.forge.Events.Subscribe(h.ctx)	319	+	events := h.forge.Bus.Subscribe(h.ctx)
320			320
321		// Two repositories, pushed at the same time with different refs.	321		// Two repositories, pushed at the same time with different refs.
322		const n = 4	322		const n = 4

M cmd/ermite/events_test.go +57 -0


@@ -18,6 +18,8 @@ import (
18	18		"google.golang.org/protobuf/types/known/fieldmaskpb"
19	19
20	20		pb "bnema.dev/ermite-api/gen/ermite"
	21	+	"bnema.dev/ermite/internal/adapters/rpc"
	22	+	"bnema.dev/ermite/internal/app"
21	23		"bnema.dev/ermite/internal/domain"
22	24		)
23	25

@@ -317,3 +319,58 @@ func TestServeShutdownEndsWatchStreams(t *testing.T) {
317	319		assert.NotContains(t, logs.String(), `"level":"error"`, "no error logged:\n%s", logs.String())
318	320		assert.False(t, errors.Is(err, context.Canceled))
319	321		}
	322	+
	323	+	// A token that is revoked while its Watch is open ends the stream with
	324	+	// unauthenticated, and the private events that follow never reach it.
	325	+	func TestWatchEndsWhenTheTokenIsRevoked(t *testing.T) {
	326	+	h := newHarness(t, harnessOpts{revalidateEvery: 20 * time.Millisecond})
	327	+	h.seed("priv", domain.VisibilityPrivate)
	328	+	created, err := h.forge.Tokens.Create(h.ctx, domain.Root(), app.CreateTokenParams{
	329	+	Label: "watcher", Scopes: scopes(domain.ScopeRepoRead),
	330	+	})
	331	+	require.NoError(t, err)
	332	+	w := h.watch(h.api(created.Secret), "")
	333	+	wt := h.workTree("main", "a.txt")
	334	+	h.mustGit(wt, "push", "-q", h.urlAuth(adminToken, "/priv.git"), "main")
	335	+	assert.Equal(t, "priv", w.next(t, 5*time.Second).GetPushed().GetRepo(), "the token may read priv")
	336	+
	337	+	require.NoError(t, h.forge.Tokens.Revoke(h.ctx, domain.Root(), created.Token.ID))
	338	+	h.writeFile(wt, "b.txt", []byte("b\n"))
	339	+	h.mustGit(wt, "push", "-q", h.urlAuth(adminToken, "/priv.git"), "main")
	340	+
	341	+	select {
	342	+	case <-w.done:
	343	+	case <-time.After(5 * time.Second):
	344	+	t.Fatal("the stream of a revoked token did not end")
	345	+	}
	346	+	assert.Equal(t, "unauthenticated", codeName(w.err))
	347	+	for ev := range w.events {
	348	+	t.Fatalf("an event reached the revoked token: %v", ev)
	349	+	}
	350	+	}
	351	+
	352	+	// Watch refuses too many topics, and too many streams from one client.
	353	+	func TestWatchLimits(t *testing.T) {
	354	+	h := newHarness(t, harnessOpts{})
	355	+	api := h.api(adminToken)
	356	+
	357	+	topics := make([]string, app.MaxWatchTopics+1)
	358	+	for i := range topics {
	359	+	topics[i] = "home"
	360	+	}
	361	+	assert.Equal(t, "invalid_argument", codeName(watchErr(h.ctx, api, topics...)))
	362	+
	363	+	for range app.MaxWatchersPerClient {
	364	+	h.watch(api, "")
	365	+	}
	366	+	require.Eventually(t, func() bool { return h.forge.EventLog.Watchers() == app.MaxWatchersPerClient }, 5*time.Second, time.Millisecond)
	367	+	assert.Equal(t, "resource_exhausted", codeName(watchErr(h.ctx, api)))
	368	+	// A stream of another client address is not affected (the harness trusts
	369	+	// the loopback proxy only when configured: use the in-process caller).
	370	+	inProc := newClientsOn(h.forge.API.InProcessClient(), rpc.InProcessURL)
	371	+	ctx, cancel := context.WithCancel(domain.ContextWithPrincipal(h.ctx, domain.Root()))
	372	+	defer cancel()
	373	+	st, err := inProc.event.Watch(ctx, connect.NewRequest(&pb.WatchRequest{}))
	374	+	require.NoError(t, err)
	375	+	_ = st.Close()
	376	+	}


@@ -18,6 +18,8 @@ import (
18	"google.golang.org/protobuf/types/known/fieldmaskpb"	18		"google.golang.org/protobuf/types/known/fieldmaskpb"
19		19
20	pb "bnema.dev/ermite-api/gen/ermite"	20		pb "bnema.dev/ermite-api/gen/ermite"
		21	+	"bnema.dev/ermite/internal/adapters/rpc"
		22	+	"bnema.dev/ermite/internal/app"
21	"bnema.dev/ermite/internal/domain"	23		"bnema.dev/ermite/internal/domain"
22	)	24		)
23		25

@@ -317,3 +319,58 @@ func TestServeShutdownEndsWatchStreams(t *testing.T) {
317	assert.NotContains(t, logs.String(), `"level":"error"`, "no error logged:\n%s", logs.String())	319		assert.NotContains(t, logs.String(), `"level":"error"`, "no error logged:\n%s", logs.String())
318	assert.False(t, errors.Is(err, context.Canceled))	320		assert.False(t, errors.Is(err, context.Canceled))
319	}	321		}
		322	+
		323	+	// A token that is revoked while its Watch is open ends the stream with
		324	+	// unauthenticated, and the private events that follow never reach it.
		325	+	func TestWatchEndsWhenTheTokenIsRevoked(t *testing.T) {
		326	+	h := newHarness(t, harnessOpts{revalidateEvery: 20 * time.Millisecond})
		327	+	h.seed("priv", domain.VisibilityPrivate)
		328	+	created, err := h.forge.Tokens.Create(h.ctx, domain.Root(), app.CreateTokenParams{
		329	+	Label: "watcher", Scopes: scopes(domain.ScopeRepoRead),
		330	+	})
		331	+	require.NoError(t, err)
		332	+	w := h.watch(h.api(created.Secret), "")
		333	+	wt := h.workTree("main", "a.txt")
		334	+	h.mustGit(wt, "push", "-q", h.urlAuth(adminToken, "/priv.git"), "main")
		335	+	assert.Equal(t, "priv", w.next(t, 5*time.Second).GetPushed().GetRepo(), "the token may read priv")
		336	+
		337	+	require.NoError(t, h.forge.Tokens.Revoke(h.ctx, domain.Root(), created.Token.ID))
		338	+	h.writeFile(wt, "b.txt", []byte("b\n"))
		339	+	h.mustGit(wt, "push", "-q", h.urlAuth(adminToken, "/priv.git"), "main")
		340	+
		341	+	select {
		342	+	case <-w.done:
		343	+	case <-time.After(5 * time.Second):
		344	+	t.Fatal("the stream of a revoked token did not end")
		345	+	}
		346	+	assert.Equal(t, "unauthenticated", codeName(w.err))
		347	+	for ev := range w.events {
		348	+	t.Fatalf("an event reached the revoked token: %v", ev)
		349	+	}
		350	+	}
		351	+
		352	+	// Watch refuses too many topics, and too many streams from one client.
		353	+	func TestWatchLimits(t *testing.T) {
		354	+	h := newHarness(t, harnessOpts{})
		355	+	api := h.api(adminToken)
		356	+
		357	+	topics := make([]string, app.MaxWatchTopics+1)
		358	+	for i := range topics {
		359	+	topics[i] = "home"
		360	+	}
		361	+	assert.Equal(t, "invalid_argument", codeName(watchErr(h.ctx, api, topics...)))
		362	+
		363	+	for range app.MaxWatchersPerClient {
		364	+	h.watch(api, "")
		365	+	}
		366	+	require.Eventually(t, func() bool { return h.forge.EventLog.Watchers() == app.MaxWatchersPerClient }, 5*time.Second, time.Millisecond)
		367	+	assert.Equal(t, "resource_exhausted", codeName(watchErr(h.ctx, api)))
		368	+	// A stream of another client address is not affected (the harness trusts
		369	+	// the loopback proxy only when configured: use the in-process caller).
		370	+	inProc := newClientsOn(h.forge.API.InProcessClient(), rpc.InProcessURL)
		371	+	ctx, cancel := context.WithCancel(domain.ContextWithPrincipal(h.ctx, domain.Root()))
		372	+	defer cancel()
		373	+	st, err := inProc.event.Watch(ctx, connect.NewRequest(&pb.WatchRequest{}))
		374	+	require.NoError(t, err)
		375	+	_ = st.Close()
		376	+	}

M cmd/ermite/forge.go +9 -5


@@ -41,6 +41,8 @@ type options struct {
41	41		maxBodyBytes int64
42	42		// eventBuffer overrides app.DefaultEventBuffer when positive.
43	43		eventBuffer int
	44	+	// revalidateEvery overrides app.DefaultRevalidateEvery when not zero.
	45	+	revalidateEvery time.Duration
44	46		}
45	47
46	48		// forge is the wired server: every adapter and use case, ready to serve.

@@ -61,7 +63,7 @@ type forge struct {
61	63		Browse *app.Browse
62	64		EventLog *app.Events
63	65		API *rpc.API
64		-	Events ports.EventBus
	66	+	Bus ports.EventBus
65	67		}
66	68
67	69		// serverVersion is the version GetVersion reports: the module version of the

@@ -116,8 +118,8 @@ func newForge(ctx context.Context, cfg config.Config, opts options) (_ *forge, e
116	118		}
117	119
118	120		clock := systemClock{}
119		-	f.Events = eventbus.New()
120		-	f.Repos = app.NewRepos(store, git, f.Events, clock)
	121	+	f.Bus = eventbus.New()
	122	+	f.Repos = app.NewRepos(store, git, f.Bus, clock)
121	123		report, err := f.Repos.Reconcile(ctx)
122	124		if err != nil {
123	125		return nil, fmt.Errorf("reconcile repositories: %w", err)

@@ -131,7 +133,7 @@ func newForge(ctx context.Context, cfg config.Config, opts options) (_ *forge, e
131	133		f.Tokens = app.NewTokens(store.Tokens(), clock)
132	134		f.Auth = app.NewAuth(cfg.AdminTokenHash, store.Tokens(), clock)
133	135		f.maint = app.NewMaintainer(ctx, git, opts.maintenanceDelay)
134		-	f.Pushes = app.NewPushes(store, f.Repos, git, f.Events, clock, f.maint, f.Auth)
	136	+	f.Pushes = app.NewPushes(store, f.Repos, git, f.Bus, clock, f.maint, f.Auth)
135	137
136	138		exe := opts.hookExe
137	139		if exe == "" {

@@ -148,7 +150,9 @@ func newForge(ctx context.Context, cfg config.Config, opts options) (_ *forge, e
148	150		}
149	151
150	152		f.Browse = app.NewBrowse(store, git)
151		-	f.EventLog = app.NewEvents(ctx, f.Repos, f.Events, opts.eventBuffer, clock)
	153	+	f.EventLog = app.NewEvents(ctx, f.Repos, f.Bus, clock, app.EventsConfig{
	154	+	Size: opts.eventBuffer, Auth: f.Auth, RevalidateEvery: opts.revalidateEvery,
	155	+	})
152	156
153	157		// One budget of failed authentications per client, for Git and the API.
154	158		limiter := clientlimit.NewLimiter(nil)

 	maxBodyBytes int64
 	// eventBuffer overrides app.DefaultEventBuffer when positive.
 	eventBuffer int
+	// revalidateEvery overrides app.DefaultRevalidateEvery when not zero.
+	revalidateEvery time.Duration
 }
 // forge is the wired server: every adapter and use case, ready to serve.
 	Browse   *app.Browse
 	EventLog *app.Events
 	API      *rpc.API
-	Events   ports.EventBus
+	Bus      ports.EventBus
 }
 // serverVersion is the version GetVersion reports: the module version of the
 	}
 	clock := systemClock{}
-	f.Events = eventbus.New()
+	f.Bus = eventbus.New()
-	f.Repos = app.NewRepos(store, git, f.Events, clock)
+	f.Repos = app.NewRepos(store, git, f.Bus, clock)
 	report, err := f.Repos.Reconcile(ctx)
 	if err != nil {
 		return nil, fmt.Errorf("reconcile repositories: %w", err)
 	f.Tokens = app.NewTokens(store.Tokens(), clock)
 	f.Auth = app.NewAuth(cfg.AdminTokenHash, store.Tokens(), clock)
 	f.maint = app.NewMaintainer(ctx, git, opts.maintenanceDelay)
-	f.Pushes = app.NewPushes(store, f.Repos, git, f.Events, clock, f.maint, f.Auth)
+	f.Pushes = app.NewPushes(store, f.Repos, git, f.Bus, clock, f.maint, f.Auth)
 	exe := opts.hookExe
 	if exe == "" {
 	}
 	f.Browse = app.NewBrowse(store, git)
-	f.EventLog = app.NewEvents(ctx, f.Repos, f.Events, opts.eventBuffer, clock)
+	f.EventLog = app.NewEvents(ctx, f.Repos, f.Bus, clock, app.EventsConfig{
+		Size: opts.eventBuffer, Auth: f.Auth, RevalidateEvery: opts.revalidateEvery,
+	})
 	// One budget of failed authentications per client, for Git and the API.
 	limiter := clientlimit.NewLimiter(nil)

M internal/adapters/rpc/errors.go +2 -0


@@ -50,6 +50,8 @@ func codeOf(err error) (connect.Code, bool) {
50	50		return connect.CodeFailedPrecondition, true
51	51		case errors.Is(err, domain.ErrConflict):
52	52		return connect.CodeAborted, true
	53	+	case errors.Is(err, domain.ErrResourceExhausted):
	54	+	return connect.CodeResourceExhausted, true
53	55		}
54	56		for _, invalid := range invalidErrors {
55	57		if errors.Is(err, invalid) {


@@ -50,6 +50,8 @@ func codeOf(err error) (connect.Code, bool) {
50	return connect.CodeFailedPrecondition, true	50		return connect.CodeFailedPrecondition, true
51	case errors.Is(err, domain.ErrConflict):	51		case errors.Is(err, domain.ErrConflict):
52	return connect.CodeAborted, true	52		return connect.CodeAborted, true
		53	+	case errors.Is(err, domain.ErrResourceExhausted):
		54	+	return connect.CodeResourceExhausted, true
53	}	55		}
54	for _, invalid := range invalidErrors {	56		for _, invalid := range invalidErrors {
55	if errors.Is(err, invalid) {	57		if errors.Is(err, invalid) {

M internal/adapters/rpc/event.go +4 -1


@@ -16,10 +16,13 @@ type eventService struct{ events *app.Events }
16	16
17	17		// Watch streams events until the client goes away or the server stops. The
18	18		// stream ends with success in both cases: the contract has no error for
19		-	// "server is shutting down".
	19	+	// "server is shutting down". It ends with unauthenticated when the token of
	20	+	// the caller is revoked or expires, and a caller that has too many streams
	21	+	// open gets resource_exhausted.
20	22		func (s eventService) Watch(ctx context.Context, req connect.Request[pb.WatchRequest], stream *connect.ServerStream[pb.WatchResponse]) error {
21	23		params := app.WatchParams{
22	24		Topics: req.Msg.GetTopics(), ResumeAfter: req.Msg.GetResumeAfter(),
	25	+	Client: clientFromContext(ctx),
23	26		// A client's call returns when the response begins: send the headers
24	27		// now, not with the first event, which may be far away. Connect flushes
25	28		// them for a nil message. A failure shows on the first real send.

 // Watch streams events until the client goes away or the server stops. The
 // stream ends with success in both cases: the contract has no error for
-// "server is shutting down".
+// "server is shutting down". It ends with unauthenticated when the token of
+// the caller is revoked or expires, and a caller that has too many streams
+// open gets resource_exhausted.
 func (s *eventService) Watch(ctx context.Context, req *connect.Request[pb.WatchRequest], stream *connect.ServerStream[pb.WatchResponse]) error {
 	params := app.WatchParams{
 		Topics: req.Msg.GetTopics(), ResumeAfter: req.Msg.GetResumeAfter(),
+		Client: clientFromContext(ctx),
 		// A client's call returns when the response begins: send the headers
 		// now, not with the first event, which may be far away. Connect flushes
 		// them for a nil message. A failure shows on the first real send.

M internal/app/events.go +206 -25


@@ -9,6 +9,7 @@ import (
9	9		"strconv"
10	10		"strings"
11	11		"sync"
	12	+	"time"
12	13
13	14		"github.com/bnema/zerowrap"
14	15

@@ -19,6 +20,20 @@ import (
19	20		// DefaultEventBuffer is the number of events the Events use case remembers.
20	21		const DefaultEventBuffer = 1024
21	22
	23	+	// Limits of the watchers.
	24	+	const (
	25	+	// MaxWatchTopics is the most topics one watch may follow.
	26	+	MaxWatchTopics = 64
	27	+	// MaxWatchers is the most watches in progress at once.
	28	+	MaxWatchers = 256
	29	+	// MaxWatchersPerClient is the most watches in progress for one client
	30	+	// address (see WatchParams.Client).
	31	+	MaxWatchersPerClient = 16
	32	+	// DefaultRevalidateEvery is how long a watcher with a stored token goes
	33	+	// without asking the token store whether the token still exists.
	34	+	DefaultRevalidateEvery = 5 * time.Second
	35	+	)
	36	+
22	37		// Topics a watcher may follow.
23	38		const (
24	39		// TopicHome follows every repository the caller may see.

@@ -27,10 +42,23 @@ const (
27	42		topicRepoPrefix = "repo:"
28	43		)
29	44
30		-	// bufferedEvent is an event with its sequence number in the buffer.
	45	+	// bufferedEvent is an event with its sequence number in the buffer, and what
	46	+	// was true of its repository when it was buffered. Access to an event is
	47	+	// decided from that, and from the repository as it is when the event is
	48	+	// delivered: a repository made public later does not reveal what happened
	49	+	// while it was private, and a repository deleted and created again under the
	50	+	// same name does not inherit the events of the old one.
31	51		type bufferedEvent struct {
32	52		seq uint64
33	53		ev domain.Event
	54	+
	55	+	// known is false when the repository did not exist any more when the
	56	+	// event was buffered: nobody is told about it.
	57	+	known bool
	58	+	// vis is the visibility of the repository at that time.
	59	+	vis domain.Visibility
	60	+	// created is the creation time of that repository.
	61	+	created time.Time
34	62		}
35	63
36	64		// Events keeps the last events of the bus in a ring buffer, numbers them, and

@@ -46,8 +74,11 @@ type bufferedEvent struct {
46	74		// NewEvents and lives until Close.
47	75		type Events struct {
48	76		repos *Repos
	77	+	auth Revalidator
	78	+	clock ports.Clock
49	79		boot string
50	80		size int
	81	+	every time.Duration // see EventsConfig.RevalidateEvery
51	82
52	83		stop context.CancelFunc
53	84		done chan struct{}

@@ -58,18 +89,39 @@ type Events struct {
58	89		changed chan struct{} // closed and replaced on every append and on close
59	90		closed bool
60	91		watches int
	92	+	clients map[string]int // watches in progress by client
61	93		}
62	94
63		-	// NewEvents subscribes to the bus and starts remembering its events, at most
64		-	// size of them (DefaultEventBuffer when not positive). ctx carries the
65		-	// logger; its cancellation is not followed, call Close.
66		-	func NewEvents(ctx context.Context, repos Repos, bus ports.EventBus, size int, clock ports.Clock) Events {
	95	+	// EventsConfig configures Events.
	96	+	type EventsConfig struct {
	97	+	// Size is the number of events remembered; DefaultEventBuffer when not
	98	+	// positive.
	99	+	Size int
	100	+	// Auth confirms that the token of a watcher still exists. Required.
	101	+	Auth Revalidator
	102	+	// RevalidateEvery is how long the answer of Auth is kept for a watcher;
	103	+	// DefaultRevalidateEvery when zero, and negative to ask for every event.
	104	+	RevalidateEvery time.Duration
	105	+	}
	106	+
	107	+	// NewEvents subscribes to the bus and starts remembering its events. ctx
	108	+	// carries the logger; its cancellation is not followed, call Close.
	109	+	func NewEvents(ctx context.Context, repos Repos, bus ports.EventBus, clock ports.Clock, cfg EventsConfig) Events {
	110	+	size := cfg.Size
67	111		if size <= 0 {
68	112		size = DefaultEventBuffer
69	113		}
	114	+	every := cfg.RevalidateEvery
	115	+	if every == 0 {
	116	+	every = DefaultRevalidateEvery
	117	+	}
70	118		run, stop := context.WithCancel(context.WithoutCancel(ctx))
71	119		e := &Events{
72	120		repos: repos,
	121	+	auth: cfg.Auth,
	122	+	clock: clock,
	123	+	every: every,
	124	+	clients: make(map[string]int),
73	125		boot: strconv.FormatInt(clock.Now().UnixMilli(), 36),
74	126		size: size,
75	127		stop: stop,

@@ -92,21 +144,48 @@ func (e *Events) feed(ctx context.Context, sub <-chan domain.Event) {
92	144		if !ok {
93	145		return
94	146		}
95		-	e.append(ev)
	147	+	e.append(ctx, ev)
96	148		case <-ctx.Done():
97	149		return
98	150		}
99	151		}
100	152		}
101	153
102		-	func (e *Events) append(ev domain.Event) {
	154	+	// snapshot records the visibility and the generation of the repository an
	155	+	// event is about. The events of the repository lifecycle carry them; for a
	156	+	// push the repository is read now.
	157	+	func (e *Events) snapshot(ctx context.Context, ev domain.Event) bufferedEvent {
	158	+	switch ev := ev.(type) {
	159	+	case domain.RepoCreated:
	160	+	return bufferedEvent{known: true, vis: ev.Visibility, created: ev.At}
	161	+	case domain.RepoUpdated:
	162	+	return bufferedEvent{known: true, vis: ev.Visibility, created: ev.CreatedAt}
	163	+	case domain.RepoDeleted:
	164	+	return bufferedEvent{known: true, vis: ev.Visibility, created: ev.CreatedAt}
	165	+	case domain.PushEvent:
	166	+	repo, err := lookupRepo(ctx, e.repos.meta, ev.Repo)
	167	+	if err != nil {
	168	+	if !errors.Is(err, domain.ErrNotFound) {
	169	+	log := zerowrap.FromCtx(ctx)
	170	+	log.Error().Err(err).Str("repo", ev.Repo.String()).Msg("read repository of a buffered event")
	171	+	}
	172	+	return bufferedEvent{} // gone, or unknown: nobody is told
	173	+	}
	174	+	return bufferedEvent{known: true, vis: repo.Visibility, created: repo.CreatedAt}
	175	+	}
	176	+	return bufferedEvent{known: true}
	177	+	}
	178	+
	179	+	func (e *Events) append(ctx context.Context, ev domain.Event) {
	180	+	be := e.snapshot(ctx, ev) // outside the lock: it reads the store
103	181		e.mu.Lock()
104	182		defer e.mu.Unlock()
105	183		if len(e.ring) == e.size {
106	184		copy(e.ring, e.ring[1:])
107	185		e.ring = e.ring[:e.size-1]
108	186		}
109		-	e.ring = append(e.ring, bufferedEvent{seq: e.nextSeq, ev: ev})
	187	+	be.seq, be.ev = e.nextSeq, ev
	188	+	e.ring = append(e.ring, be)
110	189		e.nextSeq++
111	190		e.wake()
112	191		}

@@ -126,11 +205,14 @@ func (e *Events) shutdown() {
126	205		}
127	206		}
128	207
129		-	// Close ends every watch cleanly, stops feeding and waits for the feeder.
130		-	// It is safe to call more than once.
131		-	func (e *Events) Close(context.Context) {
	208	+	// Close ends every watch cleanly, stops feeding and waits for the feeder,
	209	+	// until ctx ends. It is safe to call more than once.
	210	+	func (e *Events) Close(ctx context.Context) {
132	211		e.stop()
133		-	<-e.done
	212	+	select {
	213	+	case <-e.done:
	214	+	case <-ctx.Done():
	215	+	}
134	216		}
135	217
136	218		// Watchers returns the number of watches in progress.

@@ -159,6 +241,10 @@ type WatchParams struct {
159	241		// ResumeAfter is the id of the last event the caller saw; empty to
160	242		// receive only new events.
161	243		ResumeAfter string
	244	+	// Client identifies where the watch comes from, for the limit per client
	245	+	// (MaxWatchersPerClient): an opaque key such as a normalized address. Empty
	246	+	// for callers in this process, which only count against MaxWatchers.
	247	+	Client string
162	248		// Ready, when not nil, is called once the topics are valid and the watch
163	249		// is registered: events published from then on are not missed. Nothing is
164	250		// sent before it returns.

@@ -184,15 +270,20 @@ type watch struct {
184	270		p domain.Principal
185	271		home bool
186	272		repos map[domain.RepoName]bool
	273	+
	274	+	checked time.Time // when the token store last confirmed the principal
187	275		}
188	276
189	277		func (w *watch) wants(name domain.RepoName) bool { return w.home \|\| w.repos[name] }
190	278
191	279		// parseTopics validates the topics of a watch. A repository topic needs a
192	280		// repository the principal may read (else domain.ErrNotFound); an unknown
193		-	// topic is domain.ErrInvalid.
	281	+	// topic, or more than MaxWatchTopics, is domain.ErrInvalid.
194	282		func (e Events) parseTopics(ctx context.Context, p domain.Principal, topics []string) (watch, error) {
195		-	w := &watch{p: p, repos: make(map[domain.RepoName]bool)}
	283	+	if len(topics) > MaxWatchTopics {
	284	+	return nil, fmt.Errorf("%w: %d topics, at most %d", domain.ErrInvalid, len(topics), MaxWatchTopics)
	285	+	}
	286	+	w := &watch{p: p, repos: make(map[domain.RepoName]bool), checked: time.Now()}
196	287		if len(topics) == 0 {
197	288		w.home = true
198	289		}

@@ -220,6 +311,12 @@ func (e *Events) parseTopics(ctx context.Context, p domain.Principal, topics []s
220	311		// Watch calls send for the events the principal may see until ctx ends, the
221	312		// buffer closes (both return nil) or send fails (its error is returned).
222	313		//
	314	+	// A principal with a stored token is checked again for every event, and every
	315	+	// few seconds when idle: when the token expired or was revoked the watch ends
	316	+	// with domain.ErrUnauthenticated and nothing more is sent. More watches than
	317	+	// MaxWatchers, or than MaxWatchersPerClient for one Client, give
	318	+	// domain.ErrResourceExhausted.
	319	+	//
223	320		// Without ResumeAfter only new events are sent. With it, the buffered events
224	321		// after that id are sent first, then new ones. When the id is no longer
225	322		// buffered, or was not issued by this run, a gap comes first and the oldest

@@ -244,11 +341,23 @@ func (e *Events) Watch(ctx context.Context, p domain.Principal, params WatchPara
244	341		next, gap = oldest, true
245	342		}
246	343		}
	344	+	if e.watches >= MaxWatchers \|\| (params.Client != "" && e.clients[params.Client] >= MaxWatchersPerClient) {
	345	+	e.mu.Unlock()
	346	+	return fmt.Errorf("%w: too many watchers", domain.ErrResourceExhausted)
	347	+	}
247	348		e.watches++
	349	+	if params.Client != "" {
	350	+	e.clients[params.Client]++
	351	+	}
248	352		e.mu.Unlock()
249	353		defer func() {
250	354		e.mu.Lock()
251	355		e.watches--
	356	+	if params.Client != "" {
	357	+	if e.clients[params.Client]--; e.clients[params.Client] == 0 {
	358	+	delete(e.clients, params.Client)
	359	+	}
	360	+	}
252	361		e.mu.Unlock()
253	362		}()
254	363

@@ -263,6 +372,9 @@ func (e *Events) Watch(ctx context.Context, p domain.Principal, params WatchPara
263	372		}
264	373		}
265	374		for {
	375	+	if err := e.stillAllowed(ctx, w); err != nil {
	376	+	return err
	377	+	}
266	378		batch, changed, oldest, lost, closed := e.since(next)
267	379		if lost {
268	380		if err := send(WatchEvent{Gap: true}); err != nil {

@@ -283,12 +395,60 @@ func (e *Events) Watch(ctx context.Context, p domain.Principal, params WatchPara
283	395		if closed {
284	396		return nil
285	397		}
286		-	select {
287		-	case <-changed:
288		-	case <-ctx.Done():
	398	+	e.wait(ctx, w, changed)
	399	+	if ctx.Err() != nil {
	400	+	return nil
	401	+	}
	402	+	}
	403	+	}
	404	+
	405	+	// wait blocks until more events arrive or ctx ends. A watcher with a stored
	406	+	// token also wakes up every few seconds, so that a revocation ends the watch
	407	+	// of a quiet forge too (the caller checks the token on the next turn).
	408	+	func (e Events) wait(ctx context.Context, w watch, changed <-chan struct{}) {
	409	+	var idle <-chan time.Time
	410	+	if w.p.TokenID() != "" {
	411	+	t := time.NewTimer(e.idleCheck())
	412	+	defer t.Stop()
	413	+	idle = t.C
	414	+	}
	415	+	select {
	416	+	case <-changed:
	417	+	case <-idle:
	418	+	case <-ctx.Done():
	419	+	}
	420	+	}
	421	+
	422	+	// idleCheck is how long a watcher with a stored token waits for news before
	423	+	// checking its token anyway.
	424	+	func (e *Events) idleCheck() time.Duration {
	425	+	if e.every <= 0 {
	426	+	return DefaultRevalidateEvery
	427	+	}
	428	+	return min(e.every, DefaultRevalidateEvery)
	429	+	}
	430	+
	431	+	// stillAllowed checks that the holder of a stored token may still watch: the
	432	+	// token has not expired and, at most every few seconds, still exists. Root and
	433	+	// anonymous callers have nothing to lose.
	434	+	func (e Events) stillAllowed(ctx context.Context, w watch) error {
	435	+	if w.p.TokenID() == "" {
	436	+	return nil
	437	+	}
	438	+	if exp := w.p.ExpiresAt(); !exp.IsZero() && !e.clock.Now().Before(exp) {
	439	+	return domain.ErrUnauthenticated
	440	+	}
	441	+	if e.every > 0 && time.Since(w.checked) < e.every {
	442	+	return nil
	443	+	}
	444	+	if err := e.auth.Revalidate(ctx, w.p); err != nil {
	445	+	if ctx.Err() != nil {
289	446		return nil
290	447		}
	448	+	return err
291	449		}
	450	+	w.checked = time.Now()
	451	+	return nil
292	452		}
293	453
294	454		// since returns the buffered events from sequence next on, and the channel

@@ -308,6 +468,9 @@ func (e *Events) since(next uint64) (batch []bufferedEvent, changed <-chan struc
308	468
309	469		// deliver sends the event if the watcher follows its topic and may see it.
310	470		func (e Events) deliver(ctx context.Context, w watch, be bufferedEvent, send func(WatchEvent) error) error {
	471	+	if err := e.stillAllowed(ctx, w); err != nil {
	472	+	return err
	473	+	}
311	474		out := WatchEvent{ID: e.id(be.seq), Event: be.ev}
312	475		switch ev := be.ev.(type) {
313	476		case domain.EventGap:

@@ -320,23 +483,28 @@ func (e Events) deliver(ctx context.Context, w watch, be bufferedEvent, send f
320	483		}
321	484		return send(out)
322	485		case domain.RepoCreated:
323		-	return e.deliverRepo(ctx, w, ev.Repo, true, out, send)
	486	+	return e.deliverRepo(ctx, w, ev.Repo, true, be, out, send)
324	487		case domain.RepoUpdated:
325		-	return e.deliverRepo(ctx, w, ev.Repo, true, out, send)
	488	+	return e.deliverRepo(ctx, w, ev.Repo, true, be, out, send)
326	489		case domain.PushEvent:
327		-	return e.deliverRepo(ctx, w, ev.Repo, false, out, send)
	490	+	return e.deliverRepo(ctx, w, ev.Repo, false, be, out, send)
328	491		}
329	492		return nil // an event this version does not stream
330	493		}
331	494
332		-	// deliverRepo re-checks the access to the repository, with its present
333		-	// visibility, and sends the event. A repository that no longer exists is
	495	+	// deliverRepo sends an event about a repository if the watcher could read the
	496	+	// repository when the event happened and can read it now, and the repository
	497	+	// is still the one the event is about. A repository that no longer exists is
334	498		// skipped: its RepoDeleted follows or came before.
335		-	func (e Events) deliverRepo(ctx context.Context, w watch, name domain.RepoName, withRepo bool, out WatchEvent, send func(WatchEvent) error) error {
336		-	if !w.wants(name) {
	499	+	//
	500	+	// The exception is a repository that became private: the watchers who could
	501	+	// read it before are told once, with its name and its new visibility only,
	502	+	// so that what they show can be removed.
	503	+	func (e Events) deliverRepo(ctx context.Context, w watch, name domain.RepoName, withRepo bool, be bufferedEvent, out WatchEvent, send func(WatchEvent) error) error {
	504	+	if !w.wants(name) \|\| !be.known {
337	505		return nil
338	506		}
339		-	repo, err := e.repos.Get(ctx, w.p, name)
	507	+	repo, err := lookupRepo(ctx, e.repos.meta, name)
340	508		switch {
341	509		case errors.Is(err, domain.ErrNotFound):
342	510		return nil

@@ -345,6 +513,19 @@ func (e Events) deliverRepo(ctx context.Context, w watch, name domain.RepoName
345	513		case err != nil:
346	514		return fmt.Errorf("check access to %q: %w", name, err)
347	515		}
	516	+	if repo.CreatedAt.After(be.created) {
	517	+	return nil // the event is about a repository of this name that was deleted
	518	+	}
	519	+	if !canReadAs(w.p, name, be.vis) {
	520	+	if up, ok := be.ev.(domain.RepoUpdated); ok && canReadAs(w.p, name, up.Previous) {
	521	+	out.Repo = &RepoView{Repo: domain.Repo{Name: name, Visibility: up.Visibility}}
	522	+	return send(out)
	523	+	}
	524	+	return nil
	525	+	}
	526	+	if !canRead(w.p, repo) {
	527	+	return nil
	528	+	}
348	529		if withRepo {
349	530		view, err := e.repos.View(ctx, repo)
350	531		if err != nil {


@@ -9,6 +9,7 @@ import (
9		"strconv"	9		"strconv"
10		"strings"	10		"strings"
11		"sync"	11		"sync"
			12	+	"time"
12			13
13		"github.com/bnema/zerowrap"	14		"github.com/bnema/zerowrap"
14			15

@@ -19,6 +20,20 @@ import (
19		// DefaultEventBuffer is the number of events the Events use case remembers.	20		// DefaultEventBuffer is the number of events the Events use case remembers.
20		const DefaultEventBuffer = 1024	21		const DefaultEventBuffer = 1024
21			22
			23	+	// Limits of the watchers.
			24	+	const (
			25	+	// MaxWatchTopics is the most topics one watch may follow.
			26	+	MaxWatchTopics = 64
			27	+	// MaxWatchers is the most watches in progress at once.
			28	+	MaxWatchers = 256
			29	+	// MaxWatchersPerClient is the most watches in progress for one client
			30	+	// address (see WatchParams.Client).
			31	+	MaxWatchersPerClient = 16
			32	+	// DefaultRevalidateEvery is how long a watcher with a stored token goes
			33	+	// without asking the token store whether the token still exists.
			34	+	DefaultRevalidateEvery = 5 * time.Second
			35	+	)
			36	+
22		// Topics a watcher may follow.	37		// Topics a watcher may follow.
23		const (	38		const (
24		// TopicHome follows every repository the caller may see.	39		// TopicHome follows every repository the caller may see.

@@ -27,10 +42,23 @@ const (
27		topicRepoPrefix = "repo:"	42		topicRepoPrefix = "repo:"
28		)	43		)
29			44
30	-	// bufferedEvent is an event with its sequence number in the buffer.	45	+	// bufferedEvent is an event with its sequence number in the buffer, and what
			46	+	// was true of its repository when it was buffered. Access to an event is
			47	+	// decided from that, and from the repository as it is when the event is
			48	+	// delivered: a repository made public later does not reveal what happened
			49	+	// while it was private, and a repository deleted and created again under the
			50	+	// same name does not inherit the events of the old one.
31		type bufferedEvent struct {	51		type bufferedEvent struct {
32		seq uint64	52		seq uint64
33		ev domain.Event	53		ev domain.Event
			54	+
			55	+	// known is false when the repository did not exist any more when the
			56	+	// event was buffered: nobody is told about it.
			57	+	known bool
			58	+	// vis is the visibility of the repository at that time.
			59	+	vis domain.Visibility
			60	+	// created is the creation time of that repository.
			61	+	created time.Time
34		}	62		}
35			63
36		// Events keeps the last events of the bus in a ring buffer, numbers them, and	64		// Events keeps the last events of the bus in a ring buffer, numbers them, and

@@ -46,8 +74,11 @@ type bufferedEvent struct {
46		// NewEvents and lives until Close.	74		// NewEvents and lives until Close.
47		type Events struct {	75		type Events struct {
48		repos *Repos	76		repos *Repos
			77	+	auth Revalidator
			78	+	clock ports.Clock
49		boot string	79		boot string
50		size int	80		size int
			81	+	every time.Duration // see EventsConfig.RevalidateEvery
51			82
52		stop context.CancelFunc	83		stop context.CancelFunc
53		done chan struct{}	84		done chan struct{}

@@ -58,18 +89,39 @@ type Events struct {
58		changed chan struct{} // closed and replaced on every append and on close	89		changed chan struct{} // closed and replaced on every append and on close
59		closed bool	90		closed bool
60		watches int	91		watches int
			92	+	clients map[string]int // watches in progress by client
61		}	93		}
62			94
63	-	// NewEvents subscribes to the bus and starts remembering its events, at most	95	+	// EventsConfig configures Events.
64	-	// size of them (DefaultEventBuffer when not positive). ctx carries the	96	+	type EventsConfig struct {
65	-	// logger; its cancellation is not followed, call Close.	97	+	// Size is the number of events remembered; DefaultEventBuffer when not
66	-	func NewEvents(ctx context.Context, repos Repos, bus ports.EventBus, size int, clock ports.Clock) Events {	98	+	// positive.
			99	+	Size int
			100	+	// Auth confirms that the token of a watcher still exists. Required.
			101	+	Auth Revalidator
			102	+	// RevalidateEvery is how long the answer of Auth is kept for a watcher;
			103	+	// DefaultRevalidateEvery when zero, and negative to ask for every event.
			104	+	RevalidateEvery time.Duration
			105	+	}
			106	+
			107	+	// NewEvents subscribes to the bus and starts remembering its events. ctx
			108	+	// carries the logger; its cancellation is not followed, call Close.
			109	+	func NewEvents(ctx context.Context, repos Repos, bus ports.EventBus, clock ports.Clock, cfg EventsConfig) Events {
			110	+	size := cfg.Size
67		if size <= 0 {	111		if size <= 0 {
68		size = DefaultEventBuffer	112		size = DefaultEventBuffer
69		}	113		}
			114	+	every := cfg.RevalidateEvery
			115	+	if every == 0 {
			116	+	every = DefaultRevalidateEvery
			117	+	}
70		run, stop := context.WithCancel(context.WithoutCancel(ctx))	118		run, stop := context.WithCancel(context.WithoutCancel(ctx))
71		e := &Events{	119		e := &Events{
72		repos: repos,	120		repos: repos,
			121	+	auth: cfg.Auth,
			122	+	clock: clock,
			123	+	every: every,
			124	+	clients: make(map[string]int),
73		boot: strconv.FormatInt(clock.Now().UnixMilli(), 36),	125		boot: strconv.FormatInt(clock.Now().UnixMilli(), 36),
74		size: size,	126		size: size,
75		stop: stop,	127		stop: stop,

@@ -92,21 +144,48 @@ func (e *Events) feed(ctx context.Context, sub <-chan domain.Event) {
92		if !ok {	144		if !ok {
93		return	145		return
94		}	146		}
95	-	e.append(ev)	147	+	e.append(ctx, ev)
96		case <-ctx.Done():	148		case <-ctx.Done():
97		return	149		return
98		}	150		}
99		}	151		}
100		}	152		}
101			153
102	-	func (e *Events) append(ev domain.Event) {	154	+	// snapshot records the visibility and the generation of the repository an
			155	+	// event is about. The events of the repository lifecycle carry them; for a
			156	+	// push the repository is read now.
			157	+	func (e *Events) snapshot(ctx context.Context, ev domain.Event) bufferedEvent {
			158	+	switch ev := ev.(type) {
			159	+	case domain.RepoCreated:
			160	+	return bufferedEvent{known: true, vis: ev.Visibility, created: ev.At}
			161	+	case domain.RepoUpdated:
			162	+	return bufferedEvent{known: true, vis: ev.Visibility, created: ev.CreatedAt}
			163	+	case domain.RepoDeleted:
			164	+	return bufferedEvent{known: true, vis: ev.Visibility, created: ev.CreatedAt}
			165	+	case domain.PushEvent:
			166	+	repo, err := lookupRepo(ctx, e.repos.meta, ev.Repo)
			167	+	if err != nil {
			168	+	if !errors.Is(err, domain.ErrNotFound) {
			169	+	log := zerowrap.FromCtx(ctx)
			170	+	log.Error().Err(err).Str("repo", ev.Repo.String()).Msg("read repository of a buffered event")
			171	+	}
			172	+	return bufferedEvent{} // gone, or unknown: nobody is told
			173	+	}
			174	+	return bufferedEvent{known: true, vis: repo.Visibility, created: repo.CreatedAt}
			175	+	}
			176	+	return bufferedEvent{known: true}
			177	+	}
			178	+
			179	+	func (e *Events) append(ctx context.Context, ev domain.Event) {
			180	+	be := e.snapshot(ctx, ev) // outside the lock: it reads the store
103		e.mu.Lock()	181		e.mu.Lock()
104		defer e.mu.Unlock()	182		defer e.mu.Unlock()
105		if len(e.ring) == e.size {	183		if len(e.ring) == e.size {
106		copy(e.ring, e.ring[1:])	184		copy(e.ring, e.ring[1:])
107		e.ring = e.ring[:e.size-1]	185		e.ring = e.ring[:e.size-1]
108		}	186		}
109	-	e.ring = append(e.ring, bufferedEvent{seq: e.nextSeq, ev: ev})	187	+	be.seq, be.ev = e.nextSeq, ev
			188	+	e.ring = append(e.ring, be)
110		e.nextSeq++	189		e.nextSeq++
111		e.wake()	190		e.wake()
112		}	191		}

@@ -126,11 +205,14 @@ func (e *Events) shutdown() {
126		}	205		}
127		}	206		}
128			207
129	-	// Close ends every watch cleanly, stops feeding and waits for the feeder.	208	+	// Close ends every watch cleanly, stops feeding and waits for the feeder,
130	-	// It is safe to call more than once.	209	+	// until ctx ends. It is safe to call more than once.
131	-	func (e *Events) Close(context.Context) {	210	+	func (e *Events) Close(ctx context.Context) {
132		e.stop()	211		e.stop()
133	-	<-e.done	212	+	select {
			213	+	case <-e.done:
			214	+	case <-ctx.Done():
			215	+	}
134		}	216		}
135			217
136		// Watchers returns the number of watches in progress.	218		// Watchers returns the number of watches in progress.

@@ -159,6 +241,10 @@ type WatchParams struct {
159		// ResumeAfter is the id of the last event the caller saw; empty to	241		// ResumeAfter is the id of the last event the caller saw; empty to
160		// receive only new events.	242		// receive only new events.
161		ResumeAfter string	243		ResumeAfter string
			244	+	// Client identifies where the watch comes from, for the limit per client
			245	+	// (MaxWatchersPerClient): an opaque key such as a normalized address. Empty
			246	+	// for callers in this process, which only count against MaxWatchers.
			247	+	Client string
162		// Ready, when not nil, is called once the topics are valid and the watch	248		// Ready, when not nil, is called once the topics are valid and the watch
163		// is registered: events published from then on are not missed. Nothing is	249		// is registered: events published from then on are not missed. Nothing is
164		// sent before it returns.	250		// sent before it returns.

@@ -184,15 +270,20 @@ type watch struct {
184		p domain.Principal	270		p domain.Principal
185		home bool	271		home bool
186		repos map[domain.RepoName]bool	272		repos map[domain.RepoName]bool
			273	+
			274	+	checked time.Time // when the token store last confirmed the principal
187		}	275		}
188			276
189		func (w *watch) wants(name domain.RepoName) bool { return w.home \|\| w.repos[name] }	277		func (w *watch) wants(name domain.RepoName) bool { return w.home \|\| w.repos[name] }
190			278
191		// parseTopics validates the topics of a watch. A repository topic needs a	279		// parseTopics validates the topics of a watch. A repository topic needs a
192		// repository the principal may read (else domain.ErrNotFound); an unknown	280		// repository the principal may read (else domain.ErrNotFound); an unknown
193	-	// topic is domain.ErrInvalid.	281	+	// topic, or more than MaxWatchTopics, is domain.ErrInvalid.
194		func (e Events) parseTopics(ctx context.Context, p domain.Principal, topics []string) (watch, error) {	282		func (e Events) parseTopics(ctx context.Context, p domain.Principal, topics []string) (watch, error) {
195	-	w := &watch{p: p, repos: make(map[domain.RepoName]bool)}	283	+	if len(topics) > MaxWatchTopics {
			284	+	return nil, fmt.Errorf("%w: %d topics, at most %d", domain.ErrInvalid, len(topics), MaxWatchTopics)
			285	+	}
			286	+	w := &watch{p: p, repos: make(map[domain.RepoName]bool), checked: time.Now()}
196		if len(topics) == 0 {	287		if len(topics) == 0 {
197		w.home = true	288		w.home = true
198		}	289		}

@@ -220,6 +311,12 @@ func (e *Events) parseTopics(ctx context.Context, p domain.Principal, topics []s
220		// Watch calls send for the events the principal may see until ctx ends, the	311		// Watch calls send for the events the principal may see until ctx ends, the
221		// buffer closes (both return nil) or send fails (its error is returned).	312		// buffer closes (both return nil) or send fails (its error is returned).
222		//	313		//
			314	+	// A principal with a stored token is checked again for every event, and every
			315	+	// few seconds when idle: when the token expired or was revoked the watch ends
			316	+	// with domain.ErrUnauthenticated and nothing more is sent. More watches than
			317	+	// MaxWatchers, or than MaxWatchersPerClient for one Client, give
			318	+	// domain.ErrResourceExhausted.
			319	+	//
223		// Without ResumeAfter only new events are sent. With it, the buffered events	320		// Without ResumeAfter only new events are sent. With it, the buffered events
224		// after that id are sent first, then new ones. When the id is no longer	321		// after that id are sent first, then new ones. When the id is no longer
225		// buffered, or was not issued by this run, a gap comes first and the oldest	322		// buffered, or was not issued by this run, a gap comes first and the oldest

@@ -244,11 +341,23 @@ func (e *Events) Watch(ctx context.Context, p domain.Principal, params WatchPara
244		next, gap = oldest, true	341		next, gap = oldest, true
245		}	342		}
246		}	343		}
			344	+	if e.watches >= MaxWatchers \|\| (params.Client != "" && e.clients[params.Client] >= MaxWatchersPerClient) {
			345	+	e.mu.Unlock()
			346	+	return fmt.Errorf("%w: too many watchers", domain.ErrResourceExhausted)
			347	+	}
247		e.watches++	348		e.watches++
			349	+	if params.Client != "" {
			350	+	e.clients[params.Client]++
			351	+	}
248		e.mu.Unlock()	352		e.mu.Unlock()
249		defer func() {	353		defer func() {
250		e.mu.Lock()	354		e.mu.Lock()
251		e.watches--	355		e.watches--
			356	+	if params.Client != "" {
			357	+	if e.clients[params.Client]--; e.clients[params.Client] == 0 {
			358	+	delete(e.clients, params.Client)
			359	+	}
			360	+	}
252		e.mu.Unlock()	361		e.mu.Unlock()
253		}()	362		}()
254			363

@@ -263,6 +372,9 @@ func (e *Events) Watch(ctx context.Context, p domain.Principal, params WatchPara
263		}	372		}
264		}	373		}
265		for {	374		for {
			375	+	if err := e.stillAllowed(ctx, w); err != nil {
			376	+	return err
			377	+	}
266		batch, changed, oldest, lost, closed := e.since(next)	378		batch, changed, oldest, lost, closed := e.since(next)
267		if lost {	379		if lost {
268		if err := send(WatchEvent{Gap: true}); err != nil {	380		if err := send(WatchEvent{Gap: true}); err != nil {

@@ -283,12 +395,60 @@ func (e *Events) Watch(ctx context.Context, p domain.Principal, params WatchPara
283		if closed {	395		if closed {
284		return nil	396		return nil
285		}	397		}
286	-	select {	398	+	e.wait(ctx, w, changed)
287	-	case <-changed:	399	+	if ctx.Err() != nil {
288	-	case <-ctx.Done():	400	+	return nil
			401	+	}
			402	+	}
			403	+	}
			404	+
			405	+	// wait blocks until more events arrive or ctx ends. A watcher with a stored
			406	+	// token also wakes up every few seconds, so that a revocation ends the watch
			407	+	// of a quiet forge too (the caller checks the token on the next turn).
			408	+	func (e Events) wait(ctx context.Context, w watch, changed <-chan struct{}) {
			409	+	var idle <-chan time.Time
			410	+	if w.p.TokenID() != "" {
			411	+	t := time.NewTimer(e.idleCheck())
			412	+	defer t.Stop()
			413	+	idle = t.C
			414	+	}
			415	+	select {
			416	+	case <-changed:
			417	+	case <-idle:
			418	+	case <-ctx.Done():
			419	+	}
			420	+	}
			421	+
			422	+	// idleCheck is how long a watcher with a stored token waits for news before
			423	+	// checking its token anyway.
			424	+	func (e *Events) idleCheck() time.Duration {
			425	+	if e.every <= 0 {
			426	+	return DefaultRevalidateEvery
			427	+	}
			428	+	return min(e.every, DefaultRevalidateEvery)
			429	+	}
			430	+
			431	+	// stillAllowed checks that the holder of a stored token may still watch: the
			432	+	// token has not expired and, at most every few seconds, still exists. Root and
			433	+	// anonymous callers have nothing to lose.
			434	+	func (e Events) stillAllowed(ctx context.Context, w watch) error {
			435	+	if w.p.TokenID() == "" {
			436	+	return nil
			437	+	}
			438	+	if exp := w.p.ExpiresAt(); !exp.IsZero() && !e.clock.Now().Before(exp) {
			439	+	return domain.ErrUnauthenticated
			440	+	}
			441	+	if e.every > 0 && time.Since(w.checked) < e.every {
			442	+	return nil
			443	+	}
			444	+	if err := e.auth.Revalidate(ctx, w.p); err != nil {
			445	+	if ctx.Err() != nil {
289		return nil	446		return nil
290		}	447		}
			448	+	return err
291		}	449		}
			450	+	w.checked = time.Now()
			451	+	return nil
292		}	452		}
293			453
294		// since returns the buffered events from sequence next on, and the channel	454		// since returns the buffered events from sequence next on, and the channel

@@ -308,6 +468,9 @@ func (e *Events) since(next uint64) (batch []bufferedEvent, changed <-chan struc
308			468
309		// deliver sends the event if the watcher follows its topic and may see it.	469		// deliver sends the event if the watcher follows its topic and may see it.
310		func (e Events) deliver(ctx context.Context, w watch, be bufferedEvent, send func(WatchEvent) error) error {	470		func (e Events) deliver(ctx context.Context, w watch, be bufferedEvent, send func(WatchEvent) error) error {
			471	+	if err := e.stillAllowed(ctx, w); err != nil {
			472	+	return err
			473	+	}
311		out := WatchEvent{ID: e.id(be.seq), Event: be.ev}	474		out := WatchEvent{ID: e.id(be.seq), Event: be.ev}
312		switch ev := be.ev.(type) {	475		switch ev := be.ev.(type) {
313		case domain.EventGap:	476		case domain.EventGap:

@@ -320,23 +483,28 @@ func (e Events) deliver(ctx context.Context, w watch, be bufferedEvent, send f
320		}	483		}
321		return send(out)	484		return send(out)
322		case domain.RepoCreated:	485		case domain.RepoCreated:
323	-	return e.deliverRepo(ctx, w, ev.Repo, true, out, send)	486	+	return e.deliverRepo(ctx, w, ev.Repo, true, be, out, send)
324		case domain.RepoUpdated:	487		case domain.RepoUpdated:
325	-	return e.deliverRepo(ctx, w, ev.Repo, true, out, send)	488	+	return e.deliverRepo(ctx, w, ev.Repo, true, be, out, send)
326		case domain.PushEvent:	489		case domain.PushEvent:
327	-	return e.deliverRepo(ctx, w, ev.Repo, false, out, send)	490	+	return e.deliverRepo(ctx, w, ev.Repo, false, be, out, send)
328		}	491		}
329		return nil // an event this version does not stream	492		return nil // an event this version does not stream
330		}	493		}
331			494
332	-	// deliverRepo re-checks the access to the repository, with its present	495	+	// deliverRepo sends an event about a repository if the watcher could read the
333	-	// visibility, and sends the event. A repository that no longer exists is	496	+	// repository when the event happened and can read it now, and the repository
			497	+	// is still the one the event is about. A repository that no longer exists is
334		// skipped: its RepoDeleted follows or came before.	498		// skipped: its RepoDeleted follows or came before.
335	-	func (e Events) deliverRepo(ctx context.Context, w watch, name domain.RepoName, withRepo bool, out WatchEvent, send func(WatchEvent) error) error {	499	+	//
336	-	if !w.wants(name) {	500	+	// The exception is a repository that became private: the watchers who could
			501	+	// read it before are told once, with its name and its new visibility only,
			502	+	// so that what they show can be removed.
			503	+	func (e Events) deliverRepo(ctx context.Context, w watch, name domain.RepoName, withRepo bool, be bufferedEvent, out WatchEvent, send func(WatchEvent) error) error {
			504	+	if !w.wants(name) \|\| !be.known {
337		return nil	505		return nil
338		}	506		}
339	-	repo, err := e.repos.Get(ctx, w.p, name)	507	+	repo, err := lookupRepo(ctx, e.repos.meta, name)
340		switch {	508		switch {
341		case errors.Is(err, domain.ErrNotFound):	509		case errors.Is(err, domain.ErrNotFound):
342		return nil	510		return nil

@@ -345,6 +513,19 @@ func (e Events) deliverRepo(ctx context.Context, w watch, name domain.RepoName
345		case err != nil:	513		case err != nil:
346		return fmt.Errorf("check access to %q: %w", name, err)	514		return fmt.Errorf("check access to %q: %w", name, err)
347		}	515		}
			516	+	if repo.CreatedAt.After(be.created) {
			517	+	return nil // the event is about a repository of this name that was deleted
			518	+	}
			519	+	if !canReadAs(w.p, name, be.vis) {
			520	+	if up, ok := be.ev.(domain.RepoUpdated); ok && canReadAs(w.p, name, up.Previous) {
			521	+	out.Repo = &RepoView{Repo: domain.Repo{Name: name, Visibility: up.Visibility}}
			522	+	return send(out)
			523	+	}
			524	+	return nil
			525	+	}
			526	+	if !canRead(w.p, repo) {
			527	+	return nil
			528	+	}
348		if withRepo {	529		if withRepo {
349		view, err := e.repos.View(ctx, repo)	530		view, err := e.repos.View(ctx, repo)
350		if err != nil {	531		if err != nil {

M internal/app/events_test.go +367 -33


@@ -19,23 +19,63 @@ import (
19	19		"bnema.dev/ermite/internal/ports/mocks"
20	20		)
21	21
	22	+	// fakeAuth is the Revalidator of the rig: tokens are valid until revoked.
	23	+	type fakeAuth struct {
	24	+	mu sync.Mutex
	25	+	revoked map[string]bool
	26	+	calls int
	27	+	}
	28	+
	29	+	func (a *fakeAuth) Revalidate(_ context.Context, p domain.Principal) error {
	30	+	a.mu.Lock()
	31	+	defer a.mu.Unlock()
	32	+	a.calls++
	33	+	if a.revoked[p.TokenID()] {
	34	+	return domain.ErrUnauthenticated
	35	+	}
	36	+	return nil
	37	+	}
	38	+
	39	+	func (a *fakeAuth) revoke(id string) {
	40	+	a.mu.Lock()
	41	+	defer a.mu.Unlock()
	42	+	if a.revoked == nil {
	43	+	a.revoked = map[string]bool{}
	44	+	}
	45	+	a.revoked[id] = true
	46	+	}
	47	+
	48	+	func (a *fakeAuth) count() int {
	49	+	a.mu.Lock()
	50	+	defer a.mu.Unlock()
	51	+	return a.calls
	52	+	}
	53	+
22	54		// eventRig is an Events use case fed by a channel the test writes to.
23	55		type eventRig struct {
24	56		t *testing.T
25	57		meta *mocks.RepoMetaStore
26	58		git *mocks.GitRepos
27	59		feed chan domain.Event
	60	+	auth *fakeAuth
28	61		events *app.Events
29	62		}
30	63
31	64		func newEventRig(t testing.T, size int) eventRig {
	65	+	t.Helper()
	66	+	return newEventRigConfig(t, app.EventsConfig{Size: size})
	67	+	}
	68	+
	69	+	func newEventRigConfig(t testing.T, cfg app.EventsConfig) eventRig {
32	70		t.Helper()
33	71		r := &eventRig{
34	72		t: t,
35	73		meta: mocks.NewRepoMetaStore(t),
36	74		git: mocks.NewGitRepos(t),
37	75		feed: make(chan domain.Event, 4096),
	76	+	auth: &fakeAuth{},
38	77		}
	78	+	cfg.Auth = r.auth
39	79		bus := mocks.NewEventBus(t)
40	80		bus.EXPECT().Subscribe(mock.Anything).RunAndReturn(func(ctx context.Context) <-chan domain.Event {
41	81		go func() { <-ctx.Done(); close(r.feed) }()

@@ -45,7 +85,7 @@ func newEventRig(t testing.T, size int) eventRig {
45	85		clock.EXPECT().Now().Return(now).Maybe()
46	86		r.git.EXPECT().HeadBranch(mock.Anything, mock.Anything).Return("main", nil).Maybe()
47	87		repos := app.NewRepos(r.meta, r.git, bus, clock)
48		-	r.events = app.NewEvents(context.Background(), repos, bus, size, clock)
	88	+	r.events = app.NewEvents(context.Background(), repos, bus, clock, cfg)
49	89		t.Cleanup(func() { r.events.Close(context.Background()) })
50	90		return r
51	91		}

@@ -92,12 +132,20 @@ func (w watcher) next(t testing.T) app.WatchEvent {
92	132		return app.WatchEvent{}
93	133		}
94	134
95		-	func (w watcher) none(t testing.T) {
	135	+	// sentinelOID marks the push events that only prove the ones before them were
	136	+	// handled.
	137	+	var sentinelOID = fmt.Sprintf("%040x", 0xfeed)
	138	+
	139	+	// quiet proves that the watcher got nothing since its last read: it pushes an
	140	+	// event on n, which the watcher may see, and requires it to be the next one.
	141	+	// Events reach a watcher in order, so anything sent before would show up first.
	142	+	func (w watcher) quiet(t testing.T, r *eventRig, n domain.RepoName) {
96	143		t.Helper()
97		-	select {
98		-	case e := <-w.got:
99		-	t.Fatalf("unexpected event %+v", e)
100		-	case <-time.After(100 * time.Millisecond):
	144	+	r.feed <- domain.PushEvent{Repo: n, Updates: []domain.RefUpdate{{Old: domain.ZeroOID, New: sentinelOID, Ref: "refs/heads/sentinel"}}, At: now}
	145	+	e := w.next(t)
	146	+	push, ok := e.Event.(domain.PushEvent)
	147	+	if !ok \|\| len(push.Updates) != 1 \|\| push.Updates[0].New != sentinelOID {
	148	+	t.Fatalf("expected nothing before the sentinel, got %+v", e)
101	149		}
102	150		}
103	151

@@ -121,7 +169,7 @@ func TestWatchLiveAndFiltered(t *testing.T) {
121	169		for name, w := range map[string]*watcher{"anonymous": anon, "scoped": scoped} {
122	170		e := w.next(t)
123	171		assert.Equal(t, pub, e.Event.(domain.PushEvent).Repo, name)
124		-	w.none(t)
	172	+	w.quiet(t, r, pub)
125	173		}
126	174		first, second := root.next(t), root.next(t)
127	175		assert.Equal(t, priv, first.Event.(domain.PushEvent).Repo)

@@ -130,7 +178,7 @@ func TestWatchLiveAndFiltered(t *testing.T) {
130	178		assert.NotEqual(t, first.ID, second.ID)
131	179
132	180		assert.Equal(t, priv, onlyPriv.next(t).Event.(domain.PushEvent).Repo)
133		-	onlyPriv.none(t)
	181	+	onlyPriv.quiet(t, r, priv)
134	182		}
135	183
136	184		func TestWatchRechecksAccessPerEvent(t *testing.T) {

@@ -155,7 +203,7 @@ func TestWatchRechecksAccessPerEvent(t *testing.T) {
155	203		vis = domain.VisibilityPrivate
156	204		mu.Unlock()
157	205		r.push(n, 2)
158		-	w.none(t)
	206	+	w.quiet(t, r, r.repo("open", domain.VisibilityPublic)) // push 2 is not delivered to anonymous
159	207		}
160	208
161	209		func TestWatchResumeAndGap(t *testing.T) {

@@ -176,21 +224,8 @@ func TestWatchResumeAndGap(t *testing.T) {
176	224		}
177	225		require.Len(t, ids, 20)
178	226
179		-	t.Run("resume replays the buffered events after the id, then goes live", func(t *testing.T) {
180		-	w := r.watch(ctx, domain.Root(), app.WatchParams{ResumeAfter: ids[16]})
181		-	for _, want := range ids[17:] {
182		-	e := w.next(t)
183		-	assert.False(t, e.Gap)
184		-	assert.Equal(t, want, e.ID)
185		-	}
186		-	w.none(t)
187		-	})
188		-
189		-	t.Run("resume at the last event replays nothing", func(t *testing.T) {
190		-	w := r.watch(ctx, domain.Root(), app.WatchParams{ResumeAfter: ids[19]})
191		-	w.none(t)
192		-	})
193		-
	227	+	// The subtests that read the state of the ring come first: the ones that
	228	+	// need a sentinel add events to it.
194	229		t.Run("an id that fell out of the buffer gives a gap, then the oldest buffered event", func(t *testing.T) {
195	230		// 20 events were taken, the buffer holds the last 8: ids[12] is the
196	231		// oldest still present; ids[0] is long gone.

@@ -215,13 +250,27 @@ func TestWatchResumeAndGap(t *testing.T) {
215	250		assert.True(t, w.next(t).Gap)
216	251		})
217	252
218		-	t.Run("a resumed watch goes live after the replay", func(t *testing.T) {
219		-	w := r.watch(ctx, domain.Root(), app.WatchParams{ResumeAfter: ids[19]})
220		-	w.none(t)
	253	+	t.Run("resume replays the buffered events after the id, then goes live", func(t *testing.T) {
	254	+	w := r.watch(ctx, domain.Root(), app.WatchParams{ResumeAfter: ids[16]})
	255	+	for _, want := range ids[17:] {
	256	+	e := w.next(t)
	257	+	assert.False(t, e.Gap)
	258	+	assert.Equal(t, want, e.ID)
	259	+	}
	260	+	// Nothing else was replayed: the next event is the one pushed now.
	261	+	r.push(n, 98)
	262	+	live := w.next(t)
	263	+	assert.False(t, live.Gap)
	264	+	assert.NotContains(t, ids, live.ID)
	265	+	ids = append(ids, live.ID)
	266	+	})
	267	+
	268	+	t.Run("resume at the last event replays nothing, then goes live", func(t *testing.T) {
	269	+	w := r.watch(ctx, domain.Root(), app.WatchParams{ResumeAfter: ids[len(ids)-1]})
221	270		r.push(n, 99)
222	271		e := w.next(t)
223	272		assert.False(t, e.Gap)
224		-	assert.NotEmpty(t, e.ID)
	273	+	assert.NotContains(t, ids, e.ID, "the first event is the new one, nothing was replayed")
225	274		})
226	275		}
227	276

@@ -314,14 +363,24 @@ func TestWatchSlowWatcherGetsAGap(t *testing.T) {
314	363		return nil
315	364		})
316	365		}()
317		-	require.Eventually(t, func() bool { return r.events.Watchers() == 1 }, 5*time.Second, time.Millisecond)
	366	+	// A second, fast watcher tells when the use case has taken every event:
	367	+	// it receives them in order.
	368	+	fast := r.watch(context.Background(), domain.Root(), app.WatchParams{})
	369	+	require.Eventually(t, func() bool { return r.events.Watchers() == 2 }, 5*time.Second, time.Millisecond)
318	370		r.push(n, 0)
319	371		<-blocked
320	372		for i := 1; i < 30; i++ {
321	373		r.push(n, i)
322	374		}
323		-	// Wait for the use case to have taken them all (it does not block).
324		-	time.Sleep(100 * time.Millisecond)
	375	+	// The ring holds four events: the fast watcher may get a gap, but it ends
	376	+	// on the last push, which proves the use case took all of them.
	377	+	last := fmt.Sprintf("%040x", 29+1)
	378	+	for {
	379	+	e := fast.next(t)
	380	+	if p, ok := e.Event.(domain.PushEvent); ok && p.Updates[0].New == last {
	381	+	break
	382	+	}
	383	+	}
325	384		close(release)
326	385
327	386		require.Eventually(t, func() bool {

@@ -349,7 +408,7 @@ func TestEventsRepoPayloads(t *testing.T) {
349	408		require.Eventually(t, func() bool { return r.events.Watchers() == 2 }, 5*time.Second, time.Millisecond)
350	409
351	410		r.feed <- domain.RepoCreated{Repo: pub, Visibility: domain.VisibilityPublic, At: now}
352		-	r.feed <- domain.RepoUpdated{Repo: pub, At: now}
	411	+	r.feed <- domain.RepoUpdated{Repo: pub, At: now, Visibility: domain.VisibilityPublic, Previous: domain.VisibilityPublic}
353	412		// A deleted repository cannot be read any more: its visibility decides.
354	413		r.feed <- domain.RepoDeleted{Repo: priv, Visibility: domain.VisibilityPrivate, At: now}
355	414		r.feed <- domain.RepoDeleted{Repo: pub, Visibility: domain.VisibilityPublic, At: now}

@@ -364,5 +423,280 @@ func TestEventsRepoPayloads(t *testing.T) {
364	423		assert.Equal(t, priv, root.next(t).Event.(domain.RepoDeleted).Repo)
365	424		assert.Equal(t, pub, root.next(t).Event.(domain.RepoDeleted).Repo)
366	425		assert.Equal(t, pub, anon.next(t).Event.(domain.RepoDeleted).Repo, "the private deletion was skipped")
367		-	anon.none(t)
	426	+	anon.quiet(t, r, pub)
	427	+	}
	428	+
	429	+	// tokenWith builds the principal of a stored token with an id and an expiry.
	430	+	func tokenWith(id string, expires time.Time, repos ...string) domain.Principal {
	431	+	tok := domain.Token{ID: id, Scopes: []domain.Scope{domain.ScopeRepoRead}, ExpiresAt: expires}
	432	+	for _, r := range repos {
	433	+	n, err := domain.ParseRepoName(r)
	434	+	if err != nil {
	435	+	panic(err)
	436	+	}
	437	+	tok.Repos = append(tok.Repos, n)
	438	+	}
	439	+	return domain.TokenPrincipal(tok)
	440	+	}
	441	+
	442	+	func TestWatchEndsWhenTheTokenIsRevoked(t *testing.T) {
	443	+	// Asking at every turn keeps the test independent of the clock.
	444	+	r := newEventRigConfig(t, app.EventsConfig{RevalidateEvery: -1})
	445	+	pub := r.repo("pub", domain.VisibilityPublic)
	446	+	priv := r.repo("priv", domain.VisibilityPrivate)
	447	+	p := tokenWith("tok", time.Time{})
	448	+	ctx, cancel := context.WithCancel(context.Background())
	449	+	defer cancel()
	450	+
	451	+	w := r.watch(ctx, p, app.WatchParams{})
	452	+	require.Eventually(t, func() bool { return r.events.Watchers() == 1 }, 5*time.Second, time.Millisecond)
	453	+	r.push(pub, 1)
	454	+	w.next(t)
	455	+
	456	+	r.auth.revoke("tok")
	457	+	r.push(priv, 2) // would be delivered to a token that may read priv
	458	+	select {
	459	+	case err := <-w.done:
	460	+	assert.ErrorIs(t, err, domain.ErrUnauthenticated)
	461	+	case <-time.After(5 * time.Second):
	462	+	t.Fatal("the watch of a revoked token did not end")
	463	+	}
	464	+	select {
	465	+	case e := <-w.got:
	466	+	t.Fatalf("an event was sent after the revocation: %+v", e)
	467	+	default:
	468	+	}
	469	+	}
	470	+
	471	+	func TestWatchEndsWhenTheTokenExpires(t *testing.T) {
	472	+	r := newEventRig(t, 0)
	473	+	pub := r.repo("pub", domain.VisibilityPublic)
	474	+	// The rig's clock stands at now: this token expired at now.
	475	+	p := tokenWith("tok", now)
	476	+	w := r.watch(context.Background(), p, app.WatchParams{})
	477	+	r.push(pub, 1)
	478	+	select {
	479	+	case err := <-w.done:
	480	+	assert.ErrorIs(t, err, domain.ErrUnauthenticated)
	481	+	case <-time.After(5 * time.Second):
	482	+	t.Fatal("the watch of an expired token did not end")
	483	+	}
	484	+	assert.Empty(t, w.got)
	485	+	}
	486	+
	487	+	func TestWatchRevalidatesIdleWatchersAndCachesTheAnswer(t *testing.T) {
	488	+	t.Run("a quiet forge still notices a revocation", func(t *testing.T) {
	489	+	r := newEventRigConfig(t, app.EventsConfig{RevalidateEvery: 10 * time.Millisecond})
	490	+	w := r.watch(context.Background(), tokenWith("tok", time.Time{}), app.WatchParams{})
	491	+	require.Eventually(t, func() bool { return r.events.Watchers() == 1 }, 5*time.Second, time.Millisecond)
	492	+	r.auth.revoke("tok")
	493	+	select {
	494	+	case err := <-w.done:
	495	+	assert.ErrorIs(t, err, domain.ErrUnauthenticated)
	496	+	case <-time.After(5 * time.Second):
	497	+	t.Fatal("an idle watch of a revoked token did not end")
	498	+	}
	499	+	})
	500	+
	501	+	t.Run("the answer is kept between checks", func(t *testing.T) {
	502	+	r := newEventRigConfig(t, app.EventsConfig{RevalidateEvery: time.Hour})
	503	+	pub := r.repo("pub", domain.VisibilityPublic)
	504	+	w := r.watch(context.Background(), tokenWith("tok", time.Time{}), app.WatchParams{})
	505	+	for i := range 20 {
	506	+	r.push(pub, i)
	507	+	}
	508	+	for range 20 {
	509	+	w.next(t)
	510	+	}
	511	+	assert.Zero(t, r.auth.count(), "no store lookup within the interval")
	512	+	})
	513	+
	514	+	t.Run("root and anonymous never ask", func(t *testing.T) {
	515	+	r := newEventRigConfig(t, app.EventsConfig{RevalidateEvery: -1})
	516	+	pub := r.repo("pub", domain.VisibilityPublic)
	517	+	a := r.watch(context.Background(), domain.Anonymous(), app.WatchParams{})
	518	+	b := r.watch(context.Background(), domain.Root(), app.WatchParams{})
	519	+	r.push(pub, 1)
	520	+	a.next(t)
	521	+	b.next(t)
	522	+	assert.Zero(t, r.auth.count())
	523	+	})
	524	+	}
	525	+
	526	+	func TestWatchLimits(t *testing.T) {
	527	+	noop := func(app.WatchEvent) error { return nil }
	528	+
	529	+	t.Run("too many topics", func(t *testing.T) {
	530	+	r := newEventRig(t, 0)
	531	+	r.repo("pub", domain.VisibilityPublic)
	532	+	ok := make([]string, app.MaxWatchTopics)
	533	+	for i := range ok {
	534	+	ok[i] = "home"
	535	+	}
	536	+	ctx, cancel := context.WithCancel(context.Background())
	537	+	cancel() // returns at once once the topics are accepted
	538	+	assert.NoError(t, r.events.Watch(ctx, domain.Root(), app.WatchParams{Topics: ok}, noop))
	539	+	err := r.events.Watch(context.Background(), domain.Root(), app.WatchParams{Topics: append(ok, "home")}, noop)
	540	+	assert.ErrorIs(t, err, domain.ErrInvalid)
	541	+	})
	542	+
	543	+	t.Run("total watchers", func(t *testing.T) {
	544	+	r := newEventRig(t, 0)
	545	+	ctx, cancel := context.WithCancel(context.Background())
	546	+	defer cancel()
	547	+	for range app.MaxWatchers {
	548	+	r.watch(ctx, domain.Root(), app.WatchParams{})
	549	+	}
	550	+	require.Eventually(t, func() bool { return r.events.Watchers() == app.MaxWatchers }, 5*time.Second, time.Millisecond)
	551	+	err := r.events.Watch(ctx, domain.Root(), app.WatchParams{}, noop)
	552	+	assert.ErrorIs(t, err, domain.ErrResourceExhausted)
	553	+	assert.Equal(t, app.MaxWatchers, r.events.Watchers(), "a refused watch is not counted")
	554	+
	555	+	cancel() // all end; a new one is accepted again
	556	+	require.Eventually(t, func() bool { return r.events.Watchers() == 0 }, 5*time.Second, time.Millisecond)
	557	+	again, stop := context.WithCancel(context.Background())
	558	+	stop()
	559	+	assert.NoError(t, r.events.Watch(again, domain.Root(), app.WatchParams{}, noop))
	560	+	})
	561	+
	562	+	t.Run("watchers per client", func(t *testing.T) {
	563	+	r := newEventRig(t, 0)
	564	+	ctx, cancel := context.WithCancel(context.Background())
	565	+	defer cancel()
	566	+	for range app.MaxWatchersPerClient {
	567	+	r.watch(ctx, domain.Root(), app.WatchParams{Client: "198.51.100.1"})
	568	+	}
	569	+	require.Eventually(t, func() bool { return r.events.Watchers() == app.MaxWatchersPerClient }, 5*time.Second, time.Millisecond)
	570	+
	571	+	err := r.events.Watch(ctx, domain.Root(), app.WatchParams{Client: "198.51.100.1"}, noop)
	572	+	assert.ErrorIs(t, err, domain.ErrResourceExhausted)
	573	+
	574	+	// Another client, and callers in process, are not affected.
	575	+	r.watch(ctx, domain.Root(), app.WatchParams{Client: "198.51.100.2"})
	576	+	r.watch(ctx, domain.Root(), app.WatchParams{})
	577	+	require.Eventually(t, func() bool { return r.events.Watchers() == app.MaxWatchersPerClient+2 }, 5*time.Second, time.Millisecond)
	578	+
	579	+	cancel()
	580	+	require.Eventually(t, func() bool { return r.events.Watchers() == 0 }, 5*time.Second, time.Millisecond)
	581	+	again, stop := context.WithCancel(context.Background())
	582	+	stop()
	583	+	assert.NoError(t, r.events.Watch(again, domain.Root(), app.WatchParams{Client: "198.51.100.1"}, noop), "the slots were released")
	584	+	})
	585	+	}
	586	+
	587	+	// repoAt declares a repository whose metadata the test changes.
	588	+	type liveRepo struct {
	589	+	mu sync.Mutex
	590	+	repo domain.Repo
	591	+	}
	592	+
	593	+	func (l liveRepo) set(f func(domain.Repo)) {
	594	+	l.mu.Lock()
	595	+	defer l.mu.Unlock()
	596	+	f(&l.repo)
	597	+	}
	598	+
	599	+	func (r eventRig) liveRepo(s string, vis domain.Visibility, created time.Time) (liveRepo, domain.RepoName) {
	600	+	n := name(r.t, s)
	601	+	l := &liveRepo{repo: domain.Repo{Name: n, Visibility: vis, CreatedAt: created}}
	602	+	r.meta.EXPECT().Get(mock.Anything, n).RunAndReturn(func(context.Context, domain.RepoName) (domain.Repo, error) {
	603	+	l.mu.Lock()
	604	+	defer l.mu.Unlock()
	605	+	return l.repo, nil
	606	+	}).Maybe()
	607	+	return l, n
	608	+	}
	609	+
	610	+	func TestEventsKeepTheVisibilityOfTheirTime(t *testing.T) {
	611	+	ctx, cancel := context.WithCancel(context.Background())
	612	+	defer cancel()
	613	+
	614	+	t.Run("a repository made public does not reveal what happened while it was private", func(t *testing.T) {
	615	+	r := newEventRig(t, 0)
	616	+	l, n := r.liveRepo("secret", domain.VisibilityPrivate, now)
	617	+	root := r.watch(ctx, domain.Root(), app.WatchParams{})
	618	+	require.Eventually(t, func() bool { return r.events.Watchers() >= 1 }, 5*time.Second, time.Millisecond)
	619	+	r.push(n, 1)
	620	+	r.push(n, 2)
	621	+	root.next(t)
	622	+	root.next(t) // both are buffered now
	623	+
	624	+	l.set(func(repo *domain.Repo) { repo.Visibility = domain.VisibilityPublic })
	625	+	r.feed <- domain.RepoUpdated{Repo: n, At: now, Visibility: domain.VisibilityPublic, Previous: domain.VisibilityPrivate, CreatedAt: now}
	626	+	root.next(t)
	627	+	r.push(n, 3) // public era
	628	+
	629	+	// An anonymous caller that resumes with an id nobody issued gets what
	630	+	// is buffered, filtered: the public update and the public push, not the
	631	+	// pushes of the private era.
	632	+	anon := r.watch(ctx, domain.Anonymous(), app.WatchParams{ResumeAfter: "bogus"})
	633	+	assert.True(t, anon.next(t).Gap)
	634	+	upd := anon.next(t)
	635	+	require.IsType(t, domain.RepoUpdated{}, upd.Event)
	636	+	push := anon.next(t)
	637	+	require.IsType(t, domain.PushEvent{}, push.Event)
	638	+	assert.Equal(t, fmt.Sprintf("%040x", 3+1), push.Event.(domain.PushEvent).Updates[0].New)
	639	+	anon.quiet(t, r, n)
	640	+	})
	641	+
	642	+	t.Run("a repository deleted and created again does not inherit old events", func(t *testing.T) {
	643	+	r := newEventRig(t, 0)
	644	+	first := now.Add(-time.Hour)
	645	+	l, n := r.liveRepo("again", domain.VisibilityPublic, first)
	646	+	root := r.watch(ctx, domain.Root(), app.WatchParams{})
	647	+	require.Eventually(t, func() bool { return r.events.Watchers() >= 1 }, 5*time.Second, time.Millisecond)
	648	+	r.push(n, 1)
	649	+	root.next(t)
	650	+
	651	+	l.set(func(repo *domain.Repo) { repo.CreatedAt = now }) // deleted, created again
	652	+	r.push(n, 2)
	653	+	root.next(t)
	654	+
	655	+	anon := r.watch(ctx, domain.Anonymous(), app.WatchParams{ResumeAfter: "bogus"})
	656	+	assert.True(t, anon.next(t).Gap)
	657	+	e := anon.next(t)
	658	+	require.IsType(t, domain.PushEvent{}, e.Event)
	659	+	assert.Equal(t, fmt.Sprintf("%040x", 2+1), e.Event.(domain.PushEvent).Updates[0].New, "only the push to the new repository")
	660	+	anon.quiet(t, r, n)
	661	+	})
	662	+
	663	+	t.Run("a repository made private tells those who could read it, once", func(t *testing.T) {
	664	+	r := newEventRig(t, 0)
	665	+	l, n := r.liveRepo("hide", domain.VisibilityPublic, now)
	666	+	anon := r.watch(ctx, domain.Anonymous(), app.WatchParams{})
	667	+	reader := r.watch(ctx, tokenPrincipal(domain.ScopeRepoRead, "hide"), app.WatchParams{})
	668	+	other := r.watch(ctx, tokenPrincipal(domain.ScopeRepoRead, "elsewhere"), app.WatchParams{})
	669	+	root := r.watch(ctx, domain.Root(), app.WatchParams{})
	670	+	require.Eventually(t, func() bool { return r.events.Watchers() >= 4 }, 5*time.Second, time.Millisecond)
	671	+
	672	+	l.set(func(repo *domain.Repo) { repo.Visibility = domain.VisibilityPrivate })
	673	+	r.feed <- domain.RepoUpdated{Repo: n, At: now, Visibility: domain.VisibilityPrivate, Previous: domain.VisibilityPublic, CreatedAt: now}
	674	+
	675	+	for name, w := range map[string]*watcher{"anonymous": anon, "reader": reader, "other": other, "root": root} {
	676	+	e := w.next(t)
	677	+	up, ok := e.Event.(domain.RepoUpdated)
	678	+	require.True(t, ok, name)
	679	+	assert.Equal(t, n, up.Repo, name)
	680	+	require.NotNil(t, e.Repo, name)
	681	+	assert.Equal(t, domain.VisibilityPrivate, e.Repo.Visibility, name)
	682	+	assert.Empty(t, e.Repo.Description, "nothing but the name and the visibility", name)
	683	+	}
	684	+
	685	+	// Later events of the repository reach those who may read it now, only.
	686	+	r.push(n, 1)
	687	+	assert.IsType(t, domain.PushEvent{}, reader.next(t).Event)
	688	+	assert.IsType(t, domain.PushEvent{}, root.next(t).Event)
	689	+	anon.quiet(t, r, r.repo("open", domain.VisibilityPublic))
	690	+	other.quiet(t, r, r.repo("open2", domain.VisibilityPublic))
	691	+	})
	692	+
	693	+	t.Run("a private repository that never was public stays hidden", func(t *testing.T) {
	694	+	r := newEventRig(t, 0)
	695	+	_, n := r.liveRepo("closed", domain.VisibilityPrivate, now)
	696	+	anon := r.watch(ctx, domain.Anonymous(), app.WatchParams{})
	697	+	require.Eventually(t, func() bool { return r.events.Watchers() >= 1 }, 5*time.Second, time.Millisecond)
	698	+	r.feed <- domain.RepoUpdated{Repo: n, At: now, Visibility: domain.VisibilityPrivate, Previous: domain.VisibilityPrivate, CreatedAt: now}
	699	+	r.push(n, 1)
	700	+	anon.quiet(t, r, r.repo("open", domain.VisibilityPublic))
	701	+	})
368	702		}


@@ -19,23 +19,63 @@ import (
19		"bnema.dev/ermite/internal/ports/mocks"	19		"bnema.dev/ermite/internal/ports/mocks"
20		)	20		)
21			21
			22	+	// fakeAuth is the Revalidator of the rig: tokens are valid until revoked.
			23	+	type fakeAuth struct {
			24	+	mu sync.Mutex
			25	+	revoked map[string]bool
			26	+	calls int
			27	+	}
			28	+
			29	+	func (a *fakeAuth) Revalidate(_ context.Context, p domain.Principal) error {
			30	+	a.mu.Lock()
			31	+	defer a.mu.Unlock()
			32	+	a.calls++
			33	+	if a.revoked[p.TokenID()] {
			34	+	return domain.ErrUnauthenticated
			35	+	}
			36	+	return nil
			37	+	}
			38	+
			39	+	func (a *fakeAuth) revoke(id string) {
			40	+	a.mu.Lock()
			41	+	defer a.mu.Unlock()
			42	+	if a.revoked == nil {
			43	+	a.revoked = map[string]bool{}
			44	+	}
			45	+	a.revoked[id] = true
			46	+	}
			47	+
			48	+	func (a *fakeAuth) count() int {
			49	+	a.mu.Lock()
			50	+	defer a.mu.Unlock()
			51	+	return a.calls
			52	+	}
			53	+
22		// eventRig is an Events use case fed by a channel the test writes to.	54		// eventRig is an Events use case fed by a channel the test writes to.
23		type eventRig struct {	55		type eventRig struct {
24		t *testing.T	56		t *testing.T
25		meta *mocks.RepoMetaStore	57		meta *mocks.RepoMetaStore
26		git *mocks.GitRepos	58		git *mocks.GitRepos
27		feed chan domain.Event	59		feed chan domain.Event
			60	+	auth *fakeAuth
28		events *app.Events	61		events *app.Events
29		}	62		}
30			63
31		func newEventRig(t testing.T, size int) eventRig {	64		func newEventRig(t testing.T, size int) eventRig {
			65	+	t.Helper()
			66	+	return newEventRigConfig(t, app.EventsConfig{Size: size})
			67	+	}
			68	+
			69	+	func newEventRigConfig(t testing.T, cfg app.EventsConfig) eventRig {
32		t.Helper()	70		t.Helper()
33		r := &eventRig{	71		r := &eventRig{
34		t: t,	72		t: t,
35		meta: mocks.NewRepoMetaStore(t),	73		meta: mocks.NewRepoMetaStore(t),
36		git: mocks.NewGitRepos(t),	74		git: mocks.NewGitRepos(t),
37		feed: make(chan domain.Event, 4096),	75		feed: make(chan domain.Event, 4096),
			76	+	auth: &fakeAuth{},
38		}	77		}
			78	+	cfg.Auth = r.auth
39		bus := mocks.NewEventBus(t)	79		bus := mocks.NewEventBus(t)
40		bus.EXPECT().Subscribe(mock.Anything).RunAndReturn(func(ctx context.Context) <-chan domain.Event {	80		bus.EXPECT().Subscribe(mock.Anything).RunAndReturn(func(ctx context.Context) <-chan domain.Event {
41		go func() { <-ctx.Done(); close(r.feed) }()	81		go func() { <-ctx.Done(); close(r.feed) }()

@@ -45,7 +85,7 @@ func newEventRig(t testing.T, size int) eventRig {
45		clock.EXPECT().Now().Return(now).Maybe()	85		clock.EXPECT().Now().Return(now).Maybe()
46		r.git.EXPECT().HeadBranch(mock.Anything, mock.Anything).Return("main", nil).Maybe()	86		r.git.EXPECT().HeadBranch(mock.Anything, mock.Anything).Return("main", nil).Maybe()
47		repos := app.NewRepos(r.meta, r.git, bus, clock)	87		repos := app.NewRepos(r.meta, r.git, bus, clock)
48	-	r.events = app.NewEvents(context.Background(), repos, bus, size, clock)	88	+	r.events = app.NewEvents(context.Background(), repos, bus, clock, cfg)
49		t.Cleanup(func() { r.events.Close(context.Background()) })	89		t.Cleanup(func() { r.events.Close(context.Background()) })
50		return r	90		return r
51		}	91		}

@@ -92,12 +132,20 @@ func (w watcher) next(t testing.T) app.WatchEvent {
92		return app.WatchEvent{}	132		return app.WatchEvent{}
93		}	133		}
94			134
95	-	func (w watcher) none(t testing.T) {	135	+	// sentinelOID marks the push events that only prove the ones before them were
			136	+	// handled.
			137	+	var sentinelOID = fmt.Sprintf("%040x", 0xfeed)
			138	+
			139	+	// quiet proves that the watcher got nothing since its last read: it pushes an
			140	+	// event on n, which the watcher may see, and requires it to be the next one.
			141	+	// Events reach a watcher in order, so anything sent before would show up first.
			142	+	func (w watcher) quiet(t testing.T, r *eventRig, n domain.RepoName) {
96		t.Helper()	143		t.Helper()
97	-	select {	144	+	r.feed <- domain.PushEvent{Repo: n, Updates: []domain.RefUpdate{{Old: domain.ZeroOID, New: sentinelOID, Ref: "refs/heads/sentinel"}}, At: now}
98	-	case e := <-w.got:	145	+	e := w.next(t)
99	-	t.Fatalf("unexpected event %+v", e)	146	+	push, ok := e.Event.(domain.PushEvent)
100	-	case <-time.After(100 * time.Millisecond):	147	+	if !ok \|\| len(push.Updates) != 1 \|\| push.Updates[0].New != sentinelOID {
			148	+	t.Fatalf("expected nothing before the sentinel, got %+v", e)
101		}	149		}
102		}	150		}
103			151

@@ -121,7 +169,7 @@ func TestWatchLiveAndFiltered(t *testing.T) {
121		for name, w := range map[string]*watcher{"anonymous": anon, "scoped": scoped} {	169		for name, w := range map[string]*watcher{"anonymous": anon, "scoped": scoped} {
122		e := w.next(t)	170		e := w.next(t)
123		assert.Equal(t, pub, e.Event.(domain.PushEvent).Repo, name)	171		assert.Equal(t, pub, e.Event.(domain.PushEvent).Repo, name)
124	-	w.none(t)	172	+	w.quiet(t, r, pub)
125		}	173		}
126		first, second := root.next(t), root.next(t)	174		first, second := root.next(t), root.next(t)
127		assert.Equal(t, priv, first.Event.(domain.PushEvent).Repo)	175		assert.Equal(t, priv, first.Event.(domain.PushEvent).Repo)

@@ -130,7 +178,7 @@ func TestWatchLiveAndFiltered(t *testing.T) {
130		assert.NotEqual(t, first.ID, second.ID)	178		assert.NotEqual(t, first.ID, second.ID)
131			179
132		assert.Equal(t, priv, onlyPriv.next(t).Event.(domain.PushEvent).Repo)	180		assert.Equal(t, priv, onlyPriv.next(t).Event.(domain.PushEvent).Repo)
133	-	onlyPriv.none(t)	181	+	onlyPriv.quiet(t, r, priv)
134		}	182		}
135			183
136		func TestWatchRechecksAccessPerEvent(t *testing.T) {	184		func TestWatchRechecksAccessPerEvent(t *testing.T) {

@@ -155,7 +203,7 @@ func TestWatchRechecksAccessPerEvent(t *testing.T) {
155		vis = domain.VisibilityPrivate	203		vis = domain.VisibilityPrivate
156		mu.Unlock()	204		mu.Unlock()
157		r.push(n, 2)	205		r.push(n, 2)
158	-	w.none(t)	206	+	w.quiet(t, r, r.repo("open", domain.VisibilityPublic)) // push 2 is not delivered to anonymous
159		}	207		}
160			208
161		func TestWatchResumeAndGap(t *testing.T) {	209		func TestWatchResumeAndGap(t *testing.T) {

@@ -176,21 +224,8 @@ func TestWatchResumeAndGap(t *testing.T) {
176		}	224		}
177		require.Len(t, ids, 20)	225		require.Len(t, ids, 20)
178			226
179	-	t.Run("resume replays the buffered events after the id, then goes live", func(t *testing.T) {	227	+	// The subtests that read the state of the ring come first: the ones that
180	-	w := r.watch(ctx, domain.Root(), app.WatchParams{ResumeAfter: ids[16]})	228	+	// need a sentinel add events to it.
181	-	for _, want := range ids[17:] {
182	-	e := w.next(t)
183	-	assert.False(t, e.Gap)
184	-	assert.Equal(t, want, e.ID)
185	-	}
186	-	w.none(t)
187	-	})
188	-
189	-	t.Run("resume at the last event replays nothing", func(t *testing.T) {
190	-	w := r.watch(ctx, domain.Root(), app.WatchParams{ResumeAfter: ids[19]})
191	-	w.none(t)
192	-	})
193	-
194		t.Run("an id that fell out of the buffer gives a gap, then the oldest buffered event", func(t *testing.T) {	229		t.Run("an id that fell out of the buffer gives a gap, then the oldest buffered event", func(t *testing.T) {
195		// 20 events were taken, the buffer holds the last 8: ids[12] is the	230		// 20 events were taken, the buffer holds the last 8: ids[12] is the
196		// oldest still present; ids[0] is long gone.	231		// oldest still present; ids[0] is long gone.

@@ -215,13 +250,27 @@ func TestWatchResumeAndGap(t *testing.T) {
215		assert.True(t, w.next(t).Gap)	250		assert.True(t, w.next(t).Gap)
216		})	251		})
217			252
218	-	t.Run("a resumed watch goes live after the replay", func(t *testing.T) {	253	+	t.Run("resume replays the buffered events after the id, then goes live", func(t *testing.T) {
219	-	w := r.watch(ctx, domain.Root(), app.WatchParams{ResumeAfter: ids[19]})	254	+	w := r.watch(ctx, domain.Root(), app.WatchParams{ResumeAfter: ids[16]})
220	-	w.none(t)	255	+	for _, want := range ids[17:] {
			256	+	e := w.next(t)
			257	+	assert.False(t, e.Gap)
			258	+	assert.Equal(t, want, e.ID)
			259	+	}
			260	+	// Nothing else was replayed: the next event is the one pushed now.
			261	+	r.push(n, 98)
			262	+	live := w.next(t)
			263	+	assert.False(t, live.Gap)
			264	+	assert.NotContains(t, ids, live.ID)
			265	+	ids = append(ids, live.ID)
			266	+	})
			267	+
			268	+	t.Run("resume at the last event replays nothing, then goes live", func(t *testing.T) {
			269	+	w := r.watch(ctx, domain.Root(), app.WatchParams{ResumeAfter: ids[len(ids)-1]})
221		r.push(n, 99)	270		r.push(n, 99)
222		e := w.next(t)	271		e := w.next(t)
223		assert.False(t, e.Gap)	272		assert.False(t, e.Gap)
224	-	assert.NotEmpty(t, e.ID)	273	+	assert.NotContains(t, ids, e.ID, "the first event is the new one, nothing was replayed")
225		})	274		})
226		}	275		}
227			276

@@ -314,14 +363,24 @@ func TestWatchSlowWatcherGetsAGap(t *testing.T) {
314		return nil	363		return nil
315		})	364		})
316		}()	365		}()
317	-	require.Eventually(t, func() bool { return r.events.Watchers() == 1 }, 5*time.Second, time.Millisecond)	366	+	// A second, fast watcher tells when the use case has taken every event:
			367	+	// it receives them in order.
			368	+	fast := r.watch(context.Background(), domain.Root(), app.WatchParams{})
			369	+	require.Eventually(t, func() bool { return r.events.Watchers() == 2 }, 5*time.Second, time.Millisecond)
318		r.push(n, 0)	370		r.push(n, 0)
319		<-blocked	371		<-blocked
320		for i := 1; i < 30; i++ {	372		for i := 1; i < 30; i++ {
321		r.push(n, i)	373		r.push(n, i)
322		}	374		}
323	-	// Wait for the use case to have taken them all (it does not block).	375	+	// The ring holds four events: the fast watcher may get a gap, but it ends
324	-	time.Sleep(100 * time.Millisecond)	376	+	// on the last push, which proves the use case took all of them.
			377	+	last := fmt.Sprintf("%040x", 29+1)
			378	+	for {
			379	+	e := fast.next(t)
			380	+	if p, ok := e.Event.(domain.PushEvent); ok && p.Updates[0].New == last {
			381	+	break
			382	+	}
			383	+	}
325		close(release)	384		close(release)
326			385
327		require.Eventually(t, func() bool {	386		require.Eventually(t, func() bool {

@@ -349,7 +408,7 @@ func TestEventsRepoPayloads(t *testing.T) {
349		require.Eventually(t, func() bool { return r.events.Watchers() == 2 }, 5*time.Second, time.Millisecond)	408		require.Eventually(t, func() bool { return r.events.Watchers() == 2 }, 5*time.Second, time.Millisecond)
350			409
351		r.feed <- domain.RepoCreated{Repo: pub, Visibility: domain.VisibilityPublic, At: now}	410		r.feed <- domain.RepoCreated{Repo: pub, Visibility: domain.VisibilityPublic, At: now}
352	-	r.feed <- domain.RepoUpdated{Repo: pub, At: now}	411	+	r.feed <- domain.RepoUpdated{Repo: pub, At: now, Visibility: domain.VisibilityPublic, Previous: domain.VisibilityPublic}
353		// A deleted repository cannot be read any more: its visibility decides.	412		// A deleted repository cannot be read any more: its visibility decides.
354		r.feed <- domain.RepoDeleted{Repo: priv, Visibility: domain.VisibilityPrivate, At: now}	413		r.feed <- domain.RepoDeleted{Repo: priv, Visibility: domain.VisibilityPrivate, At: now}
355		r.feed <- domain.RepoDeleted{Repo: pub, Visibility: domain.VisibilityPublic, At: now}	414		r.feed <- domain.RepoDeleted{Repo: pub, Visibility: domain.VisibilityPublic, At: now}

@@ -364,5 +423,280 @@ func TestEventsRepoPayloads(t *testing.T) {
364		assert.Equal(t, priv, root.next(t).Event.(domain.RepoDeleted).Repo)	423		assert.Equal(t, priv, root.next(t).Event.(domain.RepoDeleted).Repo)
365		assert.Equal(t, pub, root.next(t).Event.(domain.RepoDeleted).Repo)	424		assert.Equal(t, pub, root.next(t).Event.(domain.RepoDeleted).Repo)
366		assert.Equal(t, pub, anon.next(t).Event.(domain.RepoDeleted).Repo, "the private deletion was skipped")	425		assert.Equal(t, pub, anon.next(t).Event.(domain.RepoDeleted).Repo, "the private deletion was skipped")
367	-	anon.none(t)	426	+	anon.quiet(t, r, pub)
			427	+	}
			428	+
			429	+	// tokenWith builds the principal of a stored token with an id and an expiry.
			430	+	func tokenWith(id string, expires time.Time, repos ...string) domain.Principal {
			431	+	tok := domain.Token{ID: id, Scopes: []domain.Scope{domain.ScopeRepoRead}, ExpiresAt: expires}
			432	+	for _, r := range repos {
			433	+	n, err := domain.ParseRepoName(r)
			434	+	if err != nil {
			435	+	panic(err)
			436	+	}
			437	+	tok.Repos = append(tok.Repos, n)
			438	+	}
			439	+	return domain.TokenPrincipal(tok)
			440	+	}
			441	+
			442	+	func TestWatchEndsWhenTheTokenIsRevoked(t *testing.T) {
			443	+	// Asking at every turn keeps the test independent of the clock.
			444	+	r := newEventRigConfig(t, app.EventsConfig{RevalidateEvery: -1})
			445	+	pub := r.repo("pub", domain.VisibilityPublic)
			446	+	priv := r.repo("priv", domain.VisibilityPrivate)
			447	+	p := tokenWith("tok", time.Time{})
			448	+	ctx, cancel := context.WithCancel(context.Background())
			449	+	defer cancel()
			450	+
			451	+	w := r.watch(ctx, p, app.WatchParams{})
			452	+	require.Eventually(t, func() bool { return r.events.Watchers() == 1 }, 5*time.Second, time.Millisecond)
			453	+	r.push(pub, 1)
			454	+	w.next(t)
			455	+
			456	+	r.auth.revoke("tok")
			457	+	r.push(priv, 2) // would be delivered to a token that may read priv
			458	+	select {
			459	+	case err := <-w.done:
			460	+	assert.ErrorIs(t, err, domain.ErrUnauthenticated)
			461	+	case <-time.After(5 * time.Second):
			462	+	t.Fatal("the watch of a revoked token did not end")
			463	+	}
			464	+	select {
			465	+	case e := <-w.got:
			466	+	t.Fatalf("an event was sent after the revocation: %+v", e)
			467	+	default:
			468	+	}
			469	+	}
			470	+
			471	+	func TestWatchEndsWhenTheTokenExpires(t *testing.T) {
			472	+	r := newEventRig(t, 0)
			473	+	pub := r.repo("pub", domain.VisibilityPublic)
			474	+	// The rig's clock stands at now: this token expired at now.
			475	+	p := tokenWith("tok", now)
			476	+	w := r.watch(context.Background(), p, app.WatchParams{})
			477	+	r.push(pub, 1)
			478	+	select {
			479	+	case err := <-w.done:
			480	+	assert.ErrorIs(t, err, domain.ErrUnauthenticated)
			481	+	case <-time.After(5 * time.Second):
			482	+	t.Fatal("the watch of an expired token did not end")
			483	+	}
			484	+	assert.Empty(t, w.got)
			485	+	}
			486	+
			487	+	func TestWatchRevalidatesIdleWatchersAndCachesTheAnswer(t *testing.T) {
			488	+	t.Run("a quiet forge still notices a revocation", func(t *testing.T) {
			489	+	r := newEventRigConfig(t, app.EventsConfig{RevalidateEvery: 10 * time.Millisecond})
			490	+	w := r.watch(context.Background(), tokenWith("tok", time.Time{}), app.WatchParams{})
			491	+	require.Eventually(t, func() bool { return r.events.Watchers() == 1 }, 5*time.Second, time.Millisecond)
			492	+	r.auth.revoke("tok")
			493	+	select {
			494	+	case err := <-w.done:
			495	+	assert.ErrorIs(t, err, domain.ErrUnauthenticated)
			496	+	case <-time.After(5 * time.Second):
			497	+	t.Fatal("an idle watch of a revoked token did not end")
			498	+	}
			499	+	})
			500	+
			501	+	t.Run("the answer is kept between checks", func(t *testing.T) {
			502	+	r := newEventRigConfig(t, app.EventsConfig{RevalidateEvery: time.Hour})
			503	+	pub := r.repo("pub", domain.VisibilityPublic)
			504	+	w := r.watch(context.Background(), tokenWith("tok", time.Time{}), app.WatchParams{})
			505	+	for i := range 20 {
			506	+	r.push(pub, i)
			507	+	}
			508	+	for range 20 {
			509	+	w.next(t)
			510	+	}
			511	+	assert.Zero(t, r.auth.count(), "no store lookup within the interval")
			512	+	})
			513	+
			514	+	t.Run("root and anonymous never ask", func(t *testing.T) {
			515	+	r := newEventRigConfig(t, app.EventsConfig{RevalidateEvery: -1})
			516	+	pub := r.repo("pub", domain.VisibilityPublic)
			517	+	a := r.watch(context.Background(), domain.Anonymous(), app.WatchParams{})
			518	+	b := r.watch(context.Background(), domain.Root(), app.WatchParams{})
			519	+	r.push(pub, 1)
			520	+	a.next(t)
			521	+	b.next(t)
			522	+	assert.Zero(t, r.auth.count())
			523	+	})
			524	+	}
			525	+
			526	+	func TestWatchLimits(t *testing.T) {
			527	+	noop := func(app.WatchEvent) error { return nil }
			528	+
			529	+	t.Run("too many topics", func(t *testing.T) {
			530	+	r := newEventRig(t, 0)
			531	+	r.repo("pub", domain.VisibilityPublic)
			532	+	ok := make([]string, app.MaxWatchTopics)
			533	+	for i := range ok {
			534	+	ok[i] = "home"
			535	+	}
			536	+	ctx, cancel := context.WithCancel(context.Background())
			537	+	cancel() // returns at once once the topics are accepted
			538	+	assert.NoError(t, r.events.Watch(ctx, domain.Root(), app.WatchParams{Topics: ok}, noop))
			539	+	err := r.events.Watch(context.Background(), domain.Root(), app.WatchParams{Topics: append(ok, "home")}, noop)
			540	+	assert.ErrorIs(t, err, domain.ErrInvalid)
			541	+	})
			542	+
			543	+	t.Run("total watchers", func(t *testing.T) {
			544	+	r := newEventRig(t, 0)
			545	+	ctx, cancel := context.WithCancel(context.Background())
			546	+	defer cancel()
			547	+	for range app.MaxWatchers {
			548	+	r.watch(ctx, domain.Root(), app.WatchParams{})
			549	+	}
			550	+	require.Eventually(t, func() bool { return r.events.Watchers() == app.MaxWatchers }, 5*time.Second, time.Millisecond)
			551	+	err := r.events.Watch(ctx, domain.Root(), app.WatchParams{}, noop)
			552	+	assert.ErrorIs(t, err, domain.ErrResourceExhausted)
			553	+	assert.Equal(t, app.MaxWatchers, r.events.Watchers(), "a refused watch is not counted")
			554	+
			555	+	cancel() // all end; a new one is accepted again
			556	+	require.Eventually(t, func() bool { return r.events.Watchers() == 0 }, 5*time.Second, time.Millisecond)
			557	+	again, stop := context.WithCancel(context.Background())
			558	+	stop()
			559	+	assert.NoError(t, r.events.Watch(again, domain.Root(), app.WatchParams{}, noop))
			560	+	})
			561	+
			562	+	t.Run("watchers per client", func(t *testing.T) {
			563	+	r := newEventRig(t, 0)
			564	+	ctx, cancel := context.WithCancel(context.Background())
			565	+	defer cancel()
			566	+	for range app.MaxWatchersPerClient {
			567	+	r.watch(ctx, domain.Root(), app.WatchParams{Client: "198.51.100.1"})
			568	+	}
			569	+	require.Eventually(t, func() bool { return r.events.Watchers() == app.MaxWatchersPerClient }, 5*time.Second, time.Millisecond)
			570	+
			571	+	err := r.events.Watch(ctx, domain.Root(), app.WatchParams{Client: "198.51.100.1"}, noop)
			572	+	assert.ErrorIs(t, err, domain.ErrResourceExhausted)
			573	+
			574	+	// Another client, and callers in process, are not affected.
			575	+	r.watch(ctx, domain.Root(), app.WatchParams{Client: "198.51.100.2"})
			576	+	r.watch(ctx, domain.Root(), app.WatchParams{})
			577	+	require.Eventually(t, func() bool { return r.events.Watchers() == app.MaxWatchersPerClient+2 }, 5*time.Second, time.Millisecond)
			578	+
			579	+	cancel()
			580	+	require.Eventually(t, func() bool { return r.events.Watchers() == 0 }, 5*time.Second, time.Millisecond)
			581	+	again, stop := context.WithCancel(context.Background())
			582	+	stop()
			583	+	assert.NoError(t, r.events.Watch(again, domain.Root(), app.WatchParams{Client: "198.51.100.1"}, noop), "the slots were released")
			584	+	})
			585	+	}
			586	+
			587	+	// repoAt declares a repository whose metadata the test changes.
			588	+	type liveRepo struct {
			589	+	mu sync.Mutex
			590	+	repo domain.Repo
			591	+	}
			592	+
			593	+	func (l liveRepo) set(f func(domain.Repo)) {
			594	+	l.mu.Lock()
			595	+	defer l.mu.Unlock()
			596	+	f(&l.repo)
			597	+	}
			598	+
			599	+	func (r eventRig) liveRepo(s string, vis domain.Visibility, created time.Time) (liveRepo, domain.RepoName) {
			600	+	n := name(r.t, s)
			601	+	l := &liveRepo{repo: domain.Repo{Name: n, Visibility: vis, CreatedAt: created}}
			602	+	r.meta.EXPECT().Get(mock.Anything, n).RunAndReturn(func(context.Context, domain.RepoName) (domain.Repo, error) {
			603	+	l.mu.Lock()
			604	+	defer l.mu.Unlock()
			605	+	return l.repo, nil
			606	+	}).Maybe()
			607	+	return l, n
			608	+	}
			609	+
			610	+	func TestEventsKeepTheVisibilityOfTheirTime(t *testing.T) {
			611	+	ctx, cancel := context.WithCancel(context.Background())
			612	+	defer cancel()
			613	+
			614	+	t.Run("a repository made public does not reveal what happened while it was private", func(t *testing.T) {
			615	+	r := newEventRig(t, 0)
			616	+	l, n := r.liveRepo("secret", domain.VisibilityPrivate, now)
			617	+	root := r.watch(ctx, domain.Root(), app.WatchParams{})
			618	+	require.Eventually(t, func() bool { return r.events.Watchers() >= 1 }, 5*time.Second, time.Millisecond)
			619	+	r.push(n, 1)
			620	+	r.push(n, 2)
			621	+	root.next(t)
			622	+	root.next(t) // both are buffered now
			623	+
			624	+	l.set(func(repo *domain.Repo) { repo.Visibility = domain.VisibilityPublic })
			625	+	r.feed <- domain.RepoUpdated{Repo: n, At: now, Visibility: domain.VisibilityPublic, Previous: domain.VisibilityPrivate, CreatedAt: now}
			626	+	root.next(t)
			627	+	r.push(n, 3) // public era
			628	+
			629	+	// An anonymous caller that resumes with an id nobody issued gets what
			630	+	// is buffered, filtered: the public update and the public push, not the
			631	+	// pushes of the private era.
			632	+	anon := r.watch(ctx, domain.Anonymous(), app.WatchParams{ResumeAfter: "bogus"})
			633	+	assert.True(t, anon.next(t).Gap)
			634	+	upd := anon.next(t)
			635	+	require.IsType(t, domain.RepoUpdated{}, upd.Event)
			636	+	push := anon.next(t)
			637	+	require.IsType(t, domain.PushEvent{}, push.Event)
			638	+	assert.Equal(t, fmt.Sprintf("%040x", 3+1), push.Event.(domain.PushEvent).Updates[0].New)
			639	+	anon.quiet(t, r, n)
			640	+	})
			641	+
			642	+	t.Run("a repository deleted and created again does not inherit old events", func(t *testing.T) {
			643	+	r := newEventRig(t, 0)
			644	+	first := now.Add(-time.Hour)
			645	+	l, n := r.liveRepo("again", domain.VisibilityPublic, first)
			646	+	root := r.watch(ctx, domain.Root(), app.WatchParams{})
			647	+	require.Eventually(t, func() bool { return r.events.Watchers() >= 1 }, 5*time.Second, time.Millisecond)
			648	+	r.push(n, 1)
			649	+	root.next(t)
			650	+
			651	+	l.set(func(repo *domain.Repo) { repo.CreatedAt = now }) // deleted, created again
			652	+	r.push(n, 2)
			653	+	root.next(t)
			654	+
			655	+	anon := r.watch(ctx, domain.Anonymous(), app.WatchParams{ResumeAfter: "bogus"})
			656	+	assert.True(t, anon.next(t).Gap)
			657	+	e := anon.next(t)
			658	+	require.IsType(t, domain.PushEvent{}, e.Event)
			659	+	assert.Equal(t, fmt.Sprintf("%040x", 2+1), e.Event.(domain.PushEvent).Updates[0].New, "only the push to the new repository")
			660	+	anon.quiet(t, r, n)
			661	+	})
			662	+
			663	+	t.Run("a repository made private tells those who could read it, once", func(t *testing.T) {
			664	+	r := newEventRig(t, 0)
			665	+	l, n := r.liveRepo("hide", domain.VisibilityPublic, now)
			666	+	anon := r.watch(ctx, domain.Anonymous(), app.WatchParams{})
			667	+	reader := r.watch(ctx, tokenPrincipal(domain.ScopeRepoRead, "hide"), app.WatchParams{})
			668	+	other := r.watch(ctx, tokenPrincipal(domain.ScopeRepoRead, "elsewhere"), app.WatchParams{})
			669	+	root := r.watch(ctx, domain.Root(), app.WatchParams{})
			670	+	require.Eventually(t, func() bool { return r.events.Watchers() >= 4 }, 5*time.Second, time.Millisecond)
			671	+
			672	+	l.set(func(repo *domain.Repo) { repo.Visibility = domain.VisibilityPrivate })
			673	+	r.feed <- domain.RepoUpdated{Repo: n, At: now, Visibility: domain.VisibilityPrivate, Previous: domain.VisibilityPublic, CreatedAt: now}
			674	+
			675	+	for name, w := range map[string]*watcher{"anonymous": anon, "reader": reader, "other": other, "root": root} {
			676	+	e := w.next(t)
			677	+	up, ok := e.Event.(domain.RepoUpdated)
			678	+	require.True(t, ok, name)
			679	+	assert.Equal(t, n, up.Repo, name)
			680	+	require.NotNil(t, e.Repo, name)
			681	+	assert.Equal(t, domain.VisibilityPrivate, e.Repo.Visibility, name)
			682	+	assert.Empty(t, e.Repo.Description, "nothing but the name and the visibility", name)
			683	+	}
			684	+
			685	+	// Later events of the repository reach those who may read it now, only.
			686	+	r.push(n, 1)
			687	+	assert.IsType(t, domain.PushEvent{}, reader.next(t).Event)
			688	+	assert.IsType(t, domain.PushEvent{}, root.next(t).Event)
			689	+	anon.quiet(t, r, r.repo("open", domain.VisibilityPublic))
			690	+	other.quiet(t, r, r.repo("open2", domain.VisibilityPublic))
			691	+	})
			692	+
			693	+	t.Run("a private repository that never was public stays hidden", func(t *testing.T) {
			694	+	r := newEventRig(t, 0)
			695	+	_, n := r.liveRepo("closed", domain.VisibilityPrivate, now)
			696	+	anon := r.watch(ctx, domain.Anonymous(), app.WatchParams{})
			697	+	require.Eventually(t, func() bool { return r.events.Watchers() >= 1 }, 5*time.Second, time.Millisecond)
			698	+	r.feed <- domain.RepoUpdated{Repo: n, At: now, Visibility: domain.VisibilityPrivate, Previous: domain.VisibilityPrivate, CreatedAt: now}
			699	+	r.push(n, 1)
			700	+	anon.quiet(t, r, r.repo("open", domain.VisibilityPublic))
			701	+	})
368		}	702		}

M internal/app/repos_write.go +6 -2


@@ -73,6 +73,7 @@ func (r *Repos) Update(ctx context.Context, p domain.Principal, name domain.Repo
73	73		if err := adminAccess(p, repo); err != nil {
74	74		return domain.Repo{}, err
75	75		}
	76	+	before := repo
76	77		if params.Description == nil && params.Visibility == nil && params.DefaultBranch == nil {
77	78		return domain.Repo{}, fmt.Errorf("%w: nothing to update", domain.ErrInvalid)
78	79		}

@@ -123,7 +124,10 @@ func (r *Repos) Update(ctx context.Context, p domain.Principal, name domain.Repo
123	124		Bool("default_branch", params.DefaultBranch != nil).
124	125		Str("principal", p.Label()).Str("token_id", p.TokenID()).
125	126		Msg("repository updated")
126		-	r.bus.Publish(ctx, domain.RepoUpdated{Repo: name, At: r.clock.Now().UTC()})
	127	+	r.bus.Publish(ctx, domain.RepoUpdated{
	128	+	Repo: name, At: r.clock.Now().UTC(),
	129	+	Visibility: repo.Visibility, Previous: before.Visibility, CreatedAt: repo.CreatedAt,
	130	+	})
127	131		return repo, nil
128	132		}
129	133

@@ -176,6 +180,6 @@ func (r *Repos) Delete(ctx context.Context, p domain.Principal, name domain.Repo
176	180		return fmt.Errorf("delete metadata of %q: %w", name, err)
177	181		}
178	182		log.Info().Str("repo", name.String()).Str("principal", p.Label()).Str("token_id", p.TokenID()).Msg("repository deleted")
179		-	r.bus.Publish(ctx, domain.RepoDeleted{Repo: name, Visibility: repo.Visibility, At: r.clock.Now().UTC()})
	183	+	r.bus.Publish(ctx, domain.RepoDeleted{Repo: name, Visibility: repo.Visibility, At: r.clock.Now().UTC(), CreatedAt: repo.CreatedAt})
180	184		return nil
181	185		}

 	if err := adminAccess(p, repo); err != nil {
 		return domain.Repo{}, err
 	}
+	before := repo
 	if params.Description == nil && params.Visibility == nil && params.DefaultBranch == nil {
 		return domain.Repo{}, fmt.Errorf("%w: nothing to update", domain.ErrInvalid)
 	}
 		Bool("default_branch", params.DefaultBranch != nil).
 		Str("principal", p.Label()).Str("token_id", p.TokenID()).
 		Msg("repository updated")
-	r.bus.Publish(ctx, domain.RepoUpdated{Repo: name, At: r.clock.Now().UTC()})
+	r.bus.Publish(ctx, domain.RepoUpdated{
+		Repo: name, At: r.clock.Now().UTC(),
+		Visibility: repo.Visibility, Previous: before.Visibility, CreatedAt: repo.CreatedAt,
+	})
 	return repo, nil
 }
 		return fmt.Errorf("delete metadata of %q: %w", name, err)
 	}
 	log.Info().Str("repo", name.String()).Str("principal", p.Label()).Str("token_id", p.TokenID()).Msg("repository deleted")
-	r.bus.Publish(ctx, domain.RepoDeleted{Repo: name, Visibility: repo.Visibility, At: r.clock.Now().UTC()})
+	r.bus.Publish(ctx, domain.RepoDeleted{Repo: name, Visibility: repo.Visibility, At: r.clock.Now().UTC(), CreatedAt: repo.CreatedAt})
 	return nil
 }

M internal/app/repos_write_test.go +3 -1


@@ -29,7 +29,9 @@ func TestUpdate(t *testing.T) {
29	29		f.meta.EXPECT().Get(ctx, n).Return(priv, nil).Twice()
30	30		want := domain.Repo{Name: n, Visibility: domain.VisibilityPublic, Description: "new"}
31	31		f.meta.EXPECT().Update(ctx, want).Return(nil)
32		-	f.bus.EXPECT().Publish(ctx, domain.RepoUpdated{Repo: n, At: now}).Return().Once()
	32	+	f.bus.EXPECT().Publish(ctx, domain.RepoUpdated{
	33	+	Repo: n, At: now, Visibility: domain.VisibilityPublic, Previous: domain.VisibilityPrivate,
	34	+	}).Return().Once()
33	35
34	36		got, err := f.repos.Update(ctx, admin, n, app.UpdateParams{
35	37		Description: ptr("new"), Visibility: ptr(domain.VisibilityPublic),

 		f.meta.EXPECT().Get(ctx, n).Return(priv, nil).Twice()
 		want := domain.Repo{Name: n, Visibility: domain.VisibilityPublic, Description: "new"}
 		f.meta.EXPECT().Update(ctx, want).Return(nil)
-		f.bus.EXPECT().Publish(ctx, domain.RepoUpdated{Repo: n, At: now}).Return().Once()
+		f.bus.EXPECT().Publish(ctx, domain.RepoUpdated{
+			Repo: n, At: now, Visibility: domain.VisibilityPublic, Previous: domain.VisibilityPrivate,
+		}).Return().Once()
 		got, err := f.repos.Update(ctx, admin, n, app.UpdateParams{
 			Description: ptr("new"), Visibility: ptr(domain.VisibilityPublic),

M internal/clientlimit/limiter.go +5 -0


@@ -77,6 +77,11 @@ func limiterKey(addr string) string {
77	77		return netip.PrefixFrom(a, ipv6Prefix).Masked().String()
78	78		}
79	79
	80	+	// Key returns the identity under which a client address is counted, by the
	81	+	// limiter and by the other per-client limits: an IPv4 address, or the /64
	82	+	// network of an IPv6 address.
	83	+	func Key(addr string) string { return limiterKey(addr) }
	84	+
80	85		// Blocked reports whether addr has used up its failures, and how long until
81	86		// it may try again.
82	87		func (l *Limiter) Blocked(_ context.Context, addr string) (bool, time.Duration) {


@@ -77,6 +77,11 @@ func limiterKey(addr string) string {
77	return netip.PrefixFrom(a, ipv6Prefix).Masked().String()	77		return netip.PrefixFrom(a, ipv6Prefix).Masked().String()
78	}	78		}
79		79
		80	+	// Key returns the identity under which a client address is counted, by the
		81	+	// limiter and by the other per-client limits: an IPv4 address, or the /64
		82	+	// network of an IPv6 address.
		83	+	func Key(addr string) string { return limiterKey(addr) }
		84	+
80	// Blocked reports whether addr has used up its failures, and how long until	85		// Blocked reports whether addr has used up its failures, and how long until
81	// it may try again.	86		// it may try again.
82	func (l *Limiter) Blocked(_ context.Context, addr string) (bool, time.Duration) {	87		func (l *Limiter) Blocked(_ context.Context, addr string) (bool, time.Duration) {

M internal/domain/errors.go +3 -0


@@ -33,4 +33,7 @@ var (
33	33		// ErrPreconditionFailed means the state of the forge does not allow the
34	34		// operation, for example a default branch that does not exist.
35	35		ErrPreconditionFailed = errors.New("failed precondition")
	36	+	// ErrResourceExhausted means a limit of the server is reached, for example
	37	+	// the number of event watchers; the caller may retry later.
	38	+	ErrResourceExhausted = errors.New("resource exhausted")
36	39		)


@@ -33,4 +33,7 @@ var (
33	// ErrPreconditionFailed means the state of the forge does not allow the	33		// ErrPreconditionFailed means the state of the forge does not allow the
34	// operation, for example a default branch that does not exist.	34		// operation, for example a default branch that does not exist.
35	ErrPreconditionFailed = errors.New("failed precondition")	35		ErrPreconditionFailed = errors.New("failed precondition")
		36	+	// ErrResourceExhausted means a limit of the server is reached, for example
		37	+	// the number of event watchers; the caller may retry later.
		38	+	ErrResourceExhausted = errors.New("resource exhausted")
36	)	39		)

M internal/domain/event.go +11 -0


@@ -62,9 +62,18 @@ func (RepoCreated) EventKind() string { return "repo.created" }
62	62
63	63		// RepoUpdated is published when the metadata of a repository, or its default
64	64		// branch, changed. The new state is read from the repository.
	65	+	//
	66	+	// The visibility fields tell who could read the repository before and after
	67	+	// the change, and CreatedAt tells which repository of that name it was (a
	68	+	// name can be deleted and created again): the event stream decides who may be
	69	+	// told from them.
65	70		type RepoUpdated struct {
66	71		Repo RepoName
67	72		At time.Time
	73	+	// Visibility is the visibility after the change, Previous the one before.
	74	+	Visibility, Previous Visibility
	75	+	// CreatedAt is the creation time of the repository.
	76	+	CreatedAt time.Time
68	77		}
69	78
70	79		// EventKind implements Event.

@@ -76,6 +85,8 @@ type RepoDeleted struct {
76	85		Repo RepoName
77	86		Visibility Visibility
78	87		At time.Time
	88	+	// CreatedAt is the creation time of the repository that was deleted.
	89	+	CreatedAt time.Time
79	90		}
80	91
81	92		// EventKind implements Event.


@@ -62,9 +62,18 @@ func (RepoCreated) EventKind() string { return "repo.created" }
62		62
63	// RepoUpdated is published when the metadata of a repository, or its default	63		// RepoUpdated is published when the metadata of a repository, or its default
64	// branch, changed. The new state is read from the repository.	64		// branch, changed. The new state is read from the repository.
		65	+	//
		66	+	// The visibility fields tell who could read the repository before and after
		67	+	// the change, and CreatedAt tells which repository of that name it was (a
		68	+	// name can be deleted and created again): the event stream decides who may be
		69	+	// told from them.
65	type RepoUpdated struct {	70		type RepoUpdated struct {
66	Repo RepoName	71		Repo RepoName
67	At time.Time	72		At time.Time
		73	+	// Visibility is the visibility after the change, Previous the one before.
		74	+	Visibility, Previous Visibility
		75	+	// CreatedAt is the creation time of the repository.
		76	+	CreatedAt time.Time
68	}	77		}
69		78
70	// EventKind implements Event.	79		// EventKind implements Event.

@@ -76,6 +85,8 @@ type RepoDeleted struct {
76	Repo RepoName	85		Repo RepoName
77	Visibility Visibility	86		Visibility Visibility
78	At time.Time	87		At time.Time
		88	+	// CreatedAt is the creation time of the repository that was deleted.
		89	+	CreatedAt time.Time
79	}	90		}
80		91
81	// EventKind implements Event.	92		// EventKind implements Event.