[RHCLOUD-32506] Send relevant data from incoming payloads to Kessel

connector-drawer/src/main/java/com/redhat/cloud/notifications/connector/drawer/DrawerCloudEventDataExtractor.java

@@ -16,5 +16,7 @@ public void extract(Exchange exchange, JsonObject cloudEventData) {
         exchange.setProperty(ExchangeProperty.DRAWER_ENTRY_PAYLOAD, notification.drawerEntryPayload());
         exchange.setProperty(ExchangeProperty.RECIPIENT_SETTINGS, notification.recipientSettings());
         exchange.setProperty(ExchangeProperty.UNSUBSCRIBERS, notification.unsubscribers());
+        exchange.setProperty(ExchangeProperty.UNSUBSCRIBERS, notification.unsubscribers());
+        exchange.setProperty(ExchangeProperty.AUTHZ_CRITERIA, notification.authzCriteria());
     }
 }

connector-drawer/src/main/java/com/redhat/cloud/notifications/connector/drawer/constant/ExchangeProperty.java

@@ -9,4 +9,5 @@ public class ExchangeProperty {
     public static final String UNSUBSCRIBERS = "unsubscribers";
 
     public static final String DRAWER_ENTRY_PAYLOAD = "drawer_entry_payload";
+    public static final String AUTHZ_CRITERIA = "authz_criteria";
 }

connector-drawer/src/main/java/com/redhat/cloud/notifications/connector/drawer/model/DrawerNotificationToConnector.java

@@ -1,6 +1,7 @@
 package com.redhat.cloud.notifications.connector.drawer.model;
 
 import com.fasterxml.jackson.annotation.JsonProperty;
+import io.vertx.core.json.JsonObject;
 import java.util.Collection;
 
 public record DrawerNotificationToConnector(
@@ -15,5 +16,8 @@ public record DrawerNotificationToConnector(
     Collection<RecipientSettings> recipientSettings,
 
     @JsonProperty("unsubscribers")
-    Collection<String> unsubscribers
+    Collection<String> unsubscribers,
+
+    @JsonProperty("authz_criteria")
+    JsonObject authzCriteria
 )  { }

connector-drawer/src/main/java/com/redhat/cloud/notifications/connector/drawer/recipients/RecipientsResolverPreparer.java

@@ -5,6 +5,7 @@
 import com.redhat.cloud.notifications.connector.drawer.constant.ExchangeProperty;
 import com.redhat.cloud.notifications.connector.drawer.model.RecipientSettings;
 import com.redhat.cloud.notifications.connector.drawer.recipients.pojo.RecipientsQuery;
+import io.vertx.core.json.JsonObject;
 import jakarta.enterprise.context.ApplicationScoped;
 import jakarta.inject.Inject;
 import org.apache.camel.Exchange;
@@ -26,12 +27,14 @@ public void process(final Exchange exchange) throws JsonProcessingException {
         List<RecipientSettings> recipientSettings = exchange.getProperty(ExchangeProperty.RECIPIENT_SETTINGS, List.class);
         Set<String> unsubscribers = exchange.getProperty(ExchangeProperty.UNSUBSCRIBERS, Set.class);
         final String orgId = exchange.getProperty(ORG_ID, String.class);
+        JsonObject authzCriteria = exchange.getProperty(ExchangeProperty.AUTHZ_CRITERIA, JsonObject.class);
 
         RecipientsQuery recipientsQuery = new RecipientsQuery();
         recipientsQuery.unsubscribers = unsubscribers;
         recipientsQuery.orgId = orgId;
         recipientsQuery.recipientSettings = Set.copyOf(recipientSettings);
         recipientsQuery.subscribedByDefault = true;
+        recipientsQuery.authzCriteria = authzCriteria;
 
         // Serialize the payload.
         exchange.getMessage().setBody(objectMapper.writeValueAsString(recipientsQuery));

connector-drawer/src/main/java/com/redhat/cloud/notifications/connector/drawer/recipients/pojo/RecipientsQuery.java

@@ -2,6 +2,7 @@
 
 import com.fasterxml.jackson.databind.annotation.JsonNaming;
 import com.redhat.cloud.notifications.connector.drawer.model.RecipientSettings;
+import io.vertx.core.json.JsonObject;
 import jakarta.validation.constraints.NotBlank;
 import jakarta.validation.constraints.NotNull;
 import java.util.Set;
@@ -22,4 +23,6 @@ public class RecipientsQuery {
     public Set<String> unsubscribers;
 
     public boolean subscribedByDefault;
+
+    public JsonObject authzCriteria;
 }

connector-drawer/src/test/java/com/redhat/cloud/notifications/connector/drawer/DrawerConnectorRoutesTest.java

@@ -103,7 +103,7 @@ private static DrawerNotificationToConnector buildTestDrawerNotificationToConnec
         drawerEntryPayload.setBundle("My Bundle");
 
         RecipientSettings recipientSettings = new RecipientSettings();
-        return new DrawerNotificationToConnector(orgId, drawerEntryPayload, Set.of(recipientSettings), List.of("user-1", "user-2"));
+        return new DrawerNotificationToConnector(orgId, drawerEntryPayload, Set.of(recipientSettings), List.of("user-1", "user-2"), new JsonObject());
     }
 
     private HttpRequest getMockHttpRequest(String path, ExpectationResponseCallback expectationResponseCallback) {

connector-email/src/main/java/com/redhat/cloud/notifications/connector/email/EmailCloudEventDataExtractor.java

@@ -47,6 +47,8 @@ public void extract(final Exchange exchange, final JsonObject cloudEventData) {
                 .map(String.class::cast)
                 .collect(toSet());
 
+        final JsonObject authzCriteria = cloudEventData.getJsonObject("authz_criteria");
+
         final Set<String> emails = recipientSettings.stream()
                 .filter(settings -> settings.getEmails() != null)
                 .flatMap(settings -> settings.getEmails().stream())
@@ -58,6 +60,7 @@ public void extract(final Exchange exchange, final JsonObject cloudEventData) {
         exchange.setProperty(ExchangeProperty.SUBSCRIBED_BY_DEFAULT, cloudEventData.getBoolean("subscribed_by_default"));
         exchange.setProperty(ExchangeProperty.SUBSCRIBERS, subscribers);
         exchange.setProperty(ExchangeProperty.UNSUBSCRIBERS, unsubscribers);
+        exchange.setProperty(ExchangeProperty.AUTHZ_CRITERIA, authzCriteria);
         exchange.setProperty(ExchangeProperty.EMAIL_RECIPIENTS, emails);
         exchange.setProperty(ExchangeProperty.EMAIL_SENDER, cloudEventData.getString("email_sender"));
 

connector-email/src/main/java/com/redhat/cloud/notifications/connector/email/constants/ExchangeProperty.java

@@ -48,4 +48,6 @@ public class ExchangeProperty {
     public static final String RECIPIENTS_SIZE = "recipientsSize";
 
     public static final String USE_EMAIL_BOP_V2_SERVICE = "use_email_bop_V2_service";
+    public static final String AUTHZ_CRITERIA = "authz_criteria";
+
 }

connector-email/src/main/java/com/redhat/cloud/notifications/connector/email/processors/recipients/RecipientsQuery.java

@@ -2,6 +2,7 @@
 
 import com.fasterxml.jackson.databind.annotation.JsonNaming;
 import com.redhat.cloud.notifications.connector.email.model.settings.RecipientSettings;
+import io.vertx.core.json.JsonObject;
 import jakarta.validation.constraints.NotBlank;
 import jakarta.validation.constraints.NotNull;
 import java.util.Set;
@@ -22,4 +23,6 @@ public class RecipientsQuery {
     public Set<String> unsubscribers;
 
     public boolean subscribedByDefault;
+
+    public JsonObject authzCriteria;
 }

connector-email/src/main/java/com/redhat/cloud/notifications/connector/email/processors/recipients/RecipientsResolverRequestPreparer.java

@@ -4,6 +4,7 @@
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.redhat.cloud.notifications.connector.email.constants.ExchangeProperty;
 import com.redhat.cloud.notifications.connector.email.model.settings.RecipientSettings;
+import io.vertx.core.json.JsonObject;
 import jakarta.enterprise.context.ApplicationScoped;
 import jakarta.inject.Inject;
 import org.apache.camel.Exchange;
@@ -25,6 +26,8 @@ public void process(final Exchange exchange) throws JsonProcessingException {
         List<RecipientSettings> recipientSettings = exchange.getProperty(ExchangeProperty.RECIPIENT_SETTINGS, List.class);
         Set<String> subscribers = exchange.getProperty(ExchangeProperty.SUBSCRIBERS, Set.class);
         Set<String> unsubscribers = exchange.getProperty(ExchangeProperty.UNSUBSCRIBERS, Set.class);
+        JsonObject authzCriteria = exchange.getProperty(ExchangeProperty.AUTHZ_CRITERIA, JsonObject.class);
+
         boolean subscribedByDefault = exchange.getProperty(ExchangeProperty.SUBSCRIBED_BY_DEFAULT, boolean.class);
         final String orgId = exchange.getProperty(ORG_ID, String.class);
 
@@ -34,6 +37,7 @@ public void process(final Exchange exchange) throws JsonProcessingException {
         recipientsQuery.orgId = orgId;
         recipientsQuery.recipientSettings = Set.copyOf(recipientSettings);
         recipientsQuery.subscribedByDefault = subscribedByDefault;
+        recipientsQuery.authzCriteria = authzCriteria;
 
         // Serialize the payload.
         exchange.getMessage().setBody(objectMapper.writeValueAsString(recipientsQuery));

engine/src/main/java/com/redhat/cloud/notifications/processors/AuthzCriteriaExtractor.java

@@ -0,0 +1,33 @@
+package com.redhat.cloud.notifications.processors;
+
+import com.redhat.cloud.notifications.models.EmailAggregation;
+import com.redhat.cloud.notifications.models.Event;
+import com.redhat.cloud.notifications.transformers.BaseTransformer;
+import io.vertx.core.json.JsonObject;
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.inject.Inject;
+
+@ApplicationScoped
+public class AuthzCriteriaExtractor {
+
+    public static final String AUTHZ_CRITERIA = "authz_criteria";
+
+    @Inject
+    BaseTransformer baseTransformer;
+
+    public JsonObject extract(Event event) {
+        JsonObject data = baseTransformer.toJsonObject(event);
+        if (null != data.getJsonObject(BaseTransformer.CONTEXT) && null != data.getJsonObject(BaseTransformer.CONTEXT).getJsonObject(AUTHZ_CRITERIA)) {
+            return data.getJsonObject(BaseTransformer.CONTEXT).getJsonObject(AUTHZ_CRITERIA);
+        }
+        return null;
+    }
+
+    public JsonObject extract(EmailAggregation emailAggregation) {
+        JsonObject data = emailAggregation.getPayload();
+        if (null != data.getJsonObject(BaseTransformer.CONTEXT) && null != data.getJsonObject(BaseTransformer.CONTEXT).getJsonObject(AUTHZ_CRITERIA)) {
+            return data.getJsonObject(BaseTransformer.CONTEXT).getJsonObject(AUTHZ_CRITERIA);
+        }
+        return null;
+    }
+}

engine/src/main/java/com/redhat/cloud/notifications/processors/drawer/DrawerNotificationToConnector.java

@@ -3,6 +3,7 @@
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.redhat.cloud.notifications.models.DrawerEntryPayload;
 import com.redhat.cloud.notifications.processors.email.connector.dto.RecipientSettings;
+import io.vertx.core.json.JsonObject;
 import java.util.Collection;
 
 public record DrawerNotificationToConnector(
@@ -17,5 +18,8 @@ public record DrawerNotificationToConnector(
     Collection<RecipientSettings> recipientSettings,
 
     @JsonProperty("unsubscribers")
-    Collection<String> unsubscribers
+    Collection<String> unsubscribers,
+
+    @JsonProperty("authz_criteria")
+    JsonObject authzCriteria
 )  { }

engine/src/main/java/com/redhat/cloud/notifications/processors/drawer/DrawerProcessor.java

@@ -15,6 +15,7 @@
 import com.redhat.cloud.notifications.models.EndpointType;
 import com.redhat.cloud.notifications.models.Event;
 import com.redhat.cloud.notifications.models.IntegrationTemplate;
+import com.redhat.cloud.notifications.processors.AuthzCriteriaExtractor;
 import com.redhat.cloud.notifications.processors.ConnectorSender;
 import com.redhat.cloud.notifications.processors.SystemEndpointTypeProcessor;
 import com.redhat.cloud.notifications.processors.email.connector.dto.RecipientSettings;
@@ -73,6 +74,9 @@ public class DrawerProcessor extends SystemEndpointTypeProcessor {
     @Inject
     BundleRepository bundleRepository;
 
+    @Inject
+    AuthzCriteriaExtractor authzCriteriaExtractor;
+
     @Override
     public void process(Event event, List<Endpoint> endpoints) {
         if (!engineConfig.isDrawerEnabled()) {
@@ -103,7 +107,8 @@ public void process(Event event, List<Endpoint> endpoints) {
             event.getOrgId(),
             drawerEntryPayload,
             recipientSettings,
-            unsubscribers
+            unsubscribers,
+            authzCriteriaExtractor.extract(event)
         );
 
         connectorSender.send(event, endpoint, JsonObject.mapFrom(drawerNotificationToConnector));

engine/src/main/java/com/redhat/cloud/notifications/processors/email/EmailAggregationProcessor.java

@@ -345,7 +345,9 @@ private void processBundleAggregation(List<AggregationCommand> aggregationComman
                      */
                     recipientsUsernames,
                     Collections.emptySet(),
-                    false
+                    false,
+                    // because recipient list has already been computed using authz constraints, we don't need it for the aggregated email
+                    null
                 );
 
                 connectorSender.send(aggregatorEvent, endpoint, JsonObject.mapFrom(emailNotification));

engine/src/main/java/com/redhat/cloud/notifications/processors/email/EmailAggregator.java

@@ -11,6 +11,7 @@
 import com.redhat.cloud.notifications.models.Endpoint;
 import com.redhat.cloud.notifications.models.EventType;
 import com.redhat.cloud.notifications.models.SubscriptionType;
+import com.redhat.cloud.notifications.processors.AuthzCriteriaExtractor;
 import com.redhat.cloud.notifications.processors.email.aggregators.AbstractEmailPayloadAggregator;
 import com.redhat.cloud.notifications.processors.email.aggregators.EmailPayloadAggregatorFactory;
 import com.redhat.cloud.notifications.recipients.RecipientResolver;
@@ -67,6 +68,9 @@ public class EmailAggregator {
     private static final String EVENT_TYPE_KEY = "event_type";
     private static final String RECIPIENTS_KEY = "recipients";
 
+    @Inject
+    AuthzCriteriaExtractor authzCriteriaExtractor;
+
     @ConfigProperty(name = "notifications.aggregation.max-page-size", defaultValue = "100")
     int maxPageSize;
 
@@ -89,6 +93,8 @@ public Map<User, Map<String, Object>> getAggregated(UUID appId, EmailAggregation
 
             // For each aggregation...
             for (EmailAggregation aggregation : aggregations) {
+
+                JsonObject authzCriteria = authzCriteriaExtractor.extract(aggregation);
                 // We need its event type to determine the target endpoints.
                 String eventTypeName = getEventType(aggregation);
                 EventType eventType;
@@ -128,7 +134,8 @@ public Map<User, Map<String, Object>> getAggregated(UUID appId, EmailAggregation
                             ).collect(toSet()),
                             subscribers,
                             unsubscribers,
-                            eventType.isSubscribedByDefault()
+                            eventType.isSubscribedByDefault(),
+                            authzCriteria
                         );
                     } catch (Exception ex) {
                         Log.error("Error calling external recipients resolver service", ex);

engine/src/main/java/com/redhat/cloud/notifications/processors/email/EmailProcessor.java

@@ -7,6 +7,7 @@
 import com.redhat.cloud.notifications.models.EndpointType;
 import com.redhat.cloud.notifications.models.Event;
 import com.redhat.cloud.notifications.models.InstantEmailTemplate;
+import com.redhat.cloud.notifications.processors.AuthzCriteriaExtractor;
 import com.redhat.cloud.notifications.processors.ConnectorSender;
 import com.redhat.cloud.notifications.processors.SystemEndpointTypeProcessor;
 import com.redhat.cloud.notifications.processors.email.connector.dto.EmailNotification;
@@ -51,6 +52,9 @@ public class EmailProcessor extends SystemEndpointTypeProcessor {
     @Inject
     SubscriptionRepository subscriptionRepository;
 
+    @Inject
+    AuthzCriteriaExtractor authzCriteriaExtractor;
+
     @Override
     public void process(final Event event, final List<Endpoint> endpoints) {
         // Generate an aggregation if the event supports it.
@@ -63,6 +67,8 @@ public void process(final Event event, final List<Endpoint> endpoints) {
             return;
         }
 
+        JsonObject authzCriteria = authzCriteriaExtractor.extract(event);
+
         Set<String> subscribers;
         Set<String> unsubscribers;
         if (event.getEventType().isSubscribedByDefault()) {
@@ -111,7 +117,8 @@ public void process(final Event event, final List<Endpoint> endpoints) {
             recipientSettings,
             subscribers,
             unsubscribers,
-            event.getEventType().isSubscribedByDefault()
+            event.getEventType().isSubscribedByDefault(),
+            authzCriteria
         );
 
         final JsonObject payload = JsonObject.mapFrom(emailNotification);

engine/src/main/java/com/redhat/cloud/notifications/processors/email/connector/dto/EmailNotification.java

@@ -1,7 +1,7 @@
 package com.redhat.cloud.notifications.processors.email.connector.dto;
 
 import com.fasterxml.jackson.annotation.JsonProperty;
-
+import io.vertx.core.json.JsonObject;
 import java.util.Collection;
 
 /**
@@ -30,5 +30,6 @@ public record EmailNotification(
     @JsonProperty("recipient_settings")     Collection<RecipientSettings> recipientSettings,
     @JsonProperty("subscribers")            Collection<String> subscribers,
     @JsonProperty("unsubscribers")          Collection<String> unsubscribers,
-    @JsonProperty("subscribed_by_default")  boolean subscribedByDefault
+    @JsonProperty("subscribed_by_default")  boolean subscribedByDefault,
+    @JsonProperty("authz_criteria")         JsonObject authzCriteria
 )  { }

engine/src/main/java/com/redhat/cloud/notifications/recipients/recipientsresolver/ExternalRecipientsResolver.java

@@ -8,6 +8,7 @@
 import dev.failsafe.function.CheckedSupplier;
 import io.quarkus.cache.CacheResult;
 import io.quarkus.logging.Log;
+import io.vertx.core.json.JsonObject;
 import jakarta.annotation.PostConstruct;
 import jakarta.enterprise.context.ApplicationScoped;
 import jakarta.inject.Inject;
@@ -58,7 +59,7 @@ private <T> T retryOnError(final CheckedSupplier<T> usersServiceCall) {
     }
 
     @CacheResult(cacheName = "recipients-resolver-results")
-    public Set<User> recipientUsers(String orgId, Set<RecipientSettings> recipientSettings, Set<String> subscribers, Set<String> unsubscribers, boolean subscribedByDefault) {
+    public Set<User> recipientUsers(String orgId, Set<RecipientSettings> recipientSettings, Set<String> subscribers, Set<String> unsubscribers, boolean subscribedByDefault, JsonObject authzCriteria) {
         RecipientsQuery recipientsQuery = new RecipientsQuery();
         recipientsQuery.subscribers = Set.copyOf(subscribers);
         recipientsQuery.unsubscribers = Set.copyOf(unsubscribers);
@@ -70,6 +71,7 @@ public Set<User> recipientUsers(String orgId, Set<RecipientSettings> recipientSe
 
         recipientsQuery.recipientSettings = recipientSettingsSet;
         recipientsQuery.subscribedByDefault = subscribedByDefault;
+        recipientsQuery.authzCriteria = authzCriteria;
         Set<User> recipientsList = retryOnError(() -> recipientsResolverService.getRecipients(recipientsQuery));
         return recipientsList;
     }

engine/src/main/java/com/redhat/cloud/notifications/recipients/recipientsresolver/pojo/RecipientsQuery.java

@@ -2,6 +2,7 @@
 
 import com.fasterxml.jackson.databind.annotation.JsonNaming;
 import com.redhat.cloud.notifications.processors.email.connector.dto.RecipientSettings;
+import io.vertx.core.json.JsonObject;
 import jakarta.validation.constraints.NotBlank;
 import jakarta.validation.constraints.NotNull;
 import java.util.Set;
@@ -22,4 +23,6 @@ public class RecipientsQuery {
     public Set<String> unsubscribers;
 
     public boolean subscribedByDefault;
+
+    public JsonObject authzCriteria;
 }