[SELC-7501] Updated user-cdc to monitor toAddOnAggregates flags and propagate users from the aggregator to the aggregates (#300)

Author: gaetano-miglionico

apps/user-cdc/pom.xml

@@ -59,7 +59,7 @@
         <dependency>
             <groupId>com.azure</groupId>
             <artifactId>azure-data-tables</artifactId>
-            <version>12.1.1</version>
+            <version>12.3.15</version>
         </dependency>
         <dependency>
             <groupId>io.quarkus</groupId>

apps/user-cdc/src/main/java/it/pagopa/selfcare/user/event/UserInstitutionCdcService.java

@@ -15,9 +15,17 @@
 import io.quarkus.runtime.Startup;
 import io.quarkus.runtime.configuration.ConfigUtils;
 import io.smallrye.mutiny.Multi;
+import io.smallrye.mutiny.Uni;
+import io.smallrye.mutiny.subscription.MultiEmitter;
+import it.pagopa.selfcare.onboarding.common.PartyRole;
+import it.pagopa.selfcare.product.entity.ProductRoleInfo;
+import it.pagopa.selfcare.product.service.ProductService;
 import it.pagopa.selfcare.user.UserUtils;
 import it.pagopa.selfcare.user.client.EventHubFdRestClient;
 import it.pagopa.selfcare.user.client.EventHubRestClient;
+import it.pagopa.selfcare.user.event.client.InternalDelegationApiClient;
+import it.pagopa.selfcare.user.event.client.InternalUserApiClient;
+import it.pagopa.selfcare.user.event.client.InternalUserGroupApiClient;
 import it.pagopa.selfcare.user.event.entity.UserInstitution;
 import it.pagopa.selfcare.user.event.mapper.NotificationMapper;
 import it.pagopa.selfcare.user.event.repository.UserInstitutionRepository;
@@ -28,12 +36,17 @@
 import it.pagopa.selfcare.user.model.constants.OnboardedProductState;
 import jakarta.enterprise.context.ApplicationScoped;
 import jakarta.inject.Inject;
+import jakarta.ws.rs.core.Response;
 import lombok.extern.slf4j.Slf4j;
 import org.bson.BsonDocument;
 import org.bson.conversions.Bson;
 import org.eclipse.microprofile.config.inject.ConfigProperty;
 import org.eclipse.microprofile.rest.client.inject.RestClient;
 import org.jboss.resteasy.reactive.ClientWebApplicationException;
+import org.openapi.quarkus.internal_json.model.AddMembersToUserGroupDto;
+import org.openapi.quarkus.internal_json.model.AddUserRoleDto;
+import org.openapi.quarkus.internal_json.model.DelegationResponse;
+import org.openapi.quarkus.internal_json.model.Product;
 import org.openapi.quarkus.user_registry_json.api.UserApi;
 
 import java.time.Duration;
@@ -67,7 +80,7 @@ public class UserInstitutionCdcService {
 
 
     private final TelemetryClient telemetryClient;
-
+    private final ProductService productService;
     private final TableClient tableClient;
     private final String mongodbDatabase;
     private final ReactiveMongoClient mongoClient;
@@ -78,11 +91,26 @@ public class UserInstitutionCdcService {
     private final Integer maxRetry;
     private final boolean sendEventsEnabled;
     private final boolean sendFdEventsEnabled;
+    private final boolean addOnAggregatesEnabled;
+    private final List<String> addOnAggregatesGroupProducts;
 
 
     @RestClient
     @Inject
     UserApi userRegistryApi;
+
+    @RestClient
+    @Inject
+    InternalDelegationApiClient delegationApi;
+
+    @RestClient
+    @Inject
+    InternalUserApiClient userApi;
+
+    @RestClient
+    @Inject
+    InternalUserGroupApiClient userGroupApi;
+
     @RestClient
     @Inject
     EventHubRestClient eventHubRestClient;
@@ -101,8 +129,10 @@ public UserInstitutionCdcService(ReactiveMongoClient mongoClient,
                                      @ConfigProperty(name = "user-cdc.retry") Integer maxRetry,
                                      @ConfigProperty(name = "user-cdc.send-events.watch.enabled") Boolean sendEventsEnabled,
                                      @ConfigProperty(name = "user-cdc.send-events-fd.watch.enabled") Boolean sendFdEventsEnabled,
+                                     @ConfigProperty(name = "user-cdc.add-on-aggregates.watch.enabled") Boolean addOnAggregatesEnabled,
+                                     @ConfigProperty(name = "user-cdc.add-on-aggregates.group.products") List<String> addOnAggregatesGroupProducts,
                                      UserInstitutionRepository userInstitutionRepository,
-                                     TelemetryClient telemetryClient,
+                                     TelemetryClient telemetryClient, ProductService productService,
                                      TableClient tableClient, NotificationMapper notificationMapper) {
         this.mongoClient = mongoClient;
         this.mongodbDatabase = mongodbDatabase;
@@ -111,10 +141,13 @@ public UserInstitutionCdcService(ReactiveMongoClient mongoClient,
         this.retryMaxBackOff = retryMaxBackOff;
         this.retryMinBackOff = retryMinBackOff;
         this.telemetryClient = telemetryClient;
+        this.productService = productService;
         this.tableClient = tableClient;
         this.notificationMapper = notificationMapper;
         this.sendEventsEnabled = sendEventsEnabled;
         this.sendFdEventsEnabled = sendFdEventsEnabled;
+        this.addOnAggregatesEnabled = addOnAggregatesEnabled;
+        this.addOnAggregatesGroupProducts = addOnAggregatesGroupProducts;
         telemetryClient.getContext().getOperation().setName(OPERATION_NAME);
         initOrderStream();
     }
@@ -174,17 +207,21 @@ public void propagateDocumentToConsumers(ChangeStreamDocument<UserInstitution> d
 
         boolean userMailIsChanged = isUserMailChanged(userInstitutionChanged);
 
-        if (Boolean.FALSE.equals(userMailIsChanged)) {
+        if (!userMailIsChanged) {
             consumerUserInstitutionRepositoryEvent(document);
         }
 
-        if (Boolean.TRUE.equals(sendEventsEnabled)) {
+        if (sendEventsEnabled) {
             consumerToSendScUserEvent(document);
         }
 
-        if (Boolean.TRUE.equals(sendFdEventsEnabled) && hasFdProduct) {
+        if (sendFdEventsEnabled && hasFdProduct) {
             consumerToSendUserEventForFD(document, userMailIsChanged);
         }
+
+        if (addOnAggregatesEnabled) {
+            consumerToAddOnAggregates(userInstitutionChanged);
+        }
     }
 
     private boolean isUserMailChanged(UserInstitution userInstitutionChanged) {
@@ -341,4 +378,139 @@ private TrackEventInput toTrackEventInputByUserInstitution(UserInstitution userI
                 .institutionId(userInstitution.getInstitutionId())
                 .build();
     }
+
+    public void consumerToAddOnAggregates(UserInstitution userInstitutionChanged) {
+        log.info("Starting consumerToAddOnAggregates on UserInstitution with id {}", userInstitutionChanged.getId());
+        userInstitutionChanged.getProducts().stream()
+                .filter(p -> p.getStatus() == OnboardedProductState.ACTIVE && Boolean.TRUE.equals(p.getToAddOnAggregates()))
+                .forEach(p -> {
+                    final String parentInstitutionId = userInstitutionChanged.getInstitutionId();
+                    final String userId = userInstitutionChanged.getUserId();
+                    final String userMailUuid = userInstitutionChanged.getUserMailUuid();
+                    getDelegations(parentInstitutionId, p.getProductId())
+                            .onItem().transformToUniAndConcatenate(d ->
+                                    createUserOnAggregate(d.getInstitutionId(), d.getInstitutionName(), d.getInstitutionType(), parentInstitutionId, userId, userMailUuid, p)
+                                            .onItem().transformToUni(r ->
+                                                    addUserToAggregatorGroup(d.getInstitutionId(), parentInstitutionId, p.getProductId(), userId)
+                                                            .onFailure().recoverWithNull()
+                                            )
+                                            .onFailure().recoverWithNull()
+                            )
+                            .subscribe().with(
+                                    r -> {},
+                                    t -> log.error("Error in consumerToAddOnAggregates on UserInstitution with id {}", userInstitutionChanged.getId(), t),
+                                    () -> log.info("Completed consumerToAddOnAggregates on UserInstitution with id {}", userInstitutionChanged.getId())
+                            );
+                });
+    }
+
+    private Multi<DelegationResponse> getDelegations(String parentInstitutionId, String productId) {
+        log.info("Getting delegations toAddOnAggregates of parentInstitutionId {} with productId {}", parentInstitutionId, productId);
+        return Multi.createFrom().emitter(emitter ->
+                fetchDelegationPages(parentInstitutionId, productId, 0, emitter));
+    }
+
+    private void fetchDelegationPages(String parentInstitutionId, String productId, int page,
+                                      MultiEmitter<? super DelegationResponse> emitter) {
+        delegationApi.getDelegationsUsingGET2(null, parentInstitutionId, productId, null, null, null, page, null)
+                .onFailure().retry().withBackOff(Duration.ofSeconds(retryMinBackOff), Duration.ofSeconds(retryMaxBackOff)).atMost(maxRetry)
+                .subscribe().with(r -> {
+                    r.getDelegations().stream()
+                            .filter(d -> DelegationResponse.StatusEnum.ACTIVE.equals(d.getStatus())
+                                    && DelegationResponse.TypeEnum.EA.equals(d.getType()))
+                            .forEach(d -> {
+                                log.info("Found ACTIVE EA delegation with id {} (from: {}, to: {}, product: {})", d.getId(), d.getInstitutionId(), d.getBrokerId(), d.getProductId());
+                                emitter.emit(d);
+                            });
+                    if (r.getPageInfo().getPageNo() < (r.getPageInfo().getTotalPages() - 1)) {
+                        final int nextPage = page + 1;
+                        log.debug("Switching to page {} to get more delegations for parentInstitutionId {} and productId {}", nextPage, parentInstitutionId, productId);
+                        fetchDelegationPages(parentInstitutionId, productId, nextPage, emitter);
+                    } else {
+                        log.debug("No more delegation pages for parentInstitutionId {} and productId {}", parentInstitutionId, productId);
+                        emitter.complete();
+                    }
+                }, t -> {
+                    log.error("Error fetching delegations page {} for parentInstitutionId {} and productId {}: {}", page, parentInstitutionId, productId, t.getMessage());
+                    emitter.fail(t);
+                });
+    }
+
+    private Uni<Response> createUserOnAggregate(String institutionId, String institutionDescription, String institutionType, String parentInstitutionId,
+                                                String userId, String userMailUuid, OnboardedProduct product) {
+        log.info("Creating user {} to institution {} and product {} from parentInstitutionId {}", userId, institutionId, product.getProductId(), parentInstitutionId);
+        final AddUserRoleDto addUserRoleDto = AddUserRoleDto.builder()
+                .institutionId(institutionId)
+                .institutionDescription(institutionDescription)
+                .product(Product.builder()
+                        .role(getRoleToPropagate(product.getRole()))
+                        .productId(product.getProductId())
+                        .productRoles(getProductRolesToPropagate(product.getRole(), product.getProductRole(), product.getProductId(), institutionType))
+                        .tokenId(product.getTokenId())
+                        .build())
+                .hasToSendEmail(false)
+                .userMailUuid(userMailUuid)
+                .build();
+        return userApi.createUserByUserId(userId, addUserRoleDto)
+                .onFailure().retry().withBackOff(Duration.ofSeconds(retryMinBackOff), Duration.ofSeconds(retryMaxBackOff)).atMost(maxRetry)
+                .onFailure().invoke(t -> log.error("Error creating user {} on aggregate {} for parentInstitutionId {}: {}", userId, institutionId, parentInstitutionId, t.getMessage()))
+                .onItem().invoke(r -> log.info("Created user on aggregate {} for parentInstitutionId {} - status {}", institutionId, parentInstitutionId, r.getStatus()));
+    }
+
+    /**
+     * Determines the role to assign to the new user on the aggregate.
+     * If the parent's role is OPERATOR, returns OPERATOR.
+     * For any other role (e.g., MANAGER, DELEGATE, etc.), returns ADMIN_EA.
+     *
+     * @param parentRole the role of the user on the aggregator
+     * @return the role to assign to the new user on the aggregate
+     */
+    private String getRoleToPropagate(PartyRole parentRole) {
+        return PartyRole.OPERATOR.equals(parentRole) ? PartyRole.OPERATOR.name() : PartyRole.ADMIN_EA.name();
+    }
+
+    /**
+     * Determines the productRole to assign to the new user on the aggregate.
+     * If the parentRole is OPERATOR, we simply propagate the same parentProductRole, otherwise
+     * we retrieve the productRole associated to the ADMIN_EA role for the product using the ProductService.
+     *
+     * @param parentRole the role of the user on the aggregator
+     * @param parentProductRole the productRole of the user on the aggregator
+     * @param productId the productId of the user on the aggregator
+     * @param targetInstitutionType the institutionType of the aggregate
+     * @return the productRole to assign to the new user on the aggregate
+     */
+    private List<String> getProductRolesToPropagate(PartyRole parentRole, String parentProductRole, String productId, String targetInstitutionType) {
+        if (PartyRole.OPERATOR.equals(parentRole)) {
+            return List.of(parentProductRole);
+        } else {
+            final it.pagopa.selfcare.product.entity.Product product = productService.getProduct(productId);
+            final Map<PartyRole, ProductRoleInfo> roleMappings = product.getRoleMappings(targetInstitutionType);
+            return Optional.ofNullable(roleMappings.get(PartyRole.ADMIN_EA))
+                    .map(pri -> pri.getRoles().stream().findFirst()
+                            .map(r -> List.of(r.getCode()))
+                            .orElse(List.of()))
+                    .orElse(List.of());
+        }
+    }
+
+    private Uni<Response> addUserToAggregatorGroup(String institutionId, String parentInstitutionId, String productId, String userId) {
+        if (!addOnAggregatesGroupProducts.contains(productId)) {
+            log.info("Skipping adding user {} to group of institution {}, parentInstitutionId {} and product {}", userId, institutionId, parentInstitutionId, productId);
+            return Uni.createFrom().item(Response.status(Response.Status.OK).build());
+        }
+
+        log.info("Adding user {} to group of institution {}, parentInstitutionId {} and product {}", userId, institutionId, parentInstitutionId, productId);
+        final AddMembersToUserGroupDto addMembersToUserGroupDto = AddMembersToUserGroupDto.builder()
+                .institutionId(institutionId)
+                .parentInstitutionId(parentInstitutionId)
+                .productId(productId)
+                .members(Set.of(UUID.fromString(userId)))
+                .build();
+        return userGroupApi.addMembersToUserGroupWithParentInstitutionIdUsingPUT(addMembersToUserGroupDto)
+                .onFailure().retry().withBackOff(Duration.ofSeconds(retryMinBackOff), Duration.ofSeconds(retryMaxBackOff)).atMost(maxRetry)
+                .onFailure().invoke(t -> log.error("Error adding user {} to group of institution {} with parentInstitutionId {}: {}", userId, institutionId, parentInstitutionId, t.getMessage()))
+                .onItem().invoke(r -> log.info("Added user {} on group of institution {} with parentInstitutionId {} - status {}", userId, institutionId, parentInstitutionId, r.getStatus()));
+    }
+
 }

apps/user-cdc/src/main/java/it/pagopa/selfcare/user/event/client/InternalApiHeadersFactory.java

@@ -0,0 +1,24 @@
+package it.pagopa.selfcare.user.event.client;
+
+import jakarta.inject.Inject;
+import jakarta.ws.rs.core.MultivaluedHashMap;
+import jakarta.ws.rs.core.MultivaluedMap;
+import org.eclipse.microprofile.config.inject.ConfigProperty;
+import org.eclipse.microprofile.rest.client.ext.ClientHeadersFactory;
+
+public class InternalApiHeadersFactory implements ClientHeadersFactory {
+
+    private static final String API_KEY_NAME = "Ocp-Apim-Subscription-Key";
+
+    @Inject
+    @ConfigProperty(name = "client.internal.api-key")
+    private String apiKey;
+
+    @Override
+    public MultivaluedMap<String, String> update(MultivaluedMap<String, String> multivaluedMap, MultivaluedMap<String, String> multivaluedMap1) {
+        MultivaluedMap<String, String> headers = new MultivaluedHashMap<>();
+        headers.putSingle(API_KEY_NAME, apiKey);
+        return headers;
+    }
+
+}

apps/user-cdc/src/main/java/it/pagopa/selfcare/user/event/client/InternalDelegationApiClient.java

@@ -0,0 +1,10 @@
+package it.pagopa.selfcare.user.event.client;
+
+import org.eclipse.microprofile.rest.client.annotation.RegisterClientHeaders;
+import org.eclipse.microprofile.rest.client.inject.RegisterRestClient;
+import org.openapi.quarkus.internal_json.api.DelegationApi;
+
+@RegisterRestClient(configKey = "client.internal.delegation-api")
+@RegisterClientHeaders(InternalApiHeadersFactory.class)
+public interface InternalDelegationApiClient extends DelegationApi {
+}

apps/user-cdc/src/main/java/it/pagopa/selfcare/user/event/client/InternalUserApiClient.java

@@ -0,0 +1,10 @@
+package it.pagopa.selfcare.user.event.client;
+
+import org.eclipse.microprofile.rest.client.annotation.RegisterClientHeaders;
+import org.eclipse.microprofile.rest.client.inject.RegisterRestClient;
+import org.openapi.quarkus.internal_json.api.UserApi;
+
+@RegisterRestClient(configKey = "client.internal.user-api")
+@RegisterClientHeaders(InternalApiHeadersFactory.class)
+public interface InternalUserApiClient extends UserApi {
+}

apps/user-cdc/src/main/java/it/pagopa/selfcare/user/event/client/InternalUserGroupApiClient.java

@@ -0,0 +1,10 @@
+package it.pagopa.selfcare.user.event.client;
+
+import org.eclipse.microprofile.rest.client.annotation.RegisterClientHeaders;
+import org.eclipse.microprofile.rest.client.inject.RegisterRestClient;
+import org.openapi.quarkus.internal_json.api.UserGroupApi;
+
+@RegisterRestClient(configKey = "client.internal.user-group-api")
+@RegisterClientHeaders(InternalApiHeadersFactory.class)
+public interface InternalUserGroupApiClient extends UserGroupApi {
+}

apps/user-cdc/src/main/java/it/pagopa/selfcare/user/event/config/UserCdcConfig.java

@@ -4,12 +4,25 @@
 import com.azure.data.tables.TableClientBuilder;
 import com.microsoft.applicationinsights.TelemetryClient;
 import com.microsoft.applicationinsights.TelemetryConfiguration;
+import it.pagopa.selfcare.azurestorage.AzureBlobClient;
+import it.pagopa.selfcare.azurestorage.AzureBlobClientDefault;
+import it.pagopa.selfcare.product.service.ProductService;
+import it.pagopa.selfcare.product.service.ProductServiceCacheable;
 import jakarta.enterprise.context.ApplicationScoped;
 import org.eclipse.microprofile.config.inject.ConfigProperty;
 
 @ApplicationScoped
 public class UserCdcConfig {
 
+    @ConfigProperty(name = "user-cdc.blob-storage.container-product")
+    String containerProduct;
+
+    @ConfigProperty(name = "user-cdc.blob-storage.filepath-product")
+    String filepathProduct;
+
+    @ConfigProperty(name = "user-cdc.storage.connection-string")
+    String connectionStringProduct;
+
     @ApplicationScoped
     public TelemetryClient telemetryClient(@ConfigProperty(name = "user-cdc.appinsights.connection-string") String appInsightsConnectionString) {
         TelemetryConfiguration telemetryConfiguration = TelemetryConfiguration.createDefault();
@@ -26,4 +39,14 @@ public TableClient tableClient(@ConfigProperty(name = "user-cdc.storage.connecti
                 .buildClient();
     }
 
+    @ApplicationScoped
+    public ProductService productService() {
+        AzureBlobClient azureBlobClient = new AzureBlobClientDefault(connectionStringProduct, containerProduct);
+        try{
+            return new ProductServiceCacheable(azureBlobClient, filepathProduct);
+        } catch(IllegalArgumentException e){
+            throw new IllegalArgumentException("Found an issue when trying to serialize product json string!!");
+        }
+    }
+
 }