portfolio-trading-accounts upsert

CHANGELOG.md

@@ -1,6 +1,10 @@
 # Changelog
 All notable changes to this project will be documented in this file.
 
+## [9.7.0]
+### Added
+ - Implement Portfolio Trading Account Creation;
+
 ## [9.6.0]
 ### Added
  - Implement for Document ingestion;

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

@@ -12,6 +12,7 @@
 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.stream.clients.config.CompositeApiClientConfig;
 import com.fasterxml.jackson.annotation.JsonInclude.Include;
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -117,6 +118,12 @@ public PaymentsApi paymentsApi(ApiClient investmentApiClient) {
         return new PaymentsApi(investmentApiClient);
     }
 
+    @Bean
+    @ConditionalOnMissingBean
+    public PortfolioTradingAccountsApi portfolioTradingAccountsApi(ApiClient investmentApiClient) {
+        return new PortfolioTradingAccountsApi(investmentApiClient);
+    }
+
     @Bean
     @ConditionalOnMissingBean
     public CurrencyApi currencyApi(ApiClient investmentApiClient) {

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

@@ -11,6 +11,7 @@
 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.stream.clients.autoconfigure.DbsApiClientsAutoConfiguration;
 import com.backbase.stream.investment.saga.InvestmentAssetUniverseSaga;
 import com.backbase.stream.investment.saga.InvestmentContentSaga;
@@ -59,10 +60,10 @@ public CustomIntegrationApiService customIntegrationApiService(ApiClient apiClie
 
     @Bean
     public InvestmentPortfolioService investmentPortfolioService(PortfolioApi portfolioApi,
-        InvestmentProductsApi investmentProductsApi, PaymentsApi paymentsApi,
+        InvestmentProductsApi investmentProductsApi, PaymentsApi paymentsApi, PortfolioTradingAccountsApi portfolioTradingAccountsApi,
         InvestmentIngestionConfigurationProperties configurationProperties) {
         return new InvestmentPortfolioService(investmentProductsApi, portfolioApi, paymentsApi,
-            configurationProperties);
+            portfolioTradingAccountsApi, configurationProperties);
     }
 
     @Bean

stream-investment/investment-core/src/main/java/com/backbase/stream/investment/InvestmentData.java

@@ -5,6 +5,7 @@
 import com.backbase.investment.api.service.v1.model.PortfolioList;
 import com.backbase.investment.api.service.v1.model.PortfolioProduct;
 import com.backbase.investment.api.service.v1.model.ProductTypeEnum;
+import com.backbase.stream.investment.model.InvestmentPortfolioTradingAccount;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
@@ -28,6 +29,7 @@ public class InvestmentData {
     private List<PortfolioProduct> portfolioProducts;
     private InvestmentAssetData investmentAssetData;
     private List<PortfolioList> portfolios;
+    private List<InvestmentPortfolioTradingAccount> investmentPortfolioTradingAccounts;
 
     public Map<String, List<UUID>> getClientsByLeExternalId() {
         Map<String, List<UUID>> clientsByLeExternalId = new HashMap<>();

stream-investment/investment-core/src/main/java/com/backbase/stream/investment/model/InvestmentPortfolioTradingAccount.java

@@ -0,0 +1,14 @@
+package com.backbase.stream.investment.model;
+
+import lombok.Builder;
+import lombok.Data;
+
+@Data
+@Builder
+public class InvestmentPortfolioTradingAccount {
+    private String portfolioExternalId;
+    private String accountId;
+    private String accountExternalId;
+    private Boolean isDefault;
+    private Boolean isInternal;
+}

stream-investment/investment-core/src/main/java/com/backbase/stream/investment/saga/InvestmentSaga.java

@@ -1,8 +1,10 @@
 package com.backbase.stream.investment.saga;
 
+import com.backbase.investment.api.service.v1.model.PortfolioList;
 import com.backbase.stream.configuration.InvestmentIngestionConfigurationProperties;
 import com.backbase.stream.investment.InvestmentData;
 import com.backbase.stream.investment.InvestmentTask;
+import com.backbase.stream.investment.model.InvestmentPortfolioTradingAccount;
 import com.backbase.stream.investment.service.AsyncTaskService;
 import com.backbase.stream.investment.service.InvestmentClientService;
 import com.backbase.stream.investment.service.InvestmentModelPortfolioService;
@@ -56,6 +58,7 @@ public class InvestmentSaga implements StreamTaskExecutor<InvestmentTask> {
     public static final String RESULT_FAILED = "failed";
 
     private static final String INVESTMENT_PRODUCTS = "investment-products";
+    private static final String INVESTMENT_PORTFOLIO_TRADING_ACCOUNTS = "investment-portfolio-trading-accounts";
     private static final String INVESTMENT_PORTFOLIO_MODELS = "investment-portfolio-models";
     private static final String INVESTMENT_PORTFOLIOS = "investment-portfolios";
     private static final String PROCESSING_PREFIX = "Processing ";
@@ -81,6 +84,7 @@ public Mono<InvestmentTask> executeTask(InvestmentTask streamTask) {
             .flatMap(this::upsertClients)
             .flatMap(this::upsertInvestmentProducts)
             .flatMap(this::upsertInvestmentPortfolios)
+            .flatMap(this::upsertPortfolioTradingAccounts)
             .flatMap(this::upsertInvestmentPortfolioDeposits)
             .flatMap(this::upsertPortfoliosAllocations)
             .doOnSuccess(completedTask -> log.info(
@@ -134,6 +138,14 @@ private Mono<InvestmentTask> upsertPortfoliosAllocations(InvestmentTask investme
                         data.getPortfolioProducts(),
                         investmentTask.getData().getInvestmentAssetData()))
                 .collectList()
+                .doOnError(throwable -> {
+                    log.error("Allocation generation failed for portfolios:{} taskId={}",
+                        data.getPortfolios().stream().map(PortfolioList::getUuid).toList(), investmentTask.getId(),
+                        throwable);
+                    investmentTask.error(INVESTMENT_PORTFOLIO_TRADING_ACCOUNTS, OP_UPSERT, RESULT_FAILED,
+                        investmentTask.getName(), investmentTask.getId(),
+                        "Failed to upsert investment portfolio trading accounts: " + throwable.getMessage());
+                })
                 .map(o -> investmentTask)
             );
     }
@@ -263,6 +275,36 @@ private Mono<InvestmentTask> upsertInvestmentProducts(InvestmentTask investmentT
             });
     }
 
+    private Mono<InvestmentTask> upsertPortfolioTradingAccounts(InvestmentTask investmentTask) {
+        List<InvestmentPortfolioTradingAccount> investmentPortfolioTradingAccounts = investmentTask.getData()
+            .getInvestmentPortfolioTradingAccounts();
+        int accountsCount = investmentPortfolioTradingAccounts.size();
+
+        log.info("Starting investment portfolio trading accounts upsert: taskId={}, arrangementCount={}",
+            investmentTask.getId(), accountsCount);
+
+        investmentTask.info(INVESTMENT_PORTFOLIO_TRADING_ACCOUNTS, OP_UPSERT, null, investmentTask.getName(),
+            investmentTask.getId(), PROCESSING_PREFIX + accountsCount + " investment portfolio trading accounts");
+
+        return investmentPortfolioService.upsertPortfolioTradingAccounts(investmentPortfolioTradingAccounts)
+            .map(products -> {
+                investmentTask.info(INVESTMENT_PORTFOLIO_TRADING_ACCOUNTS, OP_UPSERT, RESULT_CREATED,
+                    investmentTask.getName(), investmentTask.getId(),
+                    UPSERTED_PREFIX + products.size() + " investment portfolio trading accounts");
+                log.info("Successfully upserted all investment portfolio trading accounts: taskId={}, productCount={}",
+                    investmentTask.getId(), products.size());
+
+                return investmentTask;
+            })
+            .doOnError(throwable -> {
+                log.error("Failed to upsert investment portfolio trading accounts: taskId={}, arrangementCount={}",
+                    investmentTask.getId(), accountsCount, throwable);
+                investmentTask.error(INVESTMENT_PORTFOLIO_TRADING_ACCOUNTS, OP_UPSERT, RESULT_FAILED,
+                    investmentTask.getName(), investmentTask.getId(),
+                    "Failed to upsert investment portfolio trading accounts: " + throwable.getMessage());
+            });
+    }
+
     /**
      * Upserts investment clients for all users in the investment data.
      *

stream-investment/investment-core/src/main/java/com/backbase/stream/investment/service/InvestmentAssetUniverseService.java

@@ -104,7 +104,7 @@ public Mono<com.backbase.stream.investment.Asset> getOrCreateAsset(com.backbase.
             // If asset exists, log and return it
             .flatMap(a -> {
                 log.info("Asset already exists with Asset Identifier : {}", assetIdentifier);
-                return investmentRestAssetUniverseService.patchAsset(a, asset).thenReturn(a);
+                return investmentRestAssetUniverseService.patchAsset(a, asset, categoryIdByCode).thenReturn(a);
             })
             .map(assetMapper::map)
             // If Mono is empty (asset not found), create the asset