fix: Fix hubspot incremental syncs when >10k records have been updated (

#586) ## Test Plan Testing locally ## Deployment instructions [Add any special deployment instructions here]
supaglue-labs · Apr 6, 2023 · d2bd85c · d2bd85c
1 parent 11bb735
commit d2bd85c
Showing 1 changed file with 61 additions and 15 deletions.
diff --git a/packages/core/remotes/crm/hubspot/index.ts b/packages/core/remotes/crm/hubspot/index.ts
@@ -1,7 +1,16 @@
 import { Client } from '@hubspot/api-client';
-import { CollectionResponseSimplePublicObjectWithAssociationsForwardPaging as HubspotPaginatedCompanies } from '@hubspot/api-client/lib/codegen/crm/companies';
-import { CollectionResponseSimplePublicObjectWithAssociationsForwardPaging as HubspotPaginatedContacts } from '@hubspot/api-client/lib/codegen/crm/contacts';
-import { CollectionResponseSimplePublicObjectWithAssociationsForwardPaging as HubspotPaginatedDeals } from '@hubspot/api-client/lib/codegen/crm/deals';
+import {
+  CollectionResponseSimplePublicObjectWithAssociationsForwardPaging as HubspotPaginatedCompanies,
+  CollectionResponseWithTotalSimplePublicObjectForwardPaging as HubspotPaginatedCompaniesWithTotal,
+} from '@hubspot/api-client/lib/codegen/crm/companies';
+import {
+  CollectionResponseSimplePublicObjectWithAssociationsForwardPaging as HubspotPaginatedContacts,
+  CollectionResponseWithTotalSimplePublicObjectForwardPaging as HubspotPaginatedContactsWithTotal,
+} from '@hubspot/api-client/lib/codegen/crm/contacts';
+import {
+  CollectionResponseSimplePublicObjectWithAssociationsForwardPaging as HubspotPaginatedDeals,
+  CollectionResponseWithTotalSimplePublicObjectForwardPaging as HubspotPaginatedDealsWithTotal,
+} from '@hubspot/api-client/lib/codegen/crm/deals';
 import { CollectionResponsePublicOwnerForwardPaging as HubspotPaginatedOwners } from '@hubspot/api-client/lib/codegen/crm/owners';
 import {
   CRMConnectionUnsafe,
@@ -42,6 +51,7 @@ import {
 } from './mappers';
 
 const HUBSPOT_RECORD_LIMIT = 100;
+const HUBSPOT_SEARCH_RESULTS_LIMIT = 10000;
 
 const propertiesToFetch = {
   company: [
@@ -162,10 +172,18 @@ class HubSpotClient extends AbstractCrmRemoteClient {
   }
 
   public async listAccounts(updatedAfter?: Date): Promise<Readable> {
-    const impl = updatedAfter
-      ? this.#listAccountsIncrementalImpl.bind(this, updatedAfter)
+    // TODO(585): Incremental uses the Search endpoint which doesn't allow for more than 10000 results. We need to introduce another layer of pagination.
+    let impl = updatedAfter
+      ? this.#listAccountsIncrementalImpl.bind(this, updatedAfter, HUBSPOT_RECORD_LIMIT)
       : this.#listAccountsFullImpl.bind(this);
 
+    if (updatedAfter) {
+      const response = await this.#listAccountsIncrementalImpl(updatedAfter, 0);
+      if (response.total > HUBSPOT_SEARCH_RESULTS_LIMIT) {
+        impl = this.#listAccountsFullImpl.bind(this);
+      }
+    }
+
     const passThrough = new PassThrough({ objectMode: true });
 
     (async () => {
@@ -228,7 +246,11 @@ class HubSpotClient extends AbstractCrmRemoteClient {
     });
   }
 
-  async #listAccountsIncrementalImpl(updatedAfter: Date, after?: string): Promise<HubspotPaginatedCompanies> {
+  async #listAccountsIncrementalImpl(
+    updatedAfter: Date,
+    limit: number,
+    after?: string
+  ): Promise<HubspotPaginatedCompaniesWithTotal> {
     return await retryWhenRateLimited(async () => {
       await this.maybeRefreshAccessToken();
       const companies = await this.#client.crm.companies.searchApi.doSearch({
@@ -250,7 +272,7 @@ class HubSpotClient extends AbstractCrmRemoteClient {
           } as unknown as string, // hubspot sdk has wrong types https://github.com/HubSpot/hubspot-api-nodejs/issues/350
         ],
         properties: propertiesToFetch.company,
-        limit: HUBSPOT_RECORD_LIMIT,
+        limit,
         after: after as unknown as number, // hubspot sdk has wrong types https://github.com/HubSpot/hubspot-api-nodejs/issues/350
       });
       return companies;
@@ -280,10 +302,18 @@ class HubSpotClient extends AbstractCrmRemoteClient {
   }
 
   public async listOpportunities(updatedAfter?: Date): Promise<Readable> {
-    const impl = updatedAfter
-      ? this.#listOpportunitiesIncrementalImpl.bind(this, updatedAfter)
+    // TODO(585): Incremental uses the Search endpoint which doesn't allow for more than 10000 results. We need to introduce another layer of pagination.
+    let impl = updatedAfter
+      ? this.#listOpportunitiesIncrementalImpl.bind(this, updatedAfter, HUBSPOT_RECORD_LIMIT)
       : this.#listOpportunitiesFullImpl.bind(this);
 
+    if (updatedAfter) {
+      const response = await this.#listOpportunitiesIncrementalImpl(updatedAfter, 0);
+      if (response.total > HUBSPOT_SEARCH_RESULTS_LIMIT) {
+        impl = this.#listOpportunitiesFullImpl.bind(this);
+      }
+    }
+
     const passThrough = new PassThrough({ objectMode: true });
 
     (async () => {
@@ -346,7 +376,11 @@ class HubSpotClient extends AbstractCrmRemoteClient {
     });
   }
 
-  async #listOpportunitiesIncrementalImpl(updatedAfter: Date, after?: string): Promise<HubspotPaginatedDeals> {
+  async #listOpportunitiesIncrementalImpl(
+    updatedAfter: Date,
+    limit: number,
+    after?: string
+  ): Promise<HubspotPaginatedDealsWithTotal> {
     return await retryWhenRateLimited(async () => {
       await this.maybeRefreshAccessToken();
       const response = await this.#client.crm.deals.searchApi.doSearch({
@@ -368,7 +402,7 @@ class HubSpotClient extends AbstractCrmRemoteClient {
           } as unknown as string, // hubspot sdk has wrong types https://github.com/HubSpot/hubspot-api-nodejs/issues/350
         ],
         properties: propertiesToFetch.deal,
-        limit: HUBSPOT_RECORD_LIMIT,
+        limit,
         after: after as unknown as number, // hubspot sdk has wrong types https://github.com/HubSpot/hubspot-api-nodejs/issues/350
       });
 
@@ -435,10 +469,18 @@ class HubSpotClient extends AbstractCrmRemoteClient {
   }
 
   public async listContacts(updatedAfter?: Date): Promise<Readable> {
-    const impl = updatedAfter
-      ? this.#listContactsIncrementalImpl.bind(this, updatedAfter)
+    // TODO(585): Incremental uses the Search endpoint which doesn't allow for more than 10000 results. We need to introduce another layer of pagination.
+    let impl = updatedAfter
+      ? this.#listContactsIncrementalImpl.bind(this, updatedAfter, HUBSPOT_RECORD_LIMIT)
       : this.#listContactsFullImpl.bind(this);
 
+    if (updatedAfter) {
+      const response = await this.#listContactsIncrementalImpl(updatedAfter, 0);
+      if (response.total > HUBSPOT_SEARCH_RESULTS_LIMIT) {
+        impl = this.#listContactsFullImpl.bind(this);
+      }
+    }
+
     const passThrough = new PassThrough({ objectMode: true });
 
     (async () => {
@@ -501,7 +543,11 @@ class HubSpotClient extends AbstractCrmRemoteClient {
     });
   }
 
-  async #listContactsIncrementalImpl(updatedAfter: Date, after?: string): Promise<HubspotPaginatedContacts> {
+  async #listContactsIncrementalImpl(
+    updatedAfter: Date,
+    limit: number,
+    after?: string
+  ): Promise<HubspotPaginatedContactsWithTotal> {
     return await retryWhenRateLimited(async () => {
       await this.maybeRefreshAccessToken();
 
@@ -525,7 +571,7 @@ class HubSpotClient extends AbstractCrmRemoteClient {
           } as unknown as string, // hubspot sdk has wrong types
         ],
         properties: propertiesToFetch.contact,
-        limit: HUBSPOT_RECORD_LIMIT,
+        limit,
         after: after as unknown as number, // hubspot sdk has wrong types
       });