feat: filter destinations for events from rETL in processor

Author: mihir20

processor/processor.go

@@ -1528,6 +1528,9 @@ 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)
+
 	for _, batchEvent := range jobList {
 
 		var gatewayBatchEvent types.GatewayBatchRequest
@@ -1542,6 +1545,7 @@ func (proc *Handle) processJobsForDest(partition string, subJobs subJob) *transf
 			panic(err)
 		}
 		sourceId := eventParams.SourceId
+		destinationID := eventParams.DestinationID
 		requestIP := gatewayBatchEvent.RequestIP
 		receivedAt := gatewayBatchEvent.ReceivedAt
 
@@ -1573,6 +1577,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)
@@ -1680,8 +1687,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
 			}
 
@@ -1849,7 +1858,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
@@ -2984,7 +2999,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),
@@ -2994,7 +3009,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(
@@ -3004,7 +3019,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

@@ -2736,7 +2736,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,
@@ -2747,7 +2747,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,
@@ -2758,7 +2758,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,
@@ -2817,7 +2828,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() {
@@ -2974,7 +2985,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,
@@ -2985,7 +2996,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,
@@ -2996,7 +3007,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,
@@ -3007,7 +3018,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,
@@ -3018,7 +3029,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,
@@ -3028,6 +3039,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

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