1package daemon23import (4 "encoding/json"5 "net/http"6 "net/url"7 "regexp"8 "strings"9 "time"1011 "github.com/FelineStateMachine/tinyrelay/internal/event"12)1314const (15 roomStreamKeepAlive = 25 * time.Second16 roomStreamBuffer = 6417)1819var roomStreamPath = regexp.MustCompile(`^/rooms/([a-z0-9_-]{1,64})/stream$`)2021func isRoomStreamPath(path string) bool { return roomStreamPath.MatchString(path) }2223// roomStreamHTTP serves GET /rooms/<id>/stream as server-sent events. Each24// event the viewer may see in the room arrives as "event: message" with the25// JSON event as data; a comment keeps the connection alive. The viewer is26// identified by the browser session cookie or a NIP-98 signature.27func (t *Tenant) roomStreamHTTP(w http.ResponseWriter, r *http.Request) {28 ctx, finish := t.app.telemetry.Start(r.Context(), "rooms")29 outcome := "error"30 defer func() { finish(outcome) }()31 if r.Method != http.MethodGet {32 w.Header().Set("Allow", "GET")33 http.Error(w, "method not allowed", http.StatusMethodNotAllowed)34 outcome = "invalid"35 return36 }37 id := roomStreamPath.FindStringSubmatch(r.URL.Path)[1]38 if origin := r.Header.Get("Origin"); origin != "" && r.Header.Get("Authorization") == "" {39 base, err := url.Parse(t.requestURL(r))40 if err != nil || origin != base.Scheme+"://"+base.Host {41 http.Error(w, "auth-required: sign this request", http.StatusUnauthorized)42 outcome = "unauthorized"43 return44 }45 }46 actor, err := t.resolveUIActor(r)47 if err != nil {48 browseHTTPError(w, err)49 outcome = "unauthorized"50 return51 }52 // Hold an operation only while checking access. A stream may stay open53 // for hours and must not keep maintenance waiting; it ends instead when54 // maintenance or shutdown begins.55 opCtx, done, err := t.beginOperation(ctx)56 if err != nil {57 http.Error(w, err.Error(), http.StatusServiceUnavailable)58 outcome = "busy"59 return60 }61 err = t.browseRead(opCtx, actor)62 if err == nil {63 _, _, err = t.roomFor(opCtx, actor, id)64 }65 done()66 if err != nil {67 browseHTTPError(w, err)68 outcome = "unauthorized"69 return70 }71 flusher, ok := w.(http.Flusher)72 if !ok {73 http.Error(w, "streaming unsupported", http.StatusInternalServerError)74 return75 }76 session := browseSession(t, actor)77 events := make(chan event.Event, roomStreamBuffer)78 overflow := make(chan struct{}, 1)79 stop := t.router.Listen(func(e event.Event) {80 globalPresence := e.Kind == 20001 && event.Tag(e, "h") == "" && actor != ""81 if globalPresence {82 allowed, _, err := t.community.RoomWriteAllowed(r.Context(), id, e.PubKey)83 if err != nil || !allowed {84 return85 }86 }87 if e.Kind == event.KIND_MARMOT_GROUP || (event.Tag(e, "h") != id && !globalPresence) {88 return89 }90 select {91 case events <- e:92 default:93 select {94 case overflow <- struct{}{}:95 default:96 }97 }98 })99 defer stop()100 w.Header().Set("Content-Type", "text/event-stream; charset=utf-8")101 w.Header().Set("Cache-Control", "no-store, private")102 w.Header().Set("X-Accel-Buffering", "no")103 w.WriteHeader(http.StatusOK)104 if _, err := w.Write([]byte(": room " + id + "\n\n")); err != nil {105 return106 }107 flusher.Flush()108 keepAlive := time.NewTicker(roomStreamKeepAlive)109 defer keepAlive.Stop()110 delivered := 0111 defer func() {112 t.app.telemetry.Logger().Debug("room stream closed", "events", delivered, "outcome", outcome)113 }()114 paused := t.maintenance.watch()115 for {116 select {117 case <-r.Context().Done():118 outcome = "closed"119 return120 case <-t.workCtx.Done():121 outcome = "closed"122 return123 case <-paused:124 if t.maintenance.blocked() {125 outcome = "closed"126 return127 }128 paused = t.maintenance.watch()129 case <-overflow:130 outcome = "busy"131 return132 case <-keepAlive.C:133 if _, err := w.Write([]byte(": keep-alive\n\n")); err != nil {134 outcome = "closed"135 return136 }137 flusher.Flush()138 case e := <-events:139 if !t.gate.CanSee(r.Context(), e, session, nil) {140 continue141 }142 raw, err := json.Marshal(e)143 if err != nil {144 continue145 }146 var frame strings.Builder147 frame.WriteString("event: message\ndata: ")148 frame.Write(raw)149 frame.WriteString("\n\n")150 if _, err := w.Write([]byte(frame.String())); err != nil {151 outcome = "closed"152 return153 }154 flusher.Flush()155 delivered++156 }157 }158}