feat(crons): Backfill new broken monitor setting

migrations_lockfile.txt

@@ -10,6 +10,6 @@ hybridcloud: 0016_add_control_cacheversion
 nodestore: 0002_nodestore_no_dictfield
 remote_subscriptions: 0003_drop_remote_subscription
 replays: 0004_index_together
-sentry: 0742_backfill_alertrule_detection_type
+sentry: 0743_backfill_broken_monitor_notification_setting_option
 social_auth: 0002_default_auto_field
 uptime: 0005_uptime_status

src/sentry/migrations/0743_backfill_broken_monitor_notification_setting_option.py

@@ -0,0 +1,53 @@
+# Generated by Django 5.0.6 on 2024-07-17 08:41
+
+from django.apps.registry import Apps
+from django.db import migrations
+from django.db.backends.base.schema import BaseDatabaseSchemaEditor
+
+from sentry.new_migrations.migrations import CheckedMigration
+from sentry.utils.query import RangeQuerySetWrapperWithProgressBar
+
+
+def backfill_broken_monitor_notification_setting_option(
+    apps: Apps, schema_editor: BaseDatabaseSchemaEditor
+) -> None:
+    NotificationSettingOption = apps.get_model("sentry", "NotificationSettingOption")
+
+    # No index on the type column, so scan all rows manually to avoid timing out
+    notif_settings = NotificationSettingOption.objects.all()
+    for setting in RangeQuerySetWrapperWithProgressBar(notif_settings):
+        if setting.type == "approval":
+            NotificationSettingOption.objects.get_or_create(
+                user_id=setting.user_id,
+                scope_type=setting.scope_type,
+                scope_identifier=setting.scope_identifier,
+                type="brokenMonitors",
+                defaults={"value": setting.value},
+            )
+
+
+class Migration(CheckedMigration):
+    # This flag is used to mark that a migration shouldn't be automatically run in production.
+    # This should only be used for operations where it's safe to run the migration after your
+    # code has deployed. So this should not be used for most operations that alter the schema
+    # of a table.
+    # Here are some things that make sense to mark as post deployment:
+    # - Large data migrations. Typically we want these to be run manually so that they can be
+    #   monitored and not block the deploy for a long period of time while they run.
+    # - Adding indexes to large tables. Since this can take a long time, we'd generally prefer to
+    #   run this outside deployments so that we don't block them. Note that while adding an index
+    #   is a schema change, it's completely safe to run the operation after the code has deployed.
+    # Once deployed, run these manually via: https://develop.sentry.dev/database-migrations/#migration-deployment
+    is_post_deployment = True
+
+    dependencies = [
+        ("sentry", "0742_backfill_alertrule_detection_type"),
+    ]
+
+    operations = [
+        migrations.RunPython(
+            backfill_broken_monitor_notification_setting_option,
+            reverse_code=migrations.RunPython.noop,
+            hints={"tables": ["sentry_notificationsettingoption"]},
+        )
+    ]

tests/sentry/migrations/test_0743_backfill_broken_monitor_notification_setting_option.py

@@ -0,0 +1,77 @@
+from uuid import uuid4
+
+from sentry.models.notificationsettingoption import NotificationSettingOption
+from sentry.testutils.cases import TestMigrations
+from sentry.testutils.silo import control_silo_test
+
+
+@control_silo_test
+class BackfillBrokenMonitorNotificationSettingOptionTest(TestMigrations):
+    migrate_from = "0742_backfill_alertrule_detection_type"
+    migrate_to = "0743_backfill_broken_monitor_notification_setting_option"
+    connection = "control"
+
+    def setup_before_migration(self, apps):
+        NotificationSettingOptionModel = apps.get_model("sentry", "NotificationSettingOption")
+        User = apps.get_model("sentry", "User")
+
+        def create_user():
+            email = uuid4().hex + "@example.com"
+            return User.objects.create(
+                email=email, username=email, is_staff=True, is_active=True, is_superuser=False
+            )
+
+        self.user1 = create_user()
+        NotificationSettingOptionModel.objects.create(
+            user_id=self.user1.id,
+            scope_type="user",
+            scope_identifier=self.user1.id,
+            type="approval",
+            value="never",
+        )
+        self.user2 = create_user()
+        NotificationSettingOptionModel.objects.create(
+            user_id=self.user2.id,
+            scope_type="user",
+            scope_identifier=self.user2.id,
+            type="approval",
+            value="always",
+        )
+        # Test that an existing brokenMonitors row isn't duplicated or overridden
+        self.user3 = create_user()
+        NotificationSettingOptionModel.objects.create(
+            user_id=self.user3.id,
+            scope_type="user",
+            scope_identifier=self.user3.id,
+            type="brokenMonitors",
+            value="always",
+        )
+        NotificationSettingOptionModel.objects.create(
+            user_id=self.user3.id,
+            scope_type="user",
+            scope_identifier=self.user3.id,
+            type="approval",
+            value="never",
+        )
+
+    def test(self):
+        broken_notifs = NotificationSettingOption.objects.filter(type="brokenMonitors")
+        assert broken_notifs.count() == 3
+
+        never_notif = broken_notifs.get(user_id=self.user1.id)
+        assert never_notif.value == "never"
+        assert never_notif.user_id == self.user1.id
+        assert never_notif.scope_type == "user"
+        assert never_notif.scope_identifier == self.user1.id
+
+        always_notif = broken_notifs.get(user_id=self.user2.id)
+        assert always_notif.value == "always"
+        assert always_notif.user_id == self.user2.id
+        assert always_notif.scope_type == "user"
+        assert always_notif.scope_identifier == self.user2.id
+
+        existing_notif = broken_notifs.get(user_id=self.user3.id)
+        assert existing_notif.value == "always"
+        assert existing_notif.user_id == self.user3.id
+        assert existing_notif.scope_type == "user"
+        assert existing_notif.scope_identifier == self.user3.id