Respect survey schedules before sending SMS (#2380)

Author: mverzilli

lib/ask/runtime/channel_broker.ex

@@ -36,8 +36,8 @@ defmodule Ask.Runtime.ChannelBroker do
     call(channel_id, {:has_delivery_confirmation?})
   end
 
-  def ask(channel_id, channel_type, respondent, token, reply) do
-    cast(channel_id, {:ask, channel_type, respondent, token, reply})
+  def ask(channel_id, channel_type, respondent, token, reply, not_before \\ nil, not_after \\ nil) do
+    cast(channel_id, {:ask, channel_type, respondent, token, reply, not_before, not_after})
   end
 
   def has_queued_message?(channel_id, respondent_id) do
@@ -145,16 +145,18 @@ defmodule Ask.Runtime.ChannelBroker do
   end
 
   @impl true
-  def handle_cast({:ask, "sms", respondent, token, reply}, state) do
+  def handle_cast({:ask, "sms", respondent, token, reply, not_before, not_after}, state) do
     debug("handle_cast[ask]",
       channel_type: "sms",
       channel_id: state.channel_id,
       respondent_id: respondent.id,
       token: token,
-      reply: reply
+      reply: reply,
+      not_before: not_before,
+      not_after: not_after
     )
 
-    contact = {respondent, token, reply}
+    contact = {respondent, token, reply, not_before, not_after}
     refreshed_state = refresh_runtime_channel(state)
     size = messages_count(refreshed_state, respondent, reply)
 
@@ -410,32 +412,41 @@ defmodule Ask.Runtime.ChannelBroker do
   # Activates the next queued contact. There must be at least one contact in
   # queue. It doesn't verify if the channel capacity has been reached!
   defp activate_next_queued_contact(state) do
-    {new_state, unqueued_item} =
+    {new_broker_state, unqueued_item} =
       state
       |> refresh_runtime_channel()
       |> State.activate_next_in_queue()
 
     case unqueued_item do
       {respondent_id, token, not_before, not_after} ->
-        Respondent.with_lock(respondent_id, fn respondent ->
-          cond do
-            expired_call?(not_after) ->
-              new_state = State.deactivate_contact(new_state, respondent.id)
-              Ask.Runtime.Survey.contact_attempt_expired(respondent)
-              new_state
-
-            true ->
-              ivr_call(new_state, respondent, token, not_before, not_after)
-          end
+        contact_or_mark_as_expired(respondent_id, not_after, new_broker_state, fn (respondent) -> 
+          ivr_call(new_broker_state, respondent, token, not_before, not_after)
         end)
-      {respondent_id, token, reply} ->
-        Respondent.with_lock(respondent_id, fn respondent ->
-          channel_ask(new_state, respondent, token, reply)
+      {respondent_id, token, reply, _not_before, not_after} ->
+        contact_or_mark_as_expired(respondent_id, not_after, new_broker_state, fn (respondent) ->
+          channel_ask(new_broker_state, respondent, token, reply)
         end)
-    end
+      end
   end
 
-  defp expired_call?(not_after) do
+  defp contact_or_mark_as_expired(respondent_id, not_after, new_broker_state, do_contact) do
+    Respondent.with_lock(respondent_id, fn respondent ->
+      cond do
+        expired_contact_attempt?(not_after) ->
+          new_state = State.deactivate_contact(new_broker_state, respondent.id)
+          Ask.Runtime.Survey.contact_attempt_expired(respondent)
+          new_state
+
+        true ->
+          do_contact.(respondent)
+      end
+    end)
+  end
+
+  defp expired_contact_attempt?(nil) do
+    false
+  end
+  defp expired_contact_attempt?(not_after) do
     DateTime.compare(not_after, Ask.SystemTime.time().now) != :gt
   end
 

lib/ask/runtime/channel_broker_state.ex

@@ -94,16 +94,16 @@ defmodule Ask.Runtime.ChannelBrokerState do
   end
 
   # Adds an SMS contact to the queue with given priority (`:high`, `:normal`).
-  def queue_contact(state, {respondent, token, reply}, size, priority) do
+  def queue_contact(state, {respondent, token, reply, not_before, not_after}, size, priority) do
     Queue.upsert!(%{
       channel_id: state.channel_id,
       respondent_id: respondent.id,
       queued_at: SystemTime.time().now,
       priority: priority,
       size: size,
       token: token,
-      not_before: nil,
-      not_after: nil,
+      not_before: not_before,
+      not_after: not_after,
       reply: reply
     })
 
@@ -173,7 +173,7 @@ defmodule Ask.Runtime.ChannelBrokerState do
   end
 
   defp to_item("sms", contact) do
-    {contact.respondent_id, contact.token, contact.reply}
+    {contact.respondent_id, contact.token, contact.reply, contact.not_before, contact.not_after}
   end
 
   # Increments the number of contacts for the respondent. Activates the contact

lib/ask/runtime/session.ex

@@ -245,30 +245,27 @@ defmodule Ask.Runtime.Session do
     end
   end
 
-  def contact_respondent(%{current_mode: %SMSMode{}} = session) do
+  def contact_respondent(%{schedule: schedule, current_mode: %SMSMode{}} = session) do
     token = Ecto.UUID.generate()
 
     respondent = session.respondent
     {:ok, _flow, reply} = Flow.retry(session.flow, TextVisitor.new("sms"), respondent.disposition)
     channel = session.current_mode.channel
     log_prompts(reply, channel, session.flow.mode, respondent)
 
-    ChannelBroker.ask(channel.id, channel.type, session.respondent, token, reply)
+    {not_before, not_after} = acceptable_contact_time_window(schedule)
+
+    ChannelBroker.ask(channel.id, channel.type, session.respondent, token, reply, not_before, not_after)
 
+    # TODO: what happens with this when contact attempt falls outside acceptable window
     respondent = Respondent.update_stats(respondent.id, reply)
     %{session | token: token, respondent: respondent}
   end
 
   def contact_respondent(%{schedule: schedule, current_mode: %IVRMode{}} = session) do
     token = Ecto.UUID.generate()
 
-    next_available_date_time =
-      schedule
-      |> Schedule.next_available_date_time()
-
-    today_end_time =
-      schedule
-      |> Schedule.at_end_time(next_available_date_time)
+    {not_before, not_after} = acceptable_contact_time_window(schedule)
 
     channel = session.current_mode.channel
 
@@ -277,8 +274,8 @@ defmodule Ask.Runtime.Session do
       channel.type,
       session.respondent,
       token,
-      next_available_date_time,
-      today_end_time
+      not_before,
+      not_after
     )
 
     %{session | token: token}
@@ -388,9 +385,13 @@ defmodule Ask.Runtime.Session do
            flow: flow,
            respondent: respondent,
            token: token,
-           current_mode: %SMSMode{channel: channel}
+           current_mode: %SMSMode{channel: channel},
+           schedule: schedule
          } = session
        ) do
+
+    {not_before, not_after} = acceptable_contact_time_window(schedule)
+
     case flow
          |> Flow.step(
            session.current_mode |> SessionMode.visitor(),
@@ -401,7 +402,7 @@ defmodule Ask.Runtime.Session do
         if Reply.prompts(reply) != [] do
           log_prompts(reply, channel, flow.mode, respondent, true)
 
-          ChannelBroker.ask(channel.id, channel.type, respondent, token, reply)
+          ChannelBroker.ask(channel.id, channel.type, respondent, token, reply, not_before, not_after)
 
           respondent = Respondent.update_stats(respondent.id, reply)
           {:end, reply, respondent}
@@ -422,7 +423,7 @@ defmodule Ask.Runtime.Session do
 
         log_prompts(reply, channel, flow.mode, respondent)
 
-        ChannelBroker.ask(channel.id, channel.type, respondent, token, reply)
+        ChannelBroker.ask(channel.id, channel.type, respondent, token, reply, not_before, not_after)
 
         respondent = Respondent.update_stats(respondent.id, reply)
         {:ok, %{session | flow: flow, respondent: respondent}, reply, current_timeout(session)}
@@ -472,21 +473,16 @@ defmodule Ask.Runtime.Session do
            schedule: schedule
          } = session
        ) do
-    next_available_date_time =
-      schedule
-      |> Schedule.next_available_date_time()
 
-    today_end_time =
-      schedule
-      |> Schedule.at_end_time(next_available_date_time)
+    {not_before, not_after} = acceptable_contact_time_window(schedule)
 
     ChannelBroker.setup(
       channel.id,
       channel.type,
       respondent,
       token,
-      next_available_date_time,
-      today_end_time
+      not_before,
+      not_after
     )
 
     {:ok, %{session | respondent: respondent}, %Reply{}, current_timeout(session)}
@@ -1043,4 +1039,16 @@ defmodule Ask.Runtime.Session do
     until = Schedule.next_available_date_time(session.schedule)
     Interval.new(from: from, until: until) |> Interval.duration(:minutes)
   end
+
+  defp acceptable_contact_time_window(schedule) do
+    not_before =
+      schedule
+      |> Schedule.next_available_date_time()
+
+    not_after =
+      schedule
+      |> Schedule.at_end_time(not_before)
+
+    {not_before, not_after}
+  end
 end

test/ask/runtime/channel_broker_state_test.exs

@@ -68,10 +68,13 @@ defmodule Ask.Runtime.ChannelBrokerStateTest do
       now = DateTime.utc_now()
       mock_time(now)
 
+      five_seconds_ago = DateTime.add(now, -5, :second)
+      in_one_minute = DateTime.add(now, 60, :second)
+
       State.new(0, "sms", %{})
-      |> State.queue_contact({%{id: 2, disposition: :queued}, "secret", []}, 5)
+      |> State.queue_contact({%{id: 2, disposition: :queued}, "secret", [], five_seconds_ago, in_one_minute}, 5)
 
-      assert [%{respondent_id: 2, size: 5, queued_at: now, reply: []}] = Queue.queued_contacts(0)
+      assert [%{respondent_id: 2, size: 5, queued_at: now, reply: [], not_before: five_seconds_ago, not_after: in_one_minute}] = Queue.queued_contacts(0)
       assert [] = Queue.active_contacts(0)
     end
 
@@ -214,10 +217,10 @@ defmodule Ask.Runtime.ChannelBrokerStateTest do
 
       {_, contact} =
         State.new(0, "sms", %{})
-        |> State.queue_contact({respondent, "secret", []}, 2)
+        |> State.queue_contact({respondent, "secret", [], nil, nil}, 2)
         |> State.activate_next_in_queue()
 
-      assert {^respondent_id, "secret", []} = contact
+      assert {^respondent_id, "secret", [], nil, nil} = contact
     end
 
     @tag :time_mock

test/ask/runtime/channel_broker_test.exs

@@ -241,6 +241,49 @@ defmodule Ask.Runtime.ChannelBrokerTest do
   end
 
   describe "Nuntium" do
+    defp activate_respondent(state, "sms", respondent, not_before, not_after) do
+      {_, state, _} =
+        ChannelBroker.handle_cast(
+          {:ask, "sms", respondent, "token", "foo", not_before, not_after},
+          state
+        )
+
+      state
+    end
+
+    test "Respect schedules for SMS" do
+      %{state: state, respondents: respondents, test_channel: test_channel} = build_survey("sms")
+
+      now = SystemTime.time().now
+      not_before = DateTime.add(now, 5, :second)
+      not_after = DateTime.add(now, 60, :second)
+      expired = DateTime.add(now, -5, :second)
+
+      with_mock Ask.Runtime.Survey, [contact_attempt_expired: fn _ -> :ok end] do
+        state =
+          state
+          |> activate_respondent("sms", Enum.at(respondents, 0), now, not_after)
+          |> activate_respondent("sms", Enum.at(respondents, 1), not_before, expired)
+          |> activate_respondent("sms", Enum.at(respondents, 2), not_before, expired)
+          |> activate_respondent("sms", Enum.at(respondents, 3), not_before, not_after)
+          |> activate_respondent("sms", Enum.at(respondents, 4), not_before, not_after)
+
+        expected_active_respondents = [
+          Enum.at(respondents, 0),
+          Enum.at(respondents, 3),
+          Enum.at(respondents, 4)
+        ]
+
+        assert Enum.map(expected_active_respondents, fn r -> r.id end) == active_respondent_ids(state)
+
+        assert_sent_smss(expected_active_respondents, test_channel)
+        assert_not_sent_smss([
+          Enum.at(respondents, 1),
+          Enum.at(respondents, 2),
+        ], test_channel)
+      end
+    end
+
     test "Every SMS is sent when capacity isn't set", %{} do
       channel_capacity = nil
       [test_channel, respondents, _channel] = initialize_survey("sms", channel_capacity)
@@ -466,6 +509,12 @@ defmodule Ask.Runtime.ChannelBrokerTest do
     refute_received [:ask, ^test_channel, _respondent, _token, _reply, _channel_id]
   end
 
+  defp assert_not_sent_smss(respondents, test_channel) do
+    Enum.each(respondents, fn %{id: id} ->
+      refute_received [:ask, ^test_channel, %{id: ^id}, _token, _reply, _channel_id]
+    end)
+  end
+
   defp assert_some_sent_smss(amount, respondents, test_channel) do
     respondent_ids = Enum.map(respondents, & &1.id)
 
@@ -511,7 +560,7 @@ defmodule Ask.Runtime.ChannelBrokerTest do
       config: Config.channel_broker_config()
     }
 
-    %{state: state, respondents: respondents, channel: channel}
+    %{state: state, respondents: respondents, channel: channel, test_channel: test_channel}
   end
 
   defp start_survey(channel_type) do
@@ -547,7 +596,9 @@ defmodule Ask.Runtime.ChannelBrokerTest do
   end
 
   defp respondent_to_contact("sms", respondent) do
-    {respondent, "secret", []}
+    not_before = SystemTime.time().now |> DateTime.add(-3600, :second)
+    not_after = SystemTime.time().now |> DateTime.add(3600, :second)
+    {respondent, "secret", [], not_before, not_after}
   end
 
   defp channel_state("ivr", respondent) do