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