Repository: xataio/pgstream
Pull request #389 (2025-06-04 11:57:00)

Json transformer support
Adding JSON transformer for json and jsonb date types of postgres. This transformer executes a list of given operations on the json data to be transformed.
All operations must be either `set` or `delete`. 
`set` operations support literal values as well as templates, making use of `sprig` and `greenmask`'s function sets. Also, like the template transformer, `.GetValue` and `.GetDynamicValue` functions are added. Unlike template transformer, here `.GetValue` refers to the value at given path, rather than the entire JSON object; whereas `.GetDynamicValue` is again used for referring to other columns.
`delete` operations simply delete the object at the given path.
`Transform` errors out if the given path does not exists and the parameter `error_not_exist` - which is false by default - is set to true.

Adding helper function `getOperationsParam` to map the transformer parameters to the fields of struct `Operation`.
Also adding `JSONValue` to be used only within JSON transformer. As mentioned before, this one's `.GetValue` function - unlike the regular `Value` struct - refers to a particular path in the json object.
Using [sjson](https://github.com/tidwall/sjson) library for executing the operations. Just like `greenmask`, pathing should follow the [synxtax rules of sjson](https://github.com/tidwall/sjson#path-syntax)

Also adding tests and updating the docs.

fixes: #145 

github-actions[bot]: ### Merging this branch changes the coverage (1 decrease, 1 increase)

| Impacted Packages | Coverage Δ | :robot: |
|-------------------|------------|---------|
| github.com/xataio/pgstream/pkg/transformers | 84.87% (**+2.80%**) | :thumbsup: |
| github.com/xataio/pgstream/pkg/transformers/builder | 38.46% (**-1.01%**) | :thumbsdown: |

---

<details>

<summary>Coverage by file</summary>

### Changed files (no unit tests)

| Changed File | Coverage Δ | Total | Covered | Missed | :robot: |
|--------------|------------|-------|---------|--------|---------|
| github.com/xataio/pgstream/pkg/transformers/builder/transformer_builder.go | 38.46% (**-1.01%**) | 39 (+1) | 15 | 24 (+1) | :thumbsdown: |
| github.com/xataio/pgstream/pkg/transformers/json_transformer.go | 91.51% (**+91.51%**) | 106 (+106) | 97 (+97) | 9 (+9) | :star2: |
| github.com/xataio/pgstream/pkg/transformers/transformer.go | 86.89% (ø) | 61 | 53 | 8 |  |

_Please note that the "Total", "Covered", and "Missed" counts above refer to ***code statements*** instead of lines of code. The value in brackets refers to the test coverage of that file in the old version of the code._

### Changed unit test files

- github.com/xataio/pgstream/pkg/transformers/json_transformer_test.go

</details>

github-actions[bot]: ### Merging this branch changes the coverage (1 decrease, 1 increase)

| Impacted Packages | Coverage Δ | :robot: |
|-------------------|------------|---------|
| github.com/xataio/pgstream/pkg/transformers | 84.87% (**+2.80%**) | :thumbsup: |
| github.com/xataio/pgstream/pkg/transformers/builder | 38.46% (**-1.01%**) | :thumbsdown: |

---

<details>

<summary>Coverage by file</summary>

### Changed files (no unit tests)

| Changed File | Coverage Δ | Total | Covered | Missed | :robot: |
|--------------|------------|-------|---------|--------|---------|
| github.com/xataio/pgstream/pkg/transformers/builder/transformer_builder.go | 38.46% (**-1.01%**) | 39 (+1) | 15 | 24 (+1) | :thumbsdown: |
| github.com/xataio/pgstream/pkg/transformers/json_transformer.go | 91.51% (**+91.51%**) | 106 (+106) | 97 (+97) | 9 (+9) | :star2: |
| github.com/xataio/pgstream/pkg/transformers/transformer.go | 86.89% (ø) | 61 | 53 | 8 |  |

_Please note that the "Total", "Covered", and "Missed" counts above refer to ***code statements*** instead of lines of code. The value in brackets refers to the test coverage of that file in the old version of the code._

### Changed unit test files

- github.com/xataio/pgstream/pkg/transformers/json_transformer_test.go

</details>

github-actions[bot]: ### Merging this branch changes the coverage (1 decrease, 1 increase)

| Impacted Packages | Coverage Δ | :robot: |
|-------------------|------------|---------|
| github.com/xataio/pgstream/pkg/transformers | 84.87% (**+2.80%**) | :thumbsup: |
| github.com/xataio/pgstream/pkg/transformers/builder | 38.46% (**-1.01%**) | :thumbsdown: |

---

<details>

<summary>Coverage by file</summary>

### Changed files (no unit tests)

| Changed File | Coverage Δ | Total | Covered | Missed | :robot: |
|--------------|------------|-------|---------|--------|---------|
| github.com/xataio/pgstream/pkg/transformers/builder/transformer_builder.go | 38.46% (**-1.01%**) | 39 (+1) | 15 | 24 (+1) | :thumbsdown: |
| github.com/xataio/pgstream/pkg/transformers/json_transformer.go | 91.51% (**+91.51%**) | 106 (+106) | 97 (+97) | 9 (+9) | :star2: |
| github.com/xataio/pgstream/pkg/transformers/transformer.go | 86.89% (ø) | 61 | 53 | 8 |  |

_Please note that the "Total", "Covered", and "Missed" counts above refer to ***code statements*** instead of lines of code. The value in brackets refers to the test coverage of that file in the old version of the code._

### Changed unit test files

- github.com/xataio/pgstream/pkg/transformers/json_transformer_test.go

</details>

github-actions[bot]: ### Merging this branch changes the coverage (1 decrease, 1 increase)

| Impacted Packages | Coverage Δ | :robot: |
|-------------------|------------|---------|
| github.com/xataio/pgstream/pkg/transformers | 84.76% (**+2.69%**) | :thumbsup: |
| github.com/xataio/pgstream/pkg/transformers/builder | 38.46% (**-1.01%**) | :thumbsdown: |

---

<details>

<summary>Coverage by file</summary>

### Changed files (no unit tests)

| Changed File | Coverage Δ | Total | Covered | Missed | :robot: |
|--------------|------------|-------|---------|--------|---------|
| github.com/xataio/pgstream/pkg/transformers/builder/transformer_builder.go | 38.46% (**-1.01%**) | 39 (+1) | 15 | 24 (+1) | :thumbsdown: |
| github.com/xataio/pgstream/pkg/transformers/json_transformer.go | 91.78% (**+91.78%**) | 73 (+73) | 67 (+67) | 6 (+6) | :star2: |
| github.com/xataio/pgstream/pkg/transformers/json_transformer_operation.go | 89.19% (**+89.19%**) | 37 (+37) | 33 (+33) | 4 (+4) | :star2: |
| github.com/xataio/pgstream/pkg/transformers/transformer.go | 86.89% (ø) | 61 | 53 | 8 |  |

_Please note that the "Total", "Covered", and "Missed" counts above refer to ***code statements*** instead of lines of code. The value in brackets refers to the test coverage of that file in the old version of the code._

### Changed unit test files

- github.com/xataio/pgstream/pkg/transformers/json_transformer_test.go
- github.com/xataio/pgstream/pkg/transformers/template_transformer_test.go

</details>


Patch:
diff --git a/docs/README.md b/docs/README.md
index cf7d065..fb57e97 100644
--- a/docs/README.md
+++ b/docs/README.md
@@ -1107,6 +1107,125 @@ transformations:
 
 #### Other Transformers
 
+ <details>
+  <summary>json</summary>
+
+**Description:** Transforms json data with set and delete operations
+
+| Supported PostgreSQL types |
+| -------------------------- |
+| json, jsonb                |
+
+| Parameter   | Type   | Default | Required |
+| ----------- | ------ | ------- | -------- |
+| operations  | array  | N/A     | Yes      |
+
+Parameter for each operation:
+| Parameter       | Type    | Default | Required | Values                         |
+| --------------- | ------- | ------- | -------- | ------------------------------ |
+| operation       | string  | N/A     | Yes      | set, delete                    |
+| path            | string  | N/A     | Yes      | *sjson syntax                  |
+| error_not_exist | boolean | false   | No       | true, false                    |
+| value           | string  | N\A     | **Yes    | Any valid JSON representation  |
+| value_template  | string  | N\A     | **Yes    | Any template with valid syntax |
+
+*Paths should follow [sjson syntax](https://github.com/tidwall/sjson#path-syntax)
+
+**Either `value` or `value_template` must be provided if the operation is `set`. If both are provided, `value_template` takes precedence.
+
+JSON transformer can be used for Postgres types json and jsonb. This transformer executes a list of given operations on the json data to be transformed.
+
+All operations must be either `set` or `delete`.
+`set` operations support literal values as well as templates, making use of `sprig` and `greenmask`'s function sets. See template transformer section for more details. Also, like the template transformer, `.GetValue` and `.GetDynamicValue` functions are supported. Unlike template transformer, here `.GetValue` refers to the value at given path, rather than the entire JSON object; whereas `.GetDynamicValue` is again used for referring to other columns.
+`delete` operations simply delete the object at the given path.
+Execution of an operation errors out if the given path does not exists and the parameter `error_not_exist` - which is false by default - is set to true.
+
+JSON transformer uses [sjson](https://github.com/tidwall/sjson) library for executing the operations. Operation paths should follow the [synxtax rules of sjson](https://github.com/tidwall/sjson#path-syntax)
+
+With the below config `pgstream` transforms the json values in the column `user_info_json` of the table `users` by:
+- First, traversing all the items in the array named `purchases`, and for each element, setting value to "-" for key "item".
+- Then, deleting the object named "country" under the top-level object "address".
+- Completely masking the "city" value under object "address", using `go-masker`'s default masking function supported by `pgstream`'s templating.
+- Finally, setting the user's lastname after fetching it from some other column named `lastname`, using dynamic values support. Assuming there's such column, having the lastname info for users.
+
+Example input-output is given below the config.
+
+**Example Configuration:**
+
+```yaml
+transformations:
+  table_transformers:
+    - schema: public
+      table: users
+      column_transformers:
+        user_info_json:
+          name: json
+          parameters:
+            operations:
+              - operation: set
+                path: "purchases.#.item"
+                value: "-"
+                error_not_exists: true
+              - operation: delete
+                path: "address.country"
+              - operation: set
+                path: "address.city"
+                value_template:  "\"{{ masking \"default\" .GetValue }}\""
+              - operation: set
+                path: "user.lastname"
+                value_template:  "\"{{ .GetDynamicValue \"lastname\" }}\""
+```
+
+For input JSON value,
+```json
+{
+  "user": {
+    "firstname": "john",
+    "lastname": "unknown"
+  },
+  "residency": {
+    "city": "some city",
+    "country": "some country"
+  },
+  "purchases": [
+    {
+      "item": "book",
+      "price": 10
+    },
+    {
+      "item": "pen",
+      "price": 2
+    }
+  ]
+}
+```
+
+the JSON transformer with above config produces output:
+```json
+{
+  "user": {
+    "firstname": "john",
+    "lastname": "doe"
+  },
+  "residency": {
+    "city": "*********"
+  },
+  "purchases": [
+    {
+      "item": "-",
+      "price": 10
+    },
+    {
+      "item": "-",
+      "price": 2
+    }
+  ]
+}
+```
+
+</details>
+
+
  <details>
   <summary>literal_string</summary>
 
diff --git a/pkg/transformers/builder/transformer_builder.go b/pkg/transformers/builder/transformer_builder.go
index e8518dc..c1424a0 100644
--- a/pkg/transformers/builder/transformer_builder.go
+++ b/pkg/transformers/builder/transformer_builder.go
@@ -48,6 +48,12 @@ var TransformersMap = map[transformers.TransformerType]struct {
 			return transformers.NewTemplateTransformer(cfg.Parameters)
 		},
 	},
+	transformers.JSON: {
+		Definition: transformers.JSONTransformerDefinition(),
+		BuildFn: func(cfg *transformers.Config) (transformers.Transformer, error) {
+			return transformers.NewJSONTransformer(cfg.Parameters)
+		},
+	},
 	transformers.PhoneNumber: {
 		Definition: transformers.PhoneNumberTransformerDefinition(),
 		BuildFn: func(cfg *transformers.Config) (transformers.Transformer, error) {
diff --git a/pkg/transformers/json_transformer.go b/pkg/transformers/json_transformer.go
new file mode 100644
index 0000000..13c91f5
--- /dev/null
+++ b/pkg/transformers/json_transformer.go
@@ -0,0 +1,196 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package transformers
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"fmt"
+	"slices"
+	"text/template"
+
+	"github.com/Masterminds/sprig/v3"
+	greenmasktoolkit "github.com/eminano/greenmask/pkg/toolkit"
+	"github.com/xataio/pgstream/internal/json"
+)
+
+const (
+	jsonDeleteOpName = "delete"
+	jsonSetOpName    = "set"
+)
+
+var (
+	errOperationsMustBeProvided      = errors.New("operations parameter must be provided")
+	errOperationsCannotBeEmpty       = errors.New("operations cannot be empty")
+	errInvalidOperationsType         = errors.New("unknown operation name, must be one of 'set' or 'delete'")
+	errOperationNameMustBeProvided   = errors.New("operation name must be provided in the operation definition")
+	errValueOrTemplateMustBeProvided = errors.New("either value or value_template must be provided in the 'set' operation definition")
+	errPathMustBeProvided            = errors.New("path must be provided in the operation definition")
+	jsonCompatibleTypes              = []SupportedDataType{
+		JSONDataType,
+	}
+	jsonParams = []Parameter{
+		{
+			Name:          "operations",
+			SupportedType: "array",
+			Default:       nil,
+			Dynamic:       false,
+			Required:      true,
+		},
+	}
+)
+
+type JSONTransformer struct {
+	operations []*jsonOperation
+	jsonVal    *jsonValue
+	buf        *bytes.Buffer
+}
+
+func NewJSONTransformer(params ParameterValues) (*JSONTransformer, error) {
+	operations, err := getOperationsParam(params)
+	if err != nil {
+		return nil, fmt.Errorf("json_transformer: error finding operations parameter: %w", err)
+	}
+
+	// prepare the template objects for operations that has template values
+	for idx, o := range operations {
+		if o.valueTemplate != "" {
+			tmpl, err := template.New(fmt.Sprintf("op[%d] %s %s", idx, o.operation, o.path)).
+				Funcs(greenmasktoolkit.FuncMap()).
+				Funcs(sprig.FuncMap()).
+				Parse(o.valueTemplate)
+			if err != nil {
+				return nil, fmt.Errorf("json_transformer: error parsing template op[%d] with path \"%s\": %w", idx, o.path, err)
+			}
+			o.tmpl = tmpl
+		}
+	}
+
+	return &JSONTransformer{
+		operations: operations,
+		buf:        bytes.NewBuffer(nil),
+		jsonVal:    &jsonValue{},
+	}, nil
+}
+
+func (jt *JSONTransformer) Transform(_ context.Context, value Value) (any, error) {
+	var err error
+	var toTransform []byte
+	switch val := value.TransformValue.(type) {
+	case []byte:
+		toTransform = val
+	case string:
+		toTransform = []byte(val)
+	default:
+		toTransform, err = json.Marshal(value.TransformValue)
+		if err != nil {
+			return nil, fmt.Errorf("json_transformer: error marshalling value to JSON: %w", err)
+		}
+	}
+	// set dynamic values for the jsonValue instance, to be used in templates
+	jt.jsonVal.setDynamicValues(value.DynamicValues)
+
+	res := slices.Clone(toTransform)
+	for idx, op := range jt.operations {
+		// apply each operation in the order they were provided
+		res, err = op.apply(res, jt.jsonVal, jt.buf)
+		if err != nil {
+			return nil, fmt.Errorf("cannot apply \"%s\" operation[%d] with path %s: %w", op.operation, idx, op.path, err)
+		}
+	}
+
+	var resJSON any
+	if err = json.Unmarshal(res, &resJSON); err != nil {
+		return nil, fmt.Errorf("json_transformer: error unmarshalling the result: %w; JSON: %s", err, string(res))
+	}
+	return resJSON, nil
+}
+
+func (jt *JSONTransformer) CompatibleTypes() []SupportedDataType {
+	return jsonCompatibleTypes
+}
+
+func (jt *JSONTransformer) Type() TransformerType {
+	return JSON
+}
+
+func JSONTransformerDefinition() *Definition {
+	return &Definition{
+		SupportedTypes: jsonCompatibleTypes,
+		Parameters:     jsonParams,
+	}
+}
+
+func getOperationsParam(params ParameterValues) ([]*jsonOperation, error) {
+	arrayAny, found, err := FindParameter[[]any](params, "operations")
+	if err != nil {
+		return nil, fmt.Errorf("operations must be an array: %w", err)
+	}
+	if !found {
+		return nil, errOperationsMustBeProvided
+	}
+	if len(arrayAny) == 0 {
+		return nil, errOperationsCannotBeEmpty
+	}
+
+	operations := make([]*jsonOperation, 0, len(arrayAny))
+	for _, valAny := range arrayAny {
+		val, ok := valAny.(map[string]any)
+		if !ok {
+			return nil, fmt.Errorf("invalid element type in operations array, got %T: %w", valAny, ErrInvalidParameters)
+		}
+
+		op := &jsonOperation{}
+		op.operation, found, err = FindParameter[string](val, "operation")
+		if err != nil {
+			return nil, fmt.Errorf("operation name must be a string: %w", err)
+		}
+		if !found {
+			return nil, errOperationNameMustBeProvided
+		}
+		if op.operation != jsonSetOpName && op.operation != jsonDeleteOpName {
+			return nil, errInvalidOperationsType
+		}
+
+		op.path, found, err = FindParameter[string](val, "path")
+		if err != nil {
+			return nil, fmt.Errorf("path must be a string: %w", err)
+		}
+		if !found {
+			return nil, errPathMustBeProvided
+		}
+
+		op.errorNotExist, err = FindParameterWithDefault(val, "error_not_exist", false)
+		if err != nil {
+			return nil, fmt.Errorf("error_not_exist must be a boolean: %w", err)
+		}
+
+		if op.operation == jsonDeleteOpName {
+			operations = append(operations, op)
+			continue
+		}
+
+		// get value for set operation
+		var valueTemplateFound, valueFound bool
+		op.value, valueFound, err = FindParameter[any](val, "value")
+		if err != nil {
+			return nil, fmt.Errorf("cannot read parameter 'value': %w", err)
+		}
+
+		// get value template for set operation
+		op.valueTemplate, valueTemplateFound, err = FindParameter[string](val, "value_template")
+		if err != nil {
+			return nil, fmt.Errorf("value_template must be a string: %w", err)
+		}
+
+		// ensure that at least one of value or value_template is provided
+		if !valueFound && !valueTemplateFound {
+			return nil, errValueOrTemplateMustBeProvided
+		}
+
+		operations = append(operations, op)
+	}
+
+	return operations, nil
+}
diff --git a/pkg/transformers/json_transformer_operation.go b/pkg/transformers/json_transformer_operation.go
new file mode 100644
index 0000000..d16d308
--- /dev/null
+++ b/pkg/transformers/json_transformer_operation.go
@@ -0,0 +1,105 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package transformers
+
+import (
+	"bytes"
+	"fmt"
+	"text/template"
+
+	"github.com/tidwall/gjson"
+	"github.com/tidwall/sjson"
+)
+
+var (
+	jsonSetOpt = &sjson.Options{
+		ReplaceInPlace: true,
+	}
+	errDynamicValuesNil = fmt.Errorf("dynamic values are nil")
+)
+
+type jsonOperation struct {
+	operation     string
+	value         any
+	valueTemplate string
+	path          string
+	errorNotExist bool
+	tmpl          *template.Template
+}
+
+func (o *jsonOperation) apply(inp []byte, jsonVal *jsonValue, buf *bytes.Buffer) ([]byte, error) {
+	var res []byte
+	var err error
+
+	switch o.operation {
+	case jsonSetOpName:
+		if o.tmpl != nil {
+			buf.Reset()
+			jsonVal.setValue(inp, o.path)
+			if o.errorNotExist && !jsonVal.exists {
+				return nil, fmt.Errorf("value by path \"%s\" does not exist", o.path)
+			}
+			if err = o.tmpl.Execute(buf, jsonVal); err != nil {
+				return nil, fmt.Errorf("error executing template: %w", err)
+			}
+			newValue := buf.Bytes()
+			res, err = sjson.SetRawBytesOptions(inp, o.path, newValue, jsonSetOpt)
+			if err != nil {
+				return nil, fmt.Errorf("error applying set raw operation: %w", err)
+			}
+		} else {
+			res, err = sjson.SetBytesOptions(inp, o.path, o.value, jsonSetOpt)
+			if err != nil {
+				return nil, fmt.Errorf("error applying set operation: %w", err)
+			}
+		}
+
+	case jsonDeleteOpName:
+		if o.errorNotExist && !gjson.GetBytes(inp, o.path).Exists() {
+			return nil, fmt.Errorf("value by path \"%s\" does not exist", o.path)
+		}
+		res, err = sjson.DeleteBytes(inp, o.path)
+		if err != nil {
+			return nil, fmt.Errorf("error applying delete operation: %w", err)
+		}
+
+	default:
+		return nil, fmt.Errorf("unknown operation %s", o.operation)
+	}
+
+	return res, nil
+}
+
+type jsonValue struct {
+	exists        bool
+	value         any
+	dynamicValues map[string]any
+}
+
+func (jv *jsonValue) setValue(data []byte, path string) {
+	res := gjson.GetBytes(data, path)
+	jv.value = res.Value()
+	jv.exists = res.Exists()
+}
+
+func (jv *jsonValue) setDynamicValues(values map[string]any) {
+	jv.dynamicValues = values
+}
+
+func (jv *jsonValue) GetValue() any {
+	if !jv.exists {
+		return nil
+	}
+	return jv.value
+}
+
+func (jv *jsonValue) GetDynamicValue(key string) (any, error) {
+	if jv.dynamicValues == nil {
+		return nil, errDynamicValuesNil
+	}
+	dynValue, found := jv.dynamicValues[key]
+	if !found {
+		return nil, fmt.Errorf("dynamic value '%s' not found", key)
+	}
+	return dynValue, nil
+}
diff --git a/pkg/transformers/transformer.go b/pkg/transformers/transformer.go
index 0dd5460..a5e3731 100644
--- a/pkg/transformers/transformer.go
+++ b/pkg/transformers/transformer.go
@@ -52,6 +52,7 @@ const (
 	GreenmaskUTCTimestamp  TransformerType = "greenmask_utc_timestamp"
 	Masking                TransformerType = "masking"
 	Template               TransformerType = "template"
+	JSON                   TransformerType = "json"
 )
 
 type SupportedDataType string
diff --git a/transformers-definition.json b/transformers-definition.json
index 66a6e85..c8b4fcc 100644
--- a/transformers-definition.json
+++ b/transformers-definition.json
@@ -370,6 +370,21 @@
         }
       ]
     },
+    {
+      "name": "json",
+      "supported_types": [
+        "json"
+      ],
+      "parameters": [
+        {
+          "name": "operations",
+          "supported_type": "array",
+          "default": null,
+          "dynamic": false,
+          "required": true
+        }
+      ]
+    },
     {
       "name": "literal_string",
       "supported_types": [


Test patch:
diff --git a/pkg/transformers/json_transformer_test.go b/pkg/transformers/json_transformer_test.go
new file mode 100644
index 0000000..9b92ff2
--- /dev/null
+++ b/pkg/transformers/json_transformer_test.go
@@ -0,0 +1,495 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package transformers
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"github.com/xataio/pgstream/internal/json"
+)
+
+func TestNewJsonTransformer(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name    string
+		params  ParameterValues
+		wantErr error
+	}{
+		{
+			name: "ok - valid",
+			params: ParameterValues{
+				"operations": []any{
+					map[string]any{
+						"operation":       "set",
+						"path":            "/greeting",
+						"value":           "hello world",
+						"error_not_exist": false,
+					},
+					map[string]any{
+						"operation":       "delete",
+						"path":            "/farewell",
+						"error_not_exist": true,
+					},
+				},
+			},
+			wantErr: nil,
+		},
+		{
+			name: "error - operations invalid type",
+			params: ParameterValues{
+				"operations": "not an array",
+			},
+			wantErr: errors.New("operations must be an array"),
+		},
+		{
+			name:    "error - operations not provided",
+			params:  ParameterValues{},
+			wantErr: errOperationsMustBeProvided,
+		},
+		{
+			name: "error - operations empty",
+			params: ParameterValues{
+				"operations": []any{},
+			},
+			wantErr: errOperationsCannotBeEmpty,
+		},
+		{
+			name: "error - invalid operation type",
+			params: ParameterValues{
+				"operations": []any{
+					map[string]any{
+						"operation":       3,
+						"path":            "/greeting",
+						"value":           "hello world",
+						"error_not_exist": false,
+					},
+				},
+			},
+			wantErr: ErrInvalidParameters,
+		},
+		{
+			name: "error - operation name not provided",
+			params: ParameterValues{
+				"operations": []any{
+					map[string]any{
+						"path":            "/greeting",
+						"value":           "hello world",
+						"error_not_exist": false,
+					},
+				},
+			},
+			wantErr: errOperationNameMustBeProvided,
+		},
+		{
+			name: "error - invalid operation name",
+			params: ParameterValues{
+				"operations": []any{
+					map[string]any{
+						"operation":       "not_a_valid_operation",
+						"path":            "/greeting",
+						"value":           "hello world",
+						"error_not_exist": false,
+					},
+				},
+			},
+			wantErr: errInvalidOperationsType,
+		},
+		{
+			name: "error - invalid path type",
+			params: ParameterValues{
+				"operations": []any{
+					map[string]any{
+						"operation":       "set",
+						"path":            3,
+						"value":           "hello world",
+						"error_not_exist": false,
+					},
+				},
+			},
+			wantErr: ErrInvalidParameters,
+		},
+		{
+			name: "error - path not provided",
+			params: ParameterValues{
+				"operations": []any{
+					map[string]any{
+						"operation":       "set",
+						"value":           "hello world",
+						"error_not_exist": false,
+					},
+				},
+			},
+			wantErr: errPathMustBeProvided,
+		},
+		{
+			name: "error - invalid error_not_exist type",
+			params: ParameterValues{
+				"operations": []any{
+					map[string]any{
+						"operation":       "delete",
+						"path":            "/greeting",
+						"error_not_exist": "invalid",
+					},
+				},
+			},
+			wantErr: ErrInvalidParameters,
+		},
+		{
+			name: "error - invalid value type",
+			params: ParameterValues{
+				"operations": []any{
+					map[string]any{
+						"operation":       "set",
+						"path":            "/greeting",
+						"value":           nil,
+						"error_not_exist": false,
+					},
+				},
+			},
+			wantErr: errors.New("cannot read parameter 'value'"),
+		},
+		{
+			name: "error - invalid value_template type",
+			params: ParameterValues{
+				"operations": []any{
+					map[string]any{
+						"operation":       "set",
+						"path":            "/greeting",
+						"value_template":  3,
+						"error_not_exist": false,
+					},
+				},
+			},
+			wantErr: ErrInvalidParameters,
+		},
+		{
+			name: "error - both value and value_template absent",
+			params: ParameterValues{
+				"operations": []any{
+					map[string]any{
+						"operation":       "set",
+						"path":            "/greeting",
+						"error_not_exist": false,
+					},
+				},
+			},
+			wantErr: errValueOrTemplateMustBeProvided,
+		},
+		{
+			name: "error - template cannot be parsed",
+			params: ParameterValues{
+				"operations": []any{
+					map[string]any{
+						"operation":       "set",
+						"path":            "/greeting",
+						"error_not_exist": false,
+						"value_template":  "{{ invalid template syntax",
+					},
+				},
+			},
+			wantErr: errors.New("json_transformer: error parsing template"),
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			tt, err := NewJSONTransformer(tc.params)
+			if tc.wantErr != nil {
+				require.Error(t, err)
+				if !errors.Is(err, tc.wantErr) {
+					require.Contains(t, err.Error(), tc.wantErr.Error())
+				}
+				return
+			}
+			require.NoError(t, err)
+			require.NotNil(t, tt)
+		})
+	}
+}
+
+func TestJsonTransformer_Transform(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name          string
+		value         any
+		dynamicValues map[string]any
+		params        ParameterValues
+
+		wantOutput         any
+		wantOutputContains string
+		wantErr            error
+	}{
+		{
+			name:  "ok - basic set and delete operations",
+			value: map[string]any{"greeting": "hello", "farewell": "goodbye"},
+			params: ParameterValues{
+				"operations": []any{
+					map[string]any{
+						"operation":       "set",
+						"path":            "greeting",
+						"value":           "hello world",
+						"error_not_exist": false,
+					},
+					map[string]any{
+						"operation":       "delete",
+						"path":            "farewell",
+						"error_not_exist": true,
+					},
+				},
+			},
+			wantOutput: map[string]any{
+				"greeting": "hello world",
+			},
+			wantErr: nil,
+		},
+		{
+			name:  "ok - basic set and delete operations, []byte input",
+			value: []byte(`{"greeting":"hello", "farewell":"goodbye"}`),
+			params: ParameterValues{
+				"operations": []any{
+					map[string]any{
+						"operation":       "set",
+						"path":            "greeting",
+						"value":           "hello world",
+						"error_not_exist": false,
+					},
+					map[string]any{
+						"operation":       "delete",
+						"path":            "farewell",
+						"error_not_exist": true,
+					},
+				},
+			},
+			wantOutput: map[string]any{
+				"greeting": "hello world",
+			},
+			wantErr: nil,
+		},
+		{
+			name:  "ok - set and delete on array",
+			value: []any{map[string]any{"greeting": "hello", "farewell": "goodbye"}},
+			params: ParameterValues{
+				"operations": []any{
+					map[string]any{
+						"operation":       "set",
+						"path":            "0.greeting",
+						"value":           "hello world",
+						"error_not_exist": false,
+					},
+					map[string]any{
+						"operation":       "set",
+						"path":            "1.greeting",
+						"value":           "hello again",
+						"error_not_exist": true,
+					},
+					map[string]any{
+						"operation":       "delete",
+						"path":            "0.farewell",
+						"error_not_exist": true,
+					},
+				},
+			},
+			wantOutput: []any{map[string]any{"greeting": "hello world"}, map[string]any{"greeting": "hello again"}},
+			wantErr:    nil,
+		},
+		{
+			name: "ok - set and delete, with dynamic values, template with masking, complex path",
+			value: map[string]any{
+				"user": map[string]any{
+					"firstname": "john",
+					"lastname":  "unknown",
+				},
+				"residency": map[string]any{
+					"city":    "some city",
+					"country": "some country",
+				},
+				"purchases": []any{
+					map[string]any{"item": "book", "price": 10},
+					map[string]any{"item": "pen", "price": 2},
+				},
+			},
+			params: ParameterValues{
+				"operations": []any{
+					map[string]any{
+						"operation":       "set",
+						"path":            "purchases.#.item",
+						"value":           "-",
+						"error_not_exist": true,
+					},
+					map[string]any{
+						"operation":       "set",
+						"path":            "user.lastname",
+						"value_template":  "\"{{ .GetDynamicValue \"lastname\" }}\"",
+						"error_not_exist": true,
+					},
+					map[string]any{
+						"operation":       "set",
+						"path":            "residency.city",
+						"value_template":  "\"{{ masking \"default\" .GetValue }}\"",
+						"error_not_exist": true,
+					},
+					map[string]any{
+						"operation":       "delete",
+						"path":            "residency.country",
+						"error_not_exist": true,
+					},
+				},
+			},
+			dynamicValues: map[string]any{
+				"lastname": "doe",
+				"column":   3.1415,
+			},
+			wantOutput: map[string]any{
+				"user": map[string]any{
+					"firstname": "john",
+					"lastname":  "doe",
+				},
+				"residency": map[string]any{
+					"city": "*********",
+				},
+				"purchases": []any{
+					map[string]any{"item": "-", "price": float64(10)},
+					map[string]any{"item": "-", "price": float64(2)},
+				},
+			},
+			wantErr: nil,
+		},
+		{
+			name:  "error - cannot marshal value",
+			value: make(chan int), // invalid type for JSON
+			params: ParameterValues{
+				"operations": []any{
+					map[string]any{
+						"operation":       "set",
+						"path":            "0.greeting",
+						"value":           "hi",
+						"error_not_exist": true,
+					},
+				},
+			},
+			wantErr: errors.New("json_transformer: error marshalling value to JSON"),
+		},
+		{
+			name:  "error - setting invalid value for json",
+			value: []any{map[string]any{"greeting": "hello", "farewell": "goodbye"}},
+			params: ParameterValues{
+				"operations": []any{
+					map[string]any{
+						"operation":       "set",
+						"path":            "0.greeting",
+						"value":           make(chan int),
+						"error_not_exist": true,
+					},
+				},
+			},
+			wantErr: errors.New("json: unsupported type: chan int"),
+		},
+		{
+			name:  "error - set path not found in object",
+			value: map[string]any{"greeting": "hello"},
+			params: ParameterValues{
+				"operations": []any{
+					map[string]any{
+						"operation":       "set",
+						"path":            "farewell",
+						"value_template":  "goodbye",
+						"error_not_exist": true,
+					},
+				},
+			},
+			wantErr: errors.New("value by path \"farewell\" does not exist"),
+		},
+		{
+			name:  "error - delete path not found in array",
+			value: []any{map[string]any{"greeting": "hello", "farewell": "goodbye"}},
+			params: ParameterValues{
+				"operations": []any{
+					map[string]any{
+						"operation":       "delete",
+						"path":            "1.farewell",
+						"error_not_exist": true,
+					},
+				},
+			},
+			wantErr: errors.New("value by path \"1.farewell\" does not exist"),
+		},
+		{
+			name:  "error - invalid template, quotes missing",
+			value: map[string]any{"greeting": "hello"},
+			params: ParameterValues{
+				"operations": []any{
+					map[string]any{
+						"operation":      "set",
+						"path":           "farewell",
+						"value_template": "{{ masking default .GetValue }}",
+					},
+				},
+			},
+			wantErr: errors.New("error executing template"),
+		},
+		{
+			name:  "error - dynamic values missing",
+			value: map[string]any{"greeting": "hello"},
+			params: ParameterValues{
+				"operations": []any{
+					map[string]any{
+						"operation":      "set",
+						"path":           "farewell",
+						"value_template": "{{ .GetDynamicValue \"notexists\" }}",
+					},
+				},
+			},
+			wantErr: errors.New("dynamic values are nil"),
+		},
+		{
+			name:  "error - dynamic value not found",
+			value: map[string]any{"greeting": "hello"},
+			params: ParameterValues{
+				"operations": []any{
+					map[string]any{
+						"operation":      "set",
+						"path":           "farewell",
+						"value_template": "{{ .GetDynamicValue \"notexists\" }}",
+					},
+				},
+			},
+			dynamicValues: map[string]any{"exists": "value"},
+			wantErr:       errors.New("dynamic value 'notexists' not found"),
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			tt, err := NewJSONTransformer(tc.params)
+			require.NoError(t, err)
+			got, err := tt.Transform(context.Background(), Value{TransformValue: tc.value, DynamicValues: tc.dynamicValues})
+
+			if tc.wantErr != nil {
+				require.Error(t, err)
+				if !errors.Is(err, tc.wantErr) {
+					require.Contains(t, err.Error(), tc.wantErr.Error())
+				}
+				return
+			}
+			require.NoError(t, err)
+
+			if tc.wantOutputContains != "" {
+				marshalled, err := json.Marshal(&got)
+				require.NoError(t, err)
+				require.Contains(t, string(marshalled), tc.wantOutputContains)
+			}
+			if tc.wantOutput != nil {
+				require.Equal(t, tc.wantOutput, got)
+			}
+		})
+	}
+}
diff --git a/pkg/transformers/template_transformer_test.go b/pkg/transformers/template_transformer_test.go
index d034e6a..c2c0b9c 100644
--- a/pkg/transformers/template_transformer_test.go
+++ b/pkg/transformers/template_transformer_test.go
@@ -174,7 +174,7 @@ func TestTemplateTransformer_Transform_WithDynamicValues(t *testing.T) {
 				"template": "{{- if eq .GetValue \"hello\" -}} {{.GetDynamicValue \"value1\" }} {{- else -}} {{.GetDynamicValue \"value2\" }} {{- end -}}",
 			},
 			wantOutput: "",
-			wantErr:    errors.New("dynamic values are nil"),
+			wantErr:    errDynamicValuesNil,
 		},
 		{
 			name:  "error - dynamic value not found",