TAR-944 : Upgrade BJS to update Investment Caboose Data

stream-investment/investment-core/src/main/java/com/backbase/stream/configuration/IngestConfigProperties.java

@@ -155,6 +155,9 @@ public static class AssetConfig {
         private int marketSpecialDayConcurrency = 5;
         private int assetCategoryConcurrency = 5;
         private int assetCategoryTypeConcurrency = 5;
+        private int assetUpsertConcurrency = 5;
+        private int priceConcurrency = 5;
+        private int intradayPriceConcurrency = 5;
 
     }
 

stream-investment/investment-core/src/main/java/com/backbase/stream/configuration/InvestmentCabooseClientConfig.java

@@ -0,0 +1,154 @@
+package com.backbase.stream.configuration;
+
+import com.backbase.investment.api.service.ApiClient;
+import com.backbase.investment.api.service.v1.AllocationsApi;
+import com.backbase.investment.api.service.v1.AssetUniverseApi;
+import com.backbase.investment.api.service.v1.AsyncBulkGroupsApi;
+import com.backbase.investment.api.service.v1.ClientApi;
+import com.backbase.investment.api.service.v1.CurrencyApi;
+import com.backbase.investment.api.service.v1.FinancialAdviceApi;
+import com.backbase.investment.api.service.v1.InvestmentApi;
+import com.backbase.investment.api.service.v1.InvestmentProductsApi;
+import com.backbase.investment.api.service.v1.PaymentsApi;
+import com.backbase.investment.api.service.v1.PortfolioApi;
+import com.backbase.investment.api.service.v1.PortfolioTradingAccountsApi;
+import com.backbase.investment.api.service.v1.RiskAssessmentApi;
+import com.backbase.stream.clients.config.CompositeApiClientConfig;
+import com.fasterxml.jackson.annotation.JsonInclude.Include;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializationFeature;
+import java.text.DateFormat;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.http.client.reactive.ReactorClientHttpConnector;
+import org.springframework.http.MediaType;
+import org.springframework.http.codec.json.Jackson2JsonDecoder;
+import org.springframework.http.codec.json.Jackson2JsonEncoder;
+import org.springframework.web.reactive.function.client.WebClient;
+import org.springframework.web.reactive.function.client.WebClient.Builder;
+import reactor.netty.http.client.HttpClient;
+
+/**
+ * HTTP client configuration for investment-caboose API communication.
+ *
+ * <p>Mirrors {@link InvestmentClientConfig} exactly but targets investment-caboose via
+ * {@code backbase.communication.services.investment-caboose.directUri}. Registers a dedicated
+ * {@code cabooseInvestmentApiClient} and all individual investment API beans with a
+ * {@code caboose} prefix so they can coexist alongside primary beans in dual-write mode.
+ *
+ * <p>This configuration is only loaded when
+ * {@code backbase.bootstrap.ingestions.investment.caboose.enabled=true}.
+ */
+@Configuration
+@ConditionalOnProperty(name = "backbase.bootstrap.ingestions.investment.caboose.enabled")
+@ConfigurationProperties("backbase.communication.services.investment-caboose")
+public class InvestmentCabooseClientConfig extends CompositeApiClientConfig {
+
+    public static final String INVESTMENT_CABOOSE_SERVICE_ID = "investment-caboose";
+
+    public InvestmentCabooseClientConfig() {
+        super(INVESTMENT_CABOOSE_SERVICE_ID);
+    }
+
+    /**
+     * Reactive {@link ApiClient} pointed at investment-caboose.
+     *
+     * <p>Uses a dedicated {@code cabooseHttpClient} so the caboose connection pool is
+     * completely independent of the primary Investment Service pool.
+     */
+    @Bean("cabooseInvestmentApiClient")
+    public ApiClient cabooseInvestmentApiClient(
+        WebClient interServiceWebClient,
+        @Qualifier("cabooseHttpClient") HttpClient cabooseHttpClient,
+        ObjectMapper objectMapper,
+        DateFormat dateFormat) {
+        ObjectMapper mapper = objectMapper.copy();
+        mapper.setSerializationInclusion(Include.NON_EMPTY);
+        mapper.disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS);
+        Builder mutate = interServiceWebClient.mutate()
+            .clientConnector(new ReactorClientHttpConnector(cabooseHttpClient));
+        mutate.codecs(clientCodecConfigurer -> {
+            Jackson2JsonEncoder encoder = new Jackson2JsonEncoder(mapper, MediaType.APPLICATION_JSON);
+            Jackson2JsonDecoder decoder = new Jackson2JsonDecoder(mapper, MediaType.APPLICATION_JSON);
+            clientCodecConfigurer.defaultCodecs().jackson2JsonEncoder(encoder);
+            clientCodecConfigurer.defaultCodecs().jackson2JsonDecoder(decoder);
+        });
+        return new ApiClient(mutate.build(), mapper, dateFormat)
+            .setBasePath(createBasePath());
+    }
+
+    @Bean("cabooseClientApi")
+    public ClientApi cabooseClientApi(
+        @Qualifier("cabooseInvestmentApiClient") ApiClient cabooseInvestmentApiClient) {
+        return new ClientApi(cabooseInvestmentApiClient);
+    }
+
+    @Bean("cabooseInvestmentProductsApi")
+    public InvestmentProductsApi cabooseInvestmentProductsApi(
+        @Qualifier("cabooseInvestmentApiClient") ApiClient cabooseInvestmentApiClient) {
+        return new InvestmentProductsApi(cabooseInvestmentApiClient);
+    }
+
+    @Bean("caboosePortfolioApi")
+    public PortfolioApi caboosePortfolioApi(
+        @Qualifier("cabooseInvestmentApiClient") ApiClient cabooseInvestmentApiClient) {
+        return new PortfolioApi(cabooseInvestmentApiClient);
+    }
+
+    @Bean("cabooseFinancialAdviceApi")
+    public FinancialAdviceApi cabooseFinancialAdviceApi(
+        @Qualifier("cabooseInvestmentApiClient") ApiClient cabooseInvestmentApiClient) {
+        return new FinancialAdviceApi(cabooseInvestmentApiClient);
+    }
+
+    @Bean("cabooseAssetUniverseApi")
+    public AssetUniverseApi cabooseAssetUniverseApi(
+        @Qualifier("cabooseInvestmentApiClient") ApiClient cabooseInvestmentApiClient) {
+        return new AssetUniverseApi(cabooseInvestmentApiClient);
+    }
+
+    @Bean("cabooseAllocationsApi")
+    public AllocationsApi cabooseAllocationsApi(
+        @Qualifier("cabooseInvestmentApiClient") ApiClient cabooseInvestmentApiClient) {
+        return new AllocationsApi(cabooseInvestmentApiClient);
+    }
+
+    @Bean("cabooseInvestmentApi")
+    public InvestmentApi cabooseInvestmentApi(
+        @Qualifier("cabooseInvestmentApiClient") ApiClient cabooseInvestmentApiClient) {
+        return new InvestmentApi(cabooseInvestmentApiClient);
+    }
+
+    @Bean("caboosePaymentsApi")
+    public PaymentsApi caboosePaymentsApi(
+        @Qualifier("cabooseInvestmentApiClient") ApiClient cabooseInvestmentApiClient) {
+        return new PaymentsApi(cabooseInvestmentApiClient);
+    }
+
+    @Bean("caboosePortfolioTradingAccountsApi")
+    public PortfolioTradingAccountsApi caboosePortfolioTradingAccountsApi(
+        @Qualifier("cabooseInvestmentApiClient") ApiClient cabooseInvestmentApiClient) {
+        return new PortfolioTradingAccountsApi(cabooseInvestmentApiClient);
+    }
+
+    @Bean("cabooseCurrencyApi")
+    public CurrencyApi cabooseCurrencyApi(
+        @Qualifier("cabooseInvestmentApiClient") ApiClient cabooseInvestmentApiClient) {
+        return new CurrencyApi(cabooseInvestmentApiClient);
+    }
+
+    @Bean("cabooseRiskAssessmentApi")
+    public RiskAssessmentApi cabooseRiskAssessmentApi(
+        @Qualifier("cabooseInvestmentApiClient") ApiClient cabooseInvestmentApiClient) {
+        return new RiskAssessmentApi(cabooseInvestmentApiClient);
+    }
+
+    @Bean("cabooseAsyncBulkGroupsApi")
+    public AsyncBulkGroupsApi cabooseAsyncBulkGroupsApi(
+        @Qualifier("cabooseInvestmentApiClient") ApiClient cabooseInvestmentApiClient) {
+        return new AsyncBulkGroupsApi(cabooseInvestmentApiClient);
+    }
+}

stream-investment/investment-core/src/main/java/com/backbase/stream/configuration/InvestmentCabooseIngestionProperties.java

@@ -0,0 +1,53 @@
+package com.backbase.stream.configuration;
+
+import lombok.Data;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+/**
+ * Feature-flag properties for investment-caboose ingestion.
+ *
+ * <p>Controls whether the Bootstrap Job writes investment data to investment-caboose and which
+ * stages are active for the caboose path. All flags are independent of their primary-service
+ * counterparts in {@link InvestmentIngestionConfigurationProperties} — disabling a stage on
+ * the primary path does <strong>not</strong> automatically disable it on the caboose path.
+ *
+ * <p>Example:
+ * <pre>
+ * backbase:
+ *   bootstrap:
+ *     ingestions:
+ *       investment:
+ *         caboose:
+ *           enabled: true
+ *           asset-universe-enabled: true
+ *           wealth-enabled: true
+ * </pre>
+ */
+@Data
+@ConfigurationProperties(prefix = "backbase.bootstrap.ingestions.investment.caboose")
+public class InvestmentCabooseIngestionProperties {
+
+    /**
+     * When {@code true}, investment data is also written to investment-caboose.
+     * If caboose is unavailable a warning is logged but the bootstrap job does not fail.
+     *
+     * <p>Default: {@code false}.
+     */
+    private boolean enabled = false;
+
+    /**
+     * Controls whether the caboose asset-universe stage (currencies, markets, assets, prices)
+     * is executed. Independent of the primary {@code assetUniverseEnabled} flag.
+     *
+     * <p>Default: {@code true}.
+     */
+    private boolean assetUniverseEnabled = true;
+
+    /**
+     * Controls whether the caboose wealth / portfolio stage (clients, portfolios, trading accounts)
+     * is executed. Independent of the primary {@code wealthEnabled} flag.
+     *
+     * <p>Default: {@code true}.
+     */
+    private boolean wealthEnabled = true;
+}

stream-investment/investment-core/src/main/java/com/backbase/stream/configuration/InvestmentCabooseRestServiceApiConfiguration.java

@@ -0,0 +1,111 @@
+package com.backbase.stream.configuration;
+
+import com.backbase.stream.investment.service.resttemplate.InvestmentRestAssetUniverseService;
+import com.backbase.stream.investment.service.resttemplate.InvestmentRestModelPortfolioService;
+import com.backbase.stream.investment.service.resttemplate.InvestmentRestProductPortfolioService;
+import com.fasterxml.jackson.annotation.JsonInclude.Include;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializationFeature;
+import jakarta.validation.constraints.Pattern;
+import lombok.Setter;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.http.converter.json.MappingJackson2HttpMessageConverter;
+import org.springframework.web.client.RestTemplate;
+
+/**
+ * Sync REST-template client configuration for investment-caboose.
+ *
+ * <p>Mirrors {@link InvestmentRestServiceApiConfiguration} but targets investment-caboose.
+ * The service ID is configurable via
+ * {@code backbase.investment.communication.integration.caboose.serviceId} and defaults to
+ * {@value #DEFAULT_SERVICE_ID}.
+ *
+ * <p>This configuration is only loaded when {@link InvestmentCabooseServiceConfiguration} is
+ * active (i.e. {@code backbase.bootstrap.ingestions.investment.caboose.enabled=true}).
+ *
+ * <p>All beans carry an explicit {@code cabooseRest} / {@code cabooseInvestmentRest} name prefix
+ * so they coexist with the primary {@link InvestmentRestServiceApiConfiguration} beans.
+ */
+@Setter
+@Configuration
+@ConditionalOnBean(InvestmentCabooseServiceConfiguration.class)
+@ConfigurationProperties(prefix = "backbase.investment.communication.integration.caboose")
+public class InvestmentCabooseRestServiceApiConfiguration {
+
+    static final String DEFAULT_SERVICE_ID = "investment-caboose";
+
+    private String serviceId = DEFAULT_SERVICE_ID;
+    private String serviceUrl = "";
+
+    @Value("${backbase.communication.http.default-scheme:http}")
+    @Pattern(regexp = "https?")
+    private String scheme;
+
+    /**
+     * Sync {@link com.backbase.investment.api.service.sync.ApiClient} pointed at investment-caboose.
+     *
+     * <p>Mirrors {@link InvestmentRestServiceApiConfiguration#restInvestmentApiClient} exactly:
+     * uses the shared {@code interServiceRestTemplate} directly (no copy) so that all auth
+     * configuration — interceptors, request factory, credentials — is naturally preserved.
+     */
+    @Bean("cabooseRestInvestmentApiClient")
+    public com.backbase.investment.api.service.sync.ApiClient cabooseRestInvestmentApiClient(
+        @Qualifier("interServiceRestTemplate") RestTemplate restTemplate,
+        @Qualifier("cabooseRestInvestmentObjectMapper") ObjectMapper cabooseRestInvestmentObjectMapper) {
+
+        MappingJackson2HttpMessageConverter converter = new MappingJackson2HttpMessageConverter();
+        converter.setObjectMapper(cabooseRestInvestmentObjectMapper);
+
+        restTemplate.getMessageConverters().removeIf(m -> m instanceof MappingJackson2HttpMessageConverter);
+        restTemplate.getMessageConverters().add(converter);
+
+        com.backbase.investment.api.service.sync.ApiClient apiClient =
+            new com.backbase.investment.api.service.sync.ApiClient(restTemplate);
+        apiClient.setBasePath(scheme + "://" + serviceId + serviceUrl);
+        return apiClient;
+    }
+
+    @Bean("cabooseRestInvestmentObjectMapper")
+    public ObjectMapper cabooseRestInvestmentObjectMapper(ObjectMapper legacyObjectMapper) {
+        ObjectMapper mapper = legacyObjectMapper.copy();
+        mapper.setSerializationInclusion(Include.NON_EMPTY);
+        mapper.disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS);
+        return mapper;
+    }
+
+    @Bean("cabooseRestAssetUniverseApi")
+    public com.backbase.investment.api.service.sync.v1.AssetUniverseApi cabooseRestAssetUniverseApi(
+        @Qualifier("cabooseRestInvestmentApiClient")
+        com.backbase.investment.api.service.sync.ApiClient cabooseRestInvestmentApiClient) {
+        return new com.backbase.investment.api.service.sync.v1.AssetUniverseApi(cabooseRestInvestmentApiClient);
+    }
+
+    @Bean("cabooseInvestmentRestModelPortfolioService")
+    public InvestmentRestModelPortfolioService cabooseInvestmentRestModelPortfolioService(
+        @Qualifier("cabooseRestInvestmentApiClient")
+        com.backbase.investment.api.service.sync.ApiClient cabooseRestInvestmentApiClient) {
+        return new InvestmentRestModelPortfolioService(cabooseRestInvestmentApiClient);
+    }
+
+    @Bean("cabooseInvestmentRestProductPortfolioService")
+    public InvestmentRestProductPortfolioService cabooseInvestmentRestProductPortfolioService(
+        @Qualifier("cabooseRestInvestmentApiClient")
+        com.backbase.investment.api.service.sync.ApiClient cabooseRestInvestmentApiClient,
+        IngestConfigProperties portfolioProperties) {
+        return new InvestmentRestProductPortfolioService(cabooseRestInvestmentApiClient, portfolioProperties);
+    }
+
+    @Bean("cabooseInvestmentRestAssetUniverseService")
+    public InvestmentRestAssetUniverseService cabooseInvestmentRestAssetUniverseService(
+        @Qualifier("cabooseRestInvestmentApiClient")
+        com.backbase.investment.api.service.sync.ApiClient cabooseRestInvestmentApiClient,
+        IngestConfigProperties portfolioProperties) {
+        return new InvestmentRestAssetUniverseService(cabooseRestInvestmentApiClient, portfolioProperties);
+    }
+}
+