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