# feat(webhook): webhook-dispatcher 추가 + payout 중복 이벤트 멱등 처리 보강

docker-compose.yml

@@ -71,6 +71,20 @@ services:
       timeout: 3s
       retries: 10
 
+  mock-webhook:
+    image: wiremock/wiremock:3.13.1
+    container_name: openremit-mock-webhook
+    ports:
+      - "9997:8080"
+    volumes:
+      - ./mock-webhook/mappings:/home/wiremock/mappings:ro
+    command: ["--port", "8080"]
+    healthcheck:
+      test: ["CMD-SHELL", "wget -q --spider http://localhost:8080/__admin/health || exit 1"]
+      interval: 5s
+      timeout: 3s
+      retries: 10
+
   kafka:
     image: apache/kafka:3.8.0
     container_name: openremit-kafka

mock-webhook/mappings/webhook-success.json

@@ -0,0 +1,11 @@
+{
+  "request": {
+    "method": "POST",
+    "urlPath": "/webhook"
+  },
+  "response": {
+    "status": 200,
+    "headers": { "Content-Type": "application/json" },
+    "jsonBody": { "received": true }
+  }
+}

payout-worker/src/main/kotlin/com/openremit/payout/application/PayoutProcessor.kt

@@ -10,7 +10,6 @@ import com.openremit.payout.infrastructure.client.PayoutClient
 import com.openremit.payout.infrastructure.persistence.PayoutAttemptRepository
 import com.openremit.payout.infrastructure.persistence.PayoutOutboxRepository
 import org.slf4j.LoggerFactory
-import org.springframework.dao.DataIntegrityViolationException
 import org.springframework.stereotype.Service
 import org.springframework.transaction.annotation.Transactional
 import tools.jackson.databind.ObjectMapper
@@ -27,22 +26,26 @@ class PayoutProcessor(
 
     /**
      * 단일 트랜잭션으로 처리:
-     *   1) PayoutAttempt INSERT (UNIQUE remittance_id로 멱등성 차단)
+     *   1) PayoutAttempt 사전 SELECT(멱등성 체크) → 없으면 INSERT
      *   2) 송금사 API 호출 (외부 I/O — 트랜잭션 안이지만 단일 호출이라 짧음)
      *   3) attempts 상태 업데이트 + payout_outbox INSERT (Debezium 발행 대상)
      *
+     * 멱등성: 사전 SELECT로 차단. UNIQUE 제약 위반을 catch하던 이전 패턴은 트랜잭션을
+     *   rollback-only로 마킹해 commit 시 UnexpectedRollbackException → Kafka partition stuck.
+     *   Kafka 키(remittanceId) 기반 파티셔닝으로 동일 키가 단일 컨슈머 스레드에 직렬화되므로
+     *   사전 체크의 TOCTOU race는 실무상 발생하지 않는다.
+     *
      * 트레이드오프: 외부 I/O가 트랜잭션에 포함되어 DB 커넥션을 점유. 송금사 호출이 짧을 때만 안전.
      * 호출이 길어지면 향후 (1) attempts INSERT만 트랜잭션 1, (2) 송금사 호출, (3) outbox INSERT를
      * 트랜잭션 2로 분리 + reconciler가 PENDING attempts 회수하는 구조로 진화.
      */
     @Transactional
     fun process(event: RemittancePaidEvent) {
-        val attempt = try {
-            payoutAttemptRepository.saveAndFlush(PayoutAttempt(remittanceId = event.remittanceId))
-        } catch (e: DataIntegrityViolationException) {
+        if (payoutAttemptRepository.findByRemittanceId(event.remittanceId) != null) {
             log.info("payout attempt for remittanceId={} already exists — skipping", event.remittanceId)
             return
         }
+        val attempt = payoutAttemptRepository.saveAndFlush(PayoutAttempt(remittanceId = event.remittanceId))
 
         val outbox: PayoutOutboxEvent = try {
             val result = payoutClient.payout(

payout-worker/src/test/kotlin/com/openremit/payout/PayoutWorkerIntegrationTest.kt

@@ -111,6 +111,49 @@ class PayoutWorkerIntegrationTest @Autowired constructor(
         assertTrue(events[0].payload.contains("PAYOUT-XYZ"))
     }
 
+    @Test
+    fun `duplicate remittance paid event is deduplicated and payout API called once`() {
+        // 같은 remittanceId로 3번 produce — Kafka at-least-once 시나리오 모사.
+        // 사전 SELECT 멱등성 체크가 없으면 두 번째 메시지가 UNIQUE 위반 → rollback-only →
+        // UnexpectedRollbackException → ack 안 됨 → 무한 루프로 partition stuck.
+        val event = paidEvent(remittanceId = 77L)
+        repeat(3) {
+            kafkaTemplate.send(
+                RemittanceEventTopics.PAID,
+                event.remittanceId.toString(),
+                objectMapper.writeValueAsString(event),
+            ).get()
+        }
+
+        waitForCondition(timeoutMs = 30_000) {
+            payoutAttemptRepository.findByRemittanceId(77L)?.status == PayoutAttemptStatus.COMPLETED
+        }
+
+        // attempt 1행만, outbox 1건만, payout API 호출도 1회만.
+        val attempt = payoutAttemptRepository.findByRemittanceId(77L)
+        assertNotNull(attempt)
+        assertEquals(PayoutAttemptStatus.COMPLETED, attempt.status)
+
+        val events = payoutOutboxRepository.findByAggregateTypeAndAggregateIdOrderByIdAsc(
+            PayoutOutboxEvent.AGGREGATE_TYPE_REMITTANCE,
+            "77",
+        )
+        // 추가 메시지 처리가 모두 사전 체크에서 skip되어 outbox 1건만.
+        // 짧은 시간 내에 추가 outbox row가 안 들어오는지 한 번 더 대기.
+        Thread.sleep(1_000)
+        val eventsAfterWait = payoutOutboxRepository.findByAggregateTypeAndAggregateIdOrderByIdAsc(
+            PayoutOutboxEvent.AGGREGATE_TYPE_REMITTANCE,
+            "77",
+        )
+        assertEquals(1, events.size)
+        assertEquals(1, eventsAfterWait.size)
+        wireMock.verify(
+            1,
+            com.github.tomakehurst.wiremock.client.WireMock.postRequestedFor(urlPathEqualTo("/payouts"))
+                .withRequestBody(matchingJsonPath("$.remittance_id", equalTo("77"))),
+        )
+    }
+
     @Test
     fun `payout 5xx writes failed outbox and marks attempt failed`() {
         val event = paidEvent(remittanceId = 99L)

webhook-dispatcher/build.gradle.kts

@@ -1,6 +1,34 @@
-// Webhook 발송 워커 (재시도 포함)
-// Day 9에 Spring Boot 플러그인 + Application 추가 예정.
+plugins {
+    id("org.springframework.boot")
+    kotlin("plugin.jpa")
+}
+
+allOpen {
+    annotation("jakarta.persistence.Entity")
+    annotation("jakarta.persistence.MappedSuperclass")
+    annotation("jakarta.persistence.Embeddable")
+}
 
 dependencies {
     implementation(project(":common"))
+
+    implementation("org.springframework.boot:spring-boot-starter-actuator")
+    implementation("org.springframework.boot:spring-boot-starter-data-jpa")
+    implementation("org.springframework.boot:spring-boot-starter-flyway")
+    implementation("org.springframework.boot:spring-boot-starter-webmvc")
+    implementation("org.springframework.boot:spring-boot-starter-validation")
+    implementation("org.flywaydb:flyway-mysql")
+    implementation("org.springframework.boot:spring-boot-starter-kafka")
+    runtimeOnly("com.mysql:mysql-connector-j")
+
+    testImplementation("org.springframework.boot:spring-boot-starter-actuator-test")
+    testImplementation("org.springframework.boot:spring-boot-starter-data-jpa-test")
+    testImplementation("org.springframework.boot:spring-boot-starter-flyway-test")
+    testImplementation("org.springframework.boot:spring-boot-starter-webmvc-test")
+    testImplementation("org.springframework.boot:spring-boot-testcontainers")
+    testImplementation("org.springframework.boot:spring-boot-starter-kafka-test")
+    testImplementation("org.testcontainers:junit-jupiter")
+    testImplementation("org.testcontainers:mysql")
+    testImplementation("org.testcontainers:kafka")
+    testImplementation("org.wiremock:wiremock-standalone:3.13.1")
 }

webhook-dispatcher/src/main/kotlin/com/openremit/webhook/WebhookDispatcherApplication.kt

@@ -0,0 +1,19 @@
+package com.openremit.webhook
+
+import com.openremit.webhook.application.WebhookBackoffProperties
+import com.openremit.webhook.infrastructure.client.WebhookClientProperties
+import org.springframework.boot.autoconfigure.SpringBootApplication
+import org.springframework.boot.context.properties.EnableConfigurationProperties
+import org.springframework.boot.runApplication
+import org.springframework.kafka.annotation.EnableKafka
+import org.springframework.scheduling.annotation.EnableScheduling
+
+@SpringBootApplication
+@EnableKafka
+@EnableScheduling
+@EnableConfigurationProperties(WebhookClientProperties::class, WebhookBackoffProperties::class)
+class WebhookDispatcherApplication
+
+fun main(args: Array<String>) {
+    runApplication<WebhookDispatcherApplication>(*args)
+}

webhook-dispatcher/src/main/kotlin/com/openremit/webhook/application/WebhookBackoff.kt

@@ -0,0 +1,26 @@
+package com.openremit.webhook.application
+
+import org.springframework.boot.context.properties.ConfigurationProperties
+import java.time.Duration
+import kotlin.math.pow
+
+/**
+ * Exponential backoff. attempt(1-base)에 대한 다음 대기 시간 계산.
+ *   attempt=1 → base
+ *   attempt=2 → base * multiplier
+ *   attempt=N → base * multiplier^(N-1)
+ *
+ * 데모 기본값: 1s, 2s, 4s, 8s, 16s (총 5회 시도, 31초).
+ */
+@ConfigurationProperties(prefix = "openremit.webhook.backoff")
+data class WebhookBackoffProperties(
+    val baseMillis: Long = 1000,
+    val multiplier: Double = 2.0,
+    val maxAttempts: Int = 5,
+) {
+    fun delayFor(attempt: Int): Duration {
+        require(attempt >= 1) { "attempt must be >= 1, was $attempt" }
+        val millis = baseMillis.toDouble() * multiplier.pow(attempt - 1)
+        return Duration.ofMillis(millis.toLong())
+    }
+}

webhook-dispatcher/src/main/kotlin/com/openremit/webhook/application/WebhookDispatcher.kt

@@ -0,0 +1,42 @@
+package com.openremit.webhook.application
+
+import com.openremit.webhook.domain.WebhookStatus
+import com.openremit.webhook.infrastructure.persistence.WebhookRepository
+import org.slf4j.LoggerFactory
+import org.springframework.data.domain.PageRequest
+import org.springframework.scheduling.annotation.Scheduled
+import org.springframework.stereotype.Service
+import java.time.Instant
+
+/**
+ * 폴링 스케줄러. PENDING + next_retry_at <= now 인 webhook 행을 회수해
+ * [WebhookSender]로 한 행씩 발송 시도.
+ *
+ * 트레이드오프: SELECT 후 row-level 락 없음 → 단일 인스턴스 가정.
+ *   다중 인스턴스 운영 시 SELECT ... FOR UPDATE SKIP LOCKED 필요. 현재는 데모 범위 밖.
+ */
+@Service
+class WebhookDispatcher(
+    private val webhookRepository: WebhookRepository,
+    private val webhookSender: WebhookSender,
+) {
+    private val log = LoggerFactory.getLogger(javaClass)
+
+    @Scheduled(fixedDelayString = "\${openremit.webhook.poll-interval-millis:1000}")
+    fun dispatchDue() {
+        val due = webhookRepository.findDueForRetry(
+            status = WebhookStatus.PENDING,
+            now = Instant.now(),
+            pageable = PageRequest.of(0, BATCH_SIZE),
+        )
+        if (due.isEmpty()) return
+        log.debug("dispatching {} due webhooks", due.size)
+        for (webhook in due) {
+            webhookSender.attemptSend(webhook.id)
+        }
+    }
+
+    companion object {
+        private const val BATCH_SIZE = 50
+    }
+}

webhook-dispatcher/src/main/kotlin/com/openremit/webhook/application/WebhookEnqueueService.kt

@@ -0,0 +1,43 @@
+package com.openremit.webhook.application
+
+import com.openremit.webhook.domain.Webhook
+import com.openremit.webhook.infrastructure.client.WebhookClientProperties
+import com.openremit.webhook.infrastructure.persistence.WebhookRepository
+import org.slf4j.LoggerFactory
+import org.springframework.stereotype.Service
+import org.springframework.transaction.annotation.Transactional
+
+/**
+ * Kafka 컨슈머가 호출. event_id로 멱등성 보장:
+ *   동일 (remittanceId, eventType) 메시지 재소비 시 사전 SELECT로 skip.
+ *
+ * 사전 체크 이유: UNIQUE 제약 위반(`DataIntegrityViolationException`)을 단순 catch하면
+ * @Transactional 트랜잭션이 rollback-only로 마킹되어 commit 시 UnexpectedRollbackException이
+ * 발생한다. Kafka 컨슈머가 ack 못 하고 같은 메시지를 무한 재배송 → partition stuck.
+ *
+ * Kafka는 키 기반 파티셔닝으로 동일 event_id가 단일 컨슈머 스레드에 직렬화되므로 사전 체크의
+ * TOCTOU race는 실무상 발생하지 않는다. 만에 하나 발생해도 redelivery 시 SELECT가 잡아준다.
+ */
+@Service
+class WebhookEnqueueService(
+    private val webhookRepository: WebhookRepository,
+    private val clientProperties: WebhookClientProperties,
+) {
+    private val log = LoggerFactory.getLogger(javaClass)
+
+    @Transactional
+    fun enqueue(eventId: String, eventType: String, payload: String) {
+        if (webhookRepository.findByEventId(eventId) != null) {
+            log.info("webhook for eventId={} already enqueued — skipping", eventId)
+            return
+        }
+        webhookRepository.save(
+            Webhook(
+                eventId = eventId,
+                eventType = eventType,
+                targetUrl = clientProperties.targetUrl,
+                payload = payload,
+            )
+        )
+    }
+}

webhook-dispatcher/src/main/kotlin/com/openremit/webhook/application/WebhookSender.kt

@@ -0,0 +1,65 @@
+package com.openremit.webhook.application
+
+import com.openremit.webhook.domain.Webhook
+import com.openremit.webhook.domain.WebhookStatus
+import com.openremit.webhook.infrastructure.client.WebhookClient
+import com.openremit.webhook.infrastructure.client.WebhookSendException
+import com.openremit.webhook.infrastructure.persistence.WebhookRepository
+import org.slf4j.LoggerFactory
+import org.springframework.stereotype.Service
+import org.springframework.transaction.annotation.Transactional
+
+/**
+ * 단일 webhook 행에 대한 발송 시도. 트랜잭션을 행 단위로 분리해 한 행 실패가
+ * 같은 배치의 다른 행 처리를 방해하지 않도록 한다.
+ *
+ * 별도 빈으로 분리한 이유: WebhookDispatcher의 스케줄러 메서드가 호출하면
+ * 같은 클래스 self-invocation이라 @Transactional 프록시가 적용되지 않는다.
+ */
+@Service
+class WebhookSender(
+    private val webhookRepository: WebhookRepository,
+    private val webhookClient: WebhookClient,
+    private val backoff: WebhookBackoffProperties,
+) {
+    private val log = LoggerFactory.getLogger(javaClass)
+
+    @Transactional
+    fun attemptSend(webhookId: Long) {
+        val webhook = webhookRepository.findById(webhookId).orElse(null) ?: return
+        if (webhook.status != WebhookStatus.PENDING) return
+
+        try {
+            val result = webhookClient.send(webhook.targetUrl, webhook.payload)
+            webhook.markSuccess(result.httpStatus)
+            log.info(
+                "webhook id={} eventId={} sent successfully (httpStatus={}, attempt={})",
+                webhook.id, webhook.eventId, result.httpStatus, webhook.attemptCount,
+            )
+        } catch (e: WebhookSendException) {
+            handleFailure(webhook, e)
+        }
+    }
+
+    private fun handleFailure(webhook: Webhook, e: WebhookSendException) {
+        val nextAttempt = webhook.attemptCount + 1
+        val nextDelay = backoff.delayFor(nextAttempt)
+        webhook.markFailed(
+            httpStatus = e.httpStatus,
+            reason = e.message ?: "unknown",
+            maxAttempts = backoff.maxAttempts,
+            nextDelay = nextDelay,
+        )
+        if (webhook.status == WebhookStatus.FAILED) {
+            log.warn(
+                "webhook id={} eventId={} reached max attempts ({}) — terminal FAILED",
+                webhook.id, webhook.eventId, backoff.maxAttempts,
+            )
+        } else {
+            log.info(
+                "webhook id={} eventId={} attempt {} failed (httpStatus={}); next retry in {}",
+                webhook.id, webhook.eventId, webhook.attemptCount, e.httpStatus, nextDelay,
+            )
+        }
+    }
+}