Repository: amp-labs/connectors
Pull request #1434 (2025-03-04 11:53:59)

[CON-1560] feat: Add Bitbucket Connector
## Checklist
- [x] Ran Linter

## Testing 


### GET
URL: http://localhost:4444/2.0/repositories
<img width="1374" alt="image" src="https://github.com/user-attachments/assets/c818cb64-1452-4454-b4b5-8eea5825ead9" />





### POST
URL: http://localhost:4444/2.0/workspaces/integrationuser-bitbucket/hooks
<img width="1361" alt="image" src="https://github.com/user-attachments/assets/cc0b0705-9756-4b19-8b1c-6d8af0b82629" />






### PUT
URL: http://localhost:4444/2.0/workspaces/integrationuser-bitbucket/hooks/bf9632f3-251e-4726-ac24-8f5ecacd312d
<img width="1363" alt="image" src="https://github.com/user-attachments/assets/f40d9bf9-a1b1-44b3-ade9-3cc6fc44f536" />






### DELETE
URL: http://localhost:4444/2.0/workspaces/integrationuser-bitbucket/hooks/bf9632f3-251e-4726-ac24-8f5ecacd312d
<img width="1366" alt="image" src="https://github.com/user-attachments/assets/f4c4977b-d746-435a-ba5c-966e26d85f97" />






### Pagination
Bitbucket pagination uses a next link to load more result. The last page has no next link.
<img width="1364" alt="image" src="https://github.com/user-attachments/assets/3fac38e1-4722-4a8c-b8e8-371853f28d59" />




Next page:
<img width="1359" alt="image" src="https://github.com/user-attachments/assets/aa1206b2-6b82-4e7e-a259-6e7422e74b4d" />





### Errors


Invalid URL provided
<img width="1359" alt="image" src="https://github.com/user-attachments/assets/20e84aa8-dfd4-4fcd-aa25-da7c0fbea7f9" />



 incomplete payload provided
<img width="1362" alt="image" src="https://github.com/user-attachments/assets/ed88394a-8a1b-46cf-a677-06daab2b776c" />

 invalid method provided
<img width="1355" alt="image" src="https://github.com/user-attachments/assets/122dc575-e5f7-4670-9d29-db72a703d924" />








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/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/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/go.mod b/go.mod
index 2488b5fe..59b77a82 100644
--- a/go.mod
+++ b/go.mod
@@ -20,8 +20,8 @@ require (
 	github.com/stretchr/testify v1.10.0
 	gitlab.com/c0b/go-ordered-json v0.0.0-20201030195603-febf46534d5a
 	golang.org/x/exp v0.0.0-20250103183323-7d7fa50e5329
-	golang.org/x/oauth2 v0.27.0
-	golang.org/x/text v0.22.0
+	golang.org/x/oauth2 v0.28.0
+	golang.org/x/text v0.23.0
 )
 
 require (
diff --git a/go.sum b/go.sum
index 6bfeeb07..28c03d50 100644
--- a/go.sum
+++ b/go.sum
@@ -143,8 +143,8 @@ golang.org/x/net v0.25.0/go.mod h1:JkAGAh7GEvH74S6FOH42FLoXpXbE/aqXSrIQjXgsiwM=
 golang.org/x/net v0.33.0/go.mod h1:HXLR5J+9DxmrqMwG9qjGCxZ+zKXxBru04zlTvWlWuN4=
 golang.org/x/net v0.35.0 h1:T5GQRQb2y08kTAByq9L4/bz8cipCdA8FbRTXewonqY8=
 golang.org/x/net v0.35.0/go.mod h1:EglIi67kWsHKlRzzVMUD93VMSWGFOMSZgxFjparz1Qk=
-golang.org/x/oauth2 v0.27.0 h1:da9Vo7/tDv5RH/7nZDz1eMGS/q1Vv1N/7FCrBhI9I3M=
-golang.org/x/oauth2 v0.27.0/go.mod h1:onh5ek6nERTohokkhCD/y2cV4Do3fxFHFuAejCkRWT8=
+golang.org/x/oauth2 v0.28.0 h1:CrgCKl8PPAVtLnU3c+EDw6x11699EWlsDeWNWKdIOkc=
+golang.org/x/oauth2 v0.28.0/go.mod h1:onh5ek6nERTohokkhCD/y2cV4Do3fxFHFuAejCkRWT8=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
@@ -183,8 +183,8 @@ golang.org/x/text v0.13.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE=
 golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU=
 golang.org/x/text v0.15.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU=
 golang.org/x/text v0.21.0/go.mod h1:4IBbMaMmOPCJ8SecivzSH54+73PCFmPWxNTLm+vZkEQ=
-golang.org/x/text v0.22.0 h1:bofq7m3/HAFvbF51jz3Q9wLg3jkvSPuiZu/pD1XwgtM=
-golang.org/x/text v0.22.0/go.mod h1:YRoo4H8PVmsu+E3Ou7cqLVH8oXWIHVoX0jqUWALQhfY=
+golang.org/x/text v0.23.0 h1:D71I7dUrlY+VX0gQShAThNGHFxZ13dGLBHQLVl1mJlY=
+golang.org/x/text v0.23.0/go.mod h1:/BLNzu4aZCJ1+kcD0DNRotWKage4q2rGVAg4o22unh4=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc=
diff --git a/internal/components/operations/operations.go b/internal/components/operations/operations.go
index 640499b9..e3101e61 100644
--- a/internal/components/operations/operations.go
+++ b/internal/components/operations/operations.go
@@ -23,7 +23,7 @@ type HTTPOperation[RequestType any, ResponseType any] struct {
 // HTTPHandlers contains operation-specific HTTP handlers for building and parsing HTTP requests and responses.
 type HTTPHandlers[RequestType any, ResponseType any] struct {
 	BuildRequest  func(context.Context, RequestType) (*http.Request, error)
-	ParseResponse func(context.Context, RequestType, *common.JSONHTTPResponse) (ResponseType, error)
+	ParseResponse func(context.Context, RequestType, *http.Request, *common.JSONHTTPResponse) (ResponseType, error)
 	ErrorHandler  func(*http.Response, []byte) error
 }
 
@@ -85,5 +85,5 @@ func (op *HTTPOperation[RequestType, ResponseType]) ExecuteRequest(
 		return response, err
 	}
 
-	return op.handlers.ParseResponse(ctx, params, jsonResp)
+	return op.handlers.ParseResponse(ctx, params, req, jsonResp)
 }
diff --git a/internal/components/transport.go b/internal/components/transport.go
index bc89070c..e39fc61b 100644
--- a/internal/components/transport.go
+++ b/internal/components/transport.go
@@ -40,5 +40,10 @@ func NewTransport(
 	}, nil
 }
 
-func (c *Transport) JSONHTTPClient() *common.JSONHTTPClient { return c.json }
-func (c *Transport) HTTPClient() *common.HTTPClient         { return c.json.HTTPClient }
+func (t *Transport) SetBaseURL(newURL string) {
+	t.ProviderContext.providerInfo.BaseURL = newURL
+	t.json.HTTPClient.Base = newURL
+}
+
+func (t *Transport) JSONHTTPClient() *common.JSONHTTPClient { return t.json }
+func (t *Transport) HTTPClient() *common.HTTPClient         { return t.json.HTTPClient }
diff --git a/internal/goutils/cloning.go b/internal/goutils/cloning.go
index 560409d1..3b0830e6 100644
--- a/internal/goutils/cloning.go
+++ b/internal/goutils/cloning.go
@@ -5,8 +5,16 @@ import (
 	"encoding/gob"
 )
 
+//nolint:ireturn
+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
+//
+// nolint:ireturn
+func Clone[T any](input T) (T, error) {
 	var buf bytes.Buffer
 	enc := gob.NewEncoder(&buf)
 	dec := gob.NewDecoder(&buf)
diff --git a/providers/bitbucket.go b/providers/bitbucket.go
new file mode 100644
index 00000000..4c755397
--- /dev/null
+++ b/providers/bitbucket.go
@@ -0,0 +1,42 @@
+package providers
+
+const Bitbucket Provider = "bitbucket"
+
+func init() {
+	SetInfo(Bitbucket, ProviderInfo{
+		DisplayName: "Bitbucket",
+		AuthType:    Oauth2,
+		BaseURL:     "https://api.bitbucket.org",
+		Oauth2Opts: &Oauth2Opts{
+			GrantType:                 AuthorizationCode,
+			AuthURL:                   "https://bitbucket.org/site/oauth2/authorize",
+			TokenURL:                  "https://bitbucket.org/site/oauth2/access_token",
+			ExplicitScopesRequired:    false,
+			ExplicitWorkspaceRequired: false,
+		},
+		PostAuthInfoNeeded: false,
+		//nolint:lll
+		Media: &Media{
+			DarkMode: &MediaTypeDarkMode{
+				IconURL: "https://res.cloudinary.com/dycvts6vp/image/upload/v1741019187/media/bitbucket.org_1741019186.jpg",
+				LogoURL: "https://res.cloudinary.com/dycvts6vp/image/upload/v1741019214/media/bitbucket.org_1741019213.svg",
+			},
+			Regular: &MediaTypeRegular{
+				IconURL: "https://res.cloudinary.com/dycvts6vp/image/upload/v1741019187/media/bitbucket.org_1741019186.jpg",
+				LogoURL: "https://res.cloudinary.com/dycvts6vp/image/upload/v1741019214/media/bitbucket.org_1741019213.svg",
+			},
+		},
+		Support: Support{
+			BulkWrite: BulkWriteSupport{
+				Insert: false,
+				Update: false,
+				Upsert: false,
+				Delete: false,
+			},
+			Proxy:     false,
+			Read:      false,
+			Subscribe: false,
+			Write:     false,
+		},
+	})
+}
diff --git a/providers/blueshift/handlers.go b/providers/blueshift/handlers.go
index 0f551ac2..e3fcad29 100644
--- a/providers/blueshift/handlers.go
+++ b/providers/blueshift/handlers.go
@@ -50,6 +50,7 @@ func (c *Connector) buildReadRequest(ctx context.Context, params common.ReadPara
 func (c *Connector) parseReadResponse(
 	ctx context.Context,
 	params common.ReadParams,
+	request *http.Request,
 	response *common.JSONHTTPResponse,
 ) (*common.ReadResult, error) {
 	path, err := metadata.Schemas.LookupURLPath(c.Module(), params.ObjectName)
@@ -102,6 +103,7 @@ func (c *Connector) buildWriteRequest(ctx context.Context, params common.WritePa
 func (c *Connector) parseWriteResponse(
 	ctx context.Context,
 	params common.WriteParams,
+	request *http.Request,
 	response *common.JSONHTTPResponse,
 ) (*common.WriteResult, error) {
 	node, ok := response.Body()
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/front/handlers.go b/providers/front/handlers.go
index 006fcf18..ae2cfc3e 100644
--- a/providers/front/handlers.go
+++ b/providers/front/handlers.go
@@ -37,6 +37,7 @@ func (c *Connector) buildReadRequest(ctx context.Context, params common.ReadPara
 func (c *Connector) parseReadResponse(
 	ctx context.Context,
 	params common.ReadParams,
+	request *http.Request,
 	response *common.JSONHTTPResponse,
 ) (*common.ReadResult, error) {
 	return common.ParseResult(
@@ -73,6 +74,7 @@ func (c *Connector) buildWriteRequest(ctx context.Context, params common.WritePa
 func (c *Connector) parseWriteResponse(
 	ctx context.Context,
 	params common.WriteParams,
+	request *http.Request,
 	response *common.JSONHTTPResponse,
 ) (*common.WriteResult, error) {
 	body, ok := response.Body()
diff --git a/providers/gorgias/handlers.go b/providers/gorgias/handlers.go
index 63242757..10d7b412 100644
--- a/providers/gorgias/handlers.go
+++ b/providers/gorgias/handlers.go
@@ -35,6 +35,7 @@ func (c *Connector) buildSingleObjectMetadataRequest(ctx context.Context, object
 func (c *Connector) parseSingleObjectMetadataResponse(
 	ctx context.Context,
 	objectName string,
+	request *http.Request,
 	response *common.JSONHTTPResponse,
 ) (*common.ObjectMetadata, error) {
 	objectMetadata := common.ObjectMetadata{
diff --git a/providers/groove/handlers.go b/providers/groove/handlers.go
index e3b321ce..2684a895 100644
--- a/providers/groove/handlers.go
+++ b/providers/groove/handlers.go
@@ -34,6 +34,7 @@ func (c *Connector) buildSingleObjectMetadataRequest(ctx context.Context, object
 func (c *Connector) parseSingleObjectMetadataResponse(
 	ctx context.Context,
 	objectName string,
+	request *http.Request,
 	response *common.JSONHTTPResponse,
 ) (*common.ObjectMetadata, error) {
 	objectMetadata := common.ObjectMetadata{
@@ -106,6 +107,7 @@ func (c *Connector) buildReadRequest(ctx context.Context, params common.ReadPara
 func (c *Connector) parseReadResponse(
 	ctx context.Context,
 	params common.ReadParams,
+	request *http.Request,
 	response *common.JSONHTTPResponse,
 ) (*common.ReadResult, error) {
 	return common.ParseResult(
@@ -142,6 +144,7 @@ func (c *Connector) buildWriteRequest(ctx context.Context, params common.WritePa
 func (c *Connector) parseWriteResponse(
 	ctx context.Context,
 	params common.WriteParams,
+	request *http.Request,
 	response *common.JSONHTTPResponse,
 ) (*common.WriteResult, error) {
 	body, ok := response.Body()
diff --git a/providers/helpscout/handlers.go b/providers/helpscout/handlers.go
index 37ac8334..a05c1257 100644
--- a/providers/helpscout/handlers.go
+++ b/providers/helpscout/handlers.go
@@ -43,6 +43,7 @@ func (c *Connector) buildSingleObjectMetadataRequest(ctx context.Context, object
 func (c *Connector) parseSingleObjectMetadataResponse(
 	ctx context.Context,
 	objectName string,
+	request *http.Request,
 	response *common.JSONHTTPResponse,
 ) (*common.ObjectMetadata, error) {
 	objectMetadata := common.ObjectMetadata{
@@ -89,6 +90,7 @@ func (c *Connector) buildReadRequest(ctx context.Context, params common.ReadPara
 func (c *Connector) parseReadResponse(
 	ctx context.Context,
 	params common.ReadParams,
+	request *http.Request,
 	response *common.JSONHTTPResponse,
 ) (*common.ReadResult, error) {
 	return common.ParseResult(
@@ -125,6 +127,7 @@ func (c *Connector) buildWriteRequest(ctx context.Context, params common.WritePa
 func (c *Connector) parseWriteResponse(
 	ctx context.Context,
 	params common.WriteParams,
+	request *http.Request,
 	response *common.JSONHTTPResponse,
 ) (*common.WriteResult, error) {
 	// The response is always an empty response body.
diff --git a/providers/hubspot/subscriptionEvent.go b/providers/hubspot/subscriptionEvent.go
index b880bb8c..888e2034 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.
@@ -32,8 +31,10 @@ 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 (c *Connector) VerifyWebhookMessage(
+func (*Connector) VerifyWebhookMessage(
 	_ context.Context, params *common.WebhookVerificationParameters,
 ) (bool, error) {
 	ts := params.Headers.Get(string(xHubspotRequestTimestamp))
diff --git a/providers/netsuite/handlers.go b/providers/netsuite/handlers.go
index b457bd4c..390ff097 100644
--- a/providers/netsuite/handlers.go
+++ b/providers/netsuite/handlers.go
@@ -32,6 +32,7 @@ func (c *Connector) buildObjectMetadataRequest(ctx context.Context, object strin
 func (c *Connector) parseObjectMetadataResponse(
 	ctx context.Context,
 	object string,
+	request *http.Request,
 	resp *common.JSONHTTPResponse,
 ) (*common.ObjectMetadata, error) {
 	metadata, err := common.UnmarshalJSON[metadataResponse](resp)
diff --git a/providers/outreach/metadata.go b/providers/outreach/metadata.go
index 08aa0276..4adb8a3a 100644
--- a/providers/outreach/metadata.go
+++ b/providers/outreach/metadata.go
@@ -2,6 +2,7 @@ package outreach
 
 import (
 	"context"
+	"fmt"
 
 	"github.com/amp-labs/connectors/common"
 )
@@ -45,7 +46,9 @@ func (c *Connector) ListObjectMetadata(ctx context.Context,
 
 		metadata, err := metadataMapper(res)
 		if err != nil {
-			return nil, err
+			objMetadata.Errors[obj] = err
+
+			continue
 		}
 
 		metadata.DisplayName = obj
@@ -65,6 +68,10 @@ func metadataMapper(resp *common.JSONHTTPResponse) (*common.ObjectMetadata, erro
 		FieldsMap: make(map[string]string),
 	}
 
+	if len(response.Data) == 0 {
+		return nil, fmt.Errorf("%w: could not find a record to sample fields from", common.ErrMissingExpectedValues)
+	}
+
 	attributes := response.Data[0].Attributes
 	for k := range attributes {
 		metadata.FieldsMap[k] = k
diff --git a/providers/salesforce/record.go b/providers/salesforce/record.go
new file mode 100644
index 00000000..5fead605
--- /dev/null
+++ b/providers/salesforce/record.go
@@ -0,0 +1,76 @@
+package salesforce
+
+import (
+	"context"
+
+	"github.com/amp-labs/connectors/common"
+	"github.com/amp-labs/connectors/common/urlbuilder"
+	"github.com/amp-labs/connectors/internal/datautils"
+)
+
+// GetRecordsWithIds returns records matching identifiers.
+func (c *Connector) GetRecordsWithIds( // nolint:revive
+	ctx context.Context,
+	objectName string,
+	ids []string,
+	fields []string,
+	associations []string, // no-op, preserved to match other deep connectors
+) ([]common.ReadResultRow, error) {
+	// Sanitize method arguments.
+	config := recordsByIDsParams{
+		ReadParams: common.ReadParams{
+			ObjectName:        objectName,
+			Fields:            datautils.NewSetFromList(fields),
+			AssociatedObjects: associations,
+		},
+		RecordIdentifiers: datautils.NewSetFromList(ids),
+	}
+
+	if err := config.ValidateParams(true); err != nil {
+		return nil, err
+	}
+
+	url, err := c.buildReadByIdentifierURL(config)
+	if err != nil {
+		return nil, err
+	}
+
+	rsp, err := c.Client.Get(ctx, url.String())
+	if err != nil {
+		return nil, err
+	}
+
+	readResult, err := common.ParseResult(
+		rsp,
+		getRecords,
+		getNextRecordsURL,
+		common.GetMarshalledDataWithId,
+		config.Fields,
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	return readResult.Data, nil
+}
+
+type recordsByIDsParams struct {
+	common.ReadParams
+	RecordIdentifiers datautils.Set[string]
+}
+
+func (c *Connector) buildReadByIdentifierURL(config recordsByIDsParams) (*urlbuilder.URL, error) {
+	// Requesting record identifiers using SOQL query.
+	url, err := c.getRestApiURL("query")
+	if err != nil {
+		return nil, err
+	}
+
+	query := makeSOQL(config.ReadParams).
+		WithIDs(config.RecordIdentifiers.List()).
+		String()
+
+	url.WithQueryParam("q", query)
+
+	return url, nil
+}
diff --git a/providers/salesforce/soql.go b/providers/salesforce/soql.go
index 78389aa9..c1d27ef0 100644
--- a/providers/salesforce/soql.go
+++ b/providers/salesforce/soql.go
@@ -5,6 +5,13 @@ import (
 	"strings"
 )
 
+// nolint:lll
+const (
+	// See `Limiting Result Rows` section on this webpage:
+	// https://developer.salesforce.com/docs/atlas.en-us.soql_sosl.meta/soql_sosl/sforce_api_calls_soql_select_fields.htm
+	identifiersLimitStr = "200"
+)
+
 // soqlBuilder builder of Salesforce Object Query Language.
 // It constructs query dynamically.
 type soqlBuilder struct {
@@ -20,7 +27,7 @@ func (s *soqlBuilder) SelectFields(fields []string) *soqlBuilder {
 			s.fields = "FIELDS(ALL)"
 			// if all fields are to be returned then we must limit to avoid error.
 			// Error example: `The SOQL FIELDS function must have a LIMIT of at most 200`
-			s.limit = "200"
+			s.limit = identifiersLimitStr
 
 			return s
 		}
@@ -47,6 +54,19 @@ func (s *soqlBuilder) Where(condition string) *soqlBuilder {
 	return s
 }
 
+func (s *soqlBuilder) WithIDs(identifiers []string) *soqlBuilder {
+	// Decorate each id with quotes.
+	for index, id := range identifiers {
+		identifiers[index] = fmt.Sprintf("'%v'", id)
+	}
+
+	identifiersList := strings.Join(identifiers, ",")
+
+	// nolint:lll
+	// https://developer.salesforce.com/docs/atlas.en-us.soql_sosl.meta/soql_sosl/sforce_api_calls_soql_select_fields.htm
+	return s.Where(fmt.Sprintf("Id IN (%v)", identifiersList))
+}
+
 func (s *soqlBuilder) String() string {
 	query := fmt.Sprintf("SELECT %s FROM %s", s.fields, s.from)
 
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..568ce759
--- /dev/null
+++ b/providers/salesforce/subscriptionEvent.go
@@ -0,0 +1,181 @@
+// 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
+}
+
+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 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 CollapsedSubscriptionEvent) SubscriptionEventList() ([]common.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([]common.SubscriptionEvent, len(recordIDs))
+
+	for index, recordID := range recordIDs {
+		event, err := goutils.Clone[map[string]any](e)
+		if err != nil {
+			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 {
+			return nil, errChangeEventHeaderType
+		}
+
+		changeEventHeader[keyEventRecordID] = recordID
+		delete(changeEventHeader, keyEventRecordIdentifiers)
+
+		// Save changes.
+		event[keyEventChangeEventHeader] = changeEventHeader
+		events[index] = evt
+	}
+
+	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":
+		return common.SubscriptionEventTypeCreate, nil
+	case "UPDATE":
+		return common.SubscriptionEventTypeUpdate, nil
+	case "DELETE":
+		return common.SubscriptionEventTypeDelete, nil
+	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
+}
diff --git a/providers/smartleadv2/handlers.go b/providers/smartleadv2/handlers.go
index 98cacf0f..bc9fd388 100644
--- a/providers/smartleadv2/handlers.go
+++ b/providers/smartleadv2/handlers.go
@@ -25,6 +25,7 @@ func (c *Connector) buildReadRequest(ctx context.Context, params common.ReadPara
 func (c *Connector) parseReadResponse(
 	ctx context.Context,
 	params common.ReadParams,
+	request *http.Request,
 	resp *common.JSONHTTPResponse,
 ) (*common.ReadResult, error) {
 	return common.ParseResult(
@@ -74,6 +75,7 @@ func (c *Connector) buildWriteRequest(ctx context.Context, params common.WritePa
 func (c *Connector) parseWriteResponse(
 	ctx context.Context,
 	params common.WriteParams,
+	request *http.Request,
 	resp *common.JSONHTTPResponse,
 ) (*common.WriteResult, error) {
 	// Get the JSON node from response
@@ -133,6 +135,7 @@ func (c *Connector) buildDeleteRequest(ctx context.Context, params common.Delete
 func (c *Connector) parseDeleteResponse(
 	ctx context.Context,
 	params common.DeleteParams,
+	request *http.Request,
 	resp *common.JSONHTTPResponse,
 ) (*common.DeleteResult, error) {
 	if resp.Code != http.StatusOK && resp.Code != http.StatusNoContent {
diff --git a/scripts/utils/credutils/tokenWriter.go b/scripts/utils/credutils/tokenWriter.go
index 2328ddb7..227b1846 100644
--- a/scripts/utils/credutils/tokenWriter.go
+++ b/scripts/utils/credutils/tokenWriter.go
@@ -53,9 +53,18 @@ func internalCredsFileWrite(filePath string, token *oauth2.Token) (*ordered.Orde
 	}
 
 	orderedMap.Set(credscanning.Fields.AccessToken.PathJSON, token.AccessToken)
-	orderedMap.Set(credscanning.Fields.RefreshToken.PathJSON, token.RefreshToken)
-	orderedMap.Set(credscanning.Fields.ExpiryFormat.PathJSON, "RFC3339Nano")
-	orderedMap.Set(credscanning.Fields.Expiry.PathJSON, token.Expiry.UTC().Format(time.RFC3339Nano))
+
+	if len(token.RefreshToken) == 0 {
+		// Refresh token and data associated with are not relevant for the connector.
+		orderedMap.Delete(credscanning.Fields.RefreshToken.PathJSON)
+		orderedMap.Delete(credscanning.Fields.ExpiryFormat.PathJSON)
+		orderedMap.Delete(credscanning.Fields.Expiry.PathJSON)
+	} else {
+		// Store data related to refresh token.
+		orderedMap.Set(credscanning.Fields.RefreshToken.PathJSON, token.RefreshToken)
+		orderedMap.Set(credscanning.Fields.ExpiryFormat.PathJSON, "RFC3339Nano")
+		orderedMap.Set(credscanning.Fields.Expiry.PathJSON, token.Expiry.UTC().Format(time.RFC3339Nano))
+	}
 
 	outputData, err := json.MarshalIndent(orderedMap, "", "  ")
 	if err != 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/soql_test.go b/providers/salesforce/soql_test.go
new file mode 100644
index 00000000..50ade0ad
--- /dev/null
+++ b/providers/salesforce/soql_test.go
@@ -0,0 +1,44 @@
+package salesforce
+
+import (
+	"testing"
+
+	"github.com/amp-labs/connectors/common"
+	"github.com/amp-labs/connectors/internal/datautils"
+	"gotest.tools/v3/assert"
+)
+
+func TestSoqlBuilderWithIDs(t *testing.T) {
+	t.Parallel()
+
+	soql := makeSOQL(common.ReadParams{
+		ObjectName: "Account",
+		// Note: fields doesn't preserve order of elements.
+		// To simplify test only one element is included.
+		Fields: datautils.NewSet("shippingstreet"),
+	})
+
+	{
+		// SOQL builder must produce query matching documentation.
+		// nolint:lll
+		// https://developer.salesforce.com/docs/atlas.en-us.soql_sosl.meta/soql_sosl/sforce_api_calls_soql_select_fields.htm
+		soql.WithIDs(
+			[]string{
+				"001ak00000OQ4RxAAL",
+				"001ak00000OQ4RyAAL",
+				"001ak00000OQ4TZAA1",
+				"001ak00000OQ4TbAAL",
+				"001ak00000OQ4VBAA1",
+				"001ak00000OQ4VCAA1",
+			})
+
+		output := soql.String()
+		assert.Equal(t, output, "SELECT shippingstreet FROM Account WHERE Id IN ("+
+			"'001ak00000OQ4RxAAL',"+
+			"'001ak00000OQ4RyAAL',"+
+			"'001ak00000OQ4TZAA1',"+
+			"'001ak00000OQ4TbAAL',"+
+			"'001ak00000OQ4VBAA1',"+
+			"'001ak00000OQ4VCAA1')", "mismatching SOQL query string")
+	}
+}
diff --git a/providers/salesforce/subscriptionEvent_test.go b/providers/salesforce/subscriptionEvent_test.go
new file mode 100644
index 00000000..82fbf416
--- /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 := 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.SubscriptionEventList()
+	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/metadata/metadata.go b/test/blueshift/metadata/metadata.go
index d26cdf0c..c151d174 100644
--- a/test/blueshift/metadata/metadata.go
+++ b/test/blueshift/metadata/metadata.go
@@ -19,5 +19,6 @@ func main() {
 	if err != nil {
 		log.Fatal(err)
 	}
+
 	utils.DumpJSON(m, os.Stdout)
 }
diff --git a/test/blueshift/read/main.go b/test/blueshift/read/main.go
index 89d1ad31..e970a0ba 100644
--- a/test/blueshift/read/main.go
+++ b/test/blueshift/read/main.go
@@ -22,6 +22,7 @@ func run() error {
 	connector := blueshift.GetBlueshiftConnector(ctx)
 
 	slog.Info("Reading campaigns")
+
 	res, err := connector.Read(ctx, common.ReadParams{
 		ObjectName: "campaigns",
 		Fields:     datautils.NewStringSet("uuid", "name"),
@@ -33,6 +34,7 @@ func run() error {
 	utils.DumpJSON(res, os.Stdout)
 
 	slog.Info("Reading email templates")
+
 	res, err = connector.Read(ctx, common.ReadParams{
 		ObjectName: "email_templates",
 		Fields:     datautils.NewStringSet("uuid", "name"),
@@ -44,6 +46,7 @@ func run() error {
 	utils.DumpJSON(res, os.Stdout)
 
 	slog.Info("Reading sms templates")
+
 	res, err = connector.Read(ctx, common.ReadParams{
 		ObjectName: "sms_templates",
 		Fields:     datautils.NewStringSet("uuid", "author"),
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)
 	}
diff --git a/test/outreach/metadata/metadata.go b/test/outreach/metadata/metadata.go
index 48209f11..bfe1237d 100644
--- a/test/outreach/metadata/metadata.go
+++ b/test/outreach/metadata/metadata.go
@@ -20,7 +20,7 @@ func main() {
 
 	conn := connTest.GetOutreachConnector(ctx)
 
-	m, err := conn.ListObjectMetadata(ctx, []string{"sequences"})
+	m, err := conn.ListObjectMetadata(ctx, []string{"snippets", "duties", "phoneNumbers"})
 	if err != nil {
 		utils.Fail("error listing metadata for Outreach", "error", err)
 	}
diff --git a/test/salesforce/record/main.go b/test/salesforce/record/main.go
new file mode 100644
index 00000000..1d1bf514
--- /dev/null
+++ b/test/salesforce/record/main.go
@@ -0,0 +1,39 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"os/signal"
+	"syscall"
+
+	connTest "github.com/amp-labs/connectors/test/salesforce"
+	"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.GetSalesforceConnector(ctx)
+
+	res, err := conn.GetRecordsWithIds(ctx,
+		"Account",
+		[]string{
+			"001ak00000OQ4RxAAL",
+			"001ak00000OQ4RyAAL",
+			"001ak00000OQ4TZAA1",
+		},
+		[]string{"id", "name", "shippingstreet"},
+		nil)
+	if err != nil {
+		utils.Fail("error reading from connector", "error", err)
+	}
+
+	fmt.Println("Reading..")
+	utils.DumpJSON(res, os.Stdout)
+}