diff --git a/docker-compose.yml b/docker-compose.yml index 63d5a05..4e77a67 100644 --- a/docker-compose.yml +++ b/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 diff --git a/mock-webhook/mappings/webhook-success.json b/mock-webhook/mappings/webhook-success.json new file mode 100644 index 0000000..3116375 --- /dev/null +++ b/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 } + } +} diff --git a/payout-worker/src/main/kotlin/com/openremit/payout/application/PayoutProcessor.kt b/payout-worker/src/main/kotlin/com/openremit/payout/application/PayoutProcessor.kt index 72ee68b..05c0810 100644 --- a/payout-worker/src/main/kotlin/com/openremit/payout/application/PayoutProcessor.kt +++ b/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( diff --git a/payout-worker/src/test/kotlin/com/openremit/payout/PayoutWorkerIntegrationTest.kt b/payout-worker/src/test/kotlin/com/openremit/payout/PayoutWorkerIntegrationTest.kt index 01daa4c..c3af104 100644 --- a/payout-worker/src/test/kotlin/com/openremit/payout/PayoutWorkerIntegrationTest.kt +++ b/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) diff --git a/webhook-dispatcher/build.gradle.kts b/webhook-dispatcher/build.gradle.kts index e9cc219..cdd5e67 100644 --- a/webhook-dispatcher/build.gradle.kts +++ b/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") } diff --git a/webhook-dispatcher/src/main/kotlin/com/openremit/webhook/WebhookDispatcherApplication.kt b/webhook-dispatcher/src/main/kotlin/com/openremit/webhook/WebhookDispatcherApplication.kt new file mode 100644 index 0000000..5143d2f --- /dev/null +++ b/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) { + runApplication(*args) +} diff --git a/webhook-dispatcher/src/main/kotlin/com/openremit/webhook/application/WebhookBackoff.kt b/webhook-dispatcher/src/main/kotlin/com/openremit/webhook/application/WebhookBackoff.kt new file mode 100644 index 0000000..307c90a --- /dev/null +++ b/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()) + } +} diff --git a/webhook-dispatcher/src/main/kotlin/com/openremit/webhook/application/WebhookDispatcher.kt b/webhook-dispatcher/src/main/kotlin/com/openremit/webhook/application/WebhookDispatcher.kt new file mode 100644 index 0000000..53582b0 --- /dev/null +++ b/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 + } +} diff --git a/webhook-dispatcher/src/main/kotlin/com/openremit/webhook/application/WebhookEnqueueService.kt b/webhook-dispatcher/src/main/kotlin/com/openremit/webhook/application/WebhookEnqueueService.kt new file mode 100644 index 0000000..be60985 --- /dev/null +++ b/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, + ) + ) + } +} diff --git a/webhook-dispatcher/src/main/kotlin/com/openremit/webhook/application/WebhookSender.kt b/webhook-dispatcher/src/main/kotlin/com/openremit/webhook/application/WebhookSender.kt new file mode 100644 index 0000000..f9d044e --- /dev/null +++ b/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, + ) + } + } +} diff --git a/webhook-dispatcher/src/main/kotlin/com/openremit/webhook/domain/Webhook.kt b/webhook-dispatcher/src/main/kotlin/com/openremit/webhook/domain/Webhook.kt new file mode 100644 index 0000000..c9e348b --- /dev/null +++ b/webhook-dispatcher/src/main/kotlin/com/openremit/webhook/domain/Webhook.kt @@ -0,0 +1,94 @@ +package com.openremit.webhook.domain + +import jakarta.persistence.Column +import jakarta.persistence.Entity +import jakarta.persistence.EnumType +import jakarta.persistence.Enumerated +import jakarta.persistence.GeneratedValue +import jakarta.persistence.GenerationType +import jakarta.persistence.Id +import jakarta.persistence.Table +import org.hibernate.annotations.JdbcTypeCode +import org.hibernate.type.SqlTypes +import java.time.Instant + +@Entity +@Table(name = "webhooks") +class Webhook( + @Column(name = "event_id", nullable = false, unique = true, length = 100) + val eventId: String, + + @Column(name = "event_type", nullable = false, length = 64) + val eventType: String, + + @Column(name = "target_url", nullable = false, length = 500) + val targetUrl: String, + + @JdbcTypeCode(SqlTypes.JSON) + @Column(name = "payload", nullable = false, columnDefinition = "json") + val payload: String, + + @Column(name = "created_at", nullable = false) + val createdAt: Instant = Instant.now(), +) { + @Id + @GeneratedValue(strategy = GenerationType.IDENTITY) + var id: Long = 0 + + @Enumerated(EnumType.STRING) + @Column(name = "status", nullable = false, length = 20) + var status: WebhookStatus = WebhookStatus.PENDING + + @Column(name = "attempt_count", nullable = false) + var attemptCount: Int = 0 + + @Column(name = "next_retry_at") + var nextRetryAt: Instant? = Instant.now() + + @Column(name = "last_response_status") + var lastResponseStatus: Int? = null + + @Column(name = "last_failure_reason", length = 500) + var lastFailureReason: String? = null + + @Column(name = "updated_at", nullable = false) + var updatedAt: Instant = Instant.now() + + /** 발송 성공 — terminal 상태로 전환. */ + fun markSuccess(httpStatus: Int) { + attemptCount += 1 + status = WebhookStatus.SUCCESS + lastResponseStatus = httpStatus + lastFailureReason = null + nextRetryAt = null + updatedAt = Instant.now() + } + + /** + * 발송 실패. attempt_count를 증가시키고: + * - 남은 시도가 있으면 status=PENDING, next_retry_at = now + delay + * - 마지막 시도였으면 status=FAILED, next_retry_at=null (terminal) + * + * @param maxAttempts 최대 시도 횟수 (이 값에 도달하면 terminal) + * @param nextDelay 다음 시도까지 대기 시간 (마지막 시도면 무시) + */ + fun markFailed(httpStatus: Int?, reason: String, maxAttempts: Int, nextDelay: java.time.Duration) { + attemptCount += 1 + lastResponseStatus = httpStatus + lastFailureReason = reason.take(500) + if (attemptCount >= maxAttempts) { + status = WebhookStatus.FAILED + nextRetryAt = null + } else { + status = WebhookStatus.PENDING + nextRetryAt = Instant.now().plus(nextDelay) + } + updatedAt = Instant.now() + } +} + +enum class WebhookStatus { + PENDING, + SUCCESS, + FAILED, +} diff --git a/webhook-dispatcher/src/main/kotlin/com/openremit/webhook/infrastructure/client/WebhookClient.kt b/webhook-dispatcher/src/main/kotlin/com/openremit/webhook/infrastructure/client/WebhookClient.kt new file mode 100644 index 0000000..64d896c --- /dev/null +++ b/webhook-dispatcher/src/main/kotlin/com/openremit/webhook/infrastructure/client/WebhookClient.kt @@ -0,0 +1,80 @@ +package com.openremit.webhook.infrastructure.client + +import org.springframework.boot.context.properties.ConfigurationProperties +import org.springframework.context.annotation.Bean +import org.springframework.context.annotation.Configuration +import org.springframework.http.client.SimpleClientHttpRequestFactory +import org.springframework.stereotype.Component +import org.springframework.web.client.RestClient +import org.springframework.web.client.RestClientResponseException + +interface WebhookClient { + /** + * 외부 webhook 엔드포인트로 POST. 2xx 응답이면 성공. + * 4xx/5xx/IO는 [WebhookSendException]으로 wrap. + */ + fun send(targetUrl: String, payload: String): Result + + data class Result(val httpStatus: Int) +} + +class WebhookSendException( + val httpStatus: Int?, + message: String, + cause: Throwable? = null, +) : RuntimeException(message, cause) + +@ConfigurationProperties(prefix = "openremit.webhook") +data class WebhookClientProperties( + val targetUrl: String, +) + +@Configuration +class WebhookClientConfig { + @Bean + fun webhookRestClient(): RestClient = + // SimpleClientHttpRequestFactory(JDK HttpURLConnection)를 명시 — RestClient 기본 후보가 + // HTTP/2 negotiate를 시도해 WireMock과 RST_STREAM 호환성 문제를 일으키는 것 방지 (payout-worker 동일). + RestClient.builder() + .requestFactory(SimpleClientHttpRequestFactory()) + .build() +} + +@Component +class RestClientWebhookClient( + private val webhookRestClient: RestClient, +) : WebhookClient { + + override fun send(targetUrl: String, payload: String): WebhookClient.Result = + try { + val response = webhookRestClient.post() + .uri(targetUrl) + .header("Content-Type", "application/json") + .body(payload) + .retrieve() + .toBodilessEntity() + // RestClient.retrieve()는 4xx/5xx만 예외로 처리하므로 3xx 리다이렉트는 여기로 떨어진다. + // contract는 "2xx만 성공"이므로 비-2xx는 재시도 대상으로 wrap. + if (!response.statusCode.is2xxSuccessful) { + throw WebhookSendException( + httpStatus = response.statusCode.value(), + message = "webhook target responded with non-2xx ${response.statusCode}", + ) + } + WebhookClient.Result(httpStatus = response.statusCode.value()) + } catch (e: WebhookSendException) { + throw e + } catch (e: RestClientResponseException) { + throw WebhookSendException( + httpStatus = e.statusCode.value(), + message = "webhook target responded with ${e.statusCode}", + cause = e, + ) + } catch (e: Exception) { + throw WebhookSendException( + httpStatus = null, + message = e.message ?: e.javaClass.simpleName, + cause = e, + ) + } +} diff --git a/webhook-dispatcher/src/main/kotlin/com/openremit/webhook/infrastructure/kafka/RemittancePayoutResultConsumer.kt b/webhook-dispatcher/src/main/kotlin/com/openremit/webhook/infrastructure/kafka/RemittancePayoutResultConsumer.kt new file mode 100644 index 0000000..0ec0e78 --- /dev/null +++ b/webhook-dispatcher/src/main/kotlin/com/openremit/webhook/infrastructure/kafka/RemittancePayoutResultConsumer.kt @@ -0,0 +1,47 @@ +package com.openremit.webhook.infrastructure.kafka + +import com.openremit.common.events.RemittanceEventTopics +import com.openremit.webhook.application.WebhookEnqueueService +import org.slf4j.LoggerFactory +import org.springframework.kafka.annotation.KafkaListener +import org.springframework.messaging.handler.annotation.Header +import org.springframework.kafka.support.KafkaHeaders +import org.springframework.stereotype.Component + +/** + * payout-worker가 발행한 결과 이벤트를 받아 webhook 발송 큐에 적재. + * 같은 토픽을 remittance-api도 별도 consumer-group(`remittance-api-payout-result`)으로 소비 중. + * webhook-dispatcher는 `webhook-dispatcher` consumer-group으로 별개 오프셋. + */ +@Component +class RemittancePayoutResultConsumer( + private val webhookEnqueueService: WebhookEnqueueService, +) { + private val log = LoggerFactory.getLogger(javaClass) + + @KafkaListener( + topics = [ + RemittanceEventTopics.PAYOUT_COMPLETED, + RemittanceEventTopics.PAYOUT_FAILED, + ], + groupId = "webhook-dispatcher", + ) + fun consume( + payload: String, + @Header(KafkaHeaders.RECEIVED_TOPIC) topic: String, + @Header(KafkaHeaders.RECEIVED_KEY, required = false) key: String?, + ) { + log.debug("received topic={} key={}: {}", topic, key, payload) + // 멱등성 키: remittanceId(=Kafka key) + topic. payout-worker가 outbox 발행 시 aggregate_id를 키로 넣음. + // key가 없는 경우는 운영상 발생하지 않지만 안전장치로 payload 해시 대신 명시 skip. + if (key.isNullOrBlank()) { + log.warn("dropped payout result without key: topic={}", topic) + return + } + webhookEnqueueService.enqueue( + eventId = "$key:$topic", + eventType = topic, + payload = payload, + ) + } +} diff --git a/webhook-dispatcher/src/main/kotlin/com/openremit/webhook/infrastructure/persistence/WebhookRepository.kt b/webhook-dispatcher/src/main/kotlin/com/openremit/webhook/infrastructure/persistence/WebhookRepository.kt new file mode 100644 index 0000000..8ac77e2 --- /dev/null +++ b/webhook-dispatcher/src/main/kotlin/com/openremit/webhook/infrastructure/persistence/WebhookRepository.kt @@ -0,0 +1,32 @@ +package com.openremit.webhook.infrastructure.persistence + +import com.openremit.webhook.domain.Webhook +import com.openremit.webhook.domain.WebhookStatus +import org.springframework.data.domain.Pageable +import org.springframework.data.jpa.repository.JpaRepository +import org.springframework.data.jpa.repository.Query +import org.springframework.data.repository.query.Param +import java.time.Instant + +interface WebhookRepository : JpaRepository { + fun findByEventId(eventId: String): Webhook? + + /** + * 폴링 스케줄러가 호출. PENDING + next_retry_at <= now 인 행을 + * id 오름차순으로 한 페이지씩 회수. + */ + @Query( + """ + SELECT w FROM Webhook w + WHERE w.status = :status + AND w.nextRetryAt IS NOT NULL + AND w.nextRetryAt <= :now + ORDER BY w.id ASC + """ + ) + fun findDueForRetry( + @Param("status") status: WebhookStatus, + @Param("now") now: Instant, + pageable: Pageable, + ): List +} diff --git a/webhook-dispatcher/src/main/resources/application.yaml b/webhook-dispatcher/src/main/resources/application.yaml new file mode 100644 index 0000000..194d746 --- /dev/null +++ b/webhook-dispatcher/src/main/resources/application.yaml @@ -0,0 +1,58 @@ +spring: + application: + name: webhook-dispatcher + threads: + virtual: + enabled: true + jackson: + property-naming-strategy: SNAKE_CASE + datasource: + url: jdbc:mysql://localhost:3306/openremit?useSSL=false&allowPublicKeyRetrieval=true&serverTimezone=UTC + username: openremit + password: openremit + driver-class-name: com.mysql.cj.jdbc.Driver + jpa: + hibernate: + ddl-auto: none + properties: + hibernate: + format_sql: true + show-sql: false + flyway: + enabled: true + locations: classpath:db/migration + # 다른 모듈과 같은 DB를 공유하지만 history 테이블은 분리 (ADR-011) + table: flyway_schema_history_webhook + baseline-on-migrate: true + baseline-version: "0" + kafka: + bootstrap-servers: ${KAFKA_BOOTSTRAP:localhost:9092} + consumer: + group-id: webhook-dispatcher + auto-offset-reset: earliest + enable-auto-commit: false + key-deserializer: org.apache.kafka.common.serialization.StringDeserializer + value-deserializer: org.apache.kafka.common.serialization.StringDeserializer + listener: + ack-mode: record + +openremit: + webhook: + target-url: ${WEBHOOK_TARGET_URL:http://localhost:9997/webhook} + backoff: + base-millis: 1000 + multiplier: 2.0 + max-attempts: 5 + poll-interval-millis: 1000 + +server: + port: 8082 + +management: + endpoints: + web: + exposure: + include: health,info,metrics + endpoint: + health: + show-details: when-authorized diff --git a/webhook-dispatcher/src/main/resources/db/migration/V1__webhook_tables.sql b/webhook-dispatcher/src/main/resources/db/migration/V1__webhook_tables.sql new file mode 100644 index 0000000..3b1fd37 --- /dev/null +++ b/webhook-dispatcher/src/main/resources/db/migration/V1__webhook_tables.sql @@ -0,0 +1,23 @@ +-- webhook-dispatcher 자체 소유 테이블 (ADR-011) +-- dispatcher는 remittances 테이블에 일절 접근하지 않는다. 자체 발송 이력만 보유. + +-- webhooks: 발송 시도 + 재시도 스케줄 + 멱등성 키 +-- event_id UNIQUE → Kafka 중복 소비 시 INSERT 실패로 중복 발송 차단 +-- (status, next_retry_at) 인덱스 → 폴링 스케줄러가 due 행 효율적으로 조회 +CREATE TABLE webhooks ( + id BIGINT NOT NULL AUTO_INCREMENT, + event_id VARCHAR(100) NOT NULL, + event_type VARCHAR(64) NOT NULL, + target_url VARCHAR(500) NOT NULL, + payload JSON NOT NULL, + status VARCHAR(20) NOT NULL, + attempt_count INT NOT NULL DEFAULT 0, + next_retry_at TIMESTAMP(6) NULL, + last_response_status INT NULL, + last_failure_reason VARCHAR(500) NULL, + created_at TIMESTAMP(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6), + updated_at TIMESTAMP(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6), + PRIMARY KEY (id), + UNIQUE KEY uk_webhooks_event_id (event_id), + INDEX idx_webhooks_status_next_retry (status, next_retry_at) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; diff --git a/webhook-dispatcher/src/test/kotlin/com/openremit/webhook/WebhookDispatcherIntegrationTest.kt b/webhook-dispatcher/src/test/kotlin/com/openremit/webhook/WebhookDispatcherIntegrationTest.kt new file mode 100644 index 0000000..89d0fe5 --- /dev/null +++ b/webhook-dispatcher/src/test/kotlin/com/openremit/webhook/WebhookDispatcherIntegrationTest.kt @@ -0,0 +1,148 @@ +package com.openremit.webhook + +import com.github.tomakehurst.wiremock.WireMockServer +import com.github.tomakehurst.wiremock.client.WireMock.aResponse +import com.github.tomakehurst.wiremock.client.WireMock.post +import com.github.tomakehurst.wiremock.client.WireMock.postRequestedFor +import com.github.tomakehurst.wiremock.client.WireMock.urlPathEqualTo +import com.github.tomakehurst.wiremock.core.WireMockConfiguration.wireMockConfig +import com.openremit.common.events.RemittanceEventTopics +import com.openremit.webhook.domain.WebhookStatus +import com.openremit.webhook.infrastructure.persistence.WebhookRepository +import org.springframework.beans.factory.annotation.Autowired +import org.springframework.boot.test.context.SpringBootTest +import org.springframework.context.annotation.Import +import org.springframework.kafka.core.KafkaTemplate +import org.springframework.test.context.DynamicPropertyRegistry +import org.springframework.test.context.DynamicPropertySource +import kotlin.test.AfterTest +import kotlin.test.BeforeTest +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertNotNull + +@SpringBootTest +@Import(WebhookDispatcherTestcontainersConfig::class) +class WebhookDispatcherIntegrationTest @Autowired constructor( + private val kafkaTemplate: KafkaTemplate, + private val webhookRepository: WebhookRepository, +) { + companion object { + private val wireMock: WireMockServer = WireMockServer(wireMockConfig().dynamicPort()) + + @JvmStatic + @DynamicPropertySource + fun props(registry: DynamicPropertyRegistry) { + if (!wireMock.isRunning) wireMock.start() + registry.add("openremit.webhook.target-url") { "http://localhost:${wireMock.port()}/webhook" } + // 테스트 시 backoff 단축: 50ms × 2 = 50/100/200/400/800ms (총 5회, ~1.5초) + registry.add("openremit.webhook.backoff.base-millis") { "50" } + registry.add("openremit.webhook.backoff.multiplier") { "2.0" } + registry.add("openremit.webhook.backoff.max-attempts") { "5" } + registry.add("openremit.webhook.poll-interval-millis") { "100" } + // TestcontainersConfig 클래스 로드를 강제 (Kafka container 시작). + WebhookDispatcherTestcontainersConfig.kafka + } + } + + @BeforeTest + fun setup() { + if (!wireMock.isRunning) wireMock.start() + wireMock.resetAll() + } + + @AfterTest + fun cleanup() { + webhookRepository.deleteAllInBatch() + } + + @Test + fun `payout completed event triggers webhook send and marks SUCCESS`() { + wireMock.stubFor( + post(urlPathEqualTo("/webhook")) + .willReturn(aResponse().withStatus(200)) + ) + + val remittanceId = 100L + val payload = """{"remittance_id":$remittanceId,"payout_tx_id":"PAYOUT-X","occurred_at":"2026-05-09T00:00:00Z"}""" + kafkaTemplate.send( + RemittanceEventTopics.PAYOUT_COMPLETED, + remittanceId.toString(), + payload, + ).get() + + val webhook = waitForWebhook(eventId = "$remittanceId:${RemittanceEventTopics.PAYOUT_COMPLETED}", timeoutMs = 30_000) { + it.status == WebhookStatus.SUCCESS + } + assertEquals(WebhookStatus.SUCCESS, webhook.status) + assertEquals(1, webhook.attemptCount) + assertEquals(200, webhook.lastResponseStatus) + wireMock.verify(1, postRequestedFor(urlPathEqualTo("/webhook"))) + } + + @Test + fun `payout failed event hits 502 then retries 5 times before marking FAILED`() { + wireMock.stubFor( + post(urlPathEqualTo("/webhook")) + .willReturn(aResponse().withStatus(502).withBody("upstream down")) + ) + + val remittanceId = 200L + val payload = """{"remittance_id":$remittanceId,"reason":"timeout","occurred_at":"2026-05-09T00:00:00Z"}""" + kafkaTemplate.send( + RemittanceEventTopics.PAYOUT_FAILED, + remittanceId.toString(), + payload, + ).get() + + val webhook = waitForWebhook(eventId = "$remittanceId:${RemittanceEventTopics.PAYOUT_FAILED}", timeoutMs = 30_000) { + it.status == WebhookStatus.FAILED + } + assertEquals(WebhookStatus.FAILED, webhook.status) + assertEquals(5, webhook.attemptCount) + assertEquals(502, webhook.lastResponseStatus) + wireMock.verify(5, postRequestedFor(urlPathEqualTo("/webhook"))) + } + + @Test + fun `duplicate payout result with same key is deduplicated by event_id`() { + wireMock.stubFor( + post(urlPathEqualTo("/webhook")) + .willReturn(aResponse().withStatus(200)) + ) + + val remittanceId = 300L + val payload = """{"remittance_id":$remittanceId,"payout_tx_id":"PAYOUT-DUP","occurred_at":"2026-05-09T00:00:00Z"}""" + repeat(3) { + kafkaTemplate.send( + RemittanceEventTopics.PAYOUT_COMPLETED, + remittanceId.toString(), + payload, + ).get() + } + + val webhook = waitForWebhook(eventId = "$remittanceId:${RemittanceEventTopics.PAYOUT_COMPLETED}", timeoutMs = 30_000) { + it.status == WebhookStatus.SUCCESS + } + assertEquals(1, webhook.attemptCount) + // 같은 event_id로 INSERT 1회만 성공. Kafka 재소비가 webhook을 중복 발송시키지 않음. + wireMock.verify(1, postRequestedFor(urlPathEqualTo("/webhook"))) + } + + private fun waitForWebhook( + eventId: String, + timeoutMs: Long, + intervalMs: Long = 100, + condition: (com.openremit.webhook.domain.Webhook) -> Boolean, + ): com.openremit.webhook.domain.Webhook { + val deadline = System.currentTimeMillis() + timeoutMs + while (System.currentTimeMillis() < deadline) { + val w = webhookRepository.findByEventId(eventId) + if (w != null && condition(w)) return w + Thread.sleep(intervalMs) + } + val w = webhookRepository.findByEventId(eventId) + assertNotNull(w, "webhook with eventId=$eventId not found within $timeoutMs ms") + throw AssertionError("webhook condition not satisfied within $timeoutMs ms; status=${w.status}, attempts=${w.attemptCount}") + } +} diff --git a/webhook-dispatcher/src/test/kotlin/com/openremit/webhook/WebhookDispatcherTestcontainersConfig.kt b/webhook-dispatcher/src/test/kotlin/com/openremit/webhook/WebhookDispatcherTestcontainersConfig.kt new file mode 100644 index 0000000..1588bb2 --- /dev/null +++ b/webhook-dispatcher/src/test/kotlin/com/openremit/webhook/WebhookDispatcherTestcontainersConfig.kt @@ -0,0 +1,32 @@ +package com.openremit.webhook + +import org.springframework.boot.test.context.TestConfiguration +import org.springframework.boot.testcontainers.service.connection.ServiceConnection +import org.springframework.context.annotation.Bean +import org.testcontainers.containers.MySQLContainer +import org.testcontainers.kafka.ConfluentKafkaContainer +import org.testcontainers.utility.DockerImageName + +@TestConfiguration(proxyBeanMethods = false) +class WebhookDispatcherTestcontainersConfig { + + @Bean + @ServiceConnection + fun mysqlContainer(): MySQLContainer<*> = + MySQLContainer("mysql:8.0") + .withDatabaseName("openremit") + .withUsername("test") + .withPassword("test") + + companion object { + // Spring Boot 4.0의 @ServiceConnection이 ConfluentKafkaContainer를 인식하지 않으므로 + // singleton container + System property 주입으로 bootstrap-servers를 노출 (payout-worker 동일 패턴). + val kafka: ConfluentKafkaContainer = + ConfluentKafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:7.6.1")) + .also { it.start() } + + init { + System.setProperty("spring.kafka.bootstrap-servers", kafka.bootstrapServers) + } + } +}