Refactor Keycloak client, user repos, gRPC impl; add Kafka consumer c…

src/main/java/com/isums/userservice/configurations/KafkaConsumerConfig.java

@@ -0,0 +1,70 @@
+package com.isums.userservice.configurations;
+
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.core.DefaultKafkaProducerFactory;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.listener.DeadLetterPublishingRecoverer;
+import org.springframework.kafka.listener.DefaultErrorHandler;
+import org.springframework.util.backoff.ExponentialBackOff;
+
+import java.util.Map;
+
+@Configuration
+public class KafkaConsumerConfig {
+
+    @Value("${spring.kafka.bootstrap-servers}")
+    private String bootstrapServers;
+
+    @Bean
+    public KafkaTemplate<String, Object> objectKafkaTemplate() {
+        Map<String, Object> props = Map.of(
+                ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers,
+                ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class,
+                ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
+                org.springframework.kafka.support.serializer.JsonSerializer.class,
+                org.springframework.kafka.support.serializer.JsonSerializer.ADD_TYPE_INFO_HEADERS, false
+        );
+        return new KafkaTemplate<>(new DefaultKafkaProducerFactory<>(props));
+    }
+
+    @Bean
+    public KafkaTemplate<String, String> dltKafkaTemplate() {
+        Map<String, Object> props = Map.of(
+                ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers,
+                ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class,
+                ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class
+        );
+        return new KafkaTemplate<>(new DefaultKafkaProducerFactory<>(props));
+    }
+
+    @Bean
+    public DefaultErrorHandler kafkaErrorHandler(KafkaTemplate<String, String> dltKafkaTemplate) {
+        DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(
+                dltKafkaTemplate,
+                (record, ex) -> new TopicPartition(record.topic() + ".DLT", record.partition())
+        );
+
+        ExponentialBackOff backOff = new ExponentialBackOff(1_000L, 2.0);
+        backOff.setMaxInterval(60_000L);
+        backOff.setMaxAttempts(Long.MAX_VALUE);
+
+        DefaultErrorHandler handler = new DefaultErrorHandler(recoverer, backOff);
+
+        handler.addNotRetryableExceptions(
+                com.fasterxml.jackson.core.JsonProcessingException.class,
+                com.fasterxml.jackson.databind.exc.InvalidDefinitionException.class,
+                com.fasterxml.jackson.databind.exc.UnrecognizedPropertyException.class,
+                IllegalArgumentException.class,
+                org.springframework.messaging.converter.MessageConversionException.class,
+                org.springframework.dao.DataIntegrityViolationException.class,
+                org.hibernate.exception.ConstraintViolationException.class
+        );
+
+        return handler;
+    }
+}

src/main/java/com/isums/userservice/domains/events/CreateUserPlacedEvent.java

@@ -36,5 +36,7 @@ public class CreateUserPlacedEvent {
     private String nationality;
     private String visaType;
     private String visaExpiryDate;
+
+    private String language;
 }
 

src/main/java/com/isums/userservice/domains/events/UserActivatedEvent.java

@@ -17,6 +17,7 @@ public record UserActivatedEvent(
         String email,
         String name,
         String password,
+        String locale,
         String firstRentPaymentUrl,
         Long firstRentAmount,
         Instant firstRentDueDate

src/main/java/com/isums/userservice/infrastructures/abstracts/KeycloakClient.java

@@ -2,7 +2,7 @@
 
 import com.isums.userservice.domains.dtos.KeycloakCreateUserRequest;
 
-import java.util.List;
+import java.util.Optional;
 
 public interface KeycloakClient {
     String createUser(KeycloakCreateUserRequest req);
@@ -13,15 +13,7 @@ public interface KeycloakClient {
 
     String activateAndResetPassword(String keycloakId);
 
-    /**
-     * Trigger Keycloak's "execute actions" email flow — sends a one-click
-     * link to the user so they can perform required actions (typically
-     * UPDATE_PASSWORD + VERIFY_EMAIL) before first login. Keycloak realm
-     * SMTP must be configured for this to actually deliver.
-     *
-     * @param keycloakId  Keycloak user UUID
-     * @param actions     required actions to perform (e.g. ["UPDATE_PASSWORD"])
-     * @param lifespanSec link TTL in seconds (null = Keycloak default ≈ 12h)
-     */
-    void sendExecuteActionsEmail(String keycloakId, List<String> actions, Integer lifespanSec);
+    Optional<String> findUserIdByEmail(String email);
+
+    boolean isUserEnabled(String keycloakId);
 }

src/main/java/com/isums/userservice/infrastructures/abstracts/UserService.java

@@ -1,6 +1,7 @@
 package com.isums.userservice.infrastructures.abstracts;
 
 import com.isums.userservice.domains.dtos.*;
+import com.isums.userservice.domains.events.CreateUserPlacedEvent;
 import com.isums.userservice.domains.events.DepositPaidEvent;
 
 import java.util.List;
@@ -12,6 +13,8 @@ public interface UserService {
 
     String createUser(KeycloakCreateUserRequest req);
 
+    void applyProfileFromEvent(CreateUserPlacedEvent event);
+
     UserDto getUserByEmail(String email);
 
     UserProfileDto getMe(String keycloakId);