server:
  servlet:
    context-path: /api/v1/user/
  port: 3000

spring:
  datasource:
    url: jdbc:postgresql://localhost:5432/habr
    username: postgres
    password: postgres
    driver-class-name: org.postgresql.Driver
  jpa:
    hibernate:
      ddl-auto: update
  jackson:
    default-property-inclusion: non_default
  kafka:
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
      bootstrap-servers: localhost:9092
      properties:
        spring:
          json:
            add:
              type:
                headers: false

@Data
@NoArgsConstructor
public class UserDTO {
    private String name;
    private String surname;
    private String email;
    private LocalDateTime time = LocalDateTime.now();


    public UserDTO(String name, String surname, String email) {
        this.name = name;
        this.surname = surname;
        this.email = email;
    }
}

Представленный код представляет интерфейс Base64Service и его имплементацию, который используется для работы с кодированием и декодированием данных.

public interface Base64Service {

    <T> T decode(String data, Class<T> to);

    <T> String encode(T t);

}

@Service
public class Base64ServiceImpl implements Base64Service {


    private final ObjectMapper objectMapper;

    public Base64ServiceImpl(
            @Qualifier("customObjectMapper") ObjectMapper objectMapper
    ) {
        this.objectMapper = objectMapper;
    }

    @Override
    public <T> T decode(String data, Class<T> to) {
        try {

            byte[] decodedBytes = Base64.getDecoder().decode(data);

            String jsonData = new String(decodedBytes);

            return objectMapper.readValue(jsonData,to);

        } catch (Exception e) {
            throw new Base64OperationException("Failed to decode or convert the data", e);
        }
    }

    @Override
    public <T> String encode(T t) {
        String jsonData = null;
        try {
            jsonData = objectMapper.writeValueAsString(t);
        } catch (JsonProcessingException e) {
            throw new Base64OperationException(e.getMessage());
        }
        return Base64.getEncoder().encodeToString(jsonData.getBytes());
    }
}

Kafka Producerопределяет метод для отправки сообщений с использованием KafkaTemplate.

public interface KafkaProducer {
    <T> void produce(String topic, T t);
}

@Component
@Slf4j
@RequiredArgsConstructor
public class DefaultKafkaProducer implements KafkaProducer {

    private final KafkaTemplate<String, Object> kafkaTemplate;

    @Override
    public <T> void produce(String topic, T t) {
        kafkaTemplate.send(
                topic, t
        ).whenComplete((res, th) -> {
            log.info("produced message: " + res.getProducerRecord() + " topic: " + res.getProducerRecord().topic());
        });
    }

}

@RequiredArgsConstructor
@RestController
@CrossOrigin(origins = "*")
public class UserController {

    private final UserService userService;

    @PostMapping("register")
    ResponseEntity<?> requestToRegistration(
            @RequestBody UserDTO userDTO
    ) {
        return ResponseEntity
                .ok(userService.requestToRegistration(userDTO));
    }

    @PostMapping("confirm-registration")
    ResponseEntity<?> confirm(
            @RequestParam String data
    ) {
        return ResponseEntity
                .status(201)
                .body(userService.confirmRegistration(data));
    }


}

public interface UserService {

    StatusResponse requestToRegistration(UserDTO userDTO);

    User confirmRegistration(String hash);
}

@Service
@RequiredArgsConstructor
public class UserServiceImpl implements UserService {

    private final UserRepository userRepository;
    private final Base64Service base64;
    private final KafkaProducer kafkaProducer;

    public static final String EMAIL_TOPIC = "email_message";

    @Override
    public StatusResponse requestToRegistration(UserDTO userDTO) {
        try {
            var optionalUser = this.userRepository.findByEmail(userDTO.getEmail());

            if (optionalUser.isPresent()) {
                throw new EmailRegisteredException("email: %s registered yet".formatted(userDTO.getEmail()));
            }

            var dataToSend = base64.encode(userDTO);

            kafkaProducer.produce(EMAIL_TOPIC, new KafkaEmailMessageDTO(userDTO.getEmail(), dataToSend));

            return new StatusResponse(
                    true, null
            );
        } catch (Exception e) {
            return new StatusResponse(
                    false,
                    e.getMessage()
            );
        }
    }

    @Override
    public User confirmRegistration(String hash) {

        var userDTO = base64.decode(hash, UserDTO.class);

        if (userDTO.getTime().isBefore(LocalDateTime.now().minusDays(1))) {
            throw new LinkExpiredException();
        }

        var user = new User(
                userDTO.getName(),
                userDTO.getSurname(),
                userDTO.getEmail()
        );

        return this.userRepository.save(user);
    }


}

Если попытаться перейти по данной ссылке через 24 часа после её отправки, она не будет валидной.

spring:
  mail:
    host: smtp.gmail.com
    port: 587
    username: ${SMTP_USERNAME}
    password: ${SMTP_PASSWORD}
    properties:
      mail:
        smtp:
          auth: true
        smtp.starttls.enable: true
  kafka:
    consumer:
      bootstrap-servers: localhost:9092
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.ByteArrayDeserializer
      properties:
        spring:
          json:
            add:
              type:
                headers: false
server:
  frontend-url: http://localhost:4200

@Component
@Slf4j
@RequiredArgsConstructor
public class MailListener {

    private final MailService mailService;

    @KafkaListener(
            topics = "email_message", groupId = "some"
    )
    void listen(
            KafkaMailMessage kafkaMailMessage
    ) {
        log.info("email message: {} ", kafkaMailMessage);
        mailService.send(kafkaMailMessage, MessageMode.EMAIL_VERIFICATION);
    }
}

public interface MailService {
    void send(KafkaMailMessage kafkaMailMessage, MessageMode mode);
}

@Component
@RequiredArgsConstructor
@Slf4j
public class MailServiceImpl implements MailService {

    private final JavaMailSender mailSender;

    @Value("${server.frontend-url}")
    private String frontEndURL;

    @Override
    public void send(KafkaMailMessage kafkaMailMessage, MessageMode mode) {
        var msg = new SimpleMailMessage();

        if (mode == MessageMode.EMAIL_VERIFICATION) {
            msg.setText(frontEndURL + "/verification?data=" + kafkaMailMessage.message());
        } else {
            msg.setText(kafkaMailMessage.message());
        }

        msg.setTo(kafkaMailMessage.email());
        msg.setFrom("habrexample@gmail.com");

        try {
            mailSender.send(msg);
            log.info("email send, msg: {}, mode: {}", kafkaMailMessage, mode);
        } catch (Exception e) {
            log.error("send mail error : {}", e.getMessage());
        }


    }
}

@Component({
  selector: 'app-registration',
  templateUrl: './registration.component.html',
  styleUrls: ['./registration.component.css']
})
export class RegistrationComponent {

  registrationForm: FormGroup;

  constructor(
    private userService: UserService,
    private formBuilder: FormBuilder
  ) {
    this.registrationForm = this.formBuilder.group({
      name: ['', Validators.required],
      surname: ['', Validators.required],
      email: ['', [Validators.required, Validators.email]],
    });
  }

  register() {
    let name = this.findInRegistrationForm('name')
    let surname = this.findInRegistrationForm('surname')
    let email = this.findInRegistrationForm('email')

    let userDTO = {name, surname, email}

    this.userService.requestToRegistration(userDTO)
      .subscribe((res: StatusResponse) => {
          alert(JSON.stringify(res))
        }
      )
  }

  private findInRegistrationForm(
    controlName: string
  ) {
    return this.registrationForm.get(controlName)?.value as string
  }


}

import {Component, OnInit} from '@angular/core';
import {ActivatedRoute, Router} from "@angular/router";
import {User} from "../../model/User";
import {UserService} from "../../service/user.service";

@Component({
  selector: 'app-verification',
  templateUrl: './verification.component.html',
  styleUrls: ['./verification.component.css']
})
export class VerificationComponent implements OnInit {

  constructor(
    private route: ActivatedRoute,
    private userService: UserService
  ) {
  }

  user: User

  ngOnInit(): void {
    this.route.queryParams.subscribe(params => {
      let data = params['data'] || null;

      if (data) {
        this.userService.confirmRegistration(data)
          .subscribe(res => {
            this.user = res
            console.log(res)
          }, err => {
            if (err) {
              alert('invalid confirmation link');
            }
          })
      } else {
        alert('missing data')
      }
    })
  }


}

@Injectable({
  providedIn: 'root'
})
export class UserService {

  private http = inject(HttpClient)

  private BASE_URL = 'http://localhost:3000/api/v1/user';

  requestToRegistration(
    userDTO: UserDTO
  ): Observable<any> {
    return this.http
      .post(`${this.BASE_URL}/register`, userDTO);
  }

  confirmRegistration(
    data: string
  ): Observable<any> {
    return this.http
      .post(`${this.BASE_URL}/confirm-registration?data=` + data, {});
  }

}