Repository: amp-labs/connectors
Pull request #1402 (2025-02-26 01:52:19)

[CON-1526]fix(zendesk): Incremental objects diff write URLs
# Fix

Issues discovered during integration testing:
* Forgot to update write URL endpoints. Incremental endpoints are unique for Read not Write/Delete.
* `start_time` cannot be less than 1 minute old. If it is then Since parameter is adapted to avoid an error.

Cobalt0s: * **#1402** <a href="https://app.graphite.dev/github/pr/amp-labs/connectors/1402?utm_source=stack-comment-icon" target="_blank"><img src="https://static.graphite.dev/graphite-32x32-black.png" alt="Graphite" width="10px" height="10px"/></a> 👈 <a href="https://app.graphite.dev/github/pr/amp-labs/connectors/1402?utm_source=stack-comment-view-in-graphite" target="_blank">(View in Graphite)</a>
* `main`


<h2></h2>

This stack of pull requests is managed by <a href="https://graphite.dev?utm-source=stack-comment"><b>Graphite</b></a>. Learn more about <a href="https://stacking.dev/?utm_source=stack-comment">stacking</a>.
<!-- Current dependencies on/for this PR: -->

Patch:
diff --git a/common/types.go b/common/types.go
index 0623739d..5a9d5b6a 100644
--- a/common/types.go
+++ b/common/types.go
@@ -431,7 +431,7 @@ const (
 	// successful registration.
 	RegistrationStatusSuccess RegistrationStatus = "success"
 	// registration returned error, and failed to rollback some intermittent steps.
-	RegistrationStatusError RegistrationStatus = "error"
+	RegistrationStatusFailedToRollback RegistrationStatus = "failed_to_rollback"
 )
 
 type SubscriptionRegistrationParams struct {
@@ -452,25 +452,41 @@ type ObjectEvents struct {
 type ObjectName string
 
 type SubscribeParams struct {
-	RegistrationRef    string // optional, needed for some providers like Hubspot
+	Request any
+	// RegistrationResult is the result of the Connector.Register call.
+	// Connector.Subscribe requires information from the registration.
+	// Not all providers require registration, so this is optional.
+	// eg) Salesforce and HubSpot require registration because
+	RegistrationResult *RegistrationResult
 	SubscriptionEvents map[ObjectName]ObjectEvents
-	TargetURL          string // optional
 }
 
 type SubscriptionResult struct { // this corresponds to each API call.
-	RegistrationRef string
-	SubscriptionRef string
-	Result          any
-	Objects         []ObjectName
-	Events          []SubscriptionEventType
+	Result  any
+	Objects []ObjectName
+	Events  []SubscriptionEventType
 	// ["create", "update", "delete"]
 	// our regular CRUD operation events we translate to provider-specific names contact.creation
 	UpdateFields []string
 	// ["email", "fax"]
 	PassThroughEvents []string
 	// provider specific events ["contact.merged"] for hubspot or ["jira_issue:restored", "jira_issue:archived"] for jira.
+	Status SubscriptionStatus
 }
 
+type SubscriptionStatus string
+
+const (
+	// registration is pending and not yet complete.
+	SubscriptionStatusPending SubscriptionStatus = "pending"
+	// registration returned error, and all intermittent steps are rolled back.
+	SubscriptionStatusFailed SubscriptionStatus = "failed"
+	// successful registration.
+	SubscriptionStatusSuccess SubscriptionStatus = "success"
+	// 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
diff --git a/providers/salesforce/bulk-delete.go b/providers/salesforce/bulk-delete.go
index 89630dad..daa322f2 100644
--- a/providers/salesforce/bulk-delete.go
+++ b/providers/salesforce/bulk-delete.go
@@ -13,6 +13,8 @@ import (
 // * GetJobInfo
 // * GetJobResults
 // * GetSuccessfulJobResults.
+//
+//nolint:stylecheck
 func (c *Connector) BulkDelete(ctx context.Context, params BulkOperationParams) (*BulkOperationResult, error) {
 	if len(params.ObjectName) == 0 {
 		return nil, common.ErrMissingObjects
diff --git a/providers/salesforce/register.go b/providers/salesforce/register.go
index d91478c1..2c9803d7 100644
--- a/providers/salesforce/register.go
+++ b/providers/salesforce/register.go
@@ -61,7 +61,7 @@ func (c *Connector) rollbackRegister(ctx context.Context, res *ResultData) error
 
 func (c *Connector) Register(
 	ctx context.Context,
-	params *common.SubscriptionRegistrationParams,
+	params common.SubscriptionRegistrationParams,
 ) (*common.RegistrationResult, error) {
 	validate := validator.New()
 	if err := validate.Struct(params); err != nil {
@@ -82,7 +82,7 @@ func (c *Connector) Register(
 	if err != nil {
 		if rollbackErr := c.rollbackRegister(ctx, result); rollbackErr != nil {
 			return &common.RegistrationResult{
-				Status: common.RegistrationStatusError,
+				Status: common.RegistrationStatusFailedToRollback,
 			}, errors.Join(rollbackErr, err)
 		}
 
@@ -227,7 +227,7 @@ func GetChangeDataCaptureEventName(objName string) string {
 		return GetRawObjectName(objName) + "__ChangeEvent"
 	}
 
-	return objName + "%ChangeEvent"
+	return objName + "ChangeEvent"
 }
 
 func GetChannelName(rawChannelName string) string {
diff --git a/providers/salesforce/subscribe.go b/providers/salesforce/subscribe.go
new file mode 100644
index 00000000..acc41b41
--- /dev/null
+++ b/providers/salesforce/subscribe.go
@@ -0,0 +1,161 @@
+package salesforce
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"github.com/amp-labs/connectors/common"
+	"github.com/go-playground/validator"
+)
+
+type SubscribeResult struct {
+	EventChannelMembers map[common.ObjectName]*EventChannelMember
+}
+
+// Subscribe subscribes to the events for the given objects.
+// It creates event channel members for each object in the subscription.
+// If any of the event channel members fail to be created, it will rollback the operation.
+// If the rollback fails, it will return the partial result along with the error.
+// If the rollback is successful, it will return the original error on object.
+// Registration is required prior to subscribing.
+//
+//nolint:funlen,cyclop
+func (conn *Connector) Subscribe(
+	ctx context.Context,
+	params common.SubscribeParams,
+) (*common.SubscriptionResult, error) {
+	if params.RegistrationResult == nil {
+		return nil, fmt.Errorf("%w: missing RegistrationResult", errMissingParams)
+	}
+
+	if params.RegistrationResult.Result == nil {
+		return nil, fmt.Errorf("%w: missing RegistrationResult.Result", errMissingParams)
+	}
+
+	validate := validator.New()
+	if err := validate.Struct(params); err != nil {
+		return nil, fmt.Errorf("invalid registration result: %w", err)
+	}
+
+	regstrationParams, 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,
+			params.RegistrationResult.Result,
+		)
+	}
+
+	sfRes := &SubscribeResult{
+		EventChannelMembers: make(map[common.ObjectName]*EventChannelMember),
+	}
+
+	var failError error
+
+	for objName := range params.SubscriptionEvents {
+		eventName := GetChangeDataCaptureEventName(string(objName))
+		rawChannelName := GetRawChannelNameFromChannel(regstrationParams.EventChannel.FullName)
+
+		channelMetadata := &EventChannelMemberMetadata{
+			EventChannel:   GetChannelName(rawChannelName),
+			SelectedEntity: eventName,
+		}
+		channelMember := &EventChannelMember{
+			FullName: GetChangeDataCaptureChannelMembershipName(rawChannelName, eventName),
+			Metadata: channelMetadata,
+		}
+
+		newChannelMember, err := conn.CreateEventChannelMember(ctx, channelMember)
+		if err != nil {
+			failError = fmt.Errorf("failed to create event channel member for object %s, %w", objName, err)
+
+			break
+		}
+
+		sfRes.EventChannelMembers[objName] = newChannelMember
+	}
+
+	res := &common.SubscriptionResult{
+		// Salesforce is all or nothing for an object,
+		// so if successful, we will subscribe to all events.
+		Events: []common.SubscriptionEventType{
+			common.SubscriptionEventTypeCreate,
+			common.SubscriptionEventTypeUpdate,
+			common.SubscriptionEventTypeDelete,
+		},
+	}
+
+	var rollbackError error
+
+	if failError != nil {
+		for objName, member := range sfRes.EventChannelMembers {
+			if _, err := conn.DeleteEventChannelMember(ctx, member.Id); err != nil {
+				rollbackError = errors.Join(
+					rollbackError,
+					fmt.Errorf("failed to delete event channel member for object %s: %w",
+						objName,
+						err,
+					),
+				)
+			} else {
+				// remove the object from the map
+				delete(sfRes.EventChannelMembers, objName)
+			}
+		}
+
+		if rollbackError != nil {
+			res.Status = common.SubscriptionStatusFailedToRollback
+
+			for objName := range sfRes.EventChannelMembers {
+				res.Objects = append(res.Objects, objName)
+			}
+
+			res.Result = sfRes
+
+			// we still return the partial result along with the error
+			return res, errors.Join(failError, rollbackError)
+		}
+
+		res.Events = nil
+		res.Status = common.SubscriptionStatusFailed
+		res.Result = sfRes
+
+		return res, failError
+	}
+
+	res.Status = common.SubscriptionStatusSuccess
+	res.Result = sfRes
+
+	for objName := range sfRes.EventChannelMembers {
+		res.Objects = append(res.Objects, objName)
+	}
+
+	return res, nil
+}
+
+// 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 {
+	if params.Result == nil {
+		return fmt.Errorf("%w: missing SubscriptionResult.Result", errMissingParams)
+	}
+
+	sfRes, ok := params.Result.(*SubscribeResult)
+	if !ok {
+		return fmt.Errorf(
+			"%w: expected SubscriptionResult.Result to be type '%T', but got '%T'",
+			errInvalidRequestType,
+			sfRes,
+			params.Result,
+		)
+	}
+
+	for objectName, member := range sfRes.EventChannelMembers {
+		if _, err := conn.DeleteEventChannelMember(ctx, member.Id); err != nil {
+			return fmt.Errorf("failed to delete event channel member '%s': %w", objectName, err)
+		}
+	}
+
+	return nil
+}
diff --git a/providers/zendesksupport/connector.go b/providers/zendesksupport/connector.go
index 3367ae50..40ca5c77 100644
--- a/providers/zendesksupport/connector.go
+++ b/providers/zendesksupport/connector.go
@@ -53,7 +53,7 @@ func (c *Connector) String() string {
 	return c.Provider() + ".Connector"
 }
 
-func (c *Connector) getURL(objectName string) (*urlbuilder.URL, error) {
+func (c *Connector) getReadURL(objectName string) (*urlbuilder.URL, error) {
 	path, err := metadata.Schemas.LookupURLPath(c.Module.ID, objectName)
 	if err != nil {
 		return nil, err
@@ -62,6 +62,19 @@ func (c *Connector) getURL(objectName string) (*urlbuilder.URL, error) {
 	return urlbuilder.New(c.BaseURL, path)
 }
 
+func (c *Connector) getWriteURL(objectName string) (*urlbuilder.URL, error) {
+	if objectsUnsupportedWrite[c.Module.ID].Has(objectName) {
+		return nil, common.ErrOperationNotSupportedForObject
+	}
+
+	if path, ok := writeURLExceptions[c.Module.ID][objectName]; ok {
+		// URL for write differs from read.
+		return urlbuilder.New(c.BaseURL, path)
+	}
+
+	return c.getReadURL(objectName)
+}
+
 func (c *Connector) setBaseURL(newURL string) {
 	c.BaseURL = newURL
 	c.Client.HTTPClient.Base = newURL
diff --git a/providers/zendesksupport/delete.go b/providers/zendesksupport/delete.go
index 9a5ba26c..ae8e8ed7 100644
--- a/providers/zendesksupport/delete.go
+++ b/providers/zendesksupport/delete.go
@@ -11,7 +11,7 @@ func (c *Connector) Delete(ctx context.Context, config common.DeleteParams) (*co
 		return nil, err
 	}
 
-	url, err := c.getURL(config.ObjectName)
+	url, err := c.getWriteURL(config.ObjectName)
 	if err != nil {
 		return nil, err
 	}
diff --git a/providers/zendesksupport/objectNames.go b/providers/zendesksupport/objectNames.go
index 650060d7..44e42600 100644
--- a/providers/zendesksupport/objectNames.go
+++ b/providers/zendesksupport/objectNames.go
@@ -1,6 +1,30 @@
 package zendesksupport
 
-import "github.com/amp-labs/connectors/providers/zendesksupport/metadata"
+import (
+	"github.com/amp-labs/connectors/common"
+	"github.com/amp-labs/connectors/internal/datautils"
+	"github.com/amp-labs/connectors/providers/zendesksupport/metadata"
+)
 
 // Supported object names can be found under schemas.json.
 var supportedObjectsByRead = metadata.Schemas.ObjectNames() //nolint:gochecknoglobals
+
+var objectsUnsupportedWrite = map[common.ModuleID]datautils.Set[string]{ //nolint:gochecknoglobals
+	ModuleTicketing: datautils.NewSet(
+		"attribute_values",
+		"instance_values",
+		"ticket_events",
+		"ticket_metric_events",
+	),
+	ModuleHelpCenter: datautils.NewStringSet(),
+}
+
+var writeURLExceptions = map[common.ModuleID]datautils.Map[string, string]{ //nolint:gochecknoglobals
+	ModuleTicketing: {
+		"attributes":    "/api/v2/routing/attributes",
+		"organizations": "/api/v2/organizations",
+		"tickets":       "/api/v2/tickets",
+		"users":         "/api/v2/users",
+	},
+	ModuleHelpCenter: {},
+}
diff --git a/providers/zendesksupport/read.go b/providers/zendesksupport/read.go
index 71026a09..577989bf 100644
--- a/providers/zendesksupport/read.go
+++ b/providers/zendesksupport/read.go
@@ -3,6 +3,7 @@ package zendesksupport
 import (
 	"context"
 	"strconv"
+	"time"
 
 	"github.com/amp-labs/connectors/common"
 	"github.com/amp-labs/connectors/common/urlbuilder"
@@ -48,7 +49,7 @@ func (c *Connector) buildReadURL(config common.ReadParams) (*urlbuilder.URL, err
 	}
 
 	// First page
-	url, err := c.getURL(config.ObjectName)
+	url, err := c.getReadURL(config.ObjectName)
 	if err != nil {
 		return nil, err
 	}
@@ -59,12 +60,7 @@ func (c *Connector) buildReadURL(config common.ReadParams) (*urlbuilder.URL, err
 		// Even if no Since parameter is empty the start_time must be set to 0.
 		// This is effectively to say read everything since the beginning of time.
 		// https://developer.zendesk.com/api-reference/ticketing/ticket-management/incremental_exports/#start_time
-		startTime := "0"
-		if !config.Since.IsZero() {
-			startTime = strconv.FormatInt(config.Since.Unix(), 10)
-		}
-
-		url.WithQueryParam("start_time", startTime)
+		url.WithQueryParam("start_time", formatStartTime(config))
 		url.WithQueryParam("per_page", DefaultPageSizeStr)
 	} else {
 		// Different objects have different pagination types.
@@ -77,3 +73,19 @@ func (c *Connector) buildReadURL(config common.ReadParams) (*urlbuilder.URL, err
 
 	return url, nil
 }
+
+func formatStartTime(config common.ReadParams) string {
+	if config.Since.IsZero() {
+		return "0"
+	}
+
+	// Records cannot be requested if they are less than 1 minute old.
+	unixTime := config.Since.Unix()
+	timeWindow := time.Since(config.Since)
+
+	if timeWindow.Minutes() < 1 {
+		unixTime = time.Now().Add(-1 * time.Minute).Unix()
+	}
+
+	return strconv.FormatInt(unixTime, 10)
+}
diff --git a/providers/zendesksupport/write.go b/providers/zendesksupport/write.go
index cd58dfe0..eba07a33 100644
--- a/providers/zendesksupport/write.go
+++ b/providers/zendesksupport/write.go
@@ -15,7 +15,7 @@ func (c *Connector) Write(ctx context.Context, config common.WriteParams) (*comm
 		return nil, err
 	}
 
-	url, err := c.getURL(config.ObjectName)
+	url, err := c.getWriteURL(config.ObjectName)
 	if err != nil {
 		return nil, err
 	}


Test patch:
diff --git a/providers/zendesksupport/read_test.go b/providers/zendesksupport/read_test.go
index 48d925ca..799b4baf 100644
--- a/providers/zendesksupport/read_test.go
+++ b/providers/zendesksupport/read_test.go
@@ -153,6 +153,7 @@ func TestIncrementalReadZendeskSupportModule(t *testing.T) { //nolint:funlen,goc
 				If: mockcond.And{
 					mockcond.QueryParam("per_page", "100"),
 					mockcond.QueryParam("start_time", "0"),
+					mockcond.PathSuffix("/api/v2/incremental/tickets/cursor"),
 				},
 				Then: mockserver.Response(http.StatusOK, responseTickets),
 			}.Server(),
@@ -180,6 +181,7 @@ func TestIncrementalReadZendeskSupportModule(t *testing.T) { //nolint:funlen,goc
 				If: mockcond.And{
 					mockcond.QueryParam("per_page", "100"),
 					mockcond.QueryParam("start_time", "1726674883"),
+					mockcond.PathSuffix("/api/v2/incremental/tickets/cursor"),
 				},
 				Then: mockserver.Response(http.StatusOK, responseTickets),
 			}.Server(),
@@ -197,6 +199,7 @@ func TestIncrementalReadZendeskSupportModule(t *testing.T) { //nolint:funlen,goc
 				If: mockcond.And{
 					mockcond.QueryParam("per_page", "100"),
 					mockcond.QueryParam("start_time", "0"),
+					mockcond.PathSuffix("/api/v2/incremental/users/cursor"),
 				},
 				Then: mockserver.Response(http.StatusOK, responseUsersFirstPage),
 			}.Server(),
@@ -223,6 +226,7 @@ func TestIncrementalReadZendeskSupportModule(t *testing.T) { //nolint:funlen,goc
 				If: mockcond.And{
 					mockcond.QueryParam("per_page", "100"),
 					mockcond.QueryParam("start_time", "0"),
+					mockcond.PathSuffix("/api/v2/incremental/users/cursor"),
 				},
 				Then: mockserver.Response(http.StatusOK, responseUsersLastPage),
 			}.Server(),
@@ -249,6 +253,7 @@ func TestIncrementalReadZendeskSupportModule(t *testing.T) { //nolint:funlen,goc
 				If: mockcond.And{
 					mockcond.QueryParam("per_page", "100"),
 					mockcond.QueryParam("start_time", "0"),
+					mockcond.PathSuffix("/api/v2/incremental/organizations"),
 				},
 				Then: mockserver.Response(http.StatusOK, responseOrganizations),
 			}.Server(),
diff --git a/providers/zendesksupport/write_test.go b/providers/zendesksupport/write_test.go
index 50b33fa4..d33c3763 100644
--- a/providers/zendesksupport/write_test.go
+++ b/providers/zendesksupport/write_test.go
@@ -37,11 +37,17 @@ func TestWriteZendeskSupportModule(t *testing.T) { // nolint:funlen,cyclop
 			ExpectedErrs: []error{common.ErrMissingRecordData},
 		},
 		{
-			Name:         "Unsupported object name",
+			Name:         "URL cannot be identified for an object",
 			Input:        common.WriteParams{ObjectName: "signals", RecordData: "dummy"},
 			Server:       mockserver.Dummy(),
 			ExpectedErrs: []error{common.ErrResolvingURLPathForObject},
 		},
+		{
+			Name:         "Object is not supported for write, while it is defined for read",
+			Input:        common.WriteParams{ObjectName: "ticket_events", RecordData: "dummy"},
+			Server:       mockserver.Dummy(),
+			ExpectedErrs: []error{common.ErrOperationNotSupportedForObject},
+		},
 		{
 			Name:  "Missing write parameter",
 			Input: common.WriteParams{ObjectName: "brands", RecordData: "dummy"},
@@ -135,6 +141,23 @@ func TestWriteZendeskSupportModule(t *testing.T) { // nolint:funlen,cyclop
 			},
 			ExpectedErrs: nil,
 		},
+		{
+			Name: "Writing tickets delegates to correct URL",
+			Input: common.WriteParams{
+				ObjectName: "tickets",
+				RecordData: "dummy",
+			},
+			Server: mockserver.Conditional{
+				Setup: mockserver.ContentJSON(),
+				If: mockcond.And{
+					mockcond.MethodPOST(),
+					mockcond.PathSuffix("/api/v2/tickets"),
+				},
+				Then: mockserver.Response(http.StatusOK),
+			}.Server(),
+			Expected:     &common.WriteResult{Success: true},
+			ExpectedErrs: nil,
+		},
 	}
 
 	for _, tt := range tests {
diff --git a/test/salesforce/subscribe/register.go b/test/salesforce/subscribe/register.go
index 6cc3ab1f..cc55bb92 100644
--- a/test/salesforce/subscribe/register.go
+++ b/test/salesforce/subscribe/register.go
@@ -3,12 +3,11 @@ package main
 import (
 	"context"
 	"encoding/json"
+	"flag"
 	"fmt"
-	"os"
+	"log"
 	"os/signal"
-	"strconv"
 	"syscall"
-	"time"
 
 	"github.com/amp-labs/connectors/common"
 	"github.com/amp-labs/connectors/common/logging"
@@ -25,13 +24,22 @@ func main() {
 	// Set up slog logging.
 	utils.SetupLogging()
 
-	conn := connTest.GetSalesforceConnector(ctx)
+	var uniqueString string
+	var namedCredArn string
+
+	flag.StringVar(&uniqueString, "unique", "", "Unique string to append to the registration label")
+	flag.StringVar(&namedCredArn, "arn", "", "AWS Named Credential ARN")
+	flag.Parse()
+
+	if uniqueString == "" || namedCredArn == "" {
+		log.Fatalf("missing flags: go run <your path>.go -unique <unique string> -arn <AWS Named Credential ARN>")
+	}
 
-	uniqueString := strconv.Itoa(int(time.Now().UnixMilli()))
+	conn := connTest.GetSalesforceConnector(ctx)
 
-	arn := os.Getenv("AWS_NAMED_CRED_ARN")
+	arn := namedCredArn
 
-	params := &common.SubscriptionRegistrationParams{
+	params := common.SubscriptionRegistrationParams{
 		Request: &salesforce.RegistrationParams{
 			UniqueRef:             "Amp" + uniqueString,
 			Label:                 "Amp" + uniqueString,
@@ -47,13 +55,37 @@ func main() {
 
 	fmt.Println("Registration result:", prettyPrint(result))
 
+	subscribeParams := common.SubscribeParams{
+		RegistrationResult: result,
+		SubscriptionEvents: map[common.ObjectName]common.ObjectEvents{
+			"Account": {},
+		},
+	}
+
+	subscribeResult, err := conn.Subscribe(ctx, subscribeParams)
+	if err != nil {
+		logging.Logger(ctx).Error("Error subscribing", "error", err, "subscribeResult", prettyPrint(subscribeResult))
+	}
+
+	fmt.Println("Subscribe result:", prettyPrint(subscribeResult))
+
+	if subscribeResult != nil && subscribeResult.Status == common.SubscriptionStatusSuccess {
+		if err := conn.DeleteSubscription(ctx, *subscribeResult); err != nil {
+			logging.Logger(ctx).Error("Error unsubscribing", "error", err)
+
+			return
+		}
+	}
+
+	fmt.Println("Delete subscription successful")
+
 	if err := conn.DeleteRegistration(ctx, result); err != nil {
 		logging.Logger(ctx).Error("Error rolling back registration", "error", err)
 
 		return
 	}
 
-	fmt.Println("Delete successful")
+	fmt.Println("Delete registration successful")
 }
 
 func prettyPrint(v any) string {