# feat(reconciler): wallet ledger 정산 배치 추가 + remittance wallet_transactions 기록

README.md

@@ -80,6 +80,43 @@ worker가 송금사 호출 결과를 `KafkaTemplate.send(...)`로 바로 발행
 
 따라서 worker도 결과를 자체 `payout_outbox`에 트랜잭션 INSERT 하고 Debezium이 Kafka로 흘리는 동일한 패턴을 적용합니다. 송금사 호출 후 worker가 죽어도 outbox 행은 DB에 살아있고, 재기동 시 또는 다음 binlog tail에서 자연스럽게 발행됩니다.
 
+## 정산 배치 (`reconciler`) — 왜 A·B 두 검증을 모두 돌리는가
+
+`잔액 = Σ(거래 내역)` 이라는 한 줄짜리 정합성 명제는 실제 시스템에서는 **같은 BigDecimal 비교지만 잡는 결함의 종류가 서로 다른 두 검증** 으로 쪼개집니다. reconciler 는 매일 두 검증을 모두 수행해 `reconciliations.mismatch_count` 에 기록합니다.
+
+**A-검증 (총합 무결성)**
+
+```
+wallet.balance == Σ wallet_transactions.amount   // 양수=입금, 음수=출금
+```
+
+지갑의 모든 ledger row 의 amount 합과 현재 잔액이 일치하는지. **누락되거나 오기록된 ledger row** 를 잡습니다. 예: 송금 use case 에서 wallet.withdraw 만 하고 ledger insert 가 빠진 코드 경로 — 잔액은 차감됐지만 ledger row 가 없어 합이 안 맞습니다.
+
+**B-검증 (마지막 갱신 무결성)**
+
+```
+wallet.balance == 마지막 wallet_transactions.balance_after   // id DESC LIMIT 1
+```
+
+가장 최근 ledger row 의 `balance_after` 스냅샷과 현재 잔액이 일치하는지. **race 로 인한 last-write 손실, 트랜잭션 외부에서 일어난 balance 직접 수정** 을 잡습니다. 예: 운영자가 SQL 콘솔에서 `UPDATE wallets SET balance = ...` 를 직접 친 경우 — ledger 합은 그대로지만 마지막 balance_after 와 어긋납니다.
+
+**왜 둘 중 하나만으로는 부족한가**
+
+| 결함 시나리오 | A 만 돌리면 | B 만 돌리면 | A+B 둘 다 |
+|---|---|---|---|
+| ledger row 1건 누락 | 탐지 (합 불일치) | 탐지 못 함 (마지막 balance_after 는 우연히 일치 가능) | 탐지 |
+| balance 직접 SQL 수정 | 탐지 못 함 (합은 그대로) | 탐지 (마지막 balance_after 와 어긋남) | 탐지 |
+| 양쪽 모두 부정합 | 탐지 | 탐지 | 탐지 (두 플래그 모두 켜짐) |
+
+A 와 B 는 **서로 다른 클래스의 결함** 을 잡습니다. ledger 가 단일 출처(single source of truth)라면 둘이 동일해 보이지만, 실제 incident 는 ledger 외부 경로(콘솔, 데이터 패치, 잘못된 마이그레이션, race condition)에서도 발생합니다. 두 검증 모두 통과해야 비로소 "잔액 = Σ(거래 내역)" 의 의미적 보장이 성립합니다.
+
+mismatch 가 발견되어도 잡 자체는 SUCCESS 로 종료합니다. `reconciliations.mismatch_count > 0` 을 별도 alerting 트리거로 쓰는 편이, "잡 결함" 과 "데이터 결함" 을 메타데이터 측면에서 구분할 수 있어 운영상 명확합니다.
+
+실행:
+```bash
+./gradlew :reconciler:bootRun         # 매일 04:00 KST 자동 (openremit.reconcile.cron 으로 변경)
+```
+
 ## 테스트
 
 ```bash

reconciler/build.gradle.kts

@@ -1,6 +1,30 @@
-// 정산 배치 (Spring Batch)
-// Day 10에 Spring Boot 플러그인 + 배치 구성 추가 예정.
+plugins {
+    id("org.springframework.boot")
+    kotlin("plugin.jpa")
+}
+
+allOpen {
+    annotation("jakarta.persistence.Entity")
+    annotation("jakarta.persistence.MappedSuperclass")
+    annotation("jakarta.persistence.Embeddable")
+}
 
 dependencies {
     implementation(project(":common"))
+
+    implementation("org.springframework.boot:spring-boot-starter-actuator")
+    implementation("org.springframework.boot:spring-boot-starter-data-jpa")
+    implementation("org.springframework.boot:spring-boot-starter-flyway")
+    implementation("org.springframework.boot:spring-boot-starter-batch")
+    implementation("org.springframework.boot:spring-boot-starter-json")
+    implementation("org.flywaydb:flyway-mysql")
+    runtimeOnly("com.mysql:mysql-connector-j")
+
+    testImplementation("org.springframework.boot:spring-boot-starter-actuator-test")
+    testImplementation("org.springframework.boot:spring-boot-starter-data-jpa-test")
+    testImplementation("org.springframework.boot:spring-boot-starter-flyway-test")
+    testImplementation("org.springframework.boot:spring-boot-testcontainers")
+    testImplementation("org.springframework.boot:spring-boot-starter-batch-test")
+    testImplementation("org.testcontainers:junit-jupiter")
+    testImplementation("org.testcontainers:mysql")
 }

reconciler/src/main/kotlin/com/openremit/reconcile/ReconcilerApplication.kt

@@ -0,0 +1,13 @@
+package com.openremit.reconcile
+
+import org.springframework.boot.autoconfigure.SpringBootApplication
+import org.springframework.boot.runApplication
+import org.springframework.scheduling.annotation.EnableScheduling
+
+@SpringBootApplication
+@EnableScheduling
+class ReconcilerApplication
+
+fun main(args: Array<String>) {
+    runApplication<ReconcilerApplication>(*args)
+}

reconciler/src/main/kotlin/com/openremit/reconcile/application/WalletReconcile.kt

@@ -0,0 +1,32 @@
+package com.openremit.reconcile.application
+
+import java.math.BigDecimal
+
+/**
+ * 단일 wallet의 정산 입력 스냅샷.
+ * - balance: 현재 wallet.balance
+ * - sumOfTransactions: 해당 wallet의 모든 wallet_transactions.amount 합 (A-검증)
+ * - lastBalanceAfter: 가장 최근 wallet_transaction의 balance_after, 거래 0건이면 null (B-검증)
+ */
+data class WalletSnapshot(
+    val walletId: Long,
+    val balance: BigDecimal,
+    val sumOfTransactions: BigDecimal,
+    val lastBalanceAfter: BigDecimal?,
+)
+
+data class WalletMismatch(
+    val walletId: Long,
+    val balance: BigDecimal,
+    val sumOfTransactions: BigDecimal,
+    val lastBalanceAfter: BigDecimal?,
+    val violatesA: Boolean,
+    val violatesB: Boolean,
+)
+
+data class ReconcileResult(
+    val totalCount: Int,
+    val mismatches: List<WalletMismatch>,
+) {
+    val mismatchCount: Int get() = mismatches.size
+}

reconciler/src/main/kotlin/com/openremit/reconcile/application/WalletReconcileService.kt

@@ -0,0 +1,42 @@
+package com.openremit.reconcile.application
+
+import org.springframework.stereotype.Service
+
+/**
+ * 정산 검증 도메인 서비스.
+ *
+ * A-검증: wallet.balance == Σ wallet_transactions.amount
+ *   - "총합 무결성" — 누락/오기록된 ledger row를 잡는다.
+ *
+ * B-검증: wallet.balance == 마지막 wallet_transaction.balance_after
+ *   - "마지막 갱신 무결성" — race로 인한 last-write 손실, 트랜잭션 외부에서 일어난
+ *     balance 직접 수정을 잡는다. 거래가 0건이면 balance 도 0이어야 한다.
+ *
+ * 두 검증 중 하나라도 위반하면 mismatch.
+ */
+@Service
+class WalletReconcileService {
+
+    fun reconcile(snapshots: List<WalletSnapshot>): ReconcileResult {
+        val mismatches = snapshots.mapNotNull { evaluate(it) }
+        return ReconcileResult(totalCount = snapshots.size, mismatches = mismatches)
+    }
+
+    private fun evaluate(s: WalletSnapshot): WalletMismatch? {
+        val violatesA = s.balance.compareTo(s.sumOfTransactions) != 0
+        val violatesB = if (s.lastBalanceAfter == null) {
+            s.balance.signum() != 0
+        } else {
+            s.balance.compareTo(s.lastBalanceAfter) != 0
+        }
+        if (!violatesA && !violatesB) return null
+        return WalletMismatch(
+            walletId = s.walletId,
+            balance = s.balance,
+            sumOfTransactions = s.sumOfTransactions,
+            lastBalanceAfter = s.lastBalanceAfter,
+            violatesA = violatesA,
+            violatesB = violatesB,
+        )
+    }
+}

reconciler/src/main/kotlin/com/openremit/reconcile/domain/Reconciliation.kt

@@ -0,0 +1,36 @@
+package com.openremit.reconcile.domain
+
+import jakarta.persistence.Column
+import jakarta.persistence.Entity
+import jakarta.persistence.GeneratedValue
+import jakarta.persistence.GenerationType
+import jakarta.persistence.Id
+import jakarta.persistence.Table
+import org.hibernate.annotations.JdbcTypeCode
+import org.hibernate.type.SqlTypes
+import java.time.Instant
+import java.time.LocalDate
+
+@Entity
+@Table(name = "reconciliations")
+class Reconciliation(
+    @Column(name = "target_date", nullable = false)
+    val targetDate: LocalDate,
+
+    @Column(name = "total_count", nullable = false)
+    val totalCount: Int,
+
+    @Column(name = "mismatch_count", nullable = false)
+    val mismatchCount: Int,
+
+    @JdbcTypeCode(SqlTypes.JSON)
+    @Column(name = "details", nullable = false, columnDefinition = "json")
+    val details: String,
+
+    @Column(name = "created_at", nullable = false)
+    val createdAt: Instant = Instant.now(),
+) {
+    @Id
+    @GeneratedValue(strategy = GenerationType.IDENTITY)
+    var id: Long = 0
+}

reconciler/src/main/kotlin/com/openremit/reconcile/infrastructure/batch/ReconcileScheduler.kt

@@ -0,0 +1,30 @@
+package com.openremit.reconcile.infrastructure.batch
+
+import org.slf4j.LoggerFactory
+import org.springframework.batch.core.job.Job
+import org.springframework.batch.core.job.parameters.JobParametersBuilder
+import org.springframework.batch.core.launch.JobOperator
+import org.springframework.beans.factory.annotation.Qualifier
+import org.springframework.scheduling.annotation.Scheduled
+import org.springframework.stereotype.Component
+
+/**
+ * 매일 정해진 시각에 정산 잡을 실행한다.
+ * 같은 잡을 같은 파라미터로는 Spring Batch 가 재실행을 막으므로 timestamp 를 파라미터로 넣어 매 실행을 구분한다.
+ */
+@Component
+class ReconcileScheduler(
+    private val jobOperator: JobOperator,
+    @Qualifier("walletReconcileJob") private val walletReconcileJob: Job,
+) {
+    private val log = LoggerFactory.getLogger(javaClass)
+
+    @Scheduled(cron = "\${openremit.reconcile.cron:0 0 4 * * *}")
+    fun run() {
+        val params = JobParametersBuilder()
+            .addLong("runAt", System.currentTimeMillis())
+            .toJobParameters()
+        log.info("triggering walletReconcileJob")
+        jobOperator.start(walletReconcileJob, params)
+    }
+}

reconciler/src/main/kotlin/com/openremit/reconcile/infrastructure/batch/WalletReconcileJobConfig.kt

@@ -0,0 +1,75 @@
+package com.openremit.reconcile.infrastructure.batch
+
+import com.openremit.reconcile.application.WalletReconcileService
+import com.openremit.reconcile.domain.Reconciliation
+import com.openremit.reconcile.infrastructure.persistence.ReconciliationRepository
+import com.openremit.reconcile.infrastructure.persistence.WalletReconcileQuery
+import org.slf4j.LoggerFactory
+import org.springframework.batch.core.job.Job
+import org.springframework.batch.core.job.builder.JobBuilder
+import org.springframework.batch.core.repository.JobRepository
+import org.springframework.batch.core.step.Step
+import org.springframework.batch.core.step.builder.StepBuilder
+import org.springframework.batch.infrastructure.repeat.RepeatStatus
+import org.springframework.context.annotation.Bean
+import org.springframework.context.annotation.Configuration
+import org.springframework.transaction.PlatformTransactionManager
+import tools.jackson.databind.ObjectMapper
+import java.time.LocalDate
+import java.time.ZoneId
+
+/**
+ * 정산 배치 Job 구성.
+ *
+ * - 단일 Tasklet Step. wallet 수가 많지 않은 시연 환경에서는 chunk 가 불필요한 오버헤드.
+ * - mismatch 발견되어도 Step 은 SUCCESS 로 종료한다 (alert 트리거는 mismatch_count > 0 으로 판단).
+ *   Step 을 FAILED 로 만들면 Spring Batch 메타데이터 측면에서 "잡 자체의 결함" 과 구분이 안 된다.
+ */
+@Configuration
+class WalletReconcileJobConfig {
+
+    @Bean
+    fun walletReconcileJob(
+        jobRepository: JobRepository,
+        walletReconcileStep: Step,
+    ): Job =
+        JobBuilder("walletReconcileJob", jobRepository)
+            .start(walletReconcileStep)
+            .build()
+
+    @Bean
+    fun walletReconcileStep(
+        jobRepository: JobRepository,
+        transactionManager: PlatformTransactionManager,
+        query: WalletReconcileQuery,
+        service: WalletReconcileService,
+        reconciliationRepository: ReconciliationRepository,
+        objectMapper: ObjectMapper,
+    ): Step {
+        val log = LoggerFactory.getLogger("walletReconcileStep")
+        return StepBuilder("walletReconcileStep", jobRepository)
+            .tasklet({ _, _ ->
+                val snapshots = query.loadSnapshots()
+                val result = service.reconcile(snapshots)
+                val details = objectMapper.writeValueAsString(result.mismatches)
+                reconciliationRepository.save(
+                    Reconciliation(
+                        targetDate = LocalDate.now(ZoneId.systemDefault()),
+                        totalCount = result.totalCount,
+                        mismatchCount = result.mismatchCount,
+                        details = details,
+                    )
+                )
+                if (result.mismatchCount > 0) {
+                    log.warn(
+                        "wallet reconciliation found {} mismatches out of {}: {}",
+                        result.mismatchCount, result.totalCount, details,
+                    )
+                } else {
+                    log.info("wallet reconciliation OK ({} wallets)", result.totalCount)
+                }
+                RepeatStatus.FINISHED
+            }, transactionManager)
+            .build()
+    }
+}

reconciler/src/main/kotlin/com/openremit/reconcile/infrastructure/persistence/ReconciliationRepository.kt

@@ -0,0 +1,6 @@
+package com.openremit.reconcile.infrastructure.persistence
+
+import com.openremit.reconcile.domain.Reconciliation
+import org.springframework.data.jpa.repository.JpaRepository
+
+interface ReconciliationRepository : JpaRepository<Reconciliation, Long>

reconciler/src/main/kotlin/com/openremit/reconcile/infrastructure/persistence/WalletReconcileQuery.kt

@@ -0,0 +1,49 @@
+package com.openremit.reconcile.infrastructure.persistence
+
+import com.openremit.reconcile.application.WalletSnapshot
+import jakarta.persistence.EntityManager
+import org.springframework.stereotype.Repository
+import java.math.BigDecimal
+
+/**
+ * 정산 잡이 wallet 정합성 검증을 위해 읽는 native 쿼리.
+ *
+ * reconciler 는 JPA 엔티티로 wallets / wallet_transactions 를 매핑하지 않는다 (DB 소유권 분리, ADR-011).
+ * 매핑하면 엔티티 정의가 reconciler / remittance-api 에 중복되거나 결합되므로,
+ * 정산 잡은 read-only native 쿼리로만 두 테이블에 접근한다.
+ */
+@Repository
+class WalletReconcileQuery(
+    private val entityManager: EntityManager,
+) {
+
+    fun loadSnapshots(): List<WalletSnapshot> {
+        // wallet 마다 1행. 거래가 없으면 sum=0, last_balance_after=null.
+        // last_balance_after: id 가 가장 큰 wallet_transactions.balance_after.
+        val sql = """
+            SELECT
+                w.id                                                AS wallet_id,
+                w.balance                                           AS balance,
+                COALESCE(SUM(wt.amount), 0)                         AS sum_amount,
+                (SELECT wt2.balance_after
+                   FROM wallet_transactions wt2
+                  WHERE wt2.wallet_id = w.id
+                  ORDER BY wt2.id DESC
+                  LIMIT 1)                                          AS last_balance_after
+            FROM wallets w
+            LEFT JOIN wallet_transactions wt ON wt.wallet_id = w.id
+            GROUP BY w.id, w.balance
+        """.trimIndent()
+
+        @Suppress("UNCHECKED_CAST")
+        val rows = entityManager.createNativeQuery(sql).resultList as List<Array<Any?>>
+        return rows.map { row ->
+            WalletSnapshot(
+                walletId = (row[0] as Number).toLong(),
+                balance = row[1] as BigDecimal,
+                sumOfTransactions = row[2] as BigDecimal,
+                lastBalanceAfter = row[3] as BigDecimal?,
+            )
+        }
+    }
+}