Incremental sync for Attio provider

packages/api/scripts/init.sql

@@ -2900,6 +2900,20 @@ COMMENT ON COLUMN connections.token_type IS 'The type of the token, such as "Bea
 COMMENT ON COLUMN connections.connection_token IS 'Connection token users will put in their header to identify which service / linked_User they make request for';
 
 
+-- ************************************** vertical_objects_sync_track_data
+CREATE TABLE vertical_objects_sync_track_data
+(
+ id_vertical_objects_sync_track_data uuid NOT NULL,
+ vertical                            text NOT NULL,
+ provider_slug                       text NOT NULL,
+ object                              text NOT NULL,
+ pagination_type                     text NOT NULL,
+ id_connection                       uuid NOT NULL,
+ data                                json,
+ CONSTRAINT PK_vertical_objects_sync_track_data PRIMARY KEY ( id_vertical_objects_sync_track_data ),
+ CONSTRAINT FK_connection FOREIGN KEY ( id_connection ) REFERENCES connections ( id_connection )
+);
+
 -- ************************************** ats_scorecards
 CREATE TABLE ats_scorecards
 (

packages/api/src/crm/company/services/attio/index.ts

@@ -9,9 +9,10 @@ import { EncryptionService } from '@@core/@core-services/encryption/encryption.s
 import { ApiResponse } from '@@core/utils/types';
 import { ICompanyService } from '@crm/company/types';
 import { ServiceRegistry } from '../registry.service';
-import { AttioCompanyInput, AttioCompanyOutput } from './types';
+import { AttioCompanyInput, AttioCompanyOutput, paginationType } from './types';
 import { SyncParam } from '@@core/utils/types/interface';
 import { OriginalCompanyOutput } from '@@core/utils/types/original/original.crm';
+import { v4 as uuidv4 } from 'uuid';
 
 @Injectable()
 export class AttioService implements ICompanyService {
@@ -74,20 +75,134 @@ export class AttioService implements ICompanyService {
           vertical: 'crm',
         },
       });
-      const resp = await axios.post(
-        `${connection.account_url}/v2/objects/companies/records/query`,
-        {},
-        {
-          headers: {
-            'Content-Type': 'application/json',
-            Authorization: `Bearer ${this.cryptoService.decrypt(
-              connection.access_token,
-            )}`,
-          },
+
+      const paginationTrackInfo = await this.prisma.vertical_objects_sync_track_data.findFirst({
+        where: {
+          id_connection: connection.id_connection,
+          vertical: 'crm',
+          provider_slug: 'attio',
+          object: 'company',
         },
-      );
+      });
+
+      let respData: AttioCompanyOutput[] = [];
+      let initialOffset: number = 0;
+
+      if (!paginationTrackInfo) {
+        // Intial sync
+        try {
+          while (true) {
+            const resp = await axios.post(
+              `${connection.account_url}/v2/objects/companies/records/query`,
+              {
+                "sorts": [
+                  {
+                    "attribute": "created_at",
+                    "direction": "asc"
+                  }
+                ],
+                "offset": initialOffset,
+                "limit": 500
+              },
+              {
+                headers: {
+                  accept: 'application/json',
+                  'Content-Type': 'application/json',
+                  Authorization: `Bearer ${this.cryptoService.decrypt(
+                    connection.access_token,
+                  )}`,
+                },
+              },
+            );
+
+
+            respData.push(...resp.data.data);
+            initialOffset = initialOffset + resp.data.data.length;
+
+            if (resp.data.data.length < 500) {
+              break;
+            }
+          }
+        }
+        catch (error) {
+          this.logger.log(`Error in initial sync ${error.message} and last offset is ${initialOffset}`);
+        }
+
+      }
+      else {
+        // Incremental sync
+        const currentPaginationData = paginationTrackInfo.data as paginationType;
+        initialOffset = currentPaginationData.offset;
+
+        try {
+          while (true) {
+            const resp = await axios.post(
+              `${connection.account_url}/v2/objects/companies/records/query`,
+              {
+                "sorts": [
+                  {
+                    "attribute": "created_at",
+                    "direction": "asc"
+                  }
+                ],
+                "offset": initialOffset,
+                "limit": 500
+              },
+              {
+                headers: {
+                  accept: 'application/json',
+                  'Content-Type': 'application/json',
+                  Authorization: `Bearer ${this.cryptoService.decrypt(
+                    connection.access_token,
+                  )}`
+                }
+              }
+            );
+
+            respData.push(...resp.data.data);
+            initialOffset = initialOffset + resp.data.data.length;
+
+            if (resp.data.data.length < 500) {
+              break;
+            }
+          }
+        }
+        catch (error) {
+          this.logger.log(`Error in incremental sync ${error.message} and last offset is ${initialOffset}`);
+        }
+      }
+
+      // create or update records
+      if (paginationTrackInfo) {
+        await this.prisma.vertical_objects_sync_track_data.update({
+          where: {
+            id_vertical_objects_sync_track_data: paginationTrackInfo.id_vertical_objects_sync_track_data,
+          },
+          data: {
+            data: {
+              offset: initialOffset,
+            },
+          },
+        });
+      }
+      else {
+        await this.prisma.vertical_objects_sync_track_data.create({
+          data: {
+            id_vertical_objects_sync_track_data: uuidv4(),
+            vertical: 'crm',
+            provider_slug: 'attio',
+            object: 'company',
+            pagination_type: 'offset',
+            id_connection: connection.id_connection,
+            data: {
+              offset: initialOffset,
+            },
+          },
+        });
+      }
+
       return {
-        data: resp.data.data,
+        data: respData,
         message: 'Attio companies retrieved',
         statusCode: 200,
       };

packages/api/src/crm/company/services/attio/types.ts

@@ -127,3 +127,7 @@ export interface AttioCompany {
 
 export type AttioCompanyInput = Partial<AttioCompany>;
 export type AttioCompanyOutput = AttioCompanyInput;
+
+export type paginationType = {
+  offset: number;
+};

packages/api/src/crm/contact/services/attio/index.ts

@@ -8,7 +8,8 @@ import { ActionType, handle3rdPartyServiceError } from '@@core/utils/errors';
 import { EncryptionService } from '@@core/@core-services/encryption/encryption.service';
 import { ApiResponse } from '@@core/utils/types';
 import { ServiceRegistry } from '../registry.service';
-import { AttioContactInput, AttioContactOutput } from './types';
+import { v4 as uuidv4 } from 'uuid';
+import { AttioContactInput, AttioContactOutput, paginationType } from './types';
 import { SyncParam } from '@@core/utils/types/interface';
 
 @Injectable()
@@ -74,22 +75,133 @@ export class AttioService implements IContactService {
         },
       });
 
-      const resp = await axios.post(
-        `${connection.account_url}/v2/objects/people/records/query`,
-        {},
-        {
-          headers: {
-            accept: 'application/json',
-            'Content-Type': 'application/json',
-            Authorization: `Bearer ${this.cryptoService.decrypt(
-              connection.access_token,
-            )}`,
-          },
+      const paginationTrackInfo = await this.prisma.vertical_objects_sync_track_data.findFirst({
+        where: {
+          id_connection: connection.id_connection,
+          vertical: 'crm',
+          provider_slug: 'attio',
+          object: 'contact',
         },
-      );
+      });
+
+      let respData: AttioContactOutput[] = [];
+      let initialOffset: number = 0;
+
+      if (!paginationTrackInfo) {
+        // Intial sync
+        try {
+          while (true) {
+            const resp = await axios.post(
+              `${connection.account_url}/v2/objects/people/records/query`,
+              {
+                "sorts": [
+                  {
+                    "attribute": "created_at",
+                    "direction": "asc"
+                  }
+                ],
+                "offset": initialOffset,
+                "limit": 500
+              },
+              {
+                headers: {
+                  accept: 'application/json',
+                  'Content-Type': 'application/json',
+                  Authorization: `Bearer ${this.cryptoService.decrypt(
+                    connection.access_token,
+                  )}`,
+                },
+              },
+            );
+
+
+            respData.push(...resp.data.data);
+            initialOffset = initialOffset + resp.data.data.length;
+
+            if (resp.data.data.length < 500) {
+              break;
+            }
+          }
+        }
+        catch (error) {
+          this.logger.log(`Error in initial sync ${error.message} and last offset is ${initialOffset}`);
+        }
+
+      }
+      else {
+        // Incremental sync
+        const currentPaginationData = paginationTrackInfo.data as paginationType;
+        initialOffset = currentPaginationData.offset;
+
+        try {
+          while (true) {
+            const resp = await axios.post(
+              `${connection.account_url}/v2/objects/people/records/query`,
+              {
+                "sorts": [
+                  {
+                    "attribute": "created_at",
+                    "direction": "asc"
+                  }
+                ],
+                "offset": initialOffset,
+                "limit": 500
+              },
+              {
+                headers: {
+                  accept: 'application/json',
+                  'Content-Type': 'application/json',
+                  Authorization: `Bearer ${this.cryptoService.decrypt(
+                    connection.access_token,
+                  )}`
+                }
+              }
+            );
+
+            respData.push(...resp.data.data);
+            initialOffset = initialOffset + resp.data.data.length;
+
+            if (resp.data.data.length < 500) {
+              break;
+            }
+          }
+        }
+        catch (error) {
+          this.logger.log(`Error in incremental sync ${error.message} and last offset is ${initialOffset}`);
+        }
+      }
+
+      // create or update records
+      if (paginationTrackInfo) {
+        await this.prisma.vertical_objects_sync_track_data.update({
+          where: {
+            id_vertical_objects_sync_track_data: paginationTrackInfo.id_vertical_objects_sync_track_data,
+          },
+          data: {
+            data: {
+              offset: initialOffset,
+            },
+          },
+        });
+      }
+      else {
+        await this.prisma.vertical_objects_sync_track_data.create({
+          data: {
+            id_vertical_objects_sync_track_data: uuidv4(),
+            vertical: 'crm',
+            provider_slug: 'attio',
+            object: 'contact',
+            pagination_type: 'offset',
+            id_connection: connection.id_connection,
+            data: {
+              offset: initialOffset,
+            },
+          },
+        });
+      }
 
       return {
-        data: resp.data.data,
+        data: respData,
         message: 'Attio contacts retrieved',
         statusCode: 200,
       };

packages/api/src/crm/contact/services/attio/types.ts

@@ -126,3 +126,8 @@ export interface AttioContact {
 
 export type AttioContactInput = Partial<AttioContact>;
 export type AttioContactOutput = AttioContactInput;
+
+
+export type paginationType = {
+  offset: number;
+};