Tests passing again

components/app/app/call_controllers/call_controller.rb

@@ -13,12 +13,14 @@ def run
     execute_twiml(twiml)
   end
 
-  def redirect(url = nil, http_method = nil, params = {})
-    twiml = request_twiml(
-      url,
-      http_method,
-      params.reverse_merge("CallStatus" => "in-progress")
-    )
+  def redirect(**options)
+    twiml = options.fetch(:twiml) do
+      request_twiml(
+        options.fetch(:url),
+        options[:http_method],
+        options.fetch(:params, {}).reverse_merge("CallStatus" => "in-progress")
+      )
+    end
 
     execute_twiml(twiml)
   end

components/app/app/models/call_update_event_handler.rb

@@ -3,17 +3,34 @@ class CallUpdateEventHandler
   CHANNEL_PREFIX = "call_updates".freeze
 
   class Event
-    attr_reader :payload
+    attr_reader :call_id, :voice_url, :voice_method, :twiml
 
     def initialize(**params)
-      @payload = params.fetch(:payload)
+      @call_id = params.fetch(:call_id)
+      @voice_url = params[:voice_url]
+      @voice_method = params[:voice_method]
+      @twiml = params[:twiml]
     end
 
     def self.parse(payload)
       message = JSON.parse(payload)
 
       new(
-        payload: message
+        call_id: message.fetch("id"),
+        voice_url: message["voice_url"],
+        voice_method: message["voice_method"],
+        twiml: message["twiml"]
+      )
+    end
+
+    def serialize
+      JSON.generate(
+        {
+          id: call_id,
+          voice_url:,
+          voice_method:,
+          twiml:
+        }.compact
       )
     end
   end
@@ -36,11 +53,23 @@ def parse_event(message)
     Event.parse(message)
   end
 
+  def build_event(**)
+    Event.new(**)
+  end
+
   def perform_later(event)
     queue.push(event)
   end
 
   def perform_now(event)
+    throw(
+      :redirect,
+      {
+        url: event.voice_url,
+        http_method: event.voice_method,
+        twiml: event.twiml
+      }.compact
+    )
   end
 
   def perform_queued
@@ -51,7 +80,7 @@ def perform_queued
 
   def fake_event(**params)
     Event.new(
-      payload: {},
+      call_id: SecureRandom.uuid,
       **params
     )
   end

components/app/app/request_schemas/update_call_request_schema.rb

@@ -0,0 +1,11 @@
+class UpdateCallRequestSchema
+  attr_reader :params
+
+  def initialize(params)
+    @params = params.with_indifferent_access
+  end
+
+  def output
+    params.slice(:voice_url, :voice_method, :twiml).symbolize_keys
+  end
+end

components/app/app/web/api.rb

@@ -21,8 +21,14 @@ class API < Application
       end
 
       patch "/calls/:id" do
-        AppSettings.redis.with do |redis|
-          redis.set("call_updates:#{params[:id]}", JSON.parse(request.body.read), ex: 1.day.seconds)
+        AppSettings.redis.with do |connection|
+          event_handler = CallUpdateEventHandler.new
+          request_schema = UpdateCallRequestSchema.new(JSON.parse(request.body.read))
+
+          connection.publish(
+            event_handler.channel_for(params[:id]),
+            event_handler.build_event(call_id: params[:id], **request_schema.output).serialize
+          )
         end
 
         return status(204)

components/app/app/workflows/execute_dial.rb

@@ -20,7 +20,14 @@ def call
   private
 
   def redirect(params)
-    throw(:redirect, [ verb.action, verb.method, params ])
+    throw(
+      :redirect,
+      {
+        url: verb.action,
+        http_method: verb.method,
+        params:
+      }
+    )
   end
 
   def build_callback_params(dial_status)

components/app/app/workflows/execute_gather.rb

@@ -29,7 +29,14 @@ def call
   private
 
   def redirect(params)
-    throw(:redirect, [ verb.action, verb.method, params ])
+    throw(
+      :redirect,
+      {
+        url: verb.action,
+        http_method: verb.method,
+        params:
+      }
+    )
   end
 
   def build_callback_params(digits)

components/app/app/workflows/execute_record.rb

@@ -43,7 +43,14 @@ def normalize_recording_url(raw_recording_url)
   end
 
   def redirect(params)
-    throw(:redirect, [ verb.action, verb.method, params ])
+    throw(
+      :redirect,
+      {
+        url: verb.action,
+        http_method: verb.method,
+        params:
+      }
+    )
   end
 
   def build_callback_params(response, record_result)

components/app/app/workflows/execute_redirect.rb

@@ -10,6 +10,12 @@ def call
   private
 
   def redirect
-    throw(:redirect, [ verb.content, verb.method ])
+    throw(
+      :redirect,
+      {
+        url: verb.content,
+        http_method: verb.method
+      }
+    )
   end
 end

components/app/app/workflows/execute_twiml.rb

@@ -47,7 +47,7 @@ def call
       false
     end
 
-    context.redirect(*redirect_args) if redirect_args.present?
+    context.redirect(**redirect_args) if redirect_args.present?
   rescue Errors::TwiMLError => e
     logger.error(e.message)
   end

components/app/spec/call_controllers/connect_spec.rb

@@ -38,7 +38,6 @@
             stub_voice_commands: :play_audio,
             call_properties: { call_sid: vcr_call_sid }
           )
-
           stub_twilio_stream(controller)
           stub_twiml_request(controller, response: <<~TWIML)
             <?xml version="1.0" encoding="UTF-8"?>
@@ -91,27 +90,111 @@
             )
           end
         end
+
+        it "handles call url updates" do
+          controller = build_controller(
+            call_properties: { call_sid: vcr_call_sid }
+          )
+          call_update_event_handler = CallUpdateEventHandler.new
+          stub_twilio_stream(
+            controller,
+            other_messages: {
+              call_update_event_handler.channel_for(controller.call.id) => [
+                call_update_event_handler.build_event(
+                  call_id: controller.call.id,
+                  voice_url: "https://www.example.com/redirect.xml",
+                  voice_method: "POST"
+                ).serialize
+              ]
+            }
+          )
+          stub_twiml_request(controller, response: <<~TWIML)
+            <?xml version="1.0" encoding="UTF-8"?>
+            <Response>
+              <Connect>
+                <Stream url="wss://mystream.ngrok.io/audiostream" />
+              </Connect>
+              <Play>https://api.twilio.com/cowbell.mp3</Play>
+            </Response>
+          TWIML
+
+          stub_request(:any, "https://www.example.com/redirect.xml").to_return(body: <<~TWIML)
+          <?xml version="1.0" encoding="UTF-8" ?>
+            <Response>
+              <Hangup/>
+            </Response>
+          TWIML
+
+          controller.run
+
+          expect(controller).to have_received(:write_and_await_response).with(an_instance_of(Rayo::Command::TwilioStream::Start))
+          expect(controller).to have_received(:write_and_await_response).with(an_instance_of(Rayo::Command::TwilioStream::Stop))
+          expect(WebMock).to have_requested(:post, "https://www.example.com/redirect.xml")
+        end
+
+
+        it "handles call twiml updates" do
+          controller = build_controller(
+            call_properties: { call_sid: vcr_call_sid },
+            stub_voice_commands: :play_audio
+          )
+          call_update_event_handler = CallUpdateEventHandler.new
+          stub_twilio_stream(
+            controller,
+            other_messages: {
+              call_update_event_handler.channel_for(controller.call.id) => [
+                call_update_event_handler.build_event(
+                  call_id: controller.call.id,
+                  twiml: "<Response><Play>https://www.example.com/new-audio.mp3</Play></Response>",
+                ).serialize
+              ]
+            }
+          )
+          stub_twiml_request(controller, response: <<~TWIML)
+            <?xml version="1.0" encoding="UTF-8"?>
+            <Response>
+              <Connect>
+                <Stream url="wss://mystream.ngrok.io/audiostream" />
+              </Connect>
+              <Play>https://api.twilio.com/cowbell.mp3</Play>
+            </Response>
+          TWIML
+
+          controller.run
+
+          expect(controller).to have_received(:write_and_await_response).with(an_instance_of(Rayo::Command::TwilioStream::Start))
+          expect(controller).to have_received(:write_and_await_response).with(an_instance_of(Rayo::Command::TwilioStream::Stop))
+          expect(controller).to have_received(:play_audio).with("https://www.example.com/new-audio.mp3")
+          expect(controller).not_to have_received(:play_audio).with("https://api.twilio.com/cowbell.mp3")
+        end
       end
     end
   end
 
-  def stub_twilio_stream(controller, with_events: [])
+  def stub_twilio_stream(controller, with_events: [], **options)
     allow(controller).to receive(:write_and_await_response)
 
-    AppSettings.redis.with do |redis|
-      build_twilio_stream_events(Array(with_events)).each do |event|
-        redis.publish_later("mod_twilio_stream:*", event)
+    AppSettings.redis.with do |connection|
+      build_twilio_stream_events(Array(with_events), **options).each do |(channel, message)|
+        connection.publish_later(channel, message)
       end
     end
   end
 
-  def build_twilio_stream_events(events)
-    result = [
-      build_twilio_stream_event(event: "connect"),
-      build_twilio_stream_event(event: "start")
-    ]
-    Array(events).each { |event| result.push(build_twilio_stream_event(**event)) }
-    result.push(build_twilio_stream_event(event: "disconnect"))
+  def build_twilio_stream_events(events, **options)
+    channel_name = options.fetch(:channel_name) { ExecuteConnect::EventHandler.new.channel_for("*") }
+    other_messages = options.fetch(:other_messages, {})
+
+    result = []
+    result << [ channel_name, build_twilio_stream_event(event: "connect") ]
+    result << [ channel_name, build_twilio_stream_event(event: "start") ]
+    Array(events).each { |event| result << [ channel_name, build_twilio_stream_event(**event) ] }
+    other_messages.each do |channel, messages|
+      messages.each do |message|
+        result << [ channel, message ]
+      end
+    end
+    result << [ channel_name, build_twilio_stream_event(event: "disconnect") ]
   end
 
   def build_twilio_stream_event(event:, **attributes)