feat(radio): add radio metadata session manager

Closes #3

Signed-off-by: hiyukoim <11803292+hiyukoim@users.noreply.github.com>

core/radio/metadata_manager.go

@@ -0,0 +1,224 @@
+package radio
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"time"
+)
+
+var (
+	ErrInvalidSession = errors.New("invalid radio metadata session")
+	ErrInvalidStation = errors.New("invalid radio metadata station")
+)
+
+type Station struct {
+	ID        string
+	StreamURL string
+}
+
+type TitleUpdate struct {
+	RadioID   string    `json:"radioId"`
+	Title     string    `json:"title"`
+	UpdatedAt time.Time `json:"updatedAt"`
+}
+
+type StreamReader func(ctx context.Context, streamURL string, handleTitle func(string)) error
+
+type TitlePublisher func(ctx context.Context, update TitleUpdate)
+
+type MetadataManager struct {
+	mu sync.Mutex
+
+	reader  StreamReader
+	publish TitlePublisher
+	now     func() time.Time
+	backoff func(attempt int) time.Duration
+
+	sessions map[string]activeSession
+	readers  map[string]*activeReader
+}
+
+type ManagerOption func(*MetadataManager)
+
+type activeSession struct {
+	radioID   string
+	streamURL string
+}
+
+type activeReader struct {
+	streamURL string
+	ctx       context.Context
+	cancel    context.CancelFunc
+	radioRefs map[string]int
+	lastTitle string
+}
+
+func NewMetadataManager(reader StreamReader, publisher TitlePublisher, opts ...ManagerOption) *MetadataManager {
+	m := &MetadataManager{
+		reader:   reader,
+		publish:  publisher,
+		now:      time.Now,
+		backoff:  defaultBackoff,
+		sessions: map[string]activeSession{},
+		readers:  map[string]*activeReader{},
+	}
+	for _, opt := range opts {
+		opt(m)
+	}
+	return m
+}
+
+func WithNow(now func() time.Time) ManagerOption {
+	return func(m *MetadataManager) {
+		if now != nil {
+			m.now = now
+		}
+	}
+}
+
+func WithRetryBackoff(backoff func(attempt int) time.Duration) ManagerOption {
+	return func(m *MetadataManager) {
+		if backoff != nil {
+			m.backoff = backoff
+		}
+	}
+}
+
+func (m *MetadataManager) Start(ctx context.Context, sessionID string, station Station) error {
+	if sessionID == "" {
+		return ErrInvalidSession
+	}
+	if station.ID == "" || station.StreamURL == "" {
+		return ErrInvalidStation
+	}
+	if m.reader == nil {
+		return errors.New("missing radio metadata stream reader")
+	}
+
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	if current, ok := m.sessions[sessionID]; ok {
+		if current.radioID == station.ID && current.streamURL == station.StreamURL {
+			return nil
+		}
+		m.removeSessionLocked(sessionID)
+	}
+
+	reader := m.readers[station.StreamURL]
+	if reader == nil {
+		readerCtx, cancel := context.WithCancel(ctx)
+		reader = &activeReader{
+			streamURL: station.StreamURL,
+			ctx:       readerCtx,
+			cancel:    cancel,
+			radioRefs: map[string]int{},
+		}
+		m.readers[station.StreamURL] = reader
+		go m.runReader(reader)
+	}
+
+	reader.radioRefs[station.ID]++
+	m.sessions[sessionID] = activeSession{radioID: station.ID, streamURL: station.StreamURL}
+	return nil
+}
+
+func (m *MetadataManager) Stop(sessionID string) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.removeSessionLocked(sessionID)
+}
+
+func (m *MetadataManager) removeSessionLocked(sessionID string) {
+	session, ok := m.sessions[sessionID]
+	if !ok {
+		return
+	}
+	delete(m.sessions, sessionID)
+
+	reader := m.readers[session.streamURL]
+	if reader == nil {
+		return
+	}
+
+	if count := reader.radioRefs[session.radioID]; count <= 1 {
+		delete(reader.radioRefs, session.radioID)
+	} else {
+		reader.radioRefs[session.radioID] = count - 1
+	}
+
+	if len(reader.radioRefs) == 0 {
+		delete(m.readers, session.streamURL)
+		reader.cancel()
+	}
+}
+
+func (m *MetadataManager) runReader(reader *activeReader) {
+	attempt := 0
+	for {
+		err := m.reader(reader.ctx, reader.streamURL, func(title string) {
+			m.publishTitle(reader, title)
+		})
+		if reader.ctx.Err() != nil {
+			return
+		}
+		if err == nil {
+			attempt = 0
+		} else {
+			attempt++
+		}
+
+		delay := m.backoff(attempt)
+		if delay <= 0 {
+			continue
+		}
+
+		timer := time.NewTimer(delay)
+		select {
+		case <-reader.ctx.Done():
+			timer.Stop()
+			return
+		case <-timer.C:
+		}
+	}
+}
+
+func (m *MetadataManager) publishTitle(reader *activeReader, title string) {
+	m.mu.Lock()
+	if reader.ctx.Err() != nil || title == "" || title == reader.lastTitle {
+		m.mu.Unlock()
+		return
+	}
+
+	reader.lastTitle = title
+	radioIDs := make([]string, 0, len(reader.radioRefs))
+	for radioID := range reader.radioRefs {
+		radioIDs = append(radioIDs, radioID)
+	}
+	updatedAt := m.now()
+	publish := m.publish
+	m.mu.Unlock()
+
+	if publish == nil {
+		return
+	}
+	for _, radioID := range radioIDs {
+		publish(reader.ctx, TitleUpdate{
+			RadioID:   radioID,
+			Title:     title,
+			UpdatedAt: updatedAt,
+		})
+	}
+}
+
+func defaultBackoff(attempt int) time.Duration {
+	if attempt < 1 {
+		attempt = 1
+	}
+	delay := time.Duration(attempt) * time.Second
+	if delay > 30*time.Second {
+		return 30 * time.Second
+	}
+	return delay
+}

core/radio/metadata_manager_test.go

@@ -0,0 +1,231 @@
+package radio
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"time"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+var _ = Describe("MetadataManager", func() {
+	var (
+		reader    *fakeStreamReader
+		publisher *fakePublisher
+		manager   *MetadataManager
+	)
+
+	BeforeEach(func() {
+		reader = newFakeStreamReader()
+		publisher = newFakePublisher()
+		manager = NewMetadataManager(
+			reader.Read,
+			publisher.Publish,
+			WithRetryBackoff(func(int) time.Duration { return 0 }),
+			WithNow(func() time.Time { return time.Unix(123, 0).UTC() }),
+		)
+	})
+
+	It("starts a reader for an active radio session", func() {
+		err := manager.Start(context.Background(), "session-1", Station{
+			ID:        "rd-1",
+			StreamURL: "https://stream.example.test/radio",
+		})
+
+		Expect(err).ToNot(HaveOccurred())
+		Eventually(reader.StartCount).Should(Equal(1))
+		Expect(reader.URLs()).To(Equal([]string{"https://stream.example.test/radio"}))
+	})
+
+	It("shares one reader for sessions using the same stream URL", func() {
+		Expect(manager.Start(context.Background(), "session-1", Station{ID: "rd-1", StreamURL: "https://stream.example.test/radio"})).To(Succeed())
+		Expect(manager.Start(context.Background(), "session-2", Station{ID: "rd-1", StreamURL: "https://stream.example.test/radio"})).To(Succeed())
+
+		Eventually(reader.StartCount).Should(Equal(1))
+		Consistently(reader.StartCount).Should(Equal(1))
+	})
+
+	It("cancels the reader when the last session stops", func() {
+		Expect(manager.Start(context.Background(), "session-1", Station{ID: "rd-1", StreamURL: "https://stream.example.test/radio"})).To(Succeed())
+		Eventually(reader.StartCount).Should(Equal(1))
+
+		manager.Stop("session-1")
+
+		Eventually(reader.CancelCount).Should(Equal(1))
+	})
+
+	It("keeps the reader alive while another session still references the stream URL", func() {
+		Expect(manager.Start(context.Background(), "session-1", Station{ID: "rd-1", StreamURL: "https://stream.example.test/radio"})).To(Succeed())
+		Expect(manager.Start(context.Background(), "session-2", Station{ID: "rd-1", StreamURL: "https://stream.example.test/radio"})).To(Succeed())
+		Eventually(reader.StartCount).Should(Equal(1))
+
+		manager.Stop("session-1")
+
+		Consistently(reader.CancelCount).Should(Equal(0))
+		manager.Stop("session-2")
+		Eventually(reader.CancelCount).Should(Equal(1))
+	})
+
+	It("stops an existing session before starting that session on another station", func() {
+		Expect(manager.Start(context.Background(), "session-1", Station{ID: "rd-1", StreamURL: "https://stream.example.test/one"})).To(Succeed())
+		Eventually(reader.StartCount).Should(Equal(1))
+
+		Expect(manager.Start(context.Background(), "session-1", Station{ID: "rd-2", StreamURL: "https://stream.example.test/two"})).To(Succeed())
+
+		Eventually(reader.CancelCount).Should(Equal(1))
+		Eventually(reader.StartCount).Should(Equal(2))
+		Expect(reader.URLs()).To(Equal([]string{
+			"https://stream.example.test/one",
+			"https://stream.example.test/two",
+		}))
+	})
+
+	It("publishes title updates for active radio IDs", func() {
+		Expect(manager.Start(context.Background(), "session-1", Station{ID: "rd-1", StreamURL: "https://stream.example.test/radio"})).To(Succeed())
+		Eventually(reader.StartCount).Should(Equal(1))
+
+		reader.Emit("Live Artist - Live Track")
+
+		Eventually(publisher.Updates).Should(Equal([]TitleUpdate{{
+			RadioID:   "rd-1",
+			Title:     "Live Artist - Live Track",
+			UpdatedAt: time.Unix(123, 0).UTC(),
+		}}))
+	})
+
+	It("does not publish duplicate consecutive titles from the same reader", func() {
+		Expect(manager.Start(context.Background(), "session-1", Station{ID: "rd-1", StreamURL: "https://stream.example.test/radio"})).To(Succeed())
+		Eventually(reader.StartCount).Should(Equal(1))
+
+		reader.Emit("Same Title")
+		reader.Emit("Same Title")
+
+		Consistently(publisher.Updates).Should(Equal([]TitleUpdate{{
+			RadioID:   "rd-1",
+			Title:     "Same Title",
+			UpdatedAt: time.Unix(123, 0).UTC(),
+		}}))
+	})
+
+	It("retries reader failures until stopped", func() {
+		reader.err = errors.New("stream failed")
+
+		Expect(manager.Start(context.Background(), "session-1", Station{ID: "rd-1", StreamURL: "https://stream.example.test/radio"})).To(Succeed())
+
+		Eventually(reader.StartCount).Should(BeNumerically(">", 1))
+		manager.Stop("session-1")
+		cancelCount := reader.CancelCount()
+		Consistently(reader.StartCount).Should(BeNumerically(">=", 2))
+		Expect(reader.CancelCount()).To(BeNumerically(">=", cancelCount))
+	})
+
+	It("ignores unknown stops", func() {
+		manager.Stop("missing-session")
+		Expect(reader.StartCount()).To(Equal(0))
+	})
+
+	It("rejects invalid start requests", func() {
+		Expect(manager.Start(context.Background(), "", Station{ID: "rd-1", StreamURL: "https://stream.example.test/radio"})).To(MatchError(ErrInvalidSession))
+		Expect(manager.Start(context.Background(), "session-1", Station{ID: "", StreamURL: "https://stream.example.test/radio"})).To(MatchError(ErrInvalidStation))
+		Expect(manager.Start(context.Background(), "session-1", Station{ID: "rd-1", StreamURL: ""})).To(MatchError(ErrInvalidStation))
+	})
+})
+
+type fakeStreamReader struct {
+	mu        sync.Mutex
+	started   chan struct{}
+	cancelled chan struct{}
+	handler   func(string)
+	urls      []string
+	starts    int
+	cancels   int
+	err       error
+}
+
+func newFakeStreamReader() *fakeStreamReader {
+	return &fakeStreamReader{
+		started:   make(chan struct{}),
+		cancelled: make(chan struct{}),
+	}
+}
+
+func (r *fakeStreamReader) Read(ctx context.Context, streamURL string, handleTitle func(string)) error {
+	r.mu.Lock()
+	r.starts++
+	r.urls = append(r.urls, streamURL)
+	r.handler = handleTitle
+	if r.starts == 1 {
+		close(r.started)
+	}
+	err := r.err
+	r.mu.Unlock()
+
+	if err != nil {
+		return err
+	}
+
+	<-ctx.Done()
+
+	r.mu.Lock()
+	r.cancels++
+	if r.cancels == 1 {
+		close(r.cancelled)
+	}
+	r.mu.Unlock()
+
+	return ctx.Err()
+}
+
+func (r *fakeStreamReader) Emit(title string) {
+	Eventually(func() func(string) {
+		r.mu.Lock()
+		defer r.mu.Unlock()
+		return r.handler
+	}).ShouldNot(BeNil())
+
+	r.mu.Lock()
+	handler := r.handler
+	r.mu.Unlock()
+	handler(title)
+}
+
+func (r *fakeStreamReader) StartCount() int {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	return r.starts
+}
+
+func (r *fakeStreamReader) CancelCount() int {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	return r.cancels
+}
+
+func (r *fakeStreamReader) URLs() []string {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	return append([]string(nil), r.urls...)
+}
+
+type fakePublisher struct {
+	mu      sync.Mutex
+	updates []TitleUpdate
+}
+
+func newFakePublisher() *fakePublisher {
+	return &fakePublisher{}
+}
+
+func (p *fakePublisher) Publish(_ context.Context, update TitleUpdate) {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	p.updates = append(p.updates, update)
+}
+
+func (p *fakePublisher) Updates() []TitleUpdate {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	return append([]TitleUpdate(nil), p.updates...)
+}

core/radio/radio_suite_test.go

@@ -0,0 +1,17 @@
+package radio
+
+import (
+	"testing"
+
+	"github.com/navidrome/navidrome/log"
+	"github.com/navidrome/navidrome/tests"
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+func TestRadio(t *testing.T) {
+	tests.Init(t, false)
+	log.SetLevel(log.LevelFatal)
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "Radio Suite")
+}