feat: ✨ Add replication lookback window setting (#91)

* set `is_sorted` to true

set this so that progress can be resumed in incremental syncs.

* add boto3 as explicit requirement

* set Intercom API version to 2.5 and add sorting to conversations payload

* add timestamp normalization methods to IntercomStream and update conversation post-processing

* add replication lookback window setting to configuration options

* set `is_sorted` to false for ConversationsStream and remove unused request payload preparation

* update Intercom API version to 2.15 and enhance request payload preparation with lookback window logic

* remove sorting from request payload in prepare_request_payload method

* remove logging of prepared request payload in prepare_request_payload method

* remove timestamp handling methods and simplify request payload preparation

* remove unused datetime import from client.py

* Apply suggested changes

* Add Unix timestamp signpost defaulting to sync start time for incremental Intercom streams in get_replication_key_signpost method

* Add logging to replication key signpost logic

* Apply suggestions from code review

Co-authored-by: Tobias Cadée <tobias.cadee@ticketswap.com>

* Simplify post_process method in ConversationPartsStream to return modified row directly

* Add missing paranthesis

---------

Co-authored-by: Tobias Cadee <tobias.cadee@ticketswap.com>

Author: fcortevargas

README.md

@@ -42,15 +42,16 @@ Built with the [Meltano Singer SDK](https://sdk.meltano.com).
 
 ## Settings
 
-| Setting             | Required | Default | Description |
-|:--------------------|:--------:|:-------:|:------------|
-| access_token        | True     | None    | The token to authenticate against the API service |
-| start_date          | False    | None    | The earliest record date to sync |
-| stream_maps         | False    | None    | Config object for stream maps capability. For more information check out [Stream Maps](https://sdk.meltano.com/en/latest/stream_maps.html). |
-| stream_map_config   | False    | None    | User-defined config values to be used within map expressions. |
-| flattening_enabled  | False    | None    | 'True' to enable schema flattening and automatically expand nested properties. |
-| flattening_max_depth| False    | None    | The max depth to flatten schemas. |
-| batch_config        | False    | None    |             |
+| Setting                              | Required | Default | Description                                                                                                                          |
+|:-------------------------------------|:--------:|:-------:|:-------------------------------------------------------------------------------------------------------------------------------------|
+| access_token                         |   True   |  None   | The token to authenticate against the API service                                                                                    |
+| start_date                           |  False   |  None   | The earliest record date to sync                                                                                                     |
+| replication_lookback_window_seconds  |  False   |    0    | Overlap window in seconds for incremental replication to replay recent records and reduce misses near bookmark boundaries            |
+| stream_maps                          |  False   |  None   | Config object for stream maps capability.                                                                                            |
+| stream_map_config                    |  False   |  None   | User-defined config values to be used within map expressions.                                                                        |
+| flattening_enabled                   |  False   |  None   | 'True' to enable schema flattening and automatically expand nested properties.                                                       |
+| flattening_max_depth                 |  False   |  None   | The max depth to flatten schemas.                                                                                                    |
+| batch_config                         |  False   |  None   |                                                                                                                                      |
 
 A full list of supported settings and capabilities for this
 tap is available by running:

pyproject.toml

@@ -25,6 +25,7 @@ dependencies = [
     "singer-sdk>=0.42.1",
     "fs-s3fs",
     "requests",
+    "boto3>=1.40.18",
 ]
 
 [project.optional-dependencies]

tap_intercom/client.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import time
 import typing as t
 
 from singer_sdk.authenticators import BearerTokenAuthenticator
@@ -46,7 +47,7 @@ def http_headers(self) -> dict:
         if user_agent:
             result["User-Agent"] = user_agent
         result["Content-Type"] = "application/json"
-        result["Intercom-Version"] = "2.14"
+        result["Intercom-Version"] = "2.15"
         return result
 
     def get_url_params(self, context: dict | None, next_page_token: object) -> dict:  # noqa: ARG002
@@ -83,7 +84,7 @@ def prepare_request_payload(
                 next page of data.
         """
         if self.http_method == "POST":
-            body = {"sort": {"field": self.replication_key, "order": "ascending"}}
+            body = {}
             start_date = self.get_starting_replication_key_value(context)
             if start_date or self.config.get("filters", {}).get(self.name):
                 body["query"] = {
@@ -94,6 +95,8 @@ def prepare_request_payload(
                     ],
                 }
                 if start_date:
+                    if start_date != self.config.get("start_date"):
+                        start_date -= int(self.config["replication_lookback_window_seconds"])
                     body["query"]["value"].append(
                         {
                             "field": self.replication_key,
@@ -118,6 +121,28 @@ def compare_start_date(self, value: str, start_date_value: str) -> str:
         """
         return max(value, start_date_value)
 
+    def get_replication_key_signpost(
+        self,
+        context: dict | None,  # noqa: ARG002
+    ) -> int | None:
+        """Overrides the signpost to be the Unix integer at sync start for incremental streams.
+
+        This enables the SDK to finalize state as the lower of max(replication_key_value)
+        seen during the run and the run start time.
+
+        Args:
+            context: Stream partition or context dictionary.
+
+        Returns:
+            Unix timestamp signpost for integer replication keys, else None.
+        """
+        if not self.replication_key:
+            return None
+        signpost = int(time.time())
+        self.logger.info("Setting replication key signpost to current Unix timestamp at sync start.")
+        self.logger.info("Signpost value: %s", signpost)
+        return signpost
+
     def post_process(
         self,
         row: dict,

tap_intercom/streams.py

@@ -31,6 +31,7 @@ class ConversationsStream(IntercomStream):
     records_jsonpath = "$.conversations[*]"
     http_method = "POST"
     schema = conversations_schema
+    is_sorted = False
 
     def get_child_context(self, record: dict, context: dict | None) -> dict:  # noqa: ARG002
         """Return a context dictionary for child streams."""

tap_intercom/tap.py

@@ -26,6 +26,16 @@ class TapIntercom(Tap):
             th.IntegerType,
             description="The earliest record date to sync, in unix timestamp format",
         ),
+        th.Property(
+            "replication_lookback_window_seconds",
+            th.IntegerType,
+            default=0,
+            description=(
+                "Optional overlap window in seconds applied to incremental bookmarks. "
+                "Use this to replay recent records and reduce misses caused by API pagination drift "
+                "or near-boundary updates."
+            ),
+        ),
         th.Property(
             "filters",
             th.ObjectType(