Mitchell-JohnsonandClaude Opus 5.5
fc4126a0ed
Every video plays YouTube's way: hls.js reads the connection and picks the quality, or one is picked by hand, or Original plays the file as it is. The qualities (a ladder by size, H.264) are made on the fly in 3 s pieces from wherever the viewer seeks, one ffmpeg per play, paused when far enough ahead. 360 videos open in a look-around view (drag, pinch or scroll to zoom from 8 to 300 degrees, phone motion). Settings -> Transcode device picks the machine and device that makes them, for every machine's files: a machine transcoding another's file reads it through the hub by a token for that one file. Replaces the codec/size picker and the single-quality re-encoder. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
531 lines
16 KiB
Go
531 lines
16 KiB
Go
package api
|
|
|
|
// Playing a video in the Files viewer, YouTube's way: Auto (the player reads the connection),
|
|
// any quality by hand, or the file itself. The qualities are made on the fly as HLS by
|
|
// internal/media, on the device the Transcode device setting names — which may be on another
|
|
// machine than the file.
|
|
//
|
|
// On every machine:
|
|
// POST /api/files/play[?sudo=1] {path | url, name, device} open a play
|
|
// GET /api/files/play/{id}/master.m3u8, /{rung}.m3u8, /{rung}/{n}.ts
|
|
// GET /api/files/play/{id} what it is doing
|
|
// DELETE /api/files/play/{id} (or POST …/close, for a beacon) end it
|
|
// GET /api/files/play-devices what it can encode on
|
|
// On the hub:
|
|
// POST /api/play {node, path, sudo} open a play of a file on node, wherever the setting
|
|
// says, answering which machine the play is on
|
|
// GET|PUT /api/transcode the setting, and every machine's devices
|
|
// GET /api/play-src/{token} a file, for a machine transcoding it, by a token
|
|
// that names that one file; ranges pass through
|
|
|
|
import (
|
|
"context"
|
|
"crypto/rand"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"log"
|
|
"net"
|
|
"net/http"
|
|
"net/url"
|
|
"path/filepath"
|
|
"strconv"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"hamlab/internal/files"
|
|
"hamlab/internal/media"
|
|
"hamlab/internal/store"
|
|
)
|
|
|
|
// playRuns is how many plays may be encoding at once on one machine; a fourth ends the one
|
|
// asked of least lately. The A380 makes about two 8K-sourced streams in real time.
|
|
const playRuns = 3
|
|
|
|
type playReq struct {
|
|
Path string `json:"path,omitempty"`
|
|
URL string `json:"url,omitempty"` // a file on another machine, as the hub serves it
|
|
Name string `json:"name,omitempty"`
|
|
Device string `json:"device,omitempty"`
|
|
}
|
|
|
|
type playOut struct {
|
|
Node string `json:"node"` // the machine the play is on: "" for the hub; filled in by the hub
|
|
Play *media.Play `json:"play"`
|
|
Master string `json:"master"` // the playlist, under that machine's /api
|
|
Engine string `json:"engine"`
|
|
Host string `json:"host"` // that machine's name, for the menu
|
|
Note string `json:"note,omitempty"`
|
|
Status media.PlayStatus `json:"status"`
|
|
}
|
|
|
|
// playDevices: GET /api/files/play-devices — what this machine could make the player's H.264
|
|
// on, probed as the daemon's own user (a play from another machine never runs as root).
|
|
func (s *Server) playDevices(w http.ResponseWriter, r *http.Request) {
|
|
caps, err := s.thumbs.capsFor(r.Context(), false, false)
|
|
if err != nil {
|
|
caps.Error = err.Error()
|
|
}
|
|
jsonOK(w, 200, map[string]any{"host": s.hostName(), "devices": media.Devices(caps)})
|
|
}
|
|
|
|
// playStart: POST /api/files/play[?sudo=1] — open a play of a file here, or of a URL the hub
|
|
// serves a file on another machine at.
|
|
func (s *Server) playStart(w http.ResponseWriter, r *http.Request) {
|
|
var body playReq
|
|
if err := json.NewDecoder(io.LimitReader(r.Body, 64<<10)).Decode(&body); err != nil {
|
|
jsonErr(w, 400, "bad request")
|
|
return
|
|
}
|
|
root := r.URL.Query().Get("sudo") == "1"
|
|
if root && s.fsSudo == nil {
|
|
jsonErr(w, 403, "root access needs passwordless sudo on the host")
|
|
return
|
|
}
|
|
out, code, err := s.openPlay(r.Context(), body, root)
|
|
if err != nil {
|
|
jsonErr(w, code, err.Error())
|
|
return
|
|
}
|
|
jsonOK(w, 200, out)
|
|
}
|
|
|
|
func (s *Server) openPlay(ctx context.Context, body playReq, root bool) (playOut, int, error) {
|
|
if root && s.fsSudo == nil && body.URL == "" {
|
|
return playOut{}, 403, errors.New("root access needs passwordless sudo on the host")
|
|
}
|
|
in, name := body.Path, body.Name
|
|
if body.URL != "" {
|
|
u, err := url.Parse(body.URL)
|
|
if err != nil || (u.Scheme != "http" && u.Scheme != "https") || body.Path != "" {
|
|
return playOut{}, 400, errors.New("a play is of a path or of an http URL")
|
|
}
|
|
in, root = body.URL, false
|
|
} else {
|
|
p, err := files.Clean(body.Path)
|
|
if err != nil {
|
|
return playOut{}, 400, err
|
|
}
|
|
if media.Kind(p) != "video" {
|
|
return playOut{}, 415, errors.New("not a video")
|
|
}
|
|
fsys := s.fsLocal
|
|
if root {
|
|
fsys = s.fsSudo
|
|
}
|
|
if _, err := fsys.Stat(ctx, p); err != nil {
|
|
return playOut{}, 404, err
|
|
}
|
|
in = p
|
|
if name == "" {
|
|
name = filepath.Base(p)
|
|
}
|
|
}
|
|
caps, err := s.thumbs.capsFor(ctx, root, false)
|
|
if err == nil && caps.Error != "" {
|
|
err = errors.New(caps.Error)
|
|
}
|
|
if err != nil {
|
|
return playOut{}, 503, err
|
|
}
|
|
ictx, cancel := context.WithTimeout(ctx, 30*time.Second)
|
|
src, err := media.Inspect(ictx, caps, in, root)
|
|
cancel()
|
|
if err != nil {
|
|
var de *media.DecodeError
|
|
if errors.As(err, &de) {
|
|
return playOut{}, 422, errors.New(de.Msg)
|
|
}
|
|
return playOut{}, 504, err
|
|
}
|
|
p, err := s.plays.Open(caps, root, in, name, src, body.Device)
|
|
if err != nil {
|
|
return playOut{}, 422, err
|
|
}
|
|
log.Printf("files: play %s on %s (%d qualities, %s)", name, p.Engine(), len(p.Rungs), map[bool]string{true: "from another machine", false: "here"}[p.Remote])
|
|
return playOut{Play: p, Master: "/api/files/play/" + p.ID + "/master.m3u8", Engine: p.Engine(), Host: s.hostName(), Status: p.Status()}, 200, nil
|
|
}
|
|
|
|
// playStatus: GET /api/files/play/{id}
|
|
func (s *Server) playStatus(w http.ResponseWriter, r *http.Request) {
|
|
p, ok := s.plays.Find(r.PathValue("id"))
|
|
if !ok {
|
|
jsonErr(w, 404, "no such play")
|
|
return
|
|
}
|
|
jsonOK(w, 200, p.Status())
|
|
}
|
|
|
|
// playStop: DELETE /api/files/play/{id}, or POST …/close — the form a page can send on its
|
|
// way out (navigator.sendBeacon), so a phone closing the tab releases the machine at once.
|
|
func (s *Server) playStop(w http.ResponseWriter, r *http.Request) {
|
|
s.plays.Close(r.PathValue("id"))
|
|
w.WriteHeader(204)
|
|
}
|
|
|
|
// playPlaylist: GET /api/files/play/{id}/{file} — master.m3u8, or one quality's <name>.m3u8.
|
|
func (s *Server) playPlaylist(w http.ResponseWriter, r *http.Request) {
|
|
p, ok := s.plays.Find(r.PathValue("id"))
|
|
if !ok {
|
|
jsonErr(w, 404, "no such play")
|
|
return
|
|
}
|
|
f := r.PathValue("file")
|
|
var b []byte
|
|
if f == "master.m3u8" {
|
|
b = p.Master()
|
|
} else if name, isList := strings.CutSuffix(f, ".m3u8"); isList {
|
|
if b, ok = p.Media(name); !ok {
|
|
jsonErr(w, 404, "no such quality")
|
|
return
|
|
}
|
|
} else {
|
|
jsonErr(w, 404, "not a play file")
|
|
return
|
|
}
|
|
w.Header().Set("Content-Type", "application/vnd.apple.mpegurl")
|
|
w.Header().Set("Cache-Control", "no-store")
|
|
_, _ = w.Write(b)
|
|
}
|
|
|
|
// playPiece: GET /api/files/play/{id}/{rung}/{seg} — a piece, made first if need be.
|
|
func (s *Server) playPiece(w http.ResponseWriter, r *http.Request) {
|
|
p, ok := s.plays.Find(r.PathValue("id"))
|
|
if !ok {
|
|
jsonErr(w, 404, "no such play")
|
|
return
|
|
}
|
|
ns, isTS := strings.CutSuffix(r.PathValue("seg"), ".ts")
|
|
n, err := strconv.Atoi(ns)
|
|
if !isTS || err != nil || strconv.Itoa(n) != ns {
|
|
jsonErr(w, 404, "not a piece")
|
|
return
|
|
}
|
|
file, err := p.Segment(r.Context().Done(), r.PathValue("rung"), n)
|
|
switch {
|
|
case errors.Is(err, media.ErrNotFound):
|
|
jsonErr(w, 404, err.Error())
|
|
return
|
|
case r.Context().Err() != nil:
|
|
return // the player moved on
|
|
case err != nil:
|
|
jsonErr(w, 503, err.Error())
|
|
return
|
|
}
|
|
w.Header().Set("Content-Type", "video/mp2t")
|
|
w.Header().Set("Cache-Control", "private, max-age=3600")
|
|
http.ServeFile(w, r, file)
|
|
}
|
|
|
|
// ---- on the hub: where plays are made ----
|
|
|
|
// transcodeChoice is the setting; nil is the machine the file is on.
|
|
func (s *Server) transcodeChoice() *store.Transcode {
|
|
var t *store.Transcode
|
|
s.store.View(func(st *store.State) {
|
|
if st.Transcode != nil {
|
|
c := *st.Transcode
|
|
t = &c
|
|
}
|
|
})
|
|
return t
|
|
}
|
|
|
|
type transcodeMachine struct {
|
|
Node string `json:"node"` // "" is the hub
|
|
Host string `json:"host"`
|
|
Devices []media.Device `json:"devices"`
|
|
Error string `json:"error,omitempty"` // the machine didn't answer
|
|
}
|
|
|
|
// getTranscode: GET /api/transcode — the setting, and what every machine can encode on,
|
|
// asked of each now.
|
|
func (s *Server) getTranscode(w http.ResponseWriter, r *http.Request) {
|
|
names := []string{""}
|
|
if s.nodes != nil {
|
|
names = append(names, s.nodes.Names()...)
|
|
}
|
|
out := make([]transcodeMachine, len(names))
|
|
var wg sync.WaitGroup
|
|
for i, n := range names {
|
|
wg.Add(1)
|
|
go func() {
|
|
defer wg.Done()
|
|
m := transcodeMachine{Node: n, Host: n}
|
|
ctx, cancel := context.WithTimeout(r.Context(), 8*time.Second)
|
|
defer cancel()
|
|
if n == "" {
|
|
caps, err := s.thumbs.capsFor(ctx, false, false)
|
|
if err != nil {
|
|
caps.Error = err.Error()
|
|
}
|
|
m.Host, m.Devices = s.hostName(), media.Devices(caps)
|
|
} else {
|
|
code, b, err := s.nodes.Do(ctx, n, "GET", "/api/files/play-devices", nil, nil)
|
|
var d struct {
|
|
Host string `json:"host"`
|
|
Devices []media.Device `json:"devices"`
|
|
}
|
|
switch {
|
|
case err != nil:
|
|
m.Error = err.Error()
|
|
case code == 404:
|
|
m.Error = "runs an older Ham Lab"
|
|
case code != 200 || json.Unmarshal(b, &d) != nil:
|
|
m.Error = fmt.Sprintf("answered %d", code)
|
|
default:
|
|
m.Devices = d.Devices
|
|
}
|
|
}
|
|
out[i] = m
|
|
}()
|
|
}
|
|
wg.Wait()
|
|
jsonOK(w, 200, map[string]any{"selected": s.transcodeChoice(), "machines": out})
|
|
}
|
|
|
|
// putTranscode: PUT /api/transcode {node, device}, or {} for the machine the file is on.
|
|
func (s *Server) putTranscode(w http.ResponseWriter, r *http.Request) {
|
|
var body *store.Transcode
|
|
if err := json.NewDecoder(io.LimitReader(r.Body, 4096)).Decode(&body); err != nil {
|
|
jsonErr(w, 400, "expected {\"node\": …, \"device\": …}")
|
|
return
|
|
}
|
|
if body != nil && body.Device == "" {
|
|
body = nil
|
|
}
|
|
if body != nil {
|
|
known := body.Node == ""
|
|
if s.nodes != nil {
|
|
for _, n := range s.nodes.Names() {
|
|
known = known || n == body.Node
|
|
}
|
|
}
|
|
if !known {
|
|
jsonErr(w, 400, "no such machine")
|
|
return
|
|
}
|
|
if body.Device != media.CPU && !strings.HasPrefix(body.Device, "/dev/dri/") {
|
|
jsonErr(w, 400, "a device is a render node or \"cpu\"")
|
|
return
|
|
}
|
|
}
|
|
if err := s.store.Update(func(st *store.State) { st.Transcode = body }); err != nil {
|
|
jsonErr(w, 500, err.Error())
|
|
return
|
|
}
|
|
s.getTranscode(w, r)
|
|
}
|
|
|
|
// hubPlay: POST /api/play {node, path, sudo} — open a play of a file on node ("" the hub) on
|
|
// the device the setting names. When that is on another machine, that machine reads the file
|
|
// through this hub, by a token good for that one file. If it can't be reached, the file's own
|
|
// machine makes it instead, and the answer says so.
|
|
func (s *Server) hubPlay(w http.ResponseWriter, r *http.Request) {
|
|
var body struct {
|
|
Node string `json:"node"`
|
|
Path string `json:"path"`
|
|
Sudo bool `json:"sudo"`
|
|
}
|
|
if err := json.NewDecoder(io.LimitReader(r.Body, 64<<10)).Decode(&body); err != nil {
|
|
jsonErr(w, 400, "bad request")
|
|
return
|
|
}
|
|
p, err := files.Clean(body.Path)
|
|
if err != nil {
|
|
jsonErr(w, 400, err.Error())
|
|
return
|
|
}
|
|
if body.Node != "" && (s.nodes == nil || !has(s.nodes.Names(), body.Node)) {
|
|
jsonErr(w, 404, "no such machine")
|
|
return
|
|
}
|
|
choice := s.transcodeChoice()
|
|
var note string
|
|
if choice != nil && choice.Node != body.Node {
|
|
if s.nodes != nil && choice.Node != "" && !has(s.nodes.Names(), choice.Node) {
|
|
note = "The transcode machine " + choice.Node + " is no longer managed here; made where the file is."
|
|
} else {
|
|
src, err := s.srcURL(choice.Node, body.Node, p, body.Sudo)
|
|
if err == nil {
|
|
req := playReq{URL: src, Name: filepath.Base(p), Device: choice.Device}
|
|
out, code, err := s.playOn(r.Context(), choice.Node, req, false)
|
|
if err == nil {
|
|
jsonOK(w, 200, out)
|
|
return
|
|
}
|
|
if code == 422 || code == 415 {
|
|
jsonErr(w, code, err.Error())
|
|
return
|
|
}
|
|
}
|
|
log.Printf("files: play on %s failed (%v); making it where the file is", nodeLabel(choice.Node), err)
|
|
note = "Couldn't make it on " + nodeLabel(choice.Node) + " (" + err.Error() + "), so it is made where the file is."
|
|
}
|
|
}
|
|
device := ""
|
|
if choice != nil && choice.Node == body.Node {
|
|
device = choice.Device
|
|
}
|
|
out, code, err := s.playOn(r.Context(), body.Node, playReq{Path: p, Device: device}, body.Sudo)
|
|
if err != nil {
|
|
jsonErr(w, code, err.Error())
|
|
return
|
|
}
|
|
out.Note = note
|
|
jsonOK(w, 200, out)
|
|
}
|
|
|
|
func nodeLabel(n string) string {
|
|
if n == "" {
|
|
return "the hub"
|
|
}
|
|
return n
|
|
}
|
|
|
|
func has(list []string, x string) bool {
|
|
for _, v := range list {
|
|
if v == x {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
// playOn opens a play on a machine: here, or through that node's API.
|
|
func (s *Server) playOn(ctx context.Context, node string, req playReq, sudo bool) (playOut, int, error) {
|
|
if node == "" {
|
|
out, code, err := s.openPlay(ctx, req, sudo)
|
|
return out, code, err
|
|
}
|
|
b, _ := json.Marshal(req)
|
|
path := "/api/files/play"
|
|
if sudo {
|
|
path += "?sudo=1"
|
|
}
|
|
code, resp, err := s.nodes.Do(ctx, node, "POST", path, b, http.Header{"Content-Type": []string{"application/json"}})
|
|
if err != nil {
|
|
return playOut{}, 502, err
|
|
}
|
|
if code != 200 {
|
|
var e struct{ Error string }
|
|
_ = json.Unmarshal(resp, &e)
|
|
if e.Error == "" {
|
|
e.Error = fmt.Sprintf("%s answered %d", node, code)
|
|
}
|
|
return playOut{}, code, errors.New(e.Error)
|
|
}
|
|
var out playOut
|
|
if err := json.Unmarshal(resp, &out); err != nil {
|
|
return playOut{}, 502, err
|
|
}
|
|
out.Node = node
|
|
return out, 200, nil
|
|
}
|
|
|
|
// ---- the file, for the machine transcoding it ----
|
|
|
|
type srcGrant struct {
|
|
node, path string
|
|
sudo bool
|
|
until time.Time
|
|
}
|
|
|
|
type srcGrants struct {
|
|
mu sync.Mutex
|
|
m map[string]*srcGrant
|
|
}
|
|
|
|
const srcLife = 12 * time.Hour // from its last use: a film paused overnight still plays
|
|
|
|
var grants = &srcGrants{m: map[string]*srcGrant{}}
|
|
|
|
func (g *srcGrants) mint(node, path string, sudo bool) string {
|
|
b := make([]byte, 24)
|
|
_, _ = rand.Read(b)
|
|
tok := hex.EncodeToString(b)
|
|
now := time.Now()
|
|
g.mu.Lock()
|
|
defer g.mu.Unlock()
|
|
for k, v := range g.m {
|
|
if now.After(v.until) {
|
|
delete(g.m, k)
|
|
}
|
|
}
|
|
g.m[tok] = &srcGrant{node: node, path: path, sudo: sudo, until: now.Add(srcLife)}
|
|
return tok
|
|
}
|
|
|
|
func (g *srcGrants) use(tok string) (srcGrant, bool) {
|
|
g.mu.Lock()
|
|
defer g.mu.Unlock()
|
|
v, ok := g.m[tok]
|
|
if !ok || time.Now().After(v.until) {
|
|
return srcGrant{}, false
|
|
}
|
|
v.until = time.Now().Add(srcLife)
|
|
return *v, true
|
|
}
|
|
|
|
// srcURL is where the machine `to` can read a file on machine `from` through this hub: this
|
|
// hub's own address as `to` reaches it, and a token for that one file.
|
|
func (s *Server) srcURL(to, from, path string, sudo bool) (string, error) {
|
|
host, port, err := net.SplitHostPort(s.cfg.Listen)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
ip := net.ParseIP(host)
|
|
if host == "" || ip == nil || ip.IsUnspecified() || ip.IsLoopback() {
|
|
// Listening everywhere: the address this machine reaches `to` from is one it answers on.
|
|
target := "127.0.0.1:1"
|
|
if to != "" {
|
|
if u, err := url.Parse(s.nodes.URL(to)); err == nil && u.Host != "" {
|
|
target = u.Host
|
|
}
|
|
}
|
|
c, err := net.Dial("udp", target)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
host = c.LocalAddr().(*net.UDPAddr).IP.String()
|
|
c.Close()
|
|
}
|
|
scheme := "http"
|
|
if s.fingerprint != "" {
|
|
scheme = "https"
|
|
}
|
|
return scheme + "://" + net.JoinHostPort(host, port) + "/api/play-src/" + grants.mint(from, path, sudo), nil
|
|
}
|
|
|
|
// playSrc: GET /api/play-src/{token} — the one file a token names, ranges and all, read from
|
|
// its machine. The token is the only key: it is minted for a machine transcoding that file.
|
|
func (s *Server) playSrc(w http.ResponseWriter, r *http.Request) {
|
|
g, ok := grants.use(r.PathValue("token"))
|
|
if !ok {
|
|
jsonErr(w, 404, "no such file")
|
|
return
|
|
}
|
|
q := url.Values{"path": {g.path}}
|
|
if g.sudo {
|
|
q.Set("sudo", "1")
|
|
}
|
|
r2 := r.Clone(r.Context())
|
|
r2.URL.RawQuery = q.Encode()
|
|
r2.Header.Del("Cookie")
|
|
if g.node == "" {
|
|
r2.URL.Path = "/api/files/read"
|
|
s.filesRead(w, r2)
|
|
return
|
|
}
|
|
p, ok := s.nodes.Proxy(g.node)
|
|
if !ok {
|
|
jsonErr(w, 404, "no such machine")
|
|
return
|
|
}
|
|
r2.URL.Path = "/api/nodes/" + g.node + "/files/read"
|
|
p.ServeHTTP(w, r2)
|
|
}
|