RTSP-Streamer/internal/daemon/daemon.go

// Package daemon is the orchestrator: it reads the config, asks the compositor
// for the output geometry, launches one supervised mpv per occupied grid slot,
// tiles them, and exposes a control socket so the layout can be switched live.
package daemon

import (
	"context"
	"encoding/json"
	"fmt"
	"log/slog"
	"net"
	"os"
	"reflect"
	"sync"
	"time"

	"github.com/lwoodard/rtsp-streamer/internal/compositor"
	"github.com/lwoodard/rtsp-streamer/internal/config"
	"github.com/lwoodard/rtsp-streamer/internal/ipc"
	"github.com/lwoodard/rtsp-streamer/internal/player"
	"github.com/lwoodard/rtsp-streamer/internal/protect"
	"github.com/lwoodard/rtsp-streamer/internal/viewmap"
)

// Daemon owns the running video wall.
type Daemon struct {
	cfgPath string
	log     *slog.Logger
	comp    *compositor.Client
	runDir  string

	mu       sync.Mutex
	cfg      *config.Config
	players  []*player.Player
	layout   string
	cancelLo context.CancelFunc // cancels the current layout's supervisors
	wg       sync.WaitGroup
}

// New constructs a daemon bound to a config path.
func New(cfgPath string, log *slog.Logger) (*Daemon, error) {
	comp, err := compositor.New()
	if err != nil {
		return nil, err
	}
	runDir := ipc.RunDir()
	if err := os.MkdirAll(runDir, 0o700); err != nil {
		return nil, err
	}
	return &Daemon{cfgPath: cfgPath, log: log, comp: comp, runDir: runDir}, nil
}

// Run starts the wall and blocks until ctx is cancelled.
func (d *Daemon) Run(ctx context.Context) error {
	if err := d.comp.PrepareForMPV(ctx); err != nil {
		d.log.Warn("could not preinstall mpv window rules", "err", err)
	}
	if err := d.reload(ctx); err != nil {
		return err
	}

	go d.serveControl(ctx)
	go d.healthLoop(ctx)
	go d.viewSyncLoop(ctx)

	<-ctx.Done()
	d.log.Info("shutting down")
	d.stopLayout()
	return nil
}

// resolution returns the render size, from config or the live output.
func (d *Daemon) resolution(ctx context.Context, cfg *config.Config) (int, int) {
	if cfg.Display.Width > 0 && cfg.Display.Height > 0 {
		return cfg.Display.Width, cfg.Display.Height
	}
	if out, err := d.comp.PrimaryOutput(ctx); err == nil && out.CurrentMode.Width > 0 {
		return out.CurrentMode.Width, out.CurrentMode.Height
	}
	d.log.Warn("falling back to 1920x1080; set display.width/height to override")
	return 1920, 1080
}

// reload re-reads config from disk and applies the active layout.
func (d *Daemon) reload(ctx context.Context) error {
	cfg, err := config.Load(d.cfgPath)
	if err != nil {
		return err
	}
	d.mu.Lock()
	d.cfg = cfg
	d.mu.Unlock()
	name := cfg.ActiveLayout
	if name == "" {
		d.log.Warn("no active_layout set; nothing to display")
		d.stopLayout()
		return nil
	}
	return d.applyLayout(ctx, name)
}

// applyLayout tears down the current wall and builds the named layout.
func (d *Daemon) applyLayout(ctx context.Context, name string) error {
	d.mu.Lock()
	cfg := d.cfg
	d.mu.Unlock()

	layout := cfg.LayoutByName(name)
	if layout == nil {
		return fmt.Errorf("layout %q not found", name)
	}
	cols, rows, err := layout.Dimensions()
	if err != nil {
		return err
	}
	w, h := d.resolution(ctx, cfg)
	tiles := layout.EffectiveTiles()

	d.stopLayout()

	loCtx, cancel := context.WithCancel(ctx)
	var players []*player.Player

	for slot, tile := range tiles {
		if tile.Camera == "" {
			continue
		}
		cam := cfg.CameraByName(tile.Camera)
		if cam == nil || cam.Disabled {
			d.log.Warn("skipping tile: camera unavailable", "slot", slot, "camera", tile.Camera)
			continue
		}
		url := cam.StreamURL(tile.Quality)
		if url == "" {
			d.log.Warn("skipping tile: no stream url", "slot", slot, "camera", tile.Camera, "quality", tile.Quality)
			continue
		}
		cs, rs := tile.Span()
		rect := compositor.TileRect(w, h, cols, rows, tile.Col, tile.Row, cs, rs)
		p := player.New(slot, cam.Name, url, cfg.Player, d.runDir, d.log)
		players = append(players, p)

		d.wg.Add(1)
		go func() { defer d.wg.Done(); p.Supervise(loCtx) }()

		// Position the window once it maps. Done per-tile so a slow camera
		// doesn't block the others.
		d.wg.Add(1)
		go func(p *player.Player, rect compositor.Rect) {
			defer d.wg.Done()
			d.placeWhenReady(loCtx, p, rect)
		}(p, rect)
	}

	d.mu.Lock()
	d.players = players
	d.layout = name
	d.cancelLo = cancel
	d.cfg.ActiveLayout = name
	d.mu.Unlock()
	d.log.Info("layout applied", "layout", name, "tiles", len(players), "grid", layout.Grid, "res", fmt.Sprintf("%dx%d", w, h))
	return nil
}

// placeWhenReady waits for the mpv window to map, tiles it, then keeps
// re-asserting its geometry on a timer. The re-assertion matters because mpv
// resizes its own window to the camera's native resolution when the stream
// loads (and mpv 0.35 has no flag to disable that); re-issuing the sway
// resize snaps the window back into its cell, portably across mpv versions.
func (d *Daemon) placeWhenReady(ctx context.Context, p *player.Player, rect compositor.Rect) {
	var readyPID int
	ticker := time.NewTicker(1 * time.Second)
	defer ticker.Stop()
	for {
		if ctx.Err() != nil {
			return
		}
		pid := p.PID()
		if pid != 0 {
			// New process: wait for its window to map before positioning.
			if pid != readyPID {
				if err := d.comp.WaitForWindow(ctx, pid, 15*time.Second); err != nil {
					goto wait
				}
				readyPID = pid
				d.log.Debug("window mapped", "slot", p.Slot, "pid", pid, "rect", rect)
			}
			// Re-assert geometry every tick to override mpv's auto-resize.
			if err := d.comp.Place(ctx, pid, rect); err != nil {
				d.log.Debug("place failed", "slot", p.Slot, "err", err)
			}
		}
	wait:
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
		}
	}
}

// stopLayout cancels supervisors and kills current mpv processes.
func (d *Daemon) stopLayout() {
	d.mu.Lock()
	cancel := d.cancelLo
	players := d.players
	d.cancelLo = nil
	d.players = nil
	d.mu.Unlock()

	if cancel != nil {
		cancel()
	}
	for _, p := range players {
		p.Stop()
	}
	d.wg.Wait()
}

// healthLoop periodically nudges stalled streams. Supervise already restarts
// exited mpv; this catches the "process alive but frozen" case.
func (d *Daemon) healthLoop(ctx context.Context) {
	ticker := time.NewTicker(20 * time.Second)
	defer ticker.Stop()
	stalls := map[int]int{}
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
		}
		d.mu.Lock()
		players := append([]*player.Player(nil), d.players...)
		d.mu.Unlock()
		for _, p := range players {
			if !p.Running() {
				continue
			}
			if p.Healthy() {
				stalls[p.Slot] = 0
				continue
			}
			stalls[p.Slot]++
			// Require three consecutive unresponsive probes (~60s) before
			// forcing a restart, so a brief IPC hiccup never bounces a stream.
			if stalls[p.Slot] >= 3 {
				d.log.Warn("stream hung, forcing restart", "slot", p.Slot, "camera", p.Name, "misses", stalls[p.Slot])
				p.Stop() // Supervise relaunches
				stalls[p.Slot] = 0
			}
		}
	}
}

// viewSyncLoop periodically re-syncs the active layout from its linked UniFi
// Protect live view, so edits made in Protect (cameras, slot order) show up on
// the wall without manual re-import. Off unless view_refresh_seconds > 0 and
// controller credentials are available to the daemon process.
func (d *Daemon) viewSyncLoop(ctx context.Context) {
	d.mu.Lock()
	interval := d.cfg.ViewRefreshSeconds
	d.mu.Unlock()
	if interval <= 0 {
		return
	}
	ticker := time.NewTicker(time.Duration(interval) * time.Second)
	defer ticker.Stop()
	warnedNoCreds := false
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
		}
		if err := d.syncActiveView(ctx); err != nil {
			if err == errNoCreds {
				if !warnedNoCreds {
					d.log.Warn("view sync enabled but no controller password available to the daemon; skipping")
					warnedNoCreds = true
				}
				continue
			}
			d.log.Warn("view sync failed", "err", err)
		}
	}
}

var errNoCreds = fmt.Errorf("no controller credentials")

// syncActiveView refetches the linked live view and re-applies the layout if it
// changed. No-op when the active layout isn't linked to a Protect view.
func (d *Daemon) syncActiveView(ctx context.Context) error {
	d.mu.Lock()
	cfg := d.cfg
	name := d.layout
	d.mu.Unlock()

	active := cfg.LayoutByName(name)
	if active == nil || active.ProtectView == "" {
		return nil
	}
	if cfg.Controller.Host == "" || cfg.Controller.ResolvePassword() == "" {
		return errNoCreds
	}
	cl, err := protect.New(cfg.Controller.Host, cfg.Controller.RTSPPort, cfg.Controller.VerifyTLS)
	if err != nil {
		return err
	}
	if err := cl.Login(ctx, cfg.Controller.Username, cfg.Controller.ResolvePassword()); err != nil {
		return err
	}
	views, _, err := cl.LiveViews(ctx)
	if err != nil {
		return err
	}
	var view *protect.LiveView
	for i := range views {
		if views[i].Name == active.ProtectView {
			view = &views[i]
			break
		}
	}
	if view == nil {
		return fmt.Errorf("linked view %q no longer exists on the controller", active.ProtectView)
	}
	idToName := map[string]string{}
	for _, cam := range cfg.Cameras {
		if cam.ID != "" {
			idToName[cam.ID] = cam.Name
		}
	}
	rebuilt, _ := viewmap.LayoutFromView(*view, idToName)
	rebuilt.Name = active.Name // keep our layout name stable

	if rebuilt.Grid == active.Grid && reflect.DeepEqual(rebuilt.Tiles, active.Tiles) {
		return nil // unchanged
	}

	d.log.Info("live view changed, re-syncing layout", "view", active.ProtectView, "layout", name)
	d.mu.Lock()
	if l := d.cfg.LayoutByName(name); l != nil {
		l.Grid, l.Tiles = rebuilt.Grid, rebuilt.Tiles
	}
	d.mu.Unlock()
	if err := config.Save(d.cfgPath, cfg); err != nil {
		d.log.Warn("could not persist synced layout", "err", err)
	}
	return d.applyLayout(ctx, name)
}

// serveControl accepts control-socket connections for status/reload/set-layout.
func (d *Daemon) serveControl(ctx context.Context) {
	path := ipc.SocketPath()
	_ = os.Remove(path)
	ln, err := net.Listen("unix", path)
	if err != nil {
		d.log.Error("control socket listen failed", "err", err)
		return
	}
	go func() { <-ctx.Done(); ln.Close(); os.Remove(path) }()
	d.log.Info("control socket ready", "path", path)
	for {
		conn, err := ln.Accept()
		if err != nil {
			if ctx.Err() != nil {
				return
			}
			continue
		}
		go d.handleControl(ctx, conn)
	}
}

func (d *Daemon) handleControl(ctx context.Context, conn net.Conn) {
	defer conn.Close()
	_ = conn.SetDeadline(time.Now().Add(30 * time.Second))
	var req ipc.Request
	if err := json.NewDecoder(conn).Decode(&req); err != nil {
		return
	}
	resp := d.dispatch(ctx, req)
	_ = json.NewEncoder(conn).Encode(resp)
}

func (d *Daemon) dispatch(ctx context.Context, req ipc.Request) ipc.Response {
	switch req.Cmd {
	case "status":
		return d.status()
	case "reload":
		if err := d.reload(ctx); err != nil {
			return ipc.Response{OK: false, Error: err.Error()}
		}
		return d.status()
	case "set-layout":
		if err := d.applyLayout(ctx, req.Name); err != nil {
			return ipc.Response{OK: false, Error: err.Error()}
		}
		// Persist the choice so a restart keeps it.
		if err := d.persistActiveLayout(req.Name); err != nil {
			d.log.Warn("could not persist active layout", "err", err)
		}
		return d.status()
	default:
		return ipc.Response{OK: false, Error: fmt.Sprintf("unknown command %q", req.Cmd)}
	}
}

func (d *Daemon) persistActiveLayout(name string) error {
	cfg, err := config.Load(d.cfgPath)
	if err != nil {
		return err
	}
	cfg.ActiveLayout = name
	return config.Save(d.cfgPath, cfg)
}

func (d *Daemon) status() ipc.Response {
	d.mu.Lock()
	defer d.mu.Unlock()
	resp := ipc.Response{OK: true, ActiveLayout: d.layout}
	for _, p := range d.players {
		resp.Slots = append(resp.Slots, ipc.SlotStatus{
			Slot:    p.Slot,
			Camera:  p.Name,
			PID:     p.PID(),
			Running: p.Running(),
			Healthy: p.Healthy(),
		})
	}
	return resp
}