fix: route call accept across pods

server/calls.go

@@ -330,6 +330,23 @@ func (call *videoCall) isStaleUnanswered(now time.Time, timeoutSeconds int) bool
 	return now.Sub(call.startedAt) > time.Duration(timeoutSeconds)*time.Second
 }
 
+func (call *videoCall) hasPartySessionOrUser(sid string, uid types.Uid) bool {
+	if call == nil {
+		return false
+	}
+	if sid != "" {
+		if _, ok := call.parties[sid]; ok {
+			return true
+		}
+	}
+	for _, party := range call.parties {
+		if party.uid == uid {
+			return true
+		}
+	}
+	return false
+}
+
 func (t *Topic) auditCallCreate(call *videoCall, caller types.Uid) {
 	if store.Calls == nil || call == nil || call.callID == "" || caller.IsZero() {
 		return
@@ -645,8 +662,14 @@ func (t *Topic) handleCallEvent(msg *ClientComMessage) {
 	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 it's a call in progress, hangup may arrive from any current
-			if _, ok := t.currentCall.parties[msg.sess.sid]; !ok {
+			// session of a call participant. This also covers rtc-join
+			// auto-accept, where the accepting HTTP request has no WebSocket sid.
+			sid := ""
+			if msg.sess != nil {
+				sid = msg.sess.sid
+			}
+			if !t.currentCall.hasPartySessionOrUser(sid, asUid) {
 				return
 			}
 		case 1:
@@ -654,7 +677,11 @@ func (t *Topic) handleCallEvent(msg *ClientComMessage) {
 			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 {
+			sid := ""
+			if msg.sess != nil {
+				sid = msg.sess.sid
+			}
+			if asUid == originatorUid && originator.sid != sid {
 				return
 			}
 		default:

server/cluster.go

@@ -594,7 +594,7 @@ func (c *Cluster) TopicMaster(msg *ClusterReq, rejected *bool) error {
 			logs.Warn.Println("cluster: meta request for unknown topic", msg.RcptTo)
 		}
 
-	case ProxyReqBroadcast:
+	case ProxyReqBroadcast, ProxyReqCall:
 		select {
 		case globals.hub.routeCli <- msg.CliMsg:
 		default:

server/im_agora_http.go

@@ -10,6 +10,8 @@ import (
 	"time"
 
 	rtctokenbuilder "github.com/AgoraIO/Tools/DynamicKey/AgoraDynamicKey/go/src/rtctokenbuilder2"
+	"github.com/tinode/chat/server/auth"
+	"github.com/tinode/chat/server/logs"
 	"github.com/tinode/chat/server/store"
 	"github.com/tinode/chat/server/store/types"
 )
@@ -110,6 +112,9 @@ func serveIMCallRTCJoin(wrt http.ResponseWriter, req *http.Request) {
 		writeIMError(wrt, req, http.StatusConflict, 30064, imReasonCallStateInvalid, "call state invalid", map[string]any{"status": record.Status})
 		return
 	}
+	if body.Role == "callee" && record.Status == types.CallStatusStarted {
+		acceptIMCallFromRTCJoin(principal, record)
+	}
 	channelName := config.ChannelPrefix + record.CallID
 	uid, err := agoraUIDForUser(config, principal.Uid)
 	if err != nil {
@@ -267,6 +272,60 @@ func callStateAllowsRTCJoin(status string) bool {
 	}
 }
 
+func acceptIMCallFromRTCJoin(principal *imPrincipal, record *types.CallRecord) {
+	if principal == nil || record == nil || record.Status != types.CallStatusStarted || record.CalleeID != principal.Uid {
+		return
+	}
+	now := types.TimeNow()
+	if store.Calls != nil {
+		if err := store.Calls.Accept(principal.TenantID, record.CallID, record.SeqID, now); err != nil {
+			logs.Warn.Printf("rtc join: failed to mark call accepted tenant=%d topic=%s seq=%d call_id=%s: %v",
+				principal.TenantID, record.Topic, record.SeqID, record.CallID, err)
+		}
+	}
+	if globals.hub == nil {
+		return
+	}
+	payload, _ := json.Marshal(map[string]any{"callId": record.CallID})
+	msg := &ClientComMessage{
+		Note: &MsgClientNote{
+			Topic:   record.Topic,
+			What:    "call",
+			Event:   constCallEventAccept,
+			SeqId:   record.SeqID,
+			Payload: payload,
+		},
+		Original:  record.Topic,
+		RcptTo:    record.Topic,
+		AsUser:    principal.Uid.UserId(),
+		AuthLvl:   int(auth.LevelAuth),
+		Timestamp: now,
+		TenantID:  principal.TenantID,
+		sess: &Session{
+			proto:      NONE,
+			sid:        "rtc-join:" + principal.Uid.UserId(),
+			uid:        principal.Uid,
+			tenantID:   principal.TenantID,
+			authLvl:    auth.LevelAuth,
+			lastAction: now.UnixNano(),
+		},
+		init: true,
+	}
+	if globals.cluster != nil && globals.cluster.isRemoteTopic(types.TopicKey{TenantID: principal.TenantID, Topic: record.Topic}) {
+		if err := globals.cluster.routeToTopicMaster(ProxyReqCall, msg, record.Topic, msg.sess); err != nil {
+			logs.Warn.Printf("rtc join: failed to route call accept tenant=%d topic=%s seq=%d call_id=%s: %v",
+				principal.TenantID, record.Topic, record.SeqID, record.CallID, err)
+		}
+		return
+	}
+	select {
+	case globals.hub.routeCli <- msg:
+	default:
+		logs.Err.Printf("rtc join: hub route queue full tenant=%d topic=%s seq=%d call_id=%s",
+			principal.TenantID, record.Topic, record.SeqID, record.CallID)
+	}
+}
+
 func agoraUIDForUser(config imAgoraConfig, uid types.Uid) (any, error) {
 	switch config.UIDMode {
 	case "int":

server/im_http_test.go

@@ -201,14 +201,19 @@ type testFavoritesStore struct {
 }
 
 type testCallsStore struct {
-	items  []types.CallRecord
+	items        []types.CallRecord
-	topic  string
+	topic        string
-	listed bool
+	listed       bool
-	tenant types.TenantID
+	tenant       types.TenantID
-	user   types.Uid
+	user         types.Uid
-	limit  int
+	limit        int
-	cursor string
+	cursor       string
-	next   string
+	next         string
+	accepted     bool
+	acceptTenant types.TenantID
+	acceptCallID string
+	acceptSeq    int64
+	acceptedAt   time.Time
 }
 
 type testThirdPartyConfigStore map[string]string
@@ -830,7 +835,12 @@ func (s *testCallsStore) Create(_ types.TenantID, _ *types.CallRecord) error {
 	return nil
 }
 
-func (s *testCallsStore) Accept(_ types.TenantID, _ string, _ int64, _ time.Time) error {
+func (s *testCallsStore) Accept(tenantID types.TenantID, callID string, seq int64, acceptedAt time.Time) error {
+	s.accepted = true
+	s.acceptTenant = tenantID
+	s.acceptCallID = callID
+	s.acceptSeq = seq
+	s.acceptedAt = acceptedAt
 	return nil
 }
 
@@ -6300,6 +6310,9 @@ func TestServeIMCallRTCJoinReturnsSameAgoraChannelForCallee(t *testing.T) {
 	if callerRec.Code != http.StatusOK {
 		t.Fatalf("caller rtc join status: got %d body=%s", callerRec.Code, callerRec.Body.String())
 	}
+	if calls.accepted {
+		t.Fatal("caller rtc join must not auto-accept the call")
+	}
 
 	calleeReq := httptest.NewRequest(http.MethodPost, "/v1/im/calls/rtc-join", strings.NewReader(`{"topic":"`+topic+`","call_id":"call-1","seq":12,"call_type":"audio","role":"callee"}`))
 	calleeReq.Header.Set("Authorization", "Bearer token-callee")
@@ -6308,6 +6321,9 @@ func TestServeIMCallRTCJoinReturnsSameAgoraChannelForCallee(t *testing.T) {
 	if calleeRec.Code != http.StatusOK {
 		t.Fatalf("callee rtc join status: got %d body=%s", calleeRec.Code, calleeRec.Body.String())
 	}
+	if !calls.accepted || calls.acceptTenant != 7 || calls.acceptCallID != "call-1" || calls.acceptSeq != 12 || calls.acceptedAt.IsZero() {
+		t.Fatalf("callee rtc join did not mark call accepted: %#v", calls)
+	}
 
 	callerData := decodeIMHTTPResponse(t, callerRec).Data.(map[string]any)
 	calleeData := decodeIMHTTPResponse(t, calleeRec).Data.(map[string]any)

server/session.go

@@ -1650,6 +1650,13 @@ func (s *Session) note(msg *ClientComMessage) {
 		// the user about the call.
 		//
 		// Hub will forward to topic, if appropriate.
+		if msg.Note.What == "call" && globals.cluster != nil &&
+			globals.cluster.isRemoteTopic(types.TopicKey{TenantID: msg.TenantID, Topic: msg.RcptTo}) {
+			if err := globals.cluster.routeToTopicMaster(ProxyReqCall, msg, msg.RcptTo, s); err != nil {
+				logs.Warn.Printf("s.note: failed routing detached call event to topic master topic=%s sid=%s err=%v", msg.RcptTo, s.sid, err)
+			}
+			return
+		}
 		select {
 		case globals.hub.routeCli <- msg:
 		default: