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

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()