feat(request): add optional idempotency key for request creation

API.md

@@ -184,14 +184,18 @@ Request body:
 {
   "firstName": "John",
   "lastName": "Doe",
-  "phone": "+37061234567"
+  "phone": "+37061234567",
+  "idempotencyKey": "550e8400-e29b-41d4-a716-446655440000"
 }
 ```
 
 Validation:
 - `firstName`: required, 1..50
 - `lastName`: required, 1..50
 - `phone`: required, `+[0-9]{7,15}`
+- `idempotencyKey`: optional UUID
+
+**Idempotency**: if `idempotencyKey` is provided, the server uses it for idempotent deduplication per client. Repeating the same key with the same payload returns the existing request (202). Repeating with a different payload returns 409 Conflict.
 
 Success response (`202`):
 

CHANGELOG.md

@@ -7,6 +7,10 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and
 ## [Unreleased]
 
 ### Added
+- Optional `idempotencyKey` (UUID) field in `POST /api/clients` for idempotent request deduplication per client. Same key + same payload returns existing request (202); different payload returns 409 Conflict.
+- `auth_client_id` column on `request` table tracks the owning client from the token. `GET /api/requests/{id}` verifies ownership; old rows with `auth_client_id='unknown'` remain readable during transition.
+- `HttpMessageNotReadableException` handler in `GlobalExceptionHandler` returns 400 with localized `error.request.invalidPayload` message.
+- New error messages: `error.request.idempotencyKeyConflict`, `error.request.invalidPayload` (en/ru).
 - Reactive project documentation set aligned with `jwt-demo` structure:
   - `API.md`
   - `SECURITY.md`

src/main/java/lt/satsyuk/controller/ClientController.java

@@ -13,6 +13,7 @@
 import lt.satsyuk.dto.RequestAcceptedResponse;
 import lt.satsyuk.service.ClientService;
 import lt.satsyuk.service.RequestService;
+import lt.satsyuk.service.SecurityService;
 import lombok.RequiredArgsConstructor;
 import org.springframework.http.HttpStatus;
 import org.springframework.http.ResponseEntity;
@@ -31,10 +32,11 @@ public class ClientController {
 
     private final ClientService clientService;
     private final RequestService requestService;
+    private final SecurityService securityService;
 
     @PostMapping
     @PreAuthorize("hasRole('CLIENT_CREATE')")
-    @Operation(summary = "Create client", description = "Creates an asynchronous client creation request.")
+    @Operation(summary = "Create client", description = "Creates an asynchronous client creation request. Optional idempotencyKey (UUID) enables idempotent deduplication per client.")
     @ApiResponse(responseCode = "202", description = "Client creation request accepted",
             content = @Content(mediaType = "application/json",
                     schema = @Schema(implementation = AppResponse.class)))
@@ -46,8 +48,9 @@ public class ClientController {
             content = @Content(mediaType = "application/json"))
     public Mono<ResponseEntity<AppResponse<RequestAcceptedResponse>>> create(
             @Valid @RequestBody CreateClientRequest req) {
-        return requestService.submitClientCreateRequest(req)
-                .map(response -> ResponseEntity.status(HttpStatus.ACCEPTED).body(AppResponse.ok(response)));
+        return securityService.clientId()
+                .flatMap(clientId -> requestService.submitClientCreateRequest(req, clientId)
+                        .map(response -> ResponseEntity.status(HttpStatus.ACCEPTED).body(AppResponse.ok(response))));
     }
 
     @GetMapping("/{id}")

src/main/java/lt/satsyuk/controller/RequestController.java

@@ -9,6 +9,7 @@
 import lt.satsyuk.dto.AppResponse;
 import lt.satsyuk.dto.RequestStatusResponse;
 import lt.satsyuk.service.RequestService;
+import lt.satsyuk.service.SecurityService;
 import lombok.RequiredArgsConstructor;
 import org.springframework.security.access.prepost.PreAuthorize;
 import org.springframework.web.bind.annotation.GetMapping;
@@ -27,6 +28,7 @@
 public class RequestController {
 
     private final RequestService requestService;
+    private final SecurityService securityService;
 
     @GetMapping("/{id}")
     @PreAuthorize("hasRole('CLIENT_CREATE')")
@@ -43,8 +45,9 @@ public class RequestController {
     @ApiResponse(responseCode = "404", description = "Request not found",
             content = @Content(mediaType = "application/json"))
     public Mono<AppResponse<RequestStatusResponse>> get(@PathVariable("id") UUID id) {
-        return requestService.getRequestStatus(id)
-                .map(AppResponse::ok);
+        return securityService.clientId()
+                .flatMap(clientId -> requestService.getRequestStatus(id, clientId)
+                        .map(AppResponse::ok));
     }
 }
 

src/main/java/lt/satsyuk/dto/CreateClientRequest.java

@@ -5,6 +5,8 @@
 import jakarta.validation.constraints.Pattern;
 import jakarta.validation.constraints.Size;
 
+import java.util.UUID;
+
 public record CreateClientRequest(
 
         @NotBlank(message = "{validation.firstName.required}")
@@ -20,5 +22,9 @@ public record CreateClientRequest(
         @NotBlank(message = "{validation.phone.required}")
         @Pattern(regexp = "\\+?\\d{7,15}", message = "{validation.phone.invalid}")
         @Schema(example = "+37060000000")
-        String phone
+        String phone,
+
+        @Schema(description = "Optional idempotency key. If provided, used as request id instead of generating a new one.",
+                example = "550e8400-e29b-41d4-a716-446655440000")
+        UUID idempotencyKey
 ) {}

src/main/java/lt/satsyuk/exception/GlobalExceptionHandler.java

@@ -7,6 +7,7 @@
 import org.springframework.context.MessageSourceResolvable;
 import org.springframework.context.NoSuchMessageException;
 import org.springframework.http.HttpStatus;
+import org.springframework.http.converter.HttpMessageNotReadableException;
 import org.springframework.security.access.AccessDeniedException;
 import org.springframework.validation.method.ParameterValidationResult;
 import org.springframework.web.bind.MethodArgumentNotValidException;
@@ -230,12 +231,26 @@ public Mono<AppResponse<Void>> handlePhoneExists(PhoneAlreadyExistsException ex)
         return Mono.just(AppResponse.error(AppResponse.ErrorCode.CONFLICT.getCode(), message));
     }
 
+    @ExceptionHandler(IdempotencyKeyConflictException.class)
+    @ResponseStatus(HttpStatus.CONFLICT)
+    public Mono<AppResponse<Void>> handleIdempotencyKeyConflict(IdempotencyKeyConflictException ex) {
+        String message = messageService.getMessage(ex.getMessageCode());
+        return Mono.just(AppResponse.error(AppResponse.ErrorCode.CONFLICT.getCode(), message));
+    }
+
     @ExceptionHandler(KeycloakAuthException.class)
     public Mono<AppResponse<Void>> handleKeycloakAuthException(KeycloakAuthException ex, ServerWebExchange exchange) {
         exchange.getResponse().setStatusCode(ex.getStatus());
         return Mono.just(AppResponse.error(AppResponse.ErrorCode.UNAUTHORIZED.getCode(), ex.getKeycloakMessage()));
     }
 
+    @ExceptionHandler(HttpMessageNotReadableException.class)
+    @ResponseStatus(HttpStatus.BAD_REQUEST)
+    public Mono<AppResponse<Void>> handleHttpMessageNotReadable(HttpMessageNotReadableException ex) {
+        return Mono.just(AppResponse.error(AppResponse.ErrorCode.BAD_REQUEST.getCode(),
+                messageService.getMessage("error.request.invalidPayload")));
+    }
+
     @ExceptionHandler(Exception.class)
     @ResponseStatus(HttpStatus.INTERNAL_SERVER_ERROR)
     public Mono<AppResponse<Void>> handleGeneric(Exception ex) {

src/main/java/lt/satsyuk/exception/IdempotencyKeyConflictException.java

@@ -0,0 +1,14 @@
+package lt.satsyuk.exception;
+
+import lombok.Getter;
+
+@Getter
+public class IdempotencyKeyConflictException extends RuntimeException {
+
+    private final String messageCode;
+
+    public IdempotencyKeyConflictException(String messageCode) {
+        super(messageCode);
+        this.messageCode = messageCode;
+    }
+}

src/main/java/lt/satsyuk/model/Request.java

@@ -36,5 +36,8 @@ public class Request {
 
     @Column("response_data")
     private String responseData;
+
+    @Column("auth_client_id")
+    private String authClientId = "unknown";
 }
 

src/main/java/lt/satsyuk/repository/RequestRepository.java

@@ -13,6 +13,9 @@
 
 public interface RequestRepository extends R2dbcRepository<Request, UUID> {
 
+    @Query("SELECT * FROM request WHERE id = :id AND auth_client_id = :authClientId")
+    Mono<Request> findByIdAndAuthClientId(@Param("id") UUID id, @Param("authClientId") String authClientId);
+
     @Modifying
     @Query("""
             INSERT INTO request (
@@ -22,15 +25,17 @@ INSERT INTO request (
                 created_at,
                 status_changed_at,
                 request_data,
-                response_data
+                response_data,
+                auth_client_id
             ) VALUES (
                 :id,
                 :type,
                 :status,
                 :createdAt,
                 :statusChangedAt,
                 :requestData,
-                :responseData
+                NULL,
+                :authClientId
             )
             """)
     Mono<Integer> insertRequest(@Param("id") UUID id,
@@ -39,7 +44,7 @@ Mono<Integer> insertRequest(@Param("id") UUID id,
                                 @Param("createdAt") OffsetDateTime createdAt,
                                 @Param("statusChangedAt") OffsetDateTime statusChangedAt,
                                 @Param("requestData") String requestData,
-                                @Param("responseData") String responseData);
+                                @Param("authClientId") String authClientId);
 
     @Query("""
             WITH pending AS (
@@ -87,14 +92,16 @@ Mono<ReclaimStats> reclaimStaleClientCreateRequests(@Param("staleBefore") Offset
                                                          @Param("now") OffsetDateTime now);
 
     @Modifying
-    @Query("UPDATE request SET status = 'COMPLETED', response_data = :responseData, status_changed_at = :now WHERE id = :id AND status = 'PROCESSING'")
+    @Query("UPDATE request SET status = 'COMPLETED', response_data = :responseData, status_changed_at = :now WHERE id = :id AND auth_client_id = :authClientId AND status = 'PROCESSING'")
     Mono<Integer> markCompleted(@Param("id") UUID id,
+                                @Param("authClientId") String authClientId,
                                 @Param("responseData") String responseData,
                                 @Param("now") OffsetDateTime now);
 
     @Modifying
-    @Query("UPDATE request SET status = 'FAILED', response_data = :responseData, status_changed_at = :now WHERE id = :id AND status = 'PROCESSING'")
+    @Query("UPDATE request SET status = 'FAILED', response_data = :responseData, status_changed_at = :now WHERE id = :id AND auth_client_id = :authClientId AND status = 'PROCESSING'")
     Mono<Integer> markFailed(@Param("id") UUID id,
+                             @Param("authClientId") String authClientId,
                              @Param("responseData") String responseData,
                              @Param("now") OffsetDateTime now);
 

src/main/java/lt/satsyuk/service/RequestService.java

@@ -11,6 +11,7 @@
 import lt.satsyuk.dto.CreateClientRequest;
 import lt.satsyuk.dto.RequestAcceptedResponse;
 import lt.satsyuk.dto.RequestStatusResponse;
+import lt.satsyuk.exception.IdempotencyKeyConflictException;
 import lt.satsyuk.exception.PhoneAlreadyExistsException;
 import lt.satsyuk.exception.RequestNotFoundException;
 import lt.satsyuk.model.Request;
@@ -20,6 +21,7 @@
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.dao.DataAccessResourceFailureException;
+import org.springframework.dao.DuplicateKeyException;
 import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Service;
 import reactor.core.publisher.Mono;
@@ -106,15 +108,19 @@
     @Value("${app.request.worker.processing-timeout:2m}")
     private Duration workerProcessingTimeout;
 
-    public Mono<RequestAcceptedResponse> submitClientCreateRequest(CreateClientRequest createClientRequest) {
+    public Mono<RequestAcceptedResponse> submitClientCreateRequest(CreateClientRequest createClientRequest, String authClientId) {
+        UUID requestId = createClientRequest.idempotencyKey() != null
+                ? createClientRequest.idempotencyKey()
+                : UUID.randomUUID();
         OffsetDateTime now = now();
         Request request = Request.builder()
-                .id(UUID.randomUUID())
+                .id(requestId)
                 .type(RequestType.CLIENT_CREATE)
                 .status(RequestStatus.PENDING)
                 .createdAt(now)
                 .statusChangedAt(now)
                 .requestData(writeJson(createClientRequest))
+                .authClientId(authClientId)
                 .build();
 
         return requestRepository.insertRequest(
@@ -124,13 +130,27 @@
                         request.getCreatedAt(),
                         request.getStatusChangedAt(),
                         request.getRequestData(),
-                        null
+                        authClientId
                 )
                 .flatMap(rows -> {
                     if (rows == 1) {
                         return Mono.just(new RequestAcceptedResponse(request.getId(), request.getStatus()));
                     }
                     return Mono.error(new IllegalStateException("Failed to persist async request"));
+                })
+                .onErrorResume(DuplicateKeyException.class, ex -> {
+                    if (createClientRequest.idempotencyKey() == null) {
+                        return Mono.error(ex);
+                    }
+                    return requestRepository.findByIdAndAuthClientId(requestId, authClientId)
+                            .switchIfEmpty(Mono.error(ex))
+                            .flatMap(existing -> {
+                                if (!request.getRequestData().equals(existing.getRequestData())) {
+                                    return Mono.error(new IdempotencyKeyConflictException(
+                                            "error.request.idempotencyKeyConflict"));
+                                }
+                                return Mono.just(new RequestAcceptedResponse(existing.getId(), existing.getStatus()));
+                            });
                 });
     }
 
@@ -160,17 +180,28 @@
                 );
     }
 
-    public Mono<RequestStatusResponse> getRequestStatus(UUID requestId) {
-        return requestRepository.findById(requestId)
-                .switchIfEmpty(Mono.error(new RequestNotFoundException(requestId)))
-                .map(request -> new RequestStatusResponse(
-                        request.getId(),
-                        request.getType(),
-                        request.getStatus(),
-                        request.getCreatedAt(),
-                        request.getStatusChangedAt(),
-                        readJson(request.getResponseData())
-                ));
+    public Mono<RequestStatusResponse> getRequestStatus(UUID requestId, String authClientId) {
+        return requestRepository.findByIdAndAuthClientId(requestId, authClientId)
+                .map(request -> toStatusResponse(request))
+                .switchIfEmpty(Mono.defer(() -> requestRepository.findById(requestId)
+                        .flatMap(request -> {
+                            if ("unknown".equals(request.getAuthClientId())) {
+                                return Mono.just(toStatusResponse(request));
+                            }
+                            return Mono.error(new RequestNotFoundException(requestId));
+                        })
+                        .switchIfEmpty(Mono.error(new RequestNotFoundException(requestId)))));
+    }
+
+    private RequestStatusResponse toStatusResponse(Request request) {
+        return new RequestStatusResponse(
+                request.getId(),
+                request.getType(),
+                request.getStatus(),
+                request.getCreatedAt(),
+                request.getStatusChangedAt(),
+                readJson(request.getResponseData())
+        );
     }
 
     private Mono<Void> claimAndProcessBatch() {
@@ -248,6 +279,7 @@
 
     private Mono<Void> processClaimedRequest(Request request) {
         UUID requestId = request.getId();
+        String authClientId = request.getAuthClientId();
         long startedNanos = System.nanoTime();
         return Mono.defer(() -> {
                     if (request.getType() != RequestType.CLIENT_CREATE) {
@@ -256,14 +288,14 @@
 
                     CreateClientRequest payload = readJson(request.getRequestData(), CreateClientRequest.class);
                     return clientService.create(payload)
-                            .flatMap(clientResponse -> markCompleted(requestId, clientResponse, startedNanos));
+                            .flatMap(clientResponse -> markCompleted(requestId, authClientId, clientResponse, startedNanos));
                 })
-                .onErrorResume(ex -> markFailed(requestId, ex, startedNanos));
+                .onErrorResume(ex -> markFailed(requestId, authClientId, ex, startedNanos));
     }
 
-    private Mono<Void> markCompleted(UUID requestId, ClientResponse clientResponse, long startedNanos) {
+    private Mono<Void> markCompleted(UUID requestId, String authClientId, ClientResponse clientResponse, long startedNanos) {
         String responseJson = writeJson(AppResponse.ok(clientResponse));
-        return requestRepository.markCompleted(requestId, responseJson, now())
+        return requestRepository.markCompleted(requestId, authClientId, responseJson, now())
                 .doOnNext(updated -> {
                     if (updated == 1) {
                         completedTerminalStatusCount.increment();
@@ -279,10 +311,10 @@
     }
 
 
-    private Mono<Void> markFailed(UUID requestId, Throwable ex, long startedNanos) {
+    private Mono<Void> markFailed(UUID requestId, String authClientId, Throwable ex, long startedNanos) {
         AppResponse<Void> errorPayload = toWorkerError(ex);
         String errorJson = writeJson(errorPayload);
-        return requestRepository.markFailed(requestId, errorJson, now())
+        return requestRepository.markFailed(requestId, authClientId, errorJson, now())
                 .doOnNext(updated -> {
                     if (updated == 1) {
                         if (startedNanos > 0L) {

src/main/resources/db/migration/V5__add_request_auth_client_id.sql

@@ -0,0 +1 @@
+ALTER TABLE request ADD COLUMN auth_client_id VARCHAR(255) NOT NULL DEFAULT 'unknown';

src/main/resources/messages.properties

@@ -20,6 +20,8 @@ error.client.searchQueryTooShort=Search query must contain at least {0} characte
 error.account.notFound=Account for client id\u003D{0} not found
 error.account.optimisticLock=Too many optimistic lock retries for client id\u003D{0}
 error.request.notFound=Request with id\u003D{0} not found
+error.request.idempotencyKeyConflict=Request with the same idempotency key already exists with different payload
+error.request.invalidPayload=Invalid request payload
 error.request.schedulingFailed=Failed to schedule request processing
 error.validation.failed=Validation failed
 error.typeMismatch=Invalid value\: {0}

src/main/resources/messages_en.properties

@@ -20,6 +20,8 @@ error.client.searchQueryTooShort=Search query must contain at least {0} characte
 error.account.notFound=Account for client id\u003D{0} not found
 error.account.optimisticLock=Too many optimistic lock retries for client id\u003D{0}
 error.request.notFound=Request with id\u003D{0} not found
+error.request.idempotencyKeyConflict=Request with the same idempotency key already exists with different payload
+error.request.invalidPayload=Invalid request payload
 error.request.schedulingFailed=Failed to schedule request processing
 error.validation.failed=Validation failed
 error.typeMismatch=Invalid value: {0}