1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package app
4
5import (
6 "context"
7 "errors"
8 "fmt"
9 "strconv"
10 "strings"
11 "sync"
12 "time"
13
14 "github.com/bnema/zerowrap"
15
16 "bnema.dev/ermite/internal/domain"
17 "bnema.dev/ermite/internal/ports"
18)
19
20// DefaultEventBuffer is the number of events the Events use case remembers.
21const DefaultEventBuffer = 1024
22
23// Limits of the watchers.
24const (
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
37// Topics a watcher may follow.
38const (
39 // TopicHome follows every repository the caller may see.
40 TopicHome = "home"
41 // topicRepoPrefix starts the topic of one repository: "repo:<name>".
42 topicRepoPrefix = "repo:"
43)
44
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.
51type bufferedEvent struct {
52 seq uint64
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
62}
63
64// Events keeps the last events of the bus in a ring buffer, numbers them, and
65// lets callers watch them: live, or replayed from the id of the last event
66// they saw. Events of a private repository only reach callers that may read
67// it, checked again for every event.
68//
69// Event ids are opaque to clients: "<boot>-<seq>", where boot identifies this
70// run of the server, so that an id from before a restart is unknown rather
71// than taken for a newer event.
72//
73// The goroutine that feeds the buffer uses the logger of the context given to
74// NewEvents and lives until Close.
75type Events struct {
76 repos *Repos
77 auth Revalidator
78 clock ports.Clock
79 boot string
80 size int
81 every time.Duration // see EventsConfig.RevalidateEvery
82
83 stop context.CancelFunc
84 done chan struct{}
85
86 mu sync.Mutex
87 ring []bufferedEvent // capacity size, oldest first
88 nextSeq uint64 // sequence number of the next event, from 1
89 changed chan struct{} // closed and replaced on every append and on close
90 closed bool
91 watches int
92 clients map[string]int // watches in progress by client
93}
94
95// EventsConfig configures Events.
96type 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.
109func NewEvents(ctx context.Context, repos *Repos, bus ports.EventBus, clock ports.Clock, cfg EventsConfig) *Events {
110 size := cfg.Size
111 if size <= 0 {
112 size = DefaultEventBuffer
113 }
114 every := cfg.RevalidateEvery
115 if every == 0 {
116 every = DefaultRevalidateEvery
117 }
118 run, stop := context.WithCancel(context.WithoutCancel(ctx))
119 e := &Events{
120 repos: repos,
121 auth: cfg.Auth,
122 clock: clock,
123 every: every,
124 clients: make(map[string]int),
125 boot: strconv.FormatInt(clock.Now().UnixMilli(), 36),
126 size: size,
127 stop: stop,
128 done: make(chan struct{}),
129 nextSeq: 1,
130 changed: make(chan struct{}),
131 }
132 sub := bus.Subscribe(run) // before returning: nothing published after is missed
133 go e.feed(run, sub)
134 return e
135}
136
137// feed appends the events of the bus until it stops.
138func (e *Events) feed(ctx context.Context, sub <-chan domain.Event) {
139 defer close(e.done)
140 defer e.shutdown()
141 for {
142 select {
143 case ev, ok := <-sub:
144 if !ok {
145 return
146 }
147 e.append(ctx, ev)
148 case <-ctx.Done():
149 return
150 }
151 }
152}
153
154// snapshot records the visibility and the generation of the repository an
155// event is about. Every event of the repository carries them.
156func (e *Events) snapshot(ctx context.Context, ev domain.Event) bufferedEvent {
157 switch ev := ev.(type) {
158 case domain.RepoCreated:
159 return bufferedEvent{known: true, vis: ev.Visibility, created: ev.At}
160 case domain.RepoUpdated:
161 return bufferedEvent{known: true, vis: ev.Visibility, created: ev.CreatedAt}
162 case domain.RepoDeleted:
163 return bufferedEvent{known: true, vis: ev.Visibility, created: ev.CreatedAt}
164 case domain.PushEvent:
165 if ev.CreatedAt.IsZero() {
166 return bufferedEvent{} // repository unknown at publish time: nobody is told
167 }
168 return bufferedEvent{known: true, vis: ev.Visibility, created: ev.CreatedAt}
169 case domain.MergeRequestOpened:
170 return repoSnapshot(ev.Visibility, ev.CreatedAt)
171 case domain.MergeRequestMerged:
172 return repoSnapshot(ev.Visibility, ev.CreatedAt)
173 case domain.MergeRequestClosed:
174 return repoSnapshot(ev.Visibility, ev.CreatedAt)
175 case domain.CommentAdded:
176 return repoSnapshot(ev.Visibility, ev.CreatedAt)
177 }
178 return bufferedEvent{known: true}
179}
180
181// repoSnapshot is the snapshot of an event that carries the visibility and the
182// creation time of its repository; a zero creation time means the repository
183// was unknown when it was published, and nobody is told.
184func repoSnapshot(vis domain.Visibility, created time.Time) bufferedEvent {
185 if created.IsZero() {
186 return bufferedEvent{}
187 }
188 return bufferedEvent{known: true, vis: vis, created: created}
189}
190
191func (e *Events) append(ctx context.Context, ev domain.Event) {
192 be := e.snapshot(ctx, ev) // outside the lock: it reads the store
193 e.mu.Lock()
194 defer e.mu.Unlock()
195 if len(e.ring) == e.size {
196 copy(e.ring, e.ring[1:])
197 e.ring = e.ring[:e.size-1]
198 }
199 be.seq, be.ev = e.nextSeq, ev
200 e.ring = append(e.ring, be)
201 e.nextSeq++
202 e.wake()
203}
204
205// wake releases the watchers waiting for news. The caller holds mu.
206func (e *Events) wake() {
207 close(e.changed)
208 e.changed = make(chan struct{})
209}
210
211func (e *Events) shutdown() {
212 e.mu.Lock()
213 defer e.mu.Unlock()
214 if !e.closed {
215 e.closed = true
216 e.wake()
217 }
218}
219
220// Close ends every watch cleanly, stops feeding and waits for the feeder,
221// until ctx ends. It is safe to call more than once.
222func (e *Events) Close(ctx context.Context) {
223 e.stop()
224 select {
225 case <-e.done:
226 case <-ctx.Done():
227 }
228}
229
230// Watchers returns the number of watches in progress.
231func (e *Events) Watchers() int {
232 e.mu.Lock()
233 defer e.mu.Unlock()
234 return e.watches
235}
236
237func (e *Events) id(seq uint64) string { return e.boot + "-" + strconv.FormatUint(seq, 10) }
238
239// parseID returns the sequence number of an id of this run.
240func (e *Events) parseID(id string) (uint64, bool) {
241 boot, seq, ok := strings.Cut(id, "-")
242 if !ok || boot != e.boot {
243 return 0, false
244 }
245 n, err := strconv.ParseUint(seq, 10, 64)
246 return n, err == nil && n > 0
247}
248
249// WatchParams select the events of a watch.
250type WatchParams struct {
251 // Topics are TopicHome or "repo:<name>"; empty means TopicHome.
252 Topics []string
253 // ResumeAfter is the id of the last event the caller saw; empty to
254 // receive only new events.
255 ResumeAfter string
256 // Client identifies where the watch comes from, for the limit per client
257 // (MaxWatchersPerClient): an opaque key such as a normalized address; the
258 // web passes its visitor's. Empty only for callers with no client, which
259 // count against MaxWatchers alone.
260 Client string
261 // Ready, when not nil, is called once the topics are valid and the watch
262 // is registered: events published from then on are not missed. Nothing is
263 // sent before it returns.
264 Ready func()
265}
266
267// WatchEvent is one item of a watch.
268type WatchEvent struct {
269 // ID identifies the event, for resuming. Empty on a gap.
270 ID string
271 // Gap tells that events were lost: the caller must reload what it shows.
272 // Event is nil then.
273 Gap bool
274 // Event is a domain.RepoCreated, RepoUpdated, RepoDeleted, PushEvent,
275 // MergeRequestOpened, MergeRequestMerged, MergeRequestClosed or
276 // CommentAdded.
277 Event domain.Event
278 // Repo is the state of the repository at delivery, for RepoCreated and
279 // RepoUpdated.
280 Repo *RepoView
281}
282
283// watch is the state of one Watch call.
284type watch struct {
285 p domain.Principal
286 home bool
287 repos map[domain.RepoName]bool
288
289 checked time.Time // when the token store last confirmed the principal
290}
291
292func (w *watch) wants(name domain.RepoName) bool { return w.home || w.repos[name] }
293
294// parseTopics validates the topics of a watch. A repository topic needs a
295// repository the principal may read (else domain.ErrNotFound); an unknown
296// topic, or more than MaxWatchTopics, is domain.ErrInvalid.
297func (e *Events) parseTopics(ctx context.Context, p domain.Principal, topics []string) (*watch, error) {
298 if len(topics) > MaxWatchTopics {
299 return nil, fmt.Errorf("%w: %d topics, at most %d", domain.ErrInvalid, len(topics), MaxWatchTopics)
300 }
301 w := &watch{p: p, repos: make(map[domain.RepoName]bool), checked: time.Now()}
302 if len(topics) == 0 {
303 w.home = true
304 }
305 for _, t := range topics {
306 if t == TopicHome {
307 w.home = true
308 continue
309 }
310 raw, ok := strings.CutPrefix(t, topicRepoPrefix)
311 if !ok {
312 return nil, fmt.Errorf("%w: unknown topic %q (want %q or \"repo:<name>\")", domain.ErrInvalid, t, TopicHome)
313 }
314 name, err := domain.ParseRepoName(raw)
315 if err != nil {
316 return nil, err
317 }
318 if _, err := e.repos.Get(ctx, p, name); err != nil {
319 return nil, err
320 }
321 w.repos[name] = true
322 }
323 return w, nil
324}
325
326// Watch calls send for the events the principal may see until ctx ends, the
327// buffer closes (both return nil) or send fails (its error is returned).
328//
329// A principal with a stored token is checked again for every event, and every
330// few seconds when idle: when the token expired or was revoked the watch ends
331// with domain.ErrUnauthenticated and nothing more is sent. More watches than
332// MaxWatchers, or than MaxWatchersPerClient for one Client, give
333// domain.ErrResourceExhausted.
334//
335// Without ResumeAfter only new events are sent. With it, the buffered events
336// after that id are sent first, then new ones. When the id is no longer
337// buffered, or was not issued by this run, a gap comes first and the oldest
338// buffered event follows. A watcher that falls more than the buffer behind
339// gets a gap too, and carries on with the oldest buffered event.
340func (e *Events) Watch(ctx context.Context, p domain.Principal, params WatchParams, send func(WatchEvent) error) error {
341 w, err := e.parseTopics(ctx, p, params.Topics)
342 if err != nil {
343 return err
344 }
345
346 e.mu.Lock()
347 next := e.nextSeq // only new events
348 gap := false
349 if params.ResumeAfter != "" {
350 seq, ok := e.parseID(params.ResumeAfter)
351 oldest := e.nextSeq - uint64(len(e.ring)) // first buffered sequence, nextSeq when empty
352 switch {
353 case ok && seq >= oldest-1 && seq < e.nextSeq:
354 next = seq + 1
355 default:
356 next, gap = oldest, true
357 }
358 }
359 if e.watches >= MaxWatchers || (params.Client != "" && e.clients[params.Client] >= MaxWatchersPerClient) {
360 e.mu.Unlock()
361 return fmt.Errorf("%w: too many watchers", domain.ErrResourceExhausted)
362 }
363 e.watches++
364 if params.Client != "" {
365 e.clients[params.Client]++
366 }
367 e.mu.Unlock()
368 defer func() {
369 e.mu.Lock()
370 e.watches--
371 if params.Client != "" {
372 if e.clients[params.Client]--; e.clients[params.Client] == 0 {
373 delete(e.clients, params.Client)
374 }
375 }
376 e.mu.Unlock()
377 }()
378
379 if params.Ready != nil {
380 params.Ready()
381 }
382 log := zerowrap.FromCtx(ctx)
383 log.Debug().Strs("topics", params.Topics).Bool("resume", params.ResumeAfter != "").Bool("gap", gap).Msg("watch started")
384 if gap {
385 if err := send(WatchEvent{Gap: true}); err != nil {
386 return err
387 }
388 }
389 for {
390 if err := e.stillAllowed(ctx, w); err != nil {
391 return err
392 }
393 batch, changed, oldest, lost, closed := e.since(next)
394 if lost {
395 if err := send(WatchEvent{Gap: true}); err != nil {
396 return err
397 }
398 next = oldest
399 continue
400 }
401 for _, be := range batch {
402 if err := e.deliver(ctx, w, be, send); err != nil {
403 return err
404 }
405 next = be.seq + 1
406 }
407 if len(batch) > 0 {
408 continue
409 }
410 if closed {
411 return nil
412 }
413 e.wait(ctx, w, changed)
414 if ctx.Err() != nil {
415 return nil
416 }
417 }
418}
419
420// wait blocks until more events arrive or ctx ends. A watcher with a stored
421// token also wakes up every few seconds, so that a revocation ends the watch
422// of a quiet forge too (the caller checks the token on the next turn).
423func (e *Events) wait(ctx context.Context, w *watch, changed <-chan struct{}) {
424 var idle <-chan time.Time
425 if w.p.TokenID() != "" {
426 t := time.NewTimer(e.idleCheck())
427 defer t.Stop()
428 idle = t.C
429 }
430 select {
431 case <-changed:
432 case <-idle:
433 case <-ctx.Done():
434 }
435}
436
437// idleCheck is how long a watcher with a stored token waits for news before
438// checking its token anyway.
439func (e *Events) idleCheck() time.Duration {
440 if e.every <= 0 {
441 return DefaultRevalidateEvery
442 }
443 return min(e.every, DefaultRevalidateEvery)
444}
445
446// stillAllowed checks that the holder of a stored token may still watch: the
447// token has not expired and, at most every few seconds, still exists. Root and
448// anonymous callers have nothing to lose.
449func (e *Events) stillAllowed(ctx context.Context, w *watch) error {
450 if w.p.TokenID() == "" {
451 return nil
452 }
453 if exp := w.p.ExpiresAt(); !exp.IsZero() && !e.clock.Now().Before(exp) {
454 return domain.ErrUnauthenticated
455 }
456 if e.every > 0 && time.Since(w.checked) < e.every {
457 return nil
458 }
459 if err := e.auth.Revalidate(ctx, w.p); err != nil {
460 if ctx.Err() != nil {
461 return nil
462 }
463 return err
464 }
465 w.checked = time.Now()
466 return nil
467}
468
469// since returns the buffered events from sequence next on, and the channel
470// that closes when more arrive. lost is true when next fell out of the buffer;
471// the caller then restarts from oldest, the first buffered sequence. The
472// returned slice is a copy.
473func (e *Events) since(next uint64) (batch []bufferedEvent, changed <-chan struct{}, oldest uint64, lost, closed bool) {
474 e.mu.Lock()
475 defer e.mu.Unlock()
476 oldest = e.nextSeq - uint64(len(e.ring))
477 if next < oldest {
478 return nil, e.changed, oldest, true, e.closed
479 }
480 batch = append(batch, e.ring[next-oldest:]...)
481 return batch, e.changed, oldest, false, e.closed
482}
483
484// deliver sends the event if the watcher follows its topic and may see it.
485func (e *Events) deliver(ctx context.Context, w *watch, be bufferedEvent, send func(WatchEvent) error) error {
486 if err := e.stillAllowed(ctx, w); err != nil {
487 return err
488 }
489 out := WatchEvent{ID: e.id(be.seq), Event: be.ev}
490 switch ev := be.ev.(type) {
491 case domain.EventGap:
492 // The feeder lost events: tell every watcher.
493 return send(WatchEvent{Gap: true})
494 case domain.RepoDeleted:
495 // The repository is gone: use the visibility it had.
496 if !w.wants(ev.Repo) || !canReadAs(w.p, ev.Repo, ev.Visibility) {
497 return nil
498 }
499 return send(out)
500 case domain.RepoCreated:
501 return e.deliverRepo(ctx, w, ev.Repo, true, be, out, send)
502 case domain.RepoUpdated:
503 return e.deliverRepo(ctx, w, ev.Repo, true, be, out, send)
504 case domain.PushEvent:
505 return e.deliverRepo(ctx, w, ev.Repo, false, be, out, send)
506 case domain.MergeRequestOpened:
507 return e.deliverRepo(ctx, w, ev.MergeRequest.Repo, false, be, out, send)
508 case domain.MergeRequestMerged:
509 return e.deliverRepo(ctx, w, ev.MergeRequest.Repo, false, be, out, send)
510 case domain.MergeRequestClosed:
511 return e.deliverRepo(ctx, w, ev.MergeRequest.Repo, false, be, out, send)
512 case domain.CommentAdded:
513 return e.deliverRepo(ctx, w, ev.Repo, false, be, out, send)
514 }
515 return nil // an event this version does not stream
516}
517
518// deliverRepo sends an event about a repository if the watcher could read the
519// repository when the event happened and can read it now, and the repository
520// is still the one the event is about. A repository that no longer exists is
521// skipped: its RepoDeleted follows or came before.
522//
523// The exception is a repository that became private: the watchers who could
524// read it before are told once, with its name and its new visibility only,
525// so that what they show can be removed.
526func (e *Events) deliverRepo(ctx context.Context, w *watch, name domain.RepoName, withRepo bool, be bufferedEvent, out WatchEvent, send func(WatchEvent) error) error {
527 if !w.wants(name) || !be.known {
528 return nil
529 }
530 repo, err := lookupRepo(ctx, e.repos.meta, name)
531 switch {
532 case errors.Is(err, domain.ErrNotFound):
533 return nil
534 case ctx.Err() != nil:
535 return nil
536 case err != nil:
537 return fmt.Errorf("check access to %q: %w", name, err)
538 }
539 if repo.CreatedAt.After(be.created) {
540 return nil // the event is about a repository of this name that was deleted
541 }
542 if !canReadAs(w.p, name, be.vis) {
543 if up, ok := be.ev.(domain.RepoUpdated); ok && canReadAs(w.p, name, up.Previous) {
544 out.Repo = &RepoView{Repo: domain.Repo{Name: name, Visibility: up.Visibility}}
545 return send(out)
546 }
547 return nil
548 }
549 if !canRead(w.p, repo) {
550 return nil
551 }
552 if withRepo {
553 view, err := e.repos.View(ctx, repo)
554 if err != nil {
555 return err
556 }
557 out.Repo = &view
558 }
559 return send(out)
560}