Repository: amp-labs/connectors
Pull request #1319 (2025-01-31 22:25:20)

[CON-1039] fix(Hubspot): Separate pipelines and stages
# Tests
## Deals
![image](https://github.com/user-attachments/assets/b70aacb1-e221-4038-9ede-2814189d95ee)
![image](https://github.com/user-attachments/assets/76d31f68-d76c-48ae-bb59-6becae12e623)
![image](https://github.com/user-attachments/assets/cb6bb563-c942-444a-9e65-fd8d5d22267b)

## Contacts
![image](https://github.com/user-attachments/assets/fa4fc472-4de2-4e5c-940b-c0d0b36527fb)
![image](https://github.com/user-attachments/assets/8d658332-8158-4398-8fc2-18a9565b9962)
I didn't find a field that would hold stages for contacts. By reading [this documenation](https://knowledge.hubspot.com/object-settings/set-up-and-customize-pipelines#:~:text=The%20default%20pipeline%20has%20seven,(0%25%2C%20Lost).), I have concluded that there are no such field, because there is no mention of contacts in stages paragraph:
![image](https://github.com/user-attachments/assets/aac9e8e3-8f84-4bfd-a605-522300015bf3)

# Notes
Mock tests where updated in accordance to the format specified in the ticket.
![image](https://github.com/user-attachments/assets/22218507-9828-4fe6-8dc9-2cf3f5306405)
![image](https://github.com/user-attachments/assets/d708050a-50e3-408b-bc5b-7d7a0c1eefa8)


Cobalt0s: * **#1319** <a href="https://app.graphite.dev/github/pr/amp-labs/connectors/1319?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/1319?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 15633a82..130d82d9 100644
--- a/common/types.go
+++ b/common/types.go
@@ -104,6 +104,10 @@ var (
 
 	// ErrNextPageInvalid is returned when next page token provided in Read operation cannot be understood.
 	ErrNextPageInvalid = errors.New("next page token is invalid")
+
+	// ErrInvalidImplementation is returned when implementation assumption is broken.
+	// This is not a client issue.
+	ErrInvalidImplementation = errors.New("invalid implementation")
 )
 
 // ReadParams defines how we are reading data from a SaaS API.
@@ -332,6 +336,8 @@ type FieldValue struct {
 	DisplayValue string
 }
 
+type FieldValues []FieldValue
+
 type PostAuthInfo struct {
 	CatalogVars          *map[string]string
 	RawResponse          *JSONHTTPResponse
diff --git a/providers/hubspot/metadata.go b/providers/hubspot/metadata.go
index 57b63ee9..a9a69c39 100644
--- a/providers/hubspot/metadata.go
+++ b/providers/hubspot/metadata.go
@@ -301,19 +301,21 @@ func (o fieldDescription) implyEnumerationType() (common.ValueType, []common.Fie
 // Registry of objects to the fields with external metadata.
 // It provides information on location of data (URLs) and how to process JSON to infer enum options.
 // If you want to support more objects and their fields extend this registry.
+//
+// NOTE: ResponseProcessor may retrieve values for multiple fields in a single API call.
 var objectsWithExternalMetadataFields = datautils.Map[string, []externalFieldDiscovery]{ // nolint:gochecknoglobals
 	"contacts": {
 		{
-			FieldName:         "hs_pipeline",
+			FieldNames:        []string{"hs_pipeline"},
 			EndpointPath:      "/crm/v3/pipelines/contacts",
 			ResponseProcessor: parsePipelineFieldValues,
 		},
 	},
 	"deals": {
 		{
-			FieldName:         "pipeline",
+			FieldNames:        []string{"pipeline", "dealstage"},
 			EndpointPath:      "/crm/v3/pipelines/deals",
-			ResponseProcessor: parsePipelineFieldValues,
+			ResponseProcessor: parsePipelineFieldValuesWithStages,
 		},
 	},
 }
@@ -334,79 +336,123 @@ func (c *Connector) fetchExternalMetadataEnumValues(
 	// For each external field that we support make an API call to fetch enumeration options.
 	// Store this values for each field within each object.
 	for _, discovery := range externalFields {
-		fieldMetadata, ok := fields[discovery.FieldName]
-		if !ok {
-			// Provider no longer has this field.
-			continue
-		}
-
 		rsp, err := c.Client.Get(ctx, c.getRawURL()+discovery.EndpointPath)
 		if err != nil {
 			return nil, fmt.Errorf("error resolving external metadata values for HubSpot: %w", err)
 		}
 
-		values, err := discovery.ResponseProcessor(rsp)
+		perFieldValues, err := discovery.ResponseProcessor(rsp)
 		if err != nil {
 			return nil, err
 		}
 
-		// Store discovered values.
-		fieldMetadata.Values = values
-		fields[discovery.FieldName] = fieldMetadata
+		// Response processor returns an array for each field in that order.
+		if len(perFieldValues) != len(discovery.FieldNames) {
+			return nil, common.ErrInvalidImplementation
+		}
+
+		// Store field values associated with each field.
+		for index, discoveryFieldName := range discovery.FieldNames {
+			fieldMetadata, ok := fields[discoveryFieldName]
+			if !ok {
+				// Provider no longer has this field.
+				continue
+			}
+
+			// Store discovered values.
+			fieldMetadata.Values = perFieldValues[index]
+			fields[discoveryFieldName] = fieldMetadata
+		}
 	}
 
 	return fields, nil
 }
 
-// Holds information regarding Object's Field and how to discover external metadata.
+// Represents metadata discovery details for an object's fields.
 type externalFieldDiscovery struct {
-	// FieldName that has enum options found under EndpointPath.
-	FieldName string
-	// EndpointPath is a location where list of enum values can be found.
+	// FieldNames lists the fields that have enum options available at EndpointPath.
+	// A single API response may provide values for multiple fields.
+	FieldNames []string
+	// EndpointPath specifies the API location where the list of enum values can be retrieved.
 	EndpointPath string
-	// ResponseProcessor knows how to parse an extract []common.FieldValue for the endpoint.
+	// ResponseProcessor extracts and parses []common.FieldValue for each field.
 	ResponseProcessor externalFieldProcessor
 }
 
-type externalFieldProcessor func(response *common.JSONHTTPResponse) ([]common.FieldValue, error)
-
-func parsePipelineFieldValues(response *common.JSONHTTPResponse) ([]common.FieldValue, error) {
-	type stage struct {
-		Value       string `json:"id"`
-		DisplayName string `json:"label"`
-	}
-
-	type pipeline struct {
-		Stages       []stage `json:"stages"`
-		DisplayOrder int     `json:"displayOrder"`
-	}
-
-	// For more details, refer to the HubSpot documentation on Pipelines.
-	// https://developers.hubspot.com/docs/guides/api/crm/pipelines#retrieve-pipelines
-	type pipelineResponse struct {
-		Pipelines []pipeline `json:"results"`
-	}
+// Extracts values for multiple fields from the response.
+// The number of returned lists must match externalFieldDiscovery.FieldNames, maintaining the same order.
+type externalFieldProcessor func(response *common.JSONHTTPResponse) ([]common.FieldValues, error)
 
+// Parses field values from an API response, producing two value arrays:
+//  1. The first array contains pipeline values.
+//  2. The second array contains aggregated stage values,
+//     where each stage ID is prefixed with the ID of its corresponding pipeline.
+func parsePipelineFieldValuesWithStages(response *common.JSONHTTPResponse) ([]common.FieldValues, error) {
 	resp, err := common.UnmarshalJSON[pipelineResponse](response)
 	if err != nil {
 		return nil, fmt.Errorf("error unmarshalling pipelines into JSON: %w", err)
 	}
 
+	result := make([]common.FieldValues, 2) // nolint:mnd
+
 	// Aggregate stages across all pipelines.
 	// Note: Multiple pipelines can exist for each object type, with each instance referencing a specific pipeline.
 	//
 	// Problem: In HubSpot, field options are not defined at the object level but at the instance level.
 	// This means the object identifier is required to locate the pipeline and subsequently retrieve its options.
-	result := make([]common.FieldValue, 0)
+	pipelines := make(common.FieldValues, 0)
+	stages := make(common.FieldValues, 0)
 
 	for _, pipelineItem := range resp.Pipelines {
+		pipelines = append(pipelines, common.FieldValue{
+			Value:        pipelineItem.ID,
+			DisplayValue: pipelineItem.DisplayName,
+		})
+
 		for _, s := range pipelineItem.Stages {
-			result = append(result, common.FieldValue{
-				Value:        s.Value,
+			stages = append(stages, common.FieldValue{
+				Value:        fmt.Sprintf("%v:%v", pipelineItem.ID, s.Value),
 				DisplayValue: s.DisplayName,
 			})
 		}
 	}
 
+	result[0] = pipelines
+	result[1] = stages
+
 	return result, nil
 }
+
+// Parses and returns only pipeline values.
+// Stage values are not included, because there is no corresponding field on the object to store them.
+func parsePipelineFieldValues(response *common.JSONHTTPResponse) ([]common.FieldValues, error) {
+	listOfValues, err := parsePipelineFieldValuesWithStages(response)
+	if err != nil {
+		return nil, err
+	}
+
+	if len(listOfValues) != 2 { // nolint:mnd
+		return nil, common.ErrInvalidImplementation
+	}
+
+	return []common.FieldValues{
+		listOfValues[0],
+	}, nil
+}
+
+// For more details, refer to the HubSpot documentation on Pipelines.
+// https://developers.hubspot.com/docs/guides/api/crm/pipelines#retrieve-pipelines
+type pipelineResponse struct {
+	Pipelines []pipeline `json:"results"`
+}
+
+type pipeline struct {
+	ID          string  `json:"id"`
+	DisplayName string  `json:"label"`
+	Stages      []stage `json:"stages"`
+}
+
+type stage struct {
+	Value       string `json:"id"`
+	DisplayName string `json:"label"`
+}


Test patch:
diff --git a/providers/hubspot/metadata_test.go b/providers/hubspot/metadata_test.go
index f5a3750e..cd29e0c6 100644
--- a/providers/hubspot/metadata_test.go
+++ b/providers/hubspot/metadata_test.go
@@ -207,29 +207,8 @@ func TestListObjectMetadata(t *testing.T) { // nolint:funlen,gocognit,cyclop,mai
 								ProviderType: "enumeration.select",
 								ReadOnly:     false,
 								Values: []common.FieldValue{{
-									Value:        "subscriber",
-									DisplayValue: "Subscriber",
-								}, {
-									Value:        "lead",
-									DisplayValue: "Lead",
-								}, {
-									Value:        "marketingqualifiedlead",
-									DisplayValue: "Marketing Qualified Lead",
-								}, {
-									Value:        "salesqualifiedlead",
-									DisplayValue: "Sales Qualified Lead",
-								}, {
-									Value:        "opportunity",
-									DisplayValue: "Opportunity",
-								}, {
-									Value:        "customer",
-									DisplayValue: "Customer",
-								}, {
-									Value:        "evangelist",
-									DisplayValue: "Evangelist",
-								}, {
-									Value:        "other",
-									DisplayValue: "Other",
+									Value:        "contacts-lifecycle-pipeline",
+									DisplayValue: "Lifecycle Stage Pipeline",
 								}},
 							},
 						},
@@ -265,25 +244,35 @@ func TestListObjectMetadata(t *testing.T) { // nolint:funlen,gocognit,cyclop,mai
 								ProviderType: "enumeration.select",
 								ReadOnly:     false,
 								Values: []common.FieldValue{{
-									Value:        "appointmentscheduled",
+									Value:        "default",
+									DisplayValue: "Sales Pipeline",
+								}},
+							},
+							"dealstage": {
+								DisplayName:  "Deal Stage",
+								ValueType:    "singleSelect",
+								ProviderType: "enumeration.radio",
+								ReadOnly:     false,
+								Values: []common.FieldValue{{
+									Value:        "default:appointmentscheduled",
 									DisplayValue: "Appointment Scheduled",
 								}, {
-									Value:        "qualifiedtobuy",
+									Value:        "default:qualifiedtobuy",
 									DisplayValue: "Qualified To Buy",
 								}, {
-									Value:        "presentationscheduled",
+									Value:        "default:presentationscheduled",
 									DisplayValue: "Presentation Scheduled",
 								}, {
-									Value:        "decisionmakerboughtin",
+									Value:        "default:decisionmakerboughtin",
 									DisplayValue: "Decision Maker Bought-In",
 								}, {
-									Value:        "contractsent",
+									Value:        "default:contractsent",
 									DisplayValue: "Contract Sent",
 								}, {
-									Value:        "closedwon",
+									Value:        "default:closedwon",
 									DisplayValue: "Closed Won",
 								}, {
-									Value:        "closedlost",
+									Value:        "default:closedlost",
 									DisplayValue: "Closed Lost",
 								}},
 							},
diff --git a/providers/hubspot/test/metadata-deals-properties-sampled.json b/providers/hubspot/test/metadata-deals-properties-sampled.json
index cdffff78..7f03553b 100644
--- a/providers/hubspot/test/metadata-deals-properties-sampled.json
+++ b/providers/hubspot/test/metadata-deals-properties-sampled.json
@@ -23,6 +23,31 @@
       },
       "formField": false,
       "dataSensitivity": "non_sensitive"
+    },
+    {
+      "updatedAt": "2024-09-05T17:14:04.747Z",
+      "createdAt": "2020-06-30T15:57:38.180Z",
+      "name": "dealstage",
+      "label": "Deal Stage",
+      "type": "enumeration",
+      "fieldType": "radio",
+      "description": "The stage of the deal. Deal stages allow you to categorize and track the progress of the deals that you are working on.",
+      "groupName": "deal_activity",
+      "options": [],
+      "displayOrder": 3,
+      "calculated": false,
+      "externalOptions": true,
+      "hasUniqueValue": false,
+      "hidden": false,
+      "hubspotDefined": true,
+      "modificationMetadata": {
+        "archivable": true,
+        "readOnlyDefinition": true,
+        "readOnlyOptions": false,
+        "readOnlyValue": false
+      },
+      "formField": false,
+      "dataSensitivity": "non_sensitive"
     }
   ]
 }
diff --git a/test/hubspot/metadata/contacts/main.go b/test/hubspot/metadata/contacts/main.go
index 4c9db340..09e0494e 100644
--- a/test/hubspot/metadata/contacts/main.go
+++ b/test/hubspot/metadata/contacts/main.go
@@ -3,19 +3,14 @@ package main
 import (
 	"context"
 	"log/slog"
+	"os"
 	"os/signal"
 	"syscall"
 
-	"github.com/amp-labs/connectors/common"
-	"github.com/amp-labs/connectors/internal/datautils"
 	connTest "github.com/amp-labs/connectors/test/hubspot"
 	"github.com/amp-labs/connectors/test/utils"
 )
 
-var objectName = "contacts"
-
-// we want to compare fields returned by read and schema properties provided by metadata methods
-// they must match for all such objects
 func main() {
 	// Handle Ctrl-C gracefully.
 	ctx, done := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
@@ -27,34 +22,12 @@ func main() {
 	conn := connTest.GetHubspotConnector(ctx)
 
 	metadata, err := conn.ListObjectMetadata(ctx, []string{
-		objectName,
+		"contacts",
 	})
 	if err != nil {
 		utils.Fail("error listing metadata for Hubspot", "error", err)
 	}
 
-	slog.Info("Read object using all fields from ListObjectMetadata")
-
-	requestFields := datautils.Map[string, string](metadata.Result[objectName].FieldsMap).KeySet()
-
-	response, err := conn.Read(ctx, common.ReadParams{
-		ObjectName: objectName,
-		Fields:     requestFields,
-	})
-	if err != nil {
-		utils.Fail("error reading from Hubspot", "error", err)
-	} else {
-		if response.Rows == 0 {
-			utils.Fail("expected to read at least one record", "error", err)
-		}
-
-		givenFields := datautils.Map[string, any](response.Data[0].Fields).KeySet()
-
-		difference := givenFields.Diff(requestFields)
-		if len(difference) != 0 {
-			utils.Fail("connector read didn't match requested fields", "difference", difference)
-		}
-	}
-
-	slog.Info("==> success fields requested from ListObjectMetadata are all present in Read.")
+	slog.Info("Metadata deals..")
+	utils.DumpJSON(metadata, os.Stdout)
 }