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