go-mxl-player/internal/playback/session_controller_test.go

package playback

import (
	"context"
	"errors"
	"sync"
	"testing"
	"time"
)

type stubVideoSlot struct{}

func (stubVideoSlot) Run(context.Context, FeedConfig, <-chan FeedConfig) error {
	return nil
}

type stubAudioSlot struct{}

func (stubAudioSlot) Run(context.Context, FeedConfig, <-chan FeedConfig) error {
	return nil
}

type stubSyncSlot struct{}

func (stubSyncSlot) Run(context.Context, SyncPairConfig, <-chan SyncPairConfig) error {
	return nil
}

func TestNewSessionControllerValidatesSlots(t *testing.T) {
	video := stubVideoSlot{}
	audio := stubAudioSlot{}
	sync := stubSyncSlot{}
	tests := []struct {
		name  string
		video VideoSlotRunner
		audio AudioSlotRunner
		sync  SyncSlotRunner
		want  error
	}{
		{"missing video", nil, audio, sync, ErrVideoSlotRequired},
		{"missing audio", video, nil, sync, ErrAudioSlotRequired},
		{"missing sync", video, audio, nil, ErrSyncSlotRequired},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			controller, err := NewSessionController(tt.video, tt.audio, tt.sync, nil)
			if controller != nil {
				t.Fatalf("NewSessionController() controller = %#v, want nil", controller)
			}
			if !errors.Is(err, tt.want) {
				t.Fatalf("NewSessionController() error = %v, want %v", err, tt.want)
			}
		})
	}
}

func TestNewSessionControllerAllowsUnavailableSynchronization(t *testing.T) {
	video := stubVideoSlot{}
	audio := stubAudioSlot{}
	sync := stubSyncSlot{}
	controller, err := NewSessionController(video, audio, sync, nil)
	if err != nil {
		t.Fatalf("NewSessionController() error = %v", err)
	}
	if controller == nil {
		t.Fatal("NewSessionController() controller is nil")
	}
	if controller.videoSlot != video || controller.audioSlot != audio || controller.syncSlot != sync {
		t.Fatalf("NewSessionController() = %#v", controller)
	}
	if controller.canSync != nil {
		t.Fatal("nil sync predicate was not preserved")
	}
}

func TestNewSessionControllerStoresSyncPredicate(t *testing.T) {
	predicate := func(FeedConfig, FeedConfig) bool { return true }
	controller, err := NewSessionController(
		stubVideoSlot{}, stubAudioSlot{}, stubSyncSlot{}, predicate,
	)
	if err != nil {
		t.Fatalf("NewSessionController() error = %v", err)
	}
	if controller.canSync == nil || !controller.canSync(FeedConfig{}, FeedConfig{}) {
		t.Fatal("sync predicate was not stored")
	}
}

type controllerEvent struct {
	unit       Unit
	action     string
	generation uint64
	feed       FeedConfig
	pair       SyncPairConfig
}

type recordingVideoSlot struct{ events chan<- controllerEvent }

func (s recordingVideoSlot) Run(
	ctx context.Context,
	initial FeedConfig,
	commands <-chan FeedConfig,
) error {
	s.events <- controllerEvent{
		unit: UnitVideo, action: "start",
		generation: generationFromContext(ctx), feed: initial,
	}
	for {
		select {
		case config := <-commands:
			s.events <- controllerEvent{unit: UnitVideo, action: "command", feed: config}
		case <-ctx.Done():
			s.events <- controllerEvent{unit: UnitVideo, action: "stop"}
			return ctx.Err()
		}
	}
}

type recordingAudioSlot struct{ events chan<- controllerEvent }

func (s recordingAudioSlot) Run(
	ctx context.Context,
	initial FeedConfig,
	commands <-chan FeedConfig,
) error {
	s.events <- controllerEvent{
		unit: UnitAudio, action: "start",
		generation: generationFromContext(ctx), feed: initial,
	}
	for {
		select {
		case config := <-commands:
			s.events <- controllerEvent{unit: UnitAudio, action: "command", feed: config}
		case <-ctx.Done():
			s.events <- controllerEvent{unit: UnitAudio, action: "stop"}
			return ctx.Err()
		}
	}
}

type recordingSyncSlot struct{ events chan<- controllerEvent }

func (s recordingSyncSlot) Run(
	ctx context.Context,
	initial SyncPairConfig,
	commands <-chan SyncPairConfig,
) error {
	s.events <- controllerEvent{
		unit: UnitSync, action: "start",
		generation: generationFromContext(ctx), pair: initial,
	}
	for {
		select {
		case config := <-commands:
			s.events <- controllerEvent{unit: UnitSync, action: "command", pair: config}
		case <-ctx.Done():
			s.events <- controllerEvent{unit: UnitSync, action: "stop"}
			return ctx.Err()
		}
	}
}

func newRecordingController(t *testing.T, events chan<- controllerEvent) *SessionController {
	t.Helper()
	controller, err := NewSessionController(
		recordingVideoSlot{events},
		recordingAudioSlot{events},
		recordingSyncSlot{events},
		func(FeedConfig, FeedConfig) bool { return true },
	)
	if err != nil {
		t.Fatal(err)
	}
	return controller
}

func receiveControllerEvent(t *testing.T, events <-chan controllerEvent) controllerEvent {
	t.Helper()
	select {
	case event := <-events:
		return event
	case <-time.After(time.Second):
		t.Fatal("controller event timed out")
		return controllerEvent{}
	}
}

func receiveIndependentStarts(t *testing.T, events <-chan controllerEvent) {
	t.Helper()
	seen := map[Unit]bool{}
	for len(seen) < 2 {
		event := receiveControllerEvent(t, events)
		if event.action != "start" || (event.unit != UnitVideo && event.unit != UnitAudio) {
			t.Fatalf("unexpected initial event: %+v", event)
		}
		seen[event.unit] = true
	}
}

func waitControllerSnapshot(
	t *testing.T,
	controller *SessionController,
	match func(SessionSnapshot) bool,
) SessionSnapshot {
	t.Helper()
	deadline := time.Now().Add(time.Second)
	for {
		if snapshot, ok := controller.Snapshot(); ok && match(snapshot) {
			return snapshot
		}
		if time.Now().After(deadline) {
			snapshot, ok := controller.Snapshot()
			t.Fatalf("snapshot timed out: %#v, available=%t", snapshot, ok)
		}
		time.Sleep(time.Millisecond)
	}
}

func TestSessionControllerUpdatesOnlyChangedIndependentSlot(t *testing.T) {
	events := make(chan controllerEvent, 32)
	controller := newRecordingController(t, events)
	initial := validCommandSession()
	initial.SyncRequested = false
	commands := make(chan SessionCommand)
	done := make(chan error, 1)
	go func() { done <- controller.Run(context.Background(), initial, commands) }()
	receiveIndependentStarts(t, events)

	want := FeedConfig{Domain: "/new-video", UUID: "new-video", Active: true}
	commands <- SessionCommand{Kind: CommandSetVideo, Config: want}
	event := receiveControllerEvent(t, events)
	if event.unit != UnitVideo || event.action != "command" || event.feed != want {
		t.Fatalf("replacement event = %+v", event)
	}
	select {
	case event := <-events:
		t.Fatalf("unchanged audio slot was disturbed: %+v", event)
	case <-time.After(20 * time.Millisecond):
	}

	close(commands)
	select {
	case err := <-done:
		if err != nil {
			t.Fatalf("Run() error = %v, want nil", err)
		}
	case <-time.After(time.Second):
		t.Fatal("Run() did not stop")
	}
}

func TestSessionControllerReplacesSynchronizedPairWithOneCommand(t *testing.T) {
	events := make(chan controllerEvent, 32)
	controller := newRecordingController(t, events)
	initial := validCommandSession()
	commands := make(chan SessionCommand)
	done := make(chan error, 1)
	go func() { done <- controller.Run(context.Background(), initial, commands) }()
	start := receiveControllerEvent(t, events)
	if start.unit != UnitSync || start.action != "start" {
		t.Fatalf("initial event = %+v, want sync start", start)
	}

	next := initial
	next.Video = FeedConfig{Domain: "/next", UUID: "next-video", Active: true}
	next.Audio = FeedConfig{Domain: "/next", UUID: "next-audio", Active: true}
	commands <- SessionCommand{Kind: CommandSetSession, Session: next}
	event := receiveControllerEvent(t, events)
	wantPair := SyncPairConfig{Video: next.Video, Audio: next.Audio}
	if event.unit != UnitSync || event.action != "command" || event.pair != wantPair {
		t.Fatalf("replacement event = %+v, want one sync command for %#v", event, wantPair)
	}
	select {
	case event := <-events:
		t.Fatalf("atomic replacement emitted an extra event: %+v", event)
	case <-time.After(20 * time.Millisecond):
	}

	close(commands)
	if err := <-done; err != nil {
		t.Fatalf("Run() error = %v", err)
	}
}

func TestSessionControllerStopsIndependentSlotsBeforeStartingSync(t *testing.T) {
	events := make(chan controllerEvent, 32)
	controller := newRecordingController(t, events)
	initial := validCommandSession()
	initial.SyncRequested = false
	commands := make(chan SessionCommand)
	done := make(chan error, 1)
	go func() { done <- controller.Run(context.Background(), initial, commands) }()
	receiveIndependentStarts(t, events)

	commands <- SessionCommand{Kind: CommandEnableSync}
	stopped := map[Unit]bool{}
	for {
		event := receiveControllerEvent(t, events)
		if event.unit == UnitSync && event.action == "start" {
			if !stopped[UnitVideo] || !stopped[UnitAudio] {
				t.Fatalf("sync started before both independent slots stopped: %v", stopped)
			}
			if event.generation != 2 {
				t.Fatalf("sync runtime generation = %d, want 2", event.generation)
			}
			break
		}
		if event.action != "stop" || (event.unit != UnitVideo && event.unit != UnitAudio) {
			t.Fatalf("unexpected transition event: %+v", event)
		}
		stopped[event.unit] = true
	}

	close(commands)
	select {
	case err := <-done:
		if err != nil {
			t.Fatalf("Run() error = %v, want nil", err)
		}
	case <-time.After(time.Second):
		t.Fatal("Run() did not stop")
	}
}

func TestSessionControllerCancellationStopsAndJoinsRuntime(t *testing.T) {
	events := make(chan controllerEvent, 32)
	controller := newRecordingController(t, events)
	initial := validCommandSession()
	initial.SyncRequested = false
	ctx, cancel := context.WithCancel(context.Background())
	done := make(chan error, 1)
	go func() { done <- controller.Run(ctx, initial, make(chan SessionCommand)) }()
	receiveIndependentStarts(t, events)
	cancel()

	select {
	case err := <-done:
		if !errors.Is(err, context.Canceled) {
			t.Fatalf("Run() error = %v, want context.Canceled", err)
		}
	case <-time.After(time.Second):
		t.Fatal("Run() did not stop after cancellation")
	}
}

func TestSessionControllerSnapshotTracksDesiredPlanAndGeneration(t *testing.T) {
	events := make(chan controllerEvent, 64)
	controller := newRecordingController(t, events)
	if snapshot, ok := controller.Snapshot(); ok {
		t.Fatalf("Snapshot() before Run = %#v, true; want unavailable", snapshot)
	}

	initial := validCommandSession()
	initial.SyncRequested = false
	commands := make(chan SessionCommand)
	done := make(chan error, 1)
	go func() { done <- controller.Run(context.Background(), initial, commands) }()
	receiveIndependentStarts(t, events)

	snapshot := waitControllerSnapshot(t, controller, func(snapshot SessionSnapshot) bool {
		return snapshot.Desired == initial
	})
	if snapshot.Generation != 1 || snapshot.Plan.Topology != TopologyIndependent {
		t.Fatalf("initial snapshot = %#v", snapshot)
	}

	newVideo := FeedConfig{Domain: "/new-video", UUID: "new-video", Active: true}
	commands <- SessionCommand{Kind: CommandSetVideo, Config: newVideo}
	receiveControllerEvent(t, events)
	snapshot = waitControllerSnapshot(t, controller, func(snapshot SessionSnapshot) bool {
		return snapshot.Desired.Video == newVideo
	})
	if snapshot.Generation != 1 || snapshot.Plan.Topology != TopologyIndependent {
		t.Fatalf("same-topology snapshot = %#v", snapshot)
	}

	commands <- SessionCommand{Kind: CommandEnableSync}
	for {
		if event := receiveControllerEvent(t, events); event.unit == UnitSync && event.action == "start" {
			break
		}
	}
	snapshot = waitControllerSnapshot(t, controller, func(snapshot SessionSnapshot) bool {
		return snapshot.Plan.Topology == TopologySynchronized
	})
	if snapshot.Generation != 2 || !snapshot.Desired.SyncRequested {
		t.Fatalf("sync snapshot = %#v", snapshot)
	}

	commands <- SessionCommand{Kind: CommandDisableSync}
	for {
		event := receiveControllerEvent(t, events)
		if event.action == "start" && (event.unit == UnitVideo || event.unit == UnitAudio) {
			break
		}
	}
	snapshot = waitControllerSnapshot(t, controller, func(snapshot SessionSnapshot) bool {
		return snapshot.Plan.Topology == TopologyIndependent && !snapshot.Desired.SyncRequested
	})
	if snapshot.Generation != 3 {
		t.Fatalf("independent snapshot generation = %d, want 3", snapshot.Generation)
	}

	close(commands)
	if err := <-done; err != nil {
		t.Fatalf("Run() error = %v", err)
	}
}

func TestSessionControllerSnapshotRetainsUnavailableSyncRequest(t *testing.T) {
	events := make(chan controllerEvent, 32)
	controller, err := NewSessionController(
		recordingVideoSlot{events},
		recordingAudioSlot{events},
		recordingSyncSlot{events},
		nil,
	)
	if err != nil {
		t.Fatal(err)
	}
	initial := validCommandSession()
	initial.SyncRequested = false
	commands := make(chan SessionCommand)
	done := make(chan error, 1)
	go func() { done <- controller.Run(context.Background(), initial, commands) }()
	receiveIndependentStarts(t, events)
	commands <- SessionCommand{Kind: CommandEnableSync}

	snapshot := waitControllerSnapshot(t, controller, func(snapshot SessionSnapshot) bool {
		return snapshot.Desired.SyncRequested
	})
	if snapshot.Plan.Topology != TopologyIndependent || snapshot.Generation != 1 {
		t.Fatalf("unsupported-sync snapshot = %#v", snapshot)
	}
	close(commands)
	if err := <-done; err != nil {
		t.Fatal(err)
	}
}

func TestSessionControllerSnapshotConcurrentReads(t *testing.T) {
	events := make(chan controllerEvent, 32)
	controller := newRecordingController(t, events)
	initial := validCommandSession()
	initial.SyncRequested = false
	commands := make(chan SessionCommand)
	done := make(chan error, 1)
	go func() { done <- controller.Run(context.Background(), initial, commands) }()
	receiveIndependentStarts(t, events)
	waitControllerSnapshot(t, controller, func(SessionSnapshot) bool { return true })

	var readers sync.WaitGroup
	for range 8 {
		readers.Add(1)
		go func() {
			defer readers.Done()
			for range 1_000 {
				controller.Snapshot()
			}
		}()
	}
	readers.Wait()
	close(commands)
	if err := <-done; err != nil {
		t.Fatal(err)
	}
}