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