arozos-mirror/src/mod/media/transcoder/hls.go

package transcoder

/*
	hls.go

	HTTP Live Streaming output for the transcoder.

	TranscodeAndStream pipes a single fragmented MP4 down one long-lived HTTP
	response, which cannot answer byte-range requests and therefore cannot be
	played by WebKit clients (Safari, and every browser on iOS). HLS solves that
	by cutting the transcode into short segments listed in a playlist: each
	segment is an ordinary, finite, seekable file.

	A session owns one ffmpeg process writing segments into its own temp
	directory. Sessions are keyed by owner + source file + resolution + start
	offset so that a reload, or a second player on the same file, reuses the
	transcode already running instead of starting another. Idle sessions are
	reaped by a background janitor, which kills ffmpeg and removes the directory.

	The playlist is written with -hls_playlist_type event, so it grows as the
	transcode proceeds and the player can seek freely within whatever has been
	produced so far. Seeking past that point is done the same way the MP4 path
	does it: by starting a new session at a later -ss offset.

	That last part is why a session carries the id of the player that asked for
	it. The MP4 path kills its ffmpeg as soon as the client drops the response,
	so a seek never leaves a transcode behind; an HLS session has no such
	signal, and without one every seek would add another ffmpeg racing through
	the rest of the film. A new session therefore supersedes the ones the same
	player left behind (see GetOrCreate).
*/

import (
	"crypto/md5"
	"encoding/hex"
	"errors"
	"fmt"
	"os"
	"os/exec"
	"path/filepath"
	"strconv"
	"strings"
	"sync"
	"time"

	"imuslab.com/arozos/mod/info/logger"
)

const (
	// hlsSegmentSeconds is the target length of each segment. Four seconds is
	// the usual HLS compromise: short enough that playback starts quickly and
	// seeking lands close to the requested time, long enough that the segment
	// count (and request rate) stays sane on a feature-length file.
	hlsSegmentSeconds = 4

	hlsPlaylistName   = "index.m3u8"
	hlsSegmentPattern = "seg%05d.m4s"
	hlsSegmentPrefix  = "seg"
	hlsSegmentSuffix  = ".m4s"

	// HLSInitSegmentName is the fragmented-MP4 initialisation segment every
	// fMP4 playlist points at with #EXT-X-MAP. Exported because the handler
	// serving playlists has to rewrite that URI onto the segment endpoint.
	HLSInitSegmentName = "init.mp4"

	hlsWorkingDirName = "hls"
	// hlsIdleTimeout has to outlast a pause: a paused player stops fetching
	// segments entirely, and reaping its session under it means the next
	// segment request 404s and playback dies with an unexplained decode error.
	hlsIdleTimeout     = 15 * time.Minute
	hlsMaxSessions     = 8
	hlsJanitorInterval = 30 * time.Second

	// HLSPlaylistWaitTimeout bounds how long a playlist request should wait for
	// ffmpeg to produce the first segment before giving up. Exported so the
	// handler serving playlists uses the same budget the transcode was sized for.
	HLSPlaylistWaitTimeout = 45 * time.Second
)

// HLSSession is one running transcode writing HLS segments to disk.
type HLSSession struct {
	ID        string  // opaque identifier, also the temp directory name
	Owner     string  // username allowed to fetch this session's segments
	Client    string  // opaque id of the player that asked for this transcode
	Source    string  // source file being transcoded
	Dir       string  // directory holding the playlist and its segments
	StartTime float64 // -ss offset this session was started at, in seconds

	cmd    *exec.Cmd
	stderr *tailBuffer   // last few KB of ffmpeg's diagnostics
	exited chan struct{} // closed once the transcode process has been reaped

	mu         sync.Mutex
	lastAccess time.Time
	stopped    bool
}

// tailBuffer keeps the last hlsStderrTailBytes written to it. ffmpeg reports
// why a transcode produced nothing on stderr, and that is the only explanation
// available when a session fails; the whole stream is not worth keeping, but
// the end of it names the reason.
type tailBuffer struct {
	mu   sync.Mutex
	data []byte
}

// hlsStderrTailBytes bounds how much of ffmpeg's stderr is retained per
// session. Enough for the final error plus the surrounding context, small
// enough that eight idle sessions cost nothing worth measuring.
const hlsStderrTailBytes = 4096

func (t *tailBuffer) Write(p []byte) (int, error) {
	t.mu.Lock()
	defer t.mu.Unlock()
	t.data = append(t.data, p...)
	if len(t.data) > hlsStderrTailBytes {
		t.data = t.data[len(t.data)-hlsStderrTailBytes:]
	}
	return len(p), nil
}

func (t *tailBuffer) String() string {
	t.mu.Lock()
	defer t.mu.Unlock()
	return string(t.data)
}

// StderrTail returns the end of the transcode's diagnostic output, for logging
// when the session fails to produce anything playable.
func (s *HLSSession) StderrTail() string {
	if s.stderr == nil {
		return ""
	}
	return strings.TrimSpace(s.stderr.String())
}

// touch records activity so the janitor does not reap a session that is still
// being played.
func (s *HLSSession) touch() {
	s.mu.Lock()
	s.lastAccess = time.Now()
	s.mu.Unlock()
}

func (s *HLSSession) idleFor() time.Duration {
	s.mu.Lock()
	defer s.mu.Unlock()
	return time.Since(s.lastAccess)
}

// PlaylistPath returns the on-disk location of this session's playlist.
func (s *HLSSession) PlaylistPath() string {
	return filepath.Join(s.Dir, hlsPlaylistName)
}

// SegmentPath resolves a segment name from a playlist entry to a path inside
// this session's directory. The name is validated rather than merely cleaned:
// it must look exactly like a name this session generated, so a request can
// never address a file outside the session directory.
func (s *HLSSession) SegmentPath(name string) (string, error) {
	if !validHLSSegmentName(name) {
		return "", errors.New("invalid segment name")
	}
	return filepath.Join(s.Dir, name), nil
}

// stop kills the transcode and removes the session's directory. Safe to call
// more than once.
//
// Only the goroutine started in GetOrCreate ever calls cmd.Wait; this waits on
// the channel that goroutine closes instead, since calling Wait twice on the
// same command is an error.
func (s *HLSSession) stop() {
	s.mu.Lock()
	if s.stopped {
		s.mu.Unlock()
		return
	}
	s.stopped = true
	cmd := s.cmd
	exited := s.exited
	dir := s.Dir
	s.mu.Unlock()

	if cmd != nil && cmd.Process != nil {
		cmd.Process.Kill()
		if exited != nil {
			<-exited
		}
	}
	if dir != "" {
		os.RemoveAll(dir)
	}
}

// validHLSSegmentName reports whether name matches the segment naming this
// package generates ("seg00000.m4s", plus the fixed init segment name), rejecting anything containing a path
// separator, "..", or unexpected characters.
func validHLSSegmentName(name string) bool {
	// The fMP4 initialisation segment is fetched through the same endpoint as
	// the media segments, so it has to be accepted here too. Matching the exact
	// constant keeps the guarantee that only generated names resolve.
	if name == HLSInitSegmentName {
		return true
	}
	if !strings.HasPrefix(name, hlsSegmentPrefix) || !strings.HasSuffix(name, hlsSegmentSuffix) {
		return false
	}
	digits := strings.TrimSuffix(strings.TrimPrefix(name, hlsSegmentPrefix), hlsSegmentSuffix)
	if digits == "" {
		return false
	}
	for _, c := range digits {
		if c < '0' || c > '9' {
			return false
		}
	}
	return true
}

// hlsSessionKey identifies a reusable transcode. Two requests that would
// produce byte-identical output share a session.
func hlsSessionKey(owner string, inputFile string, resolution TranscodeOutputResolution, startTime float64) string {
	raw := strings.Join([]string{
		owner,
		inputFile,
		string(resolution),
		strconv.FormatFloat(startTime, 'f', 3, 64),
	}, "\x00")
	sum := md5.Sum([]byte(raw))
	return hex.EncodeToString(sum[:])
}

// buildHLSArgs assembles the ffmpeg command line for an HLS session. It mirrors
// the encoder selection in TranscodeAndStream - the same hardware profile when
// one is available, libx264 otherwise - and differs only in the muxer.
//
// segmentBaseURL is prepended to every segment name in the playlist, letting the
// segments be fetched from an HTTP endpoint rather than sitting next to the
// playlist on disk.
func buildHLSArgs(inputFile string, dir string, resolution TranscodeOutputResolution, startTime float64, segmentBaseURL string, hw *hwEncoderProfile) ([]string, error) {
	height, err := resolutionHeight(resolution)
	if err != nil {
		return nil, err
	}

	var args []string
	var vf string
	var videoCodecArgs []string
	if hw != nil {
		args = append(args, hw.PreInput...)
		vf = hw.ScaleFilter(height)
		videoCodecArgs = append([]string{"-vcodec", hw.Codec}, hw.EncodeArgs...)
	} else {
		if height != "" {
			vf = "scale=-1:" + height
		}
		// See TranscodeAndStream: without this a 10-bit source yields a High 10
		// stream that most browsers cannot decode.
		videoCodecArgs = []string{"-vcodec", "libx264", "-preset", "superfast", "-pix_fmt", "yuv420p"}
	}

	if startTime > 0.001 {
		// Seeking before -i is the fast path: output timestamps then start at
		// zero, which is what the growing playlist expects.
		args = append(args, "-ss", fmt.Sprintf("%.3f", startTime))
	}
	args = append(args, "-i", inputFile)

	// Take the first video and, if present, the first audio track. Without this
	// a file carrying extra streams (subtitles, attachments, second audio) can
	// fail to mux into the fMP4 segments, or put a track the player does not
	// expect into them.
	args = append(args, "-map", "0:v:0", "-map", "0:a:0?", "-sn", "-dn")

	if vf != "" {
		args = append(args, "-vf", vf)
	}
	args = append(args, videoCodecArgs...)

	// Segments can only be cut on a keyframe, so force one exactly on every
	// segment boundary; otherwise ffmpeg overshoots and segment lengths drift
	// away from hlsSegmentSeconds.
	args = append(args,
		"-force_key_frames", fmt.Sprintf("expr:gte(t,n_forced*%d)", hlsSegmentSeconds),
		"-c:a", "aac", "-b:a", "128k", "-ac", "2",
	)

	args = append(args,
		"-f", "hls",
		"-hls_time", strconv.Itoa(hlsSegmentSeconds),
		"-hls_list_size", "0", // keep every segment in the playlist so seeking back works
		"-hls_playlist_type", "event",
		"-hls_flags", "independent_segments",
		// Fragmented MP4 rather than MPEG-TS. Safari plays either, but fMP4
		// segments can be appended straight into a MediaSource buffer, which is
		// what lets Firefox and Chrome play this stream without a third-party
		// library to demux transport-stream packets first.
		"-hls_segment_type", "fmp4",
		"-hls_fmp4_init_filename", HLSInitSegmentName,
		"-hls_base_url", segmentBaseURL,
		"-hls_segment_filename", hlsOutputPath(dir, hlsSegmentPattern),
		hlsOutputPath(dir, hlsPlaylistName),
	)
	return args, nil
}

// hlsOutputPath builds an output path for the HLS muxer with forward slashes.
//
// ffmpeg places the fMP4 init segment next to the playlist by cutting the
// playlist path at its last '/', and only '/'. Given a native Windows path
// (backslashes) it finds no directory, so init.mp4 lands in ffmpeg's working
// directory instead of the session's, every session overwrites the same file,
// and the player's first request - the init segment - 404s. ffmpeg on Windows
// accepts forward slashes, and on every other OS this changes nothing.
func hlsOutputPath(dir string, name string) string {
	return filepath.ToSlash(filepath.Join(dir, name))
}

// resolutionHeight maps a requested output resolution to an ffmpeg scale
// height, returning an error for anything unrecognised.
func resolutionHeight(resolution TranscodeOutputResolution) (string, error) {
	switch resolution {
	case "360p":
		return "360", nil
	case "720p":
		return "720", nil
	case "1080p":
		return "1080", nil
	case "":
		return "", nil
	}
	return "", errors.New("invalid resolution parameter")
}

// HLSManager owns every live HLS session and the temp directory they live in.
type HLSManager struct {
	root            string // parent directory for all session directories
	segmentEndpoint string // URL path that serves segments back to the player

	mu       sync.Mutex
	sessions map[string]*HLSSession
	closed   bool
	stopChan chan struct{}
}

// NewHLSManager prepares the working directory and starts the reaper. Any
// directory left behind by a previous run is discarded, since the ffmpeg
// processes that were filling those directories died with that run.
func NewHLSManager(tmpDirectory string, segmentEndpoint string) (*HLSManager, error) {
	root := filepath.Join(tmpDirectory, hlsWorkingDirName)
	os.RemoveAll(root)
	if err := os.MkdirAll(root, 0755); err != nil {
		return nil, err
	}
	m := &HLSManager{
		root:            root,
		segmentEndpoint: segmentEndpoint,
		sessions:        map[string]*HLSSession{},
		stopChan:        make(chan struct{}),
	}
	go m.janitor()
	return m, nil
}

// janitor reaps sessions nobody has touched for hlsIdleTimeout.
func (m *HLSManager) janitor() {
	ticker := time.NewTicker(hlsJanitorInterval)
	defer ticker.Stop()
	for {
		select {
		case <-m.stopChan:
			return
		case <-ticker.C:
			m.reapIdle()
		}
	}
}

func (m *HLSManager) reapIdle() {
	m.mu.Lock()
	var expired []*HLSSession
	for key, session := range m.sessions {
		if session.idleFor() > hlsIdleTimeout {
			expired = append(expired, session)
			delete(m.sessions, key)
		}
	}
	m.mu.Unlock()

	for _, session := range expired {
		session.stop()
		logger.PrintAndLog("Transcoder", "Reaped idle HLS session "+session.ID, nil)
	}
}

// Close stops the janitor and tears down every live session.
func (m *HLSManager) Close() {
	m.mu.Lock()
	if m.closed {
		m.mu.Unlock()
		return
	}
	m.closed = true
	close(m.stopChan)
	sessions := make([]*HLSSession, 0, len(m.sessions))
	for key, session := range m.sessions {
		sessions = append(sessions, session)
		delete(m.sessions, key)
	}
	m.mu.Unlock()

	for _, session := range sessions {
		session.stop()
	}
	os.RemoveAll(m.root)
}

// Session returns a live session by ID, or nil. It counts as activity, so
// fetching segments keeps the session from being reaped mid-playback.
func (m *HLSManager) Session(id string) *HLSSession {
	m.mu.Lock()
	defer m.mu.Unlock()
	for _, session := range m.sessions {
		if session.ID == id {
			session.touch()
			return session
		}
	}
	return nil
}

// GetOrCreate returns the session for this exact transcode, starting one if it
// is not already running.
//
// client identifies the player asking, so that a seek - which arrives as a
// request for the same file at a different offset - retires the transcode it
// replaces instead of leaving it running. Pass an empty string when there is no
// such id; the previous transcode of the same file for the same user is then
// treated as the one being replaced.
func (m *HLSManager) GetOrCreate(owner string, client string, inputFile string, resolution TranscodeOutputResolution, startTime float64) (*HLSSession, error) {
	key := hlsSessionKey(owner, inputFile, resolution, startTime)

	m.mu.Lock()
	if m.closed {
		m.mu.Unlock()
		return nil, errors.New("HLS manager closed")
	}
	if existing, ok := m.sessions[key]; ok {
		existing.touch()
		m.mu.Unlock()
		return existing, nil
	}
	m.mu.Unlock()

	// Make room before starting another transcode; each one costs a process.
	m.evictToCapacity()

	session := &HLSSession{
		ID:         key,
		Owner:      owner,
		Client:     client,
		Source:     inputFile,
		Dir:        filepath.Join(m.root, key),
		StartTime:  startTime,
		stderr:     &tailBuffer{},
		lastAccess: time.Now(),
	}
	if err := os.MkdirAll(session.Dir, 0755); err != nil {
		return nil, err
	}

	args, err := buildHLSArgs(inputFile, session.Dir, resolution, startTime,
		m.segmentBaseURL(key), getHWEncoderProfile())
	if err != nil {
		os.RemoveAll(session.Dir)
		return nil, err
	}

	cmd := exec.Command("ffmpeg", args...)
	//Kept rather than discarded: when a transcode produces no playable segment,
	//ffmpeg's own message is the only thing that says why.
	cmd.Stderr = session.stderr
	if err := cmd.Start(); err != nil {
		os.RemoveAll(session.Dir)
		return nil, err
	}
	session.cmd = cmd
	session.exited = make(chan struct{})
	//Sole owner of cmd.Wait: reaps the process whether it finishes the file or
	//is killed, and signals both facts through the same channel.
	go func(exited chan struct{}) {
		cmd.Wait()
		close(exited)
	}(session.exited)

	m.mu.Lock()
	// Another request may have created the same session while ffmpeg was
	// starting; keep the winner and discard this duplicate.
	if existing, ok := m.sessions[key]; ok {
		m.mu.Unlock()
		session.stop()
		existing.touch()
		return existing, nil
	}
	m.sessions[key] = session
	superseded := m.collectSuperseded(key, session)
	m.mu.Unlock()

	for _, old := range superseded {
		old.stop()
		logger.PrintAndLog("Transcoder", "Retired superseded HLS session "+old.ID, nil)
	}
	return session, nil
}

// collectSuperseded removes and returns the sessions the newly created one
// replaces: the same player's earlier transcodes, or - when the player did not
// identify itself - the same user's earlier transcodes of the same file.
//
// Must be called with m.mu held; the returned sessions are stopped by the
// caller once the lock is released, since stopping waits on a process.
func (m *HLSManager) collectSuperseded(newKey string, session *HLSSession) []*HLSSession {
	var superseded []*HLSSession
	for key, candidate := range m.sessions {
		if key == newKey {
			continue
		}
		if candidate.Owner != session.Owner {
			continue
		}
		if session.Client != "" {
			//A player only ever plays one thing at a time, so anything else it
			//started - another offset, or the previous episode - is finished with.
			if candidate.Client != session.Client {
				continue
			}
		} else if candidate.Client != "" || candidate.Source != session.Source {
			//Without an id the most that can be assumed is that the same user
			//re-opened the same file. Never retire a session that does belong to
			//an identified player.
			continue
		}
		superseded = append(superseded, candidate)
		delete(m.sessions, key)
	}
	return superseded
}

// evictToCapacity stops the least recently used sessions until there is room
// for one more.
func (m *HLSManager) evictToCapacity() {
	for {
		m.mu.Lock()
		if len(m.sessions) < hlsMaxSessions {
			m.mu.Unlock()
			return
		}
		var oldestKey string
		var oldest *HLSSession
		for key, session := range m.sessions {
			if oldest == nil || session.idleFor() > oldest.idleFor() {
				oldestKey, oldest = key, session
			}
		}
		if oldest == nil {
			m.mu.Unlock()
			return
		}
		delete(m.sessions, oldestKey)
		m.mu.Unlock()

		oldest.stop()
		logger.PrintAndLog("Transcoder", "Evicted HLS session "+oldest.ID+" to stay within the session limit", nil)
	}
}

func (m *HLSManager) segmentBaseURL(sessionID string) string {
	return m.segmentEndpoint + "?sid=" + sessionID + "&name="
}

// ReadPlaylist returns the session's playlist with the #EXT-X-MAP init segment
// URI rewritten onto the segment endpoint.
//
// -hls_base_url only rewrites media segment URIs; ffmpeg always writes the
// init segment as a bare filename. Left alone it would resolve relative to the
// playlist URL (/media/hls?file=…) and 404, so the rewrite happens here rather
// than leaving every client to work it out.
func (m *HLSManager) ReadPlaylist(session *HLSSession) ([]byte, error) {
	content, err := os.ReadFile(session.PlaylistPath())
	if err != nil {
		return nil, err
	}
	return rewritePlaylistInitURI(content, m.segmentBaseURL(session.ID)+HLSInitSegmentName), nil
}

// rewritePlaylistInitURI replaces the URI inside an #EXT-X-MAP tag. Split out
// from the file read so the substitution can be unit-tested directly.
func rewritePlaylistInitURI(playlist []byte, initURL string) []byte {
	lines := strings.Split(string(playlist), "\n")
	for i, line := range lines {
		if !strings.HasPrefix(strings.TrimSpace(line), "#EXT-X-MAP:") {
			continue
		}
		start := strings.Index(line, `URI="`)
		if start < 0 {
			continue
		}
		start += len(`URI="`)
		end := strings.Index(line[start:], `"`)
		if end < 0 {
			continue
		}
		lines[i] = line[:start] + initURL + line[start+end:]
	}
	return []byte(strings.Join(lines, "\n"))
}

// WaitForPlaylist blocks until the session's playlist lists at least one
// segment, so the player is never handed an empty playlist to choke on.
func (s *HLSSession) WaitForPlaylist(timeout time.Duration) error {
	deadline := time.Now().Add(timeout)
	s.mu.Lock()
	exited := s.exited
	s.mu.Unlock()

	for {
		if playlistHasSegment(s.PlaylistPath()) {
			return nil
		}
		// Check for a dead transcode only after re-checking the playlist above,
		// so a process that exited right after writing its last segment still
		// counts as a success.
		select {
		case <-exited:
			if playlistHasSegment(s.PlaylistPath()) {
				return nil
			}
			return errors.New("transcode ended before producing any segment")
		default:
		}
		if time.Now().After(deadline) {
			return errors.New("timed out waiting for the first segment")
		}
		time.Sleep(200 * time.Millisecond)
	}
}

// playlistHasSegment reports whether the playlist on disk already references a
// segment. ffmpeg writes the header before the first segment is complete, so
// the file existing is not on its own enough.
func playlistHasSegment(path string) bool {
	content, err := os.ReadFile(path)
	if err != nil {
		return false
	}
	return strings.Contains(string(content), hlsSegmentSuffix)
}