Repository: amp-labs/connectors
Pull request #133 (2024-03-28 12:24:07)

[ENG-850] feat: Add Gainsight connector
## Checklist
- [x] Ran Linter
- [x] Catalog tests passing
- [x]  Created PR from non-main branch

## Catalog variables
Gainsight Provider = "gainsight"
Workspace = "sb-ampersand.us2"

## Notes
Supports batching requests and pagination as well

## Testing
### GET
URL: http://localhost:4444/v1/users/services/list
![GET](https://github.com/amp-labs/connectors/assets/163011126/daf99ab1-6304-4cdb-9d65-a919dfa8c554)

### POST
URL: http://localhost:4444/v1/users/services?notify=false
![POST](https://github.com/amp-labs/connectors/assets/163011126/95ce824d-f71c-43bd-aeaa-52a385d8452e)

### PUT
URL: http://localhost:4444/v1/users/services?key=SFDCUserName
![PUT)](https://github.com/amp-labs/connectors/assets/163011126/2fc51731-a50d-4a8f-8612-4fdd62a17dcb)



Patch:
diff --git a/common/http.go b/common/http.go
index fa1944b4..fdc0487c 100644
--- a/common/http.go
+++ b/common/http.go
@@ -134,7 +134,7 @@ func (h *HTTPClient) Delete(ctx context.Context,
 func (h *HTTPClient) httpGet(ctx context.Context,
 	url string, headers []Header,
 ) (*http.Response, []byte, error) {
-	req, err := makeGetRequest(ctx, url, headers)
+	req, err := MakeGetRequest(ctx, url, headers)
 	if err != nil {
 		return nil, nil, err
 	}
@@ -191,7 +191,7 @@ func (h *HTTPClient) httpDelete(ctx context.Context,
 }
 
 // makeGetRequest creates a GET request with the given headers.
-func makeGetRequest(ctx context.Context, url string, headers []Header) (*http.Request, error) {
+func MakeGetRequest(ctx context.Context, url string, headers []Header) (*http.Request, error) {
 	req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
 	if err != nil {
 		return nil, fmt.Errorf("error creating request: %w", err)
diff --git a/common/json.go b/common/json.go
index 6b3d0c7f..b8e19239 100644
--- a/common/json.go
+++ b/common/json.go
@@ -127,7 +127,7 @@ func UnmarshalJSON[T any](rsp *JSONHTTPResponse) (*T, error) {
 // MakeJSONGetRequest creates a GET request with the given headers and adds the
 // Accept: application/json header. It then returns the request.
 func MakeJSONGetRequest(ctx context.Context, url string, headers []Header) (*http.Request, error) {
-	return makeGetRequest(ctx, url, addAcceptJSONHeader(headers))
+	return MakeGetRequest(ctx, url, addAcceptJSONHeader(headers))
 }
 
 // addAcceptJSONHeader adds the Accept: application/json header to the given headers.
diff --git a/providers/catalog.go b/providers/catalog.go
index 0a4ea8f4..f9e19872 100644
--- a/providers/catalog.go
+++ b/providers/catalog.go
@@ -23,6 +23,7 @@ const (
 	Gong                                Provider = "gong"
 	Zoom                                Provider = "zoom"
 	Intercom                            Provider = "intercom"
+  	Capsule                             Provider = "capsule"
 	DocuSign                            Provider = "docuSign"
 	DocuSignDeveloper                   Provider = "docuSignDeveloper"
 	Calendly                            Provider = "calendly"
@@ -31,6 +32,7 @@ const (
 	ConstantContact                     Provider = "constantContact"
 	MicrosoftDynamics365Sales           Provider = "microsoftDynamics365Sales"
 	MicrosoftDynamics365BusinessCentral Provider = "microsoftDynamics365BusinessCentral"
+	Gainsight                           Provider = "gainsight"
 )
 
 // ================================================================================
@@ -164,6 +166,25 @@ var catalog = CatalogType{ // nolint:gochecknoglobals
 		},
 	},
 
+	// Capsule configuration
+	Capsule: {
+		AuthType: Oauth2,
+		BaseURL:  "https://api.capsulecrm.com/api",
+		OauthOpts: OauthOpts{
+			AuthURL:                   "https://api.capsulecrm.com/oauth/authorise",
+			TokenURL:                  "https://api.capsulecrm.com/oauth/token",
+			ExplicitScopesRequired:    true,
+			ExplicitWorkspaceRequired: false,
+		},
+		Support: Support{
+			BulkWrite: false,
+			Proxy:     false,
+			Read:      false,
+			Subscribe: false,
+			Write:     false,
+		},
+	},
+
 	// Copper configuration
 	Copper: {
 		AuthType: Oauth2,
@@ -550,4 +571,23 @@ var catalog = CatalogType{ // nolint:gochecknoglobals
 			Write:     false,
 		},
 	},
+
+	// Gainsight configuration
+	Gainsight: {
+		AuthType: Oauth2,
+		BaseURL:  "https://{{.workspace}}.gainsightcloud.com",
+		OauthOpts: OauthOpts{
+			AuthURL:                   "https://{{.workspace}}.gainsightcloud.com/v1/authorize",
+			TokenURL:                  "https://{{.workspace}}.gainsightcloud.com/v1/users/oauth/token",
+			ExplicitScopesRequired:    false,
+			ExplicitWorkspaceRequired: true,
+		},
+		Support: Support{
+			BulkWrite: false,
+			Proxy:     false,
+			Read:      false,
+			Subscribe: false,
+			Write:     false,
+		},
+	},
 }
diff --git a/salesforce/bulkwrite.go b/salesforce/bulk.go
similarity index 65%
rename from salesforce/bulkwrite.go
rename to salesforce/bulk.go
index d02c8d09..e2f4bfe2 100644
--- a/salesforce/bulkwrite.go
+++ b/salesforce/bulk.go
@@ -29,16 +29,17 @@ const (
 )
 
 var (
-	ErrKeyNotFound      = errors.New("key not found")
-	ErrUnknownNodeType  = errors.New("unknown node type when parsing JSON")
-	ErrInvalidType      = errors.New("invalid type")
-	ErrInvalidJobState  = errors.New("invalid job state")
-	ErrUnsupportedMode  = errors.New("unsupported mode")
-	ErrReadToByteFailed = errors.New("failed to read data to bytes")
+	ErrKeyNotFound          = errors.New("key not found")
+	ErrUnknownNodeType      = errors.New("unknown node type when parsing JSON")
+	ErrInvalidType          = errors.New("invalid type")
+	ErrInvalidJobState      = errors.New("invalid job state")
+	ErrUnsupportedMode      = errors.New("unsupported mode")
+	ErrReadToByteFailed     = errors.New("failed to read data to bytes")
+	ErrUnsupportedOperation = errors.New("unsupported operation")
 )
 
-// BulkWriteParams defines how we are writing data to a SaaS API.
-type BulkWriteParams struct {
+// BulkOperationParams defines how we are writing data to a SaaS API.
+type BulkOperationParams struct {
 	// The name of the object we are writing, e.g. "Account"
 	ObjectName string // required
 
@@ -54,8 +55,8 @@ type BulkWriteParams struct {
 
 type BulkWriteMode string
 
-// BulkWriteResult is what's returned from writing data via the BulkWrite call.
-type BulkWriteResult struct {
+// BulkOperation is what's returned from writing data via the BulkOperation call.
+type BulkOperationResult struct {
 	// State is the state of the bulk job process
 	State string `json:"state"`
 	// JobId is the ID of the bulk job process
@@ -66,25 +67,26 @@ type GetJobInfoResult struct {
 	Id                     string  `json:"id"`
 	Object                 string  `json:"object"`
 	CreatedById            string  `json:"createdById"`
-	ExternalIdFieldName    string  `json:"externalIdFieldName"`
+	ExternalIdFieldName    string  `json:"externalIdFieldName,omitempty"`
 	State                  string  `json:"state"`
 	Operation              string  `json:"operation"`
 	ColumnDelimiter        string  `json:"columnDelimiter"`
 	LineEnding             string  `json:"lineEnding"`
 	NumberRecordsFailed    float64 `json:"numberRecordsFailed"`
 	NumberRecordsProcessed float64 `json:"numberRecordsProcessed"`
-	ErrorMessage           string  `json:"errorMessage,omitempty"`
-
-	ApexProcessingTime      float64 `json:"apexProcessingTime"`
-	ApiActiveProcessingTime float64 `json:"apiActiveProcessingTime"`
-	ApiVersion              float64 `json:"apiVersion"`
-	ConcurrencyMode         string  `json:"concurrencyMode"`
-	ContentType             string  `json:"contentType"`
-	CreatedDate             string  `json:"createdDate"`
-	JobType                 string  `json:"jobType"`
-	Retries                 float64 `json:"retries"`
-	SystemModstamp          string  `json:"systemModstamp"`
-	TotalProcessingTime     float64 `json:"totalProcessingTime"`
+	ErrorMessage           string  `json:"errorMessage"`
+
+	ApexProcessingTime      float64 `json:"apexProcessingTime,omitempty"`
+	ApiActiveProcessingTime float64 `json:"apiActiveProcessingTime,omitempty"`
+	ApiVersion              float64 `json:"apiVersion,omitempty"`
+	ConcurrencyMode         string  `json:"concurrencyMode,omitempty"`
+	ContentType             string  `json:"contentType,omitempty"`
+	CreatedDate             string  `json:"createdDate,omitempty"`
+	JobType                 string  `json:"jobType,omitempty"`
+	Retries                 float64 `json:"retries,omitempty"`
+	SystemModstamp          string  `json:"systemModstamp,omitempty"`
+	TotalProcessingTime     float64 `json:"totalProcessingTime,omitempty"`
+	IsPkChunkingSupported   bool    `json:"isPkChunkingSupported,omitempty"`
 }
 
 type FailInfo struct {
@@ -104,8 +106,8 @@ type JobResults struct {
 
 func (c *Connector) BulkWrite( //nolint:funlen,cyclop
 	ctx context.Context,
-	config BulkWriteParams,
-) (*BulkWriteResult, error) {
+	config BulkOperationParams,
+) (*BulkOperationResult, error) {
 	// Only support upsert for now
 	switch config.Mode {
 	case Upsert:
@@ -118,87 +120,20 @@ func (c *Connector) BulkWrite( //nolint:funlen,cyclop
 		return nil, fmt.Errorf("%w: %s", ErrUnsupportedMode, config.Mode)
 	}
 
-	// cretes batch upload job, returns json with id and other info
-	res, err := c.createJob(ctx, config)
-	if err != nil {
-		return nil, fmt.Errorf("createJob failed: %w", err)
-	}
-
-	resObject, err := res.Body.GetObject()
-	if err != nil {
-		return nil, fmt.Errorf(
-			"parsing result of createJob failed: %w",
-			errors.Join(err, common.ErrParseError),
-		)
-	}
-
-	state, err := resObject["state"].GetString()
-	if err != nil {
-		unpacked, _ := resObject["state"].Unpack()
-
-		return nil, fmt.Errorf(
-			"%w: expected salesforce job state to be string in response, got %T",
-			ErrInvalidType,
-			unpacked,
-		)
-	}
-
-	if strings.ToLower(state) != "open" {
-		return nil, fmt.Errorf("%w: expected job state to be open, got %s", ErrInvalidJobState, state)
-	}
-
-	jobId, err := resObject["id"].GetString()
-	if err != nil {
-		unpacked, _ := resObject["id"].Unpack()
-
-		return nil, fmt.Errorf(
-			"%w: expected salesforce job id to be string in response, got %T",
-			ErrInvalidType,
-			unpacked)
-	}
-
-	// upload csv and there is no response body other than status code
-	_, err = c.uploadCSV(ctx, jobId, config)
-	if err != nil {
-		return nil, fmt.Errorf("uploadCSV failed: %w", err)
-	}
-
-	data, err := c.completeUpload(ctx, jobId)
-	if err != nil {
-		return nil, fmt.Errorf("completeUpload failed: %w", err)
-	}
-
-	dataObject, err := data.Body.GetObject()
-	if err != nil {
-		return nil, fmt.Errorf("parsing result of completeUpload failed: %w", errors.Join(err, common.ErrParseError))
-	}
-
-	updatedJobId, err := dataObject["id"].GetString()
-	if err != nil {
-		unpacked, _ := dataObject["id"].Unpack()
-
-		return nil, fmt.Errorf(
-			"%w: expected salesforce job id to be string in response, got %T",
-			ErrInvalidType,
-			unpacked,
-		)
+	jobBody := map[string]any{
+		"object":              config.ObjectName,
+		"externalIdFieldName": config.ExternalIdField,
+		"contentType":         "CSV",
+		"operation":           "upsert",
+		"lineEnding":          "LF",
 	}
 
-	completeState, err := dataObject["state"].GetString() //nolint:varnamelen
+	result, err := c.bulkOperation(ctx, BulkOperationParams(config), jobBody)
 	if err != nil {
-		unpacked, _ := resObject["state"].Unpack()
-
-		return nil, fmt.Errorf(
-			"%w: expected salesforce job state to be string in response, got %T",
-			ErrInvalidType,
-			unpacked,
-		)
+		return nil, fmt.Errorf("bulk write failed: %w", err)
 	}
 
-	return &BulkWriteResult{
-		JobId: updatedJobId,
-		State: completeState,
-	}, nil
+	return (*BulkOperationResult)(result), nil
 }
 
 func joinURLPath(baseURL string, paths ...string) (string, error) {
@@ -210,30 +145,22 @@ func joinURLPath(baseURL string, paths ...string) (string, error) {
 	return location, nil
 }
 
-func (c *Connector) createJob(ctx context.Context, config BulkWriteParams) (*common.JSONHTTPResponse, error) {
+func (c *Connector) createJob(ctx context.Context, body map[string]any) (*common.JSONHTTPResponse, error) {
 	location, err := joinURLPath(c.BaseURL, "jobs/ingest")
 	if err != nil {
 		return nil, err
 	}
 
-	body := map[string]interface{}{
-		"object":              config.ObjectName,
-		"externalIdFieldName": config.ExternalIdField,
-		"contentType":         "CSV",
-		"operation":           "upsert",
-		"lineEnding":          "LF",
-	}
-
 	return c.post(ctx, location, body)
 }
 
-func (c *Connector) uploadCSV(ctx context.Context, jobId string, config BulkWriteParams) ([]byte, error) {
+func (c *Connector) uploadCSV(ctx context.Context, jobId string, csvData io.Reader) ([]byte, error) {
 	location, err := joinURLPath(c.BaseURL, fmt.Sprintf("jobs/ingest/%s/batches", jobId))
 	if err != nil {
 		return nil, err
 	}
 
-	data, err := io.ReadAll(config.CSVData)
+	data, err := io.ReadAll(csvData)
 	if err != nil {
 		return nil, fmt.Errorf(
 			"failed to upload CSV to salesforce: %w",
@@ -300,6 +227,22 @@ func (c *Connector) GetJobResults(ctx context.Context, jobId string) (*JobResult
 	return c.getPartialFailureDetails(ctx, jobInfo)
 }
 
+func (c *Connector) GetSuccessfulJobResults(ctx context.Context, jobId string) (*http.Response, error) {
+	location, err := joinURLPath(c.BaseURL, fmt.Sprintf("jobs/ingest/%s/successfulResults", jobId))
+	if err != nil {
+		return nil, err
+	}
+
+	req, err := common.MakeJSONGetRequest(ctx, location, nil)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create get request: %w", err)
+	}
+
+	// Get the connector's JSONHTTPClient, which is a special HTTPClient that handles JSON responses,
+	// and use it's underlying http.Client to make the request.
+	return c.Client.HTTPClient.Client.Do(req)
+}
+
 func (c *Connector) getJobResults(ctx context.Context, jobId string) (*http.Response, error) {
 	location, err := joinURLPath(c.BaseURL, fmt.Sprintf("jobs/ingest/%s/failedResults", jobId))
 	if err != nil {
@@ -316,7 +259,7 @@ func (c *Connector) getJobResults(ctx context.Context, jobId string) (*http.Resp
 	return c.Client.HTTPClient.Client.Do(req)
 }
 
-//nolint:funlen
+//nolint:funlen,cyclop
 func (c *Connector) getPartialFailureDetails(ctx context.Context, jobInfo *GetJobInfoResult) (*JobResults, error) {
 	// Query Salesforce to get partial failure details
 	res, err := c.getJobResults(ctx, jobInfo.Id)
@@ -348,17 +291,27 @@ func (c *Connector) getPartialFailureDetails(ctx context.Context, jobInfo *GetJo
 			return nil, err
 		}
 
+		fieldNames := []string{sfIdFieldName, sfErrorFieldName}
+
+		if jobInfo.ExternalIdFieldName != "" {
+			fieldNames = append(fieldNames, jobInfo.ExternalIdFieldName)
+		}
+
 		// Get column index of sf__Id, sf__Error, and externalIdFieldName in header row
 		// Salesforce API responses may not be consistent with the order of columns, so we need to get the index
 		if rowIdx == 0 {
-			indiceMap, err := getColumnIndice(record, []string{sfIdFieldName, sfErrorFieldName, jobInfo.ExternalIdFieldName})
+			indiceMap, err := getColumnIndice(record, fieldNames)
 			if err != nil {
 				return nil, err
 			}
 
 			sfIdColIdx = indiceMap[sfIdFieldName]
 			sfErrorColIdx = indiceMap[sfErrorFieldName]
-			externalIdColIdx = indiceMap[jobInfo.ExternalIdFieldName]
+
+			if jobInfo.ExternalIdFieldName != "" {
+				externalIdColIdx = indiceMap[jobInfo.ExternalIdFieldName]
+			}
+
 			rowIdx++
 
 			continue
@@ -367,14 +320,26 @@ func (c *Connector) getPartialFailureDetails(ctx context.Context, jobInfo *GetJo
 		sfId := record[sfIdColIdx]
 		failureMap := failInfo.FailedUpdates
 		errMsg := record[sfErrorColIdx]
-		externalId := record[externalIdColIdx]
 
 		if sfId == "" {
 			// If sf__Id is empty, it means the record is not updated, so it's a create failure
 			failureMap = failInfo.FailedCreates
 		}
 
-		failureMap[errMsg] = append(failureMap[errMsg], externalId)
+		var referenceId string
+
+		switch jobInfo.Operation {
+		case "upsert":
+			// for bulkwrite, we will have ExternalIdFieldName
+			referenceId = record[externalIdColIdx]
+		case "delete":
+			// for bulkdelete, we will have sf__Id as reference
+			referenceId = sfId
+		default:
+			return nil, fmt.Errorf("%w: %s", ErrUnsupportedOperation, jobInfo.Operation)
+		}
+
+		failureMap[errMsg] = append(failureMap[errMsg], referenceId)
 	}
 
 	return &JobResults{
@@ -430,3 +395,167 @@ func (c *Connector) getIncompleteJobResults(jobInfo *GetJobInfoResult) *JobResul
 
 	return jobResult
 }
+
+func (c *Connector) BulkDelete(ctx context.Context, params BulkOperationParams) (*BulkOperationResult, error) {
+	jobBody := map[string]any{
+		"operation": "delete",
+		"object":    params.ObjectName,
+	}
+
+	return c.bulkOperation(ctx, params, jobBody)
+}
+
+//nolint:funlen,cyclop
+func (c *Connector) bulkOperation(
+	ctx context.Context,
+	params BulkOperationParams,
+	jobBody map[string]any,
+) (*BulkOperationResult, error) {
+	res, err := c.createJob(ctx, jobBody)
+	if err != nil {
+		return nil, fmt.Errorf("createJob failed: %w", err)
+	}
+
+	resObject, err := res.Body.GetObject()
+	if err != nil {
+		return nil, fmt.Errorf(
+			"parsing result of createJob failed: %w",
+			errors.Join(err, common.ErrParseError),
+		)
+	}
+
+	state, err := resObject["state"].GetString()
+	if err != nil {
+		unpacked, _ := resObject["state"].Unpack()
+
+		return nil, fmt.Errorf(
+			"%w: expected salesforce job state to be string in response, got %T",
+			ErrInvalidType,
+			unpacked,
+		)
+	}
+
+	if strings.ToLower(state) != "open" {
+		return nil, fmt.Errorf("%w: expected job state to be open, got %s", ErrInvalidJobState, state)
+	}
+
+	jobId, err := resObject["id"].GetString()
+	if err != nil {
+		unpacked, _ := resObject["id"].Unpack()
+
+		return nil, fmt.Errorf(
+			"%w: expected salesforce job id to be string in response, got %T",
+			ErrInvalidType,
+			unpacked)
+	}
+
+	// upload csv and there is no response body other than status code
+	_, err = c.uploadCSV(ctx, jobId, params.CSVData)
+	if err != nil {
+		return nil, fmt.Errorf("uploadCSV failed: %w", err)
+	}
+
+	data, err := c.completeUpload(ctx, jobId)
+	if err != nil {
+		return nil, fmt.Errorf("completeUpload failed: %w", err)
+	}
+
+	dataObject, err := data.Body.GetObject()
+	if err != nil {
+		return nil, fmt.Errorf("parsing result of completeUpload failed: %w", errors.Join(err, common.ErrParseError))
+	}
+
+	updatedJobId, err := dataObject["id"].GetString()
+	if err != nil {
+		unpacked, _ := dataObject["id"].Unpack()
+
+		return nil, fmt.Errorf(
+			"%w: expected salesforce job id to be string in response, got %T",
+			ErrInvalidType,
+			unpacked,
+		)
+	}
+
+	completeState, err := dataObject["state"].GetString() //nolint:varnamelen
+	if err != nil {
+		unpacked, _ := resObject["state"].Unpack()
+
+		return nil, fmt.Errorf(
+			"%w: expected salesforce job state to be string in response, got %T",
+			ErrInvalidType,
+			unpacked,
+		)
+	}
+
+	return &BulkOperationResult{
+		JobId: updatedJobId,
+		State: completeState,
+	}, nil
+}
+
+func (c *Connector) BulkQuery(
+	ctx context.Context,
+	query string,
+) (*GetJobInfoResult, error) {
+	location, err := joinURLPath(c.BaseURL, "jobs/query")
+	if err != nil {
+		return nil, err
+	}
+
+	jobBody := map[string]any{
+		"operation": "query",
+		"query":     query,
+	}
+
+	res, err := c.post(ctx, location, jobBody)
+	if err != nil {
+		return nil, fmt.Errorf("bulk query failed: %w", err)
+	}
+
+	return common.UnmarshalJSON[GetJobInfoResult](res)
+}
+
+func (c *Connector) GetBulkQueryInfo(
+	ctx context.Context,
+	jobId string,
+) (*GetJobInfoResult, error) {
+	location, err := joinURLPath(c.BaseURL, "jobs/query", jobId)
+	if err != nil {
+		return nil, err
+	}
+
+	res, err := c.get(ctx, location)
+	if err != nil {
+		return nil, fmt.Errorf(
+			"failed to get bulk query info for job '%s': %w",
+			jobId,
+			errors.Join(err, common.ErrRequestFailed),
+		)
+	}
+
+	return common.UnmarshalJSON[GetJobInfoResult](res)
+}
+
+func (c *Connector) GetBulkQueryResults(
+	ctx context.Context,
+	jobId string,
+) (*http.Response, error) {
+	location, err := joinURLPath(c.BaseURL, fmt.Sprintf("jobs/query/%s/results", jobId))
+	if err != nil {
+		return nil, err
+	}
+
+	req, err := common.MakeJSONGetRequest(ctx, location, []common.Header{
+		{
+			Key:   "Accept",
+			Value: "text/csv",
+		},
+	})
+	if err != nil {
+		return nil, fmt.Errorf("failed to get results for bulk query %s: %w", jobId, err)
+	}
+
+	// Get the connector's JSONHTTPClient, which is a special HTTPClient that handles JSON responses,
+	// and use it's underlying http.Client to make the request.
+	return c.Client.HTTPClient.Client.Do(req)
+}
diff --git a/salesforce/methods.go b/salesforce/methods.go
index 6dff500c..102e556f 100644
--- a/salesforce/methods.go
+++ b/salesforce/methods.go
@@ -45,16 +45,3 @@ func (c *Connector) patch(ctx context.Context, url string, body any) (*common.JS
 
 	return rsp, nil
 }
-
-// // get reads data from Salesforce. It handles retries and access token refreshes.
-// func (c *Connector) getCSV(ctx context.Context, url string) ([]byte, error) {
-// 	body, err := c.JSONClient.GetCSV(ctx, url, common.Header{
-// 		Key:   "Accept",
-// 		Value: "text/csv",
-// 	})
-// 	if err != nil {
-// 		return nil, c.HandleError(err)
-// 	}
-
-// 	return body, nil
-// }
diff --git a/scripts/oauth/token.go b/scripts/oauth/token.go
index fe6893d7..0ecd2fb9 100644
--- a/scripts/oauth/token.go
+++ b/scripts/oauth/token.go
@@ -242,7 +242,10 @@ func setup() *OAuthApp {
 	provider := registry.MustString("Provider")
 	clientId := registry.MustString("ClientId")
 	clientSecret := registry.MustString("ClientSecret")
-	state := registry.MustString("State")
+	state, err := registry.GetString("State")
+	if err != nil {
+		slog.Warn("no state attached, ensure that the provider doesn't require state")
+	}
 
 	scopes, err := registry.GetString("Scopes")
 	if err != nil {
diff --git a/scripts/proxy/proxy.go b/scripts/proxy/proxy.go
index 31d5a1c8..d7b65194 100644
--- a/scripts/proxy/proxy.go
+++ b/scripts/proxy/proxy.go
@@ -11,6 +11,7 @@ import (
 	"net/url"
 	"os"
 	"strings"
+	"time"
 
 	"github.com/amp-labs/connectors/basic"
 	"github.com/amp-labs/connectors/providers"
@@ -86,6 +87,16 @@ var readers = []utils.Reader{
 		JSONPath: "$['refreshToken']",
 		CredKey:  "RefreshToken",
 	},
+	&utils.JSONReader{
+		FilePath: DefaultCredsFile,
+		JSONPath: "$['expiry']",
+		CredKey:  "Expiry",
+	},
+	&utils.JSONReader{
+		FilePath: DefaultCredsFile,
+		JSONPath: "$['expiryFormat']",
+		CredKey:  "ExpiryFormat",
+	},
 }
 
 func main() {
@@ -99,6 +110,8 @@ func main() {
 	clientSecret := registry.MustString("ClientSecret")
 	accessToken := registry.MustString("AccessToken")
 	refreshToken := registry.MustString("RefreshToken")
+	atExpiry := registry.MustString("Expiry")
+	atExpiryTimeFormat := registry.MustString("ExpiryFormat")
 
 	scopes, err := registry.GetString("Scopes")
 	if err != nil {
@@ -118,8 +131,40 @@ func main() {
 		substitutionsMap[key] = val.MustString()
 	}
 
+	expiry := parseAccessTokenExpiry(atExpiry, atExpiryTimeFormat)
+
 	validateRequiredFlags(provider, clientId, clientSecret)
-	startProxy(provider, oauthScopes, clientId, clientSecret, substitutionsMap, DefaultPort, accessToken, refreshToken)
+	startProxy(provider, oauthScopes, clientId, clientSecret, substitutionsMap, DefaultPort, accessToken, refreshToken, expiry)
+}
+
+func parseAccessTokenExpiry(expiryStr, timeFormat string) time.Time {
+	formatEnums := map[string]string{
+		"Layout":      time.Layout,
+		"ANSIC":       time.ANSIC,
+		"UnixDate":    time.UnixDate,
+		"RubyDate":    time.RubyDate,
+		"RFC822":      time.RFC822,
+		"RFC822Z":     time.RFC822Z,
+		"RFC850":      time.RFC850,
+		"RFC1123":     time.RFC1123,
+		"RFC1123Z":    time.RFC1123Z,
+		"RFC3339":     time.RFC3339,
+		"RFC3339Nano": time.RFC3339Nano,
+		"Kitchen":     time.Kitchen,
+	}
+
+	format, found := formatEnums[timeFormat]
+	if !found {
+		// specific format is specified instead of enum
+		format = timeFormat
+	}
+
+	expiry, err := time.Parse(format, expiryStr)
+	if err != nil {
+		panic(err)
+	}
+
+	return expiry
 }
 
 func validateRequiredFlags(provider, clientId, clientSecret string) {
@@ -130,8 +175,8 @@ func validateRequiredFlags(provider, clientId, clientSecret string) {
 	}
 }
 
-func startProxy(provider string, scopes []string, clientId, clientSecret string, substitutions map[string]string, port int, accessToken, refreshToken string) {
-	proxy := buildProxy(provider, scopes, clientId, clientSecret, substitutions, accessToken, refreshToken)
+func startProxy(provider string, scopes []string, clientId, clientSecret string, substitutions map[string]string, port int, accessToken, refreshToken string, expiry time.Time) {
+	proxy := buildProxy(provider, scopes, clientId, clientSecret, substitutions, accessToken, refreshToken, expiry)
 	http.Handle("/", proxy)
 
 	fmt.Printf("\nProxy server listening on :%d\n", port)
@@ -141,10 +186,10 @@ func startProxy(provider string, scopes []string, clientId, clientSecret string,
 	}
 }
 
-func buildProxy(provider string, scopes []string, clientId, clientSecret string, substitutions map[string]string, accessToken, refreshToken string) *Proxy {
+func buildProxy(provider string, scopes []string, clientId, clientSecret string, substitutions map[string]string, accessToken, refreshToken string, expiry time.Time) *Proxy {
 	providerInfo := getProviderConfig(provider, substitutions)
 	cfg := configureOAuth(clientId, clientSecret, scopes, providerInfo)
-	httpClient := setupHttpClient(cfg, accessToken, refreshToken, provider, providerInfo)
+	httpClient := setupHttpClient(cfg, accessToken, refreshToken, provider, expiry, providerInfo)
 
 	target, err := url.Parse(providerInfo.BaseURL)
 	if err != nil {
@@ -177,12 +222,16 @@ func configureOAuth(clientId, clientSecret string, scopes []string, providerInfo
 }
 
 // This helps with refreshing tokens automatically.
-func setupHttpClient(cfg *oauth2.Config, accessToken, refreshToken string, provider string, providerInfo *providers.ProviderInfo) *http.Client {
+func setupHttpClient(cfg *oauth2.Config, accessToken, refreshToken, provider string, expiry time.Time, providerInfo *providers.ProviderInfo) *http.Client {
 	ctx := context.Background()
 
 	conn, err := basic.NewConnector(
 		provider,
-		basic.WithClient(ctx, http.DefaultClient, cfg, &oauth2.Token{AccessToken: accessToken, RefreshToken: refreshToken}),
+		basic.WithClient(ctx, http.DefaultClient, cfg, &oauth2.Token{
+			AccessToken:  accessToken,
+			RefreshToken: refreshToken,
+			Expiry:       expiry, // will trigger reuse of refresh token
+		}),
 	)
 	if err != nil {
 		panic(err)
diff --git a/utils/format.go b/utils/format.go
new file mode 100644
index 00000000..31e053aa
--- /dev/null
+++ b/utils/format.go
@@ -0,0 +1,12 @@
+package utils
+
+import "encoding/json"
+
+func PrettyFormatStruct(s any) (string, error) {
+	json, err := json.MarshalIndent(s, "", "  ")
+	if err != nil {
+		return "", err
+	}
+
+	return string(json), nil
+}


Test patch:
diff --git a/providers/catalog_test.go b/providers/catalog_test.go
index d80c6136..cfc1cbb6 100644
--- a/providers/catalog_test.go
+++ b/providers/catalog_test.go
@@ -180,6 +180,28 @@ var testCases = []struct { // nolint
 		expectedErr: nil,
 	},
 
+	{
+		provider: Capsule,
+		expected: &ProviderInfo{
+			AuthType: Oauth2,
+			OauthOpts: OauthOpts{
+				AuthURL:                   "https://api.capsulecrm.com/oauth/authorise",
+				TokenURL:                  "https://api.capsulecrm.com/oauth/token",
+				ExplicitScopesRequired:    true,
+				ExplicitWorkspaceRequired: false,
+			},
+			Support: Support{
+				BulkWrite: false,
+				Proxy:     false,
+				Read:      false,
+				Subscribe: false,
+				Write:     false,
+			},
+			BaseURL: "https://api.capsulecrm.com/api",
+		},
+		expectedErr: nil,
+	},
+
 	{
 		provider:    Copper,
 		description: "Valid Copper provider config with no substitutions",
@@ -645,6 +667,31 @@ var testCases = []struct { // nolint
 		},
 		expectedErr: nil,
 	},
+	{
+		provider:    Gainsight,
+		description: "Gainsight config with substitutions",
+		substitutions: map[string]string{
+			"workspace": "company",
+		},
+		expected: &ProviderInfo{
+			Support: Support{
+				Read:      false,
+				Write:     false,
+				BulkWrite: false,
+				Subscribe: false,
+				Proxy:     false,
+			},
+			AuthType: Oauth2,
+			OauthOpts: OauthOpts{
+				AuthURL:                   "https://company.gainsightcloud.com/v1/authorize",
+				TokenURL:                  "https://company.gainsightcloud.com/v1/users/oauth/token",
+				ExplicitScopesRequired:    false,
+				ExplicitWorkspaceRequired: true,
+			},
+			BaseURL: "https://company.gainsightcloud.com",
+		},
+		expectedErr: nil,
+	},
 }
 
 func TestReadInfo(t *testing.T) { // nolint
diff --git a/test/salesforce/bulkdelete/main.go b/test/salesforce/bulkdelete/main.go
new file mode 100644
index 00000000..ea3b219c
--- /dev/null
+++ b/test/salesforce/bulkdelete/main.go
@@ -0,0 +1,240 @@
+package main
+
+import (
+	"bytes"
+	"context"
+	"encoding/csv"
+	"fmt"
+	"io"
+	"log"
+	"log/slog"
+	"os"
+	"strings"
+	"time"
+
+	"github.com/amp-labs/connectors/salesforce"
+	testUtils "github.com/amp-labs/connectors/test/utils"
+	"github.com/amp-labs/connectors/utils"
+)
+
+func main() {
+	handler := slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{
+		Level: slog.LevelDebug,
+	})
+
+	logger := slog.New(handler)
+	slog.SetDefault(logger)
+
+	ctx := context.Background()
+
+	sfc, err := testUtils.Connector(ctx)
+	if err != nil {
+		slog.Error("Error creating Salesforce connector", "error", err)
+
+		return
+	}
+
+	defer func() {
+		_ = sfc.Close()
+	}()
+
+	// We first create objects in Salesforce,
+	// and then we generate an in-memory CSV of the Salesforce IDs of the newly created objects,
+	// so that we can bulk-delete them."
+	objectCSVToDelete, err := prepareObjectsToDelete(ctx, sfc)
+	if err != nil {
+		slog.Error("Error creating file to delete", "error", err)
+		return
+	}
+
+	deleteRes, err := sfc.BulkDelete(ctx, salesforce.BulkOperationParams{
+		ObjectName: "Touchpoint__c",
+		CSVData:    bytes.NewReader(objectCSVToDelete),
+	})
+
+	if err != nil {
+		slog.Error("Error bulk deleting", "error", err)
+		return
+	}
+
+	slog.Info("Bulk delete job created", "res", deleteRes)
+
+	// Get delete results. waits for the job to complete
+	deleteResult, err := getResultInLoop(ctx, sfc, deleteRes.JobId)
+	if err != nil {
+		slog.Error("Error getting bulk delete job results", "error", err)
+		return
+	}
+
+	slog.Info("Bulk delete job done")
+
+	prettyPrint(deleteResult)
+}
+
+func isJobDone(jobRes *salesforce.JobResults) bool {
+	return jobRes.State == salesforce.JobStateComplete || jobRes.State == salesforce.JobStateFailed || jobRes.State == salesforce.JobStateAborted
+}
+
+func prettyPrint(s any) {
+	fmt.Println(utils.PrettyFormatStruct(s))
+}
+
+func getResultInLoop(ctx context.Context, sfc *salesforce.Connector, jobId string) (*salesforce.JobResults, error) {
+	done := false
+	var jobRes *salesforce.JobResults
+	var err error
+
+	for !done {
+		fmt.Print(".")
+		time.Sleep(2 * time.Second)
+
+		jobRes, err = sfc.GetJobResults(ctx, jobId)
+		if err != nil {
+			return nil, fmt.Errorf("error getting job results: %w", err)
+		}
+
+		done = isJobDone(jobRes)
+	}
+
+	return jobRes, nil
+}
+
+func csvBytesToSlice(b []byte) ([]string, [][]string, error) {
+	reader := csv.NewReader(bytes.NewBuffer(b))
+
+	records := make([][]string, 0)
+
+	headers, err := reader.Read()
+	if err != nil {
+		return nil, nil, fmt.Errorf("error reading CSV headers: %w", err)
+	}
+
+	for {
+		row, err := reader.Read()
+		if err != nil {
+			if err == io.EOF {
+				break
+			}
+
+			return nil, nil, fmt.Errorf("error reading CSV row: %w", err)
+		}
+
+		records = append(records, row)
+	}
+
+	return headers, records, nil
+}
+
+func prepareObjectsToDelete(ctx context.Context, sfc *salesforce.Connector) ([]byte, error) {
+	testFilePath := "./test/salesforce/bulkdelete/touchpoints_for_bulkdelete_20240325.csv"
+
+	fileToWrite, err := os.Open(testFilePath)
+	if err != nil {
+		return nil, fmt.Errorf("error opening file: %w", err)
+	}
+
+	defer func() {
+		if fileToWrite != nil {
+			if closeErr := fileToWrite.Close(); closeErr != nil {
+				slog.Warn("unable to close file", "error", closeErr)
+			}
+		}
+	}()
+
+	// Write the records to Salesforce, so that we can delete them later.
+	writeRes, err := sfc.BulkWrite(ctx, salesforce.BulkOperationParams{
+		ObjectName:      "Touchpoint__c",
+		ExternalIdField: "external_id__c",
+		CSVData:         fileToWrite,
+		Mode:            "upsert",
+	})
+	if err != nil {
+		return nil, fmt.Errorf("error bulk writing to prepare bulk delete: %w", err)
+	}
+
+	slog.Info("Preparing objects to delete", "res", writeRes)
+
+	// wait for the job to complete
+	_, err = getResultInLoop(ctx, sfc, writeRes.JobId)
+	if err != nil {
+		return nil, fmt.Errorf("error getting bulk write job results: %w", err)
+	}
+
+	slog.Info("Records created, now deleting them.")
+
+	return getIdsFromJobToDelete(ctx, sfc, writeRes.JobId)
+}
+
+func getIdsFromJobToDelete(ctx context.Context, sfc *salesforce.Connector, jobId string) ([]byte, error) {
+	// Get the successful results to get the ids to use for the delete
+	successRes, err := sfc.GetSuccessfulJobResults(ctx, jobId)
+	if err != nil {
+		return nil, fmt.Errorf("error getting successfult write results: %w", err)
+	}
+
+	successBody, err := io.ReadAll(successRes.Body)
+	if err != nil {
+		return nil, fmt.Errorf("error reading success results body: %w", err)
+	}
+
+	defer func() {
+		if successRes != nil && successRes.Body != nil {
+			if closeErr := successRes.Body.Close(); closeErr != nil {
+				slog.Warn("unable to close response body", "error", closeErr)
+			}
+		}
+	}()
+
+	headers, rows, err := csvBytesToSlice(successBody)
+	if err != nil {
+		return nil, fmt.Errorf("error parsing CSV: %w", err)
+	}
+
+	// remove all columns except the id column
+	csvRecords, err := filterIds(headers, rows)
+	if err != nil {
+		return nil, fmt.Errorf("error filtering ids: %w", err)
+	}
+
+	var b []byte
+	buf := bytes.NewBuffer(b)
+	w := csv.NewWriter(buf)
+
+	for _, record := range csvRecords {
+		if err := w.Write(record); err != nil {
+			log.Fatalln("error writing record to csv:", err)
+		}
+	}
+
+	w.Flush()
+
+	if err := w.Error(); err != nil {
+		log.Fatalln(err)
+	}
+
+	return buf.Bytes(), nil
+}
+
+func filterIds(headers []string, rows [][]string) ([][]string, error) {
+	// filter ids
+	idIndex := -1
+
+	for i, header := range headers {
+		if header == "sf__Id" {
+			idIndex = i
+			break
+		}
+	}
+
+	if idIndex == -1 {
+		return nil, fmt.Errorf("sf__Id not found in successfulResults headers")
+	}
+
+	csvRecords := [][]string{{"id"}}
+
+	for _, row := range rows {
+		csvRecords = append(csvRecords, []string{strings.Trim(row[idIndex], " ")})
+	}
+
+	return csvRecords, nil
+}
diff --git a/test/salesforce/bulkdelete/touchpoints_for_bulkdelete_20240325.csv b/test/salesforce/bulkdelete/touchpoints_for_bulkdelete_20240325.csv
new file mode 100644
index 00000000..880426df
--- /dev/null
+++ b/test/salesforce/bulkdelete/touchpoints_for_bulkdelete_20240325.csv
@@ -0,0 +1,11 @@
+ContactId__c,Date__c,Description__c,External_Id__c,OwnerId,Name,Type__c
+003Dp00000I8z3pIAB,2023-11-20T11:36:22-08:00,Description-0,tobedeleted-0,005Dp000003Cd1SIAS,TouchPoint-0,phone
+003Dp00000I8z3pIAB,2023-11-20T11:36:22-08:00,Description-1,tobedeleted-1,005Dp000003Cd1SIAS,TouchPoint-1,email
+003Dp00000I8z3pIAB,2023-11-20T11:36:22-08:00,Description-2,tobedeleted-2,005Dp000003Cd1SIAS,TouchPoint-2,meeting
+003Dp00000I8z3pIAB,2023-11-20T11:36:22-08:00,Description-3,tobedeleted-3,005Dp000003Cd1SIAS,TouchPoint-3,phone
+003Dp00000I8z3pIAB,2023-11-20T11:36:22-08:00,Description-4,tobedeleted-4,005Dp000003Cd1SIAS,TouchPoint-4,email
+003Dp00000I8z3pIAB,2023-11-20T11:36:22-08:00,Description-5,tobedeleted-5,005Dp000003Cd1SIAS,TouchPoint-5,meeting
+003Dp00000I8z3pIAB,2023-11-20T11:36:22-08:00,Description-6,tobedeleted-6,005Dp000003Cd1SIAS,TouchPoint-6,phone
+003Dp00000I8z3pIAB,2023-11-20T11:36:22-08:00,Description-7,tobedeleted-7,005Dp000003Cd1SIAS,TouchPoint-7,email
+003Dp00000I8z3pIAB,2023-11-20T11:36:22-08:00,Description-8,tobedeleted-8,005Dp000003Cd1SIAS,TouchPoint-8,meeting
+003Dp00000I8z3pIAB,2023-11-20T11:36:22-08:00,Description-9,tobedeleted-9,005Dp000003Cd1SIAS,TouchPoint-9,phone
diff --git a/test/salesforce/bulkquery/main.go b/test/salesforce/bulkquery/main.go
new file mode 100644
index 00000000..e5278fa2
--- /dev/null
+++ b/test/salesforce/bulkquery/main.go
@@ -0,0 +1,108 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"log/slog"
+	"os"
+	"time"
+
+	"github.com/amp-labs/connectors/salesforce"
+	testUtils "github.com/amp-labs/connectors/test/utils"
+	"github.com/amp-labs/connectors/utils"
+)
+
+func main() {
+	handler := slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{
+		Level: slog.LevelDebug,
+	})
+
+	logger := slog.New(handler)
+	slog.SetDefault(logger)
+
+	ctx := context.Background()
+
+	sfc, err := testUtils.Connector(ctx)
+	if err != nil {
+		slog.Error("Error creating Salesforce connector", "error", err)
+
+		return
+	}
+
+	defer func() {
+		_ = sfc.Close()
+	}()
+
+	query := "SELECT Id, Name FROM Account LIMIT 10"
+
+	res, err := sfc.BulkQuery(ctx, query)
+	if err != nil {
+		slog.Error("Error querying", "error", err)
+		return
+	}
+
+	slog.Info("Query Info")
+	prettyPrint(res)
+
+	_, err = getResultInLoop(ctx, sfc, res.Id)
+	if err != nil {
+		slog.Error("Error getting job results", "error", err)
+		return
+	}
+
+	slog.Info("Job completed... fetching results")
+
+	// Get the results
+	result, err := sfc.GetBulkQueryResults(ctx, res.Id)
+	if err != nil {
+		slog.Error("Error getting query results", "error", err)
+		return
+	}
+
+	body, err := io.ReadAll(result.Body)
+	if err != nil {
+		slog.Error("Error reading response body", "error", err)
+		return
+	}
+
+	defer func() {
+		if res != nil && result.Body != nil {
+			if closeErr := result.Body.Close(); closeErr != nil {
+				slog.Warn("unable to close response body", "error", closeErr)
+			}
+		}
+	}()
+
+	slog.Info("Query results")
+	fmt.Println(string(body))
+}
+
+func getResultInLoop(ctx context.Context, sfc *salesforce.Connector, jobId string) (*salesforce.GetJobInfoResult, error) {
+	done := false
+	var jobRes *salesforce.GetJobInfoResult
+	var err error
+
+	for !done {
+		fmt.Print(".")
+		time.Sleep(2 * time.Second)
+
+		jobRes, err = sfc.GetBulkQueryInfo(ctx, jobId)
+		if err != nil {
+			return nil, fmt.Errorf("error getting job results: %w", err)
+		}
+
+		done = isJobDone(jobRes)
+	}
+	fmt.Println(".")
+
+	return jobRes, nil
+}
+
+func isJobDone(jobRes *salesforce.GetJobInfoResult) bool {
+	return jobRes.State == salesforce.JobStateComplete || jobRes.State == salesforce.JobStateFailed || jobRes.State == salesforce.JobStateAborted
+}
+
+func prettyPrint(s any) {
+	fmt.Println(utils.PrettyFormatStruct(s))
+}
diff --git a/test/salesforce/bulkwrite/main.go b/test/salesforce/bulkwrite/main.go
index bfca2ddc..50ee06e0 100644
--- a/test/salesforce/bulkwrite/main.go
+++ b/test/salesforce/bulkwrite/main.go
@@ -5,15 +5,12 @@ import (
 	"encoding/json"
 	"fmt"
 	"log/slog"
-	"net/http"
 	"os"
 	"sync"
 	"time"
 
-	"github.com/amp-labs/connectors"
 	"github.com/amp-labs/connectors/salesforce"
 	testUtils "github.com/amp-labs/connectors/test/utils"
-	"github.com/amp-labs/connectors/utils"
 )
 
 const (
@@ -30,25 +27,9 @@ func main() { //nolint:funlen
 	logger := slog.New(handler)
 	slog.SetDefault(logger)
 
-	// assumes that this code is being run from the root of the project
-	// go run test/salesforce/bulkwrite/main.go
-	filePath := os.Getenv("SALESFORCE_CRED_FILE_PATH")
-	if filePath == "" {
-		filePath = "./salesforce-creds.json"
-	}
-
-	ampConnectionSchemaReader := testUtils.JSONFileReaders(filePath)
-	credentialsRegistry := utils.NewCredentialsRegistry()
-	credentialsRegistry.AddReaders(ampConnectionSchemaReader...)
-	salesforceWorkspace := credentialsRegistry.MustString(utils.WorkspaceRef)
-
-	cfg := utils.SalesforceOAuthConfigFromRegistry(credentialsRegistry)
-	tok := utils.SalesforceOauthTokenFromRegistry(credentialsRegistry)
 	ctx := context.Background()
 
-	sfc, err := connectors.Salesforce(
-		salesforce.WithClient(ctx, http.DefaultClient, cfg, tok),
-		salesforce.WithWorkspace(salesforceWorkspace))
+	sfc, err := testUtils.Connector(ctx)
 	if err != nil {
 		slog.Error("Error creating Salesforce connector", "error", err)
 
@@ -90,7 +71,7 @@ func testBulkWrite(ctx context.Context, sfc *salesforce.Connector, filePath stri
 		return "", fmt.Errorf("error opening '%s': %w", filePath, err)
 	}
 
-	res, err := sfc.BulkWrite(ctx, salesforce.BulkWriteParams{
+	res, err := sfc.BulkWrite(ctx, salesforce.BulkOperationParams{
 		ObjectName:      "Touchpoint__c",
 		ExternalIdField: "external_id__c",
 		CSVData:         file,
@@ -162,7 +143,7 @@ func testGetJobResultsForFile(ctx context.Context, sfc *salesforce.Connector, fi
 		return "", fmt.Errorf("error opening file: %w", err)
 	}
 
-	res, err := sfc.BulkWrite(ctx, salesforce.BulkWriteParams{
+	res, err := sfc.BulkWrite(ctx, salesforce.BulkOperationParams{
 		ObjectName:      "Touchpoint__c",
 		ExternalIdField: "external_id__c",
 		CSVData:         file,
diff --git a/test/utils/connector.go b/test/utils/connector.go
new file mode 100644
index 00000000..143e8b28
--- /dev/null
+++ b/test/utils/connector.go
@@ -0,0 +1,32 @@
+package utils
+
+import (
+	"context"
+	"net/http"
+	"os"
+
+	"github.com/amp-labs/connectors"
+	"github.com/amp-labs/connectors/salesforce"
+	"github.com/amp-labs/connectors/utils"
+)
+
+func Connector(ctx context.Context) (*salesforce.Connector, error) {
+	// assumes that this code is being run from the root of the project
+	// go run test/salesforce/bulkwrite/main.go
+	filePath := os.Getenv("SALESFORCE_CRED_FILE_PATH")
+	if filePath == "" {
+		filePath = "./salesforce-creds.json"
+	}
+
+	ampConnectionSchemaReader := JSONFileReaders(filePath)
+	credentialsRegistry := utils.NewCredentialsRegistry()
+	credentialsRegistry.AddReaders(ampConnectionSchemaReader...)
+	salesforceWorkspace := credentialsRegistry.MustString(utils.WorkspaceRef)
+
+	cfg := utils.SalesforceOAuthConfigFromRegistry(credentialsRegistry)
+	tok := utils.SalesforceOauthTokenFromRegistry(credentialsRegistry)
+
+	return connectors.Salesforce(
+		salesforce.WithClient(ctx, http.DefaultClient, cfg, tok),
+		salesforce.WithWorkspace(salesforceWorkspace))
+}