fix: ingore webhook headers support when body is json array

gateway/webhook/integration_test.go

@@ -284,14 +284,18 @@ func TestIntegrationWebhook(t *testing.T) {
  assert.JSONEq(t, string(p), string(batch.Batch[0]))
  }
 
- r, err = errDB.GetUnprocessed(ctx, jobsdb.GetQueryParams{
- WorkspaceID: workspaceID,
- // ParameterFilters: []jobsdb.ParameterFilterT{{
- // Name: "source_id",
- // Value: sourceID,
- // }},
- JobsLimit: 1,
- })
+ require.Eventually(t, func() bool {
+ r, err = errDB.GetUnprocessed(ctx, jobsdb.GetQueryParams{
+ WorkspaceID: workspaceID,
+ ParameterFilters: []jobsdb.ParameterFilterT{{
+ Name: "source_id",
+ Value: sourceID,
+ }},
+ JobsLimit: 10,
+ })
+ return err == nil && len(r.Jobs) == len(tc.Output.ErrQueue)
+ }, time.Second, time.Millisecond*10)
+
  require.NoError(t, err)
  assert.Len(t, r.Jobs, len(tc.Output.ErrQueue))
  for i, p := range tc.Output.ErrQueue {

gateway/webhook/webhook.go

@@ -280,7 +280,7 @@ func getXHeaders(req *http.Request) map[string]string {
  return xHeaders
 }
 
-func prepareRequestBody(req *http.Request, sourceType string, sourceListForParsingParams []string) ([]byte, error) {
+func (webhook *HandleT) prepareRequestBody(req *http.Request, sourceType string, sourceListForParsingParams []string) ([]byte, error) {
  defer func() {
  _ = req.Body.Close()
  }()
@@ -305,11 +305,13 @@ func prepareRequestBody(req *http.Request, sourceType string, sourceListForParsi
 
  xHeaders := getXHeaders(req)
  if len(xHeaders) > 0 {
- body, err = sjson.SetBytes(body, "headers", xHeaders)
+ bodyWithHeaders, err := sjson.SetBytes(body, "headers", xHeaders)
  if err != nil {
- return nil, errors.New(response.InvalidJSON)
+ // When headers are not set in body, log the error and continue without setting headers
+ webhook.logger.Infof("Error while setting headers for source type=%s in body: %s", sourceType, err.Error())
+ } else {
+ body = bodyWithHeaders
  }
-
  }
 
  return body, nil
@@ -343,7 +345,7 @@ func (bt *batchWebhookTransformerT) batchTransformLoop() {
  var payloadArr [][]byte
  var webRequests []*webhookT
  for _, req := range breq.batchRequest {
- body, err := prepareRequestBody(req.request, breq.sourceType, bt.webhook.config.sourceListForParsingParams)
+ body, err := bt.webhook.prepareRequestBody(req.request, breq.sourceType, bt.webhook.config.sourceListForParsingParams)
  if err != nil {
  req.done <- transformerResponse{Err: response.GetStatus(err.Error())}
  continue

gateway/webhook/webhook_test.go

@@ -482,6 +482,12 @@ func TestRecordWebhookErrors(t *testing.T) {
 }
 
 func TestPrepareRequestBody(t *testing.T) {
+ initWebhook()
+ ctrl := gomock.NewController(t)
+ mockGW := mockWebhook.NewMockGateway(ctrl)
+ statsStore, err := memstats.New()
+ require.NoError(t, err)
+ webhookHandler := Setup(mockGW, transformer.NewNoOpService(), statsStore)
  type requestOpts struct {
  method string
  target string
@@ -508,6 +514,7 @@ func TestPrepareRequestBody(t *testing.T) {
  name string
  req *http.Request
  sourceType string
+ description string
  includeQueryParams bool
  wantError bool
  expectedResponse string
@@ -555,6 +562,13 @@ func TestPrepareRequestBody(t *testing.T) {
  sourceType: "shopify",
  expectedResponse: `{"key":"value","query_parameters":{},"headers":{"X-Key":"header-value"}}`,
  },
+ {
+ name: "Some payload with headers for sendgrid",
+ req: createRequest(requestOpts{method: http.MethodPost, target: "http://example.com", body: strings.NewReader(`[{"key":"value"}]`), headers: map[string]string{"X-Key": "header-value"}}),
+ sourceType: "webhook",
+ description: "Request Body is array so headers won't be included",
+ expectedResponse: `[{"key":"value"}]`,
+ },
  {
  name: "Some payload with query parameters for Adjust",
  req: createRequest(requestOpts{method: http.MethodPost, target: "http://example.com", body: strings.NewReader(`{"key1":"value1"}`), params: map[string]string{"key2": "value2"}}),
@@ -571,7 +585,7 @@ func TestPrepareRequestBody(t *testing.T) {
 
  for _, tc := range testCases {
  t.Run(tc.name, func(t *testing.T) {
- result, err := prepareRequestBody(tc.req, tc.sourceType, []string{"adjust", "shopify"})
+ result, err := webhookHandler.prepareRequestBody(tc.req, tc.sourceType, []string{"adjust", "shopify"})
  if tc.wantError {
  require.Error(t, err)
  return

go.mod

@@ -80,7 +80,7 @@ require (
  github.com/rudderlabs/rudder-go-kit v0.38.0
  github.com/rudderlabs/rudder-observability-kit v0.0.3
  github.com/rudderlabs/rudder-schemas v0.5.1
- github.com/rudderlabs/rudder-transformer/go v0.0.0-20240812044419-23196ec42acf
+ github.com/rudderlabs/rudder-transformer/go v0.0.0-20240904053719-1777c6cb23f9
  github.com/rudderlabs/sql-tunnels v0.1.7
  github.com/rudderlabs/sqlconnect-go v1.9.0
  github.com/samber/lo v1.47.0

go.sum

@@ -1134,8 +1134,8 @@ github.com/rudderlabs/rudder-observability-kit v0.0.3 h1:vZtuZRkGX+6rjaeKtxxFE2Y
 github.com/rudderlabs/rudder-observability-kit v0.0.3/go.mod h1:6UjAh3H6rkE0fFLh7z8ZGQEQbKtUkRfhWOf/OUhfqW8=
 github.com/rudderlabs/rudder-schemas v0.5.1 h1:g4I5wp2yA6ZWQZ1MjSNn4zby3XctG/TOgbYUW3dS4z4=
 github.com/rudderlabs/rudder-schemas v0.5.1/go.mod h1:JoDTB9nCDXwRz+G+aYwP3Fj42HLssKARxsFFm+qqgb4=
-github.com/rudderlabs/rudder-transformer/go v0.0.0-20240812044419-23196ec42acf h1:nsU2tKjPV/sbmOoIk39ncFT8D5HBDVppmrCWO0v9HsU=
-github.com/rudderlabs/rudder-transformer/go v0.0.0-20240812044419-23196ec42acf/go.mod h1:3NGitPz4pYRRZ6Xt09S+8hb0tHK/9pZcKJ3OgOTaSmE=
+github.com/rudderlabs/rudder-transformer/go v0.0.0-20240904053719-1777c6cb23f9 h1:ecY7YIVRITu04y8fcRUdcTASlFUtzOtjhsyu8P6b4eY=
+github.com/rudderlabs/rudder-transformer/go v0.0.0-20240904053719-1777c6cb23f9/go.mod h1:3NGitPz4pYRRZ6Xt09S+8hb0tHK/9pZcKJ3OgOTaSmE=
 github.com/rudderlabs/sql-tunnels v0.1.7 h1:wDCRl6zY4M5gfWazf7XkSTGQS3yjBzUiUgEMBIfHNDA=
 github.com/rudderlabs/sql-tunnels v0.1.7/go.mod h1:5f7+YL49JHYgteP4rAgqKnr4K2OadB0oIpUS+Tt3sPM=
 github.com/rudderlabs/sqlconnect-go v1.9.0 h1:icLgqvVQ15Vh+oP7epA0b0yK6sIzxRVwPlRzOoDNVRA=