hybridcloud(apitoken): add async delete method on apitoken model (#86532)

Continuation of #86069.
Use the newly created services in the model to process deletions.

Author: mdtro

src/sentry/models/apitoken.py

@@ -2,7 +2,7 @@
 
 import hashlib
 import secrets
-from collections.abc import Collection
+from collections.abc import Collection, Mapping
 from datetime import timedelta
 from typing import Any, ClassVar
 
@@ -259,6 +259,21 @@ def handle_async_replication(self, region_name: str, shard_identifier: int) -> N
             region_name=region_name,
         )
 
+    @classmethod
+    def handle_async_deletion(
+        cls,
+        identifier: int,
+        region_name: str,
+        shard_identifier: int,
+        payload: Mapping[str, Any] | None,
+    ) -> None:
+        from sentry.hybridcloud.services.replica import region_replica_service
+
+        region_replica_service.delete_replicated_api_token(
+            apitoken_id=identifier,
+            region_name=region_name,
+        )
+
     @classmethod
     def from_grant(cls, grant: ApiGrant):
         with transaction.atomic(router.db_for_write(cls)):

tests/sentry/models/test_apitoken.py

@@ -1,5 +1,6 @@
 import hashlib
 from datetime import timedelta
+from unittest import mock
 
 import pytest
 from django.utils import timezone
@@ -150,6 +151,41 @@ def test_replica_string_serialization(self):
                 == f"replica_token_id={replica.id}, token_id={token.id} is swug"
             )
 
+    def test_delete_token_removes_replica(self):
+        user = self.create_user()
+
+        with outbox_runner():
+            token = ApiToken.objects.create(user_id=user.id, token_type=AuthTokenType.USER)
+            token.save()
+
+        # Verify replica exists
+        with assume_test_silo_mode(SiloMode.REGION):
+            assert ApiTokenReplica.objects.filter(apitoken_id=token.id).exists()
+
+        # Delete token and verify replica is removed
+        with outbox_runner():
+            token.delete()
+
+        with assume_test_silo_mode(SiloMode.REGION):
+            assert not ApiTokenReplica.objects.filter(apitoken_id=token.id).exists()
+
+    @mock.patch(
+        "sentry.hybridcloud.services.replica.region_replica_service.delete_replicated_api_token"
+    )
+    def test_handle_async_deletion_called(self, mock_delete_replica):
+        user = self.create_user()
+        token = ApiToken.objects.create(user_id=user.id, token_type=AuthTokenType.USER)
+        token_id = token.id
+
+        # Delete token and verify handle_async_deletion was called
+        with outbox_runner():
+            token.delete()
+
+        mock_delete_replica.assert_called_once_with(
+            apitoken_id=token_id,
+            region_name=mock.ANY,
+        )
+
 
 @control_silo_test
 class ApiTokenInternalIntegrationTest(TestCase):