feat: add support for regulation from features.json (#4304)

koladilip · web-flow · commit 3c2fcf634a10 · 2024-01-12T16:43:08.000+05:30
* feat: add support for regulation from features.json

* feat: add regulation to features.json

* chore: refactor tests to use generator mock instead of custom one

* chore: refactor tests to use generator mock instead of custom one

* chore: address pr comments

* chore: address pr comments
diff --git a/gateway/webhook/webhook_test.go b/gateway/webhook/webhook_test.go
@@ -23,6 +23,7 @@ import (
 	gwtypes "github.com/rudderlabs/rudder-server/gateway/internal/types"
 	mockWebhook "github.com/rudderlabs/rudder-server/gateway/mocks"
 	"github.com/rudderlabs/rudder-server/gateway/response"
+	mock_features "github.com/rudderlabs/rudder-server/mocks/services/transformer"
 	"github.com/rudderlabs/rudder-server/services/transformer"
 	"github.com/rudderlabs/rudder-server/utils/misc"
 )
@@ -73,33 +74,13 @@ func getMockSourceTransformAdapterFunc(url string) func(ctx context.Context) (so
 	}
 }
 
-func getMockTransformerService() transformer.FeaturesService {
-	return &mockTransformerService{}
-}
-
-type mockTransformerService struct{}
-
-func (*mockTransformerService) SourceTransformerVersion() string {
-	return "random-version"
-}
-
-func (*mockTransformerService) TransformerProxyVersion() string {
-	return "random-version"
-}
-
-func (*mockTransformerService) Wait() chan struct{} {
-	return make(chan struct{})
-}
-
-func (*mockTransformerService) RouterTransform(destType string) bool {
-	return false
-}
-
 func TestWebhookBlockTillFeaturesAreFetched(t *testing.T) {
 	initWebhook()
 	ctrl := gomock.NewController(t)
 	mockGW := mockWebhook.NewMockGateway(ctrl)
-	webhookHandler := Setup(mockGW, getMockTransformerService(), stats.Default)
+	mockTransformerFeaturesService := mock_features.NewMockFeaturesService(ctrl)
+	mockTransformerFeaturesService.EXPECT().Wait().Return(make(chan struct{})).Times(1)
+	webhookHandler := Setup(mockGW, mockTransformerFeaturesService, stats.Default)
 
 	mockGW.EXPECT().TrackRequestMetrics(gomock.Any()).Times(1)
 	mockGW.EXPECT().NewSourceStat(gomock.Any(), gomock.Any()).Return(&gwStats.SourceStat{}).Times(1)
diff --git a/mocks/services/transformer/mock_features.go b/mocks/services/transformer/mock_features.go
diff --git a/processor/processor_test.go b/processor/processor_test.go
@@ -33,6 +33,7 @@ import (
 	mocksJobsDB "github.com/rudderlabs/rudder-server/mocks/jobsdb"
 	mocksTransformer "github.com/rudderlabs/rudder-server/mocks/processor/transformer"
 	mockDedup "github.com/rudderlabs/rudder-server/mocks/services/dedup"
+	mock_features "github.com/rudderlabs/rudder-server/mocks/services/transformer"
 	mockReportingTypes "github.com/rudderlabs/rudder-server/mocks/utils/types"
 	"github.com/rudderlabs/rudder-server/processor/isolation"
 	"github.com/rudderlabs/rudder-server/processor/transformer"
@@ -2583,7 +2584,8 @@ var _ = Describe("Processor", Ordered, func() {
 	Context("MainLoop Tests", func() {
 		It("Should not handle jobs when transformer features are not set", func() {
 			mockTransformer := mocksTransformer.NewMockTransformer(c.mockCtrl)
-
+			mockTransformerFeaturesService := mock_features.NewMockFeaturesService(c.mockCtrl)
+			mockTransformerFeaturesService.EXPECT().Wait().Return(make(chan struct{})).AnyTimes()
 			processor := prepareHandle(NewHandle(config.Default, mockTransformer))
 
 			// crash recover returns empty list
@@ -2601,7 +2603,7 @@ var _ = Describe("Processor", Ordered, func() {
 				transientsource.NewEmptyService(),
 				fileuploader.NewDefaultProvider(),
 				c.MockRsourcesService,
-				getMockTransformerService(),
+				mockTransformerFeaturesService,
 				destinationdebugger.NewNoOpService(),
 				transformationdebugger.NewNoOpService(),
 				[]enricher.PipelineEnricher{},
@@ -4970,25 +4972,3 @@ func TestStoreMessageMerge(t *testing.T) {
 	require.Len(t, merged.dedupKeys, 2, "dedup keys should have 2 elements")
 	require.Equal(t, merged.totalEvents, 2, "total events should be 2")
 }
-
-func getMockTransformerService() transformerFeaturesService.FeaturesService {
-	return &mockTransformerService{}
-}
-
-type mockTransformerService struct{}
-
-func (*mockTransformerService) SourceTransformerVersion() string {
-	return "random-version"
-}
-
-func (*mockTransformerService) TransformerProxyVersion() string {
-	return "random-version"
-}
-
-func (*mockTransformerService) Wait() chan struct{} {
-	return make(chan struct{})
-}
-
-func (*mockTransformerService) RouterTransform(destType string) bool {
-	return false
-}
diff --git a/regulation-worker/cmd/main.go b/regulation-worker/cmd/main.go
@@ -27,6 +27,7 @@ import (
 	"github.com/rudderlabs/rudder-server/rruntime"
 	"github.com/rudderlabs/rudder-server/services/diagnostics"
 	"github.com/rudderlabs/rudder-server/services/oauth"
+	"github.com/rudderlabs/rudder-server/services/transformer"
 	"github.com/rudderlabs/rudder-server/utils/misc"
 	"github.com/rudderlabs/rudder-server/utils/types/deployment"
 )
@@ -100,6 +101,11 @@ func Run(ctx context.Context) error {
 				DestTransformURL:             config.MustGetString("DEST_TRANSFORM_URL"),
 				OAuth:                        OAuth,
 				MaxOAuthRefreshRetryAttempts: config.GetInt("RegulationWorker.oauth.maxRefreshRetryAttempts", 1),
+				TransformerFeaturesService: transformer.NewFeaturesService(ctx, transformer.FeaturesServiceConfig{
+					PollInterval:             config.GetDuration("Transformer.pollInterval", 1, time.Second),
+					TransformerURL:           config.GetString("DEST_TRANSFORM_URL", "http://localhost:9090"),
+					FeaturesRetryMaxAttempts: 10,
+				}),
 			}),
 		MaxFailedAttempts: config.GetInt("REGULATION_DELETION_MAX_FAILED_ATTEMPTS", 4),
 	}
diff --git a/regulation-worker/internal/delete/api/api.go b/regulation-worker/internal/delete/api/api.go
@@ -22,28 +22,35 @@ import (
 	backendconfig "github.com/rudderlabs/rudder-server/backend-config"
 	"github.com/rudderlabs/rudder-server/regulation-worker/internal/model"
 	"github.com/rudderlabs/rudder-server/services/oauth"
+	"github.com/rudderlabs/rudder-server/services/transformer"
 	"github.com/rudderlabs/rudder-server/utils/httputil"
 )
 
 var (
 	pkgLogger             = logger.NewLogger().Child("api")
-	supportedDestinations = []string{"BRAZE", "AM", "INTERCOM", "CLEVERTAP", "AF", "MP", "GA", "ITERABLE", "ENGAGE", "CUSTIFY", "SENDGRID", "SPRIG"}
+	SupportedDestinations = []string{"BRAZE", "AM", "INTERCOM", "CLEVERTAP", "AF", "MP", "GA", "ITERABLE", "ENGAGE", "CUSTIFY", "SENDGRID", "SPRIG"}
 )
 
 type APIManager struct {
 	Client                       *http.Client
 	DestTransformURL             string
 	OAuth                        oauth.Authorizer
 	MaxOAuthRefreshRetryAttempts int
+	TransformerFeaturesService   transformer.FeaturesService
 }
 
 type oauthDetail struct {
 	secretToken *oauth.AuthResponse
 	id          string
 }
 
-func (*APIManager) GetSupportedDestinations() []string {
-	return supportedDestinations
+func (m *APIManager) GetSupportedDestinations() []string {
+	destinations := m.TransformerFeaturesService.Regulations()
+	if len(destinations) == 0 {
+		// Fallback to default supported destinations
+		destinations = SupportedDestinations
+	}
+	return destinations
 }
 
 func (api *APIManager) deleteWithRetry(ctx context.Context, job model.Job, destination model.Destination, currentOauthRetryAttempt int) model.JobStatus {
diff --git a/regulation-worker/internal/delete/api/api_test.go b/regulation-worker/internal/delete/api/api_test.go
@@ -16,9 +16,11 @@ import (
 
 	backendconfig "github.com/rudderlabs/rudder-server/backend-config"
 	mocksBackendConfig "github.com/rudderlabs/rudder-server/mocks/backend-config"
+	mock_features "github.com/rudderlabs/rudder-server/mocks/services/transformer"
 	"github.com/rudderlabs/rudder-server/regulation-worker/internal/delete/api"
 	"github.com/rudderlabs/rudder-server/regulation-worker/internal/model"
 	"github.com/rudderlabs/rudder-server/services/oauth"
+	"github.com/rudderlabs/rudder-server/services/transformer"
 
 	"github.com/stretchr/testify/require"
 )
@@ -136,8 +138,9 @@ func TestDelete(t *testing.T) {
 			defer svr.Close()
 			t.Setenv("DEST_TRANSFORM_URL", svr.URL)
 			api := api.APIManager{
-				Client:           &http.Client{},
-				DestTransformURL: svr.URL,
+				Client:                     &http.Client{},
+				DestTransformURL:           svr.URL,
+				TransformerFeaturesService: transformer.NewNoOpService(),
 			}
 			dest := model.Destination{
 				Config: tt.destConfig,
@@ -151,6 +154,37 @@ func TestDelete(t *testing.T) {
 	}
 }
 
+func TestGetSupportedDestinations(t *testing.T) {
+	tests := []struct {
+		name                 string
+		fromFeatures         []string
+		expectedDestinations []string
+	}{
+		{
+			name:                 "test get supported destinations when there is no transformer features",
+			expectedDestinations: api.SupportedDestinations,
+			fromFeatures:         []string{},
+		},
+		{
+			name:                 "test get supported destinations when there is transformer features",
+			expectedDestinations: []string{"AM", "GA4"},
+			fromFeatures:         []string{"AM", "GA4"},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			mockTransformerFeaturesService := mock_features.NewMockFeaturesService(gomock.NewController(t))
+			mockTransformerFeaturesService.EXPECT().Regulations().Return(tt.fromFeatures)
+			api := api.APIManager{
+				TransformerFeaturesService: mockTransformerFeaturesService,
+			}
+			destinations := api.GetSupportedDestinations()
+			require.Equal(t, tt.expectedDestinations, destinations)
+		})
+	}
+}
+
 type deleteAPI struct {
 	payload         string
 	respStatusCode  int
diff --git a/services/transformer/features.go b/services/transformer/features.go
@@ -1,3 +1,5 @@
+//go:generate mockgen --build_flags=--mod=mod -destination=../../mocks/services/transformer/mock_features.go -package mock_features github.com/rudderlabs/rudder-server/services/transformer FeaturesService
+
 package transformer
 
 import (
@@ -21,6 +23,7 @@ type FeaturesServiceConfig struct {
 }
 
 type FeaturesService interface {
+	Regulations() []string
 	SourceTransformerVersion() string
 	RouterTransform(destType string) bool
 	TransformerProxyVersion() string
@@ -31,7 +34,8 @@ var defaultTransformerFeatures = `{
 	"routerTransform": {
 	  "MARKETO": true,
 	  "HS": true
-	}
+	},
+	"regulations": ["AM"],
   }`
 
 func NewFeaturesService(ctx context.Context, config FeaturesServiceConfig) FeaturesService {
@@ -53,6 +57,10 @@ func NewNoOpService() FeaturesService {
 
 type noopService struct{}
 
+func (*noopService) Regulations() []string {
+	return []string{}
+}
+
 func (*noopService) SourceTransformerVersion() string {
 	return V0
 }
diff --git a/services/transformer/features_impl.go b/services/transformer/features_impl.go
@@ -8,6 +8,7 @@ import (
 	"net/http"
 	"time"
 
+	"github.com/samber/lo"
 	"github.com/tidwall/gjson"
 
 	"github.com/rudderlabs/rudder-go-kit/config"
@@ -42,6 +43,16 @@ func (t *featuresService) RouterTransform(destType string) bool {
 	return gjson.GetBytes(t.features, "routerTransform."+destType).Bool()
 }
 
+func (t *featuresService) Regulations() []string {
+	regulationFeatures := gjson.GetBytes(t.features, "regulations")
+	if regulationFeatures.Exists() && regulationFeatures.IsArray() {
+		return lo.Map(regulationFeatures.Array(), func(f gjson.Result, _ int) string {
+			return f.String()
+		})
+	}
+	return []string{}
+}
+
 func (t *featuresService) Wait() chan struct{} {
 	return t.waitChan
 }
diff --git a/services/transformer/features_impl_test.go b/services/transformer/features_impl_test.go
