feat(scheduler): keep participant list in sync (#5460)

src/dispatch/incident/scheduled.py

@@ -23,12 +23,16 @@
 )
 from dispatch.nlp import build_phrase_matcher, build_term_vocab, extract_terms_from_text
 from dispatch.notification import service as notification_service
+from dispatch.incident import service as incident_service
 from dispatch.plugin import service as plugin_service
 from dispatch.project.models import Project
 from dispatch.scheduler import scheduler
 from dispatch.search_filter import service as search_filter_service
 from dispatch.tag import service as tag_service
 from dispatch.tag.models import Tag
+from dispatch.participant import flows as participant_flows
+from dispatch.participant_role.models import ParticipantRoleType
+
 
 from .enums import IncidentStatus
 from .messaging import send_incident_close_reminder
@@ -344,3 +348,44 @@ def incident_report_weekly(db_session: Session, project: Project):
         notification=notification,
         notification_params=notification_params,
     )
+
+
+@scheduler.add(every(1).hour, name="incident-sync-members")
+@timer
+@scheduled_project_task
+def incident_sync_members(db_session: Session, project: Project):
+    """Checks the members of all conversations associated with active
+    and stable incidents and ensures they are in the incident."""
+    plugin = plugin_service.get_active_instance(
+        db_session=db_session,
+        project_id=project.id,
+        plugin_type="conversation",
+    )
+    if not plugin:
+        log.warning("No conversation plugin is active.")
+        return
+
+    active_incidents = incident_service.get_all_by_status(
+        db_session=db_session, project_id=project.id, status=IncidentStatus.active
+    )
+    stable_incidents = incident_service.get_all_by_status(
+        db_session=db_session, project_id=project.id, status=IncidentStatus.stable
+    )
+    incidents = active_incidents + stable_incidents
+
+    for incident in incidents:
+        if incident.conversation:
+            conversation_members = plugin.instance.get_all_member_emails(
+                incident.conversation.channel_id
+            )
+            incident_members = [m.individual.email for m in incident.participants]
+
+            for member in conversation_members:
+                if member not in incident_members:
+                    participant_flows.add_participant(
+                        member,
+                        incident,
+                        db_session,
+                        roles=[ParticipantRoleType.observer],
+                    )
+                    log.debug(f"Added missing {member} to incident {incident.name}")

src/dispatch/plugins/dispatch_slack/enums.py

@@ -11,6 +11,7 @@ class SlackAPIGetEndpoints(DispatchEnum):
     users_info = "users.info"
     users_lookup_by_email = "users.lookupByEmail"
     users_profile_get = "users.profile.get"
+    conversations_members = "conversations.members"
 
 
 class SlackAPIPostEndpoints(DispatchEnum):

src/dispatch/plugins/dispatch_slack/plugin.py

@@ -53,6 +53,7 @@
     does_user_exist,
     emails_to_user_ids,
     get_user_avatar_url,
+    get_user_info_by_id,
     get_user_profile_by_email,
     is_user,
     rename_conversation,
@@ -451,6 +452,28 @@ def get_conversation_replies(self, conversation_id: str, thread_ts: str) -> list
                 replies.append(f"{reply['text']}")
         return replies
 
+    def get_all_member_emails(self, conversation_id: str) -> list[str]:
+        """
+        Fetches all members of a Slack conversation.
+
+        Args:
+            conversation_id (str): The ID of the Slack conversation.
+
+        Returns:
+            list[str]: A list of the emails for all members in the conversation.
+        """
+        client = create_slack_client(self.configuration)
+        member_ids = client.conversations_members(channel=conversation_id).get("members", [])
+
+        member_emails = []
+        for member_id in member_ids:
+            if is_user(config=self.configuration, user_id=member_id):
+                user = get_user_info_by_id(client, member_id)
+                if user:
+                    member_emails.append(user["profile"]["email"])
+
+        return member_emails
+
 
 @apply(counter, exclude=["__init__"])
 @apply(timer, exclude=["__init__"])

src/dispatch/plugins/dispatch_slack/service.py

@@ -25,6 +25,23 @@
 log = logging.getLogger(__name__)
 
 
+class WebClientWrapper:
+    """A wrapper for WebClient to make all instances with same token equal for caching."""
+
+    def __init__(self, client):
+        self._client = client
+
+    @property
+    def client(self):
+        return self._client
+
+    def __eq__(self, other):
+        return other._client.token == self._client.token
+
+    def __hash__(self):
+        return hash(type(self._client.token))
+
+
 def create_slack_client(config: SlackConversationConfiguration) -> WebClient:
     """Creates a Slack Web API client."""
     return WebClient(token=config.api_bot_token.get_secret_value())
@@ -179,28 +196,44 @@ def list_conversation_messages(client: WebClient, conversation_id: str, **kwargs
 
 
 @functools.lru_cache()
+def _get_domain(wrapper: WebClientWrapper) -> str:
+    """Gets the team's Slack domain."""
+    return make_call(wrapper.client, SlackAPIGetEndpoints.team_info)["team"]["domain"]
+
+
 def get_domain(client: WebClient) -> str:
     """Gets the team's Slack domain."""
-    return make_call(client, SlackAPIGetEndpoints.team_info)["team"]["domain"]
+    return _get_domain(WebClientWrapper(client))
 
 
 @functools.lru_cache()
+def _get_user_info_by_id(wrapper: WebClientWrapper, user_id: str) -> dict:
+    return make_call(wrapper.client, SlackAPIGetEndpoints.users_info, user=user_id)["user"]
+
+
 def get_user_info_by_id(client: WebClient, user_id: str) -> dict:
     """Gets profile information about a user by id."""
-    return make_call(client, SlackAPIGetEndpoints.users_info, user=user_id)["user"]
+    return _get_user_info_by_id(WebClientWrapper(client), user_id)
 
 
 @functools.lru_cache()
+def _get_user_info_by_email(wrapper: WebClientWrapper, email: str) -> dict:
+    """Gets profile information about a user by email."""
+    return make_call(wrapper.client, SlackAPIGetEndpoints.users_lookup_by_email, email=email)[
+        "user"
+    ]
+
+
 def get_user_info_by_email(client: WebClient, email: str) -> dict:
     """Gets profile information about a user by email."""
-    return make_call(client, SlackAPIGetEndpoints.users_lookup_by_email, email=email)["user"]
+    return _get_user_info_by_email(WebClientWrapper(client), email)
 
 
 @functools.lru_cache()
-def does_user_exist(client: WebClient, email: str) -> bool:
+def _does_user_exist(wrapper: WebClientWrapper, email: str) -> bool:
     """Checks if a user exists in the Slack workspace by their email."""
     try:
-        get_user_info_by_email(client, email)
+        get_user_info_by_email(wrapper.client, email)
         return True
     except SlackApiError as e:
         if e.response["error"] == SlackAPIErrorCode.USERS_NOT_FOUND:
@@ -209,21 +242,38 @@ def does_user_exist(client: WebClient, email: str) -> bool:
             raise
 
 
+def does_user_exist(client: WebClient, email: str) -> bool:
+    """Checks if a user exists in the Slack workspace by their email."""
+    return _does_user_exist(WebClientWrapper(client), email)
+
+
 @functools.lru_cache()
+def _get_user_profile_by_id(wrapper: WebClientWrapper, user_id: str) -> dict:
+    """Gets profile information about a user by id."""
+    return make_call(wrapper.client, SlackAPIGetEndpoints.users_profile_get, user_id=user_id)[
+        "profile"
+    ]
+
+
 def get_user_profile_by_id(client: WebClient, user_id: str) -> dict:
     """Gets profile information about a user by id."""
-    return make_call(client, SlackAPIGetEndpoints.users_profile_get, user_id=user_id)["profile"]
+    return _get_user_profile_by_id(WebClientWrapper(client), user_id)
 
 
 @functools.lru_cache()
-def get_user_profile_by_email(client: WebClient, email: str) -> SlackResponse:
+def _get_user_profile_by_email(wrapper: WebClientWrapper, email: str) -> SlackResponse:
     """Gets extended profile information about a user by email."""
-    user = get_user_info_by_email(client, email)
-    profile = get_user_profile_by_id(client, user["id"])
+    user = get_user_info_by_email(wrapper.client, email)
+    profile = get_user_profile_by_id(wrapper.client, user["id"])
     profile["tz"] = user["tz"]
     return profile
 
 
+def get_user_profile_by_email(client: WebClient, email: str) -> SlackResponse:
+    """Gets extended profile information about a user by email."""
+    return _get_user_profile_by_email(WebClientWrapper(client), email)
+
+
 def get_user_email(client: WebClient, user_id: str) -> str:
     """Gets the user's email."""
     user_info = get_user_info_by_id(client, user_id)