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