diff --git a/Procfile.dev b/Procfile.dev index 651b84edd..239b36c41 100644 --- a/Procfile.dev +++ b/Procfile.dev @@ -2,3 +2,4 @@ backend: bin/rails s -p 3000 # https://github.com/mperham/sidekiq/issues/3090#issuecomment-389748695 worker: dotenv bundle exec sidekiq -C config/sidekiq.yml vite: bin/vite dev +media_server: cd enterprise/media-server && AUTH_TOKEN=${MEDIA_SERVER_AUTH_TOKEN:-devtoken} RAILS_CALLBACK_URL=http://localhost:3000 PUBLIC_IP=${MEDIA_SERVER_PUBLIC_IP:-127.0.0.1} RECORDINGS_DIR=/tmp/chatwoot-recordings HTTP_PORT=4000 go run ./cmd/server diff --git a/app/views/api/v1/models/_inbox.json.jbuilder b/app/views/api/v1/models/_inbox.json.jbuilder index 700d780fc..cb745d9b9 100644 --- a/app/views/api/v1/models/_inbox.json.jbuilder +++ b/app/views/api/v1/models/_inbox.json.jbuilder @@ -130,6 +130,7 @@ if resource.whatsapp? json.provider_config resource.channel.try(:provider_config) if Current.account_user&.administrator? json.reauthorization_required resource.channel.try(:reauthorization_required?) json.calling_enabled resource.channel.try(:provider_config)&.dig('calling_enabled') || false + json.media_server_enabled Call.media_server_enabled? end ## Voice Channel Attributes diff --git a/config/routes.rb b/config/routes.rb index 7ee28d8f4..3f01b93e4 100644 --- a/config/routes.rb +++ b/config/routes.rb @@ -320,12 +320,6 @@ Rails.application.routes.draw do end end - namespace :media_server do - post 'callbacks/agent_disconnected', to: 'callbacks#agent_disconnected' - post 'callbacks/recording_ready', to: 'callbacks#recording_ready' - post 'callbacks/session_terminated', to: 'callbacks#session_terminated' - end - resources :webhooks, only: [:index, :create, :update, :destroy] namespace :integrations do resources :apps, only: [:index, :show] @@ -629,6 +623,16 @@ Rails.application.routes.draw do get 'instagram/callback', to: 'instagram/callbacks#show' get 'tiktok/callback', to: 'tiktok/callbacks#show' get 'notion/callback', to: 'notion/callbacks#show' + + # Media server callbacks — authenticated by shared MEDIA_SERVER_AUTH_TOKEN, + # not a user session. Intentionally top-level and not account-scoped. + namespace :callbacks do + namespace :media_server do + post :agent_disconnected, to: '/media_server/callbacks#agent_disconnected' + post :recording_ready, to: '/media_server/callbacks#recording_ready' + post :session_terminated, to: '/media_server/callbacks#session_terminated' + end + end # ---------------------------------------------------------------------- # Routes for external service verifications get '.well-known/assetlinks.json' => 'android_app#assetlinks' diff --git a/enterprise/app/controllers/api/v1/accounts/whatsapp_calls_controller.rb b/enterprise/app/controllers/api/v1/accounts/whatsapp_calls_controller.rb index 9f0c52d65..299daa0b5 100644 --- a/enterprise/app/controllers/api/v1/accounts/whatsapp_calls_controller.rb +++ b/enterprise/app/controllers/api/v1/accounts/whatsapp_calls_controller.rb @@ -199,13 +199,14 @@ class Api::V1::Accounts::WhatsappCallsController < Api::V1::Accounts::BaseContro # Step 1: Create session on media server (generates SDP offer for Meta) session_response = client.create_session( call_id: "pending_#{SecureRandom.hex(8)}", + direction: 'outgoing', sdp_offer: nil, ice_servers: [{ urls: 'stun:stun.l.google.com:19302' }], account_id: current_account.id ) # Step 2: Send the media server's SDP offer to Meta to initiate the call - sdp_offer = session_response['meta_sdp_answer'] || session_response['sdp_offer'] + sdp_offer = session_response['meta_sdp_offer'] result = conversation.inbox.channel.provider_service.initiate_call(contact_phone.delete('+'), sdp_offer) provider_call_id = result.dig('calls', 0, 'id') || result['call_id'] diff --git a/enterprise/app/controllers/api/v1/accounts/media_server/callbacks_controller.rb b/enterprise/app/controllers/media_server/callbacks_controller.rb similarity index 83% rename from enterprise/app/controllers/api/v1/accounts/media_server/callbacks_controller.rb rename to enterprise/app/controllers/media_server/callbacks_controller.rb index ec1ae6f79..bd49ca891 100644 --- a/enterprise/app/controllers/api/v1/accounts/media_server/callbacks_controller.rb +++ b/enterprise/app/controllers/media_server/callbacks_controller.rb @@ -1,6 +1,4 @@ -class Api::V1::Accounts::MediaServer::CallbacksController < Api::V1::Accounts::BaseController - skip_before_action :authenticate_user!, raise: false - skip_before_action :authenticate_access_token!, raise: false +class MediaServer::CallbacksController < ApplicationController before_action :validate_media_server_token def agent_disconnected @@ -44,9 +42,7 @@ class Api::V1::Accounts::MediaServer::CallbacksController < Api::V1::Accounts::B } ) - # Also terminate on Meta side - provider = call.inbox.channel.provider_service - provider.terminate_call(call.provider_call_id) + call.inbox.channel.provider_service.terminate_call(call.provider_call_id) rescue StandardError => e Rails.logger.error "[MEDIA SERVER] Failed to terminate on provider: #{e.message}" ensure diff --git a/enterprise/app/services/whatsapp/call_service.rb b/enterprise/app/services/whatsapp/call_service.rb index 215771819..7d87c5311 100644 --- a/enterprise/app/services/whatsapp/call_service.rb +++ b/enterprise/app/services/whatsapp/call_service.rb @@ -86,6 +86,7 @@ class Whatsapp::CallService # Step 1: Create session on Go server with Meta's SDP session_response = client.create_session( call_id: call.provider_call_id, + direction: 'incoming', sdp_offer: call.sdp_offer, ice_servers: call.ice_servers, account_id: call.account_id diff --git a/enterprise/app/services/whatsapp/incoming_call_service.rb b/enterprise/app/services/whatsapp/incoming_call_service.rb index 28e9d872f..859186e28 100644 --- a/enterprise/app/services/whatsapp/incoming_call_service.rb +++ b/enterprise/app/services/whatsapp/incoming_call_service.rb @@ -38,7 +38,12 @@ class Whatsapp::IncomingCallService existing_call.update!(status: 'in_progress', started_at: Time.current, meta: (existing_call.meta || {}).merge('sdp_answer' => sdp_answer)) Whatsapp::CallMessageBuilder.update_status!(call: existing_call, status: 'in_progress') update_conversation_call_status(existing_call.conversation, 'in-progress', existing_call.direction_label) - broadcast_outbound_call_connected(existing_call, sdp_answer) + + if existing_call.media_session_id.present? + finalize_outbound_server_relay(existing_call, sdp_answer) + else + broadcast_outbound_call_connected(existing_call, sdp_answer) + end return end @@ -200,6 +205,31 @@ class Whatsapp::IncomingCallService ActionCable.server.broadcast("account_#{inbox.account_id}", payload) end + # Server-relay outbound: deliver Meta's SDP answer to the media server so it + # completes Peer A, then ask it for an SDP offer to send to the agent (Peer B). + # Broadcast that offer so the agent browser can negotiate directly with the + # media server. + def finalize_outbound_server_relay(call, sdp_answer) + client = Whatsapp::MediaServerClient.new + client.set_meta_answer(call.media_session_id, sdp_answer: sdp_answer) + agent_offer = client.generate_agent_offer(call.media_session_id) + + payload = { + event: 'whatsapp_call.outbound_connected', + data: { + account_id: inbox.account_id, + id: call.id, + call_id: call.provider_call_id, + conversation_id: call.conversation_id, + sdp_offer: agent_offer['sdp_offer'], + ice_servers: agent_offer['ice_servers'] + } + } + ActionCable.server.broadcast("account_#{inbox.account_id}", payload) + rescue Whatsapp::MediaServerClient::ConnectionError, Whatsapp::MediaServerClient::SessionError => e + Rails.logger.error "[WHATSAPP CALL] Failed to finalize outbound server-relay: #{e.message}" + end + # Meta sends "USER_INITIATED" / "BUSINESS_INITIATED", map to Call enum values def map_direction(raw_direction) return :outgoing if raw_direction&.upcase == 'BUSINESS_INITIATED' diff --git a/enterprise/app/services/whatsapp/media_server_client.rb b/enterprise/app/services/whatsapp/media_server_client.rb index 607a8f5a0..afbe6ecd7 100644 --- a/enterprise/app/services/whatsapp/media_server_client.rb +++ b/enterprise/app/services/whatsapp/media_server_client.rb @@ -4,8 +4,8 @@ class Whatsapp::MediaServerClient TIMEOUT = 10 - def create_session(call_id:, sdp_offer:, ice_servers:, account_id: nil) - body = { call_id: call_id, meta_sdp_offer: sdp_offer, ice_servers: ice_servers, account_id: account_id }.compact + def create_session(call_id:, direction:, sdp_offer:, ice_servers:, account_id: nil) + body = { call_id: call_id, direction: direction, meta_sdp_offer: sdp_offer, ice_servers: ice_servers, account_id: account_id }.compact post('/sessions', body) end @@ -17,6 +17,10 @@ class Whatsapp::MediaServerClient post("/sessions/#{session_id}/agent-answer", { sdp_answer: sdp_answer }) end + def set_meta_answer(session_id, sdp_answer:) + post("/sessions/#{session_id}/meta-answer", { sdp_answer: sdp_answer }) + end + def reconnect_agent(session_id) post("/sessions/#{session_id}/agent-reconnect") end diff --git a/enterprise/media-server/go.mod b/enterprise/media-server/go.mod index 82a697235..07e6c426c 100644 --- a/enterprise/media-server/go.mod +++ b/enterprise/media-server/go.mod @@ -17,7 +17,7 @@ require ( github.com/pion/mdns/v2 v2.0.7 // indirect github.com/pion/randutil v0.1.0 // indirect github.com/pion/rtcp v1.2.14 // indirect - github.com/pion/sctp v1.8.34 // indirect + github.com/pion/sctp v1.8.35 // indirect github.com/pion/sdp/v3 v3.0.9 // indirect github.com/pion/srtp/v3 v3.0.4 // indirect github.com/pion/stun/v3 v3.0.0 // indirect diff --git a/enterprise/media-server/go.sum b/enterprise/media-server/go.sum new file mode 100644 index 000000000..6c271eacb --- /dev/null +++ b/enterprise/media-server/go.sum @@ -0,0 +1,61 @@ +github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= +github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/pion/datachannel v1.5.10 h1:ly0Q26K1i6ZkGf42W7D4hQYR90pZwzFOjTq5AuCKk4o= +github.com/pion/datachannel v1.5.10/go.mod h1:p/jJfC9arb29W7WrxyKbepTU20CFgyx5oLo8Rs4Py/M= +github.com/pion/dtls/v3 v3.0.4 h1:44CZekewMzfrn9pmGrj5BNnTMDCFwr+6sLH+cCuLM7U= +github.com/pion/dtls/v3 v3.0.4/go.mod h1:R373CsjxWqNPf6MEkfdy3aSe9niZvL/JaKlGeFphtMg= +github.com/pion/ice/v4 v4.0.3 h1:9s5rI1WKzF5DRqhJ+Id8bls/8PzM7mau0mj1WZb4IXE= +github.com/pion/ice/v4 v4.0.3/go.mod h1:VfHy0beAZ5loDT7BmJ2LtMtC4dbawIkkkejHPRZNB3Y= +github.com/pion/interceptor v0.1.37 h1:aRA8Zpab/wE7/c0O3fh1PqY0AJI3fCSEM5lRWJVorwI= +github.com/pion/interceptor v0.1.37/go.mod h1:JzxbJ4umVTlZAf+/utHzNesY8tmRkM2lVmkS82TTj8Y= +github.com/pion/logging v0.2.2 h1:M9+AIj/+pxNsDfAT64+MAVgJO0rsyLnoJKCqf//DoeY= +github.com/pion/logging v0.2.2/go.mod h1:k0/tDVsRCX2Mb2ZEmTqNa7CWsQPc+YYCB7Q+5pahoms= +github.com/pion/mdns/v2 v2.0.7 h1:c9kM8ewCgjslaAmicYMFQIde2H9/lrZpjBkN8VwoVtM= +github.com/pion/mdns/v2 v2.0.7/go.mod h1:vAdSYNAT0Jy3Ru0zl2YiW3Rm/fJCwIeM0nToenfOJKA= +github.com/pion/randutil v0.1.0 h1:CFG1UdESneORglEsnimhUjf33Rwjubwj6xfiOXBa3mA= +github.com/pion/randutil v0.1.0/go.mod h1:XcJrSMMbbMRhASFVOlj/5hQial/Y8oH/HVo7TBZq+j8= +github.com/pion/rtcp v1.2.14 h1:KCkGV3vJ+4DAJmvP0vaQShsb0xkRfWkO540Gy102KyE= +github.com/pion/rtcp v1.2.14/go.mod h1:sn6qjxvnwyAkkPzPULIbVqSKI5Dv54Rv7VG0kNxh9L4= +github.com/pion/rtp v1.8.9 h1:E2HX740TZKaqdcPmf4pw6ZZuG8u5RlMMt+l3dxeu6Wk= +github.com/pion/rtp v1.8.9/go.mod h1:pBGHaFt/yW7bf1jjWAoUjpSNoDnw98KTMg+jWWvziqU= +github.com/pion/sctp v1.8.35 h1:qwtKvNK1Wc5tHMIYgTDJhfZk7vATGVHhXbUDfHbYwzA= +github.com/pion/sctp v1.8.35/go.mod h1:EcXP8zCYVTRy3W9xtOF7wJm1L1aXfKRQzaM33SjQlzg= +github.com/pion/sdp/v3 v3.0.9 h1:pX++dCHoHUwq43kuwf3PyJfHlwIj4hXA7Vrifiq0IJY= +github.com/pion/sdp/v3 v3.0.9/go.mod h1:B5xmvENq5IXJimIO4zfp6LAe1fD9N+kFv+V/1lOdz8M= +github.com/pion/srtp/v3 v3.0.4 h1:2Z6vDVxzrX3UHEgrUyIGM4rRouoC7v+NiF1IHtp9B5M= +github.com/pion/srtp/v3 v3.0.4/go.mod h1:1Jx3FwDoxpRaTh1oRV8A/6G1BnFL+QI82eK4ms8EEJQ= +github.com/pion/stun/v3 v3.0.0 h1:4h1gwhWLWuZWOJIJR9s2ferRO+W3zA/b6ijOI6mKzUw= +github.com/pion/stun/v3 v3.0.0/go.mod h1:HvCN8txt8mwi4FBvS3EmDghW6aQJ24T+y+1TKjB5jyU= +github.com/pion/transport/v3 v3.0.7 h1:iRbMH05BzSNwhILHoBoAPxoB9xQgOaJk+591KC9P1o0= +github.com/pion/transport/v3 v3.0.7/go.mod h1:YleKiTZ4vqNxVwh77Z0zytYi7rXHl7j6uPLGhhz9rwo= +github.com/pion/turn/v4 v4.0.0 h1:qxplo3Rxa9Yg1xXDxxH8xaqcyGUtbHYw4QSCvmFWvhM= +github.com/pion/turn/v4 v4.0.0/go.mod h1:MuPDkm15nYSklKpN8vWJ9W2M0PlyQZqYt1McGuxG7mA= +github.com/pion/webrtc/v4 v4.0.5 h1:8cVPojcv3cQTwVga2vF1rzCNvkiEimnYdCCG7yF317I= +github.com/pion/webrtc/v4 v4.0.5/go.mod h1:LvP8Np5b/sM0uyJIcUPvJcCvhtjHxJwzh2H2PYzE6cQ= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw= +github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo= +github.com/stretchr/objx v0.5.2/go.mod h1:FRsXN1f5AsAjCGJKqEizvkpNtU+EGNCLh3NxZ/8L+MA= +github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= +github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU= +github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo= +github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= +github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA= +github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= +github.com/wlynxg/anet v0.0.5 h1:J3VJGi1gvo0JwZ/P1/Yc/8p63SoW98B5dHkYDmpgvvU= +github.com/wlynxg/anet v0.0.5/go.mod h1:eay5PRQr7fIVAMbTbchTnO9gG65Hg/uYGdc7mguHxoA= +golang.org/x/crypto v0.29.0 h1:L5SG1JTTXupVV3n6sUqMTeWbjAyfPwoda2DLX8J8FrQ= +golang.org/x/crypto v0.29.0/go.mod h1:+F4F4N5hv6v38hfeYwTdx20oUvLLc+QfrE9Ax9HtgRg= +golang.org/x/net v0.31.0 h1:68CPQngjLL0r2AlUKiSxtQFKvzRVbnzLwMUn5SzcLHo= +golang.org/x/net v0.31.0/go.mod h1:P4fl1q7dY2hnZFxEk4pPSkDHF+QqjitcnDjUQyMM+pM= +golang.org/x/sys v0.27.0 h1:wBqf8DvsY9Y/2P8gAfPDEYNuS30J4lPHJxXSb/nJZ+s= +golang.org/x/sys v0.27.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/enterprise/media-server/internal/server/handlers.go b/enterprise/media-server/internal/server/handlers.go index 2b2f23dec..3051dd4fa 100644 --- a/enterprise/media-server/internal/server/handlers.go +++ b/enterprise/media-server/internal/server/handlers.go @@ -82,6 +82,13 @@ type AgentAnswerRequest struct { SDPAnswer string `json:"sdp_answer"` } +// MetaAnswerRequest is the JSON body for POST /sessions/:id/meta-answer. +// Used for outgoing calls to deliver Meta's SDP answer to the media server +// so it can complete the Peer A (Meta-side) WebRTC handshake. +type MetaAnswerRequest struct { + SDPAnswer string `json:"sdp_answer"` +} + // AgentAnswerResponse is the JSON response for POST /sessions/:id/agent-answer. type AgentAnswerResponse struct { Status string `json:"status"` @@ -316,6 +323,39 @@ func (h *Handlers) AgentAnswer(w http.ResponseWriter, r *http.Request) { }) } +// MetaAnswer handles POST /sessions/{id}/meta-answer. It sets Meta's SDP +// answer on the Meta-side peer connection for outbound calls. +func (h *Handlers) MetaAnswer(w http.ResponseWriter, r *http.Request) { + sessionID := r.PathValue("id") + sess := h.manager.GetSession(sessionID) + if sess == nil { + writeError(w, http.StatusNotFound, "session not found") + return + } + + var req MetaAnswerRequest + if err := readJSON(r, &req); err != nil { + writeError(w, http.StatusBadRequest, "invalid request body: "+err.Error()) + return + } + if req.SDPAnswer == "" { + writeError(w, http.StatusBadRequest, "sdp_answer is required") + return + } + + if err := sess.SetMetaAnswer(req.SDPAnswer); err != nil { + slog.Error("handler: failed to set meta answer", + "session_id", sessionID, + "error", err, + ) + writeError(w, http.StatusInternalServerError, "failed to set meta answer: "+err.Error()) + return + } + + slog.Info("handler: meta answer set", "session_id", sessionID) + writeJSON(w, http.StatusOK, map[string]string{"status": "ok"}) +} + // AgentReconnect handles POST /sessions/{id}/agent-reconnect. It tears down // the old agent peer and creates a new one, returning a fresh SDP offer. func (h *Handlers) AgentReconnect(w http.ResponseWriter, r *http.Request) { @@ -603,7 +643,13 @@ func (h *Handlers) StopInjectAudio(w http.ResponseWriter, r *http.Request) { func readJSON(r *http.Request, v any) error { defer r.Body.Close() limited := io.LimitReader(r.Body, maxRequestBodySize) - return json.NewDecoder(limited).Decode(v) + err := json.NewDecoder(limited).Decode(v) + // Tolerate empty bodies — handlers with all-optional fields treat this + // as "use defaults" rather than failing. + if err == io.EOF { + return nil + } + return err } func writeJSON(w http.ResponseWriter, status int, v any) { diff --git a/enterprise/media-server/internal/server/router.go b/enterprise/media-server/internal/server/router.go index 2ce9a886e..10f75345a 100644 --- a/enterprise/media-server/internal/server/router.go +++ b/enterprise/media-server/internal/server/router.go @@ -47,6 +47,7 @@ func (rt *Router) Build() http.Handler { mux.Handle("GET /sessions/{id}", authMw(http.HandlerFunc(rt.handler.GetSession))) mux.Handle("POST /sessions/{id}/agent-offer", authMw(http.HandlerFunc(rt.handler.AgentOffer))) mux.Handle("POST /sessions/{id}/agent-answer", authMw(http.HandlerFunc(rt.handler.AgentAnswer))) + mux.Handle("POST /sessions/{id}/meta-answer", authMw(http.HandlerFunc(rt.handler.MetaAnswer))) mux.Handle("POST /sessions/{id}/agent-reconnect", authMw(http.HandlerFunc(rt.handler.AgentReconnect))) mux.Handle("POST /sessions/{id}/terminate", authMw(http.HandlerFunc(rt.handler.TerminateSession))) mux.Handle("GET /sessions/{id}/recording", authMw(http.HandlerFunc(rt.handler.GetRecording)))