From 192bdede3326c0692479f4ff023336d365cd9d3f Mon Sep 17 00:00:00 2001 From: richjohnrambo Date: Fri, 18 Sep 2026 11:15:04 +0800 Subject: [PATCH] fix: route call accept across pods --- server/calls.go | 33 ++++++++++++++++++++--- server/cluster.go | 2 +- server/im_agora_http.go | 59 +++++++++++++++++++++++++++++++++++++++++ server/im_http_test.go | 34 +++++++++++++++++------- server/session.go | 7 +++++ 5 files changed, 122 insertions(+), 13 deletions(-) diff --git a/server/calls.go b/server/calls.go index 7d0966b..6735a0b 100644 --- a/server/calls.go +++ b/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 _, ok := t.currentCall.parties[msg.sess.sid]; !ok { + // If it's a call in progress, hangup may arrive from any current + // 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: diff --git a/server/cluster.go b/server/cluster.go index 49f6e7d..df29d9f 100644 --- a/server/cluster.go +++ b/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: diff --git a/server/im_agora_http.go b/server/im_agora_http.go index ff5dcb2..ca3e8c4 100644 --- a/server/im_agora_http.go +++ b/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": diff --git a/server/im_http_test.go b/server/im_http_test.go index 82cb4d6..7c97f2a 100644 --- a/server/im_http_test.go +++ b/server/im_http_test.go @@ -201,14 +201,19 @@ type testFavoritesStore struct { } type testCallsStore struct { - items []types.CallRecord - topic string - listed bool - tenant types.TenantID - user types.Uid - limit int - cursor string - next string + items []types.CallRecord + topic string + listed bool + tenant types.TenantID + user types.Uid + limit int + cursor 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) diff --git a/server/session.go b/server/session.go index d6b5095..25e021b 100644 --- a/server/session.go +++ b/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: