fix(charge): 충전 이벤트 보너스 지급을 안정화한다

2026-05-18 13:34:12 +09:00
parent acd0393a0e
commit 810b143c9e
15 changed files with 929 additions and 102 deletions
--- a/docs/plan-task/20260518_charge_event_job_ddl.sql
+++ b/docs/plan-task/20260518_charge_event_job_ddl.sql
@@ -0,0 +1,37 @@
+CREATE TABLE charge_event_job
+(
+    id                    BIGINT       NOT NULL AUTO_INCREMENT COMMENT 'PK',
+    source_charge_id      BIGINT       NOT NULL COMMENT '이벤트 보너스 지급의 기준이 되는 원본 충전 ID(charge.id)',
+    result_charge_id      BIGINT       NULL COMMENT '이벤트 보너스로 생성된 Charge ID(charge.id)',
+    member_id             BIGINT       NOT NULL COMMENT '이벤트 보너스를 지급받을 회원 ID(member.id)',
+    charge_event_id       BIGINT       NULL COMMENT '적용된 충전 이벤트 ID(charge_event.id), 첫 충전 이벤트처럼 별도 이벤트 row가 없으면 NULL',
+    job_type              VARCHAR(30)  NOT NULL COMMENT '작업 유형(FIRST_CHARGE, ACTIVE_CHARGE_EVENT)',
+    idempotency_key       VARCHAR(100) NOT NULL COMMENT '중복 작업 방지 키(예: charge-event:{sourceChargeId}:{jobType}:{chargeEventId 또는 none})',
+    additional_can        INT          NOT NULL COMMENT '추가 지급할 보너스 캔 수',
+    payment_gateway       VARCHAR(30)  NOT NULL COMMENT '원본 충전의 결제 게이트웨이(PG, PAYVERSE, GOOGLE_IAP, APPLE_IAP 등)',
+    container             VARCHAR(10)  NOT NULL COMMENT '회원 캔 잔액 반영 대상(pg, aos, ios)',
+    method_snapshot       VARCHAR(100) NOT NULL COMMENT '이벤트 보너스 Charge.payment.method에 기록할 지급 사유 스냅샷',
+    status                VARCHAR(20)  NOT NULL DEFAULT 'PENDING' COMMENT '작업 상태(PENDING, PROCESSING, DONE, FAILED)',
+    retry_count           INT          NOT NULL DEFAULT 0 COMMENT '재시도 횟수',
+    next_retry_at         TIMESTAMP    NULL COMMENT '다음 재시도 가능 시각',
+    processing_started_at TIMESTAMP    NULL COMMENT 'PROCESSING 상태로 선점한 시각',
+    processed_at          TIMESTAMP    NULL COMMENT '지급 성공 처리 시각',
+    last_error            TEXT         NULL COMMENT '마지막 실패 사유',
+    created_at            TIMESTAMP    NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '생성 시각',
+    updated_at            TIMESTAMP    NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '수정 시각',
+    PRIMARY KEY (id),
+    UNIQUE KEY uk_charge_event_job_idempotency_key (idempotency_key),
+    KEY idx_charge_event_job_status_next_retry_at (status, next_retry_at),
+    KEY idx_charge_event_job_source_charge_id (source_charge_id),
+    KEY idx_charge_event_job_result_charge_id (result_charge_id),
+    KEY idx_charge_event_job_member_id (member_id),
+    KEY idx_charge_event_job_charge_event_id (charge_event_id),
+    CONSTRAINT fk_charge_event_job_source_charge_id
+        FOREIGN KEY (source_charge_id) REFERENCES charge (id),
+    CONSTRAINT fk_charge_event_job_result_charge_id
+        FOREIGN KEY (result_charge_id) REFERENCES charge (id),
+    CONSTRAINT fk_charge_event_job_member_id
+        FOREIGN KEY (member_id) REFERENCES member (id),
+    CONSTRAINT fk_charge_event_job_charge_event_id
+        FOREIGN KEY (charge_event_id) REFERENCES charge_event (id)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='충전 이벤트 보너스 지급 작업';
--- a/src/main/kotlin/kr/co/vividnext/sodalive/admin/event/charge/AdminChargeEventJobController.kt
+++ b/src/main/kotlin/kr/co/vividnext/sodalive/admin/event/charge/AdminChargeEventJobController.kt
@@ -0,0 +1,25 @@
+package kr.co.vividnext.sodalive.admin.event.charge
+
+import kr.co.vividnext.sodalive.common.ApiResponse
+import org.springframework.security.access.prepost.PreAuthorize
+import org.springframework.web.bind.annotation.GetMapping
+import org.springframework.web.bind.annotation.PathVariable
+import org.springframework.web.bind.annotation.PostMapping
+import org.springframework.web.bind.annotation.RequestMapping
+import org.springframework.web.bind.annotation.RestController
+
+@RestController
+@RequestMapping("/admin/charge/event-jobs")
+@PreAuthorize("hasRole('ADMIN')")
+class AdminChargeEventJobController(
+    private val service: AdminChargeEventJobService
+) {
+    @GetMapping
+    fun getJobs() = ApiResponse.ok(service.getJobs())
+
+    @PostMapping("/{jobId}/retry")
+    fun retry(@PathVariable jobId: Long): ApiResponse<Unit> {
+        service.retry(jobId)
+        return ApiResponse.ok(Unit)
+    }
+}
--- a/src/main/kotlin/kr/co/vividnext/sodalive/admin/event/charge/AdminChargeEventJobResponse.kt
+++ b/src/main/kotlin/kr/co/vividnext/sodalive/admin/event/charge/AdminChargeEventJobResponse.kt
@@ -0,0 +1,42 @@
+package kr.co.vividnext.sodalive.admin.event.charge
+
+import kr.co.vividnext.sodalive.can.charge.event.ChargeEventJob
+import kr.co.vividnext.sodalive.can.charge.event.ChargeEventJobStatus
+import kr.co.vividnext.sodalive.can.charge.event.ChargeEventJobType
+import java.time.LocalDateTime
+
+data class AdminChargeEventJobResponse(
+    val id: Long,
+    val sourceChargeId: Long,
+    val resultChargeId: Long?,
+    val memberId: Long,
+    val chargeEventId: Long?,
+    val jobType: ChargeEventJobType,
+    val additionalCan: Int,
+    val status: ChargeEventJobStatus,
+    val retryCount: Int,
+    val nextRetryAt: LocalDateTime?,
+    val lastError: String?,
+    val createdAt: LocalDateTime?,
+    val updatedAt: LocalDateTime?
+) {
+    companion object {
+        fun from(job: ChargeEventJob): AdminChargeEventJobResponse {
+            return AdminChargeEventJobResponse(
+                id = job.id!!,
+                sourceChargeId = job.sourceChargeId,
+                resultChargeId = job.resultChargeId,
+                memberId = job.memberId,
+                chargeEventId = job.chargeEventId,
+                jobType = job.jobType,
+                additionalCan = job.additionalCan,
+                status = job.status,
+                retryCount = job.retryCount,
+                nextRetryAt = job.nextRetryAt,
+                lastError = job.lastError,
+                createdAt = job.createdAt,
+                updatedAt = job.updatedAt
+            )
+        }
+    }
+}
--- a/src/main/kotlin/kr/co/vividnext/sodalive/admin/event/charge/AdminChargeEventJobService.kt
+++ b/src/main/kotlin/kr/co/vividnext/sodalive/admin/event/charge/AdminChargeEventJobService.kt
@@ -0,0 +1,28 @@
+package kr.co.vividnext.sodalive.admin.event.charge
+
+import kr.co.vividnext.sodalive.can.charge.event.ChargeEventJobRepository
+import kr.co.vividnext.sodalive.can.charge.event.ChargeEventJobStatus
+import org.springframework.stereotype.Service
+import org.springframework.transaction.annotation.Transactional
+import java.time.LocalDateTime
+
+@Service
+@Transactional(readOnly = true)
+class AdminChargeEventJobService(
+    private val repository: ChargeEventJobRepository
+) {
+    fun getJobs(): List<AdminChargeEventJobResponse> {
+        return repository.findVisibleAdminJobs().map(AdminChargeEventJobResponse::from)
+    }
+
+    @Transactional
+    fun retry(jobId: Long) {
+        val job = repository.findByIdForUpdate(jobId) ?: return
+        if (job.status != ChargeEventJobStatus.FAILED) return
+
+        job.status = ChargeEventJobStatus.PENDING
+        job.retryCount = 0
+        job.nextRetryAt = LocalDateTime.now()
+        job.lastError = null
+    }
+}
--- a/src/main/kotlin/kr/co/vividnext/sodalive/can/charge/ChargeRepository.kt
+++ b/src/main/kotlin/kr/co/vividnext/sodalive/can/charge/ChargeRepository.kt
@@ -8,11 +8,19 @@ import kr.co.vividnext.sodalive.can.payment.PaymentStatus
 import kr.co.vividnext.sodalive.can.payment.QPayment.payment
 import kr.co.vividnext.sodalive.member.QMember.member
 import org.springframework.data.jpa.repository.JpaRepository
+import org.springframework.data.jpa.repository.Lock
+import org.springframework.data.jpa.repository.Query
+import org.springframework.data.repository.query.Param
 import org.springframework.stereotype.Repository
 import java.time.LocalDateTime
+import javax.persistence.LockModeType

@Repository
-interface ChargeRepository : JpaRepository<Charge, Long>, ChargeQueryRepository
+interface ChargeRepository : JpaRepository<Charge, Long>, ChargeQueryRepository {
+    @Lock(LockModeType.PESSIMISTIC_WRITE)
+    @Query("select c from Charge c where c.id = :chargeId")
+    fun findByIdForUpdate(@Param("chargeId") chargeId: Long): Charge?
+}

 interface ChargeQueryRepository {
    fun getOldestChargeWhereRewardCanGreaterThan0(chargeId: Long, memberId: Long, container: String): Charge?
--- a/src/main/kotlin/kr/co/vividnext/sodalive/can/charge/ChargeService.kt
+++ b/src/main/kotlin/kr/co/vividnext/sodalive/can/charge/ChargeService.kt
@@ -3,7 +3,7 @@ package kr.co.vividnext.sodalive.can.charge
 import com.fasterxml.jackson.databind.ObjectMapper
 import kr.co.bootpay.Bootpay
 import kr.co.vividnext.sodalive.can.CanRepository
-import kr.co.vividnext.sodalive.can.charge.event.ChargeSpringEvent
+import kr.co.vividnext.sodalive.can.charge.event.ChargeEventJobService
 import kr.co.vividnext.sodalive.can.coupon.CanCouponNumberRepository
 import kr.co.vividnext.sodalive.can.coupon.CouponType
 import kr.co.vividnext.sodalive.can.payment.Payment
@@ -27,7 +27,6 @@ import okhttp3.RequestBody.Companion.toRequestBody
 import org.apache.commons.codec.digest.DigestUtils
 import org.json.JSONObject
 import org.springframework.beans.factory.annotation.Value
-import org.springframework.context.ApplicationEventPublisher
 import org.springframework.data.repository.findByIdOrNull
 import org.springframework.http.HttpHeaders
 import org.springframework.retry.annotation.Backoff
@@ -52,7 +51,7 @@ class ChargeService(

    private val objectMapper: ObjectMapper,
    private val okHttpClient: OkHttpClient,
-    private val applicationEventPublisher: ApplicationEventPublisher,
+    private val chargeEventJobService: ChargeEventJobService,

    private val googlePlayService: GooglePlayService,
    private val messageSource: SodaMessageSource,
@@ -102,7 +101,7 @@ class ChargeService(
    @Transactional
    fun payverseWebhook(request: PayverseWebhookRequest): Boolean {
        val chargeId = request.orderId.toLongOrNull() ?: return false
-        val charge = chargeRepository.findByIdOrNull(chargeId) ?: return false
+        val charge = chargeRepository.findByIdForUpdate(chargeId) ?: return false

        // 결제수단 확인
        if (charge.payment?.paymentGateway != PaymentGateway.PAYVERSE) {
@@ -154,15 +153,9 @@ class ChargeService(
                    charge.payment?.status = PaymentStatus.COMPLETE
                    charge.payment?.locale = request.requestCurrency

-                    val member = charge.member!!
+                    val member = memberRepository.findByIdForUpdate(charge.member!!.id!!) ?: return false
                    member.charge(charge.chargeCan, charge.rewardCan, "pg")
-
-                    applicationEventPublisher.publishEvent(
-                        ChargeSpringEvent(
-                            chargeId = charge.id!!,
-                            memberId = member.id!!
-                        )
-                    )
+                    chargeEventJobService.createAndProcessImmediate(charge.id!!, member.id!!)
                    true
                } else {
                    false
@@ -307,9 +300,9 @@ class ChargeService(

    @Transactional
    fun payverseVerify(memberId: Long, verifyRequest: PayverseVerifyRequest): ChargeCompleteResponse {
-        val charge = chargeRepository.findByIdOrNull(verifyRequest.orderId.toLong())
+        val charge = chargeRepository.findByIdForUpdate(verifyRequest.orderId.toLong())
            ?: throw SodaException(messageKey = "can.charge.invalid_payment_info")
-        val member = memberRepository.findByIdOrNull(memberId)
+        val member = memberRepository.findByIdForUpdate(memberId)
            ?: throw SodaException(messageKey = "common.error.bad_credentials")

        val currency = charge.can?.currency
@@ -372,13 +365,7 @@ class ChargeService(
                        charge.payment?.locale = verifyResponse.requestCurrency

                        member.charge(charge.chargeCan, charge.rewardCan, "pg")
-
-                        applicationEventPublisher.publishEvent(
-                            ChargeSpringEvent(
-                                chargeId = charge.id!!,
-                                memberId = member.id!!
-                            )
-                        )
+                        chargeEventJobService.createAndProcessImmediate(charge.id!!, member.id!!)

                        return ChargeCompleteResponse(
                            price = charge.payment!!.price,
@@ -429,12 +416,16 @@ class ChargeService(

    @Transactional
    fun verify(memberId: Long, verifyRequest: VerifyRequest): ChargeCompleteResponse {
-        val charge = chargeRepository.findByIdOrNull(verifyRequest.orderId.toLong())
+        val charge = chargeRepository.findByIdForUpdate(verifyRequest.orderId.toLong())
            ?: throw SodaException(messageKey = "can.charge.invalid_payment_info")
-        val member = memberRepository.findByIdOrNull(memberId)
+        val member = memberRepository.findByIdForUpdate(memberId)
            ?: throw SodaException(messageKey = "common.error.bad_credentials")

        if (charge.payment!!.paymentGateway == PaymentGateway.PG) {
+            if (charge.payment?.status == PaymentStatus.COMPLETE) {
+                return completeResponse(charge, memberId)
+            }
+
            val bootpay = Bootpay(bootpayApplicationId, bootpayPrivateKey)

            try {
@@ -449,19 +440,9 @@ class ChargeService(
                    charge.payment?.method = verifyResult.method
                    charge.payment?.status = PaymentStatus.COMPLETE
                    member.charge(charge.chargeCan, charge.rewardCan, "pg")
+                    chargeEventJobService.createAndProcessImmediate(charge.id!!, member.id!!)

-                    applicationEventPublisher.publishEvent(
-                        ChargeSpringEvent(
-                            chargeId = charge.id!!,
-                            memberId = member.id!!
-                        )
-                    )
-
-                    return ChargeCompleteResponse(
-                        price = charge.payment!!.price,
-                        currencyCode = charge.payment!!.locale?.takeLast(3) ?: "KRW",
-                        isFirstCharged = chargeRepository.isFirstCharged(memberId)
-                    )
+                    return completeResponse(charge, memberId)
                } else {
                    throw SodaException(messageKey = "can.charge.invalid_payment_info")
                }
@@ -475,12 +456,16 @@ class ChargeService(

    @Transactional
    fun verifyHecto(memberId: Long, verifyRequest: VerifyRequest): ChargeCompleteResponse {
-        val charge = chargeRepository.findByIdOrNull(verifyRequest.orderId.toLong())
+        val charge = chargeRepository.findByIdForUpdate(verifyRequest.orderId.toLong())
            ?: throw SodaException(messageKey = "can.charge.invalid_payment_info")
-        val member = memberRepository.findByIdOrNull(memberId)
+        val member = memberRepository.findByIdForUpdate(memberId)
            ?: throw SodaException(messageKey = "common.error.bad_credentials")

        if (charge.payment!!.paymentGateway == PaymentGateway.PG) {
+            if (charge.payment?.status == PaymentStatus.COMPLETE) {
+                return completeResponse(charge, memberId)
+            }
+
            val bootpay = Bootpay(bootpayHectoApplicationId, bootpayHectoPrivateKey)

            try {
@@ -500,18 +485,8 @@ class ChargeService(
                    charge.payment?.status = PaymentStatus.COMPLETE
                    member.charge(charge.chargeCan, charge.rewardCan, "pg")

-                    applicationEventPublisher.publishEvent(
-                        ChargeSpringEvent(
-                            chargeId = charge.id!!,
-                            memberId = member.id!!
-                        )
-                    )
-
-                    return ChargeCompleteResponse(
-                        price = charge.payment!!.price,
-                        currencyCode = charge.payment!!.locale?.takeLast(3) ?: "KRW",
-                        isFirstCharged = chargeRepository.isFirstCharged(memberId)
-                    )
+                    chargeEventJobService.createAndProcessImmediate(charge.id!!, member.id!!)
+                    return completeResponse(charge, memberId)
                } else {
                    throw SodaException(messageKey = "can.charge.invalid_payment_info")
                }
@@ -547,12 +522,16 @@ class ChargeService(

    @Transactional
    fun appleVerify(memberId: Long, verifyRequest: AppleVerifyRequest): ChargeCompleteResponse {
-        val charge = chargeRepository.findByIdOrNull(verifyRequest.chargeId)
+        val charge = chargeRepository.findByIdForUpdate(verifyRequest.chargeId)
            ?: throw SodaException(messageKey = "can.charge.invalid_payment_info")
-        val member = memberRepository.findByIdOrNull(memberId)
+        val member = memberRepository.findByIdForUpdate(memberId)
            ?: throw SodaException(messageKey = "common.error.bad_credentials")

        if (charge.payment!!.paymentGateway == PaymentGateway.APPLE_IAP) {
+            if (charge.payment?.status == PaymentStatus.COMPLETE) {
+                return completeResponse(charge, memberId)
+            }
+
            // 검증로직
            if (requestRealServerVerify(verifyRequest)) {
                charge.payment?.receiptId = verifyRequest.receiptString
@@ -562,18 +541,8 @@ class ChargeService(
                charge.payment?.status = PaymentStatus.COMPLETE
                member.charge(charge.chargeCan, charge.rewardCan, "ios")

-                applicationEventPublisher.publishEvent(
-                    ChargeSpringEvent(
-                        chargeId = charge.id!!,
-                        memberId = member.id!!
-                    )
-                )
-
-                return ChargeCompleteResponse(
-                    price = charge.payment!!.price,
-                    currencyCode = charge.payment!!.locale?.takeLast(3) ?: "KRW",
-                    isFirstCharged = chargeRepository.isFirstCharged(memberId)
-                )
+                chargeEventJobService.createAndProcessImmediate(charge.id!!, member.id!!)
+                return completeResponse(charge, memberId)
            } else {
                throw SodaException(messageKey = "can.charge.invalid_payment_info")
            }
@@ -619,9 +588,9 @@ class ChargeService(
        productId: String,
        purchaseToken: String
    ): ChargeCompleteResponse {
-        val charge = chargeRepository.findByIdOrNull(id = chargeId)
+        val charge = chargeRepository.findByIdForUpdate(chargeId)
            ?: throw SodaException(messageKey = "can.charge.invalid_payment_info")
-        val member = memberRepository.findByIdOrNull(id = memberId)
+        val member = memberRepository.findByIdForUpdate(memberId)
            ?: throw SodaException(messageKey = "common.error.bad_credentials")

        if (charge.payment!!.status == PaymentStatus.REQUEST) {
@@ -631,21 +600,13 @@ class ChargeService(
                charge.payment!!.status = PaymentStatus.COMPLETE
                member.charge(charge.chargeCan, 0, "aos")

-                applicationEventPublisher.publishEvent(
-                    ChargeSpringEvent(
-                        chargeId = charge.id!!,
-                        memberId = member.id!!
-                    )
-                )
-
-                return ChargeCompleteResponse(
-                    price = charge.payment!!.price,
-                    currencyCode = charge.payment!!.locale?.takeLast(3) ?: "KRW",
-                    isFirstCharged = chargeRepository.isFirstCharged(memberId)
-                )
+                chargeEventJobService.createAndProcessImmediate(charge.id!!, member.id!!)
+                return completeResponse(charge, memberId)
            } else {
                throw SodaException(messageKey = "can.charge.purchase_failed_contact")
            }
+        } else if (charge.payment!!.status == PaymentStatus.COMPLETE) {
+            return completeResponse(charge, memberId)
        } else {
            throw SodaException(messageKey = "can.charge.invalid_payment_info")
        }
@@ -727,6 +688,14 @@ class ChargeService(
        return String.format(template, *args)
    }

+    private fun completeResponse(charge: Charge, memberId: Long): ChargeCompleteResponse {
+        return ChargeCompleteResponse(
+            price = charge.payment!!.price,
+            currencyCode = charge.payment!!.locale?.takeLast(3) ?: "KRW",
+            isFirstCharged = chargeRepository.isFirstCharged(memberId)
+        )
+    }
+
    // Payverse 결제수단 매핑: 특정 schemeCode는 "카드"로 표기, 아니면 null 반환
    private fun mapPayverseSchemeToMethodByCode(schemeCode: String?): String? {
        val cardCodes = setOf(
--- a/src/main/kotlin/kr/co/vividnext/sodalive/can/charge/event/ChargeEventJob.kt
+++ b/src/main/kotlin/kr/co/vividnext/sodalive/can/charge/event/ChargeEventJob.kt
@@ -0,0 +1,86 @@
+package kr.co.vividnext.sodalive.can.charge.event
+
+import kr.co.vividnext.sodalive.can.payment.PaymentGateway
+import kr.co.vividnext.sodalive.common.BaseEntity
+import java.time.LocalDateTime
+import javax.persistence.Column
+import javax.persistence.Entity
+import javax.persistence.EnumType
+import javax.persistence.Enumerated
+import javax.persistence.Table
+import javax.persistence.UniqueConstraint
+
+enum class ChargeEventJobStatus {
+    PENDING,
+    PROCESSING,
+    DONE,
+    FAILED
+}
+
+enum class ChargeEventJobType {
+    FIRST_CHARGE,
+    ACTIVE_CHARGE_EVENT
+}
+
+@Entity
+@Table(
+    name = "charge_event_job",
+    uniqueConstraints = [
+        UniqueConstraint(
+            name = "uk_charge_event_job_idempotency_key",
+            columnNames = ["idempotency_key"]
+        )
+    ]
+)
+class ChargeEventJob(
+    @Column(name = "source_charge_id", nullable = false)
+    val sourceChargeId: Long,
+
+    @Column(name = "result_charge_id")
+    var resultChargeId: Long? = null,
+
+    @Column(name = "member_id", nullable = false)
+    val memberId: Long,
+
+    @Column(name = "charge_event_id")
+    val chargeEventId: Long? = null,
+
+    @Enumerated(EnumType.STRING)
+    @Column(name = "job_type", nullable = false, length = 30)
+    val jobType: ChargeEventJobType,
+
+    @Column(name = "idempotency_key", nullable = false, length = 100)
+    val idempotencyKey: String,
+
+    @Column(name = "additional_can", nullable = false)
+    val additionalCan: Int,
+
+    @Enumerated(EnumType.STRING)
+    @Column(name = "payment_gateway", nullable = false, length = 30)
+    val paymentGateway: PaymentGateway,
+
+    @Column(name = "container", nullable = false, length = 10)
+    val container: String,
+
+    @Column(name = "method_snapshot", nullable = false, length = 100)
+    val methodSnapshot: String,
+
+    @Enumerated(EnumType.STRING)
+    @Column(name = "status", nullable = false, length = 20)
+    var status: ChargeEventJobStatus = ChargeEventJobStatus.PENDING,
+
+    @Column(name = "retry_count", nullable = false)
+    var retryCount: Int = 0,
+
+    @Column(name = "next_retry_at")
+    var nextRetryAt: LocalDateTime? = null,
+
+    @Column(name = "processing_started_at")
+    var processingStartedAt: LocalDateTime? = null,
+
+    @Column(name = "processed_at")
+    var processedAt: LocalDateTime? = null,
+
+    @Column(name = "last_error", columnDefinition = "text")
+    var lastError: String? = null
+) : BaseEntity()
--- a/src/main/kotlin/kr/co/vividnext/sodalive/can/charge/event/ChargeEventJobRepository.kt
+++ b/src/main/kotlin/kr/co/vividnext/sodalive/can/charge/event/ChargeEventJobRepository.kt
@@ -0,0 +1,56 @@
+package kr.co.vividnext.sodalive.can.charge.event
+
+import org.springframework.data.jpa.repository.JpaRepository
+import org.springframework.data.jpa.repository.Lock
+import org.springframework.data.jpa.repository.Query
+import org.springframework.data.repository.query.Param
+import java.time.LocalDateTime
+import javax.persistence.LockModeType
+
+interface ChargeEventJobRepository : JpaRepository<ChargeEventJob, Long> {
+    fun findByIdempotencyKey(idempotencyKey: String): ChargeEventJob?
+
+    @Lock(LockModeType.PESSIMISTIC_WRITE)
+    @Query("select j from ChargeEventJob j where j.id = :jobId")
+    fun findByIdForUpdate(@Param("jobId") jobId: Long): ChargeEventJob?
+
+    @Query(
+        value = """
+        select j.id
+        from charge_event_job j
+        where j.status = 'PENDING'
+          and (j.next_retry_at is null or j.next_retry_at <= :now)
+        order by j.created_at asc
+        limit :limit
+        for update skip locked
+        """,
+        nativeQuery = true
+    )
+    fun findNextPendingJobIdsForUpdate(
+        @Param("now") now: LocalDateTime,
+        @Param("limit") limit: Int
+    ): List<Long>
+
+    @Query(
+        value = """
+        select j.id
+        from charge_event_job j
+        where j.status = 'PENDING'
+          and j.job_type = 'FIRST_CHARGE'
+          and (j.next_retry_at is null or j.next_retry_at <= :now)
+        order by j.created_at asc
+        limit 1
+        """,
+        nativeQuery = true
+    )
+    fun findPendingFirstChargeJobId(@Param("now") now: LocalDateTime): Long?
+
+    @Query(
+        """
+        select j from ChargeEventJob j
+        where j.status in (kr.co.vividnext.sodalive.can.charge.event.ChargeEventJobStatus.PENDING, kr.co.vividnext.sodalive.can.charge.event.ChargeEventJobStatus.FAILED)
+        order by j.createdAt desc
+        """
+    )
+    fun findVisibleAdminJobs(): List<ChargeEventJob>
+}
--- a/src/main/kotlin/kr/co/vividnext/sodalive/can/charge/event/ChargeEventJobService.kt
+++ b/src/main/kotlin/kr/co/vividnext/sodalive/can/charge/event/ChargeEventJobService.kt
@@ -0,0 +1,222 @@
+package kr.co.vividnext.sodalive.can.charge.event
+
+import kr.co.vividnext.sodalive.can.charge.Charge
+import kr.co.vividnext.sodalive.can.charge.ChargeRepository
+import kr.co.vividnext.sodalive.can.charge.ChargeStatus
+import kr.co.vividnext.sodalive.can.payment.Payment
+import kr.co.vividnext.sodalive.can.payment.PaymentGateway
+import kr.co.vividnext.sodalive.can.payment.PaymentStatus
+import kr.co.vividnext.sodalive.common.SodaException
+import kr.co.vividnext.sodalive.fcm.FcmEvent
+import kr.co.vividnext.sodalive.fcm.FcmEventType
+import kr.co.vividnext.sodalive.fcm.notification.PushNotificationCategory
+import kr.co.vividnext.sodalive.member.Member
+import kr.co.vividnext.sodalive.member.MemberRepository
+import kr.co.vividnext.sodalive.member.auth.AuthRepository
+import org.springframework.context.ApplicationEventPublisher
+import org.springframework.dao.DataIntegrityViolationException
+import org.springframework.stereotype.Service
+import org.springframework.transaction.PlatformTransactionManager
+import org.springframework.transaction.annotation.Transactional
+import org.springframework.transaction.support.TransactionSynchronization
+import org.springframework.transaction.support.TransactionSynchronizationManager
+import org.springframework.transaction.support.TransactionTemplate
+import java.time.LocalDateTime
+import kotlin.math.ceil
+import kotlin.math.round
+
+@Service
+@Transactional(readOnly = true)
+class ChargeEventJobService(
+    private val jobRepository: ChargeEventJobRepository,
+    private val chargeRepository: ChargeRepository,
+    private val memberRepository: MemberRepository,
+    private val chargeEventRepository: ChargeEventRepository? = null,
+    private val authRepository: AuthRepository? = null,
+    private val applicationEventPublisher: ApplicationEventPublisher? = null,
+    transactionManager: PlatformTransactionManager? = null
+) {
+    private val transactionTemplate = transactionManager?.let { TransactionTemplate(it) }
+
+    @Transactional
+    fun createAndProcessImmediate(sourceChargeId: Long, memberId: Long): ChargeEventJob? {
+        val job = createProcessingJob(sourceChargeId, memberId) ?: return null
+        if (transactionTemplate == null || !TransactionSynchronizationManager.isSynchronizationActive()) {
+            processJob(job.id!!)
+            return job
+        }
+        TransactionSynchronizationManager.registerSynchronization(
+            object : TransactionSynchronization {
+                override fun afterCommit() {
+                    processImmediateAfterCommit(job.id!!)
+                }
+            }
+        )
+        return job
+    }
+
+    private fun processImmediateAfterCommit(jobId: Long) {
+        try {
+            transactionTemplate?.executeWithoutResult { processJob(jobId) }
+        } catch (ex: Exception) {
+            transactionTemplate?.executeWithoutResult { markImmediateFailure(jobId, ex) }
+        }
+    }
+
+    private fun markImmediateFailure(jobId: Long, ex: Exception) {
+        val job = jobRepository.findByIdForUpdate(jobId) ?: return
+        if (job.status == ChargeEventJobStatus.DONE) return
+        job.status = ChargeEventJobStatus.PENDING
+        job.nextRetryAt = LocalDateTime.now().plusMinutes(5)
+        job.lastError = ex.message?.take(MAX_ERROR_LENGTH)
+    }
+
+    @Transactional
+    fun createProcessingJob(sourceChargeId: Long, memberId: Long): ChargeEventJob? {
+        val charge = chargeRepository.findByIdForUpdate(sourceChargeId)
+            ?: throw SodaException(messageKey = "can.charge.event.not_applied_contact")
+        val member = memberRepository.findByIdForUpdate(memberId)
+            ?: throw SodaException(messageKey = "can.charge.event.not_applied_contact")
+        val snapshot = buildSnapshot(charge, member) ?: return null
+        val existing = jobRepository.findByIdempotencyKey(snapshot.idempotencyKey)
+        if (existing != null) return existing
+
+        return try {
+            jobRepository.saveAndFlush(snapshot)
+        } catch (_: DataIntegrityViolationException) {
+            jobRepository.findByIdempotencyKey(snapshot.idempotencyKey)
+        }
+    }
+
+    @Transactional
+    fun processJob(jobId: Long) {
+        val job = jobRepository.findByIdForUpdate(jobId) ?: return
+        if (job.status == ChargeEventJobStatus.DONE) return
+
+        val member = memberRepository.findByIdForUpdate(job.memberId)
+            ?: throw SodaException(messageKey = "can.charge.event.not_applied_contact")
+        val eventCharge = Charge(0, job.additionalCan, status = ChargeStatus.EVENT)
+        eventCharge.title = "${job.additionalCan} 캔"
+        eventCharge.member = member
+        eventCharge.payment = Payment(status = PaymentStatus.COMPLETE, paymentGateway = job.paymentGateway).also {
+            it.method = job.methodSnapshot
+        }
+        val savedCharge = chargeRepository.save(eventCharge)
+
+        member.charge(0, job.additionalCan, job.container)
+        job.status = ChargeEventJobStatus.DONE
+        job.resultChargeId = savedCharge.id
+        job.processedAt = LocalDateTime.now()
+        job.nextRetryAt = null
+        job.lastError = null
+        publishPaidNotification(job)
+    }
+
+    private fun buildSnapshot(charge: Charge, member: Member): ChargeEventJob? {
+        val paymentGateway = charge.payment?.paymentGateway
+            ?: throw SodaException(messageKey = "can.charge.event.not_applied_contact")
+        return if (member.auth != null) {
+            val authRepository = authRepository ?: return null
+            val authDate = authRepository.getOldestCreatedAtByDi(member.auth!!.di)
+            val chargeCount = authRepository.getMemberIdsByDi(member.auth!!.di)
+                .sumOf { id -> chargeRepository.getChargeCountAfterDate(memberId = id, authDate) }
+            if (chargeCount > 1) {
+                buildActiveEventSnapshot(charge, member, paymentGateway)
+            } else {
+                buildFirstChargeSnapshot(charge, member, paymentGateway)
+            }
+        } else {
+            buildActiveEventSnapshot(charge, member, paymentGateway)
+        }
+    }
+
+    private fun buildFirstChargeSnapshot(charge: Charge, member: Member, paymentGateway: PaymentGateway): ChargeEventJob {
+        val additionalCan = ceil(charge.chargeCan * 0.15).toInt()
+        return ChargeEventJob(
+            sourceChargeId = charge.id!!,
+            memberId = member.id!!,
+            chargeEventId = null,
+            jobType = ChargeEventJobType.FIRST_CHARGE,
+            idempotencyKey = idempotencyKey(charge.id!!, ChargeEventJobType.FIRST_CHARGE, null),
+            additionalCan = additionalCan,
+            paymentGateway = paymentGateway,
+            container = container(paymentGateway),
+            methodSnapshot = FIRST_CHARGE_METHOD,
+            status = ChargeEventJobStatus.PROCESSING,
+            processingStartedAt = LocalDateTime.now()
+        )
+    }
+
+    private fun buildActiveEventSnapshot(charge: Charge, member: Member, paymentGateway: PaymentGateway): ChargeEventJob? {
+        val repository = chargeEventRepository ?: return null
+        val chargeEvent = repository.getChargeEvent() ?: return null
+        val eventChargeCount = repository.getPaymentCount(
+            member = member,
+            method = chargeEvent.title,
+            startDate = chargeEvent.startDate,
+            endDate = chargeEvent.endDate
+        )
+        if (eventChargeCount >= chargeEvent.availableCount) return null
+
+        val additionalCan = round(charge.chargeCan * chargeEvent.addPercent).toInt()
+        return ChargeEventJob(
+            sourceChargeId = charge.id!!,
+            memberId = member.id!!,
+            chargeEventId = chargeEvent.id!!,
+            jobType = ChargeEventJobType.ACTIVE_CHARGE_EVENT,
+            idempotencyKey = idempotencyKey(charge.id!!, ChargeEventJobType.ACTIVE_CHARGE_EVENT, chargeEvent.id),
+            additionalCan = additionalCan,
+            paymentGateway = paymentGateway,
+            container = container(paymentGateway),
+            methodSnapshot = chargeEvent.title,
+            status = ChargeEventJobStatus.PROCESSING,
+            processingStartedAt = LocalDateTime.now()
+        )
+    }
+
+    private fun publishPaidNotification(job: ChargeEventJob) {
+        val publisher = applicationEventPublisher ?: return
+        if (job.jobType == ChargeEventJobType.FIRST_CHARGE) {
+            publisher.publishEvent(
+                FcmEvent(
+                    type = FcmEventType.INDIVIDUAL,
+                    category = PushNotificationCategory.SYSTEM,
+                    titleKey = "can.charge.event.first_title",
+                    messageKey = "can.charge.event.additional_can_paid",
+                    args = listOf(job.additionalCan),
+                    recipients = listOf(job.memberId),
+                    isAuth = null
+                )
+            )
+            return
+        }
+        publisher.publishEvent(
+            FcmEvent(
+                type = FcmEventType.INDIVIDUAL,
+                category = PushNotificationCategory.SYSTEM,
+                title = job.methodSnapshot,
+                messageKey = "can.charge.event.additional_can_paid",
+                args = listOf(job.additionalCan),
+                recipients = listOf(job.memberId),
+                isAuth = null
+            )
+        )
+    }
+
+    private fun container(paymentGateway: PaymentGateway): String {
+        return when (paymentGateway) {
+            PaymentGateway.GOOGLE_IAP -> "aos"
+            PaymentGateway.APPLE_IAP -> "ios"
+            else -> "pg"
+        }
+    }
+
+    private fun idempotencyKey(sourceChargeId: Long, jobType: ChargeEventJobType, chargeEventId: Long?): String {
+        return "charge-event:$sourceChargeId:$jobType:${chargeEventId ?: "none"}"
+    }
+
+    companion object {
+        private const val FIRST_CHARGE_METHOD = "첫 충전 이벤트"
+        private const val MAX_ERROR_LENGTH = 1000
+    }
+}
--- a/src/main/kotlin/kr/co/vividnext/sodalive/can/charge/event/ChargeEventJobWorker.kt
+++ b/src/main/kotlin/kr/co/vividnext/sodalive/can/charge/event/ChargeEventJobWorker.kt
@@ -0,0 +1,73 @@
+package kr.co.vividnext.sodalive.can.charge.event
+
+import org.slf4j.LoggerFactory
+import org.springframework.scheduling.annotation.Scheduled
+import org.springframework.stereotype.Component
+import org.springframework.transaction.PlatformTransactionManager
+import org.springframework.transaction.support.TransactionTemplate
+import java.time.LocalDateTime
+
+@Component
+class ChargeEventJobWorker(
+    private val jobRepository: ChargeEventJobRepository,
+    private val jobService: ChargeEventJobService,
+    transactionManager: PlatformTransactionManager
+) {
+    private val log = LoggerFactory.getLogger(javaClass)
+    private val transactionTemplate = TransactionTemplate(transactionManager)
+
+    @Scheduled(fixedDelayString = "\${sodalive.charge-event-job.fixed-delay-ms:300000}")
+    fun runPendingJobs() {
+        val now = LocalDateTime.now()
+        claimPendingJobs(now).forEach { jobId ->
+            try {
+                jobService.processJob(jobId)
+            } catch (ex: Exception) {
+                failJob(jobId, ex)
+            }
+        }
+    }
+
+    private fun claimPendingJobs(now: LocalDateTime): List<Long> {
+        return transactionTemplate.execute {
+            val jobIds = jobRepository.findNextPendingJobIdsForUpdate(now, BATCH_SIZE)
+            jobIds.forEach { jobId ->
+                val job = jobRepository.findById(jobId).orElse(null) ?: return@forEach
+                job.status = ChargeEventJobStatus.PROCESSING
+                job.processingStartedAt = LocalDateTime.now()
+                jobRepository.save(job)
+            }
+            jobIds
+        }.orEmpty()
+    }
+
+    private fun failJob(jobId: Long, ex: Exception) {
+        log.warn("Failed to process charge event job. jobId={}, error={}", jobId, ex.message)
+        transactionTemplate.executeWithoutResult {
+            val job = jobRepository.findById(jobId).orElse(null) ?: return@executeWithoutResult
+            job.retryCount += 1
+            job.lastError = ex.message?.take(MAX_ERROR_LENGTH)
+            if (job.retryCount >= MAX_RETRY_COUNT) {
+                job.status = ChargeEventJobStatus.FAILED
+                return@executeWithoutResult
+            }
+            job.status = ChargeEventJobStatus.PENDING
+            job.nextRetryAt = LocalDateTime.now().plusMinutes(backoffMinutes(job.retryCount))
+            jobRepository.save(job)
+        }
+    }
+
+    private fun backoffMinutes(retryCount: Int): Long {
+        return when (retryCount) {
+            1 -> 5L
+            2 -> 10L
+            else -> 15L
+        }
+    }
+
+    companion object {
+        private const val BATCH_SIZE = 30
+        private const val MAX_RETRY_COUNT = 3
+        private const val MAX_ERROR_LENGTH = 1000
+    }
+}
--- a/src/main/kotlin/kr/co/vividnext/sodalive/can/charge/event/ChargeEventRepository.kt
+++ b/src/main/kotlin/kr/co/vividnext/sodalive/can/charge/event/ChargeEventRepository.kt
@@ -3,6 +3,7 @@ package kr.co.vividnext.sodalive.can.charge.event
 import com.querydsl.jpa.impl.JPAQueryFactory
 import kr.co.vividnext.sodalive.admin.event.charge.ChargeEvent
 import kr.co.vividnext.sodalive.admin.event.charge.QChargeEvent.chargeEvent
+import kr.co.vividnext.sodalive.can.charge.ChargeStatus
 import kr.co.vividnext.sodalive.can.charge.QCharge.charge
 import kr.co.vividnext.sodalive.can.payment.PaymentStatus
 import kr.co.vividnext.sodalive.can.payment.QPayment.payment
@@ -14,6 +15,7 @@ interface ChargeEventRepository : JpaRepository<ChargeEvent, Long>, ChargeEventQ

 interface ChargeEventQueryRepository {
    fun getChargeEvent(): ChargeEvent?
+    fun existsActiveChargeEvent(): Boolean
    fun getPaymentCount(member: Member, method: String, startDate: LocalDateTime, endDate: LocalDateTime): Int
 }

@@ -31,6 +33,19 @@ class ChargeEventQueryRepositoryImpl(private val queryFactory: JPAQueryFactory)
            .fetchFirst()
    }

+    override fun existsActiveChargeEvent(): Boolean {
+        val now = LocalDateTime.now()
+        return queryFactory
+            .select(chargeEvent.id)
+            .from(chargeEvent)
+            .where(
+                chargeEvent.isActive.isTrue
+                    .and(chargeEvent.startDate.loe(now))
+                    .and(chargeEvent.endDate.goe(now))
+            )
+            .fetchFirst() != null
+    }
+
    override fun getPaymentCount(
        member: Member,
        method: String,
@@ -39,16 +54,19 @@ class ChargeEventQueryRepositoryImpl(private val queryFactory: JPAQueryFactory)
    ): Int {
        val where = charge.member.eq(member)
            .and(charge.payment.method.eq(method))
+            .and(charge.status.eq(ChargeStatus.EVENT))
+            .and(charge.createdAt.between(startDate, endDate))
            .and(
                charge.payment.status.eq(PaymentStatus.COMPLETE)
                    .or(charge.payment.status.eq(PaymentStatus.RETURN))
            )

        return queryFactory
-            .selectFrom(charge)
+            .select(charge.id.count())
+            .from(charge)
            .innerJoin(charge.payment, payment)
            .where(where)
-            .fetch()
-            .count()
+            .fetchOne()
+            ?.toInt() ?: 0
    }
 }
--- a/src/main/kotlin/kr/co/vividnext/sodalive/can/charge/event/ChargeSpringEvent.kt
+++ b/src/main/kotlin/kr/co/vividnext/sodalive/can/charge/event/ChargeSpringEvent.kt
@@ -1,21 +0,0 @@
-package kr.co.vividnext.sodalive.can.charge.event
-
-import org.springframework.scheduling.annotation.Async
-import org.springframework.stereotype.Component
-import org.springframework.transaction.event.TransactionalEventListener
-
-class ChargeSpringEvent(
-    val chargeId: Long,
-    val memberId: Long
-)
-
-@Component
-class ChargeSpringEventListener(
-    private val chargeEventService: ChargeEventService
-) {
-    @Async
-    @TransactionalEventListener
-    fun applyChargeEvent(event: ChargeSpringEvent) {
-        chargeEventService.applyChargeEvent(event.chargeId, event.memberId)
-    }
-}
--- a/src/test/kotlin/kr/co/vividnext/sodalive/admin/event/charge/AdminChargeEventJobServiceTest.kt
+++ b/src/test/kotlin/kr/co/vividnext/sodalive/admin/event/charge/AdminChargeEventJobServiceTest.kt
@@ -0,0 +1,73 @@
+package kr.co.vividnext.sodalive.admin.event.charge
+
+import kr.co.vividnext.sodalive.can.charge.event.ChargeEventJob
+import kr.co.vividnext.sodalive.can.charge.event.ChargeEventJobRepository
+import kr.co.vividnext.sodalive.can.charge.event.ChargeEventJobStatus
+import kr.co.vividnext.sodalive.can.charge.event.ChargeEventJobType
+import kr.co.vividnext.sodalive.can.payment.PaymentGateway
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Test
+import org.mockito.Mockito
+import java.time.LocalDateTime
+
+class AdminChargeEventJobServiceTest {
+    @Test
+    fun shouldReturnOnlyPendingAndFailedJobs() {
+        val repository = Mockito.mock(ChargeEventJobRepository::class.java)
+        val service = AdminChargeEventJobService(repository)
+        val pending = job(1L, ChargeEventJobStatus.PENDING)
+        val failed = job(2L, ChargeEventJobStatus.FAILED)
+
+        Mockito.`when`(repository.findVisibleAdminJobs()).thenReturn(listOf(pending, failed))
+
+        val responses = service.getJobs()
+
+        assertEquals(listOf(1L, 2L), responses.map { it.id })
+        assertEquals(listOf(ChargeEventJobStatus.PENDING, ChargeEventJobStatus.FAILED), responses.map { it.status })
+    }
+
+    @Test
+    fun shouldRetryOnlyFailedJob() {
+        val repository = Mockito.mock(ChargeEventJobRepository::class.java)
+        val service = AdminChargeEventJobService(repository)
+        val failed = job(1L, ChargeEventJobStatus.FAILED)
+
+        Mockito.`when`(repository.findByIdForUpdate(1L)).thenReturn(failed)
+
+        service.retry(1L)
+
+        assertEquals(ChargeEventJobStatus.PENDING, failed.status)
+        assertEquals(0, failed.retryCount)
+    }
+
+    @Test
+    fun shouldNotChangeNonFailedJobOnRetryRequest() {
+        val repository = Mockito.mock(ChargeEventJobRepository::class.java)
+        val service = AdminChargeEventJobService(repository)
+        val pending = job(1L, ChargeEventJobStatus.PENDING)
+
+        Mockito.`when`(repository.findByIdForUpdate(1L)).thenReturn(pending)
+
+        service.retry(1L)
+
+        assertEquals(ChargeEventJobStatus.PENDING, pending.status)
+        assertEquals(2, pending.retryCount)
+    }
+
+    private fun job(id: Long, status: ChargeEventJobStatus): ChargeEventJob {
+        return ChargeEventJob(
+            sourceChargeId = 100L,
+            memberId = 10L,
+            chargeEventId = 20L,
+            jobType = ChargeEventJobType.ACTIVE_CHARGE_EVENT,
+            idempotencyKey = "charge-event:100:ACTIVE_CHARGE_EVENT:20",
+            additionalCan = 10,
+            paymentGateway = PaymentGateway.PG,
+            container = "pg",
+            methodSnapshot = "봄 이벤트",
+            status = status,
+            retryCount = 2,
+            nextRetryAt = LocalDateTime.now()
+        ).also { it.id = id }
+    }
+}
--- a/src/test/kotlin/kr/co/vividnext/sodalive/can/charge/event/ChargeEventJobServiceTest.kt
+++ b/src/test/kotlin/kr/co/vividnext/sodalive/can/charge/event/ChargeEventJobServiceTest.kt
@@ -0,0 +1,89 @@
+package kr.co.vividnext.sodalive.can.charge.event
+
+import kr.co.vividnext.sodalive.can.charge.Charge
+import kr.co.vividnext.sodalive.can.charge.ChargeRepository
+import kr.co.vividnext.sodalive.can.charge.ChargeStatus
+import kr.co.vividnext.sodalive.can.payment.PaymentGateway
+import kr.co.vividnext.sodalive.member.Member
+import kr.co.vividnext.sodalive.member.MemberRepository
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Assertions.assertNotNull
+import org.junit.jupiter.api.Test
+import org.mockito.Mockito
+import java.time.LocalDateTime
+
+class ChargeEventJobServiceTest {
+    @Test
+    fun shouldPayBonusOnceFromJobSnapshot() {
+        val jobRepository = Mockito.mock(ChargeEventJobRepository::class.java)
+        val chargeRepository = Mockito.mock(ChargeRepository::class.java)
+        val memberRepository = Mockito.mock(MemberRepository::class.java)
+        val service = ChargeEventJobService(jobRepository, chargeRepository, memberRepository)
+        val member = member(id = 10L)
+        val job = job(id = 1L, memberId = 10L, additionalCan = 15, paymentGateway = PaymentGateway.PAYVERSE)
+
+        Mockito.`when`(jobRepository.findByIdForUpdate(1L)).thenReturn(job)
+        Mockito.`when`(memberRepository.findByIdForUpdate(10L)).thenReturn(member)
+        Mockito.`when`(chargeRepository.save(Mockito.any(Charge::class.java))).thenAnswer { invocation ->
+            (invocation.arguments[0] as Charge).also { it.id = 200L }
+        }
+
+        service.processJob(1L)
+
+        assertEquals(ChargeEventJobStatus.DONE, job.status)
+        assertEquals(15, member.pgRewardCan)
+        assertEquals(200L, job.resultChargeId)
+        assertNotNull(job.processedAt)
+        Mockito.verify(chargeRepository).save(
+            Mockito.argThat { charge ->
+                charge.status == ChargeStatus.EVENT &&
+                    charge.rewardCan == 15 &&
+                    charge.member == member &&
+                    charge.payment?.paymentGateway == PaymentGateway.PAYVERSE &&
+                    charge.payment?.method == "봄 이벤트"
+            }
+        )
+    }
+
+    @Test
+    fun shouldSkipAlreadyDoneJobWithoutPayingAgain() {
+        val jobRepository = Mockito.mock(ChargeEventJobRepository::class.java)
+        val chargeRepository = Mockito.mock(ChargeRepository::class.java)
+        val memberRepository = Mockito.mock(MemberRepository::class.java)
+        val service = ChargeEventJobService(jobRepository, chargeRepository, memberRepository)
+        val job = job(id = 1L, status = ChargeEventJobStatus.DONE)
+
+        Mockito.`when`(jobRepository.findByIdForUpdate(1L)).thenReturn(job)
+
+        service.processJob(1L)
+
+        Mockito.verify(chargeRepository, Mockito.never()).save(Mockito.any(Charge::class.java))
+        Mockito.verify(memberRepository, Mockito.never()).findByIdForUpdate(Mockito.anyLong())
+    }
+
+    private fun job(
+        id: Long,
+        memberId: Long = 10L,
+        additionalCan: Int = 10,
+        paymentGateway: PaymentGateway = PaymentGateway.PG,
+        status: ChargeEventJobStatus = ChargeEventJobStatus.PROCESSING
+    ): ChargeEventJob {
+        return ChargeEventJob(
+            sourceChargeId = 100L,
+            memberId = memberId,
+            chargeEventId = 20L,
+            jobType = ChargeEventJobType.ACTIVE_CHARGE_EVENT,
+            idempotencyKey = "charge-event:100:ACTIVE_CHARGE_EVENT:20",
+            additionalCan = additionalCan,
+            paymentGateway = paymentGateway,
+            container = "pg",
+            methodSnapshot = "봄 이벤트",
+            status = status,
+            nextRetryAt = LocalDateTime.now()
+        ).also { it.id = id }
+    }
+
+    private fun member(id: Long): Member {
+        return Member(password = "pw", nickname = "tester").also { it.id = id }
+    }
+}
--- a/src/test/kotlin/kr/co/vividnext/sodalive/can/charge/event/ChargeEventJobWorkerTest.kt
+++ b/src/test/kotlin/kr/co/vividnext/sodalive/can/charge/event/ChargeEventJobWorkerTest.kt
@@ -0,0 +1,122 @@
+package kr.co.vividnext.sodalive.can.charge.event
+
+import kr.co.vividnext.sodalive.can.payment.PaymentGateway
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Assertions.assertTrue
+import org.junit.jupiter.api.Test
+import org.mockito.Mockito
+import org.springframework.scheduling.annotation.Scheduled
+import org.springframework.transaction.support.AbstractPlatformTransactionManager
+import org.springframework.transaction.support.DefaultTransactionStatus
+import java.time.LocalDateTime
+import java.util.Optional
+
+class ChargeEventJobWorkerTest {
+    @Test
+    fun shouldRunEveryFiveMinutesByDefault() {
+        val scheduled = ChargeEventJobWorker::class.java
+            .getDeclaredMethod("runPendingJobs")
+            .getAnnotation(Scheduled::class.java)
+
+        assertEquals("\${sodalive.charge-event-job.fixed-delay-ms:300000}", scheduled.fixedDelayString)
+    }
+
+    @Test
+    fun shouldClaimAtMostThirtyPendingJobs() {
+        val repository = Mockito.mock(ChargeEventJobRepository::class.java)
+        val service = Mockito.mock(ChargeEventJobService::class.java)
+        val worker = ChargeEventJobWorker(repository, service, ChargeEventJobTestTransactionManager())
+        val jobIds = (1L..30L).toList()
+
+        Mockito.`when`(repository.findNextPendingJobIdsForUpdate(anyLocalDateTime(), Mockito.eq(30))).thenReturn(jobIds)
+        jobIds.forEach { jobId ->
+            Mockito.`when`(repository.findById(jobId)).thenReturn(Optional.of(job(jobId)))
+        }
+
+        worker.runPendingJobs()
+
+        Mockito.verify(repository).findNextPendingJobIdsForUpdate(anyLocalDateTime(), Mockito.eq(30))
+        Mockito.verify(service, Mockito.times(30)).processJob(Mockito.anyLong())
+    }
+
+    @Test
+    fun shouldReturnWhenNoPendingJobsExist() {
+        val repository = Mockito.mock(ChargeEventJobRepository::class.java)
+        val service = Mockito.mock(ChargeEventJobService::class.java)
+        val worker = ChargeEventJobWorker(repository, service, ChargeEventJobTestTransactionManager())
+
+        Mockito.`when`(repository.findNextPendingJobIdsForUpdate(anyLocalDateTime(), Mockito.eq(30))).thenReturn(emptyList())
+
+        worker.runPendingJobs()
+
+        Mockito.verify(service, Mockito.never()).processJob(Mockito.anyLong())
+    }
+
+    @Test
+    fun shouldRetryFailedJobWithBackoffAndFailAfterThirdFailure() {
+        val repository = Mockito.mock(ChargeEventJobRepository::class.java)
+        val service = Mockito.mock(ChargeEventJobService::class.java)
+        val worker = ChargeEventJobWorker(repository, service, ChargeEventJobTestTransactionManager())
+        val retryJob = job(1L, retryCount = 2)
+        val before = retryJob.nextRetryAt
+
+        Mockito.`when`(repository.findNextPendingJobIdsForUpdate(anyLocalDateTime(), Mockito.eq(30))).thenReturn(listOf(1L))
+        Mockito.`when`(repository.findById(1L)).thenReturn(Optional.of(retryJob))
+        Mockito.doThrow(IllegalStateException("bonus down")).`when`(service).processJob(1L)
+
+        worker.runPendingJobs()
+
+        assertEquals(ChargeEventJobStatus.FAILED, retryJob.status)
+        assertEquals(3, retryJob.retryCount)
+        assertEquals("bonus down", retryJob.lastError)
+        assertTrue(retryJob.nextRetryAt == before)
+    }
+
+    @Test
+    fun shouldUseFiveMinutesForFirstWorkerFailureBackoff() {
+        val repository = Mockito.mock(ChargeEventJobRepository::class.java)
+        val service = Mockito.mock(ChargeEventJobService::class.java)
+        val worker = ChargeEventJobWorker(repository, service, ChargeEventJobTestTransactionManager())
+        val retryJob = job(1L, retryCount = 0)
+        val before = LocalDateTime.now()
+
+        Mockito.`when`(repository.findNextPendingJobIdsForUpdate(anyLocalDateTime(), Mockito.eq(30))).thenReturn(listOf(1L))
+        Mockito.`when`(repository.findById(1L)).thenReturn(Optional.of(retryJob))
+        Mockito.doThrow(IllegalStateException("bonus down")).`when`(service).processJob(1L)
+
+        worker.runPendingJobs()
+
+        assertEquals(ChargeEventJobStatus.PENDING, retryJob.status)
+        assertEquals(1, retryJob.retryCount)
+        assertTrue(retryJob.nextRetryAt!!.isAfter(before.plusMinutes(4)))
+        assertTrue(retryJob.nextRetryAt!!.isBefore(LocalDateTime.now().plusMinutes(6)))
+    }
+
+    private fun job(id: Long, retryCount: Int = 0): ChargeEventJob {
+        return ChargeEventJob(
+            sourceChargeId = 100L,
+            memberId = 10L,
+            chargeEventId = 20L,
+            jobType = ChargeEventJobType.ACTIVE_CHARGE_EVENT,
+            idempotencyKey = "charge-event:100:ACTIVE_CHARGE_EVENT:20",
+            additionalCan = 10,
+            paymentGateway = PaymentGateway.PG,
+            container = "pg",
+            methodSnapshot = "봄 이벤트",
+            status = ChargeEventJobStatus.PENDING,
+            retryCount = retryCount,
+            nextRetryAt = LocalDateTime.now()
+        ).also { it.id = id }
+    }
+
+    private fun anyLocalDateTime(): LocalDateTime {
+        return Mockito.any(LocalDateTime::class.java) ?: LocalDateTime.now()
+    }
+}
+
+private class ChargeEventJobTestTransactionManager : AbstractPlatformTransactionManager() {
+    override fun doGetTransaction(): Any = Any()
+    override fun doBegin(transaction: Any, definition: org.springframework.transaction.TransactionDefinition) {}
+    override fun doCommit(status: DefaultTransactionStatus) {}
+    override fun doRollback(status: DefaultTransactionStatus) {}
+}