worker: Make outgoing bot read processed message.

This commit updates outgoing bots to mark messages they process as read.
Since outgoing bots create their own `UserMessage` rows, this change
enables us to track whether the bot has in fact processed the message by
adding the `read` flag to their `UserMessage` after they have been
processed.

Fixes #28869.

zerver/lib/bot_lib.py

@@ -7,6 +7,7 @@ from django.conf import settings
 from django.utils.translation import gettext as _
 from zulip_bots.lib import BotIdentity, RateLimit
 
+from zerver.actions.message_flags import do_update_message_flags
 from zerver.actions.message_send import (
     internal_send_group_direct_message,
     internal_send_private_message,
@@ -153,3 +154,10 @@ class EmbeddedBotHandler:
 
     def quit(self, message: str = "") -> None:
         raise EmbeddedBotQuitError(message)
+
+
+def do_flag_service_bots_messages_as_processed(
+    bot_profile: UserProfile, message_ids: list[int]
+) -> None:
+    assert bot_profile.is_bot is True and bot_profile.bot_type in UserProfile.SERVICE_BOT_TYPES
+    do_update_message_flags(bot_profile, "add", "read", message_ids)

zerver/tests/test_outgoing_webhook_system.py

@@ -712,3 +712,28 @@ class TestOutgoingWebhookMessaging(ZulipTestCase):
             UserMessage.objects.filter(user_profile=bot, message=event_message).exists()
         )
         self.assertTrue(has_message_access(bot, event_message, has_user_message=lambda: True))
+
+    @responses.activate
+    def test_flag_messages_outgoing_webhook_bot_has_processed(self) -> None:
+        """
+        Verifies that once an event has been processed by the outgoing webhook bot's
+        queue processor, the message is marked as processed when read by the
+        outgoing webhook bot.
+        """
+        bot_owner = self.example_user("othello")
+        bot = self.create_outgoing_bot(bot_owner)
+
+        responses.add(
+            responses.POST,
+            "https://bot.example.com/",
+            json="",
+        )
+
+        with self.assertLogs(level="INFO"):
+            self.send_stream_message(
+                bot_owner, "Denmark", content=f"@**{bot.full_name}** foo", topic_name="bar"
+            )
+
+        event_message = self.get_last_message()
+        bot_user_message = UserMessage.objects.get(user_profile=bot, message=event_message)
+        self.assertIn("read", bot_user_message.flags_list())

zerver/worker/outgoing_webhooks.py

@@ -4,8 +4,10 @@ from typing import Any
 
 from typing_extensions import override
 
+from zerver.lib.bot_lib import do_flag_service_bots_messages_as_processed
 from zerver.lib.outgoing_webhook import do_rest_call, get_outgoing_webhook_service_handler
 from zerver.models.bots import get_bot_services
+from zerver.models.users import get_user_profile_by_id
 from zerver.worker.base import QueueProcessingWorker, assign_queue
 
 logger = logging.getLogger(__name__)
@@ -17,9 +19,12 @@ class OutgoingWebhookWorker(QueueProcessingWorker):
     def consume(self, event: dict[str, Any]) -> None:
         message = event["message"]
         event["command"] = message["content"]
+        bot_profile = get_user_profile_by_id(event["user_profile_id"])
 
         services = get_bot_services(event["user_profile_id"])
         for service in services:
             event["service_name"] = str(service.name)
             service_handler = get_outgoing_webhook_service_handler(service)
             do_rest_call(service.base_url, event, service_handler)
+
+        do_flag_service_bots_messages_as_processed(bot_profile, [message["id"]])