diff --git a/stream-investment/investment-core/src/main/java/com/backbase/stream/configuration/IngestConfigProperties.java b/stream-investment/investment-core/src/main/java/com/backbase/stream/configuration/IngestConfigProperties.java index e142c48e8..b07c44259 100644 --- a/stream-investment/investment-core/src/main/java/com/backbase/stream/configuration/IngestConfigProperties.java +++ b/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; } diff --git a/stream-investment/investment-core/src/main/java/com/backbase/stream/configuration/InvestmentCabooseClientConfig.java b/stream-investment/investment-core/src/main/java/com/backbase/stream/configuration/InvestmentCabooseClientConfig.java new file mode 100644 index 000000000..d108d322e --- /dev/null +++ b/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. + * + *

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. + * + *

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. + * + *

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); + } +} diff --git a/stream-investment/investment-core/src/main/java/com/backbase/stream/configuration/InvestmentCabooseIngestionProperties.java b/stream-investment/investment-core/src/main/java/com/backbase/stream/configuration/InvestmentCabooseIngestionProperties.java new file mode 100644 index 000000000..215254b45 --- /dev/null +++ b/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. + * + *

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 not automatically disable it on the caboose path. + * + *

Example: + * 

+ * backbase:
+ *   bootstrap:
+ *     ingestions:
+ *       investment:
+ *         caboose:
+ *           enabled: true
+ *           asset-universe-enabled: true
+ *           wealth-enabled: true
+ *
+ */ +@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. + * + *

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. + * + *

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. + * + *

Default: {@code true}. + */ + private boolean wealthEnabled = true; +} diff --git a/stream-investment/investment-core/src/main/java/com/backbase/stream/configuration/InvestmentCabooseRestServiceApiConfiguration.java b/stream-investment/investment-core/src/main/java/com/backbase/stream/configuration/InvestmentCabooseRestServiceApiConfiguration.java new file mode 100644 index 000000000..185d7983b --- /dev/null +++ b/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. + * + *

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}. + * + *

This configuration is only loaded when {@link InvestmentCabooseServiceConfiguration} is + * active (i.e. {@code backbase.bootstrap.ingestions.investment.caboose.enabled=true}). + * + *

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. + * + *

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); + } +} + diff --git a/stream-investment/investment-core/src/main/java/com/backbase/stream/configuration/InvestmentCabooseServiceConfiguration.java b/stream-investment/investment-core/src/main/java/com/backbase/stream/configuration/InvestmentCabooseServiceConfiguration.java new file mode 100644 index 000000000..4385217a0 --- /dev/null +++ b/stream-investment/investment-core/src/main/java/com/backbase/stream/configuration/InvestmentCabooseServiceConfiguration.java @@ -0,0 +1,243 @@ +package com.backbase.stream.configuration; + +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.autoconfigure.DbsApiClientsAutoConfiguration; +import com.backbase.stream.investment.saga.InvestmentAssetUniverseSaga; +import com.backbase.stream.investment.saga.InvestmentSaga; +import com.backbase.stream.investment.service.AsyncTaskService; +import com.backbase.stream.investment.service.InvestmentAssetPriceService; +import com.backbase.stream.investment.service.InvestmentAssetUniverseService; +import com.backbase.stream.investment.service.InvestmentClientService; +import com.backbase.stream.investment.service.InvestmentCurrencyService; +import com.backbase.stream.investment.service.InvestmentIntradayAssetPriceService; +import com.backbase.stream.investment.service.InvestmentModelPortfolioService; +import com.backbase.stream.investment.service.InvestmentPortfolioAllocationService; +import com.backbase.stream.investment.service.InvestmentPortfolioProductService; +import com.backbase.stream.investment.service.InvestmentPortfolioService; +import com.backbase.stream.investment.service.InvestmentRiskAssessmentService; +import com.backbase.stream.investment.service.InvestmentRiskQuestionaryService; +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 lombok.RequiredArgsConstructor; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.Import; + +/** + * Spring configuration for investment-caboose ingestion — mirrors {@link InvestmentServiceConfiguration} + * in structure but targets investment-caboose via a dedicated API client. + * + *

Active only when {@code backbase.bootstrap.ingestions.investment.caboose.enabled=true}. + * + *

Can operate as a standalone path independently of + * {@link InvestmentServiceConfiguration}. When both are active the BJS dual-writes asset-universe and + * portfolio data to the primary Investment Service and investment-caboose. Content ingestion remains + * primary-only. + * + *

All beans are prefixed with {@code caboose} so they coexist alongside primary beans + * in dual-write mode without ambiguity. + * + *

Sync REST-template services are provided by {@link InvestmentCabooseRestServiceApiConfiguration}. + */ +@RequiredArgsConstructor +@Configuration +@Import({ + DbsApiClientsAutoConfiguration.class, + InvestmentCabooseClientConfig.class, + InvestmentCabooseWebClientConfiguration.class, + InvestmentCabooseRestServiceApiConfiguration.class +}) +@ConditionalOnProperty( + name = "backbase.bootstrap.ingestions.investment.caboose.enabled", + havingValue = "true" +) +@EnableConfigurationProperties({ + InvestmentCabooseIngestionProperties.class, + InvestmentIngestionConfigurationProperties.class, + IngestConfigProperties.class +}) +public class InvestmentCabooseServiceConfiguration { + + + @Bean("cabooseInvestmentClientService") + public InvestmentClientService cabooseInvestmentClientService( + @Qualifier("cabooseClientApi") ClientApi cabooseClientApi) { + return new InvestmentClientService(cabooseClientApi); + } + + @Bean("cabooseInvestmentModelPortfolioService") + public InvestmentModelPortfolioService cabooseInvestmentModelPortfolioService( + @Qualifier("cabooseFinancialAdviceApi") FinancialAdviceApi cabooseFinancialAdviceApi, + @Qualifier("cabooseInvestmentRestModelPortfolioService") + InvestmentRestModelPortfolioService cabooseInvestmentRestModelPortfolioService, + IngestConfigProperties portfolioProperties) { + return new InvestmentModelPortfolioService(cabooseFinancialAdviceApi, + cabooseInvestmentRestModelPortfolioService, portfolioProperties); + } + + @Bean("cabooseInvestmentPortfolioProductService") + public InvestmentPortfolioProductService cabooseInvestmentPortfolioProductService( + @Qualifier("cabooseInvestmentProductsApi") InvestmentProductsApi cabooseInvestmentProductsApi, + IngestConfigProperties portfolioProperties, + @Qualifier("cabooseInvestmentModelPortfolioService") + InvestmentModelPortfolioService cabooseInvestmentModelPortfolioService, + @Qualifier("cabooseInvestmentRestProductPortfolioService") + InvestmentRestProductPortfolioService cabooseInvestmentRestProductPortfolioService) { + return new InvestmentPortfolioProductService(cabooseInvestmentProductsApi, portfolioProperties, + cabooseInvestmentModelPortfolioService, cabooseInvestmentRestProductPortfolioService); + } + + @Bean("cabooseInvestmentPortfolioService") + public InvestmentPortfolioService cabooseInvestmentPortfolioService( + @Qualifier("caboosePortfolioApi") PortfolioApi caboosePortfolioApi, + @Qualifier("caboosePaymentsApi") PaymentsApi caboosePaymentsApi, + @Qualifier("caboosePortfolioTradingAccountsApi") + PortfolioTradingAccountsApi caboosePortfolioTradingAccountsApi, + IngestConfigProperties portfolioProperties) { + return new InvestmentPortfolioService(caboosePortfolioApi, caboosePaymentsApi, + caboosePortfolioTradingAccountsApi, portfolioProperties); + } + + @Bean("cabooseInvestmentAssetUniverseService") + public InvestmentAssetUniverseService cabooseInvestmentAssetUniverseService( + @Qualifier("cabooseAssetUniverseApi") AssetUniverseApi cabooseAssetUniverseApi, + @Qualifier("cabooseInvestmentRestAssetUniverseService") + InvestmentRestAssetUniverseService cabooseInvestmentRestAssetUniverseService, + IngestConfigProperties portfolioProperties) { + return new InvestmentAssetUniverseService(cabooseAssetUniverseApi, + cabooseInvestmentRestAssetUniverseService, portfolioProperties); + } + + @Bean("cabooseAsyncTaskService") + public AsyncTaskService cabooseAsyncTaskService( + @Qualifier("cabooseAsyncBulkGroupsApi") AsyncBulkGroupsApi cabooseAsyncBulkGroupsApi) { + return new AsyncTaskService(cabooseAsyncBulkGroupsApi); + } + + @Bean("cabooseInvestmentAssetPriceService") + public InvestmentAssetPriceService cabooseInvestmentAssetPriceService( + @Qualifier("cabooseAssetUniverseApi") AssetUniverseApi cabooseAssetUniverseApi, + IngestConfigProperties portfolioProperties) { + return new InvestmentAssetPriceService(cabooseAssetUniverseApi, portfolioProperties); + } + + @Bean("cabooseInvestmentIntradayAssetPriceService") + public InvestmentIntradayAssetPriceService cabooseInvestmentIntradayAssetPriceService( + @Qualifier("cabooseAssetUniverseApi") AssetUniverseApi cabooseAssetUniverseApi, + IngestConfigProperties portfolioProperties) { + return new InvestmentIntradayAssetPriceService(cabooseAssetUniverseApi, portfolioProperties); + } + + @Bean("cabooseInvestmentPortfolioAllocationService") + public InvestmentPortfolioAllocationService cabooseInvestmentPortfolioAllocationService( + @Qualifier("cabooseAllocationsApi") AllocationsApi cabooseAllocationsApi, + @Qualifier("cabooseAssetUniverseApi") AssetUniverseApi cabooseAssetUniverseApi, + @Qualifier("cabooseInvestmentApi") InvestmentApi cabooseInvestmentApi, + IngestConfigProperties portfolioProperties) { + return new InvestmentPortfolioAllocationService(cabooseAllocationsApi, cabooseAssetUniverseApi, + cabooseInvestmentApi, portfolioProperties); + } + + @Bean("cabooseInvestmentCurrencyService") + public InvestmentCurrencyService cabooseInvestmentCurrencyService( + @Qualifier("cabooseCurrencyApi") CurrencyApi cabooseCurrencyApi) { + return new InvestmentCurrencyService(cabooseCurrencyApi); + } + + @Bean("cabooseInvestmentRiskAssessmentService") + public InvestmentRiskAssessmentService cabooseInvestmentRiskAssessmentService( + @Qualifier("cabooseRiskAssessmentApi") RiskAssessmentApi cabooseRiskAssessmentApi) { + return new InvestmentRiskAssessmentService(cabooseRiskAssessmentApi); + } + + @Bean("cabooseInvestmentRiskQuestionaryService") + public InvestmentRiskQuestionaryService cabooseInvestmentRiskQuestionaryService( + @Qualifier("cabooseRiskAssessmentApi") RiskAssessmentApi cabooseRiskAssessmentApi, + IngestConfigProperties portfolioProperties) { + return new InvestmentRiskQuestionaryService(cabooseRiskAssessmentApi, portfolioProperties); + } + + @Bean("cabooseInvestmentSaga") + public InvestmentSaga cabooseInvestmentSaga( + @Qualifier("cabooseInvestmentClientService") InvestmentClientService cabooseInvestmentClientService, + @Qualifier("cabooseInvestmentRiskAssessmentService") + InvestmentRiskAssessmentService cabooseInvestmentRiskAssessmentService, + @Qualifier("cabooseInvestmentRiskQuestionaryService") + InvestmentRiskQuestionaryService cabooseInvestmentRiskQuestionaryService, + @Qualifier("cabooseInvestmentPortfolioService") + InvestmentPortfolioService cabooseInvestmentPortfolioService, + @Qualifier("cabooseInvestmentModelPortfolioService") + InvestmentModelPortfolioService cabooseInvestmentModelPortfolioService, + @Qualifier("cabooseInvestmentPortfolioProductService") + InvestmentPortfolioProductService cabooseInvestmentPortfolioProductService, + @Qualifier("cabooseInvestmentPortfolioAllocationService") + InvestmentPortfolioAllocationService cabooseInvestmentPortfolioAllocationService, + @Qualifier("cabooseAsyncTaskService") AsyncTaskService cabooseAsyncTaskService, + InvestmentCabooseIngestionProperties cabooseIngestionProperties, + @Qualifier("cabooseAssetUniverseApi") AssetUniverseApi cabooseAssetUniverseApi) { + return new InvestmentSaga(cabooseInvestmentClientService, cabooseInvestmentRiskAssessmentService, + cabooseInvestmentRiskQuestionaryService, cabooseInvestmentPortfolioService, + cabooseInvestmentPortfolioAllocationService, cabooseInvestmentModelPortfolioService, + cabooseInvestmentPortfolioProductService, cabooseAsyncTaskService, + buildCabooseIngestionConfigurationProperties(cabooseIngestionProperties), cabooseAssetUniverseApi); + } + + @Bean("cabooseInvestmentAssetUniverseSaga") + public InvestmentAssetUniverseSaga cabooseInvestmentAssetUniverseSaga( + @Qualifier("cabooseInvestmentAssetUniverseService") + InvestmentAssetUniverseService cabooseInvestmentAssetUniverseService, + @Qualifier("cabooseInvestmentAssetPriceService") + InvestmentAssetPriceService cabooseInvestmentAssetPriceService, + @Qualifier("cabooseInvestmentIntradayAssetPriceService") + InvestmentIntradayAssetPriceService cabooseInvestmentIntradayAssetPriceService, + @Qualifier("cabooseInvestmentCurrencyService") + InvestmentCurrencyService cabooseInvestmentCurrencyService, + @Qualifier("cabooseAsyncTaskService") AsyncTaskService cabooseAsyncTaskService, + InvestmentCabooseIngestionProperties cabooseIngestionProperties, + IngestConfigProperties portfolioProperties) { + return new InvestmentAssetUniverseSaga(cabooseInvestmentAssetUniverseService, + cabooseInvestmentAssetPriceService, cabooseInvestmentIntradayAssetPriceService, + cabooseInvestmentCurrencyService, cabooseAsyncTaskService, + buildCabooseIngestionConfigurationProperties(cabooseIngestionProperties), portfolioProperties); + } + + /** + * Builds a plain {@link InvestmentIngestionConfigurationProperties} instance populated + * from the caboose-specific feature flags. + * + *

This object is not registered as a Spring bean — it is created + * inline inside each saga factory method. This intentionally avoids having two beans of + * type {@link InvestmentIngestionConfigurationProperties} in the Spring context (which + * would cause autowiring ambiguity in dual-write mode). + * + *

The saga's internal guard ({@code coreConfigurationProperties.isAssetUniverseEnabled()}, + * etc.) will therefore read from + * {@code backbase.bootstrap.ingestions.investment.caboose.*} — completely independent of + * the primary {@code backbase.bootstrap.ingestions.investment.*} flags. + * + * @param cabooseIngestionProperties bound caboose properties + * @return a fresh properties instance carrying only the caboose flags + */ + private static InvestmentIngestionConfigurationProperties buildCabooseIngestionConfigurationProperties( + InvestmentCabooseIngestionProperties cabooseIngestionProperties) { + InvestmentIngestionConfigurationProperties props = new InvestmentIngestionConfigurationProperties(); + props.setAssetUniverseEnabled(cabooseIngestionProperties.isAssetUniverseEnabled()); + props.setWealthEnabled(cabooseIngestionProperties.isWealthEnabled()); + return props; + } +} diff --git a/stream-investment/investment-core/src/main/java/com/backbase/stream/configuration/InvestmentCabooseWebClientConfiguration.java b/stream-investment/investment-core/src/main/java/com/backbase/stream/configuration/InvestmentCabooseWebClientConfiguration.java new file mode 100644 index 000000000..e31453dd1 --- /dev/null +++ b/stream-investment/investment-core/src/main/java/com/backbase/stream/configuration/InvestmentCabooseWebClientConfiguration.java @@ -0,0 +1,82 @@ +package com.backbase.stream.configuration; + +import io.netty.channel.ChannelOption; +import io.netty.handler.timeout.ReadTimeoutHandler; +import io.netty.handler.timeout.WriteTimeoutHandler; +import java.time.Duration; +import java.util.concurrent.TimeUnit; +import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import reactor.netty.http.client.HttpClient; +import reactor.netty.resources.ConnectionProvider; + +/** + * WebClient configuration for investment-caboose service communication. + * + *

Mirrors {@link InvestmentWebClientConfiguration} but produces a dedicated, separately named + * connection pool and HTTP client for investment-caboose so that both targets (Investment Service + * and investment-caboose) have independent resource budgets and do not interfere with each other. + * + *

All tunable values are externalized via {@link InvestmentCabooseWebClientProperties} and can + * be overridden through {@code application.yml} without recompiling. + * + *

All beans are explicitly named ({@code cabooseConnectionProvider} and + * {@code cabooseHttpClient}) so they never clash with the primary investment beans. + */ +@Slf4j +@Configuration +@EnableConfigurationProperties(InvestmentCabooseWebClientProperties.class) +public class InvestmentCabooseWebClientConfiguration { + + /** + * Dedicated {@link ConnectionProvider} for the investment-caboose client pool. + * + * @param props investment-caboose HTTP-client configuration properties + * @return caboose-specific ConnectionProvider + */ + @Bean("cabooseConnectionProvider") + public ConnectionProvider cabooseConnectionProvider(InvestmentCabooseWebClientProperties props) { + ConnectionProvider provider = ConnectionProvider.builder("investment-caboose-client-pool") + .maxConnections(props.getMaxConnections()) + .maxIdleTime(Duration.ofMinutes(props.getMaxIdleTimeMinutes())) + .maxLifeTime(Duration.ofMinutes(props.getMaxLifeTimeMinutes())) + .pendingAcquireMaxCount(props.getMaxPendingAcquires()) + .pendingAcquireTimeout(Duration.ofMillis(props.getPendingAcquireTimeoutMillis())) + .evictInBackground(Duration.ofSeconds(props.getEvictInBackgroundSeconds())) + .build(); + + log.info("Configured investment-caboose ConnectionProvider: maxConnections={}, maxIdleTime={}min," + + " pendingAcquireMaxCount={}, pendingAcquireTimeout={}ms", + props.getMaxConnections(), props.getMaxIdleTimeMinutes(), props.getMaxPendingAcquires(), + props.getPendingAcquireTimeoutMillis()); + + return provider; + } + + /** + * Creates an {@link HttpClient} backed by the caboose-specific connection pool and with + * explicit connect / read / write timeouts. + * + * @param connectionProvider the caboose-specific connection provider + * @param props investment-caboose HTTP-client configuration properties + * @return configured HttpClient targeting investment-caboose + */ + @Bean("cabooseHttpClient") + public HttpClient cabooseHttpClient( + @Qualifier("cabooseConnectionProvider") ConnectionProvider connectionProvider, + InvestmentCabooseWebClientProperties props) { + return HttpClient.create(connectionProvider) + .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, props.getConnectTimeoutSeconds() * 1000) + .option(ChannelOption.SO_KEEPALIVE, true) + .option(ChannelOption.TCP_NODELAY, true) + .responseTimeout(Duration.ofSeconds(props.getReadTimeoutSeconds())) + .doOnConnected(connection -> connection + .addHandlerLast(new ReadTimeoutHandler(props.getReadTimeoutSeconds(), TimeUnit.SECONDS)) + .addHandlerLast(new WriteTimeoutHandler(props.getWriteTimeoutSeconds(), TimeUnit.SECONDS))); + } + +} + diff --git a/stream-investment/investment-core/src/main/java/com/backbase/stream/configuration/InvestmentCabooseWebClientProperties.java b/stream-investment/investment-core/src/main/java/com/backbase/stream/configuration/InvestmentCabooseWebClientProperties.java new file mode 100644 index 000000000..300a14d44 --- /dev/null +++ b/stream-investment/investment-core/src/main/java/com/backbase/stream/configuration/InvestmentCabooseWebClientProperties.java @@ -0,0 +1,80 @@ +package com.backbase.stream.configuration; + +import lombok.Data; +import org.springframework.boot.context.properties.ConfigurationProperties; + +/** + * Configuration properties for the investment-caboose HTTP client connection pool and timeouts. + * + *

All values can be overridden via {@code application.yml} / {@code application.properties} + * using the prefix {@code backbase.communication.services.investment-caboose.http-client}. + * + *

Example: + * 

+ * backbase:
+ *   communication:
+ *     services:
+ *       investment-caboose:
+ *         http-client:
+ *           max-connections: 20
+ *           max-idle-time-minutes: 5
+ *           max-pending-acquires: 100
+ *           pending-acquire-timeout-millis: 45000
+ *           connect-timeout-seconds: 10
+ *           read-timeout-seconds: 30
+ *           write-timeout-seconds: 30
+ *
+ */ +@Data +@ConfigurationProperties(prefix = "backbase.communication.services.investment-caboose.http-client") +public class InvestmentCabooseWebClientProperties { + + /** + * Maximum number of open TCP connections to investment-caboose. + * Limiting this prevents the service from being overwhelmed (which causes 503 responses). + */ + private int maxConnections = 50; + + /** + * Maximum time (in minutes) that a connection can remain idle in the pool before being evicted. + */ + private long maxIdleTimeMinutes = 5; + + /** + * Maximum lifetime (in minutes) of a pooled connection regardless of activity. + */ + private long maxLifeTimeMinutes = 30; + + /** + * Maximum number of requests that can wait for a free connection ({@code -1} = no queue limit). + * + * @see InvestmentWebClientProperties#getMaxPendingAcquires() + */ + private int maxPendingAcquires = -1; + + /** + * Maximum time (in milliseconds) a request will wait to acquire a connection from the pool. + */ + private long pendingAcquireTimeoutMillis = 150_000; + + /** + * Background eviction interval (in seconds) for idle/expired connections in the pool. + */ + private long evictInBackgroundSeconds = 120; + + /** + * TCP connection timeout in seconds. Prevents hanging on unresponsive servers. + */ + private int connectTimeoutSeconds = 10; + + /** + * Read timeout in seconds. Prevents indefinite hangs waiting for a response. + */ + private int readTimeoutSeconds = 30; + + /** + * Write timeout in seconds. Prevents indefinite hangs while sending a request. + */ + private int writeTimeoutSeconds = 30; +} + diff --git a/stream-investment/investment-core/src/main/java/com/backbase/stream/configuration/InvestmentClientConfig.java b/stream-investment/investment-core/src/main/java/com/backbase/stream/configuration/InvestmentClientConfig.java index 019fa5bdd..1d0ff6a21 100644 --- a/stream-investment/investment-core/src/main/java/com/backbase/stream/configuration/InvestmentClientConfig.java +++ b/stream-investment/investment-core/src/main/java/com/backbase/stream/configuration/InvestmentClientConfig.java @@ -21,10 +21,10 @@ import java.text.DateFormat; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; -import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; 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; @@ -60,8 +60,7 @@ public InvestmentClientConfig() { /** * Configuration for Investment service REST client (ClientApi). */ - @Bean - @ConditionalOnMissingBean + @Bean("investmentApiClient") public ApiClient investmentApiClient(WebClient interServiceWebClient, @Qualifier("investmentHttpClient") HttpClient investmentHttpClient, ObjectMapper objectMapper, @@ -69,7 +68,8 @@ public ApiClient investmentApiClient(WebClient interServiceWebClient, ObjectMapper mapper = objectMapper.copy(); mapper.setSerializationInclusion(Include.NON_EMPTY); mapper.disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS); - Builder mutate = interServiceWebClient.mutate(); + Builder mutate = interServiceWebClient.mutate() + .clientConnector(new ReactorClientHttpConnector(investmentHttpClient)); mutate.codecs(clientCodecConfigurer -> { Jackson2JsonEncoder encoder = new Jackson2JsonEncoder(mapper, MediaType.APPLICATION_JSON); Jackson2JsonDecoder decoder = new Jackson2JsonDecoder(mapper, MediaType.APPLICATION_JSON); @@ -81,80 +81,69 @@ public ApiClient investmentApiClient(WebClient interServiceWebClient, } @Bean - @ConditionalOnMissingBean - public ClientApi clientApi(ApiClient investmentApiClient) { + public ClientApi clientApi(@Qualifier("investmentApiClient") ApiClient investmentApiClient) { return new ClientApi(investmentApiClient); } @Bean - @ConditionalOnMissingBean - public InvestmentProductsApi investmentProductsApi(ApiClient investmentApiClient) { + public InvestmentProductsApi investmentProductsApi( + @Qualifier("investmentApiClient") ApiClient investmentApiClient) { return new InvestmentProductsApi(investmentApiClient); } @Bean - @ConditionalOnMissingBean - public PortfolioApi portfolioApi(ApiClient investmentApiClient) { + public PortfolioApi portfolioApi(@Qualifier("investmentApiClient") ApiClient investmentApiClient) { return new PortfolioApi(investmentApiClient); } @Bean - @ConditionalOnMissingBean - public FinancialAdviceApi financialAdviceApi(ApiClient investmentApiClient) { + public FinancialAdviceApi financialAdviceApi(@Qualifier("investmentApiClient") ApiClient investmentApiClient) { return new FinancialAdviceApi(investmentApiClient); } @Bean - @ConditionalOnMissingBean - public AssetUniverseApi assetUniverseApi(ApiClient investmentApiClient) { + public AssetUniverseApi assetUniverseApi(@Qualifier("investmentApiClient") ApiClient investmentApiClient) { return new AssetUniverseApi(investmentApiClient); } @Bean - @ConditionalOnMissingBean - public AllocationsApi allocationsApi(ApiClient investmentApiClient) { + public AllocationsApi allocationsApi(@Qualifier("investmentApiClient") ApiClient investmentApiClient) { return new AllocationsApi(investmentApiClient); } @Bean - @ConditionalOnMissingBean - public InvestmentApi investmentApi(ApiClient investmentApiClient) { + public InvestmentApi investmentApi(@Qualifier("investmentApiClient") ApiClient investmentApiClient) { return new InvestmentApi(investmentApiClient); } @Bean - @ConditionalOnMissingBean - public ContentApi contentApi(ApiClient investmentApiClient) { + public ContentApi contentApi(@Qualifier("investmentApiClient") ApiClient investmentApiClient) { return new ContentApi(investmentApiClient); } @Bean - @ConditionalOnMissingBean - public PaymentsApi paymentsApi(ApiClient investmentApiClient) { + public PaymentsApi paymentsApi(@Qualifier("investmentApiClient") ApiClient investmentApiClient) { return new PaymentsApi(investmentApiClient); } @Bean - @ConditionalOnMissingBean - public PortfolioTradingAccountsApi portfolioTradingAccountsApi(ApiClient investmentApiClient) { + public PortfolioTradingAccountsApi portfolioTradingAccountsApi( + @Qualifier("investmentApiClient") ApiClient investmentApiClient) { return new PortfolioTradingAccountsApi(investmentApiClient); } @Bean - @ConditionalOnMissingBean - public CurrencyApi currencyApi(ApiClient investmentApiClient) { + public CurrencyApi currencyApi(@Qualifier("investmentApiClient") ApiClient investmentApiClient) { return new CurrencyApi(investmentApiClient); } @Bean - @ConditionalOnMissingBean - public RiskAssessmentApi riskAssessmentApi(ApiClient investmentApiClient) { + public RiskAssessmentApi riskAssessmentApi(@Qualifier("investmentApiClient") ApiClient investmentApiClient) { return new RiskAssessmentApi(investmentApiClient); } @Bean - @ConditionalOnMissingBean - public AsyncBulkGroupsApi asyncBulkGroupsApi(ApiClient investmentApiClient) { + public AsyncBulkGroupsApi asyncBulkGroupsApi(@Qualifier("investmentApiClient") ApiClient investmentApiClient) { return new AsyncBulkGroupsApi(investmentApiClient); } diff --git a/stream-investment/investment-core/src/main/java/com/backbase/stream/configuration/InvestmentIngestionConfigurationProperties.java b/stream-investment/investment-core/src/main/java/com/backbase/stream/configuration/InvestmentIngestionConfigurationProperties.java index 49e75c73a..6281b9d37 100644 --- a/stream-investment/investment-core/src/main/java/com/backbase/stream/configuration/InvestmentIngestionConfigurationProperties.java +++ b/stream-investment/investment-core/src/main/java/com/backbase/stream/configuration/InvestmentIngestionConfigurationProperties.java @@ -2,7 +2,6 @@ package com.backbase.stream.configuration; import lombok.Data; -import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; import org.springframework.boot.context.properties.ConfigurationProperties; /** @@ -13,7 +12,6 @@ * {@link IngestConfigProperties}. */ @Data -@ConditionalOnBean(InvestmentServiceConfiguration.class) @ConfigurationProperties(prefix = "backbase.bootstrap.ingestions.investment") public class InvestmentIngestionConfigurationProperties { diff --git a/stream-investment/investment-core/src/main/java/com/backbase/stream/configuration/InvestmentRestServiceApiConfiguration.java b/stream-investment/investment-core/src/main/java/com/backbase/stream/configuration/InvestmentRestServiceApiConfiguration.java index cac0326d1..1dd81ad7e 100644 --- a/stream-investment/investment-core/src/main/java/com/backbase/stream/configuration/InvestmentRestServiceApiConfiguration.java +++ b/stream-investment/investment-core/src/main/java/com/backbase/stream/configuration/InvestmentRestServiceApiConfiguration.java @@ -14,10 +14,10 @@ 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.autoconfigure.condition.ConditionalOnMissingBean; import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.Primary; import org.springframework.http.converter.json.MappingJackson2HttpMessageConverter; import org.springframework.web.client.RestTemplate; @@ -37,8 +37,7 @@ public class InvestmentRestServiceApiConfiguration { /** * Configuration for Investment service REST client (ClientApi). */ - @Bean - @ConditionalOnMissingBean + @Bean("restInvestmentApiClient") public com.backbase.investment.api.service.sync.ApiClient restInvestmentApiClient( @Qualifier("interServiceRestTemplate") RestTemplate restTemplate, @Qualifier("restInvestmentObjectMapper") ObjectMapper restInvestmentObjectMapper) { @@ -55,8 +54,7 @@ public com.backbase.investment.api.service.sync.ApiClient restInvestmentApiClien return apiClient; } - @Bean - @Qualifier("restInvestmentObjectMapper") + @Bean("restInvestmentObjectMapper") public ObjectMapper restInvestmentObjectMapper(ObjectMapper legacyObjectMapper) { ObjectMapper mapper = legacyObjectMapper.copy(); mapper.setSerializationInclusion(Include.NON_EMPTY); @@ -64,47 +62,59 @@ public ObjectMapper restInvestmentObjectMapper(ObjectMapper legacyObjectMapper) return mapper; } - @Bean - @ConditionalOnMissingBean + @Bean("restContentApi") public com.backbase.investment.api.service.sync.v1.ContentApi restContentApi( + @Qualifier("restInvestmentApiClient") com.backbase.investment.api.service.sync.ApiClient restInvestmentApiClient) { return new com.backbase.investment.api.service.sync.v1.ContentApi(restInvestmentApiClient); } - @Bean - @ConditionalOnMissingBean + @Bean("restAssetUniverseApi") public com.backbase.investment.api.service.sync.v1.AssetUniverseApi restAssetUniverseApi( + @Qualifier("restInvestmentApiClient") com.backbase.investment.api.service.sync.ApiClient restInvestmentApiClient) { return new com.backbase.investment.api.service.sync.v1.AssetUniverseApi(restInvestmentApiClient); } @Bean - public InvestmentRestNewsContentService investmentNewsContentService(ContentApi restContentApi, + @Primary + public InvestmentRestNewsContentService investmentNewsContentService( + @Qualifier("restContentApi") ContentApi restContentApi, + @Qualifier("restInvestmentApiClient") com.backbase.investment.api.service.sync.ApiClient restInvestmentApiClient) { return new InvestmentRestNewsContentService(restContentApi, restInvestmentApiClient); } @Bean - public InvestmentRestDocumentContentService investmentRestContentDocumentService(ContentApi restContentApi, + @Primary + public InvestmentRestDocumentContentService investmentRestContentDocumentService( + @Qualifier("restContentApi") ContentApi restContentApi, + @Qualifier("restInvestmentApiClient") com.backbase.investment.api.service.sync.ApiClient restInvestmentApiClient) { return new InvestmentRestDocumentContentService(restContentApi, restInvestmentApiClient); } @Bean + @Primary public InvestmentRestAssetUniverseService investmentRestAssetUniverseService( + @Qualifier("restInvestmentApiClient") com.backbase.investment.api.service.sync.ApiClient restInvestmentApiClient, IngestConfigProperties portfolioProperties) { return new InvestmentRestAssetUniverseService(restInvestmentApiClient, portfolioProperties); } @Bean + @Primary public InvestmentRestModelPortfolioService investmentRestModelPortfolioService( + @Qualifier("restInvestmentApiClient") com.backbase.investment.api.service.sync.ApiClient restInvestmentApiClient) { return new InvestmentRestModelPortfolioService(restInvestmentApiClient); } @Bean + @Primary public InvestmentRestProductPortfolioService investmentRestProductPortfolioService( + @Qualifier("restInvestmentApiClient") com.backbase.investment.api.service.sync.ApiClient restInvestmentApiClient, IngestConfigProperties portfolioProperties) { return new InvestmentRestProductPortfolioService(restInvestmentApiClient, portfolioProperties); diff --git a/stream-investment/investment-core/src/main/java/com/backbase/stream/configuration/InvestmentServiceConfiguration.java b/stream-investment/investment-core/src/main/java/com/backbase/stream/configuration/InvestmentServiceConfiguration.java index 467392d24..2300ea3d5 100644 --- a/stream-investment/investment-core/src/main/java/com/backbase/stream/configuration/InvestmentServiceConfiguration.java +++ b/stream-investment/investment-core/src/main/java/com/backbase/stream/configuration/InvestmentServiceConfiguration.java @@ -34,15 +34,19 @@ import com.backbase.stream.investment.service.resttemplate.InvestmentRestNewsContentService; import com.backbase.stream.investment.service.resttemplate.InvestmentRestProductPortfolioService; import lombok.RequiredArgsConstructor; +import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Import; +import org.springframework.context.annotation.Primary; @Import({ DbsApiClientsAutoConfiguration.class, - InvestmentClientConfig.class + InvestmentClientConfig.class, + InvestmentWebClientConfiguration.class, + InvestmentRestServiceApiConfiguration.class }) @EnableConfigurationProperties({ InvestmentIngestionConfigurationProperties.class, @@ -54,12 +58,15 @@ public class InvestmentServiceConfiguration { @Bean - public InvestmentClientService investmentClientService(ClientApi clientApi) { + @Primary + public InvestmentClientService investmentClientService(@Qualifier("clientApi") ClientApi clientApi) { return new InvestmentClientService(clientApi); } @Bean - public InvestmentModelPortfolioService investmentModelPortfolioService(FinancialAdviceApi financialAdviceApi, + @Primary + public InvestmentModelPortfolioService investmentModelPortfolioService( + @Qualifier("financialAdviceApi") FinancialAdviceApi financialAdviceApi, InvestmentRestModelPortfolioService investmentRestModelPortfolioService, IngestConfigProperties portfolioProperties) { return new InvestmentModelPortfolioService(financialAdviceApi, investmentRestModelPortfolioService, @@ -67,68 +74,94 @@ public InvestmentModelPortfolioService investmentModelPortfolioService(Financial } @Bean + @Primary public InvestmentPortfolioProductService investmentPortfolioProductService( - InvestmentProductsApi investmentProductsApi, IngestConfigProperties portfolioProperties, - InvestmentModelPortfolioService modelPortfolioService, + @Qualifier("investmentProductsApi") InvestmentProductsApi investmentProductsApi, + IngestConfigProperties portfolioProperties, + InvestmentModelPortfolioService investmentModelPortfolioService, InvestmentRestProductPortfolioService investmentRestProductPortfolioService) { - return new InvestmentPortfolioProductService(investmentProductsApi, portfolioProperties, modelPortfolioService, - investmentRestProductPortfolioService); + return new InvestmentPortfolioProductService(investmentProductsApi, portfolioProperties, + investmentModelPortfolioService, investmentRestProductPortfolioService); } @Bean - public InvestmentPortfolioService investmentPortfolioService(PortfolioApi portfolioApi, PaymentsApi paymentsApi, - PortfolioTradingAccountsApi portfolioTradingAccountsApi, + @Primary + public InvestmentPortfolioService investmentPortfolioService( + @Qualifier("portfolioApi") PortfolioApi portfolioApi, + @Qualifier("paymentsApi") PaymentsApi paymentsApi, + @Qualifier("portfolioTradingAccountsApi") PortfolioTradingAccountsApi portfolioTradingAccountsApi, IngestConfigProperties portfolioProperties) { return new InvestmentPortfolioService(portfolioApi, paymentsApi, portfolioTradingAccountsApi, portfolioProperties); } @Bean - public InvestmentAssetUniverseService investmentAssetUniverseService(AssetUniverseApi assetUniverseApi, - InvestmentRestAssetUniverseService investmentRestAssetUniverseService) { - return new InvestmentAssetUniverseService(assetUniverseApi, investmentRestAssetUniverseService); + @Primary + public InvestmentAssetUniverseService investmentAssetUniverseService( + @Qualifier("assetUniverseApi") AssetUniverseApi assetUniverseApi, + InvestmentRestAssetUniverseService investmentRestAssetUniverseService, + IngestConfigProperties portfolioProperties) { + return new InvestmentAssetUniverseService(assetUniverseApi, investmentRestAssetUniverseService, + portfolioProperties); } @Bean - public AsyncTaskService asyncTaskService(AsyncBulkGroupsApi asyncBulkGroupsApi) { + @Primary + public AsyncTaskService asyncTaskService( + @Qualifier("asyncBulkGroupsApi") AsyncBulkGroupsApi asyncBulkGroupsApi) { return new AsyncTaskService(asyncBulkGroupsApi); } @Bean - public InvestmentAssetPriceService investmentAssetPriceService(AssetUniverseApi assetUniverseApi) { - return new InvestmentAssetPriceService(assetUniverseApi); + @Primary + public InvestmentAssetPriceService investmentAssetPriceService( + @Qualifier("assetUniverseApi") AssetUniverseApi assetUniverseApi, + IngestConfigProperties portfolioProperties) { + return new InvestmentAssetPriceService(assetUniverseApi, portfolioProperties); } @Bean - public InvestmentIntradayAssetPriceService investmentIntradayAssetPriceService(AssetUniverseApi assetUniverseApi) { - return new InvestmentIntradayAssetPriceService(assetUniverseApi); + @Primary + public InvestmentIntradayAssetPriceService investmentIntradayAssetPriceService( + @Qualifier("assetUniverseApi") AssetUniverseApi assetUniverseApi, + IngestConfigProperties portfolioProperties) { + return new InvestmentIntradayAssetPriceService(assetUniverseApi, portfolioProperties); } @Bean - public InvestmentPortfolioAllocationService investmentPortfolioAllocationService(AllocationsApi allocationsApi, - AssetUniverseApi assetUniverseApi, InvestmentApi investmentApi, + @Primary + public InvestmentPortfolioAllocationService investmentPortfolioAllocationService( + @Qualifier("allocationsApi") AllocationsApi allocationsApi, + @Qualifier("assetUniverseApi") AssetUniverseApi assetUniverseApi, + @Qualifier("investmentApi") InvestmentApi investmentApi, IngestConfigProperties portfolioProperties) { return new InvestmentPortfolioAllocationService(allocationsApi, assetUniverseApi, investmentApi, portfolioProperties); } @Bean - public InvestmentCurrencyService investmentCurrencyService(CurrencyApi currencyApi) { + @Primary + public InvestmentCurrencyService investmentCurrencyService(@Qualifier("currencyApi") CurrencyApi currencyApi) { return new InvestmentCurrencyService(currencyApi); } @Bean - public InvestmentRiskAssessmentService investmentRiskAssessmentService(RiskAssessmentApi riskAssessmentApi) { + @Primary + public InvestmentRiskAssessmentService investmentRiskAssessmentService( + @Qualifier("riskAssessmentApi") RiskAssessmentApi riskAssessmentApi) { return new InvestmentRiskAssessmentService(riskAssessmentApi); } @Bean - public InvestmentRiskQuestionaryService investmentRiskQuestionaryService(RiskAssessmentApi riskAssessmentApi, + @Primary + public InvestmentRiskQuestionaryService investmentRiskQuestionaryService( + @Qualifier("riskAssessmentApi") RiskAssessmentApi riskAssessmentApi, IngestConfigProperties portfolioProperties) { return new InvestmentRiskQuestionaryService(riskAssessmentApi, portfolioProperties); } @Bean + @Primary public InvestmentSaga investmentSaga(InvestmentClientService investmentClientService, InvestmentRiskAssessmentService investmentRiskAssessmentService, InvestmentRiskQuestionaryService investmentRiskQuestionaryService, @@ -137,7 +170,7 @@ public InvestmentSaga investmentSaga(InvestmentClientService investmentClientSer InvestmentPortfolioProductService investmentPortfolioProductService, InvestmentPortfolioAllocationService investmentPortfolioAllocationService, AsyncTaskService asyncTaskService, InvestmentIngestionConfigurationProperties coreConfigurationProperties, - AssetUniverseApi assetUniverseApi) { + @Qualifier("assetUniverseApi") AssetUniverseApi assetUniverseApi) { return new InvestmentSaga(investmentClientService, investmentRiskAssessmentService, investmentRiskQuestionaryService, investmentPortfolioService, investmentPortfolioAllocationService, investmentModelPortfolioService, investmentPortfolioProductService, asyncTaskService, @@ -145,6 +178,7 @@ public InvestmentSaga investmentSaga(InvestmentClientService investmentClientSer } @Bean + @Primary public InvestmentAssetUniverseSaga investmentStaticDataSaga( InvestmentAssetUniverseService investmentAssetUniverseService, InvestmentAssetPriceService investmentAssetPriceService, @@ -159,6 +193,7 @@ public InvestmentAssetUniverseSaga investmentStaticDataSaga( } @Bean + @Primary public InvestmentContentSaga investmentContentSaga( InvestmentRestNewsContentService investmentRestNewsContentService, InvestmentRestDocumentContentService investmentRestDocumentContentService, diff --git a/stream-investment/investment-core/src/main/java/com/backbase/stream/configuration/InvestmentWebClientConfiguration.java b/stream-investment/investment-core/src/main/java/com/backbase/stream/configuration/InvestmentWebClientConfiguration.java index 4ade2ad4b..87821687f 100644 --- a/stream-investment/investment-core/src/main/java/com/backbase/stream/configuration/InvestmentWebClientConfiguration.java +++ b/stream-investment/investment-core/src/main/java/com/backbase/stream/configuration/InvestmentWebClientConfiguration.java @@ -56,8 +56,9 @@ public ConnectionProvider investmentConnectionProvider(InvestmentWebClientProper .build(); log.info("Configured investment ConnectionProvider: maxConnections={}, maxIdleTime={}min," - + " pendingAcquireMaxCount={}", - props.getMaxConnections(), props.getMaxIdleTimeMinutes(), props.getMaxPendingAcquires()); + + " pendingAcquireMaxCount={}, pendingAcquireTimeout={}ms", + props.getMaxConnections(), props.getMaxIdleTimeMinutes(), props.getMaxPendingAcquires(), + props.getPendingAcquireTimeoutMillis()); return provider; } diff --git a/stream-investment/investment-core/src/main/java/com/backbase/stream/configuration/InvestmentWebClientProperties.java b/stream-investment/investment-core/src/main/java/com/backbase/stream/configuration/InvestmentWebClientProperties.java index af6c23775..74191e11c 100644 --- a/stream-investment/investment-core/src/main/java/com/backbase/stream/configuration/InvestmentWebClientProperties.java +++ b/stream-investment/investment-core/src/main/java/com/backbase/stream/configuration/InvestmentWebClientProperties.java @@ -17,9 +17,9 @@ * investment: * http-client: * max-connections: 20 + * max-pending-acquires: -1 + * pending-acquire-timeout-millis: 300000 * max-idle-time-minutes: 5 - * max-pending-acquires: 100 - * pending-acquire-timeout-millis: 45000 * connect-timeout-seconds: 10 * read-timeout-seconds: 30 * write-timeout-seconds: 30 @@ -31,9 +31,12 @@ public class InvestmentWebClientProperties { /** * Maximum number of open TCP connections to the Investment service. - * Limiting this prevents the service from being overwhelmed (which causes 503 responses). + * + *

Keep this moderate to avoid overwhelming the Investment service (503 responses). + * When all connections are busy, additional requests wait in the pending-acquire queue + * (see {@link #pendingAcquireTimeoutMillis}) rather than opening more TCP connections. */ - private int maxConnections = 20; + private int maxConnections = 50; /** * Maximum time (in minutes) that a connection can remain idle in the pool before being evicted. @@ -46,15 +49,24 @@ public class InvestmentWebClientProperties { private long maxLifeTimeMinutes = 30; /** - * Maximum number of requests that can be queued waiting for a connection. - * Bounds the in-memory queue so callers receive a fast failure rather than an unbounded backlog. + * Maximum number of requests that can wait for a free connection. + * + *

{@code -1} means no queue limit — callers block up to + * {@link #pendingAcquireTimeoutMillis} instead of failing fast with + * "Pending acquire queue has reached its maximum size". + * + *

Pair with application-level {@code flatMap} concurrency limits in the ingestion + * services so the queue does not grow without bound. */ - private int maxPendingAcquires = 100; + private int maxPendingAcquires = -1; /** * Maximum time (in milliseconds) a request will wait to acquire a connection from the pool. + * + *

When all {@link #maxConnections} are in use, new requests wait here rather than + * opening additional connections or failing immediately. */ - private long pendingAcquireTimeoutMillis = 45_000; + private long pendingAcquireTimeoutMillis = 150_000; /** * Background eviction interval (in seconds) for idle/expired connections in the pool. diff --git a/stream-investment/investment-core/src/main/java/com/backbase/stream/investment/service/InvestmentAssetPriceService.java b/stream-investment/investment-core/src/main/java/com/backbase/stream/investment/service/InvestmentAssetPriceService.java index 40a6b89db..25c0c3293 100644 --- a/stream-investment/investment-core/src/main/java/com/backbase/stream/investment/service/InvestmentAssetPriceService.java +++ b/stream-investment/investment-core/src/main/java/com/backbase/stream/investment/service/InvestmentAssetPriceService.java @@ -8,6 +8,7 @@ import com.backbase.investment.api.service.v1.model.OASPrice; import com.backbase.investment.api.service.v1.model.PaginatedOASPriceList; import com.backbase.investment.api.service.v1.model.TypeEnum; +import com.backbase.stream.configuration.IngestConfigProperties; import com.backbase.stream.investment.Asset; import com.backbase.stream.investment.RandomParam; import java.security.SecureRandom; @@ -38,6 +39,7 @@ public class InvestmentAssetPriceService { private final SecureRandom random = new SecureRandom(); private final AssetUniverseApi assetUniverseApi; + private final IngestConfigProperties ingestProperties; public Mono> ingestPrices(List assets) { return generatePrices(assets) @@ -95,7 +97,7 @@ private Mono>> generatePrices(List assets) { }); }) .filter(Predicate.not(CollectionUtils::isEmpty)); - }) + }, ingestProperties.getAsset().getPriceConcurrency()) .collectList(); } diff --git a/stream-investment/investment-core/src/main/java/com/backbase/stream/investment/service/InvestmentAssetUniverseService.java b/stream-investment/investment-core/src/main/java/com/backbase/stream/investment/service/InvestmentAssetUniverseService.java index 1455fdd39..98d935b2b 100644 --- a/stream-investment/investment-core/src/main/java/com/backbase/stream/investment/service/InvestmentAssetUniverseService.java +++ b/stream-investment/investment-core/src/main/java/com/backbase/stream/investment/service/InvestmentAssetUniverseService.java @@ -9,6 +9,7 @@ import com.backbase.investment.api.service.v1.model.MarketSpecialDay; import com.backbase.investment.api.service.v1.model.MarketSpecialDayRequest; import com.backbase.investment.api.service.v1.model.PaginatedAssetCategoryList; +import com.backbase.stream.configuration.IngestConfigProperties; import com.backbase.stream.investment.model.AssetCategoryEntry; import com.backbase.stream.investment.service.resttemplate.InvestmentRestAssetUniverseService; import java.net.URI; @@ -36,6 +37,7 @@ public class InvestmentAssetUniverseService { private final AssetUniverseApi assetUniverseApi; private final InvestmentRestAssetUniverseService investmentRestAssetUniverseService; + private final IngestConfigProperties ingestProperties; private final AssetMapper assetMapper = Mappers.getMapper(AssetMapper.class); /** @@ -262,7 +264,7 @@ public Flux upsertAssets(List log.warn( "Retrying asset upsert: key={}, attempt={}", asset.getKeyString(), signal.totalRetries() + 1))), - 5); + ingestProperties.getAsset().getAssetUpsertConcurrency()); }); } diff --git a/stream-investment/investment-core/src/main/java/com/backbase/stream/investment/service/InvestmentClientService.java b/stream-investment/investment-core/src/main/java/com/backbase/stream/investment/service/InvestmentClientService.java index 0e8779a68..39f6153a2 100644 --- a/stream-investment/investment-core/src/main/java/com/backbase/stream/investment/service/InvestmentClientService.java +++ b/stream-investment/investment-core/src/main/java/com/backbase/stream/investment/service/InvestmentClientService.java @@ -53,11 +53,11 @@ public class InvestmentClientService { *

  • Deduplicates clients by internalUserId before processing
    • *
  • Limits concurrent requests to avoid overwhelming the service and triggering 503 errors
    • *
  • Implements exponential backoff retry with 503 and 409 conflict handling
    • - *
  • Processes clients sequentially through the same ExecutorService to maintain order and prevent race conditions
    • + *
  • Failed clients are logged and skipped to prevent batch failures
    • * * * @param clientUsers the list of clients to upsert - * @return Mono emitting the list of successfully upserted clients + * @return Mono emitting the list of successfully upserted clients; failed clients are logged and skipped */ public Mono> upsertClients(List clientUsers) { // Deduplicate by internalUserId to prevent duplicate processing @@ -99,10 +99,12 @@ public Mono> upsertClients(List clientUsers) { .doOnSuccess(upsertedClient -> log.debug( "Successfully upserted client: investmentClientId={}, internalUserId={}", upsertedClient.getInvestmentClientId(), upsertedClient.getInternalUserId())) - .doOnError(throwable -> log.error( - "Failed to upsert client: internalUserId={}, externalUserId={}, legalEntityExternalId={}", - clientUser.getInternalUserId(), clientUser.getExternalUserId(), - clientUser.getLegalEntityId(), throwable)); + .onErrorResume(throwable -> { + log.warn("Skipping client due to error: internalUserId={}, externalUserId={}, legalEntityExternalId={}", + clientUser.getInternalUserId(), clientUser.getExternalUserId(), + clientUser.getLegalEntityId(), throwable); + return Mono.empty(); + }); }, 5) // Max 5 concurrent requests to avoid overwhelming the service diff --git a/stream-investment/investment-core/src/main/java/com/backbase/stream/investment/service/InvestmentIntradayAssetPriceService.java b/stream-investment/investment-core/src/main/java/com/backbase/stream/investment/service/InvestmentIntradayAssetPriceService.java index cb05e17f6..0b661663d 100644 --- a/stream-investment/investment-core/src/main/java/com/backbase/stream/investment/service/InvestmentIntradayAssetPriceService.java +++ b/stream-investment/investment-core/src/main/java/com/backbase/stream/investment/service/InvestmentIntradayAssetPriceService.java @@ -4,6 +4,7 @@ import com.backbase.investment.api.service.v1.model.GroupResult; import com.backbase.investment.api.service.v1.model.OASCreatePriceRequest; import com.backbase.investment.api.service.v1.model.TypeEnum; +import com.backbase.stream.configuration.IngestConfigProperties; import com.backbase.stream.investment.model.AssetWithMarketAndLatestPrice; import com.backbase.stream.investment.model.ExpandedLatestPrice; import com.backbase.stream.investment.model.PaginatedExpandedAssetList; @@ -43,6 +44,7 @@ public class InvestmentIntradayAssetPriceService { private final AssetUniverseApi assetUniverseApi; + private final IngestConfigProperties ingestProperties; /** * Generates and triggers ingestion of intraday prices for all assets that have a latest price. @@ -109,7 +111,7 @@ private Mono>> generateIntradayPrices() { ) ) .onErrorResume(e -> Mono.empty()); - }) + }, ingestProperties.getAsset().getIntradayPriceConcurrency()) .collectList(); }) .doOnError(error -> { diff --git a/stream-investment/investment-core/src/main/java/com/backbase/stream/investment/service/InvestmentPortfolioService.java b/stream-investment/investment-core/src/main/java/com/backbase/stream/investment/service/InvestmentPortfolioService.java index 0006816e7..6ade25f55 100644 --- a/stream-investment/investment-core/src/main/java/com/backbase/stream/investment/service/InvestmentPortfolioService.java +++ b/stream-investment/investment-core/src/main/java/com/backbase/stream/investment/service/InvestmentPortfolioService.java @@ -33,7 +33,9 @@ import javax.annotation.Nonnull; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; +import org.springframework.http.HttpStatus; import org.springframework.util.CollectionUtils; +import org.springframework.util.StringUtils; import org.springframework.web.reactive.function.client.WebClientResponseException; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; @@ -69,27 +71,38 @@ public class InvestmentPortfolioService { public Mono> upsertPortfolios(List investmentArrangements, Map> clientsByLeExternalId) { return Flux.fromIterable(investmentArrangements) - .flatMap(arrangement -> { - log.debug("Upserting investment portfolio for arrangement: externalId={}, name={}, productId={}", - arrangement.getExternalId(), arrangement.getName(), arrangement.getInvestmentProductId()); - - return upsertInvestmentPortfolios(arrangement, clientsByLeExternalId) - .map(p -> InvestmentPortfolio - .builder() - .portfolio(p) - .initialCash(arrangement.getInitialCash()) - .withdrawalAmount(arrangement.getWithdrawalAmount()) - .build()) - .doOnSuccess(ip -> log.debug( - "Successfully upserted investment portfolio: portfolioUuid={}, externalId={}, name={}", - ip.getPortfolio().getUuid(), ip.getPortfolio().getExternalId(), ip.getPortfolio().getName())) - .doOnError(throwable -> log.error( - "Failed to upsert investment portfolio: arrangementExternalId={}, arrangementName={}", - arrangement.getExternalId(), arrangement.getName(), throwable)); - }) + .flatMap(arrangement -> upsertSinglePortfolio(arrangement, clientsByLeExternalId)) .collectList(); } + /** + * Upserts a single portfolio derived from an investment arrangement. + * + *

    Errors are logged and result in an empty Mono so a single portfolio failure does not + * abort processing of the remaining arrangements in the batch. + */ + private Mono upsertSinglePortfolio(InvestmentArrangement arrangement, + Map> clientsByLeExternalId) { + log.debug("Upserting investment portfolio for arrangement: externalId={}, name={}, productId={}", + arrangement.getExternalId(), arrangement.getName(), arrangement.getInvestmentProductId()); + + return upsertInvestmentPortfolios(arrangement, clientsByLeExternalId) + .map(p -> InvestmentPortfolio + .builder() + .portfolio(p) + .initialCash(arrangement.getInitialCash()) + .withdrawalAmount(arrangement.getWithdrawalAmount()) + .build()) + .doOnSuccess(ip -> log.debug( + "Successfully upserted investment portfolio: portfolioUuid={}, externalId={}, name={}", + ip.getPortfolio().getUuid(), ip.getPortfolio().getExternalId(), ip.getPortfolio().getName())) + .onErrorResume(throwable -> { + log.warn("Skipping portfolio due to error: externalId={}, name={}", + arrangement.getExternalId(), arrangement.getName(), throwable); + return Mono.empty(); + }); + } + /** * Creates or updates an investment portfolio for the given arrangement. * @@ -237,12 +250,26 @@ private Mono createNewPortfolio(InvestmentArrangement investmentA .status(StatusA3dEnum.ACTIVE) .activated(OffsetDateTime.now().minusMonths(config.getPortfolio().getActivationPastMonths())); + String externalId = investmentArrangement.getExternalId(); return portfolioApi.createPortfolio(request, null, null, null) .doOnSuccess(created -> log.info( "Successfully created investment portfolio: uuid={}, externalId={}, name={}, clients={}", created.getUuid(), created.getExternalId(), created.getName(), associatedClients.size())) - .doOnError(throwable -> logPortfolioCreationError( - investmentArrangement.getExternalId(), investmentArrangement.getName(), throwable)); + .onErrorResume(WebClientResponseException.class, ex -> { + if (!isPortfolioAlreadyExistsError(ex)) { + logPortfolioCreationError(externalId, investmentArrangement.getName(), ex); + return Mono.error(ex); + } + log.info("Portfolio create conflict (already exists); re-fetching by externalId={}", externalId); + return listExistingPortfolios(externalId) + .flatMap(existing -> patchPortfolio(existing, investmentArrangement, clientsByLeId)) + .switchIfEmpty(Mono.defer(() -> { + log.warn( + "Portfolio create reported already exists but none found on re-list: externalId={}", + externalId); + return Mono.error(ex); + })); + }); } /** @@ -731,4 +758,15 @@ private void logPortfolioCreationError(String externalId, String name, Throwable } } + /** + * Detects portfolio-create conflicts where the portfolio already exists (e.g. arrangement or external ID taken). + */ + private static boolean isPortfolioAlreadyExistsError(WebClientResponseException ex) { + if (!HttpStatus.BAD_REQUEST.equals(ex.getStatusCode())) { + return false; + } + String body = ex.getResponseBodyAsString(); + return StringUtils.hasText(body) && body.toLowerCase().contains("already exists"); + } + } diff --git a/stream-investment/investment-core/src/test/java/com/backbase/stream/configuration/InvestmentClientConfigTest.java b/stream-investment/investment-core/src/test/java/com/backbase/stream/configuration/InvestmentClientConfigTest.java index 98242d1b2..4a1b30350 100644 --- a/stream-investment/investment-core/src/test/java/com/backbase/stream/configuration/InvestmentClientConfigTest.java +++ b/stream-investment/investment-core/src/test/java/com/backbase/stream/configuration/InvestmentClientConfigTest.java @@ -71,7 +71,7 @@ void investmentApiClient_returnsNonNullApiClientWithCodecsConfigured() { WebClient webClient = WebClient.builder().build(); ObjectMapper objectMapper = new ObjectMapper(); DateFormat dateFormat = mock(DateFormat.class); - HttpClient httpClient = mock(HttpClient.class); // declared parameter; not used by the method body + HttpClient httpClient = mock(HttpClient.class); ApiClient result = config.investmentApiClient(webClient, httpClient, objectMapper, dateFormat); diff --git a/stream-investment/investment-core/src/test/java/com/backbase/stream/configuration/InvestmentServiceConfigurationTest.java b/stream-investment/investment-core/src/test/java/com/backbase/stream/configuration/InvestmentServiceConfigurationTest.java index 0af267eb0..27df49301 100644 --- a/stream-investment/investment-core/src/test/java/com/backbase/stream/configuration/InvestmentServiceConfigurationTest.java +++ b/stream-investment/investment-core/src/test/java/com/backbase/stream/configuration/InvestmentServiceConfigurationTest.java @@ -95,7 +95,8 @@ void investmentPortfolioService_returnsPortfolioService() { void investmentAssetUniverseService_returnsAssetUniverseService() { assertThat(config.investmentAssetUniverseService( mock(AssetUniverseApi.class), - mock(InvestmentRestAssetUniverseService.class))) + mock(InvestmentRestAssetUniverseService.class), + new IngestConfigProperties())) .isNotNull().isInstanceOf(InvestmentAssetUniverseService.class); } @@ -109,14 +110,15 @@ void asyncTaskService_returnsAsyncTaskService() { @Test @DisplayName("investmentAssetPriceService — returns non-null InvestmentAssetPriceService") void investmentAssetPriceService_returnsAssetPriceService() { - assertThat(config.investmentAssetPriceService(mock(AssetUniverseApi.class))) + assertThat(config.investmentAssetPriceService(mock(AssetUniverseApi.class), new IngestConfigProperties())) .isNotNull().isInstanceOf(InvestmentAssetPriceService.class); } @Test @DisplayName("investmentIntradayAssetPriceService — returns non-null InvestmentIntradayAssetPriceService") void investmentIntradayAssetPriceService_returnsIntradayAssetPriceService() { - assertThat(config.investmentIntradayAssetPriceService(mock(AssetUniverseApi.class))) + assertThat(config.investmentIntradayAssetPriceService(mock(AssetUniverseApi.class), + new IngestConfigProperties())) .isNotNull().isInstanceOf(InvestmentIntradayAssetPriceService.class); } diff --git a/stream-investment/investment-core/src/test/java/com/backbase/stream/investment/service/InvestmentAssetPriceServiceTest.java b/stream-investment/investment-core/src/test/java/com/backbase/stream/investment/service/InvestmentAssetPriceServiceTest.java index 66610c245..d829653eb 100644 --- a/stream-investment/investment-core/src/test/java/com/backbase/stream/investment/service/InvestmentAssetPriceServiceTest.java +++ b/stream-investment/investment-core/src/test/java/com/backbase/stream/investment/service/InvestmentAssetPriceServiceTest.java @@ -13,6 +13,7 @@ import com.backbase.investment.api.service.v1.model.GroupResult; import com.backbase.investment.api.service.v1.model.OASPrice; import com.backbase.investment.api.service.v1.model.PaginatedOASPriceList; +import com.backbase.stream.configuration.IngestConfigProperties; import com.backbase.stream.investment.Asset; import com.backbase.stream.investment.RandomParam; import java.nio.charset.StandardCharsets; @@ -57,7 +58,7 @@ class InvestmentAssetPriceServiceTest { @BeforeEach void setUp() { assetUniverseApi = Mockito.mock(AssetUniverseApi.class); - service = new InvestmentAssetPriceService(assetUniverseApi); + service = new InvestmentAssetPriceService(assetUniverseApi, new IngestConfigProperties()); } // ========================================================================= diff --git a/stream-investment/investment-core/src/test/java/com/backbase/stream/investment/service/InvestmentAssetUniverseServiceTest.java b/stream-investment/investment-core/src/test/java/com/backbase/stream/investment/service/InvestmentAssetUniverseServiceTest.java index 737e512e0..ec063a08b 100644 --- a/stream-investment/investment-core/src/test/java/com/backbase/stream/investment/service/InvestmentAssetUniverseServiceTest.java +++ b/stream-investment/investment-core/src/test/java/com/backbase/stream/investment/service/InvestmentAssetUniverseServiceTest.java @@ -22,6 +22,7 @@ import com.backbase.investment.api.service.v1.model.PaginatedAssetCategoryList; import com.backbase.investment.api.service.v1.model.PaginatedAssetCategoryTypeList; import com.backbase.investment.api.service.v1.model.PaginatedMarketSpecialDayList; +import com.backbase.stream.configuration.IngestConfigProperties; import com.backbase.stream.investment.model.AssetCategoryEntry; import com.backbase.stream.investment.service.resttemplate.InvestmentRestAssetUniverseService; import java.net.URI; @@ -67,7 +68,8 @@ class InvestmentAssetUniverseServiceTest { void setUp() { assetUniverseApi = mock(AssetUniverseApi.class); investmentRestAssetUniverseService = mock(InvestmentRestAssetUniverseService.class); - service = new InvestmentAssetUniverseService(assetUniverseApi, investmentRestAssetUniverseService); + service = new InvestmentAssetUniverseService(assetUniverseApi, investmentRestAssetUniverseService, + new IngestConfigProperties()); } // ========================================================================= diff --git a/stream-investment/investment-core/src/test/java/com/backbase/stream/investment/service/InvestmentClientServiceTest.java b/stream-investment/investment-core/src/test/java/com/backbase/stream/investment/service/InvestmentClientServiceTest.java index a391ee1ab..b242c752d 100644 --- a/stream-investment/investment-core/src/test/java/com/backbase/stream/investment/service/InvestmentClientServiceTest.java +++ b/stream-investment/investment-core/src/test/java/com/backbase/stream/investment/service/InvestmentClientServiceTest.java @@ -5,18 +5,23 @@ import com.backbase.investment.api.service.v1.model.ClientCreateRequest; import com.backbase.investment.api.service.v1.model.OASClient; import com.backbase.investment.api.service.v1.model.OASClientUpdateRequest; +import com.backbase.investment.api.service.v1.model.PaginatedOASClientList; import com.backbase.investment.api.service.v1.model.PatchedOASClientUpdateRequest; +import com.backbase.stream.investment.ClientUser; import java.nio.charset.StandardCharsets; +import java.util.List; import java.util.UUID; import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Nested; import org.junit.jupiter.api.Test; import org.mockito.Mockito; import org.springframework.http.HttpHeaders; -import org.springframework.http.HttpStatus; import org.springframework.web.reactive.function.client.WebClientResponseException; import reactor.core.publisher.Mono; import reactor.test.StepVerifier; +import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.ArgumentMatchers.*; import static org.mockito.Mockito.when; @@ -73,5 +78,83 @@ void updateClient_success() { .expectNext(updated) .verifyComplete(); } + + @Nested + @DisplayName("upsertClients") + class UpsertClientsTests { + + @Test + @DisplayName("single client fails in batch — remaining clients still processed") + void upsertClients_singleFailure_doesNotStopBatch() { + String failUserId = "user-fail"; + String okUserId = "user-ok"; + UUID createdUuid = UUID.randomUUID(); + + when(clientApi.listClients(anyList(), isNull(), isNull(), isNull(), isNull(), + eq(failUserId), isNull(), eq(1), isNull(), isNull(), isNull(), isNull())) + .thenReturn(Mono.just(new PaginatedOASClientList().results(List.of()))); + when(clientApi.createClient(argThat(r -> r != null && failUserId.equals(r.getInternalUserId())))) + .thenReturn(Mono.error(new RuntimeException("API failure for client 1"))); + + OASClient created = Mockito.mock(OASClient.class); + when(created.getUuid()).thenReturn(createdUuid); + when(created.getInternalUserId()).thenReturn(okUserId); + when(created.getExtraData()).thenReturn(java.util.Map.of("user_external_id", "ext-ok")); + + when(clientApi.listClients(anyList(), isNull(), isNull(), isNull(), isNull(), + eq(okUserId), isNull(), eq(1), isNull(), isNull(), isNull(), isNull())) + .thenReturn(Mono.just(new PaginatedOASClientList().results(List.of()))); + ClientCreate clientCreate = new ClientCreate(createdUuid) + .internalUserId(okUserId) + .putExtraDataItem("user_external_id", "ext-ok"); + when(clientApi.createClient(argThat(r -> r != null && okUserId.equals(r.getInternalUserId())))) + .thenReturn(Mono.just(clientCreate)); + + List input = List.of( + ClientUser.builder() + .internalUserId(failUserId) + .externalUserId("ext-fail") + .legalEntityId("le-1") + .build(), + ClientUser.builder() + .internalUserId(okUserId) + .externalUserId("ext-ok") + .legalEntityId("le-2") + .build() + ); + + StepVerifier.create(service.upsertClients(input)) + .assertNext(clients -> { + assertThat(clients).hasSize(1); + assertThat(clients.getFirst().getInternalUserId()).isEqualTo(okUserId); + assertThat(clients.getFirst().getInvestmentClientId()).isEqualTo(createdUuid); + }) + .verifyComplete(); + } + + @Test + @DisplayName("all clients fail — returns empty list without throwing") + void upsertClients_allFail_returnsEmptyList() { + String failUserId = "user-all-fail"; + + when(clientApi.listClients(anyList(), isNull(), isNull(), isNull(), isNull(), + eq(failUserId), isNull(), eq(1), isNull(), isNull(), isNull(), isNull())) + .thenReturn(Mono.just(new PaginatedOASClientList().results(List.of()))); + when(clientApi.createClient(any())) + .thenReturn(Mono.error(new RuntimeException("API failure"))); + + List input = List.of( + ClientUser.builder() + .internalUserId(failUserId) + .externalUserId("ext-fail") + .legalEntityId("le-1") + .build() + ); + + StepVerifier.create(service.upsertClients(input)) + .assertNext(List::isEmpty) + .verifyComplete(); + } + } } diff --git a/stream-investment/investment-core/src/test/java/com/backbase/stream/investment/service/InvestmentIntradayAssetPriceServiceTest.java b/stream-investment/investment-core/src/test/java/com/backbase/stream/investment/service/InvestmentIntradayAssetPriceServiceTest.java index 41e6c0827..37949327a 100644 --- a/stream-investment/investment-core/src/test/java/com/backbase/stream/investment/service/InvestmentIntradayAssetPriceServiceTest.java +++ b/stream-investment/investment-core/src/test/java/com/backbase/stream/investment/service/InvestmentIntradayAssetPriceServiceTest.java @@ -13,6 +13,7 @@ import com.backbase.investment.api.service.v1.model.GroupResult; import com.backbase.investment.api.service.v1.model.OASCreatePriceRequest; import com.backbase.investment.api.service.v1.model.TypeEnum; +import com.backbase.stream.configuration.IngestConfigProperties; import com.backbase.stream.investment.model.AssetWithMarketAndLatestPrice; import com.backbase.stream.investment.model.ExpandedLatestPrice; import com.backbase.stream.investment.model.ExpandedMarket; @@ -59,7 +60,7 @@ class InvestmentIntradayAssetPriceServiceTest { @BeforeEach void setUp() { assetUniverseApi = mock(AssetUniverseApi.class); - service = new InvestmentIntradayAssetPriceService(assetUniverseApi); + service = new InvestmentIntradayAssetPriceService(assetUniverseApi, new IngestConfigProperties()); } // ========================================================================= diff --git a/stream-investment/investment-core/src/test/java/com/backbase/stream/investment/service/InvestmentPortfolioServiceTest.java b/stream-investment/investment-core/src/test/java/com/backbase/stream/investment/service/InvestmentPortfolioServiceTest.java index 2276346a3..67847186b 100644 --- a/stream-investment/investment-core/src/test/java/com/backbase/stream/investment/service/InvestmentPortfolioServiceTest.java +++ b/stream-investment/investment-core/src/test/java/com/backbase/stream/investment/service/InvestmentPortfolioServiceTest.java @@ -634,6 +634,57 @@ void upsertInvestmentPortfolios_patchFails_withWebClientException_fallsBackToExi verify(portfolioApi, never()).createPortfolio(any(), any(), any(), any()); } + @Test + @DisplayName("create fails with already-exists conflict — re-lists and patches existing portfolio") + void upsertInvestmentPortfolios_createAlreadyExists_relistsAndPatches() { + UUID portfolioUuid = UUID.randomUUID(); + UUID productId = UUID.randomUUID(); + String externalId = "PORTFOLIO-EXT-EXISTS"; + String leExternalId = "LE-EXISTS"; + UUID clientUuid = UUID.randomUUID(); + + InvestmentArrangement arrangement = buildArrangement(externalId, "Existing Portfolio", productId, leExternalId); + PortfolioList existing = buildPortfolioList(portfolioUuid, externalId, OffsetDateTime.now().minusMonths(6)); + PortfolioList patched = buildPortfolioList(portfolioUuid, externalId, OffsetDateTime.now().minusMonths(6)); + + String alreadyExistsBody = """ + {"code":"INVALID_INPUT","message":"The following fields contain input errors",\ + "errors":{"arrangement_id":["portfolio with this arrangement id already exists."],\ + "external_id":["portfolio with this external id already exists."]}}"""; + + PaginatedPortfolioListList emptyList = Mockito.mock(PaginatedPortfolioListList.class); + when(emptyList.getResults()).thenReturn(List.of()); + + PaginatedPortfolioListList existingList = Mockito.mock(PaginatedPortfolioListList.class); + when(existingList.getResults()).thenReturn(List.of(existing)); + + when(portfolioApi.listPortfolios(isNull(), isNull(), isNull(), + isNull(), eq(externalId), isNull(), isNull(), eq(1), + isNull(), isNull(), isNull(), isNull())) + .thenReturn(Mono.just(emptyList)) + .thenReturn(Mono.just(existingList)); + + when(portfolioApi.createPortfolio(any(), isNull(), isNull(), isNull())) + .thenReturn(Mono.error(WebClientResponseException.create( + HttpStatus.BAD_REQUEST.value(), "Bad Request", + HttpHeaders.EMPTY, alreadyExistsBody.getBytes(StandardCharsets.UTF_8), StandardCharsets.UTF_8))); + + when(portfolioApi.patchPortfolio(eq(portfolioUuid.toString()), isNull(), isNull(), isNull(), any())) + .thenReturn(Mono.just(patched)); + + Map> clientsByLeExternalId = Map.of(leExternalId, List.of(clientUuid)); + + StepVerifier.create(service.upsertInvestmentPortfolios(arrangement, clientsByLeExternalId)) + .expectNextMatches(p -> portfolioUuid.equals(p.getUuid())) + .verifyComplete(); + + verify(portfolioApi, Mockito.times(2)).listPortfolios( + isNull(), isNull(), isNull(), isNull(), eq(externalId), isNull(), isNull(), eq(1), + isNull(), isNull(), isNull(), isNull()); + verify(portfolioApi).createPortfolio(any(), isNull(), isNull(), isNull()); + verify(portfolioApi).patchPortfolio(eq(portfolioUuid.toString()), isNull(), isNull(), isNull(), any()); + } + @Test @DisplayName("multiple portfolios returned for same externalId — returns IllegalStateException") void upsertInvestmentPortfolios_multipleExistingPortfolios_returnsError() { @@ -731,6 +782,47 @@ void upsertPortfolios_multipleArrangements_returnsAllPortfolios() { .verifyComplete(); } + @Test + @DisplayName("single arrangement fails in batch — remaining arrangements still processed") + void upsertPortfolios_singleFailure_doesNotStopBatch() { + UUID portfolioUuid1 = UUID.randomUUID(); + UUID portfolioUuid2 = UUID.randomUUID(); + UUID productId = UUID.randomUUID(); + String externalId1 = "EXT-BATCH-FAIL"; + String externalId2 = "EXT-BATCH-OK"; + String leExternalId = "LE-BATCH"; + UUID clientUuid = UUID.randomUUID(); + + InvestmentArrangement arrangement1 = buildArrangement(externalId1, "Portfolio Fail", productId, leExternalId); + InvestmentArrangement arrangement2 = buildArrangement(externalId2, "Portfolio OK", productId, leExternalId); + + PaginatedPortfolioListList emptyList1 = Mockito.mock(PaginatedPortfolioListList.class); + when(emptyList1.getResults()).thenReturn(List.of()); + when(portfolioApi.listPortfolios(isNull(), isNull(), isNull(), + isNull(), eq(externalId1), isNull(), isNull(), eq(1), + isNull(), isNull(), isNull(), isNull())) + .thenReturn(Mono.just(emptyList1)); + + PaginatedPortfolioListList emptyList2 = Mockito.mock(PaginatedPortfolioListList.class); + when(emptyList2.getResults()).thenReturn(List.of()); + when(portfolioApi.listPortfolios(isNull(), isNull(), isNull(), + isNull(), eq(externalId2), isNull(), isNull(), eq(1), + isNull(), isNull(), isNull(), isNull())) + .thenReturn(Mono.just(emptyList2)); + + PortfolioList created2 = buildPortfolioList(portfolioUuid2, externalId2, OffsetDateTime.now().minusMonths(6)); + when(portfolioApi.createPortfolio(any(), isNull(), isNull(), isNull())) + .thenReturn(Mono.error(new RuntimeException("API failure for portfolio 1"))) + .thenReturn(Mono.just(created2)); + + Map> clientsByLeExternalId = Map.of(leExternalId, List.of(clientUuid)); + + StepVerifier.create(service.upsertPortfolios(List.of(arrangement1, arrangement2), clientsByLeExternalId)) + .expectNextMatches(list -> list.size() == 1 + && portfolioUuid2.equals(list.getFirst().getPortfolio().getUuid())) + .verifyComplete(); + } + @Test @DisplayName("empty arrangements list — returns empty list without calling API") void upsertPortfolios_emptyArrangements_returnsEmptyList() {