feat: isolate server ut communication

enterprise/reporting/event_sampler/badger_event_sampler.go

@@ -2,6 +2,7 @@ package event_sampler
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"os"
 	"sync"
@@ -117,7 +118,7 @@ func (es *BadgerEventSampler) Get(key string) (bool, error) {
 		return nil
 	})
 
-	if err == badger.ErrKeyNotFound {
+	if errors.Is(err, badger.ErrKeyNotFound) {
 		return false, nil
 	} else if err != nil {
 		return false, err

integration_test/reporting_dropped_events/reporting_dropped_events_test.go

@@ -15,6 +15,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/rudderlabs/rudder-server/processor/types"
+
 	"github.com/ory/dockertest/v3"
 	"github.com/stretchr/testify/require"
 	"golang.org/x/sync/errgroup"
@@ -25,7 +27,6 @@ import (
 	kithelper "github.com/rudderlabs/rudder-go-kit/testhelper"
 	"github.com/rudderlabs/rudder-go-kit/testhelper/docker/resource/postgres"
 	"github.com/rudderlabs/rudder-go-kit/testhelper/rand"
-	"github.com/rudderlabs/rudder-server/processor/transformer"
 	"github.com/rudderlabs/rudder-server/runner"
 	"github.com/rudderlabs/rudder-server/testhelper/backendconfigtest"
 	"github.com/rudderlabs/rudder-server/testhelper/health"
@@ -120,7 +121,7 @@ func TestReportingDroppedEvents(t *testing.T) {
 				transformertest.ViolationErrorTransformerHandler(
 					http.StatusBadRequest,
 					"tracking plan validation failed",
-					[]transformer.ValidationError{{Type: "Datatype-Mismatch", Message: "must be number"}},
+					[]types.ValidationError{{Type: "Datatype-Mismatch", Message: "must be number"}},
 				),
 			).
 			Build()

integration_test/reporting_error_index/reporting_error_index_test.go

@@ -15,6 +15,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/rudderlabs/rudder-server/processor/types"
+
 	_ "github.com/marcboeker/go-duckdb"
 	"github.com/ory/dockertest/v3"
 	"github.com/samber/lo"
@@ -28,7 +30,6 @@ import (
 	"github.com/rudderlabs/rudder-go-kit/testhelper/docker/resource/postgres"
 	"github.com/rudderlabs/rudder-go-kit/testhelper/rand"
 	"github.com/rudderlabs/rudder-server/jobsdb"
-	"github.com/rudderlabs/rudder-server/processor/transformer"
 	"github.com/rudderlabs/rudder-server/runner"
 	"github.com/rudderlabs/rudder-server/testhelper/backendconfigtest"
 	"github.com/rudderlabs/rudder-server/testhelper/health"
@@ -62,7 +63,7 @@ func TestReportingErrorIndex(t *testing.T) {
 				transformertest.ViolationErrorTransformerHandler(
 					http.StatusBadRequest,
 					"tracking plan validation failed",
-					[]transformer.ValidationError{{Type: "Datatype-Mismatch", Message: "must be number"}},
+					[]types.ValidationError{{Type: "Datatype-Mismatch", Message: "must be number"}},
 				),
 			).
 			Build()

integration_test/tracing/tracing_test.go

@@ -15,6 +15,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/rudderlabs/rudder-server/processor/types"
+
 	_ "github.com/marcboeker/go-duckdb"
 	"github.com/ory/dockertest/v3"
 	"github.com/samber/lo"
@@ -33,7 +35,6 @@ import (
 	"github.com/rudderlabs/rudder-server/app"
 	"github.com/rudderlabs/rudder-server/gateway/response"
 	"github.com/rudderlabs/rudder-server/jobsdb"
-	"github.com/rudderlabs/rudder-server/processor/transformer"
 	"github.com/rudderlabs/rudder-server/runner"
 	"github.com/rudderlabs/rudder-server/testhelper/backendconfigtest"
 	"github.com/rudderlabs/rudder-server/testhelper/health"
@@ -359,15 +360,15 @@ func TestTracing(t *testing.T) {
 		defer bcServer.Close()
 
 		trServer := transformertest.NewBuilder().
-			WithUserTransformHandler(func(request []transformer.TransformerEvent) (response []transformer.TransformerResponse) {
+			WithUserTransformHandler(func(request []types.TransformerEvent) (response []types.TransformerResponse) {
 				for i := range request {
 					req := request[i]
-					response = append(response, transformer.TransformerResponse{
+					response = append(response, types.TransformerResponse{
 						Metadata:   req.Metadata,
 						Output:     req.Message,
 						StatusCode: http.StatusOK,
 					})
-					response = append(response, transformer.TransformerResponse{
+					response = append(response, types.TransformerResponse{
 						Metadata:   req.Metadata,
 						Output:     req.Message,
 						StatusCode: http.StatusOK,

integration_test/transformer_contract/transformer_contract_test.go

@@ -28,7 +28,7 @@ import (
 
 	backendconfig "github.com/rudderlabs/rudder-server/backend-config"
 	"github.com/rudderlabs/rudder-server/jobsdb"
-	"github.com/rudderlabs/rudder-server/processor/transformer"
+	"github.com/rudderlabs/rudder-server/processor/types"
 	"github.com/rudderlabs/rudder-server/runner"
 	"github.com/rudderlabs/rudder-server/testhelper/health"
 	"github.com/rudderlabs/rudder-server/testhelper/transformertest"
@@ -107,7 +107,7 @@ func TestTransformerContract(t *testing.T) {
 
 		trServer := transformertest.NewBuilder().
 			WithUserTransformHandler(
-				func(request []transformer.TransformerEvent) (response []transformer.TransformerResponse) {
+				func(request []types.TransformerEvent) (response []types.TransformerResponse) {
 					for i := range request {
 						req := request[i]
 
@@ -123,15 +123,15 @@ func TestTransformerContract(t *testing.T) {
 						require.Equal(t, req.Metadata.TransformationID, "transformation-1")
 						require.Equal(t, req.Metadata.TransformationVersionID, "version-1")
 						require.Equal(t, req.Metadata.EventType, "identify")
-						require.Equal(t, req.Credentials, []transformer.Credential{
+						require.Equal(t, req.Credentials, []types.Credential{
 							{
 								ID:       "credential-1",
 								Key:      "key-1",
 								Value:    "value-1",
 								IsSecret: false,
 							},
 						})
-						response = append(response, transformer.TransformerResponse{
+						response = append(response, types.TransformerResponse{
 							Metadata:   req.Metadata,
 							Output:     req.Message,
 							StatusCode: http.StatusOK,

integration_test/warehouse/warehouse_test.go

@@ -17,6 +17,8 @@ import (
 	"text/template"
 	"time"
 
+	"github.com/rudderlabs/rudder-server/processor/types"
+
 	transformertest "github.com/rudderlabs/rudder-go-kit/testhelper/docker/resource/transformer"
 
 	"github.com/google/uuid"
@@ -1589,14 +1591,14 @@ func TestDestinationTransformation(t *testing.T) {
 		testcases := []struct {
 			name           string
 			configOverride map[string]any
-			validateEvents func(t *testing.T, events []transformer.TransformerResponse)
+			validateEvents func(t *testing.T, events []types.TransformerResponse)
 		}{
 			{
 				name: "with allowUsersContextTraits=true",
 				configOverride: map[string]any{
 					"allowUsersContextTraits": true,
 				},
-				validateEvents: func(t *testing.T, events []transformer.TransformerResponse) {
+				validateEvents: func(t *testing.T, events []types.TransformerResponse) {
 					var identifyEvent output
 					err := mapstructure.Decode(events[0].Output, &identifyEvent)
 					require.NoError(t, err)
@@ -1625,7 +1627,7 @@ func TestDestinationTransformation(t *testing.T) {
 				configOverride: map[string]any{
 					"allowUsersContextTraits": false,
 				},
-				validateEvents: func(t *testing.T, events []transformer.TransformerResponse) {
+				validateEvents: func(t *testing.T, events []types.TransformerResponse) {
 					var identifyEvent output
 					err := mapstructure.Decode(events[0].Output, &identifyEvent)
 					require.NoError(t, err)
@@ -1652,7 +1654,7 @@ func TestDestinationTransformation(t *testing.T) {
 			{
 				name:           "without allowUsersContextTraits",
 				configOverride: map[string]any{},
-				validateEvents: func(t *testing.T, events []transformer.TransformerResponse) {
+				validateEvents: func(t *testing.T, events []types.TransformerResponse) {
 					var identifyEvent output
 					err := mapstructure.Decode(events[0].Output, &identifyEvent)
 					require.NoError(t, err)
@@ -1719,7 +1721,7 @@ func TestDestinationTransformation(t *testing.T) {
 			})
 			require.NoError(t, err)
 
-			var transformerEvents []transformer.TransformerEvent
+			var transformerEvents []types.TransformerEvent
 			err = json.Unmarshal([]byte(b.String()), &transformerEvents)
 			require.NoError(t, err)
 
@@ -1735,14 +1737,14 @@ func TestDestinationTransformation(t *testing.T) {
 		testcases := []struct {
 			name           string
 			configOverride map[string]any
-			validateEvents func(t *testing.T, events []transformer.TransformerResponse)
+			validateEvents func(t *testing.T, events []types.TransformerResponse)
 		}{
 			{
 				name: "with underscoreDivideNumbers=true",
 				configOverride: map[string]any{
 					"underscoreDivideNumbers": true,
 				},
-				validateEvents: func(t *testing.T, events []transformer.TransformerResponse) {
+				validateEvents: func(t *testing.T, events []types.TransformerResponse) {
 					var trackOutput output
 					err := mapstructure.Decode(events[0].Output, &trackOutput)
 					require.NoError(t, err)
@@ -1767,7 +1769,7 @@ func TestDestinationTransformation(t *testing.T) {
 				configOverride: map[string]any{
 					"underscoreDivideNumbers": false,
 				},
-				validateEvents: func(t *testing.T, events []transformer.TransformerResponse) {
+				validateEvents: func(t *testing.T, events []types.TransformerResponse) {
 					var trackOutput output
 					err := mapstructure.Decode(events[0].Output, &trackOutput)
 					require.NoError(t, err)
@@ -1790,7 +1792,7 @@ func TestDestinationTransformation(t *testing.T) {
 			{
 				name:           "without underscoreDivideNumbers",
 				configOverride: map[string]any{},
-				validateEvents: func(t *testing.T, events []transformer.TransformerResponse) {
+				validateEvents: func(t *testing.T, events []types.TransformerResponse) {
 					var trackOutput output
 					err := mapstructure.Decode(events[0].Output, &trackOutput)
 					require.NoError(t, err)
@@ -1850,7 +1852,7 @@ func TestDestinationTransformation(t *testing.T) {
 			})
 			require.NoError(t, err)
 
-			var transformerEvents []transformer.TransformerEvent
+			var transformerEvents []types.TransformerEvent
 			err = json.Unmarshal([]byte(b.String()), &transformerEvents)
 			require.NoError(t, err)
 
@@ -1867,15 +1869,15 @@ func TestDestinationTransformation(t *testing.T) {
 			name           string
 			destType       string
 			configOverride map[string]any
-			validateEvents func(t *testing.T, events []transformer.TransformerResponse)
+			validateEvents func(t *testing.T, events []types.TransformerResponse)
 		}{
 			{
 				name:     "for non-datalake destinations should be present",
 				destType: whutils.BQ,
 				configOverride: map[string]any{
 					"allowUsersContextTraits": true,
 				},
-				validateEvents: func(t *testing.T, events []transformer.TransformerResponse) {
+				validateEvents: func(t *testing.T, events []types.TransformerResponse) {
 					var identifyEvent output
 					err := mapstructure.Decode(events[0].Output, &identifyEvent)
 					require.NoError(t, err)
@@ -1905,7 +1907,7 @@ func TestDestinationTransformation(t *testing.T) {
 				configOverride: map[string]any{
 					"allowUsersContextTraits": false,
 				},
-				validateEvents: func(t *testing.T, events []transformer.TransformerResponse) {
+				validateEvents: func(t *testing.T, events []types.TransformerResponse) {
 					var identifyEvent output
 					err := mapstructure.Decode(events[0].Output, &identifyEvent)
 					require.NoError(t, err)
@@ -1975,7 +1977,7 @@ func TestDestinationTransformation(t *testing.T) {
 			})
 			require.NoError(t, err)
 
-			var transformerEvents []transformer.TransformerEvent
+			var transformerEvents []types.TransformerEvent
 			err = json.Unmarshal([]byte(b.String()), &transformerEvents)
 			require.NoError(t, err)
 

internal/enricher/enricher.go

@@ -2,7 +2,7 @@ package enricher
 
 import (
 	backendconfig "github.com/rudderlabs/rudder-server/backend-config"
-	"github.com/rudderlabs/rudder-server/utils/types"
+	"github.com/rudderlabs/rudder-server/processor/types"
 )
 
 // PipelineEnricher is a new paradigm under which the gateway events in

internal/enricher/geolocation.go

@@ -14,8 +14,8 @@ import (
 	"github.com/rudderlabs/rudder-go-kit/logger"
 	"github.com/rudderlabs/rudder-go-kit/stats"
 	backendconfig "github.com/rudderlabs/rudder-server/backend-config"
+	"github.com/rudderlabs/rudder-server/processor/types"
 	"github.com/rudderlabs/rudder-server/services/geolocation"
-	"github.com/rudderlabs/rudder-server/utils/types"
 )
 
 const (

internal/enricher/geolocation_test.go

@@ -17,8 +17,8 @@ import (
 	svcMetric "github.com/rudderlabs/rudder-go-kit/stats/metric"
 	miniodocker "github.com/rudderlabs/rudder-go-kit/testhelper/docker/resource/minio"
 	backendconfig "github.com/rudderlabs/rudder-server/backend-config"
+	"github.com/rudderlabs/rudder-server/processor/types"
 	"github.com/rudderlabs/rudder-server/services/geolocation"
-	"github.com/rudderlabs/rudder-server/utils/types"
 )
 
 func TestGeolocationEnrichment_Setup(t *testing.T) {

processor/consent.go

@@ -6,8 +6,8 @@ import (
 	"github.com/samber/lo"
 
 	backendconfig "github.com/rudderlabs/rudder-server/backend-config"
+	"github.com/rudderlabs/rudder-server/processor/types"
 	"github.com/rudderlabs/rudder-server/utils/misc"
-	"github.com/rudderlabs/rudder-server/utils/types"
 )
 
 type ConsentManagementInfo struct {

processor/consent_test.go

@@ -4,13 +4,11 @@ import (
 	"testing"
 
 	"github.com/samber/lo"
+	"github.com/stretchr/testify/require"
 
 	"github.com/rudderlabs/rudder-go-kit/logger"
 	backendconfig "github.com/rudderlabs/rudder-server/backend-config"
-
-	"github.com/stretchr/testify/require"
-
-	"github.com/rudderlabs/rudder-server/utils/types"
+	"github.com/rudderlabs/rudder-server/processor/types"
 )
 
 func TestGetOneTrustConsentCategories(t *testing.T) {

processor/delayed/stats.go

@@ -4,10 +4,11 @@ import (
 	"strings"
 	"time"
 
+	"github.com/rudderlabs/rudder-server/processor/types"
+
 	"github.com/rudderlabs/rudder-go-kit/config"
 	"github.com/rudderlabs/rudder-go-kit/stats"
 	backendconfig "github.com/rudderlabs/rudder-server/backend-config"
-	"github.com/rudderlabs/rudder-server/processor/transformer"
 	"github.com/rudderlabs/rudder-server/utils/misc"
 )
 
@@ -25,7 +26,7 @@ func NewEventStats(stats stats.Stats, config *config.Config) *eventStats {
 	}
 }
 
-func (s *eventStats) ObserveSourceEvents(source *backendconfig.SourceT, events []transformer.TransformerEvent) {
+func (s *eventStats) ObserveSourceEvents(source *backendconfig.SourceT, events []types.TransformerEvent) {
 	statusCount := map[string]map[string]int{
 		"missing_original_timestamp": {},
 		"missing_sent_at":            {},