Repository: amp-labs/connectors
Pull request #1437 (2025-03-04 22:36:36)

[CON-1590]feat(Salesforce): Subscription Event

Cobalt0s: * **#1437** <a href="https://app.graphite.dev/github/pr/amp-labs/connectors/1437?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/1437?utm_source=stack-comment-view-in-graphite" target="_blank">(View in Graphite)</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/common/stringMap.go b/common/stringMap.go
index d39290f2..b723bf13 100644
--- a/common/stringMap.go
+++ b/common/stringMap.go
@@ -146,6 +146,25 @@ func (m StringMap) GetFloat(key string) (float64, error) {
 	}
 }
 
+func (m StringMap) GetNumber(key string) (float64, error) {
+	val, err := m.Get(key)
+	if err != nil {
+		return 0, err
+	}
+
+	t := reflect.TypeOf(val)
+
+	//nolint:exhaustive
+	switch t.Kind() {
+	case reflect.Int, reflect.Int8, reflect.Int16, reflect.Int32, reflect.Int64:
+		return float64(reflect.ValueOf(val).Int()), nil
+	case reflect.Float32, reflect.Float64:
+		return reflect.ValueOf(val).Float(), nil
+	default:
+		return 0, fmt.Errorf("%w: expected a number, but received %T", errFieldTypeMismatch, val)
+	}
+}
+
 //nolint:ireturn
 func assertType[T any](val any) (T, error) {
 	of, ok := val.(T)
diff --git a/internal/goutils/cloning.go b/internal/goutils/cloning.go
index 560409d1..53e680fc 100644
--- a/internal/goutils/cloning.go
+++ b/internal/goutils/cloning.go
@@ -5,6 +5,11 @@ import (
 	"encoding/gob"
 )
 
+func init() {
+	gob.Register(map[string]any{})
+	gob.Register([]any{})
+}
+
 // Clone uses gob to deep copy objects.
 func Clone[T any](input T) (T, error) { // nolint:ireturn
 	var buf bytes.Buffer
diff --git a/providers/brevo/connector.go b/providers/brevo/connector.go
index 14d606e3..3049518a 100644
--- a/providers/brevo/connector.go
+++ b/providers/brevo/connector.go
@@ -5,6 +5,8 @@ import (
 
 	"github.com/amp-labs/connectors/common"
 	"github.com/amp-labs/connectors/internal/components"
+	"github.com/amp-labs/connectors/internal/components/operations"
+	"github.com/amp-labs/connectors/internal/components/reader"
 	"github.com/amp-labs/connectors/internal/components/schema"
 	"github.com/amp-labs/connectors/providers"
 	"github.com/amp-labs/connectors/providers/brevo/metadata"
@@ -19,6 +21,7 @@ type Connector struct {
 
 	// supported operations
 	components.SchemaProvider
+	components.Reader
 }
 
 func NewConnector(params common.Parameters) (*Connector, error) {
@@ -30,6 +33,25 @@ func NewConnector(params common.Parameters) (*Connector, error) {
 func constructor(base *components.Connector) (*Connector, error) {
 	connector := &Connector{Connector: base}
 
+	registry, err := components.NewEndpointRegistry(supportedOperations())
+	if err != nil {
+		return nil, err
+	}
+
+	// Set the metadata provider for the connector
+	connector.SchemaProvider = schema.NewOpenAPISchemaProvider(connector.ProviderContext.Module(), metadata.Schemas)
+
+	connector.Reader = reader.NewHTTPReader(
+		connector.HTTPClient().Client,
+		registry,
+		connector.ProviderContext.Module(),
+		operations.ReadHandlers{
+			BuildRequest:  connector.buildReadRequest,
+			ParseResponse: connector.parseReadResponse,
+			ErrorHandler:  common.InterpretError,
+		},
+	)
+
 	// Set the metadata provider for the connector
 	connector.SchemaProvider = schema.NewOpenAPISchemaProvider(connector.ProviderContext.Module(), metadata.Schemas)
 
diff --git a/providers/brevo/handlers.go b/providers/brevo/handlers.go
new file mode 100644
index 00000000..7f0c70d9
--- /dev/null
+++ b/providers/brevo/handlers.go
@@ -0,0 +1,60 @@
+package brevo
+
+import (
+	"context"
+	"net/http"
+	"strconv"
+
+	"github.com/amp-labs/connectors/common"
+	"github.com/amp-labs/connectors/common/urlbuilder"
+	"github.com/amp-labs/connectors/providers/brevo/metadata"
+)
+
+var apiVersion = "v3" //nolint:gochecknoglobals
+
+func (c *Connector) buildReadRequest(ctx context.Context, params common.ReadParams) (*http.Request, error) {
+	var (
+		url *urlbuilder.URL
+		err error
+	)
+
+	if params.NextPage != "" {
+		url, err = urlbuilder.New(params.NextPage.String())
+		if err != nil {
+			return nil, err
+		}
+
+		return http.NewRequestWithContext(ctx, http.MethodGet, url.String(), nil)
+	}
+
+	url, err = urlbuilder.New(c.ProviderInfo().BaseURL, apiVersion, params.ObjectName)
+	if err != nil {
+		return nil, err
+	}
+
+	// first page pagination
+	if supportLimitAndOffset.Has(params.ObjectName) {
+		url.WithQueryParam("limit", strconv.Itoa(pageSize))
+		url.WithQueryParam("offset", "0")
+	}
+
+	return http.NewRequestWithContext(ctx, http.MethodGet, url.String(), nil)
+}
+
+func (c *Connector) parseReadResponse(
+	ctx context.Context,
+	params common.ReadParams,
+	request *http.Request,
+	response *common.JSONHTTPResponse,
+) (*common.ReadResult, error) {
+	responseFieldName := metadata.Schemas.LookupArrayFieldName(c.Module(), params.ObjectName)
+	requestURL := request.URL
+
+	return common.ParseResult(
+		response,
+		common.GetRecordsUnderJSONPath(responseFieldName),
+		nextRecordsURL(requestURL),
+		common.GetMarshaledData,
+		params.Fields,
+	)
+}
diff --git a/providers/brevo/parse.go b/providers/brevo/parse.go
new file mode 100644
index 00000000..70fb1f37
--- /dev/null
+++ b/providers/brevo/parse.go
@@ -0,0 +1,51 @@
+package brevo
+
+import (
+	"net/url"
+	"strconv"
+
+	"github.com/amp-labs/connectors/common"
+	"github.com/amp-labs/connectors/internal/jsonquery"
+	"github.com/spyzhov/ajson"
+)
+
+const (
+	pageSizeKey = "limit"
+	pageSize    = 200
+)
+
+func nextRecordsURL(previousURL *url.URL) common.NextPageFunc {
+	return func(node *ajson.Node) (string, error) {
+		totalCount, err := jsonquery.New(node).IntegerOptional("count")
+		if err != nil || totalCount == nil {
+			return "", err
+		}
+
+		nextURL := *previousURL
+
+		// parse the current offset value form previous request url
+		query := nextURL.Query()
+		currentOffset := 0
+		offsetStr := query.Get("offset")
+
+		if offsetStr != "" {
+			currentOffset, err = strconv.Atoi(offsetStr)
+			if err != nil {
+				return "", nil //nolint:nilerr
+			}
+		}
+
+		// calcuate the next offset value
+		nextOffset := currentOffset + pageSize
+		// if nextOffset would exceed total count, we've fetched all records
+		if nextOffset >= int(*totalCount) {
+			return "", nil
+		}
+
+		query.Set("offset", strconv.Itoa(nextOffset))
+		query.Set("limit", strconv.Itoa(pageSize))
+		nextURL.RawQuery = query.Encode()
+
+		return nextURL.String(), nil
+	}
+}
diff --git a/providers/brevo/support.go b/providers/brevo/support.go
new file mode 100644
index 00000000..604fb4c2
--- /dev/null
+++ b/providers/brevo/support.go
@@ -0,0 +1,49 @@
+package brevo
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/amp-labs/connectors/internal/components"
+	"github.com/amp-labs/connectors/internal/datautils"
+	"github.com/amp-labs/connectors/internal/staticschema"
+	"github.com/amp-labs/connectors/providers/brevo/metadata"
+)
+
+var supportLimitAndOffset = datautils.NewSet( //nolint:gochecknoglobals
+	"blockedContacts",
+	"categories",
+	"children",
+	"companies",
+	"contacts",
+	"deals",
+	"emailCampaigns",
+	"files",
+	"folders",
+	"inbound/events",
+	"lists",
+	"notes",
+	"processes",
+	"products",
+	"smsCampaigns",
+	"smtp/statistics/events",
+	"smtp/statistics/reports",
+	"subAccount",
+	"tasks",
+	"templates",
+	"transactionalSMS/statistics/events",
+)
+
+func supportedOperations() components.EndpointRegistryInput {
+	// We support reading everything under schema.json, so we get all the objects and join it into a pattern.
+	readSupport := metadata.Schemas.ObjectNames().GetList(staticschema.RootModuleID)
+
+	return components.EndpointRegistryInput{
+		staticschema.RootModuleID: {
+			{
+				Endpoint: fmt.Sprintf("{%s}", strings.Join(readSupport, ",")),
+				Support:  components.ReadSupport,
+			},
+		},
+	}
+}
diff --git a/providers/hubspot/subscriptionEvent.go b/providers/hubspot/subscriptionEvent.go
index b880bb8c..e675963d 100644
--- a/providers/hubspot/subscriptionEvent.go
+++ b/providers/hubspot/subscriptionEvent.go
@@ -24,7 +24,6 @@ Newer version doc: https://developers.hubspot.com/docs/guides/apps/public-apps/c
 */
 
 /*
-Note:
 SubscriptionEvent is a map[string]any as opposed to a typed struct, because the structure of the event is not known .
 We may define the latest structure of the event, but in the future, the provider may add more fields.
 In that case, we won't be receiving those fields in the event.
@@ -33,7 +32,7 @@ This form also prevents null fields to be sent out as zero values.
 type SubscriptionEvent map[string]any
 
 // VerifyWebhookMessage verifies the signature of a webhook message from Hubspot.
-func (c *Connector) VerifyWebhookMessage(
+func (*Connector) VerifyWebhookMessage(
 	_ context.Context, params *common.WebhookVerificationParameters,
 ) (bool, error) {
 	ts := params.Headers.Get(string(xHubspotRequestTimestamp))
diff --git a/providers/salesforce/subscribe.go b/providers/salesforce/subscribe.go
index 867d2a35..5b6f86db 100644
--- a/providers/salesforce/subscribe.go
+++ b/providers/salesforce/subscribe.go
@@ -13,11 +13,11 @@ type SubscribeResult struct {
 	EventChannelMembers map[common.ObjectName]*EventChannelMember
 }
 
-func (conn *Connector) EmptySubscritpionParams() *common.SubscribeParams {
+func (c *Connector) EmptySubscritpionParams() *common.SubscribeParams {
 	return &common.SubscribeParams{}
 }
 
-func (conn *Connector) EmptySubscriptionResult() *common.SubscriptionResult {
+func (c *Connector) EmptySubscriptionResult() *common.SubscriptionResult {
 	return &common.SubscriptionResult{
 		Result: &SubscribeResult{},
 	}
@@ -31,7 +31,7 @@ func (conn *Connector) EmptySubscriptionResult() *common.SubscriptionResult {
 // Registration is required prior to subscribing.
 //
 //nolint:funlen,cyclop
-func (conn *Connector) Subscribe(
+func (c *Connector) Subscribe(
 	ctx context.Context,
 	params common.SubscribeParams,
 ) (*common.SubscriptionResult, error) {
@@ -48,11 +48,11 @@ func (conn *Connector) Subscribe(
 		return nil, fmt.Errorf("invalid registration result: %w", err)
 	}
 
-	regstrationParams, ok := params.RegistrationResult.Result.(*ResultData)
+	registrationParams, ok := params.RegistrationResult.Result.(*ResultData)
 	if !ok {
 		return nil, fmt.Errorf(
 			"%w: expected SubscribeParams.RegistrationResult.Result to be type '%T', but got '%T'", errInvalidRequestType,
-			regstrationParams,
+			registrationParams,
 			params.RegistrationResult.Result,
 		)
 	}
@@ -65,7 +65,7 @@ func (conn *Connector) Subscribe(
 
 	for objName := range params.SubscriptionEvents {
 		eventName := GetChangeDataCaptureEventName(string(objName))
-		rawChannelName := GetRawChannelNameFromChannel(regstrationParams.EventChannel.FullName)
+		rawChannelName := GetRawChannelNameFromChannel(registrationParams.EventChannel.FullName)
 
 		channelMetadata := &EventChannelMemberMetadata{
 			EventChannel:   GetChannelName(rawChannelName),
@@ -76,7 +76,7 @@ func (conn *Connector) Subscribe(
 			Metadata: channelMetadata,
 		}
 
-		newChannelMember, err := conn.CreateEventChannelMember(ctx, channelMember)
+		newChannelMember, err := c.CreateEventChannelMember(ctx, channelMember)
 		if err != nil {
 			failError = fmt.Errorf("failed to create event channel member for object %s, %w", objName, err)
 
@@ -100,7 +100,7 @@ func (conn *Connector) Subscribe(
 
 	if failError != nil {
 		for objName, member := range sfRes.EventChannelMembers {
-			if _, err := conn.DeleteEventChannelMember(ctx, member.Id); err != nil {
+			if _, err := c.DeleteEventChannelMember(ctx, member.Id); err != nil {
 				rollbackError = errors.Join(
 					rollbackError,
 					fmt.Errorf("failed to delete event channel member for object %s: %w",
@@ -146,7 +146,7 @@ func (conn *Connector) Subscribe(
 
 // DeleteSubscription deletes the subscription by deleting all the event channel members.
 // If any of the event channel members fail to be deleted, it will return an error.
-func (conn *Connector) DeleteSubscription(ctx context.Context, params common.SubscriptionResult) error {
+func (c *Connector) DeleteSubscription(ctx context.Context, params common.SubscriptionResult) error {
 	if params.Result == nil {
 		return fmt.Errorf("%w: missing SubscriptionResult.Result", errMissingParams)
 	}
@@ -162,7 +162,7 @@ func (conn *Connector) DeleteSubscription(ctx context.Context, params common.Sub
 	}
 
 	for objectName, member := range sfRes.EventChannelMembers {
-		if _, err := conn.DeleteEventChannelMember(ctx, member.Id); err != nil {
+		if _, err := c.DeleteEventChannelMember(ctx, member.Id); err != nil {
 			return fmt.Errorf("failed to delete event channel member '%s': %w", objectName, err)
 		}
 	}
diff --git a/providers/salesforce/subscriptionEvent.go b/providers/salesforce/subscriptionEvent.go
new file mode 100644
index 00000000..a57b72b5
--- /dev/null
+++ b/providers/salesforce/subscriptionEvent.go
@@ -0,0 +1,177 @@
+// nolint:lll,tagliatelle
+package salesforce
+
+import (
+	"context"
+	"errors"
+
+	"github.com/amp-labs/connectors/common"
+	"github.com/amp-labs/connectors/internal/goutils"
+)
+
+const (
+	keyEventRecordID          = "recordId"
+	keyEventRecordIdentifiers = "recordIds"
+	keyEventChangeEventHeader = "ChangeEventHeader"
+)
+
+var (
+	errChangeEventHeaderType = errors.New("key ChangeEventHeader is not of map[string]any type")
+	errRecordIDsType         = errors.New("key recordIds is not of []any type")
+)
+
+func (*Connector) VerifyWebhookMessage(context.Context, *common.WebhookVerificationParameters) (bool, error) {
+	return true, nil
+}
+
+// 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
+
+// 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) {
+	eventHeaderMap, err := extractChangeEventHeader(e)
+	if err != nil {
+		return nil, err
+	}
+
+	recordIDsAny, err := eventHeaderMap.Get(keyEventRecordIdentifiers)
+	if err != nil {
+		return nil, err
+	}
+
+	recordIDs, ok := recordIDsAny.([]any)
+	if !ok {
+		return nil, errRecordIDsType
+	}
+
+	events := make([]SubscriptionEvent, len(recordIDs))
+
+	for index, recordID := range recordIDs {
+		event, err := goutils.Clone[map[string]any](e)
+		if err != nil {
+			return nil, err
+		}
+
+		// Reach out to the nested object and remove record identifiers and attach record id.
+		changeEventHeader, ok := event[keyEventChangeEventHeader].(map[string]any)
+		if !ok {
+			return nil, errChangeEventHeaderType
+		}
+
+		changeEventHeader[keyEventRecordID] = recordID
+		delete(changeEventHeader, keyEventRecordIdentifiers)
+
+		// Save changes.
+		event[keyEventChangeEventHeader] = changeEventHeader
+		events[index] = event
+	}
+
+	return events, nil
+}
+
+var _ common.SubscriptionEvent = SubscriptionEvent{}
+
+// SubscriptionEvent holds event data.
+//
+// Record ID can have wildcard symbols:
+// https://developer.salesforce.com/docs/atlas.en-us.change_data_capture.meta/change_data_capture/cdc_field_conversion_single_event.htm.
+type SubscriptionEvent map[string]any
+
+func (s SubscriptionEvent) asMap() (common.StringMap, error) {
+	return extractChangeEventHeader(s)
+}
+
+// EventType
+// https://developer.salesforce.com/docs/atlas.en-us.change_data_capture.meta/change_data_capture/cdc_event_fields_header.htm.
+func (s SubscriptionEvent) EventType() (common.SubscriptionEventType, error) {
+	changeType, err := s.RawEventName()
+	if err != nil {
+		return "", err
+	}
+
+	switch changeType {
+	case "CREATE", "GAP_CREATE":
+		return common.SubscriptionEventTypeCreate, nil
+	case "UPDATE", "GAP_UPDATE":
+		return common.SubscriptionEventTypeUpdate, nil
+	case "DELETE", "GAP_DELETE":
+		return common.SubscriptionEventTypeDelete, nil
+	case "UNDELETE", "SNAPSHOT", "GAP_UNDELETE", "GAP_OVERFLOW":
+		fallthrough
+	default:
+		return common.SubscriptionEventTypeOther, nil
+	}
+}
+
+func (s SubscriptionEvent) RawEventName() (string, error) {
+	registry, err := s.asMap()
+	if err != nil {
+		return "", err
+	}
+
+	return registry.GetString("changeType")
+}
+
+func (s SubscriptionEvent) ObjectName() (string, error) {
+	registry, err := s.asMap()
+	if err != nil {
+		return "", err
+	}
+
+	// The API name of the standard or custom object that the change pertains to.
+	// For example, Account or MyObject__c.
+	return registry.GetString("entityName")
+}
+
+func (s SubscriptionEvent) Workspace() (string, error) {
+	// Not applicable
+	return "", nil
+}
+
+// RecordId
+// https://developer.salesforce.com/docs/atlas.en-us.change_data_capture.meta/change_data_capture/cdc_field_conversion_single_event.htm.
+func (s SubscriptionEvent) RecordId() (string, error) {
+	registry, err := s.asMap()
+	if err != nil {
+		return "", err
+	}
+
+	return registry.GetString(keyEventRecordID)
+}
+
+func (s SubscriptionEvent) EventTimeStampNano() (int64, error) {
+	registry, err := s.asMap()
+	if err != nil {
+		return 0, err
+	}
+
+	// The date and time when the change occurred,
+	// represented as the number of milliseconds since January 1, 1970 00:00:00 GMT.
+	num, err := registry.GetNumber("commitTimestamp")
+	if err != nil {
+		return 0, err
+	}
+
+	return int64(num), nil
+}
+
+func extractChangeEventHeader(registry map[string]any) (common.StringMap, error) {
+	eventMap := common.StringMap(registry)
+
+	eventHeaderAny, err := eventMap.Get(keyEventChangeEventHeader)
+	if err != nil {
+		return nil, err
+	}
+
+	eventHeader, ok := eventHeaderAny.(map[string]any)
+	if !ok {
+		return nil, errChangeEventHeaderType
+	}
+
+	return eventHeader, nil
+}


Test patch:
diff --git a/providers/salesforce/subscriptionEvent_test.go b/providers/salesforce/subscriptionEvent_test.go
new file mode 100644
index 00000000..10bd2d65
--- /dev/null
+++ b/providers/salesforce/subscriptionEvent_test.go
@@ -0,0 +1,55 @@
+package salesforce
+
+import (
+	"encoding/json"
+	"testing"
+
+	"github.com/amp-labs/connectors/common"
+	"github.com/amp-labs/connectors/test/utils/testutils"
+	"gotest.tools/v3/assert"
+)
+
+func TestSubscriptionEventProperties(t *testing.T) {
+	t.Parallel()
+
+	eventNewAccountData := testutils.DataFromFile(t, "subscription/new_account.json")
+
+	changeEvent := ChangeEvent{}
+	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()
+	assert.NilError(t, err, "error should be nil")
+
+	if len(events) != 1 {
+		t.Fatalf("failed to start a test, expected to have only one event")
+	}
+
+	event := events[0]
+
+	eventType, err := event.EventType()
+	validateSubEvent(t, err, eventType, common.SubscriptionEventTypeCreate, "EventType")
+
+	rawEventType, err := event.RawEventName()
+	validateSubEvent(t, err, rawEventType, "CREATE", "RawEventName")
+
+	objectName, err := event.ObjectName()
+	validateSubEvent(t, err, objectName, "Account", "ObjectName")
+
+	workspace, err := event.Workspace()
+	validateSubEvent(t, err, workspace, "", "Workspace")
+
+	recordID, err := event.RecordId()
+	validateSubEvent(t, err, recordID, "0015f00002J9YYEAA3", "RecordId")
+
+	timestamp, err := event.EventTimeStampNano()
+	validateSubEvent(t, err, timestamp, 1712693965000, "EventTimeStampNano")
+}
+
+func validateSubEvent[V any](t *testing.T, err error, actual, expected V, methodName string) {
+	t.Helper()
+
+	assert.NilError(t, err, "error should be nil")
+	assert.Equal(t, actual, expected, "method "+methodName)
+}
diff --git a/providers/salesforce/test/subscription/new_account.json b/providers/salesforce/test/subscription/new_account.json
new file mode 100644
index 00000000..c69a0b23
--- /dev/null
+++ b/providers/salesforce/test/subscription/new_account.json
@@ -0,0 +1,60 @@
+{
+  "ChangeEventHeader": {
+    "entityName": "Account",
+    "recordIds": [
+      "0015f00002J9YYEAA3"
+    ],
+    "changeType": "CREATE",
+    "changeOrigin": "com/salesforce/api/soap/60.0;client=SfdcInternalAPI/",
+    "transactionKey": "0001ade9-3f74-0b99-dbc4-42e73424b774",
+    "sequenceNumber": 1,
+    "commitTimestamp": 1712693965000,
+    "commitNumber": 1082985383811,
+    "commitUser": "0055f000005mc66AAA",
+    "nulledFields": [],
+    "diffFields": [],
+    "changedFields": []
+  },
+  "Name": "Acme",
+  "Type": null,
+  "ParentId": null,
+  "BillingAddress": null,
+  "ShippingAddress": null,
+  "Phone": null,
+  "Fax": null,
+  "AccountNumber": null,
+  "Website": null,
+  "Sic": null,
+  "Industry": null,
+  "AnnualRevenue": null,
+  "NumberOfEmployees": null,
+  "Ownership": null,
+  "TickerSymbol": null,
+  "Description": "Sample account record.",
+  "Rating": null,
+  "Site": null,
+  "OwnerId": "0055f000005mc66AAA",
+  "CreatedDate": 1712693965000,
+  "CreatedById": "0055f000005mc66AAA",
+  "LastModifiedDate": 1712693965000,
+  "LastModifiedById": "0055f000005mc66AAA",
+  "Jigsaw": null,
+  "JigsawCompanyId": null,
+  "CleanStatus": "Pending",
+  "AccountSource": null,
+  "DunsNumber": null,
+  "Tradestyle": null,
+  "NaicsCode": null,
+  "NaicsDesc": null,
+  "YearStarted": null,
+  "SicDesc": null,
+  "DandbCompanyId": null,
+  "OperatingHoursId": null,
+  "CustomerPriority__c": null,
+  "SLA__c": null,
+  "Active__c": null,
+  "NumberofLocations__c": null,
+  "UpsellOpportunity__c": null,
+  "SLASerialNumber__c": null,
+  "SLAExpirationDate__c": null
+}
diff --git a/test/blueshift/write/main.go b/test/blueshift/write/main.go
index 9fb69a8e..45ef75f3 100644
--- a/test/blueshift/write/main.go
+++ b/test/blueshift/write/main.go
@@ -31,10 +31,10 @@ func main() {
 	}
 
 	slog.Info("> TEST Create Custom User List")
+
 	if err := createCustomUserList(ctx, conn); err != nil {
 		slog.Error(err.Error())
 	}
-
 }
 
 func createCustomer(ctx context.Context, conn *blueshift.Connector) error {
diff --git a/test/brevo/metadata/metadata.go b/test/brevo/metadata/metadata.go
index d268f85d..547c62f0 100644
--- a/test/brevo/metadata/metadata.go
+++ b/test/brevo/metadata/metadata.go
@@ -19,5 +19,6 @@ func main() {
 	if err != nil {
 		log.Fatal(err)
 	}
+
 	utils.DumpJSON(m, os.Stdout)
 }
diff --git a/test/brevo/read/main.go b/test/brevo/read/main.go
new file mode 100644
index 00000000..99a38fa9
--- /dev/null
+++ b/test/brevo/read/main.go
@@ -0,0 +1,36 @@
+package main
+
+import (
+	"context"
+	"log/slog"
+	"os"
+	"os/signal"
+	"syscall"
+
+	"github.com/amp-labs/connectors"
+	"github.com/amp-labs/connectors/common"
+	connTest "github.com/amp-labs/connectors/test/brevo"
+	"github.com/amp-labs/connectors/test/utils"
+)
+
+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.GetBrevoConnector(ctx)
+
+	res, err := conn.Read(ctx, common.ReadParams{
+		ObjectName: "contacts",
+		Fields:     connectors.Fields("id", "smsBlacklisted", "email"),
+	})
+	if err != nil {
+		utils.Fail("error reading from Brevo", "error", err)
+	}
+
+	slog.Info("Reading projects..")
+	utils.DumpJSON(res, os.Stdout)
+}
diff --git a/test/groove/read/main.go b/test/groove/read/main.go
index 059caa04..9b8954f7 100644
--- a/test/groove/read/main.go
+++ b/test/groove/read/main.go
@@ -37,7 +37,6 @@ func main() {
 	if err := testRead(ctx, conn, "agents", []string{"email", "first_name", "id"}); err != nil {
 		slog.Error(err.Error())
 	}
-
 }
 
 func testRead(ctx context.Context, conn *gr.Connector, objectName string, fields []string) error {
@@ -60,6 +59,7 @@ func testRead(ctx context.Context, conn *gr.Connector, objectName string, fields
 	if _, err := os.Stdout.Write(jsonStr); err != nil {
 		return fmt.Errorf("error writing to stdout: %w", err)
 	}
+
 	if _, err := os.Stdout.WriteString("\n"); err != nil {
 		return fmt.Errorf("error writing to stdout: %w", err)
 	}