salesforce implementation - phase 1 by ness-david-dedu · Pull Request #3939 · redpanda-data/connect

ness-david-dedu · 2026-01-29T14:38:41Z

No description provided.

CLAassistant · 2026-01-29T14:38:48Z

All committers have signed the CLA.

josephwoodward · 2026-02-05T14:20:54Z

internal/impl/salesforce/processor_salesforce.go

+
+func (*salesforceProcessor) Close(context.Context) error { return nil }
+
+func init() {


minor: It's a common convention to put init functions at the top of the file under constants/variables as the first function so it's lifecycle is clear to readers. You'll see the same convention used throughout other components.

internal/impl/salesforce/processor_salesforce.go

internal/impl/salesforce/salesforcehttp/http_metrics/http_metrics.go

josephwoodward · 2026-02-05T14:30:22Z

internal/impl/salesforce/salesforcehttp/client.go

+
+	query := apiUrl.Query()
+	query.Set("grant_type", "client_credentials")
+	query.Set("client_id", s.clientId)


Suggested change

query.Set("client_id", s.clientId)

query.Set("client_id", s.clientID)

Minor: Follow the Go convention of uppercase acronyms.

Also, aren't these supposed to be sent in the request-body instead of the query string?

done, moved to request body

internal/impl/salesforce/salesforcehttp/client.go

josephwoodward · 2026-02-05T14:44:19Z

internal/impl/salesforce/salesforcehttp/client.go

+// This is the general function that calls Salesforce API on a specific URL using the URL object.
+// It applies standard header parameters to all calls, Authorization, User-Agent and Accept.
+// It uses the helper functions to check against possible response codes and handling the retry-after mechanism
+func (s *Client) callSalesforceApi(ctx context.Context, u *url.URL) ([]byte, error) {


Suggested change

func (s *Client) callSalesforceApi(ctx context.Context, u *url.URL) ([]byte, error) {

func (s *Client) callSalesforceAPI(ctx context.Context, u *url.URL) ([]byte, error) {

Co-authored-by: Joseph Woodward <joseph.woodward@xeuse.com>

…ics.go Co-authored-by: Joseph Woodward <joseph.woodward@xeuse.com>

internal/impl/salesforce/salesforcehttp/client.go

internal/impl/salesforce/salesforcehttp/client_test.go

internal/impl/salesforce/salesforcehttp/salesforce_helper.go

internal/impl/salesforce/salesforcehttp/types.go

internal/impl/salesforce/processor_salesforce_test.go

josephwoodward · 2026-02-10T15:47:49Z

internal/impl/salesforce/salesforcehttp/client.go

+	retryBody, retryErr := s.doSalesforceRequest(ctx, u)
+
+	if retryErr != nil {
+		return nil, fmt.Errorf("request failed: %v", retryErr)


Given this module will change with the connector, it's probably better to wrap the error with %w instead of %v. Same for other instances I see exist in these files.

josephwoodward · 2026-02-10T20:13:59Z

internal/impl/salesforce/salesforcehttp/http_metrics/http_metrics.go

+	inflightVal int64 // track ourselves; gauge uses Set()
+	total       *service.MetricCounter
+	errors      *service.MetricCounter
+	status2xx   *service.MetricCounter


Do we use a metric per status code in other components? My understanding is it's better to have a status code as a label? It'd be worth looking at some other components capturing HTTP metrics to see if there's a pattern. This is definitely something we should look to standardise though.

changed to using a label

Co-authored-by: Joseph Woodward <joseph.woodward@xeuse.com>

josephwoodward · 2026-02-12T15:20:16Z

internal/impl/salesforce/salesforcehttp/types.go

+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// types.go defines core data structures, response models, and enums for the Jira processor.


Suggested change

// types.go defines core data structures, response models, and enums for the Jira processor.

// types.go defines core data structures, response models, and enums for the Salesforce processor.

josephwoodward · 2026-02-12T15:31:12Z

internal/impl/salesforce/salesforcehttp/client.go

+	if err != nil {
+		return nil, fmt.Errorf("invalid URL: %w", err)
+	}
+


We should remove the q=query param from the concatenated string and encode/escape it properly:

q := apiUrl.Query() q.Set("q", query) apiUrl.RawQuery = q.Encode()

josephwoodward · 2026-02-12T15:38:32Z

internal/impl/salesforce/salesforcehttp/salesforce_helper.go

+//  1. Basic request and auth check
+//
+//    ctx := context.Background()
+//    req, _ := http.NewRequestWithContext(ctx, http.MethodGet, "https:your-domain.atlassian.net/rest/api/3/myself", nil)


Looks like this is from the Jira example? We should update the documentation to a working salesforce example.

josephwoodward · 2026-02-12T15:45:54Z

internal/impl/salesforce/salesforcehttp/http_metrics/http_metrics.go

+// metrics such as in-flight requests, response status codes, errors, and request duration.
+func NewInstrumentedClient(m *service.Metrics, namespace string, client *http.Client) *http.Client {
+	if client == nil {
+		client = &http.Client{}


Is there any reason to use http.Client here instead of http.DefaultClient as you have done in DoRequestWithRetries?

josephwoodward · 2026-02-12T15:46:45Z

internal/impl/salesforce/salesforcehttp/client.go

+}
+
+// NewClient is the constructor for a Client object
+func NewClient(log *service.Logger, orgUrl, clientId, clientSecret, apiVersion string, maxRetries int, metrics *service.Metrics, httpClient *http.Client) (*Client, error) {


Suggested change

func NewClient(log *service.Logger, orgUrl, clientId, clientSecret, apiVersion string, maxRetries int, metrics *service.Metrics, httpClient *http.Client) (*Client, error) {

func NewClient(log *service.Logger, orgURL, clientID, clientSecret, apiVersion string, maxRetries int, metrics *service.Metrics, httpClient *http.Client) (*Client, error) {

Also, arguments are usually supplied in terms of importance and relevance, so something like logger and metrics would be at the end, for instance:

httpClient *http.Client, orgUrl, clientId, clientSecret, apiVersion string, maxRetries int, metrics *service.Metrics, log *service.Logger

updated the arguments order:

orgURL, clientID, clientSecret — connection identity

apiVersion, maxRetries — behavioral config

httpClient, log, metrics — infrastructure/dependencies

josephwoodward · 2026-02-12T16:08:22Z

internal/impl/salesforce/salesforcehttp/client.go

+
+	req.Header.Set("Accept", "application/json")
+	req.Header.Set("User-Agent", "Redpanda-Connect")
+	req.Header.Set("Authorization", "Bearer "+s.bearerToken)


Do we need to guard token access with a mutex to prevent data races with access to the token?

josephwoodward · 2026-02-12T16:37:12Z

internal/impl/salesforce/processor_salesforce.go

+	if err != nil {
+		return nil, err
+	}
+	s.log.Debugf("Fetching from Salesforce.. Input: %s", string(inputMsg))


I presume this will go once the processor implementation is complete?

josephwoodward · 2026-02-12T16:38:12Z

internal/impl/salesforce/processor_salesforce.go

+		return nil, err
+	}
+
+	maxRetries, err := conf.FieldInt("max_retries")


We should probably validate this for negative retries.

josephwoodward · 2026-02-12T16:44:02Z

internal/impl/salesforce/salesforcehttp/client.go

+func (s *Client) updateAndSetBearerToken(ctx context.Context) error {
+	apiUrl, err := url.Parse(s.orgURL + salesforceAPIBasePath + "/oauth2/token")
+	if err != nil {
+		return fmt.Errorf("invalid URL: %w", err)


Would be good to add a bit more context to these kinds of errors, such as "invalid token endpoint url" to make it clearer to readers what part is invalid.

josephwoodward · 2026-02-12T16:47:18Z

internal/impl/salesforce/salesforcehttp/client_test.go

+	}))
+	defer ts.Close()
+
+	// log := service.NewLogger("test")


Are these comments needed?

Jeffail · 2026-02-16T10:11:42Z

internal/impl/salesforce/salesforcehttp/client.go

+	clientID     string
+	clientSecret string
+	apiVersion   string
+	bearerToken  string


I'd maybe suggest having an atomic string here instead of a mutex since it seems to only be getted and setted rather than used in a multiple line interaction.

Jeffail · 2026-02-16T10:16:25Z

internal/impl/salesforce/salesforcehttp/salesforce_helper.go

+	attempt := 0
+
+	for {
+		resp, err := client.Do(req.WithContext(ctx))


After the first iteration this request will have already flushed its body, and so it'll be empty if this is a POST request. Instead you'll need to reconstruct the request for each iteration. A quick solution is to change the req param into a reqBuilder func() *http.Request.

Jeffail · 2026-02-16T10:17:26Z

internal/impl/salesforce/salesforcehttp/salesforce_helper.go

+			}
+		}
+
+		defer resp.Body.Close()


Since we're inside a loop I would move this below the read and make it immediate

Jeffail · 2026-02-16T10:18:34Z

internal/impl/salesforce/salesforcehttp/http_metrics/http_metrics.go

+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package http_metrics


We don't generally use underscores in package names, I would simply name this metrics.

salesforce implementation - phase 1

9f33aa4

ness-david-dedu added 2 commits February 1, 2026 23:12

add tests

cb6d0e8

fix lint

19992c0

josephwoodward reviewed Feb 5, 2026

View reviewed changes

internal/impl/salesforce/processor_salesforce.go Outdated Show resolved Hide resolved

josephwoodward reviewed Feb 5, 2026

View reviewed changes

internal/impl/salesforce/salesforcehttp/http_metrics/http_metrics.go Outdated Show resolved Hide resolved

josephwoodward reviewed Feb 5, 2026

View reviewed changes

internal/impl/salesforce/salesforcehttp/client.go Outdated Show resolved Hide resolved

josephwoodward reviewed Feb 5, 2026

View reviewed changes

ness-david-dedu and others added 6 commits February 6, 2026 15:56

Update internal/impl/salesforce/processor_salesforce.go

0d42fb6

Co-authored-by: Joseph Woodward <joseph.woodward@xeuse.com>

Update internal/impl/salesforce/salesforcehttp/client.go

815a971

Co-authored-by: Joseph Woodward <joseph.woodward@xeuse.com>

Update internal/impl/salesforce/salesforcehttp/http_metrics/http_metr…

0aac7fe

…ics.go Co-authored-by: Joseph Woodward <joseph.woodward@xeuse.com>

PR changes - rename method

f3d218c

pr change - move init method

4c25033

PR changes - update auth request

ba175b8

josephwoodward reviewed Feb 10, 2026

View reviewed changes

internal/impl/salesforce/salesforcehttp/client.go Outdated Show resolved Hide resolved

josephwoodward reviewed Feb 10, 2026

View reviewed changes

internal/impl/salesforce/salesforcehttp/client_test.go Outdated Show resolved Hide resolved

josephwoodward reviewed Feb 10, 2026

View reviewed changes

internal/impl/salesforce/salesforcehttp/salesforce_helper.go Outdated Show resolved Hide resolved

josephwoodward reviewed Feb 10, 2026

View reviewed changes

internal/impl/salesforce/salesforcehttp/types.go Outdated Show resolved Hide resolved

josephwoodward reviewed Feb 10, 2026

View reviewed changes

internal/impl/salesforce/processor_salesforce_test.go Outdated Show resolved Hide resolved

josephwoodward reviewed Feb 10, 2026

View reviewed changes

ness-david-dedu and others added 7 commits February 10, 2026 23:32

Update internal/impl/salesforce/salesforcehttp/types.go

0470914

Co-authored-by: Joseph Woodward <joseph.woodward@xeuse.com>

Update internal/impl/salesforce/processor_salesforce_test.go

cbd5d0d

Co-authored-by: Joseph Woodward <joseph.woodward@xeuse.com>

Update internal/impl/salesforce/salesforcehttp/salesforce_helper.go

6ce42fc

Co-authored-by: Joseph Woodward <joseph.woodward@xeuse.com>

Update internal/impl/salesforce/salesforcehttp/client_test.go

b0645c4

Co-authored-by: Joseph Woodward <joseph.woodward@xeuse.com>

Update internal/impl/salesforce/salesforcehttp/client.go

c99e95a

Co-authored-by: Joseph Woodward <joseph.woodward@xeuse.com>

change logging

54adbcc

change metrics to label

35905a6

josephwoodward reviewed Feb 12, 2026

View reviewed changes

pr changes

5b72f5b

Jeffail reviewed Feb 16, 2026

View reviewed changes

ness-david-dedu added 2 commits February 16, 2026 17:45

PR changes

f3a837e

fix lint

9fbdcbb


		func (*salesforceProcessor) Close(context.Context) error { return nil }

		func init() {

	query.Set("client_id", s.clientId)
	query.Set("client_id", s.clientID)

	func (s Client) callSalesforceApi(ctx context.Context, u url.URL) ([]byte, error) {
	func (s Client) callSalesforceAPI(ctx context.Context, u url.URL) ([]byte, error) {

	// types.go defines core data structures, response models, and enums for the Jira processor.
	// types.go defines core data structures, response models, and enums for the Salesforce processor.

	func NewClient(log service.Logger, orgUrl, clientId, clientSecret, apiVersion string, maxRetries int, metrics service.Metrics, httpClient http.Client) (Client, error) {
	func NewClient(log service.Logger, orgURL, clientID, clientSecret, apiVersion string, maxRetries int, metrics service.Metrics, httpClient http.Client) (Client, error) {

Conversation

ness-david-dedu commented Jan 29, 2026

Uh oh!

CLAassistant commented Jan 29, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

josephwoodward Feb 5, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

CLAassistant commented Jan 29, 2026 •

edited

Loading

josephwoodward Feb 5, 2026 •

edited

Loading