worker: Flag messages processed by outgoing bot.

This commit updates outgoing bots to mark messages they process as read.
Since the service bots have 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`.

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_service_bot_system.py

@@ -4,6 +4,7 @@ from typing import Any, Concatenate
 from unittest import mock
 
 import orjson
+import responses
 from django.conf import settings
 from django.test import override_settings
 from typing_extensions import ParamSpec, override
@@ -17,6 +18,7 @@ from zerver.lib.test_classes import ZulipTestCase
 from zerver.lib.test_helpers import mock_queue_publish
 from zerver.lib.validator import check_string
 from zerver.models import Recipient, UserProfile
+from zerver.models.messages import UserMessage
 from zerver.models.realms import get_realm
 from zerver.models.scheduled_jobs import NotificationTriggers
 
@@ -604,3 +606,27 @@ class TestServiceBotEventTriggers(ZulipTestCase):
         recipients = [self.user_profile, self.bot_profile]
         self.send_group_direct_message(sender, recipients)
         self.assertFalse(mock_queue_event_on_commit.called)
+
+    @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 (flagged with `read`).
+        """
+        self.bot_profile.bot_type = UserProfile.OUTGOING_WEBHOOK_BOT
+        self.bot_profile.save()
+        sender = self.user_profile
+        recipients = [self.user_profile, self.bot_profile, self.second_bot_profile]
+        responses.add(
+            responses.POST,
+            "https://bot.example.com/",
+            json="",
+        )
+        message_id = self.send_group_direct_message(
+            sender, recipients, content=f"@**{self.bot_profile.full_name}** foo"
+        )
+        # message = Message.objects.get(id=message_id, sender=sender)
+        bot_user_message = UserMessage.objects.get(
+            user_profile=self.bot_profile, message=message_id
+        )
+        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"]])