refactor: move segchanman into the bus

pkg/api/api_internal.go

···
 			errors.WriteHTTPBadRequest(w, "rendition required", nil)
 			return
 		}
-		seg := <-a.MediaManager.SubscribeSegment(ctx, user, rendition)
+		seg := <-a.Bus.SubscribeSegment(ctx, user, rendition)
 		base := filepath.Base(seg.Filepath)
 		w.Header().Set("Location", fmt.Sprintf("%s/playback/%s/%s/segment/%s\n", a.CLI.OwnInternalURL(), user, rendition, base))
 		w.WriteHeader(301)

pkg/bus/bus.go

···
 
 // Bus is a simple pub/sub system for backing websocket connections
 type Bus struct {
-	mu      sync.Mutex
-	clients map[string][]Subscription
+	mu            sync.Mutex
+	clients       map[string][]Subscription
+	segChans      map[string][]chan *Seg
+	segChansMutex sync.Mutex
 }
 
 func NewBus() *Bus {
 	return &Bus{
-		clients: make(map[string][]Subscription),
+		clients:  make(map[string][]Subscription),
+		segChans: make(map[string][]chan *Seg),
 	}
 }
 

pkg/bus/segchanman.go

···
+package bus
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"go.opentelemetry.io/otel"
+	"stream.place/streamplace/pkg/log"
+	"stream.place/streamplace/pkg/spmetrics"
+)
+
+// it's a segment channel manager, you see
+
+type Seg struct {
+	Filepath       string
+	Data           []byte
+	PacketizedData *PacketizedSegment
+}
+
+type PacketizedSegment struct {
+	Video    [][]byte
+	Audio    [][]byte
+	Duration time.Duration
+}
+
+func segChanKey(user string, rendition string) string {
+	return fmt.Sprintf("%s::%s", user, rendition)
+}
+
+func (b *Bus) SubscribeSegment(ctx context.Context, user string, rendition string) <-chan *Seg {
+	key := segChanKey(user, rendition)
+	b.segChansMutex.Lock()
+	defer b.segChansMutex.Unlock()
+	chs, ok := b.segChans[key]
+	if !ok {
+		chs = []chan *Seg{}
+		b.segChans[key] = chs
+	}
+	ch := make(chan *Seg)
+	chs = append(chs, ch)
+	b.segChans[key] = chs
+	spmetrics.SegmentSubscriptionsOpen.WithLabelValues(user, rendition).Set(float64(len(chs)))
+	return ch
+}
+
+func (b *Bus) UnsubscribeSegment(ctx context.Context, user string, rendition string, ch <-chan *Seg) {
+	key := segChanKey(user, rendition)
+	b.segChansMutex.Lock()
+	defer b.segChansMutex.Unlock()
+	chs, ok := b.segChans[key]
+	if !ok {
+		return
+	}
+	for i, c := range chs {
+		if c == ch {
+			chs = append(chs[:i], chs[i+1:]...)
+			break
+		}
+	}
+	spmetrics.SegmentSubscriptionsOpen.WithLabelValues(user, rendition).Set(float64(len(chs)))
+	b.segChans[key] = chs
+}
+
+func (b *Bus) PublishSegment(ctx context.Context, user string, rendition string, seg *Seg) {
+	ctx, span := otel.Tracer("signer").Start(ctx, "PublishSegment")
+	defer span.End()
+	key := segChanKey(user, rendition)
+	b.segChansMutex.Lock()
+	defer b.segChansMutex.Unlock()
+	chs, ok := b.segChans[key]
+	if !ok {
+		return
+	}
+	for _, ch := range chs {
+		go func(ch chan *Seg) {
+			select {
+			case ch <- seg:
+			case <-ctx.Done():
+				return
+			case <-time.After(1 * time.Minute):
+				log.Warn(ctx, "failed to send segment to channel, timing out", "user", user, "rendition", rendition)
+			}
+
+		}(ch)
+	}
+}

pkg/director/director.go

···
 					bus:         d.bus,
 					segmentChan: make(chan struct{}),
 					op:          d.op,
+					packets:     make([]bus.PacketizedSegment, 0),
 				}
 				d.streamSessions[not.Segment.RepoDID] = ss
 				g.Go(func() error {

pkg/director/stream_session.go

···
 	"stream.place/streamplace/pkg/livepeer"
 	"stream.place/streamplace/pkg/log"
 	"stream.place/streamplace/pkg/media"
-	"stream.place/streamplace/pkg/media/segchanman"
 	"stream.place/streamplace/pkg/model"
 	"stream.place/streamplace/pkg/renditions"
 	"stream.place/streamplace/pkg/spmetrics"
···
 	lastStatus     time.Time
 	lastStatusLock sync.Mutex
 	g              *errgroup.Group
+	packets        []bus.PacketizedSegment
 }
 
 func (ss *StreamSession) Start(ctx context.Context, not *media.NewSegmentNotification) error {
···
 
 	ss.bus.Publish(spseg.Creator, spseg)
 	ss.Go(ctx, func() error {
-		return ss.AddToHLS(ctx, spseg, "source", not.Data)
+		return ss.AddPlaybackSegment(ctx, spseg, "source", &bus.Seg{
+			Filepath: not.Segment.ID,
+			Data:     not.Data,
+		})
 	})
 
 	if ss.cli.Thumbnail {
···
 			return fmt.Errorf("failed to write transcoded segment file: %w", err)
 		}
 		ss.Go(ctx, func() error {
-			return ss.AddToHLS(ctx, spseg, rs[i].Name, seg)
-		})
-		ss.Go(ctx, func() error {
-			ss.mm.PublishSegment(ctx, spseg.Creator, rs[i].Name, &segchanman.Seg{
+			return ss.AddPlaybackSegment(ctx, spseg, rs[i].Name, &bus.Seg{
 				Filepath: fd.Name(),
 				Data:     seg,
 			})
-			return nil
 		})
 
 	}
+	return nil
+}
+
+func (ss *StreamSession) AddPlaybackSegment(ctx context.Context, spseg *streamplace.Segment, rendition string, seg *bus.Seg) error {
+	ss.Go(ctx, func() error {
+		return ss.AddToHLS(ctx, spseg, rendition, seg.Data)
+	})
+	ss.Go(ctx, func() error {
+		return ss.AddToWebRTC(ctx, spseg, rendition, seg)
+	})
+	return nil
+}
+
+func (ss *StreamSession) AddToWebRTC(ctx context.Context, spseg *streamplace.Segment, rendition string, seg *bus.Seg) error {
+	packet, err := media.Packetize(ctx, seg)
+	if err != nil {
+		return fmt.Errorf("failed to packetize segment: %w", err)
+	}
+	seg.PacketizedData = packet
+	ss.bus.PublishSegment(ctx, spseg.Creator, rendition, seg)
 	return nil
 }
 

pkg/media/concat.go

···
 
 	"github.com/go-gst/go-gst/gst"
 	"github.com/go-gst/go-gst/gst/app"
+	"stream.place/streamplace/pkg/bus"
 	"stream.place/streamplace/pkg/log"
-	"stream.place/streamplace/pkg/media/segchanman"
 )
 
 type ConcatStreamer interface {
-	SubscribeSegment(ctx context.Context, user string, rendition string) <-chan *segchanman.Seg
-	UnsubscribeSegment(ctx context.Context, user string, rendition string, ch <-chan *segchanman.Seg)
+	SubscribeSegment(ctx context.Context, user string, rendition string) <-chan *bus.Seg
+	UnsubscribeSegment(ctx context.Context, user string, rendition string, ch <-chan *bus.Seg)
 }
 
 // This function remains in scope for the duration of a single users' playback

pkg/media/concat2.go

···
 	"fmt"
 
 	"github.com/go-gst/go-gst/gst"
+	"stream.place/streamplace/pkg/bus"
 	"stream.place/streamplace/pkg/log"
-	"stream.place/streamplace/pkg/media/segchanman"
 )
 
 var ErrConcatDone = errors.New("concat done")
 
-func ConcatBin(ctx context.Context, segCh <-chan *segchanman.Seg) (*gst.Bin, error) {
+func ConcatBin(ctx context.Context, segCh <-chan *bus.Seg) (*gst.Bin, error) {
 	ctx = log.WithLogValues(ctx, "func", "ConcatBin")
 	bin := gst.NewBin("concat-bin")
 
···
 	return bin, nil
 }
 
-func addConcatDemuxer(ctx context.Context, bin *gst.Bin, seg *segchanman.Seg, syncPadVideoSink *gst.Pad, syncPadAudioSink *gst.Pad) error {
+func addConcatDemuxer(ctx context.Context, bin *gst.Bin, seg *bus.Seg, syncPadVideoSink *gst.Pad, syncPadAudioSink *gst.Pad) error {
 
 	log.Debug(ctx, "adding concat demuxer", "seg", seg.Filepath)
 	demuxBin, err := ConcatDemuxBin(ctx, seg)

pkg/media/concat2_test.go

···
 	"github.com/stretchr/testify/require"
 	"go.uber.org/goleak"
 	"golang.org/x/sync/errgroup"
+	"stream.place/streamplace/pkg/bus"
 	"stream.place/streamplace/pkg/gstinit"
 	"stream.place/streamplace/pkg/log"
-	"stream.place/streamplace/pkg/media/segchanman"
 )
 
 func TestConcatBin(t *testing.T) {
···
 		return fmt.Errorf("failed to read fixture file: %w", err)
 	}
 
-	testSegs := []*segchanman.Seg{}
+	testSegs := []*bus.Seg{}
 	for range 5 {
-		testSegs = append(testSegs, &segchanman.Seg{
+		testSegs = append(testSegs, &bus.Seg{
 			Data:     bs,
 			Filepath: filename,
 		})
 	}
 
-	segCh := make(chan *segchanman.Seg)
+	segCh := make(chan *bus.Seg)
 	go func() {
 		for _, seg := range testSegs {
 			segCh <- seg

pkg/media/concat_demux.go

···
 
 	"github.com/go-gst/go-gst/gst"
 	"github.com/go-gst/go-gst/gst/app"
+	"stream.place/streamplace/pkg/bus"
 	"stream.place/streamplace/pkg/log"
-	"stream.place/streamplace/pkg/media/segchanman"
 )
 
 // silly technique to avoid leaking pads
 func doNothing(self *gst.Element, pad *gst.Pad) {}
 
-func ConcatDemuxBin(ctx context.Context, seg *segchanman.Seg) (*gst.Bin, error) {
+func ConcatDemuxBin(ctx context.Context, seg *bus.Seg) (*gst.Bin, error) {
 	ctx = log.WithLogValues(ctx, "func", "SegDemuxBin")
 	bin := gst.NewBin("seg-demux-bin")
 

pkg/media/concat_demux_test.go

···
 	"github.com/stretchr/testify/require"
 	"go.uber.org/goleak"
 	"golang.org/x/sync/errgroup"
+	"stream.place/streamplace/pkg/bus"
 	"stream.place/streamplace/pkg/gstinit"
 	"stream.place/streamplace/pkg/log"
-	"stream.place/streamplace/pkg/media/segchanman"
 )
 
 func TestConcatDemuxBin(t *testing.T) {
···
 		return fmt.Errorf("failed to read fixture file: %w", err)
 	}
 
-	testSeg := &segchanman.Seg{
+	testSeg := &bus.Seg{
 		Data:     bs,
 		Filepath: filename,
 	}

pkg/media/media.go

···
 	"stream.place/streamplace/pkg/bus"
 	"stream.place/streamplace/pkg/config"
 	"stream.place/streamplace/pkg/gstinit"
-	"stream.place/streamplace/pkg/media/segchanman"
 	"stream.place/streamplace/pkg/model"
 
 	"stream.place/streamplace/pkg/replication"
···
 
 type MediaManager struct {
 	cli                 *config.CLI
-	segChanMan          *segchanman.SegChanMan
 	replicator          replication.Replicator
 	hlsRunning          map[string]*M3U8
 	hlsRunningMut       sync.Mutex
···
 	}
 	return &MediaManager{
 		cli:          cli,
-		segChanMan:   segchanman.MakeSegChanMan(),
 		replicator:   rep,
 		hlsRunning:   map[string]*M3U8{},
 		httpPipes:    map[string]io.Writer{},
···
 	defer mm.newSegmentSubsMutex.Unlock()
 	mm.newSegmentSubs = append(mm.newSegmentSubs, ch)
 	return ch
-}
-
-// subscribe to the latest segments from a given user for livestreaming purposes
-func (mm *MediaManager) SubscribeSegment(ctx context.Context, user string, rendition string) <-chan *segchanman.Seg {
-	return mm.segChanMan.SubscribeSegment(ctx, user, rendition)
-}
-
-func (mm *MediaManager) UnsubscribeSegment(ctx context.Context, user string, rendition string, ch <-chan *segchanman.Seg) {
-	mm.segChanMan.UnsubscribeSegment(ctx, user, rendition, ch)
-}
-
-// subscribe to the latest segments from a given user for livestreaming purposes
-func (mm *MediaManager) PublishSegment(ctx context.Context, user, rendition string, seg *segchanman.Seg) {
-	mm.segChanMan.PublishSegment(ctx, user, rendition, seg)
 }
 
 type obj map[string]any

pkg/media/packetize.go

···
 
 	"github.com/go-gst/go-gst/gst"
 	"github.com/go-gst/go-gst/gst/app"
+	"stream.place/streamplace/pkg/bus"
 	"stream.place/streamplace/pkg/log"
-	"stream.place/streamplace/pkg/media/segchanman"
 )
 
-type PacketizedSegment struct {
-	Video    [][]byte
-	Audio    [][]byte
-	Duration time.Duration
-}
-
 // take in a segment and return a bunch of packets suitable for webrtc
-func Packetize(ctx context.Context, seg *segchanman.Seg) (*PacketizedSegment, error) {
+func Packetize(ctx context.Context, seg *bus.Seg) (*bus.PacketizedSegment, error) {
 
 	pipelineSlice := []string{
 		"h264parse name=videoparse ! video/x-h264,stream-format=byte-stream ! appsink sync=false name=videoappsink",
···
 		return nil, fmt.Errorf("pipeline error: %w", err)
 	}
 
-	return &PacketizedSegment{
+	return &bus.PacketizedSegment{
 		Video:    videoOutput,
 		Audio:    audioOutput,
 		Duration: segDur,

pkg/media/packetize_test.go

···
 	"github.com/stretchr/testify/require"
 	"go.uber.org/goleak"
 	"golang.org/x/sync/errgroup"
+	"stream.place/streamplace/pkg/bus"
 	"stream.place/streamplace/pkg/gstinit"
-	"stream.place/streamplace/pkg/media/segchanman"
 )
 
 func TestPacketize(t *testing.T) {
···
 	bs, err := io.ReadAll(inputFile)
 	require.NoError(t, err)
 
-	testSeg := &segchanman.Seg{
+	testSeg := &bus.Seg{
 		Data:     bs,
 		Filepath: filename,
 	}

pkg/media/segchanman/segchanman.go

···
-package segchanman
-
-import (
-	"context"
-	"fmt"
-	"sync"
-	"time"
-
-	"go.opentelemetry.io/otel"
-	"stream.place/streamplace/pkg/log"
-	"stream.place/streamplace/pkg/spmetrics"
-)
-
-// it's a segment channel manager, you see
-
-type Seg struct {
-	Filepath string
-	Data     []byte
-}
-
-type SegChanMan struct {
-	segChans      map[string][]chan *Seg
-	segChansMutex sync.Mutex
-}
-
-func MakeSegChanMan() *SegChanMan {
-	return &SegChanMan{
-		segChans: make(map[string][]chan *Seg),
-	}
-}
-
-func segChanKey(user string, rendition string) string {
-	return fmt.Sprintf("%s::%s", user, rendition)
-}
-
-func (s *SegChanMan) SubscribeSegment(ctx context.Context, user string, rendition string) <-chan *Seg {
-	key := segChanKey(user, rendition)
-	s.segChansMutex.Lock()
-	defer s.segChansMutex.Unlock()
-	chs, ok := s.segChans[key]
-	if !ok {
-		chs = []chan *Seg{}
-		s.segChans[key] = chs
-	}
-	ch := make(chan *Seg)
-	chs = append(chs, ch)
-	s.segChans[key] = chs
-	spmetrics.SegmentSubscriptionsOpen.WithLabelValues(user, rendition).Set(float64(len(chs)))
-	return ch
-}
-
-func (s *SegChanMan) UnsubscribeSegment(ctx context.Context, user string, rendition string, ch <-chan *Seg) {
-	key := segChanKey(user, rendition)
-	s.segChansMutex.Lock()
-	defer s.segChansMutex.Unlock()
-	chs, ok := s.segChans[key]
-	if !ok {
-		return
-	}
-	for i, c := range chs {
-		if c == ch {
-			chs = append(chs[:i], chs[i+1:]...)
-			break
-		}
-	}
-	spmetrics.SegmentSubscriptionsOpen.WithLabelValues(user, rendition).Set(float64(len(chs)))
-	s.segChans[key] = chs
-}
-
-func (s *SegChanMan) PublishSegment(ctx context.Context, user string, rendition string, seg *Seg) {
-	ctx, span := otel.Tracer("signer").Start(ctx, "PublishSegment")
-	defer span.End()
-	key := segChanKey(user, rendition)
-	s.segChansMutex.Lock()
-	defer s.segChansMutex.Unlock()
-	chs, ok := s.segChans[key]
-	if !ok {
-		return
-	}
-	for _, ch := range chs {
-		go func(ch chan *Seg) {
-			select {
-			case ch <- seg:
-			case <-ctx.Done():
-				return
-			case <-time.After(1 * time.Minute):
-				log.Warn(ctx, "failed to send segment to channel, timing out", "user", user, "rendition", rendition)
-			}
-
-		}(ch)
-	}
-}

pkg/media/segmenter_hls.go

···
 		return fmt.Errorf("error creating ToHLS pipeline: %w", err)
 	}
 
-	outputQueue, done, err := ConcatStream(ctx, pipeline, user, rendition, mm)
+	outputQueue, done, err := ConcatStream(ctx, pipeline, user, rendition, mm.bus)
 	if err != nil {
 		return fmt.Errorf("failed to get output queue: %w", err)
 	}

pkg/media/validate.go

···
 	"stream.place/streamplace/pkg/constants"
 	"stream.place/streamplace/pkg/crypto/signers"
 	"stream.place/streamplace/pkg/log"
-	"stream.place/streamplace/pkg/media/segchanman"
 	"stream.place/streamplace/pkg/model"
 
 	"git.stream.place/streamplace/c2pa-go/pkg/c2pa"
···
 	if _, err := io.Copy(fd, r); err != nil {
 		return err
 	}
-	scmSeg := &segchanman.Seg{
-		Filepath: fd.Name(),
-		Data:     buf,
-	}
-	go mm.PublishSegment(ctx, repoDID, "source", scmSeg)
 	seg := &model.Segment{
 		ID:            *mani.Label,
 		SigningKeyDID: signingKeyDID,

pkg/media/webrtc_playback.go

···
 	"github.com/google/uuid"
 	"github.com/pion/webrtc/v4"
 	"github.com/pion/webrtc/v4/pkg/media"
+	"stream.place/streamplace/pkg/bus"
 	"stream.place/streamplace/pkg/log"
-	"stream.place/streamplace/pkg/media/segchanman"
 	"stream.place/streamplace/pkg/spmetrics"
 )
 
···
 		return nil, fmt.Errorf("failed to create GStreamer pipeline: %w", err) //nolint:all
 	}
 
-	segBuffer := make(chan *segchanman.Seg, 1024)
+	segBuffer := make(chan *bus.Seg, 1024)
 	go func() {
-		ch := mm.SubscribeSegment(ctx, user, rendition)
-		defer mm.UnsubscribeSegment(ctx, user, rendition, ch)
+		ch := mm.bus.SubscribeSegment(ctx, user, rendition)
+		defer mm.bus.UnsubscribeSegment(ctx, user, rendition, ch)
 		for {
 			select {
 			case <-ctx.Done():
···
 		}
 	}()
 
-	segCh := make(chan *segchanman.Seg)
+	segCh := make(chan *bus.Seg)
 	go func() {
 		for {
 			select {

pkg/media/webrtc_playback2.go

···
 	"github.com/pion/webrtc/v4"
 	"github.com/pion/webrtc/v4/pkg/media"
 	"golang.org/x/sync/errgroup"
+	"stream.place/streamplace/pkg/bus"
 	"stream.place/streamplace/pkg/log"
 	"stream.place/streamplace/pkg/spmetrics"
 )
···
 			}
 		}()
 
-		packetQueue := make(chan *PacketizedSegment, 1024)
+		packetQueue := make(chan *bus.PacketizedSegment, 1024)
 		go func() {
-			ch := mm.SubscribeSegment(ctx, user, rendition)
-			defer mm.UnsubscribeSegment(ctx, user, rendition, ch)
+			ch := mm.bus.SubscribeSegment(ctx, user, rendition)
+			defer mm.bus.UnsubscribeSegment(ctx, user, rendition, ch)
 			for {
 				select {
 				case <-ctx.Done():
···
 
 			p1 := <-packetQueue
 			p2 := <-packetQueue
-			bufPacketQueue := make(chan *PacketizedSegment, 1024)
+			bufPacketQueue := make(chan *bus.PacketizedSegment, 1024)
 			go func() {
 				bufPacketQueue <- p1
 				bufPacketQueue <- p2