On Repeat auto playlist generation

.gitignore

@@ -27,7 +27,9 @@ gradle/**
 out/
 !**/src/main/**/out/
 !**/src/test/**/out/
-
+**/src/main/generated
+!*.env
+local.env
 ### NetBeans ###
 /nbproject/private/
 /nbbuild/

build.gradle

@@ -45,9 +45,12 @@ dependencies {
     implementation 'org.springframework.cloud:spring-cloud-starter-netflix-eureka-client'
     implementation 'org.jetbrains:annotations:23.0.0'
     implementation 'org.flywaydb:flyway-core'
-    implementation 'com.odeyalo.sonata:suite:1.0.0'
+    implementation 'com.odeyalo.sonata.suite:common:1.0.1'
+    implementation 'com.odeyalo.sonata.suite:suite-brokers:0.0.11'
     implementation "com.odeyalo.sonata.suite:security:${suiteSecurityVersion}"
     implementation "org.mapstruct:mapstruct:${mapstructVersion}"
+    implementation 'io.projectreactor.kafka:reactor-kafka:1.3.21'
+    implementation 'org.springframework.kafka:spring-kafka:3.0.9'
     compileOnly 'org.projectlombok:lombok'
     runtimeOnly 'org.postgresql:postgresql'
     runtimeOnly 'org.postgresql:r2dbc-postgresql'
@@ -62,7 +65,7 @@ dependencies {
     testImplementation 'org.springframework.cloud:spring-cloud-contract-stub-runner'
     testImplementation 'io.r2dbc:r2dbc-h2'
     testImplementation "com.github.javafaker:javafaker:${javaFakerVersion}"
-
+    testImplementation 'org.testcontainers:postgresql'
 }
 
 dependencyManagement {

docker-compose.test.yml

@@ -17,12 +17,16 @@ services:
       # Used to download required dependencies from Suite module
       GITHUB_USERNAME: ${GH_USERNAME}
       GITHUB_ACCESS_TOKEN: ${GH_ACCESS_TOKEN}
+    volumes:
+      - /var/run/docker.sock:/var/run/docker.sock
+
 
   db:
     image: 'postgres:15.2'
     container_name: db
+    command: -c 'max_connections=200'
     ports:
-      - "5433:5432"
+      - "5435:5432"
     environment:
       - POSTGRES_USER=postgres
       - POSTGRES_PASSWORD=root

sonata-playlists-test-run.sh

@@ -1,3 +1,3 @@
 #!/bin/bash
 
-docker-compose -f docker-compose.test.yml up --abort-on-container-exit
+docker compose -f docker-compose.test.yml up --abort-on-container-exit

src/main/java/com/odeyalo/sonata/playlists/config/kafka/KafkaConsumerConfiguration.java

@@ -0,0 +1,55 @@
+package com.odeyalo.sonata.playlists.config.kafka;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
+import com.fasterxml.jackson.module.paramnames.ParameterNamesModule;
+import com.odeyalo.sonata.suite.brokers.events.playlist.gen.PlaylistImagesGeneratedEvent;
+import com.odeyalo.sonata.suite.brokers.events.playlist.gen.payload.GenerativePlaylistEvent;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.springframework.boot.autoconfigure.jackson.Jackson2ObjectMapperBuilderCustomizer;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.core.reactive.ReactiveKafkaConsumerTemplate;
+import org.springframework.kafka.support.serializer.JsonDeserializer;
+import reactor.kafka.receiver.ReceiverOptions;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+@Configuration
+public class KafkaConsumerConfiguration {
+
+    @Bean
+    public Jackson2ObjectMapperBuilderCustomizer customizer() {
+        return jacksonObjectMapperBuilder -> jacksonObjectMapperBuilder.modules(
+                new JavaTimeModule(),
+                new ParameterNamesModule()
+        );
+    }
+
+    @Bean
+    public ReceiverOptions<String, PlaylistImagesGeneratedEvent> generatedPlaylistReceiverOptions(ObjectMapper objectMapper) {
+        final Map<String, Object> consumerProps = new HashMap<>();
+        consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
+        consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "generated-playlists-consumers");
+        consumerProps.put(JsonDeserializer.TRUSTED_PACKAGES, "*");
+        consumerProps.put(JsonDeserializer.USE_TYPE_INFO_HEADERS, false);
+        consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+        consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
+
+        final JsonDeserializer<PlaylistImagesGeneratedEvent> deserializer = new JsonDeserializer<>(GenerativePlaylistEvent.class, objectMapper);
+
+        deserializer.setUseTypeHeaders(false);
+
+        return ReceiverOptions.<String, PlaylistImagesGeneratedEvent>create(consumerProps)
+                .withValueDeserializer(deserializer)
+                .subscription(Collections.singleton("playlists.gen.images"));
+    }
+
+    @Bean
+    public ReactiveKafkaConsumerTemplate<String, PlaylistImagesGeneratedEvent> reactiveKafkaConsumerTemplate(ReceiverOptions<String, PlaylistImagesGeneratedEvent> kafkaReceiverOptions) {
+        return new ReactiveKafkaConsumerTemplate<>(kafkaReceiverOptions);
+    }
+}

src/main/java/com/odeyalo/sonata/playlists/config/security/local/AuthenticationConfiguration.java

@@ -0,0 +1,72 @@
+package com.odeyalo.sonata.playlists.config.security.local;
+
+import com.odeyalo.suite.security.auth.TokenAuthenticationManager;
+import com.odeyalo.suite.security.auth.token.AccessTokenMetadata;
+import com.odeyalo.suite.security.auth.token.ReactiveAccessTokenValidator;
+import com.odeyalo.suite.security.auth.token.ValidatedAccessToken;
+import org.jetbrains.annotations.NotNull;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Primary;
+import org.springframework.context.annotation.Profile;
+import org.springframework.security.authentication.ReactiveAuthenticationManager;
+import reactor.core.publisher.Mono;
+
+import java.time.Instant;
+import java.util.Map;
+
+@Configuration
+@Profile("local")
+public class AuthenticationConfiguration {
+
+    @Bean
+    @Primary
+    public ReactiveAuthenticationManager reactiveAuthenticationManager() {
+        return new TokenAuthenticationManager(
+                new LocalDevelopmentAccessTokenValidator()
+        );
+    }
+
+    private static class LocalDevelopmentAccessTokenValidator implements ReactiveAccessTokenValidator {
+        private final Map<String, ValidatedAccessToken> tokensCache = Map.of(
+                "token1", ValidatedAccessToken.valid(
+                        AccessTokenMetadata.of("123", new String[]{"read", "write"},
+                                Instant.now().getEpochSecond(),
+                                Instant.now().plusSeconds(600).getEpochSecond()
+                        )),
+                "token1_2", ValidatedAccessToken.valid(
+                        AccessTokenMetadata.of("123", new String[]{"read", "write", "playlist"},
+                                Instant.now().getEpochSecond(),
+                                Instant.now().plusSeconds(600).getEpochSecond()
+                        )),
+                "token2", ValidatedAccessToken.valid(
+                        AccessTokenMetadata.of("miku", new String[]{"read", "write", "playlist"},
+                                Instant.now().getEpochSecond(),
+                                Instant.now().plusSeconds(600).getEpochSecond()
+                        ))
+        );
+
+        private final Logger logger = LoggerFactory.getLogger(LocalDevelopmentAccessTokenValidator.class);
+
+        public LocalDevelopmentAccessTokenValidator() {
+            logger.info("Using local 'DEV' mode, cache of available access tokens to use. Token that is not exist in cache will cause HTTP 401 UNAUTHORIZED status");
+
+            tokensCache.forEach((key, value) -> {
+                final AccessTokenMetadata tokenMetadata = value.getToken();
+                logger.info("Generated access token: '{}' for user 'ID({})' with following scopes: '({})', expire after: {} seconds", key, tokenMetadata.getUserId(), tokenMetadata.getScopes(), tokenMetadata.getExpiresIn() - Instant.now().getEpochSecond());
+            });
+
+        }
+
+        @Override
+        @NotNull
+        public Mono<ValidatedAccessToken> validateToken(@NotNull final String tokenValue) {
+
+            final ValidatedAccessToken validatedAccessToken = tokensCache.getOrDefault(tokenValue, ValidatedAccessToken.invalid());
+
+            return Mono.just(validatedAccessToken);
+        }
+    }
+}

src/main/java/com/odeyalo/sonata/playlists/entity/GeneratedPlaylistEntity.java

@@ -0,0 +1,36 @@
+package com.odeyalo.sonata.playlists.entity;
+
+import com.odeyalo.sonata.suite.brokers.events.playlist.gen.GeneratedPlaylistType;
+import lombok.*;
+import lombok.experimental.FieldDefaults;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+import org.springframework.data.annotation.Id;
+import org.springframework.data.relational.core.mapping.Column;
+import org.springframework.data.relational.core.mapping.Table;
+
+import java.time.Instant;
+
+@Data
+@AllArgsConstructor(staticName = "of")
+@NoArgsConstructor
+@Builder
+@FieldDefaults(level = AccessLevel.PRIVATE)
+@Table(name = "generated_playlists")
+public class GeneratedPlaylistEntity {
+    @Id
+    @Nullable
+    Long id;
+    @NotNull
+    @Column("playlist_id")
+    String playlistId;
+    @NotNull
+    @Column("generated_for")
+    String userId;
+    @Nullable
+    @Column("playlist_type")
+    GeneratedPlaylistType generatedPlaylistType;
+    @NotNull
+    @Column("generated_at")
+    Instant generatedAt;
+}

src/main/java/com/odeyalo/sonata/playlists/repository/GeneratedPlaylistRepository.java

@@ -0,0 +1,38 @@
+package com.odeyalo.sonata.playlists.repository;
+
+import com.odeyalo.sonata.playlists.entity.GeneratedPlaylistEntity;
+import com.odeyalo.sonata.playlists.model.Playlist;
+import org.jetbrains.annotations.NotNull;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+/**
+ * Base interface to work with {@link com.odeyalo.sonata.playlists.entity.GeneratedPlaylistEntity}
+ */
+public interface GeneratedPlaylistRepository {
+    /**
+     * Save or update the given GeneratedPlaylistEntity to the repository
+     *
+     * @param playlist - GeneratedPlaylistEntity to save
+     * @return saved GeneratedPlaylistEntity
+     */
+    @NotNull
+    Mono<GeneratedPlaylistEntity> save(@NotNull GeneratedPlaylistEntity playlist);
+
+    /**
+     * Search for the GeneratedPlaylistEntity for the user
+     *
+     * @param userId - user id to use for search
+     * @return {@link Flux} with found {@link GeneratedPlaylistEntity}
+     */
+    @NotNull
+    Flux<GeneratedPlaylistEntity> findByGeneratedFor(@NotNull String userId);
+
+    /**
+     * Clear the repository. Commonly used in tests
+     *
+     * @return - empty mono
+     */
+    @NotNull
+    Mono<Void> clear();
+}

src/main/java/com/odeyalo/sonata/playlists/repository/r2dbc/R2dbcGeneratedPlaylistRepository.java

@@ -0,0 +1,36 @@
+package com.odeyalo.sonata.playlists.repository.r2dbc;
+
+import com.odeyalo.sonata.playlists.entity.GeneratedPlaylistEntity;
+import com.odeyalo.sonata.playlists.repository.GeneratedPlaylistRepository;
+import com.odeyalo.sonata.playlists.repository.r2dbc.delegate.R2dbcGeneratedPlaylistRepositoryDelegate;
+import org.jetbrains.annotations.NotNull;
+import org.springframework.stereotype.Component;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+@Component
+public final class R2dbcGeneratedPlaylistRepository implements GeneratedPlaylistRepository {
+    private final R2dbcGeneratedPlaylistRepositoryDelegate delegate;
+
+    public R2dbcGeneratedPlaylistRepository(final R2dbcGeneratedPlaylistRepositoryDelegate delegate) {
+        this.delegate = delegate;
+    }
+
+    @Override
+    @NotNull
+    public Mono<GeneratedPlaylistEntity> save(@NotNull final GeneratedPlaylistEntity playlist) {
+        return delegate.save(playlist);
+    }
+
+    @Override
+    @NotNull
+    public Flux<GeneratedPlaylistEntity> findByGeneratedFor(@NotNull final String userId) {
+        return delegate.findByUserId(userId);
+    }
+
+    @Override
+    @NotNull
+    public Mono<Void> clear() {
+        return delegate.deleteAll();
+    }
+}

src/main/java/com/odeyalo/sonata/playlists/repository/r2dbc/R2dbcImageRepository.java

@@ -1,6 +1,9 @@
 package com.odeyalo.sonata.playlists.repository.r2dbc;
 
 import com.odeyalo.sonata.playlists.entity.ImageEntity;
+import org.jetbrains.annotations.NotNull;
+import org.springframework.data.r2dbc.repository.Query;
+import org.springframework.data.repository.query.Param;
 import org.springframework.data.repository.reactive.ReactiveCrudRepository;
 import org.springframework.stereotype.Repository;
 import reactor.core.publisher.Mono;
@@ -9,4 +12,15 @@
 public interface R2dbcImageRepository extends ReactiveCrudRepository<ImageEntity, Long> {
 
     Mono<ImageEntity> findByUrl(String url);
+
+    @Query("""
+                INSERT INTO images (url, width, height)
+                VALUES (:#{#image.url}, :#{#image.width}, :#{#image.height})
+                ON CONFLICT (url) DO UPDATE
+                SET url = excluded.url
+                RETURNING *
+            """)
+    @NotNull
+    Mono<ImageEntity> upsert(@NotNull @Param("image") ImageEntity image);
+
 }

src/main/java/com/odeyalo/sonata/playlists/repository/r2dbc/callback/write/SavePlaylistImageOnMissingAfterSaveCallback.java

@@ -1,11 +1,10 @@
 package com.odeyalo.sonata.playlists.repository.r2dbc.callback.write;
 
-import com.odeyalo.sonata.playlists.entity.PlaylistImage;
 import com.odeyalo.sonata.playlists.entity.ImageEntity;
 import com.odeyalo.sonata.playlists.entity.PlaylistEntity;
+import com.odeyalo.sonata.playlists.entity.PlaylistImage;
 import com.odeyalo.sonata.playlists.repository.PlaylistImagesRepository;
 import com.odeyalo.sonata.playlists.repository.r2dbc.R2dbcImageRepository;
-import org.apache.commons.lang3.BooleanUtils;
 import org.jetbrains.annotations.NotNull;
 import org.reactivestreams.Publisher;
 import org.springframework.context.annotation.Lazy;
@@ -40,24 +39,24 @@ public Publisher<PlaylistEntity> onAfterSave(@NotNull final PlaylistEntity entit
     @NotNull
     private Mono<List<PlaylistImage>> saveImages(PlaylistEntity playlist) {
         List<ImageEntity> images = playlist.getImages();
+
         return Flux.fromIterable(images)
-                .filterWhen(this::isImageNotExist)
-                .flatMap(entity -> playlistImagesRepository.deleteAllByPlaylistId(playlist.getId()).thenReturn(entity))
-                .flatMap(r2DbcImageRepository::save)
-                .flatMap(imageEntity -> buildAndSave(playlist, imageEntity))
+                .flatMap(image -> {
+
+                    return  playlistImagesRepository.deleteAllByPlaylistId(playlist.getId())
+                            .then(Mono.defer(() -> r2DbcImageRepository.upsert(image)
+                            .flatMap(imageEntity -> buildAndSave(playlist, imageEntity))));
+                })
                 .collectList();
     }
 
     @NotNull
-    private Mono<PlaylistImage> buildAndSave(PlaylistEntity parent, ImageEntity imageEntity) {
-        PlaylistImage imageToSave = PlaylistImage.builder().imageId(imageEntity.getId()).playlistId(parent.getId()).build();
+    private Mono<PlaylistImage> buildAndSave(@NotNull final PlaylistEntity parent,
+                                             @NotNull final ImageEntity imageEntity) {
+        final PlaylistImage imageToSave = PlaylistImage.builder()
+                .imageId(imageEntity.getId())
+                .playlistId(parent.getId())
+                .build();
         return playlistImagesRepository.save(imageToSave);
     }
-
-    @NotNull
-    private Mono<Boolean> isImageNotExist(ImageEntity entity) {
-        return r2DbcImageRepository.findByUrl(entity.getUrl())
-                .hasElement()
-                .map(BooleanUtils::negate);
-    }
 }

src/main/java/com/odeyalo/sonata/playlists/repository/r2dbc/delegate/R2dbcGeneratedPlaylistRepositoryDelegate.java

@@ -0,0 +1,15 @@
+package com.odeyalo.sonata.playlists.repository.r2dbc.delegate;
+
+import com.odeyalo.sonata.playlists.entity.GeneratedPlaylistEntity;
+import org.jetbrains.annotations.NotNull;
+import org.springframework.data.r2dbc.repository.R2dbcRepository;
+import org.springframework.stereotype.Repository;
+import reactor.core.publisher.Flux;
+
+@Repository
+public interface R2dbcGeneratedPlaylistRepositoryDelegate extends R2dbcRepository<GeneratedPlaylistEntity, Long> {
+
+    @NotNull
+    Flux<GeneratedPlaylistEntity> findByUserId(@NotNull String userId);
+
+}