RTSP-Streamer/internal/player/player.go

// Package player manages one mpv process per camera stream. Each stream runs
// independently so a single dead camera never disturbs the rest of the wall;
// the manager relaunches only the slot that failed, with backoff.
package player

import (
	"bufio"
	"context"
	"encoding/json"
	"fmt"
	"log/slog"
	"net"
	"os"
	"os/exec"
	"path/filepath"
	"regexp"
	"strings"
	"sync"
	"syscall"
	"time"

	"github.com/lwoodard/rtsp-streamer/internal/config"
	"github.com/lwoodard/rtsp-streamer/internal/logbuf"
)

// Player supervises a single mpv instance bound to one grid slot.
type Player struct {
	Slot int    // grid cell index (row-major)
	Name string // camera name, for logs/titles
	URL  string // RTSP(S) source
	// TileW/TileH, when set (before Supervise), are passed as --geometry so
	// mpv opens its window at the tile size instead of the video's native
	// size — a restarting stream then maps already-sized for its cell.
	TileW, TileH int

	cfg     config.Player
	ipcPath string
	runDir  string
	log     *slog.Logger

	mu      sync.Mutex
	cmd     *exec.Cmd
	started time.Time
	stderr  *logbuf.Writer // tail of the current process's stderr
}

// New creates a player. runDir is where the mpv IPC socket lives.
func New(slot int, name, url string, cfg config.Player, runDir string, log *slog.Logger) *Player {
	return &Player{
		Slot:    slot,
		Name:    name,
		URL:     url,
		cfg:     cfg,
		runDir:  runDir,
		ipcPath: filepath.Join(runDir, fmt.Sprintf("mpv-slot-%d.sock", slot)),
		log:     log.With("slot", slot, "camera", name),
	}
}

// Title is the window title mpv advertises; the compositor could match on it,
// though we prefer matching by PID.
func (p *Player) Title() string {
	return fmt.Sprintf("rtsp-streamer:slot-%d", p.Slot)
}

func (p *Player) args() []string {
	args := []string{
		"--no-config",
		"--force-window=yes",
		"--idle=no",
		"--keep-open=no",
		"--no-osc",
		"--no-input-default-bindings",
		"--input-cursor=no",
		"--cursor-autohide=always",
		"--no-border",
		"--fullscreen=no", // we tile via the compositor, not fullscreen
		// Reconnect in-process when the RTSP stream ends. UniFi Protect closes
		// a camera's RTSP connection periodically (every 20-40s on some
		// models); without this mpv hits EOF, exits cleanly, and the
		// supervisor relaunches it ~3s later — a visible drop. loop-file=inf
		// reopens the same URL the instant it ends, keeping the window alive,
		// so recovery is sub-second with no teardown.
		"--loop-file=inf",
		// Fit any camera into its tile regardless of the camera's aspect:
		//  * keepaspect=yes         → letterbox the video, never crop/stretch
		//  * keepaspect-window=no   → do NOT reshape the window to the video's
		//    aspect; accept the tile size the compositor assigns. Without this
		//    mpv grows the window to e.g. 4:3 and the overflow is hidden under
		//    the neighbouring tile ("bottom cut off").
		// mpv still resizes the window to the video's pixel size on load (mpv
		// 0.35 lacks --auto-window-resize), so the daemon re-asserts each
		// tile's geometry on a timer.
		"--keepaspect=yes",
		"--keepaspect-window=no",
		"--title=" + p.Title(),
		"--input-ipc-server=" + p.ipcPath,
		"--hwdec=" + p.cfg.HWDec,
		// Live-stream hygiene: TCP transport, no caching, and drop frames when
		// behind so latency self-corrects instead of accumulating a backlog.
		"--rtsp-transport=tcp",
		"--profile=low-latency",
		"--cache=no",
		"--framedrop=decoder+vo",
		"--demuxer-lavf-o=stimeout=5000000,fflags=+nobuffer,flags=+low_delay",
	}
	if !p.cfg.Audio {
		// A wall of simultaneous feeds is unwatchable with sound, and skipping
		// the audio decoder saves CPU per stream on the Pi.
		args = append(args, "--no-audio")
	}
	if p.TileW > 0 && p.TileH > 0 {
		// Open the window at the tile size; --geometry overrides mpv's
		// resize-to-video-size on load, so the compositor never has to snap
		// the window back into its cell.
		args = append(args, fmt.Sprintf("--geometry=%dx%d", p.TileW, p.TileH))
	}
	if p.cfg.MaxFPS > 0 {
		args = append(args, fmt.Sprintf("--vf=fps=%d", p.cfg.MaxFPS))
	}
	if p.cfg.Profile != "" && p.cfg.Profile != "low-latency" {
		args = append(args, "--profile="+p.cfg.Profile)
	}
	args = append(args, p.cfg.ExtraArgs...)
	args = append(args, p.URL)
	return args
}

// PID returns the running mpv process id, or 0 if not running.
func (p *Player) PID() int {
	p.mu.Lock()
	defer p.mu.Unlock()
	if p.cmd == nil || p.cmd.Process == nil {
		return 0
	}
	return p.cmd.Process.Pid
}

// Running reports whether the process is currently alive.
func (p *Player) Running() bool {
	return p.PID() != 0
}

// start launches mpv once. Caller owns retry/backoff.
func (p *Player) start(ctx context.Context) error {
	if err := os.Remove(p.ipcPath); err != nil && !os.IsNotExist(err) {
		p.log.Warn("stale ipc socket", "err", err)
	}
	cmd := exec.CommandContext(ctx, "mpv", p.args()...)
	// On ctx cancel, ask mpv to exit cleanly first; WaitDelay escalates to
	// SIGKILL if it hasn't gone away shortly after. This makes layout teardown
	// graceful and parallel — every supervisor reaps its own process.
	cmd.Cancel = func() error { return cmd.Process.Signal(syscall.SIGTERM) }
	cmd.WaitDelay = 2 * time.Second
	// Inherit the caller's environment (WAYLAND_DISPLAY etc. must be set).
	cmd.Env = os.Environ()
	// Keep the tail of mpv's stderr so a crash's reason (bad codec, dropped
	// connection, auth failure) is logged when the process exits, instead of
	// vanishing. Cheap: a small ring, not the whole firehose.
	stderr := logbuf.New(30)
	cmd.Stdout = nil
	cmd.Stderr = stderr
	if err := cmd.Start(); err != nil {
		return fmt.Errorf("starting mpv: %w", err)
	}
	p.mu.Lock()
	p.cmd = cmd
	p.started = time.Now()
	p.stderr = stderr
	p.mu.Unlock()
	p.log.Info("mpv started", "pid", cmd.Process.Pid)
	return nil
}

// Supervise runs mpv and relaunches it whenever it exits, until ctx is
// cancelled. Backoff prevents a hot loop when a camera is unreachable.
func (p *Player) Supervise(ctx context.Context) {
	backoff := time.Duration(p.cfg.RestartBackoffSeconds) * time.Second
	if backoff <= 0 {
		backoff = 3 * time.Second
	}
	for {
		if ctx.Err() != nil {
			return
		}
		if err := p.start(ctx); err != nil {
			p.log.Error("failed to start mpv", "err", err)
			if !sleep(ctx, backoff) {
				return
			}
			continue
		}
		p.mu.Lock()
		cmd := p.cmd
		stderr := p.stderr
		p.mu.Unlock()
		err := cmd.Wait()

		p.mu.Lock()
		p.cmd = nil
		p.mu.Unlock()

		if ctx.Err() != nil {
			return
		}
		// Surface the last line(s) mpv printed — that's where the failure
		// reason lives (connection refused, unsupported codec, 401, ...).
		reason := lastLine(stderr)
		p.log.Warn("mpv exited, will restart", "err", err, "after", backoff, "reason", reason)
		if !sleep(ctx, backoff) {
			return
		}
	}
}

// Stop asks the mpv process to exit (SIGTERM, escalating to SIGKILL if it
// ignores it). It never calls Wait — Supervise owns the reap, notices the
// exit, and relaunches unless its context has been cancelled. Watching p.cmd
// (which Supervise clears after reaping) instead of double-Waiting avoids a
// race on the process handle.
func (p *Player) Stop() {
	p.mu.Lock()
	cmd := p.cmd
	p.mu.Unlock()
	if cmd == nil || cmd.Process == nil {
		return
	}
	proc := cmd.Process
	_ = proc.Signal(syscall.SIGTERM)
	go func() {
		deadline := time.Now().Add(2 * time.Second)
		for time.Now().Before(deadline) {
			p.mu.Lock()
			alive := p.cmd == cmd
			p.mu.Unlock()
			if !alive {
				return
			}
			time.Sleep(100 * time.Millisecond)
		}
		_ = proc.Kill()
	}()
}

// Command sends a JSON IPC command to mpv and returns the decoded reply. Used
// for health probes and live property changes.
func (p *Player) Command(args ...any) (map[string]any, error) {
	return ipcCommand(p.ipcPath, args...)
}

// ipcCommand sends one JSON command to an mpv IPC socket and returns the
// decoded reply. Shared by Player (health probes, reload) and Clock (overlay
// updates).
func ipcCommand(path string, args ...any) (map[string]any, error) {
	conn, err := net.DialTimeout("unix", path, 2*time.Second)
	if err != nil {
		return nil, err
	}
	defer conn.Close()
	_ = conn.SetDeadline(time.Now().Add(2 * time.Second))

	payload, err := json.Marshal(map[string]any{"command": args})
	if err != nil {
		return nil, err
	}
	if _, err := conn.Write(append(payload, '\n')); err != nil {
		return nil, err
	}
	scanner := bufio.NewScanner(conn)
	for scanner.Scan() {
		var reply map[string]any
		if err := json.Unmarshal(scanner.Bytes(), &reply); err != nil {
			continue
		}
		// mpv emits async events too; the command reply carries "error".
		if _, ok := reply["error"]; ok {
			return reply, nil
		}
	}
	return nil, fmt.Errorf("no reply from mpv ipc")
}

// Healthy probes mpv over IPC and reports whether it is responsive. It queries
// a property that always exists (mpv-version) rather than time-pos, because a
// live stream with caching disabled can legitimately report time-pos as
// unavailable while playing fine — using it here caused false "unhealthy"
// verdicts and needless restarts. A false result now means mpv's IPC didn't
// answer at all, i.e. the process is genuinely hung.
func (p *Player) Healthy() bool {
	if !p.Running() {
		return false
	}
	reply, err := p.Command("get_property", "mpv-version")
	if err != nil {
		return false
	}
	return reply["error"] == "success"
}

// ansiRE matches terminal escape sequences mpv sprinkles into its output.
var ansiRE = regexp.MustCompile(`\x1b\[[0-9;?]*[ -/]*[@-~]`)

// lastLine returns the most useful recent stderr line mpv emitted: the last
// line that is not one of mpv's periodic playback-status prints (e.g.
// "V: 00:00:35 / 00:00:35 (100%)"), with ANSI escapes stripped. A real error
// (connection refused, unsupported codec, 401) is what we want to surface; the
// status line is only the fallback when nothing else was printed.
func lastLine(w *logbuf.Writer) string {
	if w == nil {
		return ""
	}
	lines := w.Lines(0)
	fallback := ""
	for i := len(lines) - 1; i >= 0; i-- {
		s := strings.TrimSpace(ansiRE.ReplaceAllString(lines[i], ""))
		if s == "" {
			continue
		}
		if fallback == "" {
			fallback = s
		}
		if isStatusLine(s) {
			continue
		}
		return s
	}
	return fallback
}

// isStatusLine reports whether s is one of mpv's transient A/V progress prints
// rather than a substantive message.
func isStatusLine(s string) bool {
	for _, p := range []string{"V:", "A:", "AV:", "(Paused)", "(Buffering)"} {
		if strings.HasPrefix(s, p) {
			return true
		}
	}
	return false
}

// Reload reconnects mpv to its stream, snapping playback back to the live
// edge. Live RTSP can't be seeked, so latency that accumulates when the Pi
// runs a hair behind real-time is only cleared by reopening the stream. The
// daemon calls this on a stagger so drift stays bounded without a visible
// wall-wide blip. It reuses the running mpv process (no window teardown).
func (p *Player) Reload() error {
	reply, err := p.Command("loadfile", p.URL, "replace")
	if err != nil {
		return err
	}
	if reply["error"] != "success" {
		return fmt.Errorf("mpv loadfile: %v", reply["error"])
	}
	return nil
}

func sleep(ctx context.Context, d time.Duration) bool {
	t := time.NewTimer(d)
	defer t.Stop()
	select {
	case <-ctx.Done():
		return false
	case <-t.C:
		return true
	}
}