[Fix] 중복 요청 시 동일 Event ID를 반환하도록 Outbox 멱등성 보장

backend/src/main/java/com/opensource/docgrid/domain/sync/repository/SyncOutboxEventRepository.java

@@ -11,6 +11,7 @@
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.jpa.repository.JpaRepository;
 import org.springframework.data.jpa.repository.Lock;
+import org.springframework.data.jpa.repository.Modifying;
 import org.springframework.data.jpa.repository.Query;
 import org.springframework.data.repository.query.Param;
 
@@ -55,6 +56,38 @@ Page<SyncOutboxEvent> findAdminEvents(
 
     Optional<SyncOutboxEvent> findByIdempotencyKey(String idempotencyKey);
 
+    /**
+     * 동일 비즈니스 변경을 여러 Transaction이 동시에 기록해도 최초 Event 한 건만 생성한다.
+     *
+     * <p>Unique 예외를 잡은 Transaction은 이미 rollback-only가 되므로 PostgreSQL의 원자적
+     * {@code ON CONFLICT DO NOTHING}을 사용하고 호출자가 같은 Key의 확정 행을 다시 읽는다.
+     */
+    @Modifying(flushAutomatically = true)
+    @Query(value = """
+        INSERT INTO sync_outbox_events (
+            event_id, idempotency_key, aggregate_type, aggregate_id, aggregate_version,
+            event_type, payload_json, status, available_at, occurred_at,
+            retry_count, max_retry_count, created_at, updated_at
+        ) VALUES (
+            :eventId, :idempotencyKey, :aggregateType, :aggregateId, :aggregateVersion,
+            :eventType, :payloadJson, 'PENDING', :availableAt, :occurredAt,
+            0, :maxRetryCount, :occurredAt, :occurredAt
+        )
+        ON CONFLICT (idempotency_key) DO NOTHING
+        """, nativeQuery = true)
+    int insertPendingIfAbsent(
+        @Param("eventId") UUID eventId,
+        @Param("idempotencyKey") String idempotencyKey,
+        @Param("aggregateType") String aggregateType,
+        @Param("aggregateId") Long aggregateId,
+        @Param("aggregateVersion") Long aggregateVersion,
+        @Param("eventType") String eventType,
+        @Param("payloadJson") String payloadJson,
+        @Param("availableAt") LocalDateTime availableAt,
+        @Param("occurredAt") LocalDateTime occurredAt,
+        @Param("maxRetryCount") int maxRetryCount
+    );
+
     /**
      * 현재 실행 가능한 가장 오래된 PENDING Event 한 건을 다른 Dispatcher의 잠금을 기다리지 않고 Claim한다.
      */

backend/src/main/java/com/opensource/docgrid/domain/sync/service/command/SyncEventWriter.java

@@ -2,6 +2,7 @@
 
 import java.time.Clock;
 import java.time.LocalDateTime;
+import java.util.Objects;
 import java.util.UUID;
 
 import org.springframework.stereotype.Service;
@@ -15,6 +16,8 @@
 import com.opensource.docgrid.domain.sync.enums.SyncEventType;
 import com.opensource.docgrid.domain.sync.enums.SyncPermissionOperation;
 import com.opensource.docgrid.domain.sync.repository.SyncOutboxEventRepository;
+import com.opensource.docgrid.global.exception.DocGridException;
+import com.opensource.docgrid.global.exception.ErrorCode;
 
 import lombok.RequiredArgsConstructor;
 
@@ -52,19 +55,14 @@ public SyncOutboxEvent recordDocumentVersionCreated(
         String payloadJson = String.format("{\"embeddingModelId\":%d}", embeddingModel.getId());
 
         // Version·Job 생성 Transaction과 함께 Commit돼야 Dispatcher가 부분 상태를 관측하지 않는다.
-        return syncOutboxEventRepository.save(
-            SyncOutboxEvent.builder()
-                .eventId(UUID.randomUUID())
-                .idempotencyKey(idempotencyKey)
-                .aggregateType(SyncAggregateType.DOCUMENT_VERSION)
-                .aggregateId(documentVersion.getId())
-                .aggregateVersion((long) documentVersion.getVersionNo())
-                .eventType(SyncEventType.DOCUMENT_VERSION_CREATED)
-                .payloadJson(payloadJson)
-                .occurredAt(occurredAt)
-                .availableAt(occurredAt)
-                .maxRetryCount(DEFAULT_MAX_RETRY_COUNT)
-                .build()
+        return saveEvent(
+            idempotencyKey,
+            SyncAggregateType.DOCUMENT_VERSION,
+            documentVersion.getId(),
+            (long) documentVersion.getVersionNo(),
+            SyncEventType.DOCUMENT_VERSION_CREATED,
+            payloadJson,
+            occurredAt
         );
     }
 
@@ -138,19 +136,42 @@ private SyncOutboxEvent saveEvent(
         String payloadJson,
         LocalDateTime occurredAt
     ) {
-        return syncOutboxEventRepository.save(
-            SyncOutboxEvent.builder()
-                .eventId(UUID.randomUUID())
-                .idempotencyKey(idempotencyKey)
-                .aggregateType(aggregateType)
-                .aggregateId(aggregateId)
-                .aggregateVersion(aggregateVersion)
-                .eventType(eventType)
-                .payloadJson(payloadJson)
-                .occurredAt(occurredAt)
-                .availableAt(occurredAt)
-                .maxRetryCount(DEFAULT_MAX_RETRY_COUNT)
-                .build()
+        // 1. DB Unique Key와 ON CONFLICT를 한 문장으로 실행해 동시 요청도 예외 없이 한 행에 수렴시킨다.
+        syncOutboxEventRepository.insertPendingIfAbsent(
+            UUID.randomUUID(),
+            idempotencyKey,
+            aggregateType.name(),
+            aggregateId,
+            aggregateVersion,
+            eventType.name(),
+            payloadJson,
+            occurredAt,
+            occurredAt,
+            DEFAULT_MAX_RETRY_COUNT
         );
+
+        // 2. 최초 생성자와 중복 요청자 모두 DB가 선택한 동일 Event를 반환한다.
+        SyncOutboxEvent event = syncOutboxEventRepository.findByIdempotencyKey(idempotencyKey)
+            .orElseThrow(() -> new DocGridException(ErrorCode.SYNC_EVENT_INCONSISTENT));
+        validateExistingEvent(event, aggregateType, aggregateId, aggregateVersion, eventType, payloadJson);
+        return event;
+    }
+
+    private void validateExistingEvent(
+        SyncOutboxEvent event,
+        SyncAggregateType aggregateType,
+        Long aggregateId,
+        Long aggregateVersion,
+        SyncEventType eventType,
+        String payloadJson
+    ) {
+        // 같은 Key가 다른 명령을 가리키면 중복 성공으로 숨기지 않고 원장 충돌로 중단한다.
+        if (event.getAggregateType() != aggregateType
+            || !Objects.equals(event.getAggregateId(), aggregateId)
+            || !Objects.equals(event.getAggregateVersion(), aggregateVersion)
+            || event.getEventType() != eventType
+            || !Objects.equals(event.getPayloadJson(), payloadJson)) {
+            throw new DocGridException(ErrorCode.SYNC_EVENT_INCONSISTENT);
+        }
     }
 }