1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package web
4
5import (
6 "context"
7 "errors"
8 "fmt"
9 "io"
10 "net/http"
11 "net/url"
12 "strconv"
13 "strings"
14 "sync"
15 "time"
16
17 "github.com/bnema/zerowrap"
18
19 "bnema.dev/ermite/internal/clientlimit"
20 "bnema.dev/ermite/internal/domain"
21 "bnema.dev/ermite/internal/ports"
22)
23
24// Live updates. A page that can change under the eyes of its reader opens
25// GET /-/events (the Live component of the hint badge). The handler follows
26// EventService.Watch as the principal of the request and turns each event into
27// the HTML fragments the page shows, sent as Server-Sent Events:
28//
29// GET /-/events?topic=home
30// GET /-/events?topic=owner&view=inbox[&repo=<name>] (the signed-in owner only)
31// GET /-/events?topic=repo:<name>&view=summary|refs
32// GET /-/events?topic=repo:<name>&view=commits&ref=<branch>&head=<sha>
33// GET /-/events?topic=repo:<name>&view=mrs
34// GET /-/events?topic=repo:<name>&view=mr&n=<number>&ref=<source branch>&head=<sha>&target=<target branch>
35// GET /-/events?topic=repo:<name>&view=issues
36// GET /-/events?topic=repo:<name>&view=issue&n=<number>
37// GET /-/events?topic=repo:<name>&view=actions
38// GET /-/events?topic=repo:<name>&view=run&n=<run id>
39// GET /-/events?topic=repo:<name>&view=job&job=<job id>
40//
41// topic is the Watch topic. view says which fragments of a repository the
42// page shows, so that the server renders only those (summary when absent).
43// On the commits page, ref is the branch it lists and head the commit it
44// lists first: the banner counts the commits between that head and the new
45// tip, so it stays right however many pushes came in and however often the
46// stream reconnected. A Last-Event-ID header resumes the stream
47// (resume_after). With no header, after=<event id> supplies the resume cursor
48// for a replacement stream. The header wins. Both use Watch's parseID
49// validation: an invalid or expired cursor produces a gap.
50//
51// Events, each with the id of the Watch event it comes from:
52//
53// home-list data: <li id="repo-<name>">…</li>, or <li id="repo-<name>" data-gone></li>
54// for a repository that is deleted or no longer visible
55// Activity rows carry data-activity: push, MR lifecycle/comment,
56// or issue lifecycle/comment (never pending moderation or CI).
57// repo-last-push data: <dd id="last-push-<name>">…</dd>
58// refs-list data: <tbody id="ref-rows">…</tbody>, and again with
59// <dl id="refs-kv">…</dl>, the facts of the rail
60// commits-head data: <div id="commits-head">N new commits — reload</div>
61// mr-mergeability data: <div id="mr-mergeability">…</div>, on the merge request pages:
62// the badge of the merge box (a slot like mr-notice), when the
63// mergeability of the request shown settles or a push moves its
64// source or target
65// mr-head data: <div id="mr-head"></div>; the source branch moved.
66// ermiteMR fetches the current page and refreshes its head-dependent
67// regions in place, preserving the reader's position and drafts.
68// The replacement stream uses the new head for CI filtering.
69// mr-notice data: <div id="mr-notice">…</div>, on the merge request and issue pages:
70// something changed, with the link that reloads the page
71// ci-run data: <tr id="run-<id>">…</tr> (the list of runs),
72// <section id="recent-actions">…</section> (the rail of the summary) or
73// <div id="run-body">…</div> (one run)
74// ci-job data: <h2 id="job-head">…</h2> and <ol id="job-steps">…</ol>; the log
75// itself is never sent in this stream: the page follows it on its own
76// feed, /<repo>/actions/runs/<id>/job/<job>/stream?from=<line>
77// gap data: gap; events were lost, or what the page shows is gone or
78// changed: the page fetches itself again (ermiteLive in ermite.js)
79//
80// The data of an event is one fragment, a top-level element with an id, one
81// "data:" line per line of HTML; the page swaps the element that has that id
82// (ermiteLive.swap). The slots of the banners (commits-head, mr-notice) are
83// status regions that stay in the page: only their children are replaced, so
84// that a screen reader announces the banner.
85// An event that changes nothing on the page is still sent as an id alone, so
86// that a reconnection resumes after it.
87
88// DefaultHeartbeat is how often an idle stream gets a comment line, so that
89// proxies keep it open and a dead peer is noticed.
90const DefaultHeartbeat = 25 * time.Second
91
92const (
93 // writeWindow bounds one write to the visitor, whatever the write timeout
94 // of the server is.
95 writeWindow = 30 * time.Second
96 // retryAfter is the delay asked from a visitor refused for lack of room.
97 retryAfter = 10
98)
99
100// ownerTopic is the stream of the inbox: the merge requests and issues of every
101// repository, for the owner.
102const ownerTopic = "owner"
103
104// parseTopic reads "home", "owner" or "repo:<name>"; repo is "" for home and
105// owner.
106func parseTopic(topic string) (repo string, ok bool) {
107 if topic == "home" || topic == ownerTopic {
108 return "", true
109 }
110 if name, found := strings.CutPrefix(topic, "repo:"); found && name != "" {
111 return name, true
112 }
113 return "", false
114}
115
116// ownerStreamOK tells whether the request may follow the owner topic: the
117// signed-in owner, with the inbox view and a valid repository filter, on a
118// server that has the inbox.
119func (s *Server) ownerStreamOK(ctx context.Context, q url.Values) bool {
120 if s.cfg.Inbox == nil || !SignedIn(ctx) || q.Get("view") != "inbox" {
121 return false
122 }
123 if f := q.Get("repo"); f != "" {
124 if _, err := domain.ParseRepoName(f); err != nil {
125 return false
126 }
127 }
128 return true
129}
130
131// events is GET /-/events.
132func (s *Server) events(w http.ResponseWriter, r *http.Request) {
133 if s.cfg.Events == nil {
134 http.NotFound(w, r)
135 return
136 }
137 q := r.URL.Query()
138 topic := q.Get("topic")
139 repo, ok := parseTopic(topic)
140 if topic == ownerTopic {
141 // Anyone but the signed-in owner gets the answer of an unknown topic.
142 ok = s.ownerStreamOK(r.Context(), q)
143 repo = ""
144 }
145 if !ok {
146 denyStream(w, http.StatusBadRequest, "topic must be home or repo:<name>")
147 return
148 }
149 ctx, cancel := context.WithCancel(r.Context())
150 defer cancel()
151 log := zerowrap.FromCtx(ctx)
152
153 if repo != "" {
154 // A repository the visitor cannot see is a missing one (AC2).
155 cctx, ccancel := context.WithTimeout(ctx, requestTimeout)
156 err := s.canSee(cctx, repo)
157 if err == nil {
158 if v := q.Get("view"); v == "issues" || v == "issue" {
159 err = s.issuesOn(cctx, repo)
160 }
161 }
162 ccancel()
163 if err != nil {
164 s.denyEvents(ctx, w, err)
165 return
166 }
167 }
168
169 // The commit of the run of a job page is read once here: the annotations of
170 // the live job fragments link to files at it, and it never changes, so a
171 // job update does not read the run again. A run that is not there or not
172 // readable refuses the stream like a missing repository.
173 var sha string
174 if repo != "" && q.Get("view") == "job" && s.cfg.Runs != nil {
175 cctx, ccancel := context.WithTimeout(ctx, requestTimeout)
176 var err error
177 sha, err = s.jobRunSha(cctx, repo, q.Get("job"))
178 ccancel()
179 if err != nil {
180 s.denyEvents(ctx, w, err)
181 return
182 }
183 }
184
185 // The watch runs in a goroutine and hands each event to this one through
186 // an unbuffered channel: a visitor that reads slowly holds the watch in
187 // its send, falls behind the buffer of the use case, and gets a gap.
188 msgs := make(chan recvd)
189 started := make(chan struct{})
190 var startOnce sync.Once
191 start := func() { startOnce.Do(func() { close(started) }) }
192 finished := make(chan struct{})
193 resume := r.Header.Get("Last-Event-ID")
194 if resume == "" {
195 resume = q.Get("after")
196 }
197 params := ports.WatchParams{
198 Topics: []string{topic}, ResumeAfter: resume,
199 // The visitor is the client of the watch, for the limit per client.
200 Client: clientlimit.Key(clientlimit.ClientAddr(r, s.cfg.TrustedProxies)),
201 Ready: start,
202 }
203 go func() {
204 defer close(finished)
205 defer close(msgs)
206 // net/http only recovers the handler goroutine: a panic in the watch
207 // would end the process. It ends this stream as an error instead.
208 defer func() {
209 if v := recover(); v != nil {
210 log.Error().Interface("panic", v).Msg("event watch panicked")
211 select {
212 case msgs <- recvd{err: fmt.Errorf("watch panicked: %v", v)}:
213 case <-ctx.Done():
214 }
215 }
216 }()
217 err := s.cfg.Events.Watch(ctx, principal(ctx), params, func(we ports.WatchEvent) error {
218 start()
219 select {
220 case msgs <- recvd{ev: we}:
221 return nil
222 case <-ctx.Done():
223 return ctx.Err()
224 }
225 })
226 if err != nil && ctx.Err() == nil {
227 select {
228 case msgs <- recvd{err: err}:
229 case <-ctx.Done():
230 }
231 }
232 }()
233 defer func() {
234 cancel()
235 <-finished
236 }()
237
238 // The watch is registered (or refused, which ends it before): only then
239 // does the answer commit to 200.
240 var first *recvd
241 select {
242 case <-started:
243 case m, open := <-msgs:
244 if !open {
245 return // the server is stopping
246 }
247 if m.err != nil {
248 s.denyEvents(ctx, w, m.err)
249 return
250 }
251 first = &m
252 case <-ctx.Done():
253 return
254 }
255
256 rc := http.NewResponseController(w)
257 hdr := w.Header()
258 hdr.Set("Content-Type", "text/event-stream")
259 cacheHeaders(ctx, hdr, "no-cache, no-transform", sessionCache+", no-transform")
260 if hdr.Get("Cache-Control") == noStore { // an invalid cookie was cleared
261 hdr.Set("Cache-Control", noStore+", no-transform")
262 }
263 hdr.Set("X-Accel-Buffering", "no")
264 _ = rc.SetWriteDeadline(time.Now().Add(writeWindow))
265 w.WriteHeader(http.StatusOK)
266 out := &sseWriter{w: w, rc: rc}
267 if err := out.raw(": connected\n\n"); err != nil {
268 return
269 }
270
271 st := &liveStream{s: s, repo: repo, view: q.Get("view"), ref: q.Get("ref"), head: q.Get("head"), n: q.Get("n"), target: q.Get("target"), job: q.Get("job"), sha: sha}
272 if topic == ownerTopic {
273 st.filter = q.Get("repo")
274 }
275 handle := func(m recvd) bool {
276 if m.err != nil {
277 if ctx.Err() == nil && !errors.Is(m.err, context.Canceled) {
278 log.Debug().Err(m.err).Str("topic", topic).Msg("event stream ended")
279 }
280 return false
281 }
282 if !liveEvent(m.ev) {
283 return true // nothing to show, not even an id to resume after
284 }
285 // An owner's stream ends when the session does, before it sends
286 // anything more: an event can carry what only the owner may see, and
287 // waiting for the next heartbeat would leave a window.
288 if SignedIn(ctx) && !s.sessionStillValid(ctx, r) {
289 return false
290 }
291 rctx, rcancel := context.WithTimeout(ctx, requestTimeout)
292 frags, err := st.render(rctx, m.ev)
293 rcancel()
294 if err != nil {
295 if ctx.Err() != nil {
296 return false
297 }
298 // What the page shows can no longer be told from the events:
299 // have it fetch itself again.
300 log.Warn().Err(err).Str("topic", topic).Msg("render live fragment")
301 frags = gap
302 }
303 return out.send(m.ev.ID, frags) == nil
304 }
305 if first != nil && !handle(*first) {
306 return
307 }
308
309 beat := s.cfg.Heartbeat
310 if beat <= 0 {
311 beat = DefaultHeartbeat
312 }
313 tick := time.NewTicker(beat)
314 defer tick.Stop()
315 for {
316 select {
317 case m, open := <-msgs:
318 if !open || !handle(m) {
319 return
320 }
321 case <-tick.C:
322 // A stream opened by the owner ends when the session does
323 // (sign out everywhere, expiry): the page reconnects as it can.
324 if SignedIn(ctx) && !s.sessionStillValid(ctx, r) {
325 return
326 }
327 if out.raw(": heartbeat\n\n") != nil {
328 return
329 }
330 case <-ctx.Done():
331 return
332 }
333 }
334}
335
336// recvd is a message of the Watch stream, or how it failed.
337type recvd struct {
338 ev ports.WatchEvent
339 err error
340}
341
342// canSee checks that the caller of the request may see the repository named in
343// a topic: one it may not is as missing as one that does not exist.
344func (s *Server) canSee(ctx context.Context, rawName string) error {
345 name, err := domain.ParseRepoName(rawName)
346 if err != nil {
347 return err
348 }
349 _, err = s.cfg.Repos.Get(ctx, principal(ctx), name)
350 return err
351}
352
353// issuesOn is domain.ErrNotFound unless the repository has issues on and
354// s.cfg.Issues is set: the same answer as for a repository the visitor cannot see.
355//
356//hexcheck:ignore suspicious-business-logic-in-adapter maps domain RepoSettings.IssuesEnabled to a not-found stream; decides nothing itself
357func (s *Server) issuesOn(ctx context.Context, rawName string) error {
358 if s.cfg.Issues == nil {
359 return domain.ErrNotFound
360 }
361 name, err := domain.ParseRepoName(rawName)
362 if err != nil {
363 return err
364 }
365 v, err := s.cfg.Repos.Get(ctx, principal(ctx), name)
366 if err != nil {
367 return err
368 }
369 if !v.EffectiveSettings().IssuesEnabled() {
370 return domain.ErrNotFound
371 }
372 return nil
373}
374
375// denyStream answers a request that gets no stream.
376func denyStream(w http.ResponseWriter, status int, msg string) {
377 w.Header().Set("Cache-Control", "no-store")
378 http.Error(w, msg, status)
379}
380
381// denyEvents maps an error of a use case, before the stream started, to a
382// status. Not found and invalid are 404 whatever the reason: a private
383// repository is a missing one.
384func (s *Server) denyEvents(ctx context.Context, w http.ResponseWriter, err error) {
385 switch {
386 case errors.Is(err, context.Canceled):
387 // The visitor is gone.
388 case errors.Is(err, domain.ErrNotFound), domain.IsInvalid(err):
389 denyStream(w, http.StatusNotFound, "not found")
390 case errors.Is(err, domain.ErrResourceExhausted):
391 w.Header().Set("Retry-After", strconv.Itoa(retryAfter))
392 denyStream(w, http.StatusServiceUnavailable, "too many event streams, retry later")
393 case errors.Is(err, domain.ErrUnauthenticated):
394 denyStream(w, http.StatusUnauthorized, "unauthorized")
395 default:
396 log := zerowrap.FromCtx(ctx)
397 log.Error().Err(err).Msg("open event stream")
398 denyStream(w, http.StatusInternalServerError, "internal error")
399 }
400}
401
402// fragment is one SSE event to send: the name and the HTML of the fragment.
403type fragment struct{ event, html string }
404
405// sseWriter writes events to a visitor, flushing each, under a deadline of
406// its own so that the write timeout of the server cannot end the stream.
407type sseWriter struct {
408 w http.ResponseWriter
409 rc *http.ResponseController
410}
411
412func (o *sseWriter) raw(s string) error {
413 _ = o.rc.SetWriteDeadline(time.Now().Add(writeWindow))
414 if _, err := io.WriteString(o.w, s); err != nil {
415 return err
416 }
417 return o.rc.Flush()
418}
419
420// send writes the fragments of one Watch event under its id. An event that
421// has no fragment still carries its id, so that a reconnection resumes after
422// it.
423func (o *sseWriter) send(id string, frags []fragment) error {
424 if len(frags) == 0 {
425 if id == "" {
426 return nil
427 }
428 return o.raw(formatEvent(id, "", "", false))
429 }
430 for _, f := range frags {
431 if err := o.raw(formatEvent(id, f.event, f.html, true)); err != nil {
432 return err
433 }
434 }
435 return nil
436}
437
438// formatEvent is one SSE event. Each line of data is a "data:" field, and a
439// browser joins them back with newlines.
440func formatEvent(id, event, data string, hasData bool) string {
441 var b strings.Builder
442 if event != "" {
443 b.WriteString("event: " + event + "\n")
444 }
445 if id != "" {
446 b.WriteString("id: " + id + "\n")
447 }
448 if hasData {
449 data = strings.ReplaceAll(strings.ReplaceAll(data, "\r\n", "\n"), "\r", "\n")
450 for line := range strings.SplitSeq(data, "\n") {
451 b.WriteString("data: " + line + "\n")
452 }
453 }
454 b.WriteString("\n")
455 return b.String()
456}
457
458// liveURL is the stream a page follows: the topic and, for a repository
459// page, what it shows (see the comment on events).
460func liveURL(topic, query string) string {
461 u := "/-/events?topic=" + url.QueryEscape(topic)
462 if query != "" {
463 u += "&" + query
464 }
465 return u
466}