Feature/add cdp to lega lentity saga

pom.xml

@@ -34,6 +34,7 @@
         <module>stream-contacts</module>
         <module>stream-loans</module>
         <module>stream-audiences</module>
+        <module>stream-cdp</module>
         <module>stream-compositions</module>
         <module>stream-plan-manager</module>
         <module>stream-customer-profile</module>

stream-cdp/README.md

@@ -0,0 +1,6 @@
+# Stream Audiences Integration
+The goal of this module is to enable ingestion of Customers into Retail Customers and Business Customers segments.
+
+The ingestion is done through an HTTP call towards `User Segments Collector` service.
+
+`UserKindSegmentationSaga` (responsible for triggering the ingestion towards the collector) is triggered from `LegalEntitySaga`

stream-cdp/cdp-core/pom.xml

@@ -0,0 +1,42 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <parent>
+        <groupId>com.backbase.stream</groupId>
+        <artifactId>stream-cdp</artifactId>
+        <version>9.5.0</version>
+    </parent>
+
+    <artifactId>cdp-core</artifactId>
+    <packaging>jar</packaging>
+    <name>Stream :: CDP Core</name>
+
+    <properties>
+        <checkstyle.disable.checks>true</checkstyle.disable.checks>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>com.backbase.stream</groupId>
+            <artifactId>stream-dbs-clients</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>com.backbase.stream</groupId>
+            <artifactId>stream-worker</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>io.projectreactor</groupId>
+            <artifactId>reactor-test</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>com.backbase.buildingblocks</groupId>
+            <artifactId>service-sdk-starter-test</artifactId>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+
+</project>

stream-cdp/cdp-core/src/main/java/com/backbase/stream/cdp/CdpSaga.java

@@ -0,0 +1,69 @@
+package com.backbase.stream.cdp;
+
+import com.backbase.cdp.ingestion.api.service.v1.CdpApi;
+import com.backbase.stream.configuration.CdpProperties;
+import com.backbase.stream.worker.StreamTaskExecutor;
+import com.backbase.stream.worker.exception.StreamTaskException;
+import lombok.extern.slf4j.Slf4j;
+import reactor.core.publisher.Mono;
+
+@Slf4j
+public class CdpSaga implements StreamTaskExecutor<CdpTask> {
+
+    public static final String ENTITY = "CdpProfile";
+    public static final String INGEST = "ingest";
+    public static final String SUCCESS = "success";
+    public static final String ERROR = "error";
+    public static final String INGESTED_SUCCESSFULLY = "Customer ingested successfully";
+    public static final String FAILED_TO_INGEST = "Failed to ingest Customer";
+
+    private final CdpApi cdpServiceApi;
+    private final CdpProperties cdpProperties;
+
+    public CdpSaga(
+        CdpApi cdpServiceApi,
+        CdpProperties cdpProperties
+    ) {
+        this.cdpServiceApi = cdpServiceApi;
+        this.cdpProperties = cdpProperties;
+    }
+
+    @Override
+    public Mono<CdpTask> executeTask(CdpTask streamTask) {
+
+        var request = streamTask.getCdpEvents();
+
+        return cdpServiceApi.ingestEvents(request)
+            .then(Mono.fromCallable(() -> {
+                streamTask.info(ENTITY, INGEST, SUCCESS, null,
+                    request.getCdpEvents().getFirst().getSourceId(), INGESTED_SUCCESSFULLY);
+                return streamTask;
+            }))
+            .onErrorResume(throwable -> {
+                streamTask.error(ENTITY, INGEST, ERROR, null,
+                    request.getCdpEvents().getFirst().getSourceId(), FAILED_TO_INGEST);
+                return Mono.error(new StreamTaskException(streamTask, throwable, FAILED_TO_INGEST));
+            });
+    }
+
+    @Override
+    public Mono<CdpTask> rollBack(CdpTask streamTask) {
+        return null;
+    }
+
+    public boolean isEnabled() {
+        if (cdpProperties == null) {
+            return false;
+        }
+
+        return cdpProperties.enabled();
+    }
+
+    public String getDefaultCustomerCategory() {
+        if (!isEnabled()) {
+            return null;
+        }
+
+        return cdpProperties.defaultCustomerCategory();
+    }
+}

stream-cdp/cdp-core/src/main/java/com/backbase/stream/cdp/CdpTask.java

@@ -0,0 +1,16 @@
+package com.backbase.stream.cdp;
+
+import com.backbase.cdp.ingestion.api.service.v1.model.CdpEvents;
+import com.backbase.stream.worker.model.StreamTask;
+import lombok.Data;
+
+@Data
+public class CdpTask extends StreamTask {
+
+    private CdpEvents cdpEvents;
+
+    @Override
+    public String getName() {
+        return "cdpProfilesIngestionTask";
+    }
+}

stream-cdp/cdp-core/src/main/java/com/backbase/stream/configuration/CdpConfiguration.java

@@ -0,0 +1,22 @@
+package com.backbase.stream.configuration;
+
+
+import com.backbase.cdp.ingestion.api.service.v1.CdpApi;
+import com.backbase.stream.cdp.CdpSaga;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+@EnableConfigurationProperties({CdpProperties.class})
+@Configuration
+public class CdpConfiguration {
+
+    @Bean
+    public CdpSaga cdpSaga(
+        CdpApi cdpServiceApi,
+        CdpProperties cdpProperties
+    ) {
+        return new CdpSaga(cdpServiceApi, cdpProperties);
+    }
+
+}

stream-cdp/cdp-core/src/main/java/com/backbase/stream/configuration/CdpProperties.java

@@ -0,0 +1,11 @@
+package com.backbase.stream.configuration;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+@ConfigurationProperties(prefix = "backbase.stream.cdp")
+public record CdpProperties(
+    boolean enabled,
+    String defaultCustomerCategory
+) {
+
+}

stream-cdp/cdp-core/src/main/java/com/backbase/stream/package-info.java

@@ -0,0 +1 @@
+package com.backbase.stream;

stream-cdp/cdp-core/src/test/java/com/backbase/stream/cdp/CdpSagaTest.java

@@ -0,0 +1,89 @@
+package com.backbase.stream.cdp;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import com.backbase.cdp.ingestion.api.service.v1.CdpApi;
+import com.backbase.cdp.ingestion.api.service.v1.model.CdpEvent;
+import com.backbase.cdp.ingestion.api.service.v1.model.CdpEvents;
+import com.backbase.stream.configuration.CdpProperties;
+import java.util.Map;
+import java.util.UUID;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import reactor.core.publisher.Mono;
+
+@ExtendWith(MockitoExtension.class)
+class CdpSagaTest {
+
+    @Mock
+    private CdpApi cdpServiceApi;
+
+    @Mock
+    private CdpProperties userKindSegmentationProperties;
+
+    @InjectMocks
+    private CdpSaga cdpSaga;
+
+    @Test
+    void testExecuteTask() {
+        var task = createTask();
+        when(cdpServiceApi.ingestEvents(any())).thenReturn(Mono.empty());
+
+        cdpSaga.executeTask(task).block();
+
+        verify(cdpServiceApi).ingestEvents(any());
+    }
+
+    @Test
+    void isDisabledByDefault() {
+        var saga = new CdpSaga(
+            cdpServiceApi,
+            null
+        );
+
+        assertFalse(saga.isEnabled());
+    }
+
+    @Test
+    void defaultCustomerCategoryIsNullWhenSagaIsDisabled() {
+        when(userKindSegmentationProperties.enabled()).thenReturn(false);
+
+        assertNull(cdpSaga.getDefaultCustomerCategory());
+    }
+
+    @Test
+    void rollbackReturnsNull() {
+        assertNull(cdpSaga.rollBack(createTask()));
+    }
+
+    @Test
+    void returnsDefaultCustomerCategoryFromProperties() {
+        when(userKindSegmentationProperties.enabled()).thenReturn(true);
+        when(userKindSegmentationProperties.defaultCustomerCategory()).thenReturn("RETAIL");
+
+        assertEquals("RETAIL", cdpSaga.getDefaultCustomerCategory());
+    }
+
+    private CdpTask createTask() {
+        var task = new CdpTask();
+        task.setCdpEvents(
+            new CdpEvents()
+                .addCdpEventsItem(new CdpEvent()
+                    .eventType("ProfileCreatedEvent")
+                    .eventId(UUID.randomUUID().toString())
+                    .sourceSystem("BACKBASE")
+                    .sourceType("USER_ID")
+                    .sourceId("internal-id")
+                    .data(Map.of()))
+        );
+        return task;
+    }
+}

stream-cdp/cdp-core/src/test/java/com/backbase/stream/configuration/CdpPropertiesTest.java

@@ -0,0 +1,17 @@
+package com.backbase.stream.configuration;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import org.junit.jupiter.api.Test;
+
+class CdpPropertiesTest {
+
+    @Test
+    void testProperties() {
+        CdpProperties properties = new CdpProperties(true, "testCategory");
+
+        assertTrue(properties.enabled());
+        assertEquals("testCategory", properties.defaultCustomerCategory());
+    }
+}

stream-cdp/pom.xml

@@ -0,0 +1,20 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <parent>
+        <groupId>com.backbase.stream</groupId>
+        <artifactId>stream-services</artifactId>
+        <version>9.5.0</version>
+    </parent>
+
+    <artifactId>stream-cdp</artifactId>
+
+    <packaging>pom</packaging>
+    <name>Stream :: CDP</name>
+
+    <modules>
+        <module>cdp-core</module>
+    </modules>
+
+</project>

stream-dbs-clients/pom.xml

@@ -745,6 +745,60 @@
                             </configOptions>
                         </configuration>
                     </execution>
+                    <execution>
+                        <id>boat-validation-cdp-spec</id>
+                        <goals>
+                            <goal>validate</goal>
+                        </goals>
+                        <phase>generate-sources</phase>
+                        <configuration>
+                            <input>${project.basedir}/src/main/openapi/cdp-ingestion-service-api-v1.0.0-beta.yaml</input>
+                            <failOnWarning>true</failOnWarning>
+                        </configuration>
+                    </execution>
+                    <execution>
+                        <id>generate-cdp-ingestion-service-api-code</id>
+                        <goals>
+                            <goal>generate-webclient-embedded</goal>
+                        </goals>
+                        <phase>generate-sources</phase>
+                        <configuration>
+                            <openapiNormalizer>REFACTOR_ALLOF_WITH_PROPERTIES_ONLY=true</openapiNormalizer>
+                            <inputSpec>${project.basedir}/src/main/openapi/cdp-ingestion-service-api-v1.0.0-beta.yaml</inputSpec>
+                            <apiPackage>com.backbase.cdp.ingestion.api.service.v1</apiPackage>
+                            <modelPackage>com.backbase.cdp.ingestion.api.service.v1.model</modelPackage>
+                            <configOptions>
+                                <useBeanValidation>false</useBeanValidation>
+                            </configOptions>
+                        </configuration>
+                    </execution>
+                    <execution>
+                        <id>boat-validation-cdp-profiles-spec</id>
+                        <goals>
+                            <goal>validate</goal>
+                        </goals>
+                        <phase>generate-sources</phase>
+                        <configuration>
+                            <input>${project.basedir}/src/main/openapi/cdp-profiles-service-api-v1.0.0-beta.yaml</input>
+                            <failOnWarning>true</failOnWarning>
+                        </configuration>
+                    </execution>
+                    <execution>
+                        <id>generate-cdp-profiles-service-api-code</id>
+                        <goals>
+                            <goal>generate-webclient-embedded</goal>
+                        </goals>
+                        <phase>generate-sources</phase>
+                        <configuration>
+                            <openapiNormalizer>REFACTOR_ALLOF_WITH_PROPERTIES_ONLY=true</openapiNormalizer>
+                            <inputSpec>${project.basedir}/src/main/openapi/cdp-profiles-service-api-v1.0.0-beta.yaml</inputSpec>
+                            <apiPackage>com.backbase.cdp.profiles.api.service.v1</apiPackage>
+                            <modelPackage>com.backbase.cdp.profiles.api.service.v1.model</modelPackage>
+                            <configOptions>
+                                <useBeanValidation>false</useBeanValidation>
+                            </configOptions>
+                        </configuration>
+                    </execution>
                     <execution>
                         <id>generate-plan-manager-service-api-code</id>
                         <goals>