im_server/server/calls.go

/******************************************************************************
 *
 *  Description :
 *    Video call handling (establishment, metadata exhange and termination).
 *
 *****************************************************************************/
package main

import (
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"strconv"
	"strings"
	"time"

	"github.com/tinode/chat/server/logs"
	"github.com/tinode/chat/server/store"
	"github.com/tinode/chat/server/store/types"
	jcr "github.com/tinode/jsonco"
)

// Video call constants.
const (
	// Events for call between users A and B.
	//
	// B has received the call but hasn't picked it up yet.
	constCallEventRinging = "ringing"
	// B has accepted the call.
	constCallEventAccept = "accept"
	// WebRTC SDP & ICE data exchange events.
	constCallEventOffer        = "offer"
	constCallEventAnswer       = "answer"
	constCallEventIceCandidate = "ice-candidate"
	// Call finished by either side or server.
	constCallEventHangUp = "hang-up"

	// Message headers representing call states.
	// Call is established.
	constCallMsgAccepted = "accepted"
	// Previously establied call has successfully finished.
	constCallMsgFinished = "finished"
	// Call is dropped (e.g. because of an error).
	constCallMsgDisconnected = "disconnected"
	// Call is missed (the callee didn't pick up the phone).
	constCallMsgMissed = "missed"
	// Call is declined (the callee hung up before picking up).
	constCallMsgDeclined = "declined"

	constCallTypeAudio = "audio"
	constCallTypeVideo = "video"
)

type callConfig struct {
	// Enable video/voice calls.
	Enabled bool `json:"enabled"`
	// Timeout in seconds before a call is dropped if not answered.
	CallEstablishmentTimeout int `json:"call_establishment_timeout"`
	// ICE servers.
	ICEServers []iceServer `json:"ice_servers"`
	// Alternative config as an external file.
	ICEServersFile string `json:"ice_servers_file"`
}

// ICE server config.
type iceServer struct {
	Username       string   `json:"username,omitempty"`
	Credential     string   `json:"credential,omitempty"`
	CredentialType string   `json:"credential_type,omitempty"`
	Urls           []string `json:"urls,omitempty"`
}

// callPartyData describes a video call participant.
type callPartyData struct {
	// ID of the call participant (asUid); not necessarily the session owner.
	uid types.Uid
	// True if this session/user initiated the call.
	isOriginator bool
	// Call party session.
	sess *Session
}

// videoCall describes video call that's being established or in progress.
type videoCall struct {
	// Call participants (session sid -> callPartyData).
	parties map[string]callPartyData
	// Call message seq ID.
	seq int64
	// Call message content.
	content any
	// Call message content mime type.
	contentMime any
	// Product-level call identifier.
	callID string
	// Product-level call type: audio or video.
	callType string
	// Time when the invite was accepted by the server.
	startedAt time.Time
	// Time when the call was accepted.
	acceptedAt time.Time
}

func normalizeCallInvite(head map[string]any, content any) (map[string]any, any, error) {
	if head == nil {
		return nil, nil, types.ErrMalformed
	}
	state, _ := head["webrtc"].(string)
	if state != "started" {
		return nil, nil, types.ErrMalformed
	}
	call := callInvitePayloadMap(content)
	if len(call) == 0 || !hasCallIdentity(call) {
		headCall := callInvitePayloadMap(head)
		if len(call) == 0 {
			call = headCall
		} else {
			for key, value := range headCall {
				if _, exists := call[key]; !exists {
					call[key] = value
				}
			}
			normalizeCallPayloadAliases(call)
		}
	}
	if len(call) == 0 {
		return nil, nil, types.ErrMalformed
	}
	callID, _ := call["callId"].(string)
	callType, _ := call["callType"].(string)
	if callType == "" {
		callType, _ = call["media_type"].(string)
	}
	if callType == "" {
		callType, _ = call["type"].(string)
	}
	if callType == "" {
		if aonly, _ := call["aonly"].(bool); aonly {
			callType = constCallTypeAudio
		} else {
			callType = constCallTypeVideo
		}
	}
	callType = strings.ToLower(strings.TrimSpace(callType))
	if strings.TrimSpace(callID) == "" || (callType != constCallTypeAudio && callType != constCallTypeVideo) {
		return nil, nil, types.ErrMalformed
	}
	normalizedHead := make(map[string]any, len(head)+2)
	for key, value := range head {
		normalizedHead[key] = value
	}
	normalizedHead["webrtc"] = "started"
	normalizedHead["x-im-type"] = "call"
	if _, ok := normalizedHead["mime"]; !ok {
		normalizedHead["mime"] = "application/json"
	}
	normalizedContent := make(map[string]any, len(call)+3)
	for key, value := range call {
		normalizedContent[key] = value
	}
	normalizedContent["callId"] = strings.TrimSpace(callID)
	normalizedContent["callType"] = callType
	normalizedContent["audio"] = true
	normalizedContent["video"] = callType == constCallTypeVideo
	return normalizedHead, normalizedContent, nil
}

func callInvitePayloadMap(content any) map[string]any {
	call := anyMap(content)
	if len(call) == 0 {
		return nil
	}
	normalizeCallPayloadAliases(call)
	if hasCallIdentity(call) {
		return call
	}
	if draftyCall := draftyVideoCallPayload(call); len(draftyCall) > 0 {
		for key, value := range call {
			if _, exists := draftyCall[key]; !exists && key != "fmt" && key != "ent" && key != "txt" {
				draftyCall[key] = value
			}
		}
		normalizeCallPayloadAliases(draftyCall)
		return draftyCall
	}
	return call
}

func anyMap(value any) map[string]any {
	if typed, ok := value.(map[string]any); ok {
		return typed
	}
	if typed, ok := value.(types.KVMap); ok {
		return map[string]any(typed)
	}
	return nil
}

func normalizeCallPayloadAliases(call map[string]any) {
	if call == nil {
		return
	}
	if _, ok := call["callId"]; !ok {
		for _, key := range []string{"call_id", "id"} {
			if value, exists := call[key]; exists {
				call["callId"] = value
				break
			}
		}
	}
	if _, ok := call["callType"]; !ok {
		for _, key := range []string{"call_type", "mediaType", "media_type"} {
			if value, exists := call[key]; exists {
				call["callType"] = value
				break
			}
		}
	}
}

func hasCallIdentity(call map[string]any) bool {
	if call == nil {
		return false
	}
	callID, _ := call["callId"].(string)
	if strings.TrimSpace(callID) == "" {
		return false
	}
	callType, _ := call["callType"].(string)
	if strings.TrimSpace(callType) != "" {
		return true
	}
	if mediaType, _ := call["media_type"].(string); strings.TrimSpace(mediaType) != "" {
		return true
	}
	if _, ok := call["aonly"].(bool); ok {
		return true
	}
	return false
}

func draftyVideoCallPayload(doc map[string]any) map[string]any {
	for _, raw := range anySlice(doc["ent"]) {
		entity := anyMap(raw)
		if entityType, _ := entity["tp"].(string); entityType != "VC" {
			continue
		}
		if data := anyMap(entity["data"]); len(data) > 0 {
			return data
		}
		return map[string]any{}
	}
	for _, raw := range anySlice(doc["fmt"]) {
		format := anyMap(raw)
		if formatType, _ := format["tp"].(string); formatType == "VC" {
			return map[string]any{}
		}
	}
	return nil
}

func anySlice(value any) []any {
	switch typed := value.(type) {
	case []any:
		return typed
	case []map[string]any:
		items := make([]any, 0, len(typed))
		for _, item := range typed {
			items = append(items, item)
		}
		return items
	case []types.KVMap:
		items := make([]any, 0, len(typed))
		for _, item := range typed {
			items = append(items, item)
		}
		return items
	default:
		return nil
	}
}

func callPayloadString(payload any, key string) string {
	data, ok := payload.(map[string]any)
	if !ok {
		if typed, typedOK := payload.(types.KVMap); typedOK {
			data = map[string]any(typed)
		}
	}
	if len(data) == 0 {
		return ""
	}
	value, _ := data[key].(string)
	return strings.TrimSpace(value)
}

func callContentString(content any, key string) string {
	return callPayloadString(content, key)
}

func normalizeCallEvent(event string) string {
	switch strings.ToLower(strings.TrimSpace(event)) {
	case "hangup", "hang_up", "hang-up", "finish", "finished", "end", "ended", "disconnect", "disconnected":
		return constCallEventHangUp
	default:
		return strings.TrimSpace(event)
	}
}

func isDetachedCallEvent(event string) bool {
	switch normalizeCallEvent(event) {
	case constCallEventRinging, constCallEventHangUp, constCallEventAccept:
		return true
	default:
		return false
	}
}

func (call *videoCall) matchesPayload(payload any) bool {
	payloadCallID := callPayloadString(payload, "callId")
	return payloadCallID == "" || payloadCallID == call.callID
}

func (call *videoCall) isStaleUnanswered(now time.Time, timeoutSeconds int) bool {
	if call == nil || len(call.parties) != 1 || call.startedAt.IsZero() || timeoutSeconds <= 0 {
		return false
	}
	return now.Sub(call.startedAt) > time.Duration(timeoutSeconds)*time.Second
}

func (t *Topic) auditCallCreate(call *videoCall, caller types.Uid) {
	if store.Calls == nil || call == nil || call.callID == "" || caller.IsZero() {
		return
	}
	callee := t.p2pOtherUser(caller)
	if callee.IsZero() {
		return
	}
	now := types.TimeNow()
	if err := store.Calls.Create(t.tenantID, &types.CallRecord{
		TenantID:  t.tenantID,
		CallID:    call.callID,
		Topic:     t.name,
		SeqID:     call.seq,
		CallerID:  caller,
		CalleeID:  callee,
		CallType:  call.callType,
		Status:    types.CallStatusStarted,
		StartedAt: now,
		CreatedAt: now,
		UpdatedAt: now,
	}); err != nil {
		logs.Warn.Printf("topic[%s]: failed to audit call start %s: %v", t.name, call.callID, err)
	}
}

func (t *Topic) auditCallAccept(call *videoCall, acceptedAt time.Time) {
	if store.Calls == nil || call == nil || call.callID == "" {
		return
	}
	if err := store.Calls.Accept(t.tenantID, call.callID, call.seq, acceptedAt); err != nil {
		logs.Warn.Printf("topic[%s]: failed to audit call accept %s: %v", t.name, call.callID, err)
	}
}

func (t *Topic) auditCallFinish(call *videoCall, status string, durationMS int64, finishedAt time.Time) {
	if store.Calls == nil || call == nil || call.callID == "" {
		return
	}
	if err := store.Calls.Finish(t.tenantID, call.callID, call.seq, status, durationMS, finishedAt); err != nil {
		logs.Warn.Printf("topic[%s]: failed to audit call finish %s: %v", t.name, call.callID, err)
	}
}

// callPartySession returns a session to be stored in the call party data.
func callPartySession(sess *Session) *Session {
	if sess.isProxy() {
		// We are on the topic host node. Make a copy of the ephemeral proxy session.
		callSess := &Session{
			proto: PROXY,
			// Multiplexing session which actually handles the communication.
			multi: sess.multi,
			// Local parameters specific to this session.
			sid:         sess.sid,
			userAgent:   sess.userAgent,
			remoteAddr:  sess.remoteAddr,
			lang:        sess.lang,
			countryCode: sess.countryCode,
			proxyReq:    ProxyReqCall,
			background:  sess.background,
			uid:         sess.uid,
		}
		return callSess
	}
	return sess
}

func initVideoCalls(jsconfig json.RawMessage) error {
	var config callConfig

	if len(jsconfig) == 0 {
		return nil
	}

	if err := json.Unmarshal([]byte(jsconfig), &config); err != nil {
		return fmt.Errorf("failed to parse config: %w", err)
	}

	if !config.Enabled {
		logs.Info.Println("Video calls disabled")
		return nil
	}

	if len(config.ICEServers) > 0 {
		globals.iceServers = config.ICEServers
	} else if config.ICEServersFile != "" {
		var iceConfig []iceServer
		file, err := os.Open(config.ICEServersFile)
		if err != nil {
			return fmt.Errorf("failed to read ICE config: %w", err)
		}

		jr := jcr.New(file)
		if err = json.NewDecoder(jr).Decode(&iceConfig); err != nil {
			switch jerr := err.(type) {
			case *json.UnmarshalTypeError:
				lnum, cnum, _ := jr.LineAndChar(jerr.Offset)
				return fmt.Errorf("unmarshall error in ICE config in %s at %d:%d (offset %d bytes): %w",
					jerr.Field, lnum, cnum, jerr.Offset, jerr)
			case *json.SyntaxError:
				lnum, cnum, _ := jr.LineAndChar(jerr.Offset)
				return fmt.Errorf("syntax error in config file at %d:%d (offset %d bytes): %w",
					lnum, cnum, jerr.Offset, jerr)
			default:
				return fmt.Errorf("failed to parse config file: %w", err)
			}
		}
		file.Close()

		globals.iceServers = iceConfig
	}

	if len(globals.iceServers) == 0 {
		return errors.New("no valid ICE cervers found")
	}

	globals.callEstablishmentTimeout = config.CallEstablishmentTimeout
	if globals.callEstablishmentTimeout <= 0 {
		globals.callEstablishmentTimeout = defaultCallEstablishmentTimeout
	}

	logs.Info.Println("Video calls enabled with", len(globals.iceServers), "ICE servers")
	return nil
}

// Add webRTC-related headers to message Head. The original Head may already contain some entries,
// like 'sender', preserve them.
func (call *videoCall) messageHead(head map[string]any, newState string, duration int) map[string]any {
	if head == nil {
		head = map[string]any{}
	}

	head["replace"] = ":" + strconv.FormatInt(call.seq, 10)
	head["webrtc"] = newState

	if duration > 0 {
		head["webrtc-duration"] = duration
	} else {
		delete(head, "webrtc-duration")
	}
	if call.contentMime != nil {
		head["mime"] = call.contentMime
	}
	return head
}

// Generates server info message template for the video call event.
func (call *videoCall) infoMessage(event string) *ServerComMessage {
	return &ServerComMessage{
		Info: &MsgServerInfo{
			What:  "call",
			Event: event,
			SeqId: call.seq,
		},
	}
}

// Returns Uid and session of the present video call originator
// if a call is being established or in progress.
func (t *Topic) getCallOriginator() (types.Uid, *Session) {
	if t.currentCall == nil {
		return types.ZeroUid, nil
	}
	for _, p := range t.currentCall.parties {
		if p.isOriginator {
			return p.uid, p.sess
		}
	}
	return types.ZeroUid, nil
}

// Handles video call invite (initiation)
// (in response to msg = {pub head=[mime: application/x-tinode-webrtc]}).
func (t *Topic) handleCallInvite(msg *ClientComMessage, asUid types.Uid) {
	// Call being establshed.
	t.currentCall = &videoCall{
		parties:     make(map[string]callPartyData),
		seq:         t.lastID,
		content:     msg.Pub.Content,
		contentMime: msg.Pub.Head["mime"],
		callID:      callContentString(msg.Pub.Content, "callId"),
		callType:    callContentString(msg.Pub.Content, "callType"),
		startedAt:   types.TimeNow(),
	}
	t.currentCall.parties[msg.sess.sid] = callPartyData{
		uid:          asUid,
		isOriginator: true,
		sess:         callPartySession(msg.sess),
	}
	t.auditCallCreate(t.currentCall, asUid)
	// Wait for constCallEstablishmentTimeout for the other side to accept the call.
	t.callEstablishmentTimer.Reset(time.Duration(imCallTimeoutSeconds()) * time.Second)
}

// Handles events on existing video call (acceptance, termination, metadata exchange).
// (in response to msg = {note what=call}).
func (t *Topic) handleCallEvent(msg *ClientComMessage) {
	if t.currentCall == nil {
		// Must initiate call first.
		logs.Warn.Printf("topic[%s]: No call in progress", t.name)
		return
	}
	if t.isInactive() {
		// Topic is paused or being deleted.
		return
	}

	call := msg.Note
	if t.currentCall.seq != call.SeqId {
		// Call not found.
		logs.Info.Printf("topic[%s]: invalid seq id - current call (%d) vs received (%d)", t.name, t.currentCall.seq, call.SeqId)
		return
	}
	if !t.currentCall.matchesPayload(call.Payload) {
		logs.Info.Printf("topic[%s]: invalid call id for seq %d", t.name, t.currentCall.seq)
		return
	}

	asUid := types.ParseUserId(msg.AsUser)

	if _, userFound := t.perUser[asUid]; !userFound {
		// User not found in topic.
		logs.Warn.Printf("topic[%s]: could not find user %s", t.name, asUid.UserId())
		return
	}

	event := normalizeCallEvent(call.Event)
	switch event {
	case constCallEventRinging, constCallEventAccept:
		// Invariants:
		// 1. Call has been initiated but not been established yet.
		if len(t.currentCall.parties) != 1 {
			return
		}
		originatorUid, originator := t.getCallOriginator()
		if originator == nil {
			// No originator session: terminating.
			t.terminateCallInProgress(false)
			return
		}
		// 2. These events may only arrive from the callee.
		if originator.sid == msg.sess.sid || originatorUid == asUid {
			return
		}
		// Prepare a {info} message to forward to the call originator.
		forwardMsg := t.currentCall.infoMessage(event)
		forwardMsg.Info.From = msg.AsUser
		forwardMsg.Info.Topic = t.original(originatorUid)
		if event == constCallEventAccept {
			// The call has been accepted.
			// Send a replacement {data} message to the topic.
			msgCopy := *msg
			msgCopy.AsUser = originatorUid.UserId()
			replaceWith := constCallMsgAccepted
			var origHead map[string]any
			if msgCopy.Pub != nil {
				origHead = msgCopy.Pub.Head
			} // else fetch the original message from store and use its head.
			head := t.currentCall.messageHead(origHead, replaceWith, 0)
			if err := t.saveAndBroadcastMessage(&msgCopy, originatorUid, false, nil,
				head, t.currentCall.content); err != nil {
				return
			}
			// Add callee data to t.currentCall.
			t.currentCall.parties[msg.sess.sid] = callPartyData{
				uid:          asUid,
				isOriginator: false,
				sess:         callPartySession(msg.sess),
			}
			t.currentCall.acceptedAt = time.Now()
			t.auditCallAccept(t.currentCall, t.currentCall.acceptedAt)

			// Notify other clients that the call has been accepted.
			t.infoCallSubsOffline(msg.AsUser, asUid, event, t.currentCall.seq, call.Payload, msg.sess.sid, false)
			t.callEstablishmentTimer.Stop()
		}
		originator.queueOut(forwardMsg)

	case constCallEventOffer, constCallEventAnswer, constCallEventIceCandidate:
		// Invariants:
		// 1. Call has been estabslied (2 participants).
		if len(t.currentCall.parties) != 2 {
			logs.Warn.Printf("topic[%s]: call participants expected 2 vs found %d", t.name, len(t.currentCall.parties))
			return
		}
		// 2. Event is coming from a call participant session.
		if _, ok := t.currentCall.parties[msg.sess.sid]; !ok {
			logs.Warn.Printf("topic[%s]: call event from non-party session %s", t.name, msg.sess.sid)
			return
		}
		// Call metadata exchange. Either side of the call may send these events.
		// Simply forward them to the other session.
		var otherUid types.Uid
		var otherEnd *Session
		for sid, p := range t.currentCall.parties {
			if sid != msg.sess.sid {
				otherUid = p.uid
				otherEnd = p.sess
				break
			}
		}
		if otherEnd == nil {
			logs.Warn.Printf("topic[%s]: could not find call peer for session %s", t.name, msg.sess.sid)
			return
		}
		// All is good. Send {info} message to the otherEnd.
		forwardMsg := t.currentCall.infoMessage(call.Event)
		forwardMsg.Info.From = msg.AsUser
		forwardMsg.Info.Topic = t.original(otherUid)
		forwardMsg.Info.Payload = call.Payload
		otherEnd.queueOut(forwardMsg)

	case constCallEventHangUp:
		switch len(t.currentCall.parties) {
		case 2:
			// If it's a call in progress, hangup may arrive only from a call participant session.
			if _, ok := t.currentCall.parties[msg.sess.sid]; !ok {
				return
			}
		case 1:
			// Call hasn't been established yet.
			originatorUid, originator := t.getCallOriginator()
			// Hangup may come from either the originating session or
			// any callee user session.
			if asUid == originatorUid && originator.sid != msg.sess.sid {
				return
			}
		default:
			break
		}
		t.maybeEndCallInProgress(msg.AsUser, msg, false)

	default:
		logs.Warn.Printf("topic[%s]: video call (seq %d) received unexpected call event: %s", t.name, t.currentCall.seq, call.Event)
	}
}

// Ends current call in response to a client hangup request (msg).
func (t *Topic) maybeEndCallInProgress(from string, msg *ClientComMessage, callDidTimeout bool) {
	if t.currentCall == nil {
		return
	}
	t.callEstablishmentTimer.Stop()
	originatorUid, _ := t.getCallOriginator()
	var replaceWith string
	var callDuration int64
	finishedAt := time.Now()
	if from != "" && len(t.currentCall.parties) == 2 {
		// This is a call in progress.
		replaceWith = constCallMsgFinished
		callDuration = finishedAt.Sub(t.currentCall.acceptedAt).Milliseconds()
	} else {
		if from != "" {
			// User originated hang-up.
			if from == originatorUid.UserId() {
				// Originator/caller requested event.
				replaceWith = constCallMsgMissed
			} else {
				// Callee requested event.
				replaceWith = constCallMsgDeclined
			}
		} else {
			// Server initiated disconnect.
			// Call hasn't been established. Just drop it.
			if callDidTimeout {
				replaceWith = constCallMsgMissed
			} else {
				replaceWith = constCallMsgDisconnected
			}
		}
	}

	// Send a message indicating the call has ended.
	msgCopy := *msg
	msgCopy.AsUser = originatorUid.UserId()
	var origHead map[string]any
	if msgCopy.Pub != nil {
		origHead = msgCopy.Pub.Head
	} // else fetch the original message from store and use its head.
	head := t.currentCall.messageHead(origHead, replaceWith, int(callDuration))
	if err := t.saveAndBroadcastMessage(&msgCopy, originatorUid, false, nil, head, t.currentCall.content); err != nil {
		logs.Err.Printf("topic[%s]: failed to write finalizing message for call seq id %d - '%s'", t.name, t.currentCall.seq, err)
	}
	t.auditCallFinish(t.currentCall, callFinalStatus(replaceWith), callDuration, finishedAt)

	// Send {info} hangup event to the subscribed sessions.
	t.broadcastToSessions(t.currentCall.infoMessage(constCallEventHangUp))

	// Let all other sessions know the call is over.
	for tgt := range t.perUser {
		t.infoCallSubsOffline(from, tgt, constCallEventHangUp, t.currentCall.seq, nil, "", true)
	}
	t.currentCall = nil
}

func callFinalStatus(replaceWith string) string {
	switch replaceWith {
	case constCallMsgFinished:
		return types.CallStatusFinished
	case constCallMsgMissed:
		return types.CallStatusMissed
	case constCallMsgDeclined:
		return types.CallStatusDeclined
	default:
		return types.CallStatusDisconnected
	}
}

// Server initiated call termination.
func (t *Topic) terminateCallInProgress(callDidTimeout bool) {
	if t.currentCall == nil {
		return
	}
	uid, sess := t.getCallOriginator()
	if sess == nil || uid.IsZero() {
		// Just drop the call.
		logs.Warn.Printf("topic[%s]: video call seq %d has no originator, terminating.", t.name, t.currentCall.seq)
		t.currentCall = nil
		return
	}
	// Dummy hangup request.
	dummy := &ClientComMessage{
		Original:  t.original(uid),
		RcptTo:    uid.UserId(),
		AsUser:    uid.UserId(),
		Timestamp: types.TimeNow(),
		sess:      sess,
	}

	logs.Info.Printf("topic[%s]: terminating call seq %d, timeout: %t", t.name, t.currentCall.seq, callDidTimeout)
	t.maybeEndCallInProgress("", dummy, callDidTimeout)
}