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"
	"sync"
	"syscall"
	"time"

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

// 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

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

	mu      sync.Mutex
	cmd     *exec.Cmd
	started time.Time
}

// 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
		// 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.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()...)
	// Inherit the caller's environment (WAYLAND_DISPLAY etc. must be set).
	cmd.Env = os.Environ()
	// Discard mpv's chatty stdout/stderr; errors surface via exit code.
	cmd.Stdout = nil
	cmd.Stderr = nil
	if err := cmd.Start(); err != nil {
		return fmt.Errorf("starting mpv: %w", err)
	}
	p.mu.Lock()
	p.cmd = cmd
	p.started = time.Now()
	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
		p.mu.Unlock()
		err := cmd.Wait()

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

		if ctx.Err() != nil {
			return
		}
		p.log.Warn("mpv exited, will restart", "err", err, "after", backoff)
		if !sleep(ctx, backoff) {
			return
		}
	}
}

// Stop terminates the mpv process (SIGTERM, then SIGKILL after a grace period).
func (p *Player) Stop() {
	p.mu.Lock()
	cmd := p.cmd
	p.mu.Unlock()
	if cmd == nil || cmd.Process == nil {
		return
	}
	_ = cmd.Process.Signal(syscall.SIGTERM)
	done := make(chan struct{})
	go func() { _, _ = cmd.Process.Wait(); close(done) }()
	select {
	case <-done:
	case <-time.After(2 * time.Second):
		_ = cmd.Process.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) {
	conn, err := net.DialTimeout("unix", p.ipcPath, 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"
}

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
	}
}