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 // presentation flat MP4
	Muxl           []byte // bare canonical .m4s (blindly concatenatable)
	PacketizedData *PacketizedSegment
	Published      bool
}

// PacketizedSample is one WebRTC-writable sample — a video access unit or an
// Opus packet — with its real duration on the source timeline. Carrying the
// per-sample duration (rather than dividing the segment evenly) keeps
// non-uniform frame spacing intact: an encoder shedding frames under
// bandwidth pressure sends bursts and gaps, and respacing those uniformly
// rubber-bands the video against the audio.
type PacketizedSample struct {
	Data     []byte
	Duration time.Duration
}

type PacketizedSegment struct {
	Video    []PacketizedSample
	Audio    []PacketizedSample
	Duration time.Duration
}

var chanSize = 1024

type SegChan struct {
	C       chan *Seg
	Context context.Context
}

var bufSize = 10

func segChanKey(user string, rendition string) string {
	return fmt.Sprintf("%s::%s", user, rendition)
}

// get a channel to subscribe to new segments for a given user and rendition
func (b *Bus) SubscribeSegment(ctx context.Context, user string, rendition string) *SegChan {
	return b.SubscribeSegmentBuf(ctx, user, rendition, 0)
}

// get a channel to subscribe to new segments for a given user and rendition,
// starting with bufSize cached segments that we already have
func (b *Bus) SubscribeSegmentBuf(ctx context.Context, user string, rendition string, bufSize int) *SegChan {
	key := segChanKey(user, rendition)
	b.segChansMutex.Lock()
	defer b.segChansMutex.Unlock()
	chs, ok := b.segChans[key]
	if !ok {
		chs = []*SegChan{}
		b.segChans[key] = chs
	}
	// Buffered, and fed in publication order by PublishSegment itself (no
	// goroutine per segment, which delivered bursts in random order); a
	// subscriber that falls chanSize segments behind loses segments rather
	// than holding the publisher. The most recent segments are replayed
	// into it first so a late subscriber can start at once.
	ch := make(chan *Seg, chanSize)
	b.segBufMutex.RLock()
	defer b.segBufMutex.RUnlock()
	curBuf, ok := b.segBuf[key]
	if ok {
		if bufSize > len(curBuf) {
			bufSize = len(curBuf)
		}
		for i := 0; i < bufSize; i += 1 {
			ch <- curBuf[len(curBuf)-bufSize+i]
		}
	}
	segChan := &SegChan{C: ch, Context: ctx}
	chs = append(chs, segChan)
	b.segChans[key] = chs
	spmetrics.SegmentSubscriptionsOpen.WithLabelValues(user, rendition).Set(float64(len(chs)))
	return segChan
}

// unsubscribe from a channel for a given user and rendition
func (b *Bus) UnsubscribeSegment(ctx context.Context, user string, rendition string, ch *SegChan) {
	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()
	b.segBufMutex.Lock()
	defer b.segBufMutex.Unlock()
	curBuf, ok := b.segBuf[key]
	if !ok {
		curBuf = []*Seg{}
		b.segBuf[key] = curBuf
	}
	curBuf = append(curBuf, seg)
	if len(curBuf) > bufSize {
		curBuf = curBuf[1:]
	}
	b.segBuf[key] = curBuf
	chs, ok := b.segChans[key]
	if !ok {
		return
	}
	for _, segChan := range chs {
		select {
		case segChan.C <- seg:
		case <-segChan.Context.Done():
		default:
			log.Warn(ctx, "segment subscriber is not keeping up, dropping a segment", "user", user, "rendition", rendition, "behind", len(segChan.C))
		}
	}
}

func (b *Bus) EndSession(ctx context.Context, user string, rendition string) {
	b.segChansMutex.Lock()
	defer b.segChansMutex.Unlock()
	b.segBufMutex.Lock()
	defer b.segBufMutex.Unlock()

	key := segChanKey(user, rendition)
	delete(b.segBuf, key)
}