refactor(charge): 충전 이벤트 작업 패키지를 정리한다
This commit is contained in:
@@ -3,7 +3,6 @@ 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.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
|
||||
@@ -20,6 +19,7 @@ import kr.co.vividnext.sodalive.point.MemberPointRepository
|
||||
import kr.co.vividnext.sodalive.point.PointGrantLog
|
||||
import kr.co.vividnext.sodalive.point.PointGrantLogRepository
|
||||
import kr.co.vividnext.sodalive.useraction.ActionType
|
||||
import kr.co.vividnext.sodalive.v2.can.charge.event.ChargeEventJobService
|
||||
import okhttp3.MediaType.Companion.toMediaTypeOrNull
|
||||
import okhttp3.OkHttpClient
|
||||
import okhttp3.Request
|
||||
|
||||
@@ -1,86 +0,0 @@
|
||||
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()
|
||||
@@ -1,56 +0,0 @@
|
||||
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>
|
||||
}
|
||||
@@ -1,222 +0,0 @@
|
||||
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
|
||||
}
|
||||
}
|
||||
@@ -1,73 +0,0 @@
|
||||
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
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user