Repository: amp-labs/connectors
Pull request #1712 (2025-06-02 18:25:37)

[CON-1736] feat(hubspot): Read until
# Changes
* Added mock tests to validate existing Since functionality.
* Extended support to include Until parameter, applied together with Since via filterGroups.filters in the request payload.

# Live tests
Validated with contacts object by filtering between two specific timestamps:
![image](https://github.com/user-attachments/assets/dfcf527a-be7a-40d8-96c2-335c71d79500)


# Exceptions
The Lists object is outside of the standard Objects API and uses a distinct request and response format.
It does not support incremental sync via Since, so Until is also not applied (for now).

Standard reading of lists remains unaffected:
![image](https://github.com/user-attachments/assets/99441b56-8446-45f0-a02f-8cdf5cb341ac)


Cobalt0s: > [!WARNING]
> <b>This pull request is not mergeable via GitHub because a downstack PR is open. Once all requirements are satisfied, merge this PR as a stack <a href="https://app.graphite.dev/github/pr/amp-labs/connectors/1712?utm_source=stack-comment-downstack-mergeability-warning" >on Graphite</a>.</b>
> <a href="https://graphite.dev/docs/merge-pull-requests">Learn more</a>

* **#1712** <a href="https://app.graphite.dev/github/pr/amp-labs/connectors/1712?utm_source=stack-comment-icon" target="_blank"><img src="https://static.graphite.dev/graphite-32x32-black.png" alt="Graphite" width="10px" height="10px"/></a> 👈 <a href="https://app.graphite.dev/github/pr/amp-labs/connectors/1712?utm_source=stack-comment-view-in-graphite" target="_blank">(View in Graphite)</a>
* **#1711** <a href="https://app.graphite.dev/github/pr/amp-labs/connectors/1711?utm_source=stack-comment-icon" target="_blank"><img src="https://static.graphite.dev/graphite-32x32-black.png" alt="Graphite" width="10px" height="10px"/></a>
* `main`


<h2></h2>

This stack of pull requests is managed by <a href="https://graphite.dev?utm-source=stack-comment"><b>Graphite</b></a>. Learn more about <a href="https://stacking.dev/?utm_source=stack-comment">stacking</a>.
<!-- Current dependencies on/for this PR: -->

Patch:
diff --git a/providers/hubspot/params.go b/providers/hubspot/params.go
index 9251c8fd..860d381a 100644
--- a/providers/hubspot/params.go
+++ b/providers/hubspot/params.go
@@ -68,7 +68,3 @@ func WithModule(module common.ModuleID) Option {
 		params.WithModule(module, supportedModules, common.ModuleRoot)
 	}
 }
-
-func requiresFiltering(config common.ReadParams) bool {
-	return !config.Since.IsZero()
-}
diff --git a/providers/hubspot/parse.go b/providers/hubspot/parse.go
index 4880f9eb..6d7eb816 100644
--- a/providers/hubspot/parse.go
+++ b/providers/hubspot/parse.go
@@ -9,6 +9,20 @@ import (
 	"github.com/spyzhov/ajson"
 )
 
+/*
+Pagination format:
+
+{
+  "results": [...],
+  "paging": {
+    "next": {
+      "after": "394",
+      "link": "https://api.hubapi.com/crm/v3/objects/contacts?limit=100&properties=listId%2Cname&after=394"
+    }
+  }
+}
+*/
+
 // getNextRecordsAfter returns the "after" value for the next page of results.
 func getNextRecordsAfter(node *ajson.Node) (string, error) {
 	var nextPage string
diff --git a/providers/hubspot/read.go b/providers/hubspot/read.go
index 5491dec9..83dee9e6 100644
--- a/providers/hubspot/read.go
+++ b/providers/hubspot/read.go
@@ -37,18 +37,22 @@ func (c *Connector) Read(ctx context.Context, config common.ReadParams) (*common
 	// The Search endpoint has a 10K record limit. In case this limit is reached,
 	// the sorting allows the caller to continue in another call by offsetting
 	// until the ID of the last record that was successfully fetched.
-	if requiresFiltering(config) {
+	filters := make(Filters, 0)
+	if !config.Since.IsZero() {
+		filters = append(filters, BuildLastModifiedFilterGroup(&config))
+	}
+
+	if !config.Until.IsZero() {
+		filters = append(filters, BuildUntilTimestampFilterGroup(&config))
+	}
+
+	if len(filters) != 0 {
 		searchParams := SearchParams{
 			ObjectName: config.ObjectName,
-			FilterGroups: []FilterGroup{
-				{
-					Filters: []Filter{
-						BuildLastModifiedFilterGroup(&config),
-						// Add more filters to AND them together
-					},
-					// Add more filter groups to OR them together
-				},
-			},
+			FilterGroups: []FilterGroup{{
+				Filters: filters,
+				// Add more filter groups to OR them together
+			}},
 			SortBy: []SortBy{
 				BuildSort(ObjectFieldHsObjectId, SortDirectionAsc),
 			},
diff --git a/providers/hubspot/search.go b/providers/hubspot/search.go
index 81295f6a..42c0ada9 100644
--- a/providers/hubspot/search.go
+++ b/providers/hubspot/search.go
@@ -113,6 +113,25 @@ func BuildLastModifiedFilterGroup(params *common.ReadParams) Filter {
 	}
 }
 
+// BuildUntilTimestampFilterGroup filters records modified until and including the given time.
+func BuildUntilTimestampFilterGroup(params *common.ReadParams) Filter {
+	if params.Until.IsZero() {
+		return Filter{}
+	}
+
+	// Use the lastmodifieddate field for contacts, and hs_lastmodifieddate for other objects.
+	lastModifiedField := ObjectFieldHsLastModifiedDate
+	if params.ObjectName == string(ObjectTypeContact) {
+		lastModifiedField = ObjectFieldLastModifiedDate
+	}
+
+	return Filter{
+		FieldName: string(lastModifiedField),
+		Operator:  FilterOperatorTypeLTE,
+		Value:     params.Until.Format(time.RFC3339),
+	}
+}
+
 // BuildIdFilterGroup filters records greater than the given id.
 func BuildIdFilterGroup(id string) Filter {
 	return Filter{
diff --git a/providers/hubspot/test/read-contacts-objects-api.json b/providers/hubspot/test/read/objects-api/contacts-response.json
similarity index 100%
rename from providers/hubspot/test/read-contacts-objects-api.json
rename to providers/hubspot/test/read/objects-api/contacts-response.json
diff --git a/providers/hubspot/types.go b/providers/hubspot/types.go
index 0babb011..78cf16de 100644
--- a/providers/hubspot/types.go
+++ b/providers/hubspot/types.go
@@ -79,6 +79,8 @@ type FilterGroup struct {
 	Filters []Filter `json:"filters,omitempty"`
 }
 
+type Filters []Filter
+
 type Filter struct {
 	FieldName string             `json:"propertyName,omitempty"`
 	Operator  FilterOperatorType `json:"operator,omitempty"`
diff --git a/providers/salesforce/read.go b/providers/salesforce/read.go
index ce135120..2b034afe 100644
--- a/providers/salesforce/read.go
+++ b/providers/salesforce/read.go
@@ -66,6 +66,10 @@ func makeSOQL(config common.ReadParams) *soqlBuilder {
 		soql.Where("SystemModstamp > " + datautils.Time.FormatRFC3339inUTC(config.Since))
 	}
 
+	if !config.Until.IsZero() {
+		soql.Where("SystemModstamp <= " + datautils.Time.FormatRFC3339inUTC(config.Until))
+	}
+
 	if config.Deleted {
 		soql.Where("IsDeleted = true")
 	}


Test patch:
diff --git a/providers/hubspot/read_test.go b/providers/hubspot/read_test.go
index c45357a2..8eddd20c 100644
--- a/providers/hubspot/read_test.go
+++ b/providers/hubspot/read_test.go
@@ -3,6 +3,7 @@ package hubspot
 import (
 	"net/http"
 	"testing"
+	"time"
 
 	"github.com/amp-labs/connectors"
 	"github.com/amp-labs/connectors/common"
@@ -15,7 +16,10 @@ import (
 func TestRead(t *testing.T) { //nolint:funlen,gocognit,cyclop,maintidx
 	t.Parallel()
 
-	responseContacts := testutils.DataFromFile(t, "read-contacts-objects-api.json")
+	requestContactsSince := testutils.DataFromFile(t, "read/objects-api/contacts-req-payload-since.json")
+	requestContactsUntil := testutils.DataFromFile(t, "read/objects-api/contacts-req-payload-until.json")
+	requestContactsSinceUntil := testutils.DataFromFile(t, "read/objects-api/contacts-req-payload-since-until.json")
+	responseContacts := testutils.DataFromFile(t, "read/objects-api/contacts-response.json")
 	responseListsFirst := testutils.DataFromFile(t, "read-lists-1-first-page.json")
 	responseListsLast := testutils.DataFromFile(t, "read-lists-2-second-page.json")
 
@@ -102,6 +106,77 @@ func TestRead(t *testing.T) { //nolint:funlen,gocognit,cyclop,maintidx
 			},
 			ExpectedErrs: nil,
 		},
+		{
+			Name: "Contacts records since time",
+			Input: common.ReadParams{
+				ObjectName: "contacts",
+				Fields:     connectors.Fields("email"),
+				Since: time.Date(2024, 9, 19, 4, 30, 45, 600,
+					time.FixedZone("UTC-8", -8*60*60)),
+			},
+			Server: mockserver.Conditional{
+				Setup: mockserver.ContentJSON(),
+				If: mockcond.And{
+					mockcond.Path("/crm/v3/objects/contacts/search"),
+					mockcond.BodyBytes(requestContactsSince),
+				},
+				Then: mockserver.Response(http.StatusOK, responseContacts),
+			}.Server(),
+			Comparator: testroutines.ComparatorPagination,
+			Expected: &common.ReadResult{
+				Rows:     3,
+				NextPage: "394",
+				Done:     false,
+			},
+		},
+		{
+			Name: "Contacts records until time",
+			Input: common.ReadParams{
+				ObjectName: "contacts",
+				Fields:     connectors.Fields("email"),
+				Until: time.Date(2025, 1, 1, 0, 0, 0, 0,
+					time.FixedZone("UTC-8", -8*60*60)),
+			},
+			Server: mockserver.Conditional{
+				Setup: mockserver.ContentJSON(),
+				If: mockcond.And{
+					mockcond.Path("/crm/v3/objects/contacts/search"),
+					mockcond.BodyBytes(requestContactsUntil),
+				},
+				Then: mockserver.Response(http.StatusOK, responseContacts),
+			}.Server(),
+			Comparator: testroutines.ComparatorPagination,
+			Expected: &common.ReadResult{
+				Rows:     3,
+				NextPage: "394",
+				Done:     false,
+			},
+		},
+		{
+			Name: "Contacts records from 'since' till 'until'",
+			Input: common.ReadParams{
+				ObjectName: "contacts",
+				Fields:     connectors.Fields("email"),
+				Since: time.Date(2024, 9, 19, 4, 30, 45, 600,
+					time.FixedZone("UTC-8", -8*60*60)),
+				Until: time.Date(2025, 1, 1, 0, 0, 0, 0,
+					time.FixedZone("UTC-8", -8*60*60)),
+			},
+			Server: mockserver.Conditional{
+				Setup: mockserver.ContentJSON(),
+				If: mockcond.And{
+					mockcond.Path("/crm/v3/objects/contacts/search"),
+					mockcond.BodyBytes(requestContactsSinceUntil),
+				},
+				Then: mockserver.Response(http.StatusOK, responseContacts),
+			}.Server(),
+			Comparator: testroutines.ComparatorPagination,
+			Expected: &common.ReadResult{
+				Rows:     3,
+				NextPage: "394",
+				Done:     false,
+			},
+		},
 		{
 			Name: "Lists first page is done via search",
 			Input: common.ReadParams{
@@ -110,8 +185,11 @@ func TestRead(t *testing.T) { //nolint:funlen,gocognit,cyclop,maintidx
 			},
 			Server: mockserver.Conditional{
 				Setup: mockserver.ContentJSON(),
-				If:    mockcond.Path("/crm/v3/lists/search"),
-				Then:  mockserver.Response(http.StatusOK, responseListsFirst),
+				If: mockcond.And{
+					mockcond.Path("/crm/v3/lists/search"),
+					mockcond.Body(`{"offset":0,"count":100}`),
+				},
+				Then: mockserver.Response(http.StatusOK, responseListsFirst),
 			}.Server(),
 			Comparator: testroutines.ComparatorSubsetRead,
 			Expected: &common.ReadResult{
diff --git a/providers/hubspot/test/read/objects-api/contacts-req-payload-since-until.json b/providers/hubspot/test/read/objects-api/contacts-req-payload-since-until.json
new file mode 100644
index 00000000..de117487
--- /dev/null
+++ b/providers/hubspot/test/read/objects-api/contacts-req-payload-since-until.json
@@ -0,0 +1,28 @@
+{
+  "filterGroups": [
+    {
+      "filters": [
+        {
+          "propertyName": "lastmodifieddate",
+          "operator": "GTE",
+          "value": "2024-09-19T04:30:45-08:00"
+        },
+        {
+          "propertyName": "lastmodifieddate",
+          "operator": "LTE",
+          "value": "2025-01-01T00:00:00-08:00"
+        }
+      ]
+    }
+  ],
+  "limit": "100",
+  "properties": [
+    "email"
+  ],
+  "sorts": [
+    {
+      "propertyName": "hs_object_id",
+      "direction": "ASCENDING"
+    }
+  ]
+}
diff --git a/providers/hubspot/test/read/objects-api/contacts-req-payload-since.json b/providers/hubspot/test/read/objects-api/contacts-req-payload-since.json
new file mode 100644
index 00000000..3941a370
--- /dev/null
+++ b/providers/hubspot/test/read/objects-api/contacts-req-payload-since.json
@@ -0,0 +1,23 @@
+{
+  "filterGroups": [
+    {
+      "filters": [
+        {
+          "propertyName": "lastmodifieddate",
+          "operator": "GTE",
+          "value": "2024-09-19T04:30:45-08:00"
+        }
+      ]
+    }
+  ],
+  "limit": "100",
+  "properties": [
+    "email"
+  ],
+  "sorts": [
+    {
+      "propertyName": "hs_object_id",
+      "direction": "ASCENDING"
+    }
+  ]
+}
diff --git a/providers/hubspot/test/read/objects-api/contacts-req-payload-until.json b/providers/hubspot/test/read/objects-api/contacts-req-payload-until.json
new file mode 100644
index 00000000..e6068836
--- /dev/null
+++ b/providers/hubspot/test/read/objects-api/contacts-req-payload-until.json
@@ -0,0 +1,23 @@
+{
+  "filterGroups": [
+    {
+      "filters": [
+        {
+          "propertyName": "lastmodifieddate",
+          "operator": "LTE",
+          "value": "2025-01-01T00:00:00-08:00"
+        }
+      ]
+    }
+  ],
+  "limit": "100",
+  "properties": [
+    "email"
+  ],
+  "sorts": [
+    {
+      "propertyName": "hs_object_id",
+      "direction": "ASCENDING"
+    }
+  ]
+}
diff --git a/test/hubspot/read/contacts/incremental/main.go b/test/hubspot/read/contacts/incremental/main.go
index 31b023f2..05617c6a 100644
--- a/test/hubspot/read/contacts/incremental/main.go
+++ b/test/hubspot/read/contacts/incremental/main.go
@@ -26,8 +26,9 @@ func main() {
 
 	res, err := conn.Read(ctx, common.ReadParams{
 		ObjectName: "contacts",
-		Fields:     connectors.Fields("email", "phone", "company", "website", "lastname", "firstname"),
-		Since:      time.Now().Add(-4 * time.Hour * 24),
+		Fields:     connectors.Fields("email", "company", "website", "lastname", "firstname", "lastmodifieddate"),
+		Since:      timestamp("2025-03-01T05:43:30.157Z"),
+		Until:      timestamp("2025-03-01T05:43:33.157Z"),
 	})
 	if err != nil {
 		utils.Fail("error reading from Hubspot", "error", err)
@@ -36,3 +37,12 @@ func main() {
 	slog.Info("Reading contacts..")
 	utils.DumpJSON(res, os.Stdout)
 }
+
+func timestamp(timeText string) time.Time {
+	result, err := time.Parse("2006-01-02T15:04:05.000Z", timeText)
+	if err != nil {
+		utils.Fail("bad timestamp", "error", err)
+	}
+
+	return result
+}
diff --git a/test/salesforce/read/account/main.go b/test/salesforce/read/account/main.go
new file mode 100644
index 00000000..01a8e16b
--- /dev/null
+++ b/test/salesforce/read/account/main.go
@@ -0,0 +1,52 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"os/signal"
+	"syscall"
+	"time"
+
+	"github.com/amp-labs/connectors"
+	connTest "github.com/amp-labs/connectors/test/salesforce"
+	"github.com/amp-labs/connectors/test/utils"
+)
+
+const TimeoutSeconds = 30
+
+func main() {
+	// Handle Ctrl-C gracefully.
+	ctx, done := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
+	defer done()
+
+	// Set up slog logging.
+	utils.SetupLogging()
+
+	conn := connTest.GetSalesforceConnector(ctx)
+
+	ctx, done = context.WithTimeout(ctx, TimeoutSeconds*time.Second)
+	defer done()
+
+	res, err := conn.Read(ctx, connectors.ReadParams{
+		ObjectName: "Account",
+		Fields:     connectors.Fields("Id", "Name", "BillingCity", "IsDeleted", "SystemModstamp"),
+		Since:      timestamp("2024-08-28T13:47:37"),
+		Until:      timestamp("2025-01-01T00:00:00"),
+	})
+	if err != nil {
+		utils.Fail("error reading", "error", err)
+	}
+
+	fmt.Println("Reading..")
+	utils.DumpJSON(res, os.Stdout)
+}
+
+func timestamp(timeText string) time.Time {
+	result, err := time.Parse("2006-01-02T15:04:05", timeText)
+	if err != nil {
+		utils.Fail("bad timestamp", "error", err)
+	}
+
+	return result
+}
diff --git a/test/utils/testroutines/comparator.go b/test/utils/testroutines/comparator.go
index 820b065b..7498247d 100644
--- a/test/utils/testroutines/comparator.go
+++ b/test/utils/testroutines/comparator.go
@@ -43,9 +43,13 @@ func ComparatorSubsetRead(serverURL string, actual, expected *common.ReadResult)
 func ComparatorPagination(serverURL string, actual *common.ReadResult, expected *common.ReadResult) bool {
 	expectedNextPage := resolveTestServerURL(expected.NextPage.String(), serverURL)
 
-	return compareNextPageToken(actual.NextPage.String(), expectedNextPage) &&
-		actual.Rows == expected.Rows &&
-		actual.Done == expected.Done
+	a := compareNextPageToken(actual.NextPage.String(), expectedNextPage)
+	b := actual.Rows == expected.Rows
+	c := actual.Done == expected.Done
+
+	return a &&
+		b &&
+		c
 }
 
 func compareNextPageToken(actual, expected string) bool {