session controller

internal/playback/session_controller.go

@@ -0,0 +1,303 @@
+package playback
+
+import (
+	"context"
+	"errors"
+	"fmt"
+)
+
+type VideoSlotRunner interface {
+	Run(
+		context.Context,
+		FeedConfig,
+		<-chan FeedConfig,
+	) error
+}
+
+type AudioSlotRunner interface {
+	Run(
+		context.Context,
+		FeedConfig,
+		<-chan FeedConfig,
+	) error
+}
+
+type SyncSlotRunner interface {
+	Run(
+		context.Context,
+		SyncPairConfig,
+		<-chan SyncPairConfig,
+	) error
+}
+
+var _ VideoSlotRunner = (*VideoSlot)(nil)
+var _ AudioSlotRunner = (*AudioSlot)(nil)
+var _ SyncSlotRunner = (*SyncSlot)(nil)
+
+var (
+	ErrVideoSlotRequired = errors.New("video slot is required")
+	ErrAudioSlotRequired = errors.New("audio slot is required")
+	ErrSyncSlotRequired  = errors.New("sync slot is required")
+)
+
+type SessionController struct {
+	videoSlot VideoSlotRunner
+	audioSlot AudioSlotRunner
+	syncSlot  SyncSlotRunner
+	canSync   SyncPredicate
+}
+
+func NewSessionController(
+	videoSlot VideoSlotRunner,
+	audioSlot AudioSlotRunner,
+	syncSlot SyncSlotRunner,
+	canSync SyncPredicate,
+) (*SessionController, error) {
+	if videoSlot == nil {
+		return nil, ErrVideoSlotRequired
+	}
+	if audioSlot == nil {
+		return nil, ErrAudioSlotRequired
+	}
+	if syncSlot == nil {
+		return nil, ErrSyncSlotRequired
+	}
+	return &SessionController{
+		videoSlot: videoSlot,
+		audioSlot: audioSlot,
+		syncSlot:  syncSlot,
+		canSync:   canSync,
+	}, nil
+}
+
+type sessionRuntime struct {
+	topology SessionTopology
+	cancel   context.CancelFunc
+	done     chan struct{}
+	result   error
+
+	videoCommands chan FeedConfig
+	audioCommands chan FeedConfig
+	syncCommands  chan SyncPairConfig
+}
+
+var (
+	ErrSessionRuntimeStopped = errors.New(
+		"session slot runtime stopped unexpectedly",
+	)
+)
+
+func (c *SessionController) Run(
+	ctx context.Context,
+	initial SessionConfig,
+	commands <-chan SessionCommand,
+) error {
+	plan, err := BuildSessionPlan(initial, c.canSync)
+	if err != nil {
+		return fmt.Errorf("build initial session plan: %w", err)
+	}
+
+	desired := initial
+	runtime := c.startSessionRuntime(ctx, plan)
+
+	for {
+		var runtimeDone <-chan struct{}
+		if runtime != nil {
+			runtimeDone = runtime.done
+		}
+
+		select {
+		case <-ctx.Done():
+			stopSessionRuntime(runtime)
+			return ctx.Err()
+
+		case <-runtimeDone:
+			return unexpectedSessionRuntimeError(runtime.result)
+
+		case command, ok := <-commands:
+			if !ok {
+				stopSessionRuntime(runtime)
+				return nil
+			}
+
+			nextDesired, err := ApplySessionCommand(desired, command)
+			if err != nil {
+				// Invalid commands must not disturb the current valid runtime.
+				continue
+			}
+			nextPlan, err := BuildSessionPlan(nextDesired, c.canSync)
+			if err != nil {
+				continue
+			}
+
+			nextRuntime, err := c.reconcileSessionRuntime(
+				ctx,
+				runtime,
+				plan,
+				nextPlan,
+			)
+			if err != nil {
+				stopSessionRuntime(runtime)
+				if ctx.Err() != nil {
+					return ctx.Err()
+				}
+				return err
+			}
+
+			desired = nextDesired
+			plan = nextPlan
+			runtime = nextRuntime
+		}
+	}
+}
+
+func (c *SessionController) startSessionRuntime(
+	ctx context.Context,
+	plan SessionPlan,
+) *sessionRuntime {
+	if plan.Topology == TopologyIdle {
+		return &sessionRuntime{topology: TopologyIdle}
+	}
+
+	runtimeCtx, cancel := context.WithCancel(ctx)
+	runtime := &sessionRuntime{
+		topology: plan.Topology,
+		cancel:   cancel,
+		done:     make(chan struct{}),
+	}
+
+	switch plan.Topology {
+	case TopologyIndependent:
+		runtime.videoCommands = make(chan FeedConfig)
+		runtime.audioCommands = make(chan FeedConfig)
+		results := make(chan error, 2)
+
+		go func() {
+			results <- c.videoSlot.Run(
+				runtimeCtx,
+				plan.Video,
+				runtime.videoCommands,
+			)
+		}()
+		go func() {
+			results <- c.audioSlot.Run(
+				runtimeCtx,
+				plan.Audio,
+				runtime.audioCommands,
+			)
+		}()
+		go func() {
+			first := <-results
+			cancel()
+			second := <-results
+			runtime.result = errors.Join(first, second)
+			close(runtime.done)
+		}()
+
+	case TopologySynchronized:
+		runtime.syncCommands = make(chan SyncPairConfig)
+		go func() {
+			runtime.result = c.syncSlot.Run(
+				runtimeCtx,
+				plan.Sync,
+				runtime.syncCommands,
+			)
+			close(runtime.done)
+		}()
+	}
+
+	return runtime
+}
+
+func stopSessionRuntime(runtime *sessionRuntime) {
+	if runtime == nil || runtime.done == nil {
+		return
+	}
+	runtime.cancel()
+	<-runtime.done
+}
+
+func unexpectedSessionRuntimeError(err error) error {
+	if err == nil {
+		return ErrSessionRuntimeStopped
+	}
+	return fmt.Errorf("%w: %v", ErrSessionRuntimeStopped, err)
+}
+
+func (c *SessionController) reconcileSessionRuntime(
+	ctx context.Context,
+	runtime *sessionRuntime,
+	current SessionPlan,
+	next SessionPlan,
+) (*sessionRuntime, error) {
+	if current.Topology != next.Topology {
+		stopSessionRuntime(runtime)
+		if err := ctx.Err(); err != nil {
+			return runtime, err
+		}
+		return c.startSessionRuntime(ctx, next), nil
+	}
+
+	switch next.Topology {
+	case TopologyIndependent:
+		if current.Video != next.Video {
+			if !sendFeedConfig(ctx, runtime.done, runtime.videoCommands, next.Video) {
+				return runtime, sessionRuntimeSendError(ctx, runtime)
+			}
+		}
+		if current.Audio != next.Audio {
+			if !sendFeedConfig(ctx, runtime.done, runtime.audioCommands, next.Audio) {
+				return runtime, sessionRuntimeSendError(ctx, runtime)
+			}
+		}
+
+	case TopologySynchronized:
+		if current.Sync != next.Sync {
+			if !sendSyncPairConfig(ctx, runtime.done, runtime.syncCommands, next.Sync) {
+				return runtime, sessionRuntimeSendError(ctx, runtime)
+			}
+		}
+	}
+
+	return runtime, nil
+}
+
+func sendFeedConfig(
+	ctx context.Context,
+	done <-chan struct{},
+	commands chan<- FeedConfig,
+	config FeedConfig,
+) bool {
+	select {
+	case commands <- config:
+		return true
+	case <-ctx.Done():
+		return false
+	case <-done:
+		return false
+	}
+}
+
+func sendSyncPairConfig(
+	ctx context.Context,
+	done <-chan struct{},
+	commands chan<- SyncPairConfig,
+	config SyncPairConfig,
+) bool {
+	select {
+	case commands <- config:
+		return true
+	case <-ctx.Done():
+		return false
+	case <-done:
+		return false
+	}
+}
+
+func sessionRuntimeSendError(ctx context.Context, runtime *sessionRuntime) error {
+	if err := ctx.Err(); err != nil {
+		return err
+	}
+	<-runtime.done
+	return unexpectedSessionRuntimeError(runtime.result)
+}

internal/playback/session_controller_test.go

@@ -0,0 +1,279 @@
+package playback
+
+import (
+	"context"
+	"errors"
+	"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
+	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", 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", 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", 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 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 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)
+			}
+			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")
+	}
+}