package web
import (
"context"
"net/http"
"time"
"craftdesign.group/quarry/internal/streamcap"
)
// relayWakes forwards each value from in to out as a bare wake, never
// blocking on out, until ctx ends or in closes.
//
// A ticker's channel is never closed by Stop, and a watch's stop only
// unsubscribes, so a relay that ranged over either alone outlived its
// stream: every closed tab left one goroutine and its ticker behind
// for the life of the process. The request's context is the one
// signal that the reader has gone, so the relay listens for it too.
func relayWakes[T any](ctx context.Context, in <-chan T, out chan<- struct{}) {
for {
select {
case <-ctx.Done():
return
case _, open := <-in:
if !open {
return
}
select {
case out <- struct{}{}:
default:
}
}
}
}
// liveWriteWait is how long one frame of a live stream may take to
// reach the reader. Frames are small and heartbeats go out every twenty
// seconds, so a connection that cannot take a frame in this long has a
// reader that stopped reading, and holding its goroutine open any
// longer only keeps a dead tab's resources alive.
const liveWriteWait = 30 * time.Second
// liveStream prepares w for a long-lived event stream: every frame
// written through the returned writer carries a write deadline, so a
// reader that stops reading is cut off instead of holding the handler
// forever. The server carries no global write timeout, because that
// would end healthy streams too; the deadline is set per frame instead.
// ok is false when w cannot flush, and then the stream cannot work.
func liveStream(w http.ResponseWriter) (http.ResponseWriter, http.Flusher, bool) {
fl, ok := w.(http.Flusher)
if !ok {
return w, nil, false
}
lw := &liveWriter{ResponseWriter: w, fl: fl, rc: http.NewResponseController(w)}
return lw, lw, true
}
// liveWriter arms the write deadline at the first write of each frame
// and again at its flush. Arming once per frame rather than once per
// Write keeps a many-line patch from paying for a deadline per line,
// and re-arming on the next write means a deadline left over from a
// quiet stretch never expires under a frame that has only just begun.
type liveWriter struct {
http.ResponseWriter
fl http.Flusher
rc *http.ResponseController
armed bool
}
func (l *liveWriter) arm() {
// A writer with no connection underneath (a test recorder) cannot
// take a deadline, and a stream there has nothing to cut off.
_ = l.rc.SetWriteDeadline(time.Now().Add(liveWriteWait))
}
func (l *liveWriter) Write(p []byte) (int, error) {
if !l.armed {
l.arm()
l.armed = true
}
return l.ResponseWriter.Write(p)
}
func (l *liveWriter) Flush() {
l.arm()
l.fl.Flush()
l.armed = false
}
func (l *liveWriter) Unwrap() http.ResponseWriter { return l.ResponseWriter }
// admitAnonymousStream draws one slot from the anonymous stream budget
// for req's address, or answers 429 and reports false when the address
// or the process is at its cap. The caller releases the slot when the
// stream ends. The address comes from the trusted-proxy rule, so a
// header the client wrote can neither dodge the cap nor aim it at
// somebody else.
func admitAnonymousStream(w http.ResponseWriter, req *http.Request) (release func(), ok bool) {
release, ok = streamcap.Anonymous.Acquire(clientIP(req))
if !ok {
// A slot frees when one of this reader's streams ends, so the
// wait is short and fixed.
w.Header().Set("Retry-After", streamcap.RetryAfter)
http.Error(w, "too many open streams from this address; close one and try again",
http.StatusTooManyRequests)
}
return release, ok
}