[feat #116] 알림 전송 배치 작업 (#142)

* fix : 테스트 버그 수정

* feat : 배치알림 조회 구현

* feat : multi job 관련 설정

* feat : 알림 전송 step에 필요한 usecase

* feat : 알림 전송 step

* feat : 알림 전송 job

* feat : Chunk size object 관리

* feat : 알림 가는 시점의 컨텐츠를 조회하는 걸로 수

* feat : AlertContent 삭제 hard delete로 변경

* feat : LocalDate Supplier 설정

* feat : 배치알림, 매핑 엔티티 저장 이벤트 로직

* feat : 배치알림, 매핑 엔티티 저장 포트, 서비

* edit : 메인 클래스 원상복구

.github/workflows/api-deploy.yml

@@ -46,6 +46,8 @@ jobs:
  echo "${{ secrets.APPLICATION_CORE }}" > ./application/src/main/resources/application-core.yml
  mkdir -p ./adapters/in-web/src/main/resources
  echo "${{ secrets.APPLICATION_IN_WEB_YML }}" > ./application/src/main/resources/application-in-web.yml
+ mkdir -p ./adapters/in-batch/src/main/resources
+ echo "${{ secrets.APPLICATION_IN_BATCH }}" > ./application/src/main/resources/application-in-batch.yml
 
  - name: Build with Gradle
  run: ./gradlew :entry:web:build --no-daemon

adapters/in-batch/src/main/kotlin/com/pokit/alert/component/AlertProcessor.kt

@@ -0,0 +1,17 @@
+package com.pokit.alert.component
+
+import com.pokit.alert.model.AlertBatch
+import com.pokit.alert.port.`in`.AlertUseCase
+import org.springframework.batch.item.ItemProcessor
+import org.springframework.stereotype.Component
+
+@Component
+class AlertProcessor(
+ private val alertUseCase: AlertUseCase,
+) : ItemProcessor<AlertBatch, AlertBatch> {
+
+ override fun process(alertBatch: AlertBatch): AlertBatch {
+ alertUseCase.sendMessage(alertBatch)
+ return alertBatch
+ }
+}

adapters/in-batch/src/main/kotlin/com/pokit/alert/component/AlertReader.kt

@@ -0,0 +1,29 @@
+package com.pokit.alert.component
+
+import com.pokit.alert.model.AlertBatch
+import com.pokit.alert.model.AlertBatchValue
+import com.pokit.alert.port.`in`.AlertUseCase
+import org.springframework.batch.item.ItemReader
+import org.springframework.stereotype.Component
+
+@Component
+class AlertReader(
+ private val alertUseCase: AlertUseCase,
+) : ItemReader<AlertBatch> {
+
+ private var currentPage: Int = 0
+ private var alertBatchList: MutableList<AlertBatch> = mutableListOf()
+
+ override fun read(): AlertBatch? {
+ if (alertBatchList.isEmpty()) {
+ val alertBatches = alertUseCase.loadAllAlertBatch(currentPage++, AlertBatchValue.CHUNK_SIZE)
+ alertBatchList.addAll(alertBatches)
+
+ if (alertBatchList.isEmpty()) {
+ return null
+ }
+ }
+
+ return alertBatchList.removeFirst()
+ }
+}

adapters/in-batch/src/main/kotlin/com/pokit/alert/component/AlertWriter.kt

@@ -0,0 +1,19 @@
+package com.pokit.alert.component
+
+import com.pokit.alert.model.AlertBatch
+import com.pokit.alert.port.`in`.AlertUseCase
+import org.springframework.batch.item.Chunk
+import org.springframework.batch.item.ItemWriter
+import org.springframework.stereotype.Component
+
+@Component
+class AlertWriter(
+ private val alertUseCase: AlertUseCase,
+) : ItemWriter<AlertBatch> {
+ override fun write(alertBatches: Chunk<out AlertBatch>) {
+ val batchIds = alertBatches.map { it.id }
+ val alertContents = alertUseCase.fetchAllAlertContent(batchIds)
+
+ alertUseCase.createAlerts(alertContents)
+ }
+}

adapters/in-batch/src/main/kotlin/com/pokit/alert/job/SendAlertConfig.kt

@@ -0,0 +1,43 @@
+package com.pokit.alert.job
+
+import com.navercorp.spring.batch.plus.kotlin.configuration.BatchDsl
+import com.navercorp.spring.batch.plus.kotlin.configuration.step.SimpleStepBuilderDsl
+import com.pokit.alert.component.AlertProcessor
+import com.pokit.alert.component.AlertReader
+import com.pokit.alert.component.AlertWriter
+import com.pokit.alert.model.AlertBatch
+import com.pokit.alert.model.AlertBatchValue
+import org.springframework.context.annotation.Bean
+import org.springframework.context.annotation.Configuration
+import org.springframework.transaction.PlatformTransactionManager
+import org.springframework.transaction.TransactionDefinition
+import org.springframework.transaction.interceptor.DefaultTransactionAttribute
+
+
+@Configuration
+class SendAlertConfig(
+ private val transactionManager: PlatformTransactionManager,
+ private val batch: BatchDsl,
+ private val alertReader: AlertReader,
+ private val alertProcessor: AlertProcessor,
+ private val alertWriter: AlertWriter
+) {
+ @Bean(name = ["sendAlertJob"])
+ fun sendAlertJob() = batch {
+ job("sendAlertJob") {
+ step(sendAlertStep())
+ }
+ }
+
+ @Bean
+ fun sendAlertStep() = batch {
+ step("sendAlertStep") {
+ chunk(AlertBatchValue.CHUNK_SIZE, transactionManager, fun SimpleStepBuilderDsl<AlertBatch, AlertBatch>.() {
+ reader(alertReader)
+ processor(alertProcessor)
+ writer(alertWriter)
+ transactionAttribute(DefaultTransactionAttribute(TransactionDefinition.PROPAGATION_NOT_SUPPORTED))
+ })
+ }
+ }
+}

adapters/in-batch/src/main/kotlin/com/pokit/alert/scheduler/SendAlertScheduler.kt

@@ -0,0 +1,32 @@
+package com.pokit.alert.scheduler
+
+import io.github.oshai.kotlinlogging.KotlinLogging
+import org.springframework.batch.core.Job
+import org.springframework.batch.core.JobParametersBuilder
+import org.springframework.batch.core.launch.JobLauncher
+import org.springframework.beans.factory.annotation.Qualifier
+import org.springframework.scheduling.annotation.Scheduled
+import org.springframework.stereotype.Component
+
+@Component
+class SendAlertScheduler(
+ private val jobLauncher: JobLauncher,
+ @Qualifier("sendAlertJob") private val sendAlertJob: Job
+) {
+ private val logger = KotlinLogging.logger { }
+
+ companion object {
+ private const val 매일_오전_8시 = "0 0 8 * * *"
+ }
+
+ @Scheduled(cron = 매일_오전_8시)
+ fun sendAlert() {
+ val jobParameters = JobParametersBuilder()
+ .addLong("run.id", System.currentTimeMillis())
+ .toJobParameters()
+
+ logger.info { "[ALERT BATCH] start daily send alert job" }
+ jobLauncher.run(sendAlertJob, jobParameters)
+ logger.info { "[ALERT BATCH] end daily send alert job" }
+ }
+}

adapters/in-batch/src/main/kotlin/com/pokit/config/SchedulerConfig.kt

@@ -0,0 +1,24 @@
+package com.pokit.config
+
+import org.springframework.context.annotation.Bean
+import org.springframework.context.annotation.Configuration
+import org.springframework.scheduling.TaskScheduler
+import org.springframework.scheduling.annotation.EnableScheduling
+import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler
+
+@Configuration
+@EnableScheduling
+class SchedulerConfig {
+ companion object {
+ private const val POOL_SIZE = 3
+ }
+
+ @Bean("schedulerTask")
+ fun taskScheduler(): TaskScheduler {
+ val executor = ThreadPoolTaskScheduler()
+ executor.poolSize = POOL_SIZE
+ executor.threadNamePrefix = "scheduler-thread-"
+ executor.initialize()
+ return executor
+ }
+}

adapters/in-batch/src/main/kotlin/com/pokit/remind/job/DailyContentConfig.kt

@@ -18,7 +18,7 @@ class DailyContentConfig(
  private val batch: BatchDsl,
 ) {
 
- @Bean
+ @Bean(name = ["updateDailyContentJob"])
  fun updateDailyContentJob(): Job = batch {
  job("updateDailyContentJob") {
  step(deleteAllStep()) {

adapters/in-batch/src/main/kotlin/com/pokit/remind/scheduler/DailyContentUpdateScheduler.kt

@@ -4,13 +4,14 @@ import io.github.oshai.kotlinlogging.KotlinLogging
 import org.springframework.batch.core.Job
 import org.springframework.batch.core.JobParametersBuilder
 import org.springframework.batch.core.launch.JobLauncher
+import org.springframework.beans.factory.annotation.Qualifier
 import org.springframework.scheduling.annotation.Scheduled
 import org.springframework.stereotype.Component
 
 @Component
 class DailyContentUpdateScheduler(
  private val jobLauncher: JobLauncher,
- private val updateDailyContent: Job,
+ @Qualifier("updateDailyContentJob") private val updateDailyContent: Job,
 ) {
  private val logger = KotlinLogging.logger { }
 

adapters/out-persistence/src/main/kotlin/com/pokit/out/persistence/alert/impl/AlertAdapter.kt

@@ -2,6 +2,8 @@ package com.pokit.out.persistence.alert.impl
 
 import com.pokit.alert.model.Alert
 import com.pokit.alert.port.out.AlertPort
+import com.pokit.out.persistence.alert.persist.AlertEntity
+import com.pokit.out.persistence.alert.persist.AlertJdbcRepository
 import com.pokit.out.persistence.alert.persist.AlertRepository
 import com.pokit.out.persistence.alert.persist.toDomain
 import org.springframework.data.domain.Pageable
@@ -11,7 +13,8 @@ import org.springframework.stereotype.Repository
 
 @Repository
 class AlertAdapter(
- private val alertRepository: AlertRepository
+ private val alertRepository: AlertRepository,
+ private val alertJdbcRepository: AlertJdbcRepository
 ) : AlertPort {
  override fun loadAllByUserId(userId: Long, pageable: Pageable): Slice<Alert> {
  return alertRepository.findAllByUserIdAndDeleted(userId, false, pageable)
@@ -28,4 +31,11 @@ class AlertAdapter(
  ?.delete()
  }
 
+ override fun persistAlerts(alerts: List<Alert>) {
+ val alertEntities = alerts.map {
+ AlertEntity.of(it)
+ }
+ alertJdbcRepository.bulkInsert(alertEntities)
+ }
+
 }

adapters/out-persistence/src/main/kotlin/com/pokit/out/persistence/alert/impl/AlertBatchAdapter.kt

@@ -0,0 +1,37 @@
+package com.pokit.out.persistence.alert.impl
+
+import com.pokit.alert.model.AlertBatch
+import com.pokit.alert.port.out.AlertBatchPort
+import com.pokit.out.persistence.alert.persist.AlertBatchEntity
+import com.pokit.out.persistence.alert.persist.AlertBatchRepository
+import com.pokit.out.persistence.alert.persist.toDomain
+import org.springframework.data.domain.Page
+import org.springframework.data.domain.Pageable
+import org.springframework.data.repository.findByIdOrNull
+import org.springframework.stereotype.Repository
+import java.time.LocalDate
+
+@Repository
+class AlertBatchAdapter(
+ private val alertBatchRepository: AlertBatchRepository
+) : AlertBatchPort {
+ override fun loadAllByShouldBeSentAt(now: LocalDate, pageable: Pageable): Page<AlertBatch> {
+ return alertBatchRepository.findAllByShouldBeSentAtAfterAndSent(now, false, pageable)
+ .map { it.toDomain() }
+ }
+
+ override fun send(alertBatch: AlertBatch) {
+ alertBatchRepository.findByIdOrNull(alertBatch.id)
+ ?.sent()
+ }
+
+ override fun loadByUserIdAndDate(userId: Long, date: LocalDate): AlertBatch? {
+ return alertBatchRepository.findByUserIdAndShouldBeSentAtAndSent(userId, date, false)
+ ?.run { toDomain() }
+ }
+
+ override fun persist(alertBatch: AlertBatch): AlertBatch {
+ val alertBatchEntity = AlertBatchEntity.of(alertBatch)
+ return alertBatchRepository.save(alertBatchEntity).toDomain()
+ }
+}

adapters/out-persistence/src/main/kotlin/com/pokit/out/persistence/alert/impl/AlertContentAdapter.kt

@@ -0,0 +1,27 @@
+package com.pokit.out.persistence.alert.impl
+
+import com.pokit.alert.model.AlertContent
+import com.pokit.alert.port.out.AlertContentPort
+import com.pokit.out.persistence.alert.persist.AlertContentEntity
+import com.pokit.out.persistence.alert.persist.AlertContentRepository
+import com.pokit.out.persistence.alert.persist.toDomain
+import org.springframework.stereotype.Repository
+
+@Repository
+class AlertContentAdapter(
+ private val alertContentRepository: AlertContentRepository
+) : AlertContentPort {
+ override fun loadAllInAlertBatchIds(ids: List<Long>): List<AlertContent> {
+ return alertContentRepository.findAllByAlertBatchIdIn(ids)
+ .map { it.toDomain() }
+ }
+
+ override fun deleteAll(ids: List<Long>) {
+ alertContentRepository.deleteAllByIdInBatch(ids)
+ }
+
+ override fun persist(alertContent: AlertContent): AlertContent {
+ val alertContentEntity = AlertContentEntity.of(alertContent)
+ return alertContentRepository.save(alertContentEntity).toDomain()
+ }
+}

adapters/out-persistence/src/main/kotlin/com/pokit/out/persistence/alert/persist/AlertBatchEntity.kt

@@ -33,3 +33,9 @@ class AlertBatchEntity(
  )
  }
 }
+
+fun AlertBatchEntity.toDomain() = AlertBatch(
+ id = this.id,
+ userId = this.userId,
+ shouldBeSentAt = this.shouldBeSentAt
+)

adapters/out-persistence/src/main/kotlin/com/pokit/out/persistence/alert/persist/AlertBatchRepository.kt

@@ -0,0 +1,12 @@
+package com.pokit.out.persistence.alert.persist
+
+import org.springframework.data.domain.Page
+import org.springframework.data.domain.Pageable
+import org.springframework.data.jpa.repository.JpaRepository
+import java.time.LocalDate
+
+interface AlertBatchRepository : JpaRepository<AlertBatchEntity, Long> {
+ fun findAllByShouldBeSentAtAfterAndSent(now: LocalDate, send: Boolean, pageable: Pageable): Page<AlertBatchEntity>
+
+ fun findByUserIdAndShouldBeSentAtAndSent(userId: Long, date: LocalDate, sent: Boolean): AlertBatchEntity?
+}

adapters/out-persistence/src/main/kotlin/com/pokit/out/persistence/alert/persist/AlertContentEntity.kt

@@ -17,18 +17,18 @@ class AlertContentEntity(
 
  @Column(name = "content_id")
  val contentId: Long,
-
- @Column(name = "is_deleted")
- var delete: Boolean = false
 ) : BaseEntity() {
- fun delete() {
- this.delete = true
- }
 
  companion object {
  fun of(alertContent: AlertContent) = AlertContentEntity(
  alertBatchId = alertContent.alertBatchId,
- contentId = alertContent.contentId
+ contentId = alertContent.contentId,
  )
  }
 }
+
+fun AlertContentEntity.toDomain() = AlertContent(
+ id = this.id,
+ alertBatchId = this.alertBatchId,
+ contentId = this.contentId,
+)

adapters/out-persistence/src/main/kotlin/com/pokit/out/persistence/alert/persist/AlertContentRepository.kt

@@ -0,0 +1,7 @@
+package com.pokit.out.persistence.alert.persist
+
+import org.springframework.data.jpa.repository.JpaRepository
+
+interface AlertContentRepository : JpaRepository<AlertContentEntity, Long> {
+ fun findAllByAlertBatchIdIn(ids: List<Long>): List<AlertContentEntity>
+}

adapters/out-persistence/src/main/kotlin/com/pokit/out/persistence/alert/persist/AlertJdbcRepository.kt

@@ -0,0 +1,5 @@
+package com.pokit.out.persistence.alert.persist
+
+interface AlertJdbcRepository {
+ fun bulkInsert(alertEntities: List<AlertEntity>)
+}

adapters/out-persistence/src/main/kotlin/com/pokit/out/persistence/alert/persist/AlertJdbcRepositoryImpl.kt

@@ -0,0 +1,31 @@
+package com.pokit.out.persistence.alert.persist
+
+import org.springframework.jdbc.core.JdbcTemplate
+import org.springframework.stereotype.Repository
+
+@Repository
+class AlertJdbcRepositoryImpl(
+ private val jdbcTemplate: JdbcTemplate
+) : AlertJdbcRepository {
+ override fun bulkInsert(alertEntities: List<AlertEntity>) {
+ val sql = """
+ INSERT INTO alert (
+ user_id, content_id, content_thumb_nail
+ , title, is_deleted, created_at, updated_at
+ ) VALUES (?, ?, ?, ?, ?, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
+ """.trimIndent()
+
+
+ val batchArgs = alertEntities.map { alert ->
+ arrayOf(
+ alert.userId,
+ alert.contentId,
+ alert.contentThumbNail,
+ alert.title,
+ alert.deleted
+ )
+ }
+
+ jdbcTemplate.batchUpdate(sql, batchArgs)
+ }
+}