Repository: xataio/pgstream
Pull request #161 (2025-03-05 11:34:36)

Add utc timestamp gm transformer
Introduce `UTCTimestampTransformer` which uses greenmask's Timestamp under the hood.

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

| Impacted Packages | Coverage Δ | :robot: |
|-------------------|------------|---------|
| github.com/xataio/pgstream/pkg/transformers | 100.00% (ø) |  |
| github.com/xataio/pgstream/pkg/transformers/builder | 10.26% (**-0.85%**) | :thumbsdown: |
| github.com/xataio/pgstream/pkg/transformers/greenmask | 95.11% (**+0.37%**) | :thumbsup: |

---

<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 | 10.26% (**-0.85%**) | 39 (+3) | 4 | 35 (+3) | :thumbsdown: |
| github.com/xataio/pgstream/pkg/transformers/greenmask/greenmask_timestamp_transformer.go | 97.14% (**+97.14%**) | 35 (+35) | 34 (+34) | 1 (+1) | :star2: |
| github.com/xataio/pgstream/pkg/transformers/transformer.go | 100.00% (ø) | 7 | 7 | 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/greenmask/greenmask_timestamp_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 | 100.00% (ø) |  |
| github.com/xataio/pgstream/pkg/transformers/builder | 10.26% (**-0.85%**) | :thumbsdown: |
| github.com/xataio/pgstream/pkg/transformers/greenmask | 95.11% (**+0.37%**) | :thumbsup: |

---

<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 | 10.26% (**-0.85%**) | 39 (+3) | 4 | 35 (+3) | :thumbsdown: |
| github.com/xataio/pgstream/pkg/transformers/greenmask/greenmask_timestamp_transformer.go | 97.14% (**+97.14%**) | 35 (+35) | 34 (+34) | 1 (+1) | :star2: |
| github.com/xataio/pgstream/pkg/transformers/transformer.go | 100.00% (ø) | 7 | 7 | 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/greenmask/greenmask_timestamp_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 | 100.00% (ø) |  |
| github.com/xataio/pgstream/pkg/transformers/builder | 10.26% (**-0.85%**) | :thumbsdown: |
| github.com/xataio/pgstream/pkg/transformers/greenmask | 95.11% (**+0.37%**) | :thumbsup: |

---

<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 | 10.26% (**-0.85%**) | 39 (+3) | 4 | 35 (+3) | :thumbsdown: |
| github.com/xataio/pgstream/pkg/transformers/greenmask/greenmask_timestamp_transformer.go | 97.14% (**+97.14%**) | 35 (+35) | 34 (+34) | 1 (+1) | :star2: |
| github.com/xataio/pgstream/pkg/transformers/transformer.go | 100.00% (ø) | 7 | 7 | 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/greenmask/greenmask_timestamp_transformer_test.go

</details>

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

| Impacted Packages | Coverage Δ | :robot: |
|-------------------|------------|---------|
| github.com/xataio/pgstream/pkg/transformers | 100.00% (ø) |  |
| github.com/xataio/pgstream/pkg/transformers/builder | 8.89% (**-1.37%**) | :thumbsdown: |
| github.com/xataio/pgstream/pkg/transformers/greenmask | 93.45% (**-1.59%**) | :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 | 8.89% (**-1.37%**) | 45 (+6) | 4 | 41 (+6) | :thumbsdown: |
| github.com/xataio/pgstream/pkg/transformers/greenmask/greenmask_timestamp_transformer.go | 94.74% (**+94.74%**) | 38 (+38) | 36 (+36) | 2 (+2) | :star2: |
| github.com/xataio/pgstream/pkg/transformers/transformer.go | 100.00% (ø) | 7 | 7 | 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/greenmask/greenmask_timestamp_transformer_test.go

</details>

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

| Impacted Packages | Coverage Δ | :robot: |
|-------------------|------------|---------|
| github.com/xataio/pgstream/pkg/transformers | 100.00% (ø) |  |
| github.com/xataio/pgstream/pkg/transformers/builder | 8.89% (**-1.37%**) | :thumbsdown: |
| github.com/xataio/pgstream/pkg/transformers/greenmask | 93.75% (**-1.30%**) | :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 | 8.89% (**-1.37%**) | 45 (+6) | 4 | 41 (+6) | :thumbsdown: |
| github.com/xataio/pgstream/pkg/transformers/greenmask/greenmask_timestamp_transformer.go | 97.14% (**+97.14%**) | 35 (+35) | 34 (+34) | 1 (+1) | :star2: |
| github.com/xataio/pgstream/pkg/transformers/transformer.go | 100.00% (ø) | 7 | 7 | 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/greenmask/greenmask_timestamp_transformer_test.go

</details>


Patch:
diff --git a/go.mod b/go.mod
index 371601b..764220b 100644
--- a/go.mod
+++ b/go.mod
@@ -6,7 +6,7 @@ require (
 	github.com/bytedance/sonic v1.12.3
 	github.com/cenkalti/backoff/v4 v4.3.0
 	github.com/elastic/go-elasticsearch/v8 v8.14.0
-	github.com/eminano/greenmask v0.0.0-20250210141616-781d3aca2262
+	github.com/eminano/greenmask v0.0.0-20250307113752-035ee2b102e6
 	github.com/go-logr/zerologr v1.2.3
 	github.com/golang-migrate/migrate/v4 v4.18.1
 	github.com/google/go-cmp v0.6.0
diff --git a/go.sum b/go.sum
index e6828b3..1862f72 100644
--- a/go.sum
+++ b/go.sum
@@ -104,6 +104,8 @@ github.com/elastic/go-elasticsearch/v8 v8.14.0 h1:1ywU8WFReLLcxE1WJqii3hTtbPUE2h
 github.com/elastic/go-elasticsearch/v8 v8.14.0/go.mod h1:WRvnlGkSuZyp83M2U8El/LGXpCjYLrvlkSgkAH4O5I4=
 github.com/eminano/greenmask v0.0.0-20250210141616-781d3aca2262 h1:IAhjExmPmCH/UWLv56J/50wrjlwJRTbwI8gBhb5Rl/A=
 github.com/eminano/greenmask v0.0.0-20250210141616-781d3aca2262/go.mod h1:gBBDtACvj/I3nzyEvCGqrWd2ox1XVlh5+Xv5JXKcKDI=
+github.com/eminano/greenmask v0.0.0-20250307113752-035ee2b102e6 h1:m/MuqhOhjJM2oWIjUMkyvrCuCmb/IMC8Ig3/uSM0M+U=
+github.com/eminano/greenmask v0.0.0-20250307113752-035ee2b102e6/go.mod h1:gBBDtACvj/I3nzyEvCGqrWd2ox1XVlh5+Xv5JXKcKDI=
 github.com/expr-lang/expr v1.16.9 h1:WUAzmR0JNI9JCiF0/ewwHB1gmcGw5wW7nWt8gc6PpCI=
 github.com/expr-lang/expr v1.16.9/go.mod h1:8/vRC7+7HBzESEqt5kKpYXxrxkr31SaO8r40VO/1IT4=
 github.com/fatih/color v1.18.0 h1:S8gINlzdQ840/4pfAwic/ZE0djQEH3wM94VfqLTZcOM=
diff --git a/pkg/transformers/builder/transformer_builder.go b/pkg/transformers/builder/transformer_builder.go
index 3bd7901..44c35f9 100644
--- a/pkg/transformers/builder/transformer_builder.go
+++ b/pkg/transformers/builder/transformer_builder.go
@@ -34,6 +34,8 @@ func New(cfg *transformers.Config) (transformers.Transformer, error) {
 		return greenmask.NewUnixTimestampTransformer(cfg.Generator, cfg.Parameters)
 	case transformers.GreenmaskDate:
 		return greenmask.NewDateTransformer(cfg.Generator, cfg.Parameters)
+	case transformers.GreenmaskUTCTimestamp:
+		return greenmask.NewUTCTimestampTransformer(cfg.Generator, cfg.Parameters)
 	default:
 		return nil, transformers.ErrUnsupportedTransformer
 	}
diff --git a/pkg/transformers/greenmask/greenmask_timestamp_transformer.go b/pkg/transformers/greenmask/greenmask_timestamp_transformer.go
new file mode 100644
index 0000000..a7293ae
--- /dev/null
+++ b/pkg/transformers/greenmask/greenmask_timestamp_transformer.go
@@ -0,0 +1,83 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package greenmask
+
+import (
+	"errors"
+	"fmt"
+	"time"
+
+	greenmasktransformers "github.com/eminano/greenmask/pkg/generators/transformers"
+	"github.com/xataio/pgstream/pkg/transformers"
+)
+
+type UTCTimestampTransformer struct {
+	transformer *greenmasktransformers.Timestamp
+}
+
+var (
+	errMinMaxTimestampNotSpecified = errors.New("greenmask_timestamp: min_timestamp and max_timestamp must be specified")
+	errInvalidTimestamp            = errors.New("greenmask_timestamp: min_timestamp and max_timestamp must be valid RFC3339 timestamps")
+)
+
+func NewUTCTimestampTransformer(generatorType transformers.GeneratorType, params transformers.Parameters) (*UTCTimestampTransformer, error) {
+	truncatePart, err := findParameter(params, "truncate_part", "")
+	if err != nil {
+		return nil, fmt.Errorf("greenmask_integer: truncate_part must be a string: %w", err)
+	}
+	minDateStr, err := findParameter(params, "min_timestamp", "")
+	if err != nil {
+		return nil, fmt.Errorf("greenmask_integer: min_timestamp must be a string: %w", err)
+	}
+	maxDateStr, err := findParameter(params, "max_timestamp", "")
+	if err != nil {
+		return nil, fmt.Errorf("greenmask_integer: max_timestamp must be a string: %w", err)
+	}
+	if minDateStr == "" || maxDateStr == "" {
+		return nil, errMinMaxTimestampNotSpecified
+	}
+	minDate, err := time.Parse(time.RFC3339, minDateStr)
+	if err != nil {
+		return nil, errInvalidTimestamp
+	}
+	minDate = minDate.UTC()
+
+	maxDate, err := time.Parse(time.RFC3339, maxDateStr)
+	if err != nil {
+		return nil, errInvalidTimestamp
+	}
+	maxDate = maxDate.UTC()
+
+	limiter, err := greenmasktransformers.NewTimestampLimiter(minDate, maxDate)
+	if err != nil {
+		return nil, err
+	}
+
+	t, err := greenmasktransformers.NewRandomTimestamp(truncatePart, limiter)
+	if err != nil {
+		return nil, err
+	}
+
+	if err := setGenerator(t, generatorType); err != nil {
+		return nil, err
+	}
+
+	return &UTCTimestampTransformer{
+		transformer: t,
+	}, nil
+}
+
+func (t *UTCTimestampTransformer) Transform(value any) (any, error) {
+	var toTransform []byte
+	switch val := value.(type) {
+	case time.Time:
+		toTransform = []byte(val.Format(time.RFC3339))
+	case []byte:
+		toTransform = val
+	case string:
+		toTransform = []byte(val)
+	default:
+		return nil, transformers.ErrUnsupportedValueType
+	}
+	return t.transformer.Transform(nil, toTransform)
+}
diff --git a/pkg/transformers/transformer.go b/pkg/transformers/transformer.go
index 514da8e..0fee9d1 100644
--- a/pkg/transformers/transformer.go
+++ b/pkg/transformers/transformer.go
@@ -31,6 +31,7 @@ const (
 	GreenmaskChoice        TransformerType = "greenmask_choice"
 	GreenmaskUnixTimestamp TransformerType = "greenmask_unix_timestamp"
 	GreenmaskDate          TransformerType = "greenmask_date"
+	GreenmaskUTCTimestamp  TransformerType = "greenmask_utc_timestamp"
 )
 
 type GeneratorType string


Test patch:
diff --git a/pkg/transformers/greenmask/greenmask_timestamp_transformer_test.go b/pkg/transformers/greenmask/greenmask_timestamp_transformer_test.go
new file mode 100644
index 0000000..99ee41c
--- /dev/null
+++ b/pkg/transformers/greenmask/greenmask_timestamp_transformer_test.go
@@ -0,0 +1,245 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package greenmask
+
+import (
+	"testing"
+	"time"
+
+	greenmasktransformers "github.com/eminano/greenmask/pkg/generators/transformers"
+	"github.com/stretchr/testify/require"
+	"github.com/xataio/pgstream/pkg/transformers"
+)
+
+func TestNewUTCTimestampTransformer(t *testing.T) {
+	t.Parallel()
+	tests := []struct {
+		name      string
+		generator transformers.GeneratorType
+		params    transformers.Parameters
+		wantErr   error
+	}{
+		{
+			name:      "ok - valid random",
+			generator: transformers.Random,
+			params: transformers.Parameters{
+				"min_timestamp": "2021-01-01T00:00:00Z",
+				"max_timestamp": "2022-01-02T00:00:00Z",
+			},
+			wantErr: nil,
+		},
+		{
+			name:      "error - invalid truncate_part",
+			generator: transformers.Deterministic,
+			params: transformers.Parameters{
+				"truncate_part": 1.2,
+			},
+			wantErr: transformers.ErrInvalidParameters,
+		},
+		{
+			name:      "error - invalid min_timestamp",
+			generator: transformers.Deterministic,
+			params: transformers.Parameters{
+				"min_timestamp": 1.2,
+			},
+			wantErr: transformers.ErrInvalidParameters,
+		},
+		{
+			name:      "error - invalid min_timestamp format",
+			generator: transformers.Deterministic,
+			params: transformers.Parameters{
+				"min_timestamp": "2021 Jan 01",
+				"max_timestamp": "2022-01-02T00:00:00Z",
+			},
+			wantErr: errInvalidTimestamp,
+		},
+		{
+			name:      "error - invalid max_timestamp",
+			generator: transformers.Random,
+			params: transformers.Parameters{
+				"min_timestamp": "2021-01-01T00:00:00Z",
+				"max_timestamp": 1.2,
+			},
+			wantErr: transformers.ErrInvalidParameters,
+		},
+		{
+			name:      "error - invalid max_timestamp format",
+			generator: transformers.Deterministic,
+			params: transformers.Parameters{
+				"min_timestamp": "2021-01-01T00:00:00Z",
+				"max_timestamp": "2021 Jan 01",
+			},
+			wantErr: errInvalidTimestamp,
+		},
+		{
+			name:      "error - min_timestamp not specified",
+			generator: transformers.Random,
+			params: transformers.Parameters{
+				"max_timestamp": "2022-01-02T00:00:00Z",
+			},
+			wantErr: errMinMaxTimestampNotSpecified,
+		},
+		{
+			name:      "error - min_timestamp equal to max_timestamp",
+			generator: transformers.Deterministic,
+			params: transformers.Parameters{
+				"min_timestamp": "2023-01-02T00:00:00Z",
+				"max_timestamp": "2023-01-02T00:01:00+01:00",
+			},
+			wantErr: greenmasktransformers.ErrWrongLimits,
+		},
+		{
+			name:      "error - invalid generator type",
+			generator: "invalid",
+			params: transformers.Parameters{
+				"min_timestamp": "2021-01-01T00:00:00Z",
+				"max_timestamp": "2022-01-02T00:00:00Z",
+			},
+			wantErr: transformers.ErrUnsupportedGenerator,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+			transformer, err := NewUTCTimestampTransformer(tt.generator, tt.params)
+			if tt.wantErr != nil {
+				require.ErrorIs(t, err, tt.wantErr)
+				return
+			}
+			require.NoError(t, err)
+			require.NotNil(t, transformer)
+		})
+	}
+}
+
+func TestUTCTimestampTransformer_Transform(t *testing.T) {
+	t.Parallel()
+	tests := []struct {
+		name          string
+		generatorType transformers.GeneratorType
+		input         any
+		params        transformers.Parameters
+		wantErr       error
+	}{
+		{
+			name:          "ok - transform string randomly",
+			generatorType: transformers.Random,
+			input:         "test",
+			params: transformers.Parameters{
+				"min_timestamp": "2022-01-01T00:00:00Z",
+				"max_timestamp": "2023-01-02T00:00:00Z",
+				"truncate_part": "month",
+			},
+			wantErr: nil,
+		},
+		{
+			name:          "ok - transform []byte deterministically",
+			generatorType: transformers.Deterministic,
+			input:         []byte("test"),
+			params: transformers.Parameters{
+				"min_timestamp": "2021-01-01T00:00:00Z",
+				"max_timestamp": "2023-01-02T00:00:00Z",
+				"truncate_part": "day",
+			},
+			wantErr: nil,
+		},
+		{
+			name:          "ok - transform time.Time deterministically",
+			generatorType: transformers.Deterministic,
+			input:         time.Now(),
+			params: transformers.Parameters{
+				"min_timestamp": "2020-01-01T00:00:00Z",
+				"max_timestamp": "2023-01-02T00:00:00Z",
+				"truncate_part": "hour",
+			},
+			wantErr: nil,
+		},
+		{
+			name:          "ok - truncate after millisecond part",
+			generatorType: transformers.Deterministic,
+			input:         "2021-01-01T00:00:00Z",
+			params: transformers.Parameters{
+				"min_timestamp": "2023-01-01T00:00:00Z",
+				"max_timestamp": "2023-01-01T01:10:00+01:00",
+				"truncate_part": "millisecond",
+			},
+			wantErr: nil,
+		},
+		{
+			name:          "error - invalid input type",
+			generatorType: transformers.Deterministic,
+			input:         1,
+			params: transformers.Parameters{
+				"min_timestamp": "2020-01-01T00:00:00Z",
+				"max_timestamp": "2023-01-02T00:00:00Z",
+			},
+			wantErr: transformers.ErrUnsupportedValueType,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+			transformer, err := NewUTCTimestampTransformer(tt.generatorType, tt.params)
+			require.NoError(t, err)
+			require.NotNil(t, transformer)
+
+			got, err := transformer.Transform(tt.input)
+			require.ErrorIs(t, err, tt.wantErr)
+			if err != nil {
+				return
+			}
+			require.NotNil(t, got)
+			result, ok := got.(time.Time)
+			require.True(t, ok, "expected got to be of type time.Time")
+
+			minStr, _ := tt.params["min_timestamp"].(string)
+			minTimestamp, _ := time.Parse(time.RFC3339, minStr)
+			maxStr, _ := tt.params["max_timestamp"].(string)
+			maxTimestamp, _ := time.Parse(time.RFC3339, maxStr)
+
+			truncatePart, _, _ := transformers.FindParameter[string](tt.params, "truncate_part")
+
+			// check if the result is truncated accordingly
+			// truncate min and max timestamps as well, so we can compare them with the result
+			switch truncatePart {
+			case "year":
+				require.Equal(t, int(result.Month()), 1)
+				minTimestamp = time.Date(minTimestamp.Year(), 1, 1, 0, 0, 0, 0, minTimestamp.Location())
+				maxTimestamp = time.Date(maxTimestamp.Year(), 1, 1, 0, 0, 0, 0, maxTimestamp.Location())
+			case "month":
+				require.Equal(t, result.Day(), 1)
+				minTimestamp = time.Date(minTimestamp.Year(), minTimestamp.Month(), 1, 0, 0, 0, 0, minTimestamp.Location())
+				maxTimestamp = time.Date(maxTimestamp.Year(), maxTimestamp.Month(), 1, 0, 0, 0, 0, maxTimestamp.Location())
+			case "day":
+				require.Equal(t, result.Hour(), 0)
+				minTimestamp = minTimestamp.Truncate(24 * time.Hour)
+				maxTimestamp = maxTimestamp.Truncate(24 * time.Hour)
+			case "hour":
+				require.Equal(t, result.Minute(), 0)
+				minTimestamp = minTimestamp.Truncate(time.Hour)
+				maxTimestamp = maxTimestamp.Truncate(time.Hour)
+			case "minute":
+				require.Equal(t, result.Second(), 0)
+				minTimestamp = minTimestamp.Truncate(time.Minute)
+				maxTimestamp = maxTimestamp.Truncate(time.Minute)
+			case "second":
+				require.Equal(t, result.Nanosecond(), 0)
+				minTimestamp = minTimestamp.Truncate(time.Second)
+				maxTimestamp = maxTimestamp.Truncate(time.Second)
+			}
+
+			require.Equal(t, result.Location(), time.UTC)
+
+			require.True(t, result.After(minTimestamp) || result.Equal(minTimestamp))
+			// require.True(t, result.Before(maxTimestamp) || result.Equal(maxTimestamp))
+			require.LessOrEqual(t, result, maxTimestamp)
+
+			// if deterministic, check that the same input always produces the same output
+			if tt.generatorType == transformers.Deterministic {
+				gotAgain, err := transformer.Transform(tt.input)
+				require.NoError(t, err)
+				require.Equal(t, got, gotAgain)
+			}
+		})
+	}
+}