package live

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"os"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	bizgift "kra/internal/biz/gift"
	bizlive "kra/internal/biz/live"
	bizchat "kra/internal/biz/livechat"
	platformws "kra/pkg/websocket"

	"github.com/olahol/melody"
	"github.com/pion/webrtc/v4"
)

type Manager struct {
	hub         platformws.Hub
	streams     *bizlive.StreamUsecase
	roomCatalog *bizlive.RoomUsecase
	chat        *bizchat.Usecase
	logger      *slog.Logger
	api         *webrtc.API
	mu          sync.RWMutex
	rooms       map[string]*room
	peers       map[*melody.Session]*peer
	observers   map[*melody.Session]struct{}
	nextSub     atomic.Uint64
	nextGuest   atomic.Uint64
	stop        chan struct{}
	stopOnce    sync.Once
}

type room struct {
	mu        sync.RWMutex
	streamID  string
	publisher *peer
	viewers   map[*peer]struct{}
	tracks    map[webrtc.RTPCodecType]*publishedTrack
}

type publishedTrack struct {
	id    string
	kind  webrtc.RTPCodecType
	local *webrtc.TrackLocalStaticRTP
}

type peer struct {
	session       *melody.Session
	manager       *Manager
	room          *room
	streamID      string
	userID        uint
	role          string
	authenticated bool
	subscriberID  string
	pcMu          sync.RWMutex
	pc            *webrtc.PeerConnection
	closing       bool
	lastBeat      atomic.Int64
	cleanup       sync.Once
}

type message struct {
	Type    string          `json:"type"`
	ID      string          `json:"id,omitempty"`
	Payload json.RawMessage `json:"payload,omitempty"`
}

type joinPayload struct {
	Role     string `json:"role"`
	StreamID string `json:"streamId"`
}

type sessionDescriptionPayload struct {
	Type string `json:"type"`
	SDP  string `json:"sdp"`
}

type chatSendPayload struct {
	Content string `json:"content"`
}

func newPionAPI() (*webrtc.API, error) {
	settings := webrtc.SettingEngine{}
	minPort := envPort("KRA_LIVE_UDP_MIN_PORT")
	maxPort := envPort("KRA_LIVE_UDP_MAX_PORT")
	if minPort > 0 || maxPort > 0 {
		if minPort == 0 || maxPort == 0 || minPort > maxPort {
			return nil, fmt.Errorf("invalid KRA_LIVE_UDP_MIN_PORT/KRA_LIVE_UDP_MAX_PORT")
		}
		if err := settings.SetEphemeralUDPPortRange(minPort, maxPort); err != nil {
			return nil, fmt.Errorf("configure live udp port range: %w", err)
		}
	}
	return webrtc.NewAPI(webrtc.WithSettingEngine(settings)), nil
}

func envPort(key string) uint16 {
	value, err := strconv.ParseUint(strings.TrimSpace(os.Getenv(key)), 10, 16)
	if err != nil {
		return 0
	}
	return uint16(value)
}

func liveICEServers() []webrtc.ICEServer {
	value := strings.TrimSpace(os.Getenv("KRA_LIVE_STUN_URL"))
	if value == "" {
		return nil
	}
	return []webrtc.ICEServer{{URLs: []string{value}}}
}

func New(hub platformws.Hub, streams *bizlive.StreamUsecase, rooms *bizlive.RoomUsecase, chat *bizchat.Usecase, logger *slog.Logger) (*Manager, func(), error) {
	if hub == nil {
		return nil, nil, errors.New("live websocket hub is unavailable")
	}
	if streams == nil {
		return nil, nil, errors.New("live stream usecase is unavailable")
	}
	if rooms == nil {
		return nil, nil, errors.New("live room usecase is unavailable")
	}
	if chat == nil {
		return nil, nil, errors.New("live chat usecase is unavailable")
	}
	api, err := newPionAPI()
	if err != nil {
		return nil, nil, err
	}
	m := &Manager{hub: hub, streams: streams, roomCatalog: rooms, chat: chat, logger: logger, api: api, rooms: make(map[string]*room), peers: make(map[*melody.Session]*peer), observers: make(map[*melody.Session]struct{}), stop: make(chan struct{})}
	hub.OnMessage(m.onMessage)
	hub.OnDisconnect(m.onDisconnect)
	go m.heartbeatLoop()
	return m, m.Close, nil
}

func (m *Manager) Close() {
	m.stopOnce.Do(func() { close(m.stop) })
	m.mu.Lock()
	peers := make([]*peer, 0, len(m.peers))
	for _, p := range m.peers {
		peers = append(peers, p)
	}
	m.mu.Unlock()
	for _, p := range peers {
		m.closePeer(p, "server_shutdown")
	}
}

func (m *Manager) onMessage(session *melody.Session, raw []byte) {
	var msg message
	if err := json.Unmarshal(raw, &msg); err != nil {
		m.send(session, "Error", map[string]any{"code": "INVALID_JSON", "message": "invalid json"})
		return
	}
	switch msg.Type {
	case "Join":
		var payload joinPayload
		if err := json.Unmarshal(msg.Payload, &payload); err != nil {
			m.send(session, "JoinRejected", map[string]any{"code": "INVALID_JOIN", "message": "invalid join payload"})
			return
		}
		m.join(session, payload)
	case "Offer":
		var payload sessionDescriptionPayload
		if err := json.Unmarshal(msg.Payload, &payload); err != nil {
			m.send(session, "Error", map[string]any{"code": "INVALID_OFFER", "message": "invalid offer payload"})
			return
		}
		m.handleOffer(session, payload)
	case "Heartbeat":
		if p := m.peer(session); p != nil {
			p.lastBeat.Store(time.Now().UnixMilli())
		}
		m.send(session, "HeartbeatAck", map[string]any{"streamId": peerStreamID(m, session)})
	case "Leave":
		if p := m.peer(session); p != nil {
			m.closePeer(p, "left")
		}
	case "ChatSend":
		var payload chatSendPayload
		if err := json.Unmarshal(msg.Payload, &payload); err != nil {
			m.send(session, "Error", map[string]any{"code": "INVALID_CHAT", "message": "invalid chat payload"})
			return
		}
		m.handleChat(session, payload)
	default:
		m.send(session, "Error", map[string]any{"code": "UNKNOWN_MESSAGE", "message": "unknown message type"})
	}
}

func (m *Manager) join(session *melody.Session, payload joinPayload) {
	role := strings.ToLower(strings.TrimSpace(payload.Role))
	streamID := strings.TrimSpace(payload.StreamID)
	if role == "observer" {
		m.mu.Lock()
		m.observers[session] = struct{}{}
		m.mu.Unlock()
		m.send(session, "Joined", map[string]any{"role": "observer"})
		m.sendStreamList(session)
		return
	}
	if streamID == "" || (role != "publisher" && role != "subscriber") {
		m.send(session, "JoinRejected", map[string]any{"code": "INVALID_JOIN", "message": "role and streamId are required"})
		return
	}
	userID, authenticated := sessionUserID(session)
	if role == "publisher" && !authenticated {
		m.send(session, "JoinRejected", map[string]any{"code": "AUTH_REQUIRED", "message": "authenticated user is required"})
		return
	}
	if role == "subscriber" && !authenticated {
		userID = m.anonymousViewerID()
	}
	if existing := m.peer(session); existing != nil {
		m.closePeer(existing, "rejoin")
	}
	if role == "publisher" {
		if err := m.authorizeRoomPublisher(streamID, userID); err != nil {
			m.send(session, "JoinRejected", map[string]any{"code": "PUBLISH_FORBIDDEN", "message": err.Error()})
			return
		}
	} else if err := m.authorizeRoomSubscriber(streamID); err != nil {
		m.send(session, "JoinRejected", map[string]any{"code": "ROOM_UNAVAILABLE", "message": err.Error()})
		return
	}
	r := m.getOrCreateRoom(streamID)
	if role == "publisher" {
		r.mu.RLock()
		occupied := r.publisher != nil
		r.mu.RUnlock()
		if occupied {
			m.send(session, "JoinRejected", map[string]any{"code": "DUPLICATE_PUBLISHER", "message": "stream already has a publisher"})
			return
		}
		if _, err := m.streams.RegisterPublisher(context.Background(), streamID, userID); err != nil {
			m.send(session, "JoinRejected", map[string]any{"code": errorCode(err), "message": err.Error()})
			return
		}
		if roomID, ok := roomIDFromStream(streamID); ok {
			_ = m.chat.Clear(context.Background(), roomID)
		}
		p := &peer{session: session, manager: m, room: r, streamID: streamID, userID: userID, role: role, authenticated: true}
		p.lastBeat.Store(time.Now().UnixMilli())
		r.mu.Lock()
		r.publisher = p
		r.mu.Unlock()
		m.mu.Lock()
		m.peers[session] = p
		m.mu.Unlock()
		m.send(session, "Joined", map[string]any{"role": role, "streamId": streamID, "mode": "pion-sfu"})
		m.sendPublisherStatus(streamID, false, "publisher_waiting_for_offer")
		m.broadcastStreamList()
		return
	}

	// Match the old behavior: a user can only watch a room once. The old
	// connection is cleaned up before registering the new one.
	var old *peer
	r.mu.RLock()
	for viewer := range r.viewers {
		if viewer.userID == userID {
			old = viewer
			break
		}
	}
	r.mu.RUnlock()
	if old != nil {
		m.send(old.session, "Kicked", map[string]any{"reason": "logged_in_elsewhere", "streamId": streamID})
		m.closePeer(old, "logged_in_elsewhere")
	}
	if _, err := m.streams.RegisterSubscriber(context.Background(), streamID, userID); err != nil {
		m.send(session, "JoinRejected", map[string]any{"code": errorCode(err), "message": err.Error()})
		return
	}
	p := &peer{session: session, manager: m, room: r, streamID: streamID, userID: userID, role: role, authenticated: authenticated, subscriberID: "sub_" + strconv.FormatUint(m.nextSub.Add(1), 10)}
	p.lastBeat.Store(time.Now().UnixMilli())
	r.mu.Lock()
	r.viewers[p] = struct{}{}
	ready := r.publisher != nil && len(r.tracks) > 0
	r.mu.Unlock()
	m.mu.Lock()
	m.peers[session] = p
	m.mu.Unlock()
	publicUserID := userID
	if !authenticated {
		publicUserID = 0
	}
	m.send(session, "Joined", map[string]any{"role": role, "streamId": streamID, "subscriberId": p.subscriberID, "userId": publicUserID, "mode": "pion-sfu"})
	m.sendPublisherStatusTo(session, streamID, ready, "publisher_offline")
	if ready {
		m.send(session, "PublisherStatus", map[string]any{"streamId": streamID, "ready": true})
	}
	if publisher := m.publisher(r); publisher != nil {
		m.send(publisher.session, "NewSubscriber", map[string]any{"streamId": streamID, "subscriberId": p.subscriberID, "userId": publicUserID})
	}
	m.broadcastStreamList()
}

func (m *Manager) handleChat(session *melody.Session, payload chatSendPayload) {
	p := m.peer(session)
	if p == nil || (p.role != "publisher" && p.role != "subscriber") {
		m.send(session, "Error", map[string]any{"code": "NOT_JOINED", "message": "join a room before chatting"})
		return
	}
	if !p.authenticated {
		m.send(session, "Error", map[string]any{"code": "AUTH_REQUIRED", "message": "authenticated user is required for chat"})
		return
	}
	roomID, ok := roomIDFromStream(p.streamID)
	if !ok {
		m.send(session, "Error", map[string]any{"code": "CHAT_UNAVAILABLE", "message": "chat is only available in live rooms"})
		return
	}
	available, err := m.roomCatalog.IsLive(context.Background(), roomID)
	if err != nil || !available {
		m.send(session, "Error", map[string]any{"code": "ROOM_UNAVAILABLE", "message": "live room is unavailable"})
		return
	}
	item, err := m.chat.Send(context.Background(), roomID, p.userID, payload.Content)
	if err != nil {
		m.send(session, "Error", map[string]any{"code": "INVALID_CHAT", "message": err.Error()})
		return
	}
	m.broadcastRoom(p.streamID, "ChatMessage", chatMessagePayload(item))
}

// GiftSent implements gift.Notifier. Gift persistence and wallet debit have
// already committed; chat persistence is therefore best-effort and must never
// turn a successful debit into a failed gift response.
func (m *Manager) GiftSent(ctx context.Context, sent *bizgift.Send) {
	if sent == nil || sent.RoomID == nil || sent.GiftID == nil {
		return
	}
	item, err := m.chat.RecordGift(ctx, &bizchat.GiftSnapshot{RoomID: *sent.RoomID, UserID: sent.UserID, Username: sent.Username, Nickname: sent.Nickname, GiftID: *sent.GiftID, GiftName: sent.GiftName, GiftIcon: sent.GiftIcon, Quantity: sent.Quantity, DiamondTotal: sent.DiamondTotal})
	if err != nil {
		m.log("record live gift chat failed", "room_id", *sent.RoomID, "gift_send_id", sent.ID, "error", err)
		return
	}
	m.broadcastRoom("room-"+strconv.FormatUint(uint64(*sent.RoomID), 10), "GiftMessage", chatMessagePayload(item))
}

func (m *Manager) broadcastRoom(streamID, msgType string, payload any) {
	m.mu.RLock()
	peers := make([]*peer, 0)
	for _, p := range m.peers {
		if p.streamID == streamID && (p.role == "publisher" || p.role == "subscriber") {
			peers = append(peers, p)
		}
	}
	m.mu.RUnlock()
	for _, p := range peers {
		m.send(p.session, msgType, payload)
	}
}

func chatMessagePayload(item *bizchat.Message) map[string]any {
	return map[string]any{"id": item.ID, "roomId": item.RoomID, "type": item.Type, "userId": item.UserID, "username": item.Username, "nickname": item.Nickname, "avatar": item.Avatar, "content": item.Content, "giftId": item.GiftID, "giftName": item.GiftName, "giftIcon": item.GiftIcon, "quantity": item.Quantity, "diamondTotal": item.DiamondTotal, "createdAt": item.CreatedAt}
}

func roomIDFromStream(streamID string) (uint, bool) {
	if !strings.HasPrefix(streamID, "room-") {
		return 0, false
	}
	value, err := strconv.ParseUint(strings.TrimPrefix(streamID, "room-"), 10, 64)
	return uint(value), err == nil && value > 0
}

func (m *Manager) authorizeRoomPublisher(streamID string, userID uint) error {
	if !strings.HasPrefix(streamID, "room-") {
		return nil
	}
	id, err := strconv.ParseUint(strings.TrimPrefix(streamID, "room-"), 10, 64)
	if err != nil || id == 0 {
		return errors.New("invalid live room stream id")
	}
	item, err := m.roomCatalog.GetPublic(context.Background(), uint(id))
	if err != nil {
		return errors.New("live room is unavailable")
	}
	if item.AnchorUserID != userID {
		return errors.New("current user is not the room anchor")
	}
	return nil
}

func (m *Manager) authorizeRoomSubscriber(streamID string) error {
	if strings.HasPrefix(streamID, "room-") {
		id, err := strconv.ParseUint(strings.TrimPrefix(streamID, "room-"), 10, 64)
		if err != nil || id == 0 {
			return errors.New("invalid live room stream id")
		}
		if _, err = m.roomCatalog.GetPublic(context.Background(), uint(id)); err != nil {
			return errors.New("live room is unavailable")
		}
	}
	stream, err := m.streams.Get(context.Background(), streamID)
	if err != nil || stream == nil || !stream.PublisherOnline {
		return errors.New("publisher is offline")
	}
	return nil
}

func (m *Manager) handleOffer(session *melody.Session, payload sessionDescriptionPayload) {
	p := m.peer(session)
	if p == nil || (p.role != "publisher" && p.role != "subscriber") {
		m.send(session, "Error", map[string]any{"code": "NOT_JOINED", "message": "join a stream first"})
		return
	}
	if strings.TrimSpace(payload.SDP) == "" {
		m.send(session, "Error", map[string]any{"code": "INVALID_SDP", "message": "sdp is required"})
		return
	}
	pc, err := m.api.NewPeerConnection(webrtc.Configuration{ICEServers: liveICEServers()})
	if err != nil {
		m.send(session, "Error", map[string]any{"code": "PEER_CONNECTION", "message": err.Error()})
		return
	}
	oldPC, installed := p.replacePeerConnection(pc)
	if !installed {
		_ = pc.Close()
		return
	}
	pc.OnICEConnectionStateChange(func(state webrtc.ICEConnectionState) {
		m.handleICEConnectionStateChange(p, pc, state)
	})
	if oldPC != nil {
		_ = oldPC.Close()
	}
	if p.role == "publisher" {
		pc.OnTrack(func(track *webrtc.TrackRemote, _ *webrtc.RTPReceiver) {
			m.onPublisherTrack(p, track)
		})
	} else {
		for _, track := range m.trackSnapshot(p.room) {
			if sender, addErr := pc.AddTrack(track.local); addErr == nil {
				go drainRTCP(sender)
			}
		}
	}
	if err = pc.SetRemoteDescription(webrtc.SessionDescription{Type: webrtc.SDPTypeOffer, SDP: payload.SDP}); err != nil {
		m.send(session, "Error", map[string]any{"code": "REMOTE_DESCRIPTION", "message": err.Error()})
		return
	}
	answer, err := pc.CreateAnswer(nil)
	if err != nil {
		m.send(session, "Error", map[string]any{"code": "CREATE_ANSWER", "message": err.Error()})
		return
	}
	if err = pc.SetLocalDescription(answer); err != nil {
		m.send(session, "Error", map[string]any{"code": "LOCAL_DESCRIPTION", "message": err.Error()})
		return
	}
	<-webrtc.GatheringCompletePromise(pc)
	local := pc.LocalDescription()
	if local == nil {
		m.send(session, "Error", map[string]any{"code": "NO_LOCAL_DESCRIPTION", "message": "local description unavailable"})
		return
	}
	m.send(session, "Answer", map[string]any{"type": local.Type.String(), "sdp": local.SDP})
	if p.role == "publisher" {
		m.sendPublisherStatus(p.streamID, false, "waiting_for_tracks")
	}
}

func (m *Manager) handleICEConnectionStateChange(p *peer, pc *webrtc.PeerConnection, state webrtc.ICEConnectionState) {
	if state != webrtc.ICEConnectionStateFailed && state != webrtc.ICEConnectionStateClosed {
		return
	}
	m.closePeerConnection(p, pc, "ice_"+state.String())
}

func (m *Manager) onPublisherTrack(p *peer, remote *webrtc.TrackRemote) {
	codec := remote.Codec().RTPCodecCapability
	local, err := webrtc.NewTrackLocalStaticRTP(codec, remote.ID(), p.streamID)
	if err != nil {
		m.log("live track creation failed", "error", err)
		return
	}
	entry := &publishedTrack{id: remote.ID(), kind: remote.Kind(), local: local}
	p.room.mu.Lock()
	p.room.tracks[remote.Kind()] = entry
	viewers := make([]*peer, 0, len(p.room.viewers))
	for viewer := range p.room.viewers {
		viewers = append(viewers, viewer)
	}
	p.room.mu.Unlock()
	m.sendPublisherStatus(p.streamID, true, "")
	for _, viewer := range viewers {
		m.send(viewer.session, "NewProducer", map[string]any{"streamId": p.streamID, "producerId": remote.ID(), "kind": strings.ToLower(remote.Kind().String())})
		m.send(viewer.session, "Renegotiate", map[string]any{"streamId": p.streamID})
	}
	go func() {
		for {
			packet, _, readErr := remote.ReadRTP()
			if readErr != nil {
				return
			}
			_ = local.WriteRTP(packet)
		}
	}()
}

func (m *Manager) onDisconnect(session *melody.Session) {
	if p := m.peer(session); p != nil {
		m.closePeer(p, "disconnected")
		return
	}
	m.mu.Lock()
	delete(m.observers, session)
	m.mu.Unlock()
}

func (m *Manager) heartbeatLoop() {
	ticker := time.NewTicker(10 * time.Second)
	defer ticker.Stop()
	for {
		select {
		case <-ticker.C:
			now := time.Now().UnixMilli()
			m.mu.RLock()
			peers := make([]*peer, 0, len(m.peers))
			for _, p := range m.peers {
				peers = append(peers, p)
			}
			m.mu.RUnlock()
			for _, p := range peers {
				if now-p.lastBeat.Load() >= (45 * time.Second).Milliseconds() {
					m.closePeer(p, "heartbeat_timeout")
				}
			}
		case <-m.stop:
			return
		}
	}
}

func (m *Manager) closePeer(p *peer, reason string) {
	m.closePeerConnection(p, nil, reason)
}

func (m *Manager) closePeerConnection(p *peer, candidate *webrtc.PeerConnection, reason string) {
	if p == nil {
		return
	}
	pc, ok := p.beginClose(candidate)
	if !ok {
		return
	}
	p.cleanup.Do(func() {
		if pc != nil {
			_ = pc.Close()
		}
		if p.role == "publisher" {
			_ = m.streams.UnregisterPublisher(context.Background(), p.streamID, p.userID)
			if roomID, ok := roomIDFromStream(p.streamID); ok {
				_ = m.chat.Clear(context.Background(), roomID)
			}
			p.room.mu.Lock()
			tracks := make([]*publishedTrack, 0, len(p.room.tracks))
			for _, track := range p.room.tracks {
				tracks = append(tracks, track)
			}
			if p.room.publisher == p {
				p.room.publisher = nil
			}
			p.room.tracks = make(map[webrtc.RTPCodecType]*publishedTrack)
			viewers := make([]*peer, 0, len(p.room.viewers))
			for viewer := range p.room.viewers {
				viewers = append(viewers, viewer)
			}
			p.room.mu.Unlock()
			for _, viewer := range viewers {
				for _, track := range tracks {
					m.send(viewer.session, "ProducerClosed", map[string]any{"streamId": p.streamID, "producerId": track.id, "kind": strings.ToLower(track.kind.String()), "reason": reason})
				}
				m.send(viewer.session, "PublisherStatus", map[string]any{"streamId": p.streamID, "ready": false, "reason": reason})
			}
		} else if p.role == "subscriber" {
			_ = m.streams.UnregisterSubscriber(context.Background(), p.streamID, p.userID)
			p.room.mu.Lock()
			delete(p.room.viewers, p)
			publisher := p.room.publisher
			p.room.mu.Unlock()
			if publisher != nil {
				m.send(publisher.session, "SubscriberLeft", map[string]any{"streamId": p.streamID, "subscriberId": p.subscriberID})
			}
		}
		m.mu.Lock()
		delete(m.peers, p.session)
		p.room.mu.RLock()
		emptyRoom := p.room.publisher == nil && len(p.room.viewers) == 0
		p.room.mu.RUnlock()
		if emptyRoom {
			delete(m.rooms, p.streamID)
		}
		m.mu.Unlock()
		m.broadcastStreamList()
	})
}

func (p *peer) replacePeerConnection(next *webrtc.PeerConnection) (*webrtc.PeerConnection, bool) {
	p.pcMu.Lock()
	defer p.pcMu.Unlock()
	if p.closing {
		return nil, false
	}
	previous := p.pc
	p.pc = next
	return previous, true
}

func (p *peer) hasPeerConnection(candidate *webrtc.PeerConnection) bool {
	p.pcMu.RLock()
	current := p.pc
	p.pcMu.RUnlock()
	return candidate != nil && current == candidate
}

func (p *peer) beginClose(candidate *webrtc.PeerConnection) (*webrtc.PeerConnection, bool) {
	p.pcMu.Lock()
	defer p.pcMu.Unlock()
	if p.closing || candidate != nil && p.pc != candidate {
		return nil, false
	}
	p.closing = true
	current := p.pc
	p.pc = nil
	return current, true
}

func (m *Manager) getOrCreateRoom(streamID string) *room {
	m.mu.Lock()
	defer m.mu.Unlock()
	if r := m.rooms[streamID]; r != nil {
		return r
	}
	r := &room{streamID: streamID, viewers: make(map[*peer]struct{}), tracks: make(map[webrtc.RTPCodecType]*publishedTrack)}
	m.rooms[streamID] = r
	return r
}

func (m *Manager) peer(session *melody.Session) *peer {
	m.mu.RLock()
	defer m.mu.RUnlock()
	return m.peers[session]
}

func (m *Manager) publisher(r *room) *peer {
	r.mu.RLock()
	defer r.mu.RUnlock()
	return r.publisher
}

func (m *Manager) trackSnapshot(r *room) []*publishedTrack {
	r.mu.RLock()
	defer r.mu.RUnlock()
	tracks := make([]*publishedTrack, 0, len(r.tracks))
	for _, track := range r.tracks {
		tracks = append(tracks, track)
	}
	return tracks
}

func (m *Manager) sendStreamList(session *melody.Session) {
	streams, err := m.streams.List(context.Background())
	if err != nil {
		return
	}
	items := make([]map[string]any, 0, len(streams))
	for _, stream := range streams {
		items = append(items, map[string]any{"streamId": stream.StreamID, "publisherOnline": stream.PublisherOnline, "subscriberCount": stream.SubscriberCount})
	}
	m.send(session, "StreamListUpdate", map[string]any{"streams": items})
}

func (m *Manager) broadcastStreamList() {
	m.mu.RLock()
	observers := make([]*melody.Session, 0, len(m.observers))
	for session := range m.observers {
		observers = append(observers, session)
	}
	m.mu.RUnlock()
	for _, session := range observers {
		m.sendStreamList(session)
	}
}

func (m *Manager) sendPublisherStatus(streamID string, ready bool, reason string) {
	m.mu.RLock()
	peers := make([]*peer, 0, len(m.peers))
	for _, p := range m.peers {
		if p.role == "subscriber" && p.streamID == streamID {
			peers = append(peers, p)
		}
	}
	m.mu.RUnlock()
	for _, p := range peers {
		m.sendPublisherStatusTo(p.session, streamID, ready, reason)
	}
}

func (m *Manager) sendPublisherStatusTo(session *melody.Session, streamID string, ready bool, reason string) {
	payload := map[string]any{"streamId": streamID, "ready": ready}
	if reason != "" {
		payload["reason"] = reason
	}
	m.send(session, "PublisherStatus", payload)
}

func (m *Manager) send(session *melody.Session, msgType string, payload any) {
	encoded, err := json.Marshal(map[string]any{"type": msgType, "payload": payload})
	if err != nil {
		return
	}
	_ = m.hub.Send(session, encoded)
}

func (m *Manager) log(message string, args ...any) {
	if m.logger != nil {
		m.logger.Warn(message, args...)
	}
}

func sessionUserID(session *melody.Session) (uint, bool) {
	if session == nil {
		return 0, false
	}
	value, ok := session.Keys["user_id"]
	if !ok || value == nil {
		return 0, false
	}
	switch typed := value.(type) {
	case uint:
		return typed, typed > 0
	case uint64:
		return uint(typed), typed > 0
	case int:
		return uint(typed), typed > 0
	case int64:
		return uint(typed), typed > 0
	case string:
		parsed, err := strconv.ParseUint(typed, 10, 64)
		return uint(parsed), err == nil && parsed > 0
	default:
		parsed, err := strconv.ParseUint(fmt.Sprint(value), 10, 64)
		return uint(parsed), err == nil && parsed > 0
	}
}

func (m *Manager) anonymousViewerID() uint {
	// Allocate downward from the maximum uint so anonymous connections never
	// share the zero ID and cannot collide with ordinary auto-increment users.
	return ^uint(0) - uint(m.nextGuest.Add(1))
}

func peerStreamID(m *Manager, session *melody.Session) string {
	if p := m.peer(session); p != nil {
		return p.streamID
	}
	return ""
}

func errorCode(err error) string {
	switch {
	case errors.Is(err, bizlive.ErrMaxStreams):
		return "MAX_STREAMS"
	case errors.Is(err, bizlive.ErrPublisherExists):
		return "DUPLICATE_PUBLISHER"
	case errors.Is(err, bizlive.ErrSubscriberExists):
		return "DUPLICATE_SUBSCRIBER"
	default:
		return "INVALID_JOIN"
	}
}

func drainRTCP(sender *webrtc.RTPSender) {
	for {
		if _, _, err := sender.ReadRTCP(); err != nil {
			return
		}
	}
}