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
	wallSig  string             // signature of the running wall, to skip no-op reloads
	cancelLo context.CancelFunc // cancels the current layout's supervisors
	wg       sync.WaitGroup
	pcl      *protect.Client // cached Protect session for view sync
}

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

// tileSpec is one resolved tile: which stream goes where. It doubles as the
// unit of the wall signature — if the specs (plus resolution and player
// settings) are unchanged, a reload need not touch the running streams.
type tileSpec struct {
	slot int
	name string
	url  string
	rect compositor.Rect
}

// resolveTiles turns the named layout into concrete tile specs and the wall
// signature for change detection.
func (d *Daemon) resolveTiles(ctx context.Context, cfg *config.Config, name string) ([]tileSpec, string, error) {
	layout := cfg.LayoutByName(name)
	if layout == nil {
		return nil, "", fmt.Errorf("layout %q not found", name)
	}
	cols, rows, err := layout.Dimensions()
	if err != nil {
		return nil, "", err
	}
	w, h := d.resolution(ctx, cfg)

	var specs []tileSpec
	for slot, tile := range layout.EffectiveTiles() {
		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)
		specs = append(specs, tileSpec{slot: slot, name: cam.Name, url: url, rect: rect})
	}
	sig := fmt.Sprintf("%s|%dx%d|%+v|%v", name, w, h, cfg.Player, specs)
	return specs, sig, nil
}

// applyLayout builds the named layout. When the resolved wall is identical to
// the one already running (same streams, geometry, and player settings), it
// leaves the streams alone — so saving an unrelated config edit never blanks
// the screen.
func (d *Daemon) applyLayout(ctx context.Context, name string) error {
	d.mu.Lock()
	cfg := d.cfg
	d.mu.Unlock()

	specs, sig, err := d.resolveTiles(ctx, cfg, name)
	if err != nil {
		return err
	}

	d.mu.Lock()
	unchanged := d.layout == name && d.wallSig == sig && len(d.players) > 0
	d.mu.Unlock()
	if unchanged {
		d.log.Info("layout unchanged; leaving streams running", "layout", name)
		return nil
	}

	d.stopLayout()

	loCtx, cancel := context.WithCancel(ctx)
	players := make([]*player.Player, 0, len(specs))
	places := make([]placement, 0, len(specs))

	for _, s := range specs {
		p := player.New(s.slot, s.name, s.url, cfg.Player, d.runDir, d.log)
		players = append(players, p)
		places = append(places, placement{p: p, rect: s.rect})

		d.wg.Add(1)
		go func() { defer d.wg.Done(); p.Supervise(loCtx) }()
	}
	if len(places) > 0 {
		d.wg.Add(1)
		go func() { defer d.wg.Done(); d.placeLoop(loCtx, places) }()
	}

	d.mu.Lock()
	d.players = players
	d.layout = name
	d.wallSig = sig
	d.cancelLo = cancel
	d.cfg.ActiveLayout = name
	d.mu.Unlock()
	d.log.Info("layout applied", "layout", name, "tiles", len(players))
	return nil
}

// placement pairs a player with the rectangle its window belongs in.
type placement struct {
	p    *player.Player
	rect compositor.Rect
}

// placeLoop keeps every tile's window at its assigned rectangle using one
// get_tree snapshot per tick, re-placing only windows that drifted. mpv
// resizes its own window to the video's native size when a stream loads (mpv
// 0.35 has no flag to disable that), so drift is expected on every stream
// (re)start; comparing against the snapshot means a settled wall costs one
// swaymsg per tick instead of one per tile. The tick runs fast while windows
// are mapping or drifting and relaxes once everything has been in place for a
// few consecutive ticks.
func (d *Daemon) placeLoop(ctx context.Context, places []placement) {
	const fast, slow = 500 * time.Millisecond, 2 * time.Second
	settled := 0
	mapped := map[int]bool{} // pids whose window we have seen mapped
	for {
		interval := fast
		if settled >= 3 {
			interval = slow
		}
		if !sleepCtx(ctx, interval) {
			return
		}
		rects, err := d.comp.WindowRects(ctx)
		if err != nil {
			d.log.Debug("get_tree failed", "err", err)
			continue
		}
		clean := true
		for _, pl := range places {
			pid := pl.p.PID()
			if pid == 0 {
				continue // not running (backoff); nothing to place yet
			}
			actual, ok := rects[pid]
			if !ok {
				// Process is up but its window hasn't mapped: poll fast so it
				// lands in its cell the moment it appears.
				if !mapped[pid] {
					clean = false
				}
				continue
			}
			if !mapped[pid] {
				mapped[pid] = true
				d.log.Debug("window mapped", "slot", pl.p.Slot, "pid", pid, "rect", pl.rect)
			}
			if actual != pl.rect {
				clean = false
				if err := d.comp.Place(ctx, pid, pl.rect); err != nil {
					d.log.Debug("place failed", "slot", pl.p.Slot, "err", err)
				}
			}
		}
		if clean {
			settled++
		} else {
			settled = 0
		}
		// Drop map entries for long-gone pids so restart churn over weeks of
		// uptime doesn't grow the set unboundedly.
		if len(mapped) > 2*len(places) {
			current := make(map[int]bool, len(places))
			for _, pl := range places {
				current[pl.p.PID()] = true
			}
			for pid := range mapped {
				if !current[pid] {
					delete(mapped, pid)
				}
			}
		}
	}
}

// sleepCtx sleeps for d or until ctx is cancelled; false means cancelled.
func sleepCtx(ctx context.Context, dur time.Duration) bool {
	t := time.NewTimer(dur)
	defer t.Stop()
	select {
	case <-ctx.Done():
		return false
	case <-t.C:
		return true
	}
}

// stopLayout tears down the current wall. Cancelling the layout context
// SIGTERMs every mpv concurrently (each supervisor's cmd.Cancel, escalating
// to SIGKILL via WaitDelay) and stops the placement loop; the WaitGroup
// confirms every process has been reaped before we build the next layout.
func (d *Daemon) stopLayout() {
	d.mu.Lock()
	cancel := d.cancelLo
	d.cancelLo = nil
	d.players = nil
	d.wallSig = ""
	d.mu.Unlock()

	if cancel != nil {
		cancel()
	}
	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()
	// Keyed by player (not slot) so strikes never carry over to a different
	// stream after a layout switch; stale entries are pruned each tick.
	stalls := map[*player.Player]int{}
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
		}
		d.mu.Lock()
		players := append([]*player.Player(nil), d.players...)
		d.mu.Unlock()
		current := make(map[*player.Player]bool, len(players))
		for _, p := range players {
			current[p] = true
			if !p.Running() {
				continue
			}
			if p.Healthy() {
				stalls[p] = 0
				continue
			}
			stalls[p]++
			// Require three consecutive unresponsive probes (~60s) before
			// forcing a restart, so a brief IPC hiccup never bounces a stream.
			if stalls[p] >= 3 {
				d.log.Warn("stream hung, forcing restart", "slot", p.Slot, "camera", p.Name, "misses", stalls[p])
				p.Stop() // Supervise relaunches
				stalls[p] = 0
			}
		}
		for p := range stalls {
			if !current[p] {
				delete(stalls, p)
			}
		}
	}
}

// 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) {
	warnedNoCreds := false
	for {
		// Re-read the interval every pass so enabling or tuning
		// view_refresh_seconds takes effect on reload, without a restart.
		d.mu.Lock()
		interval := d.cfg.ViewRefreshSeconds
		d.mu.Unlock()
		wait := time.Duration(interval) * time.Second
		if interval <= 0 {
			wait = 30 * time.Second // idle poll of the config; sync stays off
		}
		if !sleepCtx(ctx, wait) {
			return
		}
		if interval <= 0 {
			continue
		}
		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")

// loginProtect builds a fresh authenticated Protect client and caches it for
// subsequent sync ticks.
func (d *Daemon) loginProtect(ctx context.Context, cfg *config.Config) (*protect.Client, error) {
	cl, err := protect.New(cfg.Controller.Host, cfg.Controller.RTSPPort, cfg.Controller.VerifyTLS)
	if err != nil {
		return nil, err
	}
	if err := cl.Login(ctx, cfg.Controller.Username, cfg.Controller.ResolvePassword()); err != nil {
		return nil, err
	}
	d.mu.Lock()
	d.pcl = cl
	d.mu.Unlock()
	return cl, nil
}

// 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
	}
	// Reuse the cached session across sync ticks instead of logging in every
	// interval; if the session has expired (or this is the first sync), log in
	// fresh and retry once.
	d.mu.Lock()
	cl := d.pcl
	d.mu.Unlock()
	var views []protect.LiveView
	var err error
	if cl != nil {
		views, _, err = cl.LiveViews(ctx)
	}
	if cl == nil || err != nil {
		if cl, err = d.loginProtect(ctx, cfg); err != nil {
			return err
		}
		if views, _, err = cl.LiveViews(ctx); 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
			}
			// Back off briefly so a persistent accept error (e.g. the socket
			// vanishing) never becomes a hot loop.
			if !sleepCtx(ctx, time.Second) {
				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 {
	// Copy under the lock, probe outside it: Healthy() dials mpv's IPC socket
	// (up to ~2s per stream), and holding the daemon mutex through that would
	// block layout switches and reloads for the duration.
	d.mu.Lock()
	layout := d.layout
	players := append([]*player.Player(nil), d.players...)
	d.mu.Unlock()
	resp := ipc.Response{OK: true, ActiveLayout: layout}
	for _, p := range 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
}