1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package web
4
5import (
6 "bytes"
7 "context"
8 "errors"
9 "strconv"
10 "strings"
11 "sync"
12 "time"
13
14 "github.com/a-h/templ"
15
16 "bnema.dev/ermite/internal/domain"
17 "bnema.dev/ermite/internal/ports"
18)
19
20// liveStream turns the events of Watch into the fragments one page shows.
21type liveStream struct {
22 s *Server
23 repo string // "" on the home page
24 // filter is the repository the inbox is filtered on, "" for all.
25 filter string
26 view string // "summary", "refs" or "commits"
27 ref string // commits: the branch listed
28 head string // commits: the commit listed first
29 n string // mr: the number of the merge request shown; run: the run id
30 job string // job: the id of the job shown
31 // sha is the commit of the run of the job shown, read once when the stream
32 // is set up: the annotations link to files at it, and it never changes.
33 sha string
34 // headRuns says, on a merge request page, whether a run is one of the
35 // head shown (its checks are on that head); a job event does not carry
36 // its commit, so its run is read once. One stream reads it, no lock.
37 headRuns map[int64]bool
38}
39
40// maxHeadRuns bounds headRuns: past it the map starts again.
41const maxHeadRuns = 256
42
43// renderFragment renders a component to the data of an event.
44func renderFragment(ctx context.Context, event string, c templ.Component) (fragment, error) {
45 var b bytes.Buffer
46 if err := c.Render(ctx, &b); err != nil {
47 return fragment{}, err
48 }
49 return fragment{event: event, html: b.String()}, nil
50}
51
52var gap = []fragment{{event: "gap", html: "gap"}}
53
54// liveEvent reports whether the event of a watch is one the pages follow: a
55// gap, or one of the events of the forge. The use case may send others, which
56// no page shows and no stream carries, not even their id.
57func liveEvent(we ports.WatchEvent) bool {
58 if we.Gap {
59 return true
60 }
61 switch we.Event.(type) {
62 case domain.RepoCreated, domain.RepoUpdated, domain.RepoDeleted, domain.PushEvent,
63 domain.MergeRequestOpened, domain.MergeRequestMerged, domain.MergeRequestClosed, domain.MergeRequestReady, domain.CommentAdded, domain.CommentEdited, domain.CommentDeleted, domain.CommentResolved,
64 domain.RunUpdated, domain.JobUpdated,
65 domain.IssueOpened, domain.IssueClosed, domain.IssueReopened, domain.IssueCommented, domain.IssueSubmitted, domain.IssuePendingRemoved:
66 return true
67 }
68 return false
69}
70
71// render returns the fragments for an event, none when it changes nothing on
72// the page.
73func (l *liveStream) render(ctx context.Context, we ports.WatchEvent) ([]fragment, error) {
74 if we.Gap {
75 return gap, nil
76 }
77 if we.Event == nil {
78 return nil, nil
79 }
80 if l.view == "inbox" {
81 return l.inbox(ctx, we)
82 }
83 if l.repo == "" {
84 return l.home(ctx, we)
85 }
86 return l.repoPage(ctx, we)
87}
88
89// repoByName reads a repository as the caller of the request. A name that is
90// not valid is a repository that is not there.
91func (l *liveStream) repoByName(ctx context.Context, rawName string) (ports.RepoView, error) {
92 name, err := domain.ParseRepoName(rawName)
93 if err != nil {
94 return ports.RepoView{}, domain.ErrNotFound
95 }
96 return l.s.repoView(ctx, name)
97}
98
99// home upserts the row of the repository an event is about. The row is read
100// again from the use case as the principal of the stream, so a repository
101// that was deleted or that the visitor can no longer see (public to private)
102// removes the row, and the stream never carries more than the list shows.
103func (l *liveStream) home(ctx context.Context, we ports.WatchEvent) ([]fragment, error) {
104 var name string
105 fresh := false
106 switch ev := we.Event.(type) {
107 case domain.RepoDeleted:
108 return l.gone(ctx, ev.Repo.String())
109 case domain.RepoCreated:
110 name = ev.Repo.String()
111 case domain.RepoUpdated:
112 name = ev.Repo.String()
113 case domain.PushEvent:
114 name, fresh = ev.Repo.String(), true
115 case domain.MergeRequestOpened:
116 name, fresh = ev.MergeRequest.Repo.String(), true
117 case domain.MergeRequestMerged:
118 name, fresh = ev.MergeRequest.Repo.String(), true
119 case domain.MergeRequestClosed:
120 name, fresh = ev.MergeRequest.Repo.String(), true
121 case domain.MergeRequestReady:
122 name, fresh = ev.MergeRequest.Repo.String(), true
123 case domain.CommentAdded:
124 name, fresh = ev.Repo.String(), true
125 case domain.IssueOpened:
126 name, fresh = ev.Issue.Repo.String(), true
127 case domain.IssueClosed:
128 name, fresh = ev.Issue.Repo.String(), true
129 case domain.IssueReopened:
130 name, fresh = ev.Issue.Repo.String(), true
131 case domain.IssueCommented:
132 name, fresh = ev.Repo.String(), true
133 case domain.RunUpdated:
134 name = ev.Run.Repo.String() // the CI state of the row
135 case domain.JobUpdated:
136 if ev.Job.Phase != domain.CIPhaseEnded {
137 return nil, nil
138 }
139 name = ev.Job.Repo.String() // a failed job turns the state to failing
140 default:
141 return nil, nil
142 }
143 view, err := l.repoByName(ctx, name)
144 switch {
145 case errors.Is(err, domain.ErrNotFound):
146 return l.gone(ctx, name)
147 case err != nil:
148 return nil, err
149 }
150 row := l.s.homeRowOf(ctx, view, l.s.cfg.Now(), false)
151 row.Fresh = fresh
152 f, err := renderFragment(ctx, "home-list", HomeRowView(row))
153 return []fragment{f}, err
154}
155
156func (l *liveStream) gone(ctx context.Context, name string) ([]fragment, error) {
157 f, err := renderFragment(ctx, "home-list", RepoGone("repo-"+name))
158 return []fragment{f}, err
159}
160
161// repoPage handles the events of the topic of one repository.
162func (l *liveStream) repoPage(ctx context.Context, we ports.WatchEvent) ([]fragment, error) {
163 switch ev := we.Event.(type) {
164 case domain.RunUpdated:
165 return l.runEvent(ctx, ev)
166 case domain.JobUpdated:
167 return l.jobEvent(ctx, ev)
168 case domain.RepoDeleted, domain.RepoCreated:
169 // Gone, or a new repository under the same name: whatever the page
170 // shows is not true any more.
171 return gap, nil
172 case domain.RepoUpdated:
173 _, err := l.repoByName(ctx, l.repo)
174 switch {
175 case errors.Is(err, domain.ErrNotFound):
176 return gap, nil // no longer visible
177 case err != nil:
178 return nil, err
179 case l.view == "summary", l.view == "refs":
180 return gap, nil // the description, the default branch
181 }
182 return nil, nil
183 case domain.MergeRequestOpened:
184 return l.mrNotice(ctx, ev.MergeRequest.Number, "a merge request was opened — reload")
185 case domain.MergeRequestMerged:
186 return l.mrNotice(ctx, ev.MergeRequest.Number, "this merge request was merged — reload")
187 case domain.MergeRequestClosed:
188 return l.mrNotice(ctx, ev.MergeRequest.Number, "this merge request was closed — reload")
189 case domain.MergeRequestReady:
190 return l.mrNotice(ctx, ev.MergeRequest.Number, "this merge request is ready for review — reload")
191 case domain.CommentEdited:
192 return l.mrNotice(ctx, ev.Number, "a note was edited — reload")
193 case domain.CommentDeleted:
194 return l.mrNotice(ctx, ev.Number, "a note was deleted — reload")
195 case domain.CommentResolved:
196 return l.mrNotice(ctx, ev.Number, "a thread was resolved — reload")
197 case domain.CommentAdded:
198 return l.mrNotice(ctx, ev.Number, "a comment was added — reload")
199 case domain.IssueOpened:
200 return l.issueNotice(ctx, ev.Issue.Number, "an issue was opened — reload")
201 case domain.IssueClosed:
202 return l.issueNotice(ctx, ev.Issue.Number, "this issue was closed — reload")
203 case domain.IssueReopened:
204 return l.issueNotice(ctx, ev.Issue.Number, "this issue was reopened — reload")
205 case domain.IssueCommented:
206 return l.issueNotice(ctx, ev.Number, "a comment was added — reload")
207 case domain.PushEvent:
208 switch l.view {
209 case "mr":
210 return l.mrPushed(ctx, ev)
211 case "mrs":
212 return nil, nil
213 }
214 switch l.view {
215 case "refs":
216 return l.refs(ctx, we.ID)
217 case "commits":
218 return l.banner(ctx, ev)
219 }
220 return l.lastPush(ctx)
221 }
222 return nil, nil
223}
224
225// lastPush is the "last push" fact of the summary.
226func (l *liveStream) lastPush(ctx context.Context) ([]fragment, error) {
227 view, err := l.repoByName(ctx, l.repo)
228 if err != nil {
229 return nil, err
230 }
231 f, err := renderFragment(ctx, "repo-last-push", FactValue(lastPushFact(view, l.s.cfg.Now())))
232 return []fragment{f}, err
233}
234
235// refs is the table of the refs page and the facts of its rail. The fragments
236// of one event are built once for all the watchers of the repository that
237// see the same things (see fanout); eventID names the event.
238func (l *liveStream) refs(ctx context.Context, eventID string) ([]fragment, error) {
239 key := fanKey{repo: l.repo, event: eventID, owner: SignedIn(ctx)}
240 return l.s.refsFan.do(ctx, key, func() ([]fragment, error) {
241 d, err := l.s.loadRepo(ctx, l.repo)
242 if err != nil {
243 return nil, err
244 }
245 p := l.s.refsPage(ctx, d)
246 body, err := renderFragment(ctx, "refs-list", RefsBody(p))
247 if err != nil {
248 return nil, err
249 }
250 kv, err := renderFragment(ctx, "refs-list", RefsKV(p))
251 return []fragment{body, kv}, err
252 })
253}
254
255const (
256 // fanTTL is how long the fragments of an event are kept for the watchers
257 // that are told of it a little later than the first.
258 fanTTL = 2 * time.Second
259 // fanMax bounds the events kept; past it an event is built for each
260 // watcher, as without sharing.
261 fanMax = 64
262)
263
264// fanKey says which watchers may share the fragments of an event: those of
265// the same repository, for the same event, and of the same class. The class
266// is anonymous or owner (a session is the owner's alone, ADR 0018): what is
267// built as the owner can show what an anonymous visitor may not see, so the
268// two never share an entry.
269type fanKey struct {
270 repo, event string
271 owner bool
272}
273
274type fanEntry struct {
275 done chan struct{} // closed once frags or err is set
276 frags []fragment
277 err error
278 at time.Time // when it was built; zero while it is
279}
280
281// fanout builds the fragments of an event once for the many watchers that
282// receive it: a push to a repository that 256 pages watch would otherwise
283// load the refs and the commits they designate 256 times. The zero value is
284// ready. A failure is never kept: the next watcher builds it again.
285type fanout struct {
286 mu sync.Mutex
287 m map[fanKey]*fanEntry
288}
289
290// do returns the fragments of key, built by build when nobody has or is
291// building them. build runs with the context of the watcher that calls it,
292// which is of the class of the key; a watcher whose builder failed (its
293// request was cancelled, say) does not fail the others, they build again.
294func (f *fanout) do(ctx context.Context, key fanKey, build func() ([]fragment, error)) ([]fragment, error) {
295 if key.event == "" {
296 return build() // no event to share by
297 }
298 for range 3 {
299 f.mu.Lock()
300 now := time.Now()
301 for k, e := range f.m {
302 if !e.at.IsZero() && now.Sub(e.at) > fanTTL {
303 delete(f.m, k)
304 }
305 }
306 e, shared := f.m[key]
307 if !shared {
308 if len(f.m) >= fanMax {
309 f.mu.Unlock()
310 return build()
311 }
312 if f.m == nil {
313 f.m = make(map[fanKey]*fanEntry)
314 }
315 e = &fanEntry{done: make(chan struct{})}
316 f.m[key] = e
317 }
318 f.mu.Unlock()
319
320 if !shared {
321 return f.lead(key, e, build)
322 }
323 select {
324 case <-e.done:
325 case <-ctx.Done():
326 return nil, ctx.Err()
327 }
328 if e.err == nil {
329 return e.frags, nil
330 }
331 }
332 return build()
333}
334
335// lead builds the entry that the caller registered.
336func (f *fanout) lead(key fanKey, e *fanEntry, build func() ([]fragment, error)) (frags []fragment, err error) {
337 err = errors.New("web: fragments were not built") // a panic of build
338 defer func() {
339 f.mu.Lock()
340 e.frags, e.err, e.at = frags, err, time.Now()
341 if err != nil && f.m[key] == e {
342 delete(f.m, key)
343 }
344 f.mu.Unlock()
345 close(e.done)
346 }()
347 return build()
348}
349
350// banner tells the commits page how many commits its branch got since the
351// head it lists.
352//
353//hexcheck:ignore suspicious-business-logic-in-adapter counts the commits of a page for the banner text; display only
354func (l *liveStream) banner(ctx context.Context, p domain.PushEvent) ([]fragment, error) {
355 if l.ref == "" || !isFullID(l.head) {
356 return nil, nil
357 }
358 tip, touched := "", false
359 for _, u := range p.Updates {
360 if u.Ref == "refs/heads/"+l.ref {
361 tip, touched = u.New, true
362 if u.IsDelete() {
363 tip = ""
364 }
365 }
366 }
367 switch {
368 case !touched:
369 return nil, nil
370 case tip == "":
371 return gap, nil // the branch was deleted
372 }
373 text := ""
374 if tip != l.head {
375 name, err := domain.ParseRepoName(l.repo)
376 if err != nil {
377 return nil, err
378 }
379 res, err := l.s.commitList(ctx, name, tip, "", pageSize, "")
380 if err != nil {
381 return nil, err
382 }
383 // The count is approximate for a history with merges: the log is
384 // ordered by date, not as "head..tip", so commits that are not new may
385 // be counted before the head is met. And one page of the log being
386 // all that is read, "100+" cannot be told from a rewritten branch
387 // whose old head is further back: the use cases have no "is ancestor".
388 n, found := 0, false
389 for _, c := range res.Commits {
390 if c.ID.String() == l.head {
391 found = true
392 break
393 }
394 n++
395 }
396 switch {
397 case found:
398 text = newCommits(strconv.Itoa(n), n)
399 case res.NextPageToken != "":
400 text = newCommits(strconv.Itoa(n)+"+", 2)
401 default:
402 text = "the branch was rewritten — reload"
403 }
404 }
405 c := LiveSlot("commits-head")
406 if text != "" {
407 c = CommitsBanner(text, repoHref(l.repo, "commits", l.ref))
408 }
409 f, err := renderFragment(ctx, "commits-head", c)
410 return []fragment{f}, err
411}
412
413// newCommits is the text of the banner: "1 new commit — reload".
414func newCommits(count string, n int) string {
415 if n == 1 {
416 return count + " new commit — reload"
417 }
418 return count + " new commits — reload"
419}
420
421// isFullID reports whether s is a full commit id.
422func isFullID(s string) bool { return len(s) == 40 && isCommitID(s) }
423
424// mrNotice tells a merge request page that something happened to the request
425// it shows, or the list that its content changed. Only a fixed sentence is sent:
426// the text of a comment or a title never leaves through the stream.
427func (l *liveStream) mrNotice(ctx context.Context, number int64, text string) ([]fragment, error) {
428 switch l.view {
429 case "mrs":
430 text = "merge requests changed — reload"
431 return l.notice(ctx, text, mrBase(l.repo))
432 case "mr":
433 if strconv.FormatInt(number, 10) != l.n {
434 return nil, nil
435 }
436 return l.notice(ctx, text, mrHref(l.repo, number))
437 }
438 return nil, nil
439}
440
441// issueNotice tells an issue page that something happened to the issue it
442// shows, or the list that its content changed. Like mrNotice it sends a fixed
443// sentence only, in the same slot.
444func (l *liveStream) issueNotice(ctx context.Context, number int64, text string) ([]fragment, error) {
445 switch l.view {
446 case "issues":
447 return l.notice(ctx, "issues changed — reload", issuesBase(l.repo))
448 case "issue":
449 if strconv.FormatInt(number, 10) != l.n {
450 return nil, nil
451 }
452 return l.notice(ctx, text, issueHref(l.repo, number))
453 }
454 return nil, nil
455}
456
457// mrPushed announces a push to the source branch of the merge request shown.
458func (l *liveStream) mrPushed(ctx context.Context, p domain.PushEvent) ([]fragment, error) {
459 if l.ref == "" {
460 return nil, nil
461 }
462 for _, u := range p.Updates {
463 if u.Ref == "refs/heads/"+l.ref {
464 n, err := strconv.ParseInt(l.n, 10, 64)
465 if err != nil {
466 return nil, nil
467 }
468 return l.notice(ctx, "the branch changed — reload", mrHref(l.repo, n))
469 }
470 }
471 return nil, nil
472}
473
474func (l *liveStream) notice(ctx context.Context, text, href string) ([]fragment, error) {
475 f, err := renderFragment(ctx, "mr-notice", MRNotice(text, href))
476 return []fragment{f}, err
477}
478
479// runEvent updates the pages of CI that a RunUpdated changes: the row of the
480// list of runs, the recent actions of the summary, or the body of the run page
481// shown. Only a state is sent: the
482// events carry no log and no text of the workflow.
483func (l *liveStream) runEvent(ctx context.Context, ev domain.RunUpdated) ([]fragment, error) {
484 if l.s.cfg.Runs == nil {
485 return nil, nil
486 }
487 switch l.view {
488 case "actions":
489 f, err := renderFragment(ctx, "ci-run", RunListRow(l.s.runRow(l.repo, ev.Run, l.s.cfg.Now())))
490 if err != nil {
491 return nil, err
492 }
493 if ev.Run.Status == domain.CIQueued {
494 f.html = strings.Replace(f.html, "<tr id=", "<tr data-new id=", 1)
495 }
496 return []fragment{f}, nil
497 case "summary":
498 return l.recentActions(ctx)
499 case "run":
500 if strconv.FormatInt(ev.Run.ID, 10) != l.n {
501 return nil, nil
502 }
503 return l.runBody(ctx)
504 case "mr":
505 ofHead := ev.Run.CheckSha() == l.head
506 l.rememberRun(ev.Run.ID, ofHead)
507 if !ofHead {
508 return nil, nil
509 }
510 return l.mrChecks(ctx)
511 }
512 return nil, nil
513}
514
515// rememberRun records whether a run is one of the head of the page.
516func (l *liveStream) rememberRun(id int64, ofHead bool) {
517 if l.headRuns == nil || len(l.headRuns) >= maxHeadRuns {
518 l.headRuns = make(map[int64]bool)
519 }
520 l.headRuns[id] = ofHead
521}
522
523// jobOfHead tells whether the run of a job is one of the head of the page,
524// reading the run the first time.
525func (l *liveStream) jobOfHead(ctx context.Context, job domain.CIJob) (bool, error) {
526 if ofHead, known := l.headRuns[job.RunID]; known {
527 return ofHead, nil
528 }
529 run, err := l.s.cfg.Runs.GetRun(ctx, principal(ctx), job.Repo, job.RunID, 0)
530 switch {
531 case errors.Is(err, domain.ErrNotFound):
532 return false, nil
533 case err != nil:
534 return false, err
535 }
536 ofHead := run.Run.CheckSha() == l.head
537 l.rememberRun(job.RunID, ofHead)
538 return ofHead, nil
539}
540
541// mrChecks is the checks block of the merge request page, read again for the
542// head the page shows: a run or a job of that head changed. A job event does
543// not name its commit; reading the checks of the head is one store read, and
544// a job of another commit leaves them as they were.
545func (l *liveStream) mrChecks(ctx context.Context) ([]fragment, error) {
546 if !isFullID(l.head) {
547 return nil, nil
548 }
549 name, err := domain.ParseRepoName(l.repo)
550 if err != nil {
551 return nil, nil
552 }
553 repo, err := l.s.repoRecord(ctx, name)
554 switch {
555 case errors.Is(err, domain.ErrNotFound):
556 return gap, nil
557 case err != nil:
558 return nil, err
559 }
560 checks := l.s.checksFor(ctx, name, l.head)
561 f, err := renderFragment(ctx, "ci-run", MRChecks(checksSection(repo, checks)))
562 if err != nil {
563 return nil, err
564 }
565 sum, err := renderFragment(ctx, "ci-run", MRChecksSum(checksSummary(name, l.head, checks)))
566 return []fragment{f, sum}, err
567}
568
569// recentActions is the "Recent actions" block of the summary rail, read again
570// as the principal of the stream: the newest runs change with every event, and
571// a viewer who cannot read CI any more gets the block hidden and empty.
572func (l *liveStream) recentActions(ctx context.Context) ([]fragment, error) {
573 name, err := domain.ParseRepoName(l.repo)
574 if err != nil {
575 return nil, nil
576 }
577 a, err := l.s.listRecentActions(ctx, name, l.s.cfg.Now())
578 switch {
579 case errors.Is(err, domain.ErrNotFound):
580 a = RecentActions{AllHref: actionsBase(l.repo)}
581 case err != nil:
582 return nil, err
583 }
584 f, err := renderFragment(ctx, "ci-run", RecentActionsBlock(a))
585 return []fragment{f}, err
586}
587
588// jobEvent updates the body of the run page and the steps of the job page.
589func (l *liveStream) jobEvent(ctx context.Context, ev domain.JobUpdated) ([]fragment, error) {
590 if l.s.cfg.Runs == nil {
591 return nil, nil
592 }
593 switch l.view {
594 case "run":
595 if strconv.FormatInt(ev.Job.RunID, 10) != l.n {
596 return nil, nil
597 }
598 return l.runBody(ctx)
599 case "job":
600 if ev.Job.ID != l.job {
601 return nil, nil
602 }
603 return l.jobState(ctx)
604 case "mr":
605 if !isFullID(l.head) {
606 return nil, nil
607 }
608 ofHead, err := l.jobOfHead(ctx, ev.Job)
609 if err != nil || !ofHead {
610 return nil, err
611 }
612 return l.mrChecks(ctx)
613 }
614 return nil, nil
615}
616
617// runBody renders again the jobs of the run page.
618func (l *liveStream) runBody(ctx context.Context) ([]fragment, error) {
619 id, err := strconv.ParseInt(l.n, 10, 64)
620 if err != nil {
621 return nil, nil
622 }
623 d, err := l.s.loadRepo(ctx, l.repo)
624 if err != nil {
625 return nil, err
626 }
627 p, err := l.s.runPage(ctx, d, id, 0)
628 if err != nil {
629 return nil, err
630 }
631 body, err := renderFragment(ctx, "ci-run", runBodyFragment(p))
632 if err != nil {
633 return nil, err
634 }
635 // The buttons depend on the state: the page fetches itself again when the
636 // run completed, so that the owner gets the right ones.
637 if !p.Active {
638 return []fragment{body, {event: "gap", html: "gap"}}, nil
639 }
640 return []fragment{body}, nil
641}
642
643// jobState tells the job page that its state or its steps changed: the page
644// reloads the part of the log after the last line it holds (ermite.js).
645func (l *liveStream) jobState(ctx context.Context) ([]fragment, error) {
646 if l.sha == "" {
647 return gap, nil // the stream was set up without the commit of the run
648 }
649 d, err := l.s.loadRepo(ctx, l.repo)
650 if err != nil {
651 return nil, err
652 }
653 view, err := l.s.cfg.Runs.GetJob(ctx, principal(ctx), d.repo.Name, l.job, 0)
654 if err != nil {
655 return nil, err
656 }
657 now := l.s.cfg.Now()
658 row := l.s.jobRow(d.name(), view, now)
659 f, err := renderFragment(ctx, "ci-job", JobHead(row, view.Lines))
660 if err != nil {
661 return nil, err
662 }
663 steps, err := renderFragment(ctx, "ci-job", JobSteps(stepViews(view, now)))
664 if err != nil {
665 return nil, err
666 }
667 anns, more, err := l.s.jobAnnotations(ctx, d, view.RunID, l.sha, view, 0)
668 if err != nil {
669 return nil, err
670 }
671 notes, err := renderFragment(ctx, "ci-job", JobAnnotations(anns, more))
672 if err != nil {
673 return nil, err
674 }
675 return []fragment{f, steps, notes}, nil
676}