2026-09-01 23:52:36 +03:00
2 changed files with 418 additions and 0 deletions
@@ -0,0 +1,102 @@
+package playback
+
+import (
+	"context"
+	"errors"
+	"sync"
+)
+
+type PlaylistController struct {
+	playlist Playlist
+	retry    RetryPolicy
+	sessions chan<- SessionCommand
+
+	mu          sync.RWMutex
+	snapshot    PlaylistSnapshot
+	hasSnapshot bool
+}
+
+type PlaylistSnapshot struct {
+	State PlaylistState
+	Entry PlaylistEntry
+}
+
+var (
+	ErrNilSessionCommandChannel = errors.New("session-command channel is nil")
+)
+
+func NewPlaylistController(
+	playlist Playlist,
+	retry RetryPolicy,
+	sessions chan<- SessionCommand,
+) (*PlaylistController, error) {
+	if err := playlist.Validate(); err != nil {
+		return nil, err
+	}
+	if err := retry.Validate(); err != nil {
+		return nil, err
+	}
+	if sessions == nil {
+		return nil, ErrNilSessionCommandChannel
+	}
+	return &PlaylistController{
+		playlist: playlist,
+		retry:    retry,
+		sessions: sessions,
+	}, nil
+}
+
+func (c *PlaylistController) Run(
+	ctx context.Context,
+	commands <-chan PlaylistCommand,
+) error {
+	state := PlaylistState{}
+	c.publish(state)
+
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+
+		case command, ok := <-commands:
+			if !ok {
+				return nil
+			}
+
+			next, sessionCommand, apply, err := ApplyPlaylistSelection(
+				c.playlist,
+				state,
+				command,
+				c.retry,
+			)
+			if err != nil {
+				continue
+			}
+			if apply {
+				select {
+				case <-ctx.Done():
+					return ctx.Err()
+				case c.sessions <- sessionCommand:
+				}
+			}
+
+			state = next
+			c.publish(state)
+		}
+	}
+}
+
+func (c *PlaylistController) Snapshot() (PlaylistSnapshot, bool) {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+	return c.snapshot, c.hasSnapshot
+}
+
+func (c *PlaylistController) publish(state PlaylistState) {
+	entry, _ := state.Entry(c.playlist)
+
+	c.mu.Lock()
+	c.snapshot = PlaylistSnapshot{State: state, Entry: entry}
+	c.hasSnapshot = true
+	c.mu.Unlock()
+}
@@ -0,0 +1,316 @@
+package playback
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"testing"
+	"time"
+)
+
+func TestNewPlaylistControllerValidatesConfiguration(t *testing.T) {
+	validPlaylist := navigationPlaylist(false)
+	validRetry := validPlaylistRetryPolicy()
+	validSessions := make(chan SessionCommand)
+	tests := []struct {
+		name     string
+		playlist Playlist
+		retry    RetryPolicy
+		sessions chan<- SessionCommand
+		wantErr  error
+	}{
+		{
+			name: "invalid playlist",
+			playlist: Playlist{Entries: []PlaylistEntry{
+				{},
+			}},
+			retry:    validRetry,
+			sessions: validSessions,
+			wantErr:  ErrPlaylistEntryEmpty,
+		},
+		{
+			name:     "invalid retry",
+			playlist: validPlaylist,
+			retry:    RetryPolicy{},
+			sessions: validSessions,
+			wantErr:  ErrInvalidRetryDelay,
+		},
+		{
+			name:     "nil session commands",
+			playlist: validPlaylist,
+			retry:    validRetry,
+			wantErr:  ErrNilSessionCommandChannel,
+		},
+		{
+			name:     "valid",
+			playlist: validPlaylist,
+			retry:    validRetry,
+			sessions: validSessions,
+		},
+		{
+			name:     "empty playlist is valid",
+			playlist: Playlist{},
+			retry:    validRetry,
+			sessions: validSessions,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			controller, err := NewPlaylistController(test.playlist, test.retry, test.sessions)
+			if !errors.Is(err, test.wantErr) {
+				t.Fatalf("NewPlaylistController() error = %v, want %v", err, test.wantErr)
+			}
+			if test.wantErr != nil && controller != nil {
+				t.Fatalf("NewPlaylistController() controller = %#v, want nil", controller)
+			}
+			if test.wantErr == nil && controller == nil {
+				t.Fatal("NewPlaylistController() controller is nil")
+			}
+		})
+	}
+}
+
+func TestPlaylistControllerPublishesInitialSnapshot(t *testing.T) {
+	controller, commands, _, cancel, result := startPlaylistController(t, navigationPlaylist(false), 0)
+	defer cancel()
+
+	snapshot := waitForPlaylistSnapshot(t, controller, func(snapshot PlaylistSnapshot) bool {
+		return !snapshot.State.HasSelection
+	})
+	if snapshot.Entry != (PlaylistEntry{}) {
+		t.Fatalf("initial entry = %#v, want zero value", snapshot.Entry)
+	}
+
+	close(commands)
+	if err := waitForPlaylistResult(t, result); err != nil {
+		t.Fatalf("Run() error = %v", err)
+	}
+}
+
+func TestPlaylistControllerForwardsManualCommands(t *testing.T) {
+	controller, commands, sessions, cancel, result := startPlaylistController(t, navigationPlaylist(true), 4)
+	defer cancel()
+
+	tests := []struct {
+		command   PlaylistCommand
+		wantIndex int
+		wantUUID  string
+	}{
+		{command: PlaylistCommand{Kind: PlaylistSelect, Index: 1}, wantIndex: 1, wantUUID: "audio-2"},
+		{command: PlaylistCommand{Kind: PlaylistNext}, wantIndex: 2, wantUUID: "video-3"},
+		{command: PlaylistCommand{Kind: PlaylistPrevious}, wantIndex: 1, wantUUID: "audio-2"},
+		{command: PlaylistCommand{Kind: PlaylistSelect, Index: 1}, wantIndex: 1, wantUUID: "audio-2"},
+	}
+
+	for _, test := range tests {
+		commands <- test.command
+		session := receivePlaylistSession(t, sessions)
+		if session.Kind != CommandSetSession {
+			t.Fatalf("session kind = %v, want %v", session.Kind, CommandSetSession)
+		}
+		gotUUID := session.Session.Video.UUID
+		if gotUUID == "" {
+			gotUUID = session.Session.Audio.UUID
+		}
+		if gotUUID != test.wantUUID {
+			t.Fatalf("session UUID = %q, want %q", gotUUID, test.wantUUID)
+		}
+		waitForPlaylistSnapshot(t, controller, func(snapshot PlaylistSnapshot) bool {
+			return snapshot.State.HasSelection && snapshot.State.CurrentIndex == test.wantIndex
+		})
+	}
+
+	close(commands)
+	if err := waitForPlaylistResult(t, result); err != nil {
+		t.Fatalf("Run() error = %v", err)
+	}
+}
+
+func TestPlaylistControllerNonLoopingBoundarySendsNothing(t *testing.T) {
+	_, commands, sessions, cancel, result := startPlaylistController(t, navigationPlaylist(false), 2)
+	defer cancel()
+
+	commands <- PlaylistCommand{Kind: PlaylistSelect, Index: 2}
+	_ = receivePlaylistSession(t, sessions)
+	commands <- PlaylistCommand{Kind: PlaylistNext}
+
+	select {
+	case command := <-sessions:
+		t.Fatalf("unexpected session command at boundary: %#v", command)
+	case <-time.After(20 * time.Millisecond):
+	}
+
+	close(commands)
+	if err := waitForPlaylistResult(t, result); err != nil {
+		t.Fatalf("Run() error = %v", err)
+	}
+}
+
+func TestPlaylistControllerIgnoresInvalidCommand(t *testing.T) {
+	_, commands, sessions, cancel, result := startPlaylistController(t, navigationPlaylist(false), 2)
+	defer cancel()
+
+	commands <- PlaylistCommand{}
+	commands <- PlaylistCommand{Kind: PlaylistNext}
+	if session := receivePlaylistSession(t, sessions); session.Session.Video.UUID != "video-1" {
+		t.Fatalf("session after invalid command = %#v", session)
+	}
+
+	close(commands)
+	if err := waitForPlaylistResult(t, result); err != nil {
+		t.Fatalf("Run() error = %v", err)
+	}
+}
+
+func TestPlaylistControllerCommitsStateAfterSessionDelivery(t *testing.T) {
+	sessions := make(chan SessionCommand)
+	controller, err := NewPlaylistController(navigationPlaylist(false), validPlaylistRetryPolicy(), sessions)
+	if err != nil {
+		t.Fatalf("NewPlaylistController() error = %v", err)
+	}
+	commands := make(chan PlaylistCommand, 1)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	result := make(chan error, 1)
+	go func() { result <- controller.Run(ctx, commands) }()
+
+	waitForPlaylistSnapshot(t, controller, func(snapshot PlaylistSnapshot) bool {
+		return !snapshot.State.HasSelection
+	})
+	commands <- PlaylistCommand{Kind: PlaylistNext}
+	time.Sleep(time.Millisecond)
+	snapshot, ok := controller.Snapshot()
+	if !ok || snapshot.State.HasSelection {
+		t.Fatalf("snapshot before delivery = %#v, %v; want no selection", snapshot, ok)
+	}
+
+	_ = receivePlaylistSession(t, sessions)
+	waitForPlaylistSnapshot(t, controller, func(snapshot PlaylistSnapshot) bool {
+		return snapshot.State.HasSelection && snapshot.State.CurrentIndex == 0
+	})
+
+	close(commands)
+	if err := waitForPlaylistResult(t, result); err != nil {
+		t.Fatalf("Run() error = %v", err)
+	}
+}
+
+func TestPlaylistControllerClosedCommandsReturnsNil(t *testing.T) {
+	_, commands, _, cancel, result := startPlaylistController(t, navigationPlaylist(false), 0)
+	defer cancel()
+	close(commands)
+
+	if err := waitForPlaylistResult(t, result); err != nil {
+		t.Fatalf("Run() error = %v, want nil", err)
+	}
+}
+
+func TestPlaylistControllerCancellationWhileReceiving(t *testing.T) {
+	_, _, _, cancel, result := startPlaylistController(t, navigationPlaylist(false), 0)
+	cancel()
+
+	if err := waitForPlaylistResult(t, result); !errors.Is(err, context.Canceled) {
+		t.Fatalf("Run() error = %v, want %v", err, context.Canceled)
+	}
+}
+
+func TestPlaylistControllerCancellationWhileSending(t *testing.T) {
+	sessions := make(chan SessionCommand)
+	controller, err := NewPlaylistController(navigationPlaylist(false), validPlaylistRetryPolicy(), sessions)
+	if err != nil {
+		t.Fatalf("NewPlaylistController() error = %v", err)
+	}
+	commands := make(chan PlaylistCommand, 1)
+	ctx, cancel := context.WithCancel(context.Background())
+	result := make(chan error, 1)
+	go func() { result <- controller.Run(ctx, commands) }()
+
+	commands <- PlaylistCommand{Kind: PlaylistNext}
+	cancel()
+	if err := waitForPlaylistResult(t, result); !errors.Is(err, context.Canceled) {
+		t.Fatalf("Run() error = %v, want %v", err, context.Canceled)
+	}
+}
+
+func TestPlaylistControllerSnapshotConcurrentReads(t *testing.T) {
+	controller, commands, _, cancel, result := startPlaylistController(t, navigationPlaylist(true), 64)
+	defer cancel()
+
+	var readers sync.WaitGroup
+	for range 8 {
+		readers.Add(1)
+		go func() {
+			defer readers.Done()
+			for range 100 {
+				_, _ = controller.Snapshot()
+			}
+		}()
+	}
+	for range 32 {
+		commands <- PlaylistCommand{Kind: PlaylistNext}
+	}
+	readers.Wait()
+	close(commands)
+	if err := waitForPlaylistResult(t, result); err != nil {
+		t.Fatalf("Run() error = %v", err)
+	}
+}
+
+func startPlaylistController(
+	t *testing.T,
+	playlist Playlist,
+	sessionBuffer int,
+) (*PlaylistController, chan PlaylistCommand, chan SessionCommand, context.CancelFunc, <-chan error) {
+	t.Helper()
+	sessions := make(chan SessionCommand, sessionBuffer)
+	controller, err := NewPlaylistController(playlist, validPlaylistRetryPolicy(), sessions)
+	if err != nil {
+		t.Fatalf("NewPlaylistController() error = %v", err)
+	}
+	commands := make(chan PlaylistCommand, 64)
+	ctx, cancel := context.WithCancel(context.Background())
+	result := make(chan error, 1)
+	go func() { result <- controller.Run(ctx, commands) }()
+	return controller, commands, sessions, cancel, result
+}
+
+func waitForPlaylistSnapshot(
+	t *testing.T,
+	controller *PlaylistController,
+	predicate func(PlaylistSnapshot) bool,
+) PlaylistSnapshot {
+	t.Helper()
+	deadline := time.Now().Add(time.Second)
+	for time.Now().Before(deadline) {
+		if snapshot, ok := controller.Snapshot(); ok && predicate(snapshot) {
+			return snapshot
+		}
+		time.Sleep(time.Millisecond)
+	}
+	snapshot, _ := controller.Snapshot()
+	t.Fatalf("timed out waiting for playlist snapshot; latest = %#v", snapshot)
+	return PlaylistSnapshot{}
+}
+
+func receivePlaylistSession(t *testing.T, sessions <-chan SessionCommand) SessionCommand {
+	t.Helper()
+	select {
+	case command := <-sessions:
+		return command
+	case <-time.After(time.Second):
+		t.Fatal("timed out waiting for session command")
+		return SessionCommand{}
+	}
+}
+
+func waitForPlaylistResult(t *testing.T, result <-chan error) error {
+	t.Helper()
+	select {
+	case err := <-result:
+		return err
+	case <-time.After(time.Second):
+		t.Fatal("timed out waiting for playlist controller")
+		return nil
+	}
+}