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