feat: filter destinations for events from rETL in processor

Author: mihir20

processor/processor.go

@@ -1544,13 +1544,16 @@ func (proc *Handle) processJobsForDest(partition string, subJobs subJob) *transf
 
 	outCountMap := make(map[string]int64) // destinations enabled
 	destFilterStatusDetailMap := make(map[string]map[string]*types.StatusDetail)
+	// map of messageID to destinationID: for messages that needs to be delivered to a specific destinations only
+	msgToDestMap := make(map[string]string)
 
 	spans := make([]stats.TraceSpan, 0, len(jobList))
 	defer func() {
 		for _, span := range spans {
 			span.End()
 		}
 	}()
+
 	for _, batchEvent := range jobList {
 		var eventParams types.EventParams
 		err := jsonfast.Unmarshal(batchEvent.Parameters, &eventParams)
@@ -1560,6 +1563,7 @@ func (proc *Handle) processJobsForDest(partition string, subJobs subJob) *transf
 
 		sourceID := eventParams.SourceId
 		traceParent := eventParams.TraceParent
+		destinationID := eventParams.DestinationID
 
 		var span stats.TraceSpan
 		if traceParent == "" {
@@ -1617,6 +1621,9 @@ func (proc *Handle) processJobsForDest(partition string, subJobs subJob) *transf
 		// Iterate through all the events in the batch
 		for _, singularEvent := range gatewayBatchEvent.Batch {
 			messageId := misc.GetStringifiedData(singularEvent["messageId"])
+			if len(destinationID) != 0 {
+				msgToDestMap[messageId] = destinationID
+			}
 
 			payloadFunc := ro.Memoize(func() json.RawMessage {
 				payloadBytes, err := jsonfast.Marshal(singularEvent)
@@ -1724,8 +1731,10 @@ func (proc *Handle) processJobsForDest(partition string, subJobs subJob) *transf
 			// REPORTING - GATEWAY metrics - END
 
 			// Getting all the destinations which are enabled for this
-			// event
-			if !proc.isDestinationAvailable(singularEvent, sourceID) {
+			// event will be dropped if no valid destination is present
+			// if empty string destinationID is all the destinations for the source is validated
+			// else only passed destinationID will be validated
+			if !proc.isDestinationAvailable(singularEvent, sourceId, destinationID) {
 				continue
 			}
 
@@ -1893,7 +1902,13 @@ func (proc *Handle) processJobsForDest(partition string, subJobs subJob) *transf
 				destType := &enabledDestTypes[i]
 				enabledDestinationsList := proc.getConsentFilteredDestinations(
 					singularEvent,
-					proc.getEnabledDestinations(sourceId, *destType),
+					lo.Filter(proc.getEnabledDestinations(sourceId, *destType), func(item backendconfig.DestinationT, index int) bool {
+						destId := msgToDestMap[event.Metadata.MessageID]
+						if len(destId) != 0 {
+							return destId == item.ID
+						}
+						return len(destId) == 0
+					}),
 				)
 
 				// Adding a singular event multiple times if there are multiple destinations of same type
@@ -3073,7 +3088,7 @@ func (*Handle) getLimiterPriority(partition string) kitsync.LimiterPriorityValue
 // check if event has eligible destinations to send to
 //
 // event will be dropped if no destination is found
-func (proc *Handle) isDestinationAvailable(event types.SingularEventT, sourceId string) bool {
+func (proc *Handle) isDestinationAvailable(event types.SingularEventT, sourceId string, destinationID string) bool {
 	enabledDestTypes := integrations.FilterClientIntegrations(
 		event,
 		proc.getBackendEnabledDestinationTypes(sourceId),
@@ -3083,7 +3098,7 @@ func (proc *Handle) isDestinationAvailable(event types.SingularEventT, sourceId
 		return false
 	}
 
-	if enabledDestinationsList := proc.getConsentFilteredDestinations(
+	if enabledDestinationsList := lo.Filter(proc.getConsentFilteredDestinations(
 		event,
 		lo.Flatten(
 			lo.Map(
@@ -3093,7 +3108,9 @@ func (proc *Handle) isDestinationAvailable(event types.SingularEventT, sourceId
 				},
 			),
 		),
-	); len(enabledDestinationsList) == 0 {
+	), func(dest backendconfig.DestinationT, index int) bool {
+		return len(destinationID) == 0 || dest.ID == destinationID
+	}); len(enabledDestinationsList) == 0 {
 		proc.logger.Debug("No destination to route this event to")
 		return false
 	}

processor/processor_test.go

@@ -2772,7 +2772,7 @@ var _ = Describe("Processor", Ordered, func() {
 			defer cancel()
 			Expect(processor.config.asyncInit.WaitContext(ctx)).To(BeNil())
 
-			Expect(processor.isDestinationAvailable(eventWithDeniedConsents, SourceIDOneTrustConsent)).To(BeTrue())
+			Expect(processor.isDestinationAvailable(eventWithDeniedConsents, SourceIDOneTrustConsent, "")).To(BeTrue())
 			Expect(
 				len(processor.getConsentFilteredDestinations(
 					eventWithDeniedConsents,
@@ -2783,7 +2783,7 @@ var _ = Describe("Processor", Ordered, func() {
 				)),
 			).To(Equal(3)) // all except D1 and D3
 
-			Expect(processor.isDestinationAvailable(eventWithoutDeniedConsents, SourceIDOneTrustConsent)).To(BeTrue())
+			Expect(processor.isDestinationAvailable(eventWithoutDeniedConsents, SourceIDOneTrustConsent, "")).To(BeTrue())
 			Expect(
 				len(processor.getConsentFilteredDestinations(
 					eventWithoutDeniedConsents,
@@ -2794,7 +2794,18 @@ var _ = Describe("Processor", Ordered, func() {
 				)),
 			).To(Equal(5)) // all
 
-			Expect(processor.isDestinationAvailable(eventWithoutConsentManagementData, SourceIDOneTrustConsent)).To(BeTrue())
+			Expect(processor.isDestinationAvailable(eventWithoutConsentManagementData, SourceIDOneTrustConsent, "")).To(BeTrue())
+			Expect(
+				len(processor.getConsentFilteredDestinations(
+					eventWithoutConsentManagementData,
+					processor.getEnabledDestinations(
+						SourceIDOneTrustConsent,
+						"destination-definition-name-enabled",
+					),
+				)),
+			).To(Equal(5)) // all
+
+			Expect(processor.isDestinationAvailable(eventWithoutConsentManagementData, SourceIDOneTrustConsent, "dest-id-1")).To(BeTrue())
 			Expect(
 				len(processor.getConsentFilteredDestinations(
 					eventWithoutConsentManagementData,
@@ -2853,7 +2864,7 @@ var _ = Describe("Processor", Ordered, func() {
 				),
 			)
 			Expect(len(filteredDestinations)).To(Equal(4)) // all except dest-id-5 since both purpose1 and purpose2 are denied
-			Expect(processor.isDestinationAvailable(event, SourceIDKetchConsent)).To(BeTrue())
+			Expect(processor.isDestinationAvailable(event, SourceIDKetchConsent, "")).To(BeTrue())
 		})
 
 		It("should filter based on generic consent management preferences", func() {
@@ -3010,7 +3021,7 @@ var _ = Describe("Processor", Ordered, func() {
 			defer cancel()
 			Expect(processor.config.asyncInit.WaitContext(ctx)).To(BeNil())
 
-			Expect(processor.isDestinationAvailable(eventWithoutConsentManagementData, SourceIDGCM)).To(BeTrue())
+			Expect(processor.isDestinationAvailable(eventWithoutConsentManagementData, SourceIDGCM, "")).To(BeTrue())
 			Expect(
 				len(processor.getConsentFilteredDestinations(
 					eventWithoutConsentManagementData,
@@ -3021,7 +3032,7 @@ var _ = Describe("Processor", Ordered, func() {
 				)),
 			).To(Equal(9)) // all
 
-			Expect(processor.isDestinationAvailable(eventWithoutDeniedConsentsGCM, SourceIDGCM)).To(BeTrue())
+			Expect(processor.isDestinationAvailable(eventWithoutDeniedConsentsGCM, SourceIDGCM, "")).To(BeTrue())
 			Expect(
 				len(processor.getConsentFilteredDestinations(
 					eventWithoutDeniedConsentsGCM,
@@ -3032,7 +3043,7 @@ var _ = Describe("Processor", Ordered, func() {
 				)),
 			).To(Equal(9)) // all
 
-			Expect(processor.isDestinationAvailable(eventWithCustomConsentsGCM, SourceIDGCM)).To(BeTrue())
+			Expect(processor.isDestinationAvailable(eventWithCustomConsentsGCM, SourceIDGCM, "")).To(BeTrue())
 			Expect(
 				len(processor.getConsentFilteredDestinations(
 					eventWithCustomConsentsGCM,
@@ -3043,7 +3054,7 @@ var _ = Describe("Processor", Ordered, func() {
 				)),
 			).To(Equal(8)) // all except D13
 
-			Expect(processor.isDestinationAvailable(eventWithDeniedConsentsGCM, SourceIDGCM)).To(BeTrue())
+			Expect(processor.isDestinationAvailable(eventWithDeniedConsentsGCM, SourceIDGCM, "")).To(BeTrue())
 			Expect(
 				len(processor.getConsentFilteredDestinations(
 					eventWithDeniedConsentsGCM,
@@ -3054,7 +3065,7 @@ var _ = Describe("Processor", Ordered, func() {
 				)),
 			).To(Equal(7)) // all except D6 and D7
 
-			Expect(processor.isDestinationAvailable(eventWithDeniedConsentsGCMKetch, SourceIDGCM)).To(BeTrue())
+			Expect(processor.isDestinationAvailable(eventWithDeniedConsentsGCMKetch, SourceIDGCM, "")).To(BeTrue())
 			Expect(
 				len(processor.getConsentFilteredDestinations(
 					eventWithDeniedConsentsGCMKetch,
@@ -3064,6 +3075,15 @@ var _ = Describe("Processor", Ordered, func() {
 					),
 				)),
 			).To(Equal(8)) // all except D7
+
+			// some unknown destination ID is passed destination will be unavailable
+			Expect(processor.isDestinationAvailable(eventWithDeniedConsentsGCMKetch, SourceIDGCM, "unknown-destination")).To(BeFalse())
+
+			// known destination ID is passed and destination is enabled
+			Expect(processor.isDestinationAvailable(eventWithDeniedConsentsGCMKetch, SourceIDTransient, DestinationIDEnabledA)).To(BeTrue())
+
+			// know destination ID is passed and destination is not enabled
+			Expect(processor.isDestinationAvailable(eventWithDeniedConsentsGCMKetch, SourceIDTransient, DestinationIDDisabled)).To(BeFalse())
 		})
 	})
 

utils/types/types.go

@@ -36,6 +36,7 @@ type EventParams struct {
 	SourceId        string `json:"source_id"`
 	SourceTaskRunId string `json:"source_task_run_id"`
 	TraceParent     string `json:"traceparent"`
+	DestinationID   string `json:"destination_id"`
 }
 
 // UserSuppression is interface to access Suppress user feature