Implement AffinityProductsService and related components for fetching product affinities

pom.xml

@@ -18,6 +18,7 @@
 
     <properties>
         <java.version>21</java.version>
+        <resilience4j.version>2.3.0</resilience4j.version>
     </properties>
 
     <dependencies>
@@ -26,6 +27,22 @@
             <artifactId>spring-boot-starter-webflux</artifactId>
         </dependency>
 
+        <!-- Resilience4j: timeout, circuit breaker & retry for outcoming calls -->
+        <dependency>
+            <groupId>io.github.resilience4j</groupId>
+            <artifactId>resilience4j-spring-boot3</artifactId>
+            <version>${resilience4j.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>io.github.resilience4j</groupId>
+            <artifactId>resilience4j-reactor</artifactId>
+            <version>${resilience4j.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-validation</artifactId>
+        </dependency>
 
         <!-- Test -->
         <dependency>
@@ -33,6 +50,11 @@
             <artifactId>spring-boot-starter-test</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>io.projectreactor</groupId>
+            <artifactId>reactor-test</artifactId>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
     <build>

src/main/java/com/example/productaffinity/domain/exception/ExternalServiceException.java

@@ -0,0 +1,9 @@
+package com.example.productaffinity.domain.exception;
+
+//For exception on remote service.
+public class ExternalServiceException extends RuntimeException {
+
+    public ExternalServiceException(String message, Throwable cause) {
+        super(message, cause);
+    }
+}

src/main/java/com/example/productaffinity/domain/exception/ProductDetailUnavailableException.java

@@ -0,0 +1,15 @@
+package com.example.productaffinity.domain.exception;
+
+public class ProductDetailUnavailableException extends RuntimeException {
+
+    private final String productId;
+
+    public ProductDetailUnavailableException(String productId, Throwable cause) {
+        super("Product detail unavailable: " + productId, cause);
+        this.productId = productId;
+    }
+
+    public String productId() {
+        return productId;
+    }
+}

src/main/java/com/example/productaffinity/domain/exception/ProductNotFoundException.java

@@ -0,0 +1,8 @@
+package com.example.productaffinity.domain.exception;
+
+public class ProductNotFoundException extends RuntimeException {
+
+    public ProductNotFoundException(String productId) {
+        super("Product not found: " + productId);
+    }
+}

src/main/java/com/example/productaffinity/domain/port/outcoming/AffinityProductsIdsPort.java

@@ -2,8 +2,8 @@ package com.example.productaffinity.domain.port.outcoming;
 
 import reactor.core.publisher.Mono;
 
-import java.util.Collection;
+import java.util.List;
 
 public interface AffinityProductsIdsPort {
-    Mono<Collection<String>> findAffinityProductIds(String productId);
+    Mono<List<String>> findAffinityProductIds(String productId);
 }

src/main/java/com/example/productaffinity/domain/service/AffinityProductsService.java

@@ -0,0 +1,42 @@
+package com.example.productaffinity.domain.service;
+
+import com.example.productaffinity.domain.exception.ProductDetailUnavailableException;
+import com.example.productaffinity.domain.model.Product;
+import com.example.productaffinity.domain.port.incomming.GetAffinityProductsUC;
+import com.example.productaffinity.domain.port.outcoming.AffinityProductsIdsPort;
+import com.example.productaffinity.domain.port.outcoming.ProductDetailsPort;
+import org.springframework.stereotype.Service;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import java.util.Collection;
+import java.util.List;
+
+@Service
+public class AffinityProductsService implements GetAffinityProductsUC {
+
+    private final AffinityProductsIdsPort affinityProductsIdsPort;
+    private final ProductDetailsPort productDetailsPort;
+
+    //Máximum number of detail req.
+    private static final int DETAIL_FETCH_CONCURRENCY = 10;
+
+    public AffinityProductsService(final AffinityProductsIdsPort affinityProductsIdsPort,
+                                   final ProductDetailsPort productDetailsPort) {
+        this.affinityProductsIdsPort = affinityProductsIdsPort;
+        this.productDetailsPort = productDetailsPort;
+    }
+
+    // Returns similar products in similarity order. If fails resolving the request, web propagates status
+    @Override
+    public Mono<List<Product>> getAffinityProducts(final String productId) {
+        return affinityProductsIdsPort.findAffinityProductIds(productId).flatMapMany(Flux::fromIterable)
+                .distinct().flatMapSequential(this::fetchDetailOrSkip, DETAIL_FETCH_CONCURRENCY)
+                .collectList();
+    }
+
+    private Mono<Product> fetchDetailOrSkip(String productId) {
+        return productDetailsPort.findProductDetail(productId)
+                .onErrorResume(ProductDetailUnavailableException.class, error -> Mono.empty());
+    }
+}

src/main/java/com/example/productaffinity/domain/service/SimilarProductsService.java

@@ -1,20 +0,0 @@
-package com.example.productaffinity.domain.service;
-
-import com.example.productaffinity.domain.model.Product;
-import com.example.productaffinity.domain.port.incomming.GetAffinityProductsUC;
-import org.springframework.stereotype.Service;
-import reactor.core.publisher.Mono;
-
-import java.util.List;
-
-@Service
-public class SimilarProductsService implements GetAffinityProductsUC {
-
-    public SimilarProductsService() {
-    }
-
-    @Override
-    public Mono<List<Product>> getAffinityProducts(String productId) {
-        return null;
-    }
-}

src/main/java/com/example/productaffinity/infraestructure/adapter/outcomming/client/ApiAffinityIdsAdapter.java

@@ -0,0 +1,97 @@
+package com.example.productaffinity.infraestructure.adapter.outcomming.client;
+
+import com.example.productaffinity.domain.exception.ExternalServiceException;
+import com.example.productaffinity.domain.exception.ProductNotFoundException;
+import com.example.productaffinity.domain.port.outcoming.AffinityProductsIdsPort;
+import io.github.resilience4j.bulkhead.Bulkhead;
+import io.github.resilience4j.bulkhead.BulkheadRegistry;
+import io.github.resilience4j.circuitbreaker.CircuitBreaker;
+import io.github.resilience4j.circuitbreaker.CircuitBreakerRegistry;
+import io.github.resilience4j.reactor.bulkhead.operator.BulkheadOperator;
+import io.github.resilience4j.reactor.circuitbreaker.operator.CircuitBreakerOperator;
+import io.github.resilience4j.reactor.retry.RetryOperator;
+import io.github.resilience4j.reactor.timelimiter.TimeLimiterOperator;
+import io.github.resilience4j.retry.Retry;
+import io.github.resilience4j.retry.RetryRegistry;
+import io.github.resilience4j.timelimiter.TimeLimiter;
+import io.github.resilience4j.timelimiter.TimeLimiterRegistry;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.core.ParameterizedTypeReference;
+import org.springframework.core.codec.DecodingException;
+import org.springframework.stereotype.Component;
+import org.springframework.web.reactive.function.client.WebClient;
+import org.springframework.web.reactive.function.client.WebClientResponseException;
+import reactor.core.publisher.Mono;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.Objects;
+
+@Component
+public class ApiAffinityIdsAdapter implements AffinityProductsIdsPort {
+
+    private static final Logger log = LoggerFactory.getLogger(ApiAffinityIdsAdapter.class);
+    private static final String RESILIENCE_INSTANCE = "affinityIds";
+
+    private final WebClient webClient;
+    private final TimeLimiter timeLimiter;
+    private final CircuitBreaker circuitBreaker;
+    private final Retry retry;
+    private final Bulkhead bulkhead;
+    private final RequestCoalescer<String, List<String>> requestCoalescer;
+
+    public ApiAffinityIdsAdapter(
+            final @Qualifier("affinityIdsWebClient") WebClient existingApiWebClient,
+            final TimeLimiterRegistry timeLimiterRegistry,
+            final CircuitBreakerRegistry circuitBreakerRegistry,
+            final RetryRegistry retryRegistry,
+            final BulkheadRegistry bulkheadRegistry
+    ) {
+        this.webClient = existingApiWebClient;
+        this.timeLimiter = timeLimiterRegistry.timeLimiter(RESILIENCE_INSTANCE);
+        this.circuitBreaker = circuitBreakerRegistry.circuitBreaker(RESILIENCE_INSTANCE);
+        this.retry = retryRegistry.retry(RESILIENCE_INSTANCE);
+        this.bulkhead = bulkheadRegistry.bulkhead(RESILIENCE_INSTANCE);
+        this.requestCoalescer = new RequestCoalescer<>();
+    }
+
+    @Override
+    public Mono<List<String>> findAffinityProductIds(final String productId) {
+        return requestCoalescer.execute(productId, () -> fetchSimilarProductIds(productId));
+    }
+
+    private Mono<List<String>> fetchSimilarProductIds(final String productId) {
+        Mono<List<String>> call = webClient.get()
+                .uri("/product/{productId}/similarids", productId)
+                .retrieve()
+                .bodyToMono(new ParameterizedTypeReference<List<String>>() { })
+                .switchIfEmpty(Mono.error(new DecodingException("Empty similar product IDs response")))
+                .map(this::validateSimilarProductIds)
+                .onErrorMap(WebClientResponseException.NotFound.class, e -> new ProductNotFoundException(productId));
+
+        return call.transformDeferred(TimeLimiterOperator.of(timeLimiter))
+                .transformDeferred(CircuitBreakerOperator.of(circuitBreaker))
+                .transformDeferred(BulkheadOperator.of(bulkhead))
+                .transformDeferred(RetryOperator.of(retry))
+                .onErrorMap(this::isUnexpectedError,
+                        error -> new ExternalServiceException(
+                                "Could not obtain similar product IDs for " + productId, error))
+                .doOnError(ExternalServiceException.class,
+                        error -> log.warn("similar_product_ids_unavailable productId={} cause={}",
+                                productId, error.getCause().toString()));
+    }
+
+    private List<String> validateSimilarProductIds(List<String> productIds) {
+        if (productIds.stream().anyMatch(Objects::isNull)) {
+            throw new DecodingException("Similar product IDs response contains null");
+        }
+        return List.copyOf(productIds);
+    }
+
+    private boolean isUnexpectedError(Throwable error) {
+        return !(error instanceof ProductNotFoundException);
+    }
+
+}

src/main/java/com/example/productaffinity/infraestructure/adapter/outcomming/client/ApiProductDetailAdapter.java

@@ -0,0 +1,45 @@
+package com.example.productaffinity.infraestructure.adapter.outcomming.client;
+
+import com.example.productaffinity.domain.model.Product;
+import com.example.productaffinity.domain.port.outcoming.ProductDetailsPort;
+import io.github.resilience4j.bulkhead.Bulkhead;
+import io.github.resilience4j.bulkhead.BulkheadRegistry;
+import io.github.resilience4j.retry.Retry;
+import io.github.resilience4j.retry.RetryRegistry;
+import io.github.resilience4j.timelimiter.TimeLimiter;
+import io.github.resilience4j.timelimiter.TimeLimiterRegistry;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.stereotype.Component;
+import org.springframework.web.reactive.function.client.WebClient;
+import reactor.core.publisher.Mono;
+
+@Component
+public class ApiProductDetailAdapter implements ProductDetailsPort {
+
+    private static final String RESILIENCE_INSTANCE = "productDetail";
+
+    private final WebClient webClient;
+    private final TimeLimiter timeLimiter;
+    private final Retry retry;
+    private final Bulkhead bulkhead;
+    private final RequestCoalescer<String, Product> requestCoalescer;
+
+
+    public ApiProductDetailAdapter(
+        @Qualifier("productDetailWebClient") final WebClient existingApiWebClient,
+        final TimeLimiterRegistry timeLimiterRegistry,
+        final RetryRegistry retryRegistry,
+        final BulkheadRegistry bulkheadRegistry) {
+
+        this.webClient = existingApiWebClient;
+        this.timeLimiter = timeLimiterRegistry.timeLimiter(RESILIENCE_INSTANCE);
+        this.retry = retryRegistry.retry(RESILIENCE_INSTANCE);
+        this.bulkhead = bulkheadRegistry.bulkhead(RESILIENCE_INSTANCE);
+        this.requestCoalescer = new RequestCoalescer<>();
+    }
+
+    @Override
+    public Mono<Product> findProductDetail(final String productId) {
+        return null;
+    }
+}

src/main/java/com/example/productaffinity/infraestructure/adapter/outcomming/client/RequestCoalescer.java

@@ -0,0 +1,21 @@
+package com.example.productaffinity.infraestructure.adapter.outcomming.client;
+
+import reactor.core.publisher.Mono;
+
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.function.Supplier;
+
+//Shares a single active request per key and removes it inmediately when it finishes.
+public class RequestCoalescer<K, V> {
+    private final ConcurrentMap<K, Mono<V>> requests = new ConcurrentHashMap<>();
+    Mono<V> execute(K key, Supplier<Mono<V>> requestSupplier) {
+        return Mono.defer(() -> requests.computeIfAbsent(
+                key, ignored -> sharedRequest(key, requestSupplier)));
+    }
+    private Mono<V> sharedRequest(K key, Supplier<Mono<V>> requestSupplier) {
+        return Mono.defer(requestSupplier)
+                .doFinally(ignored -> requests.remove(key))
+                .cache();
+    }
+}

src/main/java/com/example/productaffinity/infraestructure/config/ApiProperties.java

@@ -0,0 +1,24 @@
+package com.example.productaffinity.infraestructure.config;
+
+import jakarta.validation.Valid;
+import jakarta.validation.constraints.Min;
+import jakarta.validation.constraints.NotBlank;
+import org.jetbrains.annotations.NotNull;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.validation.annotation.Validated;
+
+import java.time.Duration;
+
+@Validated
+@ConfigurationProperties(prefix = "api")
+public record ApiProperties(
+        @NotBlank String baseUrl,
+        @Valid @NotNull ClientProperties affinityIds,
+        @Valid @NotNull ClientProperties productDetail) {
+    public record ClientProperties(
+            @Min(1) int maxConnections,
+            @Min(1) int pendingAcquireMaxCount,
+            @NotNull Duration pendingAcquireTimeout,
+            @NotNull Duration responseTimeout) {
+    }
+}

src/main/java/com/example/productaffinity/infraestructure/config/WebClientConfig.java

@@ -0,0 +1,61 @@
+package com.example.productaffinity.infraestructure.config;
+
+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 org.springframework.http.client.reactive.ReactorClientHttpConnector;
+import org.springframework.web.reactive.function.client.WebClient;
+import reactor.netty.http.client.HttpClient;
+import reactor.netty.resources.ConnectionProvider;
+
+@Configuration(proxyBeanMethods = false)
+@EnableConfigurationProperties(ApiProperties.class)
+public class WebClientConfig {
+
+    @Bean(name = "affinityIdsConnectionProvider", destroyMethod = "dispose")
+    public ConnectionProvider affinityIdsConnectionProvider(ApiProperties properties) {
+        return connectionProvider("similar-ids-pool", properties.affinityIds());
+    }
+
+    @Bean(name = "productDetailConnectionProvider", destroyMethod = "dispose")
+    public ConnectionProvider productDetailConnectionProvider(ApiProperties properties) {
+        return connectionProvider("product-detail-pool", properties.productDetail());
+    }
+
+    @Bean("affinityIdsWebClient")
+    public WebClient affinityIdsWebClient(
+            ApiProperties properties,
+            @Qualifier("affinityIdsConnectionProvider") ConnectionProvider connectionProvider) {
+        return webClient(properties.baseUrl(), properties.affinityIds(), connectionProvider);
+    }
+
+    @Bean("productDetailWebClient")
+    public WebClient productDetailWebClient(
+            ApiProperties properties,
+            @Qualifier("productDetailConnectionProvider") ConnectionProvider connectionProvider) {
+        return webClient(properties.baseUrl(), properties.productDetail(), connectionProvider);
+    }
+
+    private ConnectionProvider connectionProvider(final String name,
+                                                  final ApiProperties.ClientProperties properties) {
+        return ConnectionProvider.builder(name)
+                .maxConnections(properties.maxConnections())
+                .pendingAcquireMaxCount(properties.pendingAcquireMaxCount())
+                .pendingAcquireTimeout(properties.pendingAcquireTimeout())
+                .build();
+    }
+
+    private WebClient webClient(
+            final String baseUrl,
+            final ApiProperties.ClientProperties properties,
+            final ConnectionProvider connectionProvider) {
+        HttpClient httpClient = HttpClient.create(connectionProvider)
+                .responseTimeout(properties.responseTimeout());
+        return WebClient.builder()
+                .baseUrl(baseUrl)
+                .clientConnector(new ReactorClientHttpConnector(httpClient))
+                .build();
+    }
+
+}

src/main/resources/application.yml

@@ -0,0 +1,12 @@
+api:
+  base-url: http://localhost:3001
+  affinity-ids:
+    max-connections: 200
+    pending-acquire-max-count: 200
+    pending-acquire-timeout: 250ms
+    response-timeout: 3s
+  product-detail:
+    max-connections: 500
+    pending-acquire-max-count: 500
+    pending-acquire-timeout: 250ms
+    response-timeout: 3s

src/test/java/com/example/productaffinity/domain/service/AffinityProductsServiceTest.java

@@ -0,0 +1,124 @@
+package com.example.productaffinity.domain.service;
+
+import com.example.productaffinity.domain.exception.ExternalServiceException;
+import com.example.productaffinity.domain.exception.ProductDetailUnavailableException;
+import com.example.productaffinity.domain.exception.ProductNotFoundException;
+import com.example.productaffinity.domain.model.Product;
+import com.example.productaffinity.domain.port.outcoming.AffinityProductsIdsPort;
+import com.example.productaffinity.domain.port.outcoming.ProductDetailsPort;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
+
+import java.math.BigDecimal;
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+
+class AffinityProductsServiceTest {
+
+    private AffinityProductsIdsPort affinityProductsIdsPort;
+    private ProductDetailsPort productDetailsPort;
+    private AffinityProductsService service;
+
+    @BeforeEach
+    void setUp() {
+        affinityProductsIdsPort = mock(AffinityProductsIdsPort.class);
+        productDetailsPort = mock(ProductDetailsPort.class);
+        service = new AffinityProductsService(affinityProductsIdsPort, productDetailsPort);
+    }
+
+    @Test
+    void returnsDetailsInSimilarityOrder() {
+        when(affinityProductsIdsPort.findAffinityProductIds("1")).thenReturn(Mono.just(List.of("2", "3", "4")));
+        when(productDetailsPort.findProductDetail("2")).thenReturn(Mono.just(product("2")));
+        when(productDetailsPort.findProductDetail("3")).thenReturn(Mono.just(product("3")));
+        when(productDetailsPort.findProductDetail("4")).thenReturn(Mono.just(product("4")));
+
+        StepVerifier.create(service.getAffinityProducts("1"))
+                .assertNext(products -> assertEquals(List.of("2", "3", "4"), ids(products)))
+                .verifyComplete();
+    }
+
+    @Test
+    void removesDuplicateProductsWithoutChangingSimilarityOrder() {
+        when(affinityProductsIdsPort.findAffinityProductIds("1"))
+                .thenReturn(Mono.just(List.of("2", "3", "2")));
+        when(productDetailsPort.findProductDetail("2")).thenReturn(Mono.just(product("2")));
+        when(productDetailsPort.findProductDetail("3")).thenReturn(Mono.just(product("3")));
+
+        StepVerifier.create(service.getAffinityProducts("1"))
+                .assertNext(products -> assertEquals(List.of("2", "3"), ids(products)))
+                .verifyComplete();
+    }
+
+    @Test
+    void skipsItemsWhoseDetailFailsWithoutFailingTheWholeRequest() {
+        when(affinityProductsIdsPort.findAffinityProductIds("4")).thenReturn(Mono.just(List.of("1", "2", "5")));
+        when(productDetailsPort.findProductDetail("1")).thenReturn(Mono.just(product("1")));
+        when(productDetailsPort.findProductDetail("2")).thenReturn(Mono.just(product("2")));
+        when(productDetailsPort.findProductDetail("5"))
+                .thenReturn(Mono.error(new ProductDetailUnavailableException("5", new RuntimeException("404"))));
+
+        StepVerifier.create(service.getAffinityProducts("4"))
+                .assertNext(products -> assertEquals(List.of("1", "2"), ids(products)))
+                .verifyComplete();
+    }
+
+    @Test
+    void propagatesUnexpectedDetailFailure() {
+        when(affinityProductsIdsPort.findAffinityProductIds("1")).thenReturn(Mono.just(List.of("2")));
+        when(productDetailsPort.findProductDetail("2"))
+                .thenReturn(Mono.error(new IllegalStateException("programming defect")));
+
+        StepVerifier.create(service.getAffinityProducts("1"))
+                .expectErrorMatches(error -> error instanceof IllegalStateException
+                        && error.getMessage().equals("programming defect"))
+                .verify();
+    }
+
+    @Test
+    void skipsItemThatTimesOut() {
+        when(affinityProductsIdsPort.findAffinityProductIds("3")).thenReturn(Mono.just(List.of("100", "1000")));
+        when(productDetailsPort.findProductDetail("100")).thenReturn(Mono.just(product("100")));
+        when(productDetailsPort.findProductDetail("1000"))
+                .thenReturn(Mono.error(new ProductDetailUnavailableException(
+                        "1000", new java.util.concurrent.TimeoutException("too slow"))));
+
+        StepVerifier.create(service.getAffinityProducts("3"))
+                .assertNext(products -> assertEquals(List.of("100"), ids(products)))
+                .verifyComplete();
+    }
+
+    @Test
+    void propagatesProductNotFoundFromMainCall() {
+        when(affinityProductsIdsPort.findAffinityProductIds("999"))
+                .thenReturn(Mono.error(new ProductNotFoundException("999")));
+
+        StepVerifier.create(service.getAffinityProducts("999"))
+                .expectError(ProductNotFoundException.class)
+                .verify();
+    }
+
+    @Test
+    void propagatesExternalServiceFailureFromMainCall() {
+        when(affinityProductsIdsPort.findAffinityProductIds("1"))
+                .thenReturn(Mono.error(new ExternalServiceException("boom", new RuntimeException())));
+
+        StepVerifier.create(service.getAffinityProducts("1"))
+                .expectError(ExternalServiceException.class)
+                .verify();
+    }
+
+    private List<String> ids(List<Product> products) {
+        return products.stream().map(Product::id).toList();
+    }
+
+    private Product product(String id) {
+        return new Product(id, "Product " + id, BigDecimal.TEN, true);
+    }
+}