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