Enable `SlackService` to update look-up table for new user identifiers as encountered in incoming messages

src/ctp_slack_bot/services/slack_service.py

@@ -3,10 +3,11 @@ from httpx import AsyncClient
 from loguru import logger
 from openai import OpenAI
 from pydantic import ConfigDict
-from re import compile as compile_re
+from re import compile as compile_re, Pattern
+from slack_sdk.errors import SlackApiError
 from slack_bolt.async_app import AsyncApp
 from slack_sdk.web.async_slack_response import AsyncSlackResponse
-from typing import Any, Mapping, MutableMapping, Self
+from typing import Any, Mapping, MutableMapping, Optional, Self, Set
 
 from ctp_slack_bot.core import HealthReportingApplicationComponentBase
 from ctp_slack_bot.enums import EventType
@@ -14,9 +15,6 @@ from ctp_slack_bot.models import SlackMessage, SlackResponse
 from ctp_slack_bot.services.event_brokerage_service import EventBrokerageService
 
 
-SLACK_USER_MENTION_PATTERN = compile_re(r"<@([A-Z0-9]+)>")
-
-
 class SlackService(HealthReportingApplicationComponentBase):
     """
     Service for interfacing with Slack.
@@ -24,24 +22,40 @@ class SlackService(HealthReportingApplicationComponentBase):
 
     model_config = ConfigDict(arbitrary_types_allowed=True, frozen=True)
 
+    SLACK_USER_ID_PATTERN: Pattern = compile_re(r"U\d+")
+    SLACK_USER_MENTION_PATTERN: Pattern = compile_re(r"<@(U[A-Z0-9]+)>")
+
     event_brokerage_service: EventBrokerageService
     http_client: AsyncClient
     slack_bolt_app: AsyncApp
-    user_id_name_map: MutableMapping[str, str]
+    id_name_map: MutableMapping[str, str] # TODO: Spin message processing out into its own service.
 
-    def model_post_init(self: Self, context: Any, /) -> None:
-        super().model_post_init(context)
+    def initialize(self: Self) -> None:
         self.event_brokerage_service.subscribe(EventType.OUTGOING_SLACK_RESPONSE, self.send_message)
+        self.slack_bolt_app.event("message")(self._handle_message_event)
+        self.slack_bolt_app.event("app_mention")(self._handle_app_mention_event)
+        logger.debug("Registered 2 handlers for Slack Bolt message and app mention events.")
+
+    async def send_message(self: Self, message: SlackResponse) -> None:
+        await self.slack_bolt_app.client.chat_postMessage(channel=message.channel, text=message.text, thread_ts=message.thread_ts)
+
+    @property
+    def name(self: Self) -> str:
+        return "slack_service"
+
+    async def is_healthy(self: Self) -> bool:
+        response = await self.http_client.get("https://slack-status.com/api/v2.0.0/current")
+        return response.status_code == 200
 
-    async def adapt_event_payload(self: Self, event: Mapping[str, Any]) -> SlackMessage:
-        text = SLACK_USER_MENTION_PATTERN.sub(lambda match: f"@{self.user_id_name_map.get(match.group(1))}", event.get("text", "")) # TODO: permit look-up of Slack again when not found.
+    async def _adapt_event_payload(self: Self, event: Mapping[str, Any]) -> SlackMessage:
+        text = await self._resolve_user_mentions(event.get("text", ""))
         user_id = event.get("user")
         return SlackMessage(
             type=event.get("type"),
             subtype=event.get("subtype"),
             channel=event.get("channel"),
             channel_type=event.get("channel_type"),
-            user=await self._get_user_display_name(user_id),
+            user=await self._get_name(user_id),
             bot_id=event.get("bot_id"),
             thread_ts=event.get("thread_ts"),
             text=text,
@@ -49,66 +63,100 @@ class SlackService(HealthReportingApplicationComponentBase):
             event_ts=event.get("event_ts")
         )
 
-    async def process_message(self: Self, event: Mapping[str, Any]) -> None:
-        slack_message = await self.adapt_event_payload(event.get("event", {}))
-        logger.debug("Received message from Slack: {}", slack_message)
-        await self.event_brokerage_service.publish(EventType.INCOMING_SLACK_MESSAGE, slack_message)
+    async def _ensure_ids_in_id_name_map(self: Self, ids: Set[str]) -> None:
+        unknown_ids = ids - self.id_name_map.keys()
+        if len(unknown_ids) == 0:
+            return
+        async with TaskGroup() as task_group:
+            update_tasks = {unknown_id: task_group.create_task(self._look_up_name(unknown_id)) for unknown_id in unknown_ids}
+        self.id_name_map.update({id: task.result() for id, task in update_tasks.items() if task.result()})
 
-    async def send_message(self: Self, message: SlackResponse) -> None:
-        await self.slack_bolt_app.client.chat_postMessage(channel=message.channel, text=message.text, thread_ts=message.thread_ts)
+    async def _get_name(self: Self, id: str) -> str:
+        await self._ensure_ids_in_id_name_map({id})
+        return self.id_name_map.get(id, id)
 
-    async def handle_message_event(self: Self, body: Mapping[str, Any]) -> None:
+    async def _handle_message_event(self: Self, body: Mapping[str, Any]) -> None:
         logger.debug("Ignored regular message: {}", body.get("event", {}).get("text"))
-        # await self.process_message(body)
+        # await self._process_message(body)
 
-    async def handle_app_mention_event(self: Self, body: Mapping[str, Any]) -> None:
+    async def _handle_app_mention_event(self: Self, body: Mapping[str, Any]) -> None:
         logger.debug("Received app mention for processing: {}", body.get("event", {}).get("text"))
-        await self.process_message(body)
-
-    def initialize(self: Self) -> None:
-        self.slack_bolt_app.event("message")(self.handle_message_event)
-        self.slack_bolt_app.event("app_mention")(self.handle_app_mention_event)
-        logger.debug("Registered 2 handlers for Slack Bolt message and app mention events.")
-
-    @property
-    def name(self: Self) -> str:
-        return "slack_service"
-
-    async def is_healthy(self: Self) -> bool:
-        response = await self.http_client.get("https://slack-status.com/api/v2.0.0/current")
-        return response.status_code == 200
+        await self._process_message(body)
+
+    async def _look_up_name(self: Self, id: str) -> Optional[str]:
+        if self.SLACK_USER_ID_PATTERN.fullmatch(id):
+            match await self.slack_bolt_app.client.users_info(id):
+                case AsyncSlackResponse(data={"ok": True, "user": user}):
+                    match user:
+                        case {"real_name": real_name}:
+                            return real_name
+                        case {"profile": {"display_name": display_name}}:
+                            return display_name
+                        case {"name": name}:
+                            return name
+                case AsyncSlackResponse(data={"ok": False, "error": "user_not_found"}):
+                    logger.error("An attempt to look up a user failed (user not found): {}", id)
+        return None
+
+    async def _process_message(self: Self, event: Mapping[str, Any]) -> None:
+        slack_message = await self._adapt_event_payload(event.get("event", {}))
+        logger.debug("Received message from Slack: {}", slack_message)
+        await self.event_brokerage_service.publish(EventType.INCOMING_SLACK_MESSAGE, slack_message)
 
-    async def _get_user_display_name(self: Self, user_id: str) -> str:
-        return self.user_id_name_map.get(user_id, f"<@{user_id}>")
-        # TODO: Handle new users.
+    async def _resolve_user_mentions(self: Self, text: str) -> str:
+        matches = tuple(self.SLACK_USER_MENTION_PATTERN.finditer(text))
+        unique_ids = frozenset(match.group(1) for match in matches)
+        await self._ensure_ids_in_id_name_map(unique_ids)
+        parts = []
+        previous_end = 0
+        for match in matches:
+            start, end = match.span()
+            parts.append(text[previous_end:start])
+            user_id = match.group(1)
+            parts.append(f"@{self.id_name_map.get(user_id, user_id)}")
+            previous_end = end
+        parts.append(text[previous_end:])
+        return ''.join(parts)
 
 
 class SlackServiceResource(AsyncResource):
     async def init(self: Self, event_brokerage_service: EventBrokerageService, http_client: AsyncClient, slack_bolt_app: AsyncApp) -> SlackService:
-        match await slack_bolt_app.client.users_list():
-            case AsyncSlackResponse(status_code=200, data={"ok": True, "members": users}):
-                user_id_name_map = {id: display_name
-                                    for id, display_name
-                                    in zip(map(SlackServiceResource._get_user_id, users), map(SlackServiceResource._get_user_display_name, users))
-                                    if display_name}
-                logger.debug("Obtained a list of {} user name(s) for the workspace: {}", len(user_id_name_map), user_id_name_map)
-            case something:
-                user_id_name_map = {}
-                logger.error("Could not obtain a list of user name for the workspace.")
-        slack_service = SlackService(event_brokerage_service=event_brokerage_service, http_client=http_client, slack_bolt_app=slack_bolt_app, user_id_name_map=user_id_name_map)
+        async def get_users_list():
+            cursor = None
+            while True:
+                try:
+                    response = await slack_bolt_app.client.users_list(cursor=cursor, limit=200)
+                except SlackApiError as e:
+                    logger.warning("Could not get a list of users: {}", e)
+                    break
+                match response:
+                    case AsyncSlackResponse(status_code=200, data={"ok": True, "members": users}):
+                        for user in users:
+                            yield user
+                        match response.data:
+                            case {"response_metadata": {"next_cursor": cursor}} if cursor:
+                                continue
+                    case AsyncSlackResponse(status_code=status_code) if status_code != 200:
+                        logger.warning("Could not get a list of users: response status {}", status_code)
+                    case AsyncSlackResponse(data={"ok": False}):
+                        logger.warning("Could not get a list of users: non-OK response")
+                    case _:
+                        logger.warning("Could not get a list of users.")
+                break
+        id_name_map = {user["id"]: self._get_name(user)
+                       async for user
+                       in get_users_list()}
+        logger.debug("Obtained a list of {} user name(s) for the workspace: {}", len(id_name_map), id_name_map)
+        slack_service = SlackService(event_brokerage_service=event_brokerage_service, http_client=http_client, slack_bolt_app=slack_bolt_app, id_name_map=id_name_map)
         slack_service.initialize()
         return slack_service
 
     @classmethod
-    def _get_user_id(cls, user: Mapping[str, Any]):
-        return user["id"]
-
-    @classmethod
-    def _get_user_display_name(cls, user: Mapping[str, Any]):
+    def _get_name(cls, user: Mapping[str, Any]):
         match user:
-            case {"profile": {"display_name": display_name}}:
-                return display_name
             case {"real_name": real_name}:
                 return real_name
-            case _:
-                None
+            case {"profile": {"display_name": display_name}}:
+                return display_name
+            case {"name": name}:
+                return name