fm-rds-tx/internal/ingest/adapters/icecast/source.go

package icecast

import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/jan/fm-rds-tx/internal/ingest"
	"github.com/jan/fm-rds-tx/internal/ingest/decoder"
	"github.com/jan/fm-rds-tx/internal/ingest/decoder/aac"
	"github.com/jan/fm-rds-tx/internal/ingest/decoder/fallback"
	"github.com/jan/fm-rds-tx/internal/ingest/decoder/mp3"
	"github.com/jan/fm-rds-tx/internal/ingest/decoder/oggvorbis"
)

type Source struct {
	id  string
	url string

	client *http.Client
	decReg *decoder.Registry
	reconn ReconnectConfig

	decoderPreference string

	chunks chan ingest.PCMChunk
	errs   chan error
	title  chan string

	cancel context.CancelFunc
	wg     sync.WaitGroup

	state           atomic.Value // string
	connected       atomic.Bool
	chunksIn        atomic.Uint64
	samplesIn       atomic.Uint64
	reconnects      atomic.Uint64
	discontinuities atomic.Uint64
	lastChunkAtUnix atomic.Int64
	lastMetaAtUnix  atomic.Int64
	metadataUpdates atomic.Uint64
	icyMetaInt      atomic.Int64
	lastError       atomic.Value // string
	streamTitle     atomic.Value // string
}

var errStreamEnded = errors.New("icecast stream ended")

type Option func(*Source)

func WithDecoderPreference(pref string) Option {
	return func(s *Source) {
		s.decoderPreference = normalizeDecoderPreference(pref)
	}
}

func WithDecoderRegistry(reg *decoder.Registry) Option {
	return func(s *Source) {
		if reg != nil {
			s.decReg = reg
		}
	}
}

func New(id, url string, client *http.Client, reconn ReconnectConfig, opts ...Option) *Source {
	if id == "" {
		id = "icecast-main"
	}
	if client == nil {
		// Streaming responses are long-lived; a global client timeout would
		// terminate the body read after a fixed duration.
		client = &http.Client{}
	}
	s := &Source{
		id:                id,
		url:               strings.TrimSpace(url),
		client:            client,
		reconn:            reconn,
		chunks:            make(chan ingest.PCMChunk, 64),
		errs:              make(chan error, 8),
		title:             make(chan string, 16),
		decReg:            defaultRegistry(),
		decoderPreference: "auto",
	}
	for _, opt := range opts {
		if opt != nil {
			opt(s)
		}
	}
	s.decoderPreference = normalizeDecoderPreference(s.decoderPreference)
	s.state.Store("idle")
	s.streamTitle.Store("")
	return s
}

func defaultRegistry() *decoder.Registry {
	r := decoder.NewRegistry()
	r.Register("mp3", func() decoder.Decoder { return mp3.New() })
	r.Register("oggvorbis", func() decoder.Decoder { return oggvorbis.New() })
	r.Register("aac", func() decoder.Decoder { return aac.New() })
	r.Register("ffmpeg", func() decoder.Decoder { return fallback.NewFFmpeg() })
	return r
}

func (s *Source) Descriptor() ingest.SourceDescriptor {
	return ingest.SourceDescriptor{
		ID:        s.id,
		Kind:      "icecast",
		Family:    "streaming",
		Transport: "http",
		Codec:     s.decoderPreference,
		Detail:    s.url,
		Origin: &ingest.SourceOrigin{
			Kind:     "url",
			Endpoint: redactURL(s.url),
		},
	}
}

func (s *Source) Start(ctx context.Context) error {
	if s.url == "" {
		return fmt.Errorf("icecast url is required")
	}
	// BUG-2 fix: recreate channels on every Start() so that Stop+Start works.
	// loop() closes chunks/errs/title when it exits; reusing closed channels panics.
	s.chunks = make(chan ingest.PCMChunk, 64)
	s.errs   = make(chan error, 8)
	s.title  = make(chan string, 16)
	runCtx, cancel := context.WithCancel(ctx)
	s.cancel = cancel
	s.lastError.Store("")
	s.state.Store("connecting")
	s.wg.Add(1)
	go s.loop(runCtx)
	return nil
}

func (s *Source) Stop() error {
	if s.cancel != nil {
		s.cancel()
	}
	s.wg.Wait()
	s.state.Store("stopped")
	return nil
}

func (s *Source) Chunks() <-chan ingest.PCMChunk { return s.chunks }
func (s *Source) Errors() <-chan error           { return s.errs }
func (s *Source) StreamTitleUpdates() <-chan string {
	return s.title
}

func (s *Source) Stats() ingest.SourceStats {
	state, _ := s.state.Load().(string)
	last := s.lastChunkAtUnix.Load()
	lastMeta := s.lastMetaAtUnix.Load()
	errStr, _ := s.lastError.Load().(string)
	streamTitle, _ := s.streamTitle.Load().(string)
	var lastChunkAt time.Time
	var lastMetaAt time.Time
	if last > 0 {
		lastChunkAt = time.Unix(0, last)
	}
	if lastMeta > 0 {
		lastMetaAt = time.Unix(0, lastMeta)
	}
	return ingest.SourceStats{
		State:           state,
		Connected:       s.connected.Load(),
		LastChunkAt:     lastChunkAt,
		LastMetaAt:      lastMetaAt,
		StreamTitle:     streamTitle,
		MetadataUpdates: s.metadataUpdates.Load(),
		IcyMetaInt:      int(s.icyMetaInt.Load()),
		ChunksIn:        s.chunksIn.Load(),
		SamplesIn:       s.samplesIn.Load(),
		Reconnects:      s.reconnects.Load(),
		Discontinuities: s.discontinuities.Load(),
		LastError:       errStr,
	}
}

func (s *Source) loop(ctx context.Context) {
	defer s.wg.Done()
	defer close(s.chunks)
	defer close(s.errs)
	defer close(s.title)
	attempt := 0
	for {
		select {
		case <-ctx.Done():
			return
		default:
		}

		s.state.Store("connecting")
		err := s.connectAndRun(ctx)
		if ctx.Err() != nil {
			return
		}
		if err == nil {
			err = errStreamEnded
		}
		s.connected.Store(false)
		s.lastError.Store(err.Error())
		select {
		case s.errs <- err:
		default:
		}
		s.state.Store("reconnecting")
		attempt++
		s.reconnects.Add(1)
		backoff := s.reconn.nextBackoff(attempt)
		if backoff <= 0 {
			s.state.Store("failed")
			return
		}
		select {
		case <-time.After(backoff):
		case <-ctx.Done():
			return
		}
	}
}

func (s *Source) connectAndRun(ctx context.Context) error {
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, s.url, nil)
	if err != nil {
		return err
	}
	req.Header.Set("Icy-MetaData", "1")
	resp, err := s.client.Do(req)
	if err != nil {
		return fmt.Errorf("icecast connect: %w", err)
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		return fmt.Errorf("icecast status: %s", resp.Status)
	}
	s.connected.Store(true)
	s.state.Store("buffering")
	s.lastError.Store("")
	icyMetaInt, _ := parseICYMetaInt(resp.Header.Get("icy-metaint"))
	s.icyMetaInt.Store(int64(icyMetaInt))
	stream := newICYReader(resp.Body, icyMetaInt, s.onMetadata)
	s.state.Store("running")
	return s.decodeWithPreference(ctx, stream, decoder.StreamMeta{
		ContentType:  resp.Header.Get("Content-Type"),
		SourceID:     s.id,
		SampleRateHz: 44100,
		Channels:     2,
	})
}

func (s *Source) onMetadata(meta icyMetadata) {
	s.streamTitle.Store(meta.StreamTitle)
	s.metadataUpdates.Add(1)
	s.lastMetaAtUnix.Store(time.Now().UnixNano())
	select {
	case s.title <- meta.StreamTitle:
	default:
	}
}

func (s *Source) emitChunk(chunk ingest.PCMChunk) error {
	select {
	case s.chunks <- chunk:
	default:
		s.discontinuities.Add(1)
		return io.ErrShortBuffer
	}
	s.chunksIn.Add(1)
	s.samplesIn.Add(uint64(len(chunk.Samples)))
	s.lastChunkAtUnix.Store(time.Now().UnixNano())
	return nil
}

func (s *Source) decodeWithPreference(ctx context.Context, stream io.Reader, meta decoder.StreamMeta) error {
	mode := normalizeDecoderPreference(s.decoderPreference)
	switch mode {
	case "ffmpeg":
		return s.decodeNamed(ctx, "ffmpeg", stream, meta)
	case "native":
		native, err := s.decReg.SelectByContentType(meta.ContentType)
		if err != nil {
			return fmt.Errorf("icecast native decoder select: %w", err)
		}
		return native.DecodeStream(ctx, stream, meta, s.emitChunk)
	case "auto":
		// Phase-1 policy: try native decoder first, then fall back to ffmpeg
		// only when native selection/decode reports "unsupported".
		native, err := s.decReg.SelectByContentType(meta.ContentType)
		if err == nil {
			captured := &capturingReader{r: stream}
			if err := native.DecodeStream(ctx, captured, meta, s.emitChunk); err == nil {
				return nil
			} else if !errors.Is(err, decoder.ErrUnsupported) {
				return err
			}
			// Native decode can consume stream bytes before returning "unsupported".
			// Reconstruct a full reader for fallback: consumed prefix + remaining stream.
			stream = io.MultiReader(bytes.NewReader(captured.Bytes()), stream)
		} else if !errors.Is(err, decoder.ErrUnsupported) {
			return fmt.Errorf("icecast decoder select: %w", err)
		}
		return s.decodeNamed(ctx, "ffmpeg", stream, meta)
	default:
		return fmt.Errorf("unsupported icecast decoder mode: %s", mode)
	}
}

// maxCaptureBytes caps the amount of stream data buffered while the native
// decoder is deciding whether it can handle the format. Without a cap, a
// decoder that reads extensively before returning ErrUnsupported could grow
// this buffer unboundedly on a corrupt or adversarial stream.
const maxCaptureBytes = 1 << 20 // 1 MiB

// errCaptureLimitExceeded is returned by capturingReader when the buffer cap
// is hit. The caller should treat it like ErrUnsupported and fall back.
var errCaptureLimitExceeded = errors.New("capture buffer limit exceeded")

type capturingReader struct {
	r   io.Reader
	buf bytes.Buffer
}

func (r *capturingReader) Read(p []byte) (int, error) {
	if r.buf.Len() >= maxCaptureBytes {
		return 0, errCaptureLimitExceeded
	}
	n, err := r.r.Read(p)
	if n > 0 {
		_, _ = r.buf.Write(p[:n])
	}
	return n, err
}

func (r *capturingReader) Bytes() []byte {
	return r.buf.Bytes()
}

func (s *Source) decodeNamed(ctx context.Context, name string, stream io.Reader, meta decoder.StreamMeta) error {
	dec, err := s.decReg.Create(name)
	if err != nil {
		return fmt.Errorf("icecast decoder=%s unavailable: %w", name, err)
	}
	return dec.DecodeStream(ctx, stream, meta, s.emitChunk)
}

func normalizeDecoderPreference(pref string) string {
	switch strings.ToLower(strings.TrimSpace(pref)) {
	case "", "auto":
		return "auto"
	case "native":
		return "native"
	case "ffmpeg", "fallback":
		return "ffmpeg"
	default:
		return strings.ToLower(strings.TrimSpace(pref))
	}
}

func redactURL(raw string) string {
	trimmed := strings.TrimSpace(raw)
	if trimmed == "" {
		return ""
	}
	u, err := url.Parse(trimmed)
	if err != nil || u.Host == "" {
		return trimmed
	}
	u.User = nil
	u.RawQuery = ""
	u.Fragment = ""
	return u.String()
}