live playlist readiness coordinator

imgui.ini

@@ -29,8 +29,8 @@ Size=550,720
 Collapsed=0
 
 [Window][Settings & Info]
-Pos=1220,0
+Pos=580,0
-Size=700,1080
+Size=700,720
 Collapsed=0
 
 [Window][Seetings & Info]

internal/playback/playlist_readiness.go

@@ -1,5 +1,151 @@
 package playback
 
+import (
+	"context"
+	"errors"
+	"time"
+)
+
+type PlaylistSnapshotSource interface {
+	Snapshot() (PlaylistSnapshot, bool)
+}
+
+type SessionSnapshotSource interface {
+	Snapshot() (SessionSnapshot, bool)
+}
+
+type PlaybackStatusSnapshotSource interface {
+	SnapshotAll() PlaybackStatusSnapshot
+}
+
+type playlistReadinessTicker interface {
+	C() <-chan time.Time
+	Stop()
+}
+
+type playlistReadinessTickerFactory func(time.Duration) playlistReadinessTicker
+
+type realPlaylistReadinessTicker struct {
+	ticker *time.Ticker
+}
+
+func (t realPlaylistReadinessTicker) C() <-chan time.Time { return t.ticker.C }
+func (t realPlaylistReadinessTicker) Stop()               { t.ticker.Stop() }
+
+var (
+	ErrPlaylistSnapshotSourceRequired  = errors.New("playlist snapshot source is required")
+	ErrSessionSnapshotSourceRequired   = errors.New("session snapshot source is required")
+	ErrStatusSnapshotSourceRequired    = errors.New("playback status snapshot source is required")
+	ErrPlaylistReadinessOutputRequired = errors.New("playlist readiness output channel is required")
+	ErrPlaylistReadinessInterval       = errors.New("playlist readiness interval must be positive")
+)
+
+type PlaylistReadinessCoordinator struct {
+	playlist PlaylistSnapshotSource
+	session  SessionSnapshotSource
+	statuses PlaybackStatusSnapshotSource
+	output   chan<- PlaylistReadiness
+	interval time.Duration
+
+	newTicker playlistReadinessTickerFactory
+}
+
+func NewPlaylistReadinessCoordinator(
+	playlist PlaylistSnapshotSource,
+	session SessionSnapshotSource,
+	statuses PlaybackStatusSnapshotSource,
+	output chan<- PlaylistReadiness,
+	interval time.Duration,
+) (*PlaylistReadinessCoordinator, error) {
+	if playlist == nil {
+		return nil, ErrPlaylistSnapshotSourceRequired
+	}
+	if session == nil {
+		return nil, ErrSessionSnapshotSourceRequired
+	}
+	if statuses == nil {
+		return nil, ErrStatusSnapshotSourceRequired
+	}
+	if output == nil {
+		return nil, ErrPlaylistReadinessOutputRequired
+	}
+	if interval <= 0 {
+		return nil, ErrPlaylistReadinessInterval
+	}
+
+	return &PlaylistReadinessCoordinator{
+		playlist: playlist,
+		session:  session,
+		statuses: statuses,
+		output:   output,
+		interval: interval,
+		newTicker: func(interval time.Duration) playlistReadinessTicker {
+			return realPlaylistReadinessTicker{ticker: time.NewTicker(interval)}
+		},
+	}, nil
+}
+
+func (c *PlaylistReadinessCoordinator) Run(ctx context.Context) error {
+	ticker := c.newTicker(c.interval)
+	defer ticker.Stop()
+
+	var emittedRevision uint64
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+
+		case <-ticker.C():
+			playlistSnapshot, ok := c.playlist.Snapshot()
+			if !ok ||
+				!playlistSnapshot.State.HasSelection ||
+				playlistSnapshot.Revision == 0 ||
+				playlistSnapshot.Entry.Duration <= 0 ||
+				playlistSnapshot.Timing.Started ||
+				playlistSnapshot.Revision == emittedRevision {
+				continue
+			}
+
+			sessionSnapshot, ok := c.session.Snapshot()
+			if !ok || !PlaylistEntryMatchesSession(
+				playlistSnapshot.Entry,
+				sessionSnapshot.Desired,
+			) {
+				continue
+			}
+			if !IsSessionPlaying(sessionSnapshot, c.statuses.SnapshotAll()) {
+				continue
+			}
+
+			ready := PlaylistReadiness{Revision: playlistSnapshot.Revision}
+			select {
+			case <-ctx.Done():
+				return ctx.Err()
+			case c.output <- ready:
+				emittedRevision = playlistSnapshot.Revision
+			}
+		}
+	}
+}
+
+func PlaylistEntryMatchesSession(entry PlaylistEntry, session SessionConfig) bool {
+	if err := entry.Validate(); err != nil {
+		return false
+	}
+	return playlistFeedMatchesSession(entry.Video, session.Video) &&
+		playlistFeedMatchesSession(entry.Audio, session.Audio) &&
+		entry.SyncRequested == session.SyncRequested
+}
+
+func playlistFeedMatchesSession(playlist PlaylistFeed, session FeedConfig) bool {
+	if !playlist.IsConfigured() {
+		return !session.IsConfigured() && !session.Active
+	}
+	return session.Active &&
+		playlist.Domain == session.Domain &&
+		playlist.UUID == session.UUID
+}
+
 func IsSessionPlaying(
 	session SessionSnapshot,
 	statuses PlaybackStatusSnapshot,

internal/playback/playlist_readiness_coordinator_test.go

@@ -0,0 +1,368 @@
+package playback
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"testing"
+	"time"
+)
+
+type fakePlaylistSnapshotSource struct {
+	mu       sync.RWMutex
+	snapshot PlaylistSnapshot
+	ok       bool
+}
+
+func (s *fakePlaylistSnapshotSource) Snapshot() (PlaylistSnapshot, bool) {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	return s.snapshot, s.ok
+}
+
+func (s *fakePlaylistSnapshotSource) set(snapshot PlaylistSnapshot, ok bool) {
+	s.mu.Lock()
+	s.snapshot = snapshot
+	s.ok = ok
+	s.mu.Unlock()
+}
+
+type fakeSessionSnapshotSource struct {
+	mu       sync.RWMutex
+	snapshot SessionSnapshot
+	ok       bool
+}
+
+func (s *fakeSessionSnapshotSource) Snapshot() (SessionSnapshot, bool) {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	return s.snapshot, s.ok
+}
+
+func (s *fakeSessionSnapshotSource) set(snapshot SessionSnapshot, ok bool) {
+	s.mu.Lock()
+	s.snapshot = snapshot
+	s.ok = ok
+	s.mu.Unlock()
+}
+
+type fakePlaybackStatusSnapshotSource struct {
+	mu       sync.RWMutex
+	snapshot PlaybackStatusSnapshot
+}
+
+func (s *fakePlaybackStatusSnapshotSource) SnapshotAll() PlaybackStatusSnapshot {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	return s.snapshot
+}
+
+func (s *fakePlaybackStatusSnapshotSource) set(snapshot PlaybackStatusSnapshot) {
+	s.mu.Lock()
+	s.snapshot = snapshot
+	s.mu.Unlock()
+}
+
+type fakePlaylistReadinessTicker struct {
+	ch chan time.Time
+
+	mu      sync.Mutex
+	stopped bool
+}
+
+func newFakePlaylistReadinessTicker() *fakePlaylistReadinessTicker {
+	return &fakePlaylistReadinessTicker{ch: make(chan time.Time, 16)}
+}
+
+func (t *fakePlaylistReadinessTicker) C() <-chan time.Time { return t.ch }
+func (t *fakePlaylistReadinessTicker) Stop() {
+	t.mu.Lock()
+	t.stopped = true
+	t.mu.Unlock()
+}
+func (t *fakePlaylistReadinessTicker) tick() { t.ch <- time.Now() }
+func (t *fakePlaylistReadinessTicker) isStopped() bool {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	return t.stopped
+}
+
+func TestNewPlaylistReadinessCoordinatorValidatesDependencies(t *testing.T) {
+	playlist := &fakePlaylistSnapshotSource{}
+	session := &fakeSessionSnapshotSource{}
+	statuses := &fakePlaybackStatusSnapshotSource{}
+	output := make(chan PlaylistReadiness)
+	tests := []struct {
+		name     string
+		playlist PlaylistSnapshotSource
+		session  SessionSnapshotSource
+		statuses PlaybackStatusSnapshotSource
+		output   chan<- PlaylistReadiness
+		interval time.Duration
+		wantErr  error
+	}{
+		{name: "playlist", session: session, statuses: statuses, output: output, interval: time.Millisecond, wantErr: ErrPlaylistSnapshotSourceRequired},
+		{name: "session", playlist: playlist, statuses: statuses, output: output, interval: time.Millisecond, wantErr: ErrSessionSnapshotSourceRequired},
+		{name: "statuses", playlist: playlist, session: session, output: output, interval: time.Millisecond, wantErr: ErrStatusSnapshotSourceRequired},
+		{name: "output", playlist: playlist, session: session, statuses: statuses, interval: time.Millisecond, wantErr: ErrPlaylistReadinessOutputRequired},
+		{name: "interval", playlist: playlist, session: session, statuses: statuses, output: output, wantErr: ErrPlaylistReadinessInterval},
+		{name: "valid", playlist: playlist, session: session, statuses: statuses, output: output, interval: time.Millisecond},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			coordinator, err := NewPlaylistReadinessCoordinator(
+				test.playlist,
+				test.session,
+				test.statuses,
+				test.output,
+				test.interval,
+			)
+			if !errors.Is(err, test.wantErr) {
+				t.Fatalf("NewPlaylistReadinessCoordinator() error = %v, want %v", err, test.wantErr)
+			}
+			if test.wantErr != nil && coordinator != nil {
+				t.Fatalf("coordinator = %#v, want nil", coordinator)
+			}
+		})
+	}
+}
+
+func TestPlaylistEntryMatchesSession(t *testing.T) {
+	entry := PlaylistEntry{
+		Video:         PlaylistFeed{Domain: "video-domain", UUID: "video"},
+		Audio:         PlaylistFeed{Domain: "audio-domain", UUID: "audio"},
+		SyncRequested: true,
+	}
+	matching := entry.SessionConfig(validPlaylistRetryPolicy())
+	tests := []struct {
+		name    string
+		entry   PlaylistEntry
+		session SessionConfig
+		want    bool
+	}{
+		{name: "matching", entry: entry, session: matching, want: true},
+		{name: "retry ignored", entry: entry, session: func() SessionConfig { value := matching; value.Retry.MaxAttempts = 99; return value }(), want: true},
+		{name: "wrong video UUID", entry: entry, session: func() SessionConfig { value := matching; value.Video.UUID = "other"; return value }()},
+		{name: "wrong audio domain", entry: entry, session: func() SessionConfig { value := matching; value.Audio.Domain = "other"; return value }()},
+		{name: "inactive video", entry: entry, session: func() SessionConfig { value := matching; value.Video.Active = false; return value }()},
+		{name: "wrong sync request", entry: entry, session: func() SessionConfig { value := matching; value.SyncRequested = false; return value }()},
+		{
+			name:    "absent audio matches unconfigured inactive",
+			entry:   PlaylistEntry{Video: entry.Video},
+			session: PlaylistEntry{Video: entry.Video}.SessionConfig(validPlaylistRetryPolicy()),
+			want:    true,
+		},
+		{
+			name:  "absent audio rejects configured audio",
+			entry: PlaylistEntry{Video: entry.Video},
+			session: SessionConfig{
+				Video: matching.Video,
+				Audio: matching.Audio,
+				Retry: matching.Retry,
+			},
+		},
+		{name: "invalid entry", entry: PlaylistEntry{}, session: matching},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			if got := PlaylistEntryMatchesSession(test.entry, test.session); got != test.want {
+				t.Fatalf("PlaylistEntryMatchesSession() = %v, want %v", got, test.want)
+			}
+		})
+	}
+}
+
+func TestPlaylistReadinessCoordinatorEmitsOncePerRevision(t *testing.T) {
+	playlist, session, statuses := readyVideoSnapshots(1)
+	output := make(chan PlaylistReadiness, 4)
+	coordinator, ticker, cancel, result := startReadinessCoordinator(
+		t,
+		playlist,
+		session,
+		statuses,
+		output,
+	)
+	_ = coordinator
+	defer cancel()
+
+	ticker.tick()
+	if got := receivePlaylistReadiness(t, output); got.Revision != 1 {
+		t.Fatalf("readiness revision = %d, want 1", got.Revision)
+	}
+	ticker.tick()
+	assertNoPlaylistReadiness(t, output)
+
+	next := playlistSnapshotForVideo(2)
+	playlist.set(next, true)
+	ticker.tick()
+	if got := receivePlaylistReadiness(t, output); got.Revision != 2 {
+		t.Fatalf("readiness revision = %d, want 2", got.Revision)
+	}
+
+	cancel()
+	if err := waitForPlaylistResult(t, result); !errors.Is(err, context.Canceled) {
+		t.Fatalf("Run() error = %v, want %v", err, context.Canceled)
+	}
+	if !ticker.isStopped() {
+		t.Fatal("ticker was not stopped")
+	}
+}
+
+func TestPlaylistReadinessCoordinatorWaitsForAllConditions(t *testing.T) {
+	playlist, session, statuses := readyVideoSnapshots(1)
+	output := make(chan PlaylistReadiness, 1)
+	_, ticker, cancel, result := startReadinessCoordinator(t, playlist, session, statuses, output)
+	defer cancel()
+
+	tests := []struct {
+		name   string
+		mutate func()
+	}{
+		{name: "no playlist snapshot", mutate: func() { playlist.set(PlaylistSnapshot{}, false) }},
+		{name: "no selection", mutate: func() {
+			value := playlistSnapshotForVideo(1)
+			value.State.HasSelection = false
+			playlist.set(value, true)
+		}},
+		{name: "zero revision", mutate: func() { value := playlistSnapshotForVideo(1); value.Revision = 0; playlist.set(value, true) }},
+		{name: "zero duration", mutate: func() { value := playlistSnapshotForVideo(1); value.Entry.Duration = 0; playlist.set(value, true) }},
+		{name: "already started", mutate: func() { value := playlistSnapshotForVideo(1); value.Timing.Started = true; playlist.set(value, true) }},
+		{name: "session mismatch", mutate: func() {
+			playlist.set(playlistSnapshotForVideo(1), true)
+			value, _ := session.Snapshot()
+			value.Desired.Video.UUID = "other"
+			session.set(value, true)
+		}},
+		{name: "stale statuses", mutate: func() {
+			playlist.set(playlistSnapshotForVideo(1), true)
+			_, validSession, _ := readyVideoSnapshots(1)
+			value, _ := validSession.Snapshot()
+			session.set(value, true)
+			current := statuses.SnapshotAll()
+			current.Generation = 2
+			current.Video.Generation = 2
+			statuses.set(current)
+		}},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			validPlaylist, validSession, validStatuses := readyVideoSnapshots(1)
+			playlist.set(validPlaylist.snapshot, true)
+			session.set(validSession.snapshot, true)
+			statuses.set(validStatuses.snapshot)
+			test.mutate()
+			ticker.tick()
+			assertNoPlaylistReadiness(t, output)
+		})
+	}
+
+	cancel()
+	_ = waitForPlaylistResult(t, result)
+}
+
+func TestPlaylistReadinessCoordinatorCancellationWhileBlockedSending(t *testing.T) {
+	playlist, session, statuses := readyVideoSnapshots(1)
+	output := make(chan PlaylistReadiness)
+	_, ticker, cancel, result := startReadinessCoordinator(t, playlist, session, statuses, output)
+
+	ticker.tick()
+	time.Sleep(time.Millisecond)
+	cancel()
+	if err := waitForPlaylistResult(t, result); !errors.Is(err, context.Canceled) {
+		t.Fatalf("Run() error = %v, want %v", err, context.Canceled)
+	}
+}
+
+func readyVideoSnapshots(revision uint64) (
+	*fakePlaylistSnapshotSource,
+	*fakeSessionSnapshotSource,
+	*fakePlaybackStatusSnapshotSource,
+) {
+	playlist := &fakePlaylistSnapshotSource{snapshot: playlistSnapshotForVideo(revision), ok: true}
+	entry := playlist.snapshot.Entry
+	desired := entry.SessionConfig(validPlaylistRetryPolicy())
+	session := &fakeSessionSnapshotSource{
+		snapshot: SessionSnapshot{
+			Desired:    desired,
+			Plan:       SessionPlan{Topology: TopologyIndependent, Video: desired.Video},
+			Generation: 5,
+		},
+		ok: true,
+	}
+	statuses := &fakePlaybackStatusSnapshotSource{
+		snapshot: PlaybackStatusSnapshot{
+			Generation: 5,
+			Video: Status{
+				Unit:       UnitVideo,
+				State:      StatePlaying,
+				Generation: 5,
+				Feed:       desired.Video,
+			},
+			HasVideo: true,
+		},
+	}
+	return playlist, session, statuses
+}
+
+func playlistSnapshotForVideo(revision uint64) PlaylistSnapshot {
+	entry := PlaylistEntry{
+		Name:     "video",
+		Video:    PlaylistFeed{Domain: "domain", UUID: "video"},
+		Duration: 10 * time.Second,
+	}
+	return PlaylistSnapshot{
+		State:    PlaylistState{CurrentIndex: 0, HasSelection: true},
+		Entry:    entry,
+		Revision: revision,
+		Timing:   NewPlaylistTiming(revision, entry.Duration),
+	}
+}
+
+func startReadinessCoordinator(
+	t *testing.T,
+	playlist PlaylistSnapshotSource,
+	session SessionSnapshotSource,
+	statuses PlaybackStatusSnapshotSource,
+	output chan<- PlaylistReadiness,
+) (*PlaylistReadinessCoordinator, *fakePlaylistReadinessTicker, context.CancelFunc, <-chan error) {
+	t.Helper()
+	coordinator, err := NewPlaylistReadinessCoordinator(
+		playlist,
+		session,
+		statuses,
+		output,
+		time.Millisecond,
+	)
+	if err != nil {
+		t.Fatalf("NewPlaylistReadinessCoordinator() error = %v", err)
+	}
+	ticker := newFakePlaylistReadinessTicker()
+	coordinator.newTicker = func(time.Duration) playlistReadinessTicker { return ticker }
+	ctx, cancel := context.WithCancel(context.Background())
+	result := make(chan error, 1)
+	go func() { result <- coordinator.Run(ctx) }()
+	return coordinator, ticker, cancel, result
+}
+
+func receivePlaylistReadiness(t *testing.T, output <-chan PlaylistReadiness) PlaylistReadiness {
+	t.Helper()
+	select {
+	case readiness := <-output:
+		return readiness
+	case <-time.After(time.Second):
+		t.Fatal("timed out waiting for playlist readiness")
+		return PlaylistReadiness{}
+	}
+}
+
+func assertNoPlaylistReadiness(t *testing.T, output <-chan PlaylistReadiness) {
+	t.Helper()
+	select {
+	case readiness := <-output:
+		t.Fatalf("unexpected playlist readiness: %#v", readiness)
+	case <-time.After(20 * time.Millisecond):
+	}
+}