feat(uptime): Add billing seat management for detector validators (#102989)

Add billing seat assignment and removal to
UptimeDomainCheckFailureValidator
to ensure uptime detectors created, updated, or deleted via the detector
APIs
correctly handle billing seats.

Changes:
- Assign seats when creating enabled detectors (gracefully disable if no
seats)
- Assign/remove seats when enabling/disabling detectors via updates
- Remove seats immediately when deleting detectors
- Raise validation errors when enabling fails due to no available seats

Also add comprehensive tests to verify all billing seat operations work
correctly across create, update, and delete operations.

Fixes [NEW-527: Ensure CRUD / enable+disable detectors in new UI handles
assigning / unassigning
seats](https://linear.app/getsentry/issue/NEW-527/ensure-crud-enabledisable-detectors-in-new-ui-handles-assigning)

Author: evanpurkhiser

src/sentry/deletions/defaults/uptime_subscription.py

@@ -1,34 +1,26 @@
 from sentry.deletions.base import ModelDeletionTask
-from sentry.uptime.models import UptimeSubscription
+from sentry.uptime.models import UptimeSubscription, get_detector
+from sentry.uptime.subscriptions.subscriptions import delete_uptime_subscription, remove_uptime_seat
 
 
 class UptimeSubscriptionDeletionTask(ModelDeletionTask[UptimeSubscription]):
     def delete_instance(self, instance: UptimeSubscription) -> None:
-        from sentry import quotas
-        from sentry.constants import DataCategory
-        from sentry.uptime.models import get_detector
-        from sentry.uptime.subscriptions.subscriptions import delete_uptime_subscription
 
         detector = get_detector(instance)
 
         # XXX: Typically quota updates would be handled by the
         # delete_uptime_detector function exposed in the
         # uptime.subscriptions.subscriptions module. However if a Detector is
         # deleted without using this, we need to still ensure the billing east
-        # is revoked.
+        # is revoked. This should never happen.
         #
         # Since the delete_uptime_detector function is already scheduling the
         # detector for deletion, you may think we could remove the quota
         # remove_seat call there, since it will happen here. But this would
         # mean the customers quota is not freed up _immediately_ when the
         # detector is deleted using that method.
+        remove_uptime_seat(detector)
 
-        # TODO(epurkhiser): It's very likely the new Workflow Engine detector
-        # APIs will NOT free up customers seats immediately. We'll probably
-        # need some other hook for this
-
-        # Ensure the billing seat for the parent detector is removed
-        quotas.backend.remove_seat(DataCategory.UPTIME, detector)
-
-        # Ensure the remote subscription is also removed
+        # Ensure the remote subscription is removed if it wasn't already (again
+        # it should have been as part of delete_uptime_detector)
         delete_uptime_subscription(instance)

src/sentry/uptime/endpoints/validators.py

@@ -7,11 +7,11 @@
 from rest_framework import serializers
 from rest_framework.fields import URLField
 
-from sentry import audit_log
+from sentry import audit_log, quotas
 from sentry.api.fields import ActorField
 from sentry.api.serializers.rest_framework import CamelSnakeSerializer
 from sentry.auth.superuser import is_active_superuser
-from sentry.constants import ObjectStatus
+from sentry.constants import DataCategory, ObjectStatus
 from sentry.models.environment import Environment
 from sentry.uptime.models import (
     UptimeSubscription,
@@ -28,6 +28,10 @@
     check_url_limits,
     create_uptime_detector,
     create_uptime_subscription,
+    delete_uptime_subscription,
+    disable_uptime_detector,
+    enable_uptime_detector,
+    remove_uptime_seat,
     update_uptime_detector,
     update_uptime_subscription,
 )
@@ -462,7 +466,47 @@ def create_source(self, validated_data: dict[str, Any]) -> UptimeSubscription:
 class UptimeDomainCheckFailureValidator(BaseDetectorTypeValidator):
     data_sources = serializers.ListField(child=UptimeMonitorDataSourceValidator(), required=False)
 
+    def validate_enabled(self, value: bool) -> bool:
+        """
+        Validate that enabling a detector is allowed based on seat availability.
+
+        This check will ONLY be performed when a detector instance is provided via
+        context (i.e., during updates). For creation, seat assignment is handled
+        in the create() method after the detector is created.
+        """
+        detector = self.instance
+
+        # Only validate on updates when trying to enable a currently disabled detector
+        if detector and value and not detector.enabled:
+            result = quotas.backend.check_assign_seat(DataCategory.UPTIME, detector)
+            if not result.assignable:
+                raise serializers.ValidationError(result.reason)
+
+        return value
+
+    def create(self, validated_data):
+        detector = super().create(validated_data)
+
+        try:
+            enable_uptime_detector(detector, ensure_assignment=True)
+        except UptimeMonitorNoSeatAvailable:
+            # No need to do anything if we failed to handle seat
+            # assignment. The monitor will be created, but not enabled
+            pass
+
+        return detector
+
     def update(self, instance: Detector, validated_data: dict[str, Any]) -> Detector:
+        # Handle seat management when enabling/disabling
+        was_enabled = instance.enabled
+        enabled = validated_data.get("enabled", was_enabled)
+
+        if was_enabled != enabled:
+            if enabled:
+                enable_uptime_detector(instance)
+            else:
+                disable_uptime_detector(instance)
+
         super().update(instance, validated_data)
 
         data_source = None
@@ -497,5 +541,11 @@ def update_data_source(self, instance: Detector, data_source: dict[str, Any]):
 
         return instance
 
-    def create(self, validated_data):
-        return super().create(validated_data)
+    def delete(self) -> None:
+        assert self.instance is not None
+
+        remove_uptime_seat(self.instance)
+        uptime_subscription = get_uptime_subscription(self.instance)
+        delete_uptime_subscription(uptime_subscription)
+
+        super().delete()

src/sentry/uptime/subscriptions/subscriptions.py

@@ -476,6 +476,18 @@ def disable_uptime_detector(detector: Detector, skip_quotas: bool = False):
             delete_remote_uptime_subscription.delay(uptime_subscription.id)
 
 
+def ensure_uptime_seat(detector: Detector) -> None:
+    """
+    Ensures that a billing seat is assigned for the uptime detector.
+
+    Raises UptimeMonitorNoSeatAvailable if no seats are available.
+    """
+    outcome = quotas.backend.assign_seat(DataCategory.UPTIME, detector)
+    if outcome != Outcome.ACCEPTED:
+        result = quotas.backend.check_assign_seat(DataCategory.UPTIME, detector)
+        raise UptimeMonitorNoSeatAvailable(result)
+
+
 def enable_uptime_detector(
     detector: Detector, ensure_assignment: bool = False, skip_quotas: bool = False
 ):
@@ -494,11 +506,11 @@ def enable_uptime_detector(
         return
 
     if not skip_quotas:
-        outcome = quotas.backend.assign_seat(DataCategory.UPTIME, detector)
-        if outcome != Outcome.ACCEPTED:
-            disable_uptime_detector(detector)
-            result = quotas.backend.check_assign_seat(DataCategory.UPTIME, detector)
-            raise UptimeMonitorNoSeatAvailable(result)
+        try:
+            ensure_uptime_seat(detector)
+        except UptimeMonitorNoSeatAvailable:
+            disable_uptime_detector(detector, skip_quotas=True)
+            raise
 
     uptime_subscription: UptimeSubscription = get_uptime_subscription(detector)
     detector.update(enabled=True)
@@ -513,9 +525,14 @@ def enable_uptime_detector(
         )
 
 
+def remove_uptime_seat(detector: Detector):
+    quotas.backend.remove_seat(DataCategory.UPTIME, detector)
+
+
 def delete_uptime_detector(detector: Detector):
     uptime_subscription = get_uptime_subscription(detector)
-    quotas.backend.remove_seat(DataCategory.UPTIME, detector)
+
+    remove_uptime_seat(detector)
     detector.update(status=ObjectStatus.PENDING_DELETION)
     RegionScheduledDeletion.schedule(detector, days=0)
     delete_uptime_subscription(uptime_subscription)

tests/sentry/uptime/endpoints/test_validators.py

@@ -1,9 +1,21 @@
+from unittest import mock
+
+from sentry.constants import DataCategory
+from sentry.quotas.base import SeatAssignmentResult
 from sentry.testutils.cases import TestCase, UptimeTestCase
 from sentry.uptime.endpoints.validators import (
+    UptimeDomainCheckFailureValidator,
     UptimeMonitorDataSourceValidator,
     compute_http_request_size,
 )
-from sentry.uptime.models import UptimeSubscription
+from sentry.uptime.grouptype import UptimeDomainCheckFailure
+from sentry.uptime.models import UptimeSubscription, get_uptime_subscription
+from sentry.uptime.types import (
+    DEFAULT_DOWNTIME_THRESHOLD,
+    DEFAULT_RECOVERY_THRESHOLD,
+    UptimeMonitorMode,
+)
+from sentry.utils.outcomes import Outcome
 
 
 class ComputeHttpRequestSizeTest(UptimeTestCase):
@@ -97,3 +109,188 @@ def test_too_big_request(self):
         data = self.get_valid_data(body="0" * 1000)
         validator = UptimeMonitorDataSourceValidator(data=data, context=self.context)
         assert not validator.is_valid()
+
+
+class UptimeDomainCheckFailureValidatorTest(UptimeTestCase):
+    def setUp(self):
+        super().setUp()
+        self.context = {
+            "organization": self.organization,
+            "project": self.project,
+            "request": self.make_request(user=self.user),
+        }
+
+    def get_valid_data(self, **kwargs):
+        return {
+            "name": kwargs.get("name", "Test Uptime Monitor"),
+            "type": UptimeDomainCheckFailure.slug,
+            "enabled": kwargs.get("enabled", True),
+            "config": kwargs.get(
+                "config",
+                {
+                    "mode": UptimeMonitorMode.MANUAL.value,
+                    "environment": None,
+                    "recovery_threshold": DEFAULT_RECOVERY_THRESHOLD,
+                    "downtime_threshold": DEFAULT_DOWNTIME_THRESHOLD,
+                },
+            ),
+            "dataSources": kwargs.get(
+                "data_sources",
+                [
+                    {
+                        "url": "https://sentry.io",
+                        "intervalSeconds": 60,
+                        "timeoutMs": 1000,
+                    }
+                ],
+            ),
+        }
+
+    @mock.patch(
+        "sentry.quotas.backend.assign_seat",
+        return_value=Outcome.ACCEPTED,
+    )
+    def test_create_enabled_assigns_seat(self, mock_assign_seat: mock.MagicMock) -> None:
+        """Test that creating an enabled detector assigns a billing seat."""
+        validator = UptimeDomainCheckFailureValidator(
+            data=self.get_valid_data(enabled=True), context=self.context
+        )
+        assert validator.is_valid(), validator.errors
+        detector = validator.save()
+
+        detector.refresh_from_db()
+        assert detector.enabled is True
+
+        # Verify seat was assigned
+        mock_assign_seat.assert_called_with(DataCategory.UPTIME, detector)
+
+    @mock.patch(
+        "sentry.quotas.backend.assign_seat",
+        return_value=Outcome.RATE_LIMITED,
+    )
+    def test_create_enabled_no_seat_available(self, mock_assign_seat: mock.MagicMock) -> None:
+        """Test that creating a detector with no seats available creates it but leaves it disabled."""
+        validator = UptimeDomainCheckFailureValidator(
+            data=self.get_valid_data(enabled=True), context=self.context
+        )
+        assert validator.is_valid(), validator.errors
+        detector = validator.save()
+
+        detector.refresh_from_db()
+        # Detector created but not enabled due to no seat assignment
+        assert detector.enabled is False
+
+        # Verify seat assignment was attempted
+        mock_assign_seat.assert_called_with(DataCategory.UPTIME, detector)
+
+        uptime_subscription = get_uptime_subscription(detector)
+        assert uptime_subscription.status == UptimeSubscription.Status.DISABLED.value
+
+    @mock.patch(
+        "sentry.quotas.backend.assign_seat",
+        return_value=Outcome.ACCEPTED,
+    )
+    def test_update_enable_assigns_seat(self, mock_assign_seat: mock.MagicMock) -> None:
+        """Test that enabling a previously disabled detector assigns a seat."""
+        # Create a disabled detector
+        detector = self.create_uptime_detector(enabled=False)
+
+        validator = UptimeDomainCheckFailureValidator(
+            instance=detector, data={"enabled": True}, context=self.context, partial=True
+        )
+        assert validator.is_valid(), validator.errors
+        validator.save()
+
+        detector.refresh_from_db()
+        assert detector.enabled is True
+
+        # Verify seat was assigned
+        mock_assign_seat.assert_called_with(DataCategory.UPTIME, detector)
+
+        uptime_subscription = get_uptime_subscription(detector)
+        assert uptime_subscription.status == UptimeSubscription.Status.ACTIVE.value
+
+    @mock.patch(
+        "sentry.quotas.backend.check_assign_seat",
+        return_value=SeatAssignmentResult(assignable=False, reason="No seats available"),
+    )
+    def test_update_enable_no_seat_available(self, mock_check_assign_seat: mock.MagicMock) -> None:
+        """Test that enabling fails with validation error when no seats are available."""
+        # Create a disabled detector
+        detector = self.create_uptime_detector(enabled=False)
+
+        validator = UptimeDomainCheckFailureValidator(
+            instance=detector, data={"enabled": True}, context=self.context, partial=True
+        )
+
+        # Validation should fail due to no seats available
+        assert not validator.is_valid()
+        assert "enabled" in validator.errors
+        assert validator.errors["enabled"] == ["No seats available"]
+
+        detector.refresh_from_db()
+        # Detector should still be disabled
+        assert detector.enabled is False
+
+        # Verify seat availability check was performed
+        mock_check_assign_seat.assert_called_with(DataCategory.UPTIME, detector)
+
+    @mock.patch("sentry.quotas.backend.disable_seat")
+    def test_update_disable_removes_seat(self, mock_disable_seat: mock.MagicMock) -> None:
+        """Test that disabling a previously enabled detector removes the seat."""
+        # Create an enabled detector
+        detector = self.create_uptime_detector(enabled=True)
+
+        validator = UptimeDomainCheckFailureValidator(
+            instance=detector, data={"enabled": False}, context=self.context, partial=True
+        )
+        assert validator.is_valid(), validator.errors
+        validator.save()
+
+        detector.refresh_from_db()
+        assert detector.enabled is False
+
+        # Verify disable_seat was called
+        mock_disable_seat.assert_called_with(DataCategory.UPTIME, detector)
+
+        uptime_subscription = get_uptime_subscription(detector)
+        assert uptime_subscription.status == UptimeSubscription.Status.DISABLED.value
+
+    @mock.patch("sentry.quotas.backend.remove_seat")
+    def test_delete_removes_seat(self, mock_remove_seat: mock.MagicMock) -> None:
+        """Test that deleting a detector removes its billing seat immediately."""
+        detector = self.create_uptime_detector(enabled=True)
+
+        validator = UptimeDomainCheckFailureValidator(
+            instance=detector, data={}, context=self.context
+        )
+
+        validator.delete()
+
+        # Verify remove_seat was called immediately
+        mock_remove_seat.assert_called_with(DataCategory.UPTIME, detector)
+
+    @mock.patch(
+        "sentry.quotas.backend.assign_seat",
+        return_value=Outcome.ACCEPTED,
+    )
+    def test_update_no_enable_change_no_seat_call(self, mock_assign_seat: mock.MagicMock) -> None:
+        """Test that updating without changing enabled status doesn't trigger seat operations."""
+        # Create an enabled detector
+        detector = self.create_uptime_detector(enabled=True)
+
+        # Clear any previous mock calls from creation
+        mock_assign_seat.reset_mock()
+
+        validator = UptimeDomainCheckFailureValidator(
+            instance=detector, data={"name": "Updated Name"}, context=self.context, partial=True
+        )
+        assert validator.is_valid(), validator.errors
+        validator.save()
+
+        detector.refresh_from_db()
+        assert detector.name == "Updated Name"
+        assert detector.enabled is True
+
+        # Verify no seat operations were called
+        mock_assign_seat.assert_not_called()