[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)
+    }
+}