diff --git a/cmd/wire_gen.go b/cmd/wire_gen.go index c4a797ef4..71986d36b 100644 --- a/cmd/wire_gen.go +++ b/cmd/wire_gen.go @@ -21,6 +21,7 @@ import ( "github.com/navidrome/navidrome/core/metrics" "github.com/navidrome/navidrome/core/playback" "github.com/navidrome/navidrome/core/playlists" + "github.com/navidrome/navidrome/core/radio" "github.com/navidrome/navidrome/core/scrobbler" "github.com/navidrome/navidrome/core/sonic" "github.com/navidrome/navidrome/core/stream" @@ -83,7 +84,9 @@ func CreateNativeAPIRouter(ctx context.Context) *nativeapi.Router { library := core.NewLibrary(dataStore, modelScanner, watcher, broker, manager) user := core.NewUser(dataStore, manager) maintenance := core.NewMaintenance(dataStore) - router := nativeapi.New(dataStore, share, playlistsPlaylists, insights, library, user, maintenance, manager, imageUploadService) + titlePublisher := events.NewRadioMetadataPublisher(broker) + metadataManager := radio.NewMetadataManagerService(titlePublisher) + router := nativeapi.New(dataStore, share, playlistsPlaylists, insights, library, user, maintenance, manager, imageUploadService, metadataManager) return router } @@ -247,7 +250,7 @@ func getPluginManager() *plugins.Manager { // wire_injectors.go: -var allProviders = wire.NewSet(core.Set, artwork.Set, server.New, subsonic.New, jellyfin.New, nativeapi.New, public.New, persistence.New, lastfm.NewRouter, listenbrainz.NewRouter, events.GetBroker, scanner.New, scanner.GetWatcher, metrics.GetPrometheusInstance, db.Db, plugins.GetManager, sonic.New, wire.Bind(new(agents.PluginLoader), new(*plugins.Manager)), wire.Bind(new(scrobbler.PluginLoader), new(*plugins.Manager)), wire.Bind(new(lyrics.PluginLoader), new(*plugins.Manager)), wire.Bind(new(sonic.PluginLoader), new(*plugins.Manager)), wire.Bind(new(sonic.Engine), new(*sonic.Sonic)), wire.Bind(new(nativeapi.PluginManager), new(*plugins.Manager)), wire.Bind(new(core.PluginUnloader), new(*plugins.Manager)), wire.Bind(new(plugins.PluginMetricsRecorder), new(metrics.Metrics)), wire.Bind(new(core.Watcher), new(scanner.Watcher))) +var allProviders = wire.NewSet(core.Set, artwork.Set, server.New, subsonic.New, jellyfin.New, nativeapi.New, public.New, persistence.New, lastfm.NewRouter, listenbrainz.NewRouter, events.GetBroker, events.NewRadioMetadataPublisher, scanner.New, scanner.GetWatcher, metrics.GetPrometheusInstance, db.Db, plugins.GetManager, sonic.New, wire.Bind(new(agents.PluginLoader), new(*plugins.Manager)), wire.Bind(new(scrobbler.PluginLoader), new(*plugins.Manager)), wire.Bind(new(lyrics.PluginLoader), new(*plugins.Manager)), wire.Bind(new(sonic.PluginLoader), new(*plugins.Manager)), wire.Bind(new(sonic.Engine), new(*sonic.Sonic)), wire.Bind(new(nativeapi.PluginManager), new(*plugins.Manager)), wire.Bind(new(nativeapi.RadioMetadataManager), new(*radio.MetadataManager)), wire.Bind(new(core.PluginUnloader), new(*plugins.Manager)), wire.Bind(new(plugins.PluginMetricsRecorder), new(metrics.Metrics)), wire.Bind(new(core.Watcher), new(scanner.Watcher))) func GetPluginManager(ctx context.Context) *plugins.Manager { manager := getPluginManager() diff --git a/cmd/wire_injectors.go b/cmd/wire_injectors.go index 94faa5af3..f639e69da 100644 --- a/cmd/wire_injectors.go +++ b/cmd/wire_injectors.go @@ -14,6 +14,7 @@ import ( "github.com/navidrome/navidrome/core/lyrics" "github.com/navidrome/navidrome/core/metrics" "github.com/navidrome/navidrome/core/playback" + "github.com/navidrome/navidrome/core/radio" "github.com/navidrome/navidrome/core/scrobbler" "github.com/navidrome/navidrome/core/sonic" "github.com/navidrome/navidrome/db" @@ -41,6 +42,7 @@ var allProviders = wire.NewSet( lastfm.NewRouter, listenbrainz.NewRouter, events.GetBroker, + events.NewRadioMetadataPublisher, scanner.New, scanner.GetWatcher, metrics.GetPrometheusInstance, @@ -53,6 +55,7 @@ var allProviders = wire.NewSet( wire.Bind(new(sonic.PluginLoader), new(*plugins.Manager)), wire.Bind(new(sonic.Engine), new(*sonic.Sonic)), wire.Bind(new(nativeapi.PluginManager), new(*plugins.Manager)), + wire.Bind(new(nativeapi.RadioMetadataManager), new(*radio.MetadataManager)), wire.Bind(new(core.PluginUnloader), new(*plugins.Manager)), wire.Bind(new(plugins.PluginMetricsRecorder), new(metrics.Metrics)), wire.Bind(new(core.Watcher), new(scanner.Watcher)), diff --git a/core/radio/icy/parser.go b/core/radio/icy/parser.go new file mode 100644 index 000000000..39371dfca --- /dev/null +++ b/core/radio/icy/parser.go @@ -0,0 +1,125 @@ +package icy + +import ( + "bytes" + "context" + "errors" + "io" + "strings" + "unicode/utf8" + + "golang.org/x/text/encoding/charmap" +) + +var ( + ErrInvalidMetaInt = errors.New("invalid icy metadata interval") + ErrMissingTitleHandler = errors.New("missing icy title handler") + errEndOfStream = errors.New("icy stream ended") +) + +const ( + maxMetaInt = 1024 * 1024 + maxMetadataLen = 255 * 16 +) + +// ReadStreamTitles reads ICY metadata blocks from r and emits changed StreamTitle values. +func ReadStreamTitles(ctx context.Context, r io.Reader, metaInt int, handleTitle func(string)) error { + if metaInt <= 0 || metaInt > maxMetaInt { + return ErrInvalidMetaInt + } + if handleTitle == nil { + return ErrMissingTitleHandler + } + + length := make([]byte, 1) + metadataBuf := make([]byte, maxMetadataLen) + lastTitle := "" + + for { + if err := ctx.Err(); err != nil { + return err + } + + if _, err := io.CopyN(io.Discard, r, int64(metaInt)); err != nil { + if errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF) { + return nil + } + return err + } + + if err := ctx.Err(); err != nil { + return err + } + + if err := readFullOrEnd(r, length); err != nil { + if errors.Is(err, errEndOfStream) { + return nil + } + return err + } + + metadataLen := int(length[0]) * 16 + if metadataLen == 0 { + continue + } + + metadata := metadataBuf[:metadataLen] + if err := readFullOrEnd(r, metadata); err != nil { + if errors.Is(err, errEndOfStream) { + return nil + } + return err + } + + title := streamTitle(metadata) + if title == "" || title == lastTitle { + continue + } + + lastTitle = title + handleTitle(title) + } +} + +func readFullOrEnd(r io.Reader, buf []byte) error { + if _, err := io.ReadFull(r, buf); err != nil { + if errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF) { + return errEndOfStream + } + return err + } + return nil +} + +func streamTitle(metadata []byte) string { + text := decodeMetadata(bytes.TrimRight(metadata, "\x00")) + const prefix = "StreamTitle='" + + start := strings.Index(text, prefix) + if start < 0 { + return "" + } + + start += len(prefix) + end := strings.Index(text[start:], "';") + if end < 0 { + end = strings.IndexByte(text[start:], '\'') + if end < 0 { + return "" + } + } + + return strings.TrimSpace(text[start : start+end]) +} + +func decodeMetadata(metadata []byte) string { + if utf8.Valid(metadata) { + return string(metadata) + } + + text, err := charmap.ISO8859_1.NewDecoder().String(string(metadata)) + if err != nil { + return string(bytes.ToValidUTF8(metadata, nil)) + } + return text +} diff --git a/core/radio/icy/parser_suite_test.go b/core/radio/icy/parser_suite_test.go new file mode 100644 index 000000000..94f1e8fdc --- /dev/null +++ b/core/radio/icy/parser_suite_test.go @@ -0,0 +1,17 @@ +package icy + +import ( + "testing" + + "github.com/navidrome/navidrome/log" + "github.com/navidrome/navidrome/tests" + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +func TestICY(t *testing.T) { + tests.Init(t, false) + log.SetLevel(log.LevelFatal) + RegisterFailHandler(Fail) + RunSpecs(t, "ICY Suite") +} diff --git a/core/radio/icy/parser_test.go b/core/radio/icy/parser_test.go new file mode 100644 index 000000000..1877c2baa --- /dev/null +++ b/core/radio/icy/parser_test.go @@ -0,0 +1,193 @@ +package icy + +import ( + "bytes" + "context" + "errors" + "io" + "strings" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +var _ = Describe("ReadStreamTitles", func() { + const metaInt = 5 + + readTitles := func(ctx context.Context, stream []byte, interval int) ([]string, error) { + var titles []string + err := ReadStreamTitles(ctx, bytes.NewReader(stream), interval, func(title string) { + titles = append(titles, title) + }) + return titles, err + } + + It("emits changed StreamTitle values", func() { + stream := icyStream(metaInt, + "StreamTitle='Artist One - Track One';", + "StreamTitle='Artist One - Track One';", + "StreamTitle='Artist Two - Track Two';", + ) + + titles, err := readTitles(context.Background(), stream, metaInt) + + Expect(err).ToNot(HaveOccurred()) + Expect(titles).To(Equal([]string{ + "Artist One - Track One", + "Artist Two - Track Two", + })) + }) + + It("ignores empty metadata blocks", func() { + stream := audioInterval(metaInt) + stream = append(stream, 0) + + titles, err := readTitles(context.Background(), stream, metaInt) + + Expect(err).ToNot(HaveOccurred()) + Expect(titles).To(BeEmpty()) + }) + + It("ignores metadata blocks without StreamTitle", func() { + stream := icyStream(metaInt, "StreamUrl='https://example.test';") + + titles, err := readTitles(context.Background(), stream, metaInt) + + Expect(err).ToNot(HaveOccurred()) + Expect(titles).To(BeEmpty()) + }) + + It("ignores empty StreamTitle values", func() { + stream := icyStream(metaInt, "StreamTitle='';") + + titles, err := readTitles(context.Background(), stream, metaInt) + + Expect(err).ToNot(HaveOccurred()) + Expect(titles).To(BeEmpty()) + }) + + It("returns an error for invalid metadata intervals", func() { + titles, err := readTitles(context.Background(), nil, 0) + + Expect(err).To(MatchError(ErrInvalidMetaInt)) + Expect(titles).To(BeEmpty()) + }) + + It("returns an error for huge metadata intervals", func() { + titles, err := readTitles(context.Background(), nil, maxMetaInt+1) + + Expect(err).To(MatchError(ErrInvalidMetaInt)) + Expect(titles).To(BeEmpty()) + }) + + It("keeps single quotes inside StreamTitle values", func() { + stream := icyStream(metaInt, "StreamTitle='L'Oiseau - Track';") + + titles, err := readTitles(context.Background(), stream, metaInt) + + Expect(err).ToNot(HaveOccurred()) + Expect(titles).To(Equal([]string{"L'Oiseau - Track"})) + }) + + It("stops when context is cancelled", func() { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + titles, err := readTitles(ctx, audioInterval(metaInt), metaInt) + + Expect(err).To(MatchError(context.Canceled)) + Expect(titles).To(BeEmpty()) + }) + + It("returns cleanly when input ends mid-audio interval", func() { + titles, err := readTitles(context.Background(), []byte("abc"), metaInt) + + Expect(err).ToNot(HaveOccurred()) + Expect(titles).To(BeEmpty()) + }) + + It("returns cleanly when input ends mid-metadata block", func() { + stream := append(audioInterval(metaInt), 2, 'S', 't', 'r') + + titles, err := readTitles(context.Background(), stream, metaInt) + + Expect(err).ToNot(HaveOccurred()) + Expect(titles).To(BeEmpty()) + }) + + It("decodes Latin-1 metadata", func() { + metadata := []byte("StreamTitle='Beyonc") + metadata = append(metadata, 0xe9) + metadata = append(metadata, " - D\xe9j\xe0 Vu';"...) + stream := icyStreamBytes(metaInt, metadata) + + titles, err := readTitles(context.Background(), stream, metaInt) + + Expect(err).ToNot(HaveOccurred()) + Expect(titles).To(Equal([]string{"Beyoncé - Déjà Vu"})) + }) + + It("returns reader errors other than EOF", func() { + errReader := errAfterReader{ + data: []byte("abcde"), + err: errors.New("read failed"), + } + var titles []string + + err := ReadStreamTitles(context.Background(), &errReader, metaInt, func(title string) { + titles = append(titles, title) + }) + + Expect(err).To(MatchError("read failed")) + Expect(titles).To(BeEmpty()) + }) +}) + +func icyStream(metaInt int, metadataBlocks ...string) []byte { + blocks := make([][]byte, 0, len(metadataBlocks)) + for _, block := range metadataBlocks { + blocks = append(blocks, []byte(block)) + } + return icyStreamBytes(metaInt, blocks...) +} + +func icyStreamBytes(metaInt int, metadataBlocks ...[]byte) []byte { + var stream []byte + for _, metadata := range metadataBlocks { + stream = append(stream, audioInterval(metaInt)...) + stream = append(stream, metadataBlock(metadata)...) + } + return stream +} + +func audioInterval(metaInt int) []byte { + return []byte(strings.Repeat("a", metaInt)) +} + +func metadataBlock(metadata []byte) []byte { + if len(metadata) == 0 { + return []byte{0} + } + blockCount := (len(metadata) + 15) / 16 + block := []byte{byte(blockCount)} + block = append(block, metadata...) + padding := blockCount*16 - len(metadata) + block = append(block, bytes.Repeat([]byte{0}, padding)...) + return block +} + +type errAfterReader struct { + data []byte + err error +} + +func (r *errAfterReader) Read(p []byte) (int, error) { + if len(r.data) > 0 { + n := copy(p, r.data) + r.data = r.data[n:] + return n, nil + } + return 0, r.err +} + +var _ io.Reader = (*errAfterReader)(nil) diff --git a/core/radio/icy/reader.go b/core/radio/icy/reader.go new file mode 100644 index 000000000..e640dd116 --- /dev/null +++ b/core/radio/icy/reader.go @@ -0,0 +1,58 @@ +package icy + +import ( + "context" + "errors" + "fmt" + "net/http" + "strconv" + + "github.com/navidrome/navidrome/consts" +) + +var ErrMissingMetaInt = errors.New("missing icy metadata interval") + +type ResponseStatusError struct { + StatusCode int + Status string +} + +func (e *ResponseStatusError) Error() string { + return fmt.Sprintf("icy stream returned %s", e.Status) +} + +// ReadHTTPStreamTitles requests ICY metadata from streamURL and emits changed StreamTitle values. +func ReadHTTPStreamTitles(ctx context.Context, client *http.Client, streamURL string, handleTitle func(string)) error { + if client == nil { + client = http.DefaultClient + } + + req, err := http.NewRequestWithContext(ctx, http.MethodGet, streamURL, nil) + if err != nil { + return err + } + req.Header.Set("Icy-MetaData", "1") + req.Header.Set("User-Agent", consts.HTTPUserAgent) + + resp, err := client.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + + if resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusMultipleChoices { + return &ResponseStatusError{StatusCode: resp.StatusCode, Status: resp.Status} + } + + metaIntHeader := resp.Header.Get("icy-metaint") + if metaIntHeader == "" { + return ErrMissingMetaInt + } + + metaInt, err := strconv.Atoi(metaIntHeader) + if err != nil || metaInt <= 0 { + return fmt.Errorf("%w: %q", ErrInvalidMetaInt, metaIntHeader) + } + + return ReadStreamTitles(ctx, resp.Body, metaInt, handleTitle) +} diff --git a/core/radio/icy/reader_test.go b/core/radio/icy/reader_test.go new file mode 100644 index 000000000..bbc3c4c6b --- /dev/null +++ b/core/radio/icy/reader_test.go @@ -0,0 +1,124 @@ +package icy + +import ( + "context" + "errors" + "net/http" + "net/http/httptest" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +var _ = Describe("ReadHTTPStreamTitles", func() { + const metaInt = 5 + + It("requests ICY metadata and parses stream titles", func() { + var icyHeader string + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + icyHeader = r.Header.Get("Icy-MetaData") + w.Header().Set("icy-metaint", "5") + _, _ = w.Write(icyStream(metaInt, "StreamTitle='Artist - Title';")) + })) + defer server.Close() + + var titles []string + err := ReadHTTPStreamTitles(context.Background(), server.Client(), server.URL, func(title string) { + titles = append(titles, title) + }) + + Expect(err).ToNot(HaveOccurred()) + Expect(icyHeader).To(Equal("1")) + Expect(titles).To(Equal([]string{"Artist - Title"})) + }) + + It("follows redirects before reading stream metadata", func() { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/redirect" { + http.Redirect(w, r, "/stream", http.StatusFound) + return + } + + Expect(r.URL.Path).To(Equal("/stream")) + Expect(r.Header.Get("Icy-MetaData")).To(Equal("1")) + w.Header().Set("Icy-MetaInt", "5") + _, _ = w.Write(icyStream(metaInt, "StreamTitle='Redirected - Title';")) + })) + defer server.Close() + + var titles []string + err := ReadHTTPStreamTitles(context.Background(), server.Client(), server.URL+"/redirect", func(title string) { + titles = append(titles, title) + }) + + Expect(err).ToNot(HaveOccurred()) + Expect(titles).To(Equal([]string{"Redirected - Title"})) + }) + + It("returns an error when icy-metaint is missing", func() { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = w.Write([]byte("not an icy stream")) + })) + defer server.Close() + + err := ReadHTTPStreamTitles(context.Background(), server.Client(), server.URL, func(string) {}) + + Expect(errors.Is(err, ErrMissingMetaInt)).To(BeTrue()) + }) + + It("returns an error when icy-metaint is invalid", func() { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("icy-metaint", "zero") + _, _ = w.Write([]byte("not an icy stream")) + })) + defer server.Close() + + err := ReadHTTPStreamTitles(context.Background(), server.Client(), server.URL, func(string) {}) + + Expect(errors.Is(err, ErrInvalidMetaInt)).To(BeTrue()) + }) + + It("returns response status errors for non-2xx responses", func() { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, "offline", http.StatusServiceUnavailable) + })) + defer server.Close() + + err := ReadHTTPStreamTitles(context.Background(), server.Client(), server.URL, func(string) {}) + + var statusErr *ResponseStatusError + Expect(errors.As(err, &statusErr)).To(BeTrue()) + Expect(statusErr.StatusCode).To(Equal(http.StatusServiceUnavailable)) + }) + + It("stops when context is cancelled", func() { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + err := ReadHTTPStreamTitles(ctx, http.DefaultClient, "https://example.test/stream", func(string) {}) + + Expect(errors.Is(err, context.Canceled)).To(BeTrue()) + }) + + It("stops a body read when context is cancelled", func() { + started := make(chan struct{}) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("icy-metaint", "5") + w.(http.Flusher).Flush() + close(started) + <-r.Context().Done() + })) + defer server.Close() + + ctx, cancel := context.WithCancel(context.Background()) + errC := make(chan error, 1) + go func() { + errC <- ReadHTTPStreamTitles(ctx, server.Client(), server.URL, func(string) {}) + }() + + Eventually(started).Should(BeClosed()) + cancel() + + Eventually(errC).Should(Receive(MatchError(ContainSubstring("context canceled")))) + }) +}) diff --git a/core/radio/metadata_manager.go b/core/radio/metadata_manager.go new file mode 100644 index 000000000..4b4ed9613 --- /dev/null +++ b/core/radio/metadata_manager.go @@ -0,0 +1,326 @@ +package radio + +import ( + "context" + "errors" + "sync" + "time" + + "github.com/navidrome/navidrome/log" +) + +var ( + ErrInvalidSession = errors.New("invalid radio metadata session") + ErrInvalidStation = errors.New("invalid radio metadata station") +) + +// defaultSessionTTL bounds how long a session survives without a refresh, so +// clients that vanish without calling Stop (e.g. a closed browser tab) do not +// keep an ICY reader connected to the station forever. +const defaultSessionTTL = 10 * time.Minute + +// permanentRetryDelay is used instead of the regular backoff when the reader +// reports a permanent error (dead URL, station without ICY metadata), so we +// do not hammer a stream that will never yield titles. +const permanentRetryDelay = 10 * time.Minute + +type permanentError struct{ err error } + +func (e *permanentError) Error() string { return e.err.Error() } +func (e *permanentError) Unwrap() error { return e.err } + +// MarkPermanent wraps err so the retry loop switches to permanentRetryDelay. +func MarkPermanent(err error) error { + if err == nil { + return nil + } + return &permanentError{err: err} +} + +func isPermanent(err error) bool { + var p *permanentError + return errors.As(err, &p) +} + +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 + sessionTTL time.Duration + + sessions map[string]activeSession + readers map[string]*activeReader +} + +type ManagerOption func(*MetadataManager) + +type activeSession struct { + radioID string + streamURL string + notifyCtx context.Context + expiresAt time.Time + expireTimer *time.Timer +} + +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, + sessionTTL: defaultSessionTTL, + 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 + } + } +} + +// WithSessionTTL sets how long a session lives without a refreshing Start +// call. A zero or negative TTL disables expiry. +func WithSessionTTL(ttl time.Duration) ManagerOption { + return func(m *MetadataManager) { + m.sessionTTL = ttl + } +} + +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 { + m.touchSessionLocked(sessionID, current) + return nil + } + m.removeSessionLocked(sessionID) + } + + reader := m.readers[station.StreamURL] + if reader == nil { + readerCtx, cancel := context.WithCancel(context.WithoutCancel(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]++ + session := activeSession{ + radioID: station.ID, + streamURL: station.StreamURL, + notifyCtx: context.WithoutCancel(ctx), + } + if m.sessionTTL > 0 { + session.expiresAt = m.now().Add(m.sessionTTL) + session.expireTimer = time.AfterFunc(m.sessionTTL, func() { m.expireSession(sessionID) }) + } + m.sessions[sessionID] = session + return nil +} + +// touchSessionLocked extends the session deadline. The pending expiry timer +// reschedules itself when it fires before the new deadline. +func (m *MetadataManager) touchSessionLocked(sessionID string, session activeSession) { + if m.sessionTTL <= 0 || session.expireTimer == nil { + return + } + session.expiresAt = m.now().Add(m.sessionTTL) + m.sessions[sessionID] = session +} + +func (m *MetadataManager) expireSession(sessionID string) { + m.mu.Lock() + defer m.mu.Unlock() + session, ok := m.sessions[sessionID] + if !ok || session.expireTimer == nil { + return + } + if remaining := session.expiresAt.Sub(m.now()); remaining > 0 { + session.expireTimer.Reset(remaining) + return + } + log.Debug("Radio metadata session expired without refresh", "sessionID", sessionID, "radioID", session.radioID) + m.removeSessionLocked(sessionID) +} + +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) + if session.expireTimer != nil { + session.expireTimer.Stop() + } + + 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++ + log.Trace("Radio metadata reader retrying", "streamURL", reader.streamURL, "attempt", attempt, err) + } + + delay := m.backoff(attempt) + if isPermanent(err) && delay < permanentRetryDelay { + delay = permanentRetryDelay + } + 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 + // Copy matching session contexts while the lock is held. + type sessionNotify struct { + radioID string + notifyCtx context.Context + } + notifies := make([]sessionNotify, 0, len(m.sessions)) + for _, session := range m.sessions { + if session.streamURL != reader.streamURL { + continue + } + for _, radioID := range radioIDs { + if session.radioID == radioID { + notifies = append(notifies, sessionNotify{radioID: radioID, notifyCtx: session.notifyCtx}) + break + } + } + } + m.mu.Unlock() + + if publish == nil { + return + } + for _, target := range notifies { + publish(target.notifyCtx, TitleUpdate{ + RadioID: target.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 +} diff --git a/core/radio/metadata_manager_test.go b/core/radio/metadata_manager_test.go new file mode 100644 index 000000000..b03bf1f44 --- /dev/null +++ b/core/radio/metadata_manager_test.go @@ -0,0 +1,317 @@ +package radio + +import ( + "context" + "errors" + "sync" + "time" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +var _ = Describe("MetadataManager", func() { + type contextKey string + + 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("keeps a reader alive after the start request context is cancelled", func() { + ctx, cancel := context.WithCancel(context.WithValue(context.Background(), contextKey("user"), "janedoe")) + Expect(manager.Start(ctx, "session-1", Station{ID: "rd-1", StreamURL: "https://stream.example.test/radio"})).To(Succeed()) + Eventually(reader.StartCount).Should(Equal(1)) + + cancel() + + Consistently(reader.CancelCount).Should(Equal(0)) + reader.Emit("Live Artist - Live Track") + Eventually(func() any { + contexts := publisher.Contexts() + if len(contexts) == 0 { + return nil + } + return contexts[0].Value(contextKey("user")) + }).Should(Equal("janedoe")) + + manager.Stop("session-1") + Eventually(reader.CancelCount).Should(Equal(1)) + }) + + 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)) + }) + + It("backs off for a long time on permanent reader errors", func() { + reader.err = MarkPermanent(errors.New("missing icy metadata interval")) + slow := NewMetadataManager( + reader.Read, + publisher.Publish, + WithRetryBackoff(func(int) time.Duration { return time.Millisecond }), + ) + + Expect(slow.Start(context.Background(), "session-1", Station{ + ID: "rd-1", + StreamURL: "https://stream.example.test/radio", + })).To(Succeed()) + + Eventually(reader.StartCount).Should(Equal(1)) + // Regular backoff is 1ms; the permanent error must stretch it, so no + // second attempt happens within the observation window. + Consistently(reader.StartCount, 300*time.Millisecond, 20*time.Millisecond).Should(Equal(1)) + }) + + It("expires sessions that are never refreshed", func() { + expiring := NewMetadataManager( + reader.Read, + publisher.Publish, + WithRetryBackoff(func(int) time.Duration { return 0 }), + WithSessionTTL(20*time.Millisecond), + ) + + Expect(expiring.Start(context.Background(), "session-1", Station{ + ID: "rd-1", + StreamURL: "https://stream.example.test/radio", + })).To(Succeed()) + + Eventually(reader.StartCount).Should(Equal(1)) + Eventually(reader.CancelCount).Should(Equal(1)) + }) + + It("keeps refreshed sessions alive past the TTL", func() { + expiring := NewMetadataManager( + reader.Read, + publisher.Publish, + WithRetryBackoff(func(int) time.Duration { return 0 }), + WithSessionTTL(50*time.Millisecond), + ) + station := Station{ID: "rd-1", StreamURL: "https://stream.example.test/radio"} + + Expect(expiring.Start(context.Background(), "session-1", station)).To(Succeed()) + Eventually(reader.StartCount).Should(Equal(1)) + + // Refresh the session on every poll, well inside the TTL window. + Consistently(func() int { + Expect(expiring.Start(context.Background(), "session-1", station)).To(Succeed()) + return reader.CancelCount() + }, 200*time.Millisecond, 10*time.Millisecond).Should(Equal(0)) + }) +}) + +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 + ctxs []context.Context +} + +func newFakePublisher() *fakePublisher { + return &fakePublisher{} +} + +func (p *fakePublisher) Publish(ctx context.Context, update TitleUpdate) { + p.mu.Lock() + defer p.mu.Unlock() + p.updates = append(p.updates, update) + p.ctxs = append(p.ctxs, ctx) +} + +func (p *fakePublisher) Updates() []TitleUpdate { + p.mu.Lock() + defer p.mu.Unlock() + return append([]TitleUpdate(nil), p.updates...) +} + +func (p *fakePublisher) Contexts() []context.Context { + p.mu.Lock() + defer p.mu.Unlock() + return append([]context.Context(nil), p.ctxs...) +} diff --git a/core/radio/provider.go b/core/radio/provider.go new file mode 100644 index 000000000..8a2558fab --- /dev/null +++ b/core/radio/provider.go @@ -0,0 +1,48 @@ +package radio + +import ( + "context" + "errors" + "net" + "net/http" + "time" + + "github.com/navidrome/navidrome/core/radio/icy" +) + +// icyHTTPClient has no overall timeout (ICY streams are long-lived), but bounds +// connection setup and header wait so a stalled station cannot hang the reader. +var icyHTTPClient = &http.Client{ + Transport: &http.Transport{ + DialContext: (&net.Dialer{ + Timeout: 10 * time.Second, + KeepAlive: 30 * time.Second, + }).DialContext, + TLSHandshakeTimeout: 10 * time.Second, + ResponseHeaderTimeout: 30 * time.Second, + IdleConnTimeout: 60 * time.Second, + MaxIdleConnsPerHost: 1, + }, +} + +func NewMetadataManagerService(publisher TitlePublisher) *MetadataManager { + return NewMetadataManager(func(ctx context.Context, streamURL string, handleTitle func(string)) error { + return classifyICYError(icy.ReadHTTPStreamTitles(ctx, icyHTTPClient, streamURL, handleTitle)) + }, publisher) +} + +// classifyICYError marks errors that a retry will not fix — a stream without +// ICY metadata or a client-error HTTP status — so the reader backs off for +// much longer instead of hammering the station. +func classifyICYError(err error) error { + if errors.Is(err, icy.ErrMissingMetaInt) || errors.Is(err, icy.ErrInvalidMetaInt) { + return MarkPermanent(err) + } + var statusErr *icy.ResponseStatusError + if errors.As(err, &statusErr) && + statusErr.StatusCode >= 400 && statusErr.StatusCode < 500 && + statusErr.StatusCode != http.StatusRequestTimeout && statusErr.StatusCode != http.StatusTooManyRequests { + return MarkPermanent(err) + } + return err +} diff --git a/core/radio/provider_test.go b/core/radio/provider_test.go new file mode 100644 index 000000000..f73ef770b --- /dev/null +++ b/core/radio/provider_test.go @@ -0,0 +1,20 @@ +package radio + +import ( + "fmt" + + "github.com/navidrome/navidrome/core/radio/icy" + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +var _ = Describe("Metadata provider", func() { + Describe("classifyICYError", func() { + It("marks invalid metadata intervals as permanent", func() { + err := classifyICYError(fmt.Errorf("%w: %q", icy.ErrInvalidMetaInt, "999999999")) + + Expect(isPermanent(err)).To(BeTrue()) + Expect(err).To(MatchError(ContainSubstring(icy.ErrInvalidMetaInt.Error()))) + }) + }) +}) diff --git a/core/radio/radio_suite_test.go b/core/radio/radio_suite_test.go new file mode 100644 index 000000000..cf7ce602e --- /dev/null +++ b/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") +} diff --git a/core/wire_providers.go b/core/wire_providers.go index a2fffa34f..b98c13d91 100644 --- a/core/wire_providers.go +++ b/core/wire_providers.go @@ -10,6 +10,7 @@ import ( "github.com/navidrome/navidrome/core/metrics" "github.com/navidrome/navidrome/core/playback" "github.com/navidrome/navidrome/core/playlists" + "github.com/navidrome/navidrome/core/radio" "github.com/navidrome/navidrome/core/scrobbler" "github.com/navidrome/navidrome/core/stream" ) @@ -36,4 +37,5 @@ var Set = wire.NewSet( playback.GetInstance, metrics.GetInstance, lyrics.NewLyrics, + radio.NewMetadataManagerService, ) diff --git a/server/events/events.go b/server/events/events.go index ff0a8a40a..d0962c59d 100644 --- a/server/events/events.go +++ b/server/events/events.go @@ -7,17 +7,24 @@ import ( "strings" "time" "unicode" + + coreradio "github.com/navidrome/navidrome/core/radio" ) type eventCtxKey string const broadcastToAllKey eventCtxKey = "broadcastToAll" +const sendToSameUserKey eventCtxKey = "sendToSameUser" // broadcastToAll is a context key that can be used to broadcast an event to all clients func broadcastToAll(ctx context.Context) context.Context { return context.WithValue(ctx, broadcastToAllKey, true) } +func sendToSameUser(ctx context.Context) context.Context { + return context.WithValue(ctx, sendToSameUserKey, true) +} + type Event interface { Name(Event) string Data(Event) string @@ -68,6 +75,26 @@ type NowPlayingCount struct { Count int `json:"count"` } +type RadioNowPlaying struct { + baseEvent + RadioID string `json:"radioId"` + Title string `json:"title"` + UpdatedAt time.Time `json:"updatedAt"` +} + +func NewRadioMetadataPublisher(broker Broker) coreradio.TitlePublisher { + return func(ctx context.Context, update coreradio.TitleUpdate) { + if broker == nil { + return + } + broker.SendMessage(sendToSameUser(ctx), &RadioNowPlaying{ + RadioID: update.RadioID, + Title: update.Title, + UpdatedAt: update.UpdatedAt, + }) + } +} + func (rr *RefreshResource) With(resource string, ids ...string) *RefreshResource { if rr.resources == nil { rr.resources = make(map[string][]string) diff --git a/server/events/events_test.go b/server/events/events_test.go index abfab9b6e..986b597f3 100644 --- a/server/events/events_test.go +++ b/server/events/events_test.go @@ -1,6 +1,12 @@ package events import ( + "context" + "net/http" + "time" + + coreradio "github.com/navidrome/navidrome/core/radio" + "github.com/navidrome/navidrome/model/request" . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" ) @@ -21,6 +27,49 @@ var _ = Describe("Events", func() { }) }) + Describe("RadioNowPlaying", func() { + It("serialises with a stable event name and payload", func() { + updatedAt := time.Date(2026, time.June, 30, 12, 34, 56, 0, time.UTC) + event := &RadioNowPlaying{ + RadioID: "rd-1", + Title: "Artist - Title", + UpdatedAt: updatedAt, + } + + Expect(event.Name(event)).To(Equal("radioNowPlaying")) + Expect(event.Data(event)).To(Equal(`{"radioId":"rd-1","title":"Artist - Title","updatedAt":"2026-06-30T12:34:56Z"}`)) + }) + + It("publishes metadata updates through the broker", func() { + ctx := request.WithClientUniqueId(request.WithUsername(context.Background(), "janedoe"), "browser-1") + updatedAt := time.Date(2026, time.June, 30, 12, 34, 56, 0, time.UTC) + capture := &capturingBroker{} + + publish := NewRadioMetadataPublisher(capture) + publish(ctx, coreradio.TitleUpdate{ + RadioID: "rd-1", + Title: "Artist - Title", + UpdatedAt: updatedAt, + }) + + username, hasUsername := request.UsernameFrom(capture.ctx) + clientID, hasClientID := request.ClientUniqueIdFrom(capture.ctx) + Expect(hasUsername).To(BeTrue()) + Expect(username).To(Equal("janedoe")) + Expect(hasClientID).To(BeTrue()) + Expect(clientID).To(Equal("browser-1")) + Expect(capture.broadcast).To(BeNil()) + Expect(capture.event).To(Equal(&RadioNowPlaying{ + RadioID: "rd-1", + Title: "Artist - Title", + UpdatedAt: updatedAt, + })) + routingBroker := broker{} + Expect(routingBroker.shouldSend(message{senderCtx: capture.ctx}, client{username: "janedoe", clientUniqueId: "browser-1"})).To(BeTrue()) + Expect(routingBroker.shouldSend(message{senderCtx: capture.ctx}, client{username: "johndoe", clientUniqueId: "browser-2"})).To(BeFalse()) + }) + }) + Describe("RefreshResource", func() { var rr *RefreshResource BeforeEach(func() { @@ -44,3 +93,20 @@ var _ = Describe("Events", func() { }) }) }) + +type capturingBroker struct { + ctx context.Context + event Event + broadcast Event +} + +func (b *capturingBroker) ServeHTTP(http.ResponseWriter, *http.Request) {} + +func (b *capturingBroker) SendMessage(ctx context.Context, event Event) { + b.ctx = ctx + b.event = event +} + +func (b *capturingBroker) SendBroadcastMessage(_ context.Context, event Event) { + b.broadcast = event +} diff --git a/server/events/sse.go b/server/events/sse.go index 565d8c016..49b9638b2 100644 --- a/server/events/sse.go +++ b/server/events/sse.go @@ -197,6 +197,15 @@ func (b *broker) shouldSend(msg message, c client) bool { if broadcastToAll, ok := msg.senderCtx.Value(broadcastToAllKey).(bool); ok && broadcastToAll { return true } + if sendToSameUser, ok := msg.senderCtx.Value(sendToSameUserKey).(bool); ok && sendToSameUser { + if username, ok := request.UsernameFrom(msg.senderCtx); ok { + return username == c.username + } + if user, ok := request.UserFrom(msg.senderCtx); ok { + return user.UserName == c.username + } + return true + } clientUniqueId, originatedFromClient := request.ClientUniqueIdFrom(msg.senderCtx) if !originatedFromClient { return true diff --git a/server/events/sse_test.go b/server/events/sse_test.go index e6a44ca15..9656e4d3d 100644 --- a/server/events/sse_test.go +++ b/server/events/sse_test.go @@ -38,6 +38,18 @@ var _ = Describe("Broker", func() { m := message{senderCtx: ctx} Expect(b.shouldSend(m, c)).To(BeFalse()) }) + It("sends same-user message for same username and same clientUniqueId", func() { + ctx = request.WithClientUniqueId(ctx, "1111") + ctx = request.WithUsername(ctx, "janedoe") + m := message{senderCtx: sendToSameUser(ctx)} + Expect(b.shouldSend(m, c)).To(BeTrue()) + }) + It("does not send same-user message for a different username", func() { + ctx = request.WithClientUniqueId(ctx, "1111") + ctx = request.WithUsername(ctx, "johndoe") + m := message{senderCtx: sendToSameUser(ctx)} + Expect(b.shouldSend(m, c)).To(BeFalse()) + }) It("does not send message for different username", func() { ctx = request.WithClientUniqueId(ctx, "3333") ctx = request.WithUsername(ctx, "johndoe") diff --git a/server/nativeapi/config_test.go b/server/nativeapi/config_test.go index 107b01e01..d1007f457 100644 --- a/server/nativeapi/config_test.go +++ b/server/nativeapi/config_test.go @@ -29,7 +29,7 @@ var _ = Describe("Config API", func() { conf.Server.DevUIShowConfig = true // Enable config endpoint for tests ds = &tests.MockDataStore{} auth.Init(ds) - nativeRouter := New(ds, nil, nil, nil, tests.NewMockLibraryService(), tests.NewMockUserService(), nil, nil, nil) + nativeRouter := New(ds, nil, nil, nil, tests.NewMockLibraryService(), tests.NewMockUserService(), nil, nil, nil, nil) router = server.JWTVerifier(nativeRouter) // Create test users diff --git a/server/nativeapi/library_test.go b/server/nativeapi/library_test.go index 9b7061845..13b33c238 100644 --- a/server/nativeapi/library_test.go +++ b/server/nativeapi/library_test.go @@ -31,7 +31,7 @@ var _ = Describe("Library API", func() { conf.Server.EnableSharing = false ds = &tests.MockDataStore{} auth.Init(ds) - nativeRouter := New(ds, nil, nil, nil, tests.NewMockLibraryService(), tests.NewMockUserService(), nil, nil, nil) + nativeRouter := New(ds, nil, nil, nil, tests.NewMockLibraryService(), tests.NewMockUserService(), nil, nil, nil, nil) router = server.JWTVerifier(nativeRouter) // Create test users diff --git a/server/nativeapi/native_api.go b/server/nativeapi/native_api.go index 5a7023eb6..8b7c930f9 100644 --- a/server/nativeapi/native_api.go +++ b/server/nativeapi/native_api.go @@ -15,6 +15,7 @@ import ( "github.com/navidrome/navidrome/core" "github.com/navidrome/navidrome/core/metrics" playlistsvc "github.com/navidrome/navidrome/core/playlists" + coreradio "github.com/navidrome/navidrome/core/radio" "github.com/navidrome/navidrome/log" "github.com/navidrome/navidrome/model" "github.com/navidrome/navidrome/model/request" @@ -34,6 +35,11 @@ type PluginManager interface { UnloadDisabledPlugins(ctx context.Context) } +type RadioMetadataManager interface { + Start(ctx context.Context, sessionID string, station coreradio.Station) error + Stop(sessionID string) +} + type Router struct { http.Handler ds model.DataStore @@ -45,10 +51,11 @@ type Router struct { maintenance core.Maintenance pluginManager PluginManager imgUpload core.ImageUploadService + radioMetadata RadioMetadataManager } -func New(ds model.DataStore, share core.Share, playlists playlistsvc.Playlists, insights metrics.Insights, libraryService core.Library, userService core.User, maintenance core.Maintenance, pluginManager PluginManager, imgUpload core.ImageUploadService) *Router { - r := &Router{ds: ds, share: share, playlists: playlists, insights: insights, libs: libraryService, users: userService, maintenance: maintenance, pluginManager: pluginManager, imgUpload: imgUpload} +func New(ds model.DataStore, share core.Share, playlists playlistsvc.Playlists, insights metrics.Insights, libraryService core.Library, userService core.User, maintenance core.Maintenance, pluginManager PluginManager, imgUpload core.ImageUploadService, radioMetadata RadioMetadataManager) *Router { + r := &Router{ds: ds, share: share, playlists: playlists, insights: insights, libs: libraryService, users: userService, maintenance: maintenance, pluginManager: pluginManager, imgUpload: imgUpload, radioMetadata: radioMetadata} r.Handler = r.routes() return r } diff --git a/server/nativeapi/native_api_song_test.go b/server/nativeapi/native_api_song_test.go index b1ed09d65..203fcd4cf 100644 --- a/server/nativeapi/native_api_song_test.go +++ b/server/nativeapi/native_api_song_test.go @@ -95,7 +95,7 @@ var _ = Describe("Song Endpoints", func() { mfRepo.SetData(testSongs) // Create the native API router and wrap it with the JWTVerifier middleware - nativeRouter := New(ds, nil, nil, nil, tests.NewMockLibraryService(), tests.NewMockUserService(), nil, nil, nil) + nativeRouter := New(ds, nil, nil, nil, tests.NewMockLibraryService(), tests.NewMockUserService(), nil, nil, nil, nil) router = server.JWTVerifier(nativeRouter) w = httptest.NewRecorder() }) diff --git a/server/nativeapi/playlists_test.go b/server/nativeapi/playlists_test.go index 9bf502687..bc7c74efc 100644 --- a/server/nativeapi/playlists_test.go +++ b/server/nativeapi/playlists_test.go @@ -99,7 +99,7 @@ var _ = Describe("Playlist Tracks Endpoint", func() { err := userRepo.Put(&testUser) Expect(err).ToNot(HaveOccurred()) - nativeRouter := New(ds, nil, plsSvc, nil, tests.NewMockLibraryService(), tests.NewMockUserService(), nil, nil, nil) + nativeRouter := New(ds, nil, plsSvc, nil, tests.NewMockLibraryService(), tests.NewMockUserService(), nil, nil, nil, nil) router = server.JWTVerifier(nativeRouter) w = httptest.NewRecorder() }) diff --git a/server/nativeapi/plugin_test.go b/server/nativeapi/plugin_test.go index aa91a7951..1683885e7 100644 --- a/server/nativeapi/plugin_test.go +++ b/server/nativeapi/plugin_test.go @@ -34,7 +34,7 @@ var _ = Describe("Plugin API", func() { ds = &tests.MockDataStore{} mockManager = &tests.MockPluginManager{} auth.Init(ds) - nativeRouter := New(ds, nil, nil, nil, tests.NewMockLibraryService(), tests.NewMockUserService(), nil, mockManager, nil) + nativeRouter := New(ds, nil, nil, nil, tests.NewMockLibraryService(), tests.NewMockUserService(), nil, mockManager, nil, nil) router = server.JWTVerifier(nativeRouter) // Create test users diff --git a/server/nativeapi/radio_nowplaying_test.go b/server/nativeapi/radio_nowplaying_test.go new file mode 100644 index 000000000..e3b2afd97 --- /dev/null +++ b/server/nativeapi/radio_nowplaying_test.go @@ -0,0 +1,141 @@ +package nativeapi + +import ( + "context" + "net/http" + "net/http/httptest" + "time" + + "github.com/navidrome/navidrome/conf" + "github.com/navidrome/navidrome/conf/configtest" + "github.com/navidrome/navidrome/consts" + "github.com/navidrome/navidrome/core/auth" + coreradio "github.com/navidrome/navidrome/core/radio" + "github.com/navidrome/navidrome/model" + "github.com/navidrome/navidrome/model/request" + "github.com/navidrome/navidrome/server" + "github.com/navidrome/navidrome/tests" + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +var _ = Describe("Radio now-playing metadata endpoints", func() { + var ( + router http.Handler + ds *tests.MockDataStore + radioRepo *tests.MockedRadioRepo + userRepo *tests.MockedUserRepo + manager *fakeRadioMetadataManager + testUser model.User + authRequest func(method, path string) *http.Request + ) + + BeforeEach(func() { + DeferCleanup(configtest.SetupConfig()) + conf.Server.SessionTimeout = time.Minute + conf.Server.EnableSharing = false + + radioRepo = &tests.MockedRadioRepo{ + Data: map[string]*model.Radio{ + "rd-1": { + ID: "rd-1", + Name: "Test Radio", + StreamUrl: "https://stream.example.test/radio", + }, + }, + } + userRepo = tests.CreateMockUserRepo() + ds = &tests.MockDataStore{ + MockedRadio: radioRepo, + MockedUser: userRepo, + MockedProperty: &tests.MockedPropertyRepo{}, + } + auth.Init(ds) + + testUser = model.User{ + ID: "user-1", + UserName: "testuser", + Name: "Test User", + IsAdmin: false, + NewPassword: "testpass", + } + Expect(userRepo.Put(&testUser)).To(Succeed()) + + manager = &fakeRadioMetadataManager{} + nativeRouter := New(ds, nil, nil, nil, tests.NewMockLibraryService(), tests.NewMockUserService(), nil, nil, nil, manager) + router = server.JWTVerifier(nativeRouter) + + authRequest = func(method, path string) *http.Request { + req := httptest.NewRequest(method, path, nil) + token, err := auth.CreateToken(&testUser) + Expect(err).ToNot(HaveOccurred()) + req.Header.Set(consts.UIAuthorizationHeader, "Bearer "+token) + return req.WithContext(request.WithClientUniqueId(req.Context(), "client-1")) + } + }) + + It("starts metadata reading for a stored radio stream URL", func() { + w := httptest.NewRecorder() + req := authRequest(http.MethodPost, "/radio/rd-1/nowplaying") + + router.ServeHTTP(w, req) + + Expect(w.Code).To(Equal(http.StatusNoContent)) + Expect(manager.starts).To(Equal([]metadataStart{{ + sessionID: "user-1:client-1", + station: coreradio.Station{ + ID: "rd-1", + StreamURL: "https://stream.example.test/radio", + }, + }})) + }) + + It("returns not found when starting metadata for a missing radio", func() { + w := httptest.NewRecorder() + req := authRequest(http.MethodPost, "/radio/missing/nowplaying") + + router.ServeHTTP(w, req) + + Expect(w.Code).To(Equal(http.StatusNotFound)) + Expect(manager.starts).To(BeEmpty()) + }) + + It("returns unauthorized when request is not authenticated", func() { + w := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodPost, "/radio/rd-1/nowplaying", nil) + + router.ServeHTTP(w, req) + + Expect(w.Code).To(Equal(http.StatusUnauthorized)) + Expect(manager.starts).To(BeEmpty()) + }) + + It("stops metadata reading idempotently for the active session", func() { + w := httptest.NewRecorder() + req := authRequest(http.MethodDelete, "/radio/rd-1/nowplaying") + + router.ServeHTTP(w, req) + + Expect(w.Code).To(Equal(http.StatusNoContent)) + Expect(manager.stops).To(Equal([]string{"user-1:client-1"})) + }) +}) + +type metadataStart struct { + sessionID string + station coreradio.Station +} + +type fakeRadioMetadataManager struct { + starts []metadataStart + stops []string +} + +func (m *fakeRadioMetadataManager) Start(_ context.Context, sessionID string, station coreradio.Station) error { + m.starts = append(m.starts, metadataStart{sessionID: sessionID, station: station}) + return nil +} + +func (m *fakeRadioMetadataManager) Stop(sessionID string) { + m.stops = append(m.stops, sessionID) +} diff --git a/server/nativeapi/radios.go b/server/nativeapi/radios.go index 701c6c926..0b7bd5d3a 100644 --- a/server/nativeapi/radios.go +++ b/server/nativeapi/radios.go @@ -9,7 +9,9 @@ import ( "github.com/deluan/rest" "github.com/go-chi/chi/v5" "github.com/navidrome/navidrome/consts" + coreradio "github.com/navidrome/navidrome/core/radio" "github.com/navidrome/navidrome/model" + "github.com/navidrome/navidrome/model/request" "github.com/navidrome/navidrome/server" ) @@ -27,10 +29,74 @@ func (api *Router) addRadioRoute(r chi.Router) { r.Delete("/", rest.Delete(constructor)) r.Post("/image", api.uploadRadioImage()) r.Delete("/image", api.deleteRadioImage()) + r.Post("/nowplaying", api.startRadioNowPlaying()) + r.Delete("/nowplaying", api.stopRadioNowPlaying()) }) }) } +func (api *Router) startRadioNowPlaying() http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + if api.radioMetadata == nil { + http.Error(w, "radio metadata is unavailable", http.StatusServiceUnavailable) + return + } + + radioID := chi.URLParam(r, "id") + radio, err := api.ds.Radio(r.Context()).Get(radioID) + if err != nil { + if errors.Is(err, model.ErrNotFound) { + http.Error(w, "not found", http.StatusNotFound) + return + } + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + + err = api.radioMetadata.Start(r.Context(), radioNowPlayingSessionID(r.Context()), coreradio.Station{ + ID: radio.ID, + StreamURL: radio.StreamUrl, + }) + if err != nil { + if errors.Is(err, coreradio.ErrInvalidSession) || errors.Is(err, coreradio.ErrInvalidStation) { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + + w.WriteHeader(http.StatusNoContent) + } +} + +func (api *Router) stopRadioNowPlaying() http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + if api.radioMetadata == nil { + http.Error(w, "radio metadata is unavailable", http.StatusServiceUnavailable) + return + } + + api.radioMetadata.Stop(radioNowPlayingSessionID(r.Context())) + w.WriteHeader(http.StatusNoContent) + } +} + +func radioNowPlayingSessionID(ctx context.Context) string { + user, hasUser := request.UserFrom(ctx) + clientID, _ := request.ClientUniqueIdFrom(ctx) + if clientID == "" { + clientID = "default" + } + if hasUser && user.ID != "" { + return user.ID + ":" + clientID + } + if username, ok := request.UsernameFrom(ctx); ok && username != "" { + return username + ":" + clientID + } + return clientID +} + func (api *Router) uploadRadioImage() http.HandlerFunc { return handleImageUpload(func(ctx context.Context, reader io.Reader, ext string) error { radioID := chi.URLParamFromCtx(ctx, "id") diff --git a/ui/src/actions/serverEvents.js b/ui/src/actions/serverEvents.js index 892f2e27e..32751bbbc 100644 --- a/ui/src/actions/serverEvents.js +++ b/ui/src/actions/serverEvents.js @@ -3,6 +3,7 @@ export const EVENT_SERVER_START = 'serverStart' export const EVENT_REFRESH_RESOURCE = 'refreshResource' export const EVENT_NOW_PLAYING_COUNT = 'nowPlayingCount' export const EVENT_NOW_PLAYING_COUNT_SYNC = 'nowPlayingCountSync' +export const EVENT_RADIO_NOW_PLAYING = 'radioNowPlaying' export const EVENT_STREAM_RECONNECTED = 'streamReconnected' export const processEvent = (type, data) => ({ diff --git a/ui/src/audioplayer/AudioTitle.jsx b/ui/src/audioplayer/AudioTitle.jsx index df37edfbb..90f2583a9 100644 --- a/ui/src/audioplayer/AudioTitle.jsx +++ b/ui/src/audioplayer/AudioTitle.jsx @@ -1,6 +1,7 @@ import React from 'react' import { useMediaQuery } from '@material-ui/core' import { Link } from 'react-router-dom' +import { useSelector } from 'react-redux' import clsx from 'clsx' import { QualityInfo } from '../common' import { decisionService } from '../transcode' @@ -23,6 +24,26 @@ const AudioTitle = React.memo(({ audioInfo, gainInfo, isMobile }) => { [song], ) + // The react-jinke-music-player library keeps its own internal snapshot of + // the queue item (audioInfo), copied when playback starts. Redux updates + // to state.player.current never reach that snapshot, so we read the live + // radio title from redux directly rather than relying on audioInfo alone. + const liveRadioTitle = useSelector((state) => { + if (!audioInfo.isRadio) { + return undefined + } + if ( + state.player?.current?.trackId === audioInfo.trackId && + state.player.current.radioTitle + ) { + return state.player.current.radioTitle + } + const queued = state.player.queue?.find( + (item) => item.isRadio && item.trackId === audioInfo.trackId, + ) + return queued?.radioTitle + }) + if (!song) { return '' } @@ -44,8 +65,19 @@ const AudioTitle = React.memo(({ audioInfo, gainInfo, isMobile }) => { } : {} + const stationTitle = song.title const subtitle = song.tags?.['subtitle'] - const title = song.title + (subtitle ? ` (${subtitle})` : '') + // Keep the configured station name until ICY delivers a StreamTitle. Stations that + // already embed the current song in the stream metadata show up via radioTitle; + // others stay on stationTitle until the backend reader picks up ICY. + const icyTitle = ( + liveRadioTitle || + audioInfo.radioTitle || + song.radioTitle || + '' + ).trim() + const displayTitle = icyTitle || stationTitle + const title = displayTitle + (subtitle ? ` (${subtitle})` : '') const linkTo = audioInfo.isRadio ? `/radio/${audioInfo.trackId}/show` diff --git a/ui/src/audioplayer/AudioTitle.test.jsx b/ui/src/audioplayer/AudioTitle.test.jsx index 7b297c07e..f2a4b574f 100644 --- a/ui/src/audioplayer/AudioTitle.test.jsx +++ b/ui/src/audioplayer/AudioTitle.test.jsx @@ -1,6 +1,8 @@ import React from 'react' import { render, screen } from '@testing-library/react' import { describe, it, expect, beforeEach, vi } from 'vitest' +import { Provider } from 'react-redux' +import { createStore } from 'redux' import AudioTitle from './AudioTitle' vi.mock('@material-ui/core', async () => { @@ -24,6 +26,11 @@ vi.mock('react-dnd', () => ({ useDrag: vi.fn(() => [null, () => {}]), })) +const renderWithStore = (ui, playerState = {}) => { + const store = createStore(() => ({ player: playerState })) + return render({ui}) +} + describe('', () => { const baseSong = { id: 'song-1', @@ -41,7 +48,9 @@ describe('', () => { it('links to playlist when playlistId is provided', () => { const audioInfo = { trackId: 'track-1', song: baseSong } - render() + renderWithStore( + , + ) const link = screen.getByRole('link') expect(link.getAttribute('href')).toBe('/playlist/playlist-1/show') }) @@ -51,8 +60,118 @@ describe('', () => { trackId: 'track-1', song: { ...baseSong, playlistId: undefined }, } - render() + renderWithStore( + , + ) const link = screen.getByRole('link') expect(link.getAttribute('href')).toBe('/album/album-1/show') }) + + it('shows live radio title without changing the station link', () => { + const audioInfo = { + trackId: 'rd-1', + isRadio: true, + radioTitle: 'Artist - Title', + song: { + id: 'rd-1', + title: 'Test Station', + artist: 'Test Station', + album: 'https://stream.example.test/radio', + }, + } + renderWithStore( + , + ) + + expect(screen.getByText('Artist - Title')).toBeInTheDocument() + expect(screen.getByRole('link').getAttribute('href')).toBe( + '/radio/rd-1/show', + ) + }) + + it('prefers the live redux title over the stale player-library snapshot', () => { + const audioInfo = { + trackId: 'radio-1', + isRadio: true, + // Note: no radioTitle here — this simulates react-jinke-music-player's + // stale internal snapshot, which never receives redux updates. + song: { + id: 'radio-1', + title: 'Station Name', + artist: 'Station Name', + album: 'https://stream.example.test/radio', + }, + } + renderWithStore( + , + { + current: { + isRadio: true, + trackId: 'radio-1', + radioTitle: 'Live Song - Artist', + }, + }, + ) + + expect(screen.getByText('Live Song - Artist')).toBeInTheDocument() + expect(screen.getByRole('link').getAttribute('href')).toBe( + '/radio/radio-1/show', + ) + }) + + it('falls back to the station name when redux current is a different radio', () => { + const audioInfo = { + trackId: 'radio-1', + isRadio: true, + song: { + id: 'radio-1', + title: 'Station Name', + // Artist deliberately differs from title so getByText stays unambiguous + artist: 'Station Artist', + album: 'https://stream.example.test/radio', + }, + } + renderWithStore( + , + { + current: { + isRadio: true, + trackId: 'radio-2', + radioTitle: 'Some Other Live Title', + }, + }, + ) + + expect(screen.getByText('Station Name')).toBeInTheDocument() + expect(screen.queryByText('Some Other Live Title')).not.toBeInTheDocument() + }) + + it('shows live title from the queue when current is not set yet', () => { + const audioInfo = { + trackId: 'radio-1', + isRadio: true, + song: { + id: 'radio-1', + title: 'Station Name', + artist: 'Station Artist', + album: 'https://stream.example.test/radio', + }, + } + renderWithStore( + , + { + current: {}, + queue: [ + { + trackId: 'radio-1', + isRadio: true, + radioTitle: 'Queued Live Title', + song: audioInfo.song, + }, + ], + }, + ) + + expect(screen.getByText('Queued Live Title')).toBeInTheDocument() + }) }) diff --git a/ui/src/audioplayer/Player.jsx b/ui/src/audioplayer/Player.jsx index c3d795b26..91d7524a1 100644 --- a/ui/src/audioplayer/Player.jsx +++ b/ui/src/audioplayer/Player.jsx @@ -34,6 +34,22 @@ import { keyMap } from '../hotkeys' import keyHandlers from './keyHandlers' import { calculateGain } from '../utils/calculateReplayGain' import { detectBrowserProfile, decisionService } from '../transcode' +import { httpVoid } from '../dataProvider' +import { REST_URL } from '../consts' + +const radioNowPlayingURL = (radioId) => + `${REST_URL}/radio/${encodeURIComponent(radioId)}/nowplaying` + +const requestRadioNowPlaying = (radioId, method) => { + if (!radioId) { + return Promise.resolve() + } + return httpVoid(radioNowPlayingURL(radioId), { method }).catch(() => {}) +} + +// Server-side now-playing sessions expire after an idle TTL; re-POST while a +// radio session is active so long listening sessions keep their live titles. +const radioSessionRefreshMs = 4 * 60 * 1000 const Player = () => { const theme = useCurrentTheme() @@ -46,6 +62,8 @@ const Player = () => { const [heartbeatTrackId, setHeartbeatTrackId] = useState(null) const lastPositionMsRef = useRef(0) const currentTrackIdRef = useRef(null) + const currentRadioIdRef = useRef(null) + const [activeRadioId, setActiveRadioId] = useState(null) const stoppedRef = useRef(false) const [audioInstance, setAudioInstance] = useState(null) const isDesktop = useMediaQuery('(min-width:810px)') @@ -63,6 +81,41 @@ const Player = () => { currentTrackIdRef.current = currentTrackId + const stopRadioNowPlaying = useCallback( + (radioId = currentRadioIdRef.current) => { + if (!radioId) { + return + } + requestRadioNowPlaying(radioId, 'DELETE') + if (currentRadioIdRef.current === radioId) { + currentRadioIdRef.current = null + setActiveRadioId(null) + } + }, + [], + ) + + const startRadioNowPlaying = useCallback((radioId) => { + if (!radioId || currentRadioIdRef.current === radioId) { + return + } + if (currentRadioIdRef.current) { + requestRadioNowPlaying(currentRadioIdRef.current, 'DELETE') + } + currentRadioIdRef.current = radioId + setActiveRadioId(radioId) + requestRadioNowPlaying(radioId, 'POST') + }, []) + + useInterval( + () => { + if (currentRadioIdRef.current) { + requestRadioNowPlaying(currentRadioIdRef.current, 'POST') + } + }, + activeRadioId ? radioSessionRefreshMs : null, + ) + useInterval( () => { if (heartbeatTrackId && !stoppedRef.current) { @@ -180,6 +233,10 @@ const Player = () => { } const handlePageHide = () => { + if (playerState.current?.isRadio) { + stopRadioNowPlaying(playerState.current.trackId) + return + } if (currentTrackIdRef.current && !playerState.current?.isRadio) { stoppedRef.current = true try { @@ -200,7 +257,7 @@ const Player = () => { window.removeEventListener('beforeunload', handleBeforeUnload) window.removeEventListener('pagehide', handlePageHide) } - }, [playerState, audioInstance]) + }, [playerState, audioInstance, stopRadioNowPlaying]) const defaultOptions = useMemo( () => ({ @@ -285,6 +342,15 @@ const Player = () => { } dispatch(currentPlaying(info)) + if (info.isRadio) { + startRadioNowPlaying(info.trackId) + setHeartbeatTrackId(null) + setCurrentTrackId(null) + return + } + if (currentRadioIdRef.current) { + stopRadioNowPlaying() + } if (info.duration) { const song = info.song document.title = `${song.title} - ${song.artist} - Navidrome` @@ -320,24 +386,44 @@ const Player = () => { } } }, - [context, dispatch, showNotifications, currentTrackId], + [ + context, + dispatch, + showNotifications, + currentTrackId, + startRadioNowPlaying, + stopRadioNowPlaying, + ], ) - const onAudioPlayTrackChange = useCallback(() => { - if (currentTrackId) { - subsonic.reportPlayback( - currentTrackId, - lastPositionMsRef.current, - 'stopped', - ) - } - setHeartbeatTrackId(null) - setCurrentTrackId(null) - }, [currentTrackId]) + const onAudioPlayTrackChange = useCallback( + (playId, audioLists, audioInfo) => { + // Switching to another radio is handled in onAudioPlay; avoid stopping the + // session here or the ICY reader is torn down before playback resumes. + if (!audioInfo?.isRadio && currentRadioIdRef.current) { + stopRadioNowPlaying() + } + if (currentTrackId) { + subsonic.reportPlayback( + currentTrackId, + lastPositionMsRef.current, + 'stopped', + ) + } + setHeartbeatTrackId(null) + setCurrentTrackId(null) + }, + [currentTrackId, stopRadioNowPlaying], + ) const onAudioPause = useCallback( (info) => { dispatch(currentPlaying(info)) + if (info.isRadio) { + stopRadioNowPlaying(info.trackId) + setHeartbeatTrackId(null) + return + } if (!info.isRadio && currentTrackId) { const posMs = Math.floor(info.currentTime * 1000) lastPositionMsRef.current = posMs @@ -345,11 +431,14 @@ const Player = () => { } setHeartbeatTrackId(null) }, - [dispatch, currentTrackId], + [dispatch, currentTrackId, stopRadioNowPlaying], ) const onAudioEnded = useCallback( (currentPlayId, audioLists, info) => { + if (info.isRadio) { + stopRadioNowPlaying(info.trackId) + } if (currentTrackId && !info.isRadio) { const posMs = Math.floor((info.duration || 0) * 1000) subsonic.reportPlayback(currentTrackId, posMs, 'stopped') @@ -362,7 +451,7 @@ const Player = () => { // eslint-disable-next-line no-console .catch((e) => console.log('Keepalive error:', e)) }, - [dispatch, dataProvider, currentTrackId], + [dispatch, dataProvider, currentTrackId, stopRadioNowPlaying], ) const onCoverClick = useCallback((mode, audioLists, audioInfo) => { @@ -395,6 +484,9 @@ const Player = () => { const onBeforeDestroy = useCallback(() => { return new Promise((resolve, reject) => { + if (currentRadioIdRef.current) { + stopRadioNowPlaying() + } if (currentTrackId && !playerStateRef.current?.current?.isRadio) { subsonic.reportPlayback( currentTrackId, @@ -407,7 +499,7 @@ const Player = () => { dispatch(clearQueue()) reject() }) - }, [dispatch, currentTrackId]) + }, [dispatch, currentTrackId, stopRadioNowPlaying]) if (!visible) { document.title = 'Navidrome' diff --git a/ui/src/dataProvider/httpClient.js b/ui/src/dataProvider/httpClient.js index 706486fa6..93cd7d8da 100644 --- a/ui/src/dataProvider/httpClient.js +++ b/ui/src/dataProvider/httpClient.js @@ -4,25 +4,54 @@ import { baseUrl } from '../utils' import config from '../config' import { jwtDecode } from 'jwt-decode' -const customAuthorizationHeader = 'X-ND-Authorization' +export const customAuthorizationHeader = 'X-ND-Authorization' export const clientUniqueIdHeader = 'X-ND-Client-Unique-Id' export const clientUniqueId = uuidv4() +const applyAuthHeaders = (headers) => { + headers.set(clientUniqueIdHeader, clientUniqueId) + const token = localStorage.getItem('token') + if (token) { + headers.set(customAuthorizationHeader, `Bearer ${token}`) + } + return headers +} + +// Endpoints that return 204 No Content (e.g. radio now-playing) must not use fetchJson. +export const httpVoid = (url, options = {}) => { + url = baseUrl(url) + const headers = applyAuthHeaders( + options.headers instanceof Headers + ? options.headers + : new Headers({ Accept: 'application/json', ...options.headers }), + ) + return fetch(url, { ...options, headers }).then((response) => { + const refreshedToken = response.headers.get(customAuthorizationHeader) + if (refreshedToken) { + const decoded = jwtDecode(refreshedToken) + localStorage.setItem('token', refreshedToken) + localStorage.setItem('userId', decoded.uid) + config.firstTime = false + removeHomeCache() + } + if (!response.ok) { + throw new Error(`Request failed with status ${response.status}`) + } + return response + }) +} + const httpClient = (url, options = {}) => { url = baseUrl(url) if (!options.headers) { options.headers = new Headers({ Accept: 'application/json' }) } - options.headers.set(clientUniqueIdHeader, clientUniqueId) - const token = localStorage.getItem('token') - if (token) { - options.headers.set(customAuthorizationHeader, `Bearer ${token}`) - } + applyAuthHeaders(options.headers) return fetchUtils.fetchJson(url, options).then((response) => { - const token = response.headers.get(customAuthorizationHeader) - if (token) { - const decoded = jwtDecode(token) - localStorage.setItem('token', token) + const refreshedToken = response.headers.get(customAuthorizationHeader) + if (refreshedToken) { + const decoded = jwtDecode(refreshedToken) + localStorage.setItem('token', refreshedToken) localStorage.setItem('userId', decoded.uid) // Avoid going to create admin dialog after logout/login without a refresh config.firstTime = false diff --git a/ui/src/dataProvider/index.js b/ui/src/dataProvider/index.js index 059023378..208484239 100644 --- a/ui/src/dataProvider/index.js +++ b/ui/src/dataProvider/index.js @@ -1,6 +1,10 @@ -import httpClient, { clientUniqueId, clientUniqueIdHeader } from './httpClient' +import httpClient, { + clientUniqueId, + clientUniqueIdHeader, + httpVoid, +} from './httpClient' import wrapperDataProvider from './wrapperDataProvider' -export { httpClient, clientUniqueId, clientUniqueIdHeader } +export { httpClient, httpVoid, clientUniqueId, clientUniqueIdHeader } export default wrapperDataProvider diff --git a/ui/src/eventStream.js b/ui/src/eventStream.js index c91dae875..21a3c1f02 100644 --- a/ui/src/eventStream.js +++ b/ui/src/eventStream.js @@ -1,6 +1,11 @@ import { baseUrl } from './utils' import throttle from 'lodash.throttle' -import { processEvent, serverDown, streamReconnected } from './actions' +import { + EVENT_RADIO_NOW_PLAYING, + processEvent, + serverDown, + streamReconnected, +} from './actions' import { REST_URL } from './consts' import config from './config' @@ -20,6 +25,7 @@ const setupHandlers = (stream, dispatchFn) => { stream.addEventListener('serverStart', eventHandler(dispatchFn)) stream.addEventListener('scanStatus', throttledEventHandler(dispatchFn)) stream.addEventListener('refreshResource', eventHandler(dispatchFn)) + stream.addEventListener(EVENT_RADIO_NOW_PLAYING, eventHandler(dispatchFn)) if (config.enableNowPlaying) { stream.addEventListener('nowPlayingCount', eventHandler(dispatchFn)) } @@ -76,6 +82,10 @@ const startEventStreamLegacy = async (dispatchFn) => { throttledEventHandler(dispatchFn), ) newStream.addEventListener('refreshResource', eventHandler(dispatchFn)) + newStream.addEventListener( + EVENT_RADIO_NOW_PLAYING, + eventHandler(dispatchFn), + ) if (config.enableNowPlaying) { newStream.addEventListener('nowPlayingCount', eventHandler(dispatchFn)) } diff --git a/ui/src/eventStream.test.js b/ui/src/eventStream.test.js index 27f53c872..7df399a20 100644 --- a/ui/src/eventStream.test.js +++ b/ui/src/eventStream.test.js @@ -1,6 +1,6 @@ import { describe, it, beforeEach, vi, expect } from 'vitest' import { startEventStream } from './eventStream' -import { serverDown } from './actions' +import { processEvent, serverDown } from './actions' import config from './config' class MockEventSource { @@ -48,4 +48,24 @@ describe('startEventStream', () => { vi.advanceTimersByTime(5000) expect(global.EventSource).toHaveBeenCalledTimes(2) }) + + it('dispatches radio now-playing events', async () => { + await startEventStream(dispatch) + instance.listeners.radioNowPlaying({ + type: 'radioNowPlaying', + data: JSON.stringify({ + radioId: 'rd-1', + title: 'Artist - Title', + updatedAt: '2026-06-30T12:34:56Z', + }), + }) + + expect(dispatch).toHaveBeenCalledWith( + processEvent('radioNowPlaying', { + radioId: 'rd-1', + title: 'Artist - Title', + updatedAt: '2026-06-30T12:34:56Z', + }), + ) + }) }) diff --git a/ui/src/reducers/playerReducer.js b/ui/src/reducers/playerReducer.js index d6ab7484b..2fd142ccb 100644 --- a/ui/src/reducers/playerReducer.js +++ b/ui/src/reducers/playerReducer.js @@ -12,6 +12,7 @@ import { PLAYER_SYNC_QUEUE, PLAYER_SET_MODE, PLAYER_REFRESH_QUEUE, + EVENT_RADIO_NOW_PLAYING, } from '../actions' import config from '../config' @@ -182,10 +183,28 @@ const reduceSyncQueue = (state, { data: { audioInfo, audioLists } }) => { } const reduceCurrent = (state, { data }) => { - const current = data.ended ? {} : data + let current = data.ended ? {} : data const savedPlayIndex = state.queue.findIndex( (item) => item.uuid === current.uuid, ) + // Preserve live radio titles when the music player snapshot omits them. + if (current.isRadio && current.trackId) { + const queued = state.queue.find( + (item) => + (current.uuid && item.uuid === current.uuid) || + item.trackId === current.trackId, + ) + // Only fall back to the previous current item when it is the same track, + // otherwise switching stations would inherit the old station's title. + const previousTitle = + state.current?.trackId === current.trackId + ? state.current?.radioTitle + : undefined + const liveTitle = queued?.radioTitle ?? previousTitle + if (liveTitle && !current.radioTitle) { + current = applyRadioTitle(current, liveTitle) + } + } // When a track selection is pending (playIndex is set), keep it alive // until the music player confirms it actually switched to the requested // track. Without this, a premature onAudioPlay callback for the @@ -208,6 +227,55 @@ const reduceMode = (state, { data: { mode } }) => { } } +const activeQueueItem = (state) => { + if (state.current?.isRadio) { + return state.current + } + const index = state.playIndex ?? state.savedPlayIndex + return state.queue[index] +} + +const applyRadioTitle = (item, title) => { + if (!item) { + return item + } + const song = { ...(item.song || {}) } + if (title) { + return { + ...item, + radioTitle: title, + song: { ...song, radioTitle: title }, + } + } + delete song.radioTitle + const next = { ...item, song } + delete next.radioTitle + return next +} + +const reduceRadioNowPlaying = (state, { data }) => { + const radioId = data?.radioId + const active = activeQueueItem(state) + if (!radioId || !active?.isRadio || active.trackId !== radioId) { + return state + } + + const title = (data.title || '').trim() + const matchesActive = (item) => + active.uuid ? item.uuid === active.uuid : item.trackId === radioId + + return { + ...state, + queue: state.queue.map((item) => + matchesActive(item) ? applyRadioTitle(item, title) : item, + ), + current: + state.current?.isRadio && state.current.trackId === radioId + ? applyRadioTitle(state.current, title) + : state.current, + } +} + export const playerReducer = (previousState = initialState, payload) => { const { type } = payload switch (type) { @@ -229,6 +297,8 @@ export const playerReducer = (previousState = initialState, payload) => { return reduceCurrent(previousState, payload) case PLAYER_SET_MODE: return reduceMode(previousState, payload) + case EVENT_RADIO_NOW_PLAYING: + return reduceRadioNowPlaying(previousState, payload) case PLAYER_REFRESH_QUEUE: { const resolvedUrls = payload.data || {} return { diff --git a/ui/src/reducers/playerReducer.test.js b/ui/src/reducers/playerReducer.test.js index 110ce8c53..1cda4be9e 100644 --- a/ui/src/reducers/playerReducer.test.js +++ b/ui/src/reducers/playerReducer.test.js @@ -4,6 +4,7 @@ import { PLAYER_SYNC_QUEUE, PLAYER_CURRENT, PLAYER_REFRESH_QUEUE, + EVENT_RADIO_NOW_PLAYING, } from '../actions' describe('playerReducer', () => { @@ -224,4 +225,161 @@ describe('playerReducer', () => { expect(result.playIndex).toBe(0) }) }) + + describe('radioNowPlaying event', () => { + const radioItem = { + trackId: 'rd-1', + uuid: 'radio-uuid', + name: 'Test Station', + musicSrc: 'https://stream.example.test/radio', + isRadio: true, + song: { + id: 'rd-1', + title: 'Test Station', + artist: 'Test Station', + album: 'https://stream.example.test/radio', + streamUrl: 'https://stream.example.test/radio', + }, + } + + it('updates the active radio with live title metadata', () => { + const state = { + queue: [radioItem], + current: radioItem, + savedPlayIndex: 0, + clear: false, + volume: 1, + } + const result = playerReducer(state, { + type: EVENT_RADIO_NOW_PLAYING, + data: { + radioId: 'rd-1', + title: 'Artist - Title', + updatedAt: '2026-06-30T12:34:56Z', + }, + }) + + expect(result.current.radioTitle).toBe('Artist - Title') + expect(result.current.song.radioTitle).toBe('Artist - Title') + expect(result.current.name).toBe('Test Station') + expect(result.current.song.title).toBe('Test Station') + expect(result.current.musicSrc).toBe('https://stream.example.test/radio') + expect(result.queue[0].radioTitle).toBe('Artist - Title') + }) + + it('ignores events for non-active radios', () => { + const state = { + queue: [radioItem], + current: radioItem, + savedPlayIndex: 0, + clear: false, + volume: 1, + } + const result = playerReducer(state, { + type: EVENT_RADIO_NOW_PLAYING, + data: { radioId: 'rd-2', title: 'Other Station - Title' }, + }) + + expect(result).toBe(state) + }) + + it('clears live title metadata when the title is empty', () => { + const liveRadio = { + ...radioItem, + radioTitle: 'Old Title', + song: { ...radioItem.song, radioTitle: 'Old Title' }, + } + const state = { + queue: [liveRadio], + current: liveRadio, + savedPlayIndex: 0, + clear: false, + volume: 1, + } + const result = playerReducer(state, { + type: EVENT_RADIO_NOW_PLAYING, + data: { radioId: 'rd-1', title: '' }, + }) + + expect(result.current.radioTitle).toBeUndefined() + expect(result.current.song.radioTitle).toBeUndefined() + expect(result.current.song.title).toBe('Test Station') + expect(result.current.musicSrc).toBe('https://stream.example.test/radio') + }) + + it('updates the queued active radio before current is set', () => { + const state = { + queue: [radioItem], + current: {}, + savedPlayIndex: 0, + clear: false, + volume: 1, + } + const result = playerReducer(state, { + type: EVENT_RADIO_NOW_PLAYING, + data: { radioId: 'rd-1', title: 'Artist - Title' }, + }) + + expect(result.current).toEqual({}) + expect(result.queue[0].radioTitle).toBe('Artist - Title') + }) + + it('preserves live radio title when the player snapshot omits it', () => { + const liveRadio = { + ...radioItem, + radioTitle: 'Artist - Title', + song: { ...radioItem.song, radioTitle: 'Artist - Title' }, + } + const state = { + queue: [liveRadio], + current: {}, + savedPlayIndex: 0, + clear: false, + volume: 1, + } + const result = playerReducer(state, { + type: PLAYER_CURRENT, + data: { ...radioItem }, + }) + + expect(result.current.radioTitle).toBe('Artist - Title') + expect(result.current.song.radioTitle).toBe('Artist - Title') + }) + + it('does not inherit the previous station title when switching stations', () => { + const previousStation = { + ...radioItem, + radioTitle: 'Old Station Song', + song: { ...radioItem.song, radioTitle: 'Old Station Song' }, + } + const nextStation = { + trackId: 'rd-2', + uuid: 'radio-uuid-2', + name: 'Other Station', + musicSrc: 'https://stream.example.test/other', + isRadio: true, + song: { + id: 'rd-2', + title: 'Other Station', + artist: 'Other Station', + album: 'https://stream.example.test/other', + streamUrl: 'https://stream.example.test/other', + }, + } + const state = { + queue: [previousStation, nextStation], + current: previousStation, + savedPlayIndex: 0, + clear: false, + volume: 1, + } + const result = playerReducer(state, { + type: PLAYER_CURRENT, + data: { ...nextStation }, + }) + + expect(result.current.radioTitle).toBeUndefined() + expect(result.current.song.title).toBe('Other Station') + }) + }) })