Live video on the AT Protocol
stream.place
1package api
2
3import (
4 "bufio"
5 "bytes"
6 "context"
7 "fmt"
8 "io"
9 "net/http"
10 "strconv"
11 "strings"
12 "time"
13
14 "github.com/julienschmidt/httprouter"
15 "github.com/pion/webrtc/v4"
16 "golang.org/x/sync/errgroup"
17 "stream.place/streamplace/pkg/aqtime"
18 "stream.place/streamplace/pkg/constants"
19 "stream.place/streamplace/pkg/errors"
20 "stream.place/streamplace/pkg/log"
21 "stream.place/streamplace/pkg/spmetrics"
22)
23
24func (a *StreamplaceAPI) NormalizeUser(ctx context.Context, user string) (string, error) {
25 alias, ok := a.Aliases[user]
26 if ok {
27 user = alias
28 }
29 // did:key, pass through unaltered
30 if strings.HasPrefix(user, constants.DID_KEY_PREFIX) {
31 return user, nil
32 }
33 // only other allowed case is a bluesky handle
34 repo, err := a.ATSync.SyncBlueskyRepoCached(ctx, user, a.Model)
35 if err != nil {
36 return "", err
37 }
38 return repo.DID, nil
39}
40
41func (a *StreamplaceAPI) HandleMP4Playback(ctx context.Context) httprouter.Handle {
42 return func(w http.ResponseWriter, r *http.Request, p httprouter.Params) {
43 user := p.ByName("user")
44 if user == "" {
45 errors.WriteHTTPBadRequest(w, "user required", nil)
46 return
47 }
48 rendition := getRendition(r)
49 user, err := a.NormalizeUser(ctx, user)
50 if err != nil {
51 errors.WriteHTTPBadRequest(w, "invalid user", err)
52 return
53 }
54 var delayMS int64 = 3000
55 userDelay := r.URL.Query().Get("delayms")
56 if userDelay != "" {
57 var err error
58 delayMS, err = strconv.ParseInt(userDelay, 10, 64)
59 if err != nil {
60 errors.WriteHTTPBadRequest(w, "error parsing delay", err)
61 return
62 }
63 if delayMS > 10000 {
64 errors.WriteHTTPBadRequest(w, "delay too large, maximum 10000", nil)
65 return
66 }
67 }
68 spmetrics.ViewerInc(user)
69 defer spmetrics.ViewerDec(user)
70 w.Header().Set("Content-Type", "video/mp4")
71 w.WriteHeader(200)
72 g, ctx := errgroup.WithContext(ctx)
73 pr, pw := io.Pipe()
74 bufw := bufio.NewWriter(pw)
75 g.Go(func() error {
76 return a.MediaManager.SegmentToMP4(ctx, user, rendition, bufw)
77 })
78 g.Go(func() error {
79 <-ctx.Done()
80 pr.Close()
81 pw.Close()
82 return nil
83 })
84 g.Go(func() error {
85 time.Sleep(time.Duration(delayMS) * time.Millisecond)
86 _, err := io.Copy(w, pr)
87 return err
88 })
89 if err := g.Wait(); err != nil {
90 errors.WriteHTTPBadRequest(w, "request failed", err)
91 }
92 }
93}
94
95func (a *StreamplaceAPI) HandleMKVPlayback(ctx context.Context) httprouter.Handle {
96 return func(w http.ResponseWriter, r *http.Request, p httprouter.Params) {
97 user := p.ByName("user")
98 if user == "" {
99 errors.WriteHTTPBadRequest(w, "user required", nil)
100 return
101 }
102 rendition := getRendition(r)
103 user, err := a.NormalizeUser(ctx, user)
104 if err != nil {
105 errors.WriteHTTPBadRequest(w, "invalid user", err)
106 return
107 }
108 var delayMS int64 = 1000
109 userDelay := r.URL.Query().Get("delayms")
110 if userDelay != "" {
111 var err error
112 delayMS, err = strconv.ParseInt(userDelay, 10, 64)
113 if err != nil {
114 errors.WriteHTTPBadRequest(w, "error parsing delay", err)
115 return
116 }
117 if delayMS > 10000 {
118 errors.WriteHTTPBadRequest(w, "delay too large, maximum 10000", nil)
119 return
120 }
121 }
122 spmetrics.ViewerInc(user)
123 defer spmetrics.ViewerDec(user)
124 w.Header().Set("Content-Type", "video/webm")
125 w.WriteHeader(200)
126 g, ctx := errgroup.WithContext(ctx)
127 pr, pw := io.Pipe()
128 bufw := bufio.NewWriter(pw)
129 g.Go(func() error {
130 return a.MediaManager.SegmentToMKV(ctx, user, rendition, bufw)
131 })
132 g.Go(func() error {
133 <-ctx.Done()
134 pr.Close()
135 pw.Close()
136 return nil
137 })
138 g.Go(func() error {
139 time.Sleep(time.Duration(delayMS) * time.Millisecond)
140 _, err := io.Copy(w, pr)
141 return err
142 })
143 if err := g.Wait(); err != nil {
144 errors.WriteHTTPBadRequest(w, "request failed", err)
145 }
146 }
147}
148
149func (a *StreamplaceAPI) HandleWebRTCPlayback(ctx context.Context) httprouter.Handle {
150 return func(w http.ResponseWriter, r *http.Request, p httprouter.Params) {
151 user := p.ByName("user")
152 if user == "" {
153 errors.WriteHTTPBadRequest(w, "user required", nil)
154 return
155 }
156 rendition := getRendition(r)
157 user, err := a.NormalizeUser(ctx, user)
158 if err != nil {
159 errors.WriteHTTPBadRequest(w, "invalid user", err)
160 return
161 }
162 body, err := io.ReadAll(r.Body)
163 if err != nil {
164 errors.WriteHTTPBadRequest(w, "error reading body", err)
165 return
166 }
167 offer := webrtc.SessionDescription{Type: webrtc.SDPTypeOffer, SDP: string(body)}
168 answer, err := a.MediaManager.WebRTCPlayback(ctx, user, rendition, &offer)
169 if err != nil {
170 errors.WriteHTTPInternalServerError(w, "error playing back", err)
171 return
172 }
173 w.WriteHeader(201)
174 w.Header().Add("Location", r.URL.Path)
175 if _, err := w.Write([]byte(answer.SDP)); err != nil {
176 log.Error(ctx, "error writing response", "error", err)
177 }
178 }
179}
180
181const BearerPrefix = "Bearer "
182const KeyPrefix = "0x"
183
184func (a *StreamplaceAPI) HandleWebRTCIngest(ctx context.Context) httprouter.Handle {
185 return func(w http.ResponseWriter, r *http.Request, p httprouter.Params) {
186 ct := r.Header.Get("Content-Type")
187 if ct != "application/sdp" {
188 errors.WriteHTTPBadRequest(w, "invalid content type", nil)
189 return
190 }
191 var encoded string
192 urlKey := p.ByName("key")
193 if urlKey != "" {
194 encoded = urlKey
195 } else {
196 auth := r.Header.Get("Authorization")
197 if auth == "" {
198 errors.WriteHTTPUnauthorized(w, "authorization header required", nil)
199 return
200 }
201 if !strings.HasPrefix(auth, BearerPrefix) {
202 errors.WriteHTTPUnauthorized(w, "invalid authorization header (needs Bearer prefix)", nil)
203 return
204 }
205 encoded = auth[len(BearerPrefix):]
206 // it's easy to copy-paste a trailing or leading space, so clear those out
207 encoded = strings.TrimSpace(encoded)
208 }
209
210 mediaSigner, err := a.MakeMediaSigner(ctx, encoded)
211 if err != nil {
212 errors.WriteHTTPUnauthorized(w, "invalid authorization key", err)
213 return
214 }
215
216 body, err := io.ReadAll(r.Body)
217 if err != nil {
218 errors.WriteHTTPBadRequest(w, "error reading body", err)
219 return
220 }
221 offer := webrtc.SessionDescription{Type: webrtc.SDPTypeOffer, SDP: string(body)}
222 answer, err := a.MediaManager.WebRTCIngest(ctx, &offer, mediaSigner)
223 if err != nil {
224 errors.WriteHTTPInternalServerError(w, "error playing back", err)
225 return
226 }
227 host := r.Host
228 if host == "" {
229 host = r.URL.Host
230 }
231 scheme := "http"
232 if r.TLS != nil {
233 scheme = "https"
234 }
235 location := fmt.Sprintf("%s://%s/api/live/webrtc", scheme, host)
236 log.Log(ctx, "location", "location", location)
237 w.Header().Set("Location", location)
238 w.WriteHeader(201)
239 if _, err := w.Write([]byte(answer.SDP)); err != nil {
240 log.Error(ctx, "error writing response", "error", err)
241 }
242 }
243}
244
245var epoch = time.Unix(0, 0).Format(time.RFC1123)
246
247var noCacheHeaders = map[string]string{
248 "Expires": epoch,
249 "Cache-Control": "no-cache, private, max-age=0",
250 "Pragma": "no-cache",
251 "X-Accel-Expires": "0",
252}
253
254var etagHeaders = []string{
255 "ETag",
256 "If-Modified-Since",
257 "If-Match",
258 "If-None-Match",
259 "If-Range",
260 "If-Unmodified-Since",
261}
262
263func NoCache(h httprouter.Handle) httprouter.Handle {
264 return func(w http.ResponseWriter, r *http.Request, p httprouter.Params) {
265 // Delete any ETag headers that may have been set
266 for _, v := range etagHeaders {
267 if r.Header.Get(v) != "" {
268 r.Header.Del(v)
269 }
270 }
271
272 // Set our NoCache headers
273 for k, v := range noCacheHeaders {
274 w.Header().Set(k, v)
275 }
276
277 h(w, r, p)
278 }
279}
280
281func (a *StreamplaceAPI) HandleHLSPlayback(ctx context.Context) httprouter.Handle {
282 return NoCache(func(w http.ResponseWriter, r *http.Request, p httprouter.Params) {
283 user := p.ByName("user")
284 if user == "" {
285 errors.WriteHTTPBadRequest(w, "user required", nil)
286 return
287 }
288 user, err := a.NormalizeUser(ctx, user)
289 if err != nil {
290 errors.WriteHTTPBadRequest(w, "invalid user", err)
291 return
292 }
293 file := p.ByName("file")
294 if file == "" {
295 errors.WriteHTTPBadRequest(w, "file required", nil)
296 return
297 }
298 m3u8, err := a.Director.GetM3U8(ctx, user)
299 if err != nil {
300 errors.WriteHTTPNotFound(w, "could not get m3u8", err)
301 return
302 }
303 session := r.URL.Query().Get("session")
304 rendition := r.URL.Query().Get("rendition")
305 buf, err := m3u8.GetFile(file, session, rendition)
306 if err != nil {
307 errors.WriteHTTPNotFound(w, "segment not found", err)
308 return
309 }
310
311 if strings.HasSuffix(file, ".m3u8") {
312 w.Header().Set("Content-Type", "application/vnd.apple.mpegurl")
313 } else {
314 if session != "" {
315 spmetrics.SessionSeen(user, session)
316 }
317 w.Header().Set("Content-Type", "video/mp2t")
318 }
319
320 http.ServeContent(w, r, file, time.Now(), bytes.NewReader(buf))
321 })
322}
323
324func (a *StreamplaceAPI) HandleThumbnailPlayback(ctx context.Context) httprouter.Handle {
325 return func(w http.ResponseWriter, r *http.Request, p httprouter.Params) {
326 user := p.ByName("user")
327 if user == "" {
328 errors.WriteHTTPBadRequest(w, "user required", nil)
329 return
330 }
331 user, err := a.NormalizeUser(ctx, user)
332 if err != nil {
333 errors.WriteHTTPNotFound(w, "user not found", err)
334 return
335 }
336 thumb, err := a.Model.LatestThumbnailForUser(user)
337 if err != nil {
338 errors.WriteHTTPInternalServerError(w, "could not query thumbnail", err)
339 return
340 }
341 if thumb == nil {
342 errors.WriteHTTPNotFound(w, "thumbnail not found", err)
343 return
344 }
345 aqt := aqtime.FromTime(thumb.Segment.StartTime)
346 fpath, err := a.CLI.SegmentFilePath(user, fmt.Sprintf("%s.%s", aqt.String(), thumb.Format))
347 if err != nil {
348 errors.WriteHTTPInternalServerError(w, "could not get segment file path", err)
349 return
350 }
351 http.ServeFile(w, r, fpath)
352 }
353}