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

package playback

import (
	"context"
	"errors"
	"fmt"
	"sync"
)

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 SessionSnapshot struct {
	Desired    SessionConfig
	Plan       SessionPlan
	Generation uint64
}

type SessionController struct {
	videoSlot VideoSlotRunner
	audioSlot AudioSlotRunner
	syncSlot  SyncSlotRunner
	canSync   SyncPredicate

	mu          sync.RWMutex
	snapshot    SessionSnapshot
	hasSnapshot bool
}

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
	generation := uint64(1)
	runtime := c.startSessionRuntime(ctx, plan, generation)
	c.publish(SessionSnapshot{
		Desired:    initial,
		Plan:       plan,
		Generation: generation,
	})

	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
			}

			nextGeneration := generation
			if plan.Topology != nextPlan.Topology {
				nextGeneration++
			}
			nextRuntime, err := c.reconcileSessionRuntime(
				ctx,
				runtime,
				plan,
				nextPlan,
				nextGeneration,
			)
			if err != nil {
				stopSessionRuntime(runtime)
				if ctx.Err() != nil {
					return ctx.Err()
				}
				return err
			}

			generation = nextGeneration
			desired = nextDesired
			plan = nextPlan
			runtime = nextRuntime
			c.publish(SessionSnapshot{
				Desired:    desired,
				Plan:       plan,
				Generation: generation,
			})
		}
	}
}

func (c *SessionController) startSessionRuntime(
	ctx context.Context,
	plan SessionPlan,
	generation uint64,
) *sessionRuntime {
	if plan.Topology == TopologyIdle {
		return &sessionRuntime{topology: TopologyIdle}
	}

	runtimeCtx, cancel := context.WithCancel(withGeneration(ctx, generation))
	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,
	nextGeneration uint64,
) (*sessionRuntime, error) {
	if current.Topology != next.Topology {
		stopSessionRuntime(runtime)
		if err := ctx.Err(); err != nil {
			return runtime, err
		}
		return c.startSessionRuntime(ctx, next, nextGeneration), 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)
}

func (c *SessionController) Snapshot() (SessionSnapshot, bool) {
	c.mu.RLock()
	snapshot := c.snapshot
	ok := c.hasSnapshot
	c.mu.RUnlock()
	return snapshot, ok
}

func (c *SessionController) publish(snapshot SessionSnapshot) {
	c.mu.Lock()
	c.snapshot = snapshot
	c.hasSnapshot = true
	c.mu.Unlock()
}