feat(notifications): Store the integration on NotificationThread

migrations_lockfile.txt

@@ -27,7 +27,7 @@ nodestore: 0001_squashed_0002_nodestore_no_dictfield
 
 notebooks: 0002_copy_from_investigations
 
-notifications: 0016_add_inboxitem_and_inboxitemevent
+notifications: 0017_add_notificationthread_integration_id
 
 preprod: 0001_squashed_0030_add_images_errored_to_snapshot_comparison
 

src/sentry/notifications/migrations/0017_add_notificationthread_integration_id.py

@@ -0,0 +1,36 @@
+# Generated by Django 5.2.17 on 2026-10-29 17:14
+
+import sentry.db.models.fields.hybrid_cloud_foreign_key
+from django.db import migrations
+
+from sentry.new_migrations.migrations import CheckedMigration
+
+
+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 = False
+
+    dependencies = [
+        ("notifications", "0016_add_inboxitem_and_inboxitemevent"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="notificationthread",
+            name="integration_id",
+            field=sentry.db.models.fields.hybrid_cloud_foreign_key.HybridCloudForeignKey(
+                "sentry.Integration", blank=True, db_index=True, null=True, on_delete="CASCADE"
+            ),
+        ),
+    ]

src/sentry/notifications/models/notificationthread.py

@@ -6,6 +6,7 @@
 
 from sentry.backup.scopes import RelocationScope
 from sentry.db.models import BoundedBigIntegerField, DefaultFieldsModel, cell_silo_model, sane_repr
+from sentry.db.models.fields.hybrid_cloud_foreign_key import HybridCloudForeignKey
 from sentry.notifications.platform.types import NotificationSource
 
 
@@ -62,6 +63,10 @@ class NotificationThread(DefaultFieldsModel):
     )
     object_id = BoundedBigIntegerField(null=True)
 
+    integration_id = HybridCloudForeignKey(
+        "sentry.Integration", blank=True, null=True, on_delete="CASCADE"
+    )
+
     provider_data = models.JSONField(default=dict)
 
     class Meta:

src/sentry/notifications/platform/service.py

@@ -21,7 +21,11 @@
 )
 from sentry.notifications.platform.registry import provider_registry, template_registry
 from sentry.notifications.platform.rollout import NotificationRolloutService
-from sentry.notifications.platform.target import deserialize_target, serialize_target
+from sentry.notifications.platform.target import (
+    IntegrationNotificationTarget,
+    deserialize_target,
+    serialize_target,
+)
 from sentry.notifications.platform.threading import (
     ThreadContext,
     ThreadingConfig,
@@ -316,6 +320,11 @@ def _handle_threading_result(
                     provider_data=None,
                     object_type=threading_options.thread_key.object_type,
                     object_id=threading_options.thread_key.object_id,
+                    integration_id=(
+                        target.integration_id
+                        if isinstance(target, IntegrationNotificationTarget)
+                        else None
+                    ),
                 )
                 ThreadingService.store_new_thread(
                     threading_config=threading_config,

src/sentry/notifications/platform/threading.py

@@ -109,6 +109,8 @@ class ThreadingConfig(ThreadingLookup):
     object_type: NotRequired[NotificationThreadObjectType | None]
     object_id: NotRequired[int | None]
 
+    integration_id: NotRequired[int | None]
+
 
 class ThreadingService:
     """
@@ -182,6 +184,7 @@ def _create_thread_and_record(
                     "key_data": threading_config["key_data"],
                     "object_type": threading_config.get("object_type"),
                     "object_id": threading_config.get("object_id"),
+                    "integration_id": threading_config.get("integration_id"),
                     "provider_data": threading_config["provider_data"] or {},
                 },
             )
@@ -218,6 +221,19 @@ def store_new_thread(
             threading_config, external_message_id, thread_key
         )
 
+    @staticmethod
+    def get_threads_for_object(
+        *,
+        object_type: NotificationThreadObjectType,
+        object_id: int,
+        provider_key: NotificationProviderKey | None = None,
+    ) -> list[NotificationThread]:
+        """Return all threads for the object. If provider_key is set, return only its threads."""
+        queryset = NotificationThread.objects.filter(object_type=object_type, object_id=object_id)
+        if provider_key is not None:
+            queryset = queryset.filter(provider_key=provider_key)
+        return list(queryset)
+
     @staticmethod
     def store_existing_thread(
         *,

tests/sentry/notifications/notification_action/test_platform_dispatch.py

@@ -145,6 +145,7 @@ def test_follow_up_replies_in_thread(self, mock_slack_client: mock.MagicMock) ->
         assert thread.object_type == NotificationThreadObjectType.GROUP
         assert thread.object_id == self.event_data.group.id
         assert thread.key_data["group_id"] == self.event_data.group.id
+        assert thread.integration_id == self.integration.id
 
     @mock.patch("sentry.integrations.slack.integration.SlackSdkClient")
     def test_metric_thread_flag_off_posts_top_level(

tests/sentry/notifications/platform/test_threading.py

@@ -183,6 +183,7 @@ def test_store_new_thread_creates_thread_and_record(self) -> None:
         assert thread.provider_data == {}
         assert thread.object_type is None
         assert thread.object_id is None
+        assert thread.integration_id is None
 
         assert record.id is not None
         assert record.thread_id == thread.id
@@ -211,6 +212,15 @@ def test_store_new_thread_stores_object(self) -> None:
             == thread
         )
 
+    def test_store_new_thread_stores_integration_id(self) -> None:
+        thread, _ = ThreadingService.store_new_thread(
+            threading_config={**self.threading_config, "integration_id": 42},
+            external_message_id=self.message_id,
+        )
+
+        assert thread.integration_id == 42
+        assert thread.thread_key == self.thread_key
+
     def test_store_new_thread_uses_existing_thread_on_race(self) -> None:
         """If two workers race, the second get_or_create returns the first's thread."""
         first_thread, _ = ThreadingService.store_new_thread(
@@ -236,6 +246,66 @@ def test_store_new_thread_uses_existing_thread_on_race(self) -> None:
         assert NotificationRecord.objects.count() == 2
 
 
+class ThreadingServiceGetThreadsForObjectTest(ThreadingServiceTestBase):
+    def store_thread(
+        self,
+        *,
+        target_id: str,
+        object_id: int,
+        provider_key: NotificationProviderKey = NotificationProviderKey.SLACK,
+    ) -> NotificationThread:
+        thread, _ = ThreadingService.store_new_thread(
+            threading_config={
+                **self.threading_config,
+                "provider_key": provider_key,
+                "target_id": target_id,
+                "object_type": NotificationThreadObjectType.GROUP,
+                "object_id": object_id,
+            },
+            external_message_id=self.message_id,
+        )
+        return thread
+
+    def test_returns_all_threads_for_object(self) -> None:
+        first = self.store_thread(target_id="C1111111111", object_id=456)
+        second = self.store_thread(target_id="C2222222222", object_id=456)
+        self.store_thread(target_id="C3333333333", object_id=789)
+        ThreadingService.store_new_thread(
+            threading_config={**self.threading_config, "target_id": "C4444444444"},
+            external_message_id=self.message_id,
+        )
+
+        threads = ThreadingService.get_threads_for_object(
+            object_type=NotificationThreadObjectType.GROUP, object_id=456
+        )
+
+        assert {thread.id for thread in threads} == {first.id, second.id}
+
+    def test_filters_by_provider_key(self) -> None:
+        slack_thread = self.store_thread(target_id="C1111111111", object_id=456)
+        self.store_thread(
+            target_id="C1111111111",
+            object_id=456,
+            provider_key=NotificationProviderKey.SLACK_STAGING,
+        )
+
+        threads = ThreadingService.get_threads_for_object(
+            object_type=NotificationThreadObjectType.GROUP,
+            object_id=456,
+            provider_key=NotificationProviderKey.SLACK,
+        )
+
+        assert [thread.id for thread in threads] == [slack_thread.id]
+
+    def test_returns_empty_when_no_threads(self) -> None:
+        assert (
+            ThreadingService.get_threads_for_object(
+                object_type=NotificationThreadObjectType.GROUP, object_id=456
+            )
+            == []
+        )
+
+
 class ThreadingServiceStoreExistingThreadTest(ThreadingServiceTestBase):
     def test_store_existing_thread_links_record_to_thread(self) -> None:
         # Create a thread first