vibesonic/internal/jukebox/mpv.go

// Package jukebox implements Subsonic's jukeboxControl endpoint using
// mpv as the actual server-side playback backend, driven over mpv's
// JSON IPC socket (--input-ipc-server). If mpv isn't installed, the
// jukebox is simply unavailable — VibeSonic never fakes playback.
package jukebox

import (
	"bufio"
	"encoding/json"
	"fmt"
	"net"
	"os"
	"os/exec"
	"path/filepath"
	"sync"
	"sync/atomic"
	"time"
)

// mpvClient owns one mpv process and its JSON IPC connection.
type mpvClient struct {
	mpvPath    string
	socketPath string

	mu    sync.Mutex
	cmd   *exec.Cmd
	conn  net.Conn
	reqID int64

	// pending correlates in-flight requests (by request_id) to the
	// goroutine waiting for their response; mpv's IPC is a single
	// bidirectional stream so responses must be demultiplexed.
	pendingMu sync.Mutex
	pending   map[int64]chan mpvResponse

	closed atomic.Bool
}

type mpvResponse struct {
	Error     string `json:"error"`
	Data      any    `json:"data"`
	RequestID int64  `json:"request_id"`
}

func newMPVClient(mpvPath string) *mpvClient {
	if mpvPath == "" {
		mpvPath = "mpv"
	}
	return &mpvClient{mpvPath: mpvPath, pending: make(map[int64]chan mpvResponse)}
}

func (c *mpvClient) start() error {
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.cmd != nil {
		return nil // already running
	}

	dir, err := os.MkdirTemp("", "vibesonic-mpv-")
	if err != nil {
		return err
	}
	c.socketPath = filepath.Join(dir, "mpv.sock")

	cmd := exec.Command(c.mpvPath,
		"--idle=yes",
		"--no-video",
		"--no-terminal",
		"--really-quiet",
		"--input-ipc-server="+c.socketPath,
	)
	if err := cmd.Start(); err != nil {
		os.RemoveAll(dir)
		return fmt.Errorf("starting mpv: %w", err)
	}
	c.cmd = cmd

	var conn net.Conn
	deadline := time.Now().Add(5 * time.Second)
	for time.Now().Before(deadline) {
		conn, err = net.Dial("unix", c.socketPath)
		if err == nil {
			break
		}
		time.Sleep(50 * time.Millisecond)
	}
	if conn == nil {
		_ = cmd.Process.Kill()
		os.RemoveAll(dir)
		return fmt.Errorf("connecting to mpv IPC socket: %w", err)
	}
	c.conn = conn

	go c.readLoop()
	return nil
}

func (c *mpvClient) readLoop() {
	scanner := bufio.NewScanner(c.conn)
	scanner.Buffer(make([]byte, 0, 64*1024), 1<<20)
	for scanner.Scan() {
		var resp mpvResponse
		if err := json.Unmarshal(scanner.Bytes(), &resp); err != nil {
			continue // ignore event lines / anything we can't parse
		}
		if resp.RequestID == 0 {
			continue // an mpv event, not a command reply
		}
		c.pendingMu.Lock()
		ch, ok := c.pending[resp.RequestID]
		if ok {
			delete(c.pending, resp.RequestID)
		}
		c.pendingMu.Unlock()
		if ok {
			ch <- resp
		}
	}
}

func (c *mpvClient) command(args ...any) (any, error) {
	if err := c.start(); err != nil {
		return nil, err
	}
	id := atomic.AddInt64(&c.reqID, 1)
	payload := map[string]any{"command": args, "request_id": id}
	encoded, err := json.Marshal(payload)
	if err != nil {
		return nil, err
	}

	ch := make(chan mpvResponse, 1)
	c.pendingMu.Lock()
	c.pending[id] = ch
	c.pendingMu.Unlock()

	c.mu.Lock()
	_, werr := c.conn.Write(append(encoded, '\n'))
	c.mu.Unlock()
	if werr != nil {
		c.pendingMu.Lock()
		delete(c.pending, id)
		c.pendingMu.Unlock()
		return nil, werr
	}

	select {
	case resp := <-ch:
		if resp.Error != "" && resp.Error != "success" {
			return nil, fmt.Errorf("mpv: %s", resp.Error)
		}
		return resp.Data, nil
	case <-time.After(5 * time.Second):
		c.pendingMu.Lock()
		delete(c.pending, id)
		c.pendingMu.Unlock()
		return nil, fmt.Errorf("mpv command timed out")
	}
}

func (c *mpvClient) getProperty(name string) (any, error) {
	return c.command("get_property", name)
}

func (c *mpvClient) setProperty(name string, value any) error {
	_, err := c.command("set_property", name, value)
	return err
}

func (c *mpvClient) close() {
	if c.closed.Swap(true) {
		return
	}
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.conn != nil {
		c.conn.Close()
	}
	if c.cmd != nil && c.cmd.Process != nil {
		_ = c.cmd.Process.Kill()
	}
	if c.socketPath != "" {
		os.RemoveAll(filepath.Dir(c.socketPath))
	}
}