fix(hybridcloud) Use celery for unlink delivery (#67838)

Move sso unlink operations and email delivery to a celery task that is
triggered by the existing RPC method.

Author: markstory

src/sentry/services/hybrid_cloud/organization/impl.py

@@ -61,6 +61,7 @@
 from sentry.services.hybrid_cloud.user import RpcUser
 from sentry.services.hybrid_cloud.util import flags_to_bits
 from sentry.silo import unguarded_write
+from sentry.tasks.auth import email_unlink_notifications
 from sentry.types.region import find_regions_for_orgs
 from sentry.utils.audit import create_org_delete_log
 
@@ -606,24 +607,14 @@ def send_sso_unlink_emails(
         from sentry.auth.exceptions import ProviderNotRegistered
 
         try:
-            provider = manager.get(provider_key)
+            manager.get(provider_key)
         except ProviderNotRegistered as e:
             logger.warning("Could not send SSO unlink emails: %s", e)
             return
 
-        with unguarded_write(using=router.db_for_write(OrganizationMember)):
-            # Flags are not replicated -- these updates are safe to skip outboxes
-            OrganizationMember.objects.filter(organization_id=organization_id).update(
-                flags=F("flags")
-                .bitand(~OrganizationMember.flags["sso:linked"])
-                .bitand(~OrganizationMember.flags["sso:invalid"])
-            )
-
-        member_list = OrganizationMember.objects.filter(
-            organization_id=organization_id,
+        email_unlink_notifications.delay(
+            org_id=organization_id, sending_user_email=sending_user_email, provider_key=provider_key
         )
-        for member in member_list:
-            member.send_sso_unlink_email(sending_user_email, provider)
 
     def count_members_without_sso(self, *, organization_id: int) -> int:
         return OrganizationMember.objects.filter(

src/sentry/tasks/auth.py

@@ -3,6 +3,8 @@
 import abc
 import logging
 
+from django.db import router
+from django.db.models import F
 from django.urls import reverse
 
 from sentry import audit_log, features, options
@@ -16,6 +18,7 @@
 from sentry.services.hybrid_cloud.user import RpcUser
 from sentry.services.hybrid_cloud.user.service import user_service
 from sentry.silo import SiloMode
+from sentry.silo.safety import unguarded_write
 from sentry.tasks.base import instrumented_task, retry
 from sentry.types.region import RegionMappingNotFound
 from sentry.utils.audit import create_audit_entry_from_user
@@ -60,18 +63,23 @@ def _email_missing_links(org_id: int, sending_user_id: int, provider_key: str) -
 @instrumented_task(
     name="sentry.tasks.email_unlink_notifications", queue="auth", silo_mode=SiloMode.REGION
 )
-def email_unlink_notifications(org_id: int, actor_id: int, provider_key: str):
+def email_unlink_notifications(
+    org_id: int, sending_user_email: str, provider_key: str, actor_id: int | None = None
+):
     try:
         org = Organization.objects.get(id=org_id)
         provider = manager.get(provider_key)
     except (Organization.DoesNotExist, ProviderNotRegistered) as e:
         logger.warning("Could not send SSO unlink emails: %s", e)
         return
 
-    user = user_service.get_user(user_id=actor_id)
-    if not user:
-        logger.warning("sso.unlink.email_failure.could_not_find_user", extra={"user_id": actor_id})
-        return
+    with unguarded_write(using=router.db_for_write(OrganizationMember)):
+        # Flags are not replicated -- these updates are safe to skip outboxes
+        OrganizationMember.objects.filter(organization_id=org_id).update(
+            flags=F("flags")
+            .bitand(~OrganizationMember.flags["sso:linked"])
+            .bitand(~OrganizationMember.flags["sso:invalid"])
+        )
 
     # Email all organization users, even if they never linked their accounts.
     # This provides a better experience in the case where SSO is enabled and
@@ -80,7 +88,7 @@ def email_unlink_notifications(org_id: int, actor_id: int, provider_key: str):
     # intermittently -- force an ordering in your test!
     members = OrganizationMember.objects.filter(organization=org, user_id__isnull=False)
     for member in members:
-        member.send_sso_unlink_email(user, provider)
+        member.send_sso_unlink_email(sending_user_email, provider)
 
 
 class OrganizationComplianceTask(abc.ABC):

tests/sentry/tasks/test_auth.py

@@ -102,13 +102,13 @@ def setUp(self):
             self.provider = AuthProvider.objects.create(
                 organization_id=self.organization.id, provider="dummy"
             )
-        om = self.create_member(
+        self.om = self.create_member(
             user_id=self.user.id,
             organization=self.organization,
             flags=OrganizationMember.flags["sso:linked"],
         )
 
-        assert om.flags["sso:linked"]
+        assert self.om.flags["sso:linked"]
         self.user2 = self.create_user(email="baz@example.com")
 
         om2 = self.create_member(user_id=self.user2.id, organization=self.organization, flags=0)
@@ -119,23 +119,31 @@ def setUp(self):
 
     def test_email_unlink_notifications_with_password(self):
         with self.tasks():
-            email_unlink_notifications(self.organization.id, self.user.id, self.provider.provider)
+            email_unlink_notifications(
+                self.organization.id, self.user.email, self.provider.provider
+            )
 
         emails = sorted(message.body for message in mail.outbox)
         assert len(emails) == 2
         assert f"can now login using your email {self.user.email}, and password" in emails[0]
         assert "you'll first have to set a password" not in emails[0]
+        self.om.refresh_from_db()
+        assert not self.om.flags["sso:linked"]
 
     def test_email_unlink_notifications_without_password(self):
         with assume_test_silo_mode(SiloMode.CONTROL):
             self.user.password = ""
             self.user.save()
 
         with self.tasks():
-            email_unlink_notifications(self.organization.id, self.user.id, self.provider.provider)
+            email_unlink_notifications(
+                self.organization.id, self.user.email, self.provider.provider
+            )
 
         emails = sorted(message.body for message in mail.outbox)
         assert len(emails) == 2
         assert "you'll first have to set a password" in emails[0]
         assert f"can now login using your email {self.user.email}, and password" not in emails[0]
         assert f"can now login using your email {self.user2.email}, and password" in emails[1]
+        self.om.refresh_from_db()
+        assert not self.om.flags["sso:linked"]