Merge branch 'main' into dipu/delighted

.github/workflows/build.yml

@@ -10,7 +10,7 @@ jobs:
       - name: Set up Go
         uses: actions/setup-go@v4
         with:
-          go-version: ">=1.23.3"
+          go-version: "^1.23.3"
           cache: false
 
       - name: Build

.github/workflows/linter.yml

@@ -8,13 +8,13 @@ jobs:
       - uses: actions/checkout@v3
       - uses: actions/setup-go@v4
         with:
-          go-version: "1.23.3"
+          go-version: "^1.23.3"
           cache: false
       - name: golangci-lint
         uses: golangci/golangci-lint-action@v3
         continue-on-error: false
         with:
-          version: v1.62.0
+          version: "v1.62.0"
       - name: Install gci
         run: "go install github.com/daixiang0/gci@latest"
       - name: List files with wrong format

Makefile

@@ -31,6 +31,10 @@ format: fix
 test:
 	go test -v ./...
 
+.PHONY: test-parallel
+test-parallel:
+	go test -v ./... -parallel=8 -count=3
+
 # Creates PR URLs for each template
 # Click on one of them or manually add ?template=<file.md> to the URL if you are creating a PR via the Github website
 # Templates: Under github/PULL_REQUEST_TEMPLATE directory you can add more templates

common/http.go

@@ -10,12 +10,14 @@ import (
 	"net/http"
 	"net/url"
 	"strings"
+
+	"github.com/amp-labs/connectors/common/logging"
 )
 
 // Header is a key/value pair that can be added to a request.
 type Header struct {
-	Key   string
-	Value string
+	Key   string `json:"key"`
+	Value string `json:"value"`
 }
 
 // ErrorHandler allows the caller to inject their own HTTP error handling logic.
@@ -41,6 +43,34 @@ func (h *HTTPClient) getURL(url string) (string, error) {
 	return getURL(h.Base, url)
 }
 
+// redactSensitiveHeaders redacts sensitive headers from the given headers.
+func redactSensitiveHeaders(hdrs []Header) []Header {
+	if hdrs == nil {
+		return nil
+	}
+
+	redacted := make([]Header, 0, len(hdrs))
+
+	for _, hdr := range hdrs {
+		switch {
+		case strings.EqualFold(hdr.Key, "Authorization"):
+			redacted = append(redacted, Header{Key: hdr.Key, Value: "<redacted>"})
+		case strings.EqualFold(hdr.Key, "Proxy-Authorization"):
+			redacted = append(redacted, Header{Key: hdr.Key, Value: "<redacted>"})
+		case strings.EqualFold(hdr.Key, "x-amz-security-token"):
+			redacted = append(redacted, Header{Key: hdr.Key, Value: "<redacted>"})
+		case strings.EqualFold(hdr.Key, "X-Api-Key"):
+			redacted = append(redacted, Header{Key: hdr.Key, Value: "<redacted>"})
+		case strings.EqualFold(hdr.Key, "X-Admin-Key"):
+			redacted = append(redacted, Header{Key: hdr.Key, Value: "<redacted>"})
+		default:
+			redacted = append(redacted, hdr)
+		}
+	}
+
+	return redacted
+}
+
 // Get makes a GET request to the given URL and returns the response. If the response is not a 2xx,
 // an error is returned. If the response is a 401, the caller should refresh the access token
 // and retry the request. If errorHandler is nil, then the default error handler is used.
@@ -50,6 +80,11 @@ func (h *HTTPClient) Get(ctx context.Context, url string, headers ...Header) (*h
 	if err != nil {
 		return nil, nil, err
 	}
+
+	logging.Logger(ctx).Debug("HTTP request",
+		"method", "GET", "url", fullURL,
+		"headers", redactSensitiveHeaders(headers))
+
 	// Make the request, get the response body
 	res, body, err := h.httpGet(ctx, fullURL, headers) //nolint:bodyclose
 	if err != nil {
@@ -71,6 +106,11 @@ func (h *HTTPClient) Post(ctx context.Context,
 		return nil, nil, err
 	}
 
+	logging.Logger(ctx).Debug("HTTP request",
+		"method", "POST", "url", fullURL,
+		"headers", redactSensitiveHeaders(headers),
+		"bodySize", len(reqBody))
+
 	// Make the request, get the response body
 	res, body, err := h.httpPost(ctx, fullURL, headers, reqBody) //nolint:bodyclose
 	if err != nil {
@@ -124,6 +164,11 @@ func (h *HTTPClient) Delete(ctx context.Context,
 	if err != nil {
 		return nil, nil, err
 	}
+
+	logging.Logger(ctx).Debug("HTTP request",
+		"method", "DELETE", "url", fullURL,
+		"headers", redactSensitiveHeaders(headers))
+
 	// Make the request, get the response body
 	res, body, err := h.httpDelete(ctx, fullURL, headers) //nolint:bodyclose
 	if err != nil {
@@ -232,6 +277,11 @@ func makePatchRequest(ctx context.Context, url string, headers []Header, body an
 
 	req.ContentLength = int64(len(jBody))
 
+	logging.Logger(ctx).Debug("HTTP request",
+		"method", "PATCH", "url", url,
+		"headers", redactSensitiveHeaders(headers),
+		"bodySize", len(jBody))
+
 	return addJSONContentTypeIfNotPresent(addHeaders(req, headers)), nil
 }
 
@@ -250,6 +300,11 @@ func makePutRequest(ctx context.Context, url string, headers []Header, body any)
 
 	req.ContentLength = int64(len(jBody))
 
+	logging.Logger(ctx).Debug("HTTP request",
+		"method", "PUT", "url", url,
+		"headers", redactSensitiveHeaders(headers),
+		"bodySize", len(jBody))
+
 	return addJSONContentTypeIfNotPresent(addHeaders(req, headers)), nil
 }
 

common/jsonquery/converters.go

@@ -1,6 +1,7 @@
 package jsonquery
 
 import (
+	"encoding/json"
 	"errors"
 
 	"github.com/spyzhov/ajson"
@@ -66,3 +67,23 @@ func (convertor) ObjectToMap(node *ajson.Node) (map[string]any, error) {
 
 	return result, nil
 }
+
+func ParseNode[T any](node *ajson.Node) (*T, error) {
+	var template T
+
+	raw, err := node.Unpack()
+	if err != nil {
+		return nil, err
+	}
+
+	data, err := json.Marshal(raw)
+	if err != nil {
+		return nil, err
+	}
+
+	if err := json.Unmarshal(data, &template); err != nil {
+		return nil, err
+	}
+
+	return &template, nil
+}

common/logging/logging.go

@@ -0,0 +1,106 @@
+package logging
+
+import (
+	"context"
+	"log/slog"
+)
+
+// WithLoggerEnabled returns a new context with the logger
+// explicitly enabled or disabled. If the key is not set, the
+// logger will be enabled by default.
+func WithLoggerEnabled(ctx context.Context, enabled bool) context.Context {
+	if ctx == nil {
+		ctx = context.Background()
+	}
+
+	return context.WithValue(ctx, contextKey("loggerEnabled"), enabled)
+}
+
+// With returns a new context with the given values added.
+// The values are added to the logger automatically.
+func With(ctx context.Context, values ...any) context.Context {
+	if len(values) == 0 && ctx != nil {
+		// Corner case, don't bother creating a new context.
+		return ctx
+	}
+
+	vals := append(getValues(ctx), values...)
+
+	return context.WithValue(ctx, contextKey("loggerValues"), vals)
+}
+
+// It's considered good practice to use unexported custom types for context keys.
+// This avoids collisions with other packages that might be using the same string
+// values for their own keys.
+type contextKey string
+
+func getValues(ctx context.Context) []any { //nolint:contextcheck
+	if ctx == nil {
+		ctx = context.Background()
+	}
+
+	// Check for a subsystem override.
+	sub := ctx.Value(contextKey("loggerValues"))
+	if sub != nil {
+		val, ok := sub.([]any)
+		if ok {
+			return val
+		} else {
+			return nil
+		}
+	} else {
+		return nil
+	}
+}
+
+// Logger returns a logger.
+//
+//nolint:contextcheck
+func Logger(ctx ...context.Context) *slog.Logger {
+	if len(ctx) == 0 {
+		return slog.Default()
+	}
+
+	var realCtx context.Context
+
+	// Honestly we only care if there's zero or one contexts.
+	// If there's more than one, we'll just use the first one.
+	for _, c := range ctx {
+		if c != nil {
+			realCtx = c //nolint:fatcontext
+
+			break
+		}
+	}
+
+	if realCtx == nil {
+		// No context provided, so we'll just use a sane default
+		realCtx = context.Background()
+	}
+
+	// Logging can be disabled by setting the loggerEnabled key to false.
+	sub := realCtx.Value(contextKey("loggerEnabled"))
+	if sub != nil {
+		val, ok := sub.(bool)
+		if ok && !val {
+			// The logger has been explicitly disabled.
+			//
+			// It's much, much simpler to just return a logger which
+			// throws everything away, than to add a check everywhere
+			// we might want to log something.
+			return nullLogger
+		}
+	}
+
+	// Get the default logger
+	logger := slog.Default()
+
+	// Check for key-values to add to the logger.
+	vals := getValues(realCtx)
+	if vals != nil {
+		logger = logger.With(vals...)
+	}
+
+	// Return the logger
+	return logger
+}

common/logging/null.go

@@ -0,0 +1,30 @@
+package logging
+
+import (
+	"context"
+	"log/slog"
+)
+
+var nullLogger *slog.Logger //nolint:gochecknoglobals
+
+func init() {
+	nullLogger = slog.New(&nullHandler{})
+}
+
+type nullHandler struct{}
+
+func (n *nullHandler) Enabled(_ context.Context, _ slog.Level) bool {
+	return false
+}
+
+func (n *nullHandler) Handle(_ context.Context, _ slog.Record) error {
+	return nil
+}
+
+func (n *nullHandler) WithAttrs(_ []slog.Attr) slog.Handler {
+	return n
+}
+
+func (n *nullHandler) WithGroup(_ string) slog.Handler {
+	return n
+}

common/parse.go

@@ -57,6 +57,12 @@ func ParseResult(
 		nextPage = ""
 	}
 
+	if len(marshaledData) == 0 {
+		// Either a JSON array is empty or it was nil.
+		// For consistency return empty array for missing records.
+		marshaledData = make([]ReadResultRow, 0)
+	}
+
 	return &ReadResult{
 		Rows:     int64(len(marshaledData)),
 		Data:     marshaledData,

common/types.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"net/http"
 	"time"
 
 	"github.com/amp-labs/connectors/internal/datautils"
@@ -92,6 +93,9 @@ var (
 	// ErrOperationNotSupportedForObject is returned when operation is not supported for this object.
 	ErrOperationNotSupportedForObject = errors.New("operation is not supported for this object in this module")
 
+	// ErrObjectNotSupported is returned when operation is not supported for this object.
+	ErrObjectNotSupported = errors.New("operation is not supported for this object")
+
 	// ErrResolvingURLPathForObject is returned when URL cannot be implied for object name.
 	ErrResolvingURLPathForObject = errors.New("cannot resolve URL path for given object name")
 
@@ -123,6 +127,10 @@ type ReadParams struct {
 	//		Note: timing is already handled by Since argument.
 	//		Reference: https://developers.klaviyo.com/en/docs/filtering_
 	Filter string // optional
+
+	// AssociatedObjects is a list of associated objects to fetch along with the main object.
+	// Only supported by HubSpot connector Read (not Search)
+	AssociatedObjects []string // optional
 }
 
 // WriteParams defines how we are writing data to a SaaS API.
@@ -136,6 +144,9 @@ type WriteParams struct {
 	// RecordData is a JSON node representing the record of data we want to insert in the case of CREATE
 	// or fields of data we want to modify in case of an update
 	RecordData any // required
+
+	// Associations contains associations between the object and other objects.
+	Associations any // optional
 }
 
 // DeleteParams defines how we are deleting data in SaaS API.
@@ -275,3 +286,12 @@ type SubscriptionEvent interface {
 	ObjectName() (string, error)
 	Workspace() (string, error)
 }
+
+// WebhookVerificationParameters is a struct that contains the parameters required to verify a webhook.
+type WebhookVerificationParameters struct {
+	Headers      http.Header
+	Body         []byte
+	URL          string
+	ClientSecret string
+	Method       string
+}

connectors.go

@@ -86,6 +86,13 @@ type AuthMetadataConnector interface {
 	GetPostAuthInfo(ctx context.Context) (*common.PostAuthInfo, error)
 }
 
+type WebhookVerifierConnector interface {
+	Connector
+
+	// VerifyWebhookMessage verifies the signature of a webhook message.
+	VerifyWebhookMessage(ctx context.Context, params *common.WebhookVerificationParameters) (bool, error)
+}
+
 // We re-export the following types so that they can be used by consumers of this library.
 type (
 	ReadParams               = common.ReadParams