Repository: amp-labs/connectors
Pull request #1448 (2025-03-06 05:48:43)

[CON-1590] `collapsedSubscriptionEvent` interface for Salesforce webhook event message
## This PR does the following
- Defines new interface type `CollapsedSubscriptionEvent` which is a type of webhook message that has multiple record changes defined in 1 message. This type separates records into multiple list of `SubscriptionEvent`s, which is existing interface
- This PR also moves around some Connector interface types from common tothe  connectors package since that is where all other connectors are defined
- Lastly, as `GetMarshalledDataWithId` to common package which is needed for salesforce webhook message parsing. 

## Note
This pr should be merged in order for related PR in server to merge. https://github.com/amp-labs/server/pull/2755


## Test Results

![Screenshot 2025-03-06 at 3 22 57 PM](https://github.com/user-attachments/assets/58ef7386-00c5-4003-bf06-da4c155ca898)



Patch:
diff --git a/common/parse.go b/common/parse.go
index 8a9ba6b9..6b46248f 100644
--- a/common/parse.go
+++ b/common/parse.go
@@ -1,6 +1,8 @@
 package common
 
 import (
+	"errors"
+	"fmt"
 	"strings"
 
 	"github.com/amp-labs/connectors/internal/datautils"
@@ -107,6 +109,40 @@ func GetMarshaledData(records []map[string]any, fields []string) ([]ReadResultRo
 	return data, nil
 }
 
+var (
+	errMissingId        = errors.New("missing id field in raw record")
+	errUnexpectedIdType = errors.New("unexpected id type")
+)
+
+// GetMarshalledDataWithId is very similar to GetMarshaledData, but it also extracts the "id" field from the raw record.
+func GetMarshalledDataWithId(records []map[string]any, fields []string) ([]ReadResultRow, error) {
+	data := make([]ReadResultRow, len(records))
+
+	fields = append(fields, "id")
+
+	//nolint:varnamelen
+	for i, record := range records {
+		data[i] = ReadResultRow{
+			Fields: ExtractLowercaseFieldsFromRaw(fields, record),
+			Raw:    record,
+		}
+
+		idAny := data[i].Fields["id"]
+		if idAny == nil {
+			return nil, errMissingId
+		}
+
+		id, ok := idAny.(string)
+		if !ok {
+			return nil, fmt.Errorf("%w: %T", errUnexpectedIdType, idAny)
+		}
+
+		data[i].Id = id
+	}
+
+	return data, nil
+}
+
 func GetRecordsUnderJSONPath(jsonPath string, nestedPath ...string) RecordsFunc {
 	return getRecords(false, jsonPath, nestedPath...)
 }
diff --git a/common/types.go b/common/types.go
index 48f52254..72f812cd 100644
--- a/common/types.go
+++ b/common/types.go
@@ -396,6 +396,13 @@ type SubscriptionEvent interface {
 	EventTimeStampNano() (int64, error)
 }
 
+// Some providers send multiple events in a single webhook payload.
+// This interface is used to extract individual events to SubscriptionEvent type
+// from a collapsed event for webhook parsing and processing.
+type CollapsedSubscriptionEvent interface {
+	SubscriptionEventList() ([]SubscriptionEvent, error)
+}
+
 // WebhookVerificationParameters is a struct that contains the parameters required to verify a webhook.
 type WebhookVerificationParameters struct {
 	Headers      http.Header
@@ -486,55 +493,3 @@ const (
 	// registration returned error, and failed to rollback some intermittent steps.
 	SubscriptionStatusFailedToRollback SubscriptionStatus = "failed_to_rollback"
 )
-
-// SubscribeConnector has 2 main responsibilities:
-// 1. Register a subscription with the provider.
-// Registering a subscription is a one-time operation that is required
-// by providers that hold some master registration of all subscriptions.
-// Not all providers require this, but some do.
-// 2. Subscribe to events from the provider.
-// This is the actual subscription to events from the provider.
-// It will subscribe for events and objects as specified in SubscribeParams.
-type SubscribeConnector interface {
-	Register(
-		ctx context.Context,
-		params SubscriptionRegistrationParams,
-	) (*RegistrationResult, error)
-	// TODO: Uncomment when we implement UpdateRegistration in Salesforce
-	// UpdateRegistration(
-	// 	ctx context.Context,
-	// 	params SubscriptionRegistrationParams,
-	// 	previousResult RegistrationResult,
-	// ) (*RegistrationResult, error)
-	DeleteRegistration(
-		ctx context.Context,
-		previousResult RegistrationResult,
-	) error
-	// EmptyRegistrationParams returns a empty instance of SubscriptionRegistrationParams.
-	// if there is any provider specific initialization required, it should be done here.
-	EmptyRegistrationParams() *SubscriptionRegistrationParams
-	// EmptyRegistrationResult returns a empty instance of RegistrationResult.
-	// if there is any provider specific initialization required, it should be done here.
-	EmptyRegistrationResult() *RegistrationResult
-
-	Subscribe(
-		ctx context.Context,
-		params SubscribeParams,
-	) (*SubscriptionResult, error)
-	// TODO: Uncomment when we implement UpdateSubscription in Salesforce
-	// UpdateSubscription(
-	// 	ctx context.Context,
-	// 	params SubscribeParams,
-	// 	previousResult SubscriptionResult,
-	// ) (*SubscriptionResult, error)
-	DeleteSubscription(
-		ctx context.Context,
-		previousResult SubscriptionResult,
-	) error
-	// EmptySubscritpionParams returns a empty instance of SubscribeParams.
-	// if there is any provider specific initialization required, it should be done here.
-	EmptySubscritpionParams() *SubscribeParams
-	// EmptySubscriptionResult returns a empty instance of SubscriptionResult.
-	// if there is any provider specific initialization required, it should be done here.
-	EmptySubscriptionResult() *SubscriptionResult
-}
diff --git a/connectors.go b/connectors.go
index 211ae851..a7cbda19 100644
--- a/connectors.go
+++ b/connectors.go
@@ -86,13 +86,80 @@ type AuthMetadataConnector interface {
 	GetPostAuthInfo(ctx context.Context) (*common.PostAuthInfo, error)
 }
 
+type BatchRecordReaderConnector interface {
+	Connector
+	GetRecordsWithIds(
+		ctx context.Context,
+		objectName string,
+		//nolint:revive
+		recordIds []string,
+		fields []string,
+		associations []string) ([]common.ReadResultRow, error)
+}
 type WebhookVerifierConnector interface {
 	Connector
+	BatchRecordReaderConnector
 
 	// VerifyWebhookMessage verifies the signature of a webhook message.
 	VerifyWebhookMessage(ctx context.Context, params *common.WebhookVerificationParameters) (bool, error)
 }
 
+//nolint:interfacebloat
+type SubscribeConnector interface {
+	WebhookVerifierConnector
+	// SubscribeConnector has 2 main responsibilities:
+	// 1. Register a subscription with the provider.
+	// Registering a subscription is a one-time operation that is required
+	// by providers that hold some master registration of all subscriptions.
+	// Not all providers require this, but some do.
+	// 2. Subscribe to events from the provider.
+	// This is the actual subscription to events from the provider.
+	// It will subscribe for events and objects as specified in SubscribeParams.
+	Register(
+		ctx context.Context,
+		params common.SubscriptionRegistrationParams,
+	) (*common.RegistrationResult, error)
+	// TODO: Uncomment when we implement UpdateRegistration in Salesforce
+	// UpdateRegistration(
+	// 	ctx context.Context,
+	// 	params SubscriptionRegistrationParams,
+	// 	previousResult RegistrationResult,
+	// ) (*RegistrationResult, error)
+	DeleteRegistration(
+		ctx context.Context,
+		previousResult common.RegistrationResult,
+	) error
+	// EmptyRegistrationParams returns a empty instance of SubscriptionRegistrationParams.
+	// if there is any provider specific initialization required, it should be done here.
+	EmptyRegistrationParams() *common.SubscriptionRegistrationParams
+	// EmptyRegistrationResult returns a empty instance of RegistrationResult.
+	// if there is any provider specific initialization required, it should be done here.
+	EmptyRegistrationResult() *common.RegistrationResult
+
+	Subscribe(
+		ctx context.Context,
+		params common.SubscribeParams,
+	) (*common.SubscriptionResult, error)
+	// TODO: Uncomment when we implement UpdateSubscription in Salesforce
+	// UpdateSubscription(
+	// 	ctx context.Context,
+	// 	params SubscribeParams,
+	// 	previousResult SubscriptionResult,
+	// ) (*SubscriptionResult, error)
+	DeleteSubscription(
+		ctx context.Context,
+		previousResult common.SubscriptionResult,
+	) error
+	// EmptySubscritpionParams returns a empty instance of SubscribeParams.
+	// if there is any provider specific initialization required, it should be done here.
+	EmptySubscritpionParams() *common.SubscribeParams
+	// EmptySubscriptionResult returns a empty instance of SubscriptionResult.
+	// if there is any provider specific initialization required, it should be done here.
+	EmptySubscriptionResult() *common.SubscriptionResult
+	// GetRecordsWithId is a helper function to get records by their IDs.
+	//nolint:revive
+}
+
 // We re-export the following types so that they can be used by consumers of this library.
 type (
 	ReadParams               = common.ReadParams
diff --git a/providers/hubspot/subscriptionEvent.go b/providers/hubspot/subscriptionEvent.go
index e675963d..888e2034 100644
--- a/providers/hubspot/subscriptionEvent.go
+++ b/providers/hubspot/subscriptionEvent.go
@@ -31,6 +31,8 @@ This form also prevents null fields to be sent out as zero values.
 */
 type SubscriptionEvent map[string]any
 
+var _ common.SubscriptionEvent = SubscriptionEvent{}
+
 // VerifyWebhookMessage verifies the signature of a webhook message from Hubspot.
 func (*Connector) VerifyWebhookMessage(
 	_ context.Context, params *common.WebhookVerificationParameters,
diff --git a/providers/salesforce/record.go b/providers/salesforce/record.go
index 389e91eb..5fead605 100644
--- a/providers/salesforce/record.go
+++ b/providers/salesforce/record.go
@@ -44,7 +44,7 @@ func (c *Connector) GetRecordsWithIds( // nolint:revive
 		rsp,
 		getRecords,
 		getNextRecordsURL,
-		common.GetMarshaledData,
+		common.GetMarshalledDataWithId,
 		config.Fields,
 	)
 	if err != nil {
diff --git a/providers/salesforce/subscriptionEvent.go b/providers/salesforce/subscriptionEvent.go
index a57b72b5..568ce759 100644
--- a/providers/salesforce/subscriptionEvent.go
+++ b/providers/salesforce/subscriptionEvent.go
@@ -24,16 +24,18 @@ func (*Connector) VerifyWebhookMessage(context.Context, *common.WebhookVerificat
 	return true, nil
 }
 
+var _ common.CollapsedSubscriptionEvent = CollapsedSubscriptionEvent{}
+
 // ChangeEvent represents data received from a subscription.
 // A single event may contain multiple record identifiers and can be expanded into multiple SubscriptionEvent instances.
 //
 // Structure reference:
 // https://developer.salesforce.com/docs/atlas.en-us.change_data_capture.meta/change_data_capture/cdc_event_fields_header.htm.
-type ChangeEvent map[string]any
+type CollapsedSubscriptionEvent map[string]any
 
 // ToRecordList splits bundled event into per record event.
 // Every property is duplicated across SubscriptionEvent. RecordIds is spread as RecordId.
-func (e ChangeEvent) ToRecordList() ([]SubscriptionEvent, error) {
+func (e CollapsedSubscriptionEvent) SubscriptionEventList() ([]common.SubscriptionEvent, error) {
 	eventHeaderMap, err := extractChangeEventHeader(e)
 	if err != nil {
 		return nil, err
@@ -49,7 +51,7 @@ func (e ChangeEvent) ToRecordList() ([]SubscriptionEvent, error) {
 		return nil, errRecordIDsType
 	}
 
-	events := make([]SubscriptionEvent, len(recordIDs))
+	events := make([]common.SubscriptionEvent, len(recordIDs))
 
 	for index, recordID := range recordIDs {
 		event, err := goutils.Clone[map[string]any](e)
@@ -57,6 +59,10 @@ func (e ChangeEvent) ToRecordList() ([]SubscriptionEvent, error) {
 			return nil, err
 		}
 
+		subEvent := SubscriptionEvent(event)
+
+		evt := common.SubscriptionEvent(subEvent)
+
 		// Reach out to the nested object and remove record identifiers and attach record id.
 		changeEventHeader, ok := event[keyEventChangeEventHeader].(map[string]any)
 		if !ok {
@@ -68,7 +74,7 @@ func (e ChangeEvent) ToRecordList() ([]SubscriptionEvent, error) {
 
 		// Save changes.
 		event[keyEventChangeEventHeader] = changeEventHeader
-		events[index] = event
+		events[index] = evt
 	}
 
 	return events, nil
@@ -95,14 +101,12 @@ func (s SubscriptionEvent) EventType() (common.SubscriptionEventType, error) {
 	}
 
 	switch changeType {
-	case "CREATE", "GAP_CREATE":
+	case "CREATE":
 		return common.SubscriptionEventTypeCreate, nil
-	case "UPDATE", "GAP_UPDATE":
+	case "UPDATE":
 		return common.SubscriptionEventTypeUpdate, nil
-	case "DELETE", "GAP_DELETE":
+	case "DELETE":
 		return common.SubscriptionEventTypeDelete, nil
-	case "UNDELETE", "SNAPSHOT", "GAP_UNDELETE", "GAP_OVERFLOW":
-		fallthrough
 	default:
 		return common.SubscriptionEventTypeOther, nil
 	}


Test patch:
diff --git a/providers/salesforce/connector_test.go b/providers/salesforce/connector_test.go
index 8d73ad66..06d32865 100644
--- a/providers/salesforce/connector_test.go
+++ b/providers/salesforce/connector_test.go
@@ -3,7 +3,7 @@ package salesforce
 import (
 	"testing"
 
-	"github.com/amp-labs/connectors/common"
+	"github.com/amp-labs/connectors"
 )
 
 func TestSubscribeConnector(t *testing.T) { //nolint:funlen,gocognit,cyclop,maintidx
@@ -14,7 +14,7 @@ func TestSubscribeConnector(t *testing.T) { //nolint:funlen,gocognit,cyclop,main
 		t.Fatalf("failed to construct test connector: %v", err)
 	}
 
-	_, ok := any(conn).(common.SubscribeConnector)
+	_, ok := any(conn).(connectors.SubscribeConnector)
 	if !ok {
 		t.Fatalf("expected SubscribeConnector, got %T", conn)
 	}
diff --git a/providers/salesforce/subscriptionEvent_test.go b/providers/salesforce/subscriptionEvent_test.go
index 10bd2d65..82fbf416 100644
--- a/providers/salesforce/subscriptionEvent_test.go
+++ b/providers/salesforce/subscriptionEvent_test.go
@@ -14,12 +14,12 @@ func TestSubscriptionEventProperties(t *testing.T) {
 
 	eventNewAccountData := testutils.DataFromFile(t, "subscription/new_account.json")
 
-	changeEvent := ChangeEvent{}
+	changeEvent := CollapsedSubscriptionEvent{}
 	if err := json.Unmarshal(eventNewAccountData, &changeEvent); err != nil {
 		t.Fatalf("failed to start a test, cannot parse data; error (%v)", err)
 	}
 
-	events, err := changeEvent.ToRecordList()
+	events, err := changeEvent.SubscriptionEventList()
 	assert.NilError(t, err, "error should be nil")
 
 	if len(events) != 1 {