fix(integrations) Deprecate ORM instances in slack task (#49009)

I ended up here trying to unwind joins between UserEmail (control) and
member/org (region). That won't be possible for a while as we haven't
backfilled orgmembershipmapping. I also noticed that we're putting ORM
instances into celery task arguments which is something we're moving
away with and is a necessary step for hybrid cloud as well.

src/sentry/integrations/slack/integration.py

@@ -200,5 +200,10 @@ class SlackIntegrationProvider(IntegrationProvider):  # type: ignore
         """
         Create Identity records for an organization's users if their emails match in Sentry and Slack
         """
-        run_args = {"integration": integration, "organization": organization}
+        run_args = {
+            "integration": integration,
+            "organization": organization,
+            "integration_id": integration.id,
+            "organization_id": organization.id,
+        }
         link_slack_user_identities.apply_async(kwargs=run_args)

src/sentry/models/useremail.py

@@ -26,6 +26,8 @@ if TYPE_CHECKING:
 
 class UserEmailManager(BaseManager):
     def get_for_organization(self, organization: Organization) -> QuerySet:
+        # TODO(hybridcloud) This join between user and member can't exist anymore.
+        # Ideally this would use the organizationmembermapping table but that table is incomplete.
         return self.filter(user__sentry_orgmember_set__organization=organization)
 
     def get_emails_by_user(self, organization: Organization) -> Mapping[User, Iterable[str]]:

src/sentry/tasks/auth.py

@@ -158,6 +158,7 @@ class VerifiedEmailComplianceTask(OrganizationComplianceTask):
     log_label = "verified email"
 
     def is_compliant(self, member: OrganizationMember) -> bool:
+        # TODO(hybridcloud) this is doing a join from member to user
         return UserEmail.objects.get_primary_email(member.user).is_verified
 
     def call_to_action(self, org: Organization, user: User, member: OrganizationMember):

src/sentry/tasks/integrations/slack/link_slack_user_identities.py

@@ -1,4 +1,5 @@
 import logging
+from typing import Optional
 
 from django.utils import timezone
 
@@ -21,7 +22,22 @@ logger = logging.getLogger("sentry.integrations.slack.tasks")
     name="sentry.integrations.slack.link_users_identities",
     queue="integrations",
 )
-def link_slack_user_identities(integration: Integration, organization: Organization) -> None:
+def link_slack_user_identities(
+    integration: Optional[Integration] = None,  # deprecated
+    organization: Optional[Organization] = None,  # deprecated
+    integration_id: Optional[int] = None,
+    organization_id: Optional[int] = None,
+) -> None:
+    if integration_id is not None:
+        integration = Integration.objects.get(id=integration_id)
+    if organization_id is not None:
+        # TODO(hybridcloud) This needs to use organization_service
+        # once member mappings are whole.
+        organization = Organization.objects.get(id=organization_id)
+    assert organization and integration  # type narrowing
+
+    # TODO(hybridcloud) This task is called from slack.integration.SlackIntegration,.post_install()
+    # which should happen in control silo, as it is part of integration install.
     emails_by_user = UserEmail.objects.get_emails_by_user(organization)
     slack_data_by_user = get_slack_data_by_user(integration, organization, emails_by_user)