Email verification in Spring Boot

@Configuration
public class ZapBounceConfig {

    @Bean
    RestClient zapBounceClient(RestClient.Builder builder,
                               @Value("${zapbounce.key}") String key) {
        var factory = new SimpleClientHttpRequestFactory();
        factory.setConnectTimeout(Duration.ofSeconds(3));
        factory.setReadTimeout(Duration.ofSeconds(12));   // no default. Set it.

        return builder
                .baseUrl("https://api.zapbounce.com/v1")
                .defaultHeader(HttpHeaders.AUTHORIZATION, "Bearer " + key)
                .requestFactory(factory)
                .build();
    }
}

@Service
public class VerificationService {

    private final RestClient client;

    public VerificationService(RestClient client) { this.client = client; }

    public VerifyResult verify(String email) {
        return client.post()
                .uri("/verify")
                .body(Map.of("email", email))
                .retrieve()
                .onStatus(HttpStatusCode::isError, (req, res) -> {
                    throw ZapBounceException.from(res);
                })
                .body(VerifyResult.class);
    }
}

// Jackson maps snake_case from the property below, so no per-field annotations.
public record VerifyResult(String email, String result, String reason,
                           String smtpCode, boolean billed) {}
// application.yaml: spring.jackson.property-naming-strategy: SNAKE_CASE

@Service
public class BatchService {

    private final RestClient client;
    private final ContactRepository contacts;

    /** Returns immediately; the webhook delivers completion. */
    @Async("verificationExecutor")
    public CompletableFuture<String> submitList(long listId) {
        List<String> emails = contacts.findUnverifiedEmails(listId);

        var created = client.post()
                .uri("/batches")
                .header("Idempotency-Key", "list-" + listId)
                .body(new BatchRequest("list-" + listId, emails, webhookUrl))
                .retrieve()
                .body(BatchCreated.class);

        contacts.recordBatchId(listId, created.batchId());
        return CompletableFuture.completedFuture(created.batchId());
    }

    /** Cursor loop, writing in chunks so a million rows never land in one list. */
    public void importResults(String batchId) {
        String cursor = null;
        do {
            String uri = "/batches/" + batchId + "/results?limit=1000"
                       + (cursor == null ? "" : "&cursor=" + URLEncoder.encode(cursor, UTF_8));

            ResultPage page = client.get().uri(uri).retrieve().body(ResultPage.class);

            contacts.applyVerdicts(page.data());
            cursor = page.hasMore() ? page.nextCursor() : null;
        } while (cursor != null);
    }
}

@Bean("verificationExecutor")
ThreadPoolTaskExecutor verificationExecutor() {
    var executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(4);
    executor.setQueueCapacity(100);
    executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
    return executor;
}

@RestController
public class ZapBounceWebhookController {

    @Value("${zapbounce.webhook-secret}")
    private String secret;

    // @RequestBody byte[] keeps the raw bytes. Bind to a DTO and Jackson has
    // already consumed the stream that was signed.
    @PostMapping("/webhooks/zapbounce")
    ResponseEntity<Void> receive(@RequestBody byte[] raw,
                                 @RequestHeader("ZapBounce-Signature") String signature)
            throws Exception {

        Map<String, String> parts = Arrays.stream(signature.split(","))
                .map(p -> p.split("=", 2))
                .collect(toMap(a -> a[0], a -> a[1]));

        if (Math.abs(Instant.now().getEpochSecond() - Long.parseLong(parts.get("t"))) > 300) {
            return ResponseEntity.status(HttpStatus.UNAUTHORIZED).build();
        }

        Mac mac = Mac.getInstance("HmacSHA256");
        mac.init(new SecretKeySpec(secret.getBytes(UTF_8), "HmacSHA256"));
        mac.update((parts.get("t") + ".").getBytes(UTF_8));
        String expected = HexFormat.of().formatHex(mac.doFinal(raw));

        if (!MessageDigest.isEqual(expected.getBytes(UTF_8), parts.get("v1").getBytes(UTF_8))) {
            return ResponseEntity.status(HttpStatus.UNAUTHORIZED).build();
        }

        Event event = mapper.readValue(raw, Event.class);

        // Deliveries repeat. A unique key on event id makes a replay harmless.
        if (events.insertIfAbsent(event.id())) {
            batchService.importResults(event.data().batchId());
        }
        return ResponseEntity.noContent().build();
    }
}