Implement deletion handling in CDC iterator

hubspot/list.go

@@ -35,18 +35,36 @@ const UpdatedAtListSortKey = "updatedAt"
 type TimestampResource struct {
 	CreatedAtFieldName string
 	UpdatedAtFieldName string
+	DeletedAtFieldName string
 }
 
 // TimestampResources holds a list of resources that support timestamp-based filtering.
 var TimestampResources = map[string]TimestampResource{
-	"cms.blogs.authors": {CreatedAtFieldName: "created", UpdatedAtFieldName: "updated"},
-	"cms.blogs.posts":   {CreatedAtFieldName: "created", UpdatedAtFieldName: "updated"},
-	"cms.blogs.tags":    {CreatedAtFieldName: "created", UpdatedAtFieldName: "updated"},
-	"cms.pages.landing": {CreatedAtFieldName: "createdAt", UpdatedAtFieldName: "updatedAt"},
-	"cms.pages.site":    {CreatedAtFieldName: "createdAt", UpdatedAtFieldName: "updatedAt"},
-	"cms.hubdb.tables":  {CreatedAtFieldName: "createdAt", UpdatedAtFieldName: "updatedAt"},
-	"cms.domains":       {CreatedAtFieldName: "createdAt", UpdatedAtFieldName: "updatedAt"},
-	"cms.urlRedirects":  {CreatedAtFieldName: "createdAt", UpdatedAtFieldName: "updatedAt"},
+	"cms.blogs.authors": {
+		CreatedAtFieldName: "created",
+		UpdatedAtFieldName: "updated",
+		DeletedAtFieldName: "deletedAt",
+	},
+	"cms.blogs.posts": {
+		CreatedAtFieldName: "created",
+		UpdatedAtFieldName: "updated",
+		DeletedAtFieldName: "deletedAt",
+	},
+	"cms.blogs.tags": {
+		CreatedAtFieldName: "created",
+		UpdatedAtFieldName: "updated",
+		DeletedAtFieldName: "deletedAt",
+	},
+	"cms.pages.landing": {
+		CreatedAtFieldName: "createdAt",
+		UpdatedAtFieldName: "updatedAt",
+		DeletedAtFieldName: "deletedAt",
+	},
+	"cms.pages.site": {
+		CreatedAtFieldName: "createdAt",
+		UpdatedAtFieldName: "updatedAt",
+		DeletedAtFieldName: "deletedAt",
+	},
 }
 
 // ResourcesListPaths holds a mapping of supported resources and their list endpoints.

source/iterator/cdc.go

@@ -19,8 +19,9 @@ import (
 	"fmt"
 	"time"
 
-	"github.com/conduitio-labs/conduit-connector-hubspot/hubspot"
 	sdk "github.com/conduitio/conduit-connector-sdk"
+
+	"github.com/conduitio-labs/conduit-connector-hubspot/hubspot"
 )
 
 // CDC is an implementation of a CDC iterator for the HubSpot API.
@@ -58,7 +59,7 @@ func NewCDC(ctx context.Context, params CDCParams) (*CDC, error) {
 	}
 
 	if cdc.position == nil || cdc.position.Timestamp == nil {
-		now := time.Now()
+		now := time.Now().UTC()
 		cdc.position = &Position{
 			Mode:      CDCPositionMode,
 			Timestamp: &now,
@@ -149,6 +150,7 @@ func (c *CDC) fetchTimestampBasedItems(
 		Limit:        c.bufferSize,
 		UpdatedAfter: &updatedAfter,
 		Sort:         hubspot.UpdatedAtListSortKey,
+		Archived:     true,
 	}
 
 	listResp, err := c.hubspotClient.List(ctx, c.resource, listOpts)
@@ -157,7 +159,11 @@ func (c *CDC) fetchTimestampBasedItems(
 	}
 
 	for _, item := range listResp.Results {
-		err = c.routeItem(item, resource.CreatedAtFieldName, resource.UpdatedAtFieldName, updatedAfter)
+		err = c.routeItem(item,
+			resource.CreatedAtFieldName,
+			resource.UpdatedAtFieldName,
+			resource.DeletedAtFieldName,
+			updatedAfter)
 		if err != nil {
 			return fmt.Errorf("route timestamp based item: %w", err)
 		}
@@ -178,7 +184,7 @@ func (c *CDC) fetchSearchBasedItems(
 	}
 
 	for _, item := range listResp.Results {
-		err = c.routeItem(item, resource.CreatedAtFieldName, resource.UpdatedAtFieldName, updatedAfter)
+		err = c.routeItem(item, resource.CreatedAtFieldName, resource.UpdatedAtFieldName, "", updatedAfter)
 		if err != nil {
 			return fmt.Errorf("route search based item: %w", err)
 		}
@@ -192,7 +198,8 @@ func (c *CDC) fetchSearchBasedItems(
 func (c *CDC) routeItem(
 	item hubspot.ListResponseResult,
 	createdAtFieldName,
-	updatedAtFieldName string,
+	updatedAtFieldName,
+	deletedAtFieldName string,
 	updatedAfter time.Time,
 ) error {
 	itemCreatedAt, err := item.GetTimeField(createdAtFieldName)
@@ -220,23 +227,54 @@ func (c *CDC) routeItem(
 		return fmt.Errorf("marshal sdk position: %w", err)
 	}
 
+	var itemDeletedAt time.Time
+	if deletedAtFieldName != "" {
+		itemDeletedAt, err = item.GetTimeField(deletedAtFieldName)
+		if err != nil {
+			return fmt.Errorf("get item's update date: %w", err)
+		}
+	}
+
+	c.records <- c.GetRecord(item,
+		itemCreatedAt,
+		itemUpdatedAt,
+		itemDeletedAt,
+		updatedAfter,
+		sdkPosition,
+		metadata)
+
+	return nil
+}
+
+func (c *CDC) GetRecord(item hubspot.ListResponseResult,
+	itemCreatedAt,
+	itemUpdatedAt,
+	itemDeletedAt,
+	updatedAfter time.Time,
+	sdkPosition sdk.Position,
+	metadata sdk.Metadata,
+) sdk.Record {
+	if itemDeletedAt != (time.Time{}) {
+		if itemDeletedAt == itemUpdatedAt && itemDeletedAt.Unix() > 0 {
+			return sdk.Util.Source.NewRecordDelete(sdkPosition, metadata,
+				sdk.StructuredData{hubspot.ResultsFieldID: item[hubspot.ResultsFieldID]},
+			)
+		}
+	}
+
 	// if the item's createdAt is after the timestamp after which we're searching items
 	// we consider the item's operation to be sdk.OperationCreate.
 	if itemCreatedAt.After(updatedAfter) {
-		c.records <- sdk.Util.Source.NewRecordCreate(sdkPosition, metadata,
+		return sdk.Util.Source.NewRecordCreate(sdkPosition, metadata,
 			sdk.StructuredData{hubspot.ResultsFieldID: item[hubspot.ResultsFieldID]},
 			sdk.StructuredData(item),
 		)
-
-		return nil
 	}
 
-	c.records <- sdk.Util.Source.NewRecordUpdate(sdkPosition, metadata,
+	return sdk.Util.Source.NewRecordUpdate(sdkPosition, metadata,
 		sdk.StructuredData{hubspot.ResultsFieldID: item[hubspot.ResultsFieldID]},
 		nil, sdk.StructuredData(item),
 	)
-
-	return nil
 }
 
 // Stop stops the iterator.