feat(jira): Add a cell RPC to apply Forge issue-updated events

src/sentry/integrations/jira/__init__.py

@@ -1,3 +1,5 @@
+# Registers the RPC service in every silo at startup, ahead of its first caller.
+from sentry.integrations.services.jira_forge import jira_forge_service  # noqa: F401
 from sentry.rules import rules
 
 from .actions import JiraCreateTicketAction

src/sentry/integrations/jira/tasks.py

@@ -1,16 +1,40 @@
+import logging
+from collections.abc import Callable, Mapping
+from typing import Any
+
 import orjson
+import sentry_sdk
+from django.conf import settings
 from django.db import router, transaction
 from taskbroker_client.retry import Retry
 
 from sentry import options
+from sentry.constants import ObjectStatus
+from sentry.hybridcloud.rpc.service import RpcRemoteException
 from sentry.integrations.jira.forge.client import ForgeInstallationClient
+from sentry.integrations.jira.utils import (
+    handle_assignee_change,
+    handle_issue_moved,
+    handle_status_change,
+)
 from sentry.integrations.models.integration import Integration
-from sentry.shared_integrations.exceptions import ApiError, IntegrationError
+from sentry.integrations.services.integration import RpcIntegration, integration_service
+from sentry.integrations.utils.scope import bind_org_context_from_integration
+from sentry.shared_integrations.exceptions import (
+    ApiConnectionResetError,
+    ApiError,
+    ApiHostError,
+    ApiRateLimitedError,
+    ApiTimeoutError,
+    IntegrationError,
+)
 from sentry.silo.base import SiloMode
 from sentry.tasks.base import instrumented_task
-from sentry.taskworker.namespaces import integrations_control_tasks
+from sentry.taskworker.namespaces import integrations_control_tasks, integrations_tasks
 from sentry.utils.security.encrypted_field_key_store import FernetKeyStore
 
+logger = logging.getLogger(__name__)
+
 FORGE_INSTALLATION_PAYLOAD_TTL = 15 * 60
 
 
@@ -96,3 +120,69 @@ def sync_forge_installation(key_id: str, encrypted_payload: str) -> None:
             return
         if existing != forge:
             integration.update(metadata={**integration.metadata, "forge": forge})
+
+
+@instrumented_task(
+    name="sentry.integrations.jira.tasks.apply_forge_issue_event",
+    namespace=integrations_tasks,
+    # Only temporary failures reach the retry (see _is_temporary). The handlers are
+    # idempotent, so the whole event is safe to apply again.
+    retry=Retry(times=5, delay=60, on=(ApiError, RpcRemoteException)),
+    # Longer than the Jira client's 30s timeout, so a slow call can still fail and retry.
+    processing_deadline_duration=60,
+    silo_mode=SiloMode.CELL,
+)
+def apply_forge_issue_event(integration_id: int, event: dict[str, Any]) -> None:
+    """Apply a verified Forge `avi:jira:updated:issue` event to the linked Sentry issues."""
+    integration = integration_service.get_integration(
+        integration_id=integration_id, status=ObjectStatus.ACTIVE
+    )
+    if integration is None:
+        return
+    bind_org_context_from_integration(integration.id, {"webhook": "forge_issue_updated"})
+
+    # Rekey first, since the other handlers look the issue up by its new key. Assignee goes
+    # last because its email lookup is the only call to Jira, so a failure there can't
+    # hold up the status change.
+    _apply_forge_change(handle_issue_moved, integration, event)
+    _apply_forge_change(handle_status_change, integration, event)
+    _apply_forge_change(
+        handle_assignee_change, integration, event, use_email_scope=settings.JIRA_USE_EMAIL_SCOPE
+    )
+
+
+def _is_temporary(error: Exception) -> bool:
+    if isinstance(
+        error,
+        (
+            ApiConnectionResetError,
+            ApiHostError,
+            ApiRateLimitedError,
+            ApiTimeoutError,
+            RpcRemoteException,
+        ),
+    ):
+        return True
+    # No status means Jira never responded, such as a broken connection.
+    return isinstance(error, ApiError) and (error.code is None or error.code >= 500)
+
+
+def _apply_forge_change(
+    handler: Callable[..., None],
+    integration: RpcIntegration,
+    event: Mapping[str, Any],
+    **kwargs: Any,
+) -> None:
+    try:
+        handler(integration, event, **kwargs)
+    except Exception as e:
+        if _is_temporary(e):
+            raise
+        # Retrying wouldn't help, such as a 404 for a deleted Jira account, so only this
+        # change is dropped and the rest of the event still applies.
+        logger.warning(
+            "jira.forge.issue_updated.failed",
+            extra={"integration_id": integration.id, "handler": handler.__name__},
+            exc_info=True,
+        )
+        sentry_sdk.capture_exception(e)

src/sentry/integrations/services/jira_forge/__init__.py

@@ -0,0 +1 @@
+from .service import *  # noqa

src/sentry/integrations/services/jira_forge/impl.py

@@ -0,0 +1,15 @@
+from __future__ import annotations
+
+from typing import Any
+
+from sentry.integrations.jira.tasks import apply_forge_issue_event
+from sentry.integrations.services.jira_forge.service import JiraForgeService
+
+
+class DatabaseBackedJiraForgeService(JiraForgeService):
+    def handle_issue_updated(
+        self, *, cell_name: str, integration_id: int, event: dict[str, Any]
+    ) -> None:
+        # Queued rather than applied here, so a temporary Jira failure is retried by the
+        # task instead of blocking the control outbox behind it.
+        apply_forge_issue_event.delay(integration_id=integration_id, event=event)

src/sentry/integrations/services/jira_forge/service.py

@@ -0,0 +1,36 @@
+# Please do not use
+#     from __future__ import annotations
+# in modules such as this one where hybrid cloud data models or service classes are
+# defined, because we want to reflect on type annotations and avoid forward references.
+
+from abc import abstractmethod
+from typing import Any
+
+from sentry.hybridcloud.rpc.resolvers import ByCellName
+from sentry.hybridcloud.rpc.service import RpcService, cell_rpc_method
+from sentry.silo.base import SiloMode
+
+
+class JiraForgeService(RpcService):
+    key = "jira_forge"
+    local_mode = SiloMode.CELL
+
+    @classmethod
+    def get_local_implementation(cls) -> RpcService:
+        from sentry.integrations.services.jira_forge.impl import DatabaseBackedJiraForgeService
+
+        return DatabaseBackedJiraForgeService()
+
+    @cell_rpc_method(resolve=ByCellName())
+    @abstractmethod
+    def handle_issue_updated(
+        self, *, cell_name: str, integration_id: int, event: dict[str, Any]
+    ) -> None:
+        """
+        Queue a Forge `avi:jira:updated:issue` event to be applied to the linked Sentry
+        issues in this cell. Control verifies the event before handing it over, so the
+        cell does no authentication of its own.
+        """
+
+
+jira_forge_service = JiraForgeService.create_delegation()

tests/sentry/integrations/jira/test_tasks.py

@@ -1,12 +1,18 @@
-from unittest.mock import patch
+from typing import Any
+from unittest.mock import MagicMock, patch
 
+import pytest
 import responses
 
+from sentry.constants import ObjectStatus
 from sentry.integrations.jira.client import JiraCloudClient
-from sentry.integrations.jira.tasks import sync_metadata
+from sentry.integrations.jira.tasks import apply_forge_issue_event, sync_metadata
 from sentry.integrations.models.integration import Integration
+from sentry.integrations.utils.external_issue_key import PROVIDER_ISSUE_ID_KEY
+from sentry.shared_integrations.exceptions import ApiConnectionResetError, ApiError
+from sentry.silo.base import SiloMode
 from sentry.testutils.cases import TestCase
-from sentry.testutils.silo import control_silo_test
+from sentry.testutils.silo import assume_test_silo_mode, cell_silo_test, control_silo_test
 
 
 @control_silo_test
@@ -90,3 +96,122 @@ def write_concurrently(client: JiraCloudClient) -> list[dict[str, object]]:
         assert integration.name == "Acme Jira"
         assert integration.metadata["icon"] == "https://example.com/avatar.jpg"
         assert integration.metadata["forge"] == {"system_token": "token"}
+
+
+EVENT: dict[str, Any] = {
+    "issue": {"id": "10001", "key": "APP-1", "fields": {}},
+    "changelog": {"id": "20001", "items": [{"field": "status", "toString": "Done"}]},
+}
+
+# The move as Forge sends it, captured from a test site: a project item plus a Key item
+# with fromString/toString set and no raw from/to.
+FORGE_MOVE_EVENT: dict[str, Any] = {
+    "eventType": "avi:jira:updated:issue",
+    "jiraEventTypeName": "issue_moved",
+    "issue": {
+        "id": "10008",
+        "key": "T2-1",
+        "fields": {
+            "assignee": None,
+            "project": {"id": "10003", "key": "T2"},
+            "status": {"id": "10011", "name": "Selected for development"},
+            "updated": "2025-11-10T12:41:22.000-0500",
+        },
+    },
+    "changelog": {
+        "id": "10026",
+        "items": [
+            {
+                "field": "project",
+                "fieldtype": "jira",
+                "fieldId": "project",
+                "from": "10002",
+                "fromString": "Forge Test",
+                "to": "10003",
+                "toString": "Test 2",
+            },
+            {"field": "Key", "fieldtype": "jira", "fromString": "FT-1", "toString": "T2-1"},
+        ],
+    },
+}
+
+
+@cell_silo_test
+@patch("sentry.integrations.jira.tasks.handle_status_change")
+@patch("sentry.integrations.jira.tasks.handle_assignee_change")
+@patch("sentry.integrations.jira.tasks.handle_issue_moved")
+class ApplyForgeIssueEventTest(TestCase):
+    def setUp(self) -> None:
+        super().setUp()
+        self.integration = self.create_integration(
+            organization=self.organization, provider="jira", external_id="cloud-id"
+        )
+
+    def test_applies_changes_in_order(
+        self, moved: MagicMock, assignee: MagicMock, status: MagicMock
+    ) -> None:
+        order = MagicMock()
+        order.attach_mock(moved, "moved")
+        order.attach_mock(assignee, "assignee")
+        order.attach_mock(status, "status")
+
+        apply_forge_issue_event(self.integration.id, EVENT)
+
+        # The move rekeys the issue first; assignee, the only Jira call, goes last.
+        assert [c[0] for c in order.mock_calls] == ["moved", "status", "assignee"]
+        assert moved.call_args.args[0].id == self.integration.id
+        assert moved.call_args.args[1] == EVENT
+
+    def test_skips_inactive_integration(
+        self, moved: MagicMock, assignee: MagicMock, status: MagicMock
+    ) -> None:
+        with assume_test_silo_mode(SiloMode.CONTROL):
+            self.integration.update(status=ObjectStatus.DISABLED)
+
+        apply_forge_issue_event(self.integration.id, EVENT)
+
+        assert not moved.called and not assignee.called and not status.called
+
+    @patch("sentry.integrations.jira.tasks.sentry_sdk.capture_exception")
+    def test_reports_other_failures_and_applies_the_rest(
+        self, capture: MagicMock, moved: MagicMock, assignee: MagicMock, status: MagicMock
+    ) -> None:
+        # A failure retrying can't fix: a deleted Jira account, or an unexpected payload.
+        status.__name__ = "handle_status_change"
+        status.side_effect = KeyError("unexpected payload")
+        assignee.__name__ = "handle_assignee_change"
+        assignee.side_effect = ApiError("Not found", code=404)
+
+        apply_forge_issue_event(self.integration.id, EVENT)
+
+        assert capture.call_count == 2
+        assert moved.call_count == 1
+
+    def test_raises_temporary_errors_for_retry(
+        self, moved: MagicMock, assignee: MagicMock, status: MagicMock
+    ) -> None:
+        for error in (ApiError("Jira is down", code=503), ApiConnectionResetError("reset")):
+            assignee.side_effect = error
+            with pytest.raises(type(error)):
+                apply_forge_issue_event(self.integration.id, EVENT)
+        # The status change is applied before each retry.
+        assert status.call_count == 2
+
+
+@cell_silo_test
+class ApplyForgeIssueEventRealHandlersTest(TestCase):
+    def test_forge_move_rekeys_external_issue(self) -> None:
+        integration = self.create_integration(
+            organization=self.organization, provider="jira", external_id="cloud-id"
+        )
+        external_issue = self.create_integration_external_issue(
+            group=self.create_group(), integration=integration, key="FT-1"
+        )
+
+        apply_forge_issue_event(integration.id, FORGE_MOVE_EVENT)
+        # Applying the event again, as a retry would, leaves the new key alone.
+        apply_forge_issue_event(integration.id, FORGE_MOVE_EVENT)
+
+        external_issue.refresh_from_db()
+        assert external_issue.key == "T2-1"
+        assert external_issue.metadata[PROVIDER_ISSUE_ID_KEY] == "10008"

tests/sentry/integrations/services/jira_forge/test_impl.py

@@ -0,0 +1,20 @@
+from unittest.mock import MagicMock, patch
+
+from django.conf import settings
+
+from sentry.integrations.services.jira_forge import jira_forge_service
+from sentry.testutils.cases import TestCase
+from sentry.testutils.silo import all_silo_test
+
+
+@all_silo_test
+class JiraForgeServiceTest(TestCase):
+    @patch("sentry.integrations.services.jira_forge.impl.apply_forge_issue_event.delay")
+    def test_queues_event(self, delay: MagicMock) -> None:
+        event = {"issue": {"key": "APP-1"}, "changelog": {"id": "1", "items": []}}
+
+        jira_forge_service.handle_issue_updated(
+            cell_name=settings.SENTRY_FALLBACK_CELL, integration_id=123, event=event
+        )
+
+        delay.assert_called_once_with(integration_id=123, event=event)