Resend Broadcast

kairon/api/app/routers/bot/channels.py

@@ -182,6 +182,19 @@ async def add_message_broadcast_event(
     return Response(message="Broadcast added!", data={"msg_broadcast_id": notification_id})
 
 
+@router.post("/broadcast/message/resend/{msg_broadcast_id}", response_model=Response)
+async def resend_message_broadcast_event(
+        msg_broadcast_id: str,
+        current_user: User = Security(Authentication.get_current_user_and_bot, scopes=DESIGNER_ACCESS)
+):
+    """
+    Resends a scheduled message broadcast.
+    """
+    event = MessageBroadcastEvent(current_user.get_bot(), current_user.get_user())
+    event.enqueue(EventRequestType.resend_broadcast.value, msg_broadcast_id=msg_broadcast_id)
+    return Response(message="Resending Broadcast!")
+
+
 @router.put("/broadcast/message/{msg_broadcast_id}", response_model=Response)
 async def update_message_broadcast_event(
         msg_broadcast_id: str, request: MessageBroadcastRequest,

kairon/events/definitions/message_broadcast.py

@@ -50,11 +50,15 @@ def execute(self, event_id: Text, **kwargs):
         reference_id = None
         status = EVENT_STATUS.FAIL.value
         exception = None
+        is_resend = kwargs.get('is_resend', False)
         try:
-            config, reference_id = self.__retrieve_config(event_id)
+            config, reference_id = self.__retrieve_config(event_id, is_resend)
             broadcast = MessageBroadcastFactory.get_instance(config["connector_type"]).from_config(config, event_id, reference_id)
-            recipients = broadcast.get_recipients()
-            broadcast.send(recipients)
+            if is_resend:
+                broadcast.resend_broadcast()
+            else:
+                recipients = broadcast.get_recipients()
+                broadcast.send(recipients)
             status = EVENT_STATUS.COMPLETED.value
         except Exception as e:
             logger.exception(e)
@@ -99,6 +103,16 @@ def _add_schedule(self, config: Dict):
                 MessageBroadcastProcessor.delete_task(msg_broadcast_id, self.bot)
             raise AppException(e)
 
+    def _resend_broadcast(self, msg_broadcast_id: Text):
+        try:
+            payload = {'bot': self.bot, 'user': self.user,
+                       "event_id": msg_broadcast_id, "is_resend": True}
+            Utility.request_event_server(EventClass.message_broadcast, payload)
+            return msg_broadcast_id
+        except Exception as e:
+            logger.error(e)
+            raise e
+
     def _update_schedule(self, msg_broadcast_id: Text, config: Dict):
         settings_updated = False
         current_settings = {}
@@ -130,13 +144,15 @@ def delete_schedule(self, msg_broadcast_id: Text):
             logger.error(e)
             raise e
 
-    def __retrieve_config(self, event_id: Text):
-        reference_id = ObjectId().__str__()
-        config = MessageBroadcastProcessor.get_settings(event_id, self.bot)
+    def __retrieve_config(self, event_id: Text, is_resend: bool):
+
+        reference_id = MessageBroadcastProcessor.get_reference_id_from_broadcasting_logs(event_id) \
+            if is_resend else ObjectId().__str__()
+        config = MessageBroadcastProcessor.get_settings(event_id, self.bot, is_resend=is_resend)
         bot_settings = MongoProcessor.get_bot_settings(self.bot, self.user)
         config["pyscript_timeout"] = bot_settings["dynamic_broadcast_execution_timeout"]
         MessageBroadcastProcessor.add_event_log(
             self.bot, MessageBroadcastLogType.common.value, reference_id, user=self.user, config=config,
-            status=EVENT_STATUS.INPROGRESS.value, event_id=event_id, is_new_log=True
+            status=EVENT_STATUS.INPROGRESS.value, event_id=event_id, is_new_log=True, is_resend=is_resend
         )
         return config, reference_id

kairon/events/definitions/scheduled_base.py

@@ -15,7 +15,8 @@ def enqueue(self, event_request_type: Text, **kwargs):
         request_implementation = {
             EventRequestType.trigger_async.value: self._trigger_async,
             EventRequestType.add_schedule.value: self._add_schedule,
-            EventRequestType.update_schedule.value: self._update_schedule
+            EventRequestType.update_schedule.value: self._update_schedule,
+            EventRequestType.resend_broadcast.value: self._resend_broadcast
         }
         if event_request_type not in request_implementation.keys():
             raise AppException(f"'{event_request_type}' is not a valid event server request!")
@@ -34,6 +35,10 @@ def _add_schedule(self, config: Dict):
     def _update_schedule(self, msg_broadcast_id: Text, config: Dict):
         raise NotImplementedError("Provider not implemented")
 
+    @abstractmethod
+    def _resend_broadcast(self, msg_broadcast_id: Text):
+        raise NotImplementedError("Provider not implemented")
+
     @abstractmethod
     def delete_schedule(self, msg_broadcast_id: Text):
         raise NotImplementedError("Provider not implemented")

kairon/shared/channels/broadcast/whatsapp.py

@@ -47,7 +47,7 @@ def send(self, recipients: List, **kwargs):
         else:
             self.__send_using_pyscript()
 
-    def __send_using_pyscript(self):
+    def __send_using_pyscript(self, **kwargs):
         from kairon.shared.concurrency.orchestrator import ActorOrchestrator
 
         script = self.config['pyscript']
@@ -62,7 +62,8 @@ def send_msg(template_id: Text, recipient, language_code: Text = "en", component
             MessageBroadcastProcessor.add_event_log(
                 self.bot, MessageBroadcastLogType.send.value, self.reference_id, api_response=response,
                 status=status, recipient=recipient, template_params=components, template=raw_template,
-                event_id=self.event_id, template_name=template_id
+                event_id=self.event_id, template_name=template_id, language_code=language_code, namespace=namespace,
+                resend_count=0
             )
 
             return response
@@ -107,21 +108,51 @@ def __send_using_configuration(self, recipients: List):
             for recipient, t_params in zip(recipients, template_params):
                 recipient = str(recipient) if recipient else ""
                 if not Utility.check_empty_string(recipient):
-                    response = channel_client.send_template_message(template_id, recipient, lang, t_params, namespace=namespace)
+                    response = channel_client.send_template_message(template_id, recipient, lang, t_params,
+                                                                    namespace=namespace)
                     status = "Failed" if response.get("errors") else "Success"
                     if status == "Failed":
                         failure_cnt = failure_cnt + 1
 
                     MessageBroadcastProcessor.add_event_log(
                         self.bot, MessageBroadcastLogType.send.value, self.reference_id, api_response=response,
                         status=status, recipient=recipient, template_params=t_params, template=raw_template,
-                        event_id=self.event_id, template_name=template_id
+                        event_id=self.event_id, template_name=template_id, language_code=lang, namespace=namespace,
+                        resend_count=0
                     )
             MessageBroadcastProcessor.add_event_log(
                 self.bot, MessageBroadcastLogType.common.value, self.reference_id, failure_cnt=failure_cnt, total=total,
                 event_id=self.event_id, **evaluation_log
             )
 
+    def resend_broadcast(self):
+        channel_client = self.__get_client()
+
+        message_broadcast_logs, resend_count = MessageBroadcastProcessor.extract_message_ids_from_broadcast_logs(self.reference_id)
+
+        required_logs = [log for log in message_broadcast_logs.values() if log["errors"]]
+        codes_to_exclude = Utility.environment["channels"]["360dialog"]["error_codes"]
+        required_logs = [log for log in required_logs if log["errors"][0]["code"] not in codes_to_exclude]
+
+        for log in required_logs:
+            template_id = log["template_name"]
+            namespace = log["namespace"]
+            language_code = log["language_code"]
+            components = log["template_params"]
+            recipient = log["recipient"]
+            template = log["template"]
+            resend_count = log["resend_count"] + 1
+            response = channel_client.send_template_message(template_id, recipient, language_code, components,
+                                                            namespace)
+            status = "Failed" if response.get("error") else "Success"
+
+            MessageBroadcastProcessor.add_event_log(
+                self.bot, MessageBroadcastLogType.send.value, self.reference_id, api_response=response,
+                status=status, recipient=recipient, template_params=components, template=template,
+                event_id=self.event_id, template_name=template_id, language_code=language_code, namespace=namespace,
+                resend_count=resend_count
+            )
+
     def __get_client(self):
         try:
             bot_settings = MongoProcessor.get_bot_settings(self.bot, self.user)

kairon/shared/chat/broadcast/constants.py

@@ -11,3 +11,4 @@ class MessageBroadcastLogType(str, Enum):
 class MessageBroadcastType(str, Enum):
     static = "static"
     dynamic = "dynamic"
+

kairon/shared/chat/broadcast/processor.py

@@ -18,9 +18,10 @@
 class MessageBroadcastProcessor:
 
     @staticmethod
-    def get_settings(notification_id: Text, bot: Text):
+    def get_settings(notification_id: Text, bot: Text, **kwargs):
         try:
-            settings = MessageBroadcastSettings.objects(id=notification_id, bot=bot, status=True).get()
+            status = not kwargs.get("is_resend", False)
+            settings = MessageBroadcastSettings.objects(id=notification_id, bot=bot, status=status).get()
             settings = settings.to_mongo().to_dict()
             settings["_id"] = settings["_id"].__str__()
             return settings
@@ -81,10 +82,12 @@ def delete_task(notification_id: Text, bot: Text, delete_permanently: bool = Tru
 
     @staticmethod
     def add_event_log(bot: Text, log_type: Text, reference_id: Text = None, status: Text = None, **kwargs):
-        event_completion_states = [EVENT_STATUS.FAIL.value, EVENT_STATUS.COMPLETED.value]
-        is_new_log = log_type in {MessageBroadcastLogType.send.value, MessageBroadcastLogType.self.value} or kwargs.pop("is_new_log", None)
+        is_resend = kwargs.pop("is_resend", False)
+        event_completion_states = [] if is_resend else [EVENT_STATUS.FAIL.value, EVENT_STATUS.COMPLETED.value]
+        is_new_log = log_type in {MessageBroadcastLogType.send.value,
+                                  MessageBroadcastLogType.self.value} or kwargs.pop("is_new_log", None)
         try:
-            if is_new_log:
+            if is_new_log and not is_resend:
                 raise DoesNotExist()
             log = MessageBroadcastLogs.objects(bot=bot, reference_id=reference_id, log_type=log_type,
                                                status__nin=event_completion_states).get()
@@ -108,18 +111,34 @@ def get_broadcast_logs(bot: Text, start_idx: int = 0, page_size: int = 10, **kwa
         logs = json.loads(logs)
         return logs, total_count
 
+    @staticmethod
+    def get_reference_id_from_broadcasting_logs(event_id):
+        log = MessageBroadcastLogs.objects(event_id=event_id, log_type=MessageBroadcastLogType.common.value).get()
+        return log.reference_id
+
+    @staticmethod
+    def get_recent_broadcasting_logs(message_broadcast_logs):
+        recent_logs, max_resend_count = [], 0
+        if message_broadcast_logs:
+            max_resend_count = max(log["resend_count"] for log in message_broadcast_logs)
+            recent_logs = [log for log in message_broadcast_logs if log["resend_count"] == max_resend_count]
+
+        return recent_logs, max_resend_count
+
     @staticmethod
     def extract_message_ids_from_broadcast_logs(reference_id: Text):
-        message_broadcast_logs = MessageBroadcastLogs.objects(reference_id=reference_id,
-                                                              log_type=MessageBroadcastLogType.send.value)
+        broadcast_logs = MessageBroadcastLogs.objects(reference_id=reference_id,
+                                                      log_type=MessageBroadcastLogType.send.value)
+
+        message_broadcast_logs, resend_count = MessageBroadcastProcessor.get_recent_broadcasting_logs(broadcast_logs)
         broadcast_logs = {
             message['id']: log
             for log in message_broadcast_logs
             if log.api_response and log.api_response.get('messages', [])
             for message in log.api_response['messages']
             if message['id']
         }
-        return broadcast_logs
+        return broadcast_logs, resend_count
 
     @staticmethod
     def get_db_client(bot: Text):
@@ -146,7 +165,8 @@ def log_broadcast_in_conversation_history(template_id, contact: Text, template_p
         })
 
     @staticmethod
-    def __add_broadcast_logs_status_and_errors(reference_id: Text, campaign_name: Text, broadcast_logs: Dict[Text, Document]):
+    def __add_broadcast_logs_status_and_errors(reference_id: Text, campaign_name: Text,
+                                               broadcast_logs: Dict[Text, Document], resend_count: int = 0):
         message_ids = list(broadcast_logs.keys())
         channel_logs = ChannelLogs.objects(message_id__in=message_ids, type=ChannelTypes.WHATSAPP.value)
         for log in channel_logs:
@@ -155,23 +175,28 @@ def __add_broadcast_logs_status_and_errors(reference_id: Text, campaign_name: Te
             client = MessageBroadcastProcessor.get_db_client(broadcast_log['bot'])
             if log['errors']:
                 status = "Failed"
-                broadcast_log.update(errors=log['errors'], status="Failed")
+                errors = log['errors']
             else:
                 status = "Success"
+                errors = []
+            broadcast_log.update(errors=errors, status=status)
 
             MessageBroadcastProcessor.log_broadcast_in_conversation_history(
                 template_id=broadcast_log['template_name'], contact=broadcast_log['recipient'],
                 template_params=broadcast_log['template_params'], template=broadcast_log['template'],
                 status=status, mongo_client=client
             )
 
-        ChannelLogs.objects(message_id__in=message_ids, type=ChannelTypes.WHATSAPP.value).update(campaign_id=reference_id, campaign_name=campaign_name)
+        ChannelLogs.objects(message_id__in=message_ids, type=ChannelTypes.WHATSAPP.value).update(
+            campaign_id=reference_id, campaign_name=campaign_name, resend_count=resend_count
+        )
 
     @staticmethod
     def insert_status_received_on_channel_webhook(reference_id: Text, broadcast_name: Text):
-        broadcast_logs = MessageBroadcastProcessor.extract_message_ids_from_broadcast_logs(reference_id)
+        broadcast_logs, resend_count = MessageBroadcastProcessor.extract_message_ids_from_broadcast_logs(reference_id)
         if broadcast_logs:
-            MessageBroadcastProcessor.__add_broadcast_logs_status_and_errors(reference_id, broadcast_name, broadcast_logs)
+            MessageBroadcastProcessor.__add_broadcast_logs_status_and_errors(reference_id, broadcast_name,
+                                                                             broadcast_logs, resend_count)
 
     @staticmethod
     def get_channel_metrics(channel_type: Text, bot: Text):

kairon/shared/constants.py

@@ -82,6 +82,7 @@ class EventRequestType(str, Enum):
     trigger_async = "trigger_async"
     update_schedule = "update_schedule"
     add_schedule = "add_schedule"
+    resend_broadcast = "resend_broadcast"
 
 
 class DataGeneratorCliTypes(str, Enum):

system.yaml

@@ -198,6 +198,7 @@ channels:
     partner_id: ${360_DIALOG_PARTNER_ID}
     partner_username: ${360_DIALOG_PARTNER_USERNAME}
     partner_password: ${360DIALOG_PARTNER_PASSWORD}
+    error_codes: ${ERROR_CODES:[131026,131021,131047,131052]}
   instagram:
     static_comment_reply: ${INSTA_STATIC_COMMENT_REPLY:Thanks for reaching us, please check your inbox}
 llm: