mautrix-whatsapp/pkg/connector/matrixrtc.go

package connector

import (
	"context"
	"strings"
	"time"

	"github.com/rs/zerolog"
	"maunium.net/go/mautrix/bridgev2"
	"maunium.net/go/mautrix/bridgev2/matrix"
	"maunium.net/go/mautrix/event"
	"maunium.net/go/mautrix/id"

	"go.mau.fi/mautrix-whatsapp/pkg/connector/voip"
	"go.mau.fi/mautrix-whatsapp/pkg/connector/wadb"
)

const (
	matrixRTCHealthcheckTimeout       = 15 * time.Second
	matrixRTCOutboundStartDedupWindow = 30 * time.Second
)

func withoutCancelOrBackground(ctx context.Context) context.Context {
	if ctx == nil {
		return context.Background()
	}
	return context.WithoutCancel(ctx)
}

func (wa *WhatsAppConnector) initMatrixRTCEventHooks() {
	matrixConnector, ok := wa.Bridge.Matrix.(*matrix.Connector)
	if !ok || matrixConnector.EventProcessor == nil {
		wa.Bridge.Log.Debug().Msg("Matrix connector does not expose an event processor for MatrixRTC hooks")
		return
	}
	for _, evtType := range voip.SupportedMatrixRTCEventTypes() {
		matrixConnector.EventProcessor.On(evtType, wa.handleMatrixRTCEvent)
	}
	wa.Bridge.Log.Debug().Int("event_type_count", len(voip.SupportedMatrixRTCEventTypes())).Msg("Registered MatrixRTC event hooks")
}

func (wa *WhatsAppConnector) startMatrixRTCHealthcheck() {
	if !wa.Config.VOIP.Enabled || !wa.Config.VOIP.Diagnostics.HealthcheckFocusOnStartup {
		return
	}
	go func() {
		ctx, cancel := context.WithTimeout(withoutCancelOrBackground(wa.Bridge.BackgroundCtx), matrixRTCHealthcheckTimeout)
		defer cancel()
		focus, err := voip.DiscoverLiveKitFocus(ctx, nil, wa.Bridge.Matrix.ServerName(), wa.Config.VOIP.MatrixRTC.LiveKitServiceURL)
		log := wa.Bridge.Log.With().Str("component", "voip_healthcheck").Logger()
		if err != nil {
			event := log.Warn()
			if wa.Config.VOIP.MatrixRTC.RequireLiveKitFocus {
				event = log.Error()
			}
			event.Err(err).Msg("MatrixRTC LiveKit focus healthcheck failed")
			return
		}
		log.Info().
			Str("focus_type", focus.Type).
			Str("livekit_service_url", focus.LiveKitServiceURL).
			Msg("MatrixRTC LiveKit focus healthcheck passed")
	}()
}

func (wa *WhatsAppConnector) handleMatrixRTCEvent(ctx context.Context, evt *event.Event) {
	if !wa.Config.VOIP.Enabled {
		return
	}
	parsed, ok := voip.ParseMatrixRTCEvent(evt)
	if !ok {
		return
	}

	log := zerolog.Ctx(ctx).With().
		Str("matrixrtc_kind", string(parsed.Kind)).
		Str("matrix_event_type", parsed.Type.Type).
		Stringer("matrix_room_id", parsed.RoomID).
		Stringer("matrix_sender", parsed.Sender).
		Str("matrix_call_id", parsed.CallID).
		Logger()

	if parsed.Sender == wa.Bridge.Bot.GetMXID() || wa.Bridge.IsGhostMXID(parsed.Sender) {
		log.Debug().Msg("Ignoring MatrixRTC event sent by the bridge")
		return
	}
	if !wa.Bridge.Config.Permissions.Get(parsed.Sender).SendEvents {
		log.Debug().Msg("Dropping MatrixRTC event from user with no permission to send events")
		wa.Bridge.Matrix.SendMessageStatus(ctx, &bridgev2.ErrNoPermissionToInteract, bridgev2.StatusEventInfoFromEvent(evt))
		return
	}

	portal, err := wa.Bridge.GetPortalByMXID(ctx, parsed.RoomID)
	if err != nil {
		log.Err(err).Msg("Failed to look up portal for MatrixRTC event")
		return
	} else if portal == nil {
		log.Debug().Msg("Ignoring MatrixRTC event outside a bridged portal")
		return
	}
	portalLogin := wa.Bridge.GetCachedUserLoginByID(portal.Receiver)
	if !matrixRTCSenderOwnsLogin(parsed.Sender, portalLogin) {
		log.Warn().
			Str("portal_receiver", string(portal.Receiver)).
			Msg("Dropping MatrixRTC event from a user who does not own the portal login")
		wa.Bridge.Matrix.SendMessageStatus(ctx, &bridgev2.ErrNoPermissionToInteract, bridgev2.StatusEventInfoFromEvent(evt))
		return
	}

	activeCalls, err := wa.DB.MatrixRTCCall.GetActiveInRoom(ctx, parsed.RoomID)
	if err != nil {
		log.Err(err).Msg("Failed to look up active MatrixRTC calls for room")
		return
	}
	if isMatrixRTCCallControlEvent(parsed) {
		wa.handleMatrixRTCCallControlEvent(ctx, parsed, activeCalls, log)
		return
	}
	if len(activeCalls) == 0 {
		if !shouldStartOutboundMatrixRTCCall(evt, parsed, wa.Config.VOIP.MatrixRTC.MembershipEventCompat) {
			log.Debug().Msg("Ignoring MatrixRTC event without active bridged calls in the room")
			return
		}
		if !wa.reserveMatrixRTCOutboundStart(parsed.RoomID.String()) {
			log.Debug().Msg("Ignoring duplicate outbound MatrixRTC start in dedupe window")
			return
		}
		if err = wa.startOutboundMatrixRTCCall(ctx, portal, parsed); err != nil {
			log.Err(err).Msg("Failed to start outbound WhatsApp call from MatrixRTC event")
			if cleanupErr := wa.cleanupFailedOutboundMatrixRTCStart(ctx, parsed); cleanupErr != nil {
				log.Err(cleanupErr).Msg("Failed to clean up failed outbound MatrixRTC event")
			}
		}
		return
	}

	var matched, handled, activated, ended int
	for _, activeCall := range activeCalls {
		matched++
		login, err := wa.Bridge.GetExistingUserLoginByID(ctx, activeCall.UserLoginID)
		callLog := log.With().
			Str("wa_call_id", activeCall.WACallID).
			Str("user_login_id", string(activeCall.UserLoginID)).
			Stringer("matrix_participant_mxid", activeCall.MatrixParticipantMXID).
			Str("matrix_session_id", activeCall.MatrixSessionID).
			Logger()
		if err != nil {
			callLog.Err(err).Msg("Failed to look up WhatsApp login for MatrixRTC call")
			continue
		} else if login == nil {
			callLog.Debug().Msg("MatrixRTC call references a missing WhatsApp login")
			continue
		}
		client, ok := login.Client.(*WhatsAppClient)
		if !ok || client == nil || client.VOIP == nil {
			callLog.Debug().Msg("WhatsApp login has no VOIP manager for MatrixRTC event")
			continue
		}
		if matrixRTCMembershipEventMatchesCall(parsed, activeCall) &&
			activeCall.SelectedMembershipEventID != parsed.EventID {
			if activeCall.SelectedHandRaiseEventID != "" {
				if err = client.VOIP.SetHandRaised(activeCall.WACallID, false); err != nil {
					callLog.Warn().Err(err).Msg("Failed to lower WhatsApp hand after MatrixRTC membership replacement")
				}
				activeCall.SelectedHandRaiseEventID = ""
			}
			activeCall.SelectedMembershipEventID = parsed.EventID
			if err = wa.DB.MatrixRTCCall.Put(ctx, activeCall); err != nil {
				callLog.Err(err).Msg("Failed to persist replacement MatrixRTC membership event")
			}
		}
		handled++
		endedCalls := client.VOIP.HandleMatrixRTCCallEvent(ctx, parsed, activeCall.WACallID)
		if shouldEndMatrixRTCCallFromMembership(parsed, activeCall.SelectedPublisherID) {
			endedCalls += client.VOIP.HandleMatrixRTCCallEvent(ctx, voip.MatrixRTCEvent{
				Kind:   voip.MatrixRTCEventKindRTCDecline,
				RoomID: parsed.RoomID,
				Sender: parsed.Sender,
				CallID: parsed.CallID,
			}, activeCall.WACallID)
		}
		ended += endedCalls
		if shouldActivateMatrixRTCCall(parsed, activeCall.State) {
			if err = client.activateMatrixRTCCall(ctx, activeCall, parsed); err != nil {
				callLog.Err(err).Msg("Failed to activate MatrixRTC LiveKit bridge for WhatsApp call")
			} else {
				activated++
			}
		}
		if endedCalls > 0 {
			err = wa.DB.MatrixRTCCall.MarkEnded(ctx, activeCall.UserLoginID, activeCall.WACallID, "ended", string(parsed.Kind), "", time.Now())
			if err != nil {
				callLog.Err(err).Msg("Failed to mark MatrixRTC call ended after MatrixRTC event")
			}
		}
	}
	log.Debug().
		Int("active_call_count", len(activeCalls)).
		Int("matched_call_count", matched).
		Int("handled_call_count", handled).
		Int("activated_call_count", activated).
		Int("ended_call_count", ended).
		Msg("Handled MatrixRTC event for active bridged calls")
}

func matrixRTCSenderOwnsLogin(sender id.UserID, login *bridgev2.UserLogin) bool {
	return sender != "" && login != nil && login.User != nil && login.User.MXID == sender
}

func matrixRTCMembershipEventMatchesCall(evt voip.MatrixRTCEvent, call *wadb.MatrixRTCCall) bool {
	if call == nil || evt.EventID == "" || !voip.MatrixRTCEventHasJoinContent(evt) {
		return false
	}
	switch evt.Kind {
	case voip.MatrixRTCEventKindRTCMembership, voip.MatrixRTCEventKindGroupCallMember:
		return matrixRTCEventMatchesParticipant(evt, call.SelectedPublisherID)
	default:
		return false
	}
}

func shouldActivateMatrixRTCCall(evt voip.MatrixRTCEvent, callState string) bool {
	if callState != "ringing" {
		return false
	}
	switch evt.Kind {
	case voip.MatrixRTCEventKindRTCMembership, voip.MatrixRTCEventKindGroupCallMember:
		return voip.MatrixRTCEventHasJoinContent(evt)
	default:
		return false
	}
}

func shouldStartOutboundMatrixRTCCall(evt *event.Event, parsed voip.MatrixRTCEvent, membershipCompat string) bool {
	if parsed.Type.Class != event.StateEventType || !voip.MatrixRTCEventHasJoinContent(parsed) {
		return false
	}
	switch parsed.Kind {
	case voip.MatrixRTCEventKindRTCMembership:
		if !matrixRTCCompatAllowsModern(membershipCompat) {
			return false
		}
	case voip.MatrixRTCEventKindGroupCallMember:
		if !matrixRTCCompatAllowsLegacy(membershipCompat) {
			return false
		}
	default:
		return false
	}
	return !matrixRTCPrevContentHasJoinContent(evt)
}

func shouldEndMatrixRTCCallFromMembership(evt voip.MatrixRTCEvent, selectedParticipantID string) bool {
	switch evt.Kind {
	case voip.MatrixRTCEventKindRTCMembership, voip.MatrixRTCEventKindGroupCallMember:
	default:
		return false
	}
	if selectedParticipantID == "" || voip.MatrixRTCEventHasJoinContent(evt) {
		return false
	}
	return matrixRTCEventMatchesParticipant(evt, selectedParticipantID)
}

func matrixRTCEventMatchesParticipant(evt voip.MatrixRTCEvent, participantID string) bool {
	if participantID == "" {
		return false
	}
	if matrixRTCTriggerParticipantID(evt) == participantID {
		return true
	}
	if evt.StateKey == "" || evt.Sender == "" {
		return false
	}
	if legacyID := legacyMatrixRTCParticipantIDFromStateKey(evt.Sender, evt.StateKey); legacyID == participantID {
		return true
	}
	if modernID := modernMatrixRTCParticipantIDFromStateKey(evt.Sender, evt.StateKey); modernID == participantID {
		return true
	}
	return false
}

func legacyMatrixRTCParticipantIDFromStateKey(sender id.UserID, stateKey string) string {
	prefix := "_" + string(sender) + "_"
	const suffix = "_m.call"
	if !strings.HasPrefix(stateKey, prefix) || !strings.HasSuffix(stateKey, suffix) {
		return ""
	}
	deviceID := strings.TrimSuffix(strings.TrimPrefix(stateKey, prefix), suffix)
	if deviceID == "" {
		return ""
	}
	return voip.MatrixRTCMemberID(sender, deviceID)
}

func modernMatrixRTCParticipantIDFromStateKey(sender id.UserID, stateKey string) string {
	prefix := string(sender) + "_"
	if !strings.HasPrefix(stateKey, prefix) {
		return ""
	}
	deviceID := strings.TrimPrefix(stateKey, prefix)
	if deviceID == "" {
		return ""
	}
	return voip.MatrixRTCMemberID(sender, deviceID)
}

func (wa *WhatsAppConnector) reserveMatrixRTCOutboundStart(roomID string) bool {
	if roomID == "" {
		return false
	}
	now := time.Now()
	wa.matrixRTCOutboundStartLock.Lock()
	defer wa.matrixRTCOutboundStartLock.Unlock()
	if wa.matrixRTCOutboundStartExpires == nil {
		wa.matrixRTCOutboundStartExpires = make(map[string]time.Time)
	}
	for trackedRoomID, expires := range wa.matrixRTCOutboundStartExpires {
		if !expires.After(now) {
			delete(wa.matrixRTCOutboundStartExpires, trackedRoomID)
		}
	}
	if expires := wa.matrixRTCOutboundStartExpires[roomID]; expires.After(now) {
		return false
	}
	wa.matrixRTCOutboundStartExpires[roomID] = now.Add(matrixRTCOutboundStartDedupWindow)
	return true
}

func matrixRTCPrevContentHasJoinContent(evt *event.Event) bool {
	if evt == nil || evt.Unsigned.PrevContent == nil {
		return false
	}
	prevEvt := *evt
	prevEvt.Content = *evt.Unsigned.PrevContent
	prevEvt.Unsigned.PrevContent = nil
	parsedPrev, ok := voip.ParseMatrixRTCEvent(&prevEvt)
	return ok && voip.MatrixRTCEventHasJoinContent(parsedPrev)
}