livekit/pkg/sfu/buffer/buffer.go

// Copyright 2023 LiveKit, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package buffer

import (
	"encoding/binary"
	"errors"
	"io"

	"github.com/pion/rtcp"
	"github.com/pion/rtp"
	"github.com/pion/webrtc/v4"

	"github.com/livekit/livekit-server/pkg/sfu/flexfec"
	sutils "github.com/livekit/livekit-server/pkg/utils"
	"github.com/livekit/mediatransportutil/pkg/bucket"
	"github.com/livekit/mediatransportutil/pkg/twcc"
	"github.com/livekit/protocol/livekit"
	"github.com/livekit/protocol/utils/mono"
)

const (
	rtcpReceiverReportDelta = 1e9

	InitPacketBufferSizeVideo = 300
	InitPacketBufferSizeAudio = 70

	// An unpaired FEC repair stream only needs a short bridge until its primary
	// stream appears. The decoder itself retains at most this many FEC states.
	maxPendingFECRepairPackets = 100
)

var (
	errInvalidCodec           = errors.New("invalid codec")
	errFECMediaPacketNotFound = errors.New("fec media packet not found")
)

var _ BufferProvider = (*Buffer)(nil)

type pendingPacket struct {
	arrivalTime int64
	packet      []byte
}

type fecRecoveryDelta struct {
	received      int
	recovered     int
	discarded     int
	bytesReceived int
}

func (d fecRecoveryDelta) invoke(callback func(received int, recovered int, discarded int, bytesReceived int)) {
	if callback != nil && (d.recovered > 0 || d.received > 0 || d.discarded > 0) {
		callback(d.received, d.recovered, d.discarded, d.bytesReceived)
	}
}

type cbFECRecovery func(received int, recovered int, discarded int, bytesReceived int)

// Buffer contains all packets
type Buffer struct {
	*BufferBase

	pPackets     []pendingPacket
	lastReportAt int64
	isBound      bool

	twcc      *twcc.Responder
	twccExtID uint8

	enableAudioLossProxying  bool
	lastFractionLostToReport uint8 // Last fraction lost from subscribers, should report to publisher; Audio only

	lastPacketRead int

	// callbacks
	onClose         func()
	onRtcpFeedback  func([]rtcp.Packet)
	onFinalRtpStats func(*livekit.RTPStats)
	onNotifyRTX     func(uint32, uint32, string)

	primaryBufferForRTX *Buffer
	rtxPktBuf           []byte

	primaryBufferForFEC *Buffer
	isFECRepair         bool
	fecSSRC             uint32
	fecDecoder          *flexfec.Decoder
	fecPktBuf           []byte
	onFECRecovery       cbFECRecovery

	streamInfoProbe       *StreamInfoProbe
	warnedPendingOverflow bool
}

// StreamInfoProbe identifies a stream from the mid/rid/rsid header extensions of its
// packets. It runs on the write path, i. e. as SRTP pushes into this buffer, because
// nothing reads remote streams through pion's interceptor chain.
type StreamInfoProbe struct {
	MidExtID  uint8
	RidExtID  uint8
	RsidExtID uint8

	// Tries bounds how many packets are inspected before giving up.
	Tries int

	// OnFound is called at most once, in a goroutine, as it can re-enter this buffer.
	OnFound func(ssrc uint32, mid, rid, rsid string)
}

func NewBuffer(ssrc uint32, maxVideoPkts, maxAudioPkts int) *Buffer {
	b := &Buffer{}
	b.BufferBase = NewBufferBase(BufferBaseParams{
		SSRC:                  ssrc,
		MaxVideoPkts:          maxVideoPkts,
		MaxAudioPkts:          maxAudioPkts,
		LoggerComponents:      []string{sutils.ComponentPub, sutils.ComponentSFU},
		SendPLI:               b.sendPLI,
		OnStreamRestartLocked: b.recreateFECDecoderLocked,
		IsReportingEnabled:    true,
		// a stream resuming after a gap, e. g. a layer after a dynacast pause, can have a
		// dependency descriptor frame number more than half the 16 bit range away
		IsDDRestartEnabled: true,
	})
	return b
}

func (b *Buffer) SetTWCCAndExtID(twcc *twcc.Responder, extID uint8) {
	b.Lock()
	defer b.Unlock()

	b.twcc = twcc
	b.twccExtID = extID
}

func (b *Buffer) SetAudioLossProxying(enable bool) {
	b.Lock()
	defer b.Unlock()

	b.enableAudioLossProxying = enable
}

func (b *Buffer) Bind(params webrtc.RTPParameters, codec webrtc.RTPCodecCapability, bitrates int) error {
	b.Lock()
	if b.isBound {
		b.Unlock()
		return nil
	}

	if err := b.BufferBase.BindLocked(params, codec, bitrates); err != nil {
		b.Unlock()
		return err
	}

	now := mono.UnixNano()
	b.lastReportAt = now

	if len(b.pPackets) != 0 {
		b.logger.Debugw("releasing queued packets on bind", "count", len(b.pPackets))
	}
	for _, pp := range b.pPackets {
		b.ingestLocked(pp.packet, nil, pp.arrivalTime, true, false)
	}
	b.pPackets = nil

	rtcpPackets := b.getRTCPPackets(now)

	b.isBound = true
	b.maybeCreateFECDecoderLocked()
	b.Unlock()

	b.packetPostProcess(rtcpPackets, fecRecoveryDelta{}, nil)
	return nil
}

// Write adds an RTP Packet, ordering is not guaranteed, newer packets may arrive later
func (b *Buffer) Write(pkt []byte) (n int, err error) {
	rtpPacket := rtpPacketFactory.Get().(*rtp.Packet)
	defer rtpPacketFactory.Put(rtpPacket)
	err = rtpPacket.Unmarshal(pkt)
	if err != nil {
		return
	}

	b.Lock()
	if b.BufferBase.IsClosed() {
		b.Unlock()
		err = io.EOF
		return
	}

	now := mono.UnixNano()
	if b.twcc != nil && b.twccExtID != 0 {
		if ext := rtpPacket.GetExtension(b.twccExtID); len(ext) >= 2 {
			b.twcc.Push(rtpPacket.SSRC, binary.BigEndian.Uint16(ext[0:2]), now, rtpPacket.Marker)
		}
	}

	// libwebrtc will use 0 ssrc for probing, don't push the packet to pending queue to avoid memory increasing since
	// the Bind will not be called to consume the pending packets. More details in https://github.com/pion/webrtc/pull/2816
	if rtpPacket.SSRC == 0 {
		b.Unlock()
		return
	}

	if b.streamInfoProbe != nil {
		b.probeStreamInfoLocked(rtpPacket)
	}

	// handle RTX packet
	if pb := b.primaryBufferForRTX; pb != nil {
		b.Unlock()

		// skip padding only packets
		if rtpPacket.Padding && len(rtpPacket.Payload) == 0 {
			return
		}

		pb.writeRTX(rtpPacket, now)
		return
	}

	// handle FlexFEC packet
	if pb := b.primaryBufferForFEC; pb != nil {
		b.Unlock()

		// skip padding only packets
		if rtpPacket.Padding && len(rtpPacket.Payload) == 0 {
			return
		}

		pb.writeFEC(rtpPacket, now)
		return
	}

	if !b.isBound {
		if len(b.pPackets) == 0 {
			b.logger.Debugw("received first packet")
		}

		pendingLimit := max(b.BufferBase.MaxVideoPkts(), b.BufferBase.MaxAudioPkts())
		if b.isFECRepair {
			pendingLimit = min(pendingLimit, maxPendingFECRepairPackets)
		}
		pendingLimit = max(pendingLimit, 1)
		if overflow := len(b.pPackets) - pendingLimit + 1; overflow > 0 {
			clear(b.pPackets[:overflow])
			b.pPackets = b.pPackets[overflow:]

			// a stream that keeps arriving but never binds drops every packet from here
			// on; for an RTX stream it means the pairing was never established
			if !b.warnedPendingOverflow {
				b.warnedPendingOverflow = true
				b.logger.Warnw(
					"unbound buffer overflowing, dropping packets", nil,
					"ssrc", b.BufferBase.SSRC(),
					"pending", len(b.pPackets),
				)
			}
		}

		packet := make([]byte, len(pkt))
		copy(packet, pkt)
		b.pPackets = append(b.pPackets, pendingPacket{
			packet:      packet,
			arrivalTime: now,
		})

		b.BufferBase.NotifyRead()
		b.Unlock()
		return
	}

	b.ingestLocked(pkt, rtpPacket, now, false, false)

	var fecDelta fecRecoveryDelta
	var onFECRecovery cbFECRecovery
	if b.fecDecoder != nil {
		// feed media into the FEC decoder, a media arrival can complete a
		// previously unrecoverable FEC window. Recovered packets update NACK
		// state before feedback is generated below, so a packet repaired by
		// FEC is not needlessly NACKed.
		fecDelta, onFECRecovery = b.feedFECLocked(rtpPacket, now)
	}

	rtcpPackets := b.getRTCPPackets(now)
	b.Unlock()

	b.packetPostProcess(rtcpPackets, fecDelta, onFECRecovery)
	return
}

// SetStreamInfoProbe installs probe and runs it over packets already queued.
func (b *Buffer) SetStreamInfoProbe(probe *StreamInfoProbe) {
	b.Lock()
	defer b.Unlock()

	b.streamInfoProbe = probe

	for _, pp := range b.pPackets {
		if b.streamInfoProbe == nil {
			return
		}

		var rtpPacket rtp.Packet
		if err := rtpPacket.Unmarshal(pp.packet); err != nil {
			continue
		}
		b.probeStreamInfoLocked(&rtpPacket)
	}
}

// probeStreamInfoLocked inspects one packet, clearing the probe once the stream is
// identified or the try budget runs out.
func (b *Buffer) probeStreamInfoLocked(rtpPacket *rtp.Packet) {
	probe := b.streamInfoProbe

	var mid, rid, rsid string
	if ext := rtpPacket.GetExtension(probe.MidExtID); ext != nil {
		mid = string(ext)
	}
	if ext := rtpPacket.GetExtension(probe.RidExtID); ext != nil {
		rid = string(ext)
	}
	if ext := rtpPacket.GetExtension(probe.RsidExtID); ext != nil {
		rsid = string(ext)
	}

	if mid != "" && (rid != "" || rsid != "") {
		b.streamInfoProbe = nil
		b.logger.Debugw("stream found", "ssrc", rtpPacket.SSRC, "mid", mid, "rid", rid, "rsid", rsid)
		go probe.OnFound(rtpPacket.SSRC, mid, rid, rsid)
		return
	}

	// ignore padding only packets for probe count
	if rtpPacket.Padding && len(rtpPacket.Payload) == 0 {
		return
	}

	if probe.Tries--; probe.Tries <= 0 {
		b.streamInfoProbe = nil
	}
}

func (b *Buffer) SetPrimaryBufferForRTX(primaryBuffer *Buffer) {
	b.Lock()
	b.primaryBufferForRTX = primaryBuffer
	pkts := b.pPackets
	b.pPackets = nil
	b.Unlock()

	for _, pp := range pkts {
		var rtpPacket rtp.Packet
		err := rtpPacket.Unmarshal(pp.packet)
		if err != nil {
			continue
		}
		if rtpPacket.Padding && len(rtpPacket.Payload) == 0 {
			continue
		}
		primaryBuffer.writeRTX(&rtpPacket, pp.arrivalTime)
	}
}

func (b *Buffer) NotifyRTX(ssrc uint32, repairSSRC uint32, rsid string) {
	if onNotifyRTX := b.getOnNotifyRTX(); onNotifyRTX != nil {
		onNotifyRTX(ssrc, repairSSRC, rsid)
	}
}

func (b *Buffer) writeRTX(rtxPkt *rtp.Packet, arrivalTime int64) {
	b.Lock()
	if !b.isBound {
		b.Unlock()
		return
	}

	if rtxPkt.PayloadType != b.rtxPayloadType {
		b.logger.Debugw("unexpected rtx payload type", "expected", b.rtxPayloadType, "actual", rtxPkt.PayloadType)
		b.Unlock()
		return
	}

	if b.rtxPktBuf == nil {
		b.rtxPktBuf = make([]byte, bucket.RTPMaxPktSize)
	}

	if len(rtxPkt.Payload) < 2 {
		b.logger.Warnw("rtx payload too short", nil, "size", len(rtxPkt.Payload))
		b.Unlock()
		return
	}

	repairedPkt := *rtxPkt
	repairedPkt.PayloadType = b.payloadType
	repairedPkt.SequenceNumber = binary.BigEndian.Uint16(rtxPkt.Payload[:2])
	repairedPkt.SSRC = b.BufferBase.SSRC()
	repairedPkt.Payload = rtxPkt.Payload[2:]
	n, err := repairedPkt.MarshalTo(b.rtxPktBuf)
	if err != nil {
		b.logger.Errorw("could not marshal repaired packet", err, "ssrc", b.BufferBase.SSRC(), "sn", repairedPkt.SequenceNumber)
		b.Unlock()
		return
	}

	b.ingestLocked(b.rtxPktBuf[:n], &repairedPkt, arrivalTime, false, true)

	var fecDelta fecRecoveryDelta
	var onFECRecovery cbFECRecovery
	if b.fecDecoder != nil {
		fecDelta, onFECRecovery = b.feedFECLocked(&repairedPkt, arrivalTime)
	}

	rtcpPackets := b.getRTCPPackets(arrivalTime)
	b.Unlock()

	b.packetPostProcess(rtcpPackets, fecDelta, onFECRecovery)
}

func (b *Buffer) SetPrimaryBufferForFEC(primaryBuffer *Buffer) {
	b.Lock()
	b.primaryBufferForFEC = primaryBuffer
	pkts := b.pPackets
	b.pPackets = nil
	ssrc := b.BufferBase.SSRC()
	b.Unlock()

	// Let the primary know the repair stream SSRC so its decoder is ready
	// before the first FEC packet arrives.
	primaryBuffer.setFECSSRC(ssrc)

	for _, pp := range pkts {
		var rtpPacket rtp.Packet
		err := rtpPacket.Unmarshal(pp.packet)
		if err != nil {
			continue
		}
		if rtpPacket.Padding && len(rtpPacket.Payload) == 0 {
			continue
		}
		primaryBuffer.writeFEC(&rtpPacket, pp.arrivalTime)
	}
}

func (b *Buffer) setFECRepairBufferLimit() {
	b.Lock()
	defer b.Unlock()

	b.isFECRepair = true
	if len(b.pPackets) <= maxPendingFECRepairPackets {
		return
	}

	// Pairing can be announced after repair packets start arriving. Compact the
	// retained tail so both the dropped packet bytes and oversized slice backing
	// array become collectible.
	start := len(b.pPackets) - maxPendingFECRepairPackets
	retained := make([]pendingPacket, maxPendingFECRepairPackets)
	copy(retained, b.pPackets[start:])
	clear(b.pPackets)
	b.pPackets = retained
}

func (b *Buffer) setFECSSRC(ssrc uint32) {
	b.Lock()
	b.fecSSRC = ssrc
	b.maybeCreateFECDecoderLocked()
	b.Unlock()
}

// maybeCreateFECDecoderLocked creates the FEC decoder as soon as the repair
// stream SSRC is known and the buffer is bound with a negotiated flexfec
// payload type. Protected media is read from the primary RTP packet bucket.
func (b *Buffer) maybeCreateFECDecoderLocked() {
	if b.fecDecoder != nil || b.fecSSRC == 0 || !b.isBound || b.fecPayloadType == 0 {
		return
	}

	b.fecDecoder = flexfec.NewDecoder(b.fecSSRC, b.BufferBase.SSRC(), b.getFECMediaPacketLocked, b.logger)
	b.logger.Debugw("flexfec decoder created", "fecSSRC", b.fecSSRC, "mediaSSRC", b.BufferBase.SSRC())
}

// recreateFECDecoderLocked drops FlexFEC state from the previous sequence
// space. Stream restart resyncs the packet bucket and sequence map, so repair
// packets retained by the old decoder would recover against the wrong media
// or be treated as duplicates of the new repair stream.
func (b *Buffer) recreateFECDecoderLocked() {
	if b.fecDecoder == nil {
		return
	}

	b.fecDecoder = nil
	b.maybeCreateFECDecoderLocked()
}

func (b *Buffer) getFECMediaPacketLocked(sequenceNumber uint16, dst []byte) (int, error) {
	if b.bucket == nil || b.rtpStats == nil {
		return 0, errFECMediaPacketNotFound
	}

	// FlexFEC masks use the publisher's sequence-number space. BufferBase
	// removes padding-only packets from the downstream space, so resolve the
	// original extended sequence number and apply the same adjustment used
	// when the packet was inserted into the bucket.
	highestSequenceNumber := b.rtpStats.ExtendedHighestSequenceNumber()
	extendedSequenceNumber := int64(highestSequenceNumber) + int64(int16(sequenceNumber-uint16(highestSequenceNumber)))
	if extendedSequenceNumber < 0 {
		return 0, errFECMediaPacketNotFound
	}

	extendedSN := uint64(extendedSequenceNumber)
	sequenceNumberAdjustment, err := b.snRangeMap.GetValue(extendedSN)
	if err != nil || sequenceNumberAdjustment > extendedSN {
		return 0, errFECMediaPacketNotFound
	}

	return b.bucket.GetPacket(dst, extendedSN-sequenceNumberAdjustment)
}

// OnFECRecovery is called with counter deltas whenever FEC packets are
// processed since the previous callback: FEC packets received, media packets
// recovered, FEC packets discarded and FEC bytes received.
func (b *Buffer) OnFECRecovery(fn cbFECRecovery) {
	b.Lock()
	b.onFECRecovery = fn
	b.Unlock()
}

// FECDecoderStats returns cumulative FlexFEC decode counters of the buffer.
func (b *Buffer) FECDecoderStats() flexfec.DecoderStats {
	b.RLock()
	defer b.RUnlock()

	if b.fecDecoder == nil {
		return flexfec.DecoderStats{}
	}
	return b.fecDecoder.Stats()
}

// writeFEC handles a packet of the coupled FlexFEC repair stream, recovered
// media packets are injected into the regular packet pipeline.
func (b *Buffer) writeFEC(fecPkt *rtp.Packet, arrivalTime int64) {
	b.Lock()
	if !b.isBound {
		b.Unlock()
		return
	}

	// the FEC stream is never bound in pion, run its TWCC accounting here so
	// publisher send side BWE sees the FEC packets acked. The repair stream
	// shares the media m-line, extension ids match the primary stream.
	if b.twcc != nil && b.twccExtID != 0 {
		if ext := fecPkt.GetExtension(b.twccExtID); len(ext) >= 2 {
			b.twcc.Push(fecPkt.SSRC, binary.BigEndian.Uint16(ext[0:2]), arrivalTime, fecPkt.Marker)
		}
	}

	if b.fecPayloadType == 0 || fecPkt.PayloadType != b.fecPayloadType {
		b.logger.Debugw("unexpected fec payload type", "expected", b.fecPayloadType, "actual", fecPkt.PayloadType)
		b.Unlock()
		return
	}

	if b.fecDecoder == nil {
		// normally created when the pair is declared, fall back to the
		// observed repair stream SSRC
		b.fecSSRC = fecPkt.SSRC
		b.maybeCreateFECDecoderLocked()
		if b.fecDecoder == nil {
			b.logger.Infow("flexfec decoder unavailable after creation attempt", "fecSSRC", b.fecSSRC, "mediaSSRC", b.BufferBase.SSRC(), "payloadType", b.fecPayloadType)
			b.Unlock()
			return
		}
	}

	fecDelta, onFECRecovery := b.feedFECLocked(fecPkt, arrivalTime)
	rtcpPackets := b.getRTCPPackets(arrivalTime)
	b.Unlock()

	b.packetPostProcess(rtcpPackets, fecDelta, onFECRecovery)
}

// feedFECLocked runs a media or FEC packet through the FEC decoder and
// injects recovered packets into the packet pipeline. Must be called with the
// buffer lock held and a non-nil decoder.
func (b *Buffer) feedFECLocked(
	pkt *rtp.Packet,
	arrivalTime int64,
) (fecRecoveryDelta, cbFECRecovery) {
	statsBefore := b.fecDecoder.Stats()
	recovered := b.fecDecoder.DecodeFEC(pkt)
	statsAfter := b.fecDecoder.Stats()

	fecPackets := statsAfter.FECPacketsReceived - statsBefore.FECPacketsReceived
	fecBytes := statsAfter.FECBytesReceived - statsBefore.FECBytesReceived
	fecPacketsDiscarded := statsAfter.FECPacketsDiscarded - statsBefore.FECPacketsDiscarded
	fecPacketsRecovered := statsAfter.PacketsRecovered - statsBefore.PacketsRecovered
	if b.rtpStats != nil {
		b.rtpStats.UpdateFEC(
			fecPackets,
			fecBytes,
			fecPacketsDiscarded,
			fecPacketsRecovered,
		)
	}

	if len(recovered) > 0 && b.fecPktBuf == nil {
		b.fecPktBuf = make([]byte, bucket.RTPMaxPktSize)
	}
	for _, rp := range recovered {
		n, err := rp.MarshalTo(b.fecPktBuf)
		if err != nil {
			b.logger.Warnw("could not marshal fec recovered packet", err, "ssrc", b.BufferBase.SSRC(), "sn", rp.SequenceNumber)
			continue
		}

		// recovered packets flow through the regular pipeline: they are
		// forwarded downstream and stop NACKs for the lost sequence numbers.
		// They do not re-enter the decoder because chained recovery already
		// completed within DecodeFEC. RTCP is generated once by the caller
		// after all recovered packets have updated NACK state.
		b.ingestLocked(b.fecPktBuf[:n], rp, arrivalTime, false, true)
	}

	if cb := b.onFECRecovery; cb != nil {
		return fecRecoveryDelta{
			received:      int(fecPackets),
			recovered:     int(fecPacketsRecovered),
			discarded:     int(fecPacketsDiscarded),
			bytesReceived: int(fecBytes),
		}, cb
	}

	return fecRecoveryDelta{}, nil
}

func (b *Buffer) Read(buff []byte) (n int, err error) {
	b.Lock()
	for {
		if b.BufferBase.IsClosed() {
			b.Unlock()
			return 0, io.EOF
		}

		if b.pPackets != nil && len(b.pPackets) > b.lastPacketRead {
			if len(buff) < len(b.pPackets[b.lastPacketRead].packet) {
				b.Unlock()
				return 0, bucket.ErrBufferTooSmall
			}

			n = copy(buff, b.pPackets[b.lastPacketRead].packet)
			b.lastPacketRead++
			b.Unlock()
			return
		}
		b.BufferBase.WaitRead()
	}
}

func (b *Buffer) Close() error {
	stats, err := b.BufferBase.CloseWithReason("close")
	if err != nil {
		return err
	}

	if stats != nil {
		if cb := b.getOnFinalRtpStats(); cb != nil {
			cb(stats)
		}
	}

	if cb := b.getOnClose(); cb != nil {
		cb()
	}

	return nil
}

func (b *Buffer) OnClose(fn func()) {
	b.Lock()
	b.onClose = fn
	b.Unlock()
}

func (b *Buffer) getOnClose() func() {
	b.RLock()
	defer b.RUnlock()

	return b.onClose
}

func (b *Buffer) sendPLI() {
	ssrc := b.BufferBase.SSRC()
	if ssrc == 0 {
		return
	}

	b.logger.Debugw("send pli", "mediaSSRC", ssrc)
	pli := []rtcp.Packet{
		&rtcp.PictureLossIndication{
			SenderSSRC: ssrc,
			MediaSSRC:  ssrc,
		},
	}

	if cb := b.getOnRtcpFeedback(); cb != nil {
		cb(pli)
	}
}

// ingestLocked runs a packet through the incoming packet pipeline, updating RTP
// stats, NACK state and the packet bucket. It intentionally does not generate
// RTCP so that FEC recovery can update NACK state before feedback is generated
// once. Must be called with the buffer lock held.
func (b *Buffer) ingestLocked(rawPkt []byte, rtpPacket *rtp.Packet, arrivalTime int64, isBuffered bool, isRTX bool) {
	b.BufferBase.HandleIncomingPacketLocked(
		rawPkt,
		rtpPacket,
		arrivalTime,
		isBuffered,
		isRTX,
		nil,
		0,
	)
}

// packetPostProcess runs processing after a packet is handled.
// `rtcpPackets` should be RTCP feedback from the current NACK state, which
// already reflects any FEC recovery performed during this write.
// This is to ensure that packets recovered via FEC is not NACKed.
func (b *Buffer) packetPostProcess(
	rtcpPackets []rtcp.Packet,
	fecDelta fecRecoveryDelta,
	onFECRecovery cbFECRecovery,
) {
	fecDelta.invoke(onFECRecovery)

	if len(rtcpPackets) != 0 {
		if cb := b.getOnRtcpFeedback(); cb != nil {
			cb(rtcpPackets)
		}
	}
}

func (b *Buffer) getRTCPPackets(arrivalTime int64) []rtcp.Packet {
	var pkts []rtcp.Packet
	if nackPkt := b.getNACKPacket(); nackPkt != nil {
		pkts = append(pkts, nackPkt)
	}
	if receiverReport := b.getRTCPReceiverReport(arrivalTime); receiverReport != nil {
		pkts = append(pkts, receiverReport)
	}

	return pkts
}

func (b *Buffer) getNACKPacket() *rtcp.TransportLayerNack {
	if nacks := b.BufferBase.GetNACKPairsLocked(); len(nacks) > 0 {
		ssrc := b.BufferBase.SSRC()
		return &rtcp.TransportLayerNack{
			SenderSSRC: ssrc,
			MediaSSRC:  ssrc,
			Nacks:      nacks,
		}
	}
	return nil
}

func (b *Buffer) getRTCPReceiverReport(arrivalTime int64) *rtcp.ReceiverReport {
	if arrivalTime-b.lastReportAt < rtcpReceiverReportDelta {
		return nil
	}
	b.lastReportAt = arrivalTime

	proxyLoss := b.lastFractionLostToReport
	if b.codecType == webrtc.RTPCodecTypeAudio && !b.enableAudioLossProxying {
		proxyLoss = 0
	}

	if receptionReport := b.BufferBase.GetRtcpReceptionReportLocked(proxyLoss); receptionReport != nil {
		return &rtcp.ReceiverReport{
			SSRC:    b.BufferBase.SSRC(),
			Reports: []rtcp.ReceptionReport{*receptionReport},
		}
	}

	return nil
}

func (b *Buffer) SetLastFractionLostReport(lost uint8) {
	b.Lock()
	defer b.Unlock()

	b.lastFractionLostToReport = lost
}

func (b *Buffer) OnRtcpFeedback(fn func(fb []rtcp.Packet)) {
	b.Lock()
	b.onRtcpFeedback = fn
	b.Unlock()
}

func (b *Buffer) getOnRtcpFeedback() func(fb []rtcp.Packet) {
	b.RLock()
	defer b.RUnlock()

	return b.onRtcpFeedback
}

func (b *Buffer) OnFinalRtpStats(fn func(*livekit.RTPStats)) {
	b.Lock()
	b.onFinalRtpStats = fn
	b.Unlock()
}

func (b *Buffer) getOnFinalRtpStats() func(*livekit.RTPStats) {
	b.RLock()
	defer b.RUnlock()

	return b.onFinalRtpStats
}

func (b *Buffer) OnNotifyRTX(fn func(ssrc uint32, repairSSRC uint32, rsid string)) {
	b.Lock()
	b.onNotifyRTX = fn
	b.Unlock()
}

func (b *Buffer) getOnNotifyRTX() func(ssrc uint32, repairSSRC uint32, rsid string) {
	b.RLock()
	defer b.RUnlock()

	return b.onNotifyRTX
}