1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package web
4
5import (
6 "context"
7 "errors"
8 "io"
9 "net/http"
10 "net/url"
11 "strconv"
12 "strings"
13 "time"
14
15 "connectrpc.com/connect"
16 "github.com/bnema/zerowrap"
17
18 pb "bnema.dev/ermite-api/gen/ermite"
19)
20
21// Live updates. A page that can change under the eyes of its reader opens
22// GET /-/events (the Live component of the status bar). The handler follows
23// EventService.Watch as the visitor of the request and turns each event into
24// the HTML fragments the page shows, sent as Server-Sent Events:
25//
26// GET /-/events?topic=home
27// GET /-/events?topic=repo:<name>&view=summary|refs
28// GET /-/events?topic=repo:<name>&view=commits&ref=<branch>&head=<sha>
29// GET /-/events?topic=repo:<name>&view=mrs
30// GET /-/events?topic=repo:<name>&view=mr&n=<number>&ref=<source branch>
31//
32// topic is the Watch topic. view says which fragments of a repository the
33// page shows, so that the server renders only those (summary when absent).
34// On the commits page, ref is the branch it lists and head the commit it
35// lists first: the banner counts the commits between that head and the new
36// tip, so it stays right however many pushes came in and however often the
37// stream reconnected. A Last-Event-ID header resumes the stream
38// (resume_after).
39//
40// Events, each with the id of the Watch event it comes from:
41//
42// home-list data: <tr id="repo-<name>">…</tr>, or <tr id="repo-<name>" data-gone></tr>
43// for a repository that is deleted or no longer visible
44// repo-last-push data: <dd id="last-push-<name>">…</dd>
45// refs-list data: <tbody id="ref-rows">…</tbody>, and again with
46// <dl id="refs-kv">…</dl>, the facts of the rail
47// commits-head data: <div id="commits-head">N new commits — reload</div>
48// mr-notice data: <div id="mr-notice">…</div>, on the merge request pages:
49// something changed, with the link that reloads the page
50// gap data: gap; events were lost, or what the page shows is gone or
51// changed: the page fetches itself again (ermiteLive in ermite.js)
52//
53// The data of an event is one fragment, a top-level element with an id, one
54// "data:" line per line of HTML; the page swaps the element that has that id.
55// An event that changes nothing on the page is still sent as an id alone, so
56// that a reconnection resumes after it.
57
58// DefaultHeartbeat is how often an idle stream gets a comment line, so that
59// proxies keep it open and a dead peer is noticed.
60const DefaultHeartbeat = 25 * time.Second
61
62const (
63 // startGrace is how long the handler waits after the Watch call answered
64 // for an early failure (too many watchers) before it commits to 200: the
65 // Connect protocol reports it in the stream, after the headers.
66 startGrace = 100 * time.Millisecond
67 // writeWindow bounds one write to the visitor, whatever the write timeout
68 // of the server is.
69 writeWindow = 30 * time.Second
70 // retryAfter is the delay asked from a visitor refused for lack of room.
71 retryAfter = 10
72)
73
74// parseTopic reads "home" or "repo:<name>"; repo is "" for home.
75func parseTopic(topic string) (repo string, ok bool) {
76 if topic == "home" {
77 return "", true
78 }
79 if name, found := strings.CutPrefix(topic, "repo:"); found && name != "" {
80 return name, true
81 }
82 return "", false
83}
84
85// events is GET /-/events.
86func (s *Server) events(w http.ResponseWriter, r *http.Request) {
87 if s.cfg.Events == nil {
88 http.NotFound(w, r)
89 return
90 }
91 q := r.URL.Query()
92 topic := q.Get("topic")
93 repo, ok := parseTopic(topic)
94 if !ok {
95 denyStream(w, http.StatusBadRequest, "topic must be home or repo:<name>")
96 return
97 }
98 ctx, cancel := context.WithCancel(r.Context())
99 defer cancel()
100 log := zerowrap.FromCtx(ctx)
101
102 if repo != "" {
103 // A repository the visitor cannot see is a missing one (AC2).
104 cctx, ccancel := context.WithTimeout(ctx, requestTimeout)
105 _, err := s.cfg.Repos.GetRepo(cctx, connect.NewRequest(&pb.GetRepoRequest{Repo: repo}))
106 ccancel()
107 if err != nil {
108 s.denyEvents(ctx, w, err)
109 return
110 }
111 }
112
113 stream, err := s.cfg.Events.Watch(ctx, connect.NewRequest(&pb.WatchRequest{
114 Topics: []string{topic}, ResumeAfter: r.Header.Get("Last-Event-ID"),
115 }))
116 if err != nil {
117 s.denyEvents(ctx, w, err)
118 return
119 }
120 // Wait for the answer of the API, which comes once the watch is registered
121 // (or refused), then read the stream from a goroutine.
122 _ = stream.ResponseHeader()
123 msgs := make(chan recvd)
124 done := make(chan struct{})
125 go func() {
126 defer close(done)
127 defer close(msgs)
128 for stream.Receive() {
129 select {
130 case msgs <- recvd{ev: stream.Msg().GetEvent()}:
131 case <-ctx.Done():
132 return
133 }
134 }
135 if err := stream.Err(); err != nil {
136 select {
137 case msgs <- recvd{err: err}:
138 case <-ctx.Done():
139 }
140 }
141 }()
142 defer func() {
143 cancel()
144 _ = stream.Close()
145 <-done
146 }()
147
148 var first *recvd
149 grace := time.NewTimer(startGrace)
150 select {
151 case m, open := <-msgs:
152 grace.Stop()
153 if !open {
154 return // the server is stopping
155 }
156 first = &m
157 case <-grace.C:
158 case <-ctx.Done():
159 grace.Stop()
160 return
161 }
162 if first != nil && first.err != nil {
163 s.denyEvents(ctx, w, first.err)
164 return
165 }
166
167 rc := http.NewResponseController(w)
168 hdr := w.Header()
169 hdr.Set("Content-Type", "text/event-stream")
170 cacheHeaders(ctx, hdr, "no-cache, no-transform", sessionCache+", no-transform")
171 if hdr.Get("Cache-Control") == noStore { // an invalid cookie was cleared
172 hdr.Set("Cache-Control", noStore+", no-transform")
173 }
174 hdr.Set("X-Accel-Buffering", "no")
175 _ = rc.SetWriteDeadline(time.Now().Add(writeWindow))
176 w.WriteHeader(http.StatusOK)
177 out := &sseWriter{w: w, rc: rc}
178 if err := out.raw(": connected\n\n"); err != nil {
179 return
180 }
181
182 st := &liveStream{s: s, repo: repo, view: q.Get("view"), ref: q.Get("ref"), head: q.Get("head"), n: q.Get("n")}
183 handle := func(m recvd) bool {
184 if m.err != nil {
185 if ctx.Err() == nil && connect.CodeOf(m.err) != connect.CodeCanceled {
186 log.Debug().Err(m.err).Str("topic", topic).Msg("event stream ended")
187 }
188 return false
189 }
190 // An owner's stream ends when the session does, before it sends
191 // anything more: an event can carry what only the owner may see, and
192 // waiting for the next heartbeat would leave a window.
193 if SignedIn(ctx) && !s.sessionStillValid(ctx, r) {
194 return false
195 }
196 rctx, rcancel := context.WithTimeout(ctx, requestTimeout)
197 frags, err := st.render(rctx, m.ev)
198 rcancel()
199 if err != nil {
200 if ctx.Err() != nil {
201 return false
202 }
203 // What the page shows can no longer be told from the events:
204 // have it fetch itself again.
205 log.Warn().Err(err).Str("topic", topic).Msg("render live fragment")
206 frags = gap
207 }
208 return out.send(m.ev.GetId(), frags) == nil
209 }
210 if first != nil && !handle(*first) {
211 return
212 }
213
214 beat := s.cfg.Heartbeat
215 if beat <= 0 {
216 beat = DefaultHeartbeat
217 }
218 tick := time.NewTicker(beat)
219 defer tick.Stop()
220 for {
221 select {
222 case m, open := <-msgs:
223 if !open || !handle(m) {
224 return
225 }
226 case <-tick.C:
227 // A stream opened by the owner ends when the session does
228 // (sign out everywhere, expiry): the page reconnects as it can.
229 if SignedIn(ctx) && !s.sessionStillValid(ctx, r) {
230 return
231 }
232 if out.raw(": heartbeat\n\n") != nil {
233 return
234 }
235 case <-ctx.Done():
236 return
237 }
238 }
239}
240
241// recvd is a message of the Watch stream, or how it failed.
242type recvd struct {
243 ev *pb.Event
244 err error
245}
246
247// denyStream answers a request that gets no stream.
248func denyStream(w http.ResponseWriter, status int, msg string) {
249 w.Header().Set("Cache-Control", "no-store")
250 http.Error(w, msg, status)
251}
252
253// denyEvents maps an error of the API, before the stream started, to a
254// status. Not found and invalid are 404 whatever the reason: a private
255// repository is a missing one.
256func (s *Server) denyEvents(ctx context.Context, w http.ResponseWriter, err error) {
257 switch code := connect.CodeOf(err); {
258 case code == connect.CodeNotFound, code == connect.CodeInvalidArgument:
259 denyStream(w, http.StatusNotFound, "not found")
260 case code == connect.CodeResourceExhausted:
261 w.Header().Set("Retry-After", strconv.Itoa(retryAfter))
262 denyStream(w, http.StatusServiceUnavailable, "too many event streams, retry later")
263 case code == connect.CodeUnauthenticated:
264 denyStream(w, http.StatusUnauthorized, "unauthorized")
265 case errors.Is(err, context.Canceled), code == connect.CodeCanceled:
266 // The visitor is gone.
267 default:
268 log := zerowrap.FromCtx(ctx)
269 log.Error().Err(err).Msg("open event stream")
270 denyStream(w, http.StatusInternalServerError, "internal error")
271 }
272}
273
274// fragment is one SSE event to send: the name and the HTML of the fragment.
275type fragment struct{ event, html string }
276
277// sseWriter writes events to a visitor, flushing each, under a deadline of
278// its own so that the write timeout of the server cannot end the stream.
279type sseWriter struct {
280 w http.ResponseWriter
281 rc *http.ResponseController
282}
283
284func (o *sseWriter) raw(s string) error {
285 _ = o.rc.SetWriteDeadline(time.Now().Add(writeWindow))
286 if _, err := io.WriteString(o.w, s); err != nil {
287 return err
288 }
289 return o.rc.Flush()
290}
291
292// send writes the fragments of one Watch event under its id. An event that
293// has no fragment still carries its id, so that a reconnection resumes after
294// it.
295func (o *sseWriter) send(id string, frags []fragment) error {
296 if len(frags) == 0 {
297 if id == "" {
298 return nil
299 }
300 return o.raw(formatEvent(id, "", "", false))
301 }
302 for _, f := range frags {
303 if err := o.raw(formatEvent(id, f.event, f.html, true)); err != nil {
304 return err
305 }
306 }
307 return nil
308}
309
310// formatEvent is one SSE event. Each line of data is a "data:" field, and a
311// browser joins them back with newlines.
312func formatEvent(id, event, data string, hasData bool) string {
313 var b strings.Builder
314 if event != "" {
315 b.WriteString("event: " + event + "\n")
316 }
317 if id != "" {
318 b.WriteString("id: " + id + "\n")
319 }
320 if hasData {
321 data = strings.ReplaceAll(strings.ReplaceAll(data, "\r\n", "\n"), "\r", "\n")
322 for line := range strings.SplitSeq(data, "\n") {
323 b.WriteString("data: " + line + "\n")
324 }
325 }
326 b.WriteString("\n")
327 return b.String()
328}
329
330// liveURL is the stream a page follows: the topic and, for a repository
331// page, what it shows (see the comment on events).
332func liveURL(topic, query string) string {
333 u := "/-/events?topic=" + url.QueryEscape(topic)
334 if query != "" {
335 u += "&" + query
336 }
337 return u
338}