implement trainer registration functionality

src/main/java/com/security/config/kafka/KafkaProducerConfig.java

@@ -48,7 +48,11 @@ Map<String, Object> producerConfigs() {
         config.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, requestTimeout);
         config.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, idempotence);
         config.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, inflightRequests);
-        config.put(JsonSerializer.TYPE_MAPPINGS, "NotificationEvent:com.security.events.notification.NotificationEvent,CreatedUserEvent:com.security.events.notification.CreatedUserEvent");
+        config.put(JsonSerializer.TYPE_MAPPINGS,
+                "NotificationEvent:com.security.events.notification.NotificationEvent," +
+                "CreatedUserEvent:com.security.events.notification.CreatedUserEvent," +
+                "TrainerCreatedEvent:com.security.events.classes.TrainerCreatedEvent"
+        );
         config.put(ProducerConfig.RETRIES_CONFIG, 10);
         return config;
     }
@@ -72,5 +76,14 @@ NewTopic createNotificationTopic() {
                 .configs(Map.of("min.insync.replicas", "1"))
                 .build();
     }
+    @Bean
+    NewTopic createNotificationTrainerTopic() {
+        return TopicBuilder
+                .name("trainer-created-event-topic")
+                .partitions(1)
+                .replicas(1)
+                .configs(Map.of("min.insync.replicas", "1"))
+                .build();
+    }
 
 }

src/main/java/com/security/controllers/AdminUserController.java

@@ -6,12 +6,17 @@
 import com.security.annotations.AdminAccess;
 import com.security.dtos.roles.RoleDetailsDto;
 import com.security.dtos.roles.RoleStatisticsDto;
+import com.security.dtos.trainer.TrainerRegistrationRequest;
+import com.security.dtos.trainer.TrainerRegistrationResponse;
+import com.security.services.TrainerService;
 import com.security.services.UserAccountService;
 import com.security.services.UserRoleService;
 import io.swagger.v3.oas.annotations.Operation;
 import io.swagger.v3.oas.annotations.tags.Tag;
 import jakarta.validation.Valid;
 import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.http.HttpStatus;
 import org.springframework.http.ResponseEntity;
 import org.springframework.security.core.Authentication;
 import org.springframework.web.bind.annotation.*;
@@ -23,11 +28,13 @@
 @RestController
 @RequestMapping("/admin/users")
 @RequiredArgsConstructor
+@Slf4j
 @Tag(name = "Autorizacion", description = "Endpoints para manejo de roles")
 public class AdminUserController {
 
     private final UserRoleService userRoleService;
     private final UserAccountService userAccountService;
+    private final TrainerService trainerService;
 
     @Operation(summary = "Listar roles de usuario")
     @GetMapping("/{id}/roles")
@@ -94,4 +101,12 @@ public ResponseEntity<Map<String, Object>> getUserProviderStatistics() {
         return ResponseEntity.ok(userAccountService.getUserStatistics());
     }
 
+    @PostMapping("/register-trainer")
+    @AdminAccess
+    public ResponseEntity<TrainerRegistrationResponse> registerTrainer(
+            @Valid @RequestBody TrainerRegistrationRequest request
+    ) {
+        return ResponseEntity.status(HttpStatus.CREATED).body(trainerService.registerTrainer(request));
+    }
+
 }

src/main/java/com/security/dtos/trainer/TrainerRegistrationRequest.java

@@ -0,0 +1,35 @@
+package com.security.dtos.trainer;
+
+import jakarta.validation.constraints.Email;
+import jakarta.validation.constraints.NotBlank;
+import jakarta.validation.constraints.Pattern;
+import jakarta.validation.constraints.Size;
+import lombok.*;
+
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class TrainerRegistrationRequest {
+
+    @NotBlank(message = "El email es obligatorio")
+    @Email(message = "Email inválido")
+    private String email;
+
+    @NotBlank(message = "La contraseña es obligatoria")
+    @Size(min = 8, message = "La contraseña debe tener al menos 8 caracteres")
+    private String password;
+
+    @NotBlank(message = "El nombre es obligatorio")
+    private String firstName;
+
+    @NotBlank(message = "El apellido es obligatorio")
+    private String lastName;
+
+    @NotBlank(message = "El DNI es obligatorio")
+    private String dni;
+
+    @NotBlank(message = "El teléfono es obligatorio")
+//    @Pattern(regexp = "^[+]?[0-9]{9,15}$", message = "Teléfono inválido")
+    private String phone;
+}

src/main/java/com/security/dtos/trainer/TrainerRegistrationResponse.java

@@ -0,0 +1,14 @@
+package com.security.dtos.trainer;
+
+import lombok.*;
+import java.util.UUID;
+
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class TrainerRegistrationResponse {
+    private UUID id;
+    private String email;
+    private String message;
+}

src/main/java/com/security/events/classes/TrainerCreatedEvent.java

@@ -0,0 +1,40 @@
+package com.security.events.classes;
+
+import com.fasterxml.jackson.annotation.JsonFormat;
+import lombok.*;
+
+import java.io.Serializable;
+import java.time.Instant;
+import java.util.UUID;
+
+@Builder
+public record TrainerCreatedEvent(
+        UUID id,
+        String email,
+        String firstName,
+        String lastName,
+        String dni,
+        String phone,
+
+        @JsonFormat(shape = JsonFormat.Shape.STRING, pattern = "yyyy-MM-dd'T'HH:mm:ss.SSSZ", timezone = "UTC")
+        Instant occurredAt,
+
+        String eventId,
+        String eventType
+) implements Serializable {
+
+    public static TrainerCreatedEvent create(UUID userId, String email, String firstName,
+                                             String lastName, String dni, String phone) {
+        return TrainerCreatedEvent.builder()
+                .id(userId)
+                .email(email)
+                .firstName(firstName)
+                .lastName(lastName)
+                .dni(dni)
+                .phone(phone)
+                .occurredAt(Instant.now())
+                .eventId(UUID.randomUUID().toString())
+                .eventType("TRAINER_CREATED")
+                .build();
+    }
+}

src/main/java/com/security/services/Impl/TrainerServiceImpl.java

@@ -0,0 +1,101 @@
+package com.security.services.Impl;
+
+import com.security.dtos.trainer.TrainerRegistrationRequest;
+import com.security.dtos.trainer.TrainerRegistrationResponse;
+import com.security.entity.RoleEntity;
+import com.security.entity.UserEntity;
+import com.security.enums.AuthProvider;
+import com.security.events.classes.TrainerCreatedEvent;
+import com.security.repository.RoleRepository;
+import com.security.repository.UserRepository;
+import com.security.services.TrainerService;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.common.errors.DuplicateResourceException;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.security.crypto.password.PasswordEncoder;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.util.Set;
+import java.util.UUID;
+
+@Service
+@RequiredArgsConstructor
+@Slf4j
+public class TrainerServiceImpl implements TrainerService {
+    private final UserRepository userRepository;
+    private final RoleRepository roleRepository;
+    private final PasswordEncoder passwordEncoder;
+    private final KafkaTemplate<String, Object> kafkaTemplate;
+
+    private static final String TRAINER_ROLE = "TRAINER";
+
+
+    @Transactional
+    @Override
+    public TrainerRegistrationResponse registerTrainer(TrainerRegistrationRequest request) {
+        log.info("Iniciando registro de trainer con email: {}", request.getEmail());
+
+        if (userRepository.existsByEmail(request.getEmail())) {
+            throw new DuplicateResourceException("El email ya está registrado");
+        }
+
+        RoleEntity trainerRole = roleRepository.findByName(TRAINER_ROLE)
+                .orElseThrow(() -> new IllegalStateException("Rol TRAINER no encontrado en el sistema"));
+
+        UserEntity user = UserEntity.builder()
+                .email(request.getEmail())
+                .password(passwordEncoder.encode(request.getPassword()))
+                .provider(AuthProvider.LOCAL)
+                .enabled(true)
+                .accountNonExpired(true)
+                .accountNonLocked(true)
+                .credentialsNonExpired(true)
+                .roles(Set.of(trainerRole))
+                .build();
+
+        UserEntity savedUser = userRepository.save(user);
+        log.info("Usuario trainer creado con ID: {}", savedUser.getId());
+
+        publishTrainerCreatedEvent(savedUser, request);
+
+        return TrainerRegistrationResponse.builder()
+                .id(savedUser.getId())
+                .email(savedUser.getEmail())
+                .message("Trainer registrado exitosamente")
+                .build();
+    }
+
+    private void publishTrainerCreatedEvent(UserEntity user, TrainerRegistrationRequest request) {
+        TrainerCreatedEvent event = TrainerCreatedEvent.create(
+                user.getId(),
+                user.getEmail(),
+                request.getFirstName(),
+                request.getLastName(),
+                request.getDni(),
+                request.getPhone()
+        );
+
+        kafkaTemplate.send("trainer-created-event-topic", event)
+                .whenComplete((result, ex) -> {
+                    if (ex == null) {
+                        log.info("Evento publicado correctamente - userId: {} , partition {} ,offset: {}",
+                                user.getId(),
+                                result.getRecordMetadata().partition(),
+                                result.getRecordMetadata().offset());
+                    } else {
+                        log.error("Error al publicar evento para userId {}", user.getId(), ex);
+                    }
+                });
+    }
+
+    public void compensateTrainerCreation(UUID userId) {
+        log.warn("Compensando creación de trainer para userId: {}", userId);
+        userRepository.findById(userId).ifPresent(user -> {
+            user.setEnabled(false);
+            userRepository.save(user);
+            log.info("Usuario deshabilitado como compensación: {}", userId);
+        });
+    }
+}

src/main/java/com/security/services/TrainerService.java

@@ -0,0 +1,8 @@
+package com.security.services;
+
+import com.security.dtos.trainer.TrainerRegistrationRequest;
+import com.security.dtos.trainer.TrainerRegistrationResponse;
+
+public interface TrainerService {
+    TrainerRegistrationResponse registerTrainer(TrainerRegistrationRequest request);
+}