hamlab/internal/api/thumbs.go

package api

// Previews for the Files tab: GET /api/files/thumb hands back one small JPEG of an image or
// a video, made by internal/thumbs and kept in <data dir>/thumbs.db. The work is bounded
// two ways — a handful of ffmpegs at once, and one per file however many rows ask for it —
// because a folder of a thousand photos asks for every row the moment previews go on.
//
// Root mode (?sudo=1) makes the preview through the same root helper the file reads use,
// so a file only root can read gets a preview too, and the probe runs as root as well: root
// opens every render node, whatever groups the daemon's user has.

import (
	"context"
	"errors"
	"log"
	"net/http"
	"os"
	"os/exec"
	"path/filepath"
	"runtime"
	"strconv"
	"strings"
	"sync"
	"time"

	"hamlab/internal/media"
	"hamlab/internal/priv"
)

const thumbSize = 256
const thumbTimeout = 30 * time.Second
const capsTTL = 10 * time.Minute // how long a probe is believed; a refresh after an install or a group change does not wait

type thumbSvc struct {
	cache *media.Cache
	self  string
	sem   chan struct{}
	mu    sync.Mutex
	caps  map[bool]*capsEntry // keyed by root
	calls map[string]*thumbCall
}

type capsEntry struct {
	mu   sync.Mutex
	caps media.Caps
	err  error
	at   time.Time
}

type thumbCall struct {
	done chan struct{}
	res  media.Result
	err  error
}

func newThumbSvc(dataDir, self string) *thumbSvc {
	n := runtime.NumCPU() / 2 // each ffmpeg is itself threaded; previews must not take the whole machine
	if n < 2 {
		n = 2
	}
	if n > 4 {
		n = 4
	}
	return &thumbSvc{cache: media.NewCache(filepath.Join(dataDir, "thumbs.db")), self: self, sem: make(chan struct{}, n), caps: map[bool]*capsEntry{}, calls: map[string]*thumbCall{}}
}

// capsFor is what previews can be made with, probed once and kept for a while.
func (t *thumbSvc) capsFor(ctx context.Context, root, refresh bool) (media.Caps, error) {
	t.mu.Lock()
	e := t.caps[root]
	if e == nil {
		e = &capsEntry{}
		t.caps[root] = e
	}
	t.mu.Unlock()
	e.mu.Lock()
	defer e.mu.Unlock()
	if !refresh && !e.at.IsZero() && time.Since(e.at) < capsTTL {
		return e.caps, e.err
	}
	if root {
		e.caps, e.err = media.ProbeRoot(ctx, t.self)
	} else {
		e.caps, e.err = media.Probe(ctx), nil
	}
	e.at = time.Now()
	mode := "as " + map[bool]string{false: "the daemon", true: "root"}[root]
	switch {
	case e.err != nil:
		log.Printf("thumbs: probe %s failed: %v", mode, e.err)
	case e.caps.Error != "":
		log.Printf("thumbs: %s: %s", mode, e.caps.Error)
	default:
		log.Printf("thumbs: %s: ffmpeg %s, %s", mode, e.caps.Version, describeCaps(e.caps))
	}
	return e.caps, e.err
}

// describeCaps is the one line that says how previews get made here.
func describeCaps(c media.Caps) string {
	if len(c.GPUs) == 0 {
		return "CPU only"
	}
	parts := make([]string, 0, len(c.GPUs))
	for _, g := range c.GPUs {
		name := g.Name
		if name == "" {
			name = g.Device
		}
		p := name + " (" + g.API
		if len(g.Codecs) > 0 {
			p += ": " + strings.Join(g.Codecs, ", ")
		}
		parts = append(parts, p+")")
	}
	return "GPU " + strings.Join(parts, ", ")
}

// make produces a preview, sharing the work when the same file is asked for twice at once.
func (t *thumbSvc) make(ctx context.Context, root bool, key, path string) (media.Result, error) {
	t.mu.Lock()
	if c, ok := t.calls[key]; ok {
		t.mu.Unlock()
		select {
		case <-c.done:
			return c.res, c.err
		case <-ctx.Done():
			return media.Result{}, ctx.Err()
		}
	}
	c := &thumbCall{done: make(chan struct{})}
	t.calls[key] = c
	t.mu.Unlock()
	defer func() {
		t.mu.Lock()
		delete(t.calls, key)
		t.mu.Unlock()
		close(c.done)
	}()
	// The work itself runs on a context of its own: the first requester going away must not
	// cancel a preview three other rows are waiting for.
	wctx, cancel := context.WithTimeout(context.Background(), thumbTimeout+5*time.Second)
	defer cancel()
	select {
	case t.sem <- struct{}{}:
		defer func() { <-t.sem }()
	case <-ctx.Done():
		c.err = ctx.Err()
		return c.res, c.err
	}
	caps, err := t.capsFor(wctx, root, false)
	if err != nil {
		c.err = err
		return c.res, c.err
	}
	o := media.Opts{Size: thumbSize, Timeout: thumbTimeout}
	if root {
		c.res, c.err = media.MakeRoot(wctx, t.self, caps, path, o)
	} else {
		c.res, c.err = media.Make(wctx, caps, path, o)
	}
	return c.res, c.err
}

// ---- handlers ----

// filesThumb: GET /api/files/thumb?path=…&v=<mtime>-<size>[&sudo=1]. v is the client's idea of
// the file, from its listing; it makes a cache hit free of a stat, and a changed file a new URL.
func (s *Server) filesThumb(w http.ResponseWriter, r *http.Request) {
	fsys, ok := s.fsFor(w, r)
	if !ok {
		return
	}
	p, ok := cleanQuery(w, r, "path")
	if !ok {
		return
	}
	if media.Kind(p) == "" {
		jsonErr(w, 415, "no preview for this kind of file")
		return
	}
	root := r.URL.Query().Get("sudo") == "1"
	if mt, sz, ok := parseV(r.URL.Query().Get("v")); ok {
		key := media.Key(p, sz, mt)
		if r.Header.Get("If-None-Match") == `"`+key+`"` {
			w.WriteHeader(304)
			return
		}
		if s.serveCachedThumb(w, key) {
			return
		}
	}
	in, err := fsys.Stat(r.Context(), p)
	if err != nil {
		fsErr(w, err)
		return
	}
	if in.Type == "dir" || in.Type == "other" || in.Broken {
		jsonErr(w, 400, "not a regular file")
		return
	}
	key := media.Key(p, in.Size, in.Mtime)
	if s.serveCachedThumb(w, key) {
		return
	}
	res, err := s.thumbs.make(r.Context(), root, key, p)
	if err != nil {
		var de *media.DecodeError
		switch {
		case errors.As(err, &de):
			if err := s.thumbs.cache.Put(key, media.Entry{Path: p, Failure: de.Msg}); err != nil {
				log.Printf("thumbs: cache: %v", err)
			}
			log.Printf("thumbs: %s: %s", p, de.Msg)
			jsonErr(w, 422, de.Msg)
		case errors.Is(err, context.DeadlineExceeded):
			log.Printf("thumbs: %s: timed out", p)
			jsonErr(w, 504, "preview timed out")
		case errors.Is(err, context.Canceled):
			jsonErr(w, 499, "cancelled")
		default:
			jsonErr(w, 503, err.Error())
		}
		return
	}
	if err := s.thumbs.cache.Put(key, media.Entry{Path: p, Engine: res.Engine, Codec: res.Codec, Width: res.Width, Height: res.Height, JPEG: res.JPEG}); err != nil {
		log.Printf("thumbs: cache: %v", err)
	}
	writeThumb(w, key, res.JPEG, res.Engine)
}

// serveCachedThumb answers from the cache when it can. A cache that cannot be read is a
// miss: the preview is made again rather than refused.
func (s *Server) serveCachedThumb(w http.ResponseWriter, key string) bool {
	e, found, err := s.thumbs.cache.Get(key)
	if err != nil {
		log.Printf("thumbs: cache: %v", err)
	}
	switch {
	case !found:
		return false
	case e.Failure != "":
		jsonErr(w, 422, e.Failure)
	default:
		writeThumb(w, key, e.JPEG, "cache ("+e.Engine+")")
	}
	return true
}

func writeThumb(w http.ResponseWriter, key string, jpeg []byte, engine string) {
	w.Header().Set("Content-Type", "image/jpeg")
	w.Header().Set("Content-Length", strconv.Itoa(len(jpeg)))
	w.Header().Set("Cache-Control", "private, max-age=604800")
	w.Header().Set("ETag", `"`+key+`"`)
	w.Header().Set("X-Hamlab-Thumb", engine)
	w.Header().Set("X-Content-Type-Options", "nosniff")
	_, _ = w.Write(jpeg)
}

func parseV(v string) (mtime, size int64, ok bool) {
	a, b, found := strings.Cut(v, "-")
	if !found {
		return 0, 0, false
	}
	mtime, err1 := strconv.ParseInt(a, 10, 64)
	size, err2 := strconv.ParseInt(b, 10, 64)
	return mtime, size, err1 == nil && err2 == nil
}

type thumbStatus struct {
	OK       bool          `json:"ok"`
	Engine   string        `json:"engine"` // gpu | cpu | ""
	Summary  string        `json:"summary"`
	FFmpeg   string        `json:"ffmpeg,omitempty"`
	GPUs     []media.GPU   `json:"gpus"`
	Unusable []media.GPU   `json:"unusable,omitempty"`
	Error    string        `json:"error,omitempty"`
	Install  *thumbInstall `json:"install,omitempty"` // ffmpeg is missing and this can put it on
	GPUFix   bool          `json:"gpu_fix,omitempty"` // a render node is there but closed to the daemon's user, and this can open it
	User     string        `json:"user"`
	Cache    thumbCache    `json:"cache"`
}

type thumbCache struct {
	Path    string `json:"path"`
	Entries int64  `json:"entries"`
	Bytes   int64  `json:"bytes"`
}

type thumbInstall struct {
	PM      string `json:"pm"`
	Package string `json:"package"`
}

// thumbsStatus: GET /api/files/thumbs[?sudo=1][&refresh=1] — how previews get made on this machine.
func (s *Server) thumbsStatus(w http.ResponseWriter, r *http.Request) {
	root := r.URL.Query().Get("sudo") == "1"
	if root && s.fsSudo == nil {
		jsonErr(w, 403, "root access needs passwordless sudo on the host")
		return
	}
	caps, err := s.thumbs.capsFor(r.Context(), root, r.URL.Query().Get("refresh") == "1")
	st := thumbStatus{GPUs: []media.GPU{}, User: s.user, Cache: thumbCache{Path: s.thumbs.cache.Path}}
	st.Cache.Entries, st.Cache.Bytes = s.thumbs.cache.Stats()
	if root {
		st.User = "root"
	}
	if err != nil {
		st.Error = err.Error()
		jsonOK(w, 200, st)
		return
	}
	st.FFmpeg, st.GPUs, st.Unusable, st.Error = caps.Version, caps.GPUs, caps.Unusable, caps.Error
	if st.GPUs == nil {
		st.GPUs = []media.GPU{}
	}
	st.OK = caps.Error == ""
	switch {
	case !st.OK:
		if pkg := ffmpegPackage(s.updates.PM()); pkg != "" && priv.Available() {
			st.Install = &thumbInstall{PM: s.updates.PM(), Package: pkg}
		}
		st.Summary = st.Error
	case len(caps.GPUs) > 0:
		st.Engine = "gpu"
		st.Summary = describeCaps(caps) + " · images on the CPU"
	default:
		st.Engine = "cpu"
		st.Summary = "CPU only"
	}
	if !root {
		for _, g := range caps.Unusable {
			if g.Permission && gpuFixPossible() {
				st.GPUFix = true
			}
		}
	}
	jsonOK(w, 200, st)
}

// ffmpegPackage is what to install for ffmpeg on a package manager. Fedora ships it as
// ffmpeg-free; the others under its own name.
func ffmpegPackage(pm string) string {
	switch pm {
	case "":
		return ""
	case "dnf":
		if b, _ := os.ReadFile("/etc/os-release"); strings.Contains(string(b), "ID=fedora") {
			return "ffmpeg-free"
		}
	}
	return "ffmpeg"
}

const hamlabUnit = "/etc/systemd/system/hamlab.service"

// gpuFixPossible: the daemon is an ordinary user under systemd, with root to hand.
func gpuFixPossible() bool {
	if os.Geteuid() == 0 || !priv.Available() {
		return false
	}
	_, err := os.Stat(hamlabUnit)
	return err == nil
}

// thumbsInstall: POST /api/files/thumbs/install — put ffmpeg on with the machine's package manager.
func (s *Server) thumbsInstall(w http.ResponseWriter, r *http.Request) {
	pkg := ffmpegPackage(s.updates.PM())
	if pkg == "" {
		jsonErr(w, 400, "no supported package manager on this machine")
		return
	}
	t, err := s.updates.Install([]string{pkg})
	if err != nil {
		jsonErr(w, 400, err.Error())
		return
	}
	log.Printf("thumbs: installing %s as task %s (from %s)", pkg, t.ID, clientIP(r))
	jsonOK(w, 202, t.Snapshot(false))
}

// gpuAccessScript gives hamlab.service the groups that own the render nodes, by a drop-in,
// then restarts it: a running process cannot gain a group.
const gpuAccessScript = `set -e
unit=` + hamlabUnit + `
[ -f "$unit" ] || { echo "no $unit: this machine does not run Ham Lab under systemd"; exit 1; }
groups=""
for g in render video; do
  if getent group "$g" >/dev/null 2>&1; then groups="$groups $g"; fi
done
[ -n "$groups" ] || { echo "neither a render nor a video group exists here"; exit 1; }
mkdir -p "$unit.d"
printf '[Service]\nSupplementaryGroups=%s\n' "${groups# }" > "$unit.d/gpu.conf"
echo "hamlab.service: supplementary groups${groups} ($unit.d/gpu.conf)"
systemctl daemon-reload
echo 'Ham Lab restarts in 3 seconds to pick up its new groups; the page reconnects by itself.'
systemd-run --quiet --on-active=3 --unit=hamlab-gpu-restart systemctl restart hamlab.service
`

// thumbsGPUAccess: POST /api/files/thumbs/gpu-access — open the render nodes to the daemon.
func (s *Server) thumbsGPUAccess(w http.ResponseWriter, r *http.Request) {
	if !gpuFixPossible() {
		jsonErr(w, 400, "nothing to do: Ham Lab is root, has no root, or is not under systemd here")
		return
	}
	if t, busy := s.tasks.RunningOf("thumbs:gpu"); busy { // a second run would only collide with the first's restart timer
		jsonErr(w, 409, "already being done (task "+t.ID+")")
		return
	}
	full := priv.Wrap("sh", "-c", gpuAccessScript)
	t, err := s.tasks.Start("thumbs:gpu", "give Ham Lab access to the GPU", exec.Command(full[0], full[1:]...))
	if err != nil {
		jsonErr(w, 400, err.Error())
		return
	}
	log.Printf("thumbs: GPU access drop-in as task %s, restart follows (from %s)", t.ID, clientIP(r))
	jsonOK(w, 202, t.Snapshot(false))
}