Repository: xataio/pgstream
Pull request #217 (2025-03-28 12:54:13)

Add masking support

github-actions[bot]: ### Merging this branch will **decrease** overall coverage

| Impacted Packages | Coverage Δ | :robot: |
|-------------------|------------|---------|
| github.com/xataio/pgstream/pkg/transformers | 98.18% (**-1.82%**) | :thumbsdown: |
| github.com/xataio/pgstream/pkg/transformers/builder | 0.00% (ø) |  |

---

<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 | 0.00% (ø) | 17 (+1) | 0 | 17 (+1) |  |
| github.com/xataio/pgstream/pkg/transformers/masking_transformer.go | 96.00% (**+96.00%**) | 25 (+25) | 24 (+24) | 1 (+1) | :star2: |
| github.com/xataio/pgstream/pkg/transformers/transformer.go | 100.00% (ø) | 20 | 20 | 0 |  |

_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/masking_transformer_test.go

</details>

github-actions[bot]: ### Merging this branch will **not change** overall coverage

| Impacted Packages | Coverage Δ | :robot: |
|-------------------|------------|---------|
| github.com/xataio/pgstream/pkg/transformers | 100.00% (ø) |  |
| github.com/xataio/pgstream/pkg/transformers/builder | 0.00% (ø) |  |

---

<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 | 0.00% (ø) | 17 (+1) | 0 | 17 (+1) |  |
| github.com/xataio/pgstream/pkg/transformers/masking_transformer.go | 100.00% (**+100.00%**) | 25 (+25) | 25 (+25) | 0 | :star2: |
| github.com/xataio/pgstream/pkg/transformers/transformer.go | 100.00% (ø) | 20 | 20 | 0 |  |

_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/masking_transformer_test.go

</details>

github-actions[bot]: ### Merging this branch will **not change** overall coverage

| Impacted Packages | Coverage Δ | :robot: |
|-------------------|------------|---------|
| github.com/xataio/pgstream/pkg/stream/integration | 0.00% (ø) |  |
| github.com/xataio/pgstream/pkg/transformers | 100.00% (ø) |  |
| github.com/xataio/pgstream/pkg/transformers/builder | 0.00% (ø) |  |

---

<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 | 0.00% (ø) | 17 (+1) | 0 | 17 (+1) |  |
| github.com/xataio/pgstream/pkg/transformers/masking_transformer.go | 100.00% (**+100.00%**) | 25 (+25) | 25 (+25) | 0 | :star2: |
| github.com/xataio/pgstream/pkg/transformers/transformer.go | 100.00% (ø) | 20 | 20 | 0 |  |

_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/stream/integration/pg_pg_integration_transformer_test.go
- github.com/xataio/pgstream/pkg/transformers/masking_transformer_test.go

</details>

eminano: Ah, and don't forget to update the documentation as well! 📚 
https://github.com/xataio/pgstream/blob/main/docs/README.md#supported-transformers
github-actions[bot]: ### Merging this branch will **not change** overall coverage

| Impacted Packages | Coverage Δ | :robot: |
|-------------------|------------|---------|
| github.com/xataio/pgstream/pkg/stream/integration | 0.00% (ø) |  |
| github.com/xataio/pgstream/pkg/transformers | 100.00% (ø) |  |
| github.com/xataio/pgstream/pkg/transformers/builder | 0.00% (ø) |  |

---

<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 | 0.00% (ø) | 17 (+1) | 0 | 17 (+1) |  |
| github.com/xataio/pgstream/pkg/transformers/masking_transformer.go | 100.00% (**+100.00%**) | 25 (+25) | 25 (+25) | 0 | :star2: |
| github.com/xataio/pgstream/pkg/transformers/transformer.go | 100.00% (ø) | 20 | 20 | 0 |  |

_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/stream/integration/pg_pg_integration_transformer_test.go
- github.com/xataio/pgstream/pkg/transformers/masking_transformer_test.go

</details>

github-actions[bot]: ### Merging this branch will **not change** overall coverage

| Impacted Packages | Coverage Δ | :robot: |
|-------------------|------------|---------|
| github.com/xataio/pgstream/pkg/stream/integration | 0.00% (ø) |  |
| github.com/xataio/pgstream/pkg/transformers | 100.00% (ø) |  |
| github.com/xataio/pgstream/pkg/transformers/builder | 0.00% (ø) |  |

---

<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 | 0.00% (ø) | 17 (+1) | 0 | 17 (+1) |  |
| github.com/xataio/pgstream/pkg/transformers/masking_transformer.go | 100.00% (**+100.00%**) | 25 (+25) | 25 (+25) | 0 | :star2: |
| github.com/xataio/pgstream/pkg/transformers/transformer.go | 100.00% (ø) | 20 | 20 | 0 |  |

_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/stream/integration/pg_pg_integration_transformer_test.go
- github.com/xataio/pgstream/pkg/transformers/masking_transformer_test.go

</details>


Patch:
diff --git a/docs/README.md b/docs/README.md
index b2dc0fa..48bebe7 100644
--- a/docs/README.md
+++ b/docs/README.md
@@ -263,12 +263,52 @@ For details on how to use and configure the snapshot mode, check the [snapshot t
 
 ![transformer diagram](img/pgstream_transformer_diagram.svg)
 
-`pgstream` supports column value transformations to anonymize sensitive data during replication and snapshots. This is particularly useful for compliance with data privacy regulations.
+`pgstream` supports column value transformations to anonymize or mask sensitive data during replication and snapshots. This is particularly useful for compliance with data privacy regulations.
 
-`pgstream` integrates with existing transformer open source libraries, such as [greenmask](https://github.com/GreenmaskIO/greenmask) and [neosync](https://github.com/nucleuscloud/neosync), to leverage a large amount of transformation capabilities, as well as having support for custom transformations.
+`pgstream` integrates with existing transformer open source libraries, such as [greenmask](https://github.com/GreenmaskIO/greenmask), [neosync](https://github.com/nucleuscloud/neosync) and [go-masker](https://github.com/ggwhite/go-masker), to leverage a large amount of transformation capabilities, as well as having support for custom transformations.
 
 ### Supported transformers
 
+#### go-masker
+ <details>
+  <summary>masking</summary>
+
+**Description:** Masks string values using the provided masking function.
+
+| Supported PostgreSQL types          |
+| ----------------------------------- |
+| `text`, `varchar`, `char`, `bpchar` |
+
+**Parameter Details:**
+
+| Parameter | Type   | Default  | Required | Values                                                                     |
+| --------- | ------ | -------- | -------- | -------------------------------------------------------------------------- |
+| type      | string | default  | No       | password, name, address, email, mobile, tel, id, credit_card, url, default |
+
+
+**Example Configuration:**
+
+```yaml
+transformations:
+  - schema: public
+    table: users
+    column_transformers:
+      email:
+        name: masking
+        parameters:
+          type: email
+```
+
+**Input-Output Examples:**
+
+| Input Value              | Configuration Parameters | Output Value           |
+| ------------------------ | ------------------------ | ---------------------- |
+| `aVeryStrongPassword123` | `type: password`         | `************`         |
+| `john.doe@example.com`   | `type: email`            | `joh****e@example.com` |
+| `Sensitive Data`         | `type: default`          | `**************`       |
+
+</details>
+
 #### Greenmask
 
  <details>
diff --git a/go.mod b/go.mod
index 34ef477..e185897 100644
--- a/go.mod
+++ b/go.mod
@@ -6,6 +6,7 @@ require (
 	github.com/cenkalti/backoff/v4 v4.3.0
 	github.com/elastic/go-elasticsearch/v8 v8.14.0
 	github.com/eminano/greenmask v0.0.0-20250307113752-035ee2b102e6
+	github.com/ggwhite/go-masker v1.1.0
 	github.com/go-logr/zerologr v1.2.3
 	github.com/golang-migrate/migrate/v4 v4.18.2
 	github.com/google/go-cmp v0.7.0
@@ -70,7 +71,6 @@ require (
 	github.com/expr-lang/expr v1.17.0 // indirect
 	github.com/felixge/httpsnoop v1.0.4 // indirect
 	github.com/fsnotify/fsnotify v1.8.0 // indirect
-	github.com/ggwhite/go-masker v1.1.0 // indirect
 	github.com/go-faker/faker/v4 v4.5.0 // indirect
 	github.com/go-logr/logr v1.4.2 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
diff --git a/pkg/transformers/builder/transformer_builder.go b/pkg/transformers/builder/transformer_builder.go
index ef07a46..b7d25bb 100644
--- a/pkg/transformers/builder/transformer_builder.go
+++ b/pkg/transformers/builder/transformer_builder.go
@@ -38,6 +38,8 @@ func New(cfg *transformers.Config) (transformers.Transformer, error) {
 		return neosync.NewFirstNameTransformer(cfg.Parameters)
 	case transformers.NeosyncEmail:
 		return neosync.NewEmailTransformer(cfg.Parameters)
+	case transformers.Masking:
+		return transformers.NewMaskingTransformer(cfg.Parameters)
 	default:
 		return nil, transformers.ErrUnsupportedTransformer
 	}
diff --git a/pkg/transformers/masking_transformer.go b/pkg/transformers/masking_transformer.go
new file mode 100644
index 0000000..d95db39
--- /dev/null
+++ b/pkg/transformers/masking_transformer.go
@@ -0,0 +1,89 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package transformers
+
+import (
+	"errors"
+	"fmt"
+	"strings"
+
+	"github.com/ggwhite/go-masker"
+)
+
+const (
+	mPassword   string = "password"
+	mName       string = "name"
+	mAddress    string = "address"
+	mEmail      string = "email"
+	mMobile     string = "mobile"
+	mTelephone  string = "tel"
+	mID         string = "id"
+	mCreditCard string = "credit_card"
+	mURL        string = "url"
+	mDefault    string = "default"
+)
+
+var errInvalidMaskingType = errors.New("masking_type must be one of 'password', 'name', 'address', 'email', 'mobile', 'tel', 'id', 'credit_card', 'url' or 'default'")
+
+type maskingFunction func(val string) string
+
+// MaskingTransformer is a transformer that masks sensitive data using the Masker library.
+type MaskingTransformer struct {
+	maskingFunction maskingFunction
+}
+
+// NewMaskingTransformer creates a new MaskingTransformer with the given masking function.
+func NewMaskingTransformer(params Parameters) (*MaskingTransformer, error) {
+	var mf maskingFunction
+	maskType, found, err := FindParameter[string](params, "type")
+	if err != nil {
+		return nil, fmt.Errorf("masking: type must be a string: %w", err)
+	}
+	if !found {
+		maskType = mDefault
+	}
+	m := masker.New()
+	switch maskType {
+	case mPassword:
+		mf = m.Password
+	case mName:
+		mf = m.Name
+	case mAddress:
+		mf = m.Address
+	case mEmail:
+		mf = m.Email
+	case mMobile:
+		mf = m.Mobile
+	case mID:
+		mf = m.ID
+	case mTelephone:
+		mf = m.Telephone
+	case mCreditCard:
+		mf = m.CreditCard
+	case mURL:
+		mf = m.URL
+	case mDefault:
+		mf = func(v string) string {
+			return strings.Repeat("*", len(v))
+		}
+	default:
+		return nil, errInvalidMaskingType
+	}
+
+	return &MaskingTransformer{
+		maskingFunction: mf,
+	}, nil
+}
+
+// Transform applies the masking function to the input value and returns the masked value.
+func (t *MaskingTransformer) Transform(value any) (any, error) {
+	switch val := value.(type) {
+	case string:
+		return t.maskingFunction(val), nil
+	case []byte:
+		return t.maskingFunction(string(val)), nil
+	// TODO: support more types, e.g. uuid.UUID
+	default:
+		return nil, ErrUnsupportedValueType
+	}
+}
diff --git a/pkg/transformers/transformer.go b/pkg/transformers/transformer.go
index bf6e477..9c4acce 100644
--- a/pkg/transformers/transformer.go
+++ b/pkg/transformers/transformer.go
@@ -33,6 +33,7 @@ const (
 	GreenmaskUnixTimestamp TransformerType = "greenmask_unix_timestamp"
 	GreenmaskDate          TransformerType = "greenmask_date"
 	GreenmaskUTCTimestamp  TransformerType = "greenmask_utc_timestamp"
+	Masking                TransformerType = "masking"
 )
 
 type Parameters map[string]any


Test patch:
diff --git a/pkg/stream/integration/config/integration_test_transformer_rules.yaml b/pkg/stream/integration/config/integration_test_transformer_rules.yaml
index 59de557..9e2cce1 100644
--- a/pkg/stream/integration/config/integration_test_transformer_rules.yaml
+++ b/pkg/stream/integration/config/integration_test_transformer_rules.yaml
@@ -18,6 +18,10 @@ transformations:
           preserve_length: false
           max_length: 15
           excluded_domains: ["example.com", "example.net"]
+      secondary_email:
+        name: masking
+        parameters:
+          type: email
       address:
         name: greenmask_string
         parameters:
diff --git a/pkg/stream/integration/pg_pg_integration_transformer_test.go b/pkg/stream/integration/pg_pg_integration_transformer_test.go
index a53a4dd..ecde0ce 100644
--- a/pkg/stream/integration/pg_pg_integration_transformer_test.go
+++ b/pkg/stream/integration/pg_pg_integration_transformer_test.go
@@ -19,6 +19,7 @@ type transformerTestTableRow struct {
 	name           string
 	lastName       string
 	email          string
+	secondaryEmail string
 	address        string
 	age            int
 	totalPurchases float64
@@ -35,6 +36,7 @@ var createTableQuery = `CREATE TABLE %s(
 	name text,
 	last_name varchar(255),
 	email varchar(255),
+	secondary_email varchar(255),
 	address text,
 	age integer,
 	total_purchases double precision,
@@ -78,6 +80,7 @@ func Test_PostgresToPostgres_Transformer(t *testing.T) {
 					name:           "John",
 					lastName:       "Doe",
 					email:          "john.doe@example.com",
+					secondaryEmail: "john.doe2@example.com",
 					address:        "123 Main St",
 					age:            30,
 					totalPurchases: 1000.50,
@@ -124,19 +127,19 @@ func Test_PostgresToPostgres_Transformer(t *testing.T) {
 }
 
 func insertQuery(table string, rows []transformerTestTableRow) string {
-	query := fmt.Sprintf("INSERT INTO %s (name, last_name, email, address, age, total_purchases, customer_id, birth_date, is_active, created_at, updated_at, gender) VALUES", table)
+	query := fmt.Sprintf("INSERT INTO %s (name, last_name, email, secondary_email, address, age, total_purchases, customer_id, birth_date, is_active, created_at, updated_at, gender) VALUES", table)
 	for i, row := range rows {
 		if i > 0 {
 			query += ","
 		}
-		query += fmt.Sprintf(" ('%s', '%s', '%s', '%s', %d, %f, '%s', '%s', %t, %d, '%s', '%s')",
-			row.name, row.lastName, row.email, row.address, row.age, row.totalPurchases, row.customerID, row.birthDate.Format(time.DateOnly), row.isActive, row.createdAt, row.updatedAt.Format(time.RFC3339), row.gender)
+		query += fmt.Sprintf(" ('%s', '%s', '%s', '%s', '%s', %d, %f, '%s', '%s', %t, %d, '%s', '%s')",
+			row.name, row.lastName, row.email, row.secondaryEmail, row.address, row.age, row.totalPurchases, row.customerID, row.birthDate.Format(time.DateOnly), row.isActive, row.createdAt, row.updatedAt.Format(time.RFC3339), row.gender)
 	}
 	return query
 }
 
 func validateRows(t *testing.T, ctx context.Context, conn *pglib.Conn, expectedRows []int, table string) bool {
-	selectQuery := fmt.Sprintf("SELECT id, name, last_name, email, address, age, total_purchases, customer_id, birth_date, is_active, created_at, updated_at, gender FROM %s WHERE id IN (", table)
+	selectQuery := fmt.Sprintf("SELECT id, name, last_name, email, secondary_email, address, age, total_purchases, customer_id, birth_date, is_active, created_at, updated_at, gender FROM %s WHERE id IN (", table)
 	for i, rowID := range expectedRows {
 		if i > 0 {
 			selectQuery += " ,"
@@ -151,7 +154,7 @@ func validateRows(t *testing.T, ctx context.Context, conn *pglib.Conn, expectedR
 	rowsFromDB := []transformerTestTableRow{}
 	for rows.Next() {
 		row := transformerTestTableRow{}
-		err := rows.Scan(&row.id, &row.name, &row.lastName, &row.email, &row.address, &row.age, &row.totalPurchases, &row.customerID, &row.birthDate, &row.isActive, &row.createdAt, &row.updatedAt, &row.gender)
+		err := rows.Scan(&row.id, &row.name, &row.lastName, &row.email, &row.secondaryEmail, &row.address, &row.age, &row.totalPurchases, &row.customerID, &row.birthDate, &row.isActive, &row.createdAt, &row.updatedAt, &row.gender)
 		require.NoError(t, err)
 		rowsFromDB = append(rowsFromDB, row)
 	}
@@ -165,6 +168,7 @@ func validateRows(t *testing.T, ctx context.Context, conn *pglib.Conn, expectedR
 		require.LessOrEqual(t, len(row.name), 5)
 		require.LessOrEqual(t, len(row.lastName), 10)
 		require.LessOrEqual(t, len(row.email), 15)
+		require.Equal(t, row.secondaryEmail, "joh****e2@example.com")
 		require.LessOrEqual(t, len(row.address), 20)
 		require.GreaterOrEqual(t, row.age, 18)
 		require.LessOrEqual(t, row.age, 75)
diff --git a/pkg/transformers/masking_transformer_test.go b/pkg/transformers/masking_transformer_test.go
new file mode 100644
index 0000000..ccce4a5
--- /dev/null
+++ b/pkg/transformers/masking_transformer_test.go
@@ -0,0 +1,181 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package transformers
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestMaskingTransformer(t *testing.T) {
+	t.Parallel()
+	tests := []struct {
+		name    string
+		params  Parameters
+		wantErr error
+	}{
+		{
+			name:    "ok - valid default parameters",
+			wantErr: nil,
+		},
+		{
+			name: "ok - valid custom parameters",
+			params: Parameters{
+				"type": "password",
+			},
+			wantErr: nil,
+		},
+		{
+			name: "error - invalid masking type",
+			params: Parameters{
+				"type": "invalid",
+			},
+			wantErr: errInvalidMaskingType,
+		},
+		{
+			name: "error - invalid parameter type",
+			params: Parameters{
+				"type": 123,
+			},
+			wantErr: ErrInvalidParameters,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+			mt, err := NewMaskingTransformer(tt.params)
+			require.ErrorIs(t, err, tt.wantErr)
+			if tt.wantErr != nil {
+				return
+			}
+			require.NoError(t, err)
+			require.NotNil(t, mt)
+		})
+	}
+}
+
+func TestMaskingTransformer_Transform(t *testing.T) {
+	t.Parallel()
+	tests := []struct {
+		name    string
+		params  Parameters
+		input   any
+		want    any
+		wantErr error
+	}{
+		{
+			name: "ok - password",
+			params: Parameters{
+				"type": "password",
+			},
+			input:   "aVeryStrongPassword123",
+			want:    "************",
+			wantErr: nil,
+		},
+		{
+			name: "ok - name",
+			params: Parameters{
+				"type": "name",
+			},
+			input:   []byte("John Doe"),
+			want:    "J**n D**e",
+			wantErr: nil,
+		},
+		{
+			name: "ok - address",
+			params: Parameters{
+				"type": "address",
+			},
+			input:   "123 Main St, Anytown, USA",
+			want:    "123 Ma******",
+			wantErr: nil,
+		},
+		{
+			name: "ok - email",
+			params: Parameters{
+				"type": "email",
+			},
+			input:   "john.doe@example.com",
+			want:    "joh****e@example.com",
+			wantErr: nil,
+		},
+		{
+			name: "ok - mobile",
+			params: Parameters{
+				"type": "mobile",
+			},
+			input:   "1234567890",
+			want:    "1234***890",
+			wantErr: nil,
+		},
+		{
+			name: "ok - tel",
+			params: Parameters{
+				"type": "tel",
+			},
+			input:   "+1-23-456-789",
+			want:    "(+1)2345-****",
+			wantErr: nil,
+		},
+		{
+			name: "ok - id",
+			params: Parameters{
+				"type": "id",
+			},
+			input:   "123456789",
+			want:    "123456****",
+			wantErr: nil,
+		},
+		{
+			name: "ok - credit_card",
+			params: Parameters{
+				"type": "credit_card",
+			},
+			input:   "4111-1111-1111-1111",
+			want:    "4111-1******11-1111",
+			wantErr: nil,
+		},
+		{
+			name: "ok - url",
+			params: Parameters{
+				"type": "url",
+			},
+			input:   "http://admin:mysecretpassword@localhost:1234/uri",
+			want:    "http://admin:xxxxx@localhost:1234/uri",
+			wantErr: nil,
+		},
+		{
+			name: "ok - default",
+			params: Parameters{
+				"type": "default",
+			},
+			input:   "Sensitive Data",
+			want:    "**************",
+			wantErr: nil,
+		},
+		{
+			name: "error - invalid input type",
+			params: Parameters{
+				"type": "default",
+			},
+			input:   123,
+			want:    nil,
+			wantErr: ErrUnsupportedValueType,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+			mt, err := NewMaskingTransformer(tt.params)
+			require.NoError(t, err)
+			got, err := mt.Transform(tt.input)
+			require.ErrorIs(t, err, tt.wantErr)
+			if tt.wantErr != nil {
+				return
+			}
+			require.Equal(t, tt.want, got)
+		})
+	}
+}