qcc/internal/call/signaling.go

package call

import (
	"encoding/binary"
	"sync"
	"time"

	"github.com/apernet/quic-go"
	"go.uber.org/zap"

	"github.com/niko/qcc/internal/identity"
	"github.com/niko/qcc/internal/protocol"
	"github.com/niko/qcc/pkg/types"
)

type CallSession interface {
	Number() string
	PubKey() [32]byte
	Stream() protocol.FrameReadWriter
	Conn() *quic.Conn
	SetCallID(id uint64)
	CallID() uint64
	ClearCall()
	IsAuthenticated() bool
}

type SessionRegistry interface {
	GetByNumber(number string) CallSession
	GetAll() []CallSession
	Register(sess CallSession, number string)
	Unregister(number string)
}

type BrutalApplier func(conn *quic.Conn, bps uint64)

type Manager struct {
	mu           sync.RWMutex
	calls        map[uint64]*Call
	userCalls    map[string]uint64
	nextCallID   uint64
	logger       *zap.Logger
	sessions     SessionRegistry
	mediaTimeout time.Duration
	brutalBPS    uint64
	applyBrutal  BrutalApplier
	stopCh       chan struct{}
}

type Call struct {
	ID           uint64
	From         string
	To           string
	FromSess     CallSession
	ToSess       CallSession
	StartedAt    int64
	Active       bool
	FromE2EEKey [32]byte
	ToE2EEKey   [32]byte
}

func NewManager(logger *zap.Logger, sessions SessionRegistry, mediaTimeout time.Duration, brutalBPS uint64, applyBrutal BrutalApplier) *Manager {
	m := &Manager{
		calls:        make(map[uint64]*Call),
		userCalls:    make(map[string]uint64),
		nextCallID:   1,
		logger:       logger,
		sessions:     sessions,
		mediaTimeout: mediaTimeout,
		brutalBPS:    brutalBPS,
		applyBrutal:  applyBrutal,
		stopCh:       make(chan struct{}),
	}
	go m.cleanupLoop()
	return m
}

func (m *Manager) cleanupLoop() {
	ticker := time.NewTicker(30 * time.Second)
	defer ticker.Stop()
	for {
		select {
		case <-ticker.C:
			m.cleanupStale()
		case <-m.stopCh:
			return
		}
	}
}

func (m *Manager) Stop() {
	close(m.stopCh)
}

func (m *Manager) cleanupStale() {
	m.mu.Lock()
	defer m.mu.Unlock()
	for id, call := range m.calls {
		if !call.Active && time.Since(time.Unix(0, call.StartedAt)) > m.mediaTimeout {
			delete(m.calls, id)
			delete(m.userCalls, call.From)
			delete(m.userCalls, call.To)
		}
	}
}

func parseDialPayload(payload []byte) (targetNumber string, e2eePubKey [32]byte) {
	if len(payload) < 1 {
		return "", [32]byte{}
	}
	numLen := int(payload[0])
	if len(payload) < 1+numLen+32 {
		return "", [32]byte{}
	}
	targetNumber = string(payload[1 : 1+numLen])
	copy(e2eePubKey[:], payload[1+numLen:1+numLen+32])
	return targetNumber, e2eePubKey
}

func marshalRingPayload(number string, e2eePubKey [32]byte) []byte {
	buf := make([]byte, 1+len(number)+32)
	buf[0] = byte(len(number))
	copy(buf[1:], number)
	copy(buf[1+len(number):], e2eePubKey[:])
	return buf
}

func (m *Manager) HandleDial(sess CallSession, payload []byte) {
	targetNumber, callerE2EE := parseDialPayload(payload)
	if targetNumber == "" {
		sendError(sess, types.ErrInvalidRequest, "empty target number")
		return
	}
	origTarget := targetNumber
	targetNumber = identity.NormalizeNumber(targetNumber)

	fromNumber := sess.Number()
	m.logger.Debug("Dial request",
		zap.String("from", fromNumber),
		zap.String("target_orig", origTarget),
		zap.String("target_norm", targetNumber))

	if fromNumber == targetNumber {
		sendError(sess, types.ErrSelfCall, "cannot call yourself")
		return
	}

	m.mu.Lock()
	if _, busy := m.userCalls[fromNumber]; busy {
		m.mu.Unlock()
		sendError(sess, types.ErrBusy, "you already have an active call")
		return
	}

	targetSess := m.sessions.GetByNumber(targetNumber)
	if targetSess == nil {
		m.mu.Unlock()
		sendError(sess, types.ErrNumberNotFound, "target offline")
		return
	}

	if _, busy := m.userCalls[targetNumber]; busy {
		m.mu.Unlock()
		sendError(sess, types.ErrBusy, "target is busy")
		return
	}

	callID := m.nextCallID
	m.nextCallID++
	call := &Call{
		ID:           callID,
		From:         fromNumber,
		To:           targetNumber,
		FromSess:     sess,
		ToSess:       targetSess,
		StartedAt:    time.Now().UnixNano(),
		FromE2EEKey: callerE2EE,
	}
	m.calls[callID] = call
	m.userCalls[fromNumber] = callID
	m.userCalls[targetNumber] = callID
	m.mu.Unlock()

	sess.SetCallID(callID)
	targetSess.SetCallID(callID)

	ringPayload := marshalRingPayload(fromNumber, callerE2EE)
	if st := targetSess.Stream(); st != nil {
		st.WriteFrame(byte(types.OpRing), ringPayload)
	}

	m.logger.Info("Call initiated",
		zap.Uint64("call_id", callID),
		zap.String("from", fromNumber),
		zap.String("to", targetNumber))
}

func (m *Manager) HandleAccept(sess CallSession, payload []byte) {
	callID := sess.CallID()
	if callID == 0 {
		sendError(sess, types.ErrCallNotFound, "no active call")
		return
	}

	var calleeE2EE [32]byte
	if len(payload) >= 32 {
		copy(calleeE2EE[:], payload[:32])
	}

	m.mu.Lock()
	call, exists := m.calls[callID]
	if !exists {
		m.mu.Unlock()
		sendError(sess, types.ErrCallNotFound, "call not found")
		return
	}
	call.ToE2EEKey = calleeE2EE
	call.Active = true
	m.mu.Unlock()

	if m.brutalBPS > 0 && m.applyBrutal != nil {
		if c := call.FromSess.Conn(); c != nil {
			m.applyBrutal(c, m.brutalBPS)
		}
		if c := call.ToSess.Conn(); c != nil {
			m.applyBrutal(c, m.brutalBPS)
		}
	}

	peerPayload := make([]byte, 32)
	copy(peerPayload, calleeE2EE[:])

	if st := call.FromSess.Stream(); st != nil {
		st.WriteFrame(byte(types.OpPeerAccept), peerPayload)
	}

	m.logger.Info("Call accepted",
		zap.Uint64("call_id", callID),
		zap.Bool("e2ee", calleeE2EE != [32]byte{}))
}

func (m *Manager) HandleReject(sess CallSession, _ []byte) {
	m.endCall(sess, false)
}

func (m *Manager) HandleEnd(sess CallSession, _ []byte) {
	m.endCall(sess, true)
}

func (m *Manager) endCall(sess CallSession, notify bool) {
	callID := sess.CallID()
	if callID == 0 {
		return
	}

	m.mu.Lock()
	call, exists := m.calls[callID]
	if !exists {
		m.mu.Unlock()
		return
	}
	delete(m.calls, callID)
	delete(m.userCalls, call.From)
	delete(m.userCalls, call.To)
	m.mu.Unlock()

	sess.ClearCall()

	var otherSess CallSession
	if sess == call.FromSess {
		otherSess = call.ToSess
	} else {
		otherSess = call.FromSess
	}

	if otherSess != nil {
		otherSess.ClearCall()
		if notify {
			if st := otherSess.Stream(); st != nil {
				st.WriteFrame(byte(types.OpPeerEnd), nil)
			}
		}
	}

	m.logger.Info("Call ended",
		zap.Uint64("call_id", callID))
}

func (m *Manager) GetCall(callID uint64) *Call {
	m.mu.RLock()
	defer m.mu.RUnlock()
	return m.calls[callID]
}

func (m *Manager) RelayMedia(senderSess CallSession, data []byte) {
	callID := senderSess.CallID()
	m.mu.RLock()
	call, exists := m.calls[callID]
	m.mu.RUnlock()
	if !exists || !call.Active {
		return
	}

	var target *quic.Conn
	if senderSess == call.FromSess {
		target = call.ToSess.Conn()
	} else if senderSess == call.ToSess {
		target = call.FromSess.Conn()
	} else {
		return
	}

	if err := target.SendDatagram(data); err != nil {
		m.logger.Warn("Failed to relay media datagram",
			zap.Uint64("call_id", callID),
			zap.Error(err))
	}
}

func sendError(sess CallSession, code types.ErrorCode, msg string) {
	payload := make([]byte, 2+len(msg))
	binary.BigEndian.PutUint16(payload[0:2], uint16(code))
	copy(payload[2:], msg)
	if st := sess.Stream(); st != nil {
		st.WriteFrame(byte(types.OpError), payload)
	}
}