diff --git a/assets/docs/configuration/overview-full-example.hcl b/assets/docs/configuration/overview-full-example.hcl
index a8f28749..a2e13703 100644
--- a/assets/docs/configuration/overview-full-example.hcl
+++ b/assets/docs/configuration/overview-full-example.hcl
@@ -90,16 +90,20 @@ stats_receiver {
   buffer_sec  = 20
 }
 
-// log level configuration (default: "info")
+# log level configuration (default: "info")
 log_level = "info"
 
-// Specifies how failed writes to the target should be retried, depending on an error type 
+# Specifies how failed writes to the target should be retried, depending on an error type 
 retry {
   transient {
+    # Initial delay (before first retry) for transient errors
     delay_ms = 1000 
+
+    # Maximum number of retries for transient errors
     max_attempts = 5 
   }
   setup {
+    # Initial delay (before first retry) for setup errors
     delay_ms = 20000
   }
 }
diff --git a/assets/docs/configuration/retry-example.hcl b/assets/docs/configuration/retry-example.hcl
index 2f0d5299..17bcc981 100644
--- a/assets/docs/configuration/retry-example.hcl
+++ b/assets/docs/configuration/retry-example.hcl
@@ -1,9 +1,13 @@
 retry {
   transient {
+    # Initial delay (before first retry) for transient errors
     delay_ms = 5000 
+
+    # Maximum number of retries for transient errors
     max_attempts = 10 
   }
   setup {
+    # Initial delay (before first retry) for setup errors
     delay_ms = 30000
   }
 }
diff --git a/assets/docs/configuration/targets/http-full-example.hcl b/assets/docs/configuration/targets/http-full-example.hcl
index 9a8f8a64..05ccb254 100644
--- a/assets/docs/configuration/targets/http-full-example.hcl
+++ b/assets/docs/configuration/targets/http-full-example.hcl
@@ -66,7 +66,8 @@ target {
     # Optional path to the file containing template which is used to build HTTP request based on a batch of input data
     template_file              = "myTemplate.file"
 
-    # 2 invalid + 1 setup error rules
+    # Optional HTTP response rules which are used to match HTTP response code/body and categorize it as either invalid data or target setup error.
+    # For example, we can have 2 invalid + 1 setup error rules:
     response_rules {
       # This one is a match when... 
       invalid {
diff --git a/assets/docs/configuration/targets/http-template-full-example.file b/assets/docs/configuration/targets/http-template-full-example.file
new file mode 100644
index 00000000..52222c24
--- /dev/null
+++ b/assets/docs/configuration/targets/http-template-full-example.file
@@ -0,0 +1,6 @@
+{
+  "secret_key": "{{ env "SECRET_KEY" }}",
+  "data": [
+        {{ range $i, $data := . }} {{if $i}},{{end}} {{ prettyPrint .foo }} {{ end }}
+  ]
+}
diff --git a/assets/docs/configuration/targets/http-template-unwrap-example.file b/assets/docs/configuration/targets/http-template-unwrap-example.file
new file mode 100644
index 00000000..0aa31145
--- /dev/null
+++ b/assets/docs/configuration/targets/http-template-unwrap-example.file
@@ -0,0 +1 @@
+{{ prettyPrint (index . 0)}}
diff --git a/assets/docs/configuration/transformations/builtin/jq-full-example.hcl b/assets/docs/configuration/transformations/builtin/jq-full-example.hcl
index 411da2c1..f9c6dd0f 100644
--- a/assets/docs/configuration/transformations/builtin/jq-full-example.hcl
+++ b/assets/docs/configuration/transformations/builtin/jq-full-example.hcl
@@ -1,5 +1,6 @@
 transform {
   use "jq" {
+    # Full JQ command which will be used to transform input data.  
     jq_command = <<JQEOT
 {
     my_app_id: .app_id,
@@ -9,7 +10,11 @@ transform {
 }
 JQEOT
 
+    # Optional. Timeout for execution of the script, in milliseconds.
     timeout_ms = 800
+
+    # Optional, may be used when the input is a Snowplow enriched TSV. 
+    # This will transform the data so that the root '.' JQ field contains JSON object representation of the event - with keys as returned by the Snowplow Analytics SDK.
     snowplow_mode = true
   }
 }
diff --git a/assets/docs/configuration/transformations/builtin/jq-minimal-example.hcl b/assets/docs/configuration/transformations/builtin/jq-minimal-example.hcl
index f34be287..c378adf1 100644
--- a/assets/docs/configuration/transformations/builtin/jq-minimal-example.hcl
+++ b/assets/docs/configuration/transformations/builtin/jq-minimal-example.hcl
@@ -1,5 +1,6 @@
 transform {
   use "jq" {
+    # Full JQ command which will be used to transform input data.  
     jq_command = "[.]"
   }
 }
diff --git a/assets/docs/configuration/transformations/builtin/jqFilter-full-example.hcl b/assets/docs/configuration/transformations/builtin/jqFilter-full-example.hcl
new file mode 100644
index 00000000..66c95674
--- /dev/null
+++ b/assets/docs/configuration/transformations/builtin/jqFilter-full-example.hcl
@@ -0,0 +1,13 @@
+transform {
+  use "jqFilter" {
+    # Full JQ command which will be used to filter input data. The output must be boolean. If 'false', data is then discarded.
+    jq_command = "has(\"app_id\")"
+
+    # Optional. Timeout for execution of the script, in milliseconds.
+    timeout_ms = 800
+
+    # Optional, may be used when the input is a Snowplow enriched TSV. 
+    # This will transform the data so that the root '.' JQ  field contains JSON object representation of the event - with keys as returned by the Snowplow Analytics SDK.
+    snowplow_mode = true
+  }
+}
diff --git a/assets/docs/configuration/transformations/builtin/jqFilter-minimal-example.hcl b/assets/docs/configuration/transformations/builtin/jqFilter-minimal-example.hcl
new file mode 100644
index 00000000..e548be93
--- /dev/null
+++ b/assets/docs/configuration/transformations/builtin/jqFilter-minimal-example.hcl
@@ -0,0 +1,6 @@
+transform {
+  use "jqFilter" {
+    # Full JQ command which will be used to filter input data. The output must be boolean. If 'false', data is then discarded.
+    jq_command = "has(\"app_id\")"
+  }
+}
diff --git a/docs/configuration_transformations_docs_test.go b/docs/configuration_transformations_docs_test.go
index 6bb6af34..b98f9a71 100644
--- a/docs/configuration_transformations_docs_test.go
+++ b/docs/configuration_transformations_docs_test.go
@@ -29,7 +29,7 @@ import (
 )
 
 func TestBuiltinTransformationDocumentation(t *testing.T) {
-	transformationsToTest := []string{"base64Decode", "base64Encode", "jq"}
+	transformationsToTest := []string{"base64Decode", "base64Encode", "jq", "jqFilter"}
 
 	for _, tfm := range transformationsToTest {
 
@@ -161,6 +161,8 @@ func testTransformationConfig(t *testing.T, filepath string, fullExample bool) {
 			configObject = &engine.JSEngineConfig{}
 		case "jq":
 			configObject = &transform.JQMapperConfig{}
+		case "jqFilter":
+			configObject = &filter.JQFilterConfig{}
 		default:
 			assert.Fail(fmt.Sprint("Source not recognised: ", use.Name))
 		}
diff --git a/pkg/target/http.go b/pkg/target/http.go
index 5ea521f2..79bb26a6 100644
--- a/pkg/target/http.go
+++ b/pkg/target/http.go
@@ -421,7 +421,6 @@ func (ht *HTTPTarget) Write(messages []*models.Message) (*models.TargetWriteResu
 
 			if findMatchingRule(response, ht.responseRules.Invalid) != nil {
 				for _, msg := range goodMsgs {
-					// can we use response body as an error message for invalid data?
 					msg.SetError(errors.New(response.Body))
 				}
 
diff --git a/pkg/transform/filter/jq_filter.go b/pkg/transform/filter/jq_filter.go
new file mode 100644
index 00000000..45b2775c
--- /dev/null
+++ b/pkg/transform/filter/jq_filter.go
@@ -0,0 +1,79 @@
+/**
+ * Copyright (c) 2020-present Snowplow Analytics Ltd.
+ * All rights reserved.
+ *
+ * This software is made available by Snowplow Analytics, Ltd.,
+ * under the terms of the Snowplow Limited Use License Agreement, Version 1.0
+ * located at https://docs.snowplow.io/limited-use-license-1.0
+ * BY INSTALLING, DOWNLOADING, ACCESSING, USING OR DISTRIBUTING ANY PORTION
+ * OF THE SOFTWARE, YOU AGREE TO THE TERMS OF SUCH LICENSE AGREEMENT.
+ */
+
+package filter
+
+import (
+	"errors"
+	"fmt"
+
+	"github.com/snowplow/snowbridge/config"
+	"github.com/snowplow/snowbridge/pkg/models"
+	"github.com/snowplow/snowbridge/pkg/transform"
+)
+
+// JQFilterConfig represents the configuration for the JQ filter transformation
+type JQFilterConfig struct {
+	JQCommand    string `hcl:"jq_command"`
+	RunTimeoutMs int    `hcl:"timeout_ms,optional"`
+	SpMode       bool   `hcl:"snowplow_mode,optional"`
+}
+
+// JQFilterConfigPair is a configuration pair for the jq filter transformation
+var JQFilterConfigPair = config.ConfigurationPair{
+	Name:   "jqFilter",
+	Handle: jqFilterAdapterGenerator(jqFilterConfigFunction),
+}
+
+func jqFilterConfigFunction(cfg *JQFilterConfig) (transform.TransformationFunction, error) {
+	return transform.GojqTransformationFunction(cfg.JQCommand, cfg.RunTimeoutMs, cfg.SpMode, filterOutput)
+}
+
+func jqFilterAdapterGenerator(f func(*JQFilterConfig) (transform.TransformationFunction, error)) jqFilterAdapter {
+	return func(i interface{}) (interface{}, error) {
+		cfg, ok := i.(*JQFilterConfig)
+		if !ok {
+			return nil, errors.New("invalid input, expected JQFilterConfig")
+		}
+
+		return f(cfg)
+	}
+}
+
+// This is where actual filtering is implemented, based on a JQ command output.
+func filterOutput(jqOutput transform.JqCommandOutput) transform.TransformationFunction {
+	return func(message *models.Message, interState interface{}) (*models.Message, *models.Message, *models.Message, interface{}) {
+		shouldKeepMessage, isBoolean := jqOutput.(bool)
+
+		if !isBoolean {
+			message.SetError(fmt.Errorf("jq filter returned '%v'; expected boolean", jqOutput))
+			return nil, nil, message, nil
+		}
+
+		if !shouldKeepMessage {
+			return nil, message, nil, nil
+		}
+
+		return message, nil, nil, interState
+	}
+}
+
+type jqFilterAdapter func(i interface{}) (interface{}, error)
+
+func (f jqFilterAdapter) ProvideDefault() (interface{}, error) {
+	return &JQFilterConfig{
+		RunTimeoutMs: 100,
+	}, nil
+}
+
+func (f jqFilterAdapter) Create(i interface{}) (interface{}, error) {
+	return f(i)
+}
diff --git a/pkg/transform/filter/jq_filter_test.go b/pkg/transform/filter/jq_filter_test.go
new file mode 100644
index 00000000..012b0d92
--- /dev/null
+++ b/pkg/transform/filter/jq_filter_test.go
@@ -0,0 +1,136 @@
+/**
+ * Copyright (c) 2020-present Snowplow Analytics Ltd.
+ * All rights reserved.
+ *
+ * This software is made available by Snowplow Analytics, Ltd.,
+ * under the terms of the Snowplow Limited Use License Agreement, Version 1.0
+ * located at https://docs.snowplow.io/limited-use-license-1.0
+ * BY INSTALLING, DOWNLOADING, ACCESSING, USING OR DISTRIBUTING ANY PORTION
+ * OF THE SOFTWARE, YOU AGREE TO THE TERMS OF SUCH LICENSE AGREEMENT.
+ */
+
+package filter
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/snowplow/snowbridge/pkg/models"
+	"github.com/snowplow/snowbridge/pkg/transform"
+)
+
+func TestJQFilter_SpMode_true_keep(t *testing.T) {
+	assert := assert.New(t)
+	input := &models.Message{
+		Data:         transform.SnowplowTsv1,
+		PartitionKey: "some-key",
+	}
+
+	config := &JQFilterConfig{JQCommand: `has("app_id")`, RunTimeoutMs: 100, SpMode: true}
+	filter := createFilter(t, config)
+
+	kept, dropped, invalid, _ := filter(input, nil)
+	assert.Empty(dropped)
+	assert.Empty(invalid)
+	assert.Equal(string(transform.SnowplowTsv1), string(kept.Data))
+}
+
+func TestJQFilter_SpMode_true_drop(t *testing.T) {
+	assert := assert.New(t)
+	input := &models.Message{
+		Data:         transform.SnowplowTsv1,
+		PartitionKey: "some-key",
+	}
+
+	config := &JQFilterConfig{JQCommand: `has("non_existent_key")`, RunTimeoutMs: 100, SpMode: true}
+	filter := createFilter(t, config)
+
+	kept, dropped, invalid, _ := filter(input, nil)
+	assert.Empty(kept)
+	assert.Empty(invalid)
+	assert.Equal(string(transform.SnowplowTsv1), string(dropped.Data))
+}
+
+func TestJQFilter_SpMode_false_keep(t *testing.T) {
+	assert := assert.New(t)
+	input := &models.Message{
+		Data:         transform.SnowplowJSON1,
+		PartitionKey: "some-key",
+	}
+
+	config := &JQFilterConfig{JQCommand: `has("app_id")`, RunTimeoutMs: 100, SpMode: false}
+	filter := createFilter(t, config)
+
+	kept, dropped, invalid, _ := filter(input, nil)
+	assert.Empty(dropped)
+	assert.Empty(invalid)
+	assert.Equal(string(transform.SnowplowJSON1), string(kept.Data))
+}
+
+func TestJQFilter_SpMode_false_drop(t *testing.T) {
+	assert := assert.New(t)
+	input := &models.Message{
+		Data:         transform.SnowplowJSON1,
+		PartitionKey: "some-key",
+	}
+
+	config := &JQFilterConfig{JQCommand: `has("non_existent_key")`, RunTimeoutMs: 100, SpMode: false}
+	filter := createFilter(t, config)
+
+	kept, dropped, invalid, _ := filter(input, nil)
+	assert.Empty(kept)
+	assert.Empty(invalid)
+	assert.Equal(string(transform.SnowplowJSON1), string(dropped.Data))
+}
+
+func TestJQFilter_epoch(t *testing.T) {
+	assert := assert.New(t)
+	input := &models.Message{
+		Data:         transform.SnowplowTsv1,
+		PartitionKey: "some-key",
+	}
+
+	config := &JQFilterConfig{JQCommand: `.collector_tstamp | epoch | . < 10`, RunTimeoutMs: 100, SpMode: true}
+	filter := createFilter(t, config)
+
+	kept, dropped, invalid, _ := filter(input, nil)
+	assert.Empty(kept)
+	assert.Empty(invalid)
+	assert.Equal(string(transform.SnowplowTsv1), string(dropped.Data))
+}
+
+func TestJQFilter_non_boolean_output(t *testing.T) {
+	assert := assert.New(t)
+	input := &models.Message{
+		Data:         transform.SnowplowTsv1,
+		PartitionKey: "some-key",
+	}
+
+	config := &JQFilterConfig{JQCommand: `.collector_tstamp | epoch`, RunTimeoutMs: 100, SpMode: true}
+	filter := createFilter(t, config)
+
+	kept, dropped, invalid, _ := filter(input, nil)
+
+	assert.Empty(kept)
+	assert.Empty(dropped)
+	assert.Equal("jq filter returned '1557499235'; expected boolean", invalid.GetError().Error())
+}
+
+func TestJQFilter_invalid_jq_command(t *testing.T) {
+	assert := assert.New(t)
+
+	config := &JQFilterConfig{JQCommand: `blabla`, RunTimeoutMs: 100, SpMode: true}
+	filter, err := jqFilterConfigFunction(config)
+
+	assert.Nil(filter)
+	assert.Equal("error compiling jq query: function not defined: blabla/0", err.Error())
+}
+
+func createFilter(t *testing.T, config *JQFilterConfig) transform.TransformationFunction {
+	filter, err := jqFilterConfigFunction(config)
+	if err != nil {
+		t.Fatalf("failed to create transformation function with error: %q", err.Error())
+	}
+	return filter
+}
diff --git a/pkg/transform/jq.go b/pkg/transform/jq.go
index 228b983f..4182d8a3 100644
--- a/pkg/transform/jq.go
+++ b/pkg/transform/jq.go
@@ -12,13 +12,8 @@
 package transform
 
 import (
-	"context"
 	"encoding/json"
 	"errors"
-	"fmt"
-	"time"
-
-	"github.com/itchyny/gojq"
 
 	"github.com/snowplow/snowbridge/config"
 	"github.com/snowplow/snowbridge/pkg/models"
@@ -31,42 +26,23 @@ type JQMapperConfig struct {
 	SpMode       bool   `hcl:"snowplow_mode,optional"`
 }
 
-// JQMapper handles jq generic mapping as a transformation
-type jqMapper struct {
-	JQCode       *gojq.Code
-	RunTimeoutMs time.Duration
-	SpMode       bool
+// JQMapperConfigPair is a configuration pair for the jq mapper transformation
+var JQMapperConfigPair = config.ConfigurationPair{
+	Name:   "jq",
+	Handle: jqMapperAdapterGenerator(jqMapperConfigFunction),
 }
 
-// RunFunction runs a jq mapper transformation
-func (jqm *jqMapper) RunFunction() TransformationFunction {
-	return func(message *models.Message, interState interface{}) (*models.Message, *models.Message, *models.Message, interface{}) {
-		input, err := mkJQInput(jqm, message, interState)
-		if err != nil {
-			message.SetError(err)
-			return nil, nil, message, nil
-		}
-
-		ctx, cancel := context.WithTimeout(context.Background(), jqm.RunTimeoutMs)
-		defer cancel()
-
-		iter := jqm.JQCode.RunWithContext(ctx, input)
-		// no looping since we only keep first value
-		v, ok := iter.Next()
-		if !ok {
-			message.SetError(errors.New("jq query got no output"))
-			return nil, nil, message, nil
-		}
-
-		if err, ok := v.(error); ok {
-			message.SetError(err)
-			return nil, nil, message, nil
-		}
+// jqMapperConfigFunction returns a jq mapper transformation function from a JQMapperConfig
+func jqMapperConfigFunction(c *JQMapperConfig) (TransformationFunction, error) {
+	return GojqTransformationFunction(c.JQCommand, c.RunTimeoutMs, c.SpMode, transformOutput)
+}
 
-		removeNullFields(v)
+func transformOutput(jqOutput JqCommandOutput) TransformationFunction {
+	return func(message *models.Message, interState interface{}) (*models.Message, *models.Message, *models.Message, interface{}) {
+		removeNullFields(jqOutput)
 
 		// here v is any, so we Marshal. alternative: gojq.Marshal
-		data, err := json.Marshal(v)
+		data, err := json.Marshal(jqOutput)
 		if err != nil {
 			message.SetError(errors.New("error encoding jq query output data"))
 			return nil, nil, message, nil
@@ -77,118 +53,6 @@ func (jqm *jqMapper) RunFunction() TransformationFunction {
 	}
 }
 
-// jqMapperAdapter implements the Pluggable interface
-type jqMapperAdapter func(i interface{}) (interface{}, error)
-
-// ProvideDefault implements the ComponentConfigurable interface
-func (f jqMapperAdapter) ProvideDefault() (interface{}, error) {
-	return &JQMapperConfig{
-		RunTimeoutMs: 100,
-	}, nil
-}
-
-// Create implements the ComponentCreator interface
-func (f jqMapperAdapter) Create(i interface{}) (interface{}, error) {
-	return f(i)
-}
-
-// jqMapperAdapterGenerator returns a jqAdapter
-func jqMapperAdapterGenerator(f func(c *JQMapperConfig) (TransformationFunction, error)) jqMapperAdapter {
-	return func(i interface{}) (interface{}, error) {
-		cfg, ok := i.(*JQMapperConfig)
-		if !ok {
-			return nil, errors.New("invalid input, expected JQMapperConfig")
-		}
-
-		return f(cfg)
-	}
-}
-
-// jqMapperConfigFunction returns a jq mapper transformation function from a JQMapperConfig
-func jqMapperConfigFunction(c *JQMapperConfig) (TransformationFunction, error) {
-	query, err := gojq.Parse(c.JQCommand)
-	if err != nil {
-		return nil, fmt.Errorf("error parsing jq command: %s", err)
-	}
-
-	// epoch converts a time.Time to an epoch in seconds, as integer type.
-	// It must be an integer in order to chain with jq-native time functions
-	withEpochFunction := gojq.WithFunction("epoch", 0, 1, func(a1 any, a2 []any) any {
-		if a1 == nil {
-			return nil
-		}
-
-		validTime, ok := a1.(time.Time)
-
-		if !ok {
-			return errors.New("Not a valid time input to 'epoch' function")
-		}
-
-		return int(validTime.Unix())
-	})
-
-	// epochMillis converts a time.Time to an epoch in milliseconds
-	withEpochMillisFunction := gojq.WithFunction("epochMillis", 0, 1, func(a1 any, a2 []any) any {
-		if a1 == nil {
-			return nil
-		}
-
-		validTime, ok := a1.(time.Time)
-
-		if !ok {
-			return errors.New("Not a valid time input to 'epochMillis' function")
-		}
-
-		return validTime.UnixMilli()
-	})
-
-	code, err := gojq.Compile(query, withEpochMillisFunction, withEpochFunction)
-	if err != nil {
-		return nil, fmt.Errorf("error compiling jq query: %s", err)
-	}
-
-	jq := &jqMapper{
-		JQCode:       code,
-		RunTimeoutMs: time.Duration(c.RunTimeoutMs) * time.Millisecond,
-		SpMode:       c.SpMode,
-	}
-
-	return jq.RunFunction(), nil
-}
-
-// JQMapperConfigPair is a configuration pair for the jq mapper transformation
-var JQMapperConfigPair = config.ConfigurationPair{
-	Name:   "jq",
-	Handle: jqMapperAdapterGenerator(jqMapperConfigFunction),
-}
-
-// mkJQInput ensures the input to JQ query is of expected type
-func mkJQInput(jqm *jqMapper, message *models.Message, interState interface{}) (map[string]interface{}, error) {
-	if !jqm.SpMode {
-		// gojq input can only be map[string]any or []any
-		// here we only consider the first, but we could also expand
-		var input map[string]interface{}
-		err := json.Unmarshal(message.Data, &input)
-		if err != nil {
-			return nil, err
-		}
-
-		return input, nil
-	}
-
-	parsedEvent, err := IntermediateAsSpEnrichedParsed(interState, message)
-	if err != nil {
-		return nil, err
-	}
-
-	spInput, err := parsedEvent.ToMap()
-	if err != nil {
-		return nil, err
-	}
-
-	return spInput, nil
-}
-
 func removeNullFields(data any) {
 	switch input := data.(type) {
 	case map[string]any:
@@ -216,3 +80,30 @@ func removeNullFromSlice(input []any) {
 		removeNullFields(item)
 	}
 }
+
+// jqMapperAdapterGenerator returns a jqAdapter
+func jqMapperAdapterGenerator(f func(c *JQMapperConfig) (TransformationFunction, error)) jqMapperAdapter {
+	return func(i interface{}) (interface{}, error) {
+		cfg, ok := i.(*JQMapperConfig)
+		if !ok {
+			return nil, errors.New("invalid input, expected JQMapperConfig")
+		}
+
+		return f(cfg)
+	}
+}
+
+// jqMapperAdapter implements the Pluggable interface
+type jqMapperAdapter func(i interface{}) (interface{}, error)
+
+// ProvideDefault implements the ComponentConfigurable interface
+func (f jqMapperAdapter) ProvideDefault() (interface{}, error) {
+	return &JQMapperConfig{
+		RunTimeoutMs: 100,
+	}, nil
+}
+
+// Create implements the ComponentCreator interface
+func (f jqMapperAdapter) Create(i interface{}) (interface{}, error) {
+	return f(i)
+}
diff --git a/pkg/transform/jq_common.go b/pkg/transform/jq_common.go
new file mode 100644
index 00000000..467171fa
--- /dev/null
+++ b/pkg/transform/jq_common.go
@@ -0,0 +1,132 @@
+/**
+ * Copyright (c) 2020-present Snowplow Analytics Ltd.
+ * All rights reserved.
+ *
+ * This software is made available by Snowplow Analytics, Ltd.,
+ * under the terms of the Snowplow Limited Use License Agreement, Version 1.0
+ * located at https://docs.snowplow.io/limited-use-license-1.0
+ * BY INSTALLING, DOWNLOADING, ACCESSING, USING OR DISTRIBUTING ANY PORTION
+ * OF THE SOFTWARE, YOU AGREE TO THE TERMS OF SUCH LICENSE AGREEMENT.
+ */
+
+package transform
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/itchyny/gojq"
+
+	"github.com/snowplow/snowbridge/pkg/models"
+	"github.com/snowplow/snowplow-golang-analytics-sdk/analytics"
+)
+
+// JqCommandOutput is a type representing output after executing JQ command. For filters for example we expect it to be boolean.
+type JqCommandOutput = interface{}
+
+// JqOutputHandler is a function which accepts JqCommandOutput and is response for doing something with it. For filters for example that would be filtering message based on boolean output.
+type JqOutputHandler func(JqCommandOutput) TransformationFunction
+
+// GojqTransformationFunction is a function returning another transformation function which allows us to do some GOJQ based mapping/filtering. Actual transformation happens in provided JqOutputHandler.
+func GojqTransformationFunction(command string, timeoutMs int, spMode bool, jqOutputHandler JqOutputHandler) (TransformationFunction, error) {
+	query, err := gojq.Parse(command)
+	if err != nil {
+		return nil, fmt.Errorf("error parsing jq command: %s", err)
+	}
+
+	// epoch converts a time.Time to an epoch in seconds, as integer type.
+	// It must be an integer in order to chain with jq-native time functions
+	withEpochFunction := gojq.WithFunction("epoch", 0, 1, func(a1 any, a2 []any) any {
+		if a1 == nil {
+			return nil
+		}
+
+		validTime, ok := a1.(time.Time)
+
+		if !ok {
+			return errors.New("Not a valid time input to 'epoch' function")
+		}
+
+		return int(validTime.Unix())
+	})
+
+	// epochMillis converts a time.Time to an epoch in milliseconds
+	withEpochMillisFunction := gojq.WithFunction("epochMillis", 0, 1, func(a1 any, a2 []any) any {
+		if a1 == nil {
+			return nil
+		}
+
+		validTime, ok := a1.(time.Time)
+
+		if !ok {
+			return errors.New("Not a valid time input to 'epochMillis' function")
+		}
+
+		return validTime.UnixMilli()
+	})
+
+	code, err := gojq.Compile(query, withEpochMillisFunction, withEpochFunction)
+	if err != nil {
+		return nil, fmt.Errorf("error compiling jq query: %s", err)
+	}
+
+	return runFunction(code, timeoutMs, spMode, jqOutputHandler), nil
+}
+
+func runFunction(jqcode *gojq.Code, timeoutMs int, spMode bool, jqOutputHandler JqOutputHandler) TransformationFunction {
+	return func(message *models.Message, interState interface{}) (*models.Message, *models.Message, *models.Message, interface{}) {
+		input, parsedEvent, err := mkJQInput(message, interState, spMode)
+		if err != nil {
+			message.SetError(err)
+			return nil, nil, message, nil
+		}
+
+		ctx, cancel := context.WithTimeout(context.Background(), time.Duration(timeoutMs)*time.Millisecond)
+		defer cancel()
+
+		iter := jqcode.RunWithContext(ctx, input)
+		// no looping since we only keep first value
+		jqOutput, ok := iter.Next()
+		if !ok {
+			message.SetError(errors.New("jq query got no output"))
+			return nil, nil, message, nil
+		}
+
+		if err, ok := jqOutput.(error); ok {
+			message.SetError(err)
+			return nil, nil, message, nil
+		}
+
+		return jqOutputHandler(jqOutput)(message, parsedEvent)
+	}
+}
+
+// mkJQInput ensures the input to JQ query is of expected type
+func mkJQInput(message *models.Message, interState interface{}, spMode bool) (map[string]interface{}, analytics.ParsedEvent, error) {
+	if !spMode {
+		// gojq input can only be map[string]any or []any
+		// here we only consider the first, but we could also expand
+		var input map[string]interface{}
+		err := json.Unmarshal(message.Data, &input)
+		if err != nil {
+			return nil, nil, err
+		}
+
+		return input, nil, nil
+	}
+
+	parsedEvent, err := IntermediateAsSpEnrichedParsed(interState, message)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	spInput, err := parsedEvent.ToMap()
+	if err != nil {
+		return nil, nil, err
+	}
+
+	return spInput, parsedEvent, nil
+}
diff --git a/pkg/transform/jq_test.go b/pkg/transform/jq_test.go
index f9178bfb..a99c53c4 100644
--- a/pkg/transform/jq_test.go
+++ b/pkg/transform/jq_test.go
@@ -212,7 +212,7 @@ func TestJQRunFunction_SpMode_false(t *testing.T) {
 			Scenario:  "happy_path",
 			JQCommand: `{foo: .app_id}`,
 			InputMsg: &models.Message{
-				Data:         snowplowJSON1,
+				Data:         SnowplowJSON1,
 				PartitionKey: "some-key",
 			},
 			InputInterState: nil,
diff --git a/pkg/transform/snowplow_enriched_to_json_test.go b/pkg/transform/snowplow_enriched_to_json_test.go
index 64958837..08a38a33 100644
--- a/pkg/transform/snowplow_enriched_to_json_test.go
+++ b/pkg/transform/snowplow_enriched_to_json_test.go
@@ -33,7 +33,7 @@ func TestSpEnrichedToJson(t *testing.T) {
 	}
 
 	var expectedGood = models.Message{
-		Data:         snowplowJSON1,
+		Data:         SnowplowJSON1,
 		PartitionKey: "some-key",
 	}
 
diff --git a/pkg/transform/transform_test.go b/pkg/transform/transform_test.go
index 5355252d..1062d5af 100644
--- a/pkg/transform/transform_test.go
+++ b/pkg/transform/transform_test.go
@@ -56,7 +56,7 @@ func TestNewTransformation_EnrichedToJson(t *testing.T) {
 
 	var expectedGood = []*models.Message{
 		{
-			Data:         snowplowJSON1,
+			Data:         SnowplowJSON1,
 			PartitionKey: "some-key",
 		},
 		{
@@ -117,7 +117,7 @@ func TestNewTransformation_Multiple(t *testing.T) {
 
 	var expectedGood = []*models.Message{
 		{
-			Data:         snowplowJSON1,
+			Data:         SnowplowJSON1,
 			PartitionKey: "test-data1",
 		},
 		{
diff --git a/pkg/transform/transform_test_variables.go b/pkg/transform/transform_test_variables.go
index 841367dc..baf800b5 100644
--- a/pkg/transform/transform_test_variables.go
+++ b/pkg/transform/transform_test_variables.go
@@ -22,7 +22,9 @@ var SnowplowTsv1 = []byte(`test-data1	pc	2019-05-10 14:40:37.436	2019-05-10 14:4
 
 // SpTsv1Parsed is test data
 var SpTsv1Parsed, _ = analytics.ParseEvent(string(SnowplowTsv1))
-var snowplowJSON1 = []byte(`{"app_id":"test-data1","collector_tstamp":"2019-05-10T14:40:35.972Z","contexts_com_acme_just_ints_1":[{"integerField":0},{"integerField":1},{"integerField":2}],"contexts_nl_basjes_yauaa_context_1":[{"agentClass":"Special","agentName":"python-requests","agentNameVersion":"python-requests 2.21.0","agentNameVersionMajor":"python-requests 2","agentVersion":"2.21.0","agentVersionMajor":"2","deviceBrand":"Unknown","deviceClass":"Unknown","deviceName":"Unknown","layoutEngineClass":"Unknown","layoutEngineName":"Unknown","layoutEngineVersion":"??","layoutEngineVersionMajor":"??","operatingSystemClass":"Unknown","operatingSystemName":"Unknown","operatingSystemVersion":"??"}],"derived_tstamp":"2019-05-10T14:40:35.972Z","dvce_created_tstamp":"2019-05-10T14:40:35.551Z","dvce_sent_tstamp":"2019-05-10T14:40:35Z","etl_tstamp":"2019-05-10T14:40:37.436Z","event":"unstruct","event_format":"jsonschema","event_id":"e9234345-f042-46ad-b1aa-424464066a33","event_name":"add_to_cart","event_vendor":"com.snowplowanalytics.snowplow","event_version":"1-0-0","network_userid":"d26822f5-52cc-4292-8f77-14ef6b7a27e2","platform":"pc","unstruct_event_com_snowplowanalytics_snowplow_add_to_cart_1":{"currency":"GBP","quantity":2,"sku":"item41","unitPrice":32.4},"user_id":"user\u003cbuilt-in function input\u003e","user_ipaddress":"18.194.133.57","useragent":"python-requests/2.21.0","v_collector":"ssc-0.15.0-googlepubsub","v_etl":"beam-enrich-0.2.0-common-0.36.0","v_tracker":"py-0.8.2"}`)
+
+// SnowplowJSON1 another test data
+var SnowplowJSON1 = []byte(`{"app_id":"test-data1","collector_tstamp":"2019-05-10T14:40:35.972Z","contexts_com_acme_just_ints_1":[{"integerField":0},{"integerField":1},{"integerField":2}],"contexts_nl_basjes_yauaa_context_1":[{"agentClass":"Special","agentName":"python-requests","agentNameVersion":"python-requests 2.21.0","agentNameVersionMajor":"python-requests 2","agentVersion":"2.21.0","agentVersionMajor":"2","deviceBrand":"Unknown","deviceClass":"Unknown","deviceName":"Unknown","layoutEngineClass":"Unknown","layoutEngineName":"Unknown","layoutEngineVersion":"??","layoutEngineVersionMajor":"??","operatingSystemClass":"Unknown","operatingSystemName":"Unknown","operatingSystemVersion":"??"}],"derived_tstamp":"2019-05-10T14:40:35.972Z","dvce_created_tstamp":"2019-05-10T14:40:35.551Z","dvce_sent_tstamp":"2019-05-10T14:40:35Z","etl_tstamp":"2019-05-10T14:40:37.436Z","event":"unstruct","event_format":"jsonschema","event_id":"e9234345-f042-46ad-b1aa-424464066a33","event_name":"add_to_cart","event_vendor":"com.snowplowanalytics.snowplow","event_version":"1-0-0","network_userid":"d26822f5-52cc-4292-8f77-14ef6b7a27e2","platform":"pc","unstruct_event_com_snowplowanalytics_snowplow_add_to_cart_1":{"currency":"GBP","quantity":2,"sku":"item41","unitPrice":32.4},"user_id":"user\u003cbuilt-in function input\u003e","user_ipaddress":"18.194.133.57","useragent":"python-requests/2.21.0","v_collector":"ssc-0.15.0-googlepubsub","v_etl":"beam-enrich-0.2.0-common-0.36.0","v_tracker":"py-0.8.2"}`)
 
 // SnowplowTsv2 is test data
 var SnowplowTsv2 = []byte(`test-data2	pc	2019-05-10 14:40:32.392	2019-05-10 14:40:31.105	2019-05-10 14:40:30.218	transaction_item	5071169f-3050-473f-b03f-9748319b1ef2			py-0.8.2	ssc-0.15.0-googlepubsub	beam-enrich-0.2.0-common-0.36.0	user<built-in function input>	18.194.133.57				68220ade-307b-4898-8e25-c4c8ac92f1d7																																																		transaction<built-in function input>	item58			35.87	1					python-requests/2.21.0																																										2019-05-10 14:40:30.000			{"schema":"iglu:com.snowplowanalytics.snowplow/contexts/jsonschema/1-0-1","data":[{"schema":"iglu:nl.basjes/yauaa_context/jsonschema/1-0-0","data":{"deviceBrand":"Unknown","deviceName":"Unknown","operatingSystemName":"Unknown","agentVersionMajor":"2","layoutEngineVersionMajor":"??","deviceClass":"Unknown","agentNameVersionMajor":"python-requests 2","operatingSystemClass":"Unknown","layoutEngineName":"Unknown","agentName":"python-requests","agentVersion":"2.21.0","layoutEngineClass":"Unknown","agentNameVersion":"python-requests 2.21.0","operatingSystemVersion":"??","agentClass":"Special","layoutEngineVersion":"??"}}]}		2019-05-10 14:40:31.105	com.snowplowanalytics.snowplow	transaction_item	jsonschema	1-0-0		`)
diff --git a/pkg/transform/transformconfig/transform_config.go b/pkg/transform/transformconfig/transform_config.go
index 580fedb3..ae6b76fc 100644
--- a/pkg/transform/transformconfig/transform_config.go
+++ b/pkg/transform/transformconfig/transform_config.go
@@ -25,6 +25,7 @@ var SupportedTransformations = []config.ConfigurationPair{
 	filter.AtomicFilterConfigPair,
 	filter.UnstructFilterConfigPair,
 	filter.ContextFilterConfigPair,
+	filter.JQFilterConfigPair,
 	transform.SetPkConfigPair,
 	transform.EnrichedToJSONConfigPair,
 	transform.Base64DecodeConfigPair,