Merge pull request #223 from splitio/one-queue

Author: nmayorsplit

dtos/notification.go

@@ -17,8 +17,9 @@ const (
 	UpdateTypeSplitChange        = "SPLIT_UPDATE"
 	UpdateTypeSplitKill          = "SPLIT_KILL"
 	UpdateTypeSegmentChange      = "SEGMENT_UPDATE"
-	UpdateTypeContol             = "CONTROL"
+	UpdateTypeControl            = "CONTROL"
 	UpdateTypeLargeSegmentChange = "LS_DEFINITION_UPDATE"
+	UpdateTypeRuleBasedChange    = "RB_SEGMENT_UPDATE"
 )
 
 // Control type constants
@@ -201,6 +202,7 @@ type SplitChangeUpdate struct {
 	BaseUpdate
 	previousChangeNumber *int64
 	featureFlag          *SplitDTO
+	ruleBasedSegment     *RuleBasedSegmentDTO
 }
 
 func NewSplitChangeUpdate(baseUpdate BaseUpdate, pcn *int64, featureFlag *SplitDTO) *SplitChangeUpdate {
@@ -211,8 +213,26 @@ func NewSplitChangeUpdate(baseUpdate BaseUpdate, pcn *int64, featureFlag *SplitD
 	}
 }
 
-// UpdateType always returns UpdateTypeSplitChange for SplitUpdate messages
-func (u *SplitChangeUpdate) UpdateType() string { return UpdateTypeSplitChange }
+func NewRuleBasedSegmentChangeUpdate(baseUpdate BaseUpdate, pcn *int64, ruleBasedSegment *RuleBasedSegmentDTO) *SplitChangeUpdate {
+	return &SplitChangeUpdate{
+		BaseUpdate:           baseUpdate,
+		previousChangeNumber: pcn,
+		ruleBasedSegment:     ruleBasedSegment,
+	}
+}
+
+// UpdateType returns the type of update
+func (u *SplitChangeUpdate) UpdateType() string {
+	if u.ruleBasedSegment != nil {
+		return UpdateTypeRuleBasedChange
+	}
+	return UpdateTypeSplitChange
+}
+
+// GetRuleBased returns rule-based segment
+func (u *SplitChangeUpdate) RuleBasedSegment() *RuleBasedSegmentDTO {
+	return u.ruleBasedSegment
+}
 
 // String returns the String representation of a split change notification
 func (u *SplitChangeUpdate) String() string {

dtos/notification_test.go

@@ -226,6 +226,72 @@ func TestControlUpdate(t *testing.T) {
 	}
 }
 
+func TestRuleBasedSegmentChangeUpdate(t *testing.T) {
+	// Test case 1: With rule-based segment data
+	ruleBasedSegment := &RuleBasedSegmentDTO{
+		Name:         "test-segment",
+		ChangeNumber: 123456,
+		Conditions: []RuleBasedConditionDTO{
+			{
+				ConditionType: "WHITELIST",
+				MatcherGroup: MatcherGroupDTO{
+					Matchers: []MatcherDTO{},
+				},
+			},
+		},
+	}
+
+	rbUpdate := NewRuleBasedSegmentChangeUpdate(NewBaseUpdate(NewBaseMessage(123456789, "rb_channel"), 123456), nil, ruleBasedSegment)
+	if rbUpdate.EventType() != SSEEventTypeMessage {
+		t.Error("Unexpected EventType")
+	}
+	if rbUpdate.Timestamp() != 123456789 {
+		t.Error("Unexpected Timestamp")
+	}
+	if rbUpdate.Channel() != "rb_channel" {
+		t.Error("Unexpected Channel")
+	}
+	if rbUpdate.MessageType() != MessageTypeUpdate {
+		t.Error("Unexpected MessageType")
+	}
+	if rbUpdate.ChangeNumber() != 123456 {
+		t.Error("Unexpected ChangeNumber")
+	}
+	if rbUpdate.UpdateType() != UpdateTypeRuleBasedChange {
+		t.Error("Unexpected UpdateType, got:", rbUpdate.UpdateType())
+	}
+	if rbUpdate.String() != "SplitChange(channel=rb_channel,changeNumber=123456,timestamp=123456789)" {
+		t.Error("Unexpected String", rbUpdate.String())
+	}
+	if rbUpdate.RuleBasedSegment() == nil {
+		t.Error("RuleBasedSegment should not be nil")
+	}
+	if rbUpdate.RuleBasedSegment().Name != "test-segment" {
+		t.Error("Unexpected RuleBasedSegment name")
+	}
+	if rbUpdate.RuleBasedSegment().ChangeNumber != 123456 {
+		t.Error("Unexpected RuleBasedSegment change number")
+	}
+	if len(rbUpdate.RuleBasedSegment().Conditions) != 1 {
+		t.Error("RuleBasedSegment should have 1 condition")
+	}
+	if rbUpdate.RuleBasedSegment().Conditions[0].ConditionType != "WHITELIST" {
+		t.Error("Unexpected condition type")
+	}
+
+	// Test case 2: Without rule-based segment data
+	rbUpdateNoSegment := NewRuleBasedSegmentChangeUpdate(NewBaseUpdate(NewBaseMessage(123456789, "rb_channel"), 123456), nil, nil)
+	if rbUpdateNoSegment.EventType() != SSEEventTypeMessage {
+		t.Error("Unexpected EventType for no segment case")
+	}
+	if rbUpdateNoSegment.UpdateType() != UpdateTypeSplitChange {
+		t.Error("Unexpected UpdateType for no segment case, got:", rbUpdateNoSegment.UpdateType())
+	}
+	if rbUpdateNoSegment.RuleBasedSegment() != nil {
+		t.Error("RuleBasedSegment should be nil for no segment case")
+	}
+}
+
 func TestLargeSegmentChangeUpdate(t *testing.T) {
 	ls := []LargeSegmentRFDResponseDTO{
 		{

dtos/split.go

@@ -6,8 +6,8 @@ import (
 
 // SplitChangesDTO structure to map JSON message sent by Split servers.
 type SplitChangesDTO struct {
-	FeatureFlags      FeatureFlagsDTO        `json:"ff"`
-	RuleBasedSegments []RuleBasedSegmentsDTO `json:"rbs"`
+	FeatureFlags      FeatureFlagsDTO      `json:"ff"`
+	RuleBasedSegments RuleBasedSegmentsDTO `json:"rbs"`
 }
 
 type FeatureFlagsDTO struct {

engine/validator/matchers.go

@@ -21,6 +21,15 @@ var unsupportedMatcherConditionReplacement []dtos.ConditionDTO = []dtos.Conditio
 	},
 }}
 
+// unsupportedMatcherRBConditionReplacement is the default condition to be used when a matcher is not supported
+var unsupportedMatcherRBConditionReplacement []dtos.RuleBasedConditionDTO = []dtos.RuleBasedConditionDTO{{
+	ConditionType: grammar.ConditionTypeWhitelist,
+	MatcherGroup: dtos.MatcherGroupDTO{
+		Combiner: "AND",
+		Matchers: []dtos.MatcherDTO{{MatcherType: grammar.MatcherTypeAllKeys, Negate: false}},
+	},
+}}
+
 func shouldOverrideConditions(conditions []dtos.ConditionDTO, logger logging.LoggerInterface) bool {
 	for _, condition := range conditions {
 		for _, matcher := range condition.MatcherGroup.Matchers {
@@ -33,13 +42,32 @@ func shouldOverrideConditions(conditions []dtos.ConditionDTO, logger logging.Log
 	return false
 }
 
+func shouldOverrideRBConditions(conditions []dtos.RuleBasedConditionDTO, logger logging.LoggerInterface) bool {
+	for _, condition := range conditions {
+		for _, matcher := range condition.MatcherGroup.Matchers {
+			_, err := grammar.BuildMatcher(&matcher, &injection.Context{}, logger)
+			if _, ok := err.(datatypes.UnsupportedMatcherError); ok {
+				return true
+			}
+		}
+	}
+	return false
+}
+
 // ProcessMatchers processes the matchers of a split and validates them
 func ProcessMatchers(split *dtos.SplitDTO, logger logging.LoggerInterface) {
 	if shouldOverrideConditions(split.Conditions, logger) {
 		split.Conditions = unsupportedMatcherConditionReplacement
 	}
 }
 
+// ProcessMatchers processes the matchers of a rule-based and validates them
+func ProcessRBMatchers(ruleBased *dtos.RuleBasedSegmentDTO, logger logging.LoggerInterface) {
+	if shouldOverrideRBConditions(ruleBased.Conditions, logger) {
+		ruleBased.Conditions = unsupportedMatcherRBConditionReplacement
+	}
+}
+
 // MakeUnsupportedMatcherConditionReplacement returns the default condition to be used when a matcher is not supported
 func MakeUnsupportedMatcherConditionReplacement() []dtos.ConditionDTO {
 	return unsupportedMatcherConditionReplacement

engine/validator/matchers_test.go

@@ -9,6 +9,60 @@ import (
 	"github.com/splitio/go-toolkit/v5/logging"
 )
 
+func TestProcessRBMatchers(t *testing.T) {
+	// Test case 1: Rule-based segment with unsupported matcher
+	ruleBased := &dtos.RuleBasedSegmentDTO{
+		Name:         "test-segment",
+		ChangeNumber: 123,
+		Conditions: []dtos.RuleBasedConditionDTO{
+			{
+				ConditionType: grammar.ConditionTypeRollout,
+				MatcherGroup: dtos.MatcherGroupDTO{
+					Matchers: []dtos.MatcherDTO{
+						{MatcherType: "NEW_MATCHER", KeySelector: nil},
+					},
+				},
+			},
+		},
+	}
+	ProcessRBMatchers(ruleBased, logging.NewLogger(nil))
+	if len(ruleBased.Conditions) != 1 {
+		t.Error("Conditions should have been overridden")
+	}
+	if ruleBased.Conditions[0].ConditionType != grammar.ConditionTypeWhitelist {
+		t.Error("ConditionType should be WHITELIST")
+	}
+	if ruleBased.Conditions[0].MatcherGroup.Matchers[0].MatcherType != grammar.MatcherTypeAllKeys {
+		t.Error("MatcherType should be ALL_KEYS")
+	}
+
+	// Test case 2: Rule-based segment with supported matcher
+	ruleBased = &dtos.RuleBasedSegmentDTO{
+		Name:         "test-segment",
+		ChangeNumber: 123,
+		Conditions: []dtos.RuleBasedConditionDTO{
+			{
+				ConditionType: grammar.ConditionTypeRollout,
+				MatcherGroup: dtos.MatcherGroupDTO{
+					Matchers: []dtos.MatcherDTO{
+						{MatcherType: grammar.MatcherTypeEndsWith, KeySelector: nil, String: common.StringRef("test")},
+					},
+				},
+			},
+		},
+	}
+	ProcessRBMatchers(ruleBased, logging.NewLogger(nil))
+	if len(ruleBased.Conditions) != 1 {
+		t.Error("Conditions should not have been overridden")
+	}
+	if ruleBased.Conditions[0].ConditionType != grammar.ConditionTypeRollout {
+		t.Error("ConditionType should be ROLLOUT")
+	}
+	if ruleBased.Conditions[0].MatcherGroup.Matchers[0].MatcherType != grammar.MatcherTypeEndsWith {
+		t.Error("MatcherType should be ENDS_WITH")
+	}
+}
+
 func TestProcessMatchers(t *testing.T) {
 	split := &dtos.SplitDTO{
 		Conditions: []dtos.ConditionDTO{

go.mod

@@ -12,8 +12,10 @@ require (
 
 require (
 	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/kr/pretty v0.1.0 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/stretchr/objx v0.5.2 // indirect
+	gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15 // indirect
 )
 
 require (

go.sum

@@ -10,6 +10,11 @@ github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78=
 github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc=
+github.com/kr/pretty v0.1.0 h1:L/CwN0zerZDmRFUapSPitk6f+Q3+0za1rQkzVuMiMFI=
+github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
+github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
+github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE=
+github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/redis/go-redis/v9 v9.0.4 h1:FC82T+CHJ/Q/PdyLW++GeCO+Ol59Y4T7R4jbgjvktgc=
@@ -26,7 +31,8 @@ golang.org/x/exp v0.0.0-20231006140011-7918f672742d h1:jtJma62tbqLibJ5sFQz8bKtEM
 golang.org/x/exp v0.0.0-20231006140011-7918f672742d/go.mod h1:ldy0pHrwJyGW56pPQzzkH36rKxoZW1tw7ZJpeKx+hdo=
 golang.org/x/sync v0.3.0 h1:ftCYgMx6zT/asHUrPw8BLLscYtGznsLAnjq5RH9P66E=
 golang.org/x/sync v0.3.0/go.mod h1:FU7BRWz2tNW+3quACPkgCx/L+uEAv1htQ0V83Z9Rj+Y=
-gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15 h1:YR8cESwS4TdDjEe65xsg0ogRM/Nc3DYOhEAlW+xobZo=
+gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

push/parser.go

@@ -138,8 +138,14 @@ func (p *NotificationParserImpl) parseUpdate(data *genericData, nested *genericM
 	case dtos.UpdateTypeLargeSegmentChange:
 		largeSegments := p.processLargeSegmentMessage(nested)
 		return nil, p.onLargeSegmentUpdate(dtos.NewLargeSegmentChangeUpdate(base, largeSegments))
-	case dtos.UpdateTypeContol:
+	case dtos.UpdateTypeControl:
 		return p.onControlUpdate(dtos.NewControlUpdate(base.BaseMessage, nested.ControlType)), nil
+	case dtos.UpdateTypeRuleBasedChange:
+		ruleBased := p.processRuleBasedMessage(nested)
+		if ruleBased == nil {
+			return nil, p.onSplitUpdate(dtos.NewRuleBasedSegmentChangeUpdate(base, nil, nil))
+		}
+		return nil, p.onSplitUpdate(dtos.NewRuleBasedSegmentChangeUpdate(base, &nested.PreviousChangeNumber, ruleBased))
 	default:
 		// TODO: log full event in debug mode
 		return nil, fmt.Errorf("invalid update type: %s", nested.Type)
@@ -157,10 +163,10 @@ func (p *NotificationParserImpl) processLargeSegmentMessage(nested *genericMessa
 
 func (p *NotificationParserImpl) processMessage(nested *genericMessageData) *dtos.SplitDTO {
 	compressType := getCompressType(nested.CompressType)
-	if nested.FeatureFlagDefinition == nil || compressType == nil {
+	if nested.Definition == nil || compressType == nil {
 		return nil
 	}
-	ffDecoded, err := p.dataUtils.Decode(common.StringFromRef(nested.FeatureFlagDefinition))
+	ffDecoded, err := p.dataUtils.Decode(common.StringFromRef(nested.Definition))
 	if err != nil {
 		p.logger.Debug(fmt.Sprintf("error decoding FeatureFlagDefinition: '%s'", err.Error()))
 		return nil
@@ -182,6 +188,33 @@ func (p *NotificationParserImpl) processMessage(nested *genericMessageData) *dto
 	return &featureFlag
 }
 
+func (p *NotificationParserImpl) processRuleBasedMessage(nested *genericMessageData) *dtos.RuleBasedSegmentDTO {
+	compressType := getCompressType(nested.CompressType)
+	if nested.Definition == nil || compressType == nil {
+		return nil
+	}
+	ruleBasedDecoded, err := p.dataUtils.Decode(common.StringFromRef(nested.Definition))
+	if err != nil {
+		p.logger.Debug(fmt.Sprintf("error decoding RuleBasedSegmentDefinition: '%s'", err.Error()))
+		return nil
+	}
+	if common.IntFromRef(compressType) != datautils.None {
+		ruleBasedDecoded, err = p.dataUtils.Decompress(ruleBasedDecoded, common.IntFromRef(compressType))
+		if err != nil {
+			p.logger.Debug(fmt.Sprintf("error decompressing RulebasedSegmentDefinition: '%s'", err.Error()))
+			return nil
+		}
+	}
+
+	var ruleBased dtos.RuleBasedSegmentDTO
+	err = json.Unmarshal([]byte(ruleBasedDecoded), &ruleBased)
+	if err != nil {
+		p.logger.Debug(fmt.Sprintf("error parsing rule-based segment json definition: '%s'", err.Error()))
+		return nil
+	}
+	return &ruleBased
+}
+
 type genericData struct {
 
 	// Error associated data
@@ -207,17 +240,17 @@ type metrics struct {
 }
 
 type genericMessageData struct {
-	Metrics               metrics                           `json:"metrics"`
-	Type                  string                            `json:"type"`
-	ChangeNumber          int64                             `json:"changeNumber"`
-	SplitName             string                            `json:"splitName"`
-	DefaultTreatment      string                            `json:"defaultTreatment"`
-	SegmentName           string                            `json:"segmentName"`
-	ControlType           string                            `json:"controlType"`
-	PreviousChangeNumber  int64                             `json:"pcn"`
-	CompressType          *int                              `json:"c"`
-	FeatureFlagDefinition *string                           `json:"d"`
-	LargeSegments         []dtos.LargeSegmentRFDResponseDTO `json:"ls"`
+	Metrics              metrics                           `json:"metrics"`
+	Type                 string                            `json:"type"`
+	ChangeNumber         int64                             `json:"changeNumber"`
+	SplitName            string                            `json:"splitName"`
+	DefaultTreatment     string                            `json:"defaultTreatment"`
+	SegmentName          string                            `json:"segmentName"`
+	ControlType          string                            `json:"controlType"`
+	PreviousChangeNumber int64                             `json:"pcn"`
+	CompressType         *int                              `json:"c"`
+	Definition           *string                           `json:"d"`
+	LargeSegments        []dtos.LargeSegmentRFDResponseDTO `json:"ls"`
 
 	// {\"type\":\"SPLIT_UPDATE\",\"changeNumber\":1612909342671}"}
 }