fix(source-intercom): use state for parent stream requests and use global state

airbyte-integrations/connectors/source-intercom/components.py

@@ -5,10 +5,11 @@
 from dataclasses import dataclass
 from functools import wraps
 from time import sleep
-from typing import Mapping, Optional, Union
+from typing import Any, Mapping, Optional, Union
 
 import requests
 
+from airbyte_cdk.sources.declarative.migrations.state_migration import StateMigration
 from airbyte_cdk.sources.declarative.requesters.error_handlers import DefaultErrorHandler
 from airbyte_cdk.sources.streams.http.error_handlers.response_models import ErrorResolution
 
@@ -152,3 +153,51 @@ class ErrorHandlerWithRateLimiter(DefaultErrorHandler):
     def interpret_response(self, response_or_exception: Optional[Union[requests.Response, Exception]]) -> ErrorResolution:
         # Check for response.headers to define the backoff time before the next api call
         return super().interpret_response(response_or_exception)
+
+
+class SubstreamStateMigration(StateMigration):
+    """
+    We require a custom state migration to move from the custom substream state that was generated via the legacy
+    cursor custom components. State was not written back to the platform in a way that is compatible with concurrent cursors.
+
+    The old state roughly had the following shape:
+    {
+        "updated_at": 1744153060,
+        "prior_state": {
+            "updated_at": 1744066660
+        }
+        "conversations": {
+            "updated_at": 1744153060
+        }
+    }
+
+    However, this was incompatible when we removed the custom cursors with the concurrent substream partition cursor
+    components that were configured with use global_substream_cursor and incremental_dependency. They rely on passing the value
+    of parent_state when getting parent records for the conversations/companies parent stream. The migration results in state:
+    {
+        "updated_at": 1744153060,
+        "prior_state": {
+            "updated_at": 1744066660
+            # There are a lot of nested elements here, but are not used or relevant to syncs
+        }
+        "conversations": {
+            "updated_at": 1744153060
+        }
+        "parent_state": {
+            "conversations": {
+                "updated_at": 1744153060
+            }
+        }
+    }
+    """
+
+    def should_migrate(self, stream_state: Mapping[str, Any]) -> bool:
+        return "parent_state" not in stream_state and ("conversations" in stream_state or "companies" in stream_state)
+
+    def migrate(self, stream_state: Mapping[str, Any]) -> Mapping[str, Any]:
+        migrated_parent_state = {}
+        if stream_state.get("conversations"):
+            migrated_parent_state["conversations"] = stream_state.get("conversations")
+        if stream_state.get("companies"):
+            migrated_parent_state["companies"] = stream_state.get("companies")
+        return {**stream_state, "parent_state": migrated_parent_state}

airbyte-integrations/connectors/source-intercom/manifest.yaml

@@ -649,6 +649,7 @@ definitions:
                 $ref: "#/definitions/streams/conversations"
               parent_key: id
               partition_field: id
+              incremental_dependency: true
       incremental_sync:
         type: DatetimeBasedCursor
         cursor_field: updated_at
@@ -661,6 +662,7 @@ definitions:
         cursor_datetime_formats:
           - "%s"
         is_client_side_incremental: true
+        global_substream_cursor: true
       transformations:
         - type: AddFields
           fields:
@@ -671,6 +673,9 @@ definitions:
         type: InlineSchemaLoader
         schema:
           $ref: "#/schemas/conversation_parts"
+      state_migrations:
+        - type: CustomStateMigration
+          class_name: source_declarative_manifest.components.SubstreamStateMigration
     company_segments:
       type: DeclarativeStream
       name: company_segments
@@ -724,10 +729,14 @@ definitions:
         cursor_datetime_formats:
           - "%s"
         is_client_side_incremental: true
+        global_substream_cursor: true
       schema_loader:
         type: InlineSchemaLoader
         schema:
           $ref: "#/schemas/company_segments"
+      state_migrations:
+        - type: CustomStateMigration
+          class_name: source_declarative_manifest.components.SubstreamStateMigration
     tickets:
       type: DeclarativeStream
       name: tickets

airbyte-integrations/connectors/source-intercom/metadata.yaml

@@ -10,7 +10,7 @@ data:
   connectorSubtype: api
   connectorType: source
   definitionId: d8313939-3782-41b0-be29-b3ca20d8dd3a
-  dockerImageTag: 0.13.0-rc.1
+  dockerImageTag: 0.13.0-rc.2
   dockerRepository: airbyte/source-intercom
   documentationUrl: https://docs.airbyte.com/integrations/sources/intercom
   githubIssueLabel: source-intercom