Add menu option to interrupt a contact

temba/contacts/models.py

@@ -1066,6 +1066,15 @@ def open_ticket(self, user, ticketer, topic, body: str, assignee=None):
         self.modify(user, [mod], refresh=False)
         return self.tickets.order_by("id").last()
 
+    def interrupt(self, user) -> bool:
+        """
+        Interrupts this contact's current flow
+        """
+        if self.current_flow:
+            sessions = mailroom.get_client().contact_interrupt(self.org.id, user.id, self.id)
+            return len(sessions) > 0
+        return False
+
     def block(self, user):
         """
         Blocks this contact removing it from all non-smart groups

temba/contacts/tests.py

@@ -801,6 +801,55 @@ def test_restore(self, mr_mocks):
         other_org_contact.refresh_from_db()
         self.assertEqual(Contact.STATUS_STOPPED, other_org_contact.status)
 
+    @mock_mailroom
+    def test_interrupt(self, mr_mocks):
+        contact = self.create_contact("Joe", phone="+593979000111")
+        other_org_contact = self.create_contact("Hans", phone="+593979123456", org=self.org2)
+
+        read_url = reverse("contacts.contact_read", args=[contact.uuid])
+        interrupt_url = reverse("contacts.contact_interrupt", args=[contact.id])
+
+        self.login(self.admin)
+
+        # no interrupt option if not in a flow
+        response = self.client.get(read_url)
+        self.assertNotContains(response, interrupt_url)
+
+        MockSessionWriter(contact, self.create_flow("Test")).wait().save()
+        MockSessionWriter(other_org_contact, self.create_flow("Test", org=self.org2)).wait().save()
+
+        # now it's an option
+        response = self.client.get(read_url)
+        self.assertContains(response, interrupt_url)
+
+        # can't interrupt if not logged in
+        self.client.logout()
+        response = self.client.post(interrupt_url, {"id": contact.id})
+        self.assertLoginRedirect(response)
+
+        self.login(self.user)
+
+        # can't interrupt if just regular user
+        response = self.client.post(interrupt_url, {"id": contact.id})
+        self.assertLoginRedirect(response)
+
+        self.login(self.admin)
+
+        response = self.client.post(interrupt_url, {"id": contact.id})
+        self.assertEqual(302, response.status_code)
+
+        contact.refresh_from_db()
+        self.assertIsNone(contact.current_flow)
+
+        # can't interrupt contact in other org
+        restore_url = reverse("contacts.contact_interrupt", args=[other_org_contact.id])
+        response = self.client.post(restore_url, {"id": other_org_contact.id})
+        self.assertLoginRedirect(response)
+
+        # contact should be unchanged
+        other_org_contact.refresh_from_db()
+        self.assertIsNotNone(other_org_contact.current_flow)
+
     def test_delete(self):
         contact = self.create_contact("Joe", phone="+593979000111")
         other_org_contact = self.create_contact("Hans", phone="+593979123456", org=self.org2)
@@ -1616,6 +1665,16 @@ def test_open_ticket(self, mock_contact_modify):
         self.assertEqual(self.org.default_ticket_topic, ticket.topic)
         self.assertEqual("Looks sus", ticket.body)
 
+    @mock_mailroom
+    def test_interrupt(self, mr_mocks):
+        # noop when contact not in a flow
+        self.assertFalse(self.joe.interrupt(self.admin))
+
+        flow = self.create_flow("Test")
+        MockSessionWriter(self.joe, flow).wait().save()
+
+        self.assertTrue(self.joe.interrupt(self.admin))
+
     @mock_mailroom
     def test_release(self, mr_mocks):
         # create a contact with a message

temba/contacts/views.py

@@ -564,6 +564,7 @@ class ContactCRUDL(SmartCRUDL):
         "update_fields",
         "update_fields_input",
         "export",
+        "interrupt",
         "block",
         "restore",
         "archive",
@@ -879,6 +880,8 @@ def build_content_menu(self, menu):
                     menu.add_modax(
                         _("Open Ticket"), "open-ticket", reverse("contacts.contact_open_ticket", args=[self.object.id])
                     )
+                if self.has_org_perm("contacts.contact_interrupt") and self.object.current_flow:
+                    menu.add_url_post(_("Interrupt"), reverse("contacts.contact_interrupt", args=(self.object.id,)))
 
             if self.has_org_perm("contacts.contact_update"):
                 menu.add_modax(
@@ -1424,6 +1427,19 @@ def save(self, obj):
         def get_success_url(self):
             return f"{reverse('tickets.ticket_list')}all/open/{self.ticket.uuid}/"
 
+    class Interrupt(OrgObjPermsMixin, SmartUpdateView):
+        """
+        Interrupt this contact
+        """
+
+        fields = ()
+        success_url = "uuid@contacts.contact_read"
+        success_message = ""
+
+        def save(self, obj):
+            obj.interrupt(self.request.user)
+            return obj
+
     class Block(OrgObjPermsMixin, SmartUpdateView):
         """
         Block this contact

temba/mailroom/client.py

@@ -230,14 +230,15 @@ def contact_modify(self, org_id, user_id, contact_ids, modifiers: list[Modifier]
         return self._request("contact/modify", payload)
 
     def contact_resolve(self, org_id: int, channel_id: int, urn: str):
-        payload = {
-            "org_id": org_id,
-            "channel_id": channel_id,
-            "urn": urn,
-        }
+        payload = {"org_id": org_id, "channel_id": channel_id, "urn": urn}
 
         return self._request("contact/resolve", payload)
 
+    def contact_interrupt(self, org_id: int, user_id: int, contact_id: int):
+        payload = {"org_id": org_id, "user_id": user_id, "contact_id": contact_id}
+
+        return self._request("contact/interrupt", payload)
+
     def contact_search(self, org_id, group_uuid, query, sort, offset=0, exclude_ids=()) -> SearchResults:
         payload = {
             "org_id": org_id,

temba/mailroom/tests.py

@@ -348,6 +348,19 @@ def test_contact_resolve(self, mock_post):
             json={"org_id": self.org.id, "channel_id": 345, "urn": "tel:+1234567890"},
         )
 
+    @patch("requests.post")
+    def test_contact_interrupt(self, mock_post):
+        mock_post.return_value = MockResponse(200, '{"sessions": 1}')
+
+        response = get_client().contact_interrupt(self.org.id, 3, 345)
+
+        self.assertEqual({"sessions": 1}, response)
+        mock_post.assert_called_once_with(
+            "http://localhost:8090/mr/contact/interrupt",
+            headers={"User-Agent": "Temba"},
+            json={"org_id": self.org.id, "user_id": 3, "contact_id": 345},
+        )
+
     @patch("requests.post")
     def test_contact_search(self, mock_post):
         mock_post.return_value = MockResponse(

temba/settings_common.py

@@ -395,6 +395,7 @@
         "export",
         "filter",
         "history",
+        "interrupt",
         "menu",
         "omnibox",
         "open_ticket",
@@ -610,6 +611,7 @@
         "contacts.contact_export",
         "contacts.contact_filter",
         "contacts.contact_history",
+        "contacts.contact_interrupt",
         "contacts.contact_list",
         "contacts.contact_menu",
         "contacts.contact_omnibox",
@@ -743,6 +745,7 @@
         "contacts.contact_export",
         "contacts.contact_filter",
         "contacts.contact_history",
+        "contacts.contact_interrupt",
         "contacts.contact_list",
         "contacts.contact_menu",
         "contacts.contact_omnibox",

temba/tests/mailroom.py

@@ -16,6 +16,7 @@
 from temba import mailroom
 from temba.campaigns.models import CampaignEvent, EventFire
 from temba.contacts.models import URN, Contact, ContactField, ContactGroup, ContactURN
+from temba.flows.models import FlowRun, FlowSession
 from temba.locations.models import AdminBoundary
 from temba.mailroom.client import ContactSpec, MailroomClient, MailroomException
 from temba.mailroom.modifiers import Modifier
@@ -172,6 +173,17 @@ def contact_resolve(self, org_id: int, channel_id: int, urn: str):
             "urn": {"id": contact_urn.id, "identity": contact_urn.identity},
         }
 
+    @_client_method
+    def contact_interrupt(self, org_id: int, user_id: int, contact_id: int):
+        contact = Contact.objects.get(id=contact_id)
+
+        # get the waiting session IDs
+        session_ids = list(contact.sessions.filter(status=FlowSession.STATUS_WAITING).values_list("id", flat=True))
+
+        exit_sessions(session_ids, FlowSession.STATUS_INTERRUPTED)
+
+        return {"sessions": len(session_ids)}
+
     @_client_method
     def parse_query(self, org_id: int, query: str, parse_only: bool = False, group_uuid: str = ""):
         org = Org.objects.get(id=org_id)
@@ -669,3 +681,22 @@ def decrement_credit(org):
             r.decr(f"org:{org.id}:cache:credits_remaining:{active_topup_id}", 1)
 
     return active_topup_id or None
+
+
+def exit_sessions(session_ids: list, status: str):
+    FlowRun.objects.filter(session_id__in=session_ids).update(
+        status=status, exited_on=timezone.now(), modified_on=timezone.now()
+    )
+    FlowSession.objects.filter(id__in=session_ids).update(
+        status=status,
+        ended_on=timezone.now(),
+        wait_started_on=None,
+        wait_expires_on=None,
+        timeout_on=None,
+        current_flow_id=None,
+    )
+
+    for session in FlowSession.objects.filter(id__in=session_ids):
+        session.contact.current_flow = None
+        session.contact.modified_on = timezone.now()
+        session.contact.save(update_fields=("current_flow", "modified_on"))