Add workers

README.md

@@ -8,5 +8,9 @@ docker compose -f kafka/docker-compose.yml up --build -d
 
 docker compose -f api/docker-compose.yml up --build -d  
 
+docker compose -f worker/docker-compose-blackwhite.yml up --build -d
+docker compose -f worker/docker-compose-rotate.yml up --build -d
+docker compose -f worker/docker-compose-imagga.yml up --build -d
+
 
 

api/.env

@@ -17,11 +17,13 @@ REDIS_CACHE_TIME=86400000
 
 # MINIO
 MINIO_BUCKET=files
+MINIO_EXPIRATION=1
 MINIO_URL=http://minio-api:9000
 MINIO_ROOT_USER=minioadmin
 MINIO_ROOT_PASSWORD=minioadmin
 MINIO_CONSOLE_PORT=9090
 MINIO_PORT=9000
+MINIO_TTL_PREFIX=temporary/
 
 # MONGO
 MONGO_INITDB_ROOT_USERNAME=admin

api/pom.xml

@@ -19,7 +19,6 @@
         <springdoc.starter.version>2.1.0</springdoc.starter.version>
         <jsonwebtoken.version>0.11.5</jsonwebtoken.version>
         <minio.version>8.5.7</minio.version>
-        <kafka-clients.version>3.7.0</kafka-clients.version>
 
 <!--        TESTS-->
         <containers.version>1.19.1</containers.version>
@@ -28,6 +27,19 @@
     </properties>
 
     <dependencies>
+
+        <dependency>
+            <groupId>com.bucket4j</groupId>
+            <artifactId>bucket4j-core</artifactId>
+            <version>8.10.1</version>
+        </dependency>
+
+        <dependency>
+            <groupId>com.bucket4j</groupId>
+            <artifactId>bucket4j-redis</artifactId>
+            <version>8.10.1</version>
+        </dependency>
+
         <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-web</artifactId>

api/src/main/java/com/github/asavershin/api/config/Bucket4jConfig.java

@@ -0,0 +1,43 @@
+package com.github.asavershin.api.config;
+
+import io.github.bucket4j.Bandwidth;
+import io.github.bucket4j.Bucket;
+import io.github.bucket4j.BucketConfiguration;
+import io.github.bucket4j.distributed.ExpirationAfterWriteStrategy;
+import io.github.bucket4j.distributed.proxy.ClientSideConfig;
+import io.github.bucket4j.distributed.proxy.ProxyManager;
+import io.github.bucket4j.redis.lettuce.cas.LettuceBasedProxyManager;
+import io.lettuce.core.RedisClient;
+import io.lettuce.core.RedisURI;
+import io.lettuce.core.api.StatefulRedisConnection;
+import io.lettuce.core.codec.ByteArrayCodec;
+import io.lettuce.core.codec.RedisCodec;
+import io.lettuce.core.codec.StringCodec;
+import org.springframework.boot.autoconfigure.data.redis.RedisProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.security.core.context.SecurityContextHolder;
+import org.springframework.security.core.userdetails.UserDetails;
+
+import java.time.Duration;
+import java.util.function.Supplier;
+
+@Configuration
+public class Bucket4jConfig {
+    @Bean
+    public RedisClient redisClient(final RedisProperties redisProperties) {
+        return RedisClient.create(
+                RedisURI.Builder.redis(redisProperties.getHost(), redisProperties.getPort())
+                        .withPassword(redisProperties.getPassword().toCharArray()).build());
+    }
+
+    @Bean
+    public ProxyManager<String> lettuceBasedProxyManager(final RedisClient redisClient) {
+        StatefulRedisConnection<String, byte[]> redisConnection = redisClient
+                .connect(RedisCodec.of(StringCodec.UTF8, ByteArrayCodec.INSTANCE));
+
+        return LettuceBasedProxyManager
+                .builderFor(redisConnection)
+                .build();
+    }
+}

api/src/main/java/com/github/asavershin/api/config/KafkaConf.java

@@ -102,6 +102,7 @@ private ProducerFactory<String, FiltersForPublisher> producerFactory(
                 StringSerializer.class);
         props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                 JsonSerializer.class);
+        props.put(JsonSerializer.ADD_TYPE_INFO_HEADERS, false);
         // Партиция одна, так что все равно как роутить
         props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,
                 RoundRobinPartitioner.class);

api/src/main/java/com/github/asavershin/api/config/properties/MinIOProperties.java

@@ -14,19 +14,24 @@ public class MinIOProperties {
      * The name of the bucket in the MinIO service.
      */
     private String bucket;
-
     /**
      * The URL of the MinIO service.
      */
     private String url;
-
     /**
      * The username for the MinIO service.
      */
     private String user;
-
     /**
      * The password for the MinIO service.
      */
     private String password;
+    /**
+     * The ttl of the MinIO objects.
+     */
+    private String expiration;
+    /**
+     * The name of the bucket with ttl.
+     */
+    private String ttlprefix;
 }

api/src/main/java/com/github/asavershin/api/domain/filter/Filter.java

@@ -10,17 +10,16 @@ public enum Filter {
     /**
      * Represents the REVERS_COLORS filter.
      */
-    REVERS_COLORS,
+    ROTATE,
 
     /**
      * Represents the CROP filter.
      */
-    CROP,
-
+    BLACKWHITE,
     /**
-     * Represents the REMOVE_BACKGROUND filter.
+     * Filter for external api imagga.
      */
-    REMOVE_BACKGROUND;
+    IMAGGA;
 
     /**
      * Converts a string representation of a filter name string

api/src/main/java/com/github/asavershin/api/infrastructure/in/controllers/ImageController.java

@@ -15,10 +15,17 @@
 import com.github.asavershin.api.infrastructure.in.controllers.dto.image.UploadImageResponse;
 import com.github.asavershin.api.infrastructure.in.security.CustomUserDetails;
 import com.github.asavershin.api.infrastructure.out.producers.KafkaProducer;
+import io.github.bucket4j.Bandwidth;
+import io.github.bucket4j.Bucket;
+import io.github.bucket4j.BucketConfiguration;
+import io.github.bucket4j.distributed.proxy.ProxyManager;
 import io.swagger.v3.oas.annotations.Operation;
 import io.swagger.v3.oas.annotations.tags.Tag;
 import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.context.annotation.Bean;
 import org.springframework.security.core.annotation.AuthenticationPrincipal;
+import org.springframework.security.core.userdetails.UserDetails;
 import org.springframework.web.bind.annotation.DeleteMapping;
 import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.PathVariable;
@@ -29,13 +36,16 @@
 import org.springframework.web.bind.annotation.RestController;
 import org.springframework.web.multipart.MultipartFile;
 
+import java.time.Duration;
 import java.util.List;
 import java.util.UUID;
+import java.util.function.Supplier;
 
 @RestController
 @RequestMapping("/api/v1/image")
 @Tag(name = "image", description = "Работа с изображениями")
 @RequiredArgsConstructor
+@Slf4j
 public class ImageController {
     /**
      * Domain query that allows you take images of specific user.
@@ -54,6 +64,10 @@ public class ImageController {
      * Service for tracking the status of image processing events.
      */
     private final GetStatusEvent statusEvent;
+    /**
+     * Using for limiting requests to imagga filter.
+     */
+    private final ProxyManager<String> pm;
 
     /**
      * Not final to allows Spring use proxy.
@@ -144,11 +158,12 @@ public byte[] downloadImage(
                 user.authenticatedUser().userId()
         );
     }
+
     /**
      * This start image processing event by authenticated user to his image.
      *
-     * @param user      Is param that injects by spring and contains
-     *                  current authenticated spring user.
+     * @param user    Is param that injects by spring and contains
+     *                current authenticated spring user.
      * @param imageId The ID of the image that will be processed.
      * @param filters The list of filters that will be applied to the image.
      * @return request id
@@ -165,7 +180,24 @@ public ApplyImageFiltersResponse applyImageFilters(
                         .toList(),
                 ImageId.fromString(imageId)
         );
-        producer.send(event, user.authenticatedUser().userId());
+        if (event.filters().contains(Filter.IMAGGA)) {
+            var b = getBucket(user);
+            var tokens = b.getAvailableTokens();
+            log.info("Available tokens {} for user {} ",
+                    tokens,
+                    user.getUsername()
+            );
+            if (tokens > 0) {
+                producer.send(event, user.authenticatedUser().userId());
+                b.tryConsume(1);
+            } else {
+                throw new RuntimeException(
+                        "You have reached your limit on \"imagga\""
+                );
+            }
+        } else {
+            producer.send(event, user.authenticatedUser().userId());
+        }
         return new ApplyImageFiltersResponse(
                 event.requestId().value().toString()
         );
@@ -198,4 +230,19 @@ public ApplyImageFiltersResponse applyImageFilters(
                         )
         );
     }
+
+    private Bucket getBucket(final UserDetails userDetails) {
+        var username = userDetails.getUsername();
+        return pm.builder().build(
+                "bucket: " + username,
+                () -> BucketConfiguration.builder()
+                        .addLimit(Bandwidth.builder().capacity(1L)
+                                .refillIntervally(1,
+                                        Duration.ofDays(1L)
+                                )
+                                .build()
+                        )
+                        .build()
+        );
+    }
 }

api/src/main/java/com/github/asavershin/api/infrastructure/out/storage/MinioServiceIml.java

@@ -9,12 +9,20 @@
 import io.minio.MinioClient;
 import io.minio.PutObjectArgs;
 import io.minio.RemoveObjectArgs;
+import io.minio.SetBucketLifecycleArgs;
+import io.minio.messages.Expiration;
+import io.minio.messages.LifecycleConfiguration;
+import io.minio.messages.LifecycleRule;
+import io.minio.messages.RuleFilter;
+import io.minio.messages.Status;
 import lombok.SneakyThrows;
 import org.apache.commons.compress.utils.IOUtils;
 import org.springframework.stereotype.Service;
 import org.springframework.web.multipart.MultipartFile;
 
 import java.io.InputStream;
+import java.time.ZonedDateTime;
+import java.util.LinkedList;
 import java.util.List;
 
 @Service
@@ -27,21 +35,23 @@ public class MinioServiceIml implements MinioService {
      * The MinIO properties from .yml.
      */
     private final MinIOProperties minioProperties;
+
     /**
      * Constructor for {@link MinioServiceIml}.
      *
-     * @param aMinioClient The {@link MinioClient}
-     *                    instance to interact with the MinIO server.
+     * @param aMinioClient     The {@link MinioClient}
+     *                         instance to interact with the MinIO server.
      * @param aMinioProperties The {@link MinIOProperties}
-     *                        instance containing the configuration
-     *                        for the MinIO server.
+     *                         instance containing the configuration
+     *                         for the MinIO server.
      */
     public MinioServiceIml(final MinioClient aMinioClient,
                            final MinIOProperties aMinioProperties) {
         this.minioClient = aMinioClient;
         this.minioProperties = aMinioProperties;
         createBucket();
     }
+
     /**
      * Not final to allow spring use proxy.
      */
@@ -66,6 +76,7 @@ public void saveFile(final MultipartFile image, final String filename) {
         }
         saveFile(inputStream, filename);
     }
+
     /**
      * Not final to allow spring use proxy.
      */
@@ -77,13 +88,14 @@ public byte[] getFile(final String link) {
         try {
             return IOUtils.toByteArray(
                     minioClient.getObject(GetObjectArgs.builder()
-                    .bucket(minioProperties.getBucket())
-                    .object(link)
-                    .build()));
+                            .bucket(minioProperties.getBucket())
+                            .object(link)
+                            .build()));
         } catch (Exception e) {
             throw new FileException("File download failed: " + e.getMessage());
         }
     }
+
     /**
      * Not final to allow spring use proxy.
      */
@@ -110,14 +122,35 @@ public void deleteFiles(final List<String> links) {
 
     @SneakyThrows
     private void createBucket() {
-        boolean found = minioClient.bucketExists(BucketExistsArgs.builder()
-                .bucket(minioProperties.getBucket())
-                .build());
+        boolean found = bucketExists(minioProperties.getBucket());
         if (!found) {
+            List<LifecycleRule> rules = new LinkedList<>();
+            rules.add(
+                    new LifecycleRule(
+                            Status.ENABLED,
+                            null,
+                            new Expiration((ZonedDateTime) null,
+                                    Integer.valueOf(
+                                            minioProperties.getExpiration()
+                                    ),
+                                    null),
+                            new RuleFilter(minioProperties.getTtlprefix()),
+                            "rule1",
+                            null,
+                            null,
+                            null));
             minioClient.makeBucket(MakeBucketArgs.builder()
                     .bucket(minioProperties.getBucket())
                     .build());
+            minioClient.setBucketLifecycle(
+                    SetBucketLifecycleArgs.builder().bucket(
+                                    minioProperties.getBucket()
+                            )
+                            .config(
+                                    new LifecycleConfiguration(rules)
+                            ).build());
         }
+
     }
 
     @SneakyThrows

api/src/main/resources/application.yml

@@ -54,6 +54,8 @@ minio:
   url: ${MINIO_URL}
   user: ${MINIO_ROOT_USER}
   password: ${MINIO_ROOT_PASSWORD}
+  expiration: ${MINIO_EXPIRATION}
+  ttlprefix: ${MINIO_TTL_PREFIX}
   bucket: ${MINIO_BUCKET}
   console-port: ${MINIO_CONSOLE_PORT}
   port: ${MINIO_PORT}