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