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 // TopicOwner follows the merge requests and issues of every repository,
42 // for the owner only (the inbox). Anyone else is told it is an unknown
43 // topic.
44 TopicOwner = "owner"
45 // topicRepoPrefix starts the topic of one repository: "repo:<name>".
46 topicRepoPrefix = "repo:"
47)
48
49// bufferedEvent is an event with its sequence number in the buffer, and what
50// was true of its repository when it was buffered. Access to an event is
51// decided from that, and from the repository as it is when the event is
52// delivered: a repository made public later does not reveal what happened
53// while it was private, and a repository deleted and created again under the
54// same name does not inherit the events of the old one.
55type bufferedEvent struct {
56 seq uint64
57 ev domain.Event
58
59 // known is false when the repository did not exist any more when the
60 // event was buffered: nobody is told about it.
61 known bool
62 // vis is the visibility of the repository at that time.
63 vis domain.Visibility
64 // created is the creation time of that repository.
65 created time.Time
66}
67
68// Events keeps the last events of the bus in a ring buffer, numbers them, and
69// lets callers watch them: live, or replayed from the id of the last event
70// they saw. Events of a private repository only reach callers that may read
71// it, checked again for every event.
72//
73// Event ids are opaque to clients: "<boot>-<seq>", where boot identifies this
74// run of the server, so that an id from before a restart is unknown rather
75// than taken for a newer event.
76//
77// The goroutine that feeds the buffer uses the logger of the context given to
78// NewEvents and lives until Close.
79type Events struct {
80 repos *Repos
81 auth ports.Revalidator
82 clock ports.Clock
83 boot string
84 size int
85 every time.Duration // see EventsConfig.RevalidateEvery
86
87 stop context.CancelFunc
88 done chan struct{}
89
90 mu sync.Mutex
91 ring []bufferedEvent // capacity size, oldest first
92 nextSeq uint64 // sequence number of the next event, from 1
93 changed chan struct{} // closed and replaced on every append and on close
94 closed bool
95 watches int
96 clients map[string]int // watches in progress by client
97}
98
99var _ ports.EventService = (*Events)(nil)
100
101// EventsConfig configures Events.
102type EventsConfig struct {
103 // Size is the number of events remembered; DefaultEventBuffer when not
104 // positive.
105 Size int
106 // Auth confirms that the token of a watcher still exists. Required.
107 Auth ports.Revalidator
108 // RevalidateEvery is how long the answer of Auth is kept for a watcher;
109 // DefaultRevalidateEvery when zero, and negative to ask for every event.
110 RevalidateEvery time.Duration
111}
112
113// NewEvents subscribes to the bus and starts remembering its events. ctx
114// carries the logger; its cancellation is not followed, call Close.
115func NewEvents(ctx context.Context, repos *Repos, bus ports.EventBus, clock ports.Clock, cfg EventsConfig) *Events {
116 size := cfg.Size
117 if size <= 0 {
118 size = DefaultEventBuffer
119 }
120 every := cfg.RevalidateEvery
121 if every == 0 {
122 every = DefaultRevalidateEvery
123 }
124 run, stop := context.WithCancel(context.WithoutCancel(ctx))
125 e := &Events{
126 repos: repos,
127 auth: cfg.Auth,
128 clock: clock,
129 every: every,
130 clients: make(map[string]int),
131 boot: strconv.FormatInt(clock.Now().UnixMilli(), 36),
132 size: size,
133 stop: stop,
134 done: make(chan struct{}),
135 nextSeq: 1,
136 changed: make(chan struct{}),
137 }
138 sub := bus.Subscribe(run) // before returning: nothing published after is missed
139 go e.feed(run, sub)
140 return e
141}
142
143// feed appends the events of the bus until it stops.
144func (e *Events) feed(ctx context.Context, sub <-chan domain.Event) {
145 defer close(e.done)
146 defer e.shutdown()
147 for {
148 select {
149 case ev, ok := <-sub:
150 if !ok {
151 return
152 }
153 e.append(ctx, ev)
154 case <-ctx.Done():
155 return
156 }
157 }
158}
159
160// snapshot records the visibility and the generation of the repository an
161// event is about. Every event of the repository carries them.
162func (e *Events) snapshot(ctx context.Context, ev domain.Event) bufferedEvent {
163 switch ev := ev.(type) {
164 case domain.RepoCreated:
165 return bufferedEvent{known: true, vis: ev.Visibility, created: ev.At}
166 case domain.RepoUpdated:
167 return bufferedEvent{known: true, vis: ev.Visibility, created: ev.CreatedAt}
168 case domain.RepoDeleted:
169 return bufferedEvent{known: true, vis: ev.Visibility, created: ev.CreatedAt}
170 case domain.MirrorFailed:
171 return repoSnapshot(ev.Visibility, ev.CreatedAt)
172 case domain.MirrorRecovered:
173 return repoSnapshot(ev.Visibility, ev.CreatedAt)
174 case domain.ReleasePublished:
175 return repoSnapshot(ev.Visibility, ev.CreatedAt)
176 case domain.ReleaseUpdated:
177 return repoSnapshot(ev.Visibility, ev.CreatedAt)
178 case domain.ReleaseDeleted:
179 return repoSnapshot(ev.Visibility, ev.CreatedAt)
180 case domain.ReleaseUnpublished:
181 return repoSnapshot(ev.Visibility, ev.CreatedAt)
182 case domain.PushEvent:
183 if ev.CreatedAt.IsZero() {
184 return bufferedEvent{} // repository unknown at publish time: nobody is told
185 }
186 return bufferedEvent{known: true, vis: ev.Visibility, created: ev.CreatedAt}
187 case domain.MergeRequestOpened:
188 return repoSnapshot(ev.Visibility, ev.CreatedAt)
189 case domain.MergeRequestMerged:
190 return repoSnapshot(ev.Visibility, ev.CreatedAt)
191 case domain.MergeRequestClosed:
192 return repoSnapshot(ev.Visibility, ev.CreatedAt)
193 case domain.MergeRequestAutoMergeEnabled:
194 return repoSnapshot(ev.Visibility, ev.CreatedAt)
195 case domain.MergeRequestAutoMergeCancelled:
196 return repoSnapshot(ev.Visibility, ev.CreatedAt)
197 case domain.MergeRequestReady:
198 return repoSnapshot(ev.Visibility, ev.CreatedAt)
199 case domain.CommentAdded:
200 return repoSnapshot(ev.Visibility, ev.CreatedAt)
201 case domain.IssueReferenced:
202 return repoSnapshot(ev.Visibility, ev.CreatedAt)
203 case domain.IssueOpened:
204 return repoSnapshot(ev.Visibility, ev.CreatedAt)
205 case domain.IssueResolved:
206 return repoSnapshot(ev.Visibility, ev.CreatedAt)
207 case domain.IssueClosed:
208 return repoSnapshot(ev.Visibility, ev.CreatedAt)
209 case domain.IssueReopened:
210 return repoSnapshot(ev.Visibility, ev.CreatedAt)
211 case domain.IssueCommented:
212 return repoSnapshot(ev.Visibility, ev.CreatedAt)
213 case domain.IssueSubmitted, domain.IssuePendingRemoved:
214 return bufferedEvent{known: true} // owner topic only: no repository snapshot
215 case domain.CommentEdited:
216 return repoSnapshot(ev.Visibility, ev.CreatedAt)
217 case domain.CommentDeleted:
218 return repoSnapshot(ev.Visibility, ev.CreatedAt)
219 case domain.CommentResolved:
220 return repoSnapshot(ev.Visibility, ev.CreatedAt)
221 case domain.MergeRequestMergeability:
222 return repoSnapshot(ev.Visibility, ev.CreatedAt)
223 case domain.RunUpdated:
224 return repoSnapshot(ev.Visibility, ev.CreatedAt)
225 case domain.JobUpdated:
226 return repoSnapshot(ev.Visibility, ev.CreatedAt)
227 }
228 return bufferedEvent{known: true}
229}
230
231// repoSnapshot is the snapshot of an event that carries the visibility and the
232// creation time of its repository; a zero creation time means the repository
233// was unknown when it was published, and nobody is told.
234func repoSnapshot(vis domain.Visibility, created time.Time) bufferedEvent {
235 if created.IsZero() {
236 return bufferedEvent{}
237 }
238 return bufferedEvent{known: true, vis: vis, created: created}
239}
240
241func (e *Events) append(ctx context.Context, ev domain.Event) {
242 be := e.snapshot(ctx, ev) // outside the lock: it reads the store
243 e.mu.Lock()
244 defer e.mu.Unlock()
245 if len(e.ring) == e.size {
246 copy(e.ring, e.ring[1:])
247 e.ring = e.ring[:e.size-1]
248 }
249 be.seq, be.ev = e.nextSeq, ev
250 e.ring = append(e.ring, be)
251 e.nextSeq++
252 e.wake()
253}
254
255// wake releases the watchers waiting for news. The caller holds mu.
256func (e *Events) wake() {
257 close(e.changed)
258 e.changed = make(chan struct{})
259}
260
261func (e *Events) shutdown() {
262 e.mu.Lock()
263 defer e.mu.Unlock()
264 if !e.closed {
265 e.closed = true
266 e.wake()
267 }
268}
269
270// Close ends every watch cleanly, stops feeding and waits for the feeder,
271// until ctx ends. It is safe to call more than once.
272func (e *Events) Close(ctx context.Context) {
273 e.stop()
274 select {
275 case <-e.done:
276 case <-ctx.Done():
277 }
278}
279
280// Watchers returns the number of watches in progress.
281func (e *Events) Watchers() int {
282 e.mu.Lock()
283 defer e.mu.Unlock()
284 return e.watches
285}
286
287func (e *Events) id(seq uint64) string { return e.boot + "-" + strconv.FormatUint(seq, 10) }
288
289// parseID returns the sequence number of an id of this run.
290func (e *Events) parseID(id string) (uint64, bool) {
291 boot, seq, ok := strings.Cut(id, "-")
292 if !ok || boot != e.boot {
293 return 0, false
294 }
295 n, err := strconv.ParseUint(seq, 10, 64)
296 return n, err == nil && n > 0
297}
298
299// watch is the state of one Watch call.
300type watch struct {
301 p domain.Principal
302 home bool
303 owner bool // the owner topic: merge request, issue, repo update and deletion events of every repository
304 repos map[domain.RepoName]bool
305
306 checked time.Time // when the token store last confirmed the principal
307}
308
309func (w *watch) wants(name domain.RepoName) bool { return w.home || w.repos[name] }
310
311// wantsEvent is wants for an event: the owner topic also follows the events
312// about merge requests and issues of every repository.
313func (w *watch) wantsEvent(name domain.RepoName, ev domain.Event) bool {
314 return w.wants(name) || (w.owner && (isItemEvent(ev) || isRepoLifecycleEvent(ev)))
315}
316
317// parseTopics validates the topics of a watch. A repository topic needs a
318// repository the principal may read (else domain.ErrNotFound); an unknown
319// topic, or more than MaxWatchTopics, is domain.ErrInvalid.
320func (e *Events) parseTopics(ctx context.Context, p domain.Principal, topics []string) (*watch, error) {
321 if len(topics) > MaxWatchTopics {
322 return nil, fmt.Errorf("%w: %d topics, at most %d", domain.ErrInvalid, len(topics), MaxWatchTopics)
323 }
324 w := &watch{p: p, repos: make(map[domain.RepoName]bool), checked: time.Now()}
325 if len(topics) == 0 {
326 w.home = true
327 }
328 for _, t := range topics {
329 if t == TopicHome {
330 w.home = true
331 continue
332 }
333 if t == TopicOwner && p.HasOwnerRights() {
334 w.owner = true
335 continue
336 }
337 raw, ok := strings.CutPrefix(t, topicRepoPrefix)
338 if !ok {
339 // Includes "owner" for anyone but the owner: the same error as
340 // any unknown topic.
341 return nil, fmt.Errorf("%w: unknown topic %q (want %q or \"repo:<name>\")", domain.ErrInvalid, t, TopicHome)
342 }
343 name, err := domain.ParseRepoName(raw)
344 if err != nil {
345 return nil, err
346 }
347 if _, err := e.repos.Get(ctx, p, name); err != nil {
348 return nil, err
349 }
350 w.repos[name] = true
351 }
352 return w, nil
353}
354
355// Watch calls send for the events the principal may see until ctx ends, the
356// buffer closes (both return nil) or send fails (its error is returned).
357//
358// A principal with a stored token is checked again for every event, and every
359// few seconds when idle: when the token expired or was revoked the watch ends
360// with domain.ErrUnauthenticated and nothing more is sent. More watches than
361// MaxWatchers, or than MaxWatchersPerClient for one Client, give
362// domain.ErrResourceExhausted.
363//
364// Without ResumeAfter only new events are sent. With it, the buffered events
365// after that id are sent first, then new ones. When the id is no longer
366// buffered, or was not issued by this run, a gap comes first and the oldest
367// buffered event follows. A watcher that falls more than the buffer behind
368// gets a gap too, and carries on with the oldest buffered event.
369func (e *Events) Watch(ctx context.Context, p domain.Principal, params ports.WatchParams, send func(ports.WatchEvent) error) error {
370 w, err := e.parseTopics(ctx, p, params.Topics)
371 if err != nil {
372 return err
373 }
374
375 e.mu.Lock()
376 next := e.nextSeq // only new events
377 gap := false
378 if params.ResumeAfter != "" {
379 seq, ok := e.parseID(params.ResumeAfter)
380 oldest := e.nextSeq - uint64(len(e.ring)) // first buffered sequence, nextSeq when empty
381 switch {
382 case ok && seq >= oldest-1 && seq < e.nextSeq:
383 next = seq + 1
384 default:
385 next, gap = oldest, true
386 }
387 }
388 if e.watches >= MaxWatchers || (params.Client != "" && e.clients[params.Client] >= MaxWatchersPerClient) {
389 e.mu.Unlock()
390 return fmt.Errorf("%w: too many watchers", domain.ErrResourceExhausted)
391 }
392 e.watches++
393 if params.Client != "" {
394 e.clients[params.Client]++
395 }
396 e.mu.Unlock()
397 defer func() {
398 e.mu.Lock()
399 e.watches--
400 if params.Client != "" {
401 if e.clients[params.Client]--; e.clients[params.Client] == 0 {
402 delete(e.clients, params.Client)
403 }
404 }
405 e.mu.Unlock()
406 }()
407
408 if params.Ready != nil {
409 params.Ready()
410 }
411 log := zerowrap.FromCtx(ctx)
412 log.Debug().Strs("topics", params.Topics).Bool("resume", params.ResumeAfter != "").Bool("gap", gap).Msg("watch started")
413 if gap {
414 if err := send(ports.WatchEvent{Gap: true}); err != nil {
415 return err
416 }
417 }
418 for {
419 if err := e.stillAllowed(ctx, w); err != nil {
420 return err
421 }
422 batch, changed, oldest, lost, closed := e.since(next)
423 if lost {
424 if err := send(ports.WatchEvent{Gap: true}); err != nil {
425 return err
426 }
427 next = oldest
428 continue
429 }
430 for _, be := range batch {
431 if err := e.deliver(ctx, w, be, send); err != nil {
432 return err
433 }
434 next = be.seq + 1
435 }
436 if len(batch) > 0 {
437 continue
438 }
439 if closed {
440 return nil
441 }
442 e.wait(ctx, w, changed)
443 if ctx.Err() != nil {
444 return nil
445 }
446 }
447}
448
449// wait blocks until more events arrive or ctx ends. A watcher with a stored
450// token also wakes up every few seconds, so that a revocation ends the watch
451// of a quiet forge too (the caller checks the token on the next turn).
452func (e *Events) wait(ctx context.Context, w *watch, changed <-chan struct{}) {
453 var idle <-chan time.Time
454 if w.p.TokenID() != "" {
455 t := time.NewTimer(e.idleCheck())
456 defer t.Stop()
457 idle = t.C
458 }
459 select {
460 case <-changed:
461 case <-idle:
462 case <-ctx.Done():
463 }
464}
465
466// idleCheck is how long a watcher with a stored token waits for news before
467// checking its token anyway.
468func (e *Events) idleCheck() time.Duration {
469 if e.every <= 0 {
470 return DefaultRevalidateEvery
471 }
472 return min(e.every, DefaultRevalidateEvery)
473}
474
475// stillAllowed checks that the holder of a stored token may still watch: the
476// token has not expired and, at most every few seconds, still exists. Root and
477// anonymous callers have nothing to lose. The owner topic needs nothing more:
478// a stored token with owner rights loses them only by ceasing to exist, which
479// is checked here, and the owner's web session, which this use case cannot see,
480// is checked by the web adapter before it sends anything.
481func (e *Events) stillAllowed(ctx context.Context, w *watch) error {
482 if w.p.TokenID() == "" {
483 return nil
484 }
485 if exp := w.p.ExpiresAt(); !exp.IsZero() && !e.clock.Now().Before(exp) {
486 return domain.ErrUnauthenticated
487 }
488 if e.every > 0 && time.Since(w.checked) < e.every {
489 return nil
490 }
491 if err := e.auth.Revalidate(ctx, w.p); err != nil {
492 if ctx.Err() != nil {
493 return nil
494 }
495 return err
496 }
497 w.checked = time.Now()
498 return nil
499}
500
501// since returns the buffered events from sequence next on, and the channel
502// that closes when more arrive. lost is true when next fell out of the buffer;
503// the caller then restarts from oldest, the first buffered sequence. The
504// returned slice is a copy.
505func (e *Events) since(next uint64) (batch []bufferedEvent, changed <-chan struct{}, oldest uint64, lost, closed bool) {
506 e.mu.Lock()
507 defer e.mu.Unlock()
508 oldest = e.nextSeq - uint64(len(e.ring))
509 if next < oldest {
510 return nil, e.changed, oldest, true, e.closed
511 }
512 batch = append(batch, e.ring[next-oldest:]...)
513 return batch, e.changed, oldest, false, e.closed
514}
515
516// deliver sends the event if the watcher follows its topic and may see it.
517func (e *Events) deliver(ctx context.Context, w *watch, be bufferedEvent, send func(ports.WatchEvent) error) error {
518 if err := e.stillAllowed(ctx, w); err != nil {
519 return err
520 }
521 out := ports.WatchEvent{ID: e.id(be.seq), Event: be.ev}
522 switch ev := be.ev.(type) {
523 case domain.EventGap:
524 // The feeder lost events: tell every watcher.
525 return send(ports.WatchEvent{Gap: true})
526 case domain.RepoDeleted:
527 // The repository is gone: use the visibility it had.
528 if !w.wantsEvent(ev.Repo, ev) || !canReadAs(w.p, ev.Repo, ev.Visibility) {
529 return nil
530 }
531 return send(out)
532 case domain.IssueSubmitted:
533 return e.deliverOwnerOnly(ctx, w, ev.Repo, out, send)
534 case domain.IssuePendingRemoved:
535 return e.deliverOwnerOnly(ctx, w, ev.Repo, out, send)
536 case domain.RepoCreated:
537 return e.deliverRepo(ctx, w, ev.Repo, true, be, out, send)
538 case domain.RepoUpdated:
539 return e.deliverRepo(ctx, w, ev.Repo, true, be, out, send)
540 case domain.MirrorFailed:
541 return e.deliverRepo(ctx, w, ev.Repo, false, be, out, send)
542 case domain.MirrorRecovered:
543 return e.deliverRepo(ctx, w, ev.Repo, false, be, out, send)
544 case domain.ReleasePublished:
545 return e.deliverRelease(ctx, w, ev.Release, be, out, send)
546 case domain.ReleaseUpdated:
547 return e.deliverRelease(ctx, w, ev.Release, be, out, send)
548 case domain.ReleaseDeleted:
549 return e.deliverRelease(ctx, w, ev.Release, be, out, send)
550 case domain.ReleaseUnpublished:
551 return e.deliverRelease(ctx, w, ev.Release, be, out, send)
552 case domain.PushEvent:
553 return e.deliverRepo(ctx, w, ev.Repo, false, be, out, send)
554 case domain.MergeRequestOpened:
555 return e.deliverRepo(ctx, w, ev.MergeRequest.Repo, false, be, out, send)
556 case domain.MergeRequestMerged:
557 return e.deliverRepo(ctx, w, ev.MergeRequest.Repo, false, be, out, send)
558 case domain.MergeRequestClosed:
559 return e.deliverRepo(ctx, w, ev.MergeRequest.Repo, false, be, out, send)
560 case domain.MergeRequestAutoMergeEnabled:
561 return e.deliverRepo(ctx, w, ev.MergeRequest.Repo, false, be, out, send)
562 case domain.MergeRequestAutoMergeCancelled:
563 return e.deliverRepo(ctx, w, ev.MergeRequest.Repo, false, be, out, send)
564 case domain.MergeRequestReady:
565 return e.deliverRepo(ctx, w, ev.MergeRequest.Repo, false, be, out, send)
566 case domain.CommentAdded:
567 return e.deliverRepo(ctx, w, ev.Repo, false, be, out, send)
568 case domain.IssueReferenced:
569 return e.deliverRepo(ctx, w, ev.Repo, false, be, out, send)
570 case domain.IssueOpened:
571 return e.deliverRepo(ctx, w, ev.Issue.Repo, false, be, out, send)
572 case domain.IssueResolved:
573 return e.deliverRepo(ctx, w, ev.Issue.Repo, false, be, out, send)
574 case domain.IssueClosed:
575 return e.deliverRepo(ctx, w, ev.Issue.Repo, false, be, out, send)
576 case domain.IssueReopened:
577 return e.deliverRepo(ctx, w, ev.Issue.Repo, false, be, out, send)
578 case domain.IssueCommented:
579 return e.deliverRepo(ctx, w, ev.Repo, false, be, out, send)
580 case domain.CommentEdited:
581 return e.deliverRepo(ctx, w, ev.Repo, false, be, out, send)
582 case domain.CommentDeleted:
583 return e.deliverRepo(ctx, w, ev.Repo, false, be, out, send)
584 case domain.CommentResolved:
585 return e.deliverRepo(ctx, w, ev.Repo, false, be, out, send)
586 case domain.MergeRequestMergeability:
587 return e.deliverRepo(ctx, w, ev.Repo, false, be, out, send)
588 case domain.RunUpdated:
589 // Home shows the CI state of each repository.
590 return e.deliverRepo(ctx, w, ev.Run.Repo, false, be, out, send)
591 case domain.JobUpdated:
592 // Home gets a job when it ends only: a failed job turns the state to
593 // failing before its run ends. Its progress stays on the topic of its
594 // repository.
595 if !w.repos[ev.Job.Repo] && ev.Job.Phase != domain.CIPhaseEnded {
596 return nil
597 }
598 return e.deliverRepo(ctx, w, ev.Job.Repo, false, be, out, send)
599 }
600 return nil // an event this version does not stream
601}
602
603// deliverOwnerOnly sends an owner-topic event: never home, never a repository
604// topic, not even to an owner who watches the repository. No content travels.
605func (e *Events) deliverOwnerOnly(ctx context.Context, w *watch, name domain.RepoName, out ports.WatchEvent, send func(ports.WatchEvent) error) error {
606 if !w.owner {
607 return nil
608 }
609 repo, err := lookupRepo(ctx, e.repos.meta, name)
610 switch {
611 case errors.Is(err, domain.ErrNotFound), ctx.Err() != nil:
612 return nil
613 case err != nil:
614 return fmt.Errorf("check access to %q: %w", name, err)
615 }
616 if !repo.Settings.IssuesEnabled() {
617 return nil
618 }
619 return send(out)
620}
621
622// deliverRepo sends an event about a repository if the watcher could read the
623// repository when the event happened and can read it now, and the repository
624// is still the one the event is about. A repository that no longer exists is
625// skipped: its RepoDeleted follows or came before.
626//
627// The exception is a repository that became private: the watchers who could
628// read it before are told once, with its name and its new visibility only,
629// so that what they show can be removed.
630func (e *Events) deliverRepo(ctx context.Context, w *watch, name domain.RepoName, withRepo bool, be bufferedEvent, out ports.WatchEvent, send func(ports.WatchEvent) error) error {
631 if !w.wantsEvent(name, be.ev) || !be.known {
632 return nil
633 }
634 repo, err := lookupRepo(ctx, e.repos.meta, name)
635 switch {
636 case errors.Is(err, domain.ErrNotFound):
637 return nil
638 case ctx.Err() != nil:
639 return nil
640 case err != nil:
641 return fmt.Errorf("check access to %q: %w", name, err)
642 }
643 if repo.CreatedAt.After(be.created) {
644 return nil // the event is about a repository of this name that was deleted
645 }
646 if !canReadAs(w.p, name, be.vis) {
647 if up, ok := be.ev.(domain.RepoUpdated); ok && canReadAs(w.p, name, up.Previous) {
648 out.Repo = &ports.RepoView{Repo: domain.Repo{Name: name, Visibility: up.Visibility}}
649 return send(out)
650 }
651 return nil
652 }
653 if !canRead(w.p, repo) {
654 return nil
655 }
656 if isIssueEvent(be.ev) && !repo.Settings.IssuesEnabled() {
657 return nil // issues were turned off since: a replay must not show them
658 }
659 if withRepo {
660 view, err := e.repos.View(ctx, repo)
661 if err != nil {
662 return err
663 }
664 out.Repo = &view
665 }
666 return send(out)
667}
668
669// deliverRelease sends a release event. The event of a draft goes only to
670// release writers: a draft is invisible to every other reader.
671func (e *Events) deliverRelease(ctx context.Context, w *watch, rel domain.Release, be bufferedEvent, out ports.WatchEvent, send func(ports.WatchEvent) error) error {
672 if rel.Draft && !releaseWriter(w.p, rel.Repo) {
673 return nil
674 }
675 return e.deliverRepo(ctx, w, rel.Repo, false, be, out, send)
676}
677
678// isIssueEvent reports whether ev is about an issue.
679func isIssueEvent(ev domain.Event) bool {
680 switch ev.(type) {
681 case domain.IssueOpened, domain.IssueResolved, domain.IssueClosed, domain.IssueReopened, domain.IssueCommented, domain.IssueReferenced:
682 return true
683 }
684 return false
685}
686
687// isRepoLifecycleEvent reports whether ev is a repository update or deletion:
688// the owner topic follows them, for they change what the inbox lists.
689func isRepoLifecycleEvent(ev domain.Event) bool {
690 switch ev.(type) {
691 case domain.RepoUpdated, domain.RepoDeleted:
692 return true
693 }
694 return false
695}
696
697// isItemEvent reports whether ev changes an inbox item, including CI runs.
698func isItemEvent(ev domain.Event) bool {
699 switch ev.(type) {
700 case domain.IssueReferenced:
701 return false // back-reference refreshes do not change the inbox
702 case domain.RunUpdated, domain.MergeRequestOpened, domain.MergeRequestMerged, domain.MergeRequestClosed,
703 domain.MergeRequestAutoMergeEnabled, domain.MergeRequestAutoMergeCancelled,
704 domain.MergeRequestReady, domain.MergeRequestMergeability, domain.CommentAdded,
705 domain.CommentEdited, domain.CommentDeleted, domain.CommentResolved:
706 return true
707 }
708 return isIssueEvent(ev)
709}